diff --git a/internal/jobs/worker_test.go b/internal/jobs/worker_test.go index 40323a9..c91c7a8 100644 --- a/internal/jobs/worker_test.go +++ b/internal/jobs/worker_test.go @@ -10,6 +10,7 @@ import ( 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 + t.Setenv("EVOBGP_JOB_MAX_CONCURRENT", "8") m := store.NewMemory() m.SeedDemo() @@ -35,8 +36,15 @@ func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) { t.Fatal(err) } + // Hold workers until both jobs are enqueued so inflightRefresh=2 before either + // finishModuleRefreshSuccess. Otherwise a fast ingest can finalize+deploy before + // the second Enqueue — sequential refreshes correctly produce two deploy_apply jobs. + start := make(chan struct{}) wk := &Worker{Store: m} - reg := NewRegistry(wk.Process) + reg := NewRegistry(func(j *Job) { + <-start + wk.Process(j) + }) wk.Registry = reg mid1 := modIP @@ -47,6 +55,7 @@ func TestParallelModuleRefresh_CoalescesDeployApply(t *testing.T) { if _, _, err := reg.Enqueue(tenant, KindModuleRefresh, nil, &mid2, map[string]any{"module_id": mod2.ID}); err != nil { t.Fatal(err) } + close(start) waitSucceededJobsByKindCount(t, reg, tenant, KindModuleRefresh, 2) diff --git a/internal/store/memory.go b/internal/store/memory.go index 6e561de..043e8f0 100644 --- a/internal/store/memory.go +++ b/internal/store/memory.go @@ -414,6 +414,9 @@ func (m *Memory) ListModules(tenantID string) []*Module { } return out[i].Name < out[j].Name }) + for i := range out { + out[i] = cloneModule(out[i]) + } return out } @@ -446,7 +449,7 @@ func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) { if mod.TenantID != tenantID { return nil, ErrTenantScope } - return mod, nil + return cloneModule(mod), nil } func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) { @@ -684,3 +687,24 @@ func (m *Memory) ListRevisions(tenantID, moduleID string, cursor string, limit i } return page, nextCursor, hasMore } + +func cloneStringPtr(s *string) *string { + if s == nil { + return nil + } + v := *s + return &v +} + +func cloneModule(m *Module) *Module { + if m == nil { + return nil + } + cp := *m + cp.DefaultCommunityID = cloneStringPtr(m.DefaultCommunityID) + cp.DohProfileID = cloneStringPtr(m.DohProfileID) + cp.DohProfileIDs = append([]string(nil), m.DohProfileIDs...) + cp.LastRefreshedAt = cloneTime(m.LastRefreshedAt) + cp.DeletedAt = cloneTime(m.DeletedAt) + return &cp +} diff --git a/internal/store/memory_crud.go b/internal/store/memory_crud.go index 022d7e4..db4e8ec 100644 --- a/internal/store/memory_crud.go +++ b/internal/store/memory_crud.go @@ -38,7 +38,7 @@ func (m *Memory) CreateModule(tenantID string, in *Module) (*Module, error) { } NormalizeModuleDoh(mod) m.modules[id] = mod - return mod, nil + return cloneModule(mod), nil } func (m *Memory) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*Module, error) { @@ -74,12 +74,14 @@ func (m *Memory) UpdateModule(tenantID, moduleID string, patch *ModulePatch) (*M mod.DefaultCommunityID = &v } } - ApplyModuleDohPatch(mod, patch) + if patch.DohProfileIDs != nil || patch.DohProfileID != nil || patch.DohResolverPolicy != nil { + ApplyModuleDohPatch(mod, patch) + } if patch.LastRefreshedAt != nil { t := patch.LastRefreshedAt.UTC() mod.LastRefreshedAt = &t } - return mod, nil + return cloneModule(mod), nil } func (m *Memory) SoftDeleteModule(tenantID, moduleID string) error {