diff --git a/pkg/api/merge_gate_record_test.go b/pkg/api/merge_gate_record_test.go index 2025b83e..25a1771f 100644 --- a/pkg/api/merge_gate_record_test.go +++ b/pkg/api/merge_gate_record_test.go @@ -4,6 +4,7 @@ import ( "context" "log/slog" "os" + "sync" "testing" "github.com/stretchr/testify/assert" @@ -22,57 +23,123 @@ func (s *staticGetApplyStore) Get(context.Context, int64) (*storage.Apply, error return s.apply, nil } +// capturingMergeGateStore records requests and serves per-kind rows, so +// both the drive tail's settle recording and the preflight gate's +// record-then-poll loop can run against it. Mutating a stored row's state +// from the consumer callback stands in for the processor completing the +// fan-out; the mutex keeps that write safe against the gate's poll reads. type capturingMergeGateStore struct { storage.MergeGateRequestStore + mu sync.Mutex + rows map[string]*storage.MergeGateRequest recorded []*storage.MergeGateRequest + reopened int } func (s *capturingMergeGateStore) Record(_ context.Context, req *storage.MergeGateRequest) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() s.recorded = append(s.recorded, req) + cp := *req + cp.State = storage.MergeGatePending + if s.rows == nil { + s.rows = map[string]*storage.MergeGateRequest{} + } + s.rows[req.Kind] = &cp return true, nil } +func (s *capturingMergeGateStore) GetByApplyAndKind(_ context.Context, _ int64, kind string) (*storage.MergeGateRequest, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.rows[kind] == nil { + return nil, nil + } + cp := *s.rows[kind] + return &cp, nil +} + +func (s *capturingMergeGateStore) ReopenForRetry(context.Context, int64) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.reopened++ + row := s.rows[storage.MergeGateKindPreflight] + if row == nil || row.State != storage.MergeGateFailed { + return false, nil + } + row.State = storage.MergeGatePending + row.RetryAfter = nil + return true, nil +} + +// setRowState mutates a stored row the way the processor's finish would. +func (s *capturingMergeGateStore) setRowState(kind, st string) { + s.mu.Lock() + defer s.mu.Unlock() + s.rows[kind].State = st +} + +func (s *capturingMergeGateStore) recordedCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.recorded) +} + +type staticTaskCountStore struct { + storage.TaskStore + count int64 +} + +func (s *staticTaskCountStore) CountByApplyID(context.Context, int64) (int64, error) { + return s.count, nil +} + type mockStorageWithMergeGate struct { mockStorage applies storage.ApplyStore + tasks storage.TaskStore mergeGate storage.MergeGateRequestStore } func (m *mockStorageWithMergeGate) Applies() storage.ApplyStore { return m.applies } +func (m *mockStorageWithMergeGate) Tasks() storage.TaskStore { return m.tasks } func (m *mockStorageWithMergeGate) MergeGateRequests() storage.MergeGateRequestStore { return m.mergeGate } +// newMergeGateTestService builds a Service over the capturing merge gate +// store with a single apply in the given state that owns taskCount task rows. +func newMergeGateTestService(applyState string, taskCount int64) (*Service, *capturingMergeGateStore) { + gateStore := &capturingMergeGateStore{} + st := &mockStorageWithMergeGate{ + applies: &staticGetApplyStore{apply: &storage.Apply{ + ID: 7, + ApplyIdentifier: "apply-gate-test", + Database: "gate_db", + DatabaseType: "mysql", + Environment: "staging", + Repository: "octocat/hello-world", + PullRequest: 1, + Caller: "cli:tester@host", + State: applyState, + }}, + tasks: &staticTaskCountStore{count: taskCount}, + mergeGate: gateStore, + } + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) + return New(st, testServerConfig(), nil, logger), gateStore +} + // TestRecordMergeGateGatedOnConsumer verifies the drive tail records a -// merge gate request only when a merge gate consumer is registered. A server -// with no GitHub runtime — a gRPC/CLI-only deployment — has no PR check state -// to refresh and no processor to drain requests, so a recorded row would sit +// settle request only when a merge gate consumer is registered. A server with no +// GitHub runtime — a gRPC/CLI-only deployment — has no PR check state to +// refresh and no processor to drain requests, so a recorded row would sit // pending forever; the drive tail must skip recording entirely there. With a -// consumer registered, the request is recorded with the apply's target and +// consumer registered, the settle is recorded with the apply's target and // attribution and the consumer is woken. func TestRecordMergeGateGatedOnConsumer(t *testing.T) { - newService := func() (*Service, *capturingMergeGateStore) { - gateStore := &capturingMergeGateStore{} - st := &mockStorageWithMergeGate{ - applies: &staticGetApplyStore{apply: &storage.Apply{ - ID: 7, - ApplyIdentifier: "apply-gate-test", - Database: "gate_db", - DatabaseType: "mysql", - Environment: "staging", - Repository: "octocat/hello-world", - PullRequest: 1, - Caller: "cli:tester@host", - State: state.Apply.Completed, - }}, - mergeGate: gateStore, - } - logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) - return New(st, testServerConfig(), nil, logger), gateStore - } - t.Run("no consumer registered skips recording", func(t *testing.T) { - svc, gateStore := newService() + svc, gateStore := newMergeGateTestService(state.Apply.Completed, 1) svc.recordMergeGateIfApplyResolved(t.Context(), 0, 7) @@ -80,8 +147,8 @@ func TestRecordMergeGateGatedOnConsumer(t *testing.T) { "a server without a merge gate consumer must not record requests nothing will drain") }) - t.Run("registered consumer records and is woken", func(t *testing.T) { - svc, gateStore := newService() + t.Run("registered consumer records a settle and is woken", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Completed, 1) woken := 0 svc.OnMergeGateRecorded = func() { woken++ } @@ -89,8 +156,8 @@ 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, "apply-gate-test", recorded.ApplyIdentifier) assert.Equal(t, "gate_db", recorded.DatabaseName) assert.Equal(t, "mysql", recorded.DatabaseType) assert.Equal(t, "staging", recorded.Environment) @@ -100,3 +167,159 @@ func TestRecordMergeGateGatedOnConsumer(t *testing.T) { assert.Equal(t, 1, woken, "the drive tail wakes the consumer exactly once per recording") }) } + +// TestRecordMergeGateOnTerminalStates verifies which terminal outcomes get +// a settle. A completed apply always does — it changed the live schema. A +// failed apply changed nothing, so it needs a settle only when its preflight +// held sibling PR checks: the settle's re-plan is what releases those holds. +// A non-terminal apply never records one. +func TestRecordMergeGateOnTerminalStates(t *testing.T) { + t.Run("non-terminal apply records nothing", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Running, 1) + svc.OnMergeGateRecorded = func() {} + + svc.recordMergeGateIfApplyResolved(t.Context(), 0, 7) + + assert.Empty(t, gateStore.recorded) + }) + + t.Run("failed apply without a preflight records nothing", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Failed, 1) + svc.OnMergeGateRecorded = func() {} + + svc.recordMergeGateIfApplyResolved(t.Context(), 0, 7) + + assert.Empty(t, gateStore.recorded, + "a failed apply that never held sibling checks has nothing to release or refresh") + }) + + t.Run("failed apply with a preflight records the releasing settle", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Failed, 1) + svc.OnMergeGateRecorded = func() {} + _, err := gateStore.Record(t.Context(), &storage.MergeGateRequest{ + ApplyID: 7, + Kind: storage.MergeGateKindPreflight, + }) + require.NoError(t, err) + + svc.recordMergeGateIfApplyResolved(t.Context(), 0, 7) + + require.Len(t, gateStore.recorded, 2) + assert.Equal(t, storage.MergeGateKindSettle, gateStore.recorded[1].Kind, + "the settle releases the holds the preflight placed") + }) +} + +// TestGateApplyStartOnCheckPreflight verifies the hard gate in front of an +// apply's engine work: the drive may start only once the preflight fan-out +// has confirmed sibling PR check holds. The gate does not apply on servers +// with no merge gate consumer or to applies that own no schema change tasks; it +// passes immediately on an already-completed preflight, records and waits +// otherwise, re-arms a terminally failed request, and fails closed when the +// drive context ends first. +func TestGateApplyStartOnCheckPreflight(t *testing.T) { + apply := &storage.Apply{ + ID: 7, + ApplyIdentifier: "apply-gate-test", + Database: "gate_db", + DatabaseType: "mysql", + Environment: "staging", + Repository: "octocat/hello-world", + PullRequest: 1, + Caller: "cli:tester@host", + State: state.Apply.Pending, + } + + t.Run("no consumer: the apply starts ungated", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 1) + + err := svc.gateApplyStartOnCheckPreflight(t.Context(), 0, apply, "default") + + require.NoError(t, err) + assert.Zero(t, gateStore.recordedCount()) + }) + + t.Run("task-less apply skips the preflight", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 0) + svc.OnMergeGateRecorded = func() {} + + err := svc.gateApplyStartOnCheckPreflight(t.Context(), 0, apply, "default") + + require.NoError(t, err) + assert.Zero(t, gateStore.recordedCount(), + "an apply with no schema change tasks cannot invalidate sibling verdicts") + }) + + t.Run("records the preflight, wakes the consumer, and passes once it completes", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 1) + woken := 0 + svc.OnMergeGateRecorded = func() { + woken++ + // Stand in for the processor: the wake-up drains the request. + gateStore.setRowState(storage.MergeGateKindPreflight, storage.MergeGateCompleted) + } + + err := svc.gateApplyStartOnCheckPreflight(t.Context(), 0, apply, "default") + + require.NoError(t, err) + require.Equal(t, 1, gateStore.recordedCount()) + assert.Equal(t, storage.MergeGateKindPreflight, gateStore.recorded[0].Kind) + assert.Equal(t, "cli:tester@host", gateStore.recorded[0].RequestedBy) + assert.Equal(t, 1, woken) + }) + + t.Run("already-completed preflight passes without recording", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 1) + svc.OnMergeGateRecorded = func() {} + _, err := gateStore.Record(t.Context(), &storage.MergeGateRequest{ + ApplyID: 7, + Kind: storage.MergeGateKindPreflight, + }) + require.NoError(t, err) + gateStore.setRowState(storage.MergeGateKindPreflight, storage.MergeGateCompleted) + gateStore.mu.Lock() + gateStore.recorded = nil + gateStore.mu.Unlock() + + err = svc.gateApplyStartOnCheckPreflight(t.Context(), 0, apply, "default") + + require.NoError(t, err) + assert.Zero(t, gateStore.recordedCount(), + "a resume or cutover drive pays one read, not a new request") + }) + + t.Run("terminally failed preflight is re-armed and the gate keeps waiting", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 1) + _, err := gateStore.Record(t.Context(), &storage.MergeGateRequest{ + ApplyID: 7, + Kind: storage.MergeGateKindPreflight, + }) + require.NoError(t, err) + gateStore.setRowState(storage.MergeGateKindPreflight, storage.MergeGateFailed) + svc.OnMergeGateRecorded = func() { + // Stand in for the processor draining the re-armed request. + gateStore.setRowState(storage.MergeGateKindPreflight, storage.MergeGateCompleted) + } + + err = svc.gateApplyStartOnCheckPreflight(t.Context(), 0, apply, "default") + + require.NoError(t, err) + gateStore.mu.Lock() + reopened := gateStore.reopened + gateStore.mu.Unlock() + assert.Equal(t, 1, reopened, + "a preflight that exhausted its retries during an outage must self-heal, not block the apply forever") + }) + + t.Run("fails closed when the drive context ends before the holds land", func(t *testing.T) { + svc, gateStore := newMergeGateTestService(state.Apply.Pending, 1) + svc.OnMergeGateRecorded = func() {} // no processor: the request stays pending + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err := svc.gateApplyStartOnCheckPreflight(ctx, 0, apply, "default") + + require.Error(t, err, "unconfirmed holds must abandon the drive attempt, never start the engine") + assert.Equal(t, 1, gateStore.recordedCount()) + }) +} diff --git a/pkg/api/operator.go b/pkg/api/operator.go index abdf073c..ae9eec79 100644 --- a/pkg/api/operator.go +++ b/pkg/api/operator.go @@ -1019,56 +1019,69 @@ func (s *Service) completePendingRequestForResolvedApply(ctx context.Context, dr return nil } -// hasMergeGateConsumer reports whether a merge gate consumer — the -// webhook handler's merge gate processor — exists on this server. The handler -// registers OnMergeGateRecorded at construction, so a nil callback means -// no GitHub runtime is configured: no PR check state to refresh and no -// processor to drain requests. +// hasMergeGateConsumer reports whether a merge gate consumer — a +// running refresh processor, registered by the code-host integration — exists +// on this server. A nil callback means no code-host integration is running a +// processor: no stored check state to refresh and nothing to drain requests. func (s *Service) hasMergeGateConsumer() bool { return s.OnMergeGateRecorded != nil } -// recordMergeGateIfApplyResolved records a durable merge gate request -// once the apply has settled to terminal success. A completed apply — +// recordMergeGateIfApplyResolved records a durable settle merge gate +// request once the apply has settled terminally. A completed apply — // including a completed rollback — changes the live schema of its // (environment, database type, database) target, which stales the stored plan -// check state of every other open PR planning against that target. The check -// merge gate processor consumes the durable request to re-plan those PRs. The -// apply is reloaded because the derived-state write operates on a copy and -// does not mutate the caller's row. Recording is idempotent (one request per -// apply) and never fails the drive tail: errors are logged and counted, and -// the backstop sweep over recently completed applies re-records anything -// missed here. No-op on a server with no merge gate consumer (no GitHub -// runtime configured) and for every settled state other than terminal -// success — only terminal success mutates the target schema. +// check state of every other open change planning against that target; the merge +// gate processor consumes the settle to re-plan those PRs. A terminal +// outcome that did not change the schema (failed, stopped, cancelled) needs a +// settle only when a preflight request held sibling checks before the apply +// started — the settle's re-plan is what releases those holds. The apply is +// reloaded because the derived-state write operates on a copy and does not +// mutate the caller's row. Recording is idempotent (one request per apply and +// kind) and never fails the drive tail: errors are logged and counted, and +// the backstop sweeps re-record anything missed here. No-op on a server with +// no merge gate consumer (no code-host integration running a processor). func (s *Service) recordMergeGateIfApplyResolved(ctx context.Context, driverID int, applyID int64) { if !s.hasMergeGateConsumer() { - // Without a GitHub webhook runtime this server has no PR check state - // to refresh and no processor to drain requests, so a recorded row - // would sit pending forever. - s.logger.Debug("operator: no merge gate consumer registered (GitHub is not configured on this server); skipping merge gate recording", + // Without a code-host integration this server has no stored check + // state to refresh and no processor to drain requests, so a recorded + // row would sit pending forever. + s.logger.Debug("operator: no merge gate consumer registered (no code-host integration on this server); skipping merge gate recording", "driver", driverID) return } apply, err := s.storage.Applies().Get(ctx, applyID) if err != nil { - s.logger.Error("operator: failed to reload apply before recording merge gate request; the backstop sweep will record it", + s.logger.Error("operator: failed to reload apply before recording settle merge gate request; the backstop sweep will record it", "driver", driverID, "error", fmt.Errorf("reload apply %d: %w", applyID, err)) return } if apply == nil { - s.logger.Error("operator: apply not found while recording merge gate request; sibling checks will not be re-planned", + s.logger.Error("operator: apply not found while recording settle merge gate request; no settle will be recorded", "driver", driverID, "error", fmt.Errorf("reload apply %d: %w", applyID, storage.ErrApplyNotFound)) return } - if !state.IsState(apply.State, state.Apply.Completed) { - // Only terminal success mutates the target schema; every other outcome - // (still running, stopped, cancelled, failed, reverted) leaves sibling - // plan checks accurate. - s.logger.Debug("operator: apply did not settle to terminal success; no merge gate recorded", + if !state.IsTerminalApplyState(apply.State) { + s.logger.Debug("operator: apply is not terminal; no settle merge gate recorded", append(apply.LogAttrs(), "driver", driverID)...) return } + if !state.IsState(apply.State, state.Apply.Completed) { + // Only terminal success mutates the target schema, so a non-success + // outcome needs a settle only to release a preflight hold. + preflight, err := s.storage.MergeGateRequests().GetByApplyAndKind(ctx, apply.ID, storage.MergeGateKindPreflight) + if err != nil { + s.logger.Error("operator: failed to look up preflight merge gate request for settled apply; the release sweep will record the settle if sibling checks are held", + append(apply.LogAttrs(), "driver", driverID, "error", err)...) + metrics.RecordMergeGateRecordFailure(ctx, apply.Database, apply.Environment) + return + } + if preflight == nil { + s.logger.Debug("operator: apply settled without terminal success and no preflight held sibling checks; no settle merge gate needed", + append(apply.LogAttrs(), "driver", driverID)...) + return + } + } recorded, err := s.storage.MergeGateRequests().Record(ctx, &storage.MergeGateRequest{ ApplyID: apply.ID, @@ -1082,23 +1095,168 @@ func (s *Service) recordMergeGateIfApplyResolved(ctx context.Context, driverID i RequestedBy: apply.Caller, }) if err != nil { - s.logger.Error("operator: failed to record merge gate request for completed apply; sibling PR checks stay stale until the backstop sweep records it", + s.logger.Error("operator: failed to record settle merge gate request for settled apply; sibling change checks stay stale or held until the backstop sweep records it", append(apply.LogAttrs(), "driver", driverID, "error", err)...) metrics.RecordMergeGateRecordFailure(ctx, apply.Database, apply.Environment) return } if !recorded { - s.logger.Debug("operator: merge gate request already recorded for completed apply", + s.logger.Debug("operator: settle merge gate request already recorded for settled apply", append(apply.LogAttrs(), "driver", driverID)...) return } - s.logger.Info("operator: recorded merge gate request for completed apply; sibling PR checks against the target will be re-planned", + s.logger.Info("operator: recorded settle merge gate request; sibling change checks against the target will be re-planned", append(apply.LogAttrs(), "driver", driverID)...) metrics.RecordMergeGateRecorded(ctx, apply.Database, apply.Environment, metrics.MergeGateSourceDriveTail) // Non-nil by the consumer gate above; wake the processor to drain now. s.OnMergeGateRecorded() } +const ( + // checkPreflightGateTimeout bounds how long one drive attempt waits for + // the preflight fan-out to hold sibling change checks before the apply's + // engine work may start. It exceeds the merge gate processor's poll + // interval so a wake-up kick lost across pods still completes within one + // wait. On expiry the drive attempt is abandoned and the apply stays + // claimable — fail closed: engine work never starts before the holds are + // confirmed. + checkPreflightGateTimeout = 90 * time.Second + + // checkPreflightGatePollInterval is the cadence at which the gate re-reads + // the preflight request while waiting for the fan-out to complete. + checkPreflightGatePollInterval = time.Second +) + +// gateApplyStartOnCheckPreflight blocks an apply's engine work until the +// preflight merge gate fan-out has durably held every sibling change's +// stored check on the apply's target action-required (with a comment on the +// change explaining the hold). Merges must not land on check verdicts this +// apply is about to invalidate, so the holds are a hard precondition of the +// drive: a nil return means the holds are confirmed (or the gate does not +// apply — no merge gate consumer, or a task-less apply that cannot change +// the schema). An error +// means the holds could not be confirmed; the caller abandons the drive +// attempt and the apply stays claimable, so the start is retried on a later +// poll and the apply never runs un-preflighted. +// +// The gate is idempotent across drive attempts, lease handovers, and resumes: +// the preflight request is unique per apply, and a completed request passes +// immediately, so mid-apply resumes and cutover drives pay one storage read. +func (s *Service) gateApplyStartOnCheckPreflight(ctx context.Context, driverID int, apply *storage.Apply, deployment string) error { + if !s.hasMergeGateConsumer() { + // Without a code-host integration there is no stored check state to + // hold and no processor to drain the request; the apply starts ungated. + s.logger.Debug("operator: no merge gate consumer registered (no code-host integration on this server); apply starts without a check preflight", + "driver", driverID) + return nil + } + + store := s.storage.MergeGateRequests() + req, err := store.GetByApplyAndKind(ctx, apply.ID, storage.MergeGateKindPreflight) + if err != nil { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("load preflight merge gate request: %w", err) + } + if req != nil && req.State == storage.MergeGateCompleted { + // The common resume/cutover fast path: holds were confirmed on an + // earlier drive attempt. + s.logger.Debug("operator: check preflight already completed; apply may start", + append(apply.LogAttrs(), "driver", driverID, "operation_deployment", deployment)...) + return nil + } + if req == nil { + // Only an apply that executes DDL invalidates sibling check verdicts. + // An apply that owns no task rows cannot change the schema (its drive + // fails closed on the no-tasks claim gate), so it holds nothing. + taskCount, err := s.storage.Tasks().CountByApplyID(ctx, apply.ID) + if err != nil { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("count apply tasks before check preflight: %w", err) + } + if taskCount == 0 { + s.logger.Debug("operator: apply owns no schema change tasks; no check preflight needed", + append(apply.LogAttrs(), "driver", driverID, "operation_deployment", deployment)...) + return nil + } + recorded, err := store.Record(ctx, &storage.MergeGateRequest{ + ApplyID: apply.ID, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: apply.ApplyIdentifier, + Environment: apply.Environment, + DatabaseType: apply.DatabaseType, + DatabaseName: apply.Database, + Repository: apply.Repository, + ChangeKey: storage.ChangeKeyForPullRequest(apply.PullRequest), + RequestedBy: apply.Caller, + }) + if err != nil { + metrics.RecordMergeGateRecordFailure(ctx, apply.Database, apply.Environment) + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("record preflight merge gate request: %w", err) + } + if recorded { + s.logger.Info("operator: recorded preflight merge gate request; apply start waits for sibling change checks on the target to be held", + append(apply.LogAttrs(), "driver", driverID, "operation_deployment", deployment)...) + metrics.RecordMergeGateRecorded(ctx, apply.Database, apply.Environment, metrics.MergeGateSourcePreflightGate) + } + // Non-nil by the consumer gate above; wake the processor to drain now. + s.OnMergeGateRecorded() + } + + return s.waitForCheckPreflight(ctx, driverID, apply, deployment) +} + +// waitForCheckPreflight polls the apply's preflight request until the fan-out +// completes, the gate deadline expires, or the drive context ends. A +// terminally failed request is re-armed to pending with a fresh attempt +// budget so a long outage (for example the code host unavailable past the +// retry cap) blocks the apply only until the cause clears, not until manual +// intervention. +func (s *Service) waitForCheckPreflight(ctx context.Context, driverID int, apply *storage.Apply, deployment string) error { + store := s.storage.MergeGateRequests() + deadline := time.Now().Add(checkPreflightGateTimeout) + for { + req, err := store.GetByApplyAndKind(ctx, apply.ID, storage.MergeGateKindPreflight) + if err != nil { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("poll preflight merge gate request: %w", err) + } + if req == nil { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("preflight merge gate request for apply %s disappeared while the gate waited on it", apply.ApplyIdentifier) + } + if req.State == storage.MergeGateCompleted { + s.logger.Info("operator: check preflight completed; sibling change checks on the target are held and the apply may start", + append(apply.LogAttrs(), "driver", driverID, "operation_deployment", deployment)...) + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "passed") + return nil + } + if req.State == storage.MergeGateFailed && req.RetryAfter == nil { + reopened, err := store.ReopenForRetry(ctx, req.ID) + if err != nil { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("re-arm terminally failed preflight merge gate request: %w", err) + } + if reopened { + s.logger.Warn("operator: preflight merge gate request had terminally failed; re-armed it for retry and the gate keeps waiting", + append(apply.LogAttrs(), "driver", driverID, "operation_deployment", deployment, "last_error", req.LastError)...) + s.OnMergeGateRecorded() + } + } + if time.Now().After(deadline) { + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "timeout") + return fmt.Errorf("preflight holds for apply %s not confirmed within %s (request state %s, attempts %d): apply start stays blocked until the merge gate processor confirms them", + apply.ApplyIdentifier, checkPreflightGateTimeout, req.State, req.Attempts) + } + select { + case <-ctx.Done(): + metrics.RecordCheckPreflightGateOutcome(ctx, apply.Database, apply.Environment, "error") + return fmt.Errorf("drive context ended while waiting for preflight holds: %w", ctx.Err()) + case <-time.After(checkPreflightGatePollInterval): + } + } +} + // reconcileUnclaimableParent handles a claimed operation whose parent apply // ClaimApplyByID refused. If the parent is terminal, the operation row is // reconciled to that terminal state so it stops being re-claimed on every poll @@ -1266,6 +1424,23 @@ func (s *Service) resumeClaimedApplyWithOptions(ctx context.Context, driverID in // the last failure and the resumed work. s.logApplyResumeClaim(ctx, driverID, apply) + // Engine work must not start until sibling change checks on the target are + // held: their verdicts were computed against the schema this apply is + // about to change, and a merge must not land on them mid-apply. The gate + // waits for the durable preflight fan-out and fails closed — an + // unconfirmed hold abandons this drive attempt and leaves the apply + // claimable for a later poll. + if err := s.gateApplyStartOnCheckPreflight(ctx, driverID, apply, deployment); err != nil { + s.logger.Error("operator: check preflight gate did not confirm sibling change check holds; the apply will not start on this attempt and stays claimable", + append(apply.LogAttrs(), + "driver", driverID, + "apply_operation_id", applyOperationID, + "operation_deployment", deployment, + "error", err)...) + metrics.RecordOperatorResumeFailure(ctx, apply.Database, deployment, apply.Environment, "check_preflight_gate") + return false, err + } + previousState := apply.State client, err := s.RoutingTernClient() diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 88ca6909..9c8d10c4 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -1520,6 +1520,9 @@ const ( // MergeGateSourceSweep marks a request recorded by the backstop sweep // over recently completed applies. MergeGateSourceSweep = "sweep" + // MergeGateSourcePreflightGate marks a preflight request recorded by + // the operator gate before an apply's engine work starts. + MergeGateSourcePreflightGate = "preflight_gate" // MergeGateSourceReleaseSweep marks a settle request backfilled by the // sweep over terminal applies whose preflight held sibling checks but // whose settle was never recorded. @@ -1596,6 +1599,26 @@ func RecordMergeGateEventOutcome(ctx context.Context, database, environment, out ) } +// RecordCheckPreflightGateOutcome counts outcomes of the operator gate that +// blocks an apply's engine work until sibling changes' stored checks on its +// target are held action-required. Outcomes: +// - "passed": the preflight fan-out confirmed the holds; the apply started. +// - "timeout": the holds were not confirmed within the gate deadline; the +// drive attempt was abandoned and the apply stays claimable. A sustained +// rate means the merge gate processor is failing to drain preflight +// requests — check its logs; applies on servers with a merge gate +// consumer will not start until it recovers. +// - "error": the gate could not read or record the preflight request +// (storage failure); the drive attempt was abandoned, fail closed. +func RecordCheckPreflightGateOutcome(ctx context.Context, database, environment, outcome string) { + addCounter(ctx, "schemabot.merge_gate.preflight_gate_total", + "Total operator preflight gate outcomes before apply engine work starts", "{gate}", + attribute.String("database", database), + EnvironmentAttribute(environment), + attribute.String("outcome", outcome), + ) +} + // RecordMergeGateTerminatedStuck counts merge gate requests terminated // by the stuck-processing sweep: rows wedged past the attempt cap with an // expired lease (a driver hard-killed on its final attempt). Each terminated diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 386774e7..f4d5221b 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -340,7 +340,7 @@ type MergeGateRequestStore interface { // 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 + // long code-host 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) @@ -400,7 +400,7 @@ type MergeGateRequestStore interface { // 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 + // preflight request row, but no settle row. A preflight holds sibling change // 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. diff --git a/pkg/webhook/merge_gate_integration_test.go b/pkg/webhook/merge_gate_integration_test.go index 32b64fdd..0ed82bd5 100644 --- a/pkg/webhook/merge_gate_integration_test.go +++ b/pkg/webhook/merge_gate_integration_test.go @@ -8,9 +8,9 @@ // terminally, a settle request re-plans those siblings against the live // schema, refreshing stale verdicts and releasing the holds. These tests // exercise the durable request lifecycle end to end against the real webhook -// harness: recording at the operator drive tail, the backstop and release -// sweeps, the hold and re-plan fan-outs with attribution, the fail-closed -// flip when a re-plan fails, the in-flight apply +// harness: the operator gate and drive tail recording both kinds, the +// backstop and release sweeps, the hold and re-plan fan-outs with +// attribution, the fail-closed flip when a re-plan fails, the in-flight apply // guard, same-target request coalescing, settle deferral behind an active // preflighted apply, and the recorded-request kick that drains without // waiting for a poll tick. @@ -85,10 +85,12 @@ func seedRefreshTargetCheck(t *testing.T, svc *api.Service, pr int, env, dbName, } // TestE2EMergeGateRecordedOnApplyTerminalSuccess drives a real apply -// through the webhook command path to terminal success and verifies the -// operator drive tail durably records a merge gate request for the apply's -// target before the apply is considered done — the request other pods' sibling -// PR checks are refreshed from. +// through the webhook command path to terminal success and verifies both +// durable requests around it: the operator gate records a preflight the +// processor must complete before the apply's engine work starts, and the +// drive tail records a settle for the apply's target before the apply is +// considered done — the request other pods' sibling PR checks are refreshed +// from. func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { clearMergeGateRequests(t) dbName := "webhook_mergegate_drivetail" @@ -111,14 +113,18 @@ func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { // The drive tail must invoke the registered recorded-notifier so a // co-located processor drains the request immediately instead of waiting - // for its next poll tick. The probe stands in for the processor's kick, - // which only a started processor registers. + // for its next poll tick. The probe chains the handler's own kick: the + // preflight gate blocks the apply until the processor completes the + // preflight fan-out, so the wake-up must still reach the processor. + kick := svc.OnMergeGateRecorded + require.NotNil(t, kick, "the handler registers the processor kick on the service at construction") kicked := make(chan struct{}, 1) svc.OnMergeGateRecorded = func() { select { case kicked <- struct{}{}: default: } + kick() } req := buildWebhookRequest(t, webhookPayloadOpts{ @@ -158,7 +164,15 @@ func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { assert.Fail(collect, "no completed apply for the target database yet") }, webhookIntegrationPollDeadline, 100*time.Millisecond) - // The drive tail records the merge gate request as part of the terminal + // The gate's preflight is a hard precondition of the drive: a completed + // apply proves its preflight fan-out finished before the engine started. + preflight, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, preflight, "the gate records a preflight request before the apply starts") + assert.Equal(t, storage.MergeGateCompleted, preflight.State, + "an apply cannot reach terminal state before its preflight fan-out completed") + + // The drive tail records the settle request as part of the terminal // transition, so it must be visible as soon as the apply is completed. var gateReq *storage.MergeGateRequest require.EventuallyWithT(t, func(collect *assert.CollectT) { @@ -176,7 +190,6 @@ func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { assert.Equal(t, "octocat/hello-world", gateReq.Repository) assert.Equal(t, "1", gateReq.ChangeKey) assert.Equal(t, apply.Caller, gateReq.RequestedBy) - assert.Equal(t, storage.MergeGatePending, gateReq.State) select { case <-kicked: @@ -232,7 +245,7 @@ func TestE2EMergeGateSweepBackfillsMissedApply(t *testing.T) { apply.CompletedAt = &completedAt require.NoError(t, svc.Storage().Applies().Update(ctx, apply)) - h := newE2EHandler(t, svc, gh.NewClient(nil)) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, gh.NewClient(nil)) h.sweepMergeGateRequests(ctx) gateReq, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(ctx, applyID, storage.MergeGateKindSettle) @@ -287,7 +300,7 @@ func TestE2EMergeGateReplansSiblingPRAndSkipsOriginator(t *testing.T) { originator := seedRefreshTargetCheck(t, svc, 2, "staging", dbName, checkStatusCompleted, checkConclusionActionRequired, "originator summary") - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) applyIdentifier := fmt.Sprintf("apply_mergegate_fanout_%d", time.Now().UnixNano()) gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ @@ -359,7 +372,7 @@ func TestE2EMergeGateReplanFailureFailsCheckClosed(t *testing.T) { seedRefreshTargetCheck(t, svc, 1, "staging", dbName, checkStatusCompleted, checkConclusionSuccess, "no changes") - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) applyIdentifier := fmt.Sprintf("apply_mergegate_failclosed_%d", time.Now().UnixNano()) gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ @@ -415,7 +428,7 @@ func TestE2EMergeGateLeavesInFlightApplyCheckUntouched(t *testing.T) { inFlight.ApplyID = 424242 require.NoError(t, svc.Storage().Checks().Upsert(t.Context(), inFlight)) - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) gateReq := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000003, @@ -460,7 +473,7 @@ func TestE2EMergeGateCoalescesPendingSiblingRequests(t *testing.T) { t.Cleanup(server.Close) client.BaseURL, _ = url.Parse(server.URL + "/") - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) first := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ ApplyID: 91000004, @@ -514,7 +527,7 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { t.Cleanup(server.Close) client.BaseURL, _ = url.Parse(server.URL + "/") - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) // A sentinel recorded before start is drained by the driver's startup // pass; its completion means the driver is parked on the (hour-long) @@ -652,7 +665,7 @@ func TestE2ECheckPreflightHoldsSiblingChecksAndComments(t *testing.T) { seedRefreshTargetCheck(t, svc, 1, "staging", dbName, checkStatusCompleted, checkConclusionSuccess, "no changes") - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) applyIdentifier := fmt.Sprintf("apply_mergegate_preflight_%d", time.Now().UnixNano()) preflight := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ @@ -768,7 +781,7 @@ func TestE2ECheckReleaseSweepSettlesFailedPreflightedApply(t *testing.T) { heldCheck.BlockingReason = applyInFlightBlock.blockingReason require.NoError(t, svc.Storage().Checks().Upsert(t.Context(), heldCheck)) - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) h.sweepPreflightedAppliesMissingSettle(t.Context()) settle, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindSettle) @@ -843,7 +856,7 @@ func TestE2ECheckSettleDefersToActivePreflightedApply(t *testing.T) { heldCheck.BlockingReason = applyInFlightBlock.blockingReason require.NoError(t, svc.Storage().Checks().Upsert(t.Context(), heldCheck)) - h := newE2EHandler(t, svc, client) + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) // An earlier apply on the same target settles while the later one runs. settle := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ diff --git a/pkg/webhook/webhook_integration_test.go b/pkg/webhook/webhook_integration_test.go index 7872f91b..80e43a89 100644 --- a/pkg/webhook/webhook_integration_test.go +++ b/pkg/webhook/webhook_integration_test.go @@ -274,9 +274,24 @@ func seedCheck(t *testing.T, svc *api.Service, dbName, env, conclusion string) { require.NoError(t, err) } -// newTestHandler creates a Handler wired to the given service and GitHub client, -// with an error-level logger to reduce test noise. +// newE2EHandler creates a Handler wired to the given service and GitHub +// client, with the merge gate processor running as the server runs it. +// The check preflight gate blocks every claimed apply until the processor +// confirms sibling PR check holds, so any test that drives a real apply +// needs the processor alive. func newE2EHandler(t *testing.T, svc *api.Service, client *gh.Client) *Handler { + t.Helper() + h := newE2EHandlerWithoutMergeGateProcessor(t, svc, client) + h.StartMergeGateProcessor(t.Context()) + t.Cleanup(h.StopMergeGateProcessor) + return h +} + +// newE2EHandlerWithoutMergeGateProcessor creates a Handler with an error-level +// logger and no background merge gate processor, for tests that drive the +// refresh request lifecycle by hand (manual sweeps and drains) and must not +// race a background pass. +func newE2EHandlerWithoutMergeGateProcessor(t *testing.T, svc *api.Service, client *gh.Client) *Handler { t.Helper() logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) installClient := ghclient.NewInstallationClientWithSlug(client, logger, "schemabot")