fix(jobs): make deploy coalescing deterministic via inflight counter

TestParallelModuleRefresh_CoalescesDeployApply падал на CI под -race
(want exactly one deploy_apply job, got 2). Локально тест проходил
стабильно (100/500 итераций с -cpu), но узкая гонка проявлялась при
замедлении под race-детектором.

Корень: коалесцирование решало «делать ли deploy» через
CountOtherActiveRefresh, который опрашивал статусы job-ов (queued/
running). Статусы меняются асинхронно относительно tenantRefreshMu,
поэтому в редких таймингах оба параллельных refresh могли решить,
что другой уже не активен, и каждый породил свой deploy_apply.

Решение — детерминированный inflight-счётчик refresh-kind job-ов в
Registry (inflightRefresh map[string]int), управляемый под r.mu:
- инкремент в Enqueue при создании нового refresh-kind job-а;
- декремент + проверка «последний ли я» в finishModuleRefreshSuccess
  через новый метод finalizeRefreshCoalesce (под tenantRefreshMu).

Последний refresh (счётчик <= 1) делает render + deploy_apply; все
остальные defer-ят. Решение больше не зависит от опроса статусов и
таймингов ingest.

Чтобы счётчик не утёк на error/cancel путях (где refresh не доходит
до finishModuleRefreshSuccess), обработка module_refresh и
tenant_refresh вынесена в runModuleRefresh / runTenantRefresh с
defer-обёрткой, которая гарантированно освобождает слот, если
finishModuleRefreshSuccess не отработал.

CountOtherActiveRefresh / CountOtherActiveModuleRefresh оставлены
как публичные методы (могут использоваться в мониторинге); из
продакшн-логики коалесцирования убраны.

Проверки: go build, go vet, go test ./internal/... -count=1 — exit 0.
Стресс-тест коалесцирования: 200 итераций с -cpu=4 — стабильно.

Co-authored-by: Cursor <[email protected]>

internal/jobs/job.go

@@ -184,6 +184,10 @@ type Registry struct {
 	onTerminal  func(j *Job)
 	onEnqueued  func(j *Job)
 	onRunning   func(j *Job)
+	// inflightRefresh counts refresh-kind jobs (module_refresh, tenant_refresh) per tenant that
+	// have been enqueued but not yet finalized in finishModuleRefreshSuccess. Used for deterministic
+	// deploy coalescing under tenantRefreshMu (instead of polling job statuses).
+	inflightRefresh map[string]int
 }
 
 type idempoKey struct {
@@ -194,10 +198,11 @@ type idempoKey struct {
 func NewRegistry(workerStart func(j *Job)) *Registry {
 	maxWorkers := registryMaxConcurrentJobs()
 	return &Registry{
-		byID:        make(map[string]*Job),
+		byID:            make(map[string]*Job),
-		byIdempo:    make(map[idempoKey]*Job),
+		byIdempo:        make(map[idempoKey]*Job),
-		workerStart: workerStart,
+		workerStart:     workerStart,
-		workerSem:   make(chan struct{}, maxWorkers),
+		workerSem:       make(chan struct{}, maxWorkers),
+		inflightRefresh: make(map[string]int),
 	}
 }
 
@@ -340,6 +345,9 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 		r.byIdempo[idempoKey{tenant: tenantID, key: *idempotencyKey}] = j
 	}
 	r.byID[j.ID] = j
+	if isRefreshKind(kind) {
+		r.inflightRefresh[tenantID]++
+	}
 	r.pruneTerminalIfOver(maxJobs)
 	enqueuedHook := r.onEnqueued
 	workerStart := r.workerStart
@@ -474,6 +482,34 @@ func (r *Registry) CountOtherActiveRefresh(tenantID, excludeJobID string) int {
 	return n
 }
 
+// isRefreshKind reports whether a job kind participates in deploy coalescing.
+func isRefreshKind(kind string) bool {
+	return kind == KindModuleRefresh || kind == KindTenantRefresh
+}
+
+// finalizeRefreshCoalesce is called from finishModuleRefreshSuccess under tenantRefreshMu.
+// It atomically decrements the per-tenant inflight refresh counter and reports whether the
+// caller is the last outstanding refresh for the tenant (and therefore should render+deploy).
+//
+// Unlike CountOtherActiveRefresh (which polls job statuses and races under -race), this counter
+// is incremented in Enqueue under r.mu and decremented here, so the "last one" decision is
+// deterministic regardless of how fast each refresh's ingest completes.
+func (r *Registry) finalizeRefreshCoalesce(tenantID string) bool {
+	if r == nil {
+		return true
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	n := r.inflightRefresh[tenantID]
+	if n <= 1 {
+		// Last (or already-balanced to zero) — clear the slot and let the caller deploy.
+		delete(r.inflightRefresh, tenantID)
+		return true
+	}
+	r.inflightRefresh[tenantID] = n - 1
+	return false
+}
+
 func parseCursor(s string, off *int) error {
 	_, err := fmt.Sscanf(s, "%d", off)
 	return err

internal/jobs/worker.go

@@ -119,26 +119,7 @@ func (w *Worker) Process(j *Job) {
 
 	switch j.Kind {
 	case KindModuleRefresh:
-		mid, _ := j.Meta["module_id"].(string)
+		w.runModuleRefresh(j)
-		if strings.TrimSpace(mid) == "" {
-			j.Fail("missing module_id in job meta")
-			return
-		}
-		ctx, cancel := j.workContext()
-		defer cancel()
-		if ctx.Err() != nil {
-			j.MarkCancelled()
-			return
-		}
-		if err := pipeline.RefreshModuleIngest(ctx, w.Store, w.httpClient(), j.TenantID, mid); err != nil {
-			if ctx.Err() != nil {
-				j.MarkCancelled()
-				return
-			}
-			j.Fail(err.Error())
-			return
-		}
-		w.finishModuleRefreshSuccess(j, mid)
 	case KindTenantRefresh:
 		w.runTenantRefresh(j)
 	case KindPeerReconcile:
@@ -295,7 +276,55 @@ func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex {
 
 // finishModuleRefreshSuccess marks the refresh job and, for the last active refresh in tenant,
 // creates one aggregate revision and enqueues a single deploy_apply.
+// runModuleRefresh handles a single module_refresh job and guarantees the per-tenant inflight
+// slot is released exactly once — even on failure/cancellation before finishModuleRefreshSuccess.
+func (w *Worker) runModuleRefresh(j *Job) {
+	coalesceFinalized := false
+	defer func() {
+		if !coalesceFinalized && w != nil && w.Registry != nil {
+			// Refresh failed/was cancelled before reaching finishModuleRefreshSuccess.
+			// Decrement the counter under the tenant mutex so the "last one" logic stays sound.
+			mu := w.tenantRefreshMu(j.TenantID)
+			mu.Lock()
+			w.Registry.finalizeRefreshCoalesce(j.TenantID)
+			mu.Unlock()
+		}
+	}()
+
+	mid, _ := j.Meta["module_id"].(string)
+	if strings.TrimSpace(mid) == "" {
+		j.Fail("missing module_id in job meta")
+		return
+	}
+	ctx, cancel := j.workContext()
+	defer cancel()
+	if ctx.Err() != nil {
+		j.MarkCancelled()
+		return
+	}
+	if err := pipeline.RefreshModuleIngest(ctx, w.Store, w.httpClient(), j.TenantID, mid); err != nil {
+		if ctx.Err() != nil {
+			j.MarkCancelled()
+			return
+		}
+		j.Fail(err.Error())
+		return
+	}
+	w.finishModuleRefreshSuccess(j, mid)
+	coalesceFinalized = true
+}
+
 func (w *Worker) runTenantRefresh(j *Job) {
+	coalesceFinalized := false
+	defer func() {
+		if !coalesceFinalized && w != nil && w.Registry != nil {
+			mu := w.tenantRefreshMu(j.TenantID)
+			mu.Lock()
+			w.Registry.finalizeRefreshCoalesce(j.TenantID)
+			mu.Unlock()
+		}
+	}()
+
 	moduleIDs := moduleIDsFromJobMeta(j.Meta)
 	if len(moduleIDs) == 0 {
 		j.Fail("missing module_ids in job meta")
@@ -318,6 +347,7 @@ func (w *Worker) runTenantRefresh(j *Job) {
 	}
 	j.mergeMeta(map[string]any{"module_ids": moduleIDs, "modules_refreshed": len(moduleIDs)})
 	w.finishModuleRefreshSuccess(j, trigger)
+	coalesceFinalized = true
 }
 
 func moduleIDsFromJobMeta(meta map[string]any) []string {
@@ -346,6 +376,9 @@ func moduleIDsFromJobMeta(meta map[string]any) []string {
 
 func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
 	if w == nil || w.Store == nil {
+		if w != nil && w.Registry != nil {
+			w.Registry.finalizeRefreshCoalesce(j.TenantID)
+		}
 		j.Succeed()
 		return
 	}
@@ -354,11 +387,15 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, triggerModuleID string) {
 	defer mu.Unlock()
 	ctx, cancel := j.workContext()
 	defer cancel()
-	deferDeploy := false
+	// Determine whether this is the last outstanding refresh for the tenant. The counter is
+	// incremented in Enqueue (under r.mu) and decremented here, so the "last one" decision is
+	// deterministic regardless of ingest timing — unlike the previous status-polling approach
+	// (CountOtherActiveRefresh) which could race under -race.
+	isLastRefresh := true
 	if w.Registry != nil {
-		deferDeploy = w.Registry.CountOtherActiveRefresh(j.TenantID, j.ID) > 0
+		isLastRefresh = w.Registry.finalizeRefreshCoalesce(j.TenantID)
 	}
-	if deferDeploy {
+	if !isLastRefresh {
 		j.mergeMeta(map[string]any{
 			"deploy_apply_deferred":     true,
 			"deploy_apply_defer_reason": "parallel_module_refresh",