From a52ac4eb613ad0de0982d02703a8129e78b9cbb7 Mon Sep 17 00:00:00 2001 From: chaoliu Date: Thu, 17 Sep 2026 14:26:21 +0800 Subject: [PATCH 1/4] fix(server): decouple dc plan sync from startup readiness Start() ran the workspace and dc_plan startup syncs before starting the HTTP listener. A slow Hilbert round therefore blocked readiness, the startup probe killed the container after 180s, the sync transaction rolled back with the sync not committed, and the next attempt repeated exactly the same work forever. Move both startup syncs into the periodic DC plan sync goroutine and start that goroutine after the listener, so convergence can never gate readiness. The goroutine now converges once immediately before entering the ticker loop, which also removes the previous wait of up to one full ticker period after a restart, and it uses the cancellable context so shutdown can interrupt an in-flight sync. --- internal/server/server.go | 27 +++++++++++++++++---------- 1 file changed, 17 insertions(+), 10 deletions(-) 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 { From 20e156e7077b6a2fe9ef658d453e08bf444aa56f Mon Sep 17 00:00:00 2001 From: chaoliu Date: Thu, 17 Sep 2026 14:28:26 +0800 Subject: [PATCH 2/4] fix(dc-plan): stop syncing workspace projections on refresh POST /operator/plans/refresh ran a full SyncWorkspace inline before returning the operator's plans. One SyncWorkspace costs O(plan count) transactions: the workstation projector begins a transaction per plan and the Ego Portal pending pool begins another, so a 150-plan workspace cost about 300 transactions and a 583-plan workspace about 1166. With every activated device polling the endpoint, requests amplified 1:1 into full workspace syncs, exhausted the 25-connection pool and produced thousands of deadlocks and lock wait timeouts per minute. Return the current projection instead and leave convergence to the periodic hilbert sync, which already runs every five minutes and repairs missing workstations idempotently. Plans and workstations can now lag by up to one sync period, which the collection workflow accepts. TestRefreshOperatorPlansFallsBackToStaleProjection tested the removed inline path (it asserted stale=true when the inline sync failed), so it is replaced with TestRefreshOperatorPlansDoesNotSyncInline, which fails on the previous behaviour with five syncs for five requests. --- internal/api/handlers/dc_plan.go | 19 +++----- internal/api/handlers/operator_plan_test.go | 50 ++++++++++++++------- 2 files changed, 41 insertions(+), 28 deletions(-) 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) } } From 9dcf29ce318cb39f89c28704a72186e6dddb4846 Mon Sep 17 00:00:00 2001 From: chaoliu Date: Thu, 17 Sep 2026 14:30:35 +0800 Subject: [PATCH 3/4] fix(dc-plan): index the dc plan sync lookups Records the indexes that were created by hand while the dc plan sync outage was being handled, so a rebuilt database cannot reproduce it. Without these indexes the planner falls 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. With SELECT ... FOR UPDATE InnoDB locks every row it scans, which is how one per plan UPDATE came to hold hundreds of row locks for seconds and how the two sync paths deadlocked. The statements are guarded with information_schema checks because databases patched during the incident 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, blocking every later migration. Verified against the production database, where all five indexes already exist, so the up migration is a no-op and exits cleanly. --- ...00023_dc_plan_sync_lookup_indexes.down.sql | 50 ++++++++++++++ .../000023_dc_plan_sync_lookup_indexes.up.sql | 67 +++++++++++++++++++ 2 files changed, 117 insertions(+) create mode 100644 internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.down.sql create mode 100644 internal/storage/database/migrations/v2/000023_dc_plan_sync_lookup_indexes.up.sql 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; From 602785e7a7f4e0e387e7d5894a24d9d2b9aed2cc Mon Sep 17 00:00:00 2001 From: chaoliu Date: Thu, 17 Sep 2026 15:13:46 +0800 Subject: [PATCH 4/4] fix(dc-plan): drop row locks on read-only sync lookups The workstation projector and the pending pool both lock rows they only read, and they reach those tables in opposite order: pending pool: workstations -> data_collectors -> robots projector: data_collectors -> robots -> workstations InnoDB reported the resulting deadlock as one transaction holding workstations while waiting for data_collectors, and the other holding data_collectors while waiting for workstations. The index added earlier shrank each footprint from hundreds of rows to one or two, which stopped the observed deadlocks, but the ordering conflict is unchanged: any second concurrent sync of the same workspace brings it back. None of the three lookups below writes the table it reads, and the only write that follows is against tasks, so the lock buys nothing. The lock in ensureProjectedWorkstation is kept deliberately. It is the one that matters for correctness: the unique indexes on workstations only cover is_current rows, so the gap lock taken on a missing row is what stops two concurrent projectors from inserting duplicate workstations. The test suite cannot cover this change. Tests run on sqlite and both helpers return an empty clause for sqlite, so FOR UPDATE is never exercised; the change rests on review and on production observation. --- internal/services/dc_plan_task_supply.go | 7 ++++++- internal/services/dc_plan_workstation_projection.go | 13 +++++++++++-- 2 files changed, 17 insertions(+), 3 deletions(-) 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