From 99e6cf39ae2b087017b8f231ac2db6aec83f9eca Mon Sep 17 00:00:00 2001 From: Thomas Petersen Date: Mon, 20 Jul 2026 17:30:27 +0200 Subject: [PATCH] perf(relay): compact Git packs before manifest limits Proactively replace accumulated delta packs with a bounded full-closure pack set so active repositories remain pushable and cheaper to hydrate. Preserve CAS safety with resource limits, fallback behavior, metrics, and updated verification docs. --- crates/buzz-relay/src/api/git/cas_publish.rs | 1108 ++++++++++++++++-- crates/buzz-relay/src/api/git/manifest.rs | 6 + crates/buzz-relay/src/metrics.rs | 20 + docs/git-on-object-storage.md | 55 +- docs/spec/GitOnObjectStore.tla | 43 +- 5 files changed, 1111 insertions(+), 121 deletions(-) diff --git a/crates/buzz-relay/src/api/git/cas_publish.rs b/crates/buzz-relay/src/api/git/cas_publish.rs index ee5b6ab273..6b0f04e7a9 100644 --- a/crates/buzz-relay/src/api/git/cas_publish.rs +++ b/crates/buzz-relay/src/api/git/cas_publish.rs @@ -9,14 +9,16 @@ //! 3. Snapshots refs + HEAD off the workspace (the receive-pack's published //! state, by which point the pre-receive hook has enforced fast-forward / //! branch-protection against the parent's refs). -//! 4. Captures the new objects as a pack via `git pack-objects --revs -//! --stdout` over `(refs_after) --not (refs_before-tips)` (§Push step 1–2). -//! Empty pack (refs-only push that doesn't introduce objects) is allowed -//! and stored with no `new_pack_keys`. +//! 4. Normally captures the new objects as a delta pack via `git pack-objects +//! --revs --stdout` over `(refs_after) --not (refs_before-tips)` (§Push +//! step 1–2). Before the manifest reaches its pack cap, compaction instead +//! captures the complete `refs_after` closure into bounded replacement +//! packs. Empty output for a ref-less repository is allowed. //! 5. `put_pack` (content-addressed, create-only, idempotent — §Push step 2). //! The key is derived from `sha256(bytes)` by the store layer. -//! 6. Composes `m_after` (parent packs ∪ new pack, parent digest, new refs) -//! via `Manifest::compose`-equivalent inline construction (§Push step 5). +//! 6. Composes `m_after`: normal pushes use parent packs ∪ new pack; +//! compaction replaces the pack list with the newly captured full closure. +//! Both retain the parent digest and post-push refs (§Push step 5). //! 7. `put_manifest` (content-addressed, create-only, idempotent — §Push //! step 6). //! 8. `put_pointer(IfMatch(e) | IfNoneMatchStar)` — the CAS (§Push step 7). @@ -59,9 +61,10 @@ //! serialization is the CAS. Adding a per-repo mutex would hide the //! exact contention `Inv_NoFork` proves safe. -use std::collections::BTreeMap; -use std::path::Path; +use std::collections::{BTreeMap, BTreeSet}; +use std::path::{Path, PathBuf}; use std::process::Stdio; +use std::time::{Duration, Instant}; use tempfile::TempDir; use tokio::io::AsyncWriteExt; @@ -69,11 +72,19 @@ use tokio::process::Command; use tracing::{debug, warn}; use crate::api::git::manifest::{ - pointer_key, Manifest, ManifestError, MANIFEST_VERSION, MAX_MANIFEST_REFS, + pointer_key, Manifest, ManifestError, MANIFEST_VERSION, MAX_MANIFEST_PACKS, MAX_MANIFEST_REFS, + PACK_COMPACTION_THRESHOLD, }; use crate::api::git::store::{CasOutcome, ETag, GitStore, Precond, StoreError}; use buzz_core::TenantContext; +const PACK_CAPTURE_TIMEOUT: Duration = Duration::from_secs(300); +const PACK_COMPACTION_OPERATION_TIMEOUT: Duration = Duration::from_secs(600); +const PACK_OBJECTS_WINDOW_MEMORY_BYTES: u64 = 64 * 1024 * 1024; +const PACK_OBJECTS_WINDOW: &str = "10"; +const MAX_COMPACTION_OBJECTS: u64 = 1_000_000; +static PACK_COMPACTION_SEMAPHORE: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1); + /// Errors `cas_publish` surfaces. Distinguished so `finalize_push` can map /// each to the right HTTP status (the spec's 412 → 409 mapping is here). #[derive(Debug, thiserror::Error)] @@ -143,6 +154,37 @@ pub struct PublishLimits { pub max_repo_bytes: u64, } +#[derive(Debug, Clone, Copy)] +struct PublishOptions { + limits: PublishLimits, + compaction_threshold: usize, +} + +struct CompactedPack { + pack_path: PathBuf, + idx_path: PathBuf, + pack_bytes: u64, +} + +struct CompactedPacks { + _tempdir: TempDir, + packs: Vec, + total_pack_bytes: u64, +} + +struct CompactionObservation { + started_at: Instant, + packs_before: usize, + packs_after: usize, + compacted_bytes: u64, +} + +struct PreparedCompaction { + pack_keys: Vec, + packs_after: usize, + compacted_bytes: u64, +} + /// Outcome of a successful CAS. Carries the composed manifest so the /// caller can derive kind:30618 against `m_after.refs` / `m_after.head` — /// these are the values that physically landed, by `Inv_RefEffectApplied`. @@ -360,6 +402,41 @@ async fn write_idx_sidecar( Ok(()) } +async fn wait_for_git_child( + child: &mut tokio::process::Child, + stdin_bytes: &[u8], + operation: &str, +) -> Result { + let result = tokio::time::timeout(PACK_CAPTURE_TIMEOUT, async { + let mut stdin = child + .stdin + .take() + .ok_or_else(|| CasError::PackCapture(format!("{operation} stdin closed")))?; + stdin + .write_all(stdin_bytes) + .await + .map_err(|e| CasError::PackCapture(format!("{operation} stdin write: {e}")))?; + drop(stdin); + child + .wait() + .await + .map_err(|e| CasError::PackCapture(format!("{operation} wait: {e}"))) + }) + .await; + match result { + Ok(status) => status, + Err(_) => { + if let Err(error) = child.kill().await { + warn!(operation, error = %error, "timed-out git pack-objects could not be killed"); + } + Err(CasError::PackCapture(format!( + "{operation} timed out after {} seconds", + PACK_CAPTURE_TIMEOUT.as_secs() + ))) + } + } +} + /// Capture the objects this push introduced as a single pack. /// /// Runs `git pack-objects --revs --stdout` reading rev-spec lines from @@ -416,36 +493,33 @@ async fn capture_pack( .reopen() .map_err(|e| CasError::PackCapture(format!("pack-objects stderr reopen: {e}")))?; + let window_memory_arg = format!("--window-memory={PACK_OBJECTS_WINDOW_MEMORY_BYTES}"); let mut cmd = Command::new("git"); - cmd.args(["pack-objects", "--revs", "--stdout", "-q"]) - .current_dir(repo_path) - .stdin(Stdio::piped()) - .stdout(Stdio::from(stdout_file)) - .stderr(Stdio::from(stderr_file)); + cmd.args([ + "pack-objects", + "--revs", + "--stdout", + "-q", + "--threads=1", + "--window", + PACK_OBJECTS_WINDOW, + &window_memory_arg, + ]) + .current_dir(repo_path) + .stdin(Stdio::piped()) + .stdout(Stdio::from(stdout_file)) + .stderr(Stdio::from(stderr_file)) + .kill_on_drop(true); super::transport::harden_git_env(&mut cmd); let mut child = cmd .spawn() .map_err(|e| CasError::PackCapture(format!("pack-objects spawn: {e}")))?; - { - let mut stdin = child - .stdin - .take() - .ok_or_else(|| CasError::PackCapture("pack-objects stdin closed".into()))?; - stdin - .write_all(stdin_lines.as_bytes()) - .await - .map_err(|e| CasError::PackCapture(format!("pack-objects stdin write: {e}")))?; - // Drop closes stdin → EOF. - } - let out = child - .wait_with_output() - .await - .map_err(|e| CasError::PackCapture(format!("pack-objects wait: {e}")))?; - if !out.status.success() { + let status = wait_for_git_child(&mut child, stdin_lines.as_bytes(), "pack-objects").await?; + if !status.success() { let stderr = read_prefix(stderr_tmp.path(), 64 * 1024).await; return Err(CasError::PackCapture(format!( "pack-objects failed: status={:?} stderr={}", - out.status.code(), + status.code(), stderr ))); } @@ -467,6 +541,311 @@ async fn capture_pack( Ok(Some(pack_bytes)) } +fn should_compact(parent_pack_count: usize, threshold: usize) -> bool { + parent_pack_count >= threshold +} + +async fn acquire_compaction_permit() -> Result, CasError> { + tokio::time::timeout(PACK_CAPTURE_TIMEOUT, PACK_COMPACTION_SEMAPHORE.acquire()) + .await + .map_err(|_| { + CasError::PackCapture(format!( + "timed out waiting {} seconds for pack compaction capacity", + PACK_CAPTURE_TIMEOUT.as_secs() + )) + })? + .map_err(|_| CasError::PackCapture("pack compaction capacity closed".into())) +} + +fn compacted_pack_set_is_usable(parent_pack_count: usize, compacted_pack_count: usize) -> bool { + compacted_pack_count < parent_pack_count + || (parent_pack_count >= MAX_MANIFEST_PACKS && compacted_pack_count <= MAX_MANIFEST_PACKS) +} + +async fn enforce_compaction_object_limit( + repo_path: &Path, + scratch_dir: &Path, +) -> Result<(), CasError> { + const MAX_COUNT_OBJECTS_OUTPUT_BYTES: u64 = 64 * 1024; + + let stdout_tmp = tempfile::NamedTempFile::new_in(scratch_dir) + .map_err(|e| CasError::PackCapture(format!("count-objects stdout tempfile: {e}")))?; + let stdout_file = stdout_tmp + .reopen() + .map_err(|e| CasError::PackCapture(format!("count-objects stdout reopen: {e}")))?; + let stderr_tmp = tempfile::NamedTempFile::new_in(scratch_dir) + .map_err(|e| CasError::PackCapture(format!("count-objects stderr tempfile: {e}")))?; + let stderr_file = stderr_tmp + .reopen() + .map_err(|e| CasError::PackCapture(format!("count-objects stderr reopen: {e}")))?; + let mut cmd = Command::new("git"); + cmd.args(["count-objects", "-v"]) + .current_dir(repo_path) + .stdin(Stdio::piped()) + .stdout(Stdio::from(stdout_file)) + .stderr(Stdio::from(stderr_file)) + .kill_on_drop(true); + super::transport::harden_git_env(&mut cmd); + let mut child = cmd + .spawn() + .map_err(|e| CasError::PackCapture(format!("count-objects spawn: {e}")))?; + let status = wait_for_git_child(&mut child, &[], "count-objects").await?; + if !status.success() { + return Err(CasError::PackCapture(format!( + "count-objects failed: status={:?} stderr={}", + status.code(), + read_prefix(stderr_tmp.path(), MAX_COUNT_OBJECTS_OUTPUT_BYTES).await + ))); + } + let output_len = tokio::fs::metadata(stdout_tmp.path()) + .await + .map_err(|e| CasError::PackCapture(format!("count-objects stdout metadata: {e}")))? + .len(); + if output_len > MAX_COUNT_OBJECTS_OUTPUT_BYTES { + return Err(CasError::ResourceLimit(format!( + "count-objects output is {output_len} bytes (max {MAX_COUNT_OBJECTS_OUTPUT_BYTES})" + ))); + } + let output = tokio::fs::read_to_string(stdout_tmp.path()) + .await + .map_err(|e| CasError::PackCapture(format!("read count-objects output: {e}")))?; + let parse_count = |name: &str| -> Result { + output + .lines() + .find_map(|line| line.strip_prefix(name)) + .and_then(|value| value.trim().parse::().ok()) + .ok_or_else(|| CasError::PackCapture(format!("count-objects omitted {name:?}"))) + }; + let loose = parse_count("count:")?; + let packed = parse_count("in-pack:")?; + let object_count = loose + .checked_add(packed) + .ok_or_else(|| CasError::ResourceLimit("compaction object count overflowed u64".into()))?; + if object_count > MAX_COMPACTION_OBJECTS { + return Err(CasError::ResourceLimit(format!( + "repository has {object_count} objects (compaction max {MAX_COMPACTION_OBJECTS})" + ))); + } + Ok(()) +} + +/// Capture the complete object closure reachable from the post-push refs. +/// +/// Unlike [`capture_pack`], this writes one or more bounded packs into a +/// private tempdir and supplies no negative revisions. The resulting pack set +/// can therefore replace the parent manifest's pack list rather than extending +/// it. Old object-store packs remain immutable and are not deleted. +async fn capture_compacted_packs( + repo_path: &Path, + refs_after: &BTreeMap, + limits: PublishLimits, + scratch_dir: &Path, +) -> Result { + let tempdir = TempDir::new_in(scratch_dir) + .map_err(|e| CasError::PackCapture(format!("compaction tempdir: {e}")))?; + if refs_after.is_empty() { + return Ok(CompactedPacks { + _tempdir: tempdir, + packs: Vec::new(), + total_pack_bytes: 0, + }); + } + enforce_compaction_object_limit(repo_path, scratch_dir).await?; + + let mut stdin_lines = String::new(); + let mut seen_oids = BTreeSet::new(); + for oid in refs_after.values() { + if seen_oids.insert(oid) { + stdin_lines.push_str(oid); + stdin_lines.push('\n'); + } + } + + let output_base = tempdir.path().join("compact"); + let output_base_str = output_base + .to_str() + .ok_or_else(|| CasError::PackCapture("compaction path is not valid utf-8".into()))?; + let max_pack_arg = format!("--max-pack-size={}", limits.max_pack_bytes); + let window_memory_arg = format!("--window-memory={PACK_OBJECTS_WINDOW_MEMORY_BYTES}"); + let stdout_tmp = tempfile::NamedTempFile::new_in(tempdir.path()) + .map_err(|e| CasError::PackCapture(format!("compaction stdout tempfile: {e}")))?; + let stdout_file = stdout_tmp + .reopen() + .map_err(|e| CasError::PackCapture(format!("compaction stdout reopen: {e}")))?; + let stderr_tmp = tempfile::NamedTempFile::new_in(tempdir.path()) + .map_err(|e| CasError::PackCapture(format!("compaction stderr tempfile: {e}")))?; + let stderr_file = stderr_tmp + .reopen() + .map_err(|e| CasError::PackCapture(format!("compaction stderr reopen: {e}")))?; + let mut cmd = Command::new("git"); + cmd.args([ + "pack-objects", + "--revs", + "-q", + "--threads=1", + "--window", + PACK_OBJECTS_WINDOW, + &window_memory_arg, + &max_pack_arg, + output_base_str, + ]) + .current_dir(repo_path) + .stdin(Stdio::piped()) + .stdout(Stdio::from(stdout_file)) + .stderr(Stdio::from(stderr_file)) + .kill_on_drop(true); + super::transport::harden_git_env(&mut cmd); + let mut child = cmd + .spawn() + .map_err(|e| CasError::PackCapture(format!("compaction pack-objects spawn: {e}")))?; + let status = wait_for_git_child( + &mut child, + stdin_lines.as_bytes(), + "compaction pack-objects", + ) + .await?; + if !status.success() { + return Err(CasError::PackCapture(format!( + "compaction pack-objects failed: status={:?} stderr={}", + status.code(), + read_prefix(stderr_tmp.path(), 64 * 1024).await + ))); + } + + let mut pack_paths = Vec::new(); + let mut entries = tokio::fs::read_dir(tempdir.path()) + .await + .map_err(|e| CasError::PackCapture(format!("read compacted pack directory: {e}")))?; + while let Some(entry) = entries + .next_entry() + .await + .map_err(|e| CasError::PackCapture(format!("read compacted pack entry: {e}")))? + { + let path = entry.path(); + if path.extension().and_then(|value| value.to_str()) == Some("pack") { + pack_paths.push(path); + } + } + pack_paths.sort(); + if pack_paths.len() > MAX_MANIFEST_PACKS { + return Err(CasError::ResourceLimit(format!( + "compaction produced {} packs (max {MAX_MANIFEST_PACKS})", + pack_paths.len() + ))); + } + + let mut packs = Vec::with_capacity(pack_paths.len()); + let mut total_pack_bytes = 0u64; + for pack_path in pack_paths { + let pack_bytes = tokio::fs::metadata(&pack_path) + .await + .map_err(|e| CasError::PackCapture(format!("stat compacted pack: {e}")))? + .len(); + if pack_bytes > limits.max_pack_bytes { + return Err(CasError::ResourceLimit(format!( + "compacted pack is {pack_bytes} bytes (max {})", + limits.max_pack_bytes + ))); + } + total_pack_bytes = total_pack_bytes + .checked_add(pack_bytes) + .ok_or_else(|| CasError::ResourceLimit("compacted byte count overflowed u64".into()))?; + if total_pack_bytes > limits.max_repo_bytes { + return Err(CasError::ResourceLimit(format!( + "compacted repo needs {total_pack_bytes} bytes (max {})", + limits.max_repo_bytes + ))); + } + let idx_path = pack_path.with_extension("idx"); + let idx_bytes = tokio::fs::metadata(&idx_path) + .await + .map_err(|e| CasError::PackCapture(format!("stat compacted idx: {e}")))? + .len(); + if idx_bytes > limits.max_pack_bytes { + return Err(CasError::ResourceLimit(format!( + "compacted idx is {idx_bytes} bytes (max {})", + limits.max_pack_bytes + ))); + } + packs.push(CompactedPack { + pack_path, + idx_path, + pack_bytes, + }); + } + + Ok(CompactedPacks { + _tempdir: tempdir, + packs, + total_pack_bytes, + }) +} + +async fn upload_compacted_packs( + store: &GitStore, + compacted: &CompactedPacks, +) -> Result, CasError> { + let mut pack_keys = Vec::with_capacity(compacted.packs.len()); + for pack in &compacted.packs { + let pack_bytes = tokio::fs::read(&pack.pack_path) + .await + .map_err(|e| CasError::PackCapture(format!("read compacted pack: {e}")))?; + if u64::try_from(pack_bytes.len()).unwrap_or(u64::MAX) != pack.pack_bytes { + return Err(CasError::PackCapture( + "compacted pack changed after size validation".into(), + )); + } + let pack_key = store.put_pack(&pack_bytes).await?; + let pack_digest = digest_from_pack_key(&pack_key)?; + match tokio::fs::read(&pack.idx_path).await { + Ok(idx_bytes) => { + if let Err(error) = store.put_idx(&pack_digest, &idx_bytes).await { + warn!( + pack_key = %pack_key, + error = %error, + "failed to write compacted git pack idx sidecar; push will continue" + ); + } + } + Err(error) => { + warn!( + pack_key = %pack_key, + error = %error, + "failed to read compacted git pack idx sidecar; push will continue" + ); + } + } + pack_keys.push(pack_key); + } + pack_keys.sort(); + pack_keys.dedup(); + Ok(pack_keys) +} + +async fn prepare_compaction( + store: &GitStore, + repo_path: &Path, + refs_after: &BTreeMap, + limits: PublishLimits, + scratch_dir: &Path, + packs_before: usize, +) -> Result { + let compacted = capture_compacted_packs(repo_path, refs_after, limits, scratch_dir).await?; + let packs_after = compacted.packs.len(); + if !compacted_pack_set_is_usable(packs_before, packs_after) { + return Err(CasError::ResourceLimit(format!( + "compaction did not reduce pack count (before {packs_before}, after {packs_after})" + ))); + } + let compacted_bytes = compacted.total_pack_bytes; + let pack_keys = upload_compacted_packs(store, &compacted).await?; + Ok(PreparedCompaction { + pack_keys, + packs_after, + compacted_bytes, + }) +} + async fn read_prefix(path: &Path, max_bytes: u64) -> String { use tokio::io::AsyncReadExt; @@ -521,6 +900,23 @@ fn compose_after( } } +fn compose_compacted_after( + parent_digest: Option, + head: String, + refs: BTreeMap, + mut compacted_pack_keys: Vec, +) -> Manifest { + compacted_pack_keys.sort(); + compacted_pack_keys.dedup(); + Manifest { + version: MANIFEST_VERSION, + head, + refs, + packs: compacted_pack_keys, + parent: parent_digest, + } +} + /// Derive `manifests/` from a returned manifest key, surfacing the /// hex digest the pointer body needs. fn digest_from_manifest_key(key: &str) -> Result { @@ -534,6 +930,25 @@ fn digest_from_manifest_key(key: &str) -> Result { }) } +fn record_compaction( + outcome: &'static str, + started_at: Instant, + packs_before: usize, + packs_after: Option, + compacted_bytes: Option, +) { + metrics::counter!("buzz_git_pack_compactions_total", "outcome" => outcome).increment(1); + metrics::histogram!("buzz_git_pack_compaction_seconds", "outcome" => outcome) + .record(started_at.elapsed().as_secs_f64()); + metrics::histogram!("buzz_git_pack_compaction_packs_before").record(packs_before as f64); + if let Some(packs_after) = packs_after { + metrics::histogram!("buzz_git_pack_compaction_packs_after").record(packs_after as f64); + } + if let Some(compacted_bytes) = compacted_bytes { + metrics::histogram!("buzz_git_pack_compaction_bytes").record(compacted_bytes as f64); + } +} + /// The function the §Push step 2–7 protocol distills to. /// /// **Caller contract — `Inv_RefDerivedFromParent` is structural.** The @@ -560,6 +975,31 @@ pub async fn cas_publish( parent_state: &ParentState, limits: PublishLimits, ) -> Result { + cas_publish_inner( + store, + ctx, + repo_path, + owner, + repo, + parent_state, + PublishOptions { + limits, + compaction_threshold: PACK_COMPACTION_THRESHOLD, + }, + ) + .await +} + +async fn cas_publish_inner( + store: &GitStore, + ctx: &TenantContext, + repo_path: &Path, + owner: &str, + repo: &str, + parent_state: &ParentState, + options: PublishOptions, +) -> Result { + let limits = options.limits; let pkey = pointer_key(ctx.community(), owner, repo); // Hydrated repositories are direct children of the configured Git scratch @@ -585,54 +1025,128 @@ pub async fn cas_publish( head_observed }; - // Capture new objects as a pack (steps 1–2). The "not" set is the - // parent manifest's refs — i.e. the set the workspace was hydrated - // against — so the delta covers exactly the objects this push - // introduced. - let pack_bytes = capture_pack( - repo_path, - &parent_state.parent.refs, - &refs_after, - limits.max_pack_bytes, - scratch_dir, - ) - .await?; - let new_pack_key = if let Some(bytes) = pack_bytes { - let new_pack_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); - let total_bytes = limits - .parent_hydrated_bytes - .checked_add(new_pack_bytes) - .ok_or_else(|| CasError::ResourceLimit("repo byte count overflowed u64".into()))?; - if total_bytes > limits.max_repo_bytes { - return Err(CasError::ResourceLimit(format!( - "repo would need {total_bytes} hydrated bytes (max {})", - limits.max_repo_bytes - ))); - } - debug!(bytes = bytes.len(), "captured push pack"); - let pack_key = store.put_pack(&bytes).await?; - if let Err(e) = write_idx_sidecar(store, &pack_key, &bytes, scratch_dir).await { - warn!( - pack_key = %pack_key, - error = %e, - "failed to write git pack idx sidecar; push will continue" - ); + let packs_before = parent_state.parent.packs.len(); + let mut compaction_failure = None; + let mut compaction_observation = None; + let mut compacted_manifest = None; + if should_compact(packs_before, options.compaction_threshold) { + let started_at = Instant::now(); + match acquire_compaction_permit().await { + Ok(_permit) => { + let operation = prepare_compaction( + store, + repo_path, + &refs_after, + limits, + scratch_dir, + packs_before, + ); + match tokio::time::timeout(PACK_COMPACTION_OPERATION_TIMEOUT, operation).await { + Ok(Ok(prepared)) => { + debug!( + packs_before, + packs_after = prepared.packs_after, + bytes = prepared.compacted_bytes, + "captured compacted repository pack set" + ); + compacted_manifest = Some(compose_compacted_after( + parent_state.parent_digest.clone(), + head.clone(), + refs_after.clone(), + prepared.pack_keys, + )); + compaction_observation = Some(CompactionObservation { + started_at, + packs_before, + packs_after: prepared.packs_after, + compacted_bytes: prepared.compacted_bytes, + }); + } + Ok(Err(error)) => compaction_failure = Some((started_at, error)), + Err(_) => { + compaction_failure = Some(( + started_at, + CasError::PackCapture(format!( + "pack compaction operation timed out after {} seconds", + PACK_COMPACTION_OPERATION_TIMEOUT.as_secs() + )), + )); + } + } + } + Err(error) => compaction_failure = Some((started_at, error)), } - Some(pack_key) + } + if let Some((started_at, error)) = &compaction_failure { + warn!( + packs_before, + error = %error, + "pack compaction failed; attempting normal delta-pack publication" + ); + record_compaction("fallback", *started_at, packs_before, None, None); + } + + let m_after = if let Some(manifest) = compacted_manifest { + manifest } else { - debug!("no new objects in push; manifest will reuse parent packs"); - None + // Capture new objects as a delta pack (steps 1–2). The "not" set is + // the parent manifest's refs — i.e. the set the workspace was hydrated + // against — so the delta covers exactly the objects this push + // introduced. + let pack_bytes = capture_pack( + repo_path, + &parent_state.parent.refs, + &refs_after, + limits.max_pack_bytes, + scratch_dir, + ) + .await?; + if pack_bytes.is_some() && packs_before >= MAX_MANIFEST_PACKS { + let error = compaction_failure + .map(|(_, error)| error) + .unwrap_or_else(|| { + CasError::ResourceLimit(format!( + "repository already names {packs_before} packs and compaction failed" + )) + }); + metrics::counter!("buzz_git_pack_compaction_required_failures_total").increment(1); + return Err(error); + } + let new_pack_key = if let Some(bytes) = pack_bytes { + let new_pack_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX); + let total_bytes = limits + .parent_hydrated_bytes + .checked_add(new_pack_bytes) + .ok_or_else(|| CasError::ResourceLimit("repo byte count overflowed u64".into()))?; + if total_bytes > limits.max_repo_bytes { + return Err(CasError::ResourceLimit(format!( + "repo would need {total_bytes} hydrated bytes (max {})", + limits.max_repo_bytes + ))); + } + debug!(bytes = bytes.len(), "captured push pack"); + let pack_key = store.put_pack(&bytes).await?; + if let Err(e) = write_idx_sidecar(store, &pack_key, &bytes, scratch_dir).await { + warn!( + pack_key = %pack_key, + error = %e, + "failed to write git pack idx sidecar; push will continue" + ); + } + Some(pack_key) + } else { + debug!("no new objects in push; manifest will reuse parent packs"); + None + }; + compose_after( + &parent_state.parent, + parent_state.parent_digest.clone(), + head, + refs_after, + new_pack_key, + ) }; - // Compose m_after (step 5). - let m_after = compose_after( - &parent_state.parent, - parent_state.parent_digest.clone(), - head, - refs_after, - new_pack_key, - ); - // **Pre-CAS validation** (Sami #2 / Max / Dawn): refuse to commit an // un-clone-able manifest. `Manifest::validate` checks every refname // against `is_safe_refname`, every oid against `is_hex_oid`, and @@ -640,11 +1154,36 @@ pub async fn cas_publish( // uses on read. Failure surfaces as `CasError::ManifestInvalid` // (4xx-class: client/input rejected) so the caller never confuses // it with `ManifestReadFailed` (5xx-class: parent corrupt). - m_after.validate()?; + if let Err(error) = m_after.validate() { + if let Some(observation) = &compaction_observation { + record_compaction( + "validation_error", + observation.started_at, + observation.packs_before, + Some(observation.packs_after), + Some(observation.compacted_bytes), + ); + } + return Err(error.into()); + } // Step 6: put_manifest. let manifest_bytes = m_after.canonical_bytes()?; - let manifest_key = store.put_manifest(&manifest_bytes).await?; + let manifest_key = match store.put_manifest(&manifest_bytes).await { + Ok(key) => key, + Err(error) => { + if let Some(observation) = &compaction_observation { + record_compaction( + "publish_error", + observation.started_at, + observation.packs_before, + Some(observation.packs_after), + Some(observation.compacted_bytes), + ); + } + return Err(error.into()); + } + }; let manifest_digest = digest_from_manifest_key(&manifest_key)?; // Step 7: CAS the pointer. @@ -652,15 +1191,50 @@ pub async fn cas_publish( Some(e) => Precond::IfMatch(e.clone()), None => Precond::IfNoneMatchStar, }; - match store + let cas_outcome = match store .put_pointer(&pkey, manifest_digest.as_bytes(), precond) - .await? + .await { - CasOutcome::Won(_new_etag) => Ok(CasSuccess { - manifest: m_after, - manifest_key, - }), + Ok(outcome) => outcome, + Err(error) => { + if let Some(observation) = &compaction_observation { + record_compaction( + "publish_error", + observation.started_at, + observation.packs_before, + Some(observation.packs_after), + Some(observation.compacted_bytes), + ); + } + return Err(error.into()); + } + }; + match cas_outcome { + CasOutcome::Won(_new_etag) => { + if let Some(observation) = &compaction_observation { + record_compaction( + "success", + observation.started_at, + observation.packs_before, + Some(observation.packs_after), + Some(observation.compacted_bytes), + ); + } + Ok(CasSuccess { + manifest: m_after, + manifest_key, + }) + } CasOutcome::LostRace => { + if let Some(observation) = &compaction_observation { + record_compaction( + "cas_conflict", + observation.started_at, + observation.packs_before, + Some(observation.packs_after), + Some(observation.compacted_bytes), + ); + } // Surface a typed Conflict carrying the winner so the caller // can reconcile the on-disk workspace without re-reading the // pointer. We re-GET the pointer here on the slow path; a @@ -848,6 +1422,374 @@ mod tests { assert_eq!(m.packs, vec![pack_key('e')]); } + #[test] + fn compaction_starts_with_manifest_headroom() { + assert!(!should_compact( + PACK_COMPACTION_THRESHOLD - 1, + PACK_COMPACTION_THRESHOLD + )); + assert!(should_compact( + PACK_COMPACTION_THRESHOLD, + PACK_COMPACTION_THRESHOLD + )); + assert!(should_compact( + MAX_MANIFEST_PACKS, + PACK_COMPACTION_THRESHOLD + )); + } + + #[test] + fn compaction_must_reduce_before_cap_but_may_replace_at_cap() { + assert!(compacted_pack_set_is_usable( + PACK_COMPACTION_THRESHOLD, + PACK_COMPACTION_THRESHOLD - 1 + )); + assert!(!compacted_pack_set_is_usable( + PACK_COMPACTION_THRESHOLD, + PACK_COMPACTION_THRESHOLD + )); + assert!(compacted_pack_set_is_usable( + MAX_MANIFEST_PACKS, + MAX_MANIFEST_PACKS + )); + assert!(!compacted_pack_set_is_usable( + MAX_MANIFEST_PACKS, + MAX_MANIFEST_PACKS + 1 + )); + } + + #[test] + fn compacted_manifest_replaces_parent_pack_set() { + let mut refs = BTreeMap::new(); + refs.insert("refs/heads/main".into(), "1".repeat(40)); + let manifest = compose_compacted_after( + Some(parent_digest()), + "refs/heads/main".into(), + refs, + vec![pack_key('9'), pack_key('8'), pack_key('9')], + ); + + assert_eq!(manifest.packs, vec![pack_key('8'), pack_key('9')]); + assert_eq!(manifest.parent, Some(parent_digest())); + manifest.validate().expect("compacted manifest"); + } + + #[test] + fn compacted_empty_repository_needs_no_packs() { + let manifest = compose_compacted_after( + Some(parent_digest()), + "refs/heads/main".into(), + BTreeMap::new(), + Vec::new(), + ); + + assert!(manifest.packs.is_empty()); + manifest.validate().expect("empty compacted manifest"); + } + + async fn run_test_git(repo: &Path, args: &[&str]) -> std::process::Output { + let mut command = Command::new("git"); + command.current_dir(repo).args(args); + super::super::transport::harden_git_env(&mut command); + let output = command.output().await.expect("spawn git"); + assert!( + output.status.success(), + "git {args:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); + output + } + + #[tokio::test] + async fn full_compaction_pack_covers_current_refs() { + let scratch = TempDir::new().expect("scratch"); + let repo = scratch.path().join("source"); + tokio::fs::create_dir(&repo).await.expect("repo dir"); + run_test_git(&repo, &["init", "--quiet", "--initial-branch=main"]).await; + run_test_git(&repo, &["config", "user.email", "compact@test"]).await; + run_test_git(&repo, &["config", "user.name", "compact"]).await; + tokio::fs::write(repo.join("file.txt"), b"reachable\n") + .await + .expect("file"); + run_test_git(&repo, &["add", "file.txt"]).await; + run_test_git(&repo, &["commit", "--quiet", "-m", "reachable"]).await; + let oid = String::from_utf8(run_test_git(&repo, &["rev-parse", "HEAD"]).await.stdout) + .expect("oid utf8") + .trim() + .to_string(); + let refs = BTreeMap::from([("refs/heads/main".to_string(), oid)]); + + let compacted = capture_compacted_packs( + &repo, + &refs, + PublishLimits { + parent_hydrated_bytes: 0, + max_pack_bytes: 1024 * 1024, + max_repo_bytes: 2 * 1024 * 1024, + }, + scratch.path(), + ) + .await + .expect("compact"); + + assert_eq!(compacted.packs.len(), 1); + assert!(compacted.total_pack_bytes > 0); + let idx = compacted.packs[0].idx_path.to_str().expect("idx utf8"); + run_test_git(&repo, &["verify-pack", idx]).await; + } + + fn probe_enabled() -> bool { + std::env::var("BUZZ_GIT_S3_PROBE").as_deref() == Ok("1") + } + + fn live_store() -> GitStore { + let endpoint = std::env::var("BUZZ_GIT_S3_ENDPOINT") + .or_else(|_| std::env::var("BUZZ_S3_ENDPOINT")) + .unwrap_or_else(|_| "http://localhost:9000".into()); + let access_key = std::env::var("BUZZ_GIT_S3_ACCESS_KEY") + .or_else(|_| std::env::var("BUZZ_S3_ACCESS_KEY")) + .unwrap_or_else(|_| "buzz_dev".into()); + let secret_key = std::env::var("BUZZ_GIT_S3_SECRET_KEY") + .or_else(|_| std::env::var("BUZZ_S3_SECRET_KEY")) + .unwrap_or_else(|_| "buzz_dev_secret".into()); + let bucket = std::env::var("BUZZ_GIT_S3_BUCKET") + .or_else(|_| std::env::var("BUZZ_S3_BUCKET")) + .unwrap_or_else(|_| "buzz-media".into()); + let region = std::env::var("BUZZ_GIT_S3_REGION") + .or_else(|_| std::env::var("BUZZ_S3_REGION")) + .unwrap_or_else(|_| "us-east-1".into()); + GitStore::new(&endpoint, &access_key, &secret_key, &bucket, ®ion).expect("connect minio") + } + + fn tenant() -> TenantContext { + TenantContext::resolved( + buzz_core::CommunityId::from_uuid(uuid::Uuid::from_u128(1)), + "git.example", + ) + } + + #[tokio::test] + async fn live_cas_publish_compacts_and_rehydrates() { + if !probe_enabled() { + return; + } + let store = live_store(); + let scratch = TempDir::new().expect("scratch"); + let source_path = scratch.path().join("source"); + tokio::fs::create_dir(&source_path).await.expect("repo dir"); + run_test_git(&source_path, &["init", "--quiet", "--initial-branch=main"]).await; + run_test_git(&source_path, &["config", "user.email", "compact@test"]).await; + run_test_git(&source_path, &["config", "user.name", "compact"]).await; + tokio::fs::write(source_path.join("file.txt"), b"one\n") + .await + .expect("file"); + run_test_git(&source_path, &["add", "file.txt"]).await; + run_test_git(&source_path, &["commit", "--quiet", "-m", "one"]).await; + let first_oid = String::from_utf8( + run_test_git(&source_path, &["rev-parse", "HEAD"]) + .await + .stdout, + ) + .expect("first oid utf8") + .trim() + .to_string(); + let first_refs = BTreeMap::from([("refs/heads/main".to_string(), first_oid)]); + let first_pack = capture_pack( + &source_path, + &BTreeMap::new(), + &first_refs, + 1024 * 1024, + scratch.path(), + ) + .await + .expect("capture first") + .expect("first pack"); + let first_pack_bytes = u64::try_from(first_pack.len()).expect("first pack length"); + let first_pack_key = store.put_pack(&first_pack).await.expect("put first pack"); + write_idx_sidecar(&store, &first_pack_key, &first_pack, scratch.path()) + .await + .expect("first idx"); + + tokio::fs::write(source_path.join("file.txt"), b"one\ntwo\n") + .await + .expect("second file"); + run_test_git(&source_path, &["add", "file.txt"]).await; + run_test_git(&source_path, &["commit", "--quiet", "-m", "two"]).await; + run_test_git(&source_path, &["tag", "-a", "v1", "-m", "annotated"]).await; + let parent_head_oid = String::from_utf8( + run_test_git(&source_path, &["rev-parse", "HEAD"]) + .await + .stdout, + ) + .expect("parent head utf8") + .trim() + .to_string(); + let parent_tag_oid = String::from_utf8( + run_test_git(&source_path, &["rev-parse", "refs/tags/v1"]) + .await + .stdout, + ) + .expect("parent tag utf8") + .trim() + .to_string(); + let parent_refs = BTreeMap::from([ + ("refs/heads/main".to_string(), parent_head_oid), + ("refs/tags/v1".to_string(), parent_tag_oid), + ]); + let second_pack = capture_pack( + &source_path, + &first_refs, + &parent_refs, + 1024 * 1024, + scratch.path(), + ) + .await + .expect("capture second") + .expect("second pack"); + let second_pack_bytes = u64::try_from(second_pack.len()).expect("second pack length"); + let second_pack_key = store.put_pack(&second_pack).await.expect("put second pack"); + write_idx_sidecar(&store, &second_pack_key, &second_pack, scratch.path()) + .await + .expect("second idx"); + + let parent = Manifest { + version: MANIFEST_VERSION, + head: "refs/heads/main".into(), + refs: parent_refs, + packs: vec![first_pack_key, second_pack_key], + parent: None, + }; + parent.validate().expect("parent manifest"); + let parent_key = store + .put_manifest(&parent.canonical_bytes().expect("parent bytes")) + .await + .expect("put parent"); + let parent_digest = digest_from_manifest_key(&parent_key).expect("parent digest"); + let ctx = tenant(); + let owner = format!("compact-{}", uuid::Uuid::new_v4()); + let repo = "history"; + let pkey = pointer_key(ctx.community(), &owner, repo); + match store + .put_pointer(&pkey, parent_digest.as_bytes(), Precond::IfNoneMatchStar) + .await + .expect("put pointer") + { + CasOutcome::Won(_) => {} + CasOutcome::LostRace => panic!("unique pointer must win"), + } + let cache_parent = scratch.path().join("cache"); + let cache = + crate::api::git::pack_cache::GitPackCache::new(&cache_parent, 4 * 1024 * 1024, 1) + .expect("cache"); + let (hydrated, parent_state) = crate::api::git::hydrate::hydrate_for_write( + &store, + &ctx, + &owner, + repo, + crate::api::git::hydrate::HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: 1024 * 1024, + max_repo_bytes: 2 * 1024 * 1024, + }, + ) + .await + .expect("hydrate parent"); + assert_eq!(parent_state.parent_digest, Some(parent_digest.clone())); + + tokio::fs::write(source_path.join("file.txt"), b"one\ntwo\nthree\n") + .await + .expect("third file"); + run_test_git(&source_path, &["add", "file.txt"]).await; + run_test_git(&source_path, &["commit", "--quiet", "-m", "three"]).await; + let hydrated_path = hydrated.path().to_str().expect("hydrated path utf8"); + run_test_git(&source_path, &["push", "--quiet", hydrated_path, "main"]).await; + let head_oid = String::from_utf8( + run_test_git(&source_path, &["rev-parse", "HEAD"]) + .await + .stdout, + ) + .expect("post-push head utf8") + .trim() + .to_string(); + let limits = PublishLimits { + parent_hydrated_bytes: hydrated.hydrated_bytes(), + max_pack_bytes: 1024 * 1024, + max_repo_bytes: 2 * 1024 * 1024, + }; + + let test_options = PublishOptions { + limits, + compaction_threshold: 2, + }; + let success = cas_publish_inner( + &store, + &ctx, + hydrated.path(), + &owner, + repo, + &parent_state, + test_options, + ) + .await + .expect("compaction publish"); + assert_eq!(success.manifest.packs.len(), 1); + assert_eq!( + success.manifest.refs.get("refs/heads/main"), + Some(&head_oid) + ); + assert!(success.manifest.refs.contains_key("refs/tags/v1")); + assert_eq!(success.manifest.parent, Some(parent_digest)); + + let conflict = cas_publish_inner( + &store, + &ctx, + hydrated.path(), + &owner, + repo, + &parent_state, + test_options, + ) + .await + .expect_err("stale parent must lose CAS"); + assert!(matches!(conflict, CasError::Conflict { .. })); + + let hydrated = crate::api::git::hydrate::hydrate_for_read( + &store, + &ctx, + &owner, + repo, + crate::api::git::hydrate::HydrationOptions { + pack_cache: &cache, + scratch_dir: scratch.path(), + max_pack_bytes: limits.max_pack_bytes, + max_repo_bytes: limits.max_repo_bytes, + }, + ) + .await + .expect("hydrate compacted") + .expect("repo exists"); + let hydrated_head = String::from_utf8( + run_test_git(hydrated.path(), &["rev-parse", "HEAD"]) + .await + .stdout, + ) + .expect("hydrated head utf8") + .trim() + .to_string(); + assert_eq!(hydrated_head, head_oid); + run_test_git(hydrated.path(), &["cat-file", "-e", "refs/tags/v1^{tag}"]).await; + assert_eq!( + parent_state.parent.packs.len(), + 2, + "test parent must exercise pack-count reduction" + ); + assert_eq!( + limits.parent_hydrated_bytes, + first_pack_bytes + second_pack_bytes + ); + } + /// `cas_publish` must invoke `Manifest::validate()` between /// `compose_after` and `put_manifest`. The unit on `validate` lives in /// `manifest.rs`; this test pins that the call site here actually diff --git a/crates/buzz-relay/src/api/git/manifest.rs b/crates/buzz-relay/src/api/git/manifest.rs index 8134aaafa8..baf109c1ad 100644 --- a/crates/buzz-relay/src/api/git/manifest.rs +++ b/crates/buzz-relay/src/api/git/manifest.rs @@ -37,6 +37,12 @@ pub const MANIFEST_VERSION: u32 = 1; /// Hydration indexes packs one at a time, but an unbounded pack list still /// turns one request into unbounded object-store and git subprocess work. pub const MAX_MANIFEST_PACKS: usize = 128; +/// Pack count that triggers proactive consolidation during the next accepted push. +/// +/// Keeping one quarter of the manifest capacity in reserve prevents a hot +/// repository from reaching the hard limit while a prior compaction attempt +/// falls back to the normal delta-pack path. +pub const PACK_COMPACTION_THRESHOLD: usize = MAX_MANIFEST_PACKS * 3 / 4; /// Maximum number of refs one manifest may advertise. pub const MAX_MANIFEST_REFS: usize = 10_000; diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 5ff33e9e45..16e521a44e 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -97,6 +97,11 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) { &GIT_DURATION_BUCKETS_S, ) .expect("valid git cache population wait bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_compaction_seconds".to_owned()), + &GIT_DURATION_BUCKETS_S, + ) + .expect("valid git compaction duration bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_git_hydrate_bytes".to_owned()), &GIT_BYTES_BUCKETS, @@ -107,11 +112,26 @@ pub fn install(port: u16, gauge_idle_timeout_secs: u64) { &GIT_BYTES_BUCKETS, ) .expect("valid git stream byte bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_compaction_bytes".to_owned()), + &GIT_BYTES_BUCKETS, + ) + .expect("valid git compaction byte bucket boundaries") .set_buckets_for_metric( Matcher::Full("buzz_git_hydrate_packs".to_owned()), &GIT_PACK_BUCKETS, ) .expect("valid git pack-count bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_compaction_packs_before".to_owned()), + &GIT_PACK_BUCKETS, + ) + .expect("valid git compaction input pack-count bucket boundaries") + .set_buckets_for_metric( + Matcher::Full("buzz_git_pack_compaction_packs_after".to_owned()), + &GIT_PACK_BUCKETS, + ) + .expect("valid git compaction output pack-count bucket boundaries") .set_buckets_for_metric(Matcher::Suffix("_seconds".to_owned()), &DURATION_BUCKETS_S) .expect("valid seconds bucket boundaries") .set_buckets_for_metric( diff --git a/docs/git-on-object-storage.md b/docs/git-on-object-storage.md index 105e4ba723..1d87720edc 100644 --- a/docs/git-on-object-storage.md +++ b/docs/git-on-object-storage.md @@ -285,24 +285,28 @@ against a pinned minimum `git` ≥ 2.31, the first release with `git index-pack --fsck-objects` defaults relied on here). It remains to show `manifest(d).packs` *covers* the reachable closure of `m.refs`. By induction on the push chain: the empty repo's manifest names ∅ and refs ∅ -(covered vacuously). Step 5 sets `m_after.packs = m_before.packs ∪ keys(O)` where -`O` is every object this push introduced; and `m_after.refs` only points at -objects in `m_before.refs`' closure (unchanged or deleted refs) or in `O` (new or -force-moved refs). So every object reachable from `m_after.refs` is in -`m_before.packs` (covered by IH) or in `keys(O)` — hence in `m_after.packs`. ∎ +(covered vacuously). A normal step sets +`m_after.packs = m_before.packs ∪ keys(O)` where `O` is every object this push +introduced; and `m_after.refs` only points at objects in `m_before.refs`' +closure (unchanged or deleted refs) or in `O` (new or force-moved refs). A +compaction step instead feeds every `m_after.refs` tip to `git pack-objects +--revs` with no negative revisions, then names only the resulting bounded pack +set. The normal case preserves coverage by induction; the compaction case +re-establishes coverage directly from the complete post-push ref closure. ∎ **Remark (force-push, delete, and GC).** Coverage is a *superset*, not equality, and that is the correct invariant. A delete-ref drops a key from `m.refs`; a -force-push repoints a ref off its old history. Neither removes packs from -`m.packs`, so objects reachable only from the old/deleted ref become unreachable +force-push repoints a ref off its old history. Neither normal ref operation +removes packs from `m.packs`, so objects reachable only from the old/deleted ref become unreachable but remain named. This is safe — reconstruction of the *current* refs is -unaffected — but it means `m.packs` grows monotonically under the protocol as -specified. Garbage collection (computing reachability from `m.refs` and -publishing a manifest with a pruned pack set) is a separate, *also CAS-guarded* -operation: it is just another `Push` whose `m_after` happens to name fewer packs, -so Theorems 1 and 3 apply to it unchanged. GC correctness (that it never prunes a -reachable pack) is an obligation on the GC's reachability computation, out of -scope here and called out as future work. +unaffected. Before the bounded manifest reaches its pack limit, an accepted push +proactively captures the complete post-push reachable closure and CAS-publishes +a replacement manifest that normally has fewer packs. At the hard cap, an +equal-count replacement is also valid when it incorporates the newly reachable +objects while remaining within the bound. The old immutable objects are not +deleted, so readers holding an earlier manifest remain valid. Physical +object-store deletion remains a separate retention concern outside this proof +boundary. ### Theorem 3 (Linearizable Refs / No Lost Update) @@ -482,6 +486,7 @@ symbol search, not line counts.) | `run_conformance_probe` (A1/A3 fail-closed startup gate) | `store.rs` + `main.rs` | | `hydrate_for_read` / `hydrate_for_write` | `hydrate.rs` | | Bounded digest-keyed pack/index cache and single-flight population | `pack_cache.rs` | +| Proactive full-closure pack compaction before manifest capacity | `cas_publish.rs` | | `ParentState { if_match, parent_digest, parent }` + `from_loaded`/`fresh` | `cas_publish.rs:154` | | `cas_publish(.., &parent_state) -> Result` | `cas_publish.rs:410` | | `CasError::Conflict { winner_manifest, winner_manifest_key }` (typed 412) | `cas_publish.rs:92` | @@ -519,7 +524,9 @@ snapshot reads succeed (`snapErr`); whether it *changes* refs is then **derived* (`DidChange == newVal ≠ value-in-the-manifest-it-read`), not a free boolean. The skip predicate is `MustPublish(p) == DidChange(p) \/ snapErr(p)` — "publish unless we *observed* no change," never "publish unless `b == a`" with failed reads -compared equal. +compared equal. A `compacted` marker distinguishes normal delta-pack stages +from stages whose own pack is the trusted full closure produced from every +post-push ref tip. Crucially the model carries the **real ref value** per manifest (`refs[m]` = the objectId `main` holds in manifest `m`) and explicit **history** (`parent[m]`). @@ -531,7 +538,7 @@ TLC checks eight invariants (a finiteness constraint, `BoundedManifests`, caps p |---|---|---| | `Inv_Fence` | T1 | an obligated push's manifest is published before success is observed | | `Inv_ChangedPublished` | T1 | a ref-changing push is always published (fallible-snapshot bite) | -| `Inv_Closed` | T2 | a published manifest's pack set *covers* its published parent's | +| `Inv_Closed` | T2 | a normal manifest covers its parent's packs; a compacted manifest names its trusted full-closure pack | | `Inv_NoFork` | T3 | no two published manifests share a parent (a fork = a lost update) | | `Inv_RefEffectApplied` | T3 | an installed push's committed ref value equals the value it proposed | | `Inv_RefDerivedFromParent` | T3 | an install is derived from the pointer it read (no build on superseded state) | @@ -541,7 +548,6 @@ TLC checks eight invariants (a finiteness constraint, `BoundedManifests`, caps p ``` $ tlc GitOnObjectStore.tla -config GitOnObjectStore.cfg Model checking completed. No error has been found. -1435102 states generated, 435745 distinct states found, 0 states left on queue. ``` **Every invariant is proven non-vacuous** by a mutation that trips it (each @@ -551,14 +557,16 @@ checked in isolation against each mutant). This is the discipline that catches | Mutation | Trips | |---|---| | skip predicate `DidChange /\ ~snapErr` (ref change + both snapshots fail → silently skipped) | `Inv_ChangedPublished` | -| `packs[m] = {m}` (manifest drops predecessor's packs) | `Inv_Closed` | +| a normal stage uses `packs[m] = {m}` without the full-closure marker | `Inv_Closed` | | drop CAS guard (two pushers install off one parent — a fork) | `Inv_NoFork` | | install records the *read* ref value, not the push's proposal (effect dropped) | `Inv_RefEffectApplied` | | record parent as root instead of the pointer actually read | `Inv_RefDerivedFromParent` (+ `Inv_NoFork`) | -The `packs[m]={m}` and CAS-guard mutations are exactly the two vacuity tests an -external reviewer ran against an earlier draft, where the then-invariants passed -unchanged; the history + real-ref-value vocabulary is what makes them fail now. +The unmarked `packs[m]={m}` and CAS-guard mutations are the two core closure and +serialization vacuity tests. The full-closure marker is only assigned by the +compaction branch; reusing it for a normal delta stage would make the closure +claim vacuous, so the model explicitly clears stale markers when manifest ids +are reused. The ref-value mutations (effect-dropped, wrong-parent) are what close the gap from "pointer CAS serializes" to "ref *updates* are linearizable" — the user-visible theorem the title promises. (A weaker mutation, `MustPublish == DidChange` alone, @@ -574,9 +582,8 @@ The model is checked at `Pushers = {p1,p2,p3}`, `MaxManifests = 3`, under the the retry loop otherwise lets pushers churn fresh manifest ids and ref values without bound, so the model is *finite-state only with the bound*. Three concurrent pushers exercise every CAS race relevant to these invariants (a fourth -adds no qualitatively new interleaving); the real-ref-value domain is what makes -even three a ~436K-state check. This is a *bounded* model check, not an unbounded -proof: it exhaustively verifies the invariants within the bound and is mutation- +adds no qualitatively new interleaving). This is a *bounded* model check, not an +unbounded proof: it exhaustively verifies the invariants within the bound and is mutation- shown non-vacuous, which is the standard claim for a TLC-checked safety spec. ## Summary diff --git a/docs/spec/GitOnObjectStore.tla b/docs/spec/GitOnObjectStore.tla index 1f61ef9d92..f5f71b357a 100644 --- a/docs/spec/GitOnObjectStore.tla +++ b/docs/spec/GitOnObjectStore.tla @@ -9,7 +9,8 @@ (* Pushers race to advance a single manifest pointer holding a ref value. *) (* We assert (see SAFETY PROPERTIES for the full set and per-invariant docs):*) (* T1 fence: observed success => the obligated push is durably published *) -(* T2 closure: a published manifest's pack set covers its parent's *) +(* T2 closure: a published manifest either covers its parent's packs or *) +(* names a trusted full-closure compaction pack *) (* T3 ref linearizability: installs form a fork-free chain, each commits *) (* exactly the value it proposed, derived from the pointer it read *) (* Each invariant is mutation-tested non-vacuous; see docs/ Mechanized §. *) @@ -28,12 +29,13 @@ VARIABLES staged, \* pusher id -> manifest id it intends to install parent, \* manifest id -> the manifest id it was derived from (history) refs, \* manifest id -> objectId that this manifest binds the ref "main" to + compacted, \* manifests whose own pack is a full closure of their refs newVal, \* pusher id -> objectId this push proposes for "main" (its effect) snapErr, \* pusher id -> did either ref-snapshot read fail? (BOOLEAN) observed \* set of pusher ids that have observed success (fence passed) vars == <> + parent, refs, compacted, newVal, snapErr, observed>> \* We model a single ref, "main", whose value is an objectId in ObjIds. This is \* enough to exhibit ref-update linearizability: a lost update is the published @@ -57,6 +59,7 @@ TypeOK == /\ staged \in [Pushers -> ManifestIds] /\ parent \in [ManifestIds -> ManifestIds] /\ refs \in [ManifestIds -> ObjIds] + /\ compacted \subseteq ManifestIds /\ newVal \in [Pushers -> ObjIds] /\ snapErr \in [Pushers -> BOOLEAN] /\ observed \subseteq Pushers @@ -70,6 +73,7 @@ Init == /\ staged = [p \in Pushers |-> 0] /\ parent = [m \in ManifestIds |-> 0] /\ refs = [m \in ManifestIds |-> 0] \* "main" starts at objectId 0 (empty) + /\ compacted = {} /\ newVal = [p \in Pushers |-> 0] /\ snapErr = [p \in Pushers |-> FALSE] /\ observed = {} @@ -99,15 +103,24 @@ Begin(p) == \* fail (e). The staged manifest binds "main" to v and is derived from the \* manifest the push READ -- so a stale reader builds on stale ref state, and \* only the CAS guard stops it from clobbering a newer published value. - /\ \E v \in ObjIds, e \in BOOLEAN : + /\ \E v \in ObjIds, e \in BOOLEAN, compact \in BOOLEAN : /\ newVal' = [newVal EXCEPT ![p] = v] /\ snapErr' = [snapErr EXCEPT ![p] = e] /\ LET m == FreshId IN /\ readEtag' = [readEtag EXCEPT ![p] = pointer] /\ staged' = [staged EXCEPT ![p] = m] /\ parent' = [parent EXCEPT ![m] = pointer] - /\ packs' = [packs EXCEPT ![m] = packs[pointer] \union {m}] + \* A compact stage models `pack-objects` over every post-push + \* ref tip. Its own pack is therefore trusted to cover the full + \* reachable closure; a normal stage extends the parent pack set. + /\ packs' = [packs EXCEPT + ![m] = IF compact + THEN {m} + ELSE packs[pointer] \union {m}] /\ refs' = [refs EXCEPT ![m] = v] + /\ compacted' = IF compact + THEN compacted \union {m} + ELSE compacted \ {m} /\ pc' = [pc EXCEPT ![p] = "staged"] /\ UNCHANGED <> @@ -117,7 +130,7 @@ SkipPublish(p) == /\ pc[p] = "staged" /\ ~MustPublish(p) /\ pc' = [pc EXCEPT ![p] = "done"] - /\ UNCHANGED <> + /\ UNCHANGED <> \* Step 7: CAS. Succeeds iff pointer still equals the etag this pusher read (A3). CasSucceed(p) == @@ -127,27 +140,27 @@ CasSucceed(p) == /\ pointer' = staged[p] /\ published' = published \union {staged[p]} /\ pc' = [pc EXCEPT ![p] = "done"] - /\ UNCHANGED <> + /\ UNCHANGED <> CasFail(p) == /\ pc[p] = "staged" /\ MustPublish(p) /\ pointer # readEtag[p] /\ pc' = [pc EXCEPT ![p] = "lost"] \* will retry from idle - /\ UNCHANGED <> + /\ UNCHANGED <> \* Step 8: the fence. Observe success ONLY after the push reached "done" \* (either via successful CAS or a legitimate skip). Observe(p) == /\ pc[p] = "done" /\ observed' = observed \union {p} - /\ UNCHANGED <> + /\ UNCHANGED <> \* A loser retries: back to idle, ready to re-read the advanced pointer. Retry(p) == /\ pc[p] = "lost" /\ pc' = [pc EXCEPT ![p] = "idle"] - /\ UNCHANGED <> + /\ UNCHANGED <> Next == \E p \in Pushers : @@ -209,14 +222,16 @@ Inv_RefDerivedFromParent == \A p \in Pushers : Installed(p) => (parent[staged[p]] = readEtag[p] /\ readEtag[p] \in published) -\* T2 (Reconstruction coverage -- non-vacuous): every published non-root manifest -\* names a pack set that COVERS its published parent's pack set plus its own pack. -\* Perci's mutation (packs[m] = {m} instead of packs[parent] U {m}) breaks this, -\* because then packs[parent[m]] is not a subset of packs[m]. +\* T2 (Reconstruction coverage -- non-vacuous): every published non-root +\* manifest either names its trusted full-closure compaction pack, or covers its +\* published parent's pack set plus its own delta pack. The model abstracts +\* Git's reachability walk as the `compacted` marker; production earns that +\* marker only by feeding every post-push ref tip to `git pack-objects --revs`. Inv_Closed == \A m \in published : (m # 0 /\ parent[m] \in published) => - (packs[parent[m]] \subseteq packs[m] /\ m \in packs[m]) + (m \in packs[m] /\ + (m \in compacted \/ packs[parent[m]] \subseteq packs[m])) \* Parent integrity: every published non-root manifest's parent is also published \* (the install chain is grounded in durable history, never in vapor).