diff --git a/internal/metastructure/plugin_coordinator/plugin_coordinator.go b/internal/metastructure/plugin_coordinator/plugin_coordinator.go index 1bc6a3260..e7e77b049 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" ) @@ -324,9 +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("PluginCallTimeout"): resource_update.PluginCallTimeout, - gen.Env("RequestedBy"): requestedBy, + gen.Env("RetryConfig"): retryConfig, + 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 d7d0e22b2..a0ed87fa7 100644 --- a/internal/metastructure/resource_update/resource_updater.go +++ b/internal/metastructure/resource_update/resource_updater.go @@ -183,19 +183,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 @@ -564,7 +563,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 3eab08b20..5f0cd73e3 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 3c6daa944..6bb743092 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)