diff --git a/internal/jobs/job.go b/internal/jobs/job.go index 513db7a..2f91b13 100644 --- a/internal/jobs/job.go +++ b/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), - byIdempo: make(map[idempoKey]*Job), - workerStart: workerStart, - workerSem: make(chan struct{}, maxWorkers), + byID: make(map[string]*Job), + byIdempo: make(map[idempoKey]*Job), + workerStart: workerStart, + 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 diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index a06f5c3..ccd20ac 100644 --- a/internal/jobs/worker.go +++ b/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) - 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) + w.runModuleRefresh(j) 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",