feat(storage): merge gate request kinds and an apply-in-flight check hold

pkg/api/merge_gate_record_test.go

@@ -90,6 +90,7 @@ func TestRecordMergeGateGatedOnConsumer(t *testing.T) {
 		require.Len(t, gateStore.recorded, 1)
 		recorded := gateStore.recorded[0]
 		assert.Equal(t, "apply-gate-test", recorded.ApplyIdentifier)
+		assert.Equal(t, storage.MergeGateKindSettle, recorded.Kind)
 		assert.Equal(t, "gate_db", recorded.DatabaseName)
 		assert.Equal(t, "mysql", recorded.DatabaseType)
 		assert.Equal(t, "staging", recorded.Environment)

pkg/api/operator.go

@@ -1072,6 +1072,7 @@ func (s *Service) recordMergeGateIfApplyResolved(ctx context.Context, driverID i
 
 	recorded, err := s.storage.MergeGateRequests().Record(ctx, &storage.MergeGateRequest{
 		ApplyID:         apply.ID,
+		Kind:            storage.MergeGateKindSettle,
 		ApplyIdentifier: apply.ApplyIdentifier,
 		Environment:     apply.Environment,
 		DatabaseType:    apply.DatabaseType,

pkg/schema/mysql/merge_gate_requests.sql

@@ -1,6 +1,7 @@
 CREATE TABLE `merge_gate_requests` (
   `id` bigint unsigned NOT NULL AUTO_INCREMENT,
   `apply_id` bigint unsigned NOT NULL,
+  `kind` varchar(20) NOT NULL,
   `apply_identifier` varchar(255) NOT NULL,
   `environment` varchar(50) NOT NULL,
   `database_type` varchar(50) NOT NULL,
@@ -20,7 +21,7 @@ CREATE TABLE `merge_gate_requests` (
   `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`),
+  UNIQUE KEY `idx_merge_gate_apply` (`apply_id`,`kind`),
   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/mysqlstore/checks.go

@@ -532,7 +532,37 @@ func (s *checkStore) GetByTarget(ctx context.Context, environment, dbType, datab
 // 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, `
+	flipped, err := s.markBlockedConditional(ctx, check, false)
+	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)
+	}
+	return flipped, nil
+}
+
+// MarkBlockedForApplyInFlight flips stored check state to a blocking
+// conclusion while an apply on the same target runs. Same conditional-write
+// contract as MarkBlockedForFailedRefresh, plus rows already holding the same
+// blocking reason are skipped so a retried preflight fan-out reports
+// flipped=false instead of re-flipping (and re-announcing) the hold.
+func (s *checkStore) MarkBlockedForApplyInFlight(ctx context.Context, check *storage.Check) (bool, error) {
+	flipped, err := s.markBlockedConditional(ctx, check, true)
+	if err != nil {
+		return false, fmt.Errorf("mark check blocked for apply in flight %s#%d %s/%s/%s (head %s): %w",
+			check.Repository, check.PullRequest, check.Environment, check.DatabaseType, check.DatabaseName, check.HeadSHA, err)
+	}
+	return flipped, nil
+}
+
+// markBlockedConditional is the shared conditional blocking flip: it writes
+// the caller's blocking conclusion only when the stored row still holds the
+// head SHA the caller read (a racing synchronize that re-planned a newer
+// commit wins) and is not an in-progress apply-owned row (a started apply's
+// lifecycle stays authoritative). With skipAlreadyHeld, rows whose
+// blocking_reason already equals the caller's are also left untouched, so
+// idempotent retries can distinguish "newly flipped" from "already held".
+func (s *checkStore) markBlockedConditional(ctx context.Context, check *storage.Check, skipAlreadyHeld bool) (bool, error) {
+	query := `
 		UPDATE checks
 		SET apply_id = NULL,
 		    has_changes = ?,
@@ -544,19 +574,23 @@ func (s *checkStore) MarkBlockedForFailedRefresh(ctx context.Context, check *sto
 		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),
+		  AND NOT (status = ? AND apply_id IS NOT NULL)`
+	args := []any{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)
+		checkStatusInProgress}
+	if skipAlreadyHeld {
+		query += `
+		  AND (blocking_reason IS NULL OR blocking_reason != ?)`
+		args = append(args, check.BlockingReason)
+	}
+	result, err := s.db.ExecContext(ctx, query, args...)
 	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)
+		return false, 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 false, fmt.Errorf("rows affected: %w", err)
 	}
 	return rows > 0, nil
 }

pkg/storage/mysqlstore/merge_gate_requests.go

@@ -17,7 +17,7 @@ import (
 	"github.com/block/schemabot/pkg/storage"
 )
 
-const mergeGateColumns = `id, apply_id, apply_identifier, environment, database_type, database_name,
+const mergeGateColumns = `id, apply_id, kind, apply_identifier, environment, database_type, database_name,
 	provider, repository, change_key, requested_by, state, attempts,
 	lease_owner, lease_token, lease_expires_at, retry_after, last_error,
 	completed_at, created_at, updated_at`
@@ -35,6 +35,9 @@ func (s *mergeGateRequestStore) Record(ctx context.Context, req *storage.MergeGa
 	if req.ApplyIdentifier == "" {
 		return false, fmt.Errorf("merge gate request for apply row %d requires the apply identifier", req.ApplyID)
 	}
+	if req.Kind != storage.MergeGateKindPreflight && req.Kind != storage.MergeGateKindSettle {
+		return false, fmt.Errorf("merge gate request for apply %s has unknown kind %q", req.ApplyIdentifier, req.Kind)
+	}
 	if req.Environment == "" || req.DatabaseType == "" || req.DatabaseName == "" {
 		return false, fmt.Errorf("merge gate request for apply %s requires environment, database type, and database name", req.ApplyIdentifier)
 	}
@@ -46,40 +49,63 @@ func (s *mergeGateRequestStore) Record(ctx context.Context, req *storage.MergeGa
 
 	id, err := s.identity.InsertID(ctx, s.db, `
 		INSERT INTO merge_gate_requests (
-			apply_id, apply_identifier, environment, database_type, database_name,
+			apply_id, kind, apply_identifier, environment, database_type, database_name,
 			provider, repository, change_key, requested_by, state
-		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-	`, req.ApplyID, req.ApplyIdentifier, req.Environment, req.DatabaseType, req.DatabaseName,
+		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	`, req.ApplyID, req.Kind, req.ApplyIdentifier, req.Environment, req.DatabaseType, req.DatabaseName,
 		provider, req.Repository, req.ChangeKey, req.RequestedBy, storage.MergeGatePending)
 	if err != nil {
-		// The unique key on apply_id is the idempotency guard: the drive tail
-		// and the backfill sweep may both record the same apply, and a request
-		// already in any state means the fan-out is recorded or done.
+		// The unique key on (apply_id, kind) is the idempotency guard: the
+		// drive tails, the operator preflight gate, and the backfill sweeps may
+		// all record the same apply and kind, and a request already in any
+		// state means the fan-out is recorded or done.
 		if isDuplicateKeyError(err) {
 			return false, nil
 		}
-		return false, fmt.Errorf("record merge gate request for apply %s (%s/%s in %s): %w",
-			req.ApplyIdentifier, req.DatabaseType, req.DatabaseName, req.Environment, err)
+		return false, fmt.Errorf("record %s merge gate request for apply %s (%s/%s in %s): %w",
+			req.Kind, req.ApplyIdentifier, req.DatabaseType, req.DatabaseName, req.Environment, err)
 	}
 	req.ID = id
 	req.Provider = provider
 	req.State = storage.MergeGatePending
 	return true, nil
 }
 
-func (s *mergeGateRequestStore) GetByApplyID(ctx context.Context, applyID int64) (*storage.MergeGateRequest, error) {
+func (s *mergeGateRequestStore) GetByApplyAndKind(ctx context.Context, applyID int64, kind string) (*storage.MergeGateRequest, error) {
 	row := s.db.QueryRowContext(ctx, `
 		SELECT `+mergeGateColumns+`
 		FROM merge_gate_requests
-		WHERE apply_id = ?
-	`, applyID)
+		WHERE apply_id = ? AND kind = ?
+	`, applyID, kind)
 	req, err := scanMergeGateRequest(row)
 	if err != nil {
-		return nil, fmt.Errorf("get merge gate request for apply row %d: %w", applyID, err)
+		return nil, fmt.Errorf("get %s merge gate request for apply row %d: %w", kind, applyID, err)
 	}
 	return req, nil
 }
 
+// ReopenForRetry re-arms a terminally failed request back to pending with a
+// fresh attempt budget, so the operator preflight gate can recover an apply
+// whose preflight exhausted its retries during an outage. Conditional on the
+// failed state: pending, processing, and completed rows are left untouched.
+func (s *mergeGateRequestStore) ReopenForRetry(ctx context.Context, id int64) (bool, error) {
+	result, err := s.db.ExecContext(ctx, `
+		UPDATE merge_gate_requests
+		SET state = ?, attempts = 0, lease_owner = NULL, lease_token = NULL,
+			lease_expires_at = NULL, retry_after = NULL, completed_at = NULL,
+			updated_at = NOW()
+		WHERE id = ? AND state = ?
+	`, storage.MergeGatePending, id, storage.MergeGateFailed)
+	if err != nil {
+		return false, fmt.Errorf("reopen merge gate request %d for retry: %w", id, err)
+	}
+	rows, err := result.RowsAffected()
+	if err != nil {
+		return false, fmt.Errorf("read reopened merge gate request %d rows affected: %w", id, err)
+	}
+	return rows > 0, nil
+}
+
 // mergeGateClaimablePredicate matches exactly the rows a processor 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
@@ -173,17 +199,17 @@ func (s *mergeGateRequestStore) ClaimNext(ctx context.Context, owner string, lea
 	return req, nil
 }
 
-func (s *mergeGateRequestStore) PendingForTarget(ctx context.Context, environment, databaseType, databaseName string, excludeID int64) ([]*storage.MergeGateRequest, error) {
+func (s *mergeGateRequestStore) PendingForTarget(ctx context.Context, environment, databaseType, databaseName, kind string, excludeID int64) ([]*storage.MergeGateRequest, error) {
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT `+mergeGateColumns+`
 		FROM merge_gate_requests
 		WHERE environment = ? AND database_type = ? AND database_name = ?
-		  AND state = ? AND id != ?
+		  AND kind = ? AND state = ? AND id != ?
 		ORDER BY created_at, id
-	`, environment, databaseType, databaseName, storage.MergeGatePending, excludeID)
+	`, environment, databaseType, databaseName, kind, storage.MergeGatePending, excludeID)
 	if err != nil {
-		return nil, fmt.Errorf("query pending merge gate requests for %s/%s in %s: %w",
-			databaseType, databaseName, environment, err)
+		return nil, fmt.Errorf("query pending %s merge gate requests for %s/%s in %s: %w",
+			kind, databaseType, databaseName, environment, err)
 	}
 	defer utils.CloseAndLog(rows)
 
@@ -278,28 +304,96 @@ func (s *mergeGateRequestStore) FindCompletedAppliesMissingRequest(ctx context.C
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT `+applyColumnsForApplyAlias+`
 		FROM applies a
-		LEFT JOIN merge_gate_requests r ON r.apply_id = a.id
+		LEFT JOIN merge_gate_requests r ON r.apply_id = a.id AND r.kind = ?
 		WHERE a.state = ?
 		  AND a.completed_at > `+s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, ParameterIntervalAmount(), IntervalSecond)+`
 		  AND r.id IS NULL
 		ORDER BY a.completed_at, a.id
-	`, state.Apply.Completed, int64(lookback.Seconds()))
+	`, storage.MergeGateKindSettle, state.Apply.Completed, int64(lookback.Seconds()))
 	if err != nil {
-		return nil, fmt.Errorf("query completed applies missing merge gate requests: %w", err)
+		return nil, fmt.Errorf("query completed applies missing settle merge gate requests: %w", err)
 	}
 	defer utils.CloseAndLog(rows)
 
 	var applies []*storage.Apply
 	for rows.Next() {
 		apply, err := scanApplyInto(rows)
 		if err != nil {
-			return nil, fmt.Errorf("scan completed apply missing merge gate request: %w", err)
+			return nil, fmt.Errorf("scan completed apply missing settle merge gate request: %w", err)
 		}
 		applies = append(applies, apply)
 	}
 	return applies, rows.Err()
 }
 
+// FindTerminalAppliesWithPreflightMissingSettle returns applies that settled
+// terminally within the lookback window with a preflight request but no
+// settle. Their preflight fan-out held sibling change checks action-required, and
+// only a settle fan-out re-plans those checks back to a live verdict — so the
+// settle must exist for every terminal outcome, including failed and
+// cancelled applies that never changed the schema.
+func (s *mergeGateRequestStore) FindTerminalAppliesWithPreflightMissingSettle(ctx context.Context, lookback time.Duration) ([]*storage.Apply, error) {
+	if lookback <= 0 {
+		return nil, fmt.Errorf("merge gate sweep lookback must be positive")
+	}
+	terminalStates := terminalApplyStates()
+	args := []any{storage.MergeGateKindPreflight, storage.MergeGateKindSettle}
+	args = append(args, stringArgs(terminalStates)...)
+	args = append(args, int64(lookback.Seconds()))
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT `+applyColumnsForApplyAlias+`
+		FROM applies a
+		JOIN merge_gate_requests pre ON pre.apply_id = a.id AND pre.kind = ?
+		LEFT JOIN merge_gate_requests settle ON settle.apply_id = a.id AND settle.kind = ?
+		WHERE a.state IN (`+placeholders(len(terminalStates))+`)
+		  AND a.updated_at > `+s.dialect.RelativeTime(TimestampPrecisionDefault, BeforeCurrentTime, ParameterIntervalAmount(), IntervalSecond)+`
+		  AND settle.id IS NULL
+		ORDER BY a.updated_at, a.id
+	`, args...)
+	if err != nil {
+		return nil, fmt.Errorf("query terminal applies with preflight missing settle merge gate requests: %w", err)
+	}
+	defer utils.CloseAndLog(rows)
+
+	var applies []*storage.Apply
+	for rows.Next() {
+		apply, err := scanApplyInto(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan terminal apply with preflight missing settle merge gate request: %w", err)
+		}
+		applies = append(applies, apply)
+	}
+	return applies, rows.Err()
+}
+
+// HasActivePreflightedApplyOnTarget reports whether any non-terminal apply on
+// the target has a recorded preflight request. Such an apply has held (or is
+// holding) sibling change checks and is guaranteed a settle of its own once it
+// settles terminally, so a settle fan-out for an earlier apply defers to it
+// rather than re-planning holds away mid-apply. Applies without a preflight
+// (queued, never started) do not count: they have invalidated nothing yet.
+func (s *mergeGateRequestStore) HasActivePreflightedApplyOnTarget(ctx context.Context, environment, databaseType, databaseName string) (bool, error) {
+	nonTerminal, nonTerminalArgs := nonTerminalApplyStatePredicate("a.state")
+	args := []any{storage.MergeGateKindPreflight, environment, databaseType, databaseName}
+	args = append(args, nonTerminalArgs...)
+	var exists bool
+	err := s.db.QueryRowContext(ctx, `
+		SELECT EXISTS (
+			SELECT 1
+			FROM merge_gate_requests pre
+			JOIN applies a ON a.id = pre.apply_id
+			WHERE pre.kind = ?
+			  AND pre.environment = ? AND pre.database_type = ? AND pre.database_name = ?
+			  AND `+nonTerminal+`
+		)
+	`, args...).Scan(&exists)
+	if err != nil {
+		return false, fmt.Errorf("query active preflighted applies for %s/%s in %s: %w",
+			databaseType, databaseName, environment, err)
+	}
+	return exists, nil
+}
+
 func (s *mergeGateRequestStore) TerminateStuckProcessing(ctx context.Context, reason string) (int64, error) {
 	result, err := s.db.ExecContext(ctx, `
 		UPDATE merge_gate_requests
@@ -354,7 +448,7 @@ func scanMergeGateRequestInto(row scanner) (*storage.MergeGateRequest, error) {
 	var leaseOwner, leaseToken, lastError sql.NullString
 	var leaseExpiresAt, retryAfter, completedAt sql.NullTime
 	err := row.Scan(
-		&req.ID, &req.ApplyID, &req.ApplyIdentifier, &req.Environment, &req.DatabaseType, &req.DatabaseName,
+		&req.ID, &req.ApplyID, &req.Kind, &req.ApplyIdentifier, &req.Environment, &req.DatabaseType, &req.DatabaseName,
 		&req.Provider, &req.Repository, &req.ChangeKey, &req.RequestedBy, &req.State, &req.Attempts,
 		&leaseOwner, &leaseToken, &leaseExpiresAt, &retryAfter, &lastError,
 		&completedAt, &req.CreatedAt, &req.UpdatedAt,