feat(api): record durable merge gate requests at apply drive tails

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

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

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
 }

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

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