feat(server): readiness witness and bounded shutdown (RFC 0049) (#612)

feat(server): readiness reports counts, the inventory names the quarantined, a thread keeps the deadline

From review: GET /readyz no longer lists graph ids, which are topology
behind the authenticated GET /graphs; that route gains `quarantined`,
derived from the ledger's applied graphs minus the registry, with sidecar
graphs intersected with the applied set so a sidecar for a graph the
revision does not name is never a phantom. The shutdown watchdog is an
operating-system thread, so a blocked runtime cannot postpone it, and the
signal listener is spawned before any graph opens, so the bound covers
startup. The flag wins over OMNIGRAPH_SHUTDOWN_GRACE_SECONDS, which is read
only when the flag is absent. openapi.json regenerated.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015cD8PEeUfrzpqYuU1jaZBq

feat(server): readiness witness and bounded shutdown (RFC 0048)

`GET /readyz` reports whether a replica is serving or draining, the
applied `config_digest` it booted from (`booted_serving_digest`), the
ledger revision and CAS it read, and its served and quarantined graphs; it
answers 503 once shutdown has begun while `/healthz` stays 200.
`--shutdown-grace-seconds` (or `OMNIGRAPH_SHUTDOWN_GRACE_SECONDS`, default
25) puts one deadline on graceful shutdown: readiness turns off at the
signal, in-flight requests drain, and at the deadline the process exits 2
with the unfinished work logged. `ServingSnapshot` carries the boot facts;
`ServerConfig` gains `witness` and `shutdown_grace`; `openapi.json` is
regenerated.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015cD8PEeUfrzpqYuU1jaZBq

Author: aaltshuler

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

@@ -1065,6 +1065,34 @@ pub struct HealthOutput {
     pub source_version: Option<String>,
 }
 
+/// The readiness witness of one replica (`GET /readyz`, RFC 0049): whether
+/// it is serving or draining, the applied revision it booted from, and how
+/// many graphs it does and does not serve. Unauthenticated, so it carries
+/// no graph id: those stay behind `GET /graphs`.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct ReadinessOutput {
+    /// False once shutdown has begun; the response is then 503.
+    pub ready: bool,
+    /// `serving` or `draining`.
+    pub status: String,
+    /// The `config_digest` of the applied revision this process booted from.
+    /// Fixed for the life of the process: the server never reloads.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub booted_serving_digest: Option<String>,
+    /// The ledger revision the process booted from.
+    pub state_revision: u64,
+    /// The ledger CAS (`sha256:<hex>`) the process booted from.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub state_cas: Option<String>,
+    /// How many graphs this process serves.
+    pub served_graph_count: usize,
+    /// How many graphs the applied revision names that this process does
+    /// not serve, for any reason. `GET /graphs` names them.
+    pub quarantined_graph_count: usize,
+    /// The bound on graceful shutdown, after which the process exits 2.
+    pub shutdown_grace_seconds: u64,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
 #[serde(rename_all = "snake_case")]
 pub enum ErrorCode {
@@ -1611,6 +1639,11 @@ pub struct GraphInfo {
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct GraphListResponse {
     pub graphs: Vec<GraphInfo>,
+    /// Graphs the applied revision names that this process does not serve,
+    /// for any reason, sorted (RFC 0049). Empty when every applied graph is
+    /// served.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub quarantined: Vec<String>,
 }
 
 #[cfg(test)]

crates/omnigraph-cluster/src/serve.rs

@@ -41,6 +41,18 @@ pub struct ServingSnapshot {
     pub queries: Vec<ServingQuery>,
     pub policies: Vec<ServingPolicy>,
     pub diagnostics: Vec<Diagnostic>,
+    /// The applied revision's `config_digest`: what a server booted from
+    /// reports as its `booted_serving_digest` (RFC 0049).
+    pub config_digest: Option<String>,
+    /// The ledger revision and CAS this snapshot was read from.
+    pub state_revision: u64,
+    pub state_cas: Option<String>,
+    /// Every graph the applied revision names, sorted.
+    pub applied_graphs: Vec<String>,
+    /// Applied graphs this snapshot does not serve because pending recovery
+    /// quarantined them, sorted. A sidecar for a graph the revision does not
+    /// name is not in this list.
+    pub quarantined_graphs: Vec<String>,
 }
 
 /// Read the applied revision as a serving snapshot — the read-only loader for
@@ -271,6 +283,10 @@ async fn read_snapshot_with_store(
         diagnostics.extend(startup_diagnostics);
         return Err(diagnostics);
     };
+    let boot_config_digest = state.applied_revision.config_digest.clone();
+    let boot_state_revision = state.state_revision;
+    let boot_state_cas = observations.state_cas.clone();
+    let boot_applied_graphs = applied_graph_ids(&state);
 
     let required_embedding_providers: BTreeSet<String> = state
         .applied_revision
@@ -457,6 +473,14 @@ async fn read_snapshot_with_store(
         queries,
         policies,
         diagnostics: startup_diagnostics,
+        config_digest: boot_config_digest,
+        state_revision: boot_state_revision,
+        state_cas: boot_state_cas,
+        quarantined_graphs: quarantined_graphs
+            .into_iter()
+            .filter(|graph_id| boot_applied_graphs.contains(graph_id))
+            .collect(),
+        applied_graphs: boot_applied_graphs,
     })
 }
 

crates/omnigraph-server/src/handlers.rs

@@ -28,6 +28,48 @@ pub(crate) async fn server_health() -> Json<HealthOutput> {
     })
 }
 
+/// Readiness witness (RFC 0049).
+///
+/// Unauthenticated, and therefore minimal: it reports whether this replica
+/// is serving or draining, the applied `config_digest` it booted from, the
+/// ledger revision and CAS it read, and how many graphs it serves and does
+/// not serve. Graph ids are topology and stay behind `GET /graphs`. Answers
+/// 503 once shutdown has begun; `/healthz` stays 200 while the process is
+/// alive.
+#[utoipa::path(
+    get,
+    path = "/readyz",
+    tag = "health",
+    operation_id = "readiness",
+    responses(
+        (status = 200, description = "Serving", body = ReadinessOutput),
+        (status = 503, description = "Draining", body = ReadinessOutput),
+    ),
+)]
+pub(crate) async fn server_ready(
+    State(state): State<AppState>,
+) -> (StatusCode, Json<ReadinessOutput>) {
+    let draining = state.draining.load(std::sync::atomic::Ordering::SeqCst);
+    let served_graph_count = state.routing().registry.list().len();
+    let quarantined_graph_count = state.quarantined_graphs().len();
+    let output = ReadinessOutput {
+        ready: !draining,
+        status: if draining { "draining" } else { "serving" }.to_string(),
+        booted_serving_digest: state.witness.booted_serving_digest.clone(),
+        state_revision: state.witness.state_revision,
+        state_cas: state.witness.state_cas.clone(),
+        served_graph_count,
+        quarantined_graph_count,
+        shutdown_grace_seconds: state.shutdown_grace.as_secs(),
+    };
+    let status = if draining {
+        StatusCode::SERVICE_UNAVAILABLE
+    } else {
+        StatusCode::OK
+    };
+    (status, Json(output))
+}
+
 #[utoipa::path(
     get,
     path = "/graphs",
@@ -81,7 +123,10 @@ pub(crate) async fn server_graphs_list(
         })
         .collect();
     graphs.sort_by(|a, b| a.graph_id.cmp(&b.graph_id));
-    Ok(Json(GraphListResponse { graphs }))
+    Ok(Json(GraphListResponse {
+        graphs,
+        quarantined: state.quarantined_graphs(),
+    }))
 }
 
 pub(crate) async fn server_openapi(
@@ -114,7 +159,7 @@ const CLUSTER_OPERATION_ID_PREFIX: &str = "cluster_";
 /// always-flat endpoints. `/graphs` is the management enumeration —
 /// it lives at the root in both single mode (405) and multi mode, and
 /// must never be rewritten to `/graphs/{graph_id}/graphs`.
-const ALWAYS_FLAT_PATHS: &[&str] = &["/healthz", "/graphs"];
+const ALWAYS_FLAT_PATHS: &[&str] = &["/healthz", "/readyz", "/graphs"];
 
 /// In multi-mode `server_openapi`, every protected path-item is
 /// reattached under the cluster prefix. Operation IDs gain the