diff --git a/pkg/api/merge_gate_record_test.go b/pkg/api/merge_gate_record_test.go new file mode 100644 index 000000000..c501fcba7 --- /dev/null +++ b/pkg/api/merge_gate_record_test.go @@ -0,0 +1,101 @@ +package api + +import ( + "context" + "log/slog" + "os" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" +) + +type staticGetApplyStore struct { + storage.ApplyStore + apply *storage.Apply +} + +func (s *staticGetApplyStore) Get(context.Context, int64) (*storage.Apply, error) { + return s.apply, nil +} + +type capturingMergeGateStore struct { + storage.MergeGateRequestStore + recorded []*storage.MergeGateRequest +} + +func (s *capturingMergeGateStore) Record(_ context.Context, req *storage.MergeGateRequest) (bool, error) { + s.recorded = append(s.recorded, req) + return true, nil +} + +type mockStorageWithMergeGate struct { + mockStorage + applies storage.ApplyStore + mergeGate storage.MergeGateRequestStore +} + +func (m *mockStorageWithMergeGate) Applies() storage.ApplyStore { return m.applies } +func (m *mockStorageWithMergeGate) MergeGateRequests() storage.MergeGateRequestStore { + return m.mergeGate +} + +// 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 +// pending forever; the drive tail must skip recording entirely there. With a +// consumer registered, the request 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.recordMergeGateIfApplyResolved(t.Context(), 0, 7) + + assert.Empty(t, gateStore.recorded, + "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() + woken := 0 + svc.OnMergeGateRecorded = func() { woken++ } + + svc.recordMergeGateIfApplyResolved(t.Context(), 0, 7) + + require.Len(t, gateStore.recorded, 1) + recorded := gateStore.recorded[0] + 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) + assert.Equal(t, "octocat/hello-world", recorded.Repository) + assert.Equal(t, "1", recorded.ChangeKey) + assert.Equal(t, "cli:tester@host", recorded.RequestedBy) + assert.Equal(t, 1, woken, "the drive tail wakes the consumer exactly once per recording") + }) +} diff --git a/pkg/api/operator.go b/pkg/api/operator.go index 358aba108..1cbe9560c 100644 --- a/pkg/api/operator.go +++ b/pkg/api/operator.go @@ -556,6 +556,11 @@ func (s *Service) recoverSingleApplyOperation(ctx context.Context, driverID int, return } + // The merge gate request depends only on the apply settling to terminal + // success; record it before control-request cleanup so a cleanup error + // cannot suppress it. + s.recordMergeGateIfApplyResolved(applyLeaseCtx, driverID, finalApply.ID) + // If the derived state above settled the apply terminally and a stop or // cancel is still pending, complete it now so the request does not linger // after the rollout has resolved. @@ -733,6 +738,11 @@ func (s *Service) driveClaimedMultiOperation(ctx context.Context, driverID int, // error must not suppress it. s.publishTerminalSummaryIfWon(operationLeaseCtx, driverID, finalApply, result) + // Like the terminal summary, the merge gate request depends only on the + // apply settling to terminal success; record it before control-request + // cleanup so a cleanup error cannot suppress it. + s.recordMergeGateIfApplyResolved(operationLeaseCtx, driverID, finalApply.ID) + if err := s.completePendingControlRequestsIfApplyResolved(operationLeaseCtx, driverID, finalApply.ID); err != nil { s.logger.Error("operator: failed to complete pending control requests for resolved apply", append(finalApply.LogAttrs(), @@ -904,6 +914,12 @@ func (s *Service) recoverApplyPendingStop(ctx context.Context, driverID int, own // summary; publish it if this projection won the terminal swap. s.publishTerminalSummaryIfWon(applyLeaseCtx, driverID, finalApply, result) + // Like the terminal summary, the merge gate request depends only on the + // apply settling to terminal success (the data-plane apply can complete + // while a stop was requested); record it before control-request cleanup so + // a cleanup error cannot suppress it. + s.recordMergeGateIfApplyResolved(applyLeaseCtx, driverID, finalApply.ID) + if err := s.completePendingControlRequestsIfApplyResolved(applyLeaseCtx, driverID, finalApply.ID); err != nil { s.logger.Error("operator: failed to complete pending control requests after stop reconciliation", append(finalApply.LogAttrs(), @@ -1003,6 +1019,85 @@ 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. +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 — +// 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. +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", + "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", + "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", + "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", + append(apply.LogAttrs(), "driver", driverID)...) + return + } + + recorded, err := s.storage.MergeGateRequests().Record(ctx, &storage.MergeGateRequest{ + ApplyID: apply.ID, + 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 { + s.logger.Error("operator: failed to record merge gate request for completed apply; sibling PR checks stay stale 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", + 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", + 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() +} + // 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 diff --git a/pkg/api/service.go b/pkg/api/service.go index 7e8c78fb3..1d7f0ba78 100644 --- a/pkg/api/service.go +++ b/pkg/api/service.go @@ -178,6 +178,19 @@ type Service struct { // against any still-live per-driver observer. OnApplyTerminalSummary ApplyTerminalSummaryCallback + // OnMergeGateRecorded is called after a drive tail durably records a + // merge gate request. Set by the webhook handler to wake its merge gate + // processor immediately instead of waiting for the next poll tick; the + // durable request row stays the source of truth, so a lost wake-up only + // costs poll latency, never the fan-out. + // + // Registration doubles as the drive tails' consumer signal: when nil, no + // GitHub webhook runtime exists on this server — there is no PR check + // state to refresh and no processor to drain requests — so drive tails + // skip recording entirely. Implementations must be non-blocking and safe + // for concurrent drivers. + OnMergeGateRecorded func() + pendingObserverMu sync.Mutex pendingObservers map[pendingObserverKey]tern.ProgressObserver } diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 940efb835..bc2cb237c 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -1511,3 +1511,39 @@ func RecordPendingDropsCleanupError(ctx context.Context, database, environment, attribute.String("reason", reason), ) } + +// Merge gate recording sources for RecordMergeGateRecorded. +const ( + // MergeGateSourceDriveTail marks a request recorded inline by the + // operator drive tail that settled the apply. + MergeGateSourceDriveTail = "drive_tail" + // MergeGateSourceSweep marks a request recorded by the backstop sweep + // over recently completed applies. + MergeGateSourceSweep = "sweep" +) + +// RecordMergeGateRecorded counts durable merge gate requests recorded +// when an apply settles to terminal success. A sustained "sweep" rate means +// drive tails are failing to record — check the operator logs for the +// recording error. +func RecordMergeGateRecorded(ctx context.Context, database, environment, source string) { + addCounter(ctx, "schemabot.merge_gate.requests_recorded_total", + "Total durable merge gate requests recorded for applies that settled to terminal success", "{request}", + attribute.String("database", database), + EnvironmentAttribute(environment), + attribute.String("source", source), + ) +} + +// RecordMergeGateRecordFailure counts failures to record a durable merge +// gate request for a completed apply. The backstop sweep retries the +// recording on its next pass, so a transient blip self-heals; a sustained rate +// means storage writes are failing and sibling PR checks are going stale — +// check the operator logs for the storage error. +func RecordMergeGateRecordFailure(ctx context.Context, database, environment string) { + addCounter(ctx, "schemabot.merge_gate.record_failures_total", + "Total failures to record a durable merge gate request for a completed apply", "{failure}", + attribute.String("database", database), + EnvironmentAttribute(environment), + ) +} diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 884d688d8..2e33f5eb9 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -1419,6 +1419,16 @@ type MergeGateRequest struct { UpdatedAt time.Time } +// ChangeKeyForPullRequest renders a GitHub pull request number as a merge +// gate change key. Zero (no originating PR) renders as the empty key, which +// the fan-out treats as "exclude nothing". +func ChangeKeyForPullRequest(pr int) string { + if pr <= 0 { + return "" + } + return strconv.Itoa(pr) +} + // WebhookEvent is a durable inbox row for one SCM/webhook delivery. type WebhookEvent struct { ID int64 diff --git a/pkg/webhook/merge_gate_integration_test.go b/pkg/webhook/merge_gate_integration_test.go new file mode 100644 index 000000000..facfd2e07 --- /dev/null +++ b/pkg/webhook/merge_gate_integration_test.go @@ -0,0 +1,142 @@ +//go:build integration + +// Merge gate guardrail integration tests. When an apply reaches terminal +// success on a (environment, database type, database) target, every other open +// PR with stored check state against that target planned against a schema that +// no longer exists. These tests exercise the durable merge gate request lifecycle +// against the real webhook harness, starting with recording at the operator +// drive tail. + +package webhook + +import ( + "context" + "database/sql" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "testing" + "time" + + gh "github.com/google/go-github/v86/github" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/state" + "github.com/block/schemabot/pkg/storage" +) + +// clearMergeGateRequests empties the shared merge_gate_requests table. +// The table is cross-test shared state: apply drive tails in earlier tests +// record requests the processor never drains there, and this test's drain +// would otherwise claim them before its own. +func clearMergeGateRequests(t *testing.T) { + t.Helper() + db, err := sql.Open("mysql", e2eSchemabotDSN) + require.NoError(t, err) + defer func() { _ = db.Close() }() + _, err = db.ExecContext(t.Context(), "DELETE FROM merge_gate_requests") + require.NoError(t, err) +} + +// 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. +func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { + clearMergeGateRequests(t) + dbName := "webhook_mergegate_drivetail" + svc := setupE2EService(t, dbName) + + mux := http.NewServeMux() + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + + client := gh.NewClient(nil) + client.BaseURL, _ = url.Parse(server.URL + "/") + + schemabotConfig := fmt.Sprintf("database: %s\ntype: mysql\n", dbName) + schemaFiles := map[string]string{ + "users.sql": "CREATE TABLE `users` (\n `id` bigint unsigned NOT NULL AUTO_INCREMENT,\n `name` varchar(255) NOT NULL,\n PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;", + } + result := setupFakeGitHubForPlan(t, mux, schemaFiles, schemabotConfig, dbName) + + h := newE2EHandler(t, svc, client) + + // 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. Installed after the handler so this probe is + // the active registration. + kicked := make(chan struct{}, 1) + svc.OnMergeGateRecorded = func() { + select { + case kicked <- struct{}{}: + default: + } + } + + req := buildWebhookRequest(t, webhookPayloadOpts{ + comment: "schemabot apply -e staging", + isPR: true, + }, nil) + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + assert.Contains(t, rr.Body.String(), "apply started") + + t.Cleanup(func() { + _ = svc.Storage().Locks().ForceRelease(context.WithoutCancel(t.Context()), dbName, "mysql") + }) + + select { + case body := <-result.comments: + assert.Contains(t, body, "## Schema Change Apply") + case <-time.After(webhookIntegrationPollDeadline): + t.Fatal("timed out waiting for apply plan comment") + } + + // The operator drives the apply to terminal success in the background. + var apply *storage.Apply + require.EventuallyWithT(t, func(collect *assert.CollectT) { + applies, err := svc.Storage().Applies().GetByPR(t.Context(), "octocat/hello-world", 1) + if !assert.NoError(collect, err) { + return + } + for _, a := range applies { + if a.Database == dbName && state.IsState(a.State, state.Apply.Completed) { + apply = a + return + } + } + 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 + // 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) + if !assert.NoError(collect, err) || !assert.NotNil(collect, req) { + return + } + gateReq = req + }, webhookIntegrationPollDeadline, 100*time.Millisecond) + + assert.Equal(t, apply.ApplyIdentifier, gateReq.ApplyIdentifier) + assert.Equal(t, "staging", gateReq.Environment) + assert.Equal(t, "mysql", gateReq.DatabaseType) + assert.Equal(t, dbName, gateReq.DatabaseName) + 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: + case <-time.After(webhookIntegrationPollDeadline): + t.Fatal("timed out waiting for the drive tail to invoke the merge gate recorded-notifier") + } +}