package api import ( "strconv" "sync" "testing" ) // TestSubscribeSnapshot_NoDuplicates regresses the subscribe/snapshot race: // subscribers arriving while events are being published must never see the // same event twice (once from the snapshot replay and once from the channel). // Events are numbered, so any duplicate breaks strict monotonicity. Slow // subscribers may legitimately drop events (publishLocked is non-blocking), // so gaps are not an error. func TestSubscribeSnapshot_NoDuplicates(t *testing.T) { job := &TraversalJob{Status: statusRunning} const total = 2000 const subscribers = 8 var pub sync.WaitGroup pub.Add(1) go func() { defer pub.Done() for i := 0; i < total; i++ { ev := ProgressEvent{RefID: strconv.Itoa(i)} job.mu.Lock() job.Progress = append(job.Progress, ev) job.publishLocked(ev) job.mu.Unlock() } job.mu.Lock() job.Status = statusComplete job.mu.Unlock() job.closeSubscribers() }() var subs sync.WaitGroup errs := make(chan string, subscribers) for s := 0; s < subscribers; s++ { subs.Add(1) go func() { defer subs.Done() sub, past, done := job.subscribeSnapshot() defer job.unsubscribe(sub) last := -1 check := func(refid string) { n, err := strconv.Atoi(refid) if err != nil { errs <- "bad refid " + refid return } if n <= last { errs <- "event " + refid + " out of order or duplicated after " + strconv.Itoa(last) return } last = n } for _, ev := range past { check(ev.RefID) } if !done { for ev := range sub { check(ev.RefID) } } }() } pub.Wait() subs.Wait() close(errs) for msg := range errs { t.Error(msg) } }