Fix: Stop recurring integration loops on repeated failures + surface unhealthy installations

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))

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))
 

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)

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))
+}

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)

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
 			},
 		},
 	}

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")