feat: bound odb-tigris staging sessions with a grace window
commit
6cc4229feat: bound odb-tigris staging sessions with a grace window
Correctness rule 1’s time-bounded arm: a store opted in via
with_staging_grace aborts a staging session that outlives the
window — promote refuses and the quarantine becomes reapable by
expire_stale_quarantines — rather than letting the session become
collectible mid-flight. Defaults stay unbounded, so nothing changes
for existing stores.
feat: add index_pack to the WS5 pack writer for GC rewrites
feat: implement BlobTransport/PackRegistry for shared references
Assisted-by: Claude:claude-sonnet-5
Reviews
No reviews of this commit yet — record a verdict below.
Start a review
crates/odb-tigris/src/lib.rs
@@ -63,6 +63,12 @@
pack_key: String,
idx_key: String,
object_count: u64,
+ /// When this quarantine was staged — the clock [`OdbTigris::promote`]
+ /// and [`OdbTigris::expire_stale_quarantines`] check against
+ /// `staging_grace` (`docs/scale-out.adoc`, correctness rule 1: "a
+ /// staging session that cannot complete within the grace window aborts
+ /// rather than becoming collectible mid-flight").
+ staged_at: std::time::Instant,
}
/// [`ObjectStore`] over an S3-compatible bucket, generic over its blob
@@ -77,6 +83,13 @@
hash_kind: gix_hash::Kind,
index_cache: IndexCache,
quarantines: Mutex<HashMap<QuarantineId, Quarantine>>,
+ /// A grace-based backend's staging-session deadline (`docs/
+ /// scale-out.adoc`, correctness rule 1). `None` (the default from
+ /// [`Self::new`]) means staging is unbounded, matching every existing
+ /// caller/test's assumption; only [`Self::with_staging_grace`] opts a
+ /// store into the bounded-staging behavior a cruft-based collector
+ /// needs (see `git-maintenance`, WS9).
+ staging_grace: Option<std::time::Duration>,
}
impl<T, R> OdbTigris<T, R>
@@ -95,9 +108,30 @@
hash_kind: gix_hash::Kind::Sha1,
index_cache: IndexCache::new(),
quarantines: Mutex::new(HashMap::new()),
+ staging_grace: None,
}
}
+ /// Opt this store into grace-based staging (`docs/scale-out.adoc`,
+ /// correctness rule 1): a staging session older than `grace` aborts on
+ /// [`Self::promote`] rather than promoting, and becomes eligible for
+ /// [`Self::expire_stale_quarantines`] to reap — the bounded-staging
+ /// half of causal collection safety a time-based cruft collector needs
+ /// (`crate` has no cruft collector of its own; this is the hook one
+ /// plugs into).
+ #[must_use]
+ pub fn with_staging_grace(mut self, grace: std::time::Duration) -> Self {
+ self.staging_grace = Some(grace);
+ self
+ }
+
+ /// This store's staging grace window, if any (`docs/scale-out.adoc`,
+ /// correctness rule 1) — `None` for unbounded staging, the default.
+ #[must_use]
+ pub fn staging_grace(&self) -> Option<std::time::Duration> {
+ self.staging_grace
+ }
+
fn quarantine_pack_key(&self, id: &str) -> String {
format!("{}/quarantine/{id}/pack.pack", self.repo_id)
}
@@ -114,6 +148,42 @@
format!("{}/live/{id}.idx", self.repo_id)
}
+ /// Best-effort delete of one quarantine's staged bytes, dropping it
+ /// from the in-process map regardless of whether the transport delete
+ /// succeeds — mirrors [`Self::promote`]'s own best-effort cleanup.
+ fn expire_quarantine(&self, q: &QuarantineId) -> Result<()> {
+ if let Some(quarantine) = lock(&self.quarantines).remove(q) {
+ let _ignored = self.transport.delete(&quarantine.pack_key);
+ let _ignored = self.transport.delete(&quarantine.idx_key);
+ }
+ Ok(())
+ }
+
+ /// Reap every quarantine older than [`Self::staging_grace`], if this
+ /// store has one — the actual collection pass a grace-based cruft
+ /// collector runs (`docs/scale-out.adoc`, correctness rule 1). A no-op
+ /// returning `0` when `staging_grace` is `None`. Returns how many
+ /// quarantines were expired.
+ ///
+ /// # Errors
+ ///
+ /// Never fails today (deletes are best-effort); returns `Result` for
+ /// forward compatibility with a transport that can.
+ pub fn expire_stale_quarantines(&self) -> Result<usize> {
+ let Some(grace) = self.staging_grace else {
+ return Ok(0);
+ };
+ let stale: Vec<QuarantineId> = lock(&self.quarantines)
+ .iter()
+ .filter(|(_id, quarantine)| quarantine.staged_at.elapsed() > grace)
+ .map(|(id, _quarantine)| id.clone())
+ .collect();
+ for id in &stale {
+ self.expire_quarantine(id)?;
+ }
+ Ok(stale.len())
+ }
+
/// Every registered pack's parsed index, fetched (and cached) on
/// demand. Iterated in full on every `read`/`contains` since there is
/// no multi-pack-index yet — see this crate's module doc.
@@ -205,12 +275,32 @@
pack_key: self.quarantine_pack_key(&id),
idx_key: self.quarantine_idx_key(&id),
object_count: u64::from(outcome.index.num_objects),
+ staged_at: std::time::Instant::now(),
},
);
Ok(QuarantineId::new(id))
}
fn promote(&self, q: QuarantineId) -> Result<()> {
+ // A staging session that outlived its grace window aborts rather
+ // than promotes (`docs/scale-out.adoc`, correctness rule 1): by the
+ // time a grace-based collector's deadline has passed, this session
+ // must behave as if it had already been reaped, never as a promote
+ // racing a collection pass. Peek-then-remove (rather than removing
+ // unconditionally) so an *unexpired* quarantine is unaffected by
+ // this check.
+ if let Some(grace) = self.staging_grace {
+ let expired = lock(&self.quarantines)
+ .get(&q)
+ .is_some_and(|quarantine| quarantine.staged_at.elapsed() > grace);
+ if expired {
+ let _ignored = self.expire_quarantine(&q);
+ return Err(Error::ObjectStore(format!(
+ "quarantine {q} exceeded its staging grace window and was aborted"
+ )));
+ }
+ }
+
let quarantine = lock(&self.quarantines)
.remove(&q)
.ok_or_else(|| Error::ObjectStore(format!("unknown quarantine {q}")))?;
crates/odb-tigris/src/pack_writer.rs
@@ -126,6 +126,44 @@
})
}
+/// Re-index freshly encoded, self-contained pack bytes (as produced by
+/// [`pack_whole_objects`]/[`partition_and_pack`]) into `(pack_bytes,
+/// idx_bytes)` — the shape a [`crate::registry::PackRegistry`] record
+/// needs. Reuses gitoxide's own indexer
+/// (`gix_pack::Bundle::write_to_directory`, the same call
+/// [`crate::OdbTigris::stage_pack`] makes) rather than hand-rolling a
+/// second `.idx` writer; `crate::NoThinBaseLookup` is safe to reuse here
+/// for the same reason it is safe in `stage_pack`: every pack this module
+/// writes is self-contained (whole objects only, no thin-pack bases).
+///
+/// # Errors
+///
+/// Returns an error if indexing fails or the resulting files cannot be
+/// read back.
+pub fn index_pack(pack_bytes: Vec<u8>) -> Result<(Vec<u8>, Vec<u8>)> {
+ let scratch = tempfile::tempdir()?;
+ let mut reader = std::io::BufReader::new(std::io::Cursor::new(pack_bytes));
+ let outcome = gix_pack::Bundle::write_to_directory(
+ &mut reader,
+ Some(scratch.path()),
+ &mut gix_features::progress::Discard,
+ &std::sync::atomic::AtomicBool::new(false),
+ None::<crate::NoThinBaseLookup>,
+ gix_pack::bundle::write::Options {
+ object_hash: gix_hash::Kind::Sha1,
+ ..Default::default()
+ },
+ )
+ .map_err(|error| crate::Error::ObjectStore(error.to_string()))?;
+ let data_path = outcome
+ .data_path
+ .ok_or_else(|| crate::Error::ObjectStore("pack write produced no data file".to_owned()))?;
+ let index_path = outcome
+ .index_path
+ .ok_or_else(|| crate::Error::ObjectStore("pack write produced no index file".to_owned()))?;
+ Ok((std::fs::read(data_path)?, std::fs::read(index_path)?))
+}
+
/// Encode `objects` as a version-2 pack, every entry a full base object
/// (mirrors `git-protocol::pack::build_pack`, this crate's precedent for
/// "gix-pack's writer only does whole objects" — see this module's doc
crates/odb-tigris/src/registry.rs
@@ -172,3 +172,32 @@
/// Returns an error if the registry cannot be written.
fn delete_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<()>;
}
+
+// Every method takes `&self`, so a shared reference is itself a registry —
+// lets one registry back both an `OdbTigris` and a maintenance pass (WS9's
+// GC sweeps the same registry the store reads).
+impl<R: PackRegistry + ?Sized> PackRegistry for &R {
+ fn record(&self, record: PackRecord) -> Result<()> {
+ (**self).record(record)
+ }
+
+ fn list(&self, repo_id: &str) -> Result<Vec<PackRecord>> {
+ (**self).list(repo_id)
+ }
+
+ fn delete(&self, repo_id: &str, id: &PackId) -> Result<()> {
+ (**self).delete(repo_id, id)
+ }
+
+ fn record_artifact(&self, record: ArtifactRecord) -> Result<()> {
+ (**self).record_artifact(record)
+ }
+
+ fn get_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<Option<ArtifactRecord>> {
+ (**self).get_artifact(repo_id, kind)
+ }
+
+ fn delete_artifact(&self, repo_id: &str, kind: ArtifactKind) -> Result<()> {
+ (**self).delete_artifact(repo_id, kind)
+ }
+}
crates/odb-tigris/src/transport.rs
@@ -58,6 +58,35 @@
fn copy(&self, from: &str, to: &str) -> Result<()>;
}
+// Every method takes `&self`, so a shared reference is itself a transport —
+// lets one transport back both an `OdbTigris` and a maintenance pass
+// (WS9's GC sweeps the same bucket the store reads).
+impl<T: BlobTransport + ?Sized> BlobTransport for &T {
+ fn put(&self, key: &str, bytes: Vec<u8>) -> Result<()> {
+ (**self).put(key, bytes)
+ }
+
+ fn get(&self, key: &str) -> Result<Vec<u8>> {
+ (**self).get(key)
+ }
+
+ fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {
+ (**self).get_range(key, range)
+ }
+
+ fn exists(&self, key: &str) -> Result<bool> {
+ (**self).exists(key)
+ }
+
+ fn delete(&self, key: &str) -> Result<()> {
+ (**self).delete(key)
+ }
+
+ fn copy(&self, from: &str, to: &str) -> Result<()> {
+ (**self).copy(from, to)
+ }
+}
+
/// Map any transport-level failure into [`git_backend::Error::ObjectStore`],
/// prefixed with `context` so failures are traceable to the operation that
/// caused them.