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/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")
+}