Recovery liveness, storage fault-injection matrix, and one storage implementation over object_store

.github/workflows/ci.yml

@@ -365,6 +365,20 @@ jobs:
       - name: Run RustFS CLI smoke
         run: cargo test --locked -p omnigraph-cli --test system_local local_cli_s3_end_to_end_init_load_read_flow -- --nocapture
 
+      - name: Run RustFS recovery-sidecar lifecycle
+        # Sidecar put/list/delete through the S3 storage backend on a
+        # real bucket (the failpoint only wedges the publisher; the
+        # sidecar I/O is exercised for real). Name filter `s3_` matches
+        # the bucket-gated tests in the failpoints target only; the
+        # grep guards against the filter going vacuous (cargo passes
+        # with 0 tests matched) if those tests are ever renamed.
+        run: |
+          output=$(cargo test --locked -p omnigraph-engine --features failpoints --test failpoints s3_ -- --nocapture 2>&1); status=$?
+          echo "$output"
+          [ "$status" -eq 0 ] || exit "$status"
+          echo "$output" | grep -Eq "test result: ok\. [1-9][0-9]* passed" \
+            || { echo "::error::filter 's3_' matched no tests — vacuous pass"; exit 1; }
+
       - name: Dump RustFS logs on failure
         if: failure()
         run: docker logs rustfs

AGENTS.md

@@ -240,7 +240,7 @@ omnigraph policy explain --actor act-alice --action change --branch main
 | Columnar storage on object store | ✅ Arrow/Lance | URI normalization, S3 env-var plumbing |
 | Per-dataset versioning + time travel | ✅ | `snapshot_at_version`, `entity_at`, snapshot-pinned reads across many tables |
 | Per-dataset branches | ✅ | **Graph-level** branches (atomic across all sub-tables), lazy fork, system branch filtering |
-| Atomic single-dataset commits | ✅ | **Multi-table publish via three layers**, NOT a single Lance primitive: (1) per-table Lance `commit_staged` for the data write, (2) `__manifest` row-level CAS via `ManifestBatchPublisher` for cross-table ordering, (3) the open-time recovery sweep for the residual gap between (1) and (2). All three layers ship; the five migrated writers (`MutationStaging::finalize`, `schema_apply`, `branch_merge`, `ensure_indices`, `optimize_all_tables`) write a `__recovery/{ulid}.json` sidecar before Phase B and delete it after Phase C. The next `Omnigraph::open` (gated on `OpenMode::ReadWrite`) runs the sweep in `db/manifest/recovery.rs`: classify, decide all-or-nothing per sidecar, roll forward via single `ManifestBatchPublisher::publish` or roll back via `Dataset::restore` followed by a manifest publish of the restored version (so both directions converge to `manifest == HEAD` — no residual drift), and record an audit row in `_graph_commit_recoveries.lance` (queryable via `omnigraph commit list --filter actor=omnigraph:recovery`). Continuous in-process recovery (no restart needed between Phase B failure and recovery) is the goal of a future background reconciler. Engine writes route through a sealed `TableStorage` trait (`db.storage()`) exposing only `stage_*` + `commit_staged` + reads; the inline-commit residuals (`delete_where`, `create_vector_index`) are split onto a separate sealed `InlineCommitResidual` trait reached via `db.storage_inline_residual()` (MR-854), so the default surface cannot couple a write with a HEAD advance — §1 holds by construction. `delete_where` and `create_vector_index` stay inline until upstream Lance ships a public two-phase API ([#6658](https://github.com/lance-format/lance/issues/6658), [#6666](https://github.com/lance-format/lance/issues/6666)); `LoadMode::Overwrite` uses Lance `Overwrite` staged transactions. |
+| Atomic single-dataset commits | ✅ | **Multi-table publish via three layers**, NOT a single Lance primitive: (1) per-table Lance `commit_staged` for the data write, (2) `__manifest` row-level CAS via `ManifestBatchPublisher` for cross-table ordering, (3) the open-time recovery sweep for the residual gap between (1) and (2). All three layers ship; the five migrated writers (`MutationStaging::finalize`, `schema_apply`, `branch_merge`, `ensure_indices`, `optimize_all_tables`) write a `__recovery/{ulid}.json` sidecar before Phase B and delete it after Phase C. The next `Omnigraph::open` (gated on `OpenMode::ReadWrite`) runs the sweep in `db/manifest/recovery.rs`: classify, decide all-or-nothing per sidecar, roll forward via single `ManifestBatchPublisher::publish` or roll back via `Dataset::restore` followed by a manifest publish of the restored version (so both directions converge to `manifest == HEAD` — no residual drift), and record an audit row in `_graph_commit_recoveries.lance` (queryable via `omnigraph commit list --filter actor=omnigraph:recovery`). The write entry points (`load_as`, `mutate_as`, `apply_schema_as`, `branch_merge_as`) and `refresh` additionally run an in-process roll-forward-only heal (serialized against live writers via the per-table write queues), so a long-lived server converges on its next write without restart; only rollback-eligible sidecars still defer to the next read-write open (a future background reconciler's goal). Engine writes route through a sealed `TableStorage` trait (`db.storage()`) exposing only `stage_*` + `commit_staged` + reads; the inline-commit residuals (`delete_where`, `create_vector_index`) are split onto a separate sealed `InlineCommitResidual` trait reached via `db.storage_inline_residual()` (MR-854), so the default surface cannot couple a write with a HEAD advance — §1 holds by construction. `delete_where` and `create_vector_index` stay inline until upstream Lance ships a public two-phase API ([#6658](https://github.com/lance-format/lance/issues/6658), [#6666](https://github.com/lance-format/lance/issues/6666)); `LoadMode::Overwrite` uses Lance `Overwrite` staged transactions. |
 | Compaction (`compact_files`) | ✅ | `omnigraph optimize` orchestrates over all node/edge tables, bounded concurrency; **publishes each compacted table's new version to `__manifest`** (so the manifest tracks the Lance HEAD — required for reads to observe compaction and for schema apply / strict writes to pass their HEAD-vs-manifest precondition), under the per-`(table, main)` write queue with `SidecarKind::Optimize` recovery coverage; **refuses on an unrecovered graph** (errors if a `__recovery` sidecar is pending); **skips uncovered HEAD > manifest drift** with `DriftNeedsRepair` instead of interpreting it; **skips blob-bearing tables** (reported via `TableOptimizeStats.skipped`, not silent), gated on `LANCE_SUPPORTS_BLOB_COMPACTION` until the upstream blob-v2 compaction-decode bug is fixed (see [docs/dev/invariants.md](docs/dev/invariants.md) Known Gaps) |
 | Repair uncovered drift | — | `omnigraph repair` explicitly classifies uncovered table `HEAD > manifest` drift: verified maintenance drift (`ReserveFragments`/`Rewrite`) can be published with `--confirm`; suspicious or unverifiable drift requires `--force --confirm`. Sidecar-covered crash residuals still recover automatically on open. |
 | Cleanup (`cleanup_old_versions`) | ✅ | `omnigraph cleanup` with `--keep` / `--older-than` policy |

Cargo.toml

@@ -63,7 +63,7 @@ base64 = "0.22"
 ariadne = "0.4"
 regex = "1"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
-object_store = { version = "0.12.5", default-features = false, features = ["aws"] }
+object_store = { version = "0.12.5", default-features = false, features = ["aws", "fs"] }
 fail = "0.5"
 time = { version = "0.3", features = ["formatting"] }
 axum = { version = "0.8", features = ["json", "macros"] }

crates/omnigraph-cluster/src/store.rs

@@ -169,31 +169,16 @@ impl ClusterStore {
             .map_err(|err| err.to_string())
     }
 
-    /// JSON object write with the strongest atomicity the backend offers:
-    /// temp + rename on the filesystem (no torn JSON after a crash; the
-    /// pre-port behavior), a single atomic PUT on object stores (where
-    /// copy+delete would be weaker, not stronger).
+    /// JSON object write. Atomic visibility is the storage adapter's
+    /// contract on every backend (staged temp + rename on the filesystem,
+    /// a single atomic PUT on object stores) — no torn JSON after a crash,
+    /// no per-backend branch needed here.
     async fn put_json(&self, relative: &str, payload: &str) -> Result<(), String> {
         let target = self.uri(relative);
-        match self.kind() {
-            StorageKind::Local => {
-                let tmp = format!("{target}.tmp.{}", Ulid::new());
-                self.adapter
-                    .write_text(&tmp, payload)
-                    .await
-                    .map_err(|err| err.to_string())?;
-                if let Err(err) = self.adapter.rename_text(&tmp, &target).await {
-                    let _ = self.adapter.delete(&tmp).await;
-                    return Err(err.to_string());
-                }
-                Ok(())
-            }
-            StorageKind::S3 => self
-                .adapter
-                .write_text(&target, payload)
-                .await
-                .map_err(|err| err.to_string()),
-        }
+        self.adapter
+            .write_text(&target, payload)
+            .await
+            .map_err(|err| err.to_string())
     }
 
     /// Shared list-and-parse for the sidecar/approval directories: id

crates/omnigraph-cluster/src/tests.rs

@@ -1950,13 +1950,29 @@ graphs:
     }
 
     #[tokio::test]
+    #[cfg(unix)]
     async fn refresh_flags_unreadable_payload_as_error() {
         let dir = fixture();
         init_derived_graph(dir.path()).await;
         let blob = converge_fixture(dir.path()).await;
-        // A same-named directory yields a non-NotFound IO error portably.
-        fs::remove_file(&blob).unwrap();
-        fs::create_dir(&blob).unwrap();
+        // Make the payload unreadable without removing it: permission
+        // denied is a genuine non-NotFound IO error. (A same-named
+        // directory no longer triggers this path: object-store semantics
+        // classify a directory at an object path as NotFound — "only
+        // objects exist" — which is the missing-payload case, not the
+        // unreadable one.)
+        let mut perms = fs::metadata(&blob).unwrap().permissions();
+        std::os::unix::fs::PermissionsExt::set_mode(&mut perms, 0o000);
+        fs::set_permissions(&blob, perms).unwrap();
+        // Root reads straight through mode 000 (container dev runners
+        // commonly run as root): skip rather than fail — the contract
+        // under test needs a genuine permission error.
+        if fs::read(&blob).is_ok() {
+            eprintln!(
+                "skipping refresh_flags_unreadable_payload_as_error:                  running as root (mode 000 is still readable)"
+            );
+            return;
+        }
 
         let out = refresh_config_dir(dir.path()).await;
         assert!(!out.ok);

crates/omnigraph/src/db/manifest.rs

@@ -36,7 +36,8 @@ use publisher::{GraphNamespacePublisher, ManifestBatchPublisher};
 pub(crate) use recovery::{
     RecoveryMode, RecoverySidecar, RecoverySidecarHandle, SidecarKind, SidecarTablePin,
     SidecarTableRegistration, SidecarTombstone, delete_sidecar, has_schema_apply_sidecar,
-    list_sidecars, new_sidecar, recover_manifest_drift, write_sidecar,
+    heal_pending_sidecars_roll_forward, list_sidecars, new_sidecar, recover_manifest_drift,
+    schema_apply_serial_queue_key, write_sidecar,
 };
 pub use state::SubTableEntry;
 #[cfg(test)]