feat(engine)!: bound optimize runs and report each table's outcome

changelog.d/ensure-indices-refreshes-coverage.changed.md

@@ -1,3 +1,3 @@
-- `ensure_indices` and `ensure_indices_on` also rebuild lagging BTREE/vector
-  indexes, report lagging full-text coverage as pending, and under a policy
-  need an actor with `Change`: use `ensure_indices_on_as`.
+- The engine's index build (`ensure_indices_as`, `ensure_indices_on_as`) also
+  rebuilds lagging BTREE and vector indexes and reports lagging full-text
+  coverage as pending.

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

@@ -0,0 +1,5 @@
+- The full-text rebuild accepts `change` on its branch for one more release;
+  after that it needs `maintain`. 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/maintenance-entry-points-take-an-actor.breaking.md

@@ -0,0 +1,10 @@
+- An embedded host with a policy checker refuses maintenance until its policy
+  grants `maintain` or `reclaim`. Engine maintenance methods take an actor:
+  `optimize_as`, `cleanup_as`, `ensure_indices_as`, `ensure_indices_on_as` and
+  `rebuild_full_text_indices_on_as`. `repair` is the read-only
+  `repair_report_as`, without `RepairOptions`, `healed`, `forced` or
+  `graph_manifest_version`. See [policy actions][maintenance-actions-upgrade]
+  and [repair][maintenance-entry-points-repair].
+
+[maintenance-actions-upgrade]: ../docs/user/operations/policy.md#actions
+[maintenance-entry-points-repair]: ../docs/user/operations/maintenance.md#repair

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

@@ -0,0 +1,5 @@
+- Cedar policy has two maintenance actions: `maintain` for optimize, index
+  builds, the full-text rebuild and repair, and `reclaim` for cleanup. `change`
+  implies neither. See [policy actions][maintenance-policy-actions].
+
+[maintenance-policy-actions]: ../docs/user/operations/policy.md#actions

changelog.d/optimize-artifact-size-limit.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can bound the adjacency artifact optimize writes with
+  `OptimizeLimits::artifact_bytes`; a larger one is skipped, before or during
+  its edge scan, and nothing is written.

changelog.d/optimize-before-publish-hook.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can pass `OptimizeRequest::with_before_publish`: it runs once
+  before an optimize run publishes, with the commit's nonce and input head, and
+  its failure publishes nothing.

changelog.d/optimize-cancellation.added.md

@@ -0,0 +1,2 @@
+- Embedded hosts can cancel an optimize run with `OptimizeRequest::with_cancellation`:
+  it stops before the next table or the publication and publishes nothing.

changelog.d/optimize-gate-hold-bound.added.md

@@ -0,0 +1,3 @@
+- `optimize --max-gate-hold` bounds how long a run holds main's writer gates and names the table the next run resumes at; `--tables` runs only the listed tables. See [bounding a run][optimize-bound].
+
+[optimize-bound]: ../docs/user/operations/maintenance.md#bounding-a-run

changelog.d/optimize-report-outcomes.changed.md

@@ -0,0 +1,3 @@
+- `optimize` reports one outcome per table, both graph commits and the gate hold; a failed configuration clear or adjacency artifact no longer fails a published run. See [bounding a run][optimize-report].
+
+[optimize-report]: ../docs/user/operations/maintenance.md#bounding-a-run

changelog.d/optimize-resource-ceilings.added.md

@@ -0,0 +1,3 @@
+- Embedded hosts can bound optimize per table (`TableCeilings`) and per run
+  (`run_source_bytes`); a table over a ceiling is deferred before its first
+  write, with nothing written for it.

changelog.d/optimize-table-failure-isolation.changed.md

@@ -0,0 +1,2 @@
+- A table whose own optimize work fails is reported `failed` with its error, and
+  the other tables still publish; before, one table's failure failed the run.

changelog.d/optimize-takes-a-request.breaking.md

@@ -0,0 +1,8 @@
+- Embedded hosts: `optimize_as` takes an `OptimizeRequest` (`OptimizeRequest::default()`
+  is a full run) and returns an `OptimizeReport` whose `tables` holds the per-table
+  stats; `committed` and `skipped` are methods. `optimize --json` reports the
+  `__manifest` row as `configuration_clear` and drops the always-null
+  `published_dataset_version` and `lance_head_version`. See
+  [bounding a run][optimize-request-upgrade].
+
+[optimize-request-upgrade]: ../docs/user/operations/maintenance.md#bounding-a-run

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

@@ -2058,6 +2058,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,32 +28,33 @@ pub async fn initialize_local_fixture(
     let optimized_user_tables = match plan.compaction_recency {
         CompactionRecency::Optimized => {
             let outcomes = db
-                .optimize()
+                .optimize_as(None, omnigraph::db::OptimizeRequest::default())
                 .await
                 .map_err(|error| fixture_error(format!("optimize fixture main branch: {error}")))?;
             let intended_keys = (0..plan.node_tables())
                 .map(node_table_key)
                 .chain((0..plan.edge_tables()).map(edge_table_key));
             for key in intended_keys {
                 let outcome = outcomes
+                    .tables
                     .iter()
                     .find(|outcome| outcome.type_key == key)
                     .ok_or_else(|| {
                         fixture_error(format!(
                             "optimized fixture returned no outcome for intended user table {key}"
                         ))
                     })?;
-                if outcome.skipped.is_some()
-                    || !outcome.committed
+                if outcome.skipped().is_some()
+                    || !outcome.committed()
                     || outcome.fragments_removed == 0
                     || outcome.fragments_added == 0
                 {
                     return Err(fixture_error(format!(
                         "optimized fixture did not productively compact {key}: committed={}, fragments_removed={}, fragments_added={}, skipped={:?}",
-                        outcome.committed,
+                        outcome.committed(),
                         outcome.fragments_removed,
                         outcome.fragments_added,
-                        outcome.skipped
+                        outcome.skipped()
                     )));
                 }
             }

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

@@ -60,6 +60,15 @@ pub(crate) struct UpgradeArgs {
 
 #[derive(Debug, Args)]
 pub(crate) struct OptimizeArgs {
+    /// Only these tables, in this order: `node:<Type>` or `edge:<Type>`,
+    /// comma-separated or repeated. Default: every table on main.
+    #[arg(long, value_delimiter = ',')]
+    pub(crate) tables: Vec<String>,
+    /// Stop starting tables once the run has held main's writer gates this
+    /// long, and report the rest deferred. The first table always runs.
+    /// Accepts Go-style durations: `30s`, `5m`.
+    #[arg(long)]
+    pub(crate) max_gate_hold: Option<String>,
     #[arg(long)]
     pub(crate) json: bool,
 }

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,17 +42,31 @@ pub(crate) async fn run_optimize(
     stdout: &mut dyn Write,
     stderr: &mut dyn Write,
 ) -> Result<Completion> {
-    let OptimizeArgs { json } = args;
+    let OptimizeArgs {
+        tables,
+        max_gate_hold,
+        json,
+    } = args;
+    let mut request = omnigraph::db::OptimizeRequest::default();
+    if !tables.is_empty() {
+        request = request.with_tables(tables);
+    }
+    if let Some(hold) = max_gate_hold {
+        request = request.with_limits(
+            omnigraph::db::OptimizeLimits::default().with_max_gate_hold(parse_duration_arg(&hold)?),
+        );
+    }
     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?;
-    write_optimize(&uri, &stats, json, stdout)?;
+    let report = db.optimize_as(actor.as_deref(), request).await?;
+    write_optimize(&uri, &report, json, stdout)?;
     Ok(Completion::success())
 }
 
@@ -141,13 +155,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()
@@ -207,13 +222,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())
 }