diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index 88c95c9c9..bed965a58 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -1640,6 +1640,21 @@ func RecordMergeGatePlanTimeHold(ctx context.Context, database, environment stri ) } +// RecordMergeGroupAdmissionBlocked counts merge-queue admission checks posted +// blocking: a queued pull request's stored check state turned blocking after +// it entered the queue — most often a preflight hold from a sibling change's +// in-flight apply on a shared target. The blocked PR leaves the queue and its +// author re-queues once its check re-plans green, so an occasional count is +// the gate working; a sustained rate on one repository means changes keep +// queueing against a busy target — check the target's apply activity. +func RecordMergeGroupAdmissionBlocked(ctx context.Context, repository, environment string) { + addCounter(ctx, "schemabot.merge_gate.merge_group_admission_blocked_total", + "Total merge-queue admission checks posted blocking because the queued PR's stored check state blocks", "{check}", + attribute.String("repository", repository), + EnvironmentAttribute(environment), + ) +} + // RecordMergeGateTerminatedStuck counts merge gate requests terminated // by the stuck-processing sweep: rows wedged past the attempt cap with an // expired lease (a driver hard-killed on its final attempt). Each terminated diff --git a/pkg/webhook/check_aggregate.go b/pkg/webhook/check_aggregate.go index 9649873ae..b482805a6 100644 --- a/pkg/webhook/check_aggregate.go +++ b/pkg/webhook/check_aggregate.go @@ -197,6 +197,22 @@ func hasBlockingCheckForEnvironment(checks []*storage.Check, environment string) return false } +// blockingChecksForEnvironment returns the stored rows that block a passing +// aggregate for the environment. Callers list them on operator-facing +// surfaces so a block names the databases behind it. +func blockingChecksForEnvironment(checks []*storage.Check, environment string) []*storage.Check { + var blocking []*storage.Check + for _, c := range checks { + if environment != aggregateSentinel && c.Environment != environment { + continue + } + if checkBlocksPassingAggregate(c) { + blocking = append(blocking, c) + } + } + return blocking +} + // awaitingCurrentCommitTitle is the aggregate title shown when only results // recorded for another commit hold the aggregate open — nothing is running for // the commit the Check Run is published on, and the current commit's own plan diff --git a/pkg/webhook/durable_merge_group_test.go b/pkg/webhook/durable_merge_group_test.go index 90f43ec81..59a662d30 100644 --- a/pkg/webhook/durable_merge_group_test.go +++ b/pkg/webhook/durable_merge_group_test.go @@ -21,7 +21,7 @@ import ( func durableMergeGroupEvent() *storage.WebhookEvent { payload := []byte(`{ "action": "checks_requested", - "merge_group": {"head_sha": "mergesha123"}, + "merge_group": {"head_sha": "mergesha123", "head_ref": "refs/heads/gh-readonly-queue/main/pr-1-mergesha123"}, "repository": {"full_name": "octocat/hello-world"}, "installation": {"id": 12345} }`) @@ -191,6 +191,132 @@ func TestDurableMergeGroupDriverParticipantStaysSilent(t *testing.T) { } } +// durableMergeGroupStorage layers a scripted check store over the durable +// webhook harness so driver tests can seed stored check state for the +// admission fold. +type durableMergeGroupStorage struct { + durableWebhookTestStorage + checks storage.CheckStore +} + +func (s *durableMergeGroupStorage) Checks() storage.CheckStore { + return s.checks +} + +func newDurableMergeGroupDriverHandler(t *testing.T, store storage.WebhookEventStore, checks storage.CheckStore, config *api.ServerConfig, factory *fakeClientFactory) *Handler { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + service := api.New(&durableMergeGroupStorage{ + durableWebhookTestStorage: durableWebhookTestStorage{webhookEvents: store}, + checks: checks, + }, config, nil, logger) + if factory == nil { + factory = &fakeClientFactory{} + } + return NewHandler(service, factory, nil, logger, WithDurableWebhookDispatch()) +} + +// When the queued PR's stored check state blocks — a sibling apply's preflight +// hold landed after the PR entered the queue — the driver posts a blocking +// admission check on the merge-group head instead of the pass the PR queued +// with, and the delivery still completes: the verdict was delivered. +func TestDurableMergeGroupDriverBlocksOnStoredHold(t *testing.T) { + client, mux := setupGitHubServer(t) + created := make(chan checkRunCapture, 10) + comments := make(chan string, 10) + mux.HandleFunc("POST /repos/octocat/hello-world/check-runs", func(w http.ResponseWriter, r *http.Request) { + var c checkRunCapture + require.NoError(t, json.NewDecoder(r.Body).Decode(&c)) + created <- c + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"id": 556}) + }) + mux.HandleFunc("GET /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode([]map[string]any{}) + }) + mux.HandleFunc("POST /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, r *http.Request) { + var c struct { + Body string `json:"body"` + } + require.NoError(t, json.NewDecoder(r.Body).Decode(&c)) + comments <- c.Body + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"id": 777}) + }) + installClient := ghclient.NewInstallationClient(client, testLogger()) + + store := newScriptedWebhookEventStore(durableMergeGroupEvent()) + checks := &foldCheckStore{byPR: []*storage.Check{{ + Repository: "octocat/hello-world", + PullRequest: 1, + HeadSHA: "prhead123", + Environment: "production", + DatabaseType: "mysql", + DatabaseName: "widgets", + Status: "completed", + Conclusion: "action_required", + }}} + config := &api.ServerConfig{ + AllowedEnvironments: []string{"production"}, + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + } + h := newDurableMergeGroupDriverHandler(t, store, checks, config, &fakeClientFactory{client: installClient}) + + h.driveNextDurableWebhook(t.Context(), 0, "test-host/1/webhook-driver-0") + + select { + case c := <-created: + assert.Equal(t, "SchemaBot (production)", c.Name) + assert.Equal(t, "mergesha123", c.HeadSHA) + assert.Equal(t, "action_required", c.Conclusion) + case <-time.After(durableWebhookTestDeadline): + t.Fatal("expected the driver to post a blocking merge_group check") + } + select { + case body := <-comments: + assert.Contains(t, body, "Removed From Merge Queue") + assert.Contains(t, body, "`widgets` in `production`") + assert.Contains(t, body, mergeQueueEjectedCommentMarker("mergesha123")) + case <-time.After(durableWebhookTestDeadline): + t.Fatal("expected the driver to post the ejection guidance comment") + } + select { + case <-store.completed: + case <-time.After(durableWebhookTestDeadline): + t.Fatal("expected the delivery to be marked completed") + } + require.Empty(t, store.failed) +} + +// A storage read failure during the admission fold is retryable: the verdict +// cannot be computed from unknown state, so the delivery is retried rather +// than completed with a guessed check. +func TestDurableMergeGroupDriverRetriesStorageFailure(t *testing.T) { + client, _ := setupGitHubServer(t) + installClient := ghclient.NewInstallationClient(client, testLogger()) + + store := newScriptedWebhookEventStore(durableMergeGroupEvent()) + checks := &foldCheckStore{byPRErr: errors.New("storage unavailable")} + config := &api.ServerConfig{ + AllowedEnvironments: []string{"production"}, + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + } + h := newDurableMergeGroupDriverHandler(t, store, checks, config, &fakeClientFactory{client: installClient}) + + before := time.Now() + h.driveNextDurableWebhook(t.Context(), 0, "test-host/1/webhook-driver-0") + + select { + case failure := <-store.failed: + require.NotNil(t, failure.retryAfter, "storage failure must stay retryable") + require.True(t, failure.retryAfter.After(before), "retry must be scheduled in the future") + require.Contains(t, failure.errMsg, "storage unavailable") + default: + t.Fatal("expected storage failure to be marked failed") + } + require.Empty(t, store.completed) +} + // A GitHub client failure while posting the check is retryable: the merge queue // blocks until the check lands, so the delivery must be retried, not dropped. func TestDurableMergeGroupDriverRetriesClientFailure(t *testing.T) { diff --git a/pkg/webhook/merge_group.go b/pkg/webhook/merge_group.go index 4ceceecf7..d46436a2d 100644 --- a/pkg/webhook/merge_group.go +++ b/pkg/webhook/merge_group.go @@ -5,12 +5,15 @@ import ( "encoding/json" "fmt" "net/http" + "regexp" "strconv" + "strings" "time" ghclient "github.com/block/schemabot/pkg/github" "github.com/block/schemabot/pkg/metrics" "github.com/block/schemabot/pkg/storage" + "github.com/block/schemabot/pkg/webhook/templates" ) // mergeGroupPayload is the subset of the GitHub merge_group webhook payload @@ -20,6 +23,7 @@ type mergeGroupPayload struct { Action string `json:"action"` MergeGroup struct { HeadSHA string `json:"head_sha"` + HeadRef string `json:"head_ref"` } `json:"merge_group"` Repository struct { FullName string `json:"full_name"` @@ -30,7 +34,7 @@ type mergeGroupPayload struct { } // handleMergeGroup responds to merge_group webhook events so SchemaBot's -// required Check Runs do not wedge a repository's merge queue. +// required Check Runs gate a repository's merge queue without wedging it. // // A merge queue tests queued pull requests combined, on a synthetic head // commit, before they land on the base branch. Branch protection re-evaluates @@ -39,11 +43,14 @@ type mergeGroupPayload struct { // required SchemaBot check would never appear on the merge-group commit and the // queue entry would block indefinitely. // -// Posting a passing check is correct: SchemaBot applies schema changes before a -// PR merges, and branch protection already required the PR-head check to pass -// before the PR could enter the queue. The merge group sits strictly downstream -// of an already-completed, already-gated apply, so there is nothing left to -// verify on the combined commit. +// The check posted here is a merge-time revalidation, not an unconditional +// pass. The PR-head check gated queue entry, but stored check state can turn +// blocking after entry — most importantly when a sibling PR's apply starts +// changing a target this PR's verdict was computed against and the preflight +// holds this PR's stored checks. The queue no longer looks at the PR head, so +// this admission check is the only surface where that hold can stop the +// queued merge: it re-folds the PR's stored check state and passes only when +// nothing blocks. func (h *Handler) handleMergeGroup(ctx context.Context, metricApp string, w http.ResponseWriter, body []byte, deliveryID string) { var payload mergeGroupPayload if err := json.Unmarshal(body, &payload); err != nil { @@ -143,7 +150,7 @@ func (h *Handler) handleMergeGroup(ctx context.Context, metricApp string, w http return } - if err := h.postPassingAggregateChecks(postCtx, client, repo, headSHA, mergeGroupCheckContent()); err != nil { + if err := h.postMergeGroupAdmissionChecks(postCtx, client, repo, headSHA, payload.MergeGroup.HeadRef); err != nil { // Return 500 so the delivery is recorded as failed and shows up in the // App's delivery log for redelivery. The merge queue blocks until the // check is posted, so the failure must be visible for retry, not @@ -162,16 +169,181 @@ func (h *Handler) handleMergeGroup(ctx context.Context, metricApp string, w http h.writeJSON(w, http.StatusOK, map[string]string{"message": "merge_group checks posted"}) } -// mergeGroupCheckContent is the passing Check Run content published on a -// merge-queue head. Both the request path and the durable driver use it so a -// redelivery updates the same check with identical output. -func mergeGroupCheckContent() passingAggregateCheckContent { +// mergeGroupHeadRefPattern matches the tail segment of a merge-queue branch +// ref, refs/heads/gh-readonly-queue//pr--. Only the +// final path segment is matched so a base branch containing slashes cannot +// shift the pull request number. +var mergeGroupHeadRefPattern = regexp.MustCompile(`^pr-(\d+)-`) + +// mergeGroupPRNumber extracts the queued pull request's number from the +// merge_group head ref. Each queue entry gets its own merge_group event whose +// head ref names that entry's PR, so the ref identifies which PR's stored +// check state this admission check must re-fold. +func mergeGroupPRNumber(headRef string) (int, error) { + tail := headRef + if idx := strings.LastIndex(headRef, "/"); idx >= 0 { + tail = headRef[idx+1:] + } + m := mergeGroupHeadRefPattern.FindStringSubmatch(tail) + if m == nil { + return 0, fmt.Errorf("merge group head ref %q does not name a pull request", headRef) + } + pr, err := strconv.Atoi(m[1]) + if err != nil { + return 0, fmt.Errorf("parse pull request number from merge group head ref %q: %w", headRef, err) + } + return pr, nil +} + +// mergeGroupPassContent is the passing Check Run content published on a +// merge-queue head when the queued PR's stored check state blocks nothing. +// Both the request path and the durable driver use it so a redelivery updates +// the same check with identical output. +func mergeGroupPassContent() passingAggregateCheckContent { return passingAggregateCheckContent{ operation: "merge_group_check", title: "Schema changes verified before merge queue", summary: "Schema changes in queued pull requests are applied and verified by SchemaBot before " + - "they enter the merge queue, so no additional verification is required for this merge group.", + "they enter the merge queue, and nothing is blocking this pull request's schema checks now, " + + "so no additional verification is required for this merge group.", + } +} + +// mergeGroupBlockedSummary is the Check Run output published on a merge-queue +// head when the queued PR's stored check state blocks admission. The blocked +// PR leaves the queue; the summary tells its author where to look and when to +// re-queue. +const ( + mergeGroupBlockedTitle = "Schema checks are blocking this pull request" + mergeGroupBlockedSummary = "This pull request's SchemaBot check state turned blocking after it entered the " + + "merge queue — most often because another change's apply is in flight on a database this pull request " + + "also changes, which invalidates the verdict it queued with. Check this pull request's SchemaBot check " + + "for the reason, and re-queue once it is green again (held checks re-plan automatically when the " + + "in-flight apply settles)." +) + +// mergeGroupUnidentifiedSummary is the fail-closed Check Run output published +// when the queued PR cannot be identified from the merge group ref. Stored +// check state cannot be consulted for a PR that cannot be named, and admission +// must never pass on uncertainty. +const ( + mergeGroupUnidentifiedTitle = "SchemaBot could not identify the queued pull request" + mergeGroupUnidentifiedSummary = "SchemaBot could not determine which pull request this merge group tests, " + + "so it cannot verify the pull request's schema check state and fails closed. See the server logs for " + + "the merge group ref that could not be parsed." +) + +// postMergeGroupAdmissionChecks publishes the admission Check Runs on a +// merge-queue head: a re-fold of the queued PR's stored check state at +// admission time. Every aggregate target this instance gates gets a run — +// passing when nothing in the PR's stored state blocks, action_required when +// something does (a preflight hold, a failed apply, a fail-closed re-plan) or +// when the queued PR cannot be identified. A storage read failure returns an +// error so the delivery retries; uncertainty never admits a merge. +func (h *Handler) postMergeGroupAdmissionChecks(ctx context.Context, client *ghclient.InstallationClient, repo, headSHA, headRef string) error { + pr, prErr := mergeGroupPRNumber(headRef) + if prErr != nil { + h.logger.Error("merge group admission failing closed: queued pull request could not be identified; the admission check will block this merge group", + "repo", repo, "head_sha", headSHA, "head_ref", headRef, "error", prErr) + return h.postAggregateChecks(ctx, client, repo, headSHA, passingAggregateCheckContent{ + operation: "merge_group_check", + title: mergeGroupUnidentifiedTitle, + summary: mergeGroupUnidentifiedSummary, + }, checkConclusionActionRequired, func(environment string) { + metrics.RecordMergeGroupAdmissionBlocked(ctx, repo, environment) + }) + } + + if h.service == nil || h.service.Storage() == nil { + return fmt.Errorf("storage unavailable for merge group admission of %s#%d@%s", repo, pr, headSHA) + } + checks, err := h.service.Storage().Checks().GetByPR(ctx, repo, pr) + if err != nil { + return fmt.Errorf("load stored check state for merge group admission of %s#%d@%s: %w", repo, pr, headSHA, err) + } + + var blockingRows []*storage.Check + for _, target := range h.aggregateCheckTargetsForRepo(repo) { + if !hasBlockingCheckForEnvironment(checks, target.environment) { + if err := h.postAggregateCheck(ctx, client, repo, headSHA, target, mergeGroupPassContent(), checkConclusionSuccess); err != nil { + return err + } + continue + } + blockingRows = append(blockingRows, blockingChecksForEnvironment(checks, target.environment)...) + h.logger.Warn("merge group admission blocked: the queued pull request's stored check state turned blocking after queue entry; the admission check will remove it from the queue", + "repo", repo, "pr", pr, "head_sha", headSHA, + "environment", target.environment, "check_name", target.name) + metrics.RecordMergeGroupAdmissionBlocked(ctx, repo, target.environment) + if err := h.postAggregateCheck(ctx, client, repo, headSHA, target, passingAggregateCheckContent{ + operation: "merge_group_check", + title: mergeGroupBlockedTitle, + summary: mergeGroupBlockedSummary, + }, checkConclusionActionRequired); err != nil { + return err + } } + if len(blockingRows) == 0 { + h.logger.Info("merge group admission passed: nothing in the queued pull request's stored check state blocks", + "repo", repo, "pr", pr, "head_sha", headSHA) + return nil + } + return h.ensureMergeQueueEjectedComment(ctx, client, repo, pr, headSHA, blockingRows) +} + +// mergeQueueEjectedCommentMarker makes the ejection comment idempotent per +// queue attempt: the merge-group head SHA identifies one queue entry, so a +// webhook redelivery finds the marker and skips the re-post, while a later +// re-queue produces a new merge-group commit and gets a fresh comment at the +// bottom of the PR timeline. The SHA comes from GitHub's payload, not user +// input, so the marker needs no sanitization. +func mergeQueueEjectedCommentMarker(mergeGroupHeadSHA string) string { + return fmt.Sprintf("", mergeGroupHeadSHA) +} + +// ensureMergeQueueEjectedComment posts the guidance comment on a pull request +// the admission check just removed from the merge queue. The blocking Check +// Run lives on the synthetic merge-group commit — invisible from the PR page — +// and the queue never re-adds a pull request on its own, so without this +// comment the author sees their merge silently vanish with no next step. The +// comment is part of the admission contract: a posting failure returns an +// error so the delivery retries (the already-posted checks reconcile +// idempotently on the retry). +func (h *Handler) ensureMergeQueueEjectedComment(ctx context.Context, client *ghclient.InstallationClient, repo string, pr int, mergeGroupHeadSHA string, blocking []*storage.Check) error { + marker := mergeQueueEjectedCommentMarker(mergeGroupHeadSHA) + exists, err := client.HasIssueCommentWithMarker(ctx, repo, pr, marker) + if err != nil { + return fmt.Errorf("search for existing merge-queue ejection comment on %s#%d (merge group %s): %w", + repo, pr, mergeGroupHeadSHA, err) + } + if exists { + h.logger.Debug("merge-queue ejection comment already posted", + "repo", repo, "pr", pr, "merge_group_head_sha", mergeGroupHeadSHA) + return nil + } + + data := templates.MergeQueueEjectedData{} + seen := make(map[string]bool, len(blocking)) + for _, c := range blocking { + key := c.DatabaseName + "\x00" + c.Environment + if c.DatabaseName == "" || seen[key] { + continue + } + seen[key] = true + data.Blocking = append(data.Blocking, templates.MergeQueueBlockedTarget{ + Database: c.DatabaseName, + Environment: c.Environment, + }) + } + body := h.renderPRComment(templates.RenderMergeQueueEjected(data)) + "\n" + marker + if _, _, err := client.CreateIssueComment(ctx, repo, pr, body); err != nil { + return fmt.Errorf("post merge-queue ejection comment on %s#%d (merge group %s): %w", + repo, pr, mergeGroupHeadSHA, err) + } + h.logger.Info("merge-queue ejection comment posted", + "repo", repo, "pr", pr, "merge_group_head_sha", mergeGroupHeadSHA, + "blocking_targets", len(data.Blocking)) + return nil } // enqueueDurableMergeGroup persists a merge_group delivery in the inbox. The @@ -190,13 +362,13 @@ func (h *Handler) enqueueDurableMergeGroup(ctx context.Context, payload mergeGro }) } -// processDurableMergeGroup posts the passing merge-queue check for a claimed +// processDurableMergeGroup posts the merge-queue admission check for a claimed // merge_group delivery. It re-validates every enqueue-time guard fail-closed — // config can change between enqueue and drive, and rows can arrive via replay — // so a now-ignored delivery completes as a no-op rather than posting a check -// the current config would not. GitHub client and post failures are retryable; -// posting is idempotent (the check is looked up by name and updated in place), -// so a retry after a partial post reconciles rather than duplicates. +// the current config would not. GitHub client, storage, and post failures are +// retryable; posting is idempotent (the check is looked up by name and updated +// in place), so a retry after a partial post reconciles rather than duplicates. func (h *Handler) processDurableMergeGroup(ctx context.Context, event *storage.WebhookEvent) (retry bool, err error) { var payload mergeGroupPayload if err := json.Unmarshal(event.Payload, &payload); err != nil { @@ -255,7 +427,7 @@ func (h *Handler) processDurableMergeGroup(ctx context.Context, event *storage.W return true, fmt.Errorf("create GitHub client for durable merge_group %s@%s: %w", repo, headSHA, err) } - if err := h.postPassingAggregateChecks(ctx, client, repo, headSHA, mergeGroupCheckContent()); err != nil { + if err := h.postMergeGroupAdmissionChecks(ctx, client, repo, headSHA, payload.MergeGroup.HeadRef); err != nil { metrics.RecordStatusCheckOperation(ctx, metrics.StatusCheckOperation{ Operation: "merge_group_check", Repository: repo, @@ -270,8 +442,8 @@ func (h *Handler) processDurableMergeGroup(ctx context.Context, event *storage.W } // passingAggregateCheckContent carries the operation name and Check Run output -// for a passing aggregate published outside the PR-head path (merge-queue -// heads, default-branch pushes). +// for an aggregate published outside the PR-head path (merge-queue heads, +// default-branch pushes). type passingAggregateCheckContent struct { operation string title string @@ -283,46 +455,67 @@ type passingAggregateCheckContent struct { // names as the PR-head aggregates so branch protection's required checks // always match. func (h *Handler) postPassingAggregateChecks(ctx context.Context, client *ghclient.InstallationClient, repo, headSHA string, content passingAggregateCheckContent) error { + return h.postAggregateChecks(ctx, client, repo, headSHA, content, checkConclusionSuccess, nil) +} + +// postAggregateChecks publishes one aggregate Check Run per gated environment +// on headSHA with the given conclusion. onEach, when non-nil, runs once per +// environment before its check posts (for per-environment metrics). +func (h *Handler) postAggregateChecks(ctx context.Context, client *ghclient.InstallationClient, repo, headSHA string, content passingAggregateCheckContent, conclusion string, onEach func(environment string)) error { for _, target := range h.aggregateCheckTargetsForRepo(repo) { - opts := ghclient.CheckRunOptions{ - Name: target.name, - Status: checkStatusCompleted, - Conclusion: checkConclusionSuccess, - Output: &ghclient.CheckRunOutput{ - Title: content.title, - Summary: content.summary, - }, + if onEach != nil { + onEach(target.environment) } - // Reuse an existing run for this name on the SHA so a webhook - // redelivery updates it rather than creating a duplicate Check Run. - // The lookup errors when the App slug is unknown; on any lookup error - // fall back to creating a new run — a duplicate Check Run is the safe - // outcome, a missing one is not. - existing, _, findErr := client.FindCheckRunByName(ctx, repo, headSHA, target.name) - if findErr != nil { - h.logger.Warn("could not look up existing check; creating a new one", - "repo", repo, "head_sha", headSHA, "check_name", target.name, - "operation", content.operation, "error", findErr) + if err := h.postAggregateCheck(ctx, client, repo, headSHA, target, content, conclusion); err != nil { + return err } - switch { - case findErr == nil && existing != nil: - if err := client.UpdateCheckRun(ctx, repo, existing.ID, opts); err != nil { - return fmt.Errorf("update %s check %q on %s@%s: %w", content.operation, target.name, repo, headSHA, err) - } - default: - if _, err := client.CreateCheckRun(ctx, repo, headSHA, opts); err != nil { - return fmt.Errorf("create %s check %q on %s@%s: %w", content.operation, target.name, repo, headSHA, err) - } - } - metrics.RecordStatusCheckOperation(ctx, metrics.StatusCheckOperation{ - Operation: content.operation, - Repository: repo, - Environment: target.environment, - Status: "success", - }) - h.logger.Info("passing aggregate check posted", + } + return nil +} + +// postAggregateCheck publishes a single aggregate Check Run on headSHA for one +// aggregate target with the given conclusion, reusing the PR-head aggregate's +// check name so branch protection's required-check names always match. +func (h *Handler) postAggregateCheck(ctx context.Context, client *ghclient.InstallationClient, repo, headSHA string, target aggregateCheckTarget, content passingAggregateCheckContent, conclusion string) error { + opts := ghclient.CheckRunOptions{ + Name: target.name, + Status: checkStatusCompleted, + Conclusion: conclusion, + Output: &ghclient.CheckRunOutput{ + Title: content.title, + Summary: content.summary, + }, + } + // Reuse an existing run for this name on the SHA so a webhook + // redelivery updates it rather than creating a duplicate Check Run. + // The lookup errors when the App slug is unknown; on any lookup error + // fall back to creating a new run — a duplicate Check Run is the safe + // outcome, a missing one is not. + existing, _, findErr := client.FindCheckRunByName(ctx, repo, headSHA, target.name) + if findErr != nil { + h.logger.Warn("could not look up existing check; creating a new one", "repo", repo, "head_sha", headSHA, "check_name", target.name, - "environment", target.environment, "operation", content.operation) + "operation", content.operation, "error", findErr) + } + switch { + case findErr == nil && existing != nil: + if err := client.UpdateCheckRun(ctx, repo, existing.ID, opts); err != nil { + return fmt.Errorf("update %s check %q on %s@%s: %w", content.operation, target.name, repo, headSHA, err) + } + default: + if _, err := client.CreateCheckRun(ctx, repo, headSHA, opts); err != nil { + return fmt.Errorf("create %s check %q on %s@%s: %w", content.operation, target.name, repo, headSHA, err) + } } + metrics.RecordStatusCheckOperation(ctx, metrics.StatusCheckOperation{ + Operation: content.operation, + Repository: repo, + Environment: target.environment, + Status: conclusion, + }) + h.logger.Info("aggregate check posted", + "repo", repo, "head_sha", headSHA, "check_name", target.name, + "environment", target.environment, "operation", content.operation, + "conclusion", conclusion) return nil } diff --git a/pkg/webhook/merge_group_test.go b/pkg/webhook/merge_group_test.go index bbcd463f3..8816f380b 100644 --- a/pkg/webhook/merge_group_test.go +++ b/pkg/webhook/merge_group_test.go @@ -16,6 +16,7 @@ import ( "github.com/block/schemabot/pkg/api" ghclient "github.com/block/schemabot/pkg/github" + "github.com/block/schemabot/pkg/storage" ) // buildMergeGroupWebhookRequest constructs a merge_group webhook POST request. @@ -32,6 +33,7 @@ func buildMergeGroupWebhookRequest(t *testing.T, action, headSHA string, secret "action": action, "merge_group": map[string]any{ "head_sha": headSHA, + "head_ref": "refs/heads/gh-readonly-queue/main/pr-1-" + headSHA, }, "repository": map[string]any{ "full_name": "octocat/hello-world", @@ -56,10 +58,11 @@ func buildMergeGroupWebhookRequest(t *testing.T, action, headSHA string, secret return req } -func mergeGroupTestHandler(t *testing.T, allowedEnvironments []string, repos map[string]api.RepoConfig) (*Handler, chan checkRunCapture) { +func mergeGroupTestHandler(t *testing.T, allowedEnvironments []string, repos map[string]api.RepoConfig, storedChecks ...*storage.Check) (*Handler, chan checkRunCapture, chan string) { t.Helper() client, mux := setupGitHubServer(t) created := make(chan checkRunCapture, 10) + comments := make(chan string, 10) mux.HandleFunc("POST /repos/octocat/hello-world/check-runs", func(w http.ResponseWriter, r *http.Request) { var c checkRunCapture require.NoError(t, json.NewDecoder(r.Body).Decode(&c)) @@ -67,9 +70,21 @@ func mergeGroupTestHandler(t *testing.T, allowedEnvironments []string, repos map w.WriteHeader(http.StatusCreated) _ = json.NewEncoder(w).Encode(map[string]any{"id": 555}) }) + mux.HandleFunc("GET /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode([]map[string]any{}) + }) + mux.HandleFunc("POST /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, r *http.Request) { + var c struct { + Body string `json:"body"` + } + require.NoError(t, json.NewDecoder(r.Body).Decode(&c)) + comments <- c.Body + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"id": 777}) + }) installClient := ghclient.NewInstallationClient(client, testLogger()) - service := api.New(nil, &api.ServerConfig{ + service := api.New(&foldStorage{checks: &foldCheckStore{byPR: storedChecks}}, &api.ServerConfig{ AllowedEnvironments: allowedEnvironments, Repos: repos, }, nil, testLogger()) @@ -79,7 +94,7 @@ func mergeGroupTestHandler(t *testing.T, allowedEnvironments []string, repos map ghClients: ghclient.NewSingleClientSet(defaultAppName, &fakeClientFactory{client: installClient}), logger: testLogger(), } - return h, created + return h, created, comments } // A merge queue evaluates the same required checks against the queue's @@ -89,7 +104,7 @@ func mergeGroupTestHandler(t *testing.T, allowedEnvironments []string, repos map // the same names as the PR-head aggregates — so a required SchemaBot check does // not block the merge queue forever. func TestWebhookMergeGroupPostsPassingChecks(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"staging", "production"}, map[string]api.RepoConfig{"octocat/hello-world": {}}, ) @@ -123,7 +138,7 @@ func TestWebhookMergeGroupPostsPassingChecks(t *testing.T) { // GitHub fires merge_group with "destroyed" when a PR leaves the queue. That // action needs no check run on any commit, so SchemaBot ignores it. func TestWebhookMergeGroupIgnoresNonChecksRequested(t *testing.T) { - h, created := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"octocat/hello-world": {}}) + h, created, _ := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"octocat/hello-world": {}}) rr := httptest.NewRecorder() h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "destroyed", "mergesha123", nil)) @@ -141,7 +156,7 @@ func TestWebhookMergeGroupIgnoresNonChecksRequested(t *testing.T) { // A merge_group event for a repository SchemaBot does not manage gets no check: // SchemaBot's check is not required on that repo, so there is nothing to unblock. func TestWebhookMergeGroupRejectsUnregisteredRepo(t *testing.T) { - h, created := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"org/allowed-repo": {}}) + h, created, _ := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"org/allowed-repo": {}}) rr := httptest.NewRecorder() h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) @@ -185,7 +200,7 @@ func TestWebhookMergeGroupUpdatesExistingCheck(t *testing.T) { }) installClient := ghclient.NewInstallationClientWithSlug(client, testLogger(), "schemabot") - service := api.New(nil, &api.ServerConfig{ + service := api.New(&foldStorage{checks: &foldCheckStore{}}, &api.ServerConfig{ AllowedEnvironments: []string{"production"}, Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, }, nil, testLogger()) @@ -216,7 +231,7 @@ func TestWebhookMergeGroupUpdatesExistingCheck(t *testing.T) { // With no environment scoping configured, SchemaBot publishes a single // non-environment-scoped aggregate check on the merge-group head SHA. func TestWebhookMergeGroupSingleAggregateWhenNoEnvScoping(t *testing.T) { - h, created := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"octocat/hello-world": {}}) + h, created, _ := mergeGroupTestHandler(t, nil, map[string]api.RepoConfig{"octocat/hello-world": {}}) rr := httptest.NewRecorder() h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) @@ -244,7 +259,7 @@ func TestWebhookMergeGroupSingleAggregateWhenNoEnvScoping(t *testing.T) { // Without this silence, every queue entry would re-grow the per-tenant check // rows the aggregate removes from PR heads. func TestWebhookMergeGroupParticipantStaysSilent(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": { Aggregate: &api.AggregateConfig{Role: api.AggregateRoleParticipant}, @@ -267,7 +282,7 @@ func TestWebhookMergeGroupParticipantStaysSilent(t *testing.T) { // The aggregate leader keeps posting its required checks on merge-group // commits — silence is participant-only, so the queue never wedges. func TestWebhookMergeGroupLeaderStillPosts(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": { Aggregate: &api.AggregateConfig{ @@ -291,3 +306,214 @@ func TestWebhookMergeGroupLeaderStillPosts(t *testing.T) { t.Fatal("timed out waiting for the leader's merge_group check run") } } + +// A queued PR whose stored check state turned blocking after it entered the +// queue — a preflight hold from a sibling change's in-flight apply on a shared +// target — must not merge on the verdict it queued with. The admission check +// re-folds the PR's stored state on the merge-group commit and blocks, so the +// queue removes the PR instead of merging it mid-apply. Because the blocking +// run lives on the synthetic merge-group commit and the queue never re-adds a +// PR by itself, the block also posts a guidance comment on the PR naming the +// blocked database and the re-queue step. +func TestWebhookMergeGroupBlocksWhenStoredCheckHolds(t *testing.T) { + h, created, comments := mergeGroupTestHandler(t, + []string{"production"}, + map[string]api.RepoConfig{"octocat/hello-world": {}}, + &storage.Check{ + Repository: "octocat/hello-world", + PullRequest: 1, + HeadSHA: "prhead123", + Environment: "production", + DatabaseType: "mysql", + DatabaseName: "widgets", + Status: "completed", + Conclusion: "action_required", + }, + ) + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) + require.Equal(t, http.StatusOK, rr.Code) + + select { + case c := <-created: + assert.Equal(t, "SchemaBot (production)", c.Name) + assert.Equal(t, "mergesha123", c.HeadSHA) + assert.Equal(t, "completed", c.Status) + assert.Equal(t, "action_required", c.Conclusion) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the blocking merge_group check run") + } + + select { + case body := <-comments: + assert.Contains(t, body, "Removed From Merge Queue") + assert.Contains(t, body, "`widgets` in `production`") + assert.Contains(t, body, "add it to the merge queue again") + assert.Contains(t, body, mergeQueueEjectedCommentMarker("mergesha123")) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the ejection guidance comment") + } +} + +// A redelivered merge_group event finds the ejection comment it already posted +// (identified by the merge-group head SHA marker) and does not post a +// duplicate; the blocking check still reconciles idempotently. +func TestWebhookMergeGroupEjectionCommentDeduplicatesRedelivery(t *testing.T) { + client, mux := setupGitHubServer(t) + created := make(chan checkRunCapture, 10) + mux.HandleFunc("POST /repos/octocat/hello-world/check-runs", func(w http.ResponseWriter, r *http.Request) { + var c checkRunCapture + require.NoError(t, json.NewDecoder(r.Body).Decode(&c)) + created <- c + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"id": 555}) + }) + mux.HandleFunc("GET /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode([]map[string]any{ + {"id": 777, "body": "guidance\n" + mergeQueueEjectedCommentMarker("mergesha123")}, + }) + }) + mux.HandleFunc("POST /repos/octocat/hello-world/issues/1/comments", func(w http.ResponseWriter, _ *http.Request) { + t.Error("redelivery must not post a duplicate ejection comment") + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(map[string]any{"id": 778}) + }) + + installClient := ghclient.NewInstallationClient(client, testLogger()) + service := api.New(&foldStorage{checks: &foldCheckStore{byPR: []*storage.Check{{ + Repository: "octocat/hello-world", + PullRequest: 1, + HeadSHA: "prhead123", + Environment: "production", + DatabaseType: "mysql", + DatabaseName: "widgets", + Status: "completed", + Conclusion: "action_required", + }}}}, &api.ServerConfig{ + AllowedEnvironments: []string{"production"}, + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + }, nil, testLogger()) + h := &Handler{ + service: service, + ghClients: ghclient.NewSingleClientSet(defaultAppName, &fakeClientFactory{client: installClient}), + logger: testLogger(), + } + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) + require.Equal(t, http.StatusOK, rr.Code) + + select { + case c := <-created: + assert.Equal(t, "action_required", c.Conclusion) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the blocking merge_group check run") + } +} + +// A queued PR with an apply currently running from its own head must not merge +// until the apply settles: an in-progress apply-owned stored row blocks the +// admission fold the same way it blocks the PR-head aggregate. +func TestWebhookMergeGroupBlocksWhenOwnApplyInFlight(t *testing.T) { + h, created, _ := mergeGroupTestHandler(t, + []string{"production"}, + map[string]api.RepoConfig{"octocat/hello-world": {}}, + &storage.Check{ + Repository: "octocat/hello-world", + PullRequest: 1, + HeadSHA: "prhead123", + Environment: "production", + DatabaseType: "mysql", + DatabaseName: "widgets", + Status: "in_progress", + ApplyID: 42, + }, + ) + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) + require.Equal(t, http.StatusOK, rr.Code) + + select { + case c := <-created: + assert.Equal(t, "action_required", c.Conclusion) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the blocking merge_group check run") + } +} + +// A merge group whose ref does not name a pull request cannot have its stored +// check state consulted, so admission fails closed: the check posts blocking +// rather than guessing that nothing blocks. +func TestWebhookMergeGroupFailsClosedOnUnidentifiablePR(t *testing.T) { + h, created, _ := mergeGroupTestHandler(t, + []string{"production"}, + map[string]api.RepoConfig{"octocat/hello-world": {}}, + ) + + payload := map[string]any{ + "action": "checks_requested", + "merge_group": map[string]any{ + "head_sha": "mergesha123", + "head_ref": "refs/heads/some-unexpected-ref", + }, + "repository": map[string]any{"full_name": "octocat/hello-world"}, + "installation": map[string]any{"id": 12345}, + } + body, err := json.Marshal(payload) + require.NoError(t, err) + req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/webhook", strings.NewReader(string(body))) + req.Header.Set("X-GitHub-Event", "merge_group") + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + require.Equal(t, http.StatusOK, rr.Code) + + select { + case c := <-created: + assert.Equal(t, "action_required", c.Conclusion) + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the fail-closed merge_group check run") + } +} + +// A storage read failure during admission is uncertainty, and uncertainty +// never admits a merge: the handler returns 500 so the delivery is retried +// rather than posting a check computed from unknown state. +func TestWebhookMergeGroupFailsClosedOnStorageError(t *testing.T) { + client, _ := setupGitHubServer(t) + installClient := ghclient.NewInstallationClient(client, testLogger()) + service := api.New(&foldStorage{checks: &foldCheckStore{byPRErr: assert.AnError}}, &api.ServerConfig{ + AllowedEnvironments: []string{"production"}, + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + }, nil, testLogger()) + h := &Handler{ + service: service, + ghClients: ghclient.NewSingleClientSet(defaultAppName, &fakeClientFactory{client: installClient}), + logger: testLogger(), + } + + rr := httptest.NewRecorder() + h.ServeHTTP(rr, buildMergeGroupWebhookRequest(t, "checks_requested", "mergesha123", nil)) + require.Equal(t, http.StatusInternalServerError, rr.Code) +} + +// mergeGroupPRNumber must identify the queued PR from the merge-queue branch +// ref GitHub generates, including base branches containing slashes, and refuse +// refs that do not name a PR. +func TestMergeGroupPRNumber(t *testing.T) { + pr, err := mergeGroupPRNumber("refs/heads/gh-readonly-queue/main/pr-123-0123abc") + require.NoError(t, err) + assert.Equal(t, 123, pr) + + pr, err = mergeGroupPRNumber("refs/heads/gh-readonly-queue/release/v1/pr-7-deadbeef") + require.NoError(t, err) + assert.Equal(t, 7, pr) + + _, err = mergeGroupPRNumber("refs/heads/feature-branch") + require.Error(t, err) + + _, err = mergeGroupPRNumber("") + require.Error(t, err) +} diff --git a/pkg/webhook/push_test.go b/pkg/webhook/push_test.go index 675d4a57a..d82b6ee7e 100644 --- a/pkg/webhook/push_test.go +++ b/pkg/webhook/push_test.go @@ -46,7 +46,7 @@ func buildPushWebhookRequest(t *testing.T, ref, after string, deleted bool) *htt // App selectable as a pinned required-check source — one check per gated // environment, with the same names as the PR-head aggregates. func TestWebhookPushPostsPassingChecksOnDefaultBranch(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"staging", "production"}, map[string]api.RepoConfig{"octocat/hello-world": {}}, ) @@ -81,7 +81,7 @@ func TestWebhookPushPostsPassingChecksOnDefaultBranch(t *testing.T) { // the PR and merge-queue check paths; only the default branch needs the // check-source seed. func TestWebhookPushIgnoresNonDefaultBranch(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": {}}, ) @@ -101,7 +101,7 @@ func TestWebhookPushIgnoresNonDefaultBranch(t *testing.T) { // A branch deletion push has no commit to publish a check on. func TestWebhookPushIgnoresBranchDeletion(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": {}}, ) @@ -123,7 +123,7 @@ func TestWebhookPushIgnoresBranchDeletion(t *testing.T) { // required aggregate — so a participant stays silent on default-branch pushes // rather than seeding an informational check on every landed commit. func TestWebhookPushParticipantStaysSilent(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": { Aggregate: &api.AggregateConfig{Role: api.AggregateRoleParticipant}, @@ -147,7 +147,7 @@ func TestWebhookPushParticipantStaysSilent(t *testing.T) { // default-branch pushes — silence is participant-only, so the leader App // stays selectable as a pinned required-check source. func TestWebhookPushLeaderStillPosts(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"octocat/hello-world": { Aggregate: &api.AggregateConfig{ @@ -175,7 +175,7 @@ func TestWebhookPushLeaderStillPosts(t *testing.T) { // A push on a repository SchemaBot does not manage gets no check: SchemaBot's // check is not required there, so there is no check source to maintain. func TestWebhookPushRejectsUnregisteredRepo(t *testing.T) { - h, created := mergeGroupTestHandler(t, + h, created, _ := mergeGroupTestHandler(t, []string{"production"}, map[string]api.RepoConfig{"org/allowed-repo": {}}, ) diff --git a/pkg/webhook/templates/merge_queue_ejected.go b/pkg/webhook/templates/merge_queue_ejected.go new file mode 100644 index 000000000..2f8d085b0 --- /dev/null +++ b/pkg/webhook/templates/merge_queue_ejected.go @@ -0,0 +1,50 @@ +package templates + +import ( + "fmt" + "strings" +) + +// MergeQueueBlockedTarget names one database whose stored check state blocked +// merge-queue admission for this PR. +type MergeQueueBlockedTarget struct { + Database string + Environment string +} + +// MergeQueueEjectedData describes a merge-queue admission block: the PR +// entered the queue, its stored check state turned out to be blocking, and the +// blocking admission check removed it from the queue. +type MergeQueueEjectedData struct { + // Blocking lists the databases whose stored check state blocked admission, + // deduplicated. May be empty when the blocking rows carry no database + // identifiers; the guidance stands on its own. + Blocking []MergeQueueBlockedTarget +} + +// RenderMergeQueueEjected renders the PR comment posted when the merge-queue +// admission check blocks a queued pull request. The blocking Check Run lives +// on the synthetic merge-group commit, not the PR head, so without this +// comment the author only sees the pull request silently leave the queue. +// It explains why, what clears on its own, and the one step that does not +// happen automatically: the queue never re-adds a pull request by itself. +func RenderMergeQueueEjected(data MergeQueueEjectedData) string { + var sb strings.Builder + sb.WriteString("## 🚦 Removed From Merge Queue\n\n") + sb.WriteString("This pull request's SchemaBot check state turned blocking after it entered the merge queue — ") + sb.WriteString("most often because another change's apply is in flight on a database this pull request also changes, ") + sb.WriteString("which invalidates the verdict it queued with. SchemaBot posted a blocking admission check on the ") + sb.WriteString("merge group, so the queue removed this pull request instead of merging it on a stale verdict.\n\n") + if len(data.Blocking) > 0 { + sb.WriteString("Blocking right now:\n\n") + for _, target := range data.Blocking { + fmt.Fprintf(&sb, "- `%s` in `%s`\n", + sanitizeInlineCode(target.Database), sanitizeInlineCode(target.Environment)) + } + sb.WriteString("\n") + } + sb.WriteString("**What happens next**\n\n") + sb.WriteString("- Check this pull request's SchemaBot check for the reason. A held check clears on its own: when the in-flight apply settles, SchemaBot re-plans this pull request and refreshes the check.\n") + sb.WriteString("- The merge queue does not re-add pull requests on its own — once this pull request's checks are green again, add it to the merge queue again.\n") + return sb.String() +}