From 765f4b03eecc849a2054bd2245f2766e6e5af1fb Mon Sep 17 00:00:00 2001 From: Devaris Date: Mon, 13 Jul 2026 12:44:58 -0700 Subject: [PATCH 1/2] fix(processor,provisioning): live in-place apply commits running-processor config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-place hot-reload path (PR1, #2613) never worked end-to-end: applyInPlace committed the new processor config through processor.Service.Update, whose running-instance guard (ErrProcessorRunning) refused it, so ApplyPlanLive returned the error and lifecycle.ReconfigureProcessor — the actual live swap — was never reached. PR1's mock-based tests returned success for the mocked Update and hid it; PR2's real-engine integration test exposed it (found in the #2616 review). Fix (Option A): add processor.Service.UpdateWhileRunning — Update minus the running guard — and route provisioning's updateProcessorAction through it. This is safe for both provisioning paths: the restart path applies it on a stopped pipeline (processor not running, so equivalent to Update), and the in-place path immediately swaps the live node via ReconfigureProcessor to match the committed config. Update keeps its guard for direct API/orchestrator callers, who do NOT swap the node (verified: pkg/orchestrator still uses Update). Regression test: TestService_UpdateWhileRunning_BypassesRunningGuard — Update refuses a running instance (ErrProcessorRunning); UpdateWhileRunning commits the config. The provisioning apply/import tests now assert the engine calls UpdateWhileRunning. The full real-engine end-to-end assertion lives in PR2's tightened integration test. Risk tier 1 (data path). Failure-mode analysis in the PR. Fixes the §4 in-place hot-reload blocker; unblocks PR2 (#2616). Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01B3cLokwNJYLLBpdyt3cgGr --- pkg/processor/service.go | 34 +++++++++++++++--- pkg/processor/service_test.go | 39 +++++++++++++++++++++ pkg/provisioning/apply_plan_inplace_test.go | 8 ++--- pkg/provisioning/import_actions.go | 9 ++++- pkg/provisioning/import_actions_test.go | 2 +- pkg/provisioning/interfaces.go | 7 +++- pkg/provisioning/mock/provisioning.go | 24 ++++++------- pkg/provisioning/service_test.go | 12 +++---- 8 files changed, 105 insertions(+), 30 deletions(-) diff --git a/pkg/processor/service.go b/pkg/processor/service.go index c0c48aa57..71d749101 100644 --- a/pkg/processor/service.go +++ b/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 } diff --git a/pkg/processor/service_test.go b/pkg/processor/service_test.go index 7ff54224f..38e1c66a0 100644 --- a/pkg/processor/service_test.go +++ b/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() diff --git a/pkg/provisioning/apply_plan_inplace_test.go b/pkg/provisioning/apply_plan_inplace_test.go index 5a4b7c634..2554da6c2 100644 --- a/pkg/provisioning/apply_plan_inplace_test.go +++ b/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) diff --git a/pkg/provisioning/import_actions.go b/pkg/provisioning/import_actions.go index 0e4afd94f..536fcf8b7 100644 --- a/pkg/provisioning/import_actions.go +++ b/pkg/provisioning/import_actions.go @@ -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, diff --git a/pkg/provisioning/import_actions_test.go b/pkg/provisioning/import_actions_test.go index 9f0b9c125..1621773ce 100644 --- a/pkg/provisioning/import_actions_test.go +++ b/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, diff --git a/pkg/provisioning/interfaces.go b/pkg/provisioning/interfaces.go index ad21a18f8..8ca5a2af8 100644 --- a/pkg/provisioning/interfaces.go +++ b/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 } diff --git a/pkg/provisioning/mock/provisioning.go b/pkg/provisioning/mock/provisioning.go index f2e448cb5..75a9b0802 100644 --- a/pkg/provisioning/mock/provisioning.go +++ b/pkg/provisioning/mock/provisioning.go @@ -907,41 +907,41 @@ func (c *ProcessorServiceMakeRunnableProcessorCall) DoAndReturn(f func(context.C return c } -// Update mocks base method. -func (m *ProcessorService) Update(ctx context.Context, id, plugin string, cfg processor.Config) (*processor.Instance, error) { +// UpdateWhileRunning mocks base method. +func (m *ProcessorService) UpdateWhileRunning(ctx context.Context, id, plugin string, cfg processor.Config) (*processor.Instance, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Update", ctx, id, plugin, cfg) + ret := m.ctrl.Call(m, "UpdateWhileRunning", ctx, id, plugin, cfg) ret0, _ := ret[0].(*processor.Instance) ret1, _ := ret[1].(error) return ret0, ret1 } -// Update indicates an expected call of Update. -func (mr *ProcessorServiceMockRecorder) Update(ctx, id, plugin, cfg any) *ProcessorServiceUpdateCall { +// UpdateWhileRunning indicates an expected call of UpdateWhileRunning. +func (mr *ProcessorServiceMockRecorder) UpdateWhileRunning(ctx, id, plugin, cfg any) *ProcessorServiceUpdateWhileRunningCall { mr.mock.ctrl.T.Helper() - call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Update", reflect.TypeOf((*ProcessorService)(nil).Update), ctx, id, plugin, cfg) - return &ProcessorServiceUpdateCall{Call: call} + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateWhileRunning", reflect.TypeOf((*ProcessorService)(nil).UpdateWhileRunning), ctx, id, plugin, cfg) + return &ProcessorServiceUpdateWhileRunningCall{Call: call} } -// ProcessorServiceUpdateCall wrap *gomock.Call -type ProcessorServiceUpdateCall struct { +// ProcessorServiceUpdateWhileRunningCall wrap *gomock.Call +type ProcessorServiceUpdateWhileRunningCall struct { *gomock.Call } // Return rewrite *gomock.Call.Return -func (c *ProcessorServiceUpdateCall) Return(arg0 *processor.Instance, arg1 error) *ProcessorServiceUpdateCall { +func (c *ProcessorServiceUpdateWhileRunningCall) Return(arg0 *processor.Instance, arg1 error) *ProcessorServiceUpdateWhileRunningCall { c.Call = c.Call.Return(arg0, arg1) return c } // Do rewrite *gomock.Call.Do -func (c *ProcessorServiceUpdateCall) Do(f func(context.Context, string, string, processor.Config) (*processor.Instance, error)) *ProcessorServiceUpdateCall { +func (c *ProcessorServiceUpdateWhileRunningCall) Do(f func(context.Context, string, string, processor.Config) (*processor.Instance, error)) *ProcessorServiceUpdateWhileRunningCall { c.Call = c.Call.Do(f) return c } // DoAndReturn rewrite *gomock.Call.DoAndReturn -func (c *ProcessorServiceUpdateCall) DoAndReturn(f func(context.Context, string, string, processor.Config) (*processor.Instance, error)) *ProcessorServiceUpdateCall { +func (c *ProcessorServiceUpdateWhileRunningCall) DoAndReturn(f func(context.Context, string, string, processor.Config) (*processor.Instance, error)) *ProcessorServiceUpdateWhileRunningCall { c.Call = c.Call.DoAndReturn(f) return c } diff --git a/pkg/provisioning/service_test.go b/pkg/provisioning/service_test.go index 15f926753..7442f2189 100644 --- a/pkg/provisioning/service_test.go +++ b/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) From 97807de3a5077e1b3f0395afd4f89139b70bd6fa Mon Sep 17 00:00:00 2001 From: Devaris Date: Mon, 13 Jul 2026 12:52:37 -0700 Subject: [PATCH 2/2] docs: fix stale godoc reference (Update -> UpdateWhileRunning) in updateProcessorAction Review NIT on #2617. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01B3cLokwNJYLLBpdyt3cgGr --- pkg/provisioning/import_actions.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/provisioning/import_actions.go b/pkg/provisioning/import_actions.go index 536fcf8b7..1e7cf278b 100644 --- a/pkg/provisioning/import_actions.go +++ b/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,