feat(github): re-plan sibling PR checks when an apply changes a target schema

pkg/metrics/metrics.go

@@ -1547,3 +1547,58 @@ func RecordMergeGateRecordFailure(ctx context.Context, database, environment str
 		EnvironmentAttribute(environment),
 	)
 }
+
+// RecordMergeGatePROutcome counts per-PR outcomes of merge gate fan-out.
+// Outcomes:
+//   - "refreshed": the PR's stored check state was re-planned against the
+//     mutated target schema.
+//   - "blocked_replan_failed": the re-plan failed, so the stored check was
+//     flipped to blocking (fail closed) — check the server logs for the
+//     re-plan error and re-plan the PR once the cause is fixed.
+//   - "skipped_in_flight": the PR's check row is owned by an in-flight apply,
+//     which stays authoritative; the row was left untouched.
+//   - "skipped_pr_closed": the PR is closed, so its stored checks no longer
+//     gate anything.
+//   - "skipped_not_managed": the PR no longer manages schema for the target
+//     (config removed or database dropped from the PR's schema files).
+//   - "skipped_superseded": a racing write (a synchronize that re-planned a
+//     newer head, or an apply that claimed the row) landed first and is
+//     authoritative; the refresh yielded to it.
+func RecordMergeGatePROutcome(ctx context.Context, repository, database, environment, outcome string) {
+	addCounter(ctx, "schemabot.merge_gate.pr_refreshes_total",
+		"Total per-PR outcomes of merge gate fan-out after a target schema changed", "{refresh}",
+		attribute.String("repository", repository),
+		attribute.String("database", database),
+		EnvironmentAttribute(environment),
+		attribute.String("outcome", outcome),
+	)
+}
+
+// RecordMergeGateEventOutcome counts terminal outcomes of driving one
+// durable merge gate request. Outcomes:
+//   - "completed": the fan-out refreshed (or safely skipped) every sibling PR.
+//   - "failed_retrying": the fan-out failed and the request will be retried.
+//   - "failed_terminal": the fan-out failed on its final attempt — sibling PR
+//     stored checks for the target may remain stale until their PRs re-plan.
+//     Check the server logs for the failing PR and re-plan it.
+//   - "lease_lost": the drive lost its lease mid-fan-out; another driver
+//     re-drives the request (re-planning the same PRs again is safe).
+func RecordMergeGateEventOutcome(ctx context.Context, database, environment, outcome string) {
+	addCounter(ctx, "schemabot.merge_gate.events_total",
+		"Total terminal outcomes of driving durable merge gate requests", "{event}",
+		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
+// request means sibling PR stored checks for its target may remain stale —
+// find the request's target in the server logs and re-plan the affected PRs.
+func RecordMergeGateTerminatedStuck(ctx context.Context, terminated int64) {
+	addCounterN(ctx, terminated, "schemabot.merge_gate.terminated_stuck_total",
+		"Total merge gate requests terminated by the stuck-processing sweep", "{request}",
+	)
+}

pkg/serve/serve.go

@@ -90,6 +90,8 @@ type webhookRuntime struct {
 	stopDurableWebhookDispatch      func()
 	drainInProcessWebhookWork       func(context.Context)
 	reconcileMissingSummaryComments func(context.Context)
+	startMergeGateProcessor         func(context.Context)
+	stopMergeGateProcessor          func()
 }
 
 func (r webhookRuntime) StartMissingSummaryReconciliation(ctx context.Context, logger *slog.Logger) {
@@ -535,10 +537,11 @@ func (s *Server) MetricsHandler() http.Handler {
 }
 
 // Start launches the server's background work: the operator driver pool
-// (dispatches queued applies and recovers stale ones), the remote-deployment
-// health monitor, the webhook inbox monitor (emits durable-inbox depth/backlog
-// metrics), and the pending-drops cleaner — all of which run until ctx is
-// canceled or Close is called. It also kicks off a one-shot missing-summary
+// (dispatches queued applies and recovers stale ones), the merge gate
+// processor (re-plans sibling PR checks after a target's schema changes), the
+// remote-deployment health monitor, the webhook inbox monitor (emits
+// durable-inbox depth/backlog metrics), and the pending-drops cleaner — all of
+// which run until ctx is canceled or Close is called. It also kicks off a one-shot missing-summary
 // reconciliation that, once started, runs to completion independently of ctx (it
 // repairs interrupted terminal comments and must not be cut short by a request
 // context); it runs before the operator so recovered applies attach observers
@@ -548,14 +551,18 @@ func (s *Server) Start(ctx context.Context) {
 	if s.webhook.startDurableWebhookDispatch != nil {
 		s.webhook.startDurableWebhookDispatch(ctx)
 	}
+	if s.webhook.startMergeGateProcessor != nil {
+		s.webhook.startMergeGateProcessor(ctx)
+	}
 	s.svc.StartOperator(ctx)
 	s.svc.StartRemoteDeploymentHealthMonitor(ctx)
 	s.svc.StartWebhookInboxMonitor(ctx)
 	s.svc.StartPendingDropsCleaner(ctx)
 }
 
 // Close releases the resources the Server owns and returns all cleanup errors
-// encountered, joined together. It stops the pending-drops cleaner, stops the
+// encountered, joined together. It stops the pending-drops cleaner and the
+// merge gate processor, stops the
 // operator (before closing the gRPC client it built, see below), shuts down
 // telemetry, closes that gRPC fallback client, and closes the service. svc.Close
 // stops the health monitor and closes the service's clients and storage (the
@@ -564,6 +571,9 @@ func (s *Server) Start(ctx context.Context) {
 // after Start.
 func (s *Server) Close() error {
 	s.svc.StopPendingDropsCleaner()
+	if s.webhook.stopMergeGateProcessor != nil {
+		s.webhook.stopMergeGateProcessor()
+	}
 	if s.webhook.stopDurableWebhookDispatch != nil {
 		s.webhook.stopDurableWebhookDispatch()
 	}
@@ -775,6 +785,8 @@ func buildSingleAppWebhookRuntime(serverConfig *api.ServerConfig, svc *api.Servi
 		drainInProcessWebhookWork:       handler.DrainInProcessWebhookWork,
 		handler:                         handler,
 		reconcileMissingSummaryComments: handler.ReconcileMissingSummaryComments,
+		startMergeGateProcessor:         handler.StartMergeGateProcessor,
+		stopMergeGateProcessor:          handler.StopMergeGateProcessor,
 	}, nil
 }
 
@@ -853,6 +865,8 @@ func buildMultiAppWebhookRuntime(serverConfig *api.ServerConfig, svc *api.Servic
 		drainInProcessWebhookWork:       handler.DrainInProcessWebhookWork,
 		handler:                         handler,
 		reconcileMissingSummaryComments: handler.ReconcileMissingSummaryComments,
+		startMergeGateProcessor:         handler.StartMergeGateProcessor,
+		stopMergeGateProcessor:          handler.StopMergeGateProcessor,
 	}, nil
 }
 

pkg/webhook/check_records.go

@@ -2,6 +2,7 @@ package webhook
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/block/schemabot/pkg/apitypes"
@@ -58,20 +59,26 @@ func (o reviewDriftOutcome) planDriftState() storage.PlanDriftState {
 	}
 }
 
+// errPlanCheckHeadStale reports that a plan check record was not stored
+// because the PR head advanced past the planned head. The newer head's own
+// plan is authoritative, so callers racing a synchronize (for example the
+// merge gate fan-out) treat this as a benign skip rather than a failure.
+var errPlanCheckHeadStale = errors.New("plan head is no longer the PR's current head")
+
 // storePlanCheckRecord stores per-database check state after a plan is generated.
 // The state is used internally by the aggregate check to compute its overall status.
 // No per-database GitHub Check Run is created — only the aggregate is visible on the PR.
 // Returns the commit SHA used for the plan. Failures are non-fatal.
 func (h *Handler) storePlanCheckRecord(ctx context.Context, client *ghclient.InstallationClient, repo string, pr int, schema *ghclient.SchemaRequestResult, planResp *apitypes.PlanResponse, environment string, drift reviewDriftOutcome) (string, error) {
-	headSHA, _, err := h.upsertPlanCheckRecord(ctx, client, repo, pr, schema, planResp, environment, drift)
+	headSHA, _, err := h.upsertPlanCheckRecord(ctx, client, repo, pr, schema, planResp, environment, drift, "")
 	return headSHA, err
 }
 
 // storeManualPlanCheckRecord stores per-database check state after a manual
 // plan and then reconciles same-head apply-owned stored check state when the manual
 // plan proves the target already matches the PR schema.
 func (h *Handler) storeManualPlanCheckRecord(ctx context.Context, client *ghclient.InstallationClient, repo string, pr int, schema *ghclient.SchemaRequestResult, planResp *apitypes.PlanResponse, environment string, drift reviewDriftOutcome) (string, bool, error) {
-	headSHA, check, err := h.upsertPlanCheckRecord(ctx, client, repo, pr, schema, planResp, environment, drift)
+	headSHA, check, err := h.upsertPlanCheckRecord(ctx, client, repo, pr, schema, planResp, environment, drift, "")
 	if err != nil {
 		return headSHA, false, err
 	}
@@ -134,7 +141,13 @@ func planCheckConclusion(hasChanges, hasPlanErrors, driftBlocked bool) string {
 	}
 }
 
-func (h *Handler) upsertPlanCheckRecord(ctx context.Context, client *ghclient.InstallationClient, repo string, pr int, schema *ghclient.SchemaRequestResult, planResp *apitypes.PlanResponse, environment string, drift reviewDriftOutcome) (string, *storage.Check, error) {
+// upsertPlanCheckRecord verifies the planned head is still the PR's current
+// head (returning errPlanCheckHeadStale when it is not) and upserts the
+// per-database stored check state for the plan. refreshNote, when non-empty,
+// is a pre-sanitized attribution line the merge gate fan-out appends to the
+// stored change summary so the Change column says why an unchanged PR was
+// re-planned; ordinary plan writes pass "".
+func (h *Handler) upsertPlanCheckRecord(ctx context.Context, client *ghclient.InstallationClient, repo string, pr int, schema *ghclient.SchemaRequestResult, planResp *apitypes.PlanResponse, environment string, drift reviewDriftOutcome, refreshNote string) (string, *storage.Check, error) {
 	headSHA := schema.HeadSHA
 	if headSHA == "" {
 		metrics.RecordStatusCheckOperation(ctx, metrics.StatusCheckOperation{
@@ -170,8 +183,8 @@ func (h *Handler) upsertPlanCheckRecord(ctx context.Context, client *ghclient.In
 			Environment:  environment,
 			Status:       "stale",
 		})
-		return headSHA, nil, fmt.Errorf("skip stale plan check record for repo %s pr %d environment %s database_type %s database %s: plan head SHA %s no longer matches current head SHA for PR %s",
-			repo, pr, environment, schema.Type, schema.Database, headSHA, prInfo.HeadSHA)
+		return headSHA, nil, fmt.Errorf("skip stale plan check record for repo %s pr %d environment %s database_type %s database %s: plan head SHA %s no longer matches current head SHA %s: %w",
+			repo, pr, environment, schema.Type, schema.Database, headSHA, prInfo.HeadSHA, errPlanCheckHeadStale)
 	}
 
 	hasChanges := planResp.HasChanges()
@@ -184,8 +197,14 @@ func (h *Handler) upsertPlanCheckRecord(ctx context.Context, client *ghclient.In
 	// the block rides on BlockingReason + Conclusion so a stored drift block is
 	// legible and durable across write paths.
 	changeSummary := summarizePlanChanges(schema, planResp, environment)
+	if refreshNote != "" {
+		changeSummary = appendRefreshNote(changeSummary, refreshNote)
+	}
 	blockingReason := ""
 	if driftBlocked {
+		// The drift summary owns the Change column when drift blocks: it names
+		// the deployment divergence the operator must resolve, which supersedes
+		// refresh attribution.
 		changeSummary = drift.summary
 		blockingReason = reviewTimeDeploymentDriftBlock.blockingReason
 	}

pkg/webhook/check_runs.go

@@ -108,6 +108,17 @@ var reviewTimeDeploymentDriftBlock = checkBlockReason{
 	message:        "One or more deployments differ from the reviewed plan, or could not be confirmed to match it; reconcile the deployment drift or replan once the deployments match before this check can pass.",
 }
 
+// schemaChangedReplanFailedBlock is used when the live schema of a check's
+// target changed after its plan was computed (another apply reached terminal
+// success there) and the merge gate fan-out could not re-plan the PR. The
+// stored check fails closed: a plan computed against a schema that no longer
+// exists must not keep passing. The raw re-plan error stays in the server
+// logs; only this fixed message is rendered on the PR.
+var schemaChangedReplanFailedBlock = checkBlockReason{
+	blockingReason: "schema_changed_replan_failed",
+	message:        "The live schema for this database changed after this plan was computed, and SchemaBot could not re-plan the PR against it. Re-run `schemabot plan` (or push a new commit) before this check can pass; see server logs for the re-plan failure.",
+}
+
 // noAllowedConfiguredEnvironmentsBlock is used when schema files changed but
 // the server-configured environments for the database do not overlap this
 // service's allowed_environments. SchemaBot cannot safely plan the schema

pkg/webhook/handler.go

@@ -173,6 +173,20 @@ type Handler struct {
 	webhookReconcileGrace    time.Duration
 	webhookReconcileMaxPages int
 
+	// Merge gate processor lifecycle (see merge_gate.go). The intervals
+	// have package defaults set at construction; tests override them directly.
+	mergeGatePollInterval  time.Duration
+	mergeGateLeaseDuration time.Duration
+	mergeGateSweepLookback time.Duration
+	mergeGateMu            sync.Mutex
+	mergeGateStop          chan struct{}
+	mergeGateCancel        context.CancelFunc
+	mergeGateWg            sync.WaitGroup
+	// mergeGateKick wakes the driver to run a pass now instead of waiting
+	// for the next poll tick; buffered so one pending kick coalesces any
+	// number of concurrent recordings.
+	mergeGateKick chan struct{}
+
 	logger                     *slog.Logger
 	priorEnvCheckMaxAttempts   int
 	priorEnvCheckRetryInterval time.Duration
@@ -258,6 +272,10 @@ func NewHandlerWithDispatch(service *api.Service, ghClients github.ClientSet, we
 		webhookReconcileLookback:    defaultWebhookReconcileLookback,
 		webhookReconcileGrace:       defaultWebhookReconcileGrace,
 		webhookReconcileMaxPages:    defaultWebhookReconcileMaxPages,
+		mergeGatePollInterval:       defaultMergeGatePollInterval,
+		mergeGateLeaseDuration:      defaultMergeGateLeaseDuration,
+		mergeGateSweepLookback:      defaultMergeGateSweepLookback,
+		mergeGateKick:               make(chan struct{}, 1),
 		priorEnvCheckMaxAttempts:    defaultPriorEnvCheckMaxAttempts,
 		priorEnvCheckRetryInterval:  defaultPriorEnvCheckRetryInterval,
 	}
@@ -344,6 +362,12 @@ func NewHandlerWithDispatch(service *api.Service, ghClients github.ClientSet, we
 			obs.OnTerminal(apply, tasks)
 			return nil
 		}
+
+		// Wake the merge gate processor as soon as a drive tail records a
+		// request, so sibling PR checks re-plan without waiting for the next
+		// poll tick. The durable request row stays the source of truth: a
+		// kick lost to a pod boundary only costs poll latency.
+		service.OnMergeGateRecorded = h.KickMergeGate
 	}
 
 	return h