diff --git a/Cargo.lock b/Cargo.lock
index 814066c3..fc2c0b54 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4920,6 +4920,7 @@ dependencies = [
  "tempfile",
  "tikv-jemallocator",
  "tokio",
+ "tokio-util",
  "tower-http",
  "tracing",
  "tracing-subscriber",
diff --git a/Cargo.toml b/Cargo.toml
index a376b34e..97d53d0c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -47,6 +47,7 @@ serde_yaml = "0.9.33"
 scylla = "1.1.0"
 tempfile = "3.13.0"
 tokio = "1.42.0"
+tokio-util = "0.7.13"
 tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 url = "2.5.4"
diff --git a/crates/hotblocks-harness/src/sut.rs b/crates/hotblocks-harness/src/sut.rs
index 40d3ad84..3eee2188 100644
--- a/crates/hotblocks-harness/src/sut.rs
+++ b/crates/hotblocks-harness/src/sut.rs
@@ -59,6 +59,7 @@ pub struct Sut {
     dir: TempDir,
     port: u16,
     child: Option<Child>,
+    shutdown_started: Option<Instant>,
     /// How long the last boot took to accept connections (SLI-5).
     pub last_startup: Duration
 }
@@ -78,6 +79,7 @@ impl Sut {
             cfg,
             dir,
             child: None,
+            shutdown_started: None,
             last_startup: Duration::ZERO
         };
         sut.write_config()?;
@@ -116,32 +118,52 @@ impl Sut {
         };
         child.kill().await.context("failed to kill the SUT")?;
         self.child = None;
+        self.shutdown_started = None;
         Ok(())
     }
 
-    /// SIGTERM and wait for the drain-and-exit (LIV-12, `P-SHUTDOWN`).
-    pub async fn stop(&mut self) -> Result<ShutdownReport> {
-        let Some(mut child) = self.child.take() else {
+    /// Send SIGTERM without waiting, so a test can observe the drain sequence in progress.
+    pub fn signal_shutdown(&mut self) -> Result<()> {
+        let Some(child) = self.child.as_ref() else {
             bail!("the SUT is not running")
         };
         let pid = child.id().context("the SUT has no pid")? as i32;
-        let started = Instant::now();
+        ensure!(self.shutdown_started.is_none(), "shutdown was already requested");
 
         // SAFETY: `pid` names a child of this process that has not been reaped.
-        unsafe { libc::kill(pid, libc::SIGTERM) };
+        if unsafe { libc::kill(pid, libc::SIGTERM) } != 0 {
+            return Err(std::io::Error::last_os_error()).context("failed to send SIGTERM to the SUT");
+        }
+        self.shutdown_started = Some(Instant::now());
+        Ok(())
+    }
 
-        match tokio::time::timeout(Duration::from_secs(60), child.wait()).await {
+    /// Wait for a previously requested shutdown to finish.
+    pub async fn wait_for_shutdown(&mut self, timeout: Duration) -> Result<ShutdownReport> {
+        let Some(_) = self.child.as_ref() else {
+            bail!("the SUT is not running")
+        };
+        let started = self.shutdown_started.take().context("shutdown was not requested")?;
+        let mut child = self.child.take().expect("checked above");
+
+        match tokio::time::timeout(timeout, child.wait()).await {
             Ok(status) => Ok(ShutdownReport {
                 status: status.context("failed to wait for the SUT")?,
                 took: started.elapsed()
             }),
             Err(_) => {
                 child.kill().await.ok();
-                bail!("the SUT did not exit within 60s of SIGTERM")
+                bail!("the SUT did not exit within {timeout:?} of SIGTERM")
             }
         }
     }
 
+    /// SIGTERM and wait for the drain-and-exit (LIV-12, `P-SHUTDOWN`).
+    pub async fn stop(&mut self) -> Result<ShutdownReport> {
+        self.signal_shutdown()?;
+        self.wait_for_shutdown(Duration::from_secs(60)).await
+    }
+
     /// The tail of the service log, minus the HTTP access log — the harness's own polling makes
     /// that the loudest thing in there and never the interesting one.
     pub fn log_tail(&self, lines: usize) -> String {
@@ -184,6 +206,7 @@ impl Sut {
             .with_context(|| format!("failed to spawn {}", self.cfg.bin.display()))?;
 
         self.child = Some(child);
+        self.shutdown_started = None;
         self.await_ready(started).await?;
         self.last_startup = started.elapsed();
         Ok(())
diff --git a/crates/hotblocks/Cargo.toml b/crates/hotblocks/Cargo.toml
index e1348635..cd2d395f 100644
--- a/crates/hotblocks/Cargo.toml
+++ b/crates/hotblocks/Cargo.toml
@@ -29,6 +29,7 @@ sqd-query = { path = "../query", features = ["storage"] }
 sqd-storage = { path = "../storage" }
 tikv-jemallocator = "0.6.0"
 tokio = { workspace = true, features = ["full"] }
+tokio-util = { workspace = true }
 tower-http = { version = "0.6.1", features = ["request-id", "trace"] }
 tracing = { workspace = true, features = ["valuable"] }
 tracing-subscriber = { workspace = true, features = ["env-filter", "json", "valuable"] }
@@ -39,7 +40,7 @@ anyhow = { workspace = true }
 reqwest = { workspace = true }
 sqd-hotblocks-harness = { path = "../hotblocks-harness" }
 tempfile = { workspace = true }
-tokio = { workspace = true, features = ["full"] }
+tokio = { workspace = true, features = ["full", "test-util"] }
 
 [lints]
 workspace = true
diff --git a/crates/hotblocks/spec/12-conformance-tdd.md b/crates/hotblocks/spec/12-conformance-tdd.md
index 2997225a..2c738a46 100644
--- a/crates/hotblocks/spec/12-conformance-tdd.md
+++ b/crates/hotblocks/spec/12-conformance-tdd.md
@@ -227,7 +227,7 @@ INV-21/22/23 (checks in parentheses):
 6. anchored continuation across responses never breaks parent-hash chains (INV-23);
 7. watermark coherence: `first ≤ fin ≤ head` whenever reported together (INV-5/30).
 
-## 5. Traceability matrix (status @ 2026-07-20)
+## 5. Traceability matrix (status @ 2026-07-21)
 
 Legend: **C** covered, **P** partial (some storage-layer or fixture coverage exists;
 service-level black-box coverage absent), **U** untested. Rows that changed with Phase 0 name
@@ -267,14 +267,14 @@ the same property under forks, crashes and retention is the business of CT-2/CT-
 | RP-19/20 lookup contract | CT-5 | **P** | hit/miss, disabled-index behavior, over-limit rejection before dataset lookup, and the 256-byte boundary are pinned; the NOT_FOUND vs UNKNOWN_DATASET split remains open (GAP-39) |
 | LIV-1/2 progress/stall | CT-6/7 | **U — known-violated** | GAP-1 |
 | LIV-3 query termination | CT-3/6 | U | |
-| LIV-4 waiter termination | CT-1/3 | P | `ct1_happy_path`: a query above the head answers `NO_DATA` within `P-HEAD-WAIT` |
+| LIV-4 waiter termination | CT-1/3 | **P — known-violated** | `ct1_happy_path`: a query above the head answers `NO_DATA` within `P-HEAD-WAIT`. The shutdown clause is unimplemented — the wait is a fixed timeout that never learns of the signal, so waiters are released by expiry rather than released *on* shutdown (GAP-17) |
 | LIV-5/6 startup/recovery | CT-2/6 | **U — known-violated** | GAP-7; `Sut::last_startup` records SLI-5 per boot |
 | LIV-7 reclamation | CT-7 | U | runtime reclaim on by default since 2026-07 (GAP-6 residual: boot-gated residue purge); convergence unmeasured |
 | LIV-8 isolation | CT-8 | **U — known-violated** | GAP-1/14 |
 | LIV-9 fork convergence/alarm | CT-4 | **U — known-violated** | GAP-5 now has a deterministic repro (`ct4_a_single_source_signalling_a_fork_above_its_tip_does_not_park_ingestion`): one bad source of three parks ingestion — no alarm, no recovery |
 | LIV-10 overload recovery | CT-6 | U | |
 | LIV-11 retention keep-up | CT-7 | U | |
-| LIV-12 shutdown | CT-2 | **U — known-suspect** | GAP-17; `Sut::stop` measures the drain |
+| LIV-12 shutdown | CT-2 | **P — known-violated** | `ct2_shutdown` drives the real binary through SIGTERM and pins unready-before-grace, continued serving during grace, the hard drain deadline with an in-flight long-poll, bounded exit and success status. The clean-truncation and ingest-wind-down halves are still open and the deployed grace still forces SIGKILL (GAP-17) |
 | RS-3/4 window floor/excess | CT-1/7 | U | |
 | RS-6 amplification | CT-7 | U | reclaim path fixed 2026-07 (GAP-6); bound unmeasured under churn |
 | RS-8 boot maintenance | CT-2/7 | P | unlink/orphan-purge behaviors have storage-level tests |
@@ -288,7 +288,7 @@ the same property under forks, crashes and retention is the business of CT-2/CT-
 | OB-2..11 | all | P | query metrics exist; stall gauges pending on PR #83 (unmerged); OB-2 heartbeat, OB-6 debt accounting, OB-9 alarms, OB-11 forensics absent |
 | OB-12 index state | CT-1 | **P** | CF-wide estimated keys / live SST bytes are exported for both indexes; per-dataset enabled/count/bytes and lookup hit/miss/latency remain absent (GAP-40) |
 
-## 6. Gap register (dated 2026-07-20, informative)
+## 6. Gap register (dated 2026-07-21, informative)
 
 Known or strongly suspected divergences between this spec and the current system, from
 incident history, code-level review, and coverage analysis. Priorities: P0 = active
@@ -312,7 +312,7 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po
 | GAP-14 | Read-side capacity (execution slots, waiter slots) is a single global pool: one dataset's query herd can starve all datasets | PF-4, LIV-8 | P2 | CT-8: herd on D′, tip-follower SLOs on D |
 | GAP-15 | No explicit store-format compatibility gate at boot; incompatibility surfaces as runtime decode errors | CN-12, INV-43 | P3 | CT-5 boot matrix with future-format fixture |
 | GAP-16 | ~~The service layer has essentially zero automated tests~~ — **closed by Phase 0**, see §6.1 | all | — | done |
-| GAP-17 | Shutdown can take a panic-class exit path in ingestion cancellation (observed at redeploy) | LIV-12, FM-PROC-4 | P2 | CT-2 shutdown class: SIGTERM under load ×100, zero panic exits |
+| GAP-17 | Shutdown can take a panic-class exit path in ingestion cancellation (observed at redeploy). **2026-07-21 addendum**: replica replacement is client-visible, in two ways that must not be conflated. (a) *Idle pooled connection* — at SIGTERM the server closes keep-alive connections at once while the endpoint removal is still propagating (chart: grace 5 s, no `preStop`, no readiness probe at all), so the portal POSTs into a socket already gone: 502 burst observed one second after SIGTERM with a Ready sibling idle. Nothing was admitted, so LIV-12 does permit it, and masking it belongs outside this spec — portal replay on a fresh connection (`sqd-portal` 13-conformance GAP-22) plus a chart `preStop`. (b) *Active stream cut* — **not** permitted: RP-15 requires the truncation to be "observationally identical to a budget stop", and a process kill leaves an unterminated chunked body and a half-written codec frame, so the client cannot decode the prefix at all, let alone as valid JSONL (IB-5's looser paraphrase is not a licence). A response that has already committed 200 and emitted bytes is unreplayable by any client, so (b) is fixable only here. And (b) is not rare: the drain floor already equals `P-HEAD-WAIT`, because head-waiters run a fixed 5 s timeout with no knowledge of the signal (`query/service.rs`) in violation of LIV-4's explicit shutdown clause — that is the whole 5 s grace, so under tip-follower load the exit is SIGKILL, i.e. FM-PROC-1, and the "clean path" this entry once assumed is not taken in production at all. The drain is also unbounded from the inside: `axum::serve(..).with_graceful_shutdown` carries no deadline and GAP-29 (no response deadline; a stalled reader pins its response) means nothing bounds `P-SHUTDOWN` whatever the grace is. Lastly the RP-15 escape hatch owes a truncation counter that does not exist (GAP-10). **Partly closed 2026-07-21**: SIGTERM now runs the same two-phase sequence as the portal (sqd-portal#113) — `/ready` reports 503 for `--pre-drain-grace-secs` while everything else serves normally, then the drain runs under a hard `--drain-timeout-secs`; SIGINT stays on the default handler so dev Ctrl-C is unaffected. That answers (a) and bounds `P-SHUTDOWN` from the inside, but only takes effect once the chart probes `/ready` and raises the grace above the sum of the two. Still open: (b), since a stream cut at the deadline is still a reset rather than an RP-15 end; LIV-4's shutdown clause, since waiters still sit out their fixed timeout; and the original ingest-cancellation path | LIV-12, LIV-4, RP-15, FM-PROC-4 | **P1** (was P2 on the "bounded or rare" premise; the addendum removes it — it fires on every replica replacement and the clean path is never taken under load) | CT-2 shutdown class: SIGTERM under load ×100 — zero panic exits, exit ≤ `P-SHUTDOWN`, every cut stream decodes as a valid JSONL prefix, every long-poll released ≤ `P-HEAD-WAIT` |
 | GAP-18 | Dual-writer detection exists only on some paths (finality/head updates), not all mutations | WP-15, FM-OP-3 | P3 | CT-5: two harness-driven writers, assert loser stops on every mutation type |
 | GAP-20 | `parent_number` linkage is never validated on any layer (the block trait exposes it; nothing reads it): a hash-linked run can claim an arbitrarily higher number for the next block, storing a false hole on a densely-numbered chain — a silent data gap served as if it were a slot gap. (The originally-filed non-monotonic-numbers scenario is unreachable today: the source-position advance forces ascending numbers.) | WP-2, DEF-4, INV-1 | P2 | CT-4/CT-9: hash-linked run with a number jump on a dense chain; the run MUST be rejected with no state change |
 | GAP-21 | An anchored query whose `from` sits mid-chunk above a number gap larger than the conflict-check lookback (a hard-coded 100 positions in the plan's base-block check) fails `INTERNAL` instead of evaluating the assertion. A >100-position hole with an anchor landing just above it is probably unrealistic, hence low priority. A correct all-predecessors scan was tried and reverted 2026-07-15 — it regressed `check_parent_block` into an unbounded per-chunk scan+sort; needs a lazy sort-desc + limit(100) | RP-11, INV-26 | P3 | CT-5 `ct5_anchor_is_evaluated_across_a_large_number_hole` (`#[ignore]` until fixed): >100-position hole in one chunk; anchored query just above must yield OK/CONFLICT, never 500 |
@@ -358,7 +358,7 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po
 
   | Built, awaiting scripts | Missing |
   |---|---|
-  | `Sut::crash/stop/restart` (same db, same port) → CT-2 | the CT-2 kill-point matrix |
+  | `Sut::crash/stop/restart` (same db, same port); `ct2_shutdown` exercises the bounded SIGTERM path | the remaining CT-2 kill-point matrix |
   | `Harness::fork` + `Model::resolve_fork` + the follower's CONFLICT recovery → CT-4 | the CT-4 fork/finality corpus |
   | `Model::predict_query` + initial `ct5_error_soundness` matrix | remaining CT-5 binding, boot, and overload rows |
   | `SimFaults` injection point → CT-9 | the rest of the FM-SRC repertoire |
@@ -370,10 +370,12 @@ rare, P3 = polish. **First test** names the cheapest failing-test-first entry po
 
 - **Phase 1 — the P0s.** Stall harness + OB-2/3/11 signals (GAP-1); churn soak + space
   accounting via OB-6 (GAP-6). These target the two production incidents.
-- **Phase 2 — correctness core.** CT-2 crash matrix (GAP-2), CT-4 fork/finality corpus
-  (GAP-3/4/5), CT-5 remaining error taxonomy + boot matrix (GAP-8/9/15/36/39).
+- **Phase 2 — correctness core.** CT-2 crash matrix (GAP-2) and shutdown class (GAP-17 —
+  pulled forward from Phase 3: it is client-visible on every replica replacement), CT-4
+  fork/finality corpus (GAP-3/4/5), CT-5 remaining error taxonomy + boot matrix
+  (GAP-8/9/15/36/39).
 - **Phase 3 — robustness.** CT-9 fuzz both surfaces (GAP-12), CT-3 concurrency swarms,
-  CT-8 isolation (GAP-14), shutdown class (GAP-17).
+  CT-8 isolation (GAP-14).
 - **Phase 4 — performance regime.** CT-6 scenarios S1–S6, SLO gates, saturation knees,
   baselines (GAP-7/13); wire into CI with tolerances.
 
diff --git a/crates/hotblocks/spec/13-interface-binding.md b/crates/hotblocks/spec/13-interface-binding.md
index 06f7ca35..a6397f7b 100644
--- a/crates/hotblocks/spec/13-interface-binding.md
+++ b/crates/hotblocks/spec/13-interface-binding.md
@@ -34,6 +34,7 @@ annotated with the relevant GAP.
 | GET-RETENTION | `GET /datasets/{id}/retention` | current policy JSON |
 | SET-RETENTION | `POST /datasets/{id}/retention` | policy JSON; only for `External` datasets, else `FORBIDDEN` (403) |
 | observability | `GET /metrics` (+ engine-diagnostic routes) | OB surface, text formats |
+| readiness | `GET /ready` | rotation gate (OB-8), distinct from the `/` liveness signal: 503 for the whole pre-drain grace window so the orchestrator withdraws the endpoint before anything closes (LIV-12). Process-level only — per-dataset readability (LIV-5c) is still absent (GAP-7) |
 
 Dialects accepted in query bodies: `evm`, `solana`, `bitcoin`, `tron`,
 `hyperliquidFills`, `hyperliquidReplicaCmds` — and, expressible in the query schema but
@@ -71,8 +72,13 @@ below" never applies below genesis).
   projected, and anchored continuation (RP-10) therefore requires projecting them.
 - **IB-5** The success status is committed before streaming begins; admission-time errors
   therefore arrive as proper error statuses, while post-admission failures surface as
-  RP-15 truncation (stream ends early; already-sent prefix remains valid JSONL). See
-  GAP-10 for the observability obligation.
+  RP-15 truncation (stream ends early; already-sent prefix remains valid JSONL). At this
+  binding a truncation is a *server-side end*, not a dropped connection: the codec stream
+  and the chunked body MUST both be terminated, so that the client sees a short but
+  well-framed response, exactly as at a budget stop (RP-15). A reset connection leaves a
+  half-written codec frame the client cannot decode at all — a transport failure, not a
+  truncation, and not licensed by this clause (GAP-17). See GAP-10 for the observability
+  obligation.
 - **IB-6** Response watermark headers (success): finalized head number+hash when defined,
   and a head-number header (for the live stream: at least the snapshot head; may be
   fresher per CN-5). `NO_DATA` (204) SHOULD carry the same watermark headers — currently
diff --git a/crates/hotblocks/spec/14-parameters.md b/crates/hotblocks/spec/14-parameters.md
index 0d200bce..48f40094 100644
--- a/crates/hotblocks/spec/14-parameters.md
+++ b/crates/hotblocks/spec/14-parameters.md
@@ -62,7 +62,7 @@ ran against.
 | `P-ALARM` | integrity fault → observable alarm (WP-17, LIV-9b, OB-9) | ∞ (no alarm states exist: GAP-5) | ≤ 10 s ⚠ |
 | `P-STARTUP-ACCEPT` | process start → accepting connections (LIV-5a, SLI-5) | ~35 s observed (GAP-7) | ≤ 3 s ⚠ |
 | `P-STARTUP-READY(state)` | per-dataset readable bound (LIV-5b, SLI-6) | — | budget curve vs state size ⚠ |
-| `P-SHUTDOWN` | drain-and-exit bound (LIV-12) | — | ≤ 30 s ⚠ |
+| `P-SHUTDOWN` | drain-and-exit bound (LIV-12) | `--pre-drain-grace-secs` 25 s + `--drain-timeout-secs` 25 s since 2026-07-21 (GAP-17), bounded in-process at last; the deployment's 5 s grace is still below the sum, so the exit remains SIGKILL until the chart is raised | deployment grace ≥ `pre_drain_grace + drain_timeout`; drain deadline ≥ `P-QUERY-TIME` + `P-SCHED-SLACK` ⚠ |
 | `P-DUR-PROCESS` | commits lost on process crash (CN-6) | 0 | 0 |
 | `P-DUR-SYSTEM` | commit-suffix loss window on host/power failure (CN-6b) | bounded, engine-managed (not explicitly configured) | make explicit ⚠ |
 | `P-QUIESCENCE` | harness settling period before model comparison (12 §1) | — | 2× `P-CLEANUP-PERIOD` ⚠ |
diff --git a/crates/hotblocks/src/api.rs b/crates/hotblocks/src/api.rs
index 96a240a6..1ae8d0ab 100644
--- a/crates/hotblocks/src/api.rs
+++ b/crates/hotblocks/src/api.rs
@@ -1,6 +1,9 @@
 use std::{
     future::Future,
-    sync::{Arc, LazyLock},
+    sync::{
+        Arc, LazyLock,
+        atomic::{AtomicBool, Ordering}
+    },
     time::Instant
 };
 
@@ -95,7 +98,7 @@ macro_rules! get_dataset {
 
 type AppRef = Arc<App>;
 
-pub fn build_api(app: App) -> Router {
+pub fn build_api(app: App, shutting_down: Arc<AtomicBool>) -> Router {
     Router::new()
         .route("/", get(|| async { "Welcome to SQD hot block data service!" }))
         .route("/datasets/{id}/stream", post(stream))
@@ -114,6 +117,19 @@ pub fn build_api(app: App) -> Router {
         .layer(axum::middleware::from_fn(middleware))
         .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid::default()))
         .layer(Extension(Arc::new(app)))
+        // Routed after the layers deliberately: axum leaves later routes unwrapped, and the
+        // grace window's 503s are a rotation signal, not faults -- inside `middleware` they
+        // would land in `http_status` as `error_class="Unclassified"` on every termination.
+        .route("/ready", get(get_readiness).layer(Extension(shutting_down)))
+}
+
+/// Rotation gate, not liveness: 503 from the moment shutdown starts. Per-dataset
+/// readability (LIV-5c) is not modelled -- serving is gated on full init today (GAP-7).
+async fn get_readiness(Extension(shutting_down): Extension<Arc<AtomicBool>>) -> impl IntoResponse {
+    if shutting_down.load(Ordering::Relaxed) {
+        return (StatusCode::SERVICE_UNAVAILABLE, "Shutting down").into_response();
+    }
+    (StatusCode::OK, "Ready").into_response()
 }
 
 const HASH_MAX_LEN: usize = 256;
@@ -421,6 +437,24 @@ fn stream_can_finish_cleanly(err: &anyhow::Error) -> bool {
     !err.is::<QueryTaskPanicked>()
 }
 
+#[cfg(test)]
+mod readiness_tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn ready_flips_to_503_once_shutdown_starts() {
+        let shutting_down = Arc::new(AtomicBool::new(false));
+
+        let res = get_readiness(Extension(shutting_down.clone())).await.into_response();
+        assert_eq!(res.status(), StatusCode::OK);
+
+        shutting_down.store(true, Ordering::Relaxed);
+
+        let res = get_readiness(Extension(shutting_down)).await.into_response();
+        assert_eq!(res.status(), StatusCode::SERVICE_UNAVAILABLE);
+    }
+}
+
 #[cfg(test)]
 mod stream_query_response_tests {
     use std::collections::VecDeque;
diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs
index 20422097..ede190f0 100644
--- a/crates/hotblocks/src/cli.rs
+++ b/crates/hotblocks/src/cli.rs
@@ -48,6 +48,16 @@ pub struct CLI {
     #[arg(long, default_value = "3000")]
     pub port: u16,
 
+    /// After SIGTERM, keep serving for this long while `/ready` reports 503, so the
+    /// orchestrator withdraws our endpoint before any connection closes.
+    #[arg(long, value_name = "SECS", default_value = "25")]
+    pub pre_drain_grace_secs: u64,
+
+    /// Hard cap on the drain that follows. The orchestrator's kill timeout must exceed
+    /// `pre_drain_grace + drain_timeout`.
+    #[arg(long, value_name = "SECS", default_value = "25")]
+    pub drain_timeout_secs: u64,
+
     /// Enable rocksdb stats collection
     #[arg(long)]
     pub rocksdb_stats: bool,
diff --git a/crates/hotblocks/src/main.rs b/crates/hotblocks/src/main.rs
index 6d31704f..da2f1e75 100644
--- a/crates/hotblocks/src/main.rs
+++ b/crates/hotblocks/src/main.rs
@@ -9,12 +9,22 @@ mod metrics;
 mod query;
 mod types;
 
-use std::time::Duration;
+use std::{
+    future::{Future, IntoFuture},
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering}
+    },
+    time::{Duration, Instant}
+};
 
+use anyhow::Context;
 use api::build_api;
 use clap::Parser;
 use cli::CLI;
-use tracing::{debug, error, info, instrument};
+use tokio::signal::unix::{Signal, SignalKind, signal};
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, instrument, warn};
 use types::DBRef;
 
 #[global_allocator]
@@ -31,29 +41,128 @@ fn main() -> anyhow::Result<()> {
 
     init_tracing();
 
-    tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()?
-        .block_on(async {
-            let app = args.build_app().await?;
+    let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build()?;
 
-            // NB: startup disk recovery (orphan purge + file unlink, gated by
-            // --startup-disk-reclaim) already ran inside `build_app` -> `DataService::start`,
-            // before any controller spawned.
-            tokio::spawn(db_cleanup_task(app.db.clone()));
+    // Outlives the runtime on purpose -- see `flush_and_abandon_db`.
+    let mut db = None;
 
-            let api = build_api(app);
+    let result = runtime.block_on(async {
+        let app = args.build_app().await?;
+        db = Some(app.db.clone());
 
-            let listener = tokio::net::TcpListener::bind(("0.0.0.0", args.port)).await?;
+        // NB: startup disk recovery (orphan purge + file unlink, gated by
+        // --startup-disk-reclaim) already ran inside `build_app` -> `DataService::start`,
+        // before any controller spawned.
+        tokio::spawn(db_cleanup_task(app.db.clone()));
 
-            info!("server started, listening at {}", listener.local_addr()?);
+        let shutting_down = Arc::new(AtomicBool::new(false));
+        let drain = CancellationToken::new();
 
-            axum::serve(listener, api)
-                .with_graceful_shutdown(shutdown_signal())
-                .await?;
+        let sigterm = signal(SignalKind::terminate()).context("failed to install the SIGTERM handler")?;
+        tokio::spawn(watch_shutdown_signal(
+            sigterm,
+            shutting_down.clone(),
+            drain.clone(),
+            Duration::from_secs(args.pre_drain_grace_secs)
+        ));
 
-            Ok::<_, anyhow::Error>(())
-        })
+        let api = build_api(app, shutting_down);
+
+        let listener = tokio::net::TcpListener::bind(("0.0.0.0", args.port)).await?;
+
+        info!("server started, listening at {}", listener.local_addr()?);
+
+        let serve = axum::serve(listener, api).with_graceful_shutdown({
+            let drain = drain.clone();
+            async move { drain.cancelled().await }
+        });
+
+        drive_serve_with_drain(serve.into_future(), drain, Duration::from_secs(args.drain_timeout_secs)).await?;
+
+        Ok::<_, anyhow::Error>(())
+    });
+
+    // Compactions run back-to-back on the blocking pool, and the runtime's destructor joins
+    // whichever is mid-flight -- that, not the drain, is what used to set the exit time.
+    // What gets cut is an uncommitted RocksDB transaction, which CN-6 already treats as lost.
+    runtime.shutdown_background();
+
+    if let Some(db) = db {
+        flush_and_abandon_db(db);
+    }
+
+    result
+}
+
+/// Exit without closing the database. `rocksdb_close` waits out the background compactions,
+/// which is the unbounded tail `--drain-timeout-secs` exists to remove; skipping it makes
+/// every exit crash-equivalent, which CN-6 requires us to survive anyway. The memtables are
+/// flushed first so the price is not simply moved to the next boot's WAL replay -- a start
+/// path with a far tighter budget than this one (GAP-7).
+fn flush_and_abandon_db(db: DBRef) {
+    let started = Instant::now();
+    match db.flush_all() {
+        Ok(()) => info!(elapsed_ms = started.elapsed().as_millis() as u64, "memtables flushed"),
+        Err(err) => warn!(error =? err, "failed to flush memtables, the next boot replays the WAL")
+    }
+
+    // Dropping the last handle is what calls `rocksdb_close`, so this one must not be
+    // dropped. Any clone still held by a detached blocking job outlives the process.
+    std::mem::forget(db)
+}
+
+/// SIGINT stays on the default handler: Ctrl-C in dev must not sit out the grace window.
+/// Only the first SIGTERM is acted on; a repeat is ignored on purpose -- anyone wanting an
+/// immediate exit already has SIGINT and SIGKILL.
+async fn watch_shutdown_signal(
+    mut sigterm: Signal,
+    shutting_down: Arc<AtomicBool>,
+    drain: CancellationToken,
+    pre_drain_grace: Duration
+) {
+    sigterm.recv().await;
+    info!("SIGTERM received, starting the shutdown sequence");
+    run_shutdown_sequence(shutting_down, drain, pre_drain_grace).await
+}
+
+/// Report unready while still serving normally, so the orchestrator withdraws our endpoint
+/// before anything closes.
+async fn run_shutdown_sequence(shutting_down: Arc<AtomicBool>, drain: CancellationToken, pre_drain_grace: Duration) {
+    shutting_down.store(true, Ordering::Relaxed);
+    info!(
+        grace_secs = pre_drain_grace.as_secs(),
+        "/ready now reports 503; serving on until the grace window elapses"
+    );
+    tokio::time::sleep(pre_drain_grace).await;
+    info!("pre-drain grace elapsed, draining the HTTP server");
+    drain.cancel()
+}
+
+/// Nothing bounds a response today (GAP-29), so without a deadline the drain just waits for
+/// the orchestrator's kill.
+async fn drive_serve_with_drain<F>(serve: F, drain: CancellationToken, drain_timeout: Duration) -> std::io::Result<()>
+where
+    F: Future<Output = std::io::Result<()>>
+{
+    let deadline = async {
+        drain.cancelled().await;
+        tokio::time::sleep(drain_timeout).await
+    };
+
+    tokio::select! {
+        res = serve => {
+            res?;
+            info!("HTTP server drained cleanly");
+        }
+        _ = deadline => {
+            warn!(
+                timeout_secs = drain_timeout.as_secs(),
+                "drain timeout exceeded, in-flight connections are aborted on exit"
+            );
+        }
+    }
+
+    Ok(())
 }
 
 fn init_tracing() {
@@ -77,25 +186,86 @@ fn init_tracing() {
     }
 }
 
-async fn shutdown_signal() {
-    let ctrl_c = async {
-        tokio::signal::ctrl_c().await.expect("failed to install Ctrl+C handler");
-    };
+#[cfg(test)]
+mod shutdown_tests {
+    use super::*;
 
-    #[cfg(unix)]
-    let terminate = async {
-        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
-            .expect("failed to install signal handler")
-            .recv()
-            .await;
-    };
+    /// Guards the shape of `main`: a runtime dropped rather than backgrounded joins the
+    /// blocking job and the drain deadline stops bounding the exit.
+    #[test]
+    fn exit_does_not_wait_for_a_running_blocking_job() {
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
 
-    #[cfg(not(unix))]
-    let terminate = std::future::pending::<()>();
+        let entered = Arc::new(std::sync::Barrier::new(2));
+        let in_job = entered.clone();
+        runtime.spawn_blocking(move || {
+            in_job.wait();
+            std::thread::sleep(Duration::from_secs(30))
+        });
+        entered.wait();
 
-    tokio::select! {
-        _ = ctrl_c => {},
-        _ = terminate => {},
+        let started = std::time::Instant::now();
+        runtime.shutdown_background();
+        assert!(
+            started.elapsed() < Duration::from_secs(1),
+            "exit joined the blocking job"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn the_grace_window_holds_the_drain_but_not_the_unready_flag() {
+        let shutting_down = Arc::new(AtomicBool::new(false));
+        let drain = CancellationToken::new();
+        let grace = Duration::from_secs(5);
+
+        let task = tokio::spawn(run_shutdown_sequence(shutting_down.clone(), drain.clone(), grace));
+        tokio::task::yield_now().await;
+
+        assert!(shutting_down.load(Ordering::Relaxed), "unready immediately");
+        assert!(!drain.is_cancelled(), "drain held for the whole window");
+
+        tokio::time::advance(grace - Duration::from_millis(1)).await;
+        assert!(!drain.is_cancelled());
+
+        tokio::time::advance(Duration::from_millis(2)).await;
+        task.await.unwrap();
+        assert!(drain.is_cancelled());
+    }
+
+    #[tokio::test]
+    async fn a_drain_that_finishes_inside_the_deadline_returns_ok() {
+        let drain = CancellationToken::new();
+        let serve = async {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            Ok(())
+        };
+
+        drain.cancel();
+        drive_serve_with_drain(serve, drain, Duration::from_secs(30))
+            .await
+            .expect("clean drain");
+    }
+
+    #[tokio::test]
+    async fn a_response_that_never_finishes_hits_the_deadline() {
+        let drain = CancellationToken::new();
+        let serve = std::future::pending::<std::io::Result<()>>();
+
+        drain.cancel();
+        drive_serve_with_drain(serve, drain, Duration::from_millis(20))
+            .await
+            .expect("the deadline path still exits Ok");
+    }
+
+    #[tokio::test]
+    async fn a_serve_error_propagates() {
+        let serve = async { Err(std::io::Error::other("listener failed")) };
+
+        let res = drive_serve_with_drain(serve, CancellationToken::new(), Duration::from_secs(30)).await;
+        assert!(res.is_err());
     }
 }
 
diff --git a/crates/hotblocks/tests/ct2_shutdown.rs b/crates/hotblocks/tests/ct2_shutdown.rs
new file mode 100644
index 00000000..f6431300
--- /dev/null
+++ b/crates/hotblocks/tests/ct2_shutdown.rs
@@ -0,0 +1,97 @@
+//! CT-2 — exercise the real process-level SIGTERM path (LIV-12, GAP-17).
+
+use std::{sync::Arc, time::Duration};
+
+use anyhow::{Context, Result, ensure};
+use reqwest::StatusCode;
+use sqd_hotblocks_harness::{
+    chain::HlFills,
+    harness::{Harness, HarnessConfig}
+};
+
+const START: u64 = 1_000;
+const PRE_DRAIN_GRACE: Duration = Duration::from_secs(2);
+const DRAIN_TIMEOUT: Duration = Duration::from_secs(1);
+
+#[tokio::test(flavor = "multi_thread")]
+async fn sigterm_marks_unready_serves_through_grace_and_bounds_the_drain() -> Result<()> {
+    let mut cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(HlFills), START);
+    cfg.sut_args.extend([
+        "--pre-drain-grace-secs".into(),
+        PRE_DRAIN_GRACE.as_secs().to_string(),
+        "--drain-timeout-secs".into(),
+        DRAIN_TIMEOUT.as_secs().to_string()
+    ]);
+    let mut harness = Harness::start(cfg).await?;
+
+    let http = reqwest::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_secs(10))
+        .build()?;
+    let base_url = harness.sut.base_url();
+    let ready_url = format!("{base_url}/ready");
+    let stream_url = format!("{base_url}/datasets/{}/stream", harness.dataset);
+
+    assert_eq!(http.get(&ready_url).send().await?.status(), StatusCode::OK);
+
+    // With no blocks produced, this real request waits for P-HEAD-WAIT (5s). It must still
+    // be in flight when the shorter 2s grace + 1s drain deadline expires.
+    let query = harness.chain.scan_query(START, None, None);
+    let long_poll = tokio::spawn({
+        let http = http.clone();
+        async move { http.post(stream_url).json(&query).send().await }
+    });
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    ensure!(
+        !long_poll.is_finished(),
+        "the long-poll request did not remain in flight"
+    );
+
+    harness.sut.signal_shutdown()?;
+    await_status(&http, &ready_url, StatusCode::SERVICE_UNAVAILABLE).await?;
+
+    // The endpoint-removal window must not interrupt traffic that reached this replica through
+    // a stale route or an existing keep-alive connection.
+    assert_eq!(http.get(&base_url).send().await?.status(), StatusCode::OK);
+
+    let report = harness
+        .sut
+        .wait_for_shutdown(PRE_DRAIN_GRACE + DRAIN_TIMEOUT + Duration::from_secs(3))
+        .await?;
+
+    assert!(report.status.success(), "shutdown exited with {}", report.status);
+    assert!(
+        report.took >= PRE_DRAIN_GRACE + DRAIN_TIMEOUT - Duration::from_millis(250),
+        "process exited before the configured grace and drain deadline: {:?}",
+        report.took
+    );
+    assert!(
+        report.took < PRE_DRAIN_GRACE + DRAIN_TIMEOUT + Duration::from_secs(2),
+        "process exceeded the configured shutdown bound: {:?}",
+        report.took
+    );
+
+    let request_result = long_poll.await.context("long-poll task panicked")?;
+    ensure!(
+        request_result.is_err(),
+        "the deadline must abort a response that is still waiting for data"
+    );
+
+    Ok(())
+}
+
+async fn await_status(http: &reqwest::Client, url: &str, expected: StatusCode) -> Result<()> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
+    loop {
+        if let Ok(response) = http.get(url).send().await
+            && response.status() == expected
+        {
+            return Ok(());
+        }
+        ensure!(
+            tokio::time::Instant::now() < deadline,
+            "{url} did not start returning {expected} after SIGTERM"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
diff --git a/crates/hotblocks/tests/readiness_metrics.rs b/crates/hotblocks/tests/readiness_metrics.rs
new file mode 100644
index 00000000..59151430
--- /dev/null
+++ b/crates/hotblocks/tests/readiness_metrics.rs
@@ -0,0 +1,80 @@
+//! `/ready` is a rotation gate for the orchestrator, not part of the served API. Its 503s are
+//! deliberate, so they must not reach `http_status` -- `middleware` classifies every unlabelled
+//! 5xx as `error_class="UNCLASSIFIED"`, which is what error-rate alerting reads, and a kubelet
+//! probing through the whole grace window would forge a burst of faults on every termination.
+
+use std::time::Duration;
+
+use anyhow::{Result, ensure};
+use reqwest::StatusCode;
+use sqd_hotblocks_harness::sut::{Sut, SutConfig};
+
+/// Long enough that the whole assertion runs inside the grace window.
+const PRE_DRAIN_GRACE: Duration = Duration::from_secs(30);
+
+#[tokio::test(flavor = "multi_thread")]
+async fn readiness_probes_are_not_counted_as_http_traffic() -> Result<()> {
+    let mut sut = Sut::start(SutConfig {
+        bin: env!("CARGO_BIN_EXE_sqd-hotblocks").into(),
+        datasets: vec![],
+        args: vec![
+            "--pre-drain-grace-secs".to_owned(),
+            PRE_DRAIN_GRACE.as_secs().to_string(),
+        ],
+        rust_log: "error".to_owned(),
+        startup_timeout: Duration::from_secs(30)
+    })
+    .await?;
+
+    let http = reqwest::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_secs(10))
+        .build()?;
+    let ready_url = format!("{}/ready", sut.base_url());
+    let metrics_url = format!("{}/metrics", sut.base_url());
+
+    assert_eq!(http.get(&ready_url).send().await?.status(), StatusCode::OK);
+
+    sut.signal_shutdown()?;
+    await_unready(&http, &ready_url).await?;
+
+    // Probe a few more times: inside the middleware each of these would add to the counter.
+    for _ in 0..4 {
+        assert_eq!(
+            http.get(&ready_url).send().await?.status(),
+            StatusCode::SERVICE_UNAVAILABLE
+        );
+    }
+
+    // Serving is unaffected during the grace window -- that is the point of the phase.
+    let metrics = http.get(&metrics_url).send().await?.text().await?;
+    let counted: Vec<&str> = metrics
+        .lines()
+        .filter(|line| line.starts_with("hotblocks_http_status_total"))
+        .filter(|line| line.contains(r#"status="503""#) || line.contains("error_class"))
+        .collect();
+
+    assert!(
+        counted.is_empty(),
+        "readiness probes leaked into the HTTP error metrics:\n{}",
+        counted.join("\n")
+    );
+
+    Ok(())
+}
+
+async fn await_unready(http: &reqwest::Client, url: &str) -> Result<()> {
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
+    loop {
+        if let Ok(response) = http.get(url).send().await
+            && response.status() == StatusCode::SERVICE_UNAVAILABLE
+        {
+            return Ok(());
+        }
+        ensure!(
+            tokio::time::Instant::now() < deadline,
+            "SIGTERM did not withdraw {url} from rotation"
+        );
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+}
diff --git a/crates/storage/src/db/db.rs b/crates/storage/src/db/db.rs
index c1fb6b1b..fb02de9f 100644
--- a/crates/storage/src/db/db.rs
+++ b/crates/storage/src/db/db.rs
@@ -30,6 +30,18 @@ pub const CF_DELETED_TABLES: Name = "DELETED_TABLES";
 pub const CF_BLOCK_HASHES: Name = "BLOCK_HASHES";
 pub const CF_TRANSACTION_HASHES: Name = "TRANSACTION_HASHES";
 
+/// Every column family [`DatabaseSettings::open`] creates. Pinned by a test, because a
+/// family missing here would silently keep [`Database::flush_all`] from retiring the WAL.
+const ALL_CFS: [Name; 7] = [
+    CF_DATASETS,
+    CF_CHUNKS,
+    CF_TABLES,
+    CF_DIRTY_TABLES,
+    CF_DELETED_TABLES,
+    CF_BLOCK_HASHES,
+    CF_TRANSACTION_HASHES
+];
+
 /// Whole-file rewrite cadence for `CF_TABLES`. RocksDB leaves `periodic_compaction_seconds`
 /// disabled for leveled compaction without a compaction filter, so the effective baseline
 /// is the separate 30-day `ttl` default. 7 days buys ~4x that rewrite rate.
@@ -534,6 +546,21 @@ impl Database {
         Ok(())
     }
 
+    /// Flush every memtable, so the next boot has almost no WAL to replay. Every family has
+    /// to go: a WAL file lives until the last one that wrote into it has flushed.
+    ///
+    /// Costs the memtables and nothing else -- unlike closing the database, which waits out
+    /// the background compactions. That is the trade at exit: pay a bounded flush here
+    /// rather than an unbounded close, and never close at all.
+    pub fn flush_all(&self) -> anyhow::Result<()> {
+        let cfs: Vec<_> = ALL_CFS
+            .iter()
+            .map(|name| self.db.cf_handle(name).expect("column family opened at startup"))
+            .collect();
+        self.db.flush_cfs_opt(&cfs, &rocksdb::FlushOptions::default())?;
+        Ok(())
+    }
+
     /// Force a full compaction of the table-data column family. Test support: production
     /// relies on background compaction. Rewrites files, so it needs scratch space -- the
     /// very thing that deadlocks on the full disk [`Database::reclaim_disk_space`] exists for.
@@ -681,6 +708,49 @@ mod tests {
         measure_hash_index_compression(rocksdb::DBCompressionType::Lz4, "lz4");
     }
 
+    fn active_memtable_entries(db: &Database, cf_name: &str) -> u64 {
+        db.get_int_property(cf_name, "rocksdb.num-entries-active-mem-table")
+            .unwrap()
+            .unwrap()
+    }
+
+    #[test]
+    fn flush_all_empties_every_memtable() {
+        let dir = tempfile::tempdir().unwrap();
+        let db = DatabaseSettings::default().open(dir.path()).unwrap();
+
+        for cf_name in ALL_CFS {
+            db.db.put_cf(db.db.cf_handle(cf_name).unwrap(), b"k", b"v").unwrap();
+            assert_ne!(active_memtable_entries(&db, cf_name), 0, "{cf_name} was not written");
+        }
+
+        db.flush_all().unwrap();
+
+        for cf_name in ALL_CFS {
+            assert_eq!(
+                active_memtable_entries(&db, cf_name),
+                0,
+                "{cf_name} still holds the WAL"
+            );
+        }
+    }
+
+    #[test]
+    fn all_cfs_covers_every_family_the_database_opens() {
+        let dir = tempfile::tempdir().unwrap();
+        // Dropped at once: `list_cf` needs the LOCK back.
+        DatabaseSettings::default().open(dir.path()).unwrap();
+
+        let mut opened = RocksDB::list_cf(&RocksOptions::default(), dir.path()).unwrap();
+        opened.retain(|name| name != "default");
+        opened.sort();
+
+        let mut expected: Vec<_> = ALL_CFS.iter().map(|name| name.to_string()).collect();
+        expected.sort();
+
+        assert_eq!(opened, expected);
+    }
+
     #[test]
     fn create_purges_crash_residue_before_publishing_the_dataset_label() {
         let dir = tempfile::tempdir().unwrap();