Bugfix: Fix bounded range error

README.md

@@ -106,13 +106,19 @@ flowctl pipelines stop <run-id>
 ```
 
 `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

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.

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
 

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

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=

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)
 		}
 	})
 }