From ec659a959c7a39565ebd140f6ccdc1c8517ce1b9 Mon Sep 17 00:00:00 2001 From: Devaris Date: Sat, 4 Jul 2026 16:20:43 -0700 Subject: [PATCH] =?UTF-8?q?docs(adr):=20decide=20pipeline=20architecture?= =?UTF-8?q?=20v2=20=E2=80=94=20adopt=20as=20target,=20gate=20graduation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resolve the open question of what to do with the v2 pipeline architecture (pkg/lifecycle-poc), which shipped as an undocumented, incomplete, unbenchmarked preview that taxes every data-path change. Benchmarked both architectures on the same 100k-record workload (committed: BenchmarkStreamOld vs the existing BenchmarkStreamNew). v2 allocates ~6.3× fewer objects and ~3.3× less memory per record — a real, batching-driven data-path efficiency win. So: adopt v2 as the target, keep it behind the preview flag with v1 default until explicit graduation criteria are met (multi-destination, recovery parity, chaos-tested correctness, an end-to-end benchi confirmation, Tier-1 sign-off), and do not delete it. Documents the decision and its finish line so the fork stops being a silent tax on future PRs. Adds the v1 counterpart benchmark for reproducibility of the claim. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../20260704-pipeline-architecture-v2.md | 117 ++++++++++++++++++ pkg/lifecycle/stream/benchmark_test.go | 101 +++++++++++++++ 2 files changed, 218 insertions(+) create mode 100644 docs/architecture-decision-records/20260704-pipeline-architecture-v2.md create mode 100644 pkg/lifecycle/stream/benchmark_test.go diff --git a/docs/architecture-decision-records/20260704-pipeline-architecture-v2.md b/docs/architecture-decision-records/20260704-pipeline-architecture-v2.md new file mode 100644 index 000000000..4b3f435eb --- /dev/null +++ b/docs/architecture-decision-records/20260704-pipeline-architecture-v2.md @@ -0,0 +1,117 @@ +# Pipeline architecture v2: adopt as target, graduate behind explicit criteria + +## Summary + +Adopt the v2 pipeline architecture (`pkg/lifecycle-poc`, a batch-oriented +"funnel") as the target replacement for v1 (`pkg/lifecycle`, a record-at-a-time +node graph). The decision is driven by a measured **~6.3× reduction in +allocations and ~3.3× reduction in memory per record** in the data path. v2 stays +behind the `-preview.pipeline-arch-v2` flag with v1 as the default until it meets +the graduation criteria below. We do **not** delete v2 (the win is proven) and do +**not** make it the default prematurely (it is incomplete and lacks recovery +parity). + +## Context + +Two pipeline lifecycle implementations exist in the tree: + +- **v1** — `pkg/lifecycle`, the default. A node graph + (`SourceNode → SourceAckerNode → DestinationNode → DestinationAckerNode`) + that moves records one at a time across per-node goroutines and channels, with + per-record acking. +- **v2** — `pkg/lifecycle-poc`, opt-in via `-preview.pipeline-arch-v2`. A + `funnel` of tasks driven by a worker that processes records in batches. + +v2 was introduced as a preview in #1913 with **no ADR, no completion criteria, +and no committed benchmark**. That left it an open question that silently taxes +every data-path change: the metrics fix (#2268), the SIGTERM graceful-shutdown +work, and the force-stop ack-safety follow-up (#2519) each have to be reasoned +about in both implementations. v2 is also **incomplete** — its own flag help +states it "currently supports only 1 source and 1 destination per pipeline." + +To decide its fate on evidence rather than assertion, we benchmarked both +architectures on the same workload. + +### Benchmark + +Committed, reproducible microbenchmarks isolating data-path orchestration +(no-op mocked connectors, no real I/O), each processing exactly 100,000 records +generator → printer: + +- v1: `pkg/lifecycle/stream/benchmark_test.go` `BenchmarkStreamOld` +- v2: `pkg/lifecycle-poc/funnel/funnel_test.go` `BenchmarkStreamNew` + +| Metric (per 100k records) | v1 (record-at-a-time) | v2 (1000-record batches) | v2 advantage | +| --- | --- | --- | --- | +| Allocations | ~6,905,000 | ~1,104,000 | **~6.3× fewer** | +| Memory | ~250 MB | ~76 MB | **~3.3× less** | +| Per record | ~69 allocs / ~2497 B | ~11 allocs / ~764 B | — | + +The driver is batching: v2's source returns 1000-record batches per read and the +worker moves them as a unit, amortizing the per-record channel hops, function +calls, and acking that v1 pays on every single record across four nodes. + +Allocation and GC pressure are primary drivers of sustained throughput and tail +latency under load, so this efficiency gap implies a material throughput +advantage for the data-path orchestration itself. A full end-to-end records/sec +figure via benchi on a reference pipeline is **not yet measured** and is a +graduation criterion below — the win is proven at the allocation level, and the +end-to-end magnitude (which is also bounded by connector I/O in real pipelines) +must be confirmed before v1 is removed. + +## Decision + +1. **Adopt v2 as the target architecture.** v1 is legacy and will be removed once + v2 graduates. The ~6× allocation win makes deleting v2 (discarding a proven + efficiency architecture and the work already done) the wrong call. +2. **v2 remains behind `-preview.pipeline-arch-v2` with v1 as default** until it + meets all graduation criteria: + - Multi-source and multi-destination support (v1 feature parity). + - Error-recovery parity with the behavior in + [20240812-recover-from-pipeline-errors](20240812-recover-from-pipeline-errors.md). + - A chaos-test correctness bar: SIGKILL/SIGTERM mid-batch, mid-checkpoint; + verify data-integrity invariants 1–7 hold on recovery. This seeds + `tests/chaos` (which does not exist yet). + - A committed benchi throughput comparison confirming the win end-to-end on a + reference pipeline. + - Human Tier-1 sign-off. Per the process-maturity table, the chaos suite and a + second maintainer are Phase 2 gates — so realistically v2 graduates in + Phase 2, not Phase 1. +3. **Bound the dual-maintenance tax until graduation.** New data-path behavior is + designed to be portable to both implementations. Where a change can only land + in one, the v2 gap is tracked explicitly (an issue) so graduation is never + blocked by silent drift. We do not add v1-only data-path features that widen + the parity gap without recording them. +4. **Do not rush graduation.** Making an incomplete, unrecovered architecture the + default would trade a data-integrity guarantee for throughput — forbidden + without the design-doc-and-sign-off path. + +## Consequences + +- The fork is now a **documented, criteria-gated decision** instead of an open + question. Every future data-path PR knows the target and the finish line. +- The dual-maintenance tax continues until graduation, but is bounded by the + portability rule and the explicit criteria. Prioritizing v2 completion shortens + the window; letting it drift lengthens it. +- Graduation depends on a second maintainer for the Tier-1 completion and review, + consistent with the process-maturity table. This ADR does not manufacture that + capacity — it commits the direction so the work is ready when the capacity + exists. +- Phase 1 data-path items (the #2519 force-stop follow-up, v0.17 hot-reload) must + be reasoned about in both implementations; this ADR makes that explicit rather + than incidental. +- If a future benchi run fails to confirm a meaningful end-to-end throughput win + (e.g. connector I/O dominates on all realistic reference pipelines), this ADR + is superseded by a new one that reconsiders deletion — the allocation win alone + justifies keeping v2, but not necessarily finishing it if end-to-end shows no + benefit. + +## Related + +- #1913 — Pipeline architecture v2 (preview) +- #2268 — metrics change carried into both implementations +- #2519 — SIGTERM graceful shutdown; force-stop ack-safety follow-up spans both +- [20240812-recover-from-pipeline-errors](20240812-recover-from-pipeline-errors.md) +- Benchmarks: `pkg/lifecycle/stream/benchmark_test.go`, + `pkg/lifecycle-poc/funnel/funnel_test.go` +- Phase 1 execution plan: `docs/design-documents/20260704-phase-1-execution-plan.md` diff --git a/pkg/lifecycle/stream/benchmark_test.go b/pkg/lifecycle/stream/benchmark_test.go new file mode 100644 index 000000000..c8c0b5aea --- /dev/null +++ b/pkg/lifecycle/stream/benchmark_test.go @@ -0,0 +1,101 @@ +// Copyright © 2026 Meroxa, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package stream_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/conduitio/conduit-commons/csync" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/foundation/metrics/noop" + "github.com/conduitio/conduit/pkg/lifecycle/stream" +) + +// BenchmarkStreamOld measures the v1 (record-at-a-time node graph) data path +// moving records generator -> printer, mirroring BenchmarkStreamNew in the v2 +// funnel package (which processes batchCount*batchSize = 100_000 records). Both +// process exactly recordCount records (enforced by the source Ack expectation), +// so allocs/op and B/op are directly comparable across v1 and v2; ns/op is +// dominated by the stop window and is NOT a throughput measure. +func BenchmarkStreamOld(b *testing.B) { + ctx, killAll := context.WithCancel(context.Background()) + defer killAll() + + logger := log.Nop() + + b.ReportAllocs() + b.StopTimer() + for i := 0; i < b.N; i++ { + const recordCount = 100000 + + ctrl := gomockCtrl(logger) + dlqNode := &stream.DLQHandlerNode{ + Name: "dlq", + Handler: noopDLQHandler(ctrl), + WindowSize: 1, + WindowNackThreshold: 0, + } + dlqNode.Add(1) // 1 source + node1 := &stream.SourceNode{ + Name: "generator", + Source: generatorSource(ctrl, logger, "generator", recordCount, 0), + PipelineTimer: noop.Timer{}, + } + node2 := &stream.SourceAckerNode{ + Name: "generator-acker", + Source: node1.Source, + DLQHandlerNode: dlqNode, + } + node3 := &stream.DestinationNode{ + Name: "printer", + Destination: printerDestination(ctrl, logger, "printer"), + ConnectorTimer: noop.Timer{}, + } + node4 := &stream.DestinationAckerNode{ + Name: "printer-acker", + Destination: node3.Destination, + } + stream.SetLogger(node1, logger) + stream.SetLogger(node2, logger) + stream.SetLogger(node3, logger) + stream.SetLogger(node4, logger) + node2.Sub(node1.Pub()) + node3.Sub(node2.Pub()) + node4.Sub(node3.Pub()) + + b.StartTimer() + var wg sync.WaitGroup + wg.Add(5) + go runNode(ctx, &wg, dlqNode) + go runNode(ctx, &wg, node4) + go runNode(ctx, &wg, node3) + go runNode(ctx, &wg, node2) + go runNode(ctx, &wg, node1) + + // Stop the source after all records have streamed through (recordCount + // completes well within the window). Because the source blocks once + // drained, the window only bounds ns/op — it never processes more than + // recordCount, so allocs/op and B/op measure the cost of exactly + // recordCount records and are comparable to BenchmarkStreamNew. + time.AfterFunc(3*time.Second, func() { _ = node1.Stop(ctx, nil) }) + if (*csync.WaitGroup)(&wg).WaitTimeout(ctx, 10*time.Second) != nil { + killAll() + } + b.StopTimer() + } +}