Integrate native Git publication and collaboration with resident owner routing

.github/workflows/verify.yml

@@ -21,6 +21,12 @@ jobs:
           sudo apt-get update
           sudo apt-get install -y git-lfs openssh-client
           git lfs install
+      - name: Report tool versions
+        run: |
+          rustc --version --verbose
+          cargo --version
+          rustup show active-toolchain
+          git --version
       - name: Check formatting
         run: cargo fmt --all -- --check
       - name: Check lints

crates/canopy-git-format/src/pack_index/mod.rs

@@ -161,6 +161,20 @@ impl PackIndex {
         self.ids_at(0)
     }
 
+    /// Scan native offsets in hash order using one fixed page. Callers may
+    /// build an admitted disk index of packed extents without retaining an
+    /// object-count-sized heap vector or performing a hash lookup per entry.
+    pub fn offsets(&self) -> IndexOffsets<'_> {
+        IndexOffsets {
+            index: self,
+            next: 0,
+            buffer: Box::new([0; PAGE]),
+            start: 0,
+            end: 0,
+            failed: false,
+        }
+    }
+
     /// Start a bounded sequential read at a checked native ordinal. Immutable
     /// metadata shards use this to cover contiguous ranges without rescanning
     /// earlier index entries or materializing all IDs.
@@ -268,6 +282,45 @@ pub struct IndexIds<'a> {
     end: usize,
     failed: bool,
 }
+
+pub struct IndexOffsets<'a> {
+    index: &'a PackIndex,
+    next: u32,
+    buffer: Box<[u8; PAGE]>,
+    start: usize,
+    end: usize,
+    failed: bool,
+}
+impl Iterator for IndexOffsets<'_> {
+    type Item = io::Result<u64>;
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.failed || self.next == self.index.count {
+            return None;
+        }
+        if self.start == self.end {
+            let records = (self.index.count - self.next).min((PAGE / 4) as u32) as usize;
+            self.end = records * 4;
+            self.start = 0;
+            if let Err(error) = read_at(
+                &self.index.file,
+                &mut self.buffer[..self.end],
+                self.index.offsets + u64::from(self.next) * 4,
+            ) {
+                self.failed = true;
+                return Some(Err(error));
+            }
+        }
+        let mut encoded = [0; 4];
+        encoded.copy_from_slice(&self.buffer[self.start..self.start + 4]);
+        self.start += 4;
+        self.next += 1;
+        let result = self.index.offset(u32::from_be_bytes(encoded));
+        if result.is_err() {
+            self.failed = true;
+        }
+        Some(result)
+    }
+}
 impl Iterator for IndexIds<'_> {
     type Item = io::Result<ObjectId>;
     fn next(&mut self) -> Option<Self::Item> {

crates/canopy-git-format/src/pack_index/tests.rs

@@ -60,10 +60,11 @@ fn validates_empty_and_multi_page_indexes_for_both_formats() -> io::Result<()> {
             index.pack_checksum(),
             ObjectId::try_from(vec![7; format.bytes()]).unwrap()
         );
-        for (n, oid) in index.ids().enumerate() {
+        for (n, (oid, offset)) in index.ids().zip(index.offsets()).enumerate() {
             let oid = oid?;
             assert_eq!(u32::from_be_bytes(oid[..4].try_into().unwrap()), n as u32);
             assert_eq!(index.find(oid)?.unwrap().offset, 12 + n as u64);
+            assert_eq!(index.find(oid)?.unwrap().offset, offset?);
         }
         // Shards start at a native ordinal, including positions crossing the
         // iterator's buffer boundary. The end is a valid empty iterator.
@@ -97,6 +98,20 @@ fn validates_empty_and_multi_page_indexes_for_both_formats() -> io::Result<()> {
     Ok(())
 }
 
+#[test]
+fn offset_pages_cover_native_positions_across_a_page_boundary() -> io::Result<()> {
+    for format in [ObjectFormat::Sha1, ObjectFormat::Sha256] {
+        let index = open(&fixture(format, 20_001), format)?;
+        let mut count = 0;
+        for offset in index.offsets() {
+            assert_eq!(offset?, 12 + count);
+            count += 1;
+        }
+        assert_eq!(count, 20_001);
+    }
+    Ok(())
+}
+
 #[test]
 fn rejects_truncation_and_checksum_tampering() {
     for format in [ObjectFormat::Sha1, ObjectFormat::Sha256] {
@@ -159,6 +174,7 @@ fn supports_large_offsets_and_rejects_out_of_range_references() -> io::Result<()
         rehash(&mut bytes, format);
         let index = open(&bytes, format)?;
         assert_eq!(index.find(format.zero())?.unwrap().offset, 0x1_0000_0010);
+        assert_eq!(index.offsets().next().unwrap()?, 0x1_0000_0010);
         bytes[offsets..offsets + 4].copy_from_slice(&0x8000_0001_u32.to_be_bytes());
         rehash(&mut bytes, format);
         assert!(open(&bytes, format).is_err());

crates/canopy-object-storage/src/artifact.rs

@@ -142,6 +142,18 @@ impl ArtifactStore {
         size: u64,
         digest: [u8; 32],
         input: &mut (impl AsyncRead + Unpin),
+    ) -> Result<ArtifactDescriptor, ArtifactError> {
+        self.put_owned(key, size, digest, input, Arc::new(())).await
+    }
+    /// Queued hashing retains the caller's physical admission after cancellation.
+    /// The pin grants no artifact namespace or publication authority.
+    pub async fn put_owned(
+        &self,
+        key: ArtifactKey,
+        size: u64,
+        digest: [u8; 32],
+        input: &mut (impl AsyncRead + Unpin),
+        owner: Arc<dyn Send + Sync>,
     ) -> Result<ArtifactDescriptor, ArtifactError> {
         if size > MAX_ARTIFACT_BYTES {
             return Err(ArtifactError::TooLarge);
@@ -158,7 +170,8 @@ impl ArtifactStore {
             uuid::Uuid::from_bytes(key.operation),
             uuid::Uuid::new_v4()
         ));
-        let mut upload = external::Upload::new(Arc::clone(&self.store), stage).await?;
+        let mut upload =
+            external::Upload::new_owned(Arc::clone(&self.store), stage, owner.clone()).await?;
         let result = async {
             let mut hash = blake3::Hasher::new();
             let mut remaining = size;
@@ -169,7 +182,9 @@ impl ArtifactStore {
                 tokio::time::timeout(Duration::from_secs(120), input.read_exact(&mut bytes))
                     .await
                     .map_err(|_| ArtifactError::Timeout)??;
+                let activity = owner.clone();
                 let (next, part, bytes) = tokio::task::spawn_blocking(move || {
+                    let _activity = activity;
                     hash.update(&bytes);
                     let part = *blake3::hash(&bytes).as_bytes();
                     (hash, part, Bytes::from(bytes))
@@ -213,6 +228,14 @@ impl ArtifactStore {
         &self,
         key: ArtifactKey,
         descriptor: ArtifactDescriptor,
+    ) -> Result<ArtifactRead, ArtifactError> {
+        self.read_owned(key, descriptor, Arc::new(())).await
+    }
+    pub async fn read_owned(
+        &self,
+        key: ArtifactKey,
+        descriptor: ArtifactDescriptor,
+        owner: Arc<dyn Send + Sync>,
     ) -> Result<ArtifactRead, ArtifactError> {
         if descriptor.size > MAX_ARTIFACT_BYTES {
             return Err(ArtifactError::TooLarge);
@@ -238,7 +261,68 @@ impl ArtifactStore {
             descriptor,
             offset: 0,
             hash: Some(blake3::Hasher::new()),
+            owner,
+        })
+    }
+
+    /// Authenticate the manifest before reading independently selected parts.
+    /// Each returned part is checked against that manifest. This capability
+    /// does not claim to recompute the digest of the entire artifact; callers
+    /// must already hold its certified descriptor and verify decoded objects.
+    pub async fn ranges_owned(
+        &self,
+        key: ArtifactKey,
+        descriptor: ArtifactDescriptor,
+        owner: Arc<dyn Send + Sync>,
+    ) -> Result<ArtifactRanges, ArtifactError> {
+        Ok(ArtifactRanges {
+            read: self.read_owned(key, descriptor, owner).await?,
+        })
+    }
+}
+
+/// Independent bounded reads retain the caller's physical owner through
+/// provider I/O and detached hashing. Unrequested parts are never fetched.
+pub struct ArtifactRanges {
+    read: ArtifactRead,
+}
+impl ArtifactRanges {
+    pub async fn part(&self, index: u64) -> Result<Bytes, ArtifactError> {
+        self.part_owned(index, Arc::new(())).await
+    }
+    /// Cached range capabilities retain idle file admission; the active caller
+    /// supplies its work owner separately so an idle cache cannot pin a lease.
+    pub async fn part_owned(
+        &self,
+        index: u64,
+        work: Arc<dyn Send + Sync>,
+    ) -> Result<Bytes, ArtifactError> {
+        let offset = index
+            .checked_mul(PART_BYTES as u64)
+            .filter(|offset| *offset < self.read.descriptor.size)
+            .ok_or(ArtifactError::Corrupt)?;
+        let expected = self
+            .read
+            .manifest
+            .part_digest(index)
+            .ok_or(ArtifactError::Corrupt)?;
+        let bytes = external::read(
+            self.read.store.as_ref(),
+            &self.read.path,
+            &self.read.manifest,
+            self.read.descriptor.size,
+            offset,
+        )
+        .await?;
+        let owner = (self.read.owner.clone(), work);
+        tokio::task::spawn_blocking(move || {
+            let _owner = owner;
+            if blake3::hash(&bytes).as_bytes() != &expected {
+                return Err(ArtifactError::Corrupt);
+            }
+            Ok(bytes)
         })
+        .await?
     }
 }
 
@@ -251,6 +335,7 @@ pub struct ArtifactRead {
     descriptor: ArtifactDescriptor,
     offset: u64,
     hash: Option<blake3::Hasher>,
+    owner: Arc<dyn Send + Sync>,
 }
 impl ArtifactRead {
     pub fn descriptor(&self) -> ArtifactDescriptor {
@@ -273,7 +358,9 @@ impl ArtifactRead {
             .manifest
             .part_digest(self.offset / PART_BYTES as u64)
             .ok_or(ArtifactError::Corrupt)?;
+        let activity = self.owner.clone();
         let (next, valid, bytes) = tokio::task::spawn_blocking(move || {
+            let _activity = activity;
             let valid = blake3::hash(&bytes).as_bytes() == &expected;
             hash.update(&bytes);
             (hash, valid, bytes)