Process contract: a bounded drain that exits 15, and a four-state health endpoint

CHANGELOG.md

@@ -1,5 +1,33 @@
 # Changelog
 
+## Unreleased
+
+### Added
+
+- `pipeline.drain_deadline_seconds` bounds the whole shutdown after SIGTERM.
+  The final batch, the managers' final poll and the state syncs share one
+  deadline, 30 seconds by default. When it passes the process exits 15 with
+  `system.lifecycle.drain_incomplete`. Nothing unwritten was committed, and
+  the next start replays it.
+- `/healthz` reports `starting`, `healthy`, `degraded` or `failed`, with a
+  `reason`. `degraded` means a sink's retry ladder is running, or the
+  pipeline recorded an error inside the last flush interval. `failed` answers
+  503 and the other three answer 200. The previous `ok` and `stuck` bodies
+  are now `healthy` and `failed`, and their HTTP codes did not change.
+- `sqlflow validate` warns when a sink's `retry.deadline_seconds` is longer
+  than the drain deadline.
+- `/stats` carries `errors` and `last_error` under `progress`.
+
+### Fixed
+
+- A SIGTERM that arrived while a batch was being flushed aborted the flush,
+  and the process exited with the sink's error instead of draining. The
+  batch now finishes inside the drain deadline.
+- A cancel during a table manager's regular poll skipped its final poll, and
+  a final poll that failed was logged while the process exited 0. The final
+  poll always runs, and its failure is the exit code.
+- One failed flush counted as three errors.
+
 ## v1.0.0 — sqlflow, the Go engine
 
 SQLFlow now ships a second engine: **sqlflow**, a Go rewrite of the Python

docs/coverage/features.yml

@@ -120,6 +120,10 @@ features:
     description: The process exit status carries the error code a supervisor reads.
     requires: [unit, release]
 
+  - id: lifecycle.health
+    description: /healthz reports starting, healthy, degraded or failed, with a reason.
+    requires: [unit, release]
+
   # --- The batch loop -----------------------------------------------------
   - id: core.consume_loop
     description: Accumulates a batch, flushes it, and commits in that order.

docs/coverage/invariants.yml

@@ -353,10 +353,14 @@ invariants:
     family: lifecycle
     class: liveness
     applies_to: pipeline
-    claim: The drain finishes or fails inside its deadline.
+    claim: >
+      A drain finishes or fails inside pipeline.drain_deadline_seconds. A sink
+      that never answers cannot hold the process past it: the loop returns
+      system.lifecycle.drain_incomplete, commits nothing for the batch it
+      could not write, and the next start replays it.
     verified_by: harness
     requires: []
-    tracked_by: "#161"
+    enforced: true
 
   - id: lifecycle.close.idempotent
     family: lifecycle
@@ -452,3 +456,15 @@ invariants:
     requires: []
     enforced: true
     violated_once: ["#267"]
+
+  - id: manager.drain.bounded
+    family: lifecycle
+    class: liveness
+    applies_to: manager
+    claim: >
+      The final poll after a cancel finishes or fails inside the drain
+      deadline. A sink that never answers cannot hold the process past it,
+      and every closed window it did not deliver stays in the state table.
+    verified_by: harness
+    requires: []
+    enforced: true

docs/coverage/matrix.md

@@ -41,6 +41,7 @@ added, and this page changes only when a status does.
 | `state.corruption` | A damaged state file fails the start rather than silently resetting. | ✅ | — | ✅ |
 | `lifecycle.drain` | SIGTERM writes the buffered batch before exiting. | ✅ | — | ✅ |
 | `lifecycle.exit_codes` | The process exit status carries the error code a supervisor reads. | ✅ | — | ✅ |
+| `lifecycle.health` | /healthz reports starting, healthy, degraded or failed, with a reason. | ✅ | — | ✅ |
 | `core.consume_loop` | Accumulates a batch, flushes it, and commits in that order. | ✅ | — | — |
 | `error.taxonomy` | Every failure carries a class.domain.reason code. | ✅ | — | — |
 | `error.raise` | Policy RAISE stops the pipeline on a bad record. | ✅ | — | — |
@@ -60,15 +61,15 @@ added, and this page changes only when a status does.
 | `tooling.conformance` | The harness proves the declared invariants for any integration. | ✅ | — | — |
 | `tooling.coverage` | Tests attribute to features and invariants, and the registries match the code. | ✅ | — | — |
 
-**37 features declared. 37 have at least one passing test attributed at every level they require, so 0 gap(s).**
+**38 features declared. 38 have at least one passing test attributed at every level they require, so 0 gap(s).**
 
 That sentence counts attribution, not proof. A feature is green here when
 a test named for it ran and passed; it says nothing about whether the
 integration behind it keeps a batch it could not deliver, or commits
 offsets only after a flush. Those are invariants, they are counted
 separately below, and the two numbers are not interchangeable.
 
-**37 invariants declared: 31 safety and 6 liveness. Of 142 (invariant, integration) cells: 63 proven, 51 missing, 0 skipped, 0 failing, 28 exempt. 0 gap(s).**
+**38 invariants declared: 31 safety and 7 liveness. Of 143 (invariant, integration) cells: 66 proven, 49 missing, 0 skipped, 0 failing, 28 exempt. 0 gap(s).**
 
 Safety says nothing bad happens. Liveness says something good
 eventually does, and the two are not interchangeable: a sink that
@@ -185,6 +186,7 @@ drains. An invariant holds only if it holds on all four.
 | --- | --- | --- |
 | `manager.publish.eventually` | A closed window reaches the sink without anything else happening. The loop polls on its own, and a window that closes is published. | ✅ u |
 | `manager.failure.exits` | A poll the sink refuses stops the manager with the sink's error, after one attempt, and the process exits with its code. The sink ran its retry ladder before the error arrived, so the manager does not retry in place, and a window the destination will not take is never collected, written and refused every tick while the process reports healthy. *(violated once: #267)* | ✅ u |
+| `manager.drain.bounded` | The final poll after a cancel finishes or fails inside the drain deadline. A sink that never answers cannot hold the process past it, and every closed window it did not deliver stays in the state table. | ✅ u |
 
 These lifecycle invariants are properties of the consume loop
 rather than of anything a config file names. The columns are
@@ -197,6 +199,6 @@ drains. An invariant holds only if it holds on all four.
 | --- | --- | --- | --- |
 | `pipeline.flush.eventually` | A batch that never reaches batchSize still reaches the sink, within the flush interval. | ✅ u | ✅ u |
 | `pipeline.progress.no_silent_stall` | A configuration cannot remove the flush ticker. flush_interval_seconds absent, zero or negative all run with the thirty second default, so a batch a low-traffic topic never fills still leaves on time. This entry previously claimed the opposite, that zero removed the ticker and stalled such a topic forever; the run command has always defaulted it. Pinned by TestCliInvocation_FlushIntervalNeverZero, not by the harness, and unenforced for that reason: the harness drives a pipeline that is already constructed, and this is a property of resolving the config before construction. There is nothing per-subject to observe, so demanding a cell from every subject would buy a fake rather than a proof. The liveness the harness can see is pipeline.flush.eventually, which it proves. | ❌ missing | ❌ missing |
-| `lifecycle.drain.bounded` | The drain finishes or fails inside its deadline. *(declared, tracked by #161)* | ❌ missing | ❌ missing |
+| `lifecycle.drain.bounded` | A drain finishes or fails inside pipeline.drain_deadline_seconds. A sink that never answers cannot hold the process past it: the loop returns system.lifecycle.drain_incomplete, commits nothing for the batch it could not write, and the next start replays it. | ✅ u | ✅ u |
 | `pipeline.batch.timeout` | A batch whose query exceeds the timeout fails the batch, not the process. *(declared, tracked by #163)* | ❌ missing | ❌ missing |
 

docs/coverage/status/features.yml

@@ -15,6 +15,7 @@ handler.inferred_mem: {unit: covered, integration: not_required, release: covere
 handler.structured: {unit: covered, integration: not_required, release: covered}
 lifecycle.drain: {unit: covered, integration: not_required, release: covered}
 lifecycle.exit_codes: {unit: covered, integration: not_required, release: covered}
+lifecycle.health: {unit: covered, integration: not_required, release: covered}
 manager.tumbling_window: {unit: covered, integration: not_required, release: covered}
 observability.debug_api: {unit: covered, integration: not_required, release: not_required}
 observability.metrics: {unit: covered, integration: not_required, release: not_required}

docs/coverage/status/manager.tumbling_window.yml

@@ -2,5 +2,6 @@
 # One line per invariant that applies to manager.tumbling_window, sorted by id: the status of each level.
 manager.delete.after_flush: {unit: covered, integration: missing, release: missing}
 manager.delete.nothing_on_failure: {unit: covered, integration: missing, release: missing}
+manager.drain.bounded: {unit: covered, integration: missing, release: missing}
 manager.failure.exits: {unit: covered, integration: missing, release: missing}
 manager.publish.eventually: {unit: covered, integration: missing, release: missing}

docs/coverage/status/pipeline.stateful.yml

@@ -2,7 +2,7 @@
 # One line per invariant that applies to pipeline.stateful, sorted by id: the status of each level.
 error.bad_record.threshold: {unit: missing, integration: missing, release: missing}
 error.dlq.carries_provenance: {unit: missing, integration: missing, release: missing}
-lifecycle.drain.bounded: {unit: missing, integration: missing, release: missing}
+lifecycle.drain.bounded: {unit: covered, integration: missing, release: missing}
 lifecycle.drain.on_cancel: {unit: covered, integration: missing, release: missing}
 pipeline.batch.timeout: {unit: missing, integration: missing, release: missing}
 pipeline.commit.after_flush: {unit: covered, integration: missing, release: missing}

docs/coverage/status/pipeline.stateless.yml

@@ -2,7 +2,7 @@
 # One line per invariant that applies to pipeline.stateless, sorted by id: the status of each level.
 error.bad_record.threshold: {unit: missing, integration: missing, release: missing}
 error.dlq.carries_provenance: {unit: missing, integration: missing, release: missing}
-lifecycle.drain.bounded: {unit: missing, integration: missing, release: missing}
+lifecycle.drain.bounded: {unit: covered, integration: missing, release: missing}
 lifecycle.drain.on_cancel: {unit: covered, integration: missing, release: missing}
 pipeline.batch.timeout: {unit: missing, integration: missing, release: missing}
 pipeline.commit.after_flush: {unit: covered, integration: missing, release: missing}