feat(engine): give maintenance a lane, cancel points, budgets and completion evidence

changelog.d/full-text-rebuild-change-grant.deprecated.md

@@ -0,0 +1,6 @@
+- The full-text rebuild accepts `maintain` or `change` on its branch in this
+  release; a later release accepts `maintain` only. Grant `maintain` to the
+  accounts that rebuild full-text indexes. See
+  [rebuild full-text indexes][full-text-rebuild-change-grant].
+
+[full-text-rebuild-change-grant]: ../docs/user/operations/maintenance.md#rebuild-full-text-indexes

changelog.d/lease-beat-failures-logged.added.md

@@ -0,0 +1,7 @@
+- **Failed cluster lease heartbeats are logged.** A server logs every
+  heartbeat request that fails, every failed check for a successor's claim and
+  a heartbeat left unanswered at its deadline, each with the failure's class and
+  storage kind and before any fence. See the [storage failure
+  logs][lease-beat-logs].
+
+[lease-beat-logs]: ../docs/user/operations/storage-failures.md#logs

changelog.d/maintenance-budgets.changed.md

@@ -0,0 +1,10 @@
+- **Maintenance runs one table at a time under declared memory caps.**
+  Optimize, cleanup, index builds and the full-text rebuild now run one at a
+  time per graph in a process, and a run rewrites or indexes one table at a
+  time by default (`OMNIGRAPH_MAINTENANCE_CONCURRENCY`, which defaulted to 8,
+  is now capped at 2). Compaction scans queue at most 64 MiB in Lance's I/O
+  buffer and full-text builds use 512 MiB over two workers; each has its own
+  variable. Expect a direct `optimize` on a graph with many tables to take
+  longer. See [maintenance budgets][maintenance-budgets].
+
+[maintenance-budgets]: ../docs/user/operations/maintenance.md#budgets

changelog.d/maintenance-entry-points-take-an-actor.breaking.md

@@ -0,0 +1,12 @@
+- Rust callers of the engine's maintenance methods move to their actor-taking
+  forms: `optimize_as(actor)`, `cleanup_as(options, actor)`,
+  `ensure_indices_as(actor)`, `ensure_indices_on_as(branch, actor)` and
+  `rebuild_full_text_indices_on_as(branch, actor)`; pass `None` where no policy
+  is installed. `repair` is now the read-only `repair_report_as(actor)`:
+  `RepairOptions`, the `healed` and `forced` actions and
+  `RepairStats::graph_manifest_version` are gone, and `repair --json` no longer
+  prints the always-null `graph_manifest_version`. An exhaustive match on
+  `PolicyAction` adds arms for `Maintain` and `Reclaim`. See
+  [repair][maintenance-entry-points-repair].
+
+[maintenance-entry-points-repair]: ../docs/user/operations/maintenance.md#repair

changelog.d/maintenance-failures-are-definite.changed.md

@@ -0,0 +1,8 @@
+- **A maintenance failure before its publication says the graph did not
+  move.** Optimize, cleanup, index builds, the full-text rebuild and the repair
+  report mark every such failure, so the direct CLI exits `1` (no effects) for
+  a storage or internal failure where it exited `2` (outcome unknown), and a
+  host serving the graph keeps serving through one. A publication in doubt is
+  still reported as one. See the [write path][maintenance-failure-evidence].
+
+[maintenance-failure-evidence]: ../docs/dev/writes.md#policy-and-attribution

changelog.d/maintenance-policy-actions.added.md

@@ -0,0 +1,10 @@
+- **Maintenance has its own Cedar actions.** `maintain` covers optimize, index
+  builds, the full-text rebuild and the repair report on the branch they
+  maintain, and `reclaim` covers cleanup with no branch scope; `change` implies
+  neither, so automation can compact a graph without the right to write data or
+  delete history. An embedded host that installs a policy checker enforces them
+  on every maintenance entry point, optimize and the index writers record the
+  requesting actor on their commit, and existing policies grant neither action.
+  See [policy actions][maintenance-policy-actions].
+
+[maintenance-policy-actions]: ../docs/user/operations/policy.md#actions

crates/omnigraph-api-types/src/lib.rs

@@ -2013,6 +2013,9 @@ fn access_slot(action: omnigraph_policy::PolicyAction) -> Option<AccessSlot> {
         P::InvokeQuery => Some(AccessSlot::GraphWide),
         // Server- or operator-scoped: no graph's access report names them.
         P::Admin | P::GraphList | P::ConfigManage => None,
+        // Maintenance runs only through the direct CLI and embedded callers;
+        // no served operation performs it, so the report does not name it.
+        P::Maintain | P::Reclaim => None,
     }
 }
 

crates/omnigraph-bench/src/dataset_worker.rs

@@ -803,7 +803,10 @@ pub(crate) async fn build_dataset(
                 Engine::V2,
             )?;
             if plan.needs_indices {
-                session.ensure_indices().await.map_err(|e| e.to_string())?;
+                session
+                    .ensure_indices_as(None)
+                    .await
+                    .map_err(|e| e.to_string())?;
             }
             (session, case, Some(exact))
         }

crates/omnigraph-bench/src/legacy/branch_merge_oracle.rs

@@ -28,7 +28,7 @@ pub async fn initialize_local_fixture(
     let optimized_user_tables = match plan.compaction_recency {
         CompactionRecency::Optimized => {
             let outcomes = db
-                .optimize()
+                .optimize_as(None)
                 .await
                 .map_err(|error| fixture_error(format!("optimize fixture main branch: {error}")))?;
             let intended_keys = (0..plan.node_tables())

crates/omnigraph-cli/src/commands/maintenance/mod.rs

@@ -3,7 +3,7 @@
 use std::io::Write;
 
 use color_eyre::eyre::{Result, bail};
-use omnigraph::db::{CleanupPolicyOptions, Omnigraph, RepairAction, RepairOptions};
+use omnigraph::db::{CleanupPolicyOptions, Omnigraph, RepairAction};
 
 use crate::app::Completion;
 use crate::helpers::{confirm_destructive, parse_duration_arg, resolve_actor};
@@ -42,13 +42,14 @@ pub(crate) async fn run_optimize(
     let OptimizeArgs { json } = args;
     let uri = crate::scope::direct_graph_uri(scope.target, scope.graph).await?;
     let binding = crate::scope::capture_storage_binding(scope.target)?.for_uri(&uri)?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.optimize().await?;
+    let stats = db.optimize_as(actor.as_deref()).await?;
     write_optimize(&uri, &stats, json, stdout)?;
     Ok(Completion::success())
 }
@@ -93,13 +94,14 @@ pub(crate) async fn run_repair(
     let RepairArgs { json } = args;
     let uri = crate::scope::direct_graph_uri(scope.target, scope.graph).await?;
     let binding = crate::scope::capture_storage_binding(scope.target)?.for_uri(&uri)?;
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.repair(RepairOptions::default()).await?;
+    let stats = db.repair_report_as(actor.as_deref()).await?;
     let refused_count = stats
         .datasets
         .iter()
@@ -159,13 +161,14 @@ pub(crate) async fn run_cleanup(
         keep_versions: keep,
         older_than: older_than_dur,
     };
+    let actor = resolve_actor(scope.operator, scope.actor, stderr)?;
     crate::admission::ensure_graph_with_binding(&binding, stderr).await?;
     let db = Omnigraph::open_with_storage(
         binding.uri(),
         omnigraph::storage::storage_for_binding(&binding)?,
     )
     .await?;
-    let stats = db.cleanup(options.clone()).await?;
+    let stats = db.cleanup_as(options.clone(), actor.as_deref()).await?;
     write_cleanup(&uri, &options, &policy_desc, &stats, json, stdout)?;
     Ok(Completion::success())
 }

crates/omnigraph-cli/src/output.rs

@@ -133,7 +133,6 @@ pub(crate) fn write_repair(
     if json {
         let value = serde_json::json!({
             "uri": uri,
-            "graph_manifest_version": stats.graph_manifest_version,
             "datasets": stats.datasets.iter().map(|s| serde_json::json!({
                 "type_key": s.type_key,
                 "published_dataset_version": s.published_dataset_version,
@@ -1867,7 +1866,7 @@ mod tests {
     #[tokio::test]
     async fn maintenance_optimize_and_repair_reports_preserve_all_result_fields() {
         use omnigraph::db::{
-            Omnigraph, PendingIndex, RepairAction, RepairClassification, RepairOptions, SkipReason,
+            Omnigraph, PendingIndex, RepairAction, RepairClassification, SkipReason,
         };
         // These engine result types are non_exhaustive. Obtain one valid value
         // from a tiny graph, then set its public presentation fields; no engine
@@ -1881,7 +1880,7 @@ mod tests {
         let db = Omnigraph::init(&uri, "node Person {\n    slug: String @key\n}\n")
             .await
             .unwrap();
-        let mut compacted = db.optimize().await.unwrap().pop().unwrap();
+        let mut compacted = db.optimize_as(None).await.unwrap().pop().unwrap();
         compacted.type_key = "node:Person".into();
         compacted.fragments_removed = 3;
         compacted.fragments_added = 1;
@@ -1933,7 +1932,7 @@ mod tests {
             |json, writer| super::write_optimize("graph", &[], json, writer),
         );
 
-        let mut repair = db.repair(RepairOptions::default()).await.unwrap();
+        let mut repair = db.repair_report_as(None).await.unwrap();
         let mut clean = repair.datasets.pop().unwrap();
         clean.type_key = "node:Person".into();
         clean.published_dataset_version = 4;
@@ -1950,15 +1949,14 @@ mod tests {
         refused.operations = vec!["Append".into(), "Delete".into()];
         refused.error = Some("foreign write".into());
         repair.datasets = vec![clean, refused];
-        repair.graph_manifest_version = Some(9);
         assert_maintenance_report(
             concat!(
                 "repair graph — 2 datasets\n",
                 "  node type 'Person'                       no_op        no_drift               4\n",
                 "  edge type 'Knows'                        refused      foreign_drift          4 → 6 [Append, Delete] (foreign write)\n",
             ),
             serde_json::json!({
-                "uri":"graph", "graph_manifest_version":9,
+                "uri":"graph",
                 "datasets":[
                     {"type_key":"node:Person","published_dataset_version":4,"lance_head_version":4,"classification":"no_drift","action":"no_op","operations":[],"error":null},
                     {"type_key":"edge:Knows","published_dataset_version":4,"lance_head_version":6,"classification":"foreign_drift","action":"refused","operations":["Append","Delete"],"error":"foreign write"},
@@ -1967,10 +1965,9 @@ mod tests {
             |json, writer| super::write_repair("graph", &repair, json, writer),
         );
         repair.datasets.clear();
-        repair.graph_manifest_version = None;
         assert_maintenance_report(
             "repair graph — 0 datasets\n",
-            serde_json::json!({"uri":"graph","graph_manifest_version":null,"datasets":[]}),
+            serde_json::json!({"uri":"graph","datasets":[]}),
             |json, writer| super::write_repair("graph", &repair, json, writer),
         );
     }

crates/omnigraph-cli/tests/integration/cli_data.rs

@@ -1211,7 +1211,8 @@ fn repair_json_reports_noop_on_clean_graph() {
 
     assert!(payload.get("confirm").is_none());
     assert!(payload.get("force").is_none());
-    assert_eq!(payload["graph_manifest_version"], Value::Null);
+    // The report publishes nothing, so it names no graph manifest version.
+    assert!(payload.get("graph_manifest_version").is_none());
     assert!(payload.get("manifest_version").is_none());
     assert!(payload.get("tables").is_none());
     let datasets = payload["datasets"].as_array().unwrap();
@@ -1282,7 +1283,7 @@ fn repair_reports_foreign_drift_and_refuses_removed_flags() {
             .arg(&graph),
     );
     let payload: Value = serde_json::from_slice(&output.stdout).unwrap();
-    assert_eq!(payload["graph_manifest_version"], Value::Null);
+    assert!(payload.get("graph_manifest_version").is_none());
     let person = payload["datasets"]
         .as_array()
         .unwrap()

crates/omnigraph-cluster/Cargo.toml

@@ -42,6 +42,7 @@ serial_test = "3"
 tempfile = { workspace = true }
 # Paused time for the lease tests, so the crate's own test build runs them.
 tokio = { workspace = true, features = ["test-util"] }
+tracing-subscriber = { workspace = true }
 
 [lints]
 workspace = true