diff --git a/internal/app/segment/service.go b/internal/app/segment/service.go index 02613417..d8ed18ad 100644 --- a/internal/app/segment/service.go +++ b/internal/app/segment/service.go @@ -69,13 +69,23 @@ type service struct { fields CustomFieldLister waker CampaignWaker starter CampaignStarter - // orgSyncInFlight dedupes concurrent org-wide enrolment passes, keyed by - // org id. - orgSyncInFlight sync.Map + // orgSync coalesces org-wide enrolment passes, one entry per org that is + // currently syncing. Guarded by syncMu, which owns every transition so an + // entry is only dropped when nothing is running or queued. + syncMu sync.Mutex + orgSync map[uuid.UUID]*orgSyncState +} + +// orgSyncState is a running pass plus a "do it once more" flag. A write that +// lands mid-pass cannot be served by that pass (it read the old membership), +// so it asks for a follow-up instead of being dropped. +type orgSyncState struct { + running bool + again bool } func NewService(repo repository.SegmentRepository, fields CustomFieldLister) Service { - return &service{repo: repo, fields: fields} + return &service{repo: repo, fields: fields, orgSync: map[uuid.UUID]*orgSyncState{}} } func (s *service) SetCampaignWaker(w CampaignWaker) { s.waker = w } @@ -411,27 +421,55 @@ func (s *service) syncLinkedCampaignsForSegments(ctx context.Context, orgID uuid } func (s *service) SyncOrgLinkedCampaigns(ctx context.Context, orgID uuid.UUID) { - // One in-flight pass per org: every contact write calls this, and a burst - // of writes must not stack org-wide enrolment scans. A write landing while - // a pass runs is picked up by the sweep within its interval. - if _, running := s.orgSyncInFlight.LoadOrStore(orgID, struct{}{}); running { + // One pass per org at a time, so a burst of contact writes cannot stack + // org-wide enrolment scans. Requests that arrive mid-pass are coalesced + // into a single follow-up rather than dropped: an import writes its + // segment membership after the contact rows, so the pass already running + // read the old membership and would leave those contacts to the sweep. + s.syncMu.Lock() + st := s.orgSync[orgID] + if st == nil { + st = &orgSyncState{} + s.orgSync[orgID] = st + } + if st.running { + st.again = true + s.syncMu.Unlock() return } + st.running = true + s.syncMu.Unlock() + bg := context.WithoutCancel(ctx) go func() { - defer s.orgSyncInFlight.Delete(orgID) - rctx, cancel := context.WithTimeout(bg, 2*time.Minute) - defer cancel() - links, xerr := s.repo.LinkedCampaigns(rctx, &orgID) - if xerr != nil { + for { + s.runOrgSyncPass(bg, orgID) + s.syncMu.Lock() + if st.again { + st.again = false + s.syncMu.Unlock() + continue + } + delete(s.orgSync, orgID) + s.syncMu.Unlock() return } - for _, lc := range links { - s.syncLinkedCampaign(rctx, lc) - } }() } +// runOrgSyncPass enrols every linked campaign in one organization once. +func (s *service) runOrgSyncPass(ctx context.Context, orgID uuid.UUID) { + rctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + links, xerr := s.repo.LinkedCampaigns(rctx, &orgID) + if xerr != nil { + return + } + for _, lc := range links { + s.syncLinkedCampaign(rctx, lc) + } +} + func (s *service) StartCampaignSegmentSync(ctx context.Context, interval time.Duration) { ticker := time.NewTicker(interval) defer ticker.Stop() diff --git a/internal/app/segment/service_test.go b/internal/app/segment/service_test.go new file mode 100644 index 00000000..45d81698 --- /dev/null +++ b/internal/app/segment/service_test.go @@ -0,0 +1,125 @@ +package segment + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/repository" +) + +// syncRepo is a SegmentRepository that only answers LinkedCampaigns, blocking +// there until released so a second request provably arrives mid-pass. Every +// other method is left to the embedded nil interface: this test must not reach +// them, and a panic is a clearer failure than a silent zero value. +type syncRepo struct { + repository.SegmentRepository + entered chan struct{} + release chan struct{} + + mu sync.Mutex + calls int +} + +func (r *syncRepo) LinkedCampaigns(context.Context, *uuid.UUID) ([]models.LinkedCampaign, *errx.Error) { + r.mu.Lock() + r.calls++ + r.mu.Unlock() + r.entered <- struct{}{} + <-r.release + return nil, nil +} + +func (r *syncRepo) callCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return r.calls +} + +func waitFor(t *testing.T, c chan struct{}, what string) { + t.Helper() + select { + case <-c: + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for %s", what) + } +} + +// A sync requested while a pass is running must run again afterwards. The +// import path depends on it: ImportCommit's chunked Add starts a pass, then +// writes segment membership, then asks for a sync. Dropping that request +// leaves freshly pinned contacts out of their linked campaigns until the +// periodic sweep. +func TestSyncOrgLinkedCampaignsCoalescesMidPassRequest(t *testing.T) { + repo := &syncRepo{entered: make(chan struct{}, 4), release: make(chan struct{})} + svc := NewService(repo, nil).(*service) + org := uuid.New() + ctx := context.Background() + + svc.SyncOrgLinkedCampaigns(ctx, org) + waitFor(t, repo.entered, "the first pass to start") + + // Two requests land while the first pass is blocked: both are folded into + // one follow-up, not stacked and not lost. + svc.SyncOrgLinkedCampaigns(ctx, org) + svc.SyncOrgLinkedCampaigns(ctx, org) + if got := repo.callCount(); got != 1 { + t.Fatalf("passes started while one was running = %d, want 1", got) + } + + repo.release <- struct{}{} + waitFor(t, repo.entered, "the coalesced follow-up pass") + repo.release <- struct{}{} + + // The follow-up drains the flag, so nothing runs a third time. + deadline := time.After(300 * time.Millisecond) + for { + select { + case <-repo.entered: + t.Fatalf("a third pass ran; the follow-up flag was not cleared") + case <-deadline: + if got := repo.callCount(); got != 2 { + t.Fatalf("total passes = %d, want 2", got) + } + svc.syncMu.Lock() + _, still := svc.orgSync[org] + svc.syncMu.Unlock() + if still { + t.Errorf("org sync state was not released once idle") + } + return + } + } +} + +// Back-to-back requests with no overlap each get their own pass. +func TestSyncOrgLinkedCampaignsRunsAgainAfterIdle(t *testing.T) { + repo := &syncRepo{entered: make(chan struct{}, 4), release: make(chan struct{})} + svc := NewService(repo, nil).(*service) + org := uuid.New() + ctx := context.Background() + + for i := 0; i < 2; i++ { + svc.SyncOrgLinkedCampaigns(ctx, org) + waitFor(t, repo.entered, "a pass to start") + repo.release <- struct{}{} + // Let the goroutine retire the state before asking again. + for j := 0; j < 100; j++ { + svc.syncMu.Lock() + _, running := svc.orgSync[org] + svc.syncMu.Unlock() + if !running { + break + } + time.Sleep(5 * time.Millisecond) + } + } + if got := repo.callCount(); got != 2 { + t.Fatalf("passes = %d, want 2", got) + } +}