diff --git a/internal/graphapi/integration_lifecycle_test.go b/internal/graphapi/integration_lifecycle_test.go index 4aba5e10d6..a802c640fc 100644 --- a/internal/graphapi/integration_lifecycle_test.go +++ b/internal/graphapi/integration_lifecycle_test.go @@ -4,20 +4,18 @@ package graphapi_test import ( "context" - "encoding/json" "testing" "github.com/stretchr/testify/require" "github.com/theopenlane/core/common/enums" - "github.com/theopenlane/core/common/openapi" ent "github.com/theopenlane/core/internal/ent/generated" "github.com/theopenlane/core/internal/ent/generated/notification" "github.com/theopenlane/core/internal/ent/generated/privacy" - slackdef "github.com/theopenlane/core/internal/integrations/definitions/slack" intobvs "github.com/theopenlane/core/internal/integrations/observability" "github.com/theopenlane/core/internal/integrations/operations" integrationtypes "github.com/theopenlane/core/internal/integrations/types" + testint "github.com/theopenlane/core/internal/testutils/integrations" ) // notification object types mirrored from internal/integrations/runtime/health.go @@ -26,25 +24,65 @@ const ( integrationReconnectedObjectType = "INTEGRATION_RECONNECTED" ) -// slackReconcileOperation resolves the slack definition's reconcile-policy operation name -// from the live registry so the test matches the exact name the loops are keyed on -func slackReconcileOperation(t *testing.T) string { +// harnessReconcileOperation returns the reconcile operation name for one harness mode +func harnessReconcileOperation(t *testing.T, mode string) string { t.Helper() - def, ok := suite.integrationsRT.Registry().Definition(slackdef.DefinitionID.ID()) - require.True(t, ok, "slack definition must be registered") - - for _, op := range def.Operations { - if op.Policy.Reconcile { - return op.Name - } + switch mode { + case testint.ModeRecurring: + return testint.RecurringOp.Name() + case testint.ModeExhausting: + return testint.ExhaustingOp.Name() + case testint.ModeUnresolvable: + return testint.UnresolvableOp.Name() } - t.Fatal("slack definition has no reconcile operation") + t.Fatalf("unknown harness mode %q", mode) return "" } +// newHarnessInstallation installs the test integration in the given mode through the prod +// connect flow; the unresolvable mode stores a non-token credential so the client cannot build +func newHarnessInstallation(t *testing.T, ctx context.Context, mode string) (*ent.Integration, string) { + t.Helper() + + installation, err := suite.client.db.Integration.Create(). + SetName(randomName(t)). + SetKind("testintegration"). + SetDefinitionID(testint.DefinitionID.ID()). + Save(ctx) + require.NoError(t, err) + + credentialRef := testint.TokenCredential.ID() + credential := testint.TokenCredentialSet("test-token") + + if mode == testint.ModeUnresolvable { + credentialRef = testint.ServiceAccountCredential.ID() + credential = testint.ServiceAccountCredentialSet("test-project", "svc@example.com") + } + + require.NoError(t, suite.integrationsRT.Reconcile(ctx, installation, testint.ModeInput(mode), credentialRef, &credential, nil)) + + fragment := reconcileLoopFragment(t, installation.ID, harnessReconcileOperation(t, mode)) + + return reloadIntegration(t, ctx, installation.ID), fragment +} + +// seedHarnessLoop installs the test integration in recurring mode and asserts the connect flow +// seeded exactly one loop +func seedHarnessLoop(t *testing.T, ctx context.Context) (*ent.Integration, string) { + t.Helper() + + installation, fragment := newHarnessInstallation(t, ctx, testint.ModeRecurring) + + suite.WaitForEvents() + + require.Equal(t, 1, activeReconcileJobs(t, fragment)) + + return installation, fragment +} + // reconcileLoopFragment builds the metadata containment fragment identifying the recurring // loop jobs for one installation and operation, matching the keys ResetReconcileLoops uses func reconcileLoopFragment(t *testing.T, integrationID, operation string) string { @@ -104,22 +142,10 @@ func TestIntegrationLifecycle(t *testing.T) { allowCtx := privacy.DecisionContext(setContext(org.owner.UserCtx, suite.client.db), privacy.Allow) ownerCtx := setContext(org.owner.UserCtx, suite.client.db) - // empty UserInput keeps the reconcile operation enabled - clientConfig, err := json.Marshal(slackdef.UserInput{}) - require.NoError(t, err) - - installation, err := suite.client.db.Integration.Create(). - SetName("Slack Lifecycle Test"). - SetKind("slack"). - SetDefinitionID(slackdef.DefinitionID.ID()). - SetStatus(enums.IntegrationStatusConnected). - SetConfig(openapi.IntegrationConfig{ClientConfig: clientConfig}). - Save(allowCtx) - require.NoError(t, err) + installation, fragment := newHarnessInstallation(t, allowCtx, testint.ModeRecurring) require.Equal(t, org.owner.OrganizationID, installation.OwnerID) - opName := slackReconcileOperation(t) - fragment := reconcileLoopFragment(t, installation.ID, opName) + opName := harnessReconcileOperation(t, testint.ModeRecurring) t.Run("seeding creates exactly one loop", func(t *testing.T) { require.NoError(t, suite.integrationsRT.ResetReconcileLoops(allowCtx, installation)) diff --git a/internal/graphapi/listeners_integration_cleanup_test.go b/internal/graphapi/listeners_integration_cleanup_test.go index 75370dc21c..a3874e9306 100644 --- a/internal/graphapi/listeners_integration_cleanup_test.go +++ b/internal/graphapi/listeners_integration_cleanup_test.go @@ -3,54 +3,20 @@ package graphapi_test import ( - "context" - "encoding/json" "testing" "github.com/theopenlane/entx" "gotest.tools/v3/assert" - "github.com/theopenlane/core/common/enums" - "github.com/theopenlane/core/common/openapi" - ent "github.com/theopenlane/core/internal/ent/generated" "github.com/theopenlane/core/internal/ent/generated/integration" "github.com/theopenlane/core/internal/ent/generated/privacy" - slackdef "github.com/theopenlane/core/internal/integrations/definitions/slack" ) -// seedConnectedSlackInstallation creates a connected slack installation for the ctx org and -// seeds exactly one reconcile loop, returning the installation and its loop metadata fragment -func seedConnectedSlackInstallation(t *testing.T, ctx context.Context) (*ent.Integration, string) { - t.Helper() - - clientConfig, err := json.Marshal(slackdef.UserInput{}) - assert.NilError(t, err) - - installation, err := suite.client.db.Integration.Create(). - SetName(randomName(t)). - SetKind("slack"). - SetDefinitionID(slackdef.DefinitionID.ID()). - SetStatus(enums.IntegrationStatusConnected). - SetConfig(openapi.IntegrationConfig{ClientConfig: clientConfig}). - Save(ctx) - assert.NilError(t, err) - - fragment := reconcileLoopFragment(t, installation.ID, slackReconcileOperation(t)) - - assert.NilError(t, suite.integrationsRT.ResetReconcileLoops(ctx, installation)) - - suite.WaitForEvents() - - assert.Equal(t, 1, activeReconcileJobs(t, fragment)) - - return installation, fragment -} - func TestIntegrationCleanupListenerHardDelete(t *testing.T) { org := suite.seedFreshMinimalOrgUsers(t, false) allowCtx := privacy.DecisionContext(setContext(org.owner.UserCtx, suite.client.db), privacy.Allow) - installation, fragment := seedConnectedSlackInstallation(t, allowCtx) + installation, fragment := seedHarnessLoop(t, allowCtx) hardDeleteCtx := entx.SkipSoftDelete(allowCtx) @@ -69,7 +35,7 @@ func TestIntegrationCleanupListenerNonStatusUpdateKeepsLoops(t *testing.T) { org := suite.seedFreshMinimalOrgUsers(t, false) allowCtx := privacy.DecisionContext(setContext(org.owner.UserCtx, suite.client.db), privacy.Allow) - installation, fragment := seedConnectedSlackInstallation(t, allowCtx) + installation, fragment := seedHarnessLoop(t, allowCtx) assert.NilError(t, suite.client.db.Integration.UpdateOneID(installation.ID).SetName(randomName(t)).Exec(allowCtx)) diff --git a/internal/graphapi/listeners_organization_cleanup_test.go b/internal/graphapi/listeners_organization_cleanup_test.go index 704cf9d100..0e3b9fa83f 100644 --- a/internal/graphapi/listeners_organization_cleanup_test.go +++ b/internal/graphapi/listeners_organization_cleanup_test.go @@ -32,7 +32,7 @@ func TestOrganizationCleanupListenerCascadeWithIntegrations(t *testing.T) { task1 := (&TaskBuilder{client: suite.client}).MustNew(ownerCtx, t) contact1 := (&ContactBuilder{client: suite.client}).MustNew(ownerCtx, t) - installation, fragment := seedConnectedSlackInstallation(t, allowCtx) + installation, fragment := seedHarnessLoop(t, allowCtx) assert.Equal(t, orgID, installation.OwnerID) resp, err := suite.client.api.DeleteOrganization(ownerCtx, orgID) diff --git a/internal/graphapi/recurring_schedule_test.go b/internal/graphapi/recurring_schedule_test.go new file mode 100644 index 0000000000..79244f17cf --- /dev/null +++ b/internal/graphapi/recurring_schedule_test.go @@ -0,0 +1,73 @@ +//go:build test + +package graphapi_test + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/theopenlane/core/common/enums" + "github.com/theopenlane/core/internal/ent/generated/privacy" + testint "github.com/theopenlane/core/internal/testutils/integrations" +) + +// waitForInstallationErrored polls until the installation is marked unhealthy; the exhausting +// loop reschedules through River's scheduler across several cycles, so it needs a longer window +// than the shared waitForCondition helper allows +func waitForInstallationErrored(t *testing.T, ctx context.Context, id string) { + t.Helper() + + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + inst, err := suite.client.db.Integration.Get(ctx, id) + if err == nil && inst.Status == enums.IntegrationStatusErrored { + return + } + + time.Sleep(200 * time.Millisecond) + } + + t.Fatal("timed out waiting for exhausting loop to mark the installation unhealthy") +} + +// TestReconcileLoopExhaustsToUnhealthy drives a loop whose every cycle fails and asserts the +// runtime stops rescheduling after the error budget and marks the installation unhealthy +func TestReconcileLoopExhaustsToUnhealthy(t *testing.T) { + org := suite.seedFreshMinimalOrgUsers(t, false) + allowCtx := privacy.DecisionContext(setContext(org.owner.UserCtx, suite.client.db), privacy.Allow) + ownerCtx := setContext(org.owner.UserCtx, suite.client.db) + + installation, fragment := newHarnessInstallation(t, allowCtx, testint.ModeExhausting) + + require.NoError(t, suite.integrationsRT.ResetReconcileLoops(allowCtx, installation)) + + waitForInstallationErrored(t, allowCtx, installation.ID) + + suite.WaitForEvents() + + require.Equal(t, 0, activeReconcileJobs(t, fragment)) + require.Equal(t, 1, integrationNotificationCount(t, ownerCtx, installation.OwnerID, integrationReconfigurationRequiredObjectType)) +} + +// TestReconcileLoopUnresolvableClientMarksUnhealthy asserts a loop whose client cannot be built +// is never seeded and the installation is marked unhealthy at seed time +func TestReconcileLoopUnresolvableClientMarksUnhealthy(t *testing.T) { + org := suite.seedFreshMinimalOrgUsers(t, false) + allowCtx := privacy.DecisionContext(setContext(org.owner.UserCtx, suite.client.db), privacy.Allow) + ownerCtx := setContext(org.owner.UserCtx, suite.client.db) + + installation, fragment := newHarnessInstallation(t, allowCtx, testint.ModeUnresolvable) + + require.NoError(t, suite.integrationsRT.ResetReconcileLoops(allowCtx, installation)) + + suite.WaitForEvents() + + require.Equal(t, 0, activeReconcileJobs(t, fragment)) + + reloaded := reloadIntegration(t, allowCtx, installation.ID) + require.Equal(t, enums.IntegrationStatusErrored, reloaded.Status) + require.Equal(t, 1, integrationNotificationCount(t, ownerCtx, installation.OwnerID, integrationReconfigurationRequiredObjectType)) +} diff --git a/internal/graphapi/tools_test.go b/internal/graphapi/tools_test.go index 84352c4a4a..101158cf7c 100644 --- a/internal/graphapi/tools_test.go +++ b/internal/graphapi/tools_test.go @@ -52,6 +52,7 @@ import ( "github.com/theopenlane/core/internal/graphapi/testclient" "github.com/theopenlane/core/internal/httpserve/config" emaildef "github.com/theopenlane/core/internal/integrations/definitions/email" + testint "github.com/theopenlane/core/internal/testutils/integrations" slackdef "github.com/theopenlane/core/internal/integrations/definitions/slack" systemdef "github.com/theopenlane/core/internal/integrations/definitions/system" "github.com/theopenlane/core/internal/integrations/registry" @@ -388,6 +389,7 @@ func (suite *GraphTestSuite) SetupSuite(t *testing.T) { emaildef.Builder(emaildef.MockRuntimeConfig(), false), slackdef.Builder(slackdef.Config{}, &slackdef.RuntimeSlackConfig{WebhookURL: "https://hooks.slack.com/services/test/mock/url"}, false), systemdef.Builder(systemdef.PaymentReminderConfig{}, systemdef.OrganizationDeleteConfig{}), + testint.Builder(), }, }) requireNoError(t, err) diff --git a/internal/integrations/operations/reconcile.go b/internal/integrations/operations/reconcile.go index 2164528a42..bb6fc36703 100644 --- a/internal/integrations/operations/reconcile.go +++ b/internal/integrations/operations/reconcile.go @@ -46,12 +46,13 @@ func LegacyTopicRenames() map[gala.TopicName]gala.TopicName { // ReconcileDefinition builds the Gala listener definition driving every recurring operation // cycle: installation-bound reconciliation and runtime-bound scheduled operations -func ReconcileDefinition(reg *registry.Registry, handle func(context.Context, ReconcileEnvelope) (int, error), schedule gala.Schedule) gala.Definition[ReconcileEnvelope] { +func ReconcileDefinition(reg *registry.Registry, handle func(context.Context, ReconcileEnvelope) (int, error), onExhausted func(context.Context, ReconcileEnvelope, error), schedule gala.Schedule) gala.Definition[ReconcileEnvelope] { return gala.Definition[ReconcileEnvelope]{ Topic: ReconcileTopic, Cancel: func(ctx context.Context, e ReconcileEnvelope, err error) bool { return reconcileShouldCancel(ctx, reg, e, err) }, + OnExhausted: onExhausted, Schedule: &gala.ScheduleSpec[ReconcileEnvelope]{ Schedule: schedule, Handle: handle, @@ -67,16 +68,30 @@ func ReconcileDefinition(reg *registry.Registry, handle func(context.Context, Re return intobvs.EmitContext(ctx, e.OperationContext) }, Override: func(e ReconcileEnvelope) *gala.Schedule { - if reg == nil { - return nil + src := types.IntegrationSourceFrom(e.OperationContext) + + var opSchedule *gala.Schedule + + if reg != nil { + if op, err := reg.Operation(src.DefinitionID, e.Operation); err == nil { + opSchedule = op.Schedule + } } - op, err := reg.Operation(types.IntegrationSourceFrom(e.OperationContext).DefinitionID, e.Operation) - if err != nil { - return nil + if !src.Runtime { + return opSchedule } - return op.Schedule + // runtime-bound sweeps have no installation to mark unhealthy and no reseed + // path besides startup, so they back off forever instead of exhausting + override := schedule + if opSchedule != nil { + override = *opSchedule + } + + override.MaxErrorStreak = gala.UnlimitedErrorStreak + + return &override }, }, } diff --git a/internal/integrations/runtime/execution.go b/internal/integrations/runtime/execution.go index 3b3f32ee01..95ff6cc9d8 100644 --- a/internal/integrations/runtime/execution.go +++ b/internal/integrations/runtime/execution.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "time" "github.com/riverqueue/river" @@ -526,32 +527,6 @@ func (r *Runtime) seedReconcileJobsForInstallation(ctx context.Context, inst *en opCtx := intobvs.WithOperation(ctx, op.Name) - // successor cycles carry per-cycle unique keys, so a live loop only surfaces - // through its metadata, never through the seed's insert-time key - fragment, err := types.PropertiesFragment(map[string]string{ - "entityId": inst.ID, - "operation": op.Name, - "runType": enums.IntegrationRunTypeReconcile.String(), - }) - if err != nil { - errs = append(errs, err) - continue - } - - active, err := r.Gala().HasActiveJobWithMetadata(opCtx, fragment) - if err != nil { - logx.FromContext(opCtx).Error().Err(err).Msg("failed to check for active reconcile job") - errs = append(errs, err) - - continue - } - - if active { - continue - } - - logx.FromContext(opCtx).Info().Msg("seeding reconcile loop") - if err := r.emitReconcileLoop(opCtx, inst, op.Name); err != nil { logx.FromContext(opCtx).Error().Err(err).Msg("failed to seed reconcile job") errs = append(errs, err) @@ -696,7 +671,7 @@ func (r *Runtime) resolveOperationClient(ctx context.Context, integration *ent.I if err != nil { logx.FromContext(ctx).Error().Err(err).Msg("client build failed") - return nil, credentials, integration.DefinitionID, err + return nil, credentials, integration.DefinitionID, types.Unhealthy(err, fmt.Sprintf(clientUnresolvedReasonFmt, err)) } logx.FromContext(ctx).Debug().Msg("client initialized") diff --git a/internal/integrations/runtime/reconcile_loops.go b/internal/integrations/runtime/reconcile_loops.go index f77f0472e8..e8cb609f9b 100644 --- a/internal/integrations/runtime/reconcile_loops.go +++ b/internal/integrations/runtime/reconcile_loops.go @@ -3,6 +3,7 @@ package runtime import ( "context" "errors" + "fmt" "github.com/theopenlane/core/common/enums" ent "github.com/theopenlane/core/internal/ent/generated" @@ -12,8 +13,8 @@ import ( "github.com/theopenlane/core/pkg/logx" ) -// emitReconcileLoop emits the recurring loop for one operation on an installation; the topic's -// UniqueKey derivation collapses concurrent seeds of the same loop to one job +// emitReconcileLoop starts one operation's loop unless a live one exists; the metadata guard is +// what stops seeds from spawning parallel chains, since successor cycles change their unique key func (r *Runtime) emitReconcileLoop(ctx context.Context, installation *ent.Integration, operationName string) error { oc := types.NewOperationContext(installation.OwnerID, operationName, types.IntegrationSource{ IntegrationID: installation.ID, @@ -23,6 +24,37 @@ func (r *Runtime) emitReconcileLoop(ctx context.Context, installation *ent.Integ ctx, headers := intobvs.EmitContext(ctx, oc) + fragment, err := types.PropertiesFragment(map[string]string{ + "entityId": installation.ID, + "operation": operationName, + "runType": enums.IntegrationRunTypeReconcile.String(), + }) + if err != nil { + return err + } + + active, err := r.Gala().HasActiveJobWithMetadata(ctx, fragment) + if err != nil { + return err + } + + if active { + return nil + } + + op, err := r.Registry().Operation(installation.DefinitionID, operationName) + if err != nil { + return err + } + + if op.ClientRef.Valid() { + if _, err := r.BuildClientForIntegration(ctx, installation, op.ClientRef); err != nil { + logx.FromContext(ctx).Error().Err(err).Msg("client unresolved, marking unhealthy instead of seeding loop") + + return r.MarkIntegrationUnhealthy(ctx, installation, fmt.Sprintf(clientUnresolvedReasonFmt, err)) + } + } + if _, err := r.Gala().EmitWithHeaders(ctx, operations.ReconcileTopic.Name, operations.ReconcileEnvelope{OperationContext: oc}, headers); err != nil { return err } @@ -32,6 +64,35 @@ func (r *Runtime) emitReconcileLoop(ctx context.Context, installation *ent.Integ return nil } +// clientUnresolvedReasonFmt formats the actionable reason recorded when an integration cannot establish its client +const clientUnresolvedReasonFmt = "the integration could not establish a connection and needs to be reconnected: %s" + +// reconcileExhaustedReasonFmt formats the user-facing reason recorded on the unhealthy installation +const reconcileExhaustedReasonFmt = "repeated sync failures due to %s" + +// markReconcileExhausted marks the installation unhealthy when its loop exhausts its error budget +func (r *Runtime) markReconcileExhausted(ctx context.Context, e operations.ReconcileEnvelope, cause error) { + src := types.IntegrationSourceFrom(e.OperationContext) + if src.IntegrationID == "" { + return + } + + ctx = intobvs.WithContext(ctx, e.OperationContext) + + installation, err := r.ResolveIntegration(ctx, IntegrationLookup{IntegrationID: src.IntegrationID}) + if err != nil { + logx.FromContext(ctx).Error().Err(err).Msg("failed resolving integration after exhausted reconcile loop") + + return + } + + logx.FromContext(ctx).Error().Err(cause).Msg("reconcile loop exhausted error budget, marking integration unhealthy") + + if err := r.MarkIntegrationUnhealthy(ctx, installation, fmt.Sprintf(reconcileExhaustedReasonFmt, cause)); err != nil { + logx.FromContext(ctx).Error().Err(err).Msg("failed marking integration unhealthy after exhausted reconcile loop") + } +} + // ResetReconcileLoops collapses each reconcilable operation on the installation to exactly one // recurring loop: an operation already running a single loop is left untouched (preserving its // adaptive schedule state), while zero or multiple loops are cancelled and reseeded as one fresh diff --git a/internal/integrations/runtime/runtime.go b/internal/integrations/runtime/runtime.go index bbe2d07c94..d0fb3414c6 100644 --- a/internal/integrations/runtime/runtime.go +++ b/internal/integrations/runtime/runtime.go @@ -265,7 +265,7 @@ func New(config Config) (*Runtime, error) { return nil, err } - if _, err := gala.Register(rt.Gala(), operations.ReconcileDefinition(rt.Registry(), rt.HandleReconcile, gala.Schedule{})); err != nil { + if _, err := gala.Register(rt.Gala(), operations.ReconcileDefinition(rt.Registry(), rt.HandleReconcile, rt.markReconcileExhausted, gala.Schedule{})); err != nil { return nil, err } diff --git a/internal/testutils/integrations/auth.go b/internal/testutils/integrations/auth.go new file mode 100644 index 0000000000..7bff97032b --- /dev/null +++ b/internal/testutils/integrations/auth.go @@ -0,0 +1,69 @@ +//go:build test + +package integrations + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/theopenlane/utils/ulids" + + "github.com/theopenlane/core/internal/integrations/types" +) + +const ( + // OAuthAccessToken is the access token minted by the OAuth completion fixture + OAuthAccessToken = "test-access-token" + // OAuthRefreshToken is the refresh token minted by the OAuth completion fixture + OAuthRefreshToken = "test-refresh-token" +) + +// oauthCallbackPayload carries the OAuth state and code +type oauthCallbackPayload struct { + State string `json:"state"` + Code string `json:"code,omitempty"` +} + +// oauthStart mints an authorize URL and opaque state +func oauthStart(_ context.Context, _ json.RawMessage) (types.AuthStartResult, error) { + oauthState := ulids.New().String() + + stateBytes, err := json.Marshal(oauthCallbackPayload{State: oauthState}) + if err != nil { + return types.AuthStartResult{}, err + } + + return types.AuthStartResult{ + URL: fmt.Sprintf("https://example.com/oauth/authorize?state=%s", oauthState), + State: stateBytes, + }, nil +} + +// oauthComplete validates the callback and mints the OAuth credential +func oauthComplete(_ context.Context, callbackState json.RawMessage, input types.AuthCallbackInput) (types.AuthCompleteResult, error) { + var stored oauthCallbackPayload + if err := json.Unmarshal(callbackState, &stored); err != nil { + return types.AuthCompleteResult{}, err + } + + if input.First("code") == "" { + return types.AuthCompleteResult{}, ErrOAuthCodeMissing + } + + if stored.State != input.First("state") { + return types.AuthCompleteResult{}, ErrOAuthStateMismatch + } + + tokenData, err := json.Marshal(oauthTokenCred{ + AccessToken: OAuthAccessToken, + RefreshToken: OAuthRefreshToken, + }) + if err != nil { + return types.AuthCompleteResult{}, err + } + + return types.AuthCompleteResult{ + Credential: types.CredentialSet{Data: tokenData}, + }, nil +} diff --git a/internal/testutils/integrations/builder.go b/internal/testutils/integrations/builder.go new file mode 100644 index 0000000000..82dc2c8043 --- /dev/null +++ b/internal/testutils/integrations/builder.go @@ -0,0 +1,170 @@ +//go:build test + +package integrations + +import ( + "context" + + "github.com/theopenlane/core/internal/integrations/registry" + "github.com/theopenlane/core/internal/integrations/types" + "github.com/theopenlane/core/pkg/gala" + "github.com/theopenlane/core/pkg/jsonx" +) + +// Builder returns the shared test integration definition; only the reconcile operations are +// input-gated, since seeding sweeps every reconcile-policy operation on a connected installation +func Builder() registry.Builder { + return registry.Builder(func() (types.Definition, error) { + return types.Definition{ + DefinitionSpec: types.DefinitionSpec{ + ID: DefinitionID.ID(), + Family: "Openlane", + DisplayName: "Test Integration", + Description: "Shared test integration definition.", + Category: "system", + Active: true, + Visible: true, + }, + UserInput: &types.UserInputRegistration{ + Schema: jsonx.SchemaFrom[UserInput](), + }, + CredentialRegistrations: []types.CredentialRegistration{ + { + Ref: TokenCredential.ID(), + Name: "Test Token", + Description: "API token the test client is built from.", + Schema: tokenSchema, + }, + { + Ref: OAuthCredential.ID(), + Name: "Test OAuth", + Description: "Auth-managed credential slot filled by the OAuth fixture.", + }, + { + Ref: ServiceAccountCredential.ID(), + Name: "Test Service Account", + Description: "Strict-schema credential slot used by config flows.", + Schema: serviceAccountSchema, + }, + }, + Connections: []types.ConnectionRegistration{ + { + CredentialRef: OAuthCredential.ID(), + Name: "Test OAuth", + Description: "Authenticate through the OAuth callback fixture.", + CredentialRefs: []types.CredentialSlotID{OAuthCredential.ID()}, + Auth: &types.AuthRegistration{ + CredentialRef: OAuthCredential.ID(), + Start: oauthStart, + Complete: oauthComplete, + }, + Disconnect: &types.DisconnectRegistration{ + CredentialRef: OAuthCredential.ID(), + Description: "Remove the persisted OAuth credential and disconnect this installation.", + }, + }, + { + CredentialRef: TokenCredential.ID(), + Name: "Test Token", + Description: "Connect with an API token validated by the health check.", + CredentialRefs: []types.CredentialSlotID{TokenCredential.ID()}, + ValidationOperation: HealthOp.Name(), + Disconnect: &types.DisconnectRegistration{ + CredentialRef: TokenCredential.ID(), + Description: "Remove the persisted token credential and disconnect this installation.", + }, + }, + { + CredentialRef: ServiceAccountCredential.ID(), + Name: "Test Service Account", + Description: "Connect with a service account validated by the health check.", + CredentialRefs: []types.CredentialSlotID{ServiceAccountCredential.ID()}, + ValidationOperation: HealthOp.Name(), + Disconnect: &types.DisconnectRegistration{ + CredentialRef: ServiceAccountCredential.ID(), + Description: "Remove the persisted service account credential and disconnect this installation.", + }, + }, + }, + Clients: []types.ClientRegistration{ + { + Ref: testClient.ID(), + CredentialRefs: []types.CredentialSlotID{TokenCredential.ID()}, + Description: "Test client built from the stored token credential.", + Build: buildClient, + }, + }, + Webhooks: []types.WebhookRegistration{ + { + Name: "inbound.events", + Event: webhookInboundEvent, + Events: []types.WebhookEventRegistration{ + { + Name: WebhookAlertCreated.Name(), + Topic: DefinitionID.WebhookEventTopic(WebhookAlertCreated.Name()), + Handle: func(context.Context, types.WebhookHandleRequest) error { return nil }, + }, + }, + }, + }, + Operations: []types.OperationRegistration{ + { + Name: HealthOp.Name(), + Description: "Validate the bound credential", + Topic: DefinitionID.OperationTopic(HealthOp.Name()), + ConfigSchema: healthSchema, + Policy: types.ExecutionPolicy{Inline: true}, + Handle: healthHandler, + }, + { + Name: RepoSyncOp.Name(), + Description: "Async operation running with the built client", + Topic: DefinitionID.OperationTopic(RepoSyncOp.Name()), + ClientRef: testClient.ID(), + ConfigSchema: repoSyncSchema, + Policy: types.ExecutionPolicy{}, + Handle: repoSyncHandler, + }, + { + Name: ValidatedOp.Name(), + Description: "Inline operation with a required config field", + Topic: DefinitionID.OperationTopic(ValidatedOp.Name()), + ConfigSchema: validatedSchema, + Policy: types.ExecutionPolicy{Inline: true}, + Handle: validatedHandler, + }, + { + Name: RecurringOp.Name(), + Description: "Healthy idle reconcile loop", + Topic: DefinitionID.OperationTopic(RecurringOp.Name()), + ConfigSchema: recurringSchema, + Policy: types.ExecutionPolicy{Reconcile: true}, + Schedule: &gala.Schedule{MinInterval: recurringInterval}, + Handle: idleCycle, + Disabled: disabledUnlessMode(ModeRecurring), + }, + { + Name: ExhaustingOp.Name(), + Description: "Always-failing reconcile loop for exhaustion", + Topic: DefinitionID.OperationTopic(ExhaustingOp.Name()), + ConfigSchema: exhaustingSchema, + Policy: types.ExecutionPolicy{Reconcile: true}, + Schedule: &gala.Schedule{MinInterval: exhaustingInterval, MaxErrorStreak: exhaustingMaxErrorStreak}, + Handle: failingCycle, + Disabled: disabledUnlessMode(ModeExhausting), + }, + { + Name: UnresolvableOp.Name(), + Description: "Reconcile loop whose client cannot resolve without a stored credential", + Topic: DefinitionID.OperationTopic(UnresolvableOp.Name()), + ClientRef: testClient.ID(), + ConfigSchema: unresolvableSchema, + Policy: types.ExecutionPolicy{Reconcile: true}, + Schedule: &gala.Schedule{MinInterval: recurringInterval}, + Handle: idleCycle, + Disabled: disabledUnlessMode(ModeUnresolvable), + }, + }, + }, nil + }) +} diff --git a/internal/testutils/integrations/builder_test.go b/internal/testutils/integrations/builder_test.go new file mode 100644 index 0000000000..0d761911af --- /dev/null +++ b/internal/testutils/integrations/builder_test.go @@ -0,0 +1,52 @@ +//go:build test + +package integrations + +import ( + "testing" + + "github.com/theopenlane/core/internal/integrations/registry" +) + +func TestBuilderRegistersAllSurfaces(t *testing.T) { + reg := registry.New() + if err := reg.RegisterAll(Builder()); err != nil { + t.Fatalf("register testdef definition: %v", err) + } + + names := map[string]struct{}{} + for _, name := range []string{HealthOp.Name(), RepoSyncOp.Name(), ValidatedOp.Name(), RecurringOp.Name(), ExhaustingOp.Name(), UnresolvableOp.Name()} { + if name == "" { + t.Fatal("operation registered under empty name") + } + + if _, dup := names[name]; dup { + t.Fatalf("operation name %q is not unique", name) + } + + names[name] = struct{}{} + + if _, err := reg.Operation(DefinitionID.ID(), name); err != nil { + t.Fatalf("operation %q not registered: %v", name, err) + } + } + + for _, reconcileOp := range []string{RecurringOp.Name(), ExhaustingOp.Name(), UnresolvableOp.Name()} { + op, err := reg.Operation(DefinitionID.ID(), reconcileOp) + if err != nil { + t.Fatalf("operation %q not registered: %v", reconcileOp, err) + } + + if !op.Policy.Reconcile { + t.Fatalf("operation %q is not a reconcile operation", reconcileOp) + } + } + + if _, err := reg.Client(DefinitionID.ID(), testClient.ID()); err != nil { + t.Fatalf("test client not registered: %v", err) + } + + if _, err := reg.Webhook(DefinitionID.ID(), "inbound.events"); err != nil { + t.Fatalf("webhook contract not registered: %v", err) + } +} diff --git a/internal/testutils/integrations/client.go b/internal/testutils/integrations/client.go new file mode 100644 index 0000000000..b1b2e3aa48 --- /dev/null +++ b/internal/testutils/integrations/client.go @@ -0,0 +1,25 @@ +//go:build test + +package integrations + +import ( + "context" + + "github.com/theopenlane/core/internal/integrations/types" +) + +// Client is the functional test client built from the stored token credential +type Client struct { + // Token is the resolved API token + Token string +} + +// buildClient constructs the client from the stored token credential +func buildClient(_ context.Context, req types.ClientBuildRequest) (any, error) { + cred, ok, err := TokenCredential.Resolve(req.Credentials) + if err != nil || !ok || cred.Token == "" { + return nil, ErrTokenMissing + } + + return &Client{Token: cred.Token}, nil +} diff --git a/internal/testutils/integrations/doc.go b/internal/testutils/integrations/doc.go new file mode 100644 index 0000000000..c177463331 --- /dev/null +++ b/internal/testutils/integrations/doc.go @@ -0,0 +1,4 @@ +//go:build test + +// Package integrations provides the shared prod-shaped test integration definition +package integrations diff --git a/internal/testutils/integrations/errors.go b/internal/testutils/integrations/errors.go new file mode 100644 index 0000000000..74ff68e831 --- /dev/null +++ b/internal/testutils/integrations/errors.go @@ -0,0 +1,18 @@ +//go:build test + +package integrations + +import "errors" + +var ( + // ErrCycleFailed is returned by the always-failing reconcile operation + ErrCycleFailed = errors.New("integrations: cycle failed") + // ErrTokenMissing indicates no usable token credential is stored for the installation + ErrTokenMissing = errors.New("integrations: token missing") + // ErrHealthFailed is returned by the health check when a failure marker credential is bound + ErrHealthFailed = errors.New("integrations: health failed") + // ErrOAuthCodeMissing indicates the OAuth callback carried no code + ErrOAuthCodeMissing = errors.New("integrations: missing oauth code") + // ErrOAuthStateMismatch indicates the OAuth callback state did not match + ErrOAuthStateMismatch = errors.New("integrations: oauth state mismatch") +) diff --git a/internal/testutils/integrations/handlers.go b/internal/testutils/integrations/handlers.go new file mode 100644 index 0000000000..924f4fe0f4 --- /dev/null +++ b/internal/testutils/integrations/handlers.go @@ -0,0 +1,75 @@ +//go:build test + +package integrations + +import ( + "context" + "encoding/json" + + "github.com/theopenlane/core/internal/integrations/providerkit" + + "github.com/theopenlane/core/internal/integrations/types" +) + +// healthHandler validates the bound credential, failing on the marker values +func healthHandler(_ context.Context, req types.OperationRequest) (json.RawMessage, error) { + if sa, ok, err := ServiceAccountCredential.Resolve(req.Credentials); err == nil && ok && sa.ProjectID == FailProjectID { + return nil, ErrHealthFailed + } + + if tok, ok, err := TokenCredential.Resolve(req.Credentials); err == nil && ok && tok.Token == FailToken { + return nil, ErrHealthFailed + } + + return json.RawMessage(`{"ok":true}`), nil +} + +// repoSyncHandler runs with the built client +func repoSyncHandler(_ context.Context, req types.OperationRequest) (json.RawMessage, error) { + if _, ok := req.Client.(*Client); !ok { + return nil, ErrTokenMissing + } + + return json.RawMessage(`{"synced":true}`), nil +} + +// validatedHandler backs the inline operation with a required config field +func validatedHandler(context.Context, types.OperationRequest) (json.RawMessage, error) { + return json.RawMessage(`{"validated":true}`), nil +} + +// idleCycle reports zero drift +func idleCycle(context.Context, types.OperationRequest) (json.RawMessage, error) { + return nil, nil +} + +// failingCycle always fails +func failingCycle(context.Context, types.OperationRequest) (json.RawMessage, error) { + return nil, ErrCycleFailed +} + +// webhookInboundEvent decodes the inbound webhook payload +func webhookInboundEvent(req types.WebhookInboundRequest) (types.WebhookReceivedEvent, error) { + var envelope struct { + Event string `json:"event"` + DeliveryID string `json:"delivery_id"` + } + if err := json.Unmarshal(req.Payload, &envelope); err != nil { + return types.WebhookReceivedEvent{}, err + } + + if envelope.Event == "" { + return types.WebhookReceivedEvent{}, nil + } + + return types.WebhookReceivedEvent{ + Name: envelope.Event, + DeliveryID: envelope.DeliveryID, + Payload: req.Payload, + }, nil +} + +// disabledUnlessMode gates a reconcile operation on the installation's mode input +func disabledUnlessMode(mode string) func(json.RawMessage) bool { + return providerkit.DisabledWhen(func(u UserInput) bool { return u.Mode != mode }) +} diff --git a/internal/testutils/integrations/types.go b/internal/testutils/integrations/types.go new file mode 100644 index 0000000000..f550eb0503 --- /dev/null +++ b/internal/testutils/integrations/types.go @@ -0,0 +1,142 @@ +//go:build test + +package integrations + +import ( + "encoding/json" + "time" + + "github.com/theopenlane/core/internal/integrations/providerkit" + "github.com/theopenlane/core/internal/integrations/types" +) + +// DefinitionID is the stable identifier for the shared test integration definition +var DefinitionID = types.NewDefinitionRef("def_01K0TESTDEF0000000000000001") + +var ( + // HealthOp is the inline health check used as every connection's validation operation + healthSchema, HealthOp = providerkit.OperationSchema[healthCheck]() + // RepoSyncOp is the async client-resolving operation + repoSyncSchema, RepoSyncOp = providerkit.OperationSchema[repoSync]() + // ValidatedOp is the inline operation with a required config field + validatedSchema, ValidatedOp = providerkit.OperationSchema[validatedRun]() + // RecurringOp is the healthy idle loop + recurringSchema, RecurringOp = providerkit.OperationSchema[recurringCycle]() + // ExhaustingOp is the always-failing loop + exhaustingSchema, ExhaustingOp = providerkit.OperationSchema[exhaustingCycle]() + // UnresolvableOp is the client-resolving loop seeded without a credential + unresolvableSchema, UnresolvableOp = providerkit.OperationSchema[unresolvableCycle]() + + // TokenCredential is the credential slot the test client is built from + tokenSchema, TokenCredential = providerkit.CredentialSchema[tokenCred]() + // OAuthCredential is the auth-managed slot filled by the OAuth fixture + _, OAuthCredential = providerkit.CredentialSchema[oauthTokenCred]() + // ServiceAccountCredential is the strict-schema slot used by config flows + serviceAccountSchema, ServiceAccountCredential = providerkit.CredentialSchema[serviceAccountCred]() + + // testClient builds from the token credential + testClient = types.NewClientRef[*Client]() + + // WebhookAlertCreated is the webhook event contract + WebhookAlertCreated = types.NewWebhookEventRef[webhookAlertEnvelope]("alert.created") +) + +const ( + // ModeRecurring seeds a healthy idle loop + ModeRecurring = "recurring" + // ModeExhausting seeds a loop whose every cycle fails + ModeExhausting = "exhausting" + // ModeUnresolvable seeds a client-resolving loop without a credential + ModeUnresolvable = "unresolvable" +) + +const ( + // FailProjectID fails the health check when set as the service-account project id + FailProjectID = "fail-project" + // FailToken fails the health check when set as the token value + FailToken = "fail" +) + +const ( + recurringInterval = time.Hour + exhaustingInterval = time.Millisecond + exhaustingMaxErrorStreak = 3 +) + +type healthCheck struct{} + +type repoSync struct{} + +// validatedRun is the config for the inline operation with a required field +type validatedRun struct { + // Target is the required target field + Target string `json:"target" jsonschema:"required"` +} + +type recurringCycle struct{} + +type exhaustingCycle struct{} + +type unresolvableCycle struct{} + +// tokenCred is the credential material the test client is built from +type tokenCred struct { + // Token is the API token + Token string `json:"token"` +} + +// oauthTokenCred is the credential material minted by the OAuth fixture +type oauthTokenCred struct { + // AccessToken is the OAuth2 access token + AccessToken string `json:"access_token"` + // RefreshToken is the OAuth2 refresh token + RefreshToken string `json:"refresh_token,omitempty"` +} + +// serviceAccountCred is the strict credential material used by config flows +type serviceAccountCred struct { + // ProjectID is the required project identifier + ProjectID string `json:"projectId" jsonschema:"required"` + // ServiceAccountEmail is the required service account email + ServiceAccountEmail string `json:"serviceAccountEmail" jsonschema:"required"` +} + +// UserInput is the installation-scoped user input for the test definition +type UserInput struct { + // Mode selects which recurring loop operation is active + Mode string `json:"mode,omitempty" jsonschema:"title=Scheduling Mode"` + // FilterExpr is a free-form filter expression + FilterExpr string `json:"filterExpr,omitempty" jsonschema:"title=Filter Expression"` +} + +type webhookAlertEnvelope struct{} + +// ModeInput returns the installation user input selecting one scheduling mode +func ModeInput(mode string) json.RawMessage { + raw, err := json.Marshal(UserInput{Mode: mode}) + if err != nil { + panic(err) + } + + return raw +} + +// TokenCredentialSet builds the token credential payload +func TokenCredentialSet(token string) types.CredentialSet { + raw, err := json.Marshal(tokenCred{Token: token}) + if err != nil { + panic(err) + } + + return types.CredentialSet{Data: raw} +} + +// ServiceAccountCredentialSet builds the strict credential payload +func ServiceAccountCredentialSet(projectID, email string) types.CredentialSet { + raw, err := json.Marshal(serviceAccountCred{ProjectID: projectID, ServiceAccountEmail: email}) + if err != nil { + panic(err) + } + + return types.CredentialSet{Data: raw} +} diff --git a/pkg/gala/listener.go b/pkg/gala/listener.go index 340d023374..fc82257ecf 100644 --- a/pkg/gala/listener.go +++ b/pkg/gala/listener.go @@ -71,6 +71,8 @@ type Definition[T any] struct { LogFields func(T) map[string]any // Cancel optionally classifies a handler error as terminal, converting it to river.JobCancel Cancel func(context.Context, T, error) bool + // OnExhausted runs when a scheduled loop stops on its error-streak budget + OnExhausted func(context.Context, T, error) // Schedule makes this listener a self-sustaining adaptive re-emit loop when non-nil; // exactly one of Handle and Schedule.Handle must be set Schedule *ScheduleSpec[T] diff --git a/pkg/gala/schedule.go b/pkg/gala/schedule.go index 585f027a73..534c7425e9 100644 --- a/pkg/gala/schedule.go +++ b/pkg/gala/schedule.go @@ -23,11 +23,14 @@ const ( defaultHighDriftThreshold = 200 // intervalHalving is the divisor used to halve the interval on positive drift intervalHalving = 2 - + // defaultMaxErrorStreak stops a loop after this many consecutive failed cycles + defaultMaxErrorStreak = 5 // fullFetchMinInterval is the minimum interval for operations that always fetch all records fullFetchMinInterval = time.Hour // FullHighDriftThreshold is the delta above which a full-fetch schedule snaps to minimum FullHighDriftThreshold = 1000 + // UnlimitedErrorStreak disables error-streak exhaustion + UnlimitedErrorStreak = -1 ) // Schedule defines the adaptive scheduling policy for recurring work @@ -40,6 +43,8 @@ type Schedule struct { BackoffFactor float64 `json:"backoff_factor"` // HighDriftThreshold is the delta count above which the interval resets to MinInterval HighDriftThreshold int `json:"high_drift_threshold"` + // MaxErrorStreak stops the loop after this many consecutive failed cycles; UnlimitedErrorStreak disables exhaustion + MaxErrorStreak int `json:"max_error_streak"` } // ScheduleSpec declares the adaptive re-emit loop for a scheduled listener definition @@ -66,19 +71,32 @@ func scheduleHandler[T any](g *Gala, definition Definition[T]) Handler[T] { delta, execErr := spec.Handle(ctx.Context, payload) state := spec.State(payload) + effectiveSchedule := spec.Schedule + if spec.Override != nil { + if override := spec.Override(payload); override != nil { + effectiveSchedule = *override + } + } + if execErr != nil { if definition.Cancel != nil && definition.Cancel(ctx.Context, payload, execErr) { + logx.FromContext(ctx.Context).Error().Err(execErr).Msg("scheduled listener cycle failed, canceling loop") + return river.JobCancel(execErr) } - logx.FromContext(ctx.Context).Warn().Err(execErr).Int("error_streak", state.ErrorStreak+1).Msg("scheduled listener cycle failed, scheduling retry with backoff") - } + streak := state.ErrorStreak + 1 + if effectiveSchedule.exhausted(streak) { + if definition.OnExhausted != nil { + definition.OnExhausted(ctx.Context, payload, execErr) + } - effectiveSchedule := spec.Schedule - if spec.Override != nil { - if override := spec.Override(payload); override != nil { - effectiveSchedule = *override + logx.FromContext(ctx.Context).Error().Err(execErr).Int("error_streak", streak).Msg("scheduled listener exhausted error budget, stopping loop") + + return river.JobCancel(execErr) } + + logx.FromContext(ctx.Context).Warn().Err(execErr).Int("error_streak", streak).Msg("scheduled listener cycle failed, scheduling retry with backoff") } next := effectiveSchedule.Next(state, delta, execErr) @@ -222,5 +240,16 @@ func (s Schedule) withDefaults() Schedule { s.HighDriftThreshold = defaultHighDriftThreshold } + if s.MaxErrorStreak == 0 { + s.MaxErrorStreak = defaultMaxErrorStreak + } + return s } + +// exhausted reports whether a consecutive-error streak has reached the stop threshold +func (s Schedule) exhausted(streak int) bool { + limit := s.withDefaults().MaxErrorStreak + + return limit > 0 && streak >= limit +} diff --git a/pkg/gala/schedule_test.go b/pkg/gala/schedule_test.go index b0f0899b21..9500fdbca6 100644 --- a/pkg/gala/schedule_test.go +++ b/pkg/gala/schedule_test.go @@ -196,6 +196,10 @@ func TestScheduleWithDefaultsFillsZeroValues(t *testing.T) { if filled.HighDriftThreshold != defaultHighDriftThreshold { t.Fatalf("expected HighDriftThreshold %v, got %v", defaultHighDriftThreshold, filled.HighDriftThreshold) } + + if filled.MaxErrorStreak != defaultMaxErrorStreak { + t.Fatalf("expected MaxErrorStreak %v, got %v", defaultMaxErrorStreak, filled.MaxErrorStreak) + } } func TestScheduleHandlerMarksSuccessorUniqueOnce(t *testing.T) { @@ -354,3 +358,73 @@ func TestScheduleStateIncarnationJSONCompatibility(t *testing.T) { t.Fatalf("incarnation missing from encoded state: %s", encoded) } } + +func TestScheduleHandlerStopsLoopWhenErrorStreakExhausted(t *testing.T) { + execErr := errors.New("cycle failed") + + cases := []struct { + slug string + priorStreak int + wantSuccessor bool + wantExhausted bool + }{ + {slug: "below", priorStreak: 1, wantSuccessor: true, wantExhausted: false}, + {slug: "at", priorStreak: 2, wantSuccessor: false, wantExhausted: true}, + } + + for _, tc := range cases { + t.Run(tc.slug, func(t *testing.T) { + dispatcher := &runtimeTestDispatcher{} + runtime := newTestGala(t, dispatcher) + topic := Topic[runtimeTestPayload]{ + Name: TopicName("gala.test.schedule.exhausted." + tc.slug), + Kind: System.Kind(), + UniqueKey: func(payload runtimeTestPayload) string { return payload.Message }, + } + if err := registerTopic(runtime.registry, topic); err != nil { + t.Fatalf("failed to register topic: %v", err) + } + + exhaustedCalls := 0 + + var exhaustedErr error + + definition := Definition[runtimeTestPayload]{ + Topic: topic, + OnExhausted: func(_ context.Context, _ runtimeTestPayload, err error) { + exhaustedCalls++ + exhaustedErr = err + }, + Schedule: &ScheduleSpec[runtimeTestPayload]{ + Schedule: Schedule{MinInterval: time.Millisecond, MaxErrorStreak: 3}, + Handle: func(context.Context, runtimeTestPayload) (int, error) { return 0, execErr }, + State: func(runtimeTestPayload) ScheduleState { return ScheduleState{ErrorStreak: tc.priorStreak} }, + Wrap: func(payload runtimeTestPayload, _ ScheduleState) runtimeTestPayload { return payload }, + }, + } + + err := scheduleHandler(runtime, definition)(HandlerContext{Context: context.Background()}, runtimeTestPayload{Message: "loop"}) + + // both paths cancel the current job; the difference is whether a successor was scheduled first + if _, ok := errors.AsType[*river.JobCancelError](err); !ok { + t.Fatalf("expected JobCancel, got %v", err) + } + if !errors.Is(err, execErr) { + t.Fatalf("expected execution error in chain, got %v", err) + } + + if gotSuccessor := len(dispatcher.envelopes) > 0; gotSuccessor != tc.wantSuccessor { + t.Fatalf("successor emitted = %v, want %v", gotSuccessor, tc.wantSuccessor) + } + + switch { + case tc.wantExhausted && exhaustedCalls != 1: + t.Fatalf("expected OnExhausted called once, got %d", exhaustedCalls) + case tc.wantExhausted && !errors.Is(exhaustedErr, execErr): + t.Fatalf("OnExhausted received %v, want chain containing %v", exhaustedErr, execErr) + case !tc.wantExhausted && exhaustedCalls != 0: + t.Fatalf("expected OnExhausted not called, got %d", exhaustedCalls) + } + }) + } +}