fix(dc-plan): unblock startup and stop refresh sync amplification

internal/api/handlers/dc_plan.go

@@ -402,18 +402,13 @@ func (h *DCPlanHandler) RefreshOperatorPlans(c *gin.Context) {
 		return
 	}
 
-	stale := false
-	if h.syncService == nil {
-		stale = true
-	} else if _, err := h.syncService.SyncWorkspace(c.Request.Context(), claims.WorkspaceID); err != nil {
-		stale = true
-		logger.Printf(
-			"[DC_PLAN] Operator plan refresh using stale projection: workspace_id=%d operator=%s error=%v",
-			claims.WorkspaceID,
-			claims.OperatorID,
-			err,
-		)
-	}
+	// The projection is converged by the periodic hilbert sync only (see
+	// server.startPeriodicDCPlanSync). This endpoint deliberately does not sync:
+	// one SyncWorkspace costs O(plan count) transactions (the workstation
+	// projector and the pending pool each begin a transaction per plan), so
+	// N devices polling used to amplify into thousands of concurrent
+	// transactions, exhaust the connection pool and deadlock.
+	stale := h.syncService == nil
 
 	rows := []struct {
 		ID                    int64        `db:"id"`

internal/api/handlers/operator_plan_test.go

@@ -7,9 +7,9 @@ package handlers
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"net/http"
 	"net/http/httptest"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -22,7 +22,8 @@ import (
 )
 
 type fakeOperatorPlanSyncer struct {
-	err error
+	err   error
+	calls int32
 }
 
 func (s *fakeOperatorPlanSyncer) Configured() bool {
@@ -33,12 +34,17 @@ func (s *fakeOperatorPlanSyncer) SyncWorkspace(
 	context.Context,
 	int64,
 ) (*services.DCPlanSyncResult, error) {
+	atomic.AddInt32(&s.calls, 1)
 	if s.err != nil {
 		return nil, s.err
 	}
 	return &services.DCPlanSyncResult{LastSyncedAt: time.Now().UTC()}, nil
 }
 
+func (s *fakeOperatorPlanSyncer) callCount() int {
+	return int(atomic.LoadInt32(&s.calls))
+}
+
 func TestRefreshOperatorPlansFiltersAssignmentAndReportsProgress(t *testing.T) {
 	db := newTestOperatorPlanDB(t)
 	defer db.Close()
@@ -94,25 +100,37 @@ func TestRefreshOperatorPlansExcludesCollectedPlans(t *testing.T) {
 		t.Fatalf("unexpected response: %#v", response)
 	}
 }
-func TestRefreshOperatorPlansFallsBackToStaleProjection(t *testing.T) {
+
+// Regression: RefreshOperatorPlans used to run a full SyncWorkspace inline. A single
+// SyncWorkspace costs O(plan count) transactions because the workstation projector
+// and the pending pool each begin one transaction per plan, so N devices polling the
+// endpoint amplified into thousands of concurrent transactions, exhausted the
+// connection pool and deadlocked. The projection is converged by the periodic
+// hilbert sync (server.startPeriodicDCPlanSync) instead.
+func TestRefreshOperatorPlansDoesNotSyncInline(t *testing.T) {
 	db := newTestOperatorPlanDB(t)
 	defer db.Close()
 	seedOperatorPlanFixture(t, db)
 
-	router := newTestOperatorPlanRouter(db, &fakeOperatorPlanSyncer{err: errors.New("hilbert down")})
-	w := httptest.NewRecorder()
-	req := httptest.NewRequest(http.MethodPost, "/api/v1/operator/plans/refresh", nil)
-	router.ServeHTTP(w, req)
-	if w.Code != http.StatusOK {
-		t.Fatalf("status=%d want=%d body=%s", w.Code, http.StatusOK, w.Body.String())
-	}
-
-	var response OperatorPlanRefreshResponse
-	if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
-		t.Fatalf("unmarshal response: %v", err)
+	syncer := &fakeOperatorPlanSyncer{}
+	router := newTestOperatorPlanRouter(db, syncer)
+	for i := 0; i < 5; i++ {
+		w := httptest.NewRecorder()
+		req := httptest.NewRequest(http.MethodPost, "/api/v1/operator/plans/refresh", nil)
+		router.ServeHTTP(w, req)
+		if w.Code != http.StatusOK {
+			t.Fatalf("request %d status=%d want=%d body=%s", i, w.Code, http.StatusOK, w.Body.String())
+		}
+		var response OperatorPlanRefreshResponse
+		if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil {
+			t.Fatalf("request %d unmarshal response: %v", i, err)
+		}
+		if len(response.Items) != 1 {
+			t.Fatalf("request %d expected the cached projection, got %#v", i, response)
+		}
 	}
-	if !response.Stale || len(response.Items) != 1 || response.LastSyncedAt == "" {
-		t.Fatalf("unexpected stale response: %#v", response)
+	if calls := syncer.callCount(); calls != 0 {
+		t.Fatalf("refresh must not sync, but SyncWorkspace was called %d time(s)", calls)
 	}
 }
 

internal/server/server.go

@@ -621,10 +621,6 @@ func (s *Server) Start() error {
 	s.isRunning = true
 	s.shutdownMu.Unlock()
 
-	s.syncWorkspacesOnStartup()
-	s.syncDCPlansOnStartup()
-	s.startPeriodicDCPlanSync()
-
 	logger.Printf("[SERVER] Starting HTTP server on %s", s.cfg.Server.BindAddr)
 	logger.Printf("[SERVER] Swagger UI: http://localhost%s/swagger/index.html", s.cfg.Server.BindAddr)
 
@@ -634,6 +630,12 @@ func (s *Server) Start() error {
 		}
 	}()
 
+	// Hilbert plan/workstation/pending-pool convergence runs in the background only.
+	// It must be started after the listener: a slow Hilbert round used to block
+	// readiness, so the startup probe killed the container mid-sync, the transaction
+	// rolled back and the whole sync repeated forever on the next attempt.
+	s.startPeriodicDCPlanSync()
+
 	// Start WebSocket server on separate port
 	logger.Printf("[SERVER] Transfer WebSocket server listening on %d", s.cfg.AxonTransfer.WSPort)
 
@@ -671,26 +673,24 @@ func (s *Server) EpisodeQAEnqueuer() interface {
 	return s.qa
 }
 
-func (s *Server) syncWorkspacesOnStartup() {
+func (s *Server) syncWorkspacesOnStartup(ctx context.Context) {
 	if s.workspaceSync == nil || !s.workspaceSync.Configured() {
 		logger.Printf("[WORKSPACE] Startup Hilbert workspace sync skipped: service identity config incomplete")
 		return
 	}
-	result, err := s.workspaceSync.Sync(context.Background())
+	result, err := s.workspaceSync.Sync(ctx)
 	if err != nil {
 		logger.Printf("[WORKSPACE] Startup Hilbert workspace sync failed: %v", err)
 		return
 	}
 	logger.Printf("[WORKSPACE] Startup Hilbert workspace sync completed: synced_count=%d", result.SyncedCount)
 }
 
-func (s *Server) syncDCPlansOnStartup() {
-	s.syncDCPlansOnce(context.Background(), "Startup")
-}
-
 func (s *Server) startPeriodicDCPlanSync() {
 	if s.dcPlanSync == nil || !s.dcPlanSync.Configured() {
 		logger.Printf("[DC_PLAN] Periodic Hilbert dc plan sync skipped: service identity config incomplete")
+		// Workspace resource sync does not depend on the dc plan service identity.
+		go s.syncWorkspacesOnStartup(context.Background())
 		return
 	}
 
@@ -704,6 +704,13 @@ func (s *Server) startPeriodicDCPlanSync() {
 
 	go func() {
 		defer close(done)
+
+		// Converge once immediately instead of waiting a full ticker period, so a
+		// restart does not leave workstations and pending pools stale for minutes.
+		// Both calls run in this goroutine and no longer block the listener.
+		s.syncWorkspacesOnStartup(ctx)
+		s.syncDCPlansOnce(ctx, "Startup")
+
 		ticker := time.NewTicker(dcPlanAutoSyncInterval)
 		defer ticker.Stop()
 		for {

internal/services/dc_plan_task_supply.go

@@ -458,7 +458,12 @@ func loadTaskSupplyWorkstation(
 	} else {
 		query += " ORDER BY ws.is_current DESC, ws.id DESC"
 	}
-	query += " LIMIT 1" + taskSupplyForUpdateClause(tx)
+	// Read-only lookup: this transaction goes on to write tasks only, so the
+	// workstations, data_collectors and robots rows read here are left unlocked.
+	// With FOR UPDATE this three-table join locked every row it scanned and, since
+	// this path reaches workstations before data_collectors while the projector
+	// reaches data_collectors first, the two paths could deadlock.
+	query += " LIMIT 1"
 
 	var workstation taskSupplyWorkstationRow
 	if err := tx.GetContext(ctx, &workstation, query, args...); err != nil {

internal/services/dc_plan_workstation_projection.go

@@ -146,11 +146,15 @@ func resolveProjectionCollector(
 	plan auth.HilbertDCPlan,
 ) (planProjectionCollector, error) {
 	var collector planProjectionCollector
+	// Read-only lookup: this transaction writes workstations, never data_collectors,
+	// so no row lock is taken here. Locking it made the projector hold
+	// data_collectors while waiting for workstations, which is the opposite of the
+	// order the pending pool path uses, and the two paths deadlocked.
 	if err := tx.GetContext(ctx, &collector, `
 		SELECT id, name, operator_id
 		FROM data_collectors
 		WHERE operator_id = ? AND deleted_at IS NULL
-		LIMIT 1`+projectionForUpdateClause(tx), strings.TrimSpace(plan.Operator)); err != nil {
+		LIMIT 1`, strings.TrimSpace(plan.Operator)); err != nil {
 		if err == sql.ErrNoRows {
 			return collector, fmt.Errorf("collector missing")
 		}
@@ -179,11 +183,12 @@ func resolveProjectionRobot(
 	plan auth.HilbertDCPlan,
 ) (planProjectionRobot, error) {
 	var robot planProjectionRobot
+	// Read-only lookup, for the same reason as resolveProjectionCollector.
 	if err := tx.GetContext(ctx, &robot, `
 		SELECT id, device_id, COALESCE(device_name, '') AS device_name, workspace_id
 		FROM robots
 		WHERE device_id = ? AND deleted_at IS NULL
-		LIMIT 1`+projectionForUpdateClause(tx), strconv.FormatInt(*plan.DCDeviceID, 10)); err != nil {
+		LIMIT 1`, strconv.FormatInt(*plan.DCDeviceID, 10)); err != nil {
 		if err == sql.ErrNoRows {
 			return robot, fmt.Errorf("robot missing")
 		}
@@ -204,6 +209,10 @@ func ensureProjectedWorkstation(
 	now time.Time,
 ) (bool, error) {
 	var workstation planProjectionWorkstation
+	// The lock stays here: this call either reactivates the row below or inserts a
+	// new workstation, and the partial unique indexes only cover is_current rows,
+	// so the gap lock on a missing row is what keeps two concurrent projectors
+	// from inserting duplicate workstations.
 	err := tx.GetContext(ctx, &workstation, `
 		SELECT id, superseded_at
 		FROM workstations

internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.down.sql

@@ -0,0 +1,50 @@
+-- SPDX-FileCopyrightText: 2026 ArcheBase
+-- SPDX-License-Identifier: MulanPSL-2.0
+
+-- Guarded for the same reason as the up migration: a database may carry these
+-- indexes already, and a failed statement aborts the migration.
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'tasks'
+          AND index_name = 'idx_tasks_plan_status_del') > 0,
+    'ALTER TABLE tasks DROP INDEX idx_tasks_plan_status_del',
+    'DO 0');
+PREPARE drop_idx FROM @idx_ddl;
+EXECUTE drop_idx;
+DEALLOCATE PREPARE drop_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'data_collectors'
+          AND index_name = 'idx_collectors_operator_del') > 0,
+    'ALTER TABLE data_collectors DROP INDEX idx_collectors_operator_del',
+    'DO 0');
+PREPARE drop_idx FROM @idx_ddl;
+EXECUTE drop_idx;
+DEALLOCATE PREPARE drop_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'robots'
+          AND index_name = 'idx_robots_device_del') > 0,
+    'ALTER TABLE robots DROP INDEX idx_robots_device_del',
+    'DO 0');
+PREPARE drop_idx FROM @idx_ddl;
+EXECUTE drop_idx;
+DEALLOCATE PREPARE drop_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'workstations'
+          AND index_name = 'idx_workstations_ws_del') > 0,
+    'ALTER TABLE workstations DROP INDEX idx_workstations_ws_del',
+    'DO 0');
+PREPARE drop_idx FROM @idx_ddl;
+EXECUTE drop_idx;
+DEALLOCATE PREPARE drop_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'episodes'
+          AND index_name = 'idx_episodes_del_plan_synced_qa') > 0,
+    'ALTER TABLE episodes DROP INDEX idx_episodes_del_plan_synced_qa',
+    'DO 0');
+PREPARE drop_idx FROM @idx_ddl;
+EXECUTE drop_idx;
+DEALLOCATE PREPARE drop_idx;

internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.up.sql

@@ -0,0 +1,67 @@
+-- SPDX-FileCopyrightText: 2026 ArcheBase
+-- SPDX-License-Identifier: MulanPSL-2.0
+
+-- Composite indexes backing the dc plan sync lookups.
+--
+-- Without them the planner can only fall back to single column indexes whose
+-- leading column is non selective (status = 'pending', deleted_at IS NULL), so a
+-- statement addressing one plan still scans a large index range. When that
+-- statement is a SELECT ... FOR UPDATE, InnoDB locks every row it scans: one per
+-- plan UPDATE held hundreds of row locks and took seconds, the dc plan sync could
+-- not finish inside the 180s startup probe and the container was killed before it
+-- committed, and two concurrent syncs acquiring the same rows in opposite order
+-- deadlocked.
+--
+-- Every statement is guarded because databases patched while the incident was
+-- handled already carry these indexes. A plain ADD INDEX would abort the
+-- migration, and a failed migration is fatal at startup and leaves
+-- schema_migrations_v2 dirty, which blocks every later migration.
+--
+-- The adds are pinned to ALGORITHM=INPLACE, LOCK=NONE so a database that does not
+-- carry the indexes yet builds them online instead of taking a table lock that
+-- could outlive the startup probe.
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'tasks'
+          AND index_name = 'idx_tasks_plan_status_del') = 0,
+    'ALTER TABLE tasks ADD INDEX idx_tasks_plan_status_del (dc_plan_id, status, deleted_at), ALGORITHM=INPLACE, LOCK=NONE',
+    'DO 0');
+PREPARE add_idx FROM @idx_ddl;
+EXECUTE add_idx;
+DEALLOCATE PREPARE add_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'data_collectors'
+          AND index_name = 'idx_collectors_operator_del') = 0,
+    'ALTER TABLE data_collectors ADD INDEX idx_collectors_operator_del (operator_id, deleted_at), ALGORITHM=INPLACE, LOCK=NONE',
+    'DO 0');
+PREPARE add_idx FROM @idx_ddl;
+EXECUTE add_idx;
+DEALLOCATE PREPARE add_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'robots'
+          AND index_name = 'idx_robots_device_del') = 0,
+    'ALTER TABLE robots ADD INDEX idx_robots_device_del (device_id, deleted_at), ALGORITHM=INPLACE, LOCK=NONE',
+    'DO 0');
+PREPARE add_idx FROM @idx_ddl;
+EXECUTE add_idx;
+DEALLOCATE PREPARE add_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'workstations'
+          AND index_name = 'idx_workstations_ws_del') = 0,
+    'ALTER TABLE workstations ADD INDEX idx_workstations_ws_del (workspace_id, deleted_at), ALGORITHM=INPLACE, LOCK=NONE',
+    'DO 0');
+PREPARE add_idx FROM @idx_ddl;
+EXECUTE add_idx;
+DEALLOCATE PREPARE add_idx;
+
+SET @idx_ddl := IF((SELECT COUNT(*) FROM information_schema.statistics
+        WHERE table_schema = DATABASE() AND table_name = 'episodes'
+          AND index_name = 'idx_episodes_del_plan_synced_qa') = 0,
+    'ALTER TABLE episodes ADD INDEX idx_episodes_del_plan_synced_qa (deleted_at, dc_plan_id, cloud_synced, qa_status, duration_sec), ALGORITHM=INPLACE, LOCK=NONE',
+    'DO 0');
+PREPARE add_idx FROM @idx_ddl;
+EXECUTE add_idx;
+DEALLOCATE PREPARE add_idx;