feat(startup): wire gate-based sequencer and gateway into server bootstrap

main.rs is updated to construct StartupSequencer gates before each
subsystem, install the real startup gate on SharedState after open(),
and construct the Gateway + PlanCacheInvalidator before listeners bind.

Additional bootstrap changes:
- WAL validation runs before wal_gate fires; corrupt segments are now
  fatal rather than silently ignored, preventing startup on a corrupted WAL
- WAL replay failure is now fatal (was a warn + empty state)
- ShutdownBus is wired with system metrics for per-phase drain telemetry
- A NODEDB_TEST_SLOW_DRAIN_TASK env hook enables drain-abort integration tests

Supporting changes:
- recovery_check module added for catalog sanity check on startup
- catalog_entry post_apply and invalidation tests updated
- system metrics expanded for startup and shutdown phase telemetry
- WAL manager gains validate_for_startup
- Error types extended for new startup and gateway error variants

Author: farhan-syah

nodedb/Cargo.toml

@@ -144,6 +144,7 @@ tempfile = "3"
 tokio-postgres = { workspace = true }
 proptest = "1"
 nodedb-types = { workspace = true }
+reqwest = { workspace = true }
 
 [features]
 default = []

nodedb/src/control/catalog_entry/post_apply/mod.rs

@@ -52,6 +52,11 @@ use crate::control::state::SharedState;
 /// is infallible today (all typed functions log on failure and
 /// return).
 pub fn apply_post_apply_side_effects_sync(entry: &CatalogEntry, shared: &Arc<SharedState>) {
+    // Gateway plan-cache invalidation: on any descriptor mutation, evict
+    // stale cached plans that reference the changed descriptor.
+    // This is a single, unconditional call per DDL commit — negligible overhead.
+    invalidate_gateway_cache_for_entry(entry, shared);
+
     match entry {
         CatalogEntry::PutCollection(stored) => {
             // Owner record install is sync; Data Plane register is
@@ -189,3 +194,175 @@ pub fn spawn_post_apply_async_side_effects(entry: CatalogEntry, shared: Arc<Shar
         });
     }
 }
+
+/// Notify the gateway plan-cache invalidator after a DDL descriptor mutation.
+///
+/// Extracts the descriptor name and new version from the entry and calls
+/// `PlanCacheInvalidator::invalidate`. This is best-effort: if the gateway
+/// has not been constructed yet (`gateway_invalidator == None`) the call is
+/// a no-op.
+///
+/// ## Invalidation decision table (all 31 variants — exhaustive, no `_ => {}`)
+///
+/// The gateway plan cache keys on `(sql_hash, ph_hash, GatewayVersionSet)`.
+/// A `GatewayVersionSet` lists `(collection_name, descriptor_version)` pairs
+/// extracted from the `PhysicalPlan` by `touched_collections`. A DDL entry
+/// requires invalidation only if it changes the observable plan shape for
+/// an already-cached plan. Verified against `planner/`, `rls_injection.rs`,
+/// and the `PhysicalPlan` definition.
+///
+/// | Entry kind                              | Invalidate? | Reason |
+/// |-----------------------------------------|-------------|--------|
+/// | PutCollection / DeactivateCollection    | ✅ yes      | collection schema baked into plan |
+/// | PutSequence / DeleteSequence            | ❌ no       | sequences resolved at handler level (pgwire `transaction_cmds.rs`), not in PhysicalPlan |
+/// | PutSequenceState                        | ❌ no       | runtime counter state, not plan shape |
+/// | PutTrigger / DeleteTrigger              | ❌ no       | triggers dispatched by Event Plane post-execution; no trigger fields in any PhysicalPlan variant |
+/// | PutFunction / DeleteFunction            | ❌ no       | functions looked up at eval time, not inlined |
+/// | PutProcedure / DeleteProcedure          | ❌ no       | same as functions |
+/// | PutSchedule / DeleteSchedule            | ❌ no       | scheduler runs independently |
+/// | PutChangeStream / DeleteChangeStream    | ❌ no       | CDC Event Plane concern |
+/// | PutUser / DeactivateUser                | ❌ no       | authz checked at exec time |
+/// | PutRole / DeleteRole                    | ❌ no       | same |
+/// | PutApiKey / RevokeApiKey                | ❌ no       | same |
+/// | PutMaterializedView / DeleteMaterializedView | ❌ no  | MV definition is its own catalog object; write-path `materialized_sum_sources` is set at collection-register time via PutCollection, not updated by PutMaterializedView independently |
+/// | PutTenant / DeleteTenant                | ❌ no       | tenant identity does not affect plan shape |
+/// | PutRlsPolicy / DeleteRlsPolicy          | ❌ no       | `execute_sql` is only called from CDC path (no RLS injection via `inject_rls`); per-session pgwire cache has its own DDL invalidation |
+/// | PutPermission / DeletePermission        | ❌ no       | permission checked at exec time |
+/// | PutOwner / DeleteOwner                  | ❌ no       | ownership does not affect plan shape |
+pub(crate) fn invalidate_gateway_cache_for_entry(entry: &CatalogEntry, shared: &Arc<SharedState>) {
+    let Some(ref inv) = shared.gateway_invalidator else {
+        return;
+    };
+    match entry {
+        // ── Collection mutations that change the plan shape ──────────────────
+        CatalogEntry::PutCollection(stored) => {
+            inv.invalidate(&stored.name, stored.descriptor_version.max(1));
+        }
+        CatalogEntry::DeactivateCollection { name, .. } => {
+            // Treat deactivation as version 0 (collection gone — any cached
+            // plan for it is stale).
+            inv.invalidate(name, 0);
+        }
+
+        // ── Sequence: resolved at handler level, not baked into PhysicalPlan ─
+        CatalogEntry::PutSequence(_) => {
+            // no-op: sequences resolved in pgwire transaction_cmds.rs before
+            // planning; StoredSequence never appears in a PhysicalPlan variant.
+        }
+        CatalogEntry::DeleteSequence { .. } => {
+            // no-op: same reason as PutSequence.
+        }
+        CatalogEntry::PutSequenceState(_) => {
+            // no-op: runtime counter state — the planner never reads seq state.
+        }
+
+        // ── Trigger: dispatched by Event Plane post-execution ────────────────
+        CatalogEntry::PutTrigger(_) => {
+            // no-op: triggers are AFTER-fire; no trigger field exists in any
+            // PhysicalPlan variant; Event Plane reads the trigger registry
+            // directly at fire time.
+        }
+        CatalogEntry::DeleteTrigger { .. } => {
+            // no-op: same as PutTrigger.
+        }
+
+        // ── Function / Procedure: looked up at eval time, not inlined ────────
+        CatalogEntry::PutFunction(_) => {
+            // no-op: UDFs looked up in function_registry at eval time via
+            // `wasm/` executor; never inlined into a PhysicalPlan.
+        }
+        CatalogEntry::DeleteFunction { .. } => {
+            // no-op: same as PutFunction.
+        }
+        CatalogEntry::PutProcedure(_) => {
+            // no-op: stored procedures parsed and executed at CALL time via
+            // `procedural/executor`; body not baked into any PhysicalPlan.
+        }
+        CatalogEntry::DeleteProcedure { .. } => {
+            // no-op: same as PutProcedure.
+        }
+
+        // ── Schedule: cron runs independently of the plan cache ──────────────
+        CatalogEntry::PutSchedule(_) => {
+            // no-op: ScheduleRegistry drives the scheduler loop; no plan shape
+            // changes result from a new/updated schedule definition.
+        }
+        CatalogEntry::DeleteSchedule { .. } => {
+            // no-op: same as PutSchedule.
+        }
+
+        // ── Change stream: CDC Event Plane concern ────────────────────────────
+        CatalogEntry::PutChangeStream(_) => {
+            // no-op: CDC stream definitions route WriteEvents in the Event
+            // Plane; they do not alter how a collection's plan is constructed.
+        }
+        CatalogEntry::DeleteChangeStream { .. } => {
+            // no-op: same as PutChangeStream.
+        }
+
+        // ── User / Role / ApiKey: authz checked at exec, not baked into plan ─
+        CatalogEntry::PutUser(_) => {
+            // no-op: user identity checked in credential store at exec time.
+        }
+        CatalogEntry::DeactivateUser { .. } => {
+            // no-op: same as PutUser.
+        }
+        CatalogEntry::PutRole(_) => {
+            // no-op: role membership checked at exec time via RoleStore.
+        }
+        CatalogEntry::DeleteRole { .. } => {
+            // no-op: same as PutRole.
+        }
+        CatalogEntry::PutApiKey(_) => {
+            // no-op: API key checked at connection/exec time via ApiKeyStore.
+        }
+        CatalogEntry::RevokeApiKey { .. } => {
+            // no-op: same as PutApiKey.
+        }
+
+        // ── Materialized view: MV definition is a separate catalog object ────
+        CatalogEntry::PutMaterializedView(_) => {
+            // no-op: MaterializedView metadata is its own catalog object and
+            // does not directly modify any PhysicalPlan. The `materialized_sum_sources`
+            // field in DocumentOp::Register is set at collection-register time
+            // (driven by PutCollection), not updated independently by
+            // PutMaterializedView. Any schema change that would affect plans
+            // cascades through PutCollection instead.
+        }
+        CatalogEntry::DeleteMaterializedView { .. } => {
+            // no-op: same as PutMaterializedView.
+        }
+
+        // ── Tenant: identity does not affect plan shape ───────────────────────
+        CatalogEntry::PutTenant(_) => {
+            // no-op: tenant identity used for quota enforcement at exec time.
+        }
+        CatalogEntry::DeleteTenant { .. } => {
+            // no-op: same as PutTenant.
+        }
+
+        // ── RLS policy: execute_sql callers (CDC) do not inject RLS ──────────
+        CatalogEntry::PutRlsPolicy(_) => {
+            // no-op: the gateway execute_sql path (CDC consume_remote) calls
+            // plan_sql without RLS injection; per-session pgwire plan cache
+            // has its own DDL-aware invalidation that handles RLS changes.
+        }
+        CatalogEntry::DeleteRlsPolicy { .. } => {
+            // no-op: same as PutRlsPolicy.
+        }
+
+        // ── Permission / Owner: not baked into plan ───────────────────────────
+        CatalogEntry::PutPermission(_) => {
+            // no-op: permission grants checked at exec time via PermissionStore.
+        }
+        CatalogEntry::DeletePermission { .. } => {
+            // no-op: same as PutPermission.
+        }
+        CatalogEntry::PutOwner(_) => {
+            // no-op: ownership does not influence plan structure.
+        }
+        CatalogEntry::DeleteOwner { .. } => {
+            // no-op: same as PutOwner.
+        }
+    }
+}