apunkt/omnigraph
1
0
Fork 0
mirror of https://github.com/ModernRelay/omnigraph.git synced 2026-06-12 01:45:14 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

mr-668: Omnigraph::init error-path cleanup + three failpoints (PR 2a/10)

Browse source 
PR 2a of the MR-668 multi-graph server work. Bug fix: a partially-failed
`Omnigraph::init` previously left orphan schema files at the graph URI,
making the URI unusable for a retry (the next `init` would refuse because
`_schema.pg` already exists).

Changes:

1. `init_with_storage` now wraps the I/O phase. On any error from
   `init_storage_phase`, calls `best_effort_cleanup_init_artifacts` to
   remove the three schema files before returning the original error:
   - `_schema.pg`
   - `_schema.ir.json`
   - `__schema_state.json`
   Cleanup is best-effort: a failure to delete is logged via
   `tracing::warn` but does NOT mask the init error.

2. Three failpoints added at the init phase boundaries:
   - `init.after_schema_pg_written`
   - `init.after_schema_contract_written`
   - `init.after_coordinator_init`

3. Four new failpoint tests in `tests/failpoints.rs` pin the cleanup
   behavior at each boundary plus the "original error wins over cleanup
   error" contract. All 23 failpoint tests pass.

Coverage gap (documented in code comments):
Lance per-type datasets and `__manifest/` directory created by
`GraphCoordinator::init` are NOT cleaned up after a coordinator-init-phase
failure. Recursive directory deletion requires `StorageAdapter::delete_prefix`,
which was deferred along with `DELETE /graphs/{id}` (originally PR 2b). When
that primitive lands, the third failpoint test can be tightened to assert
the graph root is fully empty.

Tests: 4 new (init_failpoint_*), all 23 failpoint tests green. No
regression in the 105 engine library tests or 64 end_to_end tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ragnor Comerford 2026-05-25 19:25:18 +02:00
parent 1ed3eea26d
commit 3fb6fa35d5
No known key found for this signature in database
2 changed files with 230 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

97
crates/omnigraph/src/db/omnigraph.rs
View file

@ -183,13 +183,36 @@ impl Omnigraph {
let mut catalog = build_catalog_from_ir(&schema_ir)?;
fixup_blob_schemas(&mut catalog);
// Write _schema.pg
let schema_path = join_uri(&root, SCHEMA_SOURCE_FILENAME);
storage.write_text(&schema_path, schema_source).await?;
write_schema_contract(&root, storage.as_ref(), &schema_ir).await?;
// Create manifest + per-type datasets
let coordinator = GraphCoordinator::init(&root, &catalog, Arc::clone(&storage)).await?;
// Run the I/O phase. On any error, best-effort-clean the schema
// artifacts that may have been written to disk before returning
// the original error. The cleanup is best-effort: a failure to
// delete is logged via `tracing::warn` but does not mask the
// original init error.
//
// Coverage gap: Lance per-type datasets and `__manifest/`
// directory created by `GraphCoordinator::init` are NOT cleaned
// up here — fully recursive directory deletion requires a
// `StorageAdapter::delete_prefix` primitive that's deferred
// along with `DELETE /graphs/{id}` (PR 2b in the MR-668 plan
// is currently deferred). If `init` fails after coordinator
// init succeeds, operators may need to remove the graph
// directory manually before retrying `init` on the same URI.
// Documented in the PR 2a commit message and `init` rustdoc.
let coordinator = match init_storage_phase(
&root,
schema_source,
&schema_ir,
&catalog,
&storage,
)
.await
{
Ok(coordinator) => coordinator,
Err(err) => {
best_effort_cleanup_init_artifacts(&root, storage.as_ref()).await;
return Err(err);
}
};
Ok(Self {
root_uri: root.clone(),
@ -1477,6 +1500,66 @@ fn read_schema_ir_from_source(schema_source: &str) -> Result<SchemaIR> {
build_schema_ir(&schema_ast).map_err(|err| OmniError::manifest(err.to_string()))
}
/// I/O phase of `Omnigraph::init_with_storage`. Split out so the caller
/// can pattern-match on the result and run cleanup on error before
/// returning the original error.
///
/// Failpoints fire at the three phase boundaries:
/// * `init.after_schema_pg_written` — `_schema.pg` is on disk.
/// * `init.after_schema_contract_written` — `_schema.pg` + `_schema.ir.json`
/// + `__schema_state.json` are on disk.
/// * `init.after_coordinator_init` — all schema files plus Lance per-type
/// datasets and `__manifest/` are on disk. (The cleanup wrapper can only
/// remove the schema files; Lance directories need `delete_prefix` —
/// deferred along with `DELETE /graphs/{id}`.)
async fn init_storage_phase(
root: &str,
schema_source: &str,
schema_ir: &SchemaIR,
catalog: &Catalog,
storage: &Arc<dyn StorageAdapter>,
) -> Result<GraphCoordinator> {
let schema_path = join_uri(root, SCHEMA_SOURCE_FILENAME);
storage.write_text(&schema_path, schema_source).await?;
crate::failpoints::maybe_fail("init.after_schema_pg_written")?;
write_schema_contract(root, storage.as_ref(), schema_ir).await?;
crate::failpoints::maybe_fail("init.after_schema_contract_written")?;
let coordinator = GraphCoordinator::init(root, catalog, Arc::clone(storage)).await?;
crate::failpoints::maybe_fail("init.after_coordinator_init")?;
Ok(coordinator)
}
/// Best-effort cleanup of init-phase artifacts. Called from
/// `init_with_storage` on any error returned by `init_storage_phase`.
///
/// Removes the three schema files: `_schema.pg`, `_schema.ir.json`,
/// `__schema_state.json`. Lance datasets and `__manifest/` are not
/// touched here — recursive directory deletion requires a
/// `StorageAdapter::delete_prefix` primitive that's deferred along
/// with `DELETE /graphs/{id}` (MR-668 PR 2b).
///
/// Failures to delete are logged via `tracing::warn` and do not mask
/// the original init error.
async fn best_effort_cleanup_init_artifacts(root: &str, storage: &dyn StorageAdapter) {
for uri in [
schema_source_uri(root),
schema_ir_uri(root),
schema_state_uri(root),
] {
if let Err(err) = storage.delete(&uri).await {
tracing::warn!(
target: "omnigraph::init",
uri = %uri,
error = %err,
"init failed; best-effort cleanup could not delete artifact",
);
}
}
}
fn schema_table_key(type_kind: SchemaTypeKind, name: &str) -> String {
match type_kind {
SchemaTypeKind::Node => format!("node:{}", name),

140
crates/omnigraph/tests/failpoints.rs
View file

@ -1666,3 +1666,143 @@ async fn ensure_indices_phase_b_failure_does_not_leak_sidecar_when_no_work_neede
"_graph_commit_recoveries.lance must NOT exist when no sidecar was processed"
);
}
// ─── MR-668 PR 2a: Omnigraph::init cleanup on partial failure ──────────────
//
// `init_with_storage` writes three schema artifacts before invoking
// `GraphCoordinator::init`. Without cleanup, a failure between any of those
// steps left orphan files behind, making the URI unusable for a retry of
// `init` (it would refuse because `_schema.pg` already exists). The tests
// below pin: on failpoint trigger at each of the three phase boundaries,
// the three schema files are removed before the error is returned.
//
// Coverage note: the third boundary (`init.after_coordinator_init`) only
// asserts cleanup of the schema files. Lance per-type directories and
// `__manifest/` are NOT cleaned up — that requires a recursive
// `StorageAdapter::delete_prefix` primitive deferred along with
// `DELETE /graphs/{id}` (MR-668 PR 2b). The orphan Lance directories
// after a coordinator-init-phase failure are documented as a known
// limitation.
#[tokio::test]
async fn init_failpoint_after_schema_pg_written_cleans_up_schema_file() {
let _scenario = FailScenario::setup();
let dir = tempfile::tempdir().unwrap();
let uri = dir.path().to_str().unwrap();
let _failpoint = ScopedFailPoint::new("init.after_schema_pg_written", "return");
let err = match Omnigraph::init(uri, helpers::TEST_SCHEMA).await {
Ok(_) => panic!("expected Omnigraph::init to fail at the configured failpoint"),
Err(e) => e,
};
assert!(
err.to_string()
.contains("injected failpoint triggered: init.after_schema_pg_written"),
"got: {err}"
);
// Only `_schema.pg` was written at this phase boundary, but the
// cleanup attempts all three — `delete` treats not-found as Ok,
// so the other two deletes are no-ops.
assert!(
!dir.path().join("_schema.pg").exists(),
"_schema.pg must be cleaned up after init failure"
);
}
#[tokio::test]
async fn init_failpoint_after_schema_contract_written_cleans_up_all_schema_files() {
let _scenario = FailScenario::setup();
let dir = tempfile::tempdir().unwrap();
let uri = dir.path().to_str().unwrap();
let _failpoint = ScopedFailPoint::new("init.after_schema_contract_written", "return");
let err = match Omnigraph::init(uri, helpers::TEST_SCHEMA).await {
Ok(_) => panic!("expected Omnigraph::init to fail at the configured failpoint"),
Err(e) => e,
};
assert!(
err.to_string()
.contains("injected failpoint triggered: init.after_schema_contract_written"),
"got: {err}"
);
assert!(
!dir.path().join("_schema.pg").exists(),
"_schema.pg must be cleaned up"
);
assert!(
!dir.path().join("_schema.ir.json").exists(),
"_schema.ir.json must be cleaned up"
);
assert!(
!dir.path().join("__schema_state.json").exists(),
"__schema_state.json must be cleaned up"
);
}
#[tokio::test]
async fn init_failpoint_after_coordinator_init_cleans_up_schema_files() {
let _scenario = FailScenario::setup();
let dir = tempfile::tempdir().unwrap();
let uri = dir.path().to_str().unwrap();
let _failpoint = ScopedFailPoint::new("init.after_coordinator_init", "return");
let err = match Omnigraph::init(uri, helpers::TEST_SCHEMA).await {
Ok(_) => panic!("expected Omnigraph::init to fail at the configured failpoint"),
Err(e) => e,
};
assert!(
err.to_string()
.contains("injected failpoint triggered: init.after_coordinator_init"),
"got: {err}"
);
// Schema files are cleaned up by `best_effort_cleanup_init_artifacts`.
assert!(
!dir.path().join("_schema.pg").exists(),
"_schema.pg must be cleaned up after late-phase init failure"
);
assert!(
!dir.path().join("_schema.ir.json").exists(),
"_schema.ir.json must be cleaned up after late-phase init failure"
);
assert!(
!dir.path().join("__schema_state.json").exists(),
"__schema_state.json must be cleaned up after late-phase init failure"
);
// Documented limitation: Lance per-type datasets and `__manifest/`
// created by `GraphCoordinator::init` are NOT cleaned up — recursive
// deletion requires the deferred `delete_prefix` primitive. This
// assertion does NOT check for their absence; it merely documents
// the boundary by noting we don't validate orphan directories here.
// When PR 2b lands, this test can be tightened to assert the graph
// root is fully empty.
}
#[tokio::test]
async fn init_failpoint_returns_original_error_not_cleanup_error() {
// The cleanup is best-effort. If `storage.delete` fails (e.g. transient
// network blip on S3), the original init failpoint error must still
// surface — not be masked by a cleanup failure. This test triggers the
// failpoint and asserts the returned error references the failpoint,
// not the cleanup. (The cleanup currently logs via `tracing::warn`;
// we can't easily fault-inject delete failures without another seam,
// so this is a smoke test for the precedence contract.)
let _scenario = FailScenario::setup();
let dir = tempfile::tempdir().unwrap();
let uri = dir.path().to_str().unwrap();
let _failpoint = ScopedFailPoint::new("init.after_schema_pg_written", "return");
let err = match Omnigraph::init(uri, helpers::TEST_SCHEMA).await {
Ok(_) => panic!("expected Omnigraph::init to fail at the configured failpoint"),
Err(e) => e,
};
// Failpoint message wins; no "cleanup" substring expected.
let msg = err.to_string();
assert!(
msg.contains("init.after_schema_pg_written"),
"init error must surface the failpoint cause, got: {msg}"
);
}