diff --git a/docs/github-app-setup.md b/docs/github-app-setup.md index 3f93007a6..14a82b433 100644 --- a/docs/github-app-setup.md +++ b/docs/github-app-setup.md @@ -78,15 +78,18 @@ Under **Organization permissions**, grant: | Event | Purpose | |-------|---------| +| **Check run** | Receive `check_run.rerequested` so a human clicking **Re-run** on a SchemaBot check re-plans the PR head | +| **Check suite** | Receive `check_suite.requested` as a redundant auto-plan convergence signal: when the organic `pull_request` delivery for a push is lost upstream, SchemaBot recovers the missing auto-plan from the check suite instead of waiting for the next reconciler scan | | **Issue comment** | Receive `schemabot plan`, `schemabot help`, etc. from PR comments | | **Merge group** | Publish a passing SchemaBot check on a merge-queue commit so a required SchemaBot check does not block the merge queue (only needed if a repo uses a merge queue) | +| **Pull request** | Auto-plan each affected database on PR open/synchronize/reopen and clean up locks and stored check state on close | | **Push** | Publish a passing SchemaBot check on default-branch commits so branch rulesets can select the App as a pinned required-check source (rulesets only index Apps whose check suites ran against the target branch) | Subscribe to **Merge group**, not **Merge queue entry** — the two are distinct events, and SchemaBot handles only `merge_group`. The **Merge group** event requires the **Merge queues: Read** repository permission above. Because that is a new permission, adding it to an existing App marks the App as requesting new permissions, which an org or repository admin must approve on each installation before it takes effect. The **Push** event needs no new permission — **Contents: Read** above already unlocks it — so subscribing an existing App takes effect immediately on every installation, with no admin re-approval. Expect webhook delivery volume to rise once subscribed: GitHub sends a `push` event for every branch and tag push on installed repositories, and SchemaBot discards everything but default-branch pushes. -Future phases will also use **Check run** (action buttons) and **Pull request** (auto-plan on open/sync). +The **Check suite** event likewise needs no new permission — **Checks: Read & Write** above already unlocks it — so subscribing takes effect immediately. It also raises delivery volume (GitHub sends `check_suite.requested` for every push to every branch), and SchemaBot acknowledges and drops every suite that has no open PR at its head. Without this subscription the recovery feature receives no deliveries and is silently inert: lost auto-plans then wait for the reconciler's next scan instead of converging within the recovery grace. ### Where Can This GitHub App Be Installed? diff --git a/pkg/metrics/README.md b/pkg/metrics/README.md index 4344103ac..7e16e7772 100644 --- a/pkg/metrics/README.md +++ b/pkg/metrics/README.md @@ -31,6 +31,7 @@ available, such as `repository`, `github_app`, and `installation_id`. | `schemabot.webhook.inbox_oldest_claimable_age_seconds` | Gauge | environment | Age of the oldest ready-to-claim durable webhook inbox row | | `schemabot.webhook.inbox_stuck_processing` | Gauge | environment | Durable webhook inbox rows stuck in processing past the attempt cap | | `schemabot.webhook.inbox_stats_collection_failures` | Counter | environment | Failed durable webhook inbox metric snapshots (liveness signal for the inbox gauges) | +| `schemabot.webhook.check_suite_recovery_total` | Counter | environment, repository, outcome | Durable check_suite recovery outcomes (`covered`, `synthesized`, `resynthesized`, `already_queued` per candidate PR; `no_open_pr` per delivery) | | `schemabot.webhook.inbox_dispatch_lag_seconds` | Histogram | environment, event_type, repository | Time from webhook receipt to the delivery's first dispatch claim | | `schemabot.webhook.dispatch_duration_seconds` | Histogram | environment, event_type, outcome | Duration of one durable webhook dispatch claim by outcome (deliberately no repository label — see the ledger note below) | | `schemabot.github.requests_total` | Counter | environment, operation, category, resource, status, repository, github_app, installation_id | GitHub API responses observed by SchemaBot | @@ -102,7 +103,7 @@ available, such as `repository`, `github_app`, and `installation_id`. **status** (locks): `success`, `conflict`, `not_found`, `not_owned`, `error` -**event_type** (webhooks): `create`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, `pull_request_review_comment`, `check_run`, `ping`, `push` +**event_type** (webhooks): `create`, `issues`, `issue_comment`, `pull_request`, `pull_request_review`, `pull_request_review_comment`, `check_run`, `check_suite`, `ping`, `push` **action** (webhooks): common GitHub actions for the subscribed webhook events, such as `created`, `opened`, `synchronize`, `submitted`, `edited`, `closed`, `requested`, `completed` (omitted for events without actions like `ping` and `push`) diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index b8d91b595..25205f22d 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -1524,7 +1524,10 @@ var knownWebhookInboxStates = func() map[string]bool { // given state. A rising pending/processing depth means dispatch is falling // behind ingestion; a rising failed_retryable depth means deliveries are // retrying; a rising completed/failed depth means terminal rows are accumulating -// and retention has not reclaimed them. +// and retention has not reclaimed them. One caveat on pending: a row created +// with a not-before time sits in pending by design until it becomes due, so +// steady nonzero pending depth alone does not imply dispatch lag — cross-check +// the oldest-claimable-age gauge, which counts only dispatchable rows. func RecordWebhookInboxDepth(ctx context.Context, state string, count int64) { if !knownWebhookInboxStates[state] { state = "unknown" @@ -1537,9 +1540,11 @@ func RecordWebhookInboxDepth(ctx context.Context, state string, count int64) { } // RecordWebhookInboxOldestClaimableAge records how long the oldest -// ready-to-claim-but-unclaimed inbox row has been waiting, in seconds. It is the -// inbox's backlog latency: a value climbing past the dispatch cadence means work -// is acked but not being picked up. +// ready-to-claim-but-unclaimed inbox row has been claimable, in seconds. Age +// counts from when the row became claimable — the later of receipt and its +// not-before/retry time — so a deliberately deferred row does not report its +// grace period as backlog. It is the inbox's backlog latency: a value climbing +// past the dispatch cadence means work is acked but not being picked up. func RecordWebhookInboxOldestClaimableAge(ctx context.Context, age time.Duration) { recordGauge(ctx, "schemabot.webhook.inbox_oldest_claimable_age_seconds", int64(age.Seconds()), "Age in seconds of the oldest ready-to-claim durable webhook inbox row", "s", @@ -1571,16 +1576,19 @@ func RecordWebhookInboxStatsCollectionFailure(ctx context.Context) { } // RecordWebhookInboxDispatchLag records how long an accepted delivery waited -// in the durable inbox before a driver first claimed it. This is the -// per-delivery counterpart of the oldest-claimable-age gauge: the gauge shows -// the backlog's worst case right now, while this histogram shows the lag every -// delivery actually experienced. A distribution drifting past the dispatch -// poll cadence means accepted deliveries are waiting on driver capacity — -// investigate driver-pool sizing and claim-query health. One caveat: a -// shutdown-released claim refunds its attempt, so the delivery's reclaim -// counts as the first attempt again and records another sample measured from -// the original receipt — deploy churn therefore adds extra, longer samples -// without any driver-capacity problem. Unknown event types fold to "unknown" +// dispatchable in the durable inbox before a driver first claimed it — from +// when the row became eligible (receipt, or its not-before time for a +// deliberately deferred row), so a deferral's grace period never counts as +// lag. This is the per-delivery counterpart of the oldest-claimable-age +// gauge: the gauge shows the backlog's worst case right now, while this +// histogram shows the lag every delivery actually experienced. A distribution +// drifting past the dispatch poll cadence means accepted deliveries are +// waiting on driver capacity — investigate driver-pool sizing and claim-query +// health. One caveat: a shutdown-released claim refunds its attempt, so the +// delivery's reclaim counts as the first attempt again and records another +// sample measured from the original eligibility time — deploy churn therefore +// adds extra, longer samples without any driver-capacity problem. Unknown +// event types fold to "unknown" // and negative lags (cross-pod clock skew between the enqueueing and claiming // replica) clamp to zero so the histogram stays trustworthy. appName is the // resolved GitHub App name used only for fold-log attribution; it is not a @@ -1594,7 +1602,7 @@ func RecordWebhookInboxDispatchLag(ctx context.Context, appName, eventType, repo lag = 0 } recordHistogram(ctx, "schemabot.webhook.inbox_dispatch_lag_seconds", lag.Seconds(), - "Time from webhook receipt to the delivery's first dispatch claim", + "Time a webhook delivery spent dispatchable before its first dispatch claim", EnvironmentAttribute(""), attribute.String("event_type", eventType), attribute.String("repository", repo), @@ -1705,6 +1713,33 @@ func RecordWebhookReconcileSynthesizedEvent(ctx context.Context, repo string, re attribute.String("outcome", outcome)) } +// RecordWebhookCheckSuiteRecovery counts outcomes of durable +// check_suite.requested recovery processing. "covered", "synthesized", +// "resynthesized", and "already_queued" increment once per candidate PR (a +// delivery can carry several), while "no_open_pr" increments once per +// delivery when resolution finds no candidate at all; a delivery retried +// under its attempt budget re-counts PRs it already observed on an earlier +// attempt, so outcomes are per observation, not per unique PR. "covered" is +// the healthy steady state — the organic pull_request delivery arrived +// during the recovery grace and planned the head, so the redundant signal +// no-oped. "synthesized" means the auto-plan delivery for an open PR head +// was genuinely lost and the check_suite signal recovered it — investigate +// the upstream loss (edge auth, GitHub send failures), not the recovery. +// "resynthesized" means the recovery reopened a terminally failed +// synthesized row, so a sustained rate is the same head failing repeatedly +// after recovery — investigate that head's processing failure. +// "already_queued" means another recovery producer (the reconciler or an +// earlier check_suite delivery) got there first. "no_open_pr" means no PR +// named by the delivery was still open at the suite head — its PRs closed or +// moved on during the grace, or a fork head matched no open PR. +func RecordWebhookCheckSuiteRecovery(ctx context.Context, repo string, outcome string) { + addCounter(ctx, "schemabot.webhook.check_suite_recovery_total", + "Total number of outcomes from durable check_suite recovery processing", "{event}", + EnvironmentAttribute(""), + attribute.String("repository", repo), + attribute.String("outcome", outcome)) +} + // RecordWebhookReconcileStuckTerminated counts webhook inbox rows the // reconciler terminated because they were parked in processing with an expired // lease at the attempt cap — a driver hard-killed on its final attempt. A diff --git a/pkg/serve/serve.go b/pkg/serve/serve.go index f86c9c2b9..e314926d6 100644 --- a/pkg/serve/serve.go +++ b/pkg/serve/serve.go @@ -767,6 +767,7 @@ func buildSingleAppWebhookRuntime(serverConfig *api.ServerConfig, svc *api.Servi webhook.WithDurableWebhookDispatch(), webhook.WithWebhookReconciler(), }, webhookReconcileSynthesisOptions(logger)...) + handlerOpts = append(handlerOpts, checkSuiteRecoveryOptions(logger)...) handler := webhook.NewHandler(svc, ghClient, []byte(ghWebhookSecret), logger, handlerOpts...) svc.SetCheckRunBackfiller(handler) logger.Info("GitHub webhook endpoint registered", @@ -843,6 +844,7 @@ func buildMultiAppWebhookRuntime(serverConfig *api.ServerConfig, svc *api.Servic webhook.WithDurableWebhookDispatch(), webhook.WithWebhookReconciler(), }, webhookReconcileSynthesisOptions(logger)...) + handlerOpts = append(handlerOpts, checkSuiteRecoveryOptions(logger)...) handler := webhook.NewHandlerWithDispatch( svc, ghclient.NewClientSet(clients), @@ -895,6 +897,32 @@ func webhookReconcileSynthesisOptions(logger *slog.Logger) []webhook.HandlerOpti return []webhook.HandlerOption{webhook.WithWebhookReconcileSynthesis()} } +// checkSuiteRecoveryOptions returns the handler option enabling durable +// check_suite.requested recovery, unless the operator disabled it with +// WEBHOOK_CHECK_SUITE_RECOVERY=false. The kill switch makes the webhook +// endpoint acknowledge and ignore check_suite deliveries with a restart +// instead of a code revert; the reconciler's missing-head scan remains the +// recovery backstop either way. An unparseable value also disables recovery: +// the only reason to set the variable is to turn recovery off, so a malformed +// value ("off", "disabled") is treated as intent to disable — an operator +// reaching for a kill switch mid-incident must get the switched-off behavior, +// not a warning in pod logs they are not watching. +func checkSuiteRecoveryOptions(logger *slog.Logger) []webhook.HandlerOption { + if value := os.Getenv("WEBHOOK_CHECK_SUITE_RECOVERY"); value != "" { + enabled, err := strconv.ParseBool(value) + if err != nil { + logger.Error("invalid WEBHOOK_CHECK_SUITE_RECOVERY value; check-suite recovery disabled (fail-safe) — lost auto-plan deliveries recover only via the reconciler", + "value", value, "error", err) + return nil + } + if !enabled { + logger.Info("check-suite recovery disabled by WEBHOOK_CHECK_SUITE_RECOVERY; lost auto-plan deliveries recover only via the reconciler") + return nil + } + } + return []webhook.HandlerOption{webhook.WithCheckSuiteRecovery()} +} + // buildServerAuthorizer constructs the API authorizer exactly as the server // wires it: admin teams from PR command authorization, and the operator-group // union that widens forward-auth write admission. Every server build and any diff --git a/pkg/serve/serve_check_suite_recovery_test.go b/pkg/serve/serve_check_suite_recovery_test.go new file mode 100644 index 000000000..cd4c026ab --- /dev/null +++ b/pkg/serve/serve_check_suite_recovery_test.go @@ -0,0 +1,43 @@ +package serve + +import ( + "io" + "log/slog" + "testing" + + "github.com/stretchr/testify/require" +) + +// TestCheckSuiteRecoveryOptions pins the kill-switch contract: durable +// check-suite recovery defaults on, WEBHOOK_CHECK_SUITE_RECOVERY set to false +// drops the webhook endpoint to acknowledge-and-ignore for check_suite +// deliveries, and an unparseable value fails safe to disabled — setting the +// variable at all signals intent to turn recovery off, so a malformed value +// must not leave it running. +func TestCheckSuiteRecoveryOptions(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + for _, tc := range []struct { + name string + value string + enabled bool + }{ + {"default enabled when unset", "", true}, + {"explicitly enabled", "true", true}, + {"disabled", "false", false}, + {"invalid value fails safe to disabled", "not-a-bool", false}, + {"non-ParseBool disable spelling fails safe to disabled", "off", false}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("WEBHOOK_CHECK_SUITE_RECOVERY", tc.value) + + opts := checkSuiteRecoveryOptions(logger) + + if tc.enabled { + require.Len(t, opts, 1) + } else { + require.Empty(t, opts) + } + }) + } +} diff --git a/pkg/storage/internal/sqlstore/sql_helpers.go b/pkg/storage/internal/sqlstore/sql_helpers.go index 0e4345d50..499acdafa 100644 --- a/pkg/storage/internal/sqlstore/sql_helpers.go +++ b/pkg/storage/internal/sqlstore/sql_helpers.go @@ -6,6 +6,7 @@ import ( "database/sql" "errors" "log/slog" + "time" ) // rollbackTx rolls back tx, logging a warning if the rollback fails for a @@ -39,6 +40,14 @@ func nullInt64Ptr(v *int64) sql.NullInt64 { return sql.NullInt64{Int64: *v, Valid: true} } +// nullTimePtr returns a sql.NullTime for a *time.Time value. +func nullTimePtr(v *time.Time) sql.NullTime { + if v == nil { + return sql.NullTime{} + } + return sql.NullTime{Time: *v, Valid: true} +} + // nullJSON returns valid JSON from []byte, defaulting to "{}" if nil/empty. func nullJSON(b []byte) string { if len(b) == 0 { diff --git a/pkg/storage/internal/sqlstore/webhook_events.go b/pkg/storage/internal/sqlstore/webhook_events.go index e2a8bc4ca..fc62eb9f5 100644 --- a/pkg/storage/internal/sqlstore/webhook_events.go +++ b/pkg/storage/internal/sqlstore/webhook_events.go @@ -55,13 +55,18 @@ func (s *webhookEventStore) Create(ctx context.Context, event *storage.WebhookEv receivedAt = time.Now() } + // A caller-set RetryAfter is a not-before time: the row is durable + // immediately but stays invisible to FindNext until the time passes. + // Producers of deferred work — a redundant convergence signal that should + // lose the race to the primary delivery — use it to schedule dispatch + // without holding the delivery outside the inbox. id, err := s.identity.InsertID(ctx, s.db, ` INSERT INTO webhook_events ( provider, delivery_id, event, action, repository, pull_request, head_sha, tenant_id, - payload, state, attempts, received_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + payload, state, attempts, retry_after, received_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `, provider, event.DeliveryID, event.Event, event.Action, event.Repository, event.PullRequest, event.HeadSHA, event.TenantID, - payload, state, event.Attempts, receivedAt) + payload, state, event.Attempts, nullTimePtr(event.RetryAfter), receivedAt) if err != nil { if s.classifier.IsDuplicateKey(err) { return s.reopenTerminalWebhookEvent(ctx, provider, event.DeliveryID, payload, receivedAt) @@ -100,7 +105,9 @@ func (s *webhookEventStore) Create(ctx context.Context, event *storage.WebhookEv // // pending/retryable rows and processing rows with a live (unexpired) lease are // genuinely in flight, so they dedup (return false). last_error is kept for -// forensics until the next attempt overwrites it. +// forensics until the next attempt overwrites it. The reopen clears +// retry_after: Redeliver is an operator recovery lever, so the reopened row is +// claimable immediately rather than re-deferred. func (s *webhookEventStore) reopenTerminalWebhookEvent(ctx context.Context, provider, deliveryID, payload string, receivedAt time.Time) (bool, error) { result, err := s.db.ExecContext(ctx, ` UPDATE webhook_events @@ -134,15 +141,16 @@ func (s *webhookEventStore) GetByDeliveryID(ctx context.Context, provider, deliv } // webhookClaimablePredicate matches exactly the rows a driver would claim: a -// pending row, a retryable row whose retry window has elapsed and is under the -// attempt cap, or a processing row whose lease has expired and is under the -// attempt cap. FindNext and the InboxStats backlog-age query derive from this -// single source so the "ready to claim" definition cannot drift between what a -// driver picks up and what the backlog gauge measures. Bind its placeholders -// with webhookClaimableArgs. +// pending row whose not-before time (retry_after) is unset or has passed, a +// retryable row whose retry window has elapsed and is under the attempt cap, +// or a processing row whose lease has expired and is under the attempt cap. +// FindNext and the InboxStats backlog-age query derive from this single source +// so the "ready to claim" definition cannot drift between what a driver picks +// up and what the backlog gauge measures. Bind its placeholders with +// webhookClaimableArgs. func (s *webhookEventStore) webhookClaimablePredicate() string { return `( - state = ? + (state = ? AND (retry_after IS NULL OR retry_after <= ` + s.dialect.CurrentTimestamp(TimestampPrecisionDefault) + `)) OR (state = ? AND (retry_after IS NULL OR retry_after <= ` + s.dialect.CurrentTimestamp(TimestampPrecisionDefault) + `) AND attempts < ?) OR (state = ? AND lease_expires_at <= ` + s.dialect.CurrentTimestamp(TimestampPrecisionMicrosecond) + ` AND attempts < ?) )` @@ -275,6 +283,10 @@ func (s *webhookEventStore) FindNext(ctx context.Context, owner string, leaseDur event.LeaseOwner = owner event.LeaseToken = leaseToken event.LeaseExpiresAt = &leaseExpiresAt + event.ClaimableSince = event.ReceivedAt + if event.RetryAfter != nil && event.RetryAfter.After(event.ClaimableSince) { + event.ClaimableSince = *event.RetryAfter + } event.RetryAfter = nil if event.StartedAt == nil { event.StartedAt = &now @@ -340,7 +352,7 @@ func (s *webhookEventStore) MarkFailed(ctx context.Context, id int64, leaseToken completed_at = CASE WHEN ? THEN COALESCE(completed_at, NOW()) ELSE completed_at END, updated_at = NOW() WHERE id = ? AND lease_token = ? - `, state, nullString(errMsg), retryAfter, retryAfter == nil, id, leaseToken) + `, state, nullString(errMsg), nullTimePtr(retryAfter), retryAfter == nil, id, leaseToken) if err != nil { return fmt.Errorf("mark webhook event %d failed: %w", id, err) } @@ -471,10 +483,16 @@ func (s *webhookEventStore) InboxStats(ctx context.Context) (*storage.WebhookInb // what is claimable: a cap-exhausted retryable row is not counted (a driver // won't take it, so it isn't backlog), and an expired-lease processing row a // driver would reclaim is counted (real backlog when its driver crashed). - // NULL (nothing waiting) scans into a zero age. + // Age is measured from when the row became claimable — the later of + // receipt and its retry_after — so a row that spent time deliberately + // deferred (or waiting out a retry window) counts only its time spent + // claimable, not its grace period, as backlog. The COALESCE keeps GREATEST + // dialect-safe: MySQL's GREATEST returns NULL when any argument is NULL + // while PostgreSQL's ignores NULLs. NULL (nothing waiting) scans into a + // zero age. var oldestAgeSeconds sql.NullFloat64 err = s.db.QueryRowContext(ctx, ` - SELECT TIMESTAMPDIFF(MICROSECOND, MIN(received_at), `+s.dialect.CurrentTimestamp(TimestampPrecisionMicrosecond)+`) / 1e6 + SELECT TIMESTAMPDIFF(MICROSECOND, MIN(GREATEST(received_at, COALESCE(retry_after, received_at))), `+s.dialect.CurrentTimestamp(TimestampPrecisionMicrosecond)+`) / 1e6 FROM webhook_events WHERE `+s.webhookClaimablePredicate()+` `, webhookClaimableArgs()...).Scan(&oldestAgeSeconds) diff --git a/pkg/storage/internal/sqlstore/webhook_events_test.go b/pkg/storage/internal/sqlstore/webhook_events_test.go index 96fc9d724..21a579f29 100644 --- a/pkg/storage/internal/sqlstore/webhook_events_test.go +++ b/pkg/storage/internal/sqlstore/webhook_events_test.go @@ -232,6 +232,169 @@ func TestWebhookEventStore_FindNextClaimsOldestPendingEvent(t *testing.T) { assert.Equal(t, "delivery-2", next.DeliveryID) } +// A pending delivery created with a not-before time is durable immediately but +// invisible to dispatch until the time passes: a deferred producer — a +// redundant convergence signal that should lose the race to the primary +// delivery — must not have its row claimed early, while a past or unset +// not-before time leaves the row immediately claimable. The backlog gauge +// agrees: a deferred row is not backlog, because no driver would take it yet. +func TestWebhookEventStore_FindNextHonorsPendingNotBefore(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + future := time.Now().Add(time.Hour) + inserted, err := store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "deferred", Event: "check_suite", Payload: []byte(`{}`), RetryAfter: &future, + }) + require.NoError(t, err) + require.True(t, inserted) + + deferred, err := store.WebhookEvents().GetByDeliveryID(ctx, storage.WebhookProviderGitHub, "deferred") + require.NoError(t, err) + require.NotNil(t, deferred) + require.NotNil(t, deferred.RetryAfter, "the not-before time must be persisted with the row") + assert.WithinDuration(t, future, *deferred.RetryAfter, 2*time.Second) + + none, err := store.WebhookEvents().FindNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.Nil(t, none, "a pending row with a future not-before time must not be claimable") + + stats, err := store.WebhookEvents().InboxStats(ctx) + require.NoError(t, err) + assert.Zero(t, stats.OldestClaimableAge, "a deferred pending row is not claimable backlog") + assert.Equal(t, int64(1), stats.CountsByState[storage.WebhookEventPending]) + + // An older sibling whose not-before time has already elapsed is claimed + // while the deferred row stays invisible. + past := time.Now().Add(-time.Hour) + inserted, err = store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "due", Event: "check_suite", Payload: []byte(`{}`), RetryAfter: &past, + }) + require.NoError(t, err) + require.True(t, inserted) + + claimed, err := store.WebhookEvents().FindNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + assert.Equal(t, "due", claimed.DeliveryID, "an elapsed not-before time leaves the row claimable") + + none, err = store.WebhookEvents().FindNext(ctx, "driver-b", time.Minute) + require.NoError(t, err) + require.Nil(t, none, "the deferred row must stay invisible while its not-before time is in the future") + + // Once the not-before time elapses, the deferred row becomes ordinary + // pending backlog and is claimed like any other row. + _, err = testDB.ExecContext(ctx, ` + UPDATE webhook_events SET retry_after = NOW() - INTERVAL 1 SECOND + WHERE provider = ? AND delivery_id = ? + `, storage.WebhookProviderGitHub, "deferred") + require.NoError(t, err) + + ready, err := store.WebhookEvents().FindNext(ctx, "driver-b", time.Minute) + require.NoError(t, err) + require.NotNil(t, ready) + assert.Equal(t, "deferred", ready.DeliveryID, "an elapsed not-before time makes the deferred row claimable") + assert.Equal(t, storage.WebhookEventProcessing, ready.State) + assert.Nil(t, ready.RetryAfter, "the claim consumes the not-before time") + + persisted, err := store.WebhookEvents().GetByDeliveryID(ctx, storage.WebhookProviderGitHub, "deferred") + require.NoError(t, err) + require.NotNil(t, persisted) + assert.Nil(t, persisted.RetryAfter, "the claim consumes the persisted not-before time, not just the returned mirror") +} + +// A row that spent time deferred measures its backlog age and dispatch-lag +// basis from when it became claimable, not from receipt: the deferral's grace +// period is by design, and reporting it as backlog would spike the gauge to +// the full grace duration the instant the row becomes due. +func TestWebhookEventStore_DeferredRowAgeMeasuredFromDueTime(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + future := time.Now().Add(time.Hour) + inserted, err := store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "was-deferred", Event: "check_suite", Payload: []byte(`{}`), RetryAfter: &future, + }) + require.NoError(t, err) + require.True(t, inserted) + + // The row was received 60s ago and its not-before time elapsed 10s ago: + // only the 10s spent claimable is backlog, not the 50s grace period. + _, err = testDB.ExecContext(ctx, ` + UPDATE webhook_events + SET received_at = NOW() - INTERVAL 60 SECOND, retry_after = NOW() - INTERVAL 10 SECOND + WHERE provider = ? AND delivery_id = ? + `, storage.WebhookProviderGitHub, "was-deferred") + require.NoError(t, err) + + stats, err := store.WebhookEvents().InboxStats(ctx) + require.NoError(t, err) + assert.Greater(t, stats.OldestClaimableAge, 5*time.Second, "age must count the time spent claimable") + assert.Less(t, stats.OldestClaimableAge, 40*time.Second, "age must not count the deferral's grace period since receipt") + + claimed, err := store.WebhookEvents().FindNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + assert.WithinDuration(t, time.Now().Add(-10*time.Second), claimed.ClaimableSince, 5*time.Second, + "the claimed event carries when it became dispatchable, not its receipt time") + + // A row that was never deferred is dispatchable from receipt. + inserted, err = store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "never-deferred", Event: "check_suite", Payload: []byte(`{}`), + }) + require.NoError(t, err) + require.True(t, inserted) + + prompt, err := store.WebhookEvents().FindNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, prompt) + require.Equal(t, "never-deferred", prompt.DeliveryID) + assert.WithinDuration(t, prompt.ReceivedAt, prompt.ClaimableSince, 2*time.Second, + "an undeferred row is dispatchable from receipt") +} + +// A duplicate Create whose GUID matches a terminally failed row reopens that +// row immediately claimable even when the incoming event carries a future +// not-before time: redelivery is an operator recovery lever, so the reopened +// row must never be re-deferred behind the very deferral whose loss it +// recovers from. +func TestWebhookEventStore_CreateReopenIgnoresIncomingNotBefore(t *testing.T) { + clearTables(t) + ctx := t.Context() + store := New(testDB) + + inserted, err := store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "delivery-1", Event: "pull_request", Payload: []byte(`{"attempt":1}`), + }) + require.NoError(t, err) + require.True(t, inserted) + + claimed, err := store.WebhookEvents().FindNext(ctx, "driver-a", time.Minute) + require.NoError(t, err) + require.NotNil(t, claimed) + require.NoError(t, store.WebhookEvents().MarkFailed(ctx, claimed.ID, claimed.LeaseToken, "permanent failure", nil)) + + future := time.Now().Add(time.Hour) + inserted, err = store.WebhookEvents().Create(ctx, &storage.WebhookEvent{ + DeliveryID: "delivery-1", Event: "pull_request", Payload: []byte(`{"attempt":2}`), RetryAfter: &future, + }) + require.NoError(t, err) + require.True(t, inserted) + + reopened, err := store.WebhookEvents().GetByDeliveryID(ctx, storage.WebhookProviderGitHub, "delivery-1") + require.NoError(t, err) + require.NotNil(t, reopened) + assert.Equal(t, storage.WebhookEventPending, reopened.State) + assert.Nil(t, reopened.RetryAfter, "the reopen discards the incoming not-before time") + + reclaimed, err := store.WebhookEvents().FindNext(ctx, "driver-b", time.Minute) + require.NoError(t, err) + require.NotNil(t, reclaimed) + assert.Equal(t, "delivery-1", reclaimed.DeliveryID, "the reopened row is immediately claimable") +} + // A delivery's claimability must not depend on its payload width. Ordering the // claimable set must keep the payload out of the sort, so a single delivery // whose payload exceeds the server's sort buffer is claimed like any other row diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go index 3e8e4d499..b060df1b5 100644 --- a/pkg/storage/storage.go +++ b/pkg/storage/storage.go @@ -224,6 +224,17 @@ type WebhookEventStore interface { // payload) and returns inserted=true, so GitHub's "Redeliver" button — // which reuses the original delivery GUID — is a real remediation for a // terminally failed delivery instead of a permanent no-op. + // + // A non-nil RetryAfter on the event is persisted as a not-before time: the + // delivery is durable immediately but FindNext will not claim it until the + // time passes. Producers of deferred work — a redundant convergence signal + // that should lose the race to the primary delivery — set it to schedule + // dispatch; nil means immediately claimable. A redelivery reopen clears it. + // + // Only a fresh insert populates event.ID; a reopen returns inserted=true + // with event.ID left zero, so callers whose behavior differs between the + // two — a deferred producer whose not-before time a reopen discards, for + // example — can tell them apart. Create(ctx context.Context, event *WebhookEvent) (inserted bool, err error) // GetByDeliveryID returns a webhook event by provider + delivery GUID, or nil if not found. @@ -246,13 +257,23 @@ type WebhookEventStore interface { // FindNext atomically claims one pending, retryable, or lease-expired event. // The claim rotates lease_owner/lease_token, increments attempts, and sets a - // lease expiry in the same transaction. Retryable and lease-expired rows are - // only reclaimed while attempts < MaxWebhookEventAttempts, so a poison event - // cannot be reclaimed forever. Returns nil when no event is claimable. + // lease expiry in the same transaction. A pending row with a future + // retry_after (not-before time) is not claimable until it passes. Retryable + // and lease-expired rows are only reclaimed while attempts < + // MaxWebhookEventAttempts, so a poison event cannot be reclaimed forever. + // Returns nil when no event is claimable. + // + // The claim consumes retry_after (the persisted row's is cleared); the + // returned event carries ClaimableSince — the later of receipt and the + // consumed not-before time — so the dispatcher can measure dispatch lag + // from when the row became eligible rather than from receipt. // - // Ordering is currently global FIFO (created_at, id). A contemplated - // evolution is per-(repository, pull_request) claiming with coalescing of - // superseded deliveries; callers should not depend on cross-key ordering. + // Ordering is currently global FIFO (created_at, id). A row that spent + // time deferred re-enters dispatch at its original insertion position once + // due — ahead of rows created during its deferral — not at its due time. A + // contemplated evolution is per-(repository, pull_request) claiming with + // coalescing of superseded deliveries; callers should not depend on + // cross-key ordering. FindNext(ctx context.Context, owner string, leaseDuration time.Duration) (*WebhookEvent, error) // Heartbeat extends the current lease. Returns ErrWebhookEventNotFound when diff --git a/pkg/storage/types.go b/pkg/storage/types.go index 3fe550e75..a34607ad2 100644 --- a/pkg/storage/types.go +++ b/pkg/storage/types.go @@ -1380,8 +1380,12 @@ type WebhookInboxStats struct { CountsByState map[string]int64 // OldestClaimableAge is how long the oldest ready-to-claim-but-unclaimed row - // (pending, or retryable with an elapsed retry window) has been waiting. It - // is the inbox's backlog latency; zero when nothing is waiting. + // (pending with no future not-before time, or retryable with an elapsed + // retry window) has been claimable. Age is measured from when the row + // became claimable — the later of receipt and its retry_after — not from + // receipt, so a row that spent time deliberately deferred does not report + // its grace period as backlog. It is the inbox's backlog latency; zero + // when nothing is waiting. OldestClaimableAge time.Duration // StuckProcessing is the number of rows wedged in processing with an expired @@ -1407,6 +1411,12 @@ type WebhookEvent struct { LeaseToken string LeaseExpiresAt *time.Time RetryAfter *time.Time + // ClaimableSince is when the row became eligible for dispatch: the later + // of receipt and its not-before time (retry_after). It is not a column — + // FindNext derives it on the returned event because the claim consumes + // retry_after, and the dispatcher needs the original eligibility time to + // measure dispatch lag without counting a deliberate deferral as backlog. + ClaimableSince time.Time LastError string ReceivedAt time.Time StartedAt *time.Time diff --git a/pkg/webhook/README.md b/pkg/webhook/README.md index bbed947be..e597065b5 100644 --- a/pkg/webhook/README.md +++ b/pkg/webhook/README.md @@ -67,9 +67,16 @@ GitHub sends POST /webhook | | | +-- Release locks held by the PR and delete stored check state | - +-- check_run + +-- check_run.rerequested + | | + | +-- Re-plan the PR head when a human clicks "Re-run" on a + | SchemaBot check + | + +-- check_suite.requested | - +-- Accepted by the webhook but not implemented today + +-- Enqueue a grace-deferred recovery pass that synthesizes the + auto-plan delivery for open PR heads whose pull_request + delivery was lost ``` ## GitHub App Authentication diff --git a/pkg/webhook/check_suite.go b/pkg/webhook/check_suite.go new file mode 100644 index 000000000..842eafbba --- /dev/null +++ b/pkg/webhook/check_suite.go @@ -0,0 +1,354 @@ +// check_suite.go feeds GitHub's check_suite.requested delivery into the +// durable inbox as a redundant convergence signal. For every push to a PR +// branch, a GitHub App with check permissions receives two independent +// deliveries seconds apart: pull_request (what auto-plan acts on) and +// check_suite.requested. When the pull_request delivery is lost upstream of +// the inbox, the check suite sits queued with zero check runs and the PR is +// blocked with no recourse until the reconciler's next scan. The check_suite +// delivery is GitHub's purpose-built "populate your check runs for this SHA +// now" signal — handling it converges a lost auto-plan in minutes instead of +// waiting for the reconcile interval. +// +// The signal is deliberately deferred, not raced: the ingress enqueues the +// delivery with a not-before time (the recovery grace), so the organic +// pull_request delivery — which almost always arrives and plans within +// seconds — wins. When the grace passes, processing re-resolves the suite +// head against GitHub's *current* PR state and synthesizes a recovery +// delivery only for open PRs still at that head whose auto-plan coverage is +// missing from the inbox. Recovery rows use the same deterministic GUID as +// the reconciler's missing-head synthesis, so the two recovery producers +// dedupe against each other naturally. +package webhook + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "time" + + ghclient "github.com/block/schemabot/pkg/github" + "github.com/block/schemabot/pkg/metrics" + "github.com/block/schemabot/pkg/storage" +) + +// checkSuiteRecoveryAction is the only check_suite action the recovery path +// consumes. GitHub sends "requested" on every push; "rerequested" (a human +// clicked Re-run on the suite) already re-plans through check_run.rerequested, +// and "completed" carries no work. +const checkSuiteRecoveryAction = "requested" + +// defaultCheckSuiteRecoveryGrace is how long a check_suite delivery waits in +// the inbox before it becomes claimable. The organic pull_request delivery +// for the same push normally arrives within seconds, plans the head, and the +// recovery pass then no-ops on coverage. The grace only delays recovery of a +// genuinely lost delivery, so it trades a few minutes of convergence lag for +// never racing the primary signal. +const defaultCheckSuiteRecoveryGrace = 2 * time.Minute + +// checkSuitePayload represents the relevant fields from a GitHub check_suite +// webhook. HeadBranch is null (decoded as empty) when the suite head lives in +// a fork, and PullRequests is empty in the same cases, so processing falls +// back to resolving PRs by head SHA via the API. +type checkSuitePayload struct { + Action string `json:"action"` + CheckSuite struct { + HeadSHA string `json:"head_sha"` + HeadBranch string `json:"head_branch"` + PullRequests []struct { + Number int `json:"number"` + } `json:"pull_requests"` + } `json:"check_suite"` + Repository struct { + FullName string `json:"full_name"` + } `json:"repository"` + Installation struct { + ID int64 `json:"id"` + } `json:"installation"` +} + +// handleCheckSuite enqueues a check_suite.requested delivery into the durable +// inbox with a not-before time and ACKs fast. All resolution work — which PRs +// the suite head belongs to, whether their auto-plan coverage exists — runs +// later under a driver lease, after the grace has given the organic +// pull_request delivery time to win. +func (h *Handler) handleCheckSuite(ctx context.Context, metricApp string, w http.ResponseWriter, body []byte, deliveryID string) { + var payload checkSuitePayload + if err := json.Unmarshal(body, &payload); err != nil { + h.writeError(w, http.StatusBadRequest, "invalid check_suite payload") + return + } + repo := payload.Repository.FullName + headSHA := payload.CheckSuite.HeadSHA + + if payload.Action != checkSuiteRecoveryAction { + h.logger.Debug("check_suite delivery ignored because action carries no recovery work", + "action", payload.Action, "repo", repo, "delivery_id", deliveryID) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite action ignored"}) + return + } + if !h.checkSuiteRecovery || !h.durableWebhookDispatch { + h.logger.Debug("check_suite delivery ignored because check-suite recovery is disabled", + "repo", repo, "head_sha", headSHA, "delivery_id", deliveryID, + "recovery_enabled", h.checkSuiteRecovery, "durable_dispatch", h.durableWebhookDispatch) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite recovery disabled"}) + return + } + if repo == "" || headSHA == "" { + h.writeError(w, http.StatusBadRequest, "check_suite payload missing repository or head SHA") + return + } + if h.service != nil && !h.service.Config().IsRepoAllowed(repo) { + h.logger.Warn("webhook from unregistered repository", + "event", "check_suite", "action", payload.Action, "repo", repo, + "delivery_id", deliveryID) + metrics.RecordUnregisteredRepositoryWebhook(ctx, metricApp, "check_suite", payload.Action, repo) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "repository not registered"}) + return + } + // check_suite.requested fires for every push to every branch, but GitHub + // names the open PRs at a same-repository head in the payload when it + // sends the delivery, so a non-fork suite with an empty list means no + // open PR existed at delivery time and there is nothing to recover. + // Skipping it here keeps the dominant non-PR traffic — pushes to branches + // (including the default branch) without an open PR — from occupying + // inbox rows and grace-delayed processing passes. A PR opened after the + // push produces its own pull_request.opened delivery, and the + // reconciler's missing-head scan backstops the exotic residue. Fork heads + // arrive with an empty head_branch and an empty list, so they pass + // through to processing's API-resolution fallback. + if payload.CheckSuite.HeadBranch != "" && len(payload.CheckSuite.PullRequests) == 0 { + h.logger.Debug("check_suite delivery skipped because no open PR existed at the suite head", + "repo", repo, "head_sha", headSHA, "branch", payload.CheckSuite.HeadBranch, + "delivery_id", deliveryID) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite without an open PR skipped"}) + return + } + + installationID := h.effectiveInstallationID(ctx, payload.Installation.ID) + if installationID == 0 { + h.writeError(w, http.StatusBadRequest, "missing installation ID in webhook payload") + return + } + + retryAfter := time.Now().Add(h.checkSuiteRecoveryGrace) + event := &storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: deliveryID, + Event: "check_suite", + Action: payload.Action, + Repository: repo, + HeadSHA: headSHA, + TenantID: strconv.FormatInt(installationID, 10), + Payload: body, + RetryAfter: &retryAfter, + } + inserted, err := h.enqueueDurableWebhookEvent(ctx, event) + if err != nil { + h.logger.Error("failed to enqueue durable check_suite recovery", + "repo", repo, "head_sha", headSHA, "delivery_id", deliveryID, "error", err) + metrics.RecordWebhookEvent(ctx, metricApp, "check_suite", payload.Action, repo, "durable_enqueue_failed") + h.writeError(w, http.StatusInternalServerError, "failed to enqueue webhook delivery") + return + } + if !inserted { + h.logger.Info("durable check_suite delivery already queued", + "repo", repo, "head_sha", headSHA, "delivery_id", deliveryID) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite recovery already queued"}) + return + } + if event.ID == 0 { + // A zero ID after inserted=true means Create reopened a redelivered + // terminal row rather than inserting a fresh one, and a reopen clears + // the not-before time: Redeliver is an operator recovery lever, so + // the reopened row is claimable immediately. + h.logger.Info("durable check_suite recovery reopened by redelivery and claimable immediately", + "repo", repo, "head_sha", headSHA, "delivery_id", deliveryID) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite recovery queued"}) + return + } + h.logger.Info("durable check_suite recovery queued", + "repo", repo, "head_sha", headSHA, "delivery_id", deliveryID, + "not_before", retryAfter.UTC().Format(time.RFC3339)) + h.writeJSON(w, http.StatusOK, map[string]string{"message": "check_suite recovery queued"}) +} + +// processDurableCheckSuite drives a claimed check_suite delivery: it resolves +// which open PRs currently sit at the suite's head SHA and synthesizes a +// recovery delivery for each one whose auto-plan coverage is missing from the +// inbox. In the steady state — the organic pull_request delivery arrived and +// planned during the grace — every candidate is covered and the pass no-ops. +func (h *Handler) processDurableCheckSuite(ctx context.Context, event *storage.WebhookEvent) (retry bool, err error) { + var payload checkSuitePayload + if err := json.Unmarshal(event.Payload, &payload); err != nil { + return false, fmt.Errorf("decode durable check_suite delivery %s: %w", event.DeliveryID, err) + } + // Re-validate the action fail-closed: rows can arrive via replay or a + // future producer, and a non-requested action carries no recovery work. + if payload.Action != checkSuiteRecoveryAction { + h.logger.Info("durable check_suite delivery ignored because action needs no work", + "delivery_id", event.DeliveryID, "action", payload.Action, "repo", event.Repository) + return false, nil + } + // The kill switch must stop synthesis with a restart, including rows + // enqueued before the operator flipped it, so re-validate the flag here + // and not just at ingress. + if !h.checkSuiteRecovery { + h.logger.Info("durable check_suite delivery ignored because check-suite recovery is disabled", + "delivery_id", event.DeliveryID, "repo", event.Repository, "head_sha", event.HeadSHA) + return false, nil + } + repo := event.Repository + headSHA := event.HeadSHA + if repo == "" || headSHA == "" { + return false, fmt.Errorf("durable check_suite delivery %s missing repo or head SHA", event.DeliveryID) + } + // The allowlist can change between enqueue and claim; re-validate so a + // deregistered repository cannot have work synthesized for it. + if h.service != nil && !h.service.Config().IsRepoAllowed(repo) { + h.logger.Warn("durable check_suite delivery from unregistered repository", + "delivery_id", event.DeliveryID, "repo", repo, "head_sha", headSHA) + metrics.RecordUnregisteredRepositoryWebhook(ctx, h.metricAppForRepo(repo), "check_suite", payload.Action, repo) + return false, nil + } + installationID, err := durableInstallationID(event) + if err != nil { + return false, err + } + client, err := h.clientForRepo(repo, installationID) + if err != nil { + return true, fmt.Errorf("create GitHub client for durable check_suite delivery %s (%s@%s): %w", + event.DeliveryID, repo, headSHA, err) + } + store := h.webhookEventStore() + if store == nil { + return true, fmt.Errorf("webhook event storage is unavailable for durable check_suite delivery %s", event.DeliveryID) + } + + candidates, retryable, err := h.resolveCheckSuitePRs(ctx, client, repo, headSHA, payload) + if err != nil { + return retryable, err + } + if len(candidates) == 0 { + h.logger.Info("durable check_suite delivery matched no open PR at the suite head", + "delivery_id", event.DeliveryID, "repo", repo, "head_sha", headSHA) + metrics.RecordWebhookCheckSuiteRecovery(ctx, repo, "no_open_pr") + return false, nil + } + + var synthesized int + for _, pr := range candidates { + covered, err := store.HasEventForHead(ctx, storage.WebhookProviderGitHub, repo, pr, headSHA) + if err != nil { + return true, fmt.Errorf("query inbox coverage for %s#%d@%s (durable check_suite delivery %s): %w", + repo, pr, headSHA, event.DeliveryID, err) + } + if covered { + h.logger.Debug("check_suite recovery skipped PR head with auto-plan coverage", + "delivery_id", event.DeliveryID, "repo", repo, "pr", pr, "head_sha", headSHA) + metrics.RecordWebhookCheckSuiteRecovery(ctx, repo, "covered") + continue + } + inserted, resynthesized, err := h.synthesizeMissingHeadDelivery(ctx, repo, pr, headSHA, installationID) + if err != nil { + return true, fmt.Errorf("synthesize recovery delivery for %s#%d@%s (durable check_suite delivery %s): %w", + repo, pr, headSHA, event.DeliveryID, err) + } + if !inserted { + // A live recovery row for this head already exists — the + // reconciler (or an earlier check_suite delivery) got there + // first and the dispatcher will plan it. + h.logger.Debug("check_suite recovery found an already-queued recovery delivery", + "delivery_id", event.DeliveryID, "repo", repo, "pr", pr, "head_sha", headSHA) + metrics.RecordWebhookCheckSuiteRecovery(ctx, repo, "already_queued") + continue + } + synthesized++ + outcome := "synthesized" + if resynthesized { + outcome = "resynthesized" + } + h.logger.Warn("check_suite recovery synthesized inbox delivery for open PR head missing its auto-plan delivery", + "delivery_id", event.DeliveryID, "repo", repo, "pr", pr, "head_sha", headSHA, + "resynthesized", resynthesized) + metrics.RecordWebhookCheckSuiteRecovery(ctx, repo, outcome) + } + h.logger.Info("durable check_suite recovery pass completed", + "delivery_id", event.DeliveryID, "repo", repo, "head_sha", headSHA, + "candidate_prs", len(candidates), "synthesized", synthesized) + return false, nil +} + +// resolveCheckSuitePRs resolves the open PRs whose *current* head is the +// check suite's head SHA. The payload's pull_requests entries are +// point-in-time snapshots taken when GitHub sent the delivery — and the +// recovery grace means minutes have passed — so each is re-fetched and kept +// only when the PR is still open at the same head; a moved or closed PR is +// skipped because a fresher signal owns its new state. Only a fork head — an +// empty head_branch, whose payload cannot name PRs — falls back to a bounded +// walk of the repository's open PRs matched by head SHA, whose listing is +// current at call time. +func (h *Handler) resolveCheckSuitePRs(ctx context.Context, client *ghclient.InstallationClient, repo, headSHA string, payload checkSuitePayload) (prs []int, retry bool, err error) { + if len(payload.CheckSuite.PullRequests) > 0 { + for _, candidate := range payload.CheckSuite.PullRequests { + if candidate.Number <= 0 { + h.logger.Debug("check_suite recovery skipped payload PR entry without a number", + "repo", repo, "head_sha", headSHA) + continue + } + info, err := client.FetchPullRequestNoCache(ctx, repo, candidate.Number) + if err != nil { + return nil, true, fmt.Errorf("fetch pull request %s#%d for check_suite head %s: %w", + repo, candidate.Number, headSHA, err) + } + if info.IsClosed() { + h.logger.Debug("check_suite recovery skipped closed PR", + "repo", repo, "pr", candidate.Number, "head_sha", headSHA) + continue + } + if info.HeadSHA != headSHA { + h.logger.Debug("check_suite recovery skipped PR whose head moved past the suite head", + "repo", repo, "pr", candidate.Number, "suite_head_sha", headSHA, + "current_head_sha", info.HeadSHA) + continue + } + prs = append(prs, candidate.Number) + } + return prs, false, nil + } + // GitHub names the open PRs for a same-repository head in the payload, so + // an empty list on a non-fork suite (a non-empty head_branch) means no + // open PR existed when the delivery was sent — a PR opened later produces + // its own pull_request.opened delivery. The API walk below exists only + // for fork heads, whose payload cannot name PRs. + if payload.CheckSuite.HeadBranch != "" { + h.logger.Debug("check_suite recovery resolved no PRs because the same-repository suite head had no open PR at delivery time", + "repo", repo, "head_sha", headSHA, "head_branch", payload.CheckSuite.HeadBranch) + return nil, false, nil + } + page := 1 + for range h.webhookReconcileMaxPages { + open, nextPage, _, err := client.ListOpenPullRequestsPage(ctx, repo, page, webhookReconcilePageSize) + if err != nil { + return nil, true, fmt.Errorf("list open pull requests for %s while resolving check_suite head %s: %w", + repo, headSHA, err) + } + for _, pr := range open { + if pr.HeadSHA == headSHA { + prs = append(prs, pr.Number) + } + } + if nextPage == 0 { + return prs, false, nil + } + page = nextPage + } + // The page budget ran out before the listing was exhausted, so a matching + // PR beyond the budget goes unresolved this delivery. The reconciler's + // missing-head scan remains the backstop for anything missed here. + h.logger.Warn("check_suite recovery exhausted its page budget while resolving the suite head; open PR coverage for this head is truncated", + "repo", repo, "head_sha", headSHA, "max_pages", h.webhookReconcileMaxPages, + "page_size", webhookReconcilePageSize) + return prs, false, nil +} diff --git a/pkg/webhook/check_suite_test.go b/pkg/webhook/check_suite_test.go new file mode 100644 index 000000000..1344571bc --- /dev/null +++ b/pkg/webhook/check_suite_test.go @@ -0,0 +1,654 @@ +package webhook + +import ( + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/schemabot/pkg/api" + ghclient "github.com/block/schemabot/pkg/github" + "github.com/block/schemabot/pkg/storage" +) + +// buildCheckSuiteWebhookRequest constructs a check_suite webhook POST request +// for octocat/hello-world, whose default branch is main. +func buildCheckSuiteWebhookRequest(t *testing.T, action, headSHA, headBranch string, prNumbers ...int) *http.Request { + t.Helper() + + prs := make([]map[string]any, 0, len(prNumbers)) + for _, n := range prNumbers { + prs = append(prs, map[string]any{"number": n}) + } + payload := map[string]any{ + "action": action, + "check_suite": map[string]any{ + "head_sha": headSHA, + "head_branch": headBranch, + "pull_requests": prs, + }, + "repository": map[string]any{ + "full_name": "octocat/hello-world", + "default_branch": "main", + }, + "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", "check_suite") + return req +} + +// newCheckSuiteIngressHandler builds a recovery-enabled handler with no +// GitHub server: the ingress path must never touch the GitHub API. +func newCheckSuiteIngressHandler(t *testing.T, store storage.WebhookEventStore, repos map[string]api.RepoConfig, opts ...HandlerOption) *Handler { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + service := api.New(&durableWebhookTestStorage{webhookEvents: store}, &api.ServerConfig{Repos: repos}, nil, logger) + factory := &fakeClientFactory{} + return NewHandler(service, factory, nil, logger, + append([]HandlerOption{WithDurableWebhookDispatch(), WithCheckSuiteRecovery()}, opts...)...) +} + +// newCheckSuiteProcessHandler builds a recovery-enabled handler whose GitHub +// client talks to a fake server; register PR responses on the returned mux. +func newCheckSuiteProcessHandler(t *testing.T, store storage.WebhookEventStore, repos map[string]api.RepoConfig, opts ...HandlerOption) (*Handler, *http.ServeMux) { + t.Helper() + ghc, mux := setupGitHubServer(t) + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + service := api.New(&durableWebhookTestStorage{webhookEvents: store}, &api.ServerConfig{Repos: repos}, nil, logger) + factory := &fakeClientFactory{client: ghclient.NewInstallationClient(ghc, logger)} + h := NewHandler(service, factory, nil, logger, + append([]HandlerOption{WithDurableWebhookDispatch(), WithCheckSuiteRecovery()}, opts...)...) + return h, mux +} + +// durableCheckSuiteEvent builds a claimed inbox row as the ingress path +// stores it: a check_suite delivery for octocat/hello-world with the given +// payload PR snapshot. +func durableCheckSuiteEvent(t *testing.T, headSHA string, prNumbers ...int) *storage.WebhookEvent { + t.Helper() + + prs := make([]map[string]any, 0, len(prNumbers)) + for _, n := range prNumbers { + prs = append(prs, map[string]any{"number": n}) + } + payload, err := json.Marshal(map[string]any{ + "action": "requested", + "check_suite": map[string]any{ + "head_sha": headSHA, + "head_branch": "feature", + "pull_requests": prs, + }, + "repository": map[string]any{ + "full_name": "octocat/hello-world", + "default_branch": "main", + }, + "installation": map[string]any{"id": 12345}, + }) + require.NoError(t, err) + return &storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: "delivery-check-suite-1", + Event: "check_suite", + Action: "requested", + Repository: "octocat/hello-world", + HeadSHA: headSHA, + TenantID: "12345", + Payload: payload, + } +} + +// durableCheckSuiteForkEvent builds a claimed inbox row for a fork-headed +// suite: head_branch is null in the payload and GitHub cannot name the PRs, +// so processing must resolve them via the open-PR listing. +func durableCheckSuiteForkEvent(t *testing.T, headSHA string) *storage.WebhookEvent { + t.Helper() + + payload, err := json.Marshal(map[string]any{ + "action": "requested", + "check_suite": map[string]any{ + "head_sha": headSHA, + "head_branch": "", + "pull_requests": []map[string]any{}, + }, + "repository": map[string]any{ + "full_name": "octocat/hello-world", + }, + "installation": map[string]any{"id": 12345}, + }) + require.NoError(t, err) + return &storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: "delivery-check-suite-1", + Event: "check_suite", + Action: "requested", + Repository: "octocat/hello-world", + HeadSHA: headSHA, + TenantID: "12345", + Payload: payload, + } +} + +// writeSinglePR responds to a GET /repos/{repo}/pulls/{n} fetch with a PR in +// the given lifecycle state at the given current head. +func writeSinglePR(t *testing.T, w http.ResponseWriter, number int, state, headSHA string) { + t.Helper() + w.Header().Set("Content-Type", "application/json") + require.NoError(t, json.NewEncoder(w).Encode(map[string]any{ + "number": number, + "state": state, + "merged": state == "closed", + "head": map[string]any{"sha": headSHA, "ref": "feature"}, + "base": map[string]any{"ref": "main"}, + "user": map[string]any{"login": "octocat"}, + })) +} + +// coveringAutoPlanRow is an organic pull_request delivery that already plans +// the head, so recovery must no-op on it. +func coveringAutoPlanRow(t *testing.T, store storage.WebhookEventStore, pr int, headSHA string) { + t.Helper() + _, err := store.Create(t.Context(), &storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: "delivery-organic-1", + Event: "pull_request", + Action: "synchronize", + Repository: "octocat/hello-world", + PullRequest: pr, + HeadSHA: headSHA, + Payload: []byte(`{}`), + }) + require.NoError(t, err) +} + +// A check_suite.requested delivery is persisted with a not-before time (the +// recovery grace) and ACKed fast — no GitHub call happens on the request +// path, and the row only becomes claimable after the organic pull_request +// delivery has had time to win. +func TestCheckSuiteWebhookQueuesWithGrace(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + req := buildCheckSuiteWebhookRequest(t, "requested", "suite-sha", "feature", 7) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + before := time.Now() + + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusOK, rr.Code) + require.JSONEq(t, `{"message":"check_suite recovery queued"}`, rr.Body.String()) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + require.NotNil(t, row) + assert.Equal(t, "check_suite", row.Event) + assert.Equal(t, "requested", row.Action) + assert.Equal(t, "octocat/hello-world", row.Repository) + assert.Equal(t, "suite-sha", row.HeadSHA) + assert.Equal(t, "12345", row.TenantID) + require.NotNil(t, row.RetryAfter, "check_suite rows must carry the recovery grace as a not-before time") + assert.WithinDuration(t, before.Add(defaultCheckSuiteRecoveryGrace), *row.RetryAfter, 10*time.Second) +} + +// Only "requested" carries recovery work: "rerequested" re-plans through +// check_run.rerequested and "completed" is pure noise, so neither may occupy +// an inbox row. +func TestCheckSuiteWebhookIgnoresNonRequestedActions(t *testing.T) { + for _, action := range []string{"rerequested", "completed"} { + t.Run(action, func(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + req := buildCheckSuiteWebhookRequest(t, action, "suite-sha", "feature", 7) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusOK, rr.Code) + require.JSONEq(t, `{"message":"check_suite action ignored"}`, rr.Body.String()) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + require.Nil(t, row) + }) + } +} + +// Without the recovery option the delivery is acknowledged and dropped — the +// kill switch must silence the feature without failing GitHub's delivery. +func TestCheckSuiteWebhookIgnoredWhenRecoveryDisabled(t *testing.T) { + store := newRecordingWebhookEventStore() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + service := api.New(&durableWebhookTestStorage{webhookEvents: store}, &api.ServerConfig{ + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + }, nil, logger) + h := NewHandler(service, &fakeClientFactory{}, nil, logger, WithDurableWebhookDispatch()) + + req := buildCheckSuiteWebhookRequest(t, "requested", "suite-sha", "feature", 7) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusOK, rr.Code) + require.JSONEq(t, `{"message":"check_suite recovery disabled"}`, rr.Body.String()) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + require.Nil(t, row) +} + +// A delivery from a repository outside the allowlist is acknowledged without +// occupying an inbox row. +func TestCheckSuiteWebhookRejectsUnregisteredRepo(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/other-repo": {}}) + + req := buildCheckSuiteWebhookRequest(t, "requested", "suite-sha", "feature", 7) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusOK, rr.Code) + require.JSONEq(t, `{"message":"repository not registered"}`, rr.Body.String()) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + require.Nil(t, row) +} + +// check_suite.requested fires for every push to every branch, but GitHub +// names the open PRs at a same-repository head in the payload — so a +// non-fork suite with an empty list had no open PR at delivery time and is +// dropped before it occupies an inbox row, regardless of branch. A suite +// that does name a PR is queued even when its head is the default branch (a +// promotion/backport PR), and fork heads — empty head_branch, whose payload +// cannot name PRs — pass through for API resolution. +func TestCheckSuiteWebhookSkipsHeadsWithoutOpenPRs(t *testing.T) { + tests := []struct { + name string + headBranch string + prNumbers []int + wantMessage string + wantQueued bool + }{ + {name: "branch push without a PR", headBranch: "feature", + wantMessage: `{"message":"check_suite without an open PR skipped"}`}, + {name: "default-branch push without a PR", headBranch: "main", + wantMessage: `{"message":"check_suite without an open PR skipped"}`}, + {name: "default-branch head with an open PR", headBranch: "main", prNumbers: []int{7}, + wantMessage: `{"message":"check_suite recovery queued"}`, wantQueued: true}, + {name: "fork head", headBranch: "", + wantMessage: `{"message":"check_suite recovery queued"}`, wantQueued: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + req := buildCheckSuiteWebhookRequest(t, "requested", "suite-sha", tt.headBranch, tt.prNumbers...) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusOK, rr.Code) + require.JSONEq(t, tt.wantMessage, rr.Body.String()) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + if tt.wantQueued { + require.NotNil(t, row) + } else { + require.Nil(t, row) + } + }) + } +} + +// A payload missing its repository is malformed and must be rejected with a +// 400 — not misfiled under the unregistered-repository metric with an empty +// repo attribute. +func TestCheckSuiteWebhookRejectsPayloadMissingRepo(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + body, err := json.Marshal(map[string]any{ + "action": "requested", + "check_suite": map[string]any{"head_sha": "suite-sha", "head_branch": "feature"}, + }) + require.NoError(t, err) + req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/webhook", strings.NewReader(string(body))) + req.Header.Set("X-GitHub-Event", "check_suite") + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + require.Equal(t, http.StatusBadRequest, rr.Code) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, "delivery-cs-1") + require.NoError(t, err) + require.Nil(t, row) +} + +// GitHub redeliveries reuse the delivery GUID; the second arrival dedupes +// against the live row instead of inserting a duplicate. +func TestCheckSuiteWebhookDedupesRedelivery(t *testing.T) { + store := newRecordingWebhookEventStore() + h := newCheckSuiteIngressHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + for i, want := range []string{ + `{"message":"check_suite recovery queued"}`, + `{"message":"check_suite recovery already queued"}`, + } { + req := buildCheckSuiteWebhookRequest(t, "requested", "suite-sha", "feature", 7) + req.Header.Set(headerDeliveryID, "delivery-cs-1") + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + require.Equal(t, http.StatusOK, rr.Code, "request %d", i) + require.JSONEq(t, want, rr.Body.String(), "request %d", i) + } +} + +// The steady state: the organic pull_request delivery arrived during the +// grace and planned the head, so processing finds coverage and synthesizes +// nothing. +func TestDurableCheckSuiteCoveredHeadNoOps(t *testing.T) { + store := newRecordingWebhookEventStore() + coveringAutoPlanRow(t, store, 7, "suite-sha") + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls/7", func(w http.ResponseWriter, _ *http.Request) { + writeSinglePR(t, w, 7, "open", "suite-sha") + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, row, "a covered head must not get a recovery row") +} + +// The recovery case: an open PR still at the suite head has no auto-plan +// coverage in the inbox, so processing synthesizes the deterministic +// recovery delivery the dispatcher will plan. +func TestDurableCheckSuiteSynthesizesMissingCoverage(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls/7", func(w http.ResponseWriter, _ *http.Request) { + writeSinglePR(t, w, 7, "open", "suite-sha") + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.NotNil(t, row) + assert.Equal(t, "pull_request", row.Event) + assert.Equal(t, 7, row.PullRequest) + assert.Equal(t, "suite-sha", row.HeadSHA) + assert.Equal(t, "12345", row.TenantID) +} + +// Every payload PR still open at the suite head gets its own PR-scoped +// recovery row — multiple PRs can share one head SHA. +func TestDurableCheckSuiteSynthesizesPerPR(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + for _, pr := range []string{"7", "8"} { + mux.HandleFunc("/repos/octocat/hello-world/pulls/"+pr, func(w http.ResponseWriter, r *http.Request) { + num := 7 + if strings.HasSuffix(r.URL.Path, "/8") { + num = 8 + } + writeSinglePR(t, w, num, "open", "suite-sha") + }) + } + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7, 8)) + + require.NoError(t, err) + require.False(t, retry) + for _, pr := range []int{7, 8} { + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", pr, "suite-sha")) + require.NoError(t, err) + require.NotNil(t, row, "PR %d must get its own recovery row", pr) + } +} + +// The payload PR list is a point-in-time snapshot and the grace means +// minutes have passed: a PR that closed or moved to a newer head is skipped, +// because a fresher signal owns its new state. +func TestDurableCheckSuiteSkipsClosedAndMovedPRs(t *testing.T) { + tests := []struct { + name string + state string + head string + }{ + {name: "closed PR", state: "closed", head: "suite-sha"}, + {name: "moved head", state: "open", head: "newer-sha"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls/7", func(w http.ResponseWriter, _ *http.Request) { + writeSinglePR(t, w, 7, tt.state, tt.head) + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, row) + }) + } +} + +// Fork heads arrive with an empty payload PR list and a null head_branch, so +// processing falls back to walking the repository's open PRs and matching by +// current head SHA. +func TestDurableCheckSuiteEmptyPayloadFallsBackToOpenPRScan(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls", func(w http.ResponseWriter, _ *http.Request) { + writeOpenPRs(t, w, + openPR(7, "suite-sha", time.Now().Add(-time.Hour)), + openPR(8, "other-sha", time.Now().Add(-time.Hour)), + ) + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteForkEvent(t, "suite-sha")) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.NotNil(t, row, "the open PR at the suite head must be recovered") + other, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 8, "other-sha")) + require.NoError(t, err) + require.Nil(t, other, "PRs at other heads must be untouched") +} + +// A fork suite head that matches no open PR is a clean no-op after the +// listing walk comes back empty. +func TestDurableCheckSuiteNoOpenPRMatchNoOps(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls", func(w http.ResponseWriter, _ *http.Request) { + writeOpenPRs(t, w) + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteForkEvent(t, "suite-sha")) + + require.NoError(t, err) + require.False(t, retry) +} + +// A same-repository suite (non-empty head_branch) whose payload names no PRs +// had no open PR at delivery time, so processing resolves nothing without +// touching the open-PR listing — no route is registered on the fake server, +// so any listing call would surface as an error. +func TestDurableCheckSuiteSameRepoEmptyPRListSkipsOpenPRWalk(t *testing.T) { + store := newRecordingWebhookEventStore() + h, _ := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha")) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, row) +} + +// One delivery can cover several PRs at the same head, and each PR's outcome +// is independent: a PR whose auto-plan coverage already exists no-ops while +// an uncovered sibling in the same pass still gets its recovery row. +func TestDurableCheckSuiteMixedCoverageSynthesizesOnlyUncovered(t *testing.T) { + store := newRecordingWebhookEventStore() + coveringAutoPlanRow(t, store, 7, "suite-sha") + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + for _, pr := range []string{"7", "8"} { + mux.HandleFunc("/repos/octocat/hello-world/pulls/"+pr, func(w http.ResponseWriter, r *http.Request) { + num := 7 + if strings.HasSuffix(r.URL.Path, "/8") { + num = 8 + } + writeSinglePR(t, w, num, "open", "suite-sha") + }) + } + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7, 8)) + + require.NoError(t, err) + require.False(t, retry) + covered, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, covered, "the covered PR must not get a recovery row") + uncovered, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 8, "suite-sha")) + require.NoError(t, err) + require.NotNil(t, uncovered, "the uncovered sibling must still be recovered") +} + +// A malformed check_suite payload cannot be decoded, so the driver fails it +// terminally (no retry) rather than crash-looping the fleet on a poison row. +func TestDurableCheckSuiteDriverFailsMalformedTerminally(t *testing.T) { + store := newScriptedWebhookEventStore(&storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: "delivery-check-suite-malformed", + Event: "check_suite", + Payload: []byte(`{not json`), + }) + h := newDurableDriverHandler(t, store, nil, nil) + + h.driveNextDurableWebhook(t.Context(), 0, "test-host/1/webhook-driver-0") + + select { + case failure := <-store.failed: + require.Nil(t, failure.retryAfter, "malformed payload must not be retried") + require.Contains(t, failure.errMsg, "decode durable check_suite delivery") + default: + t.Fatal("expected malformed check_suite event to be marked failed") + } + require.Empty(t, store.completed) +} + +// A terminally failed synthesized row does not cover its head; the +// check_suite signal reopens it as a fresh pending delivery so the head gets +// another recovery attempt. +func TestDurableCheckSuiteResynthesizesFailedRecoveryRow(t *testing.T) { + store := newRecordingWebhookEventStore() + guid := synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha") + _, err := store.Create(t.Context(), &storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: guid, + Event: "pull_request", + Action: webhookReconcileSynthesizedAction, + Repository: "octocat/hello-world", + PullRequest: 7, + HeadSHA: "suite-sha", + State: storage.WebhookEventFailed, + Payload: []byte(`{}`), + }) + require.NoError(t, err) + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls/7", func(w http.ResponseWriter, _ *http.Request) { + writeSinglePR(t, w, 7, "open", "suite-sha") + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, guid) + require.NoError(t, err) + require.NotNil(t, row) + assert.Equal(t, storage.WebhookEventPending, row.State, "the failed recovery row must be reopened") +} + +// GitHub API failures while resolving the suite head are transient: the +// delivery must be retried under its attempt budget, not dropped. +func TestDurableCheckSuiteGitHubFailureRetries(t *testing.T) { + store := newRecordingWebhookEventStore() + h, mux := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/hello-world": {}}) + mux.HandleFunc("/repos/octocat/hello-world/pulls/7", func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "boom", http.StatusInternalServerError) + }) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.Error(t, err) + require.True(t, retry) +} + +// Config can change between enqueue and claim: a repository deregistered in +// the interim must not have work synthesized for it. +func TestDurableCheckSuiteRevalidatesAllowlist(t *testing.T) { + store := newRecordingWebhookEventStore() + h, _ := newCheckSuiteProcessHandler(t, store, map[string]api.RepoConfig{"octocat/other-repo": {}}) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, row) +} + +// The kill switch must stop synthesis with a restart, including rows already +// sitting in the inbox when the operator flipped it. +func TestDurableCheckSuiteHonorsKillSwitchForQueuedRows(t *testing.T) { + store := newRecordingWebhookEventStore() + ghc, _ := setupGitHubServer(t) + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + service := api.New(&durableWebhookTestStorage{webhookEvents: store}, &api.ServerConfig{ + Repos: map[string]api.RepoConfig{"octocat/hello-world": {}}, + }, nil, logger) + factory := &fakeClientFactory{client: ghclient.NewInstallationClient(ghc, logger)} + h := NewHandler(service, factory, nil, logger, WithDurableWebhookDispatch()) + + retry, err := h.processDurableCheckSuite(t.Context(), durableCheckSuiteEvent(t, "suite-sha", 7)) + + require.NoError(t, err) + require.False(t, retry) + row, err := store.GetByDeliveryID(t.Context(), storage.WebhookProviderGitHub, synthesizedDeliveryGUID("octocat/hello-world", 7, "suite-sha")) + require.NoError(t, err) + require.Nil(t, row) +} diff --git a/pkg/webhook/durable_dispatch.go b/pkg/webhook/durable_dispatch.go index 2f6cc5fcc..ba2b979c1 100644 --- a/pkg/webhook/durable_dispatch.go +++ b/pkg/webhook/durable_dispatch.go @@ -211,9 +211,18 @@ func (h *Handler) driveClaimedDurableWebhook(ctx context.Context, driverID int, // first claim, and a retry claim's wait is the retry window, not // backlog latency. A shutdown-released claim refunds its attempt — // the refund means no genuine attempt happened — so the reclaim is - // the effective first attempt and records lag again, measured from - // the original receipt. - metrics.RecordWebhookInboxDispatchLag(ctx, appName, event.Event, event.Repository, time.Since(event.ReceivedAt)) + // the effective first attempt and records lag again. Lag is measured + // from when the row became eligible for dispatch (ClaimableSince): + // a row created with a not-before time waits out its grace period by + // design, and counting that deferral as lag would saturate the + // histogram's upper percentiles and mask real backlog regressions. + lagSince := event.ClaimableSince + if lagSince.IsZero() { + // A store that does not derive ClaimableSince falls back to + // receipt — the pre-deferral behavior, never a bogus epoch lag. + lagSince = event.ReceivedAt + } + metrics.RecordWebhookInboxDispatchLag(ctx, appName, event.Event, event.Repository, time.Since(lagSince)) } runCtx, cancelRun := context.WithCancel(ctx) @@ -451,6 +460,8 @@ func (h *Handler) processDurableWebhookEvent(ctx context.Context, event *storage return h.processDurablePullRequest(ctx, event) case "check_run": return h.processDurableCheckRun(ctx, event) + case "check_suite": + return h.processDurableCheckSuite(ctx, event) case "merge_group": return h.processDurableMergeGroup(ctx, event) case "push": diff --git a/pkg/webhook/durable_dispatch_metrics_test.go b/pkg/webhook/durable_dispatch_metrics_test.go index 4cf6dd34c..c12a0187f 100644 --- a/pkg/webhook/durable_dispatch_metrics_test.go +++ b/pkg/webhook/durable_dispatch_metrics_test.go @@ -85,6 +85,32 @@ func TestDurableWebhookDispatchMetricsFirstClaimCompleted(t *testing.T) { assertStringAttr(t, durationPoints[0].Attributes, "outcome", "completed") } +// A delivery that spent time deferred by a not-before time measures dispatch +// lag from when it became dispatchable, not from receipt: the grace period is +// by design, and counting it as lag would saturate the histogram's upper +// percentiles and mask real backlog regressions. +func TestDurableWebhookDispatchMetricsDeferredClaimMeasuresLagFromDueTime(t *testing.T) { + reader := newDispatchMetricsReader(t) + notBefore := time.Now().Add(-10 * time.Second) + store := newScriptedWebhookEventStore(&storage.WebhookEvent{ + Provider: storage.WebhookProviderGitHub, + DeliveryID: "delivery-metrics-deferred", + Event: "issue_comment", + Payload: []byte(`{}`), + ReceivedAt: time.Now().Add(-90 * time.Second), + RetryAfter: ¬Before, + }) + h := newDurableDriverHandler(t, store, nil, nil) + + h.driveNextDurableWebhook(t.Context(), 0, "test-host/1/webhook-driver-0") + + require.Len(t, store.completed, 1) + lagPoints := collectDispatchHistogramPoints(t, reader, "schemabot.webhook.inbox_dispatch_lag_seconds") + require.Len(t, lagPoints, 1, "first claim must record inbox dispatch lag") + assert.GreaterOrEqual(t, lagPoints[0].Sum, 10.0, "lag must count the wait since the not-before time") + assert.Less(t, lagPoints[0].Sum, 60.0, "lag must not count the deferral's grace period since receipt") +} + // A retry claim measures the retry window, not backlog latency, so only the // first claim records inbox dispatch lag. func TestDurableWebhookDispatchMetricsRetryClaimSkipsInboxLag(t *testing.T) { diff --git a/pkg/webhook/durable_dispatch_test.go b/pkg/webhook/durable_dispatch_test.go index 5cc725f87..67e1d779d 100644 --- a/pkg/webhook/durable_dispatch_test.go +++ b/pkg/webhook/durable_dispatch_test.go @@ -34,6 +34,7 @@ func (s *durableWebhookTestStorage) WebhookEvents() storage.WebhookEventStore { type recordingWebhookEventStore struct { mu sync.Mutex + nextID int64 events map[string]*storage.WebhookEvent } @@ -63,6 +64,11 @@ func (s *recordingWebhookEventStore) Create(_ context.Context, event *storage.We } return false, nil } + // Mirror the real store's insert contract: a fresh insert populates the + // caller's event.ID, while the reopen branch above leaves it zero — the + // discriminator callers use to tell the two apart. + s.nextID++ + event.ID = s.nextID copy := *event copy.Payload = append([]byte(nil), event.Payload...) if copy.State == "" { @@ -266,6 +272,14 @@ func (s *scriptedWebhookEventStore) FindNext(_ context.Context, owner string, _ claimed.LeaseOwner = owner claimed.LeaseToken = fmt.Sprintf("token-%d", claimed.ID) claimed.Attempts = event.Attempts + 1 + // Mirror the real claim contract: the claim consumes retry_after and the + // returned event carries ClaimableSince, the later of receipt and the + // consumed not-before time. + claimed.ClaimableSince = claimed.ReceivedAt + if claimed.RetryAfter != nil && claimed.RetryAfter.After(claimed.ClaimableSince) { + claimed.ClaimableSince = *claimed.RetryAfter + } + claimed.RetryAfter = nil return &claimed, nil } diff --git a/pkg/webhook/handler.go b/pkg/webhook/handler.go index 96cf3cb76..06e6db74b 100644 --- a/pkg/webhook/handler.go +++ b/pkg/webhook/handler.go @@ -174,6 +174,9 @@ type Handler struct { webhookReconcileGrace time.Duration webhookReconcileMaxPages int + checkSuiteRecovery bool + checkSuiteRecoveryGrace time.Duration + logger *slog.Logger priorEnvCheckMaxAttempts int priorEnvCheckRetryInterval time.Duration @@ -230,6 +233,19 @@ func WithWebhookReconcileSynthesis() HandlerOption { } } +// WithCheckSuiteRecovery feeds check_suite.requested deliveries into the +// durable inbox as a redundant convergence signal: each is enqueued with a +// not-before time (the recovery grace) and, once claimable, synthesizes a +// recovery delivery for any open PR still at the suite head whose auto-plan +// coverage is missing. It takes effect only alongside +// WithDurableWebhookDispatch; without that, check_suite deliveries are +// acknowledged and ignored. +func WithCheckSuiteRecovery() HandlerOption { + return func(h *Handler) { + h.checkSuiteRecovery = true + } +} + // NewHandler creates a new webhook handler for the legacy single-App // configuration. The provided factory is registered in the internal // ClientSet under the "default" App name so per-repo client resolution @@ -272,6 +288,7 @@ func NewHandlerWithDispatch(service *api.Service, ghClients github.ClientSet, we webhookReconcileLookback: defaultWebhookReconcileLookback, webhookReconcileGrace: defaultWebhookReconcileGrace, webhookReconcileMaxPages: defaultWebhookReconcileMaxPages, + checkSuiteRecoveryGrace: defaultCheckSuiteRecoveryGrace, priorEnvCheckMaxAttempts: defaultPriorEnvCheckMaxAttempts, priorEnvCheckRetryInterval: defaultPriorEnvCheckRetryInterval, } @@ -725,6 +742,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { case "check_run": h.handleCheckRun(ctx, metricApp, sw, body, r.Header.Get(headerDeliveryID)) recordProcessed() + case "check_suite": + h.handleCheckSuite(ctx, metricApp, sw, body, r.Header.Get(headerDeliveryID)) + recordProcessed() case "pull_request": h.handlePullRequest(ctx, metricApp, sw, body, r.Header.Get(headerDeliveryID)) recordProcessed()