diff --git a/README.md b/README.md index 19183ff..3739092 100644 --- a/README.md +++ b/README.md @@ -106,13 +106,19 @@ flowctl pipelines stop ``` `flowctl run` waits for delivery to complete or for an operator to stop it. A -bounded run completes successfully once every edge closes and sinks acknowledge; -a source, processor, or sink delivery failure marks the run failed and exits -nonzero. An unexpected EOF from an unbounded source is a failure. Currently -bounded completion is inferred from the source's `END_LEDGER` environment -variable; a source with a different finite stopping condition will be treated -as unbounded. On interruption or `pipelines stop`, the run is stopped rather +bounded run completes successfully only after every edge closes and each sink +acknowledges exactly the events sent to its stream (zero failures and errors). +Sinks must use flowctl-sdk v0.1.5 or implement the equivalent contract: +`EventsConsumed` counts successfully finished handlers, which return only after +durable commit. A source, processor, or sink delivery failure marks the run +failed and exits nonzero. An unexpected EOF from an unbounded source is a +failure. Set `FLOWCTL_SOURCE_BOUNDED: "true"` in each finite source's `env`; +`END_LEDGER` alone does not declare boundedness (`0` can mean unbounded). On interruption or `pipelines stop`, the run is stopped rather than failed. Restart and checkpoints remain the operator's responsibility. +The current `events_processed` run field is not a transport, handler-completion, +or durable-commit metric (currently it remains zero). Fan-out counts are per +sink stream, not distinct ledgers; use catalog watermarks for durable proof. +flowctl v0.0.11 with SDK v0.1.4 cannot establish complete bounded ingestion. ### Recommended operator flags @@ -312,9 +318,10 @@ spec: ``` A real source-to-sink pipeline should omit `registrationOnly` and supply -stream-capable components. For now, bounded sources are recognized by -`END_LEDGER` in the source's `env` (including when a component reference -translates `config.end_ledger` into that environment variable). +stream-capable components. Bounded sources must declare +`FLOWCTL_SOURCE_BOUNDED: "true"` in each finite source's `env`, even when a +component reference translates `config.end_ledger` to `END_LEDGER`. Never set +this for an unbounded source (including `END_LEDGER: "0"`). For more examples, see: - `examples/quickstart/testnet-duckdb-pipeline.yaml` - Starter DuckDB pipeline diff --git a/docs/bounded-drain-remediation.md b/docs/bounded-drain-remediation.md new file mode 100644 index 0000000..650999c --- /dev/null +++ b/docs/bounded-drain-remediation.md @@ -0,0 +1,120 @@ +# Bounded drain: flowctl runner remediation + +**Priority:** P0 correctness; `RUN_STATUS_COMPLETED` is not proof of durable +sink handling with the current SDK. + +**Affected release:** v0.0.11 with sinks using `flowctl-sdk` v0.1.4 + +**Implementation status:** strict per-stream acknowledgements, explicit source +boundedness, and SDK v0.1.5 process-driver drain/failure/stop tests are in this +checkout. Release is **blocked** pending rebuilt real sink binaries and the +120-ledger DuckLake telemetry gate below; no new flowctl tag is justified yet. +The process fixture is a serial 10 ms SDK handler, not a DuckLake commit test. +Run metric redesign is still pending; `events_processed` is not a commit count. + +**Depends on:** [SDK consumer/processor remediation](https://github.com/withObsrvr/flowctl-sdk/blob/main/docs/bounded-drain-remediation.md) and upgraded/rebuilt component binaries. + +## Reproduction and scope + +See `obsrvr-stellar-components/docs/flowctl-bounded-drain-handoff.md` for the +120-ledger pubnet archive run at 62,080,000–62,080,119. The source completed, +flowctl recorded `RUN_STATUS_COMPLETED` and shut components down, but DuckLake +had committed only 32 contiguous watermarks; 88 events were outstanding. The +script correctly failed its external acknowledgement gate. The timed 30/60s +archive probes in that repository are **not** evidence of bounded completion, +sustained throughput, or RPC/tip performance. Treat local `/tmp/` logs as +non-durable evidence. + +`internal/runner/stream_orchestrator.go` does wait for forwarding goroutines and +calls each sink's `CloseAndRecv` on upstream EOF. However, `fanOut` currently +accepts a response with `EventsFailed == 0` and no `Errors` as success even +when `EventsConsumed` is too small or the response is nil. `CloseAndRecv` +reflects what the sink **claims**, not what a gRPC `Send` has durably stored. +The v0.1.4 consumer claims accepted/queued events and responds before its +handlers finish. `internal/runner/pipeline_runner.go` therefore selects a nil +terminal error, marks `COMPLETED`, and stops the processes, canceling pending +sink callbacks. Its `eventsCount` is not incremented by delivery, explaining +`events_processed: 0`; it is not a commit count. + +## Required flowctl contract + +- In each sink stream count only events actually routed to that sink and + successfully passed to its `Send` (after `inputs`/event-type filtering). + Count *per stream*, not by source or across all fan-out sinks. This is a + transport-accepted count, **not** a durable-write count. +- After upstream EOF, half-close every sink stream and wait for its final + `CloseAndRecv` before declaring delivery complete. Require a **non-nil** + response, zero `EventsFailed`, empty `Errors`, and `EventsConsumed` equal to + that stream's sent count. A short count, excess count, failed/unknown status, + gRPC error, or cancellation before all acks is not `COMPLETED`. Include sink + ID, expected and acknowledged counts, and operation in the failure context + without logging payloads or credentials. A sink receiving zero matching + events must acknowledge zero. +- Require the SDK contract that `EventsConsumed` counts **finished successful + handlers**; until upgraded SDK-based sinks satisfy it, count comparison + alone cannot fix the false-success race. A durable sink must return from its + handler only after commit. No retries/drop-to-survive logic in flowctl. +- The terminal result must reach `PipelineRunner` **before** `COMPLETED` is + persisted or components are stopped. Errors mark `FAILED` and exit nonzero; + intentional stop while work is in flight is `STOPPED` (unless a real error + was already captured), never `COMPLETED`. Wait for all forwarding loops to + exit on either path. Preserve the first error when stop and delivery failure + race. +- Do not label gRPC sends as "processed" or "committed". Decide explicit + metrics for source-emitted, processor-output, sink transport-accepted, sink + handler-completed, and application-durable watermarks. Do not aggregate a + fan-out sink count as distinct ledgers. The authoritative durable range + remains the sink/catalog watermark proof, not the runner's current + `events_processed` value. + +## Work and tests + +1. **P0 — Fail closed on final sink responses.** Update `fanOut` and its tests + in `internal/runner/stream_orchestrator.go` and + `internal/runner/stream_orchestrator_test.go`. Existing `fakeSink` success + cases return `(nil, nil)` or an empty response and must be changed to + acknowledge the actual matching sends. Add missing response, short count, + excess count, failed count, gRPC error, zero-event, filtered-event, + multi-source, and fan-out cases; ensure one failed sink fails the run. +2. **P0 — Test the *real SDK* drain semantics across process boundaries.** The + existing `test/testdata/delivery_component.go` fixture replies to EOF with + an empty `ConsumeResponse` and does not simulate a slow callback. Replace + or extend it with an upgraded SDK-backed consumer, a bounded source of 120 + numbered events, and a 10 ms/event serial durable stand-in. Assert no + `COMPLETED` status or component shutdown before all 120 are acknowledged; + verify contiguous, distinct IDs and absence of leaked workers. Inject a + handler failure at N and cancellation during drain: expect `FAILED` and + `STOPPED` respectively, never a success claim for unfinished work. Check + that CLI exit status and stored run status/error match the outcome. +3. **P1 — Clarify boundedness and metrics.** `WireAll` currently treats any + nonempty `END_LEDGER` as bounded, including `"0"`, which the archive probe + uses for unbounded streaming. Require a documented, explicit source + boundedness signal rather than inferring it from the mere presence of an + endpoint value; test ordinary bounded EOF, unbounded unexpected EOF, and + the `END_LEDGER=0` case. Define and implement useful run metrics without + conflating transport acceptance with durable commit. +4. **P1 — Update user-facing docs.** Until fixes are released, qualify the + completion promise in `README.md` and `docs/modernization-plan.md`: v0.0.11 + plus v0.1.4 cannot establish complete bounded ingestion. Document the new + acknowledgement contract and telemetry semantics after they are verified. + +## Release gate and sequencing + +1. Release the SDK fix, then update `obsrvr-stellar-components`' `go.mod` and + `vendor/` (currently v0.1.4) and rebuild the actual processor and sink. + Do not test the new flowctl against stale vendored binaries. +2. Add flowctl's strict ack checks and process-driver tests; run + `go test ./...`, `go vet ./...`, and `go build ./...`. +3. Run `obsrvr-stellar-components/scripts/ducklake-telemetry-gate.sh` with the + documented `TELEMETRY_GATE_START_LEDGER=62080000`, + `TELEMETRY_GATE_END_LEDGER=62080119`, and `TELEMETRY_GATE_TIMEOUT=300` + using the rebuilt binaries. Choose a dedicated + `TELEMETRY_GATE_RUNTIME_DIR` **under `/tmp`** as required by the script. Record a successful bounded process exit, + sink ack count **120**, and offline `ingest_watermarks` min/max/count/ + distinct/gap proof of all 120 ledgers *before* releasing a new flowctl tag. + Retain pipeline, server and sink logs/catalog checks outside `/tmp` for + review. Recheck cancellation/failure paths, not only the happy path. + +A source-only archive benchmark (~12 ledgers/s on this workstation) is not a +substitute for this E2E gate. Longer sustained catch-up/tip measurements remain +separate operational work after correctness is proven. diff --git a/docs/modernization-plan.md b/docs/modernization-plan.md index 2eca7fd..414944d 100644 --- a/docs/modernization-plan.md +++ b/docs/modernization-plan.md @@ -7,8 +7,9 @@ ## Implementation notes (current checkout) The runner now waits for supervised source/processor/sink edges and returns -transport failures as CLI errors. Bounded completion is inferred from the -source's `END_LEDGER` environment setting; an unbounded source EOF is an error. +transport failures as CLI errors. Finite sources must set +`FLOWCTL_SOURCE_BOUNDED: "true"` in their environment; `END_LEDGER` alone is +not a boundedness signal (`0` can mean unbounded). Unbounded EOF is an error. The first failing edge cancels all streams. Fan-out sends synchronously, so a slow sink applies backpressure instead of dropping events. The run is marked failed on delivery error, completed on bounded completion, and stopped on @@ -39,10 +40,12 @@ remains a live signal, not the authoritative historical failure reason. ## Remaining limitation -**Boundedness is inferred from `END_LEDGER`.** Sources with other finite -termination criteria are currently classified as unbounded and their normal -EOF fails the run. Introduce an explicit bounded-source setting before -supporting other finite source types. +**Bounded completion requires a trustworthy sink.** Each sink stream must +acknowledge exactly its successful sends, with zero failures/errors. SDK v0.1.5 +counts finished successful handlers; handlers must return only after durable +commit. v0.0.11 with SDK v0.1.4 could report success before commit. The runner's +`events_processed` is not a commit metric (currently zero); fan-out acks are +per stream, not distinct ledgers. Use catalog watermarks for durable proof. ## Objective diff --git a/go.mod b/go.mod index c7cff1b..bba5610 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ require ( github.com/withObsrvr/flow-proto v0.1.3 go.etcd.io/bbolt v1.4.0 go.uber.org/zap v1.27.0 + golang.org/x/sync v0.19.0 google.golang.org/grpc v1.79.2 google.golang.org/protobuf v1.36.11 gopkg.in/yaml.v3 v3.0.1 @@ -33,6 +34,7 @@ require ( github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/stargz-snapshotter/estargz v0.16.3 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/distribution/reference v0.6.0 // indirect github.com/docker/cli v28.2.2+incompatible // indirect github.com/docker/distribution v2.8.3+incompatible // indirect @@ -62,6 +64,7 @@ require ( github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/sagikazarmark/locafero v0.7.0 // indirect github.com/sirupsen/logrus v1.9.3 // indirect @@ -71,6 +74,7 @@ require ( github.com/spf13/pflag v1.0.9 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/vbatts/tar-split v0.12.1 // indirect + github.com/withObsrvr/flowctl-sdk v0.1.5 // indirect github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect @@ -82,8 +86,8 @@ require ( go.opentelemetry.io/otel/trace v1.42.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.uber.org/multierr v1.11.0 // indirect + golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect golang.org/x/net v0.50.0 // indirect - golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.41.0 // indirect golang.org/x/text v0.34.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect diff --git a/go.sum b/go.sum index 9d26037..2f5e4ed 100644 --- a/go.sum +++ b/go.sum @@ -30,8 +30,9 @@ github.com/containerd/stargz-snapshotter/estargz v0.16.3 h1:7evrXtoh1mSbGj/pfRcc github.com/containerd/stargz-snapshotter/estargz v0.16.3/go.mod h1:uyr4BfYfOj3G9WBVE8cOlQmXAbPN9VEQpBBeJIuOipU= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= github.com/docker/cli v28.2.2+incompatible h1:qzx5BNUDFqlvyq4AHzdNB7gSyVTmU4cgsyN9SdInc1A= @@ -122,8 +123,9 @@ github.com/pelletier/go-toml/v2 v2.2.3 h1:YmeHyLY8mFWbdkNWwpr+qIL2bEqT0o95WSdkNH github.com/pelletier/go-toml/v2 v2.2.3/go.mod h1:MfCQTFTvCcUyyvvwm1+G6H/jORL20Xlb6rzQu9GuUkc= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ= github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88= @@ -154,9 +156,10 @@ github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8 github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/vbatts/tar-split v0.12.1 h1:CqKoORW7BUWBe7UL/iqTVvkTBOF8UvOMKOIZykxnnbo= github.com/vbatts/tar-split v0.12.1/go.mod h1:eF6B6i6ftWQcDqEn3/iGFRFRo8cBIMSJVOpnNdfTMFA= -github.com/withObsrvr/flow-proto v0.1.0/go.mod h1:RtbIj03ntZdVxGit+48SNPE0/K15AFAtaNtI1JF68Yo= github.com/withObsrvr/flow-proto v0.1.3 h1:L6uCl7HgnHS7g81TX2faQnYLjKJAXRjgudI0elCxW6M= github.com/withObsrvr/flow-proto v0.1.3/go.mod h1:gCX0x2CG0FR7Ndg6yPzeQjK1qRIr2K5zgw4d+uFHeNw= +github.com/withObsrvr/flowctl-sdk v0.1.5 h1:laJmnPdzHEH8m5y2/RCxsD4Mfksvx9uAFHWJlY/Fxl4= +github.com/withObsrvr/flowctl-sdk v0.1.5/go.mod h1:U8LHvFUaTTPLFHnBHBLTd+xGmQaiMkRLHzgKpnxcpro= github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e h1:JVG44RsyaB9T2KIHavMF/ppJZNG9ZpyihvCd0w101no= github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e/go.mod h1:RbqR21r5mrJuqunuUZ/Dhy/avygyECGrLceyNeo4LiM= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -193,8 +196,8 @@ go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561 h1:MDc5xs78ZrZr3HMQugiXOAkSZtfTpbJLDr/lwfgO53E= -golang.org/x/exp v0.0.0-20220909182711-5c715a9e8561/go.mod h1:cyybsKvd6eL0RnXn6p/Grxp8F5bW7iYuBgsNCOHpMYE= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d h1:jtJma62tbqLibJ5sFQz8bKtEM8rJBtfilJ2qTU199MI= +golang.org/x/exp v0.0.0-20231006140011-7918f672742d/go.mod h1:ldy0pHrwJyGW56pPQzzkH36rKxoZW1tw7ZJpeKx+hdo= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= diff --git a/internal/runner/delivery_shutdown_test.go b/internal/runner/delivery_shutdown_test.go index 3181edf..039d1ab 100644 --- a/internal/runner/delivery_shutdown_test.go +++ b/internal/runner/delivery_shutdown_test.go @@ -11,24 +11,24 @@ func TestWaitForTerminalOnStop(t *testing.T) { terminal := make(chan error, 1) terminal <- failure cancelled := false - err, reaped := waitForTerminalOnStop(terminal, func() { cancelled = true }) - if !reaped || !errors.Is(err, failure) || cancelled { - t.Fatalf("err=%v reaped=%t cancelled=%t", err, reaped, cancelled) + err := waitForTerminalOnStop(terminal, func() { cancelled = true }) + if !errors.Is(err, failure) || cancelled { + t.Fatalf("err=%v cancelled=%t", err, cancelled) } }) t.Run("failure captured during cancellation", func(t *testing.T) { terminal := make(chan error, 1) cancelled := false - err, reaped := waitForTerminalOnStop(terminal, func() { cancelled = true; terminal <- failure }) - if !reaped || !errors.Is(err, failure) || !cancelled { - t.Fatalf("err=%v reaped=%t cancelled=%t", err, reaped, cancelled) + err := waitForTerminalOnStop(terminal, func() { cancelled = true; terminal <- failure }) + if !errors.Is(err, failure) || !cancelled { + t.Fatalf("err=%v cancelled=%t", err, cancelled) } }) t.Run("operator cancellation", func(t *testing.T) { terminal := make(chan error, 1) - err, reaped := waitForTerminalOnStop(terminal, func() { terminal <- nil }) - if !reaped || err != nil { - t.Fatalf("err=%v reaped=%t", err, reaped) + err := waitForTerminalOnStop(terminal, func() { terminal <- nil }) + if err != nil { + t.Fatal(err) } }) } diff --git a/internal/runner/pipeline_runner.go b/internal/runner/pipeline_runner.go index 3cde007..79afde3 100644 --- a/internal/runner/pipeline_runner.go +++ b/internal/runner/pipeline_runner.go @@ -321,11 +321,7 @@ func (r *PipelineRunner) Run() error { // Reap delivery before stopping components. Prefer an already-reported // failure over a simultaneous interrupt, but don't fail on cancellation. if shutdownStatus == flowctlpb.RunStatus_RUN_STATUS_STOPPED && !registrationOnly { - var reaped bool - deliveryErr, reaped = waitForTerminalOnStop(terminal, streamOrch.cancel) - if !reaped { - logger.Warn("Timed out waiting for forwarding loops to exit") - } + deliveryErr = waitForTerminalOnStop(terminal, streamOrch.cancel) if deliveryErr != nil { shutdownStatus = flowctlpb.RunStatus_RUN_STATUS_FAILED } @@ -346,19 +342,14 @@ func (r *PipelineRunner) Run() error { // waitForTerminalOnStop preserves a failure already reported or captured by // the errgroup while cancellation is in progress. A cancelled edge returns nil. -func waitForTerminalOnStop(terminal <-chan error, cancel func()) (error, bool) { +func waitForTerminalOnStop(terminal <-chan error, cancel func()) error { select { case err := <-terminal: - return err, true + return err default: } cancel() - select { - case err := <-terminal: - return err, true - case <-time.After(10 * time.Second): - return nil, false - } + return <-terminal // never stop components with delivery workers still running } // markDeliveryFailure records which registered component failed. Store only @@ -369,7 +360,7 @@ func (r *PipelineRunner) markDeliveryFailure(err error) { return } id := edge.From - if edge.Operation == "send" || edge.Operation == "close" || edge.Operation == "connect" || edge.Operation == "registration" || edge.Operation == "open" { + if edge.Operation == "send" || (edge.Operation == "close" || edge.Operation == "close acknowledgement") || edge.Operation == "connect" || edge.Operation == "registration" || edge.Operation == "open" { id = edge.To } if id == "sinks" { @@ -1071,7 +1062,7 @@ func deliveryFailureRecord(err error) string { cause = "EOF" } fields := map[string]string{"component": edge.From, "from": edge.From, "to": edge.To, "edge": edge.Kind, "operation": edge.Operation, "cause": cause} - if edge.Operation == "send" || edge.Operation == "close" || edge.Operation == "open" || edge.Operation == "connect" || edge.Operation == "registration" { + if edge.Operation == "send" || (edge.Operation == "close" || edge.Operation == "close acknowledgement") || edge.Operation == "open" || edge.Operation == "connect" || edge.Operation == "registration" { fields["component"] = edge.To } encoded, _ := json.Marshal(fields) diff --git a/internal/runner/stream_orchestrator.go b/internal/runner/stream_orchestrator.go index 5907191..a2d3e41 100644 --- a/internal/runner/stream_orchestrator.go +++ b/internal/runner/stream_orchestrator.go @@ -166,14 +166,14 @@ func (s *StreamOrchestrator) WireAll() error { s.cancel() return fmt.Errorf("pipeline %s source %s has no sink declaring it in inputs", s.pipeline.Metadata.Name, source.ID) } - bounded := params["end_ledger"] != "" + bounded := sourceBounded(source.Env) s.launch(func() error { return s.fanOut(sourceReceiver{sourceStream, s.ctx, bounded}, source.ID, sinks, "source-sink") }) continue } first := s.pipeline.Spec.Processors[0].ID - bounded := params["end_ledger"] != "" + bounded := sourceBounded(source.Env) s.launch(func() error { return s.forwardSource(sourceStream, streams[0], source.ID, first, bounded) }) for i := 0; i < len(streams)-1; i++ { from, to := s.pipeline.Spec.Processors[i].ID, s.pipeline.Spec.Processors[i+1].ID @@ -185,6 +185,12 @@ func (s *StreamOrchestrator) WireAll() error { return nil } +// Boundedness is declared, not inferred from component-specific endpoints. +// END_LEDGER=0 commonly means unbounded. +func sourceBounded(env map[string]string) bool { + return env["FLOWCTL_SOURCE_BOUNDED"] == "true" +} + func hasInput(inputs []string, id string) bool { for _, input := range inputs { if input == id { @@ -281,16 +287,20 @@ func (s *StreamOrchestrator) fanOut(from interface { if len(edgeKind) > 0 { kind = edgeKind[0] } + sent := make([]int64, len(sinks)) for { event, err := from.Recv() if err == io.EOF { - for _, sink := range sinks { + for i, sink := range sinks { response, err := sink.stream.CloseAndRecv() if err != nil { - return s.edge(fromID, sink.id, kind, "close", nil, err) + return s.edge(fromID, sink.id, kind, "close acknowledgement", nil, fmt.Errorf("expected %d: %w", sent[i], err)) + } + if response == nil { + return s.edge(fromID, sink.id, kind, "close acknowledgement", nil, fmt.Errorf("expected %d: missing response", sent[i])) } - if response.GetEventsFailed() > 0 || len(response.GetErrors()) > 0 { - return s.edge(fromID, sink.id, kind, "close", nil, fmt.Errorf("sink reported %d failed events (%d errors)", response.GetEventsFailed(), len(response.GetErrors()))) + if response.GetEventsFailed() != 0 || len(response.GetErrors()) != 0 || response.GetEventsConsumed() != sent[i] { + return s.edge(fromID, sink.id, kind, "close acknowledgement", nil, fmt.Errorf("expected %d acknowledged %d, failed %d, errors %d", sent[i], response.GetEventsConsumed(), response.GetEventsFailed(), len(response.GetErrors()))) } } return nil @@ -298,7 +308,7 @@ func (s *StreamOrchestrator) fanOut(from interface { if err != nil { return s.edge(fromID, "sinks", kind, "recv", nil, err) } - for _, sink := range sinks { + for i, sink := range sinks { if accepts(sink.accepted, event.Type) { if err := sink.stream.Send(event); err != nil { // gRPC may report EOF on Send while the actual server error is @@ -310,6 +320,7 @@ func (s *StreamOrchestrator) fanOut(from interface { } return s.edge(fromID, sink.id, kind, "send", event, err) } + sent[i]++ } } } diff --git a/internal/runner/stream_orchestrator_test.go b/internal/runner/stream_orchestrator_test.go index 4755aef..4120187 100644 --- a/internal/runner/stream_orchestrator_test.go +++ b/internal/runner/stream_orchestrator_test.go @@ -38,6 +38,7 @@ type fakeSink struct { sent int sendErr, closeErr error response *flowctlv1.ConsumeResponse + nilResponse bool blocked, release chan struct{} } @@ -49,7 +50,15 @@ func (f *fakeSink) Send(*flowctlv1.Event) error { f.sent++ return f.sendErr } -func (f *fakeSink) CloseAndRecv() (*flowctlv1.ConsumeResponse, error) { return f.response, f.closeErr } +func (f *fakeSink) CloseAndRecv() (*flowctlv1.ConsumeResponse, error) { + if f.nilResponse { + return nil, f.closeErr + } + if f.response != nil { + return f.response, f.closeErr + } + return &flowctlv1.ConsumeResponse{EventsConsumed: int64(f.sent)}, f.closeErr +} func testOrchestrator(t *testing.T) *StreamOrchestrator { t.Helper() @@ -117,7 +126,7 @@ func TestFanOutFailuresAndCompletion(t *testing.T) { {name: "sink send", sendErr: failure, want: "send event 1"}, {name: "sink close", closeErr: failure, want: "close"}, {name: "success"}, - {name: "sink reports failed event", want: "reported 1 failed events"}, + {name: "sink reports failed event", want: "failed 1"}, } { t.Run(tc.name, func(t *testing.T) { s := testOrchestrator(t) @@ -140,6 +149,53 @@ func TestFanOutFailuresAndCompletion(t *testing.T) { }) } } +func TestFanOutAcknowledgements(t *testing.T) { + for _, tc := range []struct { + name string + response *flowctlv1.ConsumeResponse + missing bool + want string + }{ + {"short", &flowctlv1.ConsumeResponse{}, false, "expected 1 acknowledged 0"}, + {"excess", &flowctlv1.ConsumeResponse{EventsConsumed: 2}, false, "expected 1 acknowledged 2"}, + {"failed", &flowctlv1.ConsumeResponse{EventsConsumed: 1, EventsFailed: 1}, false, "failed 1"}, + {"errors", &flowctlv1.ConsumeResponse{EventsConsumed: 1, Errors: []string{"failure"}}, false, "errors 1"}, + {"nil", nil, true, "missing response"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := testOrchestrator(t) + sink := &fakeSink{response: tc.response, nilResponse: tc.missing} + err := s.fanOut(&fakeProcessor{events: []*flowctlv1.Event{{Type: "matching"}}}, "source", []sinkStream{{id: "bad-sink", stream: sink}}) + if err == nil || !strings.Contains(err.Error(), tc.want) || !strings.Contains(err.Error(), "bad-sink") { + t.Fatalf("err=%v", err) + } + }) + } + for _, events := range [][]*flowctlv1.Event{nil, {{Type: "ignored"}}, {{Type: "matching"}, {Type: "ignored"}}} { + s := testOrchestrator(t) + a, b := &fakeSink{}, &fakeSink{} + err := s.fanOut(&fakeProcessor{events: events}, "source", []sinkStream{{id: "a", accepted: []string{"matching"}, stream: a}, {id: "b", accepted: []string{"ignored"}, stream: b}}) + if err != nil { + t.Fatal(err) + } + } +} + +func TestSourceBoundedDeclaration(t *testing.T) { + for _, tc := range []struct { + env map[string]string + want bool + }{ + {map[string]string{"END_LEDGER": "0"}, false}, + {map[string]string{"END_LEDGER": "120"}, false}, + {map[string]string{"FLOWCTL_SOURCE_BOUNDED": "true"}, true}, + } { + if got := sourceBounded(tc.env); got != tc.want { + t.Fatalf("%v: %v", tc.env, got) + } + } +} + func TestFanOutBackpressureDoesNotDropEvent(t *testing.T) { s := testOrchestrator(t) slow := &fakeSink{blocked: make(chan struct{}), release: make(chan struct{})} diff --git a/test/delivery_process_test.go b/test/delivery_process_test.go index 74d85cf..4be7e54 100644 --- a/test/delivery_process_test.go +++ b/test/delivery_process_test.go @@ -34,6 +34,9 @@ func TestProcessDeliveryTerminalResult(t *testing.T) { {"fail-before", "", "", false, true, false, "synthetic source failure", "source -> processor"}, {"fail-after", "", "", false, true, false, "synthetic source failure", "source -> processor"}, {"bounded", "", "", true, false, false, "RUN_STATUS_COMPLETED", ""}, + {"drain", "", "drain", true, false, true, "RUN_STATUS_COMPLETED", ""}, + {"drain", "", "drain-fail", true, true, true, "failed 1", "source -> sink"}, + {"drain-stop", "", "drain", true, false, true, "RUN_STATUS_STOPPED", ""}, {"unbounded-eof", "", "", false, true, false, "unexpected EOF", "source -> processor"}, {"bounded", "processor-fail", "", true, true, false, "synthetic processor failure", "processor -> sinks"}, {"bounded", "", "sink-fail", true, true, false, "synthetic sink failure", "processor -> sink"}, @@ -47,7 +50,7 @@ func TestProcessDeliveryTerminalResult(t *testing.T) { t.Run(fmt.Sprintf("%s-%s-%s-direct=%t", tc.mode, tc.processorMode, tc.sinkMode, tc.direct), func(t *testing.T) { sourceEnv := fmt.Sprintf(" DELIVERY_MODE: %s\n", tc.mode) if tc.bounded { - sourceEnv += " END_LEDGER: \"2\"\n" + sourceEnv += " END_LEDGER: \"2\"\n FLOWCTL_SOURCE_BOUNDED: \"true\"\n" } processorSection := fmt.Sprintf(` processors: - id: processor @@ -89,16 +92,31 @@ spec: ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second) defer cancel() cmd := exec.CommandContext(ctx, flowctl, "run", "--show-status=false", "--db-path", db, "--control-plane-port=0", "--log-dir", filepath.Join(dir, tc.mode+"-logs"), file) - cmd.Dir = root + cmd.Dir = filepath.Join(dir, strings.ReplaceAll(t.Name(), "/", "-")) + if err := os.MkdirAll(cmd.Dir, 0700); err != nil { + t.Fatal(err) + } var out []byte var err error - if tc.mode == "wait" { + if tc.mode == "wait" || tc.mode == "drain-stop" { var buf bytes.Buffer cmd.Stdout, cmd.Stderr = &buf, &buf if err = cmd.Start(); err != nil { t.Fatal(err) } - time.Sleep(2 * time.Second) + if tc.mode == "drain-stop" { + logPath := filepath.Join(cmd.Dir, "logs", "sink.log") + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + contents, _ := os.ReadFile(logPath) + if strings.Contains(string(contents), "DELIVERY_RECEIVED id=") { + break + } + time.Sleep(10 * time.Millisecond) + } + } else { + time.Sleep(2 * time.Second) + } if err = cmd.Process.Signal(os.Interrupt); err != nil { t.Fatal(err) } @@ -122,6 +140,32 @@ spec: if tc.failed && !strings.Contains(string(out), tc.edge) { t.Fatalf("missing failing edge:\n%s", out) } + if tc.mode == "drain-stop" { + contents, readErr := os.ReadFile(filepath.Join(cmd.Dir, "logs", "sink.log")) + if readErr != nil || strings.Contains(string(contents), "DELIVERY_RECEIVED id=119") { + t.Fatalf("drain did not stop in flight: %v %s", readErr, contents) + } + } + if tc.mode == "drain" && !tc.failed { + contents, readErr := os.ReadFile(filepath.Join(cmd.Dir, "logs", "sink.log")) + if readErr != nil { + t.Fatal(readErr) + } + seen := make(map[string]bool) + for _, line := range strings.Split(string(contents), "\n") { + if i := strings.Index(line, "DELIVERY_RECEIVED id="); i >= 0 { + seen[strings.TrimSpace(line[i+len("DELIVERY_RECEIVED id="):])] = true + } + } + if len(seen) != 120 { + t.Fatalf("sink did not finish all 120: %d unique\n%s", len(seen), contents) + } + for i := 0; i < 120; i++ { + if !seen[fmt.Sprint(i)] { + t.Fatalf("missing %d", i) + } + } + } if tc.failed && !strings.Contains(string(out), "RUN_STATUS_FAILED") { t.Fatalf("run was not marked failed:\n%s", out) } @@ -134,6 +178,9 @@ spec: if err != nil || len(runs) != 1 { t.Fatalf("stored runs=%d err=%v", len(runs), err) } + if (tc.failed && runs[0].Run.Status != flowctlpb.RunStatus_RUN_STATUS_FAILED) || (tc.mode == "drain-stop" && runs[0].Run.Status != flowctlpb.RunStatus_RUN_STATUS_STOPPED) || (!tc.failed && tc.mode != "wait" && tc.mode != "drain-stop" && runs[0].Run.Status != flowctlpb.RunStatus_RUN_STATUS_COMPLETED) { + t.Fatalf("wrong stored status: %s", runs[0].Run.Status) + } if tc.failed { if (!strings.Contains(runs[0].Run.Error, `"from":"source"`) && !strings.Contains(runs[0].Run.Error, `"from":"processor"`)) || !strings.Contains(runs[0].Run.Error, `"cause":`) || strings.Contains(runs[0].Run.Error, "synthetic source failure") { t.Fatalf("missing or unsafe persisted structured failure: %s", runs[0].Run.Error) diff --git a/test/direct_sink_routing_test.go b/test/direct_sink_routing_test.go index 7dbc775..6e59acd 100644 --- a/test/direct_sink_routing_test.go +++ b/test/direct_sink_routing_test.go @@ -36,12 +36,14 @@ spec: DELIVERY_ROLE: source DELIVERY_MODE: bounded END_LEDGER: "2" + FLOWCTL_SOURCE_BOUNDED: "true" - id: source-b command: [%q] env: DELIVERY_ROLE: source DELIVERY_MODE: bounded END_LEDGER: "2" + FLOWCTL_SOURCE_BOUNDED: "true" sinks: - id: sink-a command: [%q] diff --git a/test/testdata/delivery_component.go b/test/testdata/delivery_component.go index 0b5c490..cdf39c7 100644 --- a/test/testdata/delivery_component.go +++ b/test/testdata/delivery_component.go @@ -3,15 +3,18 @@ package main import ( "context" + "fmt" "io" "log" "net" "os" "os/signal" + "strconv" "syscall" "time" flowctlv1 "github.com/withObsrvr/flow-proto/go/gen/flowctl/v1" + "github.com/withObsrvr/flowctl-sdk/pkg/consumer" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" @@ -29,6 +32,14 @@ type sinkService struct { } func (sourceService) StreamEvents(_ *flowctlv1.StreamRequest, stream grpc.ServerStreamingServer[flowctlv1.Event]) error { + if os.Getenv("DELIVERY_MODE") == "drain" || os.Getenv("DELIVERY_MODE") == "drain-stop" { + for i := 0; i < 120; i++ { + if err := stream.Send(&flowctlv1.Event{Id: strconv.Itoa(i), Type: "test.event"}); err != nil { + return err + } + } + return nil + } if os.Getenv("DELIVERY_MODE") != "fail-before" { if err := stream.Send(&flowctlv1.Event{Id: os.Getenv("FLOWCTL_COMPONENT_ID"), Type: "test.event"}); err != nil { return err @@ -65,13 +76,16 @@ func (sinkService) Consume(stream grpc.ClientStreamingServer[flowctlv1.Event, fl if os.Getenv("DELIVERY_MODE") == "sink-fail" { return status.Error(codes.Internal, "synthetic sink failure") } + var count int64 for { event, err := stream.Recv() if err == nil { + count++ log.Printf("DELIVERY_RECEIVED id=%s", event.Id) } if err == io.EOF { - return stream.SendAndClose(&flowctlv1.ConsumeResponse{}) + log.Print(fmt.Sprintf("DELIVERY_ACK count=%d", count)) + return stream.SendAndClose(&flowctlv1.ConsumeResponse{EventsConsumed: count}) } if err != nil { return err @@ -97,7 +111,22 @@ func main() { flowctlv1.RegisterProcessorServiceServer(server, processorService{}) case "sink": kind = flowctlv1.ComponentType_COMPONENT_TYPE_CONSUMER - flowctlv1.RegisterConsumerServiceServer(server, sinkService{}) + if os.Getenv("DELIVERY_MODE") == "drain" || os.Getenv("DELIVERY_MODE") == "drain-fail" { + flowctlv1.RegisterConsumerServiceServer(server, consumer.NewConsumerService(consumer.DefaultConfig(), id, []string{"test.event"}, func(ctx context.Context, event *flowctlv1.Event) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(10 * time.Millisecond): + } + if os.Getenv("DELIVERY_MODE") == "drain-fail" && event.Id == "60" { + return fmt.Errorf("synthetic handler failure") + } + log.Printf("DELIVERY_RECEIVED id=%s", event.Id) + return nil + })) + } else { + flowctlv1.RegisterConsumerServiceServer(server, sinkService{}) + } default: log.Fatal("unknown fixture role") }