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,
@@ -16,11 +17,12 @@ CREATE TABLE `merge_gate_requests` (
   `lease_expires_at` datetime(6) DEFAULT NULL,
   `retry_after` datetime DEFAULT NULL,
   `last_error` text,
+  `holds_recorded_at` datetime DEFAULT NULL,
   `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`),
+  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
 }