feat: implement module refresh job management with concurrency control. Add CountOtherActiveModuleRefresh method to track active jobs and enhance finishModuleRefreshSuccess to defer deploy_apply when parallel refreshes are detected, improving job processing efficiency.

internal/jobs/job.go

@@ -267,6 +267,29 @@ func (r *Registry) List(tenantID, statusFilter, kindFilter, cursor string, limit
 	return all[off:end], next, hasMore
 }
 
+// CountOtherActiveModuleRefresh returns how many module_refresh jobs for the tenant are still
+// queued or running, excluding excludeJobID (the current job). Used to batch deploy_apply.
+func (r *Registry) CountOtherActiveModuleRefresh(tenantID, excludeJobID string) int {
+	if r == nil {
+		return 0
+	}
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	n := 0
+	for _, j := range r.byID {
+		if j.TenantID != tenantID || j.Kind != KindModuleRefresh {
+			continue
+		}
+		if j.ID == excludeJobID {
+			continue
+		}
+		if j.Status == StatusQueued || j.Status == StatusRunning {
+			n++
+		}
+	}
+	return n
+}
+
 func parseCursor(s string, off *int) error {
 	_, err := fmt.Sscanf(s, "%d", off)
 	return err

internal/jobs/worker.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"evobgp/internal/birddeploy"
@@ -52,6 +53,8 @@ type Worker struct {
 	HTTPClient *http.Client // optional; CDN refresh uses this (default 45s timeout).
 	// Registry is set after BootstrapWorkers creates the job queue; used to chain deploy_apply after refresh/rollback.
 	Registry *Registry
+	// refreshGate serializes deploy_apply gating after module_refresh per tenant (see finishModuleRefreshSuccess).
+	refreshGate sync.Map // map[string]*sync.Mutex
 }
 
 type revisionLogEntry struct {
@@ -111,8 +114,7 @@ func (w *Worker) Process(j *Job) {
 		} else {
 			j.mergeMeta(map[string]any{"log_build_error": err.Error()})
 		}
-		w.enqueueDeployAllSpeakers(j, j.TenantID, rev)
+		w.finishModuleRefreshSuccess(j, rev)
-		j.Succeed()
 	case KindDeployApply:
 		w.runDeployApply(j)
 	case KindRevisionRollback:
@@ -138,6 +140,34 @@ func (w *Worker) Process(j *Job) {
 	}
 }
 
+func (w *Worker) tenantRefreshMu(tenantID string) *sync.Mutex {
+	v, _ := w.refreshGate.LoadOrStore(tenantID, &sync.Mutex{})
+	return v.(*sync.Mutex)
+}
+
+// finishModuleRefreshSuccess marks the job succeeded and enqueues deploy_apply only when no other
+// module_refresh is still queued or running for the same tenant (coalesces parallel refreshes).
+func (w *Worker) finishModuleRefreshSuccess(j *Job, rev string) {
+	if w == nil || w.Registry == nil {
+		j.Succeed()
+		return
+	}
+	mu := w.tenantRefreshMu(j.TenantID)
+	mu.Lock()
+	deferDeploy := w.Registry.CountOtherActiveModuleRefresh(j.TenantID, j.ID) > 0
+	if deferDeploy {
+		j.mergeMeta(map[string]any{
+			"deploy_apply_deferred":        true,
+			"deploy_apply_defer_reason":    "parallel_module_refresh",
+		})
+	}
+	j.Succeed()
+	mu.Unlock()
+	if !deferDeploy {
+		w.enqueueDeployAllSpeakers(j, j.TenantID, rev)
+	}
+}
+
 // enqueueDeployAllSpeakers queues the same work as POST /v1/apply (all speakers, no speaker_id).
 func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) {
 	if w == nil || w.Registry == nil {

internal/jobs/worker_test.go

@@ -0,0 +1,88 @@
+package jobs
+
+import (
+	"testing"
+	"time"
+
+	"evobgp/internal/store"
+)
+
+func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) {
+	t.Setenv("EVOBGP_ASN_RESOLVE", "0")
+	t.Setenv("EVOBGP_BIRD_ACTIVE_DIR", "") // skip bird binary path in deploy_apply
+
+	m := store.NewMemory()
+	m.SeedDemo()
+	tenant, _, modIP, _, _ := m.DemoIDs()
+	for _, mod := range m.ListModules(tenant) {
+		if mod == nil || mod.ID == modIP {
+			continue
+		}
+		disabled := false
+		if _, err := m.UpdateModule(tenant, mod.ID, &store.ModulePatch{Enabled: &disabled}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	mod2, err := m.CreateModule(tenant, &store.Module{Type: "IP_RANGES", Name: "extra-ip", Enabled: true, Priority: 30})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := m.CreateIPRangeEntry(tenant, modIP, &store.IPRangeEntry{Prefix: "10.0.0.0/24"}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := m.CreateIPRangeEntry(tenant, mod2.ID, &store.IPRangeEntry{Prefix: "192.168.0.0/24"}); err != nil {
+		t.Fatal(err)
+	}
+
+	wk := &Worker{Store: m}
+	reg := NewRegistry(wk.Process)
+	wk.Registry = reg
+
+	mid1 := modIP
+	mid2 := mod2.ID
+	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid1, map[string]any{"module_id": modIP}); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid2, map[string]any{"module_id": mod2.ID}); err != nil {
+		t.Fatal(err)
+	}
+
+	waitSucceededModuleRefreshCount(t, reg, tenant, 2)
+
+	deployJobs, _, _ := reg.List(tenant, "", KindDeployApply, "", 100)
+	if len(deployJobs) != 1 {
+		t.Fatalf("want exactly one deploy_apply job, got %d", len(deployJobs))
+	}
+
+	var deferred, withDeployID int
+	refreshJobs, _, _ := reg.List(tenant, "", KindModuleRefresh, "", 100)
+	for _, j := range refreshJobs {
+		if j.Status != StatusSucceeded {
+			continue
+		}
+		meta := j.Snapshot()["meta"].(map[string]any)
+		if v, ok := meta["deploy_apply_deferred"].(bool); ok && v {
+			deferred++
+		}
+		if _, ok := meta["deploy_apply_job_id"].(string); ok {
+			withDeployID++
+		}
+	}
+	if deferred != 1 || withDeployID != 1 {
+		t.Fatalf("want one deferred and one with deploy_apply_job_id, got deferred=%d deploy_meta=%d", deferred, withDeployID)
+	}
+}
+
+func waitSucceededModuleRefreshCount(t *testing.T, reg *Registry, tenant string, want int) {
+	t.Helper()
+	deadline := time.Now().Add(30 * time.Second)
+	for time.Now().Before(deadline) {
+		jobs, _, _ := reg.List(tenant, StatusSucceeded, KindModuleRefresh, "", 100)
+		if len(jobs) >= want {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatal("timeout waiting for module_refresh jobs")
+}