fix(engine): complete live in-place processor reconfigure and report its true mode

pkg/lifecycle/reconfigure.go

@@ -73,7 +73,7 @@ func (s *Service) ReconfigureProcessor(ctx context.Context, pipelineID, processo
 	if err != nil {
 		return cerrors.Errorf("could not fetch processor %q for live reconfigure: %w", processorID, err)
 	}
-	runnableProc, err := s.processors.MakeRunnableProcessor(ctx, instance)
+	runnableProc, err := s.processors.MakeRunnableProcessorForReconfigure(ctx, instance)
 	if err != nil {
 		return cerrors.Errorf("could not build runnable processor %q for live reconfigure: %w", processorID, err)
 	}

pkg/lifecycle/service.go

@@ -133,6 +133,11 @@ type ConnectorService interface {
 type ProcessorService interface {
 	Get(ctx context.Context, id string) (*processor.Instance, error)
 	MakeRunnableProcessor(ctx context.Context, i *processor.Instance) (*processor.RunnableProcessor, error)
+	// MakeRunnableProcessorForReconfigure builds a runnable for an
+	// already-running instance without the running guard — for the live in-place
+	// reconfigure swap (ReconfigureProcessor). See
+	// processor.Service.MakeRunnableProcessorForReconfigure.
+	MakeRunnableProcessorForReconfigure(ctx context.Context, i *processor.Instance) (*processor.RunnableProcessor, error)
 }
 
 // ConnectorPluginService can create a connector plugin dispenser.

pkg/lifecycle/service_test.go

@@ -1107,6 +1107,10 @@ func (s testProcessorService) MakeRunnableProcessor(context.Context, *processor.
 	return nil, cerrors.New("not implemented")
 }
 
+func (s testProcessorService) MakeRunnableProcessorForReconfigure(context.Context, *processor.Instance) (*processor.RunnableProcessor, error) {
+	return nil, cerrors.New("not implemented")
+}
+
 func (s testProcessorService) Get(_ context.Context, id string) (*processor.Instance, error) {
 	proc, ok := s[id]
 	if !ok {

pkg/lifecycle/stream/processor.go

@@ -282,7 +282,11 @@ func (n *ProcessorNode) applyPendingSwap(ctx context.Context) {
 		// 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 {
+		// reconfigureTeardown, not Teardown: the current (old) processor and this
+		// failed new one share the same processor.Instance, so clearing the
+		// instance's running flag here would mark the STILL-running old processor
+		// stopped. See teardownForReconfigure.
+		if tdErr := teardownForReconfigure(ctx, p.newProcessor); 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)
@@ -291,14 +295,37 @@ func (n *ProcessorNode) applyPendingSwap(ctx context.Context) {
 
 	old := n.Processor
 	n.Processor = p.newProcessor
-	if tdErr := old.Teardown(ctx); tdErr != nil {
+	// reconfigureTeardown, not Teardown: old and the new (now-live) processor
+	// share the same processor.Instance, which stays running via the new one.
+	// A plain Teardown here would clear that shared instance's running flag,
+	// wrongly marking the live processor stopped and disarming the Update/Delete
+	// guards. Only a real pipeline stop clears running (via the node's normal
+	// Teardown of its current processor).
+	if tdErr := teardownForReconfigure(ctx, old); 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
 }
 
+// teardownForReconfigure tears down proc during a live swap without clearing the
+// running state of a shared processor Instance. The built-in RunnableProcessor
+// embeds a single *Instance shared across the runnables of a swap, so tearing one
+// down with the plain Teardown would clear the instance's running flag even though
+// the instance stays running via the swap's other runnable. A Processor that needs
+// this distinction implements TeardownForReconfigure; any other implementation
+// (e.g. a test mock) falls back to the plain Teardown, which is correct for
+// processors that don't share instance state across a swap.
+func teardownForReconfigure(ctx context.Context, proc Processor) error {
+	if rp, ok := proc.(interface {
+		TeardownForReconfigure(context.Context) error
+	}); ok {
+		return rp.TeardownForReconfigure(ctx)
+	}
+	return proc.Teardown(ctx)
+}
+
 // 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

pkg/lifecycle/stream/processor_reconfigure_test.go

@@ -198,3 +198,130 @@ func TestProcessorNode_Reconfigure_ContextCancelled(t *testing.T) {
 	err := n.Reconfigure(ctx, mock.NewProcessor(ctrl))
 	is.True(cerrors.Is(err, context.Canceled))
 }
+
+// teardownSpy is a Processor that records whether it was torn down via the plain
+// Teardown (which, for the real RunnableProcessor, clears the shared Instance's
+// running flag) or via TeardownForReconfigure (which does not). It lets the swap
+// tests below assert that applyPendingSwap never uses the running-clearing
+// Teardown on a processor that is being swapped while its instance stays running.
+type teardownSpy struct {
+	openErr error
+
+	mu            sync.Mutex
+	teardownCalls int
+	reconfigCalls int
+}
+
+func (s *teardownSpy) Open(context.Context) error { return s.openErr }
+
+func (s *teardownSpy) Process(_ context.Context, recs []opencdc.Record) []sdk.ProcessedRecord {
+	out := make([]sdk.ProcessedRecord, len(recs))
+	for i, r := range recs {
+		out[i] = sdk.SingleRecord(r)
+	}
+	return out
+}
+
+func (s *teardownSpy) Teardown(context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.teardownCalls++
+	return nil
+}
+
+func (s *teardownSpy) TeardownForReconfigure(context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.reconfigCalls++
+	return nil
+}
+
+func (s *teardownSpy) counts() (teardown, reconfig int) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.teardownCalls, s.reconfigCalls
+}
+
+// TestProcessorNode_Reconfigure_SuccessfulSwap_UsesReconfigureTeardown proves the
+// swap tears the OLD processor down via TeardownForReconfigure (not the running
+// -clearing Teardown), and that a real pipeline STOP still uses the plain Teardown
+// on the node's current processor. This is the wiring half of the shared-instance
+// running-flag regression (the processor-package half lives in
+// TestReconfigureSwap_KeepsInstanceRunning_GuardsStayArmed).
+func TestProcessorNode_Reconfigure_SuccessfulSwap_UsesReconfigureTeardown(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+
+	old := &teardownSpy{}
+	newp := &teardownSpy{}
+	n := &ProcessorNode{Name: "test", Processor: old, 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))
+	}()
+
+	is.NoErr(n.Reconfigure(ctx, newp)) // blocks until the swap is applied
+
+	oldTd, oldRc := old.counts()
+	is.Equal(oldTd, 0) // the running-clearing Teardown must NOT run on a swap
+	is.Equal(oldRc, 1) // the old processor was torn down via TeardownForReconfigure
+
+	close(in)
+	wg.Wait()
+
+	newTd, newRc := newp.counts()
+	is.Equal(newTd, 1) // a real stop uses the plain, running-clearing Teardown
+	is.Equal(newRc, 0)
+
+	_, ok := <-out
+	is.Equal(false, ok)
+}
+
+// TestProcessorNode_Reconfigure_OpenFailure_UsesReconfigureTeardown proves that
+// when the NEW processor fails to open, its best-effort cleanup also uses
+// TeardownForReconfigure — the old processor (sharing the instance) keeps running,
+// so its running flag must not be cleared by tearing down the failed new one.
+func TestProcessorNode_Reconfigure_OpenFailure_UsesReconfigureTeardown(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+
+	old := &teardownSpy{}
+	newp := &teardownSpy{openErr: cerrors.New("bad new config")}
+	n := &ProcessorNode{Name: "test", Processor: old, 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))
+	}()
+
+	err := n.Reconfigure(ctx, newp)
+	is.True(err != nil) // open failed, swap rejected
+
+	newTd, newRc := newp.counts()
+	is.Equal(newTd, 0) // failed-new cleanup must not use the running-clearing Teardown
+	is.Equal(newRc, 1) // it used TeardownForReconfigure
+
+	oldTd, oldRc := old.counts()
+	is.Equal(oldTd, 0) // old kept running, untouched by the failed swap
+	is.Equal(oldRc, 0)
+
+	close(in)
+	wg.Wait()
+
+	oldTd2, _ := old.counts()
+	is.Equal(oldTd2, 1) // stop tears the still-current old processor down fully
+
+	_, ok := <-out
+	is.Equal(false, ok)
+}

pkg/processor/instance.go

@@ -18,6 +18,7 @@ package processor
 
 import (
 	"context"
+	"sync/atomic"
 	"time"
 
 	"github.com/conduitio/conduit/pkg/foundation/log"
@@ -62,7 +63,13 @@ type Instance struct {
 	// before the processor is actually running.
 	inInsp  *inspector.Inspector
 	outInsp *inspector.Inspector
-	running bool
+	// running guards the instance against a concurrent Update/Delete while it is
+	// live in a pipeline. It is read on the API goroutine (Service.Update/Delete)
+	// and written on the pipeline's Run goroutine (RunnableProcessor.Teardown at
+	// stop), so it must be accessed atomically. MakeRunnableProcessor uses
+	// CompareAndSwap to reserve it, which also makes the "refuse if already
+	// running" guard a single atomic test-and-set rather than a racy read+write.
+	running atomic.Bool
 }
 
 func (i *Instance) init(logger log.CtxLogger) {

pkg/processor/runnable_processor.go

@@ -142,6 +142,20 @@ func (p *RunnableProcessor) Process(ctx context.Context, records []opencdc.Recor
 
 func (p *RunnableProcessor) Teardown(ctx context.Context) error {
 	err := p.proc.Teardown(ctx)
-	p.running = false
+	p.running.Store(false)
 	return err
 }
+
+// TeardownForReconfigure tears down the processor plugin WITHOUT clearing the
+// shared Instance.running flag. It exists for the live in-place reconfigure swap
+// (ProcessorNode.applyPendingSwap), where a RunnableProcessor is torn down but
+// the underlying Instance stays running via the other runnable in the swap: on
+// success the new runnable takes over; on a failed open the old runnable keeps
+// running. Because both runnables embed the SAME *Instance, the plain Teardown's
+// `running = false` would wrongly mark the still-running instance stopped —
+// disarming the Update/Delete running-guards (service.go) for the rest of the
+// pipeline's life. Only a real pipeline stop, which tears down the node's
+// current processor via the plain Teardown, may clear running.
+func (p *RunnableProcessor) TeardownForReconfigure(ctx context.Context) error {
+	return p.proc.Teardown(ctx)
+}

pkg/processor/service.go

@@ -102,20 +102,46 @@ func (s *Service) Get(_ context.Context, id string) (*Instance, error) {
 }
 
 func (s *Service) MakeRunnableProcessor(ctx context.Context, i *Instance) (*RunnableProcessor, error) {
-	if i.running {
+	// Reserve the instance as running with a single atomic test-and-set: this both
+	// enforces the guard (refuse an already-running instance) and closes the
+	// read-then-write race two concurrent callers would otherwise have. On any
+	// subsequent build failure we release the reservation so the instance is not
+	// left wrongly marked running.
+	if !i.running.CompareAndSwap(false, true) {
 		return nil, ErrProcessorRunning
 	}
 
 	p, err := s.registry.NewProcessor(ctx, i.Plugin, i.ID)
 	if err != nil {
+		i.running.Store(false)
 		return nil, err
 	}
 	cond, err := newProcessorCondition(i.Condition)
 	if err != nil {
+		i.running.Store(false)
 		return nil, cerrors.Errorf("invalid condition: %w", err)
 	}
 
-	i.running = true
+	return newRunnableProcessor(p, cond, i), nil
+}
+
+// MakeRunnableProcessorForReconfigure builds a fresh RunnableProcessor for an
+// ALREADY-RUNNING processor instance, for the live in-place reconfigure swap
+// (lifecycle.ReconfigureProcessor). Unlike MakeRunnableProcessor it does not
+// refuse a running instance and does not flip the running flag: the instance
+// legitimately stays running across the swap, and the caller opens this new
+// runnable before tearing down the old one (open-before-teardown), so no record
+// is dropped. It dispenses a new plugin from the instance's current (already
+// updated) config, exactly as MakeRunnableProcessor does for a fresh start.
+func (s *Service) MakeRunnableProcessorForReconfigure(ctx context.Context, i *Instance) (*RunnableProcessor, error) {
+	p, err := s.registry.NewProcessor(ctx, i.Plugin, i.ID)
+	if err != nil {
+		return nil, err
+	}
+	cond, err := newProcessorCondition(i.Condition)
+	if err != nil {
+		return nil, cerrors.Errorf("invalid condition: %w", err)
+	}
 	return newRunnableProcessor(p, cond, i), nil
 }
 
@@ -183,7 +209,7 @@ func (s *Service) Update(ctx context.Context, id string, plugin string, cfg Conf
 	// callers (the orchestrator / HTTP+gRPC API) are refused here; the sole
 	// exception is provisioning's live in-place reconfigure, which uses
 	// UpdateWhileRunning and immediately swaps the node to match.
-	if instance.running {
+	if instance.running.Load() {
 		return nil, cerrors.Errorf("could not update processor instance (ID: %s): %w", id, ErrProcessorRunning)
 	}
 
@@ -235,7 +261,7 @@ func (s *Service) Delete(ctx context.Context, id string) error {
 		return err
 	}
 
-	if instance.running {
+	if instance.running.Load() {
 		return cerrors.Errorf("could not delete processor instance (ID: %s): %w", id, ErrProcessorRunning)
 	}