feat: coalesce org segment syncs instead of dropping them, so an import's pinned contacts reach their linked campaigns in the same request: SyncOrgLinkedCampaigns now records a follow-up when a request lands while a pass is running (ImportCommit writes segment membership after its chunked Add already started one, so that pass read the old membership and left the contacts to the 2-minute sweep), the per-org state moves from a sync.Map to a mutex-guarded map so running, queued and released are one atomic transition, and two hermetic tests cover the mid-pass coalesce and the idle re-run

2026-09-05 16:02:48 +00:00 · 2026-09-01 04:19:59 -07:00
parent 4385f6ec02
commit 271b4b1625
2 changed files with 179 additions and 16 deletions
@@ -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()
@@ -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)
+	}
+}