diff --git a/pkg/github/client.go b/pkg/github/client.go index 1f8b913f1..096f9998d 100644 --- a/pkg/github/client.go +++ b/pkg/github/client.go @@ -637,6 +637,37 @@ func (ic *InstallationClient) CreateIssueComment(ctx context.Context, repo strin return comment.GetID(), comment.GetNodeID(), nil } +// HasIssueCommentWithMarker reports whether any of the PR's most recent +// comments contains the given marker string. Callers use it to make +// at-least-once comment posting idempotent: a hidden HTML marker in the body +// identifies the comment, and a retry that finds the marker skips the +// re-post. Only the newest page of comments is inspected — the markers +// callers search for are posted moments before the search, so an older +// occurrence beyond the page means a duplicate is possible but harmless. +func (ic *InstallationClient) HasIssueCommentWithMarker(ctx context.Context, repo string, pr int, marker string) (bool, error) { + owner, repoName := splitRepo(repo) + comments, err := retryGitHubUnavailableRead(ctx, ic.logger, "list issue comments", []any{"repo", repo, "pr", pr}, func(ctx context.Context) ([]*gh.IssueComment, error) { + list, _, callErr := ic.client.Issues.ListComments(ctx, owner, repoName, pr, &gh.IssueListCommentsOptions{ + Sort: new("created"), + Direction: new("desc"), + ListOptions: gh.ListOptions{PerPage: 100}, + }) + if callErr != nil { + return nil, classifyGitHubAPIError(callErr) + } + return list, nil + }) + if err != nil { + return false, fmt.Errorf("list issue comments for %s#%d: %w", repo, pr, err) + } + for _, comment := range comments { + if strings.Contains(comment.GetBody(), marker) { + return true, nil + } + } + return false, nil +} + // GetIssueComment returns the current body of an existing PR/issue comment. func (ic *InstallationClient) GetIssueComment(ctx context.Context, repo string, commentID int64) (string, error) { owner, repoName := splitRepo(repo) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index d59c7a9a4..233d77e7c 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -1520,6 +1520,10 @@ const ( // MergeGateSourceSweep marks a request recorded by the backstop sweep // over recently completed applies. MergeGateSourceSweep = "sweep" + // MergeGateSourceReleaseSweep marks a settle request backfilled by the + // sweep over terminal applies whose preflight held sibling checks but + // whose settle was never recorded. + MergeGateSourceReleaseSweep = "release_sweep" ) // RecordMergeGateRecorded counts durable merge gate requests recorded @@ -1602,3 +1606,14 @@ func RecordMergeGateTerminatedStuck(ctx context.Context, terminated int64) { "Total merge gate requests terminated by the stuck-processing sweep", "{request}", ) } + +// RecordMergeGatePreflightRearmed counts terminally failed preflight renders +// re-armed because their apply is still active. A sustained rate means the +// code host keeps rejecting the hold rendering (outage, auth failure) while +// applies run on stored holds — sibling PRs' visible checks stay stale until +// a render lands, so find the failing render in the merge gate logs. +func RecordMergeGatePreflightRearmed(ctx context.Context, reopened int64) { + addCounterN(ctx, reopened, "schemabot.merge_gate.preflight_renders_rearmed_total", + "Total terminally failed preflight renders re-armed for still-active applies", "{request}", + ) +} diff --git a/pkg/webhook/check_runs.go b/pkg/webhook/check_runs.go index a72f08136..469e935fc 100644 --- a/pkg/webhook/check_runs.go +++ b/pkg/webhook/check_runs.go @@ -119,6 +119,17 @@ var schemaChangedReplanFailedBlock = checkBlockReason{ 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.", } +// applyInFlightBlock is used while an apply is running against a check's +// target: the stored verdict was computed against the pre-apply live schema, +// so a merge must not land on it before the apply finishes. The preflight +// fan-out writes it before the apply's engine work starts, and the apply's +// settle fan-out re-plans the check against the resulting schema, which +// replaces this hold with a live verdict. +var applyInFlightBlock = checkBlockReason{ + blockingReason: "apply_in_flight_on_target", + message: "An apply is currently changing this database's live schema, so this check is held until it finishes. SchemaBot then re-plans this PR against the resulting schema and refreshes this check automatically.", +} + // 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 diff --git a/pkg/webhook/handler.go b/pkg/webhook/handler.go index 652459311..272234910 100644 --- a/pkg/webhook/handler.go +++ b/pkg/webhook/handler.go @@ -363,11 +363,6 @@ func NewHandlerWithDispatch(service *api.Service, ghClients github.ClientSet, we 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 diff --git a/pkg/webhook/merge_gate.go b/pkg/webhook/merge_gate.go index 4917dcd0e..f806c30e1 100644 --- a/pkg/webhook/merge_gate.go +++ b/pkg/webhook/merge_gate.go @@ -1,14 +1,17 @@ -// merge_gate.go drives the merge gate guardrail: when an apply reaches -// terminal success on a (environment, database type, database) target, every -// other open PR with stored plan check state against that target planned -// against a schema that no longer exists. The operator drive tail (and a -// backstop sweep here) records a durable merge_gate_requests row, and this -// processor consumes it: it finds the sibling PRs' stored check state through -// the checks reverse index, re-plans each PR against the live schema, and — -// when a re-plan fails — fails the stored check closed so a stale plan can -// never keep passing. CLI/gRPC applies carry no PR surface at all, so this -// processor is the only path that keeps their targets' sibling PR checks -// honest. +// merge_gate.go drives the merge gate guardrail around applies on a +// (environment, database type, database) target. Before an apply's engine +// work starts, the operator gate records a durable preflight request; this +// processor consumes it by holding every sibling PR's stored check on the +// target action-required (with a PR comment explaining the hold), so a merge +// cannot land on a verdict the apply is about to invalidate — the gate blocks +// the apply's start until the holds are confirmed. Once the apply settles +// terminally, the drive tail (and backstop sweeps here) records a settle +// request; this processor consumes it by re-planning each sibling PR against +// the live schema, which refreshes stale verdicts and releases the holds. A +// re-plan that fails leaves the stored check failed closed so a stale plan +// can never keep passing. CLI/gRPC applies carry no PR surface at all, so +// this processor is the only path that keeps their targets' sibling PR +// checks honest. package webhook import ( @@ -25,6 +28,7 @@ import ( "github.com/block/schemabot/pkg/metrics" "github.com/block/schemabot/pkg/storage" "github.com/block/schemabot/pkg/webhook/action" + "github.com/block/schemabot/pkg/webhook/templates" ) const ( @@ -60,17 +64,30 @@ const ( mergeGateOutcomeSkippedPRClosed = "skipped_pr_closed" mergeGateOutcomeSkippedNotManaged = "skipped_not_managed" mergeGateOutcomeSkippedSuperseded = "skipped_superseded" + mergeGateOutcomeHeld = "held" + mergeGateOutcomeHoldSuperseded = "hold_superseded" ) // StartMergeGateProcessor starts the background driver that consumes -// durable merge gate requests. Idempotent; StopMergeGateProcessor stops -// it and waits for the in-flight pass to finish. +// durable merge gate requests, and registers this handler as the +// service's merge gate consumer. Registration lives here rather than at +// handler construction because the consumer callback is what tells the +// operator a processor exists: the drive tail records settles and the apply +// gate records (and waits on) preflights only when something will drain +// them. Idempotent; StopMergeGateProcessor stops the driver and waits for +// the in-flight pass to finish. func (h *Handler) StartMergeGateProcessor(ctx context.Context) { if h.mergeGateStore() == nil { h.logger.Warn("merge gate processor not started: storage is unavailable; sibling PR checks will go stale after applies until it recovers") return } + // Wake the driver as soon as a drive tail or apply gate records a + // request, so it is consumed 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. + h.service.OnMergeGateRecorded = h.KickMergeGate + h.mergeGateMu.Lock() if h.mergeGateStop != nil { h.mergeGateMu.Unlock() @@ -160,10 +177,34 @@ func (h *Handler) mergeGateDriver(ctx context.Context, stop <-chan struct{}) { // claim and drive requests until none remain claimable. func (h *Handler) runMergeGatePass(ctx context.Context, owner string) { h.sweepMergeGateRequests(ctx) + h.sweepPreflightedAppliesMissingSettle(ctx) h.terminateStuckMergeGateRequests(ctx) + h.rearmPreflightRendersForActiveApplies(ctx) h.drainMergeGateRequests(ctx, owner) } +// rearmPreflightRendersForActiveApplies re-arms terminally failed preflight +// requests whose apply is still active. The operator gate starts applies on +// stored holds, so a code-host rendering that exhausted its retries (for +// example through a code-host outage) has nothing else retrying it until the +// apply settles — and once the code host recovers, sibling PRs' visible +// Check Runs would sit stale-green for the rest of the apply. Re-arming +// keeps the render retrying while the apply runs; the settle re-plan +// supersedes it after that. +func (h *Handler) rearmPreflightRendersForActiveApplies(ctx context.Context) { + reopened, err := h.mergeGateStore().ReopenTerminalPreflightsForActiveApplies(ctx) + if err != nil { + h.logger.Error("merge gate re-arm sweep failed; terminally failed preflight renders stay unretried until the next pass", "error", err) + return + } + if reopened == 0 { + return + } + h.logger.Warn("merge gate re-armed terminally failed preflight renders for still-active applies; their sibling PRs' visible checks stay unrendered until a render succeeds", + "reopened", reopened) + metrics.RecordMergeGatePreflightRearmed(ctx, reopened) +} + // sweepMergeGateRequests backfills merge gate requests for completed applies // that have none — the applies table is the outbox, so a pod crash between an // apply's terminal write and its drive-tail recording cannot lose the fan-out. @@ -189,23 +230,67 @@ func (h *Handler) sweepMergeGateRequests(ctx context.Context) { if err != nil { // Each apply's backfill is independent; a failed one is retried on // the next sweep pass. - h.logger.Error("merge gate sweep failed to backfill a merge gate request for a completed apply", + h.logger.Error("merge gate sweep failed to backfill a settle request for a completed apply", append(apply.LogAttrs(), "error", err)...) metrics.RecordMergeGateRecordFailure(ctx, apply.Database, apply.Environment) continue } if !recorded { // A drive tail recorded it between the sweep query and this insert. - h.logger.Debug("merge gate sweep found the merge gate request already recorded", + h.logger.Debug("merge gate sweep found the settle request already recorded", apply.LogAttrs()...) continue } - h.logger.Info("merge gate sweep backfilled a merge gate request the drive tail did not record", + h.logger.Info("merge gate sweep backfilled a settle request the drive tail did not record", apply.LogAttrs()...) metrics.RecordMergeGateRecorded(ctx, apply.Database, apply.Environment, metrics.MergeGateSourceSweep) } } +// sweepPreflightedAppliesMissingSettle backfills settle requests for applies +// that settled terminally — in any state — after a preflight held sibling PR +// checks, but whose settle was never recorded. The completed-applies sweep +// only covers terminal success; this one exists because a preflight hold must +// always be released, even when the apply failed, was stopped, or was +// cancelled before changing the schema. Without it, an apply cancelled while +// queued (whose drive tail never ran) would leave sibling checks +// action-required forever. +func (h *Handler) sweepPreflightedAppliesMissingSettle(ctx context.Context) { + store := h.mergeGateStore() + applies, err := store.FindTerminalAppliesWithPreflightMissingSettle(ctx, h.mergeGateSweepLookback) + if err != nil { + h.logger.Error("merge gate release sweep failed to find preflighted terminal applies missing a settle; held sibling PR checks stay blocked until the next pass", "error", err) + return + } + 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, + DatabaseName: apply.Database, + Repository: apply.Repository, + ChangeKey: storage.ChangeKeyForPullRequest(apply.PullRequest), + RequestedBy: apply.Caller, + }) + if err != nil { + h.logger.Error("merge gate release sweep failed to backfill a settle request for a preflighted terminal apply; its held sibling PR checks stay blocked until the next pass", + append(apply.LogAttrs(), "error", err)...) + metrics.RecordMergeGateRecordFailure(ctx, apply.Database, apply.Environment) + continue + } + if !recorded { + h.logger.Debug("merge gate release sweep found the settle request already recorded", + apply.LogAttrs()...) + continue + } + h.logger.Info("merge gate release sweep backfilled a settle request for a preflighted terminal apply; its held sibling PR checks will be re-planned", + apply.LogAttrs()...) + metrics.RecordMergeGateRecorded(ctx, apply.Database, apply.Environment, metrics.MergeGateSourceReleaseSweep) + } +} + // terminateStuckMergeGateRequests terminalizes requests wedged in // processing past the attempt cap with an expired lease, so a poison request // cannot be reclaimed forever. Each terminated request means sibling PR @@ -255,6 +340,7 @@ func (h *Handler) driveNextMergeGate(ctx context.Context, owner string) (claimed h.logger.Info("merge gate driver claimed a request", "lease_owner", owner, + "kind", req.Kind, "apply_id", req.ApplyIdentifier, "environment", req.Environment, "database_type", req.DatabaseType, @@ -272,8 +358,8 @@ func (h *Handler) driveNextMergeGate(ctx context.Context, owner string) (claimed // a freshly claimed request, coalescing pending sibling requests for the same // target once the fan-out succeeds. func (h *Handler) driveClaimedMergeGate(ctx context.Context, store storage.MergeGateRequestStore, req *storage.MergeGateRequest) { - // Capture the pending siblings before the fan-out starts: the fan-out - // re-plans against the live schema, so it covers every schema change + // Capture the pending same-kind siblings before the fan-out starts: the + // fan-out acts on the target's current state, so it covers every request // 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.Kind, req.ID) @@ -402,33 +488,76 @@ func (h *Handler) safeFanOutMergeGate(ctx context.Context, req *storage.MergeGat return h.fanOutMergeGate(ctx, req) } -// fanOutMergeGate re-plans every sibling PR whose stored check state -// targets the request's (environment, database type, database). A returned -// error means at least one PR was neither refreshed nor safely failed closed, -// so the request must be retried; re-planning already-refreshed PRs on that -// retry is safe. +// fanOutMergeGate dispatches a claimed request to the fan-out its kind +// selects: a preflight holds sibling PR checks before the apply starts, a +// settle re-plans them after it finishes. func (h *Handler) fanOutMergeGate(ctx context.Context, req *storage.MergeGateRequest) error { - // Aggregate rows never match: their database type and name are the - // aggregate sentinel, not a real target. + switch req.Kind { + case storage.MergeGateKindPreflight: + return h.fanOutCheckPreflight(ctx, req) + case storage.MergeGateKindSettle: + return h.fanOutCheckSettle(ctx, req) + default: + // Fail the request rather than guess: an unknown kind means a newer + // (or corrupted) writer recorded something this build cannot drive. + return fmt.Errorf("unknown merge gate request kind %q for apply %s", req.Kind, req.ApplyIdentifier) + } +} + +// siblingChecksForTarget lists the stored check state the request fans out +// to: every check on the request's (environment, database type, database) +// except the originating PR's own rows — its own apply lifecycle keeps those +// current. Aggregate rows never match the target query: their database type +// and name are the aggregate sentinel, not a real target. +func (h *Handler) siblingChecksForTarget(ctx context.Context, req *storage.MergeGateRequest) ([]*storage.Check, error) { checks, err := h.service.Storage().Checks().GetByTarget(ctx, req.Environment, req.DatabaseType, req.DatabaseName) if err != nil { - return fmt.Errorf("list stored check state for target %s/%s in %s: %w", + return nil, fmt.Errorf("list stored check state for target %s/%s in %s: %w", req.DatabaseType, req.DatabaseName, req.Environment, err) } - var targets []*storage.Check for _, check := range checks { if isOriginatingChange(check, req) { - // The originating PR's own apply lifecycle already updated its - // stored check state; re-planning it here would be redundant. h.logger.Debug("merge gate skipping the originating PR", - "apply_id", req.ApplyIdentifier, "repo", check.Repository, "pr", check.PullRequest, + "apply_id", req.ApplyIdentifier, "kind", req.Kind, + "repo", check.Repository, "pr", check.PullRequest, "environment", req.Environment, "database_type", req.DatabaseType, "database", req.DatabaseName) continue } targets = append(targets, check) } + return targets, nil +} + +// fanOutCheckSettle re-plans every sibling PR whose stored check state +// targets the request's (environment, database type, database). A returned +// error means at least one PR was neither refreshed nor safely failed closed, +// so the request must be retried; re-planning already-refreshed PRs on that +// retry is safe. +func (h *Handler) fanOutCheckSettle(ctx context.Context, req *storage.MergeGateRequest) error { + // While a later preflighted apply on the same target is still active, its + // holds must survive: re-planning now would overwrite them with verdicts + // computed against a schema that apply is about to change. Defer to that + // apply's own settle — every preflighted apply records one when it settles + // (drive tail or release sweep) — which re-plans the target's siblings + // against the final schema. + activeHold, err := h.mergeGateStore().HasActivePreflightedApplyOnTarget(ctx, req.Environment, req.DatabaseType, req.DatabaseName) + if err != nil { + return fmt.Errorf("check for active preflighted applies on target %s/%s in %s: %w", + req.DatabaseType, req.DatabaseName, req.Environment, err) + } + if activeHold { + h.logger.Info("merge gate settle deferred: a preflighted apply on the target is still active, so sibling holds stay in place until its own settle re-plans them", + "apply_id", req.ApplyIdentifier, "environment", req.Environment, + "database_type", req.DatabaseType, "database", req.DatabaseName) + return nil + } + + targets, err := h.siblingChecksForTarget(ctx, req) + if err != nil { + return err + } if len(targets) == 0 { h.logger.Info("merge gate found no sibling PR check state for the target", "apply_id", req.ApplyIdentifier, "environment", req.Environment, @@ -460,6 +589,282 @@ func (h *Handler) fanOutMergeGate(ctx context.Context, req *storage.MergeGateReq return errors.Join(errs...) } +// prTargetKey identifies one PR within a fan-out. +type prTargetKey struct { + repo string + pr int +} + +// fanOutCheckPreflight holds every sibling PR's stored check on the target +// action-required before the originating apply's engine work starts, in two +// phases. The hold phase flips the stored checks and stamps +// holds_recorded_at — storage-only writes, which is what the operator gate +// waits on: the code host being unreachable must never block an apply on the +// rendering of its own holds. The render phase then surfaces the holds on +// the code host (the aggregate Check Run recompute and one hold comment per +// PR); a render failure keeps the request retryable without re-blocking the +// apply, and while the code host is fully down its merge surface is down +// with it. Every step is idempotent (conditional flips, a set-once stamp, an +// aggregate recompute, and a marker-deduplicated comment), so retries +// converge without duplicate comments. +func (h *Handler) fanOutCheckPreflight(ctx context.Context, req *storage.MergeGateRequest) error { + targets, err := h.siblingChecksForTarget(ctx, req) + if err != nil { + return err + } + + byPR := make(map[prTargetKey][]*storage.Check) + for _, check := range targets { + key := prTargetKey{repo: check.Repository, pr: check.PullRequest} + byPR[key] = append(byPR[key], check) + } + + if req.HoldsRecordedAt == nil { + if err := h.recordPreflightHolds(ctx, req, byPR); err != nil { + return err + } + } else { + h.logger.Debug("check preflight stored holds already recorded; resuming the code-host rendering", + "apply_id", req.ApplyIdentifier, "environment", req.Environment, + "database_type", req.DatabaseType, "database", req.DatabaseName) + } + + if len(byPR) == 0 { + h.logger.Info("check preflight found no sibling PR check state to hold for the target", + "apply_id", req.ApplyIdentifier, "environment", req.Environment, + "database_type", req.DatabaseType, "database", req.DatabaseName) + return nil + } + + sem := make(chan struct{}, mergeGatePRConcurrency) + var wg sync.WaitGroup + var mu sync.Mutex + var errs []error + for key, checks := range byPR { + wg.Go(func() { + sem <- struct{}{} + defer func() { <-sem }() + if err := h.renderCheckHoldOnPR(ctx, req, key.repo, key.pr, checks); err != nil { + mu.Lock() + errs = append(errs, err) + mu.Unlock() + } + }) + } + wg.Wait() + return errors.Join(errs...) +} + +// recordPreflightHolds is the storage-only hold phase: it flips every +// actionable sibling stored check on the target action-required, then stamps +// holds_recorded_at on the request. No code-host call happens here, so the +// phase — and the operator gate waiting on the stamp — succeeds or fails on +// storage alone. +func (h *Handler) recordPreflightHolds(ctx context.Context, req *storage.MergeGateRequest, byPR map[prTargetKey][]*storage.Check) error { + h.logger.Info("merge gate preflight holding sibling PR stored checks before the apply starts", + "apply_id", req.ApplyIdentifier, "environment", req.Environment, + "database_type", req.DatabaseType, "database", req.DatabaseName, + "requested_by", req.RequestedBy, "sibling_prs", len(byPR)) + + for key, checks := range byPR { + if err := h.holdStoredPRChecks(ctx, req, key.repo, key.pr, checks); err != nil { + return err + } + } + if err := h.mergeGateStore().MarkPreflightHoldsRecorded(ctx, req.ID, req.LeaseToken); err != nil { + return fmt.Errorf("record preflight holds for apply %s (target %s/%s in %s): %w", + req.ApplyIdentifier, req.DatabaseType, req.DatabaseName, req.Environment, err) + } + now := time.Now() + req.HoldsRecordedAt = &now + h.logger.Info("check preflight stored holds recorded; the apply may start while the code-host rendering completes", + "apply_id", req.ApplyIdentifier, "environment", req.Environment, + "database_type", req.DatabaseType, "database", req.DatabaseName, + "sibling_prs", len(byPR)) + return nil +} + +// preflightActionableChecks filters one PR's checks to the rows a preflight +// acts on. In-flight apply-owned rows are excluded — they already block the +// aggregate and their apply's lifecycle stays authoritative — and logged +// only when the exclusion decides the hold (logSkips true, the hold phase), +// not again during the render. +func (h *Handler) preflightActionableChecks(ctx context.Context, req *storage.MergeGateRequest, repo string, pr int, checks []*storage.Check, logSkips bool) []*storage.Check { + actionable := make([]*storage.Check, 0, len(checks)) + for _, check := range checks { + if check.Status == checkStatusInProgress { + if logSkips { + h.logger.Info("check preflight leaving in-flight apply-owned check state untouched; it already blocks the PR", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName, "check_apply_id", check.ApplyID, + "check_head_sha", check.HeadSHA) + metrics.RecordMergeGatePROutcome(ctx, repo, req.DatabaseName, req.Environment, mergeGateOutcomeSkippedInFlight) + } + continue + } + actionable = append(actionable, check) + } + return actionable +} + +// holdStoredPRChecks flips one sibling PR's stored checks on the target +// action-required. Storage-only — the code-host rendering happens separately +// in renderCheckHoldOnPR. A flip refused by the head-SHA condition means a +// racing synchronize re-planned a newer head against the pre-apply schema; +// the hold yields to it (logged and counted — the settle re-plans that head +// when the apply finishes). +func (h *Handler) holdStoredPRChecks(ctx context.Context, req *storage.MergeGateRequest, repo string, pr int, checks []*storage.Check) error { + actionable := h.preflightActionableChecks(ctx, req, repo, pr, checks, true) + if len(actionable) == 0 { + return nil + } + + held := 0 + for _, check := range actionable { + hold := *check + hold.Status = checkStatusCompleted + hold.Conclusion = checkConclusionActionRequired + hold.BlockingReason = applyInFlightBlock.blockingReason + hold.ErrorMessage = applyInFlightBlock.message + hold.ChangeSummary = clampDriftSummary(fmt.Sprintf("held: apply %s by %s is changing %s in %s", + req.ApplyIdentifier, req.RequestedBy, req.DatabaseName, req.Environment)) + flipped, err := h.service.Storage().Checks().MarkBlockedForApplyInFlight(ctx, &hold) + if err != nil { + return fmt.Errorf("hold stored check for %s#%d (target %s/%s in %s, apply %s): %w", + repo, pr, req.DatabaseType, req.DatabaseName, req.Environment, req.ApplyIdentifier, err) + } + if !flipped { + // Already held (an idempotent retry or an overlapping apply's + // preflight), or superseded by a racing write on a newer head. The + // distinction does not change this fan-out's behavior — the row is + // either blocking already or owned by a newer verdict the settle + // will re-plan — but it matters for triage, so log and count it. + h.logger.Warn("check preflight did not flip a stored check; it is already held or a racing write on a newer head superseded the hold", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName, "check_head_sha", check.HeadSHA, + "check_blocking_reason", check.BlockingReason) + metrics.RecordMergeGatePROutcome(ctx, repo, req.DatabaseName, req.Environment, mergeGateOutcomeHoldSuperseded) + continue + } + held++ + metrics.RecordMergeGatePROutcome(ctx, repo, req.DatabaseName, req.Environment, mergeGateOutcomeHeld) + } + + h.logger.Info("merge gate preflight held sibling PR stored checks for the target", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName, "requested_by", req.RequestedBy, + "held", held, "rows", len(actionable)) + return nil +} + +// renderCheckHoldOnPR surfaces one sibling PR's already-durable hold on the +// code host: it recomputes the visible aggregate Check Run from the stored +// rows so the hold blocks the merge button, and posts the +// marker-deduplicated comment explaining the hold. A failure here keeps the +// request retryable but never re-blocks the apply — the stored holds are in +// place, and while the code host is fully down its merge surface is down +// with it; the re-arm sweep keeps the render retrying for as long as the +// apply runs. +func (h *Handler) renderCheckHoldOnPR(ctx context.Context, req *storage.MergeGateRequest, repo string, pr int, checks []*storage.Check) error { + actionable := h.preflightActionableChecks(ctx, req, repo, pr, checks, false) + if len(actionable) == 0 { + h.logger.Debug("check preflight has nothing to render for the PR; only in-flight apply-owned rows, which already block it", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName) + return nil + } + + installationID, err := h.resolveRepoWebhookInstallation(ctx, repo) + if err != nil { + return fmt.Errorf("resolve installation for check preflight of %s#%d (target %s/%s in %s, apply %s): %w", + repo, pr, req.DatabaseType, req.DatabaseName, req.Environment, req.ApplyIdentifier, err) + } + prCtx, cancel, client, err := h.commandBootstrap(repo, installationID) + defer cancel() + if err != nil { + return fmt.Errorf("bootstrap check preflight of %s#%d (target %s/%s in %s, apply %s): %w", + repo, pr, req.DatabaseType, req.DatabaseName, req.Environment, req.ApplyIdentifier, err) + } + + // A GitHub failure here is uncertainty, not staleness: keep the request + // retryable rather than guessing at the PR's state. + prInfo, err := client.FetchPullRequestNoCache(prCtx, repo, pr) + if err != nil { + return fmt.Errorf("verify PR state for check preflight of %s#%d (target %s/%s in %s, apply %s): %w", + repo, pr, req.DatabaseType, req.DatabaseName, req.Environment, req.ApplyIdentifier, err) + } + if prInfo.IsClosed() { + h.logger.Info("check preflight skipping closed PR; its held stored checks no longer gate a merge, so the hold needs no rendering", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName, "merged", prInfo.Merged) + metrics.RecordMergeGatePROutcome(prCtx, repo, req.DatabaseName, req.Environment, mergeGateOutcomeSkippedPRClosed) + return nil + } + + // Recompute the PR's visible aggregate Check Run from the stored rows so + // the hold blocks the merge button, not just the database record. + // Idempotent, so retries converge the Check Run. + h.updateAggregateCheck(prCtx, client, repo, pr, actionable[0].HeadSHA) + + if err := h.ensureCheckHoldComment(prCtx, client, req, repo, pr); err != nil { + return err + } + + h.logger.Info("merge gate preflight rendered the hold on the sibling PR", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName, "requested_by", req.RequestedBy) + return nil +} + +// checkHoldCommentMarker is the hidden marker that makes the hold comment +// idempotent per (PR, apply): a retried preflight fan-out searches recent PR +// comments for it before posting. ApplyIdentifier is server-generated, so the +// marker needs no sanitization. +func checkHoldCommentMarker(req *storage.MergeGateRequest) string { + return fmt.Sprintf("", req.ApplyIdentifier) +} + +// ensureCheckHoldComment posts the comment explaining the hold, exactly once +// per PR and originating apply: the hidden marker in the body deduplicates +// retries after partial failures. Comment failures fail the fan-out — the +// hold's explanation is part of the preflight contract, so the apply does not +// start until the operator-facing surface is complete. +func (h *Handler) ensureCheckHoldComment(ctx context.Context, client *ghclient.InstallationClient, req *storage.MergeGateRequest, repo string, pr int) error { + marker := checkHoldCommentMarker(req) + exists, err := client.HasIssueCommentWithMarker(ctx, repo, pr, marker) + if err != nil { + return fmt.Errorf("search for existing hold comment on %s#%d (apply %s): %w", + repo, pr, req.ApplyIdentifier, err) + } + if exists { + h.logger.Debug("check preflight hold comment already posted", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr) + return nil + } + body := templates.RenderCheckHold(templates.CheckHoldData{ + ApplyIdentifier: req.ApplyIdentifier, + RequestedBy: req.RequestedBy, + Database: req.DatabaseName, + Environment: req.Environment, + }) + body = h.renderPRComment(body) + "\n" + marker + if _, _, err := client.CreateIssueComment(ctx, repo, pr, body); err != nil { + return fmt.Errorf("post hold comment on %s#%d (target %s/%s in %s, apply %s): %w", + repo, pr, req.DatabaseType, req.DatabaseName, req.Environment, req.ApplyIdentifier, err) + } + h.logger.Info("check preflight posted the hold comment", + "apply_id", req.ApplyIdentifier, "repo", repo, "pr", pr, + "environment", req.Environment, "database_type", req.DatabaseType, + "database", req.DatabaseName) + return nil +} + // refreshPRPlanForTarget re-plans one sibling PR's stored check state against // the target's live schema. It returns nil when the PR was refreshed or safely // skipped (closed, in-flight-owned, no longer managed, or superseded by a diff --git a/pkg/webhook/merge_gate_integration_test.go b/pkg/webhook/merge_gate_integration_test.go index 2ebbb0c84..18c5623ca 100644 --- a/pkg/webhook/merge_gate_integration_test.go +++ b/pkg/webhook/merge_gate_integration_test.go @@ -1,24 +1,31 @@ //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 -// end to end against the real webhook harness: recording at the operator drive -// tail, the backstop sweep, the sibling PR fan-out with attribution, the -// fail-closed flip when a re-plan fails, the in-flight apply guard, -// same-target request coalescing, and the recorded-request kick that drains -// without waiting for a poll tick. +// Merge gate guardrail integration tests. Before an apply's engine work +// starts on a (environment, database type, database) target, a preflight +// request holds every sibling PR's stored check on that target +// action-required — with a PR comment explaining the hold — so a merge cannot +// land on a verdict the apply is about to invalidate. Once the apply settles +// 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 +// 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. package webhook import ( "context" "database/sql" + "encoding/json" "fmt" "net/http" "net/http/httptest" "net/url" + "sync" "testing" "time" @@ -31,7 +38,7 @@ import ( "github.com/block/schemabot/pkg/storage" ) -const mergeGateTestLeaseOwner = "check-refresh-test-driver" +const mergeGateTestLeaseOwner = "merge-gate-test-driver" // clearMergeGateRequests empties the shared merge_gate_requests table. // The table is cross-test shared state: apply drive tails in earlier tests @@ -46,7 +53,7 @@ func clearMergeGateRequests(t *testing.T) { require.NoError(t, err) } -// recordRefreshRequest records a pending refresh request directly, standing in +// recordRefreshRequest records a pending merge gate request directly, standing in // for the operator drive tail so the processor side can be exercised in // isolation. func recordRefreshRequest(t *testing.T, svc *api.Service, req *storage.MergeGateRequest) *storage.MergeGateRequest { @@ -104,8 +111,8 @@ 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. Installed after the handler so this probe is - // the active registration. + // for its next poll tick. The probe stands in for the processor's kick, + // which only a started processor registers. kicked := make(chan struct{}, 1) svc.OnMergeGateRecorded = func() { select { @@ -179,7 +186,7 @@ func TestE2EMergeGateRecordedOnApplyTerminalSuccess(t *testing.T) { } // TestE2EMergeGateSweepBackfillsMissedApply verifies the outbox backstop: a -// completed apply with no refresh request (a pod crash between the terminal +// completed apply with no merge gate request (a pod crash between the terminal // write and the drive-tail recording) is found by the processor's sweep and // its request backfilled with full attribution, so the fan-out is never lost. func TestE2EMergeGateSweepBackfillsMissedApply(t *testing.T) { @@ -230,7 +237,7 @@ func TestE2EMergeGateSweepBackfillsMissedApply(t *testing.T) { 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") + require.NotNil(t, gateReq, "the sweep must backfill a merge gate request for a completed apply that has none") assert.Equal(t, apply.ApplyIdentifier, gateReq.ApplyIdentifier) assert.Equal(t, "staging", gateReq.Environment) assert.Equal(t, dbName, gateReq.DatabaseName) @@ -508,8 +515,6 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { client.BaseURL, _ = url.Parse(server.URL + "/") h := newE2EHandler(t, svc, client) - require.NotNil(t, svc.OnMergeGateRecorded, - "the handler registers the drive-tail kick on the service at construction") // A sentinel recorded before start is drained by the driver's startup // pass; its completion means the driver is parked on the (hour-long) @@ -527,6 +532,8 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { h.mergeGatePollInterval = time.Hour h.StartMergeGateProcessor(t.Context()) t.Cleanup(h.StopMergeGateProcessor) + require.NotNil(t, svc.OnMergeGateRecorded, + "starting the processor registers the drive-tail kick on the service") require.EventuallyWithT(t, func(collect *assert.CollectT) { got, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), sentinel.ApplyID, storage.MergeGateKindSettle) @@ -558,3 +565,374 @@ func TestE2EMergeGateKickDrainsWithoutTick(t *testing.T) { }, webhookIntegrationPollDeadline, 100*time.Millisecond, "the kick drains the request without a poll tick") } + +// seedApplyWithLock creates a lock and an apply in the given state against +// the target, standing in for an apply the operator drove there. +func seedApplyWithLock(t *testing.T, svc *api.Service, dbName, applyState string, pr int) *storage.Apply { + t.Helper() + ctx := t.Context() + lock := &storage.Lock{ + DatabaseName: dbName, + DatabaseType: "mysql", + Repository: "octocat/hello-world", + PullRequest: pr, + Owner: fmt.Sprintf("octocat/hello-world#%d", pr), + } + require.NoError(t, svc.Storage().Locks().Acquire(ctx, lock)) + lock, err := svc.Storage().Locks().Get(ctx, dbName, "mysql") + require.NoError(t, err) + t.Cleanup(func() { + _ = svc.Storage().Locks().ForceRelease(context.WithoutCancel(t.Context()), dbName, "mysql") + }) + + apply := &storage.Apply{ + ApplyIdentifier: fmt.Sprintf("apply_mergegate_%s_%d", applyState, time.Now().UnixNano()), + LockID: lock.ID, + PlanID: 1, + Database: dbName, + DatabaseType: "mysql", + Repository: "octocat/hello-world", + PullRequest: pr, + Environment: "staging", + Caller: "cli:preflighter@host", + InstallationID: 12345, + Engine: "spirit", + State: applyState, + } + applyID, err := svc.Storage().Applies().Create(ctx, apply) + require.NoError(t, err) + apply.ID = applyID + return apply +} + +// TestE2ECheckPreflightHoldsSiblingChecksAndComments verifies the preflight +// fan-out that runs before an apply's engine work starts: a sibling PR whose +// stored check on the target is green — the only kind of check a merge can +// land on — is flipped to action required with the apply-in-flight blocking +// reason, and one comment explaining the hold is posted on the PR. A retried +// fan-out converges without flipping the row again or posting a duplicate +// comment: the flip skips already-held rows and the comment is deduplicated +// by its hidden marker. +func TestE2ECheckPreflightHoldsSiblingChecksAndComments(t *testing.T) { + clearMergeGateRequests(t) + dbName := "webhook_mergegate_preflight" + // The hold flips stored state and talks to GitHub; no re-plan runs, so the + // lighter storage-backed service is enough — no target database. + svc := setupE2EServiceWithConfig(t, &api.ServerConfig{}) + + 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 PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;", + } + result := setupFakeGitHubForPlan(t, mux, schemaFiles, schemabotConfig, dbName) + + // Serve the PR's issue comments for the marker search that makes the hold + // comment idempotent; posted comments are fed back via servedComments. + var commentsMu sync.Mutex + var servedComments []string + mux.HandleFunc("GET /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, _ *http.Request) { + commentsMu.Lock() + defer commentsMu.Unlock() + comments := make([]*gh.IssueComment, 0, len(servedComments)) + for i, body := range servedComments { + comments = append(comments, &gh.IssueComment{ID: new(int64(i + 1)), Body: new(body)}) + } + _ = json.NewEncoder(w).Encode(comments) + }) + + // The sibling PR's green check is the merge-vulnerable state the hold + // exists for. + seedRefreshTargetCheck(t, svc, 1, "staging", dbName, + checkStatusCompleted, checkConclusionSuccess, "no changes") + + h := newE2EHandler(t, svc, client) + + applyIdentifier := fmt.Sprintf("apply_mergegate_preflight_%d", time.Now().UnixNano()) + preflight := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ + ApplyID: 91000008, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: applyIdentifier, + Environment: "staging", + DatabaseType: "mysql", + DatabaseName: dbName, + Repository: "octocat/hello-world", + ChangeKey: "2", + RequestedBy: "cli:preflighter@host", + }) + + h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) + + held, err := svc.Storage().Checks().Get(t.Context(), "octocat/hello-world", 1, "staging", "mysql", dbName) + require.NoError(t, err) + require.NotNil(t, held) + assert.Equal(t, checkStatusCompleted, held.Status) + assert.Equal(t, checkConclusionActionRequired, held.Conclusion) + assert.Equal(t, applyInFlightBlock.blockingReason, held.BlockingReason) + assert.Equal(t, applyInFlightBlock.message, held.ErrorMessage) + assert.Contains(t, held.ChangeSummary, "held: apply "+applyIdentifier) + assert.Contains(t, held.ChangeSummary, "cli:preflighter@host") + + select { + case body := <-result.comments: + assert.Contains(t, body, "Schema Check On Hold") + assert.Contains(t, body, applyIdentifier) + assert.Contains(t, body, "cli:preflighter@host") + assert.Contains(t, body, checkHoldCommentMarker(preflight)) + commentsMu.Lock() + servedComments = append(servedComments, body) + commentsMu.Unlock() + default: + t.Fatal("the preflight fan-out must post a comment explaining the hold") + } + + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), preflight.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, finished) + assert.Equal(t, storage.MergeGateCompleted, finished.State) + + // A retried fan-out (for example after a lease handover) converges: the + // already-held row is not re-flipped and the marker search suppresses a + // second comment. + db, err := sql.Open("mysql", e2eSchemabotDSN) + require.NoError(t, err) + defer func() { _ = db.Close() }() + _, err = db.ExecContext(t.Context(), ` + UPDATE merge_gate_requests + SET state = 'pending', attempts = 0, lease_owner = NULL, lease_token = NULL, + lease_expires_at = NULL, completed_at = NULL + WHERE id = ?`, finished.ID) + require.NoError(t, err) + + h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) + + refinished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), preflight.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + assert.Equal(t, storage.MergeGateCompleted, refinished.State) + select { + case body := <-result.comments: + t.Fatalf("a retried preflight fan-out must not post a duplicate hold comment, got: %s", body) + default: + } + stillHeld, err := svc.Storage().Checks().Get(t.Context(), "octocat/hello-world", 1, "staging", "mysql", dbName) + require.NoError(t, err) + assert.Equal(t, applyInFlightBlock.blockingReason, stillHeld.BlockingReason) +} + +// TestE2ECheckPreflightStoredHoldsLandDuringGitHubOutage exercises the +// preflight fan-out against a fully unavailable GitHub API. The hold phase is +// storage-only, so the sibling PR's green stored check still flips +// action-required and holds_recorded_at — the signal the operator gate starts +// the apply on — still lands; the render phase fails and keeps the request +// retryable for when GitHub recovers. A GitHub outage must never leave a +// sibling's green stored check standing, and must never block the apply on +// the rendering of its own holds. +func TestE2ECheckPreflightStoredHoldsLandDuringGitHubOutage(t *testing.T) { + clearMergeGateRequests(t) + dbName := "webhook_mergegate_outage" + svc := setupE2EServiceWithConfig(t, &api.ServerConfig{}) + + // Every GitHub call fails: the API is down for the whole fan-out. + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "github unavailable", http.StatusServiceUnavailable) + }) + server := httptest.NewServer(mux) + t.Cleanup(server.Close) + client := gh.NewClient(nil) + client.BaseURL, _ = url.Parse(server.URL + "/") + + // The sibling PR's green check is the merge-vulnerable state the hold + // exists for. + seedRefreshTargetCheck(t, svc, 1, "staging", dbName, + checkStatusCompleted, checkConclusionSuccess, "no changes") + + h := newE2EHandler(t, svc, client) + + applyIdentifier := fmt.Sprintf("apply_mergegate_outage_%d", time.Now().UnixNano()) + preflight := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ + ApplyID: 91000012, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: applyIdentifier, + Environment: "staging", + DatabaseType: "mysql", + DatabaseName: dbName, + Repository: "octocat/hello-world", + ChangeKey: "2", + RequestedBy: "cli:preflighter@host", + }) + + h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) + + // The stored hold landed despite the outage. + held, err := svc.Storage().Checks().Get(t.Context(), "octocat/hello-world", 1, "staging", "mysql", dbName) + require.NoError(t, err) + require.NotNil(t, held) + assert.Equal(t, checkStatusCompleted, held.Status) + assert.Equal(t, checkConclusionActionRequired, held.Conclusion) + assert.Equal(t, applyInFlightBlock.blockingReason, held.BlockingReason) + assert.Contains(t, held.ChangeSummary, "held: apply "+applyIdentifier) + + // The stamp the operator gate waits on is set, while the request itself + // stays retryable so the render lands once GitHub recovers. + req, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), preflight.ApplyID, storage.MergeGateKindPreflight) + require.NoError(t, err) + require.NotNil(t, req) + assert.NotNil(t, req.HoldsRecordedAt, "the hold phase must stamp holds_recorded_at even when GitHub is down") + assert.Equal(t, storage.MergeGateFailed, req.State, "the render failure keeps the request in its retry lifecycle") + assert.NotNil(t, req.RetryAfter, "the render failure is retryable, not terminal") + assert.Contains(t, req.LastError, "verify PR state for check preflight") +} + +// TestE2ECheckReleaseSweepSettlesFailedPreflightedApply verifies holds always +// release: an apply that held sibling PR checks and then failed — its drive +// tail may never run, for example when it is cancelled while queued — is +// found by the release sweep, which backfills a settle; the settle's fan-out +// re-plans the held sibling against the live schema, replacing the hold with +// a live verdict. +func TestE2ECheckReleaseSweepSettlesFailedPreflightedApply(t *testing.T) { + clearMergeGateRequests(t) + dbName := "webhook_mergegate_release" + 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;", + } + setupFakeGitHubForPlan(t, mux, schemaFiles, schemabotConfig, dbName) + + apply := seedApplyWithLock(t, svc, dbName, state.Apply.Failed, 2) + recordRefreshRequest(t, svc, &storage.MergeGateRequest{ + ApplyID: apply.ID, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: apply.ApplyIdentifier, + Environment: "staging", + DatabaseType: "mysql", + DatabaseName: dbName, + Repository: "octocat/hello-world", + ChangeKey: "2", + RequestedBy: apply.Caller, + }) + + // The sibling PR's check is held, as the apply's preflight left it. + heldCheck := seedRefreshTargetCheck(t, svc, 1, "staging", dbName, + checkStatusCompleted, checkConclusionActionRequired, "held: apply in flight") + heldCheck.BlockingReason = applyInFlightBlock.blockingReason + require.NoError(t, svc.Storage().Checks().Upsert(t.Context(), heldCheck)) + + h := newE2EHandler(t, svc, client) + + h.sweepPreflightedAppliesMissingSettle(t.Context()) + settle, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindSettle) + require.NoError(t, err) + require.NotNil(t, settle, "the release sweep must backfill a settle for a preflighted terminal apply") + assert.Equal(t, apply.ApplyIdentifier, settle.ApplyIdentifier) + assert.Equal(t, apply.Caller, settle.RequestedBy) + + // Recording is idempotent per apply and kind: a second sweep pass must not + // duplicate the settle. + h.sweepPreflightedAppliesMissingSettle(t.Context()) + again, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindSettle) + require.NoError(t, err) + require.NotNil(t, again) + assert.Equal(t, settle.ID, again.ID) + + h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) + + released, err := svc.Storage().Checks().Get(t.Context(), "octocat/hello-world", 1, "staging", "mysql", dbName) + require.NoError(t, err) + require.NotNil(t, released) + assert.Empty(t, released.BlockingReason, "the settle re-plan replaces the hold with a live verdict") + assert.Contains(t, released.ChangeSummary, "re-planned:") + + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), apply.ID, storage.MergeGateKindSettle) + require.NoError(t, err) + assert.Equal(t, storage.MergeGateCompleted, finished.State) +} + +// TestE2ECheckSettleDefersToActivePreflightedApply verifies hold ordering +// when applies overlap on a target: an earlier apply's settle must not +// re-plan sibling checks while a later preflighted apply is still running +// there — the re-plan would compute verdicts against a schema that apply is +// about to change, overwriting its holds. The settle completes without +// touching the checks; the active apply's own settle re-plans them when it +// finishes. +func TestE2ECheckSettleDefersToActivePreflightedApply(t *testing.T) { + clearMergeGateRequests(t) + dbName := "webhook_mergegate_defer" + // The deferral decision is storage-only, so the lighter storage-backed + // service is enough — no target database and no GitHub fixtures: any + // GitHub call would fail the drain loudly. + svc := setupE2EServiceWithConfig(t, &api.ServerConfig{}) + + client := gh.NewClient(nil) + server := httptest.NewServer(http.NewServeMux()) + t.Cleanup(server.Close) + client.BaseURL, _ = url.Parse(server.URL + "/") + + // A later apply on the target is mid-drive with its preflight already + // completed, so its holds are live. + active := seedApplyWithLock(t, svc, dbName, state.Apply.Running, 2) + activePreflight := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ + ApplyID: active.ID, + Kind: storage.MergeGateKindPreflight, + ApplyIdentifier: active.ApplyIdentifier, + Environment: "staging", + DatabaseType: "mysql", + DatabaseName: dbName, + Repository: "octocat/hello-world", + ChangeKey: "2", + RequestedBy: active.Caller, + }) + claimed, err := svc.Storage().MergeGateRequests().ClaimNext(t.Context(), mergeGateTestLeaseOwner, time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + require.Equal(t, activePreflight.ID, claimed.ID) + require.NoError(t, svc.Storage().MergeGateRequests().MarkCompleted(t.Context(), claimed.ID, claimed.LeaseToken)) + + heldCheck := seedRefreshTargetCheck(t, svc, 1, "staging", dbName, + checkStatusCompleted, checkConclusionActionRequired, "held: apply in flight") + heldCheck.BlockingReason = applyInFlightBlock.blockingReason + require.NoError(t, svc.Storage().Checks().Upsert(t.Context(), heldCheck)) + + h := newE2EHandler(t, svc, client) + + // An earlier apply on the same target settles while the later one runs. + settle := recordRefreshRequest(t, svc, &storage.MergeGateRequest{ + ApplyID: 91000009, + Kind: storage.MergeGateKindSettle, + ApplyIdentifier: fmt.Sprintf("apply_mergegate_defer_%d", time.Now().UnixNano()), + Environment: "staging", + DatabaseType: "mysql", + DatabaseName: dbName, + RequestedBy: "cli:tester@host", + }) + + h.drainMergeGateRequests(t.Context(), mergeGateTestLeaseOwner) + + finished, err := svc.Storage().MergeGateRequests().GetByApplyAndKind(t.Context(), settle.ApplyID, storage.MergeGateKindSettle) + require.NoError(t, err) + require.NotNil(t, finished) + assert.Equal(t, storage.MergeGateCompleted, finished.State, + "the deferred settle completes; the active apply's own settle covers the target") + + stillHeld, err := svc.Storage().Checks().Get(t.Context(), "octocat/hello-world", 1, "staging", "mysql", dbName) + require.NoError(t, err) + require.NotNil(t, stillHeld) + assert.Equal(t, checkConclusionActionRequired, stillHeld.Conclusion) + assert.Equal(t, applyInFlightBlock.blockingReason, stillHeld.BlockingReason, + "the active apply's holds must survive an earlier apply's settle") +} diff --git a/pkg/webhook/templates/check_hold.go b/pkg/webhook/templates/check_hold.go new file mode 100644 index 000000000..64f23cfd5 --- /dev/null +++ b/pkg/webhook/templates/check_hold.go @@ -0,0 +1,54 @@ +package templates + +import ( + "fmt" + "strings" +) + +// CheckHoldData describes an apply that is about to change the live schema of +// a database this PR also targets, so the PR's SchemaBot check is held until +// the apply finishes. +type CheckHoldData struct { + // ApplyIdentifier is the user-facing identifier of the apply causing the + // hold. + ApplyIdentifier string + // RequestedBy is the apply's caller. Caller-influenced text — the + // renderer neutralizes markdown-sensitive characters. + RequestedBy string + Database string + Environment string +} + +// RenderCheckHold renders the PR comment posted when an apply on the same +// database moves this PR's check to action required before it starts. It +// explains why the check flipped without any activity on the PR itself, and +// what happens when the apply finishes. +func RenderCheckHold(data CheckHoldData) string { + var sb strings.Builder + sb.WriteString("## ⏸️ Schema Check On Hold") + sb.WriteString(environmentTitleSuffix(data.Environment)) + sb.WriteString("\n\n") + fmt.Fprintf(&sb, "Apply `%s`", sanitizeInlineCode(data.ApplyIdentifier)) + if data.RequestedBy != "" { + fmt.Fprintf(&sb, " (requested by `%s`)", sanitizeInlineCode(data.RequestedBy)) + } + fmt.Fprintf(&sb, " has started changing the live schema of `%s` in `%s` — the same database this PR's schema check was evaluated against.\n\n", + sanitizeInlineCode(data.Database), sanitizeInlineCode(data.Environment)) + sb.WriteString("This PR's check verdict was computed against the schema that apply is replacing, so the check has been moved to **action required** to keep a merge from landing on a stale verdict while the apply runs.\n\n") + sb.WriteString("**What happens next**\n\n") + sb.WriteString("- No action is needed right now; the hold clears automatically.\n") + sb.WriteString("- When the apply finishes, SchemaBot re-plans this PR against the resulting schema and refreshes this check.\n") + sb.WriteString("- If the refreshed plan still matches the live schema, the check returns to passing on its own; if it shows new differences, review the refreshed plan before applying or merging.\n") + return sb.String() +} + +// sanitizeInlineCode neutralizes text rendered inside single-backtick inline +// code: backticks would terminate the span and newlines would break the +// surrounding markdown, so both are replaced with spaces and the result is +// trimmed. +func sanitizeInlineCode(s string) string { + s = strings.ReplaceAll(s, "`", " ") + s = strings.ReplaceAll(s, "\r", " ") + s = strings.ReplaceAll(s, "\n", " ") + return strings.TrimSpace(s) +}