diff --git a/pkg/conduit/runtime.go b/pkg/conduit/runtime.go index 87549b6d3..d04540dcd 100644 --- a/pkg/conduit/runtime.go +++ b/pkg/conduit/runtime.go @@ -120,6 +120,10 @@ type lifecycleService interface { // lifecycle-poc(pkg/lifecycle-poc).Service.StopAndWait for why the // Preview.PipelineArchV2 implementation always refuses. StopAndWait(ctx context.Context, pipelineID string) error + // ReconfigureProcessor keeps this interface a superset of + // provisioning.LifecycleService (see StopAndWait above) — it is used by the + // live in-place apply path. See lifecycle.Service.ReconfigureProcessor. + ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error Init(ctx context.Context) error } diff --git a/pkg/lifecycle-poc/service.go b/pkg/lifecycle-poc/service.go index cf0f646cc..eaa397441 100644 --- a/pkg/lifecycle-poc/service.go +++ b/pkg/lifecycle-poc/service.go @@ -353,6 +353,18 @@ func (s *Service) StopAndWait(context.Context, string) error { return ce } +// ReconfigureProcessor refuses under the experimental Preview.PipelineArchV2 +// lifecycle service, mirroring StopAndWait: live in-place apply is not supported +// under this arch. Returning the same coded error means a live apply is cleanly +// refused rather than silently downgraded. +func (s *Service) ReconfigureProcessor(context.Context, string, string) error { + ce := conduiterr.New(CodeStopAndWaitUnsupported, + "live in-place apply (processor reconfigure on a running pipeline) is not supported under "+ + "the experimental Preview.PipelineArchV2 lifecycle service") + ce.Suggestion = "disable Preview.PipelineArchV2, or stop the pipeline manually before applying changes" + return ce +} + // buildRunnablePipeline will build and connect all tasks configured in the pipeline. func (s *Service) buildRunnablePipeline( ctx context.Context, diff --git a/pkg/lifecycle-poc/stop_and_wait_test.go b/pkg/lifecycle-poc/stop_and_wait_test.go index 348d5b6a1..fb0cde095 100644 --- a/pkg/lifecycle-poc/stop_and_wait_test.go +++ b/pkg/lifecycle-poc/stop_and_wait_test.go @@ -49,3 +49,23 @@ func TestServiceLifecycle_StopAndWait_Unsupported(t *testing.T) { is.Equal(ce.Code.Reason(), CodeStopAndWaitUnsupported.Reason()) is.True(ce.Suggestion != "") } + +// TestServiceLifecycle_ReconfigureProcessor_Unsupported pins the same parity +// refusal for the live in-place hot-reload path (PR1 of §4): this preview +// lifecycle service must refuse ReconfigureProcessor with the same stable code +// rather than silently no-op, so provisioning.applyInPlace never runs an +// in-place swap against the unaudited arch. +func TestServiceLifecycle_ReconfigureProcessor_Unsupported(t *testing.T) { + is := is.New(t) + + logger := log.New(zerolog.Nop()) + ls := NewService(logger, testConnectorService{}, testProcessorService{}, testConnectorPluginService{}, testPipelineService{}, true) + + err := ls.ReconfigureProcessor(context.Background(), uuid.NewString(), uuid.NewString()) + is.True(err != nil) + + ce, ok := conduiterr.Get(err) + is.True(ok) + is.Equal(ce.Code.Reason(), CodeStopAndWaitUnsupported.Reason()) + is.True(ce.Suggestion != "") +} diff --git a/pkg/lifecycle/reconfigure.go b/pkg/lifecycle/reconfigure.go new file mode 100644 index 000000000..ab82a9b1b --- /dev/null +++ b/pkg/lifecycle/reconfigure.go @@ -0,0 +1,82 @@ +// 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 lifecycle + +import ( + "context" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/lifecycle/stream" +) + +// ErrProcessorNotLiveReconfigurable signals that a processor cannot be swapped in +// place on the running pipeline, so the caller must fall back to a restart-class +// apply. It is returned when the processor is not running as a plain, +// single-worker stream.ProcessorNode — e.g. it is parallelized (Workers > 1, and +// so runs as a stream.ParallelNode), or no matching node is present. Restarting +// is always a safe superset: the restart path re-imports the config and rebuilds +// every node, so it applies the change regardless of the running node's shape. +var ErrProcessorNotLiveReconfigurable = cerrors.New("processor is not live-reconfigurable, a restart is required") + +// ReconfigureProcessor swaps a processor's configuration in a running pipeline in +// place — without stopping or restarting the pipeline — by rebuilding the +// runnable processor from its (already-updated) stored instance and swapping it +// into the live stream.ProcessorNode via ProcessorNode.Reconfigure. See the +// design doc "Live in-place hot-reload". +// +// Ordering contract: the caller MUST persist the processor's new config to its +// instance before calling this (ReconfigureProcessor reads the current instance +// from the processor service). The swap happens at a record boundary; open of +// the new processor happens before teardown of the old, so if the new config +// fails to open, the old processor keeps running and the error is returned — the +// pipeline never drops (invariant 3). +// +// Returns ErrProcessorNotLiveReconfigurable if the processor is not a plain +// single-worker node in the running pipeline; the caller should then fall back to +// a restart. Returns an error (old processor kept) if building or opening the new +// processor fails. +func (s *Service) ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error { + rp, ok := s.runningPipelines.Get(pipelineID) + if !ok { + return cerrors.Errorf("pipeline %q is not running, cannot reconfigure processor %q in place", pipelineID, processorID) + } + + // A single-worker processor runs as a *stream.ProcessorNode named for the + // processor ID. A parallel processor (Workers > 1) runs as a + // *stream.ParallelNode instead and is not matched here, so it falls through + // to the restart fallback below — as does a processor with no live node. + var node *stream.ProcessorNode + for _, n := range rp.n { + pn, isProcNode := n.(*stream.ProcessorNode) + if isProcNode && pn.ID() == processorID { + node = pn + break + } + } + if node == nil { + return cerrors.Errorf("%w: processor %q in pipeline %q", ErrProcessorNotLiveReconfigurable, processorID, pipelineID) + } + + instance, err := s.processors.Get(ctx, processorID) + if err != nil { + return cerrors.Errorf("could not fetch processor %q for live reconfigure: %w", processorID, err) + } + runnableProc, err := s.processors.MakeRunnableProcessor(ctx, instance) + if err != nil { + return cerrors.Errorf("could not build runnable processor %q for live reconfigure: %w", processorID, err) + } + + return node.Reconfigure(ctx, runnableProc) +} diff --git a/pkg/lifecycle/reconfigure_test.go b/pkg/lifecycle/reconfigure_test.go new file mode 100644 index 000000000..9317d29d3 --- /dev/null +++ b/pkg/lifecycle/reconfigure_test.go @@ -0,0 +1,69 @@ +// 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 lifecycle + +import ( + "context" + "strings" + "testing" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/lifecycle/stream" + "github.com/conduitio/conduit/pkg/pipeline" + "github.com/matryer/is" +) + +func newReconfigureTestService() *Service { + return NewService( + log.Nop(), + testErrRecoveryCfg(), + testConnectorService{}, + testProcessorService{}, + testConnectorPluginService{}, + testPipelineService{}, + ) +} + +// TestService_ReconfigureProcessor_NotRunning: reconfiguring a processor in a +// pipeline that is not currently running is an error (nothing live to swap). +func TestService_ReconfigureProcessor_NotRunning(t *testing.T) { + is := is.New(t) + s := newReconfigureTestService() + + err := s.ReconfigureProcessor(context.Background(), "does-not-exist", "proc1") + is.True(err != nil) + is.True(strings.Contains(err.Error(), "not running")) +} + +// TestService_ReconfigureProcessor_NoPlainProcessorNode_FallsBackToRestart: when +// the running pipeline has no plain single-worker ProcessorNode for the target +// (it is absent, or parallelized as a ParallelNode), ReconfigureProcessor returns +// ErrProcessorNotLiveReconfigurable so the caller falls back to a restart. +func TestService_ReconfigureProcessor_NoPlainProcessorNode_FallsBackToRestart(t *testing.T) { + is := is.New(t) + s := newReconfigureTestService() + + // A running pipeline whose nodes include no *stream.ProcessorNode named + // "proc1" — stands in for both the "parallelized" and "absent" cases, which + // both must degrade to a restart. + s.runningPipelines.Set("pl1", &runnablePipeline{ + pipeline: &pipeline.Instance{ID: "pl1"}, + n: []stream.Node{&stream.FaninNode{}, &stream.FanoutNode{}}, + }) + + err := s.ReconfigureProcessor(context.Background(), "pl1", "proc1") + is.True(cerrors.Is(err, ErrProcessorNotLiveReconfigurable)) +} diff --git a/pkg/lifecycle/stream/base.go b/pkg/lifecycle/stream/base.go index ba5063baa..dbc422e40 100644 --- a/pkg/lifecycle/stream/base.go +++ b/pkg/lifecycle/stream/base.go @@ -82,6 +82,15 @@ func (n *pubSubNodeBase) Sub(in <-chan *Message) { n.subNodeBase.Sub(in) } +// In returns the inbound message channel this node reads from. It is exposed so +// a PubSubNode (e.g. ProcessorNode) can select over the inbound channel together +// with its own control signals (such as a live-reconfigure wake) instead of +// blocking solely in Trigger's Receive. The channel is set by Sub during wiring; +// In returns nil until then. +func (n *pubSubNodeBase) In() <-chan *Message { + return n.subNodeBase.in +} + func (n *pubSubNodeBase) Pub() <-chan *Message { return n.pubNodeBase.Pub() } diff --git a/pkg/lifecycle/stream/processor.go b/pkg/lifecycle/stream/processor.go index e71440546..d16330255 100644 --- a/pkg/lifecycle/stream/processor.go +++ b/pkg/lifecycle/stream/processor.go @@ -19,6 +19,7 @@ package stream import ( "bytes" "context" + "sync" "time" "github.com/conduitio/conduit-commons/opencdc" @@ -35,6 +36,24 @@ type ProcessorNode struct { base pubSubNodeBase logger log.CtxLogger + + // Live-reconfigure state (see Reconfigure). swapMu guards pending and the + // lazy creation of wakeCh; Processor itself is never guarded because it is + // only ever read or swapped by the Run goroutine (applyPendingSwap runs there, + // at a record boundary), so there is no concurrent access to guard. + swapMu sync.Mutex + pending *pendingSwap + wakeCh chan struct{} +} + +// pendingSwap is a staged live-reconfigure request. done carries the outcome back +// to the Reconfigure caller: nil on success, or an error if the new processor +// failed to open (in which case the old processor is kept). done is buffered +// (cap 1) so the Run goroutine never blocks delivering the result even if the +// caller has already given up (e.g. its context was cancelled). +type pendingSwap struct { + newProcessor Processor + done chan error } type Processor interface { @@ -51,11 +70,17 @@ func (n *ProcessorNode) ID() string { } func (n *ProcessorNode) Run(ctx context.Context) error { - trigger, cleanup, err := n.base.Trigger(ctx, n.logger, nil) + _, cleanup, err := n.base.Trigger(ctx, n.logger, nil) if err != nil { return err } defer cleanup() + // Read the inbound channel directly and select over it together with the + // wake signal (below), rather than using Trigger's blocking Receive, so a + // live reconfigure applies promptly even when no records are flowing. + in := n.base.In() + wake := n.wake() + // Teardown needs to be called even if Open() fails // (to mark the processor as not running) defer func() { @@ -74,9 +99,32 @@ func (n *ProcessorNode) Run(ctx context.Context) error { } for { - msg, err := trigger() - if err != nil || msg == nil { - return err + // Invariant 4/1: apply a staged live reconfigure at this record boundary, + // before receiving the next record. applyPendingSwap runs in THIS + // goroutine, so n.Processor is never swapped while Process (below) is + // executing — no record straddles a swap, and the swap sees no in-flight + // record inside this node. Records already forwarded used the old config; + // records received after the swap use the new one, in order. + n.applyPendingSwap(ctx) + + var msg *Message + select { + case <-ctx.Done(): + // Mirrors nodeBase.Receive's logging for observability parity. + n.logger.Debug(ctx).Msg("context closed while waiting for message") + return ctx.Err() + case <-wake: + // Woken by Reconfigure to apply a staged swap promptly (the pipeline + // may be idle, with nothing arriving on in). Loop back to + // applyPendingSwap above. + continue + case m, ok := <-in: + if !ok { + // Inbound channel closed: upstream is done, so are we. + n.logger.Debug(ctx).Msg("incoming messages channel closed") + return nil + } + msg = m } if msg.filtered { @@ -149,6 +197,108 @@ func (n *ProcessorNode) SetLogger(logger log.CtxLogger) { n.logger = logger } +// wake returns the node's wake channel, creating it on first use. It is a +// buffered (cap 1) signal channel used by Reconfigure to nudge an idle Run loop. +// Guarded by swapMu because Run and Reconfigure may first touch it concurrently. +func (n *ProcessorNode) wake() chan struct{} { + n.swapMu.Lock() + defer n.swapMu.Unlock() + if n.wakeCh == nil { + n.wakeCh = make(chan struct{}, 1) + } + return n.wakeCh +} + +// Reconfigure performs a live, in-place swap of this node's Processor to +// newProcessor, applied by the Run goroutine at the next record boundary without +// restarting the pipeline. It blocks until the swap is applied (returns nil), the +// new processor fails to Open (returns the error; the old processor is kept +// running so the pipeline never drops), or ctx is done (returns ctx.Err()). +// +// Concurrency: Reconfigure only stages the request and signals Run; the swap +// itself — Open of the new processor, the switch, and Teardown of the old — is +// performed exclusively by the Run goroutine (see applyPendingSwap), so +// n.Processor is never mutated concurrently with Process. Only one reconfigure +// may be in flight at a time; a second concurrent call returns an error. +// +// Invariant 3: a reconfigure never drops or reorders records. The swap happens at +// a record boundary; during it, backpressure pauses the source, so no record is +// lost and no position advances. +func (n *ProcessorNode) Reconfigure(ctx context.Context, newProcessor Processor) error { + done := make(chan error, 1) + wake := n.wake() + + n.swapMu.Lock() + if n.pending != nil { + n.swapMu.Unlock() + return cerrors.New("a processor reconfigure is already in progress") + } + n.pending = &pendingSwap{newProcessor: newProcessor, done: done} + n.swapMu.Unlock() + + // Nudge an idle Run loop so the swap applies promptly even with no records + // flowing. Non-blocking on a cap-1 buffered channel: never blocks the caller, + // and a token already in the buffer is enough to wake the loop. + select { + case wake <- struct{}{}: + default: + } + + select { + case err := <-done: + return err + case <-ctx.Done(): + // Caller gave up. Best-effort withdraw so a later reconfigure isn't + // blocked; if Run already claimed the request, this is a no-op and the + // swap still completes (done is buffered, so Run never blocks delivering + // its now-ignored result). + n.swapMu.Lock() + if n.pending != nil && n.pending.done == done { + n.pending = nil + } + n.swapMu.Unlock() + return ctx.Err() + } +} + +// applyPendingSwap applies a staged Reconfigure request, if any. It MUST be +// called only from the Run goroutine, between records, so that n.Processor is +// never swapped while Process is executing. +// +// Open-before-teardown: the new processor is opened first; only if Open succeeds +// is it switched in and the old one torn down. If Open fails, the old processor +// keeps running and the error is reported to the Reconfigure caller — a bad edit +// never drops the pipeline (invariant 3). +func (n *ProcessorNode) applyPendingSwap(ctx context.Context) { + n.swapMu.Lock() + p := n.pending + n.pending = nil + n.swapMu.Unlock() + if p == nil { + return + } + + if err := p.newProcessor.Open(ctx); err != nil { + // Keep the current processor running; report the failure. Best-effort + // teardown of the failed new processor (Open may have partially + // initialized it, e.g. started a WASM module), mirroring Run's defer. + if tdErr := p.newProcessor.Teardown(ctx); tdErr != nil { + n.logger.Warn(ctx).Err(tdErr).Msg("could not tear down new processor after failed live-reconfigure open") + } + p.done <- cerrors.Errorf("could not open new processor for live reconfigure, keeping current processor: %w", err) + return + } + + old := n.Processor + n.Processor = p.newProcessor + if tdErr := old.Teardown(ctx); tdErr != nil { + // The swap already succeeded; a teardown error on the old processor is + // logged, not surfaced as a swap failure. + n.logger.Warn(ctx).Err(tdErr).Msg("could not tear down previous processor after live reconfigure") + } + p.done <- nil +} + // handleSingleRecord handles a sdk.SingleRecord by checking the position, // setting the new record on the message and sending it downstream. // If there are any errors, the method nacks the message and returns diff --git a/pkg/lifecycle/stream/processor_reconfigure_property_test.go b/pkg/lifecycle/stream/processor_reconfigure_property_test.go new file mode 100644 index 000000000..012d39c9b --- /dev/null +++ b/pkg/lifecycle/stream/processor_reconfigure_property_test.go @@ -0,0 +1,209 @@ +// 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 + +import ( + "context" + "fmt" + "strconv" + "sync" + "testing" + "time" + + "github.com/conduitio/conduit-commons/opencdc" + sdk "github.com/conduitio/conduit-processor-sdk" + "github.com/conduitio/conduit/pkg/foundation/metrics/noop" + "github.com/matryer/is" +) + +// recordingProc is a hand-written Processor (not a gomock) for the property and +// fault tests, where random interleavings and blocking-mid-Open control are +// awkward to express as up-front gomock expectations. It tags each record with +// its version, records the size of every Process call, and can block in Open. +type recordingProc struct { + version int + + mu sync.Mutex + callSizes []int // len(recs) of each Process call, to pin the 1-in-1-out contract + + // reachedOpen, if non-nil, is closed when Open is entered; blockOpen makes + // Open then block until ctx is done (to hold a swap open mid-Open). + reachedOpen chan struct{} + blockOpen bool +} + +func (f *recordingProc) Open(ctx context.Context) error { + if f.reachedOpen != nil { + close(f.reachedOpen) + } + if f.blockOpen { + <-ctx.Done() + return ctx.Err() + } + return nil +} + +func (f *recordingProc) Process(_ context.Context, recs []opencdc.Record) []sdk.ProcessedRecord { + f.mu.Lock() + f.callSizes = append(f.callSizes, len(recs)) + f.mu.Unlock() + out := make([]sdk.ProcessedRecord, len(recs)) + for i, r := range recs { + if r.Metadata == nil { + r.Metadata = map[string]string{} + } + r.Metadata["version"] = strconv.Itoa(f.version) + out[i] = sdk.SingleRecord(r) + } + return out +} + +func (f *recordingProc) Teardown(context.Context) error { return nil } + +// TestProcessorNode_OneRecordPerProcessCall_LiveSwapContract pins the non-buffering +// 1-in-1-out processor contract the live-swap zero-drop guarantee depends on (see +// the design doc's Data-safety §): the node must hand Process exactly one record +// per call and never batch. If a future change makes the node call Process with +// more than one record (a batching/windowing model), this test breaks — forcing +// whoever makes that change to confront the hot-swap interaction (a buffered +// processor must be restart-class, or flush before teardown) rather than +// silently introducing a drop. +func TestProcessorNode_OneRecordPerProcessCall_LiveSwapContract(t *testing.T) { + is := is.New(t) + ctx := context.Background() + + const n = 8 + proc := &recordingProc{version: 0} + node := &ProcessorNode{Name: "t", Processor: proc, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + node.Sub(in) + out := node.Pub() + + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); is.NoErr(node.Run(ctx)) }() + + go func() { + for i := 0; i < n; i++ { + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte(strconv.Itoa(i)), Metadata: map[string]string{}}} + } + close(in) + }() + for i := 0; i < n; i++ { + <-out + } + wg.Wait() + + proc.mu.Lock() + defer proc.mu.Unlock() + is.Equal(len(proc.callSizes), n) // one Process call per record + for _, sz := range proc.callSizes { + is.Equal(sz, 1) // never batched — the contract the live swap relies on + } +} + +// TestProcessorNode_Reconfigure_PreservesOrderingAndDelivery is the property-style +// test: it interleaves a stream of records with many live reconfigures and +// asserts every record comes out exactly once, in order (invariants 3 and 4). The +// version that tags any given record is nondeterministic (it depends on which +// swap won the race at that boundary) — that is expected; what must hold is that +// no record is dropped, duplicated, or reordered across the swaps. +func TestProcessorNode_Reconfigure_PreservesOrderingAndDelivery(t *testing.T) { + is := is.New(t) + ctx := context.Background() + + const n = 200 + node := &ProcessorNode{Name: "t", Processor: &recordingProc{version: 0}, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + node.Sub(in) + out := node.Pub() + + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); is.NoErr(node.Run(ctx)) }() + + // Consumer: collect the positions in the order they arrive. + got := make([]int, 0, n) + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < n; i++ { + m := <-out + p, _ := strconv.Atoi(string(m.Record.Position)) + got = append(got, p) + } + }() + + // Producer: send records 1..n, reconfiguring at a deterministic mix of points + // so many boundaries are exercised without a random seed. + for i := 1; i <= n; i++ { + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte(fmt.Sprintf("%06d", i)), Metadata: map[string]string{}}} + if i%17 == 0 || i%23 == 0 { + is.NoErr(node.Reconfigure(ctx, &recordingProc{version: i})) + } + } + close(in) + <-done + wg.Wait() + + // Every record delivered exactly once, in order — no drop, no reorder. + is.Equal(len(got), n) + for i, p := range got { + is.Equal(p, i+1) + } +} + +// TestProcessorNode_Reconfigure_ContextCancelledMidSwap is the targeted mid-swap +// fault test (the design doc's PR1 gate for the new crash surface): cancelling the +// node context while a swap is mid-Open must not lose the already-processed record, +// must fail the reconfigure cleanly (old processor kept), and must let the node +// shut down without panic or deadlock. +func TestProcessorNode_Reconfigure_ContextCancelledMidSwap(t *testing.T) { + is := is.New(t) + ctx, cancel := context.WithCancel(context.Background()) + + old := &recordingProc{version: 0} + node := &ProcessorNode{Name: "t", Processor: old, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + node.Sub(in) + out := node.Pub() + + var runErr error + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); runErr = node.Run(ctx) }() + + // One record flows through the old processor. + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte("1"), Metadata: map[string]string{}}} + m := <-out + is.Equal(m.Record.Metadata["version"], "0") + + // Start a swap whose Open blocks; wait until it is genuinely mid-Open, then + // cancel the node context. + newProc := &recordingProc{version: 1, reachedOpen: make(chan struct{}), blockOpen: true} + reconfErr := make(chan error, 1) + go func() { reconfErr <- node.Reconfigure(ctx, newProc) }() + + select { + case <-newProc.reachedOpen: + case <-time.After(2 * time.Second): + t.Fatal("swap never reached Open") + } + cancel() // cancel mid-swap + + is.True(<-reconfErr != nil) // the swap did not complete + wg.Wait() + is.True(runErr != nil) // node shut down on the cancelled context (no panic/deadlock) +} diff --git a/pkg/lifecycle/stream/processor_reconfigure_test.go b/pkg/lifecycle/stream/processor_reconfigure_test.go new file mode 100644 index 000000000..4430e3cc3 --- /dev/null +++ b/pkg/lifecycle/stream/processor_reconfigure_test.go @@ -0,0 +1,200 @@ +// 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 + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/conduitio/conduit-commons/opencdc" + sdk "github.com/conduitio/conduit-processor-sdk" + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/metrics/noop" + "github.com/conduitio/conduit/pkg/lifecycle/stream/mock" + "github.com/matryer/is" + "go.uber.org/mock/gomock" +) + +// tagProcessor returns a mock Processor whose Process tags each record's metadata +// with by=, so a test can prove which processor handled a given record. +func tagProcessor(ctrl *gomock.Controller, tag string) *mock.Processor { + p := mock.NewProcessor(ctrl) + p.EXPECT(). + Process(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, got []opencdc.Record) []sdk.ProcessedRecord { + got[0].Metadata["by"] = tag + return []sdk.ProcessedRecord{sdk.SingleRecord(got[0])} + }).AnyTimes() + return p +} + +// TestProcessorNode_Reconfigure_SwapAtRecordBoundary proves the core in-place +// swap: record N goes through the old processor, and after Reconfigure returns, +// record N+1 goes through the new one — the source never restarts, and the swap +// happens at a clean record boundary. +func TestProcessorNode_Reconfigure_SwapAtRecordBoundary(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + + procA := tagProcessor(ctrl, "A") + procA.EXPECT().Open(gomock.Any()) + procA.EXPECT().Teardown(gomock.Any()) // torn down when swapped out + + procB := tagProcessor(ctrl, "B") + procB.EXPECT().Open(gomock.Any()) // opened during the swap + procB.EXPECT().Teardown(gomock.Any()) // torn down when the node stops + + n := &ProcessorNode{Name: "test", Processor: procA, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + n.Sub(in) + out := n.Pub() + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + is.NoErr(n.Run(ctx)) + }() + + // record 1 -> processor A + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte("p1"), Metadata: map[string]string{}}} + got1 := <-out + is.Equal(got1.Record.Metadata["by"], "A") + + // swap to B; Reconfigure blocks until the Run goroutine applies it + is.NoErr(n.Reconfigure(ctx, procB)) + + // record 2 -> processor B + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte("p2"), Metadata: map[string]string{}}} + got2 := <-out + is.Equal(got2.Record.Metadata["by"], "B") + + close(in) + wg.Wait() + _, ok := <-out + is.Equal(false, ok) // out closed on stop +} + +// TestProcessorNode_Reconfigure_OpenFailureKeepsOldProcessor proves a bad edit +// never drops the pipeline: if the new processor fails to Open, the old one keeps +// running and the error is returned to the caller (open-before-teardown). +func TestProcessorNode_Reconfigure_OpenFailureKeepsOldProcessor(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + + procA := tagProcessor(ctrl, "A") + procA.EXPECT().Open(gomock.Any()) + procA.EXPECT().Teardown(gomock.Any()) // kept through the failed swap, torn down on stop + + openErr := cerrors.New("bad new config") + procB := mock.NewProcessor(ctrl) + procB.EXPECT().Open(gomock.Any()).Return(openErr) + procB.EXPECT().Teardown(gomock.Any()) // best-effort cleanup of the failed processor + + n := &ProcessorNode{Name: "test", Processor: procA, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + n.Sub(in) + out := n.Pub() + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + is.NoErr(n.Run(ctx)) + }() + + // idle reconfigure that fails to open -> returns the error, keeps A + err := n.Reconfigure(ctx, procB) + is.True(err != nil) + is.True(cerrors.Is(err, openErr)) + + // A is still active: a record must still flow through it + in <- &Message{Ctx: ctx, Record: opencdc.Record{Position: []byte("p"), Metadata: map[string]string{}}} + got := <-out + is.Equal(got.Record.Metadata["by"], "A") + + close(in) + wg.Wait() +} + +// TestProcessorNode_Reconfigure_IdlePipelineAppliesPromptly proves the wake +// signal applies a swap even when no records are flowing (the interruptible-wait +// requirement): with nothing on the inbound channel, Reconfigure must still +// return promptly rather than block until the next record. +func TestProcessorNode_Reconfigure_IdlePipelineAppliesPromptly(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + + procA := mock.NewProcessor(ctrl) + procA.EXPECT().Open(gomock.Any()) + procA.EXPECT().Teardown(gomock.Any()) // torn down during the swap + + procB := mock.NewProcessor(ctrl) + procB.EXPECT().Open(gomock.Any()) + procB.EXPECT().Teardown(gomock.Any()) // torn down on stop + + n := &ProcessorNode{Name: "test", Processor: procA, ProcessorTimer: noop.Timer{}} + in := make(chan *Message) + n.Sub(in) + out := n.Pub() + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + is.NoErr(n.Run(ctx)) + }() + + // no records are sent; the pipeline is idle + done := make(chan error, 1) + go func() { done <- n.Reconfigure(ctx, procB) }() + select { + case err := <-done: + is.NoErr(err) + case <-time.After(2 * time.Second): + t.Fatal("reconfigure did not apply on an idle pipeline within 2s") + } + + close(in) + wg.Wait() + _, ok := <-out + is.Equal(false, ok) +} + +// TestProcessorNode_Reconfigure_ContextCancelled proves Reconfigure unblocks and +// returns ctx.Err() if its context is cancelled before the swap is applied +// (here the node is not running, so nothing applies it). +func TestProcessorNode_Reconfigure_ContextCancelled(t *testing.T) { + is := is.New(t) + ctrl := gomock.NewController(t) + + // Neither processor is ever opened: Run is not started, and the request is + // withdrawn on cancel. No expectations => any call would fail the test. + n := &ProcessorNode{Name: "test", Processor: mock.NewProcessor(ctrl), ProcessorTimer: noop.Timer{}} + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + err := n.Reconfigure(ctx, mock.NewProcessor(ctrl)) + is.True(cerrors.Is(err, context.Canceled)) +} diff --git a/pkg/provisioning/apply_plan_inplace_test.go b/pkg/provisioning/apply_plan_inplace_test.go new file mode 100644 index 000000000..5a4b7c634 --- /dev/null +++ b/pkg/provisioning/apply_plan_inplace_test.go @@ -0,0 +1,183 @@ +// 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 provisioning + +import ( + "context" + "testing" + + "github.com/conduitio/conduit/pkg/foundation/cerrors" + "github.com/conduitio/conduit/pkg/foundation/log" + "github.com/conduitio/conduit/pkg/lifecycle" + "github.com/conduitio/conduit/pkg/processor" + "github.com/conduitio/conduit/pkg/provisioning/config" + "github.com/matryer/is" + "go.uber.org/mock/gomock" +) + +// processorUpdateFixture returns an old/desired pair whose only difference is a +// single pipeline-level processor's settings — a live-swappable change (see +// Change.liveSwappable), so the diff is LiveEligible and ApplyPlanLive applies it +// in place rather than via a restart. +func processorUpdateFixture() (old, desired config.Pipeline) { + old = config.Pipeline{ + ID: "p1", + DLQ: dlqFixture(), + Connectors: []config.Connector{ + {ID: "p1:conn:src", Type: config.TypeSource, Plugin: "builtin:postgres", Settings: map[string]string{"table": "users"}}, + }, + Processors: []config.Processor{ + {ID: "p1:proc:1", Plugin: "builtin:field.set", Settings: map[string]string{"field": ".Key", "value": "a"}}, + }, + } + desired = old + desired.Processors = []config.Processor{ + {ID: "p1:proc:1", Plugin: "builtin:field.set", Settings: map[string]string{"field": ".Key", "value": "b"}}, + } + return old, desired +} + +// TestApplyPlanLive_ProcessorUpdate_AppliesInPlace_NoRestart proves the payoff of +// the live-in-place path: a processor config change on a running pipeline commits +// the new config and swaps the live node via ReconfigureProcessor — with NO +// StopAndWait and NO Start (no restart). The absence of those two expectations is +// the assertion: gomock fails the test if the restart path runs. +func TestApplyPlanLive_ProcessorUpdate_AppliesInPlace_NoRestart(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + srv, pipSrv, connSrv, procSrv, _, lifecycleSrv := newTestService(ctrl, log.Nop()) + + old, desired := processorUpdateFixture() + expectExportRunning(pipSrv, connSrv, procSrv, old) + + procSrv.EXPECT().Update(gomock.Any(), "p1:proc:1", "builtin:field.set", gomock.Any()). + Return(&processor.Instance{ID: "p1:proc:1"}, nil) + lifecycleSrv.EXPECT().ReconfigureProcessor(gomock.Any(), "p1", "p1:proc:1").Return(nil) + + diff, err := srv.Plan(ctx, desired) + is.NoErr(err) + is.True(diff.LiveEligible()) // sanity: a processor-only update is live-eligible + + applied, err := srv.ApplyPlanLive(ctx, desired, diff.Hash, true) // authorized + is.NoErr(err) + is.Equal(applied.Hash, diff.Hash) +} + +// TestApplyPlanLive_ProcessorUpdate_InPlace_CompletesDespiteCallerCancel proves +// the in-place apply detaches from the caller's context once it starts committing +// live state (review MINOR-1): cancelling the caller context mid-apply must not +// abandon the swap (which would report a failure the apply actually landed) — the +// apply runs to a consistent end on a non-cancelled context and reports success. +func TestApplyPlanLive_ProcessorUpdate_InPlace_CompletesDespiteCallerCancel(t *testing.T) { + is := is.New(t) + ctx, cancel := context.WithCancel(context.Background()) + ctrl := gomock.NewController(t) + srv, pipSrv, connSrv, procSrv, _, lifecycleSrv := newTestService(ctrl, log.Nop()) + + old, desired := processorUpdateFixture() + expectExportRunning(pipSrv, connSrv, procSrv, old) + + // The forward import runs on the detached context; cancel the caller context + // as a side-effect to prove the rest of the apply proceeds regardless. + procSrv.EXPECT().Update(gomock.Any(), "p1:proc:1", "builtin:field.set", gomock.Any()). + DoAndReturn(func(importCtx context.Context, _, _ string, _ processor.Config) (*processor.Instance, error) { + cancel() + is.NoErr(importCtx.Err()) // detached: not cancelled by the caller + return &processor.Instance{ID: "p1:proc:1"}, nil + }) + lifecycleSrv.EXPECT().ReconfigureProcessor(gomock.Any(), "p1", "p1:proc:1"). + DoAndReturn(func(swapCtx context.Context, _, _ string) error { + is.NoErr(swapCtx.Err()) // swap runs on the detached context too + return nil + }) + + diff, err := srv.Plan(ctx, desired) + is.NoErr(err) + applied, err := srv.ApplyPlanLive(ctx, desired, diff.Hash, true) + is.NoErr(err) + is.Equal(applied.Hash, diff.Hash) +} + +// TestApplyPlanLive_ProcessorUpdate_NotLiveReconfigurable_FallsBackToRestart: +// when the live node can't be swapped (ReconfigureProcessor reports the processor +// isn't live-reconfigurable, e.g. it's parallelized), the config is already +// committed, so ApplyPlanLive falls back to a full StopAndWait -> import -> Start. +func TestApplyPlanLive_ProcessorUpdate_NotLiveReconfigurable_FallsBackToRestart(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + srv, pipSrv, connSrv, procSrv, _, lifecycleSrv := newTestService(ctrl, log.Nop()) + + old, desired := processorUpdateFixture() + expectExportRunning(pipSrv, connSrv, procSrv, old) + + // procSrv.Update is called by applyInPlace's import and again by the restart + // path's import (the stateless mock Export keeps returning old, so both + // imports recompute the same action — in production the second is a no-op). + procSrv.EXPECT().Update(gomock.Any(), "p1:proc:1", "builtin:field.set", gomock.Any()). + Return(&processor.Instance{ID: "p1:proc:1"}, nil).Times(2) + lifecycleSrv.EXPECT().ReconfigureProcessor(gomock.Any(), "p1", "p1:proc:1"). + Return(lifecycle.ErrProcessorNotLiveReconfigurable) + + var order []string + lifecycleSrv.EXPECT().StopAndWait(gomock.Any(), "p1").DoAndReturn(func(context.Context, string) error { + order = append(order, "stop") + return nil + }) + lifecycleSrv.EXPECT().Start(gomock.Any(), "p1").DoAndReturn(func(context.Context, string) error { + order = append(order, "start") + return nil + }) + + diff, err := srv.Plan(ctx, desired) + is.NoErr(err) + + _, err = srv.ApplyPlanLive(ctx, desired, diff.Hash, true) + is.NoErr(err) + is.Equal(order, []string{"stop", "start"}) // fell back to a restart +} + +// TestApplyPlanLive_ProcessorUpdate_OpenFails_RollsBack: when the new processor +// fails to open, the old processor keeps running (open-before-teardown). The +// in-place apply rolls back — restoring the old config to the store — and returns +// the error, without ever restarting the pipeline. +func TestApplyPlanLive_ProcessorUpdate_OpenFails_RollsBack(t *testing.T) { + is := is.New(t) + ctx := context.Background() + ctrl := gomock.NewController(t) + srv, pipSrv, connSrv, procSrv, _, lifecycleSrv := newTestService(ctrl, log.Nop()) + + old, desired := processorUpdateFixture() + expectExportRunning(pipSrv, connSrv, procSrv, old) + + // Forward import commits the new config (one Update). The rollback then calls + // transactionalImport(old); against the stateless mock Export (which keeps + // returning old) that diffs old-vs-old = empty, so no second Update is + // recorded here — in production, where the store reflects the forward import, + // the rollback re-imports old and reverses it. + procSrv.EXPECT().Update(gomock.Any(), "p1:proc:1", "builtin:field.set", gomock.Any()). + Return(&processor.Instance{ID: "p1:proc:1"}, nil) + openErr := cerrors.New("new processor failed to open") + lifecycleSrv.EXPECT().ReconfigureProcessor(gomock.Any(), "p1", "p1:proc:1").Return(openErr) + // No StopAndWait / Start: a failed in-place swap never restarts the pipeline. + + diff, err := srv.Plan(ctx, desired) + is.NoErr(err) + + _, err = srv.ApplyPlanLive(ctx, desired, diff.Hash, true) + is.True(err != nil) + is.True(cerrors.Is(err, openErr)) +} diff --git a/pkg/provisioning/interfaces.go b/pkg/provisioning/interfaces.go index fed5f3555..ad21a18f8 100644 --- a/pkg/provisioning/interfaces.go +++ b/pkg/provisioning/interfaces.go @@ -80,4 +80,12 @@ type LifecycleService interface { // (rather than Stop) to know it is safe to mutate a running pipeline's // stored config; see docs/design-documents/20260708-live-server-deploy-apply.md. StopAndWait(ctx context.Context, pipelineID string) error + // ReconfigureProcessor swaps a processor's config in a running pipeline in + // place, without a restart — see lifecycle.Service.ReconfigureProcessor. + // ApplyPlanLiveInPlace uses it for live-eligible diffs; the caller persists + // the new config first. Returns lifecycle.ErrProcessorNotLiveReconfigurable + // when the processor cannot be swapped live (e.g. it is parallelized), so + // the caller falls back to a restart. See the live in-place hot-reload + // design doc. + ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error } diff --git a/pkg/provisioning/mock/provisioning.go b/pkg/provisioning/mock/provisioning.go index 571ee9323..f2e448cb5 100644 --- a/pkg/provisioning/mock/provisioning.go +++ b/pkg/provisioning/mock/provisioning.go @@ -1033,6 +1033,44 @@ func (m *LifecycleService) EXPECT() *LifecycleServiceMockRecorder { return m.recorder } +// ReconfigureProcessor mocks base method. +func (m *LifecycleService) ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReconfigureProcessor", ctx, pipelineID, processorID) + ret0, _ := ret[0].(error) + return ret0 +} + +// ReconfigureProcessor indicates an expected call of ReconfigureProcessor. +func (mr *LifecycleServiceMockRecorder) ReconfigureProcessor(ctx, pipelineID, processorID any) *LifecycleServiceReconfigureProcessorCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReconfigureProcessor", reflect.TypeOf((*LifecycleService)(nil).ReconfigureProcessor), ctx, pipelineID, processorID) + return &LifecycleServiceReconfigureProcessorCall{Call: call} +} + +// LifecycleServiceReconfigureProcessorCall wrap *gomock.Call +type LifecycleServiceReconfigureProcessorCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *LifecycleServiceReconfigureProcessorCall) Return(arg0 error) *LifecycleServiceReconfigureProcessorCall { + c.Call = c.Call.Return(arg0) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *LifecycleServiceReconfigureProcessorCall) Do(f func(context.Context, string, string) error) *LifecycleServiceReconfigureProcessorCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *LifecycleServiceReconfigureProcessorCall) DoAndReturn(f func(context.Context, string, string) error) *LifecycleServiceReconfigureProcessorCall { + c.Call = c.Call.DoAndReturn(f) + return c +} + // Start mocks base method. func (m *LifecycleService) Start(ctx context.Context, pipelineID string) error { m.ctrl.T.Helper() diff --git a/pkg/provisioning/plan.go b/pkg/provisioning/plan.go index 06805f652..5f0818a8e 100644 --- a/pkg/provisioning/plan.go +++ b/pkg/provisioning/plan.go @@ -49,6 +49,7 @@ import ( "github.com/conduitio/conduit/pkg/foundation/cerrors" "github.com/conduitio/conduit/pkg/foundation/cerrors/conduiterr" + "github.com/conduitio/conduit/pkg/lifecycle" "github.com/conduitio/conduit/pkg/pipeline" "github.com/conduitio/conduit/pkg/provisioning/config" json "github.com/goccy/go-json" @@ -97,12 +98,72 @@ type Change struct { Action ChangeAction `json:"action"` Effect Effect `json:"effect"` ConfigPaths []string `json:"configPaths,omitempty"` + // LiveSwappable reports whether this change can be applied to a *running* + // pipeline in place — without a stop/restart — because it touches no source + // position, ack/durability, or connection state. It is a strict subset of + // EffectInPlace: only a processor config update and a pipeline + // Name/Description-only update qualify (see liveSwappable). Additive field; + // consumers that predate it default to the safe interpretation (false = + // treat as restart-class). The apply path keys on Diff.LiveEligible, not on + // this field per-change, but it is surfaced so CLI/MCP/UI can show "this + // change applies live" vs "this change restarts the pipeline". + LiveSwappable bool `json:"liveSwappable"` // Code is a stable, dotted identifier for this kind of change (e.g. // "provisioning.connector.update"), namespaced the same way // conduiterr.Code reasons are, for consistent agent/UI consumption. Code string `json:"code"` } +// liveSwappable reports whether this Change can be applied to a running pipeline +// in place (via a ProcessorNode live swap, or an in-memory pipeline metadata +// update) without restarting it. Only two kinds qualify, both free of source +// position, ack/durability, and connection state: +// +// - a processor config update (updateProcessorAction — including a plugin +// change, since ProcessorNode.Reconfigure rebuilds the processor), and +// - a pipeline update touching only Name and/or Description. +// +// Everything else forces a restart-class apply: connector create/update/delete +// (position + open connection), the DLQ (a live destination with ack +// semantics — the "dlq" config path disqualifies a pipeline update), processor +// create/delete and pipeline membership changes (topology), and pipeline +// delete. See the design doc's "Data-safety" and classification sections. +func (c Change) liveSwappable() bool { + switch c.Resource { + case ResourceProcessor: + if c.Action != ChangeActionUpdate { + return false + } + // A Workers change alters the node topology (a single-worker + // ProcessorNode vs a parallel ParallelNode) — an in-place processor swap + // cannot make that change (ReconfigureProcessor only swaps the processor + // inside an existing node), so it must restart. Every other processor + // field (plugin, condition, settings) is a live swap. + for _, p := range c.ConfigPaths { + if p == configPathWorkers { + return false + } + } + return true + case ResourcePipeline: + if c.Action != ChangeActionUpdate { + return false + } + for _, p := range c.ConfigPaths { + if p != configPathName && p != configPathDescription { + return false + } + } + return true + case ResourceConnector: + // Connectors own source position, ack/durability, and an open plugin + // connection — never live-swappable (see the method doc). + return false + default: + return false + } +} + // Diff is Plan's result: every Change needed to reconcile the pipeline // currently stored with the desired config, plus a Hash binding this exact // Diff — ApplyPlan refuses to run unless the caller presents this Hash. @@ -116,6 +177,25 @@ type Diff struct { // already matches the current state (an idempotent re-apply). func (d Diff) Empty() bool { return len(d.Changes) == 0 } +// LiveEligible reports whether this whole diff can be applied to a running +// pipeline in place — without a stop/restart — because it is non-empty and +// *every* change in it is LiveSwappable. It is all-or-nothing on purpose: a diff +// that mixes a processor tweak with a source change needs a restart for the +// source anyway, so the whole apply takes the restart path. An empty diff is not +// live-eligible because there is nothing to apply (it is handled as an idempotent +// no-op upstream). See ApplyPlanLiveInPlace for how the apply path uses this. +func (d Diff) LiveEligible() bool { + if d.Empty() { + return false + } + for _, c := range d.Changes { + if !c.LiveSwappable { + return false + } + } + return true +} + // computeHash returns a deterministic digest of PipelineID, Changes (in the // order actionsBuilder.Build produced them — the same order ApplyPlan would // execute them in, not re-sorted, since the hash must be over exactly what @@ -191,6 +271,15 @@ func (s *Service) Plan(ctx context.Context, desired config.Pipeline) (Diff, erro } } + // Classify each change for live in-place applicability. Derived from + // Resource/Action/ConfigPaths, so it is stable for a given desired config + // and folds into the hash consistently. Computed after the brand-new + // override above (which only touches Effect, not the fields liveSwappable + // reads). + for i := range changes { + changes[i].LiveSwappable = changes[i].liveSwappable() + } + d := Diff{PipelineID: desired.ID, Changes: changes} d.Hash = d.computeHash(desired) return d, nil @@ -433,6 +522,30 @@ func (s *Service) ApplyPlanLive(ctx context.Context, desired config.Pipeline, ha return fresh, nil } + // Live-eligible diff on a running pipeline: apply in place (swap the changed + // processor nodes + update name/description metadata) without a stop/restart. + // This is the payoff of the LiveSwappable classification — a processor tweak + // takes effect with no availability blip, no position replay, no record loss. + // applyInPlace returns swappedAll=false (err=nil) when a processor cannot be + // swapped live (e.g. it is parallelized): the config is already committed, so + // we fall through to the restart path below, which rebuilds every node from + // it, applying the whole diff uniformly. + if fresh.LiveEligible() { + oldConfig, err := s.Export(ctx, desired.ID) + if err != nil { + return fresh, cerrors.Errorf("could not export current config of pipeline %q for in-place apply: %w", desired.ID, err) + } + swappedAll, err := s.applyInPlace(ctx, desired, oldConfig, fresh) + if err != nil { + return fresh, err + } + if swappedAll { + return fresh, nil + } + // Fall through to the restart path: the config is committed, so + // StopAndWait -> (idempotent) import -> Start rebuilds from it. + } + // Invariant 7 / Tier-1 safety: StopAndWait — not Stop — is the only // primitive that proves the pipeline is fully drained and its positions // are durably persisted before importPipeline runs against it. See @@ -468,6 +581,84 @@ func (s *Service) ApplyPlanLive(ctx context.Context, desired config.Pipeline, ha return fresh, nil } +// applyInPlace applies a live-eligible diff to a running pipeline without a +// stop/restart. It commits the new config to the store (invariant 5), then swaps +// each changed processor into the live node graph via the lifecycle service; +// name/description changes take effect through the committed store instance and +// need no node swap. +// +// It returns swappedAll=true when every change was applied in place. It returns +// (false, nil) when a processor cannot be swapped live (e.g. it is parallelized): +// the config is already committed, so the caller falls back to a restart that +// rebuilds every node from it. On a genuine swap failure (the new processor fails +// to open), it rolls back — restoring the old config and re-swapping any +// already-swapped processors back — so the store and the live pipeline agree on +// the old config and the pipeline keeps running unchanged, and returns the error. +func (s *Service) applyInPlace(ctx context.Context, desired, oldConfig config.Pipeline, diff Diff) (bool, error) { + // Once we commit to an in-place apply it mutates live node state that must + // reach a consistent end, so detach from the caller's cancellation (keeping + // values). Otherwise a caller context cancelled mid-swap would abandon + // ProcessorNode.Reconfigure's wait while the Run goroutine still completes the + // swap — reporting a failure the apply actually landed — and would make the + // rollback's transactional import fail, skipping it. The work here is bounded + // (a config commit plus a few processor Opens). The apply is already + // authorized and lock-held (see ApplyPlanLive); the caller learns the outcome + // from the return value, not by cancelling. + ctx = context.WithoutCancel(ctx) + + if err := s.transactionalImport(ctx, desired); err != nil { + return false, err + } + + var swapped []string + for _, c := range diff.Changes { + if c.Resource != ResourceProcessor || c.Action != ChangeActionUpdate { + // The only other live-swappable change is a pipeline + // name/description update, applied by the store import above. Caveat: + // that updates the stored instance; a running pipeline's + // metric-label name (baked in at node-build time) reflects it only on + // the next restart. Cosmetic — not a data-path concern. + continue + } + err := s.lifecycleService.ReconfigureProcessor(ctx, desired.ID, c.ID) + switch { + case cerrors.Is(err, lifecycle.ErrProcessorNotLiveReconfigurable): + // Not swappable live. Signal a restart fallback; no rollback needed — + // the restart tears down and rebuilds every node from the committed + // config anyway. + return false, nil + case err != nil: + // The new processor failed to open; the old one is still running + // (open-before-teardown). Roll back so store and live agree on the + // old config, then surface the error. + s.rollbackInPlace(ctx, desired.ID, oldConfig, swapped) + return false, cerrors.Errorf("could not apply processor %q to running pipeline %q in place: %w", c.ID, desired.ID, err) + } + swapped = append(swapped, c.ID) + } + return true, nil +} + +// rollbackInPlace best-effort undoes a partial in-place apply after a mid-diff +// swap failure: it restores oldConfig to the store and re-swaps the +// already-swapped processors back to it. Errors are logged, not returned — the +// caller is already returning the original failure, and the pipeline is still +// running its old processors (open-before-teardown), so the worst case is a +// store/live mismatch that a restart reconciles. +func (s *Service) rollbackInPlace(ctx context.Context, pipelineID string, oldConfig config.Pipeline, swapped []string) { + if err := s.transactionalImport(ctx, oldConfig); err != nil { + s.logger.Err(ctx, err).Str("pipeline_id", pipelineID). + Msg("could not restore previous config during in-place apply rollback") + return + } + for _, procID := range swapped { + if err := s.lifecycleService.ReconfigureProcessor(ctx, pipelineID, procID); err != nil { + s.logger.Err(ctx, err).Str("pipeline_id", pipelineID).Str("processor_id", procID). + Msg("could not re-swap processor to previous config during in-place apply rollback") + } + } +} + // isRunning reports whether the pipeline identified by id currently has // live, in-process work per isRunningStatus. A not-yet-existing pipeline // (pipeline.ErrInstanceNotFound) is reported as not running: there is @@ -530,22 +721,37 @@ func isRunningStatus(status pipeline.Status) bool { // actionsBuilder.preparePipelineActions's own cmp.Equal check (same ignored // field: Status) so a Change's ConfigPaths can never claim a field changed // that the builder itself considered equal. +// Pipeline-level Change.ConfigPaths names. Shared by diffPipelineFields (which +// emits them) and Change.liveSwappable (which reads them to decide that only +// name/description changes are live-swappable), so the producer and consumer +// can never disagree on a spelling. +const ( + configPathName = "name" + configPathDescription = "description" + configPathConnectors = "connectors" + configPathProcessors = "processors" + configPathDLQ = "dlq" + // configPathWorkers is a processor-level path (see diffProcessorFields); a + // change to it is a node-topology change, not a live swap. + configPathWorkers = "workers" +) + func diffPipelineFields(oldCfg, newCfg config.Pipeline) []string { var paths []string if oldCfg.Name != newCfg.Name { - paths = append(paths, "name") + paths = append(paths, configPathName) } if oldCfg.Description != newCfg.Description { - paths = append(paths, "description") + paths = append(paths, configPathDescription) } if !equalConnectorIDs(oldCfg.Connectors, newCfg.Connectors) { - paths = append(paths, "connectors") + paths = append(paths, configPathConnectors) } if !equalProcessorIDs(oldCfg.Processors, newCfg.Processors) { - paths = append(paths, "processors") + paths = append(paths, configPathProcessors) } if !cmp.Equal(oldCfg.DLQ, newCfg.DLQ) { - paths = append(paths, "dlq") + paths = append(paths, configPathDLQ) } return paths } @@ -581,7 +787,7 @@ func diffProcessorFields(oldCfg, newCfg config.Processor) []string { paths = append(paths, "plugin") } if oldCfg.Workers != newCfg.Workers { - paths = append(paths, "workers") + paths = append(paths, configPathWorkers) } if oldCfg.Condition != newCfg.Condition { paths = append(paths, "condition") diff --git a/pkg/provisioning/plan_liveswappable_test.go b/pkg/provisioning/plan_liveswappable_test.go new file mode 100644 index 000000000..23a5074f6 --- /dev/null +++ b/pkg/provisioning/plan_liveswappable_test.go @@ -0,0 +1,96 @@ +// 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 provisioning + +import ( + "testing" + + "github.com/matryer/is" +) + +// TestChange_liveSwappable covers every (Resource, Action) combination plus the +// pipeline-update ConfigPaths cases (name/description swappable; dlq/connectors/ +// processors not), which is the exact classification the live-apply decision +// depends on. +func TestChange_liveSwappable(t *testing.T) { + testCases := []struct { + name string + ch Change + want bool + }{ + // processors: an update is live-swappable, EXCEPT a Workers change (a + // node-topology change a swap can't make — regression for the review's + // MAJOR-1) + {"processor update (no paths)", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate}, true}, + {"processor update settings", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, ConfigPaths: []string{"settings.field"}}, true}, + {"processor update plugin", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, ConfigPaths: []string{"plugin"}}, true}, + {"processor update condition", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, ConfigPaths: []string{"condition"}}, true}, + {"processor update workers", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, ConfigPaths: []string{"workers"}}, false}, + {"processor update settings+workers", Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, ConfigPaths: []string{"settings.field", "workers"}}, false}, + {"processor create", Change{Resource: ResourceProcessor, Action: ChangeActionCreate}, false}, + {"processor delete", Change{Resource: ResourceProcessor, Action: ChangeActionDelete}, false}, + + // pipeline updates: only name/description-only qualify + {"pipeline update name only", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"name"}}, true}, + {"pipeline update description only", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"description"}}, true}, + {"pipeline update name+description", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"name", "description"}}, true}, + {"pipeline update dlq", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"dlq"}}, false}, + {"pipeline update name+dlq", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"name", "dlq"}}, false}, + {"pipeline update connectors", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"connectors"}}, false}, + {"pipeline update processors", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"processors"}}, false}, + {"pipeline update no paths", Change{Resource: ResourcePipeline, Action: ChangeActionUpdate}, true}, // vacuously name/description-only + {"pipeline create", Change{Resource: ResourcePipeline, Action: ChangeActionCreate}, false}, + {"pipeline delete", Change{Resource: ResourcePipeline, Action: ChangeActionDelete}, false}, + + // connectors are never live-swappable (position/connection/ack state) + {"connector update", Change{Resource: ResourceConnector, Action: ChangeActionUpdate}, false}, + {"connector create", Change{Resource: ResourceConnector, Action: ChangeActionCreate}, false}, + {"connector delete", Change{Resource: ResourceConnector, Action: ChangeActionDelete}, false}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + is := is.New(t) + is.Equal(tc.ch.liveSwappable(), tc.want) + }) + } +} + +// TestDiff_LiveEligible proves the all-or-nothing rule: a diff applies in place +// only if it is non-empty and every change is live-swappable. +func TestDiff_LiveEligible(t *testing.T) { + procUpdate := Change{Resource: ResourceProcessor, Action: ChangeActionUpdate, LiveSwappable: true} + nameUpdate := Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"name"}, LiveSwappable: true} + connUpdate := Change{Resource: ResourceConnector, Action: ChangeActionUpdate, LiveSwappable: false} + dlqUpdate := Change{Resource: ResourcePipeline, Action: ChangeActionUpdate, ConfigPaths: []string{"dlq"}, LiveSwappable: false} + + testCases := []struct { + name string + diff Diff + want bool + }{ + {"empty", Diff{}, false}, + {"single processor update", Diff{Changes: []Change{procUpdate}}, true}, + {"processor + name updates", Diff{Changes: []Change{procUpdate, nameUpdate}}, true}, + {"single connector update", Diff{Changes: []Change{connUpdate}}, false}, + {"processor + connector (mixed)", Diff{Changes: []Change{procUpdate, connUpdate}}, false}, + {"processor + dlq (mixed)", Diff{Changes: []Change{procUpdate, dlqUpdate}}, false}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + is := is.New(t) + is.Equal(tc.diff.LiveEligible(), tc.want) + }) + } +}