feat(storage): durable merge gate requests for schema-mutating applies

.github/workflows/test.yaml

@@ -137,7 +137,7 @@ jobs:
     needs: changes
     if: needs.changes.outputs.code == 'true'
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@1af3b93b6815bc44a9784bd300feb67ff0d1eeb3 # actions/checkout@v6
       - name: "Configure Go"

integration/operator_test.go

@@ -798,11 +798,13 @@ func TestOperator_OperationWithoutTasksFailsClosed(t *testing.T) {
 	require.NotNil(t, op.CompletedAt, "a failed operation stamps completed_at")
 
 	// The parent apply state is derived from its child operations, so a single
-	// failed operation drives the parent to failed.
-	parent, err := ts.Storage.Applies().Get(ctx, applyDBID)
-	require.NoError(t, err)
-	require.NotNil(t, parent)
-	assert.Equal(t, state.Apply.Failed, parent.State,
+	// failed operation drives the parent to failed. The re-derivation is a
+	// separate write after the operation is terminalized, so poll for it.
+	require.Eventually(t, func() bool {
+		parent, err := ts.Storage.Applies().Get(ctx, applyDBID)
+		require.NoError(t, err)
+		return parent != nil && state.IsState(parent.State, state.Apply.Failed)
+	}, 10*time.Second, 200*time.Millisecond,
 		"parent apply is derived from its operations and must reflect the failed child")
 }
 

pkg/api/handlers_test.go

@@ -46,8 +46,11 @@ func (m *mockStorage) ApplyOperations() storage.ApplyOperationStore { return nil
 func (m *mockStorage) Checks() storage.CheckStore                   { return nil }
 func (m *mockStorage) Settings() storage.SettingsStore              { return nil }
 func (m *mockStorage) WebhookEvents() storage.WebhookEventStore     { return m.webhookEvents }
-func (m *mockStorage) Ping(ctx context.Context) error               { return m.pingErr }
-func (m *mockStorage) Close() error                                 { return nil }
+func (m *mockStorage) MergeGateRequests() storage.MergeGateRequestStore {
+	return nil
+}
+func (m *mockStorage) Ping(ctx context.Context) error { return m.pingErr }
+func (m *mockStorage) Close() error                   { return nil }
 
 type mockPlanLookupStore struct {
 	plan *storage.Plan

pkg/schema/mysql/checks.sql

@@ -19,6 +19,7 @@ CREATE TABLE `checks` (
   PRIMARY KEY (`id`),
   UNIQUE KEY `idx_check_key` (`repository`,`pull_request`,`environment`,`database_type`,`database_name`),
   KEY `idx_repo_env_db` (`repository`,`environment`,`database_type`,`database_name`),
+  KEY `idx_env_db` (`environment`,`database_type`,`database_name`),
   KEY `idx_repo_pr` (`repository`,`pull_request`),
   KEY `idx_check_run` (`check_run_id`),
   KEY `idx_apply_id` (`apply_id`)

pkg/schema/mysql/merge_gate_requests.sql

@@ -0,0 +1,26 @@
+CREATE TABLE `merge_gate_requests` (
+  `id` bigint unsigned NOT NULL AUTO_INCREMENT,
+  `apply_id` bigint unsigned NOT NULL,
+  `apply_identifier` varchar(255) NOT NULL,
+  `environment` varchar(50) NOT NULL,
+  `database_type` varchar(50) NOT NULL,
+  `database_name` varchar(255) NOT NULL,
+  `provider` varchar(50) NOT NULL DEFAULT 'github',
+  `repository` varchar(255) NOT NULL DEFAULT '',
+  `change_key` varchar(255) NOT NULL DEFAULT '',
+  `requested_by` varchar(255) NOT NULL DEFAULT '',
+  `state` varchar(50) NOT NULL,
+  `attempts` int unsigned NOT NULL DEFAULT '0',
+  `lease_owner` varchar(255) DEFAULT NULL,
+  `lease_token` varchar(64) DEFAULT NULL,
+  `lease_expires_at` datetime(6) DEFAULT NULL,
+  `retry_after` datetime DEFAULT NULL,
+  `last_error` text,
+  `completed_at` datetime DEFAULT NULL,
+  `created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `idx_merge_gate_apply` (`apply_id`),
+  KEY `idx_merge_gate_claimable` (`state`,`retry_after`,`lease_expires_at`,`created_at`),
+  KEY `idx_merge_gate_target` (`environment`,`database_type`,`database_name`,`state`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

pkg/storage/errors.go

@@ -68,4 +68,10 @@ var (
 
 	// ErrWebhookEventLeaseLost is returned when a driver no longer owns a durable webhook event.
 	ErrWebhookEventLeaseLost = errors.New("webhook event lease lost")
+
+	// ErrMergeGateNotFound is returned when a durable merge gate request does not exist.
+	ErrMergeGateNotFound = errors.New("merge gate request not found")
+
+	// ErrMergeGateLeaseLost is returned when a driver no longer owns a durable merge gate request.
+	ErrMergeGateLeaseLost = errors.New("merge gate request lease lost")
 )

pkg/storage/mysqlstore/checks.go

@@ -508,6 +508,59 @@ func (s *checkStore) GetByDatabase(ctx context.Context, repo, environment, dbTyp
 	return scanChecks(rows)
 }
 
+// GetByTarget returns all checks for a target across all repositories and PRs.
+// The merge gate fan-out uses it: a CLI/gRPC apply carries no repository, so
+// the fan-out must find every PR planned against the target regardless of repo.
+func (s *checkStore) GetByTarget(ctx context.Context, environment, dbType, database string) ([]*storage.Check, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+checkColumns+`
+		FROM checks
+		WHERE environment = ? AND database_type = ? AND database_name = ?
+		ORDER BY repository, pull_request
+	`, environment, dbType, database)
+	if err != nil {
+		return nil, fmt.Errorf("query checks for target %s/%s in %s: %w", dbType, database, environment, err)
+	}
+	defer utils.CloseAndLog(rows)
+
+	return scanChecks(rows)
+}
+
+// MarkBlockedForFailedRefresh flips stored check state to a blocking conclusion
+// after a merge gate re-plan failed. The head SHA predicate makes the write
+// optimistic-concurrency: a racing synchronize that already stored a result for
+// a newer commit does not match and is preserved. An in-progress apply-owned
+// row is never touched — the started apply's lifecycle stays authoritative.
+func (s *checkStore) MarkBlockedForFailedRefresh(ctx context.Context, check *storage.Check) (bool, error) {
+	result, err := s.db.ExecContext(ctx, `
+		UPDATE checks
+		SET apply_id = NULL,
+		    has_changes = ?,
+		    status = ?,
+		    conclusion = ?,
+		    blocking_reason = ?,
+		    error_message = ?,
+		    change_summary = ?
+		WHERE repository = ? AND pull_request = ?
+		  AND environment = ? AND database_type = ? AND database_name = ?
+		  AND head_sha = ?
+		  AND NOT (status = ? AND apply_id IS NOT NULL)
+	`, check.HasChanges, check.Status, check.Conclusion, check.BlockingReason, check.ErrorMessage, nullString(check.ChangeSummary),
+		check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName,
+		check.HeadSHA,
+		checkStatusInProgress)
+	if err != nil {
+		return false, fmt.Errorf("mark check blocked for failed refresh %s#%d %s/%s/%s (head %s): %w",
+			check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("rows affected marking check blocked for failed refresh %s#%d %s/%s/%s: %w",
+			check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, err)
+	}
+	return rows > 0, nil
+}
+
 // Delete removes stored check state by ID.
 func (s *checkStore) Delete(ctx context.Context, id int64) error {
 	result, err := s.db.ExecContext(ctx, `DELETE FROM checks WHERE id = ?`, id)