diff --git a/pkg/api/merge_gate_record_test.go b/pkg/api/merge_gate_record_test.go index c501fcba7..2025b83e2 100644 --- a/pkg/api/merge_gate_record_test.go +++ b/pkg/api/merge_gate_record_test.go @@ -90,6 +90,7 @@ func TestRecordMergeGateGatedOnConsumer(t *testing.T) { require.Len(t, gateStore.recorded, 1) recorded := gateStore.recorded[0] assert.Equal(t, "apply-gate-test", recorded.ApplyIdentifier) + assert.Equal(t, storage.MergeGateKindSettle, recorded.Kind) assert.Equal(t, "gate_db", recorded.DatabaseName) assert.Equal(t, "mysql", recorded.DatabaseType) assert.Equal(t, "staging", recorded.Environment) diff --git a/pkg/api/operator.go b/pkg/api/operator.go index 1cbe9560c..abdf073cd 100644 --- a/pkg/api/operator.go +++ b/pkg/api/operator.go @@ -1072,6 +1072,7 @@ func (s *Service) recordMergeGateIfApplyResolved(ctx context.Context, driverID i recorded, err := s.storage.MergeGateRequests().Record(ctx, &storage.MergeGateRequest{ ApplyID: apply.ID, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: apply.ApplyIdentifier, Environment: apply.Environment, DatabaseType: apply.DatabaseType, diff --git a/pkg/schema/mysql/merge_gate_requests.sql b/pkg/schema/mysql/merge_gate_requests.sql index 28aef2ce9..6eff4c984 100644 --- a/pkg/schema/mysql/merge_gate_requests.sql +++ b/pkg/schema/mysql/merge_gate_requests.sql @@ -1,6 +1,7 @@ CREATE TABLE `merge_gate_requests` ( `id` bigint unsigned NOT NULL AUTO_INCREMENT, `apply_id` bigint unsigned NOT NULL, + `kind` varchar(20) NOT NULL, `apply_identifier` varchar(255) NOT NULL, `environment` varchar(50) NOT NULL, `database_type` varchar(50) NOT NULL, @@ -16,11 +17,12 @@ CREATE TABLE `merge_gate_requests` ( `lease_expires_at` datetime(6) DEFAULT NULL, `retry_after` datetime DEFAULT NULL, `last_error` text, + `holds_recorded_at` datetime DEFAULT NULL, `completed_at` datetime DEFAULT NULL, `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), - UNIQUE KEY `idx_merge_gate_apply` (`apply_id`), + UNIQUE KEY `idx_merge_gate_apply` (`apply_id`,`kind`), KEY `idx_merge_gate_claimable` (`state`,`retry_after`,`lease_expires_at`,`created_at`), KEY `idx_merge_gate_target` (`environment`,`database_type`,`database_name`,`state`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci diff --git a/pkg/storage/mysqlstore/checks.go b/pkg/storage/mysqlstore/checks.go index 310a462b7..4591e055f 100644 --- a/pkg/storage/mysqlstore/checks.go +++ b/pkg/storage/mysqlstore/checks.go @@ -532,7 +532,37 @@ func (s *checkStore) GetByTarget(ctx context.Context, environment, dbType, datab // a newer commit does not match and is preserved. An in-progress apply-owned // row is never touched — the started apply's lifecycle stays authoritative. func (s *checkStore) MarkBlockedForFailedRefresh(ctx context.Context, check *storage.Check) (bool, error) { - result, err := s.db.ExecContext(ctx, ` + flipped, err := s.markBlockedConditional(ctx, check, false) + if err != nil { + return false, fmt.Errorf("mark check blocked for failed refresh %s#%d %s/%s/%s (head %s): %w", + check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, err) + } + return flipped, nil +} + +// MarkBlockedForApplyInFlight flips stored check state to a blocking +// conclusion while an apply on the same target runs. Same conditional-write +// contract as MarkBlockedForFailedRefresh, plus rows already holding the same +// blocking reason are skipped so a retried preflight fan-out reports +// flipped=false instead of re-flipping (and re-announcing) the hold. +func (s *checkStore) MarkBlockedForApplyInFlight(ctx context.Context, check *storage.Check) (bool, error) { + flipped, err := s.markBlockedConditional(ctx, check, true) + if err != nil { + return false, fmt.Errorf("mark check blocked for apply in flight %s#%d %s/%s/%s (head %s): %w", + check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, err) + } + return flipped, nil +} + +// markBlockedConditional is the shared conditional blocking flip: it writes +// the caller's blocking conclusion only when the stored row still holds the +// head SHA the caller read (a racing synchronize that re-planned a newer +// commit wins) and is not an in-progress apply-owned row (a started apply's +// lifecycle stays authoritative). With skipAlreadyHeld, rows whose +// blocking_reason already equals the caller's are also left untouched, so +// idempotent retries can distinguish "newly flipped" from "already held". +func (s *checkStore) markBlockedConditional(ctx context.Context, check *storage.Check, skipAlreadyHeld bool) (bool, error) { + query := ` UPDATE checks SET apply_id = NULL, has_changes = ?, @@ -544,19 +574,23 @@ func (s *checkStore) MarkBlockedForFailedRefresh(ctx context.Context, check *sto WHERE repository = ? AND pull_request = ? AND environment = ? AND database_type = ? AND database_name = ? AND head_sha = ? - AND NOT (status = ? AND apply_id IS NOT NULL) - `, check.HasChanges, check.Status, check.Conclusion, check.BlockingReason, check.ErrorMessage, nullString(check.ChangeSummary), + AND NOT (status = ? AND apply_id IS NOT NULL)` + args := []any{check.HasChanges, check.Status, check.Conclusion, check.BlockingReason, check.ErrorMessage, nullString(check.ChangeSummary), check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, - checkStatusInProgress) + checkStatusInProgress} + if skipAlreadyHeld { + query += ` + AND (blocking_reason IS NULL OR blocking_reason != ?)` + args = append(args, check.BlockingReason) + } + result, err := s.db.ExecContext(ctx, query, args...) if err != nil { - return false, fmt.Errorf("mark check blocked for failed refresh %s#%d %s/%s/%s (head %s): %w", - check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, err) + return false, err } rows, err := result.RowsAffected() if err != nil { - return false, fmt.Errorf("rows affected marking check blocked for failed refresh %s#%d %s/%s/%s: %w", - check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, err) + return false, fmt.Errorf("rows affected: %w", err) } return rows > 0, nil } diff --git a/pkg/storage/mysqlstore/merge_gate_requests.go b/pkg/storage/mysqlstore/merge_gate_requests.go index b30914618..3831346bb 100644 --- a/pkg/storage/mysqlstore/merge_gate_requests.go +++ b/pkg/storage/mysqlstore/merge_gate_requests.go @@ -17,10 +17,10 @@ import ( "github.com/block/schemabot/pkg/storage" ) -const mergeGateColumns = `id, apply_id, apply_identifier, environment, database_type, database_name, +const mergeGateColumns = `id, apply_id, kind, apply_identifier, environment, database_type, database_name, provider, repository, change_key, requested_by, state, attempts, lease_owner, lease_token, lease_expires_at, retry_after, last_error, - completed_at, created_at, updated_at` + holds_recorded_at, completed_at, created_at, updated_at` type mergeGateRequestStore struct { db *sql.DB @@ -35,6 +35,9 @@ func (s *mergeGateRequestStore) Record(ctx context.Context, req *storage.MergeGa if req.ApplyIdentifier == "" { return false, fmt.Errorf("merge gate request for apply row %d requires the apply identifier", req.ApplyID) } + if req.Kind != storage.MergeGateKindPreflight && req.Kind != storage.MergeGateKindSettle { + return false, fmt.Errorf("merge gate request for apply %s has unknown kind %q", req.ApplyIdentifier, req.Kind) + } if req.Environment == "" || req.DatabaseType == "" || req.DatabaseName == "" { return false, fmt.Errorf("merge gate request for apply %s requires environment, database type, and database name", req.ApplyIdentifier) } @@ -46,20 +49,21 @@ func (s *mergeGateRequestStore) Record(ctx context.Context, req *storage.MergeGa id, err := s.identity.InsertID(ctx, s.db, ` INSERT INTO merge_gate_requests ( - apply_id, apply_identifier, environment, database_type, database_name, + apply_id, kind, apply_identifier, environment, database_type, database_name, provider, repository, change_key, requested_by, state - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `, req.ApplyID, req.ApplyIdentifier, req.Environment, req.DatabaseType, req.DatabaseName, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, req.ApplyID, req.Kind, req.ApplyIdentifier, req.Environment, req.DatabaseType, req.DatabaseName, provider, req.Repository, req.ChangeKey, req.RequestedBy, storage.MergeGatePending) if err != nil { - // The unique key on apply_id is the idempotency guard: the drive tail - // and the backfill sweep may both record the same apply, and a request - // already in any state means the fan-out is recorded or done. + // The unique key on (apply_id, kind) is the idempotency guard: the + // drive tails, the operator preflight gate, and the backfill sweeps may + // all record the same apply and kind, and a request already in any + // state means the fan-out is recorded or done. if isDuplicateKeyError(err) { return false, nil } - return false, fmt.Errorf("record merge gate request for apply %s (%s/%s in %s): %w", - req.ApplyIdentifier, req.DatabaseType, req.DatabaseName, req.Environment, err) + return false, fmt.Errorf("record %s merge gate request for apply %s (%s/%s in %s): %w", + req.Kind, req.ApplyIdentifier, req.DatabaseType, req.DatabaseName, req.Environment, err) } req.ID = id req.Provider = provider @@ -67,19 +71,75 @@ func (s *mergeGateRequestStore) Record(ctx context.Context, req *storage.MergeGa return true, nil } -func (s *mergeGateRequestStore) GetByApplyID(ctx context.Context, applyID int64) (*storage.MergeGateRequest, error) { +func (s *mergeGateRequestStore) GetByApplyAndKind(ctx context.Context, applyID int64, kind string) (*storage.MergeGateRequest, error) { row := s.db.QueryRowContext(ctx, ` SELECT `+mergeGateColumns+` FROM merge_gate_requests - WHERE apply_id = ? - `, applyID) + WHERE apply_id = ? AND kind = ? + `, applyID, kind) req, err := scanMergeGateRequest(row) if err != nil { - return nil, fmt.Errorf("get merge gate request for apply row %d: %w", applyID, err) + return nil, fmt.Errorf("get %s merge gate request for apply row %d: %w", kind, applyID, err) } return req, nil } +// ReopenForRetry re-arms a terminally failed request back to pending with a +// fresh attempt budget, so the operator preflight gate can recover an apply +// whose preflight exhausted its retries during an outage. Conditional on the +// failed state: pending, processing, and completed rows are left untouched. +func (s *mergeGateRequestStore) ReopenForRetry(ctx context.Context, id int64) (bool, error) { + result, err := s.db.ExecContext(ctx, ` + UPDATE merge_gate_requests + SET state = ?, attempts = 0, lease_owner = NULL, lease_token = NULL, + lease_expires_at = NULL, retry_after = NULL, completed_at = NULL, + updated_at = NOW() + WHERE id = ? AND state = ? + `, storage.MergeGatePending, id, storage.MergeGateFailed) + if err != nil { + return false, fmt.Errorf("reopen merge gate request %d for retry: %w", id, err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read reopened merge gate request %d rows affected: %w", id, err) + } + return rows > 0, nil +} + +// ReopenTerminalPreflightsForActiveApplies re-arms terminally failed +// preflight requests whose apply is still non-terminal, so a code-host +// rendering that exhausted its retries (the stored holds themselves are +// storage-only and cannot fail on the code host) keeps retrying for as long +// as the apply runs. Terminal means unclaimable: no retry window, or attempts +// at the cap. Requests for terminal applies are left alone — the apply's +// settle re-plan supersedes the render. +func (s *mergeGateRequestStore) ReopenTerminalPreflightsForActiveApplies(ctx context.Context) (int64, error) { + nonTerminal, nonTerminalArgs := nonTerminalApplyStatePredicate("a.state") + args := []any{ + storage.MergeGatePending, + storage.MergeGateKindPreflight, storage.MergeGateFailed, storage.MaxMergeGateAttempts, + } + args = append(args, nonTerminalArgs...) + result, err := s.db.ExecContext(ctx, ` + UPDATE merge_gate_requests r + JOIN applies a ON a.id = r.apply_id + SET r.state = ?, r.attempts = 0, r.lease_owner = NULL, r.lease_token = NULL, + r.lease_expires_at = NULL, r.retry_after = NULL, r.completed_at = NULL, + r.updated_at = NOW() + WHERE r.kind = ? AND r.state = ? + AND (r.retry_after IS NULL OR r.attempts >= ?) + AND `+nonTerminal+` + `, args...) + if err != nil { + return 0, fmt.Errorf("reopen terminal preflight merge gate requests for active applies: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return 0, fmt.Errorf("read reopened terminal preflight merge gate rows affected: %w", err) + } + return rows, nil +} + // mergeGateClaimablePredicate matches exactly the rows a processor would // claim: a pending row, a retryable row whose retry window has elapsed and is // under the attempt cap, or a processing row whose lease has expired and is @@ -173,17 +233,17 @@ func (s *mergeGateRequestStore) ClaimNext(ctx context.Context, owner string, lea return req, nil } -func (s *mergeGateRequestStore) PendingForTarget(ctx context.Context, environment, databaseType, databaseName string, excludeID int64) ([]*storage.MergeGateRequest, error) { +func (s *mergeGateRequestStore) PendingForTarget(ctx context.Context, environment, databaseType, databaseName, kind string, excludeID int64) ([]*storage.MergeGateRequest, error) { rows, err := s.db.QueryContext(ctx, ` SELECT `+mergeGateColumns+` FROM merge_gate_requests WHERE environment = ? AND database_type = ? AND database_name = ? - AND state = ? AND id != ? + AND kind = ? AND state = ? AND id != ? ORDER BY created_at, id - `, environment, databaseType, databaseName, storage.MergeGatePending, excludeID) + `, environment, databaseType, databaseName, kind, storage.MergeGatePending, excludeID) if err != nil { - return nil, fmt.Errorf("query pending merge gate requests for %s/%s in %s: %w", - databaseType, databaseName, environment, err) + return nil, fmt.Errorf("query pending %s merge gate requests for %s/%s in %s: %w", + kind, databaseType, databaseName, environment, err) } defer utils.CloseAndLog(rows) @@ -238,6 +298,23 @@ func (s *mergeGateRequestStore) MarkCompleted(ctx context.Context, id int64, lea return s.mergeGateLeaseResult(ctx, result, id, leaseToken) } +// MarkPreflightHoldsRecorded stamps holds_recorded_at on a claimed preflight +// request once its storage-only hold phase has flipped every sibling change's +// stored check. Set-once (COALESCE-preserved), so retries after a partial +// render keep the original stamp. Idempotent for the same lease token, on the +// same rationale as MarkCompleted. +func (s *mergeGateRequestStore) MarkPreflightHoldsRecorded(ctx context.Context, id int64, leaseToken string) error { + result, err := s.db.ExecContext(ctx, ` + UPDATE merge_gate_requests + SET holds_recorded_at = COALESCE(holds_recorded_at, NOW()), updated_at = NOW() + WHERE id = ? AND lease_token = ? + `, id, leaseToken) + if err != nil { + return fmt.Errorf("mark merge gate request %d preflight holds recorded: %w", id, err) + } + return s.mergeGateLeaseResult(ctx, result, id, leaseToken) +} + func (s *mergeGateRequestStore) CompletePendingCoalesced(ctx context.Context, id int64) (bool, error) { result, err := s.db.ExecContext(ctx, ` UPDATE merge_gate_requests @@ -278,14 +355,14 @@ func (s *mergeGateRequestStore) FindCompletedAppliesMissingRequest(ctx context.C rows, err := s.db.QueryContext(ctx, ` SELECT `+applyColumnsForApplyAlias+` FROM applies a - LEFT JOIN merge_gate_requests r ON r.apply_id = a.id + LEFT JOIN merge_gate_requests r ON r.apply_id = a.id AND r.kind = ? WHERE a.state = ? AND a.completed_at > `+s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, ParameterIntervalAmount(), IntervalSecond)+` AND r.id IS NULL ORDER BY a.completed_at, a.id - `, state.Apply.Completed, int64(lookback.Seconds())) + `, storage.MergeGateKindSettle, state.Apply.Completed, int64(lookback.Seconds())) if err != nil { - return nil, fmt.Errorf("query completed applies missing merge gate requests: %w", err) + return nil, fmt.Errorf("query completed applies missing settle merge gate requests: %w", err) } defer utils.CloseAndLog(rows) @@ -293,13 +370,81 @@ func (s *mergeGateRequestStore) FindCompletedAppliesMissingRequest(ctx context.C for rows.Next() { apply, err := scanApplyInto(rows) if err != nil { - return nil, fmt.Errorf("scan completed apply missing merge gate request: %w", err) + return nil, fmt.Errorf("scan completed apply missing settle merge gate request: %w", err) } applies = append(applies, apply) } return applies, rows.Err() } +// FindTerminalAppliesWithPreflightMissingSettle returns applies that settled +// terminally within the lookback window with a preflight request but no +// settle. Their preflight fan-out held sibling change checks action-required, and +// only a settle fan-out re-plans those checks back to a live verdict — so the +// settle must exist for every terminal outcome, including failed and +// cancelled applies that never changed the schema. +func (s *mergeGateRequestStore) FindTerminalAppliesWithPreflightMissingSettle(ctx context.Context, lookback time.Duration) ([]*storage.Apply, error) { + if lookback <= 0 { + return nil, fmt.Errorf("merge gate sweep lookback must be positive") + } + terminalStates := terminalApplyStates() + args := []any{storage.MergeGateKindPreflight, storage.MergeGateKindSettle} + args = append(args, stringArgs(terminalStates)...) + args = append(args, int64(lookback.Seconds())) + rows, err := s.db.QueryContext(ctx, ` + SELECT `+applyColumnsForApplyAlias+` + FROM applies a + JOIN merge_gate_requests pre ON pre.apply_id = a.id AND pre.kind = ? + LEFT JOIN merge_gate_requests settle ON settle.apply_id = a.id AND settle.kind = ? + WHERE a.state IN (`+placeholders(len(terminalStates))+`) + AND a.updated_at > `+s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, ParameterIntervalAmount(), IntervalSecond)+` + AND settle.id IS NULL + ORDER BY a.updated_at, a.id + `, args...) + if err != nil { + return nil, fmt.Errorf("query terminal applies with preflight missing settle merge gate requests: %w", err) + } + defer utils.CloseAndLog(rows) + + var applies []*storage.Apply + for rows.Next() { + apply, err := scanApplyInto(rows) + if err != nil { + return nil, fmt.Errorf("scan terminal apply with preflight missing settle merge gate request: %w", err) + } + applies = append(applies, apply) + } + return applies, rows.Err() +} + +// HasActivePreflightedApplyOnTarget reports whether any non-terminal apply on +// the target has a recorded preflight request. Such an apply has held (or is +// holding) sibling change checks and is guaranteed a settle of its own once it +// settles terminally, so a settle fan-out for an earlier apply defers to it +// rather than re-planning holds away mid-apply. Applies without a preflight +// (queued, never started) do not count: they have invalidated nothing yet. +func (s *mergeGateRequestStore) HasActivePreflightedApplyOnTarget(ctx context.Context, environment, databaseType, databaseName string) (bool, error) { + nonTerminal, nonTerminalArgs := nonTerminalApplyStatePredicate("a.state") + args := []any{storage.MergeGateKindPreflight, environment, databaseType, databaseName} + args = append(args, nonTerminalArgs...) + var exists bool + err := s.db.QueryRowContext(ctx, ` + SELECT EXISTS ( + SELECT 1 + FROM merge_gate_requests pre + JOIN applies a ON a.id = pre.apply_id + WHERE pre.kind = ? + AND pre.environment = ? AND pre.database_type = ? AND pre.database_name = ? + AND `+nonTerminal+` + ) + `, args...).Scan(&exists) + if err != nil { + return false, fmt.Errorf("query active preflighted applies for %s/%s in %s: %w", + databaseType, databaseName, environment, err) + } + return exists, nil +} + func (s *mergeGateRequestStore) TerminateStuckProcessing(ctx context.Context, reason string) (int64, error) { result, err := s.db.ExecContext(ctx, ` UPDATE merge_gate_requests @@ -352,12 +497,12 @@ func scanMergeGateRequest(row *sql.Row) (*storage.MergeGateRequest, error) { func scanMergeGateRequestInto(row scanner) (*storage.MergeGateRequest, error) { var req storage.MergeGateRequest var leaseOwner, leaseToken, lastError sql.NullString - var leaseExpiresAt, retryAfter, completedAt sql.NullTime + var leaseExpiresAt, retryAfter, holdsRecordedAt, completedAt sql.NullTime err := row.Scan( - &req.ID, &req.ApplyID, &req.ApplyIdentifier, &req.Environment, &req.DatabaseType, &req.DatabaseName, + &req.ID, &req.ApplyID, &req.Kind, &req.ApplyIdentifier, &req.Environment, &req.DatabaseType, &req.DatabaseName, &req.Provider, &req.Repository, &req.ChangeKey, &req.RequestedBy, &req.State, &req.Attempts, &leaseOwner, &leaseToken, &leaseExpiresAt, &retryAfter, &lastError, - &completedAt, &req.CreatedAt, &req.UpdatedAt, + &holdsRecordedAt, &completedAt, &req.CreatedAt, &req.UpdatedAt, ) if err != nil { return nil, err @@ -371,6 +516,9 @@ func scanMergeGateRequestInto(row scanner) (*storage.MergeGateRequest, error) { if retryAfter.Valid { req.RetryAfter = &retryAfter.Time } + if holdsRecordedAt.Valid { + req.HoldsRecordedAt = &holdsRecordedAt.Time + } if completedAt.Valid { req.CompletedAt = &completedAt.Time } diff --git a/pkg/storage/mysqlstore/merge_gate_requests_test.go b/pkg/storage/mysqlstore/merge_gate_requests_test.go index 6e64d5098..b0d7f9c29 100644 --- a/pkg/storage/mysqlstore/merge_gate_requests_test.go +++ b/pkg/storage/mysqlstore/merge_gate_requests_test.go @@ -15,17 +15,24 @@ import ( "github.com/block/schemabot/pkg/storage" ) -// recordTestMergeGateRequest records a pending merge gate request for a synthetic -// completed apply and returns it. Each request needs its own applies row -// because apply_id is the idempotency key; the anchor apply gets its own lock -// database (locks are unique per database) while the request carries the -// target under test, so several requests can share one target. +// recordTestMergeGateRequest records a pending settle merge gate request for a +// synthetic completed apply and returns it. Each request needs its own +// applies row because (apply_id, kind) is the idempotency key; the anchor +// apply gets its own lock database (locks are unique per database) while the +// request carries the target under test, so several requests can share one +// target. func recordTestMergeGateRequest(t *testing.T, store *Storage, name, env, dbType, dbName, repo string, pr int) *storage.MergeGateRequest { t.Helper() lock := createTestLockWithPR(t, store, name+"_lock_db", dbType, env, repo, pr) apply := createTestApplyWithStateAndEnv(t, store, lock, name, 0, state.Apply.Completed, env) + return recordTestMergeGateRequestForApply(t, store, apply, storage.MergeGateKindSettle, env, dbType, dbName, repo, pr) +} + +func recordTestMergeGateRequestForApply(t *testing.T, store *Storage, apply *storage.Apply, kind, env, dbType, dbName, repo string, pr int) *storage.MergeGateRequest { + t.Helper() req := &storage.MergeGateRequest{ ApplyID: apply.ID, + Kind: kind, ApplyIdentifier: apply.ApplyIdentifier, Environment: env, DatabaseType: dbType, @@ -42,10 +49,11 @@ func recordTestMergeGateRequest(t *testing.T, store *Storage, name, env, dbType, } // Scenario: the drive tail and the backstop sweep may both record a merge -// gate request for the same apply. The unique key on the apply makes recording -// idempotent, so the second recording reports recorded=false instead of -// duplicating the fan-out. -func TestMergeGateStore_RecordIsIdempotentPerApply(t *testing.T) { +// gate request for the same apply. The unique key on the apply and kind makes +// recording idempotent per kind, so a duplicate recording reports +// recorded=false instead of duplicating the fan-out — while the same apply's +// preflight and settle coexist as separate rows. +func TestMergeGateStore_RecordIsIdempotentPerApplyAndKind(t *testing.T) { clearTables(t) ctx := t.Context() store := New(testDB) @@ -54,6 +62,7 @@ func TestMergeGateStore_RecordIsIdempotentPerApply(t *testing.T) { again := &storage.MergeGateRequest{ ApplyID: req.ApplyID, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: req.ApplyIdentifier, Environment: req.Environment, DatabaseType: req.DatabaseType, @@ -63,10 +72,11 @@ func TestMergeGateStore_RecordIsIdempotentPerApply(t *testing.T) { require.NoError(t, err) assert.False(t, recorded) - got, err := store.MergeGateRequests().GetByApplyID(ctx, req.ApplyID) + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, got) assert.Equal(t, storage.MergeGatePending, got.State) + assert.Equal(t, storage.MergeGateKindSettle, got.Kind) assert.Equal(t, "apply_refresh_1", got.ApplyIdentifier) assert.Equal(t, "testdb", got.DatabaseName) assert.Equal(t, storage.DatabaseTypeMySQL, got.DatabaseType) @@ -75,6 +85,25 @@ func TestMergeGateStore_RecordIsIdempotentPerApply(t *testing.T) { assert.Equal(t, storage.ProviderGitHub, got.Provider) assert.Equal(t, "11", got.ChangeKey) assert.Equal(t, "cli:user@host", got.RequestedBy) + + // The same apply's preflight is a separate row, not a duplicate. + preflight := &storage.MergeGateRequest{ + ApplyID: req.ApplyID, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: req.ApplyIdentifier, + Environment: req.Environment, + DatabaseType: req.DatabaseType, + DatabaseName: req.DatabaseName, + } + recorded, err = store.MergeGateRequests().Record(ctx, preflight) + require.NoError(t, err) + assert.True(t, recorded) + + gotPre, err := store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, gotPre) + assert.Equal(t, storage.MergeGateKindPreflight, gotPre.Kind) + assert.NotEqual(t, got.ID, gotPre.ID) } func TestMergeGateStore_RecordRejectsIncompleteRequests(t *testing.T) { @@ -83,9 +112,11 @@ func TestMergeGateStore_RecordRejectsIncompleteRequests(t *testing.T) { store := New(testDB) for name, req := range map[string]*storage.MergeGateRequest{ - "missing apply row id": {ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, - "missing apply id": {ApplyID: 1, Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, - "missing target database": {ApplyID: 1, ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql"}, + "missing apply row id": {Kind: storage.MergeGateKindSettle, ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, + "missing apply id": {Kind: storage.MergeGateKindSettle, ApplyID: 1, Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, + "missing target database": {Kind: storage.MergeGateKindSettle, ApplyID: 1, ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql"}, + "missing kind": {ApplyID: 1, ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, + "unknown kind": {Kind: "bogus", ApplyID: 1, ApplyIdentifier: "a", Environment: "staging", DatabaseType: "mysql", DatabaseName: "db"}, } { _, err := store.MergeGateRequests().Record(ctx, req) assert.Error(t, err, name) @@ -170,12 +201,10 @@ func TestMergeGateStore_MarkFailedRetryableAndTerminal(t *testing.T) { require.NoError(t, err) require.NotNil(t, claimed) - // The claim predicate compares retry_after against the database clock, so - // place it far enough in the past to be immune to client/server skew. - past := time.Now().Add(-time.Minute) + past := time.Now().Add(-time.Second) require.NoError(t, store.MergeGateRequests().MarkFailed(ctx, claimed.ID, claimed.LeaseToken, "plan engine unavailable", &past)) - got, err := store.MergeGateRequests().GetByApplyID(ctx, req.ApplyID) + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) assert.Equal(t, storage.MergeGateFailed, got.State) assert.Equal(t, "plan engine unavailable", got.LastError) @@ -188,7 +217,7 @@ func TestMergeGateStore_MarkFailedRetryableAndTerminal(t *testing.T) { assert.Equal(t, claimed.ID, reclaimed.ID) require.NoError(t, store.MergeGateRequests().MarkFailed(ctx, reclaimed.ID, reclaimed.LeaseToken, "still failing", nil)) - got, err = store.MergeGateRequests().GetByApplyID(ctx, req.ApplyID) + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) assert.Equal(t, storage.MergeGateFailed, got.State) assert.Nil(t, got.RetryAfter) @@ -262,7 +291,7 @@ func TestMergeGateStore_PendingForTargetAndCoalesce(t *testing.T) { sibling := recordTestMergeGateRequest(t, store, "apply_coalesce_2", "staging", storage.DatabaseTypeMySQL, "db_coalesce", "org/repo", 72) recordTestMergeGateRequest(t, store, "apply_other_target", "production", storage.DatabaseTypeMySQL, "db_coalesce", "org/repo", 73) - pending, err := store.MergeGateRequests().PendingForTarget(ctx, "staging", storage.DatabaseTypeMySQL, "db_coalesce", first.ID) + pending, err := store.MergeGateRequests().PendingForTarget(ctx, "staging", storage.DatabaseTypeMySQL, "db_coalesce", storage.MergeGateKindSettle, first.ID) require.NoError(t, err) require.Len(t, pending, 1, "same target only, excluding the claimed request") assert.Equal(t, sibling.ID, pending[0].ID) @@ -271,7 +300,7 @@ func TestMergeGateStore_PendingForTargetAndCoalesce(t *testing.T) { require.NoError(t, err) assert.True(t, coalesced) - got, err := store.MergeGateRequests().GetByApplyID(ctx, sibling.ApplyID) + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, sibling.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) assert.Equal(t, storage.MergeGateCompleted, got.State) assert.NotNil(t, got.CompletedAt) @@ -284,8 +313,8 @@ func TestMergeGateStore_PendingForTargetAndCoalesce(t *testing.T) { } // Scenario: a pod crashes between an apply's terminal write and the drive -// tail's merge gate recording. The applies table is the outbox: the sweep finds -// completed applies in the lookback window with no merge gate request, and only +// tail's refresh recording. The applies table is the outbox: the sweep finds +// completed applies in the lookback window with no settle request, and only // those — recorded, non-completed, and out-of-window applies stay out. func TestMergeGateStore_FindCompletedAppliesMissingRequest(t *testing.T) { clearTables(t) @@ -304,6 +333,7 @@ func TestMergeGateStore_FindCompletedAppliesMissingRequest(t *testing.T) { recorded := completeApply("apply_sweep_recorded", "db_sweep_2", 82, time.Now()) _, err := store.MergeGateRequests().Record(ctx, &storage.MergeGateRequest{ ApplyID: recorded.ID, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: recorded.ApplyIdentifier, Environment: recorded.Environment, DatabaseType: recorded.DatabaseType, @@ -352,18 +382,18 @@ func TestMergeGateStore_TerminateStuckProcessing(t *testing.T) { require.NoError(t, err) assert.Equal(t, int64(1), terminated) - got, err := store.MergeGateRequests().GetByApplyID(ctx, stuck.ApplyID) + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, stuck.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) assert.Equal(t, storage.MergeGateFailed, got.State) assert.Nil(t, got.RetryAfter, "terminated rows must not be retryable") assert.Equal(t, "attempt cap with expired lease", got.LastError) - still, err := store.MergeGateRequests().GetByApplyID(ctx, reclaimable.ApplyID) + still, err := store.MergeGateRequests().GetByApplyAndKind(ctx, reclaimable.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) assert.Equal(t, storage.MergeGateProcessing, still.State, "an under-cap wedged row stays reclaimable") } -// Scenario: the refresh fan-out must find every PR planning against a target +// Scenario: the settle fan-out must find every PR planning against a target // across repositories — a CLI apply carries no repository, so the reverse // index cannot be scoped to one. func TestCheckStore_GetByTargetSpansRepositories(t *testing.T) { @@ -401,8 +431,8 @@ func TestCheckStore_GetByTargetSpansRepositories(t *testing.T) { assert.Equal(t, 2, checks[1].PullRequest) } -// Scenario: after a settle re-plan fails, the stored check is failed closed — -// but only while the row still holds the head SHA the processor read (a racing +// Scenario: after a refresh re-plan fails, the stored check is failed closed — +// but only while the row still holds the head SHA the refresher read (a racing // synchronize that stored a newer head wins) and never while an in-progress // apply owns the row (the started apply's lifecycle stays authoritative). func TestCheckStore_MarkBlockedForFailedRefresh(t *testing.T) { @@ -473,3 +503,345 @@ func TestCheckStore_MarkBlockedForFailedRefresh(t *testing.T) { assert.Equal(t, "in_progress", got.Status, "the in-flight apply-owned row is untouched") assert.Equal(t, int64(424242), got.ApplyID) } + +// Scenario: an apply's preflight exhausted its retry budget during a GitHub +// outage and failed terminally. The gate re-arms exactly that row back to +// pending with a fresh attempt budget so the next drive attempt can succeed; +// rows in any other state are left to their own lifecycle. +func TestMergeGateStore_ReopenForRetry(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + req := recordTestMergeGateRequest(t, store, "apply_reopen_1", "staging", storage.DatabaseTypeMySQL, "db_reopen", "org/repo", 111) + + // A pending row is not reopenable. + reopened, err := store.MergeGateRequests().ReopenForRetry(ctx, req.ID) + require.NoError(t, err) + assert.False(t, reopened, "a pending row must not be reopened") + + claimed, err := store.MergeGateRequests().ClaimNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + require.NoError(t, store.MergeGateRequests().MarkFailed(ctx, claimed.ID, claimed.LeaseToken, "github unavailable", nil)) + + reopened, err = store.MergeGateRequests().ReopenForRetry(ctx, req.ID) + require.NoError(t, err) + assert.True(t, reopened) + + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindSettle) + require.NoError(t, err) + assert.Equal(t, storage.MergeGatePending, got.State) + assert.Equal(t, 0, got.Attempts) + assert.Nil(t, got.RetryAfter) + assert.Nil(t, got.CompletedAt) + + reclaimed, err := store.MergeGateRequests().ClaimNext(ctx, "driver-b", time.Minute) + require.NoError(t, err) + require.NotNil(t, reclaimed, "a reopened row is claimable again") + assert.Equal(t, req.ID, reclaimed.ID) +} + +// Scenario: the preflight fan-out finishes its storage-only hold phase and +// stamps holds_recorded_at so the operator gate can start the apply before +// the code-host rendering lands. The stamp is set-once (render retries keep +// the original), lease-guarded, and survives both a terminal render failure +// and the re-arm that follows it. +func TestMergeGateStore_MarkPreflightHoldsRecorded(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + lock := createTestLockWithPR(t, store, "db_holds_lock", storage.DatabaseTypeMySQL, "staging", "org/repo", 141) + apply := createTestApplyWithStateAndEnv(t, store, lock, "apply_holds_1", 0, state.Apply.Running, "staging") + req := recordTestMergeGateRequestForApply(t, store, apply, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, "db_holds", "org/repo", 141) + + claimed, err := store.MergeGateRequests().ClaimNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + require.Nil(t, claimed.HoldsRecordedAt, "a fresh preflight has no holds recorded yet") + + require.NoError(t, store.MergeGateRequests().MarkPreflightHoldsRecorded(ctx, claimed.ID, claimed.LeaseToken)) + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, got.HoldsRecordedAt) + assert.Equal(t, storage.MergeGateProcessing, got.State, "recording holds does not complete the request; the render phase remains") + + // A retry after a partial render keeps the original stamp. + _, err = testDB.ExecContext(ctx, `UPDATE merge_gate_requests SET holds_recorded_at = NOW() - INTERVAL 1 HOUR WHERE id = ?`, claimed.ID) + require.NoError(t, err) + require.NoError(t, store.MergeGateRequests().MarkPreflightHoldsRecorded(ctx, claimed.ID, claimed.LeaseToken)) + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, got.HoldsRecordedAt) + assert.WithinDuration(t, time.Now().Add(-time.Hour), *got.HoldsRecordedAt, 5*time.Minute, "the first stamp is preserved") + + err = store.MergeGateRequests().MarkPreflightHoldsRecorded(ctx, claimed.ID, "stale-token") + assert.ErrorIs(t, err, storage.ErrMergeGateLeaseLost) + err = store.MergeGateRequests().MarkPreflightHoldsRecorded(ctx, claimed.ID+9999, "any") + assert.ErrorIs(t, err, storage.ErrMergeGateNotFound) + + // A terminal render failure and its re-arm both leave the stamp in place: + // the stored holds are real regardless of how the rendering fared. + require.NoError(t, store.MergeGateRequests().MarkFailed(ctx, claimed.ID, claimed.LeaseToken, "github unavailable", nil)) + reopened, err := store.MergeGateRequests().ReopenForRetry(ctx, claimed.ID) + require.NoError(t, err) + require.True(t, reopened) + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, req.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + assert.NotNil(t, got.HoldsRecordedAt) + assert.Equal(t, storage.MergeGatePending, got.State) +} + +// Scenario: the gate started an apply on stored holds while GitHub was down, +// and the render phase then exhausted its retry budget. Nothing else would +// retry the render until the apply settles, so the sweep re-arms terminally +// failed preflights whose apply is still active — and only those: retryable +// failures, settles, and preflights of settled applies keep their own +// lifecycles. +func TestMergeGateStore_ReopenTerminalPreflightsForActiveApplies(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + makePreflight := func(name, dbName string, pr int, applyState string) *storage.MergeGateRequest { + lock := createTestLockWithPR(t, store, dbName, storage.DatabaseTypeMySQL, "staging", "org/repo", pr) + apply := createTestApplyWithStateAndEnv(t, store, lock, name, 0, applyState, "staging") + return recordTestMergeGateRequestForApply(t, store, apply, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, dbName, "org/repo", pr) + } + failTerminally := func(req *storage.MergeGateRequest) { + _, err := testDB.ExecContext(ctx, ` + UPDATE merge_gate_requests + SET state = ?, attempts = 1, retry_after = NULL, last_error = 'github unavailable', + completed_at = NOW(), lease_owner = NULL, lease_token = NULL, lease_expires_at = NULL + WHERE id = ?`, storage.MergeGateFailed, req.ID) + require.NoError(t, err) + } + + // Terminal render failure on a running apply: re-armed. + rearm := makePreflight("apply_rearm_active", "db_rearm_1", 151, state.Apply.Running) + failTerminally(rearm) + + // A settle in the same terminal-failed shape is not a render: not re-armed. + settleApply, err := store.Applies().Get(ctx, rearm.ApplyID) + require.NoError(t, err) + settleReq := recordTestMergeGateRequestForApply(t, store, settleApply, storage.MergeGateKindSettle, "staging", storage.DatabaseTypeMySQL, "db_rearm_1", "org/repo", 151) + failTerminally(settleReq) + + // Terminal render failure on a settled apply: the settle re-plan covers it. + settled := makePreflight("apply_rearm_settled", "db_rearm_2", 152, state.Apply.Completed) + failTerminally(settled) + + // Retryable failure on a running apply: still claimable on its own. + retryable := makePreflight("apply_rearm_retryable", "db_rearm_3", 153, state.Apply.Running) + future := time.Now().Add(time.Minute) + _, err = testDB.ExecContext(ctx, `UPDATE merge_gate_requests SET state = ?, attempts = 1, retry_after = ? WHERE id = ?`, + storage.MergeGateFailed, future, retryable.ID) + require.NoError(t, err) + + reopened, err := store.MergeGateRequests().ReopenTerminalPreflightsForActiveApplies(ctx) + require.NoError(t, err) + assert.Equal(t, int64(1), reopened) + + got, err := store.MergeGateRequests().GetByApplyAndKind(ctx, rearm.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + assert.Equal(t, storage.MergeGatePending, got.State) + assert.Equal(t, 0, got.Attempts) + assert.Nil(t, got.CompletedAt) + + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, rearm.ApplyID, storage.MergeGateKindSettle) + require.NoError(t, err) + assert.Equal(t, storage.MergeGateFailed, got.State, "a terminally failed settle is not a render; not re-armed") + + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, settled.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + assert.Equal(t, storage.MergeGateFailed, got.State, "a settled apply's preflight is not re-armed") + + got, err = store.MergeGateRequests().GetByApplyAndKind(ctx, retryable.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + assert.Equal(t, storage.MergeGateFailed, got.State, "a retryable failure is not re-armed") + assert.Equal(t, 1, got.Attempts) + + // A second pass finds nothing new. + reopened, err = store.MergeGateRequests().ReopenTerminalPreflightsForActiveApplies(ctx) + require.NoError(t, err) + assert.Zero(t, reopened) +} + +// Scenario: an apply that recorded a preflight held sibling PR checks, then +// settled terminally without a settle request — for example it was cancelled +// while queued, so the drive tail never ran. The release sweep must find +// exactly those applies so a settle fan-out can release the holds; applies +// whose settle exists, applies without a preflight, and still-running applies +// stay out. +func TestMergeGateStore_FindTerminalAppliesWithPreflightMissingSettle(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + makeApply := func(name, dbName string, pr int, applyState string) *storage.Apply { + lock := createTestLockWithPR(t, store, dbName, storage.DatabaseTypeMySQL, "staging", "org/repo", pr) + return createTestApplyWithStateAndEnv(t, store, lock, name, 0, applyState, "staging") + } + + // Failed apply with a preflight and no settle: the sweep must find it. + missing := makeApply("apply_release_missing", "db_release_1", 121, state.Apply.Failed) + recordTestMergeGateRequestForApply(t, store, missing, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, "db_release_1", "org/repo", 121) + + // Terminal apply whose settle was already recorded: covered. + settled := makeApply("apply_release_settled", "db_release_2", 122, state.Apply.Completed) + recordTestMergeGateRequestForApply(t, store, settled, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, "db_release_2", "org/repo", 122) + recordTestMergeGateRequestForApply(t, store, settled, storage.MergeGateKindSettle, "staging", storage.DatabaseTypeMySQL, "db_release_2", "org/repo", 122) + + // Terminal apply that never recorded a preflight: it held nothing. + makeApply("apply_release_no_preflight", "db_release_3", 123, state.Apply.Failed) + + // Still-running apply with a preflight: its own settle comes later. + running := makeApply("apply_release_running", "db_release_4", 124, state.Apply.Running) + recordTestMergeGateRequestForApply(t, store, running, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, "db_release_4", "org/repo", 124) + + applies, err := store.MergeGateRequests().FindTerminalAppliesWithPreflightMissingSettle(ctx, time.Hour) + require.NoError(t, err) + require.Len(t, applies, 1) + assert.Equal(t, missing.ApplyIdentifier, applies[0].ApplyIdentifier) + assert.Equal(t, "db_release_1", applies[0].Database) + + // Outside the lookback window the apply is no longer swept. + _, err = testDB.ExecContext(ctx, `UPDATE applies SET updated_at = NOW() - INTERVAL 2 HOUR WHERE id = ?`, missing.ID) + require.NoError(t, err) + applies, err = store.MergeGateRequests().FindTerminalAppliesWithPreflightMissingSettle(ctx, time.Hour) + require.NoError(t, err) + assert.Empty(t, applies) +} + +// Scenario: an earlier apply's settle fan-out is about to re-plan sibling PR +// checks while a newer preflighted apply is still running on the same target. +// Re-planning now would overwrite the newer apply's holds with pre-cutover +// verdicts, so the settle defers whenever a non-terminal preflighted apply +// exists on the target — and only then: queued applies that never recorded a +// preflight have invalidated nothing. +func TestMergeGateStore_HasActivePreflightedApplyOnTarget(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + // Running apply on the target without a preflight: no deferral. + lockNoPre := createTestLockWithPR(t, store, "db_active_1", storage.DatabaseTypeMySQL, "staging", "org/repo", 131) + createTestApplyWithStateAndEnv(t, store, lockNoPre, "apply_active_no_pre", 0, state.Apply.Running, "staging") + active, err := store.MergeGateRequests().HasActivePreflightedApplyOnTarget(ctx, "staging", storage.DatabaseTypeMySQL, "db_active_target") + require.NoError(t, err) + assert.False(t, active, "a running apply without a preflight has held nothing") + + // Running apply with a preflight on the target: defer. + lockPre := createTestLockWithPR(t, store, "db_active_2", storage.DatabaseTypeMySQL, "staging", "org/repo", 132) + preflighted := createTestApplyWithStateAndEnv(t, store, lockPre, "apply_active_pre", 0, state.Apply.Running, "staging") + recordTestMergeGateRequestForApply(t, store, preflighted, storage.MergeGateKindPreflight, "staging", storage.DatabaseTypeMySQL, "db_active_target", "org/repo", 132) + + active, err = store.MergeGateRequests().HasActivePreflightedApplyOnTarget(ctx, "staging", storage.DatabaseTypeMySQL, "db_active_target") + require.NoError(t, err) + assert.True(t, active) + + // A different target is unaffected. + active, err = store.MergeGateRequests().HasActivePreflightedApplyOnTarget(ctx, "production", storage.DatabaseTypeMySQL, "db_active_target") + require.NoError(t, err) + assert.False(t, active) + + // Once the preflighted apply settles terminally, the deferral lifts. + preflighted.State = state.Apply.Failed + require.NoError(t, store.Applies().Update(ctx, preflighted)) + active, err = store.MergeGateRequests().HasActivePreflightedApplyOnTarget(ctx, "staging", storage.DatabaseTypeMySQL, "db_active_target") + require.NoError(t, err) + assert.False(t, active) +} + +// Scenario: the preflight fan-out holds a sibling PR's green check +// action-required before an apply starts. The flip is conditional the same way +// a failed-refresh block is (head SHA, no in-flight apply-owned rows), and +// additionally skips rows already holding for an apply so a retried fan-out +// reports the hold as already in place instead of newly flipped. +func TestCheckStore_MarkBlockedForApplyInFlight(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + check := &storage.Check{ + Repository: "org/repo", + PullRequest: 141, + HeadSHA: "head-hold", + Environment: "staging", + DatabaseType: storage.DatabaseTypeMySQL, + DatabaseName: "db_hold", + Status: "completed", + Conclusion: "success", + } + require.NoError(t, store.Checks().Upsert(ctx, check)) + + hold := *check + hold.Status = "completed" + hold.Conclusion = "action_required" + hold.BlockingReason = "apply_in_flight_on_target" + hold.ChangeSummary = "held: apply apply_hold_1 is changing db_hold in staging" + + flipped, err := store.Checks().MarkBlockedForApplyInFlight(ctx, &hold) + require.NoError(t, err) + assert.True(t, flipped) + got, err := store.Checks().Get(ctx, "org/repo", 141, "staging", storage.DatabaseTypeMySQL, "db_hold") + require.NoError(t, err) + assert.Equal(t, "action_required", got.Conclusion) + assert.Equal(t, "apply_in_flight_on_target", got.BlockingReason) + + // A retried fan-out sees the hold already in place and does not re-flip. + flipped, err = store.Checks().MarkBlockedForApplyInFlight(ctx, &hold) + require.NoError(t, err) + assert.False(t, flipped, "an already-held row is not flipped again") + + // A racing synchronize that stored a newer head wins over the hold. + stale := &storage.Check{ + Repository: "org/repo", + PullRequest: 142, + HeadSHA: "head-old", + Environment: "staging", + DatabaseType: storage.DatabaseTypeMySQL, + DatabaseName: "db_hold", + Status: "completed", + Conclusion: "success", + } + require.NoError(t, store.Checks().Upsert(ctx, stale)) + _, err = testDB.ExecContext(ctx, `UPDATE checks SET head_sha = 'head-new' WHERE repository = 'org/repo' AND pull_request = 142`) + require.NoError(t, err) + staleHold := *stale + staleHold.Conclusion = "action_required" + staleHold.BlockingReason = "apply_in_flight_on_target" + flipped, err = store.Checks().MarkBlockedForApplyInFlight(ctx, &staleHold) + require.NoError(t, err) + assert.False(t, flipped) + got, err = store.Checks().Get(ctx, "org/repo", 142, "staging", storage.DatabaseTypeMySQL, "db_hold") + require.NoError(t, err) + assert.Equal(t, "success", got.Conclusion, "the newer head's stored result is preserved") + + // An in-progress apply-owned row stays authoritative. + owned := &storage.Check{ + Repository: "org/repo", + PullRequest: 143, + HeadSHA: "head-owned", + Environment: "staging", + DatabaseType: storage.DatabaseTypeMySQL, + DatabaseName: "db_hold", + Status: "completed", + Conclusion: "success", + } + require.NoError(t, store.Checks().Upsert(ctx, owned)) + _, err = testDB.ExecContext(ctx, `UPDATE checks SET status = 'in_progress', apply_id = 424243 WHERE repository = 'org/repo' AND pull_request = 143`) + require.NoError(t, err) + ownedHold := *owned + ownedHold.Status = "in_progress" + ownedHold.Conclusion = "action_required" + ownedHold.BlockingReason = "apply_in_flight_on_target" + flipped, err = store.Checks().MarkBlockedForApplyInFlight(ctx, &ownedHold) + require.NoError(t, err) + assert.False(t, flipped) + got, err = store.Checks().Get(ctx, "org/repo", 143, "staging", storage.DatabaseTypeMySQL, "db_hold") + require.NoError(t, err) + assert.Equal(t, "in_progress", got.Status, "the in-flight apply-owned row is untouched") + assert.Equal(t, int64(424243), got.ApplyID) +} diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index f5bdddf09..0e197ded4 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -191,6 +191,17 @@ type CheckStore interface { // lifecycle stays authoritative. Returns true when the row was flipped. MarkBlockedForFailedRefresh(ctx context.Context, check *Check) (bool, error) + // MarkBlockedForApplyInFlight flips stored check state to a blocking + // conclusion because an apply on the same target is about to change the + // live schema: the check's verdict was computed against a schema that will + // not survive the apply, so a merge must not land on it while the apply + // runs. Same conditional-write contract as MarkBlockedForFailedRefresh + // (head-SHA optimistic concurrency, in-progress apply-owned rows are never + // touched), plus it skips rows already holding this blocking reason so a + // retried preflight fan-out reports flipped=false instead of re-flipping. + // Returns true when the row was flipped. + MarkBlockedForApplyInFlight(ctx context.Context, check *Check) (bool, error) + // Delete removes stored check state by ID. Delete(ctx context.Context, id int64) error @@ -308,19 +319,42 @@ type WebhookEventStore interface { } // MergeGateRequestStore manages durable merge gate requests. A request -// records that an apply successfully changed a target's live schema, so stored -// check state on every other open change against that target must be re-planned. -// The row is behavioral state, not just audit: the merge gate processor -// consumes pending rows to recover the fan-out after process restarts. +// records that an apply is changing (preflight) or has changed (settle) a +// target's live schema, so stored check state on every other open PR against +// that target must be held or re-planned. The row is behavioral state, not +// just audit: the merge gate processor consumes pending rows to recover +// the fan-out after process restarts, and the operator's preflight gate waits +// on the preflight row before starting an apply's engine work. type MergeGateRequestStore interface { - // Record records a pending merge gate request for a completed apply. Returns - // recorded=false when a request for the apply already exists (any state), - // so recording is idempotent across drive tails and the backfill sweep. + // Record records a pending merge gate request. Kind is required. Returns + // recorded=false when a request for the same apply and kind already exists + // (any state), so recording is idempotent across drive tails, the operator + // preflight gate, and the backfill sweeps. Record(ctx context.Context, req *MergeGateRequest) (recorded bool, err error) - // GetByApplyID returns the request for an originating apply, or nil if not - // found. - GetByApplyID(ctx context.Context, applyID int64) (*MergeGateRequest, error) + // GetByApplyAndKind returns the request of one kind for an originating + // apply, or nil if not found. The operator preflight gate polls it while + // waiting for the preflight fan-out to complete. + GetByApplyAndKind(ctx context.Context, applyID int64, kind string) (*MergeGateRequest, error) + + // ReopenForRetry re-arms a terminally failed request back to pending with a + // fresh attempt budget. The operator preflight gate uses it so an apply + // blocked on a preflight that exhausted its retries (for example during a + // long GitHub outage) becomes claimable again on the apply's next start + // attempt instead of staying blocked until manual intervention. Returns + // true when the row was re-armed; false when it was not terminally failed. + ReopenForRetry(ctx context.Context, id int64) (bool, error) + + // ReopenTerminalPreflightsForActiveApplies re-arms every terminally failed + // preflight request whose apply is still non-terminal, returning the number + // re-armed. The gate may have started such an apply on stored holds while + // the code-host rendering kept failing (for example through a code-host + // outage); once the render exhausts its retries nothing else would retry it + // until the apply settles, leaving sibling changes' visible checks stale + // for the rest of the apply. This sweep keeps the render retrying for as + // long as the apply is active; the apply's settle re-plan covers the target + // after that. + ReopenTerminalPreflightsForActiveApplies(ctx context.Context) (int64, error) // ClaimNext atomically claims one pending, retryable, or lease-expired // request. The claim rotates lease_owner/lease_token, increments attempts, @@ -330,12 +364,14 @@ type MergeGateRequestStore interface { // Returns nil when no request is claimable. ClaimNext(ctx context.Context, owner string, leaseDuration time.Duration) (*MergeGateRequest, error) - // PendingForTarget returns the pending requests for the same + // PendingForTarget returns the pending requests of one kind for the same // (environment, database_type, database_name) target, excluding the given // request id. The processor coalesces them: one fan-out covers every - // schema change recorded before it started, so the siblings complete - // together with the claimed request. - PendingForTarget(ctx context.Context, environment, databaseType, databaseName string, excludeID int64) ([]*MergeGateRequest, error) + // same-kind request recorded before it started, so the siblings complete + // together with the claimed request. Kind-scoped because a preflight + // fan-out (hold checks) does not do a settle's work (re-plan them), or + // vice versa. + PendingForTarget(ctx context.Context, environment, databaseType, databaseName, kind string, excludeID int64) ([]*MergeGateRequest, error) // Heartbeat extends the lease on a claimed request so a fan-out that spans // many PRs can outlive the initial lease without being reclaimed @@ -347,6 +383,15 @@ type MergeGateRequestStore interface { // ErrMergeGateLeaseLost when the lease token is stale. MarkCompleted(ctx context.Context, id int64, leaseToken string) error + // MarkPreflightHoldsRecorded records that a preflight fan-out has durably + // held every sibling change's stored check on the target. Set-once: the + // first write stamps holds_recorded_at and retries preserve it. The + // operator gate starts the apply on this stamp, so it must land as soon as + // the storage-only hold phase finishes — before the code-host rendering, + // which can outlast a code-host outage. Returns ErrMergeGateLeaseLost when + // the lease token is stale. + MarkPreflightHoldsRecorded(ctx context.Context, id int64, leaseToken string) error + // CompletePendingCoalesced marks a still-pending sibling request completed // because a fan-out for the same target covered it. Returns true when the // row was completed; false when it was no longer pending (for example a @@ -359,12 +404,30 @@ type MergeGateRequestStore interface { MarkFailed(ctx context.Context, id int64, leaseToken string, errMsg string, retryAfter *time.Time) error // FindCompletedAppliesMissingRequest returns applies that reached the - // completed state within the lookback window but have no merge gate request + // completed state within the lookback window but have no settle request // row. The applies table is the outbox: a pod crash between an apply's - // terminal write and its merge gate recording loses the in-line record, and + // terminal write and its refresh recording loses the in-line record, and // this sweep is how the processor backfills it. FindCompletedAppliesMissingRequest(ctx context.Context, lookback time.Duration) ([]*Apply, error) + // HasActivePreflightedApplyOnTarget reports whether any non-terminal apply + // on the target has a recorded preflight request. The settle fan-out + // checks it before re-planning: while such an apply exists, re-planning + // would overwrite its holds with verdicts computed against a schema it is + // about to change, so the settle defers to that apply's own eventual + // settle instead. + HasActivePreflightedApplyOnTarget(ctx context.Context, environment, databaseType, databaseName string) (bool, error) + + // FindTerminalAppliesWithPreflightMissingSettle returns applies that + // settled to any terminal state within the lookback window, have a + // preflight request row, but no settle row. A preflight holds sibling PR + // checks action-required, and only the settle fan-out re-plans them back + // to a live verdict — so every preflighted apply must eventually record a + // settle, even when it failed or was cancelled before changing the schema. + // This sweep backfills settles the drive tails missed so held checks + // cannot stay blocked forever. + FindTerminalAppliesWithPreflightMissingSettle(ctx context.Context, lookback time.Duration) ([]*Apply, error) + // TerminateStuckProcessing marks as terminally failed every processing row // whose lease has expired and whose attempts have reached // MaxMergeGateAttempts — a driver hard-killed on its final attempt. diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 2e33f5eb9..a22e6f01f 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -1370,23 +1370,39 @@ const ( MergeGateFailed = "failed" ) +// Merge gate request kinds. A preflight request runs before an apply's +// engine work starts: it holds every sibling change's stored check on the +// target action-required so a merge cannot land on a verdict the apply is +// about to invalidate, and the operator gate blocks the apply's start until +// it completes. A settle request runs after the apply settles terminally: it +// re-plans each sibling against the (possibly changed) live schema, which +// both refreshes the verdicts and releases the preflight holds. +const ( + MergeGateKindPreflight = "preflight" + MergeGateKindSettle = "settle" +) + // MergeGateRequest is a durable request to re-evaluate stored check state // for every open change targeting one (environment, database_type, -// database_name) after an apply successfully changed that target's live -// schema. Plans and merge-gate statuses on sibling changes were computed -// against the previous live schema, so each request fans out to those changes -// and re-plans them; the request row is the durable record that the fan-out -// must happen, surviving pod restarts and lease handovers. +// database_name) around an apply against that target. Plans and merge-gate +// statuses on sibling changes were computed against the pre-apply live +// schema, so each request fans out to those changes — a preflight holds +// their checks while the apply runs, a settle re-plans them once it +// finishes. The request row is the durable record that the fan-out must +// happen, surviving pod restarts and lease handovers. // -// One row exists per originating apply (unique on apply_id), so recording is -// idempotent and a sweep over recently completed applies can backfill any -// request lost between the apply's terminal write and its recording. +// One row exists per originating apply and kind (unique on apply_id + kind), +// so recording is idempotent and the backstop sweeps can backfill any request +// lost between the apply's state write and its recording. type MergeGateRequest struct { ID int64 // ApplyID is the internal row id of the originating apply, used only for // the uniqueness guard and sweep join. Logs and operator-facing text use // ApplyIdentifier. ApplyID int64 + // Kind selects the fan-out the processor runs for this request: + // MergeGateKindPreflight or MergeGateKindSettle. + Kind string // ApplyIdentifier is the originating apply's user-facing string identifier, // carried for attribution in refreshed check summaries and logs. ApplyIdentifier string @@ -1414,9 +1430,16 @@ type MergeGateRequest struct { LeaseExpiresAt *time.Time RetryAfter *time.Time LastError string - CompletedAt *time.Time - CreatedAt time.Time - UpdatedAt time.Time + // HoldsRecordedAt is set once by the preflight fan-out when every sibling + // change's stored check hold is durably in place. Stored holds are + // storage-only writes, so this lands even when the code host is + // unreachable; the operator gate starts the apply on it rather than on + // request completion, which additionally requires the code-host rendering + // (Check Run update and hold comment). Always nil for settle requests. + HoldsRecordedAt *time.Time + CompletedAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time } // ChangeKeyForPullRequest renders a GitHub pull request number as a merge diff --git a/pkg/webhook/merge_gate.go b/pkg/webhook/merge_gate.go index 54ad784b6..4917dcd0e 100644 --- a/pkg/webhook/merge_gate.go +++ b/pkg/webhook/merge_gate.go @@ -177,6 +177,7 @@ func (h *Handler) sweepMergeGateRequests(ctx context.Context) { for _, apply := range applies { recorded, err := store.Record(ctx, &storage.MergeGateRequest{ ApplyID: apply.ID, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: apply.ApplyIdentifier, Environment: apply.Environment, DatabaseType: apply.DatabaseType, @@ -275,7 +276,7 @@ func (h *Handler) driveClaimedMergeGate(ctx context.Context, store storage.Merge // re-plans against the live schema, so it covers every schema change // recorded before it began. A request recorded mid-fan-out is not covered // and stays pending for the next drain. - siblings, err := store.PendingForTarget(ctx, req.Environment, req.DatabaseType, req.DatabaseName, req.ID) + siblings, err := store.PendingForTarget(ctx, req.Environment, req.DatabaseType, req.DatabaseName, req.Kind, req.ID) if err != nil { // Coalescing is an optimization: without the sibling list each pending // request runs its own fan-out, which re-plans the same PRs again — diff --git a/pkg/webhook/merge_gate_integration_test.go b/pkg/webhook/merge_gate_integration_test.go index 37ed24a23..2ebbb0c84 100644 --- a/pkg/webhook/merge_gate_integration_test.go +++ b/pkg/webhook/merge_gate_integration_test.go @@ -155,7 +155,7 @@ func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { // transition, so it must be visible as soon as the apply is completed. var gateReq *storage.MergeGateRequest require.EventuallyWithT(t, func(collect *assert.CollectT) { - req, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), apply.ID) + req, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindSettle) if !assert.NoError(collect, err) || !assert.NotNil(collect, req) { return } @@ -228,7 +228,7 @@ func TestE2EMergeGateSweepBackfillsMissedApply(t *testing.T) { h := newE2EHandler(t, svc, gh.NewClient(nil)) h.sweepMergeGateRequests(ctx) - gateReq, err := svc.Storage().MergeGateRequests().GetByApplyID(ctx, applyID) + gateReq, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(ctx, applyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, gateReq, "the sweep must backfill a refresh request for a completed apply that has none") assert.Equal(t, apply.ApplyIdentifier, gateReq.ApplyIdentifier) @@ -240,7 +240,7 @@ func TestE2EMergeGateSweepBackfillsMissedApply(t *testing.T) { // Recording is idempotent per apply: a second sweep pass over the same // window must not duplicate or reset the request. h.sweepMergeGateRequests(ctx) - again, err := svc.Storage().MergeGateRequests().GetByApplyID(ctx, applyID) + again, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(ctx, applyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, again) assert.Equal(t, gateReq.ID, again.ID) @@ -285,6 +285,7 @@ func TestE2EMergeGateReplansSiblingPRAndSkipsOriginator(t *testing.T) { applyIdentifier := fmt.Sprintf("apply_mergegate_fanout_%d", time.Now().UnixNano()) gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000001, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: applyIdentifier, Environment: "staging", DatabaseType: "mysql", @@ -318,7 +319,7 @@ func TestE2EMergeGateReplansSiblingPRAndSkipsOriginator(t *testing.T) { assert.Empty(t, originatorAfter.BlockingReason) // The request itself is terminal-successful. - finished, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), gateReq.ApplyID) + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), gateReq.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, finished) assert.Equal(t, storage.MergeGateCompleted, finished.State) @@ -356,6 +357,7 @@ func TestE2EMergeGateReplanFailureFailsCheckClosed(t *testing.T) { applyIdentifier := fmt.Sprintf("apply_mergegate_failclosed_%d", time.Now().UnixNano()) gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000002, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: applyIdentifier, Environment: "staging", DatabaseType: "mysql", @@ -376,7 +378,7 @@ func TestE2EMergeGateReplanFailureFailsCheckClosed(t *testing.T) { assert.Contains(t, blocked.ChangeSummary, "re-plan failed — see server logs") assert.Contains(t, blocked.ChangeSummary, applyIdentifier) - finished, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), gateReq.ApplyID) + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), gateReq.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, finished) assert.Equal(t, storage.MergeGateCompleted, finished.State) @@ -410,6 +412,7 @@ func TestE2EMergeGateLeavesInFlightApplyCheckUntouched(t *testing.T) { gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000003, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: fmt.Sprintf("apply_mergegate_inflight_%d", time.Now().UnixNano()), Environment: "staging", DatabaseType: "mysql", @@ -427,7 +430,7 @@ func TestE2EMergeGateLeavesInFlightApplyCheckUntouched(t *testing.T) { assert.Equal(t, "apply in flight", untouched.ChangeSummary) assert.Empty(t, untouched.BlockingReason) - finished, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), gateReq.ApplyID) + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), gateReq.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, finished) assert.Equal(t, storage.MergeGateCompleted, finished.State) @@ -454,6 +457,7 @@ func TestE2EMergeGateCoalescesPendingSiblingRequests(t *testing.T) { first := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000004, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: fmt.Sprintf("apply_mergegate_coalesce_a_%d", time.Now().UnixNano()), Environment: "staging", DatabaseType: "mysql", @@ -462,6 +466,7 @@ func TestE2EMergeGateCoalescesPendingSiblingRequests(t *testing.T) { }) second := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000005, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: fmt.Sprintf("apply_mergegate_coalesce_b_%d", time.Now().UnixNano()), Environment: "staging", DatabaseType: "mysql", @@ -471,13 +476,13 @@ func TestE2EMergeGateCoalescesPendingSiblingRequests(t *testing.T) { h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) - driven, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), first.ApplyID) + driven, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), first.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, driven) assert.Equal(t, storage.MergeGateCompleted, driven.State) assert.Equal(t, 1, driven.Attempts, "the older request runs the fan-out") - coalesced, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), second.ApplyID) + coalesced, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), second.ApplyID, storage.MergeGateKindSettle) require.NoError(t, err) require.NotNil(t, coalesced) assert.Equal(t, storage.MergeGateCompleted, coalesced.State) @@ -511,6 +516,7 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { // ticker, so nothing but a kick can drain the next request. sentinel := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000006, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: fmt.Sprintf("apply_mergegate_kick_sentinel_%d", time.Now().UnixNano()), Environment: "staging", DatabaseType: "mysql", @@ -523,7 +529,7 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { t.Cleanup(h.StopMergeGateProcessor) require.EventuallyWithT(t, func(collect *assert.CollectT) { - got, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), sentinel.ApplyID) + got, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), sentinel.ApplyID, storage.MergeGateKindSettle) if !assert.NoError(collect, err) || !assert.NotNil(collect, got) { return } @@ -533,6 +539,7 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { kicked := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000007, + Kind: storage.MergeGateKindSettle, ApplyIdentifier: fmt.Sprintf("apply_mergegate_kick_%d", time.Now().UnixNano()), Environment: "staging", DatabaseType: "mysql", @@ -543,7 +550,7 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { svc.OnMergeGateRecorded() require.EventuallyWithT(t, func(collect *assert.CollectT) { - got, err := svc.Storage().MergeGateRequests().GetByApplyID(t.Context(), kicked.ApplyID) + got, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), kicked.ApplyID, storage.MergeGateKindSettle) if !assert.NoError(collect, err) || !assert.NotNil(collect, got) { return }