From fc557a9352869c7298d4b6257e1bcbddf59fe522 Mon Sep 17 00:00:00 2001 From: forhappy Date: Sun, 16 Aug 2026 18:43:25 -0700 Subject: [PATCH 1/2] feat(s3): drive GC payload discovery from ingest journals --- extensions/s3/README.md | 3 + extensions/s3/client/README.md | 13 + extensions/s3/client/src/client.rs | 9 + extensions/s3/core/src/gc.rs | 43 ++ extensions/s3/core/src/lib.rs | 6 +- extensions/s3/core/src/physical_journal.rs | 409 +++++++++++++++++ extensions/s3/core/src/repository.rs | 419 +++++++++++++++++- extensions/s3/core/tests/gc.rs | 164 ++++++- extensions/s3/core/tests/repository.rs | 3 +- extensions/s3/spec/prolly-s3/paths.md | 9 + .../s3/spec/prolly-s3/state-machines.md | 6 +- 11 files changed, 1073 insertions(+), 11 deletions(-) create mode 100644 extensions/s3/core/src/physical_journal.rs diff --git a/extensions/s3/README.md b/extensions/s3/README.md index a0a76812..708f6dec 100644 --- a/extensions/s3/README.md +++ b/extensions/s3/README.md @@ -54,3 +54,6 @@ remain operational limits. Bounded GC reclaims unreachable immutable data, and history-transfer APIs preserve a source commit DAG with destination-local IDs and payload bindings. GC currently coordinates concurrent writer handles inside one authoritative process; quiesce separately running writer processes. +Journaled ingest windows can opt into payload candidate discovery without a +payload namespace scan; legacy/direct writers should continue using the default +GC mode until they are migrated. diff --git a/extensions/s3/client/README.md b/extensions/s3/client/README.md index 8e157ae6..ca26e9f1 100644 --- a/extensions/s3/client/README.md +++ b/extensions/s3/client/README.md @@ -497,6 +497,19 @@ while gc.phase != GcPhase::Complete { } ``` +Bulk ingest callers that use the journaled batch APIs can opt into +journal-driven payload discovery: + +```rust +let mut gc = client.start_gc_journaled(two_hours_millis).await?; +``` + +This mode reads one immutable creation-intent manifest per ingest window and +does not list the payload namespace. Payloads from direct or pre-journal +writers remain retained, so use `start_gc` while those writers remain active. +A manifest contains whole-object paths, +sizes, and checksums only—payload bytes are never packed or chunked. + The collector sweeps only immutable commit, direct-node, and whole-payload objects. It never sweeps mutable refs, derived indexes, publication journals, format diff --git a/extensions/s3/client/src/client.rs b/extensions/s3/client/src/client.rs index d4bb4d8f..46480258 100644 --- a/extensions/s3/client/src/client.rs +++ b/extensions/s3/client/src/client.rs @@ -633,6 +633,15 @@ impl Client { self.repository.start_gc(grace_millis).await } + /// Start GC with payload candidates sourced from immutable ingest-window + /// journals. Payloads from legacy/direct writers remain retained; use + /// `start_gc` while migrating those writers. + pub async fn start_gc_journaled(&self, grace_millis: u64) -> Result { + self.ensure_provider_qualified()?; + self.attached_branch()?; + self.repository.start_gc_journaled(grace_millis).await + } + pub async fn resume_gc(&self) -> Result> { self.ensure_provider_qualified()?; self.attached_branch()?; diff --git a/extensions/s3/core/src/gc.rs b/extensions/s3/core/src/gc.rs index 8c07654e..bfaaa8e8 100644 --- a/extensions/s3/core/src/gc.rs +++ b/extensions/s3/core/src/gc.rs @@ -6,12 +6,43 @@ use crate::{ CommitId, ObjectPath, OperationId, PhysicalVersion, RefGeneration, RepositoryId, RootManifest, }; +/// Payload candidate discovery policy for a GC epoch. +/// +/// `LegacyScan` is the compatibility-safe default because direct/single-object +/// writers from older clients do not emit creation intents. `JournalOnly` is +/// intended for repositories whose payload ingestion uses the journaled batch +/// APIs; it avoids listing the payload namespace entirely. Unjournaled payloads +/// are retained until a legacy-scan epoch discovers them. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum GcCandidateDiscovery { + #[default] + LegacyScan, + JournalOnly, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum GcCandidateNamespace { + #[default] + Repository, + Commits, + Nodes, + Complete, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub enum GcInventorySource { + #[default] + Completions, + Intents, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum GcPhase { DiscoverBranches, DiscoverTags, MarkCommits, MarkNodes, + ScanInventory, ScanCandidates, CatchUpDirtyRoots, Ready, @@ -27,6 +58,14 @@ pub struct GcCursor { pub cutoff_millis: u64, pub phase: GcPhase, pub continuation: Option, + #[serde(default)] + pub payload_discovery: GcCandidateDiscovery, + #[serde(default)] + pub candidate_namespace: GcCandidateNamespace, + #[serde(default)] + pub journal_object_offset: usize, + #[serde(default)] + pub inventory_source: GcInventorySource, pub work: RootManifest, pub dirty_sequence: u64, pub dirty_target_sequence: u64, @@ -58,6 +97,10 @@ pub struct GcReport { pub deleted_by_kind: BTreeMap, #[serde(default)] pub protected_by_kind: BTreeMap, + #[serde(default)] + pub journal_batches: u64, + #[serde(default)] + pub journal_objects: u64, } #[derive(Clone, Debug, PartialEq, Eq)] diff --git a/extensions/s3/core/src/lib.rs b/extensions/s3/core/src/lib.rs index 89b121b9..03b74b80 100644 --- a/extensions/s3/core/src/lib.rs +++ b/extensions/s3/core/src/lib.rs @@ -13,6 +13,7 @@ mod model; mod object_plane; mod operation_index; mod payload; +mod physical_journal; mod publication; mod ref_catalog; mod repository; @@ -33,7 +34,10 @@ pub use control_versions::{ MutableControlObserver, MutableControlStore, DEFAULT_MUTABLE_CONTROL_VERSIONS_TO_RETAIN, }; pub use error::{Error, ErrorCode, Result, RetryAdvice}; -pub use gc::{GcCursor, GcPage, GcPhase, GcReport}; +pub use gc::{ + GcCandidateDiscovery, GcCandidateNamespace, GcCursor, GcInventorySource, GcPage, GcPhase, + GcReport, +}; pub use journal_indexes::{ JournalDerivedIndexes, JournalIndexAdvanceReport, JournalIndexRebuildCleanup, JournalIndexRebuildCursor, JournalIndexRebuildPhase, JournalIndexRebuildStep, diff --git a/extensions/s3/core/src/physical_journal.rs b/extensions/s3/core/src/physical_journal.rs new file mode 100644 index 00000000..c489371d --- /dev/null +++ b/extensions/s3/core/src/physical_journal.rs @@ -0,0 +1,409 @@ +use std::sync::Arc; + +use serde::{Deserialize, Serialize}; + +use crate::{ + codec::sha256, decode_canonical, encode_canonical, Error, ErrorCode, ImmutablePut, + ImmutablePutOutcome, ListRequest, ObjectPath, ObjectPlane, OperationId, RepositoryId, Result, +}; + +/// One complete immutable physical object expected to be created by an ingest +/// window. The journal contains identity only; it never contains payload bytes +/// or transfer-manager state. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct PhysicalObjectIntent { + pub(crate) path: ObjectPath, + pub(crate) size: u64, + pub(crate) checksum_sha256: [u8; 32], +} + +/// Immutable, batch-addressed creation intent for physical objects. +/// +/// A record is written once before the corresponding payload uploads begin. +/// Replaying the same operation and identity list is idempotent, while a +/// process crash after the record and before an upload leaves only a harmless +/// missing-object intent for GC to skip. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct PhysicalObjectIntentBatch { + pub(crate) repository: RepositoryId, + pub(crate) operation: OperationId, + pub(crate) created_at_millis: u64, + pub(crate) objects: Vec, +} + +/// Provider identity captured after a payload batch has completed. This keeps +/// GC off the per-object HEAD path for the normal completion case while the +/// pre-upload intent remains the recovery source for a crash in between. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct PhysicalObjectCompletion { + pub(crate) path: ObjectPath, + pub(crate) size: u64, + pub(crate) checksum_sha256: [u8; 32], + pub(crate) provider_version_id: Option, + pub(crate) provider_etag: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct PhysicalObjectCompletionBatch { + pub(crate) repository: RepositoryId, + pub(crate) operation: OperationId, + pub(crate) completed_at_millis: u64, + pub(crate) objects: Vec, +} + +fn validate_identity_paths( + repository: RepositoryId, + payload_prefix: &str, + operation: OperationId, + object_count: usize, + max_objects: usize, + objects: impl IntoIterator, +) -> Result<()> { + if repository.as_bytes() == &[0; 32] + || operation.is_nil() + || object_count == 0 + || object_count > max_objects + { + return Err(Error::new( + ErrorCode::CorruptContent, + "physical-object journal batch is malformed", + )); + } + let mut previous: Option = None; + for (path, checksum_sha256) in objects { + let encoded = hex::encode(checksum_sha256); + let expected_path = format!( + "{payload_prefix}/sha256/{}/{}/{}", + &encoded[..2], + &encoded[2..4], + encoded + ); + if checksum_sha256 == [0; 32] + || path.as_str() != expected_path + || previous.as_ref().is_some_and(|previous| previous >= &path) + { + return Err(Error::new( + ErrorCode::CorruptContent, + "physical-object journal paths are not canonical", + )); + } + previous = Some(path); + } + Ok(()) +} + +impl PhysicalObjectIntentBatch { + fn validate( + &self, + repository: RepositoryId, + payload_prefix: &str, + max_objects: usize, + ) -> Result<()> { + validate_identity_paths( + repository, + payload_prefix, + self.operation, + self.objects.len(), + max_objects, + self.objects + .iter() + .map(|object| (object.path.clone(), object.checksum_sha256)), + )?; + if self.repository != repository { + return Err(Error::new( + ErrorCode::CorruptContent, + "physical-object journal repository identity changed", + )); + } + Ok(()) + } +} + +impl PhysicalObjectCompletionBatch { + fn validate( + &self, + repository: RepositoryId, + payload_prefix: &str, + max_objects: usize, + ) -> Result<()> { + validate_identity_paths( + repository, + payload_prefix, + self.operation, + self.objects.len(), + max_objects, + self.objects + .iter() + .map(|object| (object.path.clone(), object.checksum_sha256)), + )?; + if self.repository != repository + || self + .objects + .iter() + .any(|object| object.provider_etag.is_empty()) + { + return Err(Error::new( + ErrorCode::CorruptContent, + "physical-object completion batch is malformed", + )); + } + Ok(()) + } +} + +#[derive(Clone)] +pub(crate) struct PhysicalObjectJournal { + plane: Arc

, + prefix: String, + repository: RepositoryId, + max_objects: usize, +} + +impl PhysicalObjectJournal

{ + pub(crate) fn new( + plane: Arc

, + prefix: impl Into, + repository: RepositoryId, + max_objects: usize, + ) -> Result { + if max_objects == 0 { + return Err(Error::new( + ErrorCode::InvalidLimit, + "physical-object journal requires a positive object limit", + )); + } + Ok(Self { + plane, + prefix: prefix.into(), + repository, + max_objects, + }) + } + + pub(crate) async fn record( + &self, + operation: OperationId, + created_at_millis: u64, + mut objects: Vec, + ) -> Result<()> { + objects.sort_by(|left, right| left.path.cmp(&right.path)); + objects.dedup_by(|left, right| left.path == right.path); + let batch = PhysicalObjectIntentBatch { + repository: self.repository, + operation, + created_at_millis, + objects, + }; + batch.validate(self.repository, &self.payload_prefix(), self.max_objects)?; + let bytes = encode_canonical(&batch)?; + let path = self.batch_path(operation, &batch.objects)?; + match self + .plane + .put_immutable(ImmutablePut { + path: path.clone(), + expected_sha256: sha256(&bytes), + bytes: bytes.clone(), + }) + .await + { + Ok(ImmutablePutOutcome::Created(_) | ImmutablePutOutcome::AlreadyPresent(_)) => Ok(()), + Err(original) => { + let stored = self + .plane + .get(crate::GetRequest { + path, + range: None, + physical_version: None, + }) + .await?; + match stored { + Some(stored) if stored.bytes == bytes => Ok(()), + Some(stored) => { + let existing: PhysicalObjectIntentBatch = decode_canonical(&stored.bytes)?; + (existing.repository == batch.repository + && existing.operation == batch.operation + && existing.objects == batch.objects) + .then_some(()) + .ok_or(original) + } + _ => Err(original), + } + } + } + } + + pub(crate) async fn load(&self, path: ObjectPath) -> Result { + let stored = self + .plane + .get(crate::GetRequest { + path, + range: None, + physical_version: None, + }) + .await? + .ok_or_else(|| { + Error::new( + ErrorCode::OutcomeUnknown, + "physical-object journal disappeared", + ) + })?; + let batch: PhysicalObjectIntentBatch = decode_canonical(&stored.bytes)?; + batch.validate(self.repository, &self.payload_prefix(), self.max_objects)?; + Ok(batch) + } + + pub(crate) async fn record_completion( + &self, + operation: OperationId, + completed_at_millis: u64, + mut objects: Vec, + ) -> Result<()> { + objects.sort_by(|left, right| left.path.cmp(&right.path)); + objects.dedup_by(|left, right| left.path == right.path); + let batch = PhysicalObjectCompletionBatch { + repository: self.repository, + operation, + completed_at_millis, + objects, + }; + batch.validate(self.repository, &self.payload_prefix(), self.max_objects)?; + let bytes = encode_canonical(&batch)?; + let path = self.completion_path(operation, &batch.objects)?; + match self + .plane + .put_immutable(ImmutablePut { + path: path.clone(), + expected_sha256: sha256(&bytes), + bytes: bytes.clone(), + }) + .await + { + Ok(ImmutablePutOutcome::Created(_) | ImmutablePutOutcome::AlreadyPresent(_)) => Ok(()), + Err(original) => { + let stored = self + .plane + .get(crate::GetRequest { + path, + range: None, + physical_version: None, + }) + .await?; + match stored { + Some(stored) if stored.bytes == bytes => Ok(()), + Some(stored) => { + let existing: PhysicalObjectCompletionBatch = + decode_canonical(&stored.bytes)?; + (existing.repository == batch.repository + && existing.operation == batch.operation + && existing.objects == batch.objects) + .then_some(()) + .ok_or(original) + } + _ => Err(original), + } + } + } + } + + pub(crate) async fn load_completion( + &self, + path: ObjectPath, + ) -> Result { + let stored = self + .plane + .get(crate::GetRequest { + path, + range: None, + physical_version: None, + }) + .await? + .ok_or_else(|| { + Error::new( + ErrorCode::OutcomeUnknown, + "physical-object completion disappeared", + ) + })?; + let batch: PhysicalObjectCompletionBatch = decode_canonical(&stored.bytes)?; + batch.validate(self.repository, &self.payload_prefix(), self.max_objects)?; + Ok(batch) + } + + pub(crate) fn prefix(&self) -> String { + format!( + "{}/administration/physical-object-journal/{}/intents/", + self.prefix, + hex::encode(self.repository.as_bytes()) + ) + } + + pub(crate) fn completion_prefix(&self) -> String { + format!( + "{}/administration/physical-object-journal/{}/completions/", + self.prefix, + hex::encode(self.repository.as_bytes()) + ) + } + + fn payload_prefix(&self) -> String { + format!( + "{}/payloads/{}", + self.prefix, + hex::encode(self.repository.as_bytes()) + ) + } + + fn batch_path( + &self, + operation: OperationId, + objects: &[PhysicalObjectIntent], + ) -> Result { + let identity = encode_canonical(&(operation, objects))?; + ObjectPath::new(format!( + "{}{}.cbor", + self.prefix(), + hex::encode(sha256(&identity)) + )) + } + + fn completion_path( + &self, + operation: OperationId, + objects: &[PhysicalObjectCompletion], + ) -> Result { + let objects = objects + .iter() + .map(|object| PhysicalObjectIntent { + path: object.path.clone(), + size: object.size, + checksum_sha256: object.checksum_sha256, + }) + .collect::>(); + self.completion_path_for_intents(operation, &objects) + } + + pub(crate) fn completion_path_for_intents( + &self, + operation: OperationId, + objects: &[PhysicalObjectIntent], + ) -> Result { + let identity = encode_canonical(&(operation, objects))?; + ObjectPath::new(format!( + "{}{}.cbor", + self.completion_prefix(), + hex::encode(sha256(&identity)) + )) + } +} + +pub(crate) fn journal_list_request( + prefix: String, + continuation: Option, + limit: usize, +) -> ListRequest { + ListRequest { + prefix, + continuation, + limit, + include_versions: false, + } +} diff --git a/extensions/s3/core/src/repository.rs b/extensions/s3/core/src/repository.rs index a79e90b1..e1f1e721 100644 --- a/extensions/s3/core/src/repository.rs +++ b/extensions/s3/core/src/repository.rs @@ -15,8 +15,14 @@ use prolly::{ }; use sha2::Sha256; -use crate::gc::{GcCandidate, GcCoordinator, GcDirtyRoot, GcNodeWork, GcPublicationTicket}; +use crate::gc::{ + GcCandidate, GcCandidateDiscovery, GcCandidateNamespace, GcCoordinator, GcDirtyRoot, + GcInventorySource, GcNodeWork, GcPublicationTicket, +}; use crate::merge::{MergeBaseCandidate, MergePlanEntry, MergeQueueEntry, MergeSeenEntry}; +use crate::physical_journal::{ + journal_list_request, PhysicalObjectCompletion, PhysicalObjectIntent, PhysicalObjectJournal, +}; use crate::publication::BranchMovement; use crate::store::{LocatedPackedNode, NodeCacheNamespace, NodeLocator, PreparedNodePack}; use crate::transfer::{commit_mapping_key, version_mapping_key}; @@ -42,7 +48,7 @@ use crate::{ ProviderPerKeyVersionLimit, RandomIdSource, RefCatalogCursor, RefGeneration, RefKind, RepositoryFormat, Result, RootManifest, SegmentedOperationIndex, ShardWriterAuthority, ShardedBranchPublisher, ShardedRefCatalog, StagedMutation, StagedMutationBody, StagedPut, - SystemClock, TagStore, TakeoverRequest, + StorageToken, SystemClock, TagStore, TakeoverRequest, }; /// Keep ordinary commit descriptors small enough for one bounded metadata @@ -657,6 +663,13 @@ struct GcProcessState { publication_barrier: Arc>, } +struct GcInventoryResolved { + path: ObjectPath, + size: u64, + physical_version: PhysicalVersion, + last_modified_millis: u64, +} + fn gc_process_state(repository: crate::RepositoryId) -> Arc { static STATES: OnceLock>>> = OnceLock::new(); @@ -824,6 +837,7 @@ pub struct Repository { publisher: ShardedBranchPublisher

, payloads: ImmutablePayloadStore

, commit_sessions: CommitSessionStore

, + physical_journal: PhysicalObjectJournal

, tags: TagStore

, ref_catalog: Arc>, operation_index: SegmentedOperationIndex

, @@ -1090,6 +1104,12 @@ impl Repository

{ format.repository_id, format.canonical_limits.max_mutations_per_commit as usize, )?; + let physical_journal = PhysicalObjectJournal::new( + plane.clone(), + options.repository_prefix.clone(), + format.repository_id, + format.canonical_limits.max_mutations_per_commit as usize, + )?; let operation_index = SegmentedOperationIndex::new_with_limits( plane.clone(), options.repository_prefix.clone(), @@ -1131,6 +1151,7 @@ impl Repository

{ publisher, payloads, commit_sessions, + physical_journal, tags, ref_catalog, operation_index, @@ -1891,6 +1912,8 @@ impl Repository

{ } } + self.record_payload_intents(session, &objects).await?; + let staged = stream::iter(objects) .map(|(key, bytes, headers, user_metadata)| async move { self.stage_commit_session_put_validated(key, bytes, headers, user_metadata) @@ -1901,6 +1924,7 @@ impl Repository

{ .await; let mut staged = staged.into_iter().collect::>>()?; staged.sort_by(|left, right| left.key().cmp(right.key())); + self.record_payload_completions(session, &staged).await?; Ok(staged) } @@ -1921,6 +1945,24 @@ impl Repository

{ "payload staging concurrency is outside 1..=1024", )); } + let valid_intents = objects + .iter() + .filter_map(|(key, bytes, _, _)| { + self.validate_key(key) + .ok() + .filter(|_| bytes.len() as u64 <= self.format.canonical_limits.max_object_bytes) + .and_then(|_| self.payload_intent(bytes).ok()) + }) + .collect::>(); + if !valid_intents.is_empty() { + self.physical_journal + .record( + session.identity.operation, + session.created_at_millis, + valid_intents, + ) + .await?; + } let mut staged = stream::iter(objects.into_iter().enumerate()) .map(|(index, (key, bytes, headers, user_metadata))| async move { ( @@ -1933,9 +1975,74 @@ impl Repository

{ .collect::>() .await; staged.sort_by_key(|(index, _)| *index); + let completed = staged + .iter() + .filter_map(|(_, result)| result.as_ref().ok()) + .cloned() + .collect::>(); + if !completed.is_empty() { + self.record_payload_completions(session, &completed).await?; + } Ok(staged.into_iter().map(|(_, result)| result).collect()) } + async fn record_payload_intents( + &self, + session: &CommitSessionManifest, + objects: &[CommitSessionPutInput], + ) -> Result<()> { + if objects.is_empty() { + return Ok(()); + } + let intents = objects + .iter() + .map(|(_, bytes, _, _)| self.payload_intent(bytes)) + .collect::>>()?; + self.physical_journal + .record( + session.identity.operation, + session.created_at_millis, + intents, + ) + .await + } + + fn payload_intent(&self, bytes: &[u8]) -> Result { + let checksum_sha256 = crate::codec::sha256(bytes); + Ok(PhysicalObjectIntent { + path: self.payloads.path(checksum_sha256)?, + size: bytes.len() as u64, + checksum_sha256, + }) + } + + async fn record_payload_completions( + &self, + session: &CommitSessionManifest, + mutations: &[StagedMutation], + ) -> Result<()> { + let completed_at_millis = self.options.clock.now_millis()?; + let objects = mutations + .iter() + .filter_map(|mutation| match &mutation.body { + StagedMutationBody::Put(staged) => Some(PhysicalObjectCompletion { + path: staged.binding.path.clone(), + size: staged.size, + checksum_sha256: staged.binding.checksum_sha256, + provider_version_id: staged.binding.provider_version_id.clone(), + provider_etag: staged.binding.provider_etag.clone(), + }), + StagedMutationBody::Delete { .. } => None, + }) + .collect::>(); + if objects.is_empty() { + return Ok(()); + } + self.physical_journal + .record_completion(session.identity.operation, completed_at_millis, objects) + .await + } + #[allow(clippy::too_many_arguments)] pub async fn stage_commit_session_file( &self, @@ -3637,6 +3744,25 @@ impl Repository

{ /// Start a bounded concurrent collector for immutable repository data. /// `grace_millis` must exceed the longest allowed unpublished operation. pub async fn start_gc(&self, grace_millis: u64) -> Result { + self.start_gc_with_discovery(grace_millis, GcCandidateDiscovery::LegacyScan) + .await + } + + /// Start GC using immutable physical-object creation intents for payload + /// candidates. Payloads created by direct/legacy writers are intentionally + /// retained rather than listed in this mode; run the default `start_gc` + /// epoch during migration. Commits and node objects remain discovered by + /// their normal namespace scan. + pub async fn start_gc_journaled(&self, grace_millis: u64) -> Result { + self.start_gc_with_discovery(grace_millis, GcCandidateDiscovery::JournalOnly) + .await + } + + async fn start_gc_with_discovery( + &self, + grace_millis: u64, + payload_discovery: GcCandidateDiscovery, + ) -> Result { if grace_millis == 0 { return Err(Error::new( ErrorCode::InvalidLimit, @@ -3684,6 +3810,10 @@ impl Repository

{ cutoff_millis, phase: GcPhase::DiscoverBranches, continuation: None, + payload_discovery, + candidate_namespace: GcCandidateNamespace::Repository, + journal_object_offset: 0, + inventory_source: GcInventorySource::Completions, work: RootManifest::from_tree(&work)?, dirty_sequence: self.gc_dirty_sequence.load(Ordering::Acquire), dirty_target_sequence: 0, @@ -3822,6 +3952,7 @@ impl Repository

{ GcPhase::DiscoverTags => self.gc_discover_refs(&mut next, true, max_steps).await?, GcPhase::MarkCommits => self.gc_mark_commits(&mut next, max_steps).await?, GcPhase::MarkNodes => self.gc_mark_nodes(&mut next, max_steps).await?, + GcPhase::ScanInventory => self.gc_scan_inventory(&mut next, max_steps).await?, GcPhase::ScanCandidates => self.gc_scan_candidates(&mut next, max_steps).await?, GcPhase::CatchUpDirtyRoots => { self.gc_catch_up_dirty_roots(&mut next, max_steps).await? @@ -7184,6 +7315,8 @@ impl Repository

{ if records.is_empty() { cursor.phase = if cursor.initial_scan_complete { GcPhase::CatchUpDirtyRoots + } else if cursor.payload_discovery == GcCandidateDiscovery::JournalOnly { + GcPhase::ScanInventory } else { GcPhase::ScanCandidates }; @@ -7272,11 +7405,262 @@ impl Repository

{ Ok(mark_keys.len()) } + async fn gc_scan_inventory(&self, cursor: &mut GcCursor, max_steps: usize) -> Result { + let prefix = match cursor.inventory_source { + GcInventorySource::Completions => self.physical_journal.completion_prefix(), + GcInventorySource::Intents => self.physical_journal.prefix(), + }; + let page = self + .plane + .list(journal_list_request(prefix, cursor.continuation.clone(), 1)) + .await?; + let Some(entry) = page.entries.into_iter().next() else { + cursor.continuation = None; + cursor.journal_object_offset = 0; + match cursor.inventory_source { + GcInventorySource::Completions => { + cursor.inventory_source = GcInventorySource::Intents; + } + GcInventorySource::Intents => { + cursor.phase = GcPhase::ScanCandidates; + cursor.candidate_namespace = GcCandidateNamespace::Commits; + } + } + return Ok(0); + }; + let engine = self.gc_work_engine(cursor.epoch)?; + let mut tree = self.tree_from_root(&cursor.work)?; + let (all_intents, completion_marker, completed_paths) = match cursor.inventory_source { + GcInventorySource::Completions => { + let path = entry.path; + let batch = self.physical_journal.load_completion(path.clone()).await?; + let completed_at_millis = batch.completed_at_millis; + let intents = batch + .objects + .iter() + .map(|object| crate::physical_journal::PhysicalObjectIntent { + path: object.path.clone(), + size: object.size, + checksum_sha256: object.checksum_sha256, + }) + .collect::>(); + let resolved = batch + .objects + .into_iter() + .map(|object| { + let token = StorageToken { + etag: object.provider_etag, + version_id: object.provider_version_id.clone(), + }; + GcInventoryResolved { + path: object.path, + size: object.size, + physical_version: object + .provider_version_id + .map(|version_id| PhysicalVersion::Versioned { version_id }) + .unwrap_or_else(|| PhysicalVersion::Unversioned { + token: Some(token), + }), + last_modified_millis: completed_at_millis, + } + }) + .collect::>(); + ( + intents, + Some((gc_journal_completion_key(&path), resolved)), + None, + ) + } + GcInventorySource::Intents => { + let batch = self.physical_journal.load(entry.path).await?; + let completion_path = self + .physical_journal + .completion_path_for_intents(batch.operation, &batch.objects)?; + let marker = gc_journal_completion_key(&completion_path); + if engine.get(&tree, &marker).await?.is_some() { + let completion = self + .physical_journal + .load_completion(completion_path) + .await?; + let completed_paths = completion + .objects + .into_iter() + .map(|object| object.path) + .collect::>(); + if completed_paths.len() == batch.objects.len() + && batch + .objects + .iter() + .all(|intent| completed_paths.contains(&intent.path)) + { + cursor.continuation = page.continuation; + cursor.journal_object_offset = 0; + return Ok(batch.objects.len()); + } + (batch.objects, None, Some(completed_paths)) + } else { + (batch.objects, None, None) + } + } + }; + if cursor.journal_object_offset == 0 { + cursor.report.journal_batches = cursor.report.journal_batches.saturating_add(1); + } + let start = cursor.journal_object_offset.min(all_intents.len()); + let end = start.saturating_add(max_steps).min(all_intents.len()); + let intents = &all_intents[start..end]; + cursor.report.journal_objects = cursor + .report + .journal_objects + .saturating_add(intents.len() as u64); + + let path_keys = intents + .iter() + .map(|intent| gc_path_mark_key(&intent.path)) + .collect::>(); + let path_marks = engine.get_many(&tree, &path_keys).await?; + let pending = intents + .iter() + .zip(path_marks) + .filter_map(|(intent, mark)| { + (mark.is_none() + && completed_paths + .as_ref() + .is_none_or(|completed| !completed.contains(&intent.path))) + .then_some(intent.clone()) + }) + .collect::>(); + let mut resolved = if let Some((marker, resolved)) = completion_marker { + tree = engine + .batch( + &tree, + vec![Mutation::Upsert { + key: marker, + val: Vec::new(), + }], + ) + .await?; + resolved + .into_iter() + .filter(|resolved| intents.iter().any(|intent| intent.path == resolved.path)) + .collect::>() + } else { + let metadata = stream::iter(pending.into_iter().map(|intent| async move { + let head = self.plane.head(&intent.path).await?; + Ok::<_, Error>((intent, head)) + })) + .buffered(32) + .collect::>() + .await; + let mut resolved = Vec::new(); + for result in metadata { + let (intent, Some(metadata)) = result? else { + cursor.report.already_missing = cursor.report.already_missing.saturating_add(1); + continue; + }; + if metadata.delete_marker { + continue; + } + if metadata.len != intent.size + || (metadata.sha256 != [0; 32] && metadata.sha256 != intent.checksum_sha256) + { + return Err(Error::new( + ErrorCode::CorruptContent, + format!( + "physical-object journal identity disagrees with {}", + intent.path + ), + )); + } + if metadata.last_modified_millis <= cursor.cutoff_millis { + let physical_version = metadata + .token + .version_id + .clone() + .map(|version_id| PhysicalVersion::Versioned { version_id }) + .unwrap_or_else(|| PhysicalVersion::Unversioned { + token: Some(metadata.token.clone()), + }); + resolved.push(GcInventoryResolved { + path: intent.path, + size: metadata.len, + physical_version, + last_modified_millis: metadata.last_modified_millis, + }); + } + } + resolved + }; + resolved.retain(|resolved| resolved.last_modified_millis <= cursor.cutoff_millis); + let physical_keys = resolved + .iter() + .map(|resolved| match &resolved.physical_version { + PhysicalVersion::Versioned { version_id } => { + gc_physical_mark_key(&resolved.path, version_id) + } + PhysicalVersion::Unversioned { .. } => gc_path_mark_key(&resolved.path), + }) + .collect::>(); + let physical_marks = engine.get_many(&tree, &physical_keys).await?; + let mut mutations = Vec::new(); + for (resolved, physical_mark) in resolved.into_iter().zip(physical_marks) { + if physical_mark.is_some() { + continue; + } + let candidate = GcCandidate { + path: resolved.path, + physical_version: resolved.physical_version, + len: resolved.size, + last_modified_millis: resolved.last_modified_millis, + kind: "payloads".to_string(), + }; + let key = gc_candidate_key(&candidate)?; + mutations.push(Mutation::Upsert { + key, + val: encode_canonical(&candidate)?, + }); + cursor.report.candidates = cursor.report.candidates.saturating_add(1); + cursor.report.candidate_bytes = + cursor.report.candidate_bytes.saturating_add(candidate.len); + *cursor + .report + .candidates_by_kind + .entry("payloads".to_string()) + .or_default() += 1; + } + if !mutations.is_empty() { + tree = engine.batch(&tree, mutations).await?; + } + cursor.work = RootManifest::from_tree(&tree)?; + if end == all_intents.len() { + cursor.continuation = page.continuation; + cursor.journal_object_offset = 0; + } else { + cursor.journal_object_offset = end; + } + Ok(intents.len()) + } + async fn gc_scan_candidates(&self, cursor: &mut GcCursor, max_steps: usize) -> Result { + let (prefix, namespace) = match cursor.candidate_namespace { + GcCandidateNamespace::Repository => ( + format!("{}/", self.options.repository_prefix), + GcCandidateNamespace::Repository, + ), + GcCandidateNamespace::Commits => ( + format!("{}/commits/sha256/", self.options.repository_prefix), + GcCandidateNamespace::Commits, + ), + GcCandidateNamespace::Nodes => ( + format!("{}/nodes/sha256/", self.options.repository_prefix), + GcCandidateNamespace::Nodes, + ), + GcCandidateNamespace::Complete => return Ok(0), + }; let page = self .plane .list(ListRequest { - prefix: format!("{}/", self.options.repository_prefix), + prefix, continuation: cursor.continuation.clone(), limit: max_steps, include_versions: true, @@ -7289,6 +7673,11 @@ impl Repository

{ .iter() .filter_map(|entry| { let kind = gc_managed_kind(&self.options.repository_prefix, &entry.path)?; + if cursor.payload_discovery == GcCandidateDiscovery::JournalOnly + && kind == "payloads" + { + return None; + } (entry.metadata.last_modified_millis <= cursor.cutoff_millis) .then_some((entry, kind)) }) @@ -7356,8 +7745,22 @@ impl Repository

{ cursor.work = RootManifest::from_tree(&tree)?; cursor.continuation = page.continuation; if cursor.continuation.is_none() { - cursor.initial_scan_complete = true; - cursor.phase = GcPhase::CatchUpDirtyRoots; + match namespace { + GcCandidateNamespace::Repository => { + cursor.initial_scan_complete = true; + cursor.phase = GcPhase::CatchUpDirtyRoots; + } + GcCandidateNamespace::Commits => { + cursor.candidate_namespace = GcCandidateNamespace::Nodes; + cursor.continuation = None; + } + GcCandidateNamespace::Nodes => { + cursor.candidate_namespace = GcCandidateNamespace::Complete; + cursor.initial_scan_complete = true; + cursor.phase = GcPhase::CatchUpDirtyRoots; + } + GcCandidateNamespace::Complete => {} + } } Ok(page.entries.len()) } @@ -10131,6 +10534,12 @@ fn gc_physical_mark_key(path: &ObjectPath, version: &str) -> Vec { key } +fn gc_journal_completion_key(path: &ObjectPath) -> Vec { + let mut key = b"ji/".to_vec(); + key.extend_from_slice(&crate::codec::sha256(path.as_str().as_bytes())); + key +} + fn gc_candidate_key(candidate: &GcCandidate) -> Result> { let mut key = b"d/".to_vec(); key.extend_from_slice(&crate::codec::sha256(&encode_canonical(candidate)?)); diff --git a/extensions/s3/core/tests/gc.rs b/extensions/s3/core/tests/gc.rs index 76f91fc4..39bf7e97 100644 --- a/extensions/s3/core/tests/gc.rs +++ b/extensions/s3/core/tests/gc.rs @@ -1,8 +1,8 @@ use std::{collections::BTreeMap, sync::Arc, time::Duration}; use prolly_s3_core::{ - GcPhase, ImmutablePut, MemoryObjectPlane, ObjectHeaders, ObjectPath, ObjectPlane, Repository, - RepositoryOptions, + FixedClock, GcPhase, ImmutablePut, ListRequest, MemoryObjectPlane, ObjectHeaders, ObjectPath, + ObjectPlane, PhysicalVersion, Repository, RepositoryOptions, }; use sha2::{Digest, Sha256}; @@ -222,3 +222,163 @@ async fn gc_fences_cross_handle_publications_and_deletes_exact_orphans() { ); repository.commit(pinned).await.unwrap(); } + +#[tokio::test] +async fn journaled_gc_discovers_unpublished_batch_payloads_without_namespace_payload_scan() { + let plane = Arc::new(MemoryObjectPlane::new(true)); + let clock = Arc::new(FixedClock::new(10_000)); + let options = RepositoryOptions { + repository_prefix: ".tests/gc-journal".to_string(), + clock: clock.clone(), + provider_per_key_version_limit: prolly_s3_core::ProviderPerKeyVersionLimit::Finite(10_000), + ..RepositoryOptions::default() + }; + let repository = Repository::initialize(plane.clone(), options) + .await + .unwrap(); + let session = repository + .begin_commit_session("main", "journaled orphan", 60_000) + .await + .unwrap(); + let _staged = repository + .stage_commit_session_put_batch( + &session, + vec![ + ( + b"orphan-a".to_vec(), + b"alpha".to_vec(), + ObjectHeaders::default(), + BTreeMap::new(), + ), + ( + b"orphan-b".to_vec(), + b"bravo".to_vec(), + ObjectHeaders::default(), + BTreeMap::new(), + ), + ], + 2, + ) + .await + .unwrap(); + let orphan_paths = plane + .list(ListRequest { + prefix: ".tests/gc-journal/payloads/".to_string(), + continuation: None, + limit: 10, + include_versions: false, + }) + .await + .unwrap() + .entries + .into_iter() + .map(|entry| entry.path) + .collect::>(); + assert_eq!(orphan_paths.len(), 2); + + plane.reset_request_counts(); + clock.advance(10).unwrap(); + let mut gc = repository.start_gc_journaled(1).await.unwrap(); + for _ in 0..20_000 { + gc = match gc.phase { + GcPhase::Ready | GcPhase::Sweeping => { + repository.sweep_gc(&gc, 100).await.unwrap().cursor + } + GcPhase::Complete => break, + _ => { + repository + .advance_gc(&gc, 100) + .await + .unwrap_or_else(|error| panic!("phase {:?}: {error:?}", gc.phase)) + .cursor + } + }; + } + assert_eq!(gc.phase, GcPhase::Complete); + assert_eq!(gc.report.journal_batches, 1); + assert_eq!(gc.report.journal_objects, 2); + assert_eq!(plane.request_snapshot().head, 0); + assert_eq!(gc.report.candidates_by_kind.get("payloads"), Some(&2)); + for path in orphan_paths { + assert!(plane.head(&path).await.unwrap().is_none()); + } +} + +#[tokio::test] +async fn journaled_gc_falls_back_to_precompletion_intents_after_an_interrupted_upload() { + let plane = Arc::new(MemoryObjectPlane::new(true)); + let clock = Arc::new(FixedClock::new(20_000)); + let options = RepositoryOptions { + repository_prefix: ".tests/gc-journal-intent".to_string(), + clock: clock.clone(), + provider_per_key_version_limit: prolly_s3_core::ProviderPerKeyVersionLimit::Finite(10_000), + ..RepositoryOptions::default() + }; + let repository = Repository::initialize(plane.clone(), options) + .await + .unwrap(); + let session = repository + .begin_commit_session("main", "interrupted journal completion", 60_000) + .await + .unwrap(); + repository + .stage_commit_session_put_batch( + &session, + vec![( + b"orphan".to_vec(), + b"payload".to_vec(), + ObjectHeaders::default(), + BTreeMap::new(), + )], + 1, + ) + .await + .unwrap(); + let completion = plane + .list(ListRequest { + prefix: ".tests/gc-journal-intent/administration/physical-object-journal/".to_string(), + continuation: None, + limit: 10, + include_versions: true, + }) + .await + .unwrap() + .entries + .into_iter() + .find(|entry| entry.path.as_str().contains("/completions/")) + .unwrap(); + let completion_version = completion + .metadata + .token + .version_id + .clone() + .map(|version_id| PhysicalVersion::Versioned { version_id }) + .unwrap_or_else(|| PhysicalVersion::Unversioned { + token: Some(completion.metadata.token.clone()), + }); + assert_eq!( + plane + .delete_exact(&completion.path, completion_version) + .await + .unwrap(), + prolly_s3_core::DeleteOutcome::Deleted + ); + + plane.reset_request_counts(); + clock.advance(10).unwrap(); + let mut gc = repository.start_gc_journaled(1).await.unwrap(); + for _ in 0..20_000 { + gc = match gc.phase { + GcPhase::Ready | GcPhase::Sweeping => { + repository.sweep_gc(&gc, 100).await.unwrap().cursor + } + GcPhase::Complete => break, + _ => repository.advance_gc(&gc, 100).await.unwrap().cursor, + }; + } + assert_eq!(gc.phase, GcPhase::Complete); + assert_eq!(gc.report.journal_batches, 1); + assert_eq!(gc.report.journal_objects, 1); + assert_eq!(plane.request_snapshot().head, 1); + assert_eq!(gc.report.candidates_by_kind.get("payloads"), Some(&1)); +} diff --git a/extensions/s3/core/tests/repository.rs b/extensions/s3/core/tests/repository.rs index 8b7ddb33..d7ffd35a 100644 --- a/extensions/s3/core/tests/repository.rs +++ b/extensions/s3/core/tests/repository.rs @@ -1202,7 +1202,8 @@ async fn repository_batch_results_isolates_invalid_objects_after_one_session_val results[1].as_ref().unwrap_err().code, prolly_s3_core::ErrorCode::InvalidKey ); - assert_eq!(plane.request_snapshot().immutable_put, 2); + // Two whole-object payloads plus immutable intent and completion manifests. + assert_eq!(plane.request_snapshot().immutable_put, 4); let staged = results.into_iter().filter_map(Result::ok).collect(); let receipt = repository .publish_commit_session(session, staged) diff --git a/extensions/s3/spec/prolly-s3/paths.md b/extensions/s3/spec/prolly-s3/paths.md index 8b7700b2..16816c8e 100644 --- a/extensions/s3/spec/prolly-s3/paths.md +++ b/extensions/s3/spec/prolly-s3/paths.md @@ -26,6 +26,8 @@ repository format. | commit-closure work tree | `P/administration/closure/E/tree/nodes/sha256/...` | | fsck cursor | `P/administration/fsck/E/cursor.cbor` | | fsck distinct-payload work tree | `P/administration/fsck/E/payloads/nodes/sha256/...` | +| physical-object creation intent batch | `P/administration/physical-object-journal/R/intents/H.cbor` | +| physical-object completion batch | `P/administration/physical-object-journal/R/completions/H.cbor` | | GC coordinator | `P/gc/coordinator.cbor` | | GC epoch cursor | `P/gc/epochs/E/cursor.cbor` | | GC reachability work tree | `P/administration/gc/E/tree/nodes/sha256/...` | @@ -47,6 +49,13 @@ byte pairs. `SS` is a two-digit ref-catalog shard. bytes in one repository may reuse a complete payload object. A payload path never stores multiple logical bodies or a chunk of one logical body. - Delete markers do not have payload paths. +- Journaled bulk ingest writes one immutable creation-intent manifest per + bounded window before uploading its complete payload objects. The manifest + contains paths, sizes, and checksums only; it never packs, chunks, or stores + payload bytes. +- Journal-only GC is opt-in. It discovers payloads emitted by journaled batch + APIs and retains direct or pre-journal payloads; the default GC mode keeps a + legacy repository scan for complete migration coverage. - Ordinary reads and index maintenance must not discover nodes by namespace listing. - Mutable control records retain a bounded number of provider versions. diff --git a/extensions/s3/spec/prolly-s3/state-machines.md b/extensions/s3/spec/prolly-s3/state-machines.md index 60937bb0..e850b4e9 100644 --- a/extensions/s3/spec/prolly-s3/state-machines.md +++ b/extensions/s3/spec/prolly-s3/state-machines.md @@ -103,8 +103,10 @@ preserve commit topology. 2. Discover live branches, tags, and retention-pin tags in bounded LIST pages. 3. Mark the complete commit closure, direct nodes, packed commit containers, and exact payload versions in a job-scoped Prolly tree. -4. Scan only immutable commit, direct-node, and payload namespaces older than - the configured grace cutoff. +4. In legacy mode, scan immutable commit, direct-node, and payload namespaces + older than the configured grace cutoff. Journal-only mode scans commit and + direct-node namespaces and resolves payload candidates from immutable + ingest-window creation intents. 5. Fence publication, capture the dirty-root watermark, and catch up any new roots before deletion. 6. Recheck reachability and exact-delete a bounded physical-version batch. From 91eef63e38fb536d2edf62f43ea456ce3f5999ae Mon Sep 17 00:00:00 2001 From: forhappy Date: Sun, 16 Aug 2026 18:54:28 -0700 Subject: [PATCH 2/2] fix(s3): advance journal GC on terminal pages --- .../s3/client/tests/rustfs_repository.rs | 55 +++++++++++++++++++ extensions/s3/core/src/repository.rs | 15 +++++ 2 files changed, 70 insertions(+) diff --git a/extensions/s3/client/tests/rustfs_repository.rs b/extensions/s3/client/tests/rustfs_repository.rs index e017d642..fd385753 100644 --- a/extensions/s3/client/tests/rustfs_repository.rs +++ b/extensions/s3/client/tests/rustfs_repository.rs @@ -1017,6 +1017,61 @@ async fn rustfs_streaming_bulk_write_is_bounded_batched_and_ordered() { } } +#[tokio::test(flavor = "multi_thread", worker_threads = 8)] +async fn rustfs_journaled_gc_resolves_completed_batches_without_payload_heads() { + if !rustfs_enabled() { + eprintln!("set PROLLY_S3_RUSTFS=1 to run RustFS integration tests"); + return; + } + let (aws, bucket) = rustfs_client().await; + let repository_prefix = unique_name("journaled-gc"); + let client = Client::builder() + .aws_client(aws) + .bucket(&bucket) + .repository_prefix(&repository_prefix) + .writer("rustfs-journaled-gc-writer") + .provider_identity(provider_identity()) + .attestation_signer(attestation_signer()) + .provider_per_key_version_limit(ProviderPerKeyVersionLimit::Finite(10_000)) + .initialize() + .await + .unwrap(); + let objects = stream::iter((0..64).map(|index| { + Ok(PutObjectInput { + key: format!("journal/{index:04}.txt"), + bytes: format!("value-{index}").into_bytes(), + headers: Default::default(), + user_metadata: Default::default(), + }) + })); + client + .put_object_stream( + objects, + BulkWriteOptions { + batch_size: 32, + concurrency: 8, + checkpoint_every: 16, + }, + ) + .await + .unwrap(); + + client.reset_s3_operation_metrics(); + let mut gc = client.start_gc_journaled(1).await.unwrap(); + for _ in 0..1_000 { + gc = match gc.phase { + GcPhase::Ready | GcPhase::Sweeping => client.sweep_gc(&gc, 1_000).await.unwrap().cursor, + GcPhase::Complete => break, + _ => client.advance_gc(&gc, 1_000).await.unwrap().cursor, + }; + } + assert_eq!(gc.phase, GcPhase::Complete); + assert_eq!(gc.report.journal_objects, 64); + assert!(gc.report.journal_batches >= 4); + let metrics = client.reset_s3_operation_metrics(); + assert_eq!(metrics.head_object, 0); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 8)] async fn rustfs_ordered_publication_queue_groups_unique_keys_and_orders_duplicates() { if !rustfs_enabled() { diff --git a/extensions/s3/core/src/repository.rs b/extensions/s3/core/src/repository.rs index e1f1e721..bf8c133f 100644 --- a/extensions/s3/core/src/repository.rs +++ b/extensions/s3/core/src/repository.rs @@ -7495,6 +7495,10 @@ impl Repository

{ { cursor.continuation = page.continuation; cursor.journal_object_offset = 0; + if cursor.continuation.is_none() { + cursor.phase = GcPhase::ScanCandidates; + cursor.candidate_namespace = GcCandidateNamespace::Commits; + } return Ok(batch.objects.len()); } (batch.objects, None, Some(completed_paths)) @@ -7635,6 +7639,17 @@ impl Repository

{ if end == all_intents.len() { cursor.continuation = page.continuation; cursor.journal_object_offset = 0; + if cursor.continuation.is_none() { + match cursor.inventory_source { + GcInventorySource::Completions => { + cursor.inventory_source = GcInventorySource::Intents; + } + GcInventorySource::Intents => { + cursor.phase = GcPhase::ScanCandidates; + cursor.candidate_namespace = GcCandidateNamespace::Commits; + } + } + } } else { cursor.journal_object_offset = end; }