test(cleanup): run the sweep crash windows on S3 and Azure

.github/workflows/ci.yml

@@ -1334,8 +1334,9 @@ jobs:
       - name: Run RustFS detached publication retention
         if: matrix.shard == 'failpoints'
         # Exercise detached write and compaction pins through S3 after a lost
-        # publication acknowledgement, then prove another write succeeds.
-        # Require both owners and reject unconfigured early returns.
+        # publication acknowledgement, then prove another write succeeds, and
+        # retry cleanup after a pass stopped in each window of its sweep.
+        # Require every owner and reject unconfigured early returns.
         run: |
           set -euo pipefail
           test_log="$RUNNER_TEMP/rustfs-failpoints.log"
@@ -1347,7 +1348,8 @@ jobs:
           fi
           for test_name in \
             s3_write_detached_pin_survives_next_write \
-            s3_optimize_detached_pin_survives_next_write
+            s3_optimize_detached_pin_survives_next_write \
+            s3_collector_retry_converges_after_every_sweep_window
           do
             grep -Fq "test ${test_name} ... ok" "$test_log" \
               || { echo "::error::required RustFS failpoint cell did not pass: $test_name"; exit 1; }
@@ -1548,6 +1550,7 @@ jobs:
           run_cell azure_write_detached_pin_survives_next_write --test failpoints
           run_cell azure_optimize_detached_pin_survives_next_write --test failpoints
           run_cell azure_schema_apply_post_commit_failure_leaves_readable_contract --test failpoints
+          run_cell azure_collector_retry_converges_after_every_sweep_window --test failpoints
           run_cell azure_cluster_full_lifecycle_import_apply_serve_evolve_delete --test s3_cluster
           run_cell s3::server_opens_azure_graph_directly_and_serves_snapshot_and_read --test integration
           run_cell s3::server_boots_azure_cluster_from_bare_storage_uri_and_serves_query --test integration

crates/omnigraph/src/db/omnigraph/collector.rs

@@ -58,6 +58,12 @@ decide_seam! {
     pub static CLEANUP_SWEEP_PRE_FILE_DELETE = ("cleanup.sweep_pre_file_delete", Unreachable, [Fail]);
 }
 
+decide_seam! {
+    /// After the files only a table's dead stagings reference are deleted and
+    /// before the first of their manifests is.
+    pub static CLEANUP_SWEEP_PRE_DEAD_MANIFEST_DELETE = ("cleanup.sweep_pre_dead_manifest_delete", Unreachable, [Fail]);
+}
+
 /// Requests the run made, for the two ceilings in `maintenance.rs`.
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct CollectorCost {
@@ -319,6 +325,7 @@ pub(crate) async fn sweep_table(
     let dead = plan.dead_stagings();
     if !dead.is_empty() {
         stats.bytes_removed += delete_paths(&store, &base, &plan.dead_paths).await?;
+        fail(&CLEANUP_SWEEP_PRE_DEAD_MANIFEST_DELETE)?;
     }
     for version in dead {
         let manifest = naming.manifest_path(&base, version);

crates/omnigraph/src/seams/catalog.rs

@@ -87,6 +87,7 @@ omnigraph_seams::catalog! {
     crate::db::omnigraph::collector::CLEANUP_SWEEP_PRE_MANIFEST_DELETE,
     crate::db::omnigraph::collector::CLEANUP_SWEEP_BETWEEN_MANIFESTS,
     crate::db::omnigraph::collector::CLEANUP_SWEEP_PRE_FILE_DELETE,
+    crate::db::omnigraph::collector::CLEANUP_SWEEP_PRE_DEAD_MANIFEST_DELETE,
 }
 
 /// The one string-keyed lookup: a case file or the harness names a seam, the catalog answers.