Managers: a refused poll stops the process, and the manager is a conformance subject

docs/coverage/integrations/manager.tumbling_window.yml

@@ -0,0 +1,14 @@
+# The tumbling window manager. A second loop that reaches a sink: collect
+# the closed windows, flush them, delete them from the state table. The
+# delete is its commit, so its invariants are the consume loop's restated
+# for that commit, plus the liveness a poll loop owes on its own.
+#
+# `constructed: false` because no constructor switch lists managers:
+# buildManagedTables builds the one kind there is. Kinds() must not be held
+# equal to this entry.
+id: manager.tumbling_window
+kind: manager
+constructed: false
+implements: [Manager]
+feature: manager.tumbling_window
+exempt: []

docs/coverage/invariants.yml

@@ -1,7 +1,7 @@
 # The invariants every integration must hold, and the evidence each requires.
 #
 # This file is declared, not derived, for the reason features.yml gives: the
-# failure it catches is an integration with no evidence at all. Seven of these
+# failure it catches is an integration with no evidence at all. Eight of these
 # have been violated and fixed since v1.0.4, and each of those rows names the
 # fix.
 #
@@ -377,3 +377,58 @@ invariants:
     verified_by: harness
     requires: []
     tracked_by: "#166"
+
+  # --- Managers: a second loop that reaches a sink ------------------------
+  # A table manager collects the windows that have closed, flushes them, and
+  # deletes them from the state table. The delete is its commit, so these are
+  # the consume loop's checkpoint claims restated for it, plus the liveness a
+  # poll loop owes on its own. The manager is not a sink and not a handler:
+  # it has a commit step and a liveness obligation, and a handler has
+  # neither. internal/conformance drives it the way it drives the loop.
+  - id: manager.delete.after_flush
+    family: checkpoint
+    class: safety
+    applies_to: manager
+    claim: >
+      Closed windows leave the state table only after the sink acknowledged
+      them. The table still holds every one of them when Flush runs.
+    verified_by: harness
+    requires: []
+    enforced: true
+
+  - id: manager.delete.nothing_on_failure
+    family: checkpoint
+    class: safety
+    applies_to: manager
+    claim: >
+      A failed flush deletes nothing. Every closed window stays in the state
+      table for the next attempt.
+    verified_by: harness
+    requires: []
+    enforced: true
+
+  - id: manager.publish.eventually
+    family: lifecycle
+    class: liveness
+    applies_to: manager
+    claim: >
+      A closed window reaches the sink without anything else happening. The
+      loop polls on its own, and a window that closes is published.
+    verified_by: harness
+    requires: []
+    enforced: true
+
+  - id: manager.failure.exits
+    family: lifecycle
+    class: liveness
+    applies_to: manager
+    claim: >
+      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.
+    verified_by: harness
+    requires: []
+    enforced: true
+    violated_once: ["#267"]

docs/coverage/matrix.md

@@ -68,7 +68,7 @@ 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.
 
-**32 invariants declared: 28 safety and 4 liveness. Of 136 (invariant, integration) cells: 57 proven, 51 missing, 0 skipped, 0 failing, 28 exempt. 0 gap(s).**
+**36 invariants declared: 30 safety and 6 liveness. Of 140 (invariant, integration) cells: 61 proven, 51 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
@@ -113,12 +113,14 @@ until an invariant's `requires` is filled in, and none is yet.
 
 ## Safety invariants: checkpoint
 
-| Invariant | Claim | `source.kafka` | `source.webhook` | `source.websocket` |
-| --- | --- | --- | --- | --- |
-| `source.commit.only_processed` | A source commits the marks the pipeline processed, never what it fetched. *(violated once: #154)* | ❌ missing | — exempt | — exempt |
-| `source.resume.from_committed` | Restart resumes at the committed position. No gap, and no replay before it. | ❌ missing | — exempt | — exempt |
-| `source.marks.never_regress` | A committed position never moves backwards. | ❌ missing | — exempt | — exempt |
-| `source.commit.on_revoke` | Marks commit when a partition is revoked, before the rebalance completes. *(declared, tracked by #183)* | ❌ missing | — exempt | — exempt |
+| Invariant | Claim | `source.kafka` | `source.webhook` | `source.websocket` | `manager.tumbling_window` |
+| --- | --- | --- | --- | --- | --- |
+| `source.commit.only_processed` | A source commits the marks the pipeline processed, never what it fetched. *(violated once: #154)* | ❌ missing | — exempt | — exempt | · |
+| `source.resume.from_committed` | Restart resumes at the committed position. No gap, and no replay before it. | ❌ missing | — exempt | — exempt | · |
+| `source.marks.never_regress` | A committed position never moves backwards. | ❌ missing | — exempt | — exempt | · |
+| `source.commit.on_revoke` | Marks commit when a partition is revoked, before the rebalance completes. *(declared, tracked by #183)* | ❌ missing | — exempt | — exempt | · |
+| `manager.delete.after_flush` | Closed windows leave the state table only after the sink acknowledged them. The table still holds every one of them when Flush runs. | · | · | · | ✅ u |
+| `manager.delete.nothing_on_failure` | A failed flush deletes nothing. Every closed window stays in the state table for the next attempt. | · | · | · | ✅ u |
 
 These checkpoint invariants are properties of the consume loop
 rather than of anything a config file names. The columns are
@@ -178,6 +180,11 @@ drains. An invariant holds only if it holds on all four.
 
 ## Liveness invariants: lifecycle
 
+| Invariant | Claim | `manager.tumbling_window` |
+| --- | --- | --- |
+| `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 |
+
 These lifecycle invariants are properties of the consume loop
 rather than of anything a config file names. The columns are
 its configurations, and `internal/conformance` runs each one

docs/coverage/status/manager.tumbling_window.yml

@@ -0,0 +1,6 @@
+# Generated by `make coverage-matrix`. Do not edit by hand.
+# 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.failure.exits: {unit: covered, integration: missing, release: missing}
+manager.publish.eventually: {unit: covered, integration: missing, release: missing}

internal/cli/run/root.go

@@ -408,6 +408,16 @@ func NewCommand() *cobra.Command {
 			// Managers run for the lifetime of the pipeline. Cancelling their
 			// context makes each publish one final time before returning, so
 			// windows that close during shutdown are not stranded.
+			//
+			// A manager that stops is a pipeline that has stopped publishing,
+			// whatever the consume loop is still doing: a windowed pipeline's
+			// entire output goes through its manager. So a manager failure
+			// cancels the run, the loop drains as it would on SIGTERM, and the
+			// manager's error is what the process exits with. Before #267 the
+			// failure was logged and the loop kept consuming into a table
+			// nothing would ever publish.
+			runCtx, failRun := context.WithCancelCause(ctx)
+			defer failRun(nil)
 			managerCtx, stopManagers := context.WithCancel(context.Background())
 			var managerWG sync.WaitGroup
 			for _, m := range managedTables {
@@ -416,6 +426,7 @@ func NewCommand() *cobra.Command {
 					defer managerWG.Done()
 					if err := m.Start(managerCtx); err != nil {
 						l.Error("table manager stopped", zap.Error(err))
+						failRun(err)
 					}
 				}(m)
 			}
@@ -456,12 +467,20 @@ func NewCommand() *cobra.Command {
 				statusWG.Wait()
 			}()
 
-			stats, err := turbine.ConsumeLoop(ctx, maxMsgs)
+			stats, err := turbine.ConsumeLoop(runCtx, maxMsgs)
 			// Restore default signal handling for the rest of the shutdown.
 			// The deferred drain below still has to run. Leaving the handler
 			// installed would swallow a second SIGTERM, so an operator could
 			// not interrupt a drain that hangs.
 			stopSignals()
+			// A cause other than plain cancellation is a manager's error. It
+			// outranks whatever the loop returned: the loop was stopped on
+			// purpose, and the manager's error carries the code a supervisor
+			// reads.
+			if cause := context.Cause(runCtx); cause != nil && cause != context.Canceled {
+				l.Error("table manager failed, pipeline stopped", zap.Error(cause))
+				return cause
+			}
 			if err != nil {
 				l.Error("failed to consume loop", zap.Error(err))
 				return err