fix: live in-place apply commits running-processor config (§4 blocker, Tier 1)

pkg/processor/service.go

@@ -177,14 +177,39 @@ func (s *Service) Update(ctx context.Context, id string, plugin string, cfg Conf
 	if err != nil {
 		return nil, err
 	}
-	if plugin == "" {
-		return nil, cerrors.Errorf("could not update processor instance (ID: %s): plugin name is empty", id)
-	}
 
+	// A running processor's stored config must not be changed out from under its
+	// live node — that would leave the node and the store divergent. Ordinary
+	// 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 {
 		return nil, cerrors.Errorf("could not update processor instance (ID: %s): %w", id, ErrProcessorRunning)
 	}
 
+	return s.updateConfig(ctx, instance, plugin, cfg)
+}
+
+// UpdateWhileRunning updates a processor's stored config WITHOUT Update's
+// running-instance guard. It exists solely for provisioning's live in-place
+// reconfigure path (provisioning.applyInPlace -> lifecycle.ReconfigureProcessor):
+// updating a running processor's stored config is safe ONLY when the caller
+// immediately swaps the live node to the new config, so this method must never be
+// called except paired with that swap. Every other caller uses Update, which
+// refuses a running instance.
+func (s *Service) UpdateWhileRunning(ctx context.Context, id string, plugin string, cfg Config) (*Instance, error) {
+	instance, err := s.Get(ctx, id)
+	if err != nil {
+		return nil, err
+	}
+	return s.updateConfig(ctx, instance, plugin, cfg)
+}
+
+func (s *Service) updateConfig(ctx context.Context, instance *Instance, plugin string, cfg Config) (*Instance, error) {
+	if plugin == "" {
+		return nil, cerrors.Errorf("could not update processor instance (ID: %s): plugin name is empty", instance.ID)
+	}
+
 	if instance.Plugin != plugin {
 		s.logger.Warn(ctx).Msgf("processor plugin changing from %v to %v, "+
 			"this may lead to unexpected behavior and configuration issues.", instance.Plugin, plugin)
@@ -195,8 +220,7 @@ func (s *Service) Update(ctx context.Context, id string, plugin string, cfg Conf
 	instance.UpdatedAt = time.Now()
 
 	// persist instance
-	err = s.store.Set(ctx, instance.ID, instance)
-	if err != nil {
+	if err := s.store.Set(ctx, instance.ID, instance); err != nil {
 		return nil, err
 	}
 

pkg/processor/service_test.go

@@ -407,6 +407,45 @@ func TestService_Update_Success(t *testing.T) {
 	is.Equal(newProcType, got.Plugin)
 }
 
+// TestService_UpdateWhileRunning_BypassesRunningGuard is the regression test for
+// the live in-place hot-reload bug: Update refuses a RUNNING processor
+// (ErrProcessorRunning), which is exactly the guard that made
+// provisioning.applyInPlace fail before it committed the new config through
+// UpdateWhileRunning instead — so ReconfigureProcessor was never reached and
+// in-place hot-reload never worked. UpdateWhileRunning must bypass that guard (it
+// is paired with an immediate node swap) and commit the config.
+func TestService_UpdateWhileRunning_BypassesRunningGuard(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+	db := &inmemory.DB{}
+
+	procType := "processor-type"
+	p := proc_mock.NewProcessor(gomock.NewController(t))
+	p.EXPECT().Teardown(gomock.Any()).Return(nil).AnyTimes()
+	registry := newPluginService(t, map[string]sdk.Processor{procType: p})
+	service := NewService(log.Nop(), db, registry)
+
+	inst, err := service.Create(ctx, uuid.NewString(), procType, Parent{}, Config{}, ProvisionTypeAPI, "")
+	is.NoErr(err)
+	inst.running = true // as it is inside a running pipeline
+
+	newConfig := Config{Settings: map[string]string{"k": "v"}}
+
+	// Update refuses a running instance — the guard direct API callers rely on.
+	_, err = service.Update(ctx, inst.ID, procType, newConfig)
+	is.True(err != nil)
+	is.True(cerrors.Is(err, ErrProcessorRunning))
+
+	// UpdateWhileRunning bypasses it and commits the new config.
+	got, err := service.UpdateWhileRunning(ctx, inst.ID, procType, newConfig)
+	is.NoErr(err)
+	is.Equal(newConfig, got.Config)
+
+	reread, err := service.Get(ctx, inst.ID)
+	is.NoErr(err)
+	is.Equal(newConfig, reread.Config)
+}
+
 func TestService_Update_NonExistentProcessor(t *testing.T) {
 	is := is.New(t)
 	ctx := context.Background()

pkg/provisioning/apply_plan_inplace_test.go

@@ -63,7 +63,7 @@ func TestApplyPlanLive_ProcessorUpdate_AppliesInPlace_NoRestart(t *testing.T) {
 	old, desired := processorUpdateFixture()
 	expectExportRunning(pipSrv, connSrv, procSrv, old)
 
-	procSrv.EXPECT().Update(gomock.Any(), "p1:proc:1", "builtin:field.set", gomock.Any()).
+	procSrv.EXPECT().UpdateWhileRunning(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)
 
@@ -92,7 +92,7 @@ func TestApplyPlanLive_ProcessorUpdate_InPlace_CompletesDespiteCallerCancel(t *t
 
 	// 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()).
+	procSrv.EXPECT().UpdateWhileRunning(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
@@ -127,7 +127,7 @@ func TestApplyPlanLive_ProcessorUpdate_NotLiveReconfigurable_FallsBackToRestart(
 	// 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()).
+	procSrv.EXPECT().UpdateWhileRunning(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)
@@ -168,7 +168,7 @@ func TestApplyPlanLive_ProcessorUpdate_OpenFails_RollsBack(t *testing.T) {
 	// 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()).
+	procSrv.EXPECT().UpdateWhileRunning(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)

pkg/provisioning/import_actions.go

@@ -478,7 +478,7 @@ func (a updateProcessorAction) String() string {
 // processors have no immutable-field classification (see
 // prepareProcessorActions — "all parts of a processor are updateable"), so
 // every processor change, including Plugin, runs through
-// processorService.Update rather than a delete+create pair.
+// processorService.UpdateWhileRunning rather than a delete+create pair.
 func (a updateProcessorAction) Describe() Change {
 	return Change{
 		Resource:    ResourceProcessor,
@@ -499,7 +499,14 @@ func (a updateProcessorAction) Rollback(ctx context.Context) error {
 }
 
 func (a updateProcessorAction) update(ctx context.Context, cfg config.Processor) error {
-	_, err := a.processorService.Update(
+	// UpdateWhileRunning, not Update: provisioning applies this action either on a
+	// stopped pipeline (the restart path, where the processor isn't running so the
+	// two are equivalent) or as part of a live in-place reconfigure, where the
+	// processor IS running and applyInPlace immediately swaps the node to match
+	// via lifecycle.ReconfigureProcessor. Update's running-instance guard exists
+	// for direct API callers who do NOT swap the node; it must not block the
+	// provisioning path, which always does. See processor.Service.UpdateWhileRunning.
+	_, err := a.processorService.UpdateWhileRunning(
 		ctx,
 		cfg.ID,
 		cfg.Plugin,

pkg/provisioning/import_actions_test.go

@@ -577,7 +577,7 @@ func TestUpdateProcessorAction(t *testing.T) {
 			}
 
 			connSrv := mock.NewProcessorService(ctrl)
-			connSrv.EXPECT().Update(ctx, haveCfg.ID, haveCfg.Plugin, wantCfg).Return(instance, nil)
+			connSrv.EXPECT().UpdateWhileRunning(ctx, haveCfg.ID, haveCfg.Plugin, wantCfg).Return(instance, nil)
 
 			a := updateProcessorAction{
 				oldConfig:        tc.oldConfig,

pkg/provisioning/interfaces.go

@@ -63,7 +63,12 @@ type ProcessorService interface {
 		condition string,
 	) (*processor.Instance, error)
 	MakeRunnableProcessor(ctx context.Context, i *processor.Instance) (*processor.RunnableProcessor, error)
-	Update(ctx context.Context, id string, plugin string, cfg processor.Config) (*processor.Instance, error)
+	// UpdateWhileRunning updates a processor's stored config without the
+	// running-instance guard Update enforces — provisioning applies processor
+	// updates either on a stopped pipeline or as a live in-place reconfigure that
+	// immediately swaps the node, so the guard (meant for direct API callers who
+	// do not swap) must not block it. See processor.Service.UpdateWhileRunning.
+	UpdateWhileRunning(ctx context.Context, id string, plugin string, cfg processor.Config) (*processor.Instance, error)
 	Delete(ctx context.Context, id string) error
 }
 

pkg/provisioning/service_test.go

@@ -193,8 +193,8 @@ func TestService_Init_Update(t *testing.T) {
 	pipelineService.EXPECT().Update(anyCtx, p1.P1.ID, p1.P1.Config).Return(oldPipelineInstance, nil)
 	pipelineService.EXPECT().UpdateDLQ(anyCtx, p1.P1.ID, p1.P1.DLQ)
 	connService.EXPECT().Update(anyCtx, p1.P1C1.ID, p1.P1C1.Plugin, p1.P1C1.Config).Return(oldConnector1Instance, nil)
-	procService.EXPECT().Update(anyCtx, p1.P1C2P1.ID, p1.P1C2P1.Plugin, p1.P1C2P1.Config)
-	procService.EXPECT().Update(anyCtx, p1.P1P1.ID, p1.P1P1.Plugin, p1.P1P1.Config)
+	procService.EXPECT().UpdateWhileRunning(anyCtx, p1.P1C2P1.ID, p1.P1C2P1.Plugin, p1.P1C2P1.Config)
+	procService.EXPECT().UpdateWhileRunning(anyCtx, p1.P1P1.ID, p1.P1P1.Plugin, p1.P1P1.Config)
 
 	// start pipeline
 	lifecycleService.EXPECT().Start(anyCtx, p1.P1.ID)
@@ -316,13 +316,13 @@ func TestService_Init_RollbackUpdate(t *testing.T) {
 	pipelineService.EXPECT().Update(anyCtx, p1.P1.ID, p1.P1.Config).Return(oldPipelineInstance, nil)
 	pipelineService.EXPECT().UpdateDLQ(anyCtx, p1.P1.ID, p1.P1.DLQ)
 	connService.EXPECT().Update(anyCtx, p1.P1C1.ID, p1.P1C1.Plugin, p1.P1C1.Config).Return(oldConnector1Instance, nil)
-	procService.EXPECT().Update(anyCtx, p1.P1C2P1.ID, p1.P1C2P1.Plugin, p1.P1C2P1.Config)
+	procService.EXPECT().UpdateWhileRunning(anyCtx, p1.P1C2P1.ID, p1.P1C2P1.Plugin, p1.P1C2P1.Config)
 	wantErr := cerrors.New("err")
-	procService.EXPECT().Update(anyCtx, p1.P1P1.ID, p1.P1P1.Plugin, p1.P1P1.Config).Return(nil, wantErr) // fails
+	procService.EXPECT().UpdateWhileRunning(anyCtx, p1.P1P1.ID, p1.P1P1.Plugin, p1.P1P1.Config).Return(nil, wantErr) // fails
 
 	// rollback changes
-	procService.EXPECT().Update(anyCtx, oldPipelineProcessorInstance.ID, oldPipelineProcessorInstance.Plugin, oldPipelineProcessorInstance.Config)
-	procService.EXPECT().Update(anyCtx, oldConnectorProcessorInstance.ID, oldPipelineProcessorInstance.Plugin, oldConnectorProcessorInstance.Config)
+	procService.EXPECT().UpdateWhileRunning(anyCtx, oldPipelineProcessorInstance.ID, oldPipelineProcessorInstance.Plugin, oldPipelineProcessorInstance.Config)
+	procService.EXPECT().UpdateWhileRunning(anyCtx, oldConnectorProcessorInstance.ID, oldPipelineProcessorInstance.Plugin, oldConnectorProcessorInstance.Config)
 	connService.EXPECT().Update(anyCtx, oldConnector1Instance.ID, oldConnector1Instance.Plugin, oldConnector1Instance.Config).Return(oldConnector1Instance, nil)
 	pipelineService.EXPECT().Update(anyCtx, oldPipelineInstance.ID, oldPipelineInstance.Config).Return(oldPipelineInstance, nil)
 	pipelineService.EXPECT().UpdateDLQ(anyCtx, oldPipelineInstance.ID, oldPipelineInstance.DLQ)