From 50bae4941cadf09ea3fe2287584d25e19ffc83c5 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Sat, 12 Sep 2026 15:06:10 -0400 Subject: [PATCH 1/3] Managers: a refused poll stops the process, and the manager is a conformance subject Fixes #267. The tumbling window manager logged a failed poll and polled again. A window the destination rejected was collected, written and refused every tick for as long as the process lived, with the container reporting healthy throughout. Observed against the live Bluesky firehose: three minutes, a failure every ten seconds, zero rows written, container Up. Start now returns the first failed poll, and run treats a manager's error as the pipeline's: it cancels the consume loop, which drains as it would on SIGTERM, and the process exits with the manager's error. The sink ran its retry ladder before the error reached the manager, so what arrives is final, and a retry in place repeats one the ladder already exhausted. The rows stay in the state table, because Poll deletes only after the sink accepted them, so a restart republishes the window. The manager is its own kind in the registry. It is not a sink and not a handler: it collects, flushes and deletes, and the delete is its commit. So its invariants are the consume loop's checkpoint claims restated for that commit, plus the liveness a poll loop owes on its own: - manager.delete.after_flush: the state table still holds every closed window when Flush runs, and none once the poll returns. - manager.delete.nothing_on_failure: a failed flush deletes nothing. - manager.publish.eventually: the loop publishes a closed window on its own. - manager.failure.exits: a refused poll stops the loop after one attempt. internal/conformance drives a ManagerSubject the way it drives a PipelineSubject: the harness owns a recording sink, fails its flush, and judges the event order and the state table. The tumbling manager supplies build, seed and count. All four are enforced. The status file for the new integration follows from CI's reports. --- .../integrations/manager.tumbling_window.yml | 14 + docs/coverage/invariants.yml | 55 ++++ internal/cli/run/root.go | 21 +- internal/conformance/manager.go | 301 ++++++++++++++++++ internal/coverage/registry_test.go | 10 +- internal/managers/conformance_test.go | 52 +++ internal/managers/tumbling.go | 15 +- internal/managers/tumbling_test.go | 81 +++-- scripts/coverage_matrix/registries.py | 7 +- tests/tooling/test_registries.py | 2 +- 10 files changed, 503 insertions(+), 55 deletions(-) create mode 100644 docs/coverage/integrations/manager.tumbling_window.yml create mode 100644 internal/conformance/manager.go create mode 100644 internal/managers/conformance_test.go diff --git a/docs/coverage/integrations/manager.tumbling_window.yml b/docs/coverage/integrations/manager.tumbling_window.yml new file mode 100644 index 00000000..d8a1e2b9 --- /dev/null +++ b/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: [] diff --git a/docs/coverage/invariants.yml b/docs/coverage/invariants.yml index 843b7e40..1ef94fa4 100644 --- a/docs/coverage/invariants.yml +++ b/docs/coverage/invariants.yml @@ -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"] diff --git a/internal/cli/run/root.go b/internal/cli/run/root.go index 2e9a75eb..9d0e8af2 100644 --- a/internal/cli/run/root.go +++ b/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 diff --git a/internal/conformance/manager.go b/internal/conformance/manager.go new file mode 100644 index 00000000..3c065d7f --- /dev/null +++ b/internal/conformance/manager.go @@ -0,0 +1,301 @@ +package conformance + +// The manager half of the harness. +// +// A table manager is a second loop that reaches a sink: it collects the +// windows that have closed, flushes them, and deletes them from the state +// table. The delete is its commit. So its claims are the consume loop's, +// restated for that commit, plus the two things a poll loop owes on its own: +// a closed window leaves, and a poll that cannot deliver stops the process +// rather than repeating forever. +// +// The second of those is #267. The manager logged a failed poll and polled +// again, so a window the destination rejected was collected, written and +// refused every tick for as long as the process lived, and the container +// reported healthy throughout. + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/turbolytics/sql-flow/internal/core" + "github.com/turbolytics/sql-flow/internal/coverage" + "go.opentelemetry.io/otel/metric/noop" +) + +// Manager is what a subject builds: the poll loop, and one poll of it. +type Manager interface { + Start(ctx context.Context) error + Poll(ctx context.Context) error +} + +// ManagerSubject is one manager the harness can drive. +type ManagerSubject struct { + // Integration is the registry id, e.g. "manager.tumbling_window". + Integration string + + // New builds the manager around the sink the harness supplies, polling at + // the given interval. The harness owns the sink so it can fail a flush + // and record what the manager asked of it. + New func(t *testing.T, sink core.Sink, poll time.Duration) Manager + + // Seed replaces the state table's contents with n closed windows. + Seed func(t *testing.T, n int) + + // Remaining returns how many closed windows the state table still holds. + Remaining func(t *testing.T) int64 +} + +// Managers proves every manager invariant against the subject. +func Managers(t *testing.T, s ManagerSubject) { + t.Helper() + if s.Integration == "" { + t.Fatal("conformance: ManagerSubject.Integration is required") + } + if s.New == nil || s.Seed == nil || s.Remaining == nil { + t.Fatal("conformance: ManagerSubject needs New, Seed and Remaining") + } + + feature, hasFeature, err := coverage.FeatureFor(s.Integration) + if err != nil { + t.Fatalf("conformance: %v", err) + } + + for _, v := range managerVerdicts(t, s) { + t.Run(v.invariant, func(t *testing.T) { + if hasFeature { + coverage.Covers(t, feature) + } + if v.failure != "" { + t.Fatal(v.failure) + } + coverage.Invariant(t, v.invariant, s.Integration) + }) + } + + if hasFeature { + coverage.Covers(t, feature) + } +} + +const ( + deleteAfterFlush = "manager.delete.after_flush" + deleteNothingOnFail = "manager.delete.nothing_on_failure" + publishEventually = "manager.publish.eventually" + failureExits = "manager.failure.exits" +) + +// seededWindows is how many closed windows each check starts with. Two, +// so a manager that deletes one row per flush is caught. +const seededWindows = 2 + +// managerPoll is the interval the harness runs the loop at. +const managerPoll = 20 * time.Millisecond + +// managerWait bounds every wait on the loop. It is many times the poll +// interval on purpose: a tighter bound asserts how fast the machine is, +// which is how #245 failed on CI. +const managerWait = 5 * time.Second + +func managerVerdicts(t *testing.T, s ManagerSubject) []verdict { + t.Helper() + + afterFlush := verdict{invariant: deleteAfterFlush} + onFailure := verdict{invariant: deleteNothingOnFail} + eventually := verdict{invariant: publishEventually} + exits := verdict{invariant: failureExits} + + if err := checkDeleteAfterFlush(t, s); err != nil { + afterFlush.failure = err.Error() + } + if err := checkDeleteNothingOnFailure(t, s); err != nil { + onFailure.failure = err.Error() + } + if err := checkPublishEventually(t, s); err != nil { + eventually.failure = err.Error() + } + if err := checkFailureExits(t, s); err != nil { + exits.failure = err.Error() + } + + return []verdict{afterFlush, onFailure, eventually, exits} +} + +// newManagerRun seeds the state table and builds the manager on a recording +// sink. The sink discards rows and records flushes, and fails every flush +// when asked, the way the pipeline harness's double-only subjects do. +func newManagerRun(t *testing.T, s ManagerSubject, fail bool) (Manager, *Recorder, *recordingSink) { + t.Helper() + s.Seed(t, seededWindows) + rec := &Recorder{} + sink := newRecordingSink(rec, nil, noop.NewMeterProvider()) + sink.fail = fail + return s.New(t, sink.counted, managerPoll), rec, sink +} + +// checkDeleteAfterFlush holds that the state table still has every closed +// window at the moment the sink is flushed, and none once the poll returns. +// +// Observed from inside the flush rather than inferred from the count +// afterwards: a manager that deleted first and flushed second leaves the +// same empty table, and the difference is whether a failed flush loses the +// window. +func checkDeleteAfterFlush(t *testing.T, s ManagerSubject) error { + t.Helper() + m, rec, _ := newManagerRun(t, s, false) + + atFlush := int64(-1) + rec.onEvent = func(event string) { + if event == "flush" { + atFlush = s.Remaining(t) + } + } + + if err := m.Poll(context.Background()); err != nil { + return fmt.Errorf("a poll with no fault injected failed: %v", err) + } + if !sameOrder(rec.Events(), []string{"flush"}) { + return fmt.Errorf("the poll did %s; want one flush", list(rec.Events())) + } + if atFlush != seededWindows { + return fmt.Errorf( + "the state table held %d of %d closed windows when the sink was "+ + "flushed. The delete ran first, so a flush that failed would "+ + "have lost them with no record anywhere", + atFlush, seededWindows) + } + if left := s.Remaining(t); left != 0 { + return fmt.Errorf( + "the sink accepted %d windows and %d are still in the state "+ + "table, so the next poll publishes them again", + seededWindows, left) + } + return nil +} + +// checkDeleteNothingOnFailure fails the flush and holds every window still. +func checkDeleteNothingOnFailure(t *testing.T, s ManagerSubject) error { + t.Helper() + m, rec, _ := newManagerRun(t, s, true) + + if err := m.Poll(context.Background()); err == nil { + return fmt.Errorf("the flush failed and the poll did not") + } + if !sameOrder(rec.Events(), []string{"flush-failed"}) { + return fmt.Errorf("after a failed flush the manager did %s; want "+ + "the failed flush and nothing else", list(rec.Events())) + } + if left := s.Remaining(t); left != seededWindows { + return fmt.Errorf( + "the sink refused the windows and %d of %d are gone from the "+ + "state table. A window the destination never took is lost, "+ + "and nothing downstream can tell", + seededWindows-left, seededWindows) + } + return nil +} + +// checkPublishEventually starts the loop over closed windows and holds that +// they reach the sink without anything else happening. +// +// The only liveness claim the delete checks leave open. A manager that never +// polled would satisfy both of them and fail this one alone. +func checkPublishEventually(t *testing.T, s ManagerSubject) error { + t.Helper() + m, rec, _ := newManagerRun(t, s, false) + + flushed := make(chan struct{}, 1) + rec.onEvent = func(event string) { + if event == "flush" { + select { + case flushed <- struct{}{}: + default: + } + } + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- m.Start(ctx) }() + + select { + case <-flushed: + case <-time.After(managerWait): + cancel() + <-done + return fmt.Errorf( + "%d closed windows sat in the state table for %s and the manager "+ + "never flushed them. A window that closes and is never "+ + "published is an aggregate nobody sees", + seededWindows, managerWait) + } + + cancel() + select { + case err := <-done: + if err != nil { + return fmt.Errorf("Start returned an error after a clean cancel: %v", err) + } + case <-time.After(managerWait): + return fmt.Errorf("Start did not return within %s of its cancel", managerWait) + } + + if left := s.Remaining(t); left != 0 { + return fmt.Errorf("the loop flushed and %d windows are still in the "+ + "state table", left) + } + return nil +} + +// checkFailureExits starts the loop over a sink that rejects every flush and +// holds that Start returns, on its own, with the error. +// +// The sink runs its own retry ladder before an error reaches the manager, +// so what arrives is final: the destination refused the rows, or stayed +// unreachable past the deadline. A second attempt here is a retry the ladder +// already exhausted, and an unbounded series of them is #267: the same +// windows collected, written and refused every tick, with the process +// reporting healthy throughout. +func checkFailureExits(t *testing.T, s ManagerSubject) error { + t.Helper() + m, _, sink := newManagerRun(t, s, true) + + // Cancellable so a loop that never stops can still be torn down, but + // never cancelled before the verdict: the claim is that it stops itself. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- m.Start(ctx) }() + + select { + case err := <-done: + if err == nil { + return fmt.Errorf("the sink rejected every flush and Start returned nil") + } + case <-time.After(managerWait): + cancel() + <-done + return fmt.Errorf( + "the sink rejected every flush for %s and the manager kept "+ + "polling, %d attempts. Each poll collected the same windows and "+ + "failed the same way, and the process stayed up. A supervisor "+ + "never notices, and the table never drains", + managerWait, sink.Flushes()) + } + + if n := sink.Flushes(); n != 1 { + return fmt.Errorf( + "the manager attempted %d flushes before stopping. The sink ran "+ + "its retry ladder before the first one failed, so every attempt "+ + "after it repeats a retry the ladder already exhausted", n) + } + if left := s.Remaining(t); left != seededWindows { + return fmt.Errorf( + "the manager stopped and %d of %d windows are gone from the state "+ + "table. They were never delivered, so a restart cannot publish "+ + "them", seededWindows-left, seededWindows) + } + return nil +} diff --git a/internal/coverage/registry_test.go b/internal/coverage/registry_test.go index 0133fffe..466aaba3 100644 --- a/internal/coverage/registry_test.go +++ b/internal/coverage/registry_test.go @@ -44,18 +44,20 @@ func TestToolingCoverageRegistry_ReadsEveryFileInTheDirectory(t *testing.T) { all, err := loadIntegrations() assert.NoError(t, err) - // Every kind is present, including the two no constructor builds. A - // loader that silently skipped a file would leave an integration with no - // cells, which is the sink.iceberg failure. + // Every kind is present, including the two no constructor builds: + // pipeline configurations and managers. A loader that silently skipped a + // file would leave an integration with no cells, which is the + // sink.iceberg failure. kinds := map[string]int{} for _, entry := range all { assert.That(t, entry.ID != "") assert.That(t, entry.Kind != "") kinds[entry.Kind]++ } - assert.Equal(t, 4, len(kinds)) + assert.Equal(t, 5, len(kinds)) assert.That(t, kinds["sink"] >= 6) assert.That(t, kinds["pipeline"] >= 2) + assert.That(t, kinds["manager"] >= 1) } func TestToolingCoverageRegistry_ReadsInIDOrder(t *testing.T) { diff --git a/internal/managers/conformance_test.go b/internal/managers/conformance_test.go new file mode 100644 index 00000000..0dfd44de --- /dev/null +++ b/internal/managers/conformance_test.go @@ -0,0 +1,52 @@ +package managers + +// The tumbling window manager under the conformance harness. +// +// The subject supplies three things: build the manager on the sink the +// harness hands it, put closed windows in the state table, and count what +// is left. The harness owns the sink, the faults and the verdicts, so a +// second manager kind supplies the same three things and inherits every +// manager invariant. + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/turbolytics/sql-flow/internal/conformance" + "github.com/turbolytics/sql-flow/internal/core" + "github.com/turbolytics/sql-flow/internal/coverage" +) + +func TestManagerTumblingWindow_Conformance(t *testing.T) { + coverage.Covers(t, "manager.tumbling_window") + + conn, cleanup := newTestConn(t) + defer cleanup() + exec(t, conn, `CREATE TABLE agg_cities_count (bucket TIMESTAMPTZ, city VARCHAR, count INT);`) + + conformance.Managers(t, conformance.ManagerSubject{ + Integration: "manager.tumbling_window", + + New: func(t *testing.T, sink core.Sink, poll time.Duration) conformance.Manager { + return NewTumbling(conn, collectSQL, deleteSQL, poll, sink, &sync.Mutex{}) + }, + + // Every seeded row is ten minutes old, so the close predicate the + // package's other tests use selects all of them. + Seed: func(t *testing.T, n int) { + exec(t, conn, `DELETE FROM agg_cities_count`) + for i := 0; i < n; i++ { + exec(t, conn, fmt.Sprintf( + `INSERT INTO agg_cities_count VALUES + (now()::timestamptz - INTERVAL '600' SECOND, 'city-%d', %d)`, + i, i+1)) + } + }, + + Remaining: func(t *testing.T) int64 { + return countRows(t, conn, "agg_cities_count") + }, + }) +} diff --git a/internal/managers/tumbling.go b/internal/managers/tumbling.go index 4ef40ca2..2a439b78 100644 --- a/internal/managers/tumbling.go +++ b/internal/managers/tumbling.go @@ -72,6 +72,15 @@ func WithLogger(l *zap.Logger) TumblingOption { // Start polls until the context is cancelled, then polls once more so windows // that closed during the final interval are not stranded in the table. +// +// A failed poll returns. The rows are still in the table, because Poll +// deletes only after the sink accepted them, so a restart republishes the +// same window. What Start does not do is retry in place: the sink already ran +// its retry ladder before the error reached here, so what arrives is either a +// destination that rejected the rows or one that stayed unreachable past the +// deadline. Before #267 the loop logged the error and polled again, and a +// rejected window was collected, written and refused every tick for as long +// as the process lived, with the container reporting healthy throughout. func (m *Tumbling) Start(ctx context.Context) error { m.logger.Info("starting tumbling window manager", zap.Duration("poll_interval", m.pollInterval)) @@ -83,13 +92,13 @@ func (m *Tumbling) Start(ctx context.Context) error { select { case <-ticker.C: if err := m.Poll(ctx); err != nil { - // A failed poll must not kill the manager: the next tick - // retries, and the rows are still in the table. - m.logger.Error("poll failed", zap.Error(err)) + m.logger.Error("poll failed, stopping the manager", zap.Error(err)) + return fmt.Errorf("tumbling window manager: %w", err) } case <-ctx.Done(): if err := m.Poll(context.Background()); err != nil { m.logger.Error("final poll failed", zap.Error(err)) + return fmt.Errorf("tumbling window manager: final poll: %w", err) } return nil } diff --git a/internal/managers/tumbling_test.go b/internal/managers/tumbling_test.go index 31af8d93..e0772ef3 100644 --- a/internal/managers/tumbling_test.go +++ b/internal/managers/tumbling_test.go @@ -307,6 +307,44 @@ func TestManagerTumblingWindow__RetriesAfterAFailedPoll(t *testing.T) { assert.Equal(t, int64(1), countRows(t, conn, "agg_cities_count")) } +// A poll the sink refuses stops the loop. Start returns the error after one +// attempt, with the windows still in the table for a restart to publish. +// +// Before #267 Start logged the error and polled again, so a window the +// destination rejected was collected, written and refused every tick for as +// long as the process lived, and the container reported healthy throughout. +// The sink runs its own retry ladder before an error reaches the manager, so +// a second attempt here repeats a retry the ladder already exhausted. +func TestManagerTumblingWindow__StartReturnsTheFirstFailedPoll(t *testing.T) { + coverage.Covers(t, "manager.tumbling_window") + conn, cleanup := newTestConn(t) + defer cleanup() + seedWindows(t, conn) + + sink := &failingSink{} + sink.failFlush.Store(true) + m := newTestTumbling(conn, &sink.recordingSink) + m.sink = sink + + // Never cancelled: the claim is that the loop stops on its own. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- m.Start(ctx) }() + + select { + case err := <-done: + assert.Error(t, err) + case <-time.After(5 * time.Second): + cancel() + <-done + t.Fatal("the sink rejected every flush and Start kept polling") + } + + // Two closed and one open: nothing was deleted. + assert.Equal(t, int64(3), countRows(t, conn, "agg_cities_count")) +} + // A broken delete statement must surface as an error rather than silently // republishing the same window on every poll. func TestManagerTumblingWindow__DeleteFailureIsReported(t *testing.T) { @@ -436,49 +474,6 @@ func TestManagerTumblingWindow__FinalPollOnShutdownPublishesAClosedWindow(t *tes assert.Equal(t, int64(1), countRows(t, conn, "agg_cities_count")) } -// A failing poll must not kill the manager: the rows stay and the next tick -// retries them. -func TestManagerTumblingWindow__StartSurvivesAFailedPoll(t *testing.T) { - coverage.Covers(t, "manager.tumbling_window") - conn, cleanup := newTestConn(t) - defer cleanup() - seedWindows(t, conn) - - sink := &failingSink{} - sink.failFlush.Store(true) - m := NewTumbling(conn, collectSQL, deleteSQL, 5*time.Millisecond, - &sink.recordingSink, &sync.Mutex{}) - m.sink = sink - - ctx, cancel := context.WithCancel(context.Background()) - done := make(chan error, 1) - go func() { done <- m.Start(ctx) }() - - time.Sleep(60 * time.Millisecond) - sink.failFlush.Store(false) - - deadline := time.After(5 * time.Second) - for { - if rows, _ := sink.recordingSink.counts(); rows >= 2 { - break - } - select { - case <-deadline: - t.Fatal("manager stopped polling after a failure") - default: - time.Sleep(5 * time.Millisecond) - } - } - - cancel() - select { - case err := <-done: - assert.NoError(t, err) - case <-time.After(5 * time.Second): - t.Fatal("Start did not return after cancellation") - } -} - // --- Interaction with the pipeline's state transaction --------------------- // A stateful pipeline runs with autocommit disabled, and the manager shares diff --git a/scripts/coverage_matrix/registries.py b/scripts/coverage_matrix/registries.py index 95d7b4ca..87b8c1bf 100644 --- a/scripts/coverage_matrix/registries.py +++ b/scripts/coverage_matrix/registries.py @@ -93,7 +93,7 @@ CLASSES = ("safety", "liveness") -KINDS = ("sink", "source", "handler", "pipeline") +KINDS = ("sink", "source", "handler", "pipeline", "manager") # How an invariant collects evidence. Both are explicit markers: a test says @@ -113,8 +113,9 @@ # A pipeline configuration is an integration of the harness, though no # constructor switch builds one. `constructed: false` says so, and the Kinds() -# agreement tests skip those entries. -INTEGRATION_KINDS = ("sink", "source", "handler", "pipeline") +# agreement tests skip those entries. A manager is the same: buildManagedTables +# builds the one kind there is, and no switch lists it. +INTEGRATION_KINDS = ("sink", "source", "handler", "pipeline", "manager") def load_features(): diff --git a/tests/tooling/test_registries.py b/tests/tooling/test_registries.py index efff601e..e37127a5 100644 --- a/tests/tooling/test_registries.py +++ b/tests/tooling/test_registries.py @@ -91,7 +91,7 @@ def test_the_committed_registries_load_and_validate(): assert validate_registries(invariants, integrations, features) == [] assert len(invariants) >= 20 assert {i["kind"] for i in integrations} == { - "sink", "source", "handler", "pipeline"} + "sink", "source", "handler", "pipeline", "manager"} def test_every_committed_invariant_is_unenforced_in_this_revision(): From bac3b3147b7a1c43436d15fa0cb6dff0fe2a1203 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Sat, 12 Sep 2026 15:12:37 -0400 Subject: [PATCH 2/3] coverage: the manager's status file, from CI's reports --- docs/coverage/invariants.yml | 2 +- docs/coverage/status/manager.tumbling_window.yml | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) create mode 100644 docs/coverage/status/manager.tumbling_window.yml diff --git a/docs/coverage/invariants.yml b/docs/coverage/invariants.yml index 1ef94fa4..71201bf9 100644 --- a/docs/coverage/invariants.yml +++ b/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. # diff --git a/docs/coverage/status/manager.tumbling_window.yml b/docs/coverage/status/manager.tumbling_window.yml new file mode 100644 index 00000000..c73b7c29 --- /dev/null +++ b/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} From cd84bc0d8ff562fc6a56dbd14ae24cd532a05762 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Sat, 12 Sep 2026 15:13:38 -0400 Subject: [PATCH 3/3] coverage: regenerate the page after rebasing on #269 --- docs/coverage/matrix.md | 21 ++++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/docs/coverage/matrix.md b/docs/coverage/matrix.md index 33dea611..81e65572 100644 --- a/docs/coverage/matrix.md +++ b/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