From 149fb0a18c9e6b768d91dd2c1fc063465268443e Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 6 Apr 2026 23:59:26 +0700 Subject: [PATCH] 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 | 23 ++++++++++ internal/jobs/worker.go | 34 +++++++++++++- internal/jobs/worker_test.go | 88 ++++++++++++++++++++++++++++++++++++ 3 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 internal/jobs/worker_test.go diff --git a/internal/jobs/job.go b/internal/jobs/job.go index 3cf0262..b33f5a3 100644 --- a/internal/jobs/job.go +++ b/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 diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index babe217..0131b7c 100644 --- a/internal/jobs/worker.go +++ b/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) - j.Succeed() + w.finishModuleRefreshSuccess(j, rev) 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 { diff --git a/internal/jobs/worker_test.go b/internal/jobs/worker_test.go new file mode 100644 index 0000000..d1f6bba --- /dev/null +++ b/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") +}