Drive GC payload discovery from ingest journals

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.

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

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<GcCursor> {
+        self.ensure_provider_qualified()?;
+        self.attached_branch()?;
+        self.repository.start_gc_journaled(grace_millis).await
+    }
+
     pub async fn resume_gc(&self) -> Result<Option<GcCursor>> {
         self.ensure_provider_qualified()?;
         self.attached_branch()?;

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() {

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<String>,
+    #[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<String, u64>,
     #[serde(default)]
     pub protected_by_kind: BTreeMap<String, u64>,
+    #[serde(default)]
+    pub journal_batches: u64,
+    #[serde(default)]
+    pub journal_objects: u64,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]

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,