From 90a7b093df5f4b635d29b14ba8cd3660eea0e6e3 Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Mon, 10 Aug 2026 08:59:41 -0700 Subject: [PATCH 1/2] revert(plugin): drop the per-call plugin deadline The retry-aware watchdog work bounded every watched plugin call by a deadline the agent supplied through the operator's spawn environment. Removing it again, and keeping the rest of that work. The deadline never fixed the reported failure. Healthy resources were declared missing in action because one plugin RPC blocked inside its own retry loop far longer than the status-poll interval, starving the loop that produces heartbeats. Bounding that retry in the plugin is what fixed it; the agent-side deadline only enforced the invariant the plugin had violated. The two workflow regression tests for the reported bug pass without it. What it cost was disproportionate. A deadline is not an additive interface change a plugin author can ignore, the way an added request field is. It is a cross-cutting obligation that applies inside every method, is violated by omission rather than commission, fails silently, and cannot be verified by reading your own code. It also cannot be enforced: Go has no goroutine preemption, so a plugin that does not thread the context still blocks, and the agent still waits. Paying that much interface surface for a guarantee that is not actually a guarantee is the wrong trade, especially as plugins are increasingly written by agents that will implement an interface faithfully and observe a prose invariant unreliably. The watchdog window keeps a per-call allowance, now named for what it is: the silence the agent tolerates for one call, matching the longest the updater itself waits for a reply, rather than a bound anything enforces. Kept from the same work: the window derived from the operator's actual retry cadence, the fix for status calls terminating without sending a final progress, and sizing the window from the per-plugin config the operator was really spawned with. The test that a failed status call never reissues the original mutation is kept and narrowed to the recoverable path, which is the only one that can reschedule. --- .../plugin_coordinator/plugin_coordinator.go | 2 - .../plugin_coordinator_test.go | 7 +- .../missing_in_action_timeout_test.go | 34 +- .../resource_update/resource_updater.go | 25 +- .../local/missing_in_action_test.go | 10 +- .../local/supervision_cascade_test.go | 10 +- pkg/plugin/plugin_operator.go | 109 +-- pkg/plugin/plugin_operator_deadline_test.go | 653 ------------------ pkg/plugin/plugin_operator_double_test.go | 306 ++++++++ pkg/plugin/plugin_operator_progress_test.go | 31 +- 10 files changed, 364 insertions(+), 823 deletions(-) delete mode 100644 pkg/plugin/plugin_operator_deadline_test.go create mode 100644 pkg/plugin/plugin_operator_double_test.go diff --git a/internal/metastructure/plugin_coordinator/plugin_coordinator.go b/internal/metastructure/plugin_coordinator/plugin_coordinator.go index 1bc6a3260..1a1a980ac 100644 --- a/internal/metastructure/plugin_coordinator/plugin_coordinator.go +++ b/internal/metastructure/plugin_coordinator/plugin_coordinator.go @@ -17,7 +17,6 @@ import ( "github.com/platform-engineering-labs/formae/internal/metastructure/canonicalize" "github.com/platform-engineering-labs/formae/internal/metastructure/changeset" "github.com/platform-engineering-labs/formae/internal/metastructure/messages" - "github.com/platform-engineering-labs/formae/internal/metastructure/resource_update" "github.com/platform-engineering-labs/formae/pkg/model" "github.com/platform-engineering-labs/formae/pkg/plugin" ) @@ -325,7 +324,6 @@ func (c *PluginCoordinator) spawnPluginOperator(req messages.SpawnPluginOperator func pluginOperatorEnv(retryConfig model.RetryConfig, requestedBy gen.PID) map[gen.Env]any { return map[gen.Env]any{ gen.Env("RetryConfig"): retryConfig, - gen.Env("PluginCallTimeout"): resource_update.PluginCallTimeout, gen.Env("RequestedBy"): requestedBy, } } diff --git a/internal/metastructure/plugin_coordinator/plugin_coordinator_test.go b/internal/metastructure/plugin_coordinator/plugin_coordinator_test.go index e6d1fd4a6..765143443 100644 --- a/internal/metastructure/plugin_coordinator/plugin_coordinator_test.go +++ b/internal/metastructure/plugin_coordinator/plugin_coordinator_test.go @@ -11,7 +11,6 @@ import ( "ergo.services/ergo/gen" "ergo.services/ergo/testing/unit" "github.com/platform-engineering-labs/formae/internal/metastructure/messages" - "github.com/platform-engineering-labs/formae/internal/metastructure/resource_update" "github.com/platform-engineering-labs/formae/internal/testplugin/fakeaws" "github.com/platform-engineering-labs/formae/pkg/model" "github.com/platform-engineering-labs/formae/pkg/plugin" @@ -235,8 +234,8 @@ func TestPluginCoordinator_SpawnReportsResolvedRetryConfig(t *testing.T) { } // TestPluginCoordinator_LocalSpawnEnvMatchesReportedConfig asserts a spawned -// operator is handed the same retry config the spawn result reports, plus the -// plugin call deadline the agent sizes its watchdog window from. It drives the +// operator is handed the same retry config the spawn result reports, which is +// the config the agent sizes its watchdog window from. It drives the // local path, which is the observable one; the remote path spawns with the same // environment. func TestPluginCoordinator_LocalSpawnEnvMatchesReportedConfig(t *testing.T) { @@ -261,6 +260,4 @@ func TestPluginCoordinator_LocalSpawnEnvMatchesReportedConfig(t *testing.T) { require.NotNil(t, env, "the coordinator must have spawned a plugin operator") assert.Equal(t, *spawned.RetryConfig, env[gen.Env("RetryConfig")], "the operator must poll on the config the result reports") - assert.Equal(t, resource_update.PluginCallTimeout, env[gen.Env("PluginCallTimeout")], - "the operator must bound its plugin calls by the deadline the watchdog window is built from") } diff --git a/internal/metastructure/resource_update/missing_in_action_timeout_test.go b/internal/metastructure/resource_update/missing_in_action_timeout_test.go index f60059f0a..cfec0257f 100644 --- a/internal/metastructure/resource_update/missing_in_action_timeout_test.go +++ b/internal/metastructure/resource_update/missing_in_action_timeout_test.go @@ -40,7 +40,7 @@ func TestMissingInActionTimeout_CoversLongestOperatorSleep(t *testing.T) { strategy := resource.RetryStrategy{MaxRetries: cfg.MaxRetries, BaseDelay: cfg.RetryDelay} longestSleep := max(cfg.StatusCheckInterval, cfg.RetryDelay, strategy.Backoff(cfg.MaxRetries+1)) - assert.Equal(t, longestSleep+PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg)) + assert.Equal(t, longestSleep+PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg)) assert.Equal(t, 100*time.Second, missingInActionTimeout(cfg), "the shipped defaults must yield a 100s window") assert.Greater(t, missingInActionTimeout(cfg), 2*cfg.StatusCheckInterval, @@ -62,7 +62,7 @@ func TestMissingInActionTimeout_DrivenByFlatRetryDelay(t *testing.T) { require.Greater(t, retryDelay, strategy.Backoff(cfg.MaxRetries+1), "this case only bites when the flat delay outlasts the capped backoff") - assert.Equal(t, retryDelay+PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg)) + assert.Equal(t, retryDelay+PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg)) } // TestMissingInActionTimeout_DrivenByLastScheduledBackoff covers the throttling @@ -80,9 +80,9 @@ func TestMissingInActionTimeout_DrivenByLastScheduledBackoff(t *testing.T) { require.Greater(t, lastBackoff, strategy.Backoff(cfg.MaxRetries), "the last scheduled backoff must outlast the one before it") - assert.Equal(t, lastBackoff+PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg)) + assert.Equal(t, lastBackoff+PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg)) assert.Greater(t, missingInActionTimeout(cfg), - strategy.Backoff(cfg.MaxRetries)+PluginCallTimeout+missingInActionMargin, + strategy.Backoff(cfg.MaxRetries)+PluginCallAllowance+missingInActionMargin, "a window built on Backoff(MaxRetries) would fire one backoff short") } @@ -93,37 +93,33 @@ func TestMissingInActionTimeout_SmallAndDegenerateConfigs(t *testing.T) { t.Run("MaxRetriesZero", func(t *testing.T) { cfg := pkgmodel.RetryConfig{StatusCheckInterval: 2 * time.Second, MaxRetries: 0, RetryDelay: 7 * time.Second} // The single scheduled backoff is Backoff(1), which is the base delay. - assert.Equal(t, 7*time.Second+PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg)) + assert.Equal(t, 7*time.Second+PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg)) }) t.Run("MaxRetriesOne", func(t *testing.T) { cfg := pkgmodel.RetryConfig{StatusCheckInterval: 2 * time.Second, MaxRetries: 1, RetryDelay: 7 * time.Second} // Backoff(2) doubles the base delay, still under DefaultMaxBackoff. - assert.Equal(t, 14*time.Second+PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg)) + assert.Equal(t, 14*time.Second+PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg)) }) t.Run("ZeroConfig", func(t *testing.T) { - assert.Equal(t, PluginCallTimeout+missingInActionMargin, missingInActionTimeout(pkgmodel.RetryConfig{})) + assert.Equal(t, PluginCallAllowance+missingInActionMargin, missingInActionTimeout(pkgmodel.RetryConfig{})) }) t.Run("NegativeDurations", func(t *testing.T) { cfg := pkgmodel.RetryConfig{StatusCheckInterval: -5 * time.Second, MaxRetries: 0, RetryDelay: -5 * time.Second} - assert.Equal(t, PluginCallTimeout+missingInActionMargin, missingInActionTimeout(cfg), + assert.Equal(t, PluginCallAllowance+missingInActionMargin, missingInActionTimeout(cfg), "a negative duration in config must not shrink the window") }) } -// TestPluginCallTimeouts_OperatorDeadlineExpiresFirst pins both deadlines: the -// one the agent hands the operator for a single plugin call, and the one the -// agent puts on its own call to the operator. The operator's must expire first -// so its attributable failure progress wins the race, and it must equal the -// operator's compiled defaultPluginCallTimeout fallback (pkg/plugin), which it -// stands in for whenever the deadline is not supplied. -func TestPluginCallTimeouts_OperatorDeadlineExpiresFirst(t *testing.T) { - assert.Equal(t, 60*time.Second, PluginCallTimeout, - "must track the plugin operator's compiled defaultPluginCallTimeout") - assert.Equal(t, 70, PluginOperationCallTimeout, - "the agent's call timeout is the operator's deadline plus a margin, in seconds") +// TestPluginCallAllowance_MatchesTheUpdatersOwnCallTimeout pins the call +// allowance in the watchdog window to the longest the updater itself waits for +// a reply from an operator. Nothing enforces a bound inside the plugin, so the +// window has to assume the most the agent is willing to wait. +func TestPluginCallAllowance_MatchesTheUpdatersOwnCallTimeout(t *testing.T) { + assert.Equal(t, time.Duration(PluginOperationCallTimeout)*time.Second, PluginCallAllowance, + "the window's call allowance must match the updater's own call timeout") } // armingProcess is a gen.Process double for the two watchdog-arming handlers. diff --git a/internal/metastructure/resource_update/resource_updater.go b/internal/metastructure/resource_update/resource_updater.go index b7eb28aae..e363a398e 100644 --- a/internal/metastructure/resource_update/resource_updater.go +++ b/internal/metastructure/resource_update/resource_updater.go @@ -182,19 +182,18 @@ const ( StateRejected = gen.Atom("rejected") ) -// PluginCallTimeout is the deadline the agent hands each plugin operator for a -// single watched plugin call. It matches the operator's own compiled fallback, -// so the watchdog window derived from it holds whether the operator runs on the -// supplied deadline or on its fallback. Exposed as a variable so tests can -// reduce it. -var PluginCallTimeout = 60 * time.Second - // PluginOperationCallTimeout is the maximum time (in seconds) to wait for a -// plugin operator to respond to a resource operation. It outlasts -// PluginCallTimeout so the operator's own deadline expires first and its -// attributable failure progress wins the race with this call. Exposed as a -// variable so tests can reduce it. -var PluginOperationCallTimeout = int((PluginCallTimeout + 10*time.Second) / time.Second) +// plugin operator to respond to a resource operation. Exposed as a variable so +// tests can reduce it. +var PluginOperationCallTimeout = 60 + +// PluginCallAllowance is how long a single plugin call may take before the +// watchdog stops treating an operator's silence as work in progress. Nothing +// enforces it: a plugin call runs to completion inside the plugin's own +// process, so this is the silence the agent tolerates, not a bound on the call. +// It matches the outer call timeout above, which is the longest the updater +// itself waits for a reply. Exposed as a variable so tests can reduce it. +var PluginCallAllowance = 60 * time.Second type ResourceUpdateData struct { resourceUpdate *ResourceUpdate @@ -563,7 +562,7 @@ const missingInActionMargin = 10 * time.Second func missingInActionTimeout(cfg pkgmodel.RetryConfig) time.Duration { strategy := resource.RetryStrategy{MaxRetries: cfg.MaxRetries, BaseDelay: cfg.RetryDelay} longestSleep := max(cfg.StatusCheckInterval, cfg.RetryDelay, strategy.Backoff(cfg.MaxRetries+1), 0) - return longestSleep + PluginCallTimeout + missingInActionMargin + return longestSleep + PluginCallAllowance + missingInActionMargin } // watchdogRetryConfig returns the config the watchdog window is derived from: diff --git a/internal/workflow_tests/local/missing_in_action_test.go b/internal/workflow_tests/local/missing_in_action_test.go index a592adc0f..6c37d1705 100644 --- a/internal/workflow_tests/local/missing_in_action_test.go +++ b/internal/workflow_tests/local/missing_in_action_test.go @@ -71,7 +71,7 @@ func TestSlowHeartbeatIsNotDeclaredMissingInAction(t *testing.T) { testutil.RunTestFromProjectRoot(t, func(t *testing.T) { const ( statusCheckInterval = 1 * time.Second - pluginCallTimeout = 5 * time.Second + pluginCallAllowance = 5 * time.Second slowStatusCall = 2500 * time.Millisecond ) @@ -80,7 +80,7 @@ func TestSlowHeartbeatIsNotDeclaredMissingInAction(t *testing.T) { "the gap must outlast the flat twice-the-interval window, or the old rule would not have fired") require.Less(t, heartbeatGap, watchdogMarginFloor, "the gap must stay inside the derived window, of which the fixed margin is only one term") - require.LessOrEqual(t, slowStatusCall, pluginCallTimeout/2, + require.LessOrEqual(t, slowStatusCall, pluginCallAllowance/2, "the slow call must leave as much headroom again inside the deadline the agent hands the "+ "operator, or a loaded runner turns this into a call that outran its deadline instead of "+ "a slow one that reported") @@ -125,9 +125,9 @@ func TestSlowHeartbeatIsNotDeclaredMissingInAction(t *testing.T) { }, } - origCallTimeout := resource_update.PluginCallTimeout - resource_update.PluginCallTimeout = pluginCallTimeout - t.Cleanup(func() { resource_update.PluginCallTimeout = origCallTimeout }) + origCallAllowance := resource_update.PluginCallAllowance + resource_update.PluginCallAllowance = pluginCallAllowance + t.Cleanup(func() { resource_update.PluginCallAllowance = origCallAllowance }) cfg := test_helpers.NewTestMetastructureConfig() cfg.Agent.Retry.StatusCheckInterval = statusCheckInterval diff --git a/internal/workflow_tests/local/supervision_cascade_test.go b/internal/workflow_tests/local/supervision_cascade_test.go index a45ac942b..14c06acc8 100644 --- a/internal/workflow_tests/local/supervision_cascade_test.go +++ b/internal/workflow_tests/local/supervision_cascade_test.go @@ -634,11 +634,11 @@ func TestPluginOperatorCrashConvergesViaTimeout(t *testing.T) { } // Shorten every term the PluginOperatorMissingInAction window is derived - // from — the operator's retry cadence and the plugin call deadline — so - // the watchdog fires quickly in the test. - origCallTimeout := resource_update.PluginCallTimeout - resource_update.PluginCallTimeout = 100 * time.Millisecond - t.Cleanup(func() { resource_update.PluginCallTimeout = origCallTimeout }) + // from, the operator's retry cadence and the per-call allowance, so the + // watchdog fires quickly in the test. + origCallAllowance := resource_update.PluginCallAllowance + resource_update.PluginCallAllowance = 100 * time.Millisecond + t.Cleanup(func() { resource_update.PluginCallAllowance = origCallAllowance }) cfg := test_helpers.NewTestMetastructureConfig() cfg.Agent.Retry.StatusCheckInterval = 1 * time.Second diff --git a/pkg/plugin/plugin_operator.go b/pkg/plugin/plugin_operator.go index 06fd69405..cc5a101e4 100644 --- a/pkg/plugin/plugin_operator.go +++ b/pkg/plugin/plugin_operator.go @@ -7,7 +7,6 @@ package plugin import ( "context" "encoding/json" - "errors" "fmt" "strings" "time" @@ -62,10 +61,6 @@ type PluginOperator struct { // PluginOperatorFactoryName is the factory name for remote spawning const PluginOperatorFactoryName = "PluginOperator" -// defaultPluginCallTimeout bounds a single watched plugin call when the agent -// supplies no 'PluginCallTimeout' environment variable. -const defaultPluginCallTimeout = 60 * time.Second - func NewPluginOperator() gen.ProcessBehavior { return &PluginOperator{} } @@ -367,10 +362,6 @@ type PluginUpdateData struct { context context.Context requestedBy gen.PID - // callTimeout bounds a single watched plugin call. The agent supplies it via - // the process environment so a plugin that retries internally still has to - // report back within a known budget. - callTimeout time.Duration // Metrics operationStartTime time.Time @@ -407,41 +398,12 @@ func (data PluginUpdateData) newUnforeseenError() TrackedProgress { } } -// callContext derives the context for a single watched plugin call, bounded by -// the deadline the agent supplied. Callers must invoke the returned cancel. -func (data PluginUpdateData) callContext() (context.Context, context.CancelFunc) { - return context.WithTimeout(data.context, data.callTimeout) -} - -// callDeadlineMessage describes a plugin call that outran its per-call deadline. -func (data PluginUpdateData) callDeadlineMessage(operation resource.Operation) string { - return fmt.Sprintf("plugin %s call exceeded its %s deadline", operation, data.callTimeout) -} - -// callDeadlineProgress builds the failure for a read-only plugin call that -// outran its per-call deadline. ServiceTimeout is recoverable, so the operation -// goes back through the retry ladder and heartbeats on every attempt. -func (data PluginUpdateData) callDeadlineProgress(check PluginOperatorCheckStatus) *resource.ProgressResult { - return &resource.ProgressResult{ - Operation: check.ResourceOperation, - OperationStatus: resource.OperationStatusFailure, - RequestID: check.RequestID, - NativeID: check.NativeID, - ErrorCode: resource.OperationErrorCodeServiceTimeout, - StatusMessage: data.callDeadlineMessage(resource.OperationCheckStatus), - } -} - // terminalCallError builds the terminal progress for a failed mutating plugin -// call. A call that outran its per-call deadline may already have reached the +// call. A create, update or delete that failed may already have reached the // provider, so it is reported as terminal rather than retried. func (data PluginUpdateData) terminalCallError(operation resource.Operation, err error) TrackedProgress { progress := data.newUnforeseenError() - if errors.Is(err, context.DeadlineExceeded) { - progress.StatusMessage = fmt.Sprintf("%s: %v", data.callDeadlineMessage(operation), err) - } else { - progress.StatusMessage = err.Error() - } + progress.StatusMessage = err.Error() return progress } @@ -463,10 +425,6 @@ func statusCheckAfterFailedCall(check PluginOperatorCheckStatus) PluginOperatorC // stays terminal. The check's identifiers are carried over because the retry // ladder rebuilds the next poll from the progress. func (data PluginUpdateData) statusCallFailure(check PluginOperatorCheckStatus, err error) *resource.ProgressResult { - if errors.Is(err, context.DeadlineExceeded) { - return data.callDeadlineProgress(check) - } - errorCode := resource.OperationErrorCodeUnforeseenError if isThrottlingError(err) { errorCode = resource.OperationErrorCodeThrottling @@ -520,19 +478,6 @@ func (o *PluginOperator) Init(args ...any) (statemachine.StateMachineSpec[Plugin } data.config = cfg.(model.RetryConfig) - // PluginCallTimeout bounds every watched plugin call. An absent, non-positive - // or non-duration value falls back to the compiled default. - data.callTimeout = defaultPluginCallTimeout - callTimeout, ok := o.Env("PluginCallTimeout") - if !ok { - callTimeout, ok = o.Node().Env("PluginCallTimeout") - } - if ok { - if timeout, isDuration := callTimeout.(time.Duration); isDuration && timeout > 0 { - data.callTimeout = timeout - } - } - // RequestedBy: the requesting ResourceUpdater's PID, threaded in via Env so // we can establish the operator→RU link asynchronously (off the operation // critical path). Linking here in Init would block, so we defer it via a @@ -646,10 +591,7 @@ func read(from gen.PID, state gen.Atom, data PluginUpdateData, operation ReadRes } proc.Log().Debug("PluginOperator: starting read operation for %s", operation.NativeID) - callCtx, cancel := data.callContext() - defer cancel() - - result, err := data.plugin.Read(callCtx, &resource.ReadRequest{ + result, err := data.plugin.Read(data.context, &resource.ReadRequest{ NativeID: operation.NativeID, ResourceType: operation.ResourceType, TargetConfig: operation.TargetConfig, @@ -658,14 +600,8 @@ func read(from gen.PID, state gen.Atom, data PluginUpdateData, operation ReadRes if err != nil { proc.Log().Debug("PluginOperator: failed to read resource: %v", err) progressResult.OperationStatus = resource.OperationStatusFailure - if errors.Is(err, context.DeadlineExceeded) { - // A read is safe to repeat, so it stays recoverable. - progressResult.ErrorCode = resource.OperationErrorCodeServiceTimeout - progressResult.StatusMessage = data.callDeadlineMessage(resource.OperationRead) - } else { - progressResult.ErrorCode = resource.OperationErrorCodeUnforeseenError - progressResult.StatusMessage = err.Error() - } + progressResult.ErrorCode = resource.OperationErrorCodeUnforeseenError + progressResult.StatusMessage = err.Error() } else if result.ErrorCode != "" && !(operation.TreatNotFoundAsSuccess() && result.ErrorCode == resource.OperationErrorCodeNotFound) { progressResult.OperationStatus = resource.OperationStatusFailure progressResult.ErrorCode = result.ErrorCode @@ -690,11 +626,8 @@ func create(from gen.PID, state gen.Atom, data PluginUpdateData, operation Creat proc.Log().Debug("PluginOperator: starting create operation for %s", operation.ResourceType) - callCtx, cancel := data.callContext() - defer cancel() - // Properties are expected to be pre-resolved by ResourceUpdater - result, err := data.plugin.Create(callCtx, &resource.CreateRequest{ + result, err := data.plugin.Create(data.context, &resource.CreateRequest{ ResourceType: operation.ResourceType, Label: operation.Label, Properties: operation.Properties, @@ -715,10 +648,7 @@ func update(from gen.PID, state gen.Atom, data PluginUpdateData, operation Updat return StateFinishedWithError, data, data.newNamespaceMismatchError(), nil, nil } - callCtx, cancel := data.callContext() - defer cancel() - - result, err := data.plugin.Update(callCtx, &resource.UpdateRequest{ + result, err := data.plugin.Update(data.context, &resource.UpdateRequest{ NativeID: operation.NativeID, ResourceType: operation.ResourceType, Label: operation.Label, @@ -742,10 +672,7 @@ func delete(from gen.PID, state gen.Atom, data PluginUpdateData, operation Delet return StateFinishedWithError, data, data.newNamespaceMismatchError(), nil, nil } - callCtx, cancel := data.callContext() - defer cancel() - - result, err := data.plugin.Delete(callCtx, &resource.DeleteRequest{ + result, err := data.plugin.Delete(data.context, &resource.DeleteRequest{ NativeID: operation.NativeID, ResourceType: operation.ResourceType, TargetConfig: operation.TargetConfig, @@ -774,10 +701,7 @@ func status(from gen.PID, state gen.Atom, data PluginUpdateData, operation Plugi proc.Log().Debug("PluginOperator: checking status of resource %s", operation.RequestID) - callCtx, cancel := data.callContext() - defer cancel() - - result, err := data.plugin.Status(callCtx, &resource.StatusRequest{ + result, err := data.plugin.Status(data.context, &resource.StatusRequest{ RequestID: operation.RequestID, NativeID: operation.NativeID, ResourceType: operation.ResourceType, @@ -859,23 +783,13 @@ func resume(from gen.PID, state gen.Atom, data PluginUpdateData, operation Resum data.attempts = operation.PreviousAttempts proc.Log().Debug("PluginOperator: resume waiting for resource %s", operation.Request.RequestID) - callCtx, cancel := data.callContext() - defer cancel() - - result, err := data.plugin.Status(callCtx, &resource.StatusRequest{ + result, err := data.plugin.Status(data.context, &resource.StatusRequest{ RequestID: operation.Request.RequestID, NativeID: operation.Request.NativeID, ResourceType: operation.Request.ResourceType, TargetConfig: operation.Request.TargetConfig, }) if err != nil { - if errors.Is(err, context.DeadlineExceeded) { - proc.Log().Error("PluginOperator: resumed status check of resource %s exceeded its %s call deadline: %v", operation.Request.RequestID, data.callTimeout, err) - // A status check is safe to repeat, so it stays recoverable and is - // rescheduled as another check. - check := statusCheckAfterFailedCall(operation.Request) - return handlePluginResult(data, check, proc, data.callDeadlineProgress(check)) - } proc.Log().Error("PluginOperator: failed to get resume waiting for resource: %v", err) errProgress := data.newUnforeseenError() errProgress.StatusMessage = err.Error() @@ -1002,9 +916,6 @@ func list(from gen.PID, state gen.Atom, data PluginUpdateData, operation ListRes // Retry loop with exponential backoff for throttling for attempt := 1; attempt <= maxListAttempts; attempt++ { - // Discovery has no operator watchdog and owns an intentionally long - // paging and retry budget, so a list call is not bounded by the - // agent-supplied per-call deadline. result, err = data.plugin.List(data.context, &resource.ListRequest{ ResourceType: operation.ResourceType, TargetConfig: operation.TargetConfig, diff --git a/pkg/plugin/plugin_operator_deadline_test.go b/pkg/plugin/plugin_operator_deadline_test.go deleted file mode 100644 index 674ece9de..000000000 --- a/pkg/plugin/plugin_operator_deadline_test.go +++ /dev/null @@ -1,653 +0,0 @@ -// © 2026 Platform Engineering Labs Inc. -// -// SPDX-License-Identifier: FSL-1.1-ALv2 - -//go:build unit - -package plugin - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "sync" - "testing" - "time" - - "ergo.services/ergo/gen" - "github.com/masterminds/semver" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - - pkgmodel "github.com/platform-engineering-labs/formae/pkg/model" - "github.com/platform-engineering-labs/formae/pkg/plugin/resource" -) - -const ( - deadlineTestNamespace = "test" - // deadlineSlack absorbs the time between deriving a call context and - // observing its deadline inside the test. - deadlineSlack = time.Second -) - -// recordingPlugin is a FullResourcePlugin double that records the context handed -// to each operation and optionally fails every call with a canned error. -type recordingPlugin struct { - mu sync.Mutex - ctxs map[resource.Operation]context.Context - err error -} - -func newRecordingPlugin() *recordingPlugin { - return &recordingPlugin{ctxs: make(map[resource.Operation]context.Context)} -} - -func (p *recordingPlugin) record(operation resource.Operation, ctx context.Context) { - p.mu.Lock() - defer p.mu.Unlock() - p.ctxs[operation] = ctx -} - -// contextFor returns the context the plugin received for an operation. -func (p *recordingPlugin) contextFor(t *testing.T, operation resource.Operation) context.Context { - t.Helper() - p.mu.Lock() - defer p.mu.Unlock() - ctx, ok := p.ctxs[operation] - require.True(t, ok, "plugin was never called for operation %s", operation) - return ctx -} - -func (p *recordingPlugin) RateLimit() pkgmodel.RateLimitConfig { - return pkgmodel.RateLimitConfig{Scope: pkgmodel.RateLimitScopeNamespace, MaxRequestsPerSecondForNamespace: 10} -} - -func (p *recordingPlugin) DiscoveryFilters() []pkgmodel.MatchFilter { return nil } - -func (p *recordingPlugin) LabelConfig() pkgmodel.LabelConfig { - return pkgmodel.LabelConfig{DefaultQuery: "$.Name"} -} - -func (p *recordingPlugin) Name() string { return "recording-plugin" } -func (p *recordingPlugin) Namespace() string { return deadlineTestNamespace } -func (p *recordingPlugin) Version() *semver.Version { - return semver.MustParse("1.0.0") -} -func (p *recordingPlugin) SupportedResources() []ResourceDescriptor { return nil } -func (p *recordingPlugin) SchemaForResourceType(string) (pkgmodel.Schema, error) { - return pkgmodel.Schema{}, nil -} - -func (p *recordingPlugin) Create(ctx context.Context, _ *resource.CreateRequest) (*resource.CreateResult, error) { - p.record(resource.OperationCreate, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.CreateResult{ProgressResult: inProgress(resource.OperationCreate)}, nil -} - -func (p *recordingPlugin) Read(ctx context.Context, _ *resource.ReadRequest) (*resource.ReadResult, error) { - p.record(resource.OperationRead, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.ReadResult{Properties: `{"Name":"resource"}`}, nil -} - -func (p *recordingPlugin) Update(ctx context.Context, _ *resource.UpdateRequest) (*resource.UpdateResult, error) { - p.record(resource.OperationUpdate, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.UpdateResult{ProgressResult: inProgress(resource.OperationUpdate)}, nil -} - -func (p *recordingPlugin) Delete(ctx context.Context, _ *resource.DeleteRequest) (*resource.DeleteResult, error) { - p.record(resource.OperationDelete, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.DeleteResult{ProgressResult: inProgress(resource.OperationDelete)}, nil -} - -func (p *recordingPlugin) Status(ctx context.Context, _ *resource.StatusRequest) (*resource.StatusResult, error) { - p.record(resource.OperationCheckStatus, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.StatusResult{ProgressResult: inProgress(resource.OperationCheckStatus)}, nil -} - -func (p *recordingPlugin) List(ctx context.Context, _ *resource.ListRequest) (*resource.ListResult, error) { - p.record(resource.OperationList, ctx) - if p.err != nil { - return nil, p.err - } - return &resource.ListResult{NativeIDs: []string{"resource-1"}}, nil -} - -func inProgress(operation resource.Operation) *resource.ProgressResult { - return &resource.ProgressResult{ - Operation: operation, - OperationStatus: resource.OperationStatusInProgress, - NativeID: "resource-1", - RequestID: "request-1", - } -} - -// stubOperatorLog swallows all log output for plugin operator tests. -type stubOperatorLog struct{ gen.Log } - -func (stubOperatorLog) Trace(string, ...any) {} -func (stubOperatorLog) Debug(string, ...any) {} -func (stubOperatorLog) Info(string, ...any) {} -func (stubOperatorLog) Warning(string, ...any) {} -func (stubOperatorLog) Error(string, ...any) {} -func (stubOperatorLog) Panic(string, ...any) {} - -// stubOperatorNode exposes a node-level environment so tests can exercise the -// operator's node fallback. -type stubOperatorNode struct { - gen.Node - env map[gen.Env]any -} - -func (n stubOperatorNode) Name() gen.Atom { return gen.Atom("test-node") } - -func (n stubOperatorNode) Env(name gen.Env) (any, bool) { - v, ok := n.env[name] - return v, ok -} - -// stubOperatorProcess is a hand-rolled gen.Process double for PluginOperator -// tests. It serves the process environment, records every proc.Send message and -// every proc.SendAfter reschedule. -type stubOperatorProcess struct { - gen.Process - - behavior gen.ProcessBehavior - env map[gen.Env]any - node stubOperatorNode - - mu sync.Mutex - sends []any - sendsAfter []any -} - -func (p *stubOperatorProcess) Log() gen.Log { return stubOperatorLog{} } -func (p *stubOperatorProcess) Node() gen.Node { return p.node } -func (p *stubOperatorProcess) PID() gen.PID { return gen.PID{Node: "test-node", ID: 1} } -func (p *stubOperatorProcess) Behavior() gen.ProcessBehavior { return p.behavior } -func (p *stubOperatorProcess) Mailbox() gen.ProcessMailbox { return gen.ProcessMailbox{} } -func (p *stubOperatorProcess) Env(name gen.Env) (any, bool) { - v, ok := p.env[name] - return v, ok -} - -func (p *stubOperatorProcess) Send(_ any, message any) error { - p.mu.Lock() - defer p.mu.Unlock() - p.sends = append(p.sends, message) - return nil -} - -func (p *stubOperatorProcess) SendAfter(_ any, message any, _ time.Duration) (gen.CancelFunc, error) { - p.mu.Lock() - defer p.mu.Unlock() - p.sendsAfter = append(p.sendsAfter, message) - return func() bool { return true }, nil -} - -// sentProgress returns all TrackedProgress messages sent via proc.Send, in order. -func (p *stubOperatorProcess) sentProgress() []TrackedProgress { - p.mu.Lock() - defer p.mu.Unlock() - var out []TrackedProgress - for _, s := range p.sends { - if progress, ok := s.(TrackedProgress); ok { - out = append(out, progress) - } - } - return out -} - -// scheduled returns all messages the operator rescheduled via proc.SendAfter, -// in order. -func (p *stubOperatorProcess) scheduled() []any { - p.mu.Lock() - defer p.mu.Unlock() - return append([]any(nil), p.sendsAfter...) -} - -// sentListings returns all Listing messages sent via proc.Send, in order. -func (p *stubOperatorProcess) sentListings() []Listing { - p.mu.Lock() - defer p.mu.Unlock() - var out []Listing - for _, s := range p.sends { - if listing, ok := s.(Listing); ok { - out = append(out, listing) - } - } - return out -} - -func newOperatorProcess(env map[gen.Env]any, nodeEnv map[gen.Env]any) *stubOperatorProcess { - return &stubOperatorProcess{ - env: env, - node: stubOperatorNode{env: nodeEnv}, - } -} - -// assertCallDeadline asserts the context bounds the call by budget: the deadline -// must not exceed budget from now, and must not be materially shorter either. -func assertCallDeadline(t *testing.T, ctx context.Context, budget time.Duration) { - t.Helper() - deadline, ok := ctx.Deadline() - require.True(t, ok, "a watched plugin call must receive a context with a deadline") - remaining := time.Until(deadline) - assert.LessOrEqual(t, remaining, budget, "call deadline must not exceed the supplied budget") - assert.Greater(t, remaining, budget-deadlineSlack, "call deadline must span the supplied budget") -} - -func deadlineTestData(plugin FullResourcePlugin, callTimeout time.Duration) PluginUpdateData { - return PluginUpdateData{ - attempts: 1, - callTimeout: callTimeout, - context: context.Background(), - plugin: plugin, - config: pkgmodel.RetryConfig{ - MaxRetries: 3, - RetryDelay: time.Millisecond, - StatusCheckInterval: 20 * time.Second, - }, - } -} - -func TestPluginOperatorInit_UsesCallTimeoutFromProcessEnv(t *testing.T) { - operator := &PluginOperator{} - proc := newOperatorProcess(map[gen.Env]any{ - "Plugin": newRecordingPlugin(), - "Context": context.Background(), - "RetryConfig": pkgmodel.RetryConfig{MaxRetries: 3}, - "PluginCallTimeout": 45 * time.Second, - }, nil) - proc.behavior = operator - - require.NoError(t, operator.ProcessInit(proc)) - assert.Equal(t, 45*time.Second, operator.Data().callTimeout) -} - -func TestPluginOperatorInit_FallsBackToNodeCallTimeout(t *testing.T) { - operator := &PluginOperator{} - proc := newOperatorProcess(map[gen.Env]any{ - "Plugin": newRecordingPlugin(), - "Context": context.Background(), - "RetryConfig": pkgmodel.RetryConfig{MaxRetries: 3}, - }, map[gen.Env]any{ - "PluginCallTimeout": 90 * time.Second, - }) - proc.behavior = operator - - require.NoError(t, operator.ProcessInit(proc)) - assert.Equal(t, 90*time.Second, operator.Data().callTimeout) -} - -func TestPluginOperatorInit_FallsBackToDefaultCallTimeout(t *testing.T) { - tests := []struct { - name string - env map[gen.Env]any - }{ - {name: "absent", env: nil}, - {name: "zero", env: map[gen.Env]any{"PluginCallTimeout": time.Duration(0)}}, - {name: "negative", env: map[gen.Env]any{"PluginCallTimeout": -5 * time.Second}}, - {name: "wrong type", env: map[gen.Env]any{"PluginCallTimeout": "45s"}}, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - env := map[gen.Env]any{ - "Plugin": newRecordingPlugin(), - "Context": context.Background(), - "RetryConfig": pkgmodel.RetryConfig{MaxRetries: 3}, - } - for k, v := range tt.env { - env[k] = v - } - - operator := &PluginOperator{} - proc := newOperatorProcess(env, nil) - proc.behavior = operator - - require.NoError(t, operator.ProcessInit(proc)) - assert.Equal(t, defaultPluginCallTimeout, operator.Data().callTimeout) - }) - } -} - -func TestWatchedOperationsBoundPluginCallByDeadline(t *testing.T) { - const callTimeout = 90 * time.Second - - tests := []struct { - name string - operation resource.Operation - invoke func(data PluginUpdateData, proc gen.Process) - }{ - { - name: "read", - operation: resource.OperationRead, - invoke: func(data PluginUpdateData, proc gen.Process) { - read(gen.PID{}, StateNotStarted, data, ReadResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, proc) - }, - }, - { - name: "create", - operation: resource.OperationCreate, - invoke: func(data PluginUpdateData, proc gen.Process) { - create(gen.PID{}, StateNotStarted, data, CreateResource{Namespace: deadlineTestNamespace, ResourceType: "Test::Resource"}, proc) - }, - }, - { - name: "update", - operation: resource.OperationUpdate, - invoke: func(data PluginUpdateData, proc gen.Process) { - update(gen.PID{}, StateNotStarted, data, UpdateResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, proc) - }, - }, - { - name: "delete", - operation: resource.OperationDelete, - invoke: func(data PluginUpdateData, proc gen.Process) { - delete(gen.PID{}, StateNotStarted, data, DeleteResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, proc) - }, - }, - { - name: "status", - operation: resource.OperationCheckStatus, - invoke: func(data PluginUpdateData, proc gen.Process) { - status(gen.PID{}, StateWaitingForResource, data, PluginOperatorCheckStatus{Namespace: deadlineTestNamespace, RequestID: "request-1"}, proc) - }, - }, - { - name: "resume", - operation: resource.OperationCheckStatus, - invoke: func(data PluginUpdateData, proc gen.Process) { - resume(gen.PID{}, StateNotStarted, data, ResumeWaitingForResource{ - Namespace: deadlineTestNamespace, - ResourceOperation: resource.OperationCreate, - Request: PluginOperatorCheckStatus{Namespace: deadlineTestNamespace, RequestID: "request-1"}, - PreviousAttempts: 1, - }, proc) - }, - }, - { - name: "retried create", - operation: resource.OperationCreate, - invoke: func(data PluginUpdateData, proc gen.Process) { - retry(gen.PID{}, StateRetrying, data, PluginOperatorRetry{ - ResourceOperation: resource.OperationCreate, - Request: CreateResource{Namespace: deadlineTestNamespace, ResourceType: "Test::Resource"}, - }, proc) - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - plugin := newRecordingPlugin() - proc := newOperatorProcess(nil, nil) - - tt.invoke(deadlineTestData(plugin, callTimeout), proc) - - assertCallDeadline(t, plugin.contextFor(t, tt.operation), callTimeout) - }) - } -} - -func TestList_PluginCallHasNoDeadline(t *testing.T) { - plugin := newRecordingPlugin() - proc := newOperatorProcess(nil, nil) - - state, _, _, err := list(gen.PID{}, StateNotStarted, deadlineTestData(plugin, 90*time.Second), - ListResources{Namespace: deadlineTestNamespace, ResourceType: "Test::Resource"}, proc) - - require.NoError(t, err) - assert.Equal(t, StateFinishedSuccessfully, state) - require.Len(t, proc.sentListings(), 1) - - _, hasDeadline := plugin.contextFor(t, resource.OperationList).Deadline() - assert.False(t, hasDeadline, "discovery owns its own paging budget and must not be bounded by the per-call deadline") -} - -func TestReadOnlyOperationsClassifyDeadlineAsServiceTimeout(t *testing.T) { - const callTimeout = 90 * time.Second - - tests := []struct { - name string - wantState gen.Atom - wantOperation resource.Operation - wantScheduled any - invoke func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) - }{ - { - name: "read", - wantState: StateRetrying, - wantOperation: resource.OperationRead, - // A read carries no provider side effect, so the ladder re-issues it. - wantScheduled: PluginOperatorRetry{ - ResourceOperation: resource.OperationRead, - Request: ReadResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, - }, - invoke: func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) { - state, _, progress, _, _ := read(gen.PID{}, StateNotStarted, data, - ReadResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, proc) - return state, progress - }, - }, - { - name: "resume", - wantState: StateWaitingForResource, - wantOperation: resource.OperationCreate, - wantScheduled: PluginOperatorCheckStatus{ - Namespace: deadlineTestNamespace, - RequestID: "request-1", - NativeID: "resource-1", - ResourceOperation: resource.OperationCreate, - }, - invoke: func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) { - state, _, progress, _, _ := resume(gen.PID{}, StateNotStarted, data, ResumeWaitingForResource{ - Namespace: deadlineTestNamespace, - ResourceOperation: resource.OperationCreate, - Request: PluginOperatorCheckStatus{ - Namespace: deadlineTestNamespace, - RequestID: "request-1", - NativeID: "resource-1", - ResourceOperation: resource.OperationCreate, - }, - PreviousAttempts: 1, - }, proc) - return state, progress - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - plugin := newRecordingPlugin() - plugin.err = fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded) - proc := newOperatorProcess(nil, nil) - - state, progress := tt.invoke(deadlineTestData(plugin, callTimeout), proc) - - assert.Equal(t, tt.wantState, state) - assert.Equal(t, tt.wantOperation, progress.Operation) - assert.Equal(t, resource.OperationStatusFailure, progress.OperationStatus) - assert.Equal(t, resource.OperationErrorCodeServiceTimeout, progress.ErrorCode) - assert.False(t, progress.Failed(), "retries remain, so the operation must not be terminal") - assert.Contains(t, progress.StatusMessage, callTimeout.String()) - - scheduled := proc.scheduled() - require.Len(t, scheduled, 1, "the retry ladder must reschedule the operation so every attempt heartbeats") - assert.Equal(t, tt.wantScheduled, scheduled[0]) - }) - } -} - -func TestStatus_ClassifiesDeadlineAsServiceTimeout(t *testing.T) { - const callTimeout = 90 * time.Second - - plugin := newRecordingPlugin() - plugin.err = fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded) - proc := newOperatorProcess(nil, nil) - - state, _, _, err := status(gen.PID{}, StateWaitingForResource, deadlineTestData(plugin, callTimeout), - PluginOperatorCheckStatus{ - Namespace: deadlineTestNamespace, - RequestID: "request-1", - NativeID: "resource-1", - ResourceOperation: resource.OperationCreate, - }, proc) - - require.NoError(t, err) - assert.Equal(t, StateWaitingForResource, state, "a status check that outran its deadline must not fail the operation") - - sent := proc.sentProgress() - require.Len(t, sent, 1, "the resource updater must be told the status check outran its deadline") - assert.Equal(t, resource.OperationCreate, sent[0].Operation) - assert.Equal(t, resource.OperationStatusFailure, sent[0].OperationStatus) - assert.Equal(t, resource.OperationErrorCodeServiceTimeout, sent[0].ErrorCode) - assert.False(t, sent[0].Failed(), "retries remain, so the operation must not be terminal") - assert.Contains(t, sent[0].StatusMessage, callTimeout.String()) -} - -// TestFailedStatusCallNeverReissuesTheOriginalOperation covers a status check -// that carries the request that started it — the local-path shape, where the -// retry ladder would otherwise re-issue that request. A create that may already -// have reached the provider must never be sent again, so a status call that -// fails recoverably reschedules another status check instead. -func TestFailedStatusCallNeverReissuesTheOriginalOperation(t *testing.T) { - const callTimeout = 90 * time.Second - - originalCreate := CreateResource{Namespace: deadlineTestNamespace, ResourceType: "Test::Resource"} - check := PluginOperatorCheckStatus{ - Namespace: deadlineTestNamespace, - RequestID: "request-1", - NativeID: "resource-1", - ResourceType: "Test::Resource", - ResourceOperation: resource.OperationCreate, - Request: originalCreate, - } - - tests := []struct { - name string - err error - invoke func(data PluginUpdateData, proc gen.Process) gen.Atom - }{ - { - name: "status past its deadline", - err: fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded), - invoke: func(data PluginUpdateData, proc gen.Process) gen.Atom { - state, _, _, _ := status(gen.PID{}, StateWaitingForResource, data, check, proc) - return state - }, - }, - { - name: "throttled status", - err: errors.New("ThrottlingException: Rate exceeded"), - invoke: func(data PluginUpdateData, proc gen.Process) gen.Atom { - state, _, _, _ := status(gen.PID{}, StateWaitingForResource, data, check, proc) - return state - }, - }, - { - name: "resume past its deadline", - err: fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded), - invoke: func(data PluginUpdateData, proc gen.Process) gen.Atom { - state, _, _, _, _ := resume(gen.PID{}, StateNotStarted, data, ResumeWaitingForResource{ - Namespace: deadlineTestNamespace, - ResourceOperation: resource.OperationCreate, - Request: check, - PreviousAttempts: 1, - }, proc) - return state - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - plugin := newRecordingPlugin() - plugin.err = tt.err - proc := newOperatorProcess(nil, nil) - - state := tt.invoke(deadlineTestData(plugin, callTimeout), proc) - - assert.Equal(t, StateWaitingForResource, state) - - scheduled := proc.scheduled() - require.Len(t, scheduled, 1) - rescheduledCheck, ok := scheduled[0].(PluginOperatorCheckStatus) - require.True(t, ok, "a status check that outran its deadline must reschedule a status check, got %T", scheduled[0]) - assert.Nil(t, rescheduledCheck.Request, "the rescheduled check must not carry the request that would re-issue the operation") - assert.Equal(t, check.RequestID, rescheduledCheck.RequestID) - assert.Equal(t, check.NativeID, rescheduledCheck.NativeID) - assert.Equal(t, check.ResourceType, rescheduledCheck.ResourceType) - assert.Equal(t, check.ResourceOperation, rescheduledCheck.ResourceOperation) - - _, called := plugin.ctxs[resource.OperationCreate] - assert.False(t, called, "the mutating operation must never be re-issued") - }) - } -} - -func TestMutatingOperationsClassifyDeadlineAsTerminal(t *testing.T) { - const callTimeout = 90 * time.Second - - tests := []struct { - name string - invoke func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) - }{ - { - name: "create", - invoke: func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) { - state, _, progress, _, _ := create(gen.PID{}, StateNotStarted, data, - CreateResource{Namespace: deadlineTestNamespace, ResourceType: "Test::Resource"}, proc) - return state, progress - }, - }, - { - name: "update", - invoke: func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) { - state, _, progress, _, _ := update(gen.PID{}, StateNotStarted, data, - UpdateResource{Namespace: deadlineTestNamespace, NativeID: "resource-1", DesiredProperties: json.RawMessage(`{}`)}, proc) - return state, progress - }, - }, - { - name: "delete", - invoke: func(data PluginUpdateData, proc gen.Process) (gen.Atom, TrackedProgress) { - state, _, progress, _, _ := delete(gen.PID{}, StateNotStarted, data, - DeleteResource{Namespace: deadlineTestNamespace, NativeID: "resource-1"}, proc) - return state, progress - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - plugin := newRecordingPlugin() - plugin.err = fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded) - proc := newOperatorProcess(nil, nil) - - state, progress := tt.invoke(deadlineTestData(plugin, callTimeout), proc) - - assert.Equal(t, StateFinishedWithError, state, "a mutating call that may have reached the provider must not be repeated") - assert.Equal(t, resource.OperationErrorCodeUnforeseenError, progress.ErrorCode) - assert.True(t, progress.Failed()) - assert.Contains(t, progress.StatusMessage, callTimeout.String()) - assert.Contains(t, progress.StatusMessage, "calling the cloud API", "the plugin's own error text must stay diagnosable") - assert.Empty(t, proc.sendsAfter, "no retry may be scheduled for a mutating call that outran its deadline") - }) - } -} diff --git a/pkg/plugin/plugin_operator_double_test.go b/pkg/plugin/plugin_operator_double_test.go new file mode 100644 index 000000000..f31c1ad87 --- /dev/null +++ b/pkg/plugin/plugin_operator_double_test.go @@ -0,0 +1,306 @@ +// © 2026 Platform Engineering Labs Inc. +// +// SPDX-License-Identifier: FSL-1.1-ALv2 + +//go:build unit + +package plugin + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "ergo.services/ergo/gen" + "github.com/masterminds/semver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + pkgmodel "github.com/platform-engineering-labs/formae/pkg/model" + "github.com/platform-engineering-labs/formae/pkg/plugin/resource" +) + +const ( + operatorTestNamespace = "test" +) + +// recordingPlugin is a FullResourcePlugin double that records the context handed +// to each operation and optionally fails every call with a canned error. +type recordingPlugin struct { + mu sync.Mutex + ctxs map[resource.Operation]context.Context + err error +} + +func newRecordingPlugin() *recordingPlugin { + return &recordingPlugin{ctxs: make(map[resource.Operation]context.Context)} +} + +func (p *recordingPlugin) record(operation resource.Operation, ctx context.Context) { + p.mu.Lock() + defer p.mu.Unlock() + p.ctxs[operation] = ctx +} + +// contextFor returns the context the plugin received for an operation. +func (p *recordingPlugin) contextFor(t *testing.T, operation resource.Operation) context.Context { + t.Helper() + p.mu.Lock() + defer p.mu.Unlock() + ctx, ok := p.ctxs[operation] + require.True(t, ok, "plugin was never called for operation %s", operation) + return ctx +} + +func (p *recordingPlugin) RateLimit() pkgmodel.RateLimitConfig { + return pkgmodel.RateLimitConfig{Scope: pkgmodel.RateLimitScopeNamespace, MaxRequestsPerSecondForNamespace: 10} +} + +func (p *recordingPlugin) DiscoveryFilters() []pkgmodel.MatchFilter { return nil } + +func (p *recordingPlugin) LabelConfig() pkgmodel.LabelConfig { + return pkgmodel.LabelConfig{DefaultQuery: "$.Name"} +} + +func (p *recordingPlugin) Name() string { return "recording-plugin" } +func (p *recordingPlugin) Namespace() string { return operatorTestNamespace } +func (p *recordingPlugin) Version() *semver.Version { + return semver.MustParse("1.0.0") +} +func (p *recordingPlugin) SupportedResources() []ResourceDescriptor { return nil } +func (p *recordingPlugin) SchemaForResourceType(string) (pkgmodel.Schema, error) { + return pkgmodel.Schema{}, nil +} + +func (p *recordingPlugin) Create(ctx context.Context, _ *resource.CreateRequest) (*resource.CreateResult, error) { + p.record(resource.OperationCreate, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.CreateResult{ProgressResult: inProgress(resource.OperationCreate)}, nil +} + +func (p *recordingPlugin) Read(ctx context.Context, _ *resource.ReadRequest) (*resource.ReadResult, error) { + p.record(resource.OperationRead, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.ReadResult{Properties: `{"Name":"resource"}`}, nil +} + +func (p *recordingPlugin) Update(ctx context.Context, _ *resource.UpdateRequest) (*resource.UpdateResult, error) { + p.record(resource.OperationUpdate, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.UpdateResult{ProgressResult: inProgress(resource.OperationUpdate)}, nil +} + +func (p *recordingPlugin) Delete(ctx context.Context, _ *resource.DeleteRequest) (*resource.DeleteResult, error) { + p.record(resource.OperationDelete, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.DeleteResult{ProgressResult: inProgress(resource.OperationDelete)}, nil +} + +func (p *recordingPlugin) Status(ctx context.Context, _ *resource.StatusRequest) (*resource.StatusResult, error) { + p.record(resource.OperationCheckStatus, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.StatusResult{ProgressResult: inProgress(resource.OperationCheckStatus)}, nil +} + +func (p *recordingPlugin) List(ctx context.Context, _ *resource.ListRequest) (*resource.ListResult, error) { + p.record(resource.OperationList, ctx) + if p.err != nil { + return nil, p.err + } + return &resource.ListResult{NativeIDs: []string{"resource-1"}}, nil +} + +func inProgress(operation resource.Operation) *resource.ProgressResult { + return &resource.ProgressResult{ + Operation: operation, + OperationStatus: resource.OperationStatusInProgress, + NativeID: "resource-1", + RequestID: "request-1", + } +} + +// stubOperatorLog swallows all log output for plugin operator tests. +type stubOperatorLog struct{ gen.Log } + +func (stubOperatorLog) Trace(string, ...any) {} +func (stubOperatorLog) Debug(string, ...any) {} +func (stubOperatorLog) Info(string, ...any) {} +func (stubOperatorLog) Warning(string, ...any) {} +func (stubOperatorLog) Error(string, ...any) {} +func (stubOperatorLog) Panic(string, ...any) {} + +// stubOperatorNode exposes a node-level environment so tests can exercise the +// operator's node fallback. +type stubOperatorNode struct { + gen.Node + env map[gen.Env]any +} + +func (n stubOperatorNode) Name() gen.Atom { return gen.Atom("test-node") } + +func (n stubOperatorNode) Env(name gen.Env) (any, bool) { + v, ok := n.env[name] + return v, ok +} + +// stubOperatorProcess is a hand-rolled gen.Process double for PluginOperator +// tests. It serves the process environment, records every proc.Send message and +// every proc.SendAfter reschedule. +type stubOperatorProcess struct { + gen.Process + + behavior gen.ProcessBehavior + env map[gen.Env]any + node stubOperatorNode + + mu sync.Mutex + sends []any + sendsAfter []any +} + +func (p *stubOperatorProcess) Log() gen.Log { return stubOperatorLog{} } +func (p *stubOperatorProcess) Node() gen.Node { return p.node } +func (p *stubOperatorProcess) PID() gen.PID { return gen.PID{Node: "test-node", ID: 1} } +func (p *stubOperatorProcess) Behavior() gen.ProcessBehavior { return p.behavior } +func (p *stubOperatorProcess) Mailbox() gen.ProcessMailbox { return gen.ProcessMailbox{} } +func (p *stubOperatorProcess) Env(name gen.Env) (any, bool) { + v, ok := p.env[name] + return v, ok +} + +func (p *stubOperatorProcess) Send(_ any, message any) error { + p.mu.Lock() + defer p.mu.Unlock() + p.sends = append(p.sends, message) + return nil +} + +func (p *stubOperatorProcess) SendAfter(_ any, message any, _ time.Duration) (gen.CancelFunc, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.sendsAfter = append(p.sendsAfter, message) + return func() bool { return true }, nil +} + +// sentProgress returns all TrackedProgress messages sent via proc.Send, in order. +func (p *stubOperatorProcess) sentProgress() []TrackedProgress { + p.mu.Lock() + defer p.mu.Unlock() + var out []TrackedProgress + for _, s := range p.sends { + if progress, ok := s.(TrackedProgress); ok { + out = append(out, progress) + } + } + return out +} + +// scheduled returns all messages the operator rescheduled via proc.SendAfter, +// in order. +func (p *stubOperatorProcess) scheduled() []any { + p.mu.Lock() + defer p.mu.Unlock() + return append([]any(nil), p.sendsAfter...) +} + +// sentListings returns all Listing messages sent via proc.Send, in order. +func (p *stubOperatorProcess) sentListings() []Listing { + p.mu.Lock() + defer p.mu.Unlock() + var out []Listing + for _, s := range p.sends { + if listing, ok := s.(Listing); ok { + out = append(out, listing) + } + } + return out +} + +func newOperatorProcess(env map[gen.Env]any, nodeEnv map[gen.Env]any) *stubOperatorProcess { + return &stubOperatorProcess{ + env: env, + node: stubOperatorNode{env: nodeEnv}, + } +} + +func operatorTestData(plugin FullResourcePlugin) PluginUpdateData { + return PluginUpdateData{ + attempts: 1, + context: context.Background(), + plugin: plugin, + config: pkgmodel.RetryConfig{ + MaxRetries: 3, + RetryDelay: time.Millisecond, + StatusCheckInterval: 20 * time.Second, + }, + } +} + +// TestFailedStatusCallNeverReissuesTheOriginalOperation covers a status check +// that carries the request that started it — the local-path shape, where the +// retry ladder would otherwise re-issue that request. A create that may already +// have reached the provider must never be sent again, so a status call that +// fails recoverably reschedules another status check instead. +func TestFailedStatusCallNeverReissuesTheOriginalOperation(t *testing.T) { + originalCreate := CreateResource{Namespace: operatorTestNamespace, ResourceType: "Test::Resource"} + check := PluginOperatorCheckStatus{ + Namespace: operatorTestNamespace, + RequestID: "request-1", + NativeID: "resource-1", + ResourceType: "Test::Resource", + ResourceOperation: resource.OperationCreate, + Request: originalCreate, + } + + tests := []struct { + name string + err error + invoke func(data PluginUpdateData, proc gen.Process) gen.Atom + }{ + { + name: "throttled status", + err: errors.New("ThrottlingException: Rate exceeded"), + invoke: func(data PluginUpdateData, proc gen.Process) gen.Atom { + state, _, _, _ := status(gen.PID{}, StateWaitingForResource, data, check, proc) + return state + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plugin := newRecordingPlugin() + plugin.err = tt.err + proc := newOperatorProcess(nil, nil) + + state := tt.invoke(operatorTestData(plugin), proc) + + assert.Equal(t, StateWaitingForResource, state) + + scheduled := proc.scheduled() + require.Len(t, scheduled, 1) + rescheduledCheck, ok := scheduled[0].(PluginOperatorCheckStatus) + require.True(t, ok, "a status check that outran its deadline must reschedule a status check, got %T", scheduled[0]) + assert.Nil(t, rescheduledCheck.Request, "the rescheduled check must not carry the request that would re-issue the operation") + assert.Equal(t, check.RequestID, rescheduledCheck.RequestID) + assert.Equal(t, check.NativeID, rescheduledCheck.NativeID) + assert.Equal(t, check.ResourceType, rescheduledCheck.ResourceType) + assert.Equal(t, check.ResourceOperation, rescheduledCheck.ResourceOperation) + + _, called := plugin.ctxs[resource.OperationCreate] + assert.False(t, called, "the mutating operation must never be re-issued") + }) + } +} diff --git a/pkg/plugin/plugin_operator_progress_test.go b/pkg/plugin/plugin_operator_progress_test.go index 23974d0ff..8930c66e3 100644 --- a/pkg/plugin/plugin_operator_progress_test.go +++ b/pkg/plugin/plugin_operator_progress_test.go @@ -11,7 +11,6 @@ import ( "errors" "fmt" "testing" - "time" "ergo.services/ergo/gen" "github.com/stretchr/testify/assert" @@ -24,7 +23,7 @@ import ( // create to finish, carrying the create that started it. func statusCheckFor(request any) PluginOperatorCheckStatus { return PluginOperatorCheckStatus{ - Namespace: deadlineTestNamespace, + Namespace: operatorTestNamespace, RequestID: "request-1", NativeID: "resource-1", ResourceType: "Test::Resource", @@ -34,7 +33,6 @@ func statusCheckFor(request any) PluginOperatorCheckStatus { } func TestStatus_NamespaceMismatchReportsBeforeTerminating(t *testing.T) { - const callTimeout = 90 * time.Second plugin := newRecordingPlugin() proc := newOperatorProcess(nil, nil) @@ -43,7 +41,7 @@ func TestStatus_NamespaceMismatchReportsBeforeTerminating(t *testing.T) { check.Namespace = "other" state, _, _, err := status(gen.PID{}, StateWaitingForResource, - deadlineTestData(plugin, callTimeout), check, proc) + operatorTestData(plugin), check, proc) require.NoError(t, err) assert.Equal(t, StateFinishedWithError, state) @@ -60,7 +58,6 @@ func TestStatus_NamespaceMismatchReportsBeforeTerminating(t *testing.T) { } func TestStatus_ClassifiesStatusCallError(t *testing.T) { - const callTimeout = 90 * time.Second tests := []struct { name string @@ -69,13 +66,6 @@ func TestStatus_ClassifiesStatusCallError(t *testing.T) { wantState gen.Atom wantTerminal bool }{ - { - name: "deadline", - err: fmt.Errorf("calling the cloud API: %w", context.DeadlineExceeded), - wantCode: resource.OperationErrorCodeServiceTimeout, - wantState: StateWaitingForResource, - wantTerminal: false, - }, { name: "throttling", err: errors.New("ThrottlingException: Rate exceeded"), @@ -84,9 +74,9 @@ func TestStatus_ClassifiesStatusCallError(t *testing.T) { wantTerminal: false, }, { - name: "throttled past its deadline", - err: fmt.Errorf("ThrottlingException: Rate exceeded: %w", context.DeadlineExceeded), - wantCode: resource.OperationErrorCodeServiceTimeout, + name: "throttled while the call was cancelled", + err: fmt.Errorf("ThrottlingException: Rate exceeded: %w", context.Canceled), + wantCode: resource.OperationErrorCodeThrottling, wantState: StateWaitingForResource, wantTerminal: false, }, @@ -107,7 +97,7 @@ func TestStatus_ClassifiesStatusCallError(t *testing.T) { check := statusCheckFor(nil) state, _, _, err := status(gen.PID{}, StateWaitingForResource, - deadlineTestData(plugin, callTimeout), check, proc) + operatorTestData(plugin), check, proc) require.NoError(t, err) assert.Equal(t, tt.wantState, state) @@ -131,7 +121,6 @@ func TestStatus_ClassifiesStatusCallError(t *testing.T) { // its backoff and eventually gives up, and every attempt must report its own // error rather than repeating the first one. func TestStatus_FailingCallConsumesTheRetryLadder(t *testing.T) { - const callTimeout = 90 * time.Second plugin := newRecordingPlugin() proc := newOperatorProcess(nil, nil) @@ -139,7 +128,7 @@ func TestStatus_FailingCallConsumesTheRetryLadder(t *testing.T) { plugin.err = errors.New("ThrottlingException: Rate exceeded polling for the first time") _, data, _, err := status(gen.PID{}, StateWaitingForResource, - deadlineTestData(plugin, callTimeout), check, proc) + operatorTestData(plugin), check, proc) require.NoError(t, err) plugin.err = errors.New("ThrottlingException: Rate exceeded polling for the second time") @@ -158,14 +147,13 @@ func TestStatus_FailingCallConsumesTheRetryLadder(t *testing.T) { // recovers: the operation must give up once its attempts are spent instead of // polling forever. func TestStatus_FailingCallsExhaustTheRetryLadder(t *testing.T) { - const callTimeout = 90 * time.Second plugin := newRecordingPlugin() plugin.err = errors.New("ThrottlingException: Rate exceeded") proc := newOperatorProcess(nil, nil) check := statusCheckFor(nil) - data := deadlineTestData(plugin, callTimeout) + data := operatorTestData(plugin) maxAttempts := int(data.config.MaxRetries) + 1 var state gen.Atom @@ -184,12 +172,11 @@ func TestStatus_FailingCallsExhaustTheRetryLadder(t *testing.T) { } func TestRetry_UnsupportedOperationReportsTerminalFailure(t *testing.T) { - const callTimeout = 90 * time.Second plugin := newRecordingPlugin() proc := newOperatorProcess(nil, nil) - state, _, _, err := retry(gen.PID{}, StateRetrying, deadlineTestData(plugin, callTimeout), + state, _, _, err := retry(gen.PID{}, StateRetrying, operatorTestData(plugin), PluginOperatorRetry{ResourceOperation: resource.OperationNotSupported}, proc) require.NoError(t, err) From d91be3284043e4c741674999ced0c651c980008f Mon Sep 17 00:00:00 2001 From: Jeroen Soeters Date: Tue, 18 Aug 2026 08:17:21 -0700 Subject: [PATCH 2/2] style: gofmt the coordinator spawn env after removing the call-timeout key --- .../metastructure/plugin_coordinator/plugin_coordinator.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/metastructure/plugin_coordinator/plugin_coordinator.go b/internal/metastructure/plugin_coordinator/plugin_coordinator.go index 1a1a980ac..e7e77b049 100644 --- a/internal/metastructure/plugin_coordinator/plugin_coordinator.go +++ b/internal/metastructure/plugin_coordinator/plugin_coordinator.go @@ -323,8 +323,8 @@ func (c *PluginCoordinator) spawnPluginOperator(req messages.SpawnPluginOperator // ResourceUpdater sizes its watchdog window from — and the requesting process. func pluginOperatorEnv(retryConfig model.RetryConfig, requestedBy gen.PID) map[gen.Env]any { return map[gen.Env]any{ - gen.Env("RetryConfig"): retryConfig, - gen.Env("RequestedBy"): requestedBy, + gen.Env("RetryConfig"): retryConfig, + gen.Env("RequestedBy"): requestedBy, } }