diff --git a/internal/api/handlers/dc_plan.go b/internal/api/handlers/dc_plan.go index e7ff42c..a8aba37 100644 --- a/internal/api/handlers/dc_plan.go +++ b/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"` diff --git a/internal/api/handlers/operator_plan_test.go b/internal/api/handlers/operator_plan_test.go index c2a9d91..9fd5c7b 100644 --- a/internal/api/handlers/operator_plan_test.go +++ b/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) } } diff --git a/internal/server/server.go b/internal/server/server.go index 016eebd..3ce98d4 100644 --- a/internal/server/server.go +++ b/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,12 +673,12 @@ 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 @@ -684,13 +686,11 @@ func (s *Server) syncWorkspacesOnStartup() { 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 { diff --git a/internal/services/dc_plan_task_supply.go b/internal/services/dc_plan_task_supply.go index 0477c7c..28fa70c 100644 --- a/internal/services/dc_plan_task_supply.go +++ b/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 { diff --git a/internal/services/dc_plan_workstation_projection.go b/internal/services/dc_plan_workstation_projection.go index c922f5b..fcff05f 100644 --- a/internal/services/dc_plan_workstation_projection.go +++ b/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 diff --git a/internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.down.sql b/internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.down.sql new file mode 100644 index 0000000..81702c8 --- /dev/null +++ b/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; diff --git a/internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.up.sql b/internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.up.sql new file mode 100644 index 0000000..be854cb --- /dev/null +++ b/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;