From 82d817c331fb40ba826bd288b0082915d25197fd Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Fri, 2 Oct 2026 11:50:07 -0700 Subject: [PATCH] Flush the records writer tests synchronously static-test failed TestWriterBackoffGrowsAndResets and TestWriterLivestreamSubjects on CI's slower runner. The harness flushed by sending the session's tick twice, so a second flush was still running when flushed() returned. On a loaded machine it took in the words the test said next and published them before buffered() could see them, and buffered() waited out its 5 s. Delaying that flush by 20 ms and the test by 50 ms after each say reproduces both failures locally. flushed() now runs one flush of the session to completion and the tick never fires, so sessions flush only there and when they end. The test that needs the session's own flush stuck in the publisher keeps a tick of its own. --- pkg/captions/records/writer_test.go | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/pkg/captions/records/writer_test.go b/pkg/captions/records/writer_test.go index 7a10ebaf1..06de31ee8 100644 --- a/pkg/captions/records/writer_test.go +++ b/pkg/captions/records/writer_test.go @@ -82,7 +82,6 @@ type harness struct { hub *captions.Hub repos *fakeRepos w *Writer - tick chan time.Time now time.Time nowMu sync.Mutex } @@ -101,7 +100,7 @@ func (h *harness) advance(d time.Duration) { func newHarness(t *testing.T, mutate func(*Config)) *harness { t.Helper() - h := &harness{t: t, hub: captions.NewHub(0), repos: &fakeRepos{}, tick: make(chan time.Time), now: t0.Add(time.Hour)} + h := &harness{t: t, hub: captions.NewHub(0), repos: &fakeRepos{}, now: t0.Add(time.Hour)} cfg := Config{ Hub: h.hub, NodeDID: nodeDID, @@ -109,7 +108,7 @@ func newHarness(t *testing.T, mutate func(*Config)) *harness { return comatproto.RepoStrongRef{LexiconTypeID: "com.atproto.repo.strongRef", Uri: "at://" + streamer + "/place.stream.livestream/3abc", Cid: "bafylive"}, nil }, Publisher: h.repos, - Tick: h.tick, + Tick: make(chan time.Time), // never fires; see flushed Now: h.clock, RetryDelay: time.Millisecond, StopTimeout: 5 * time.Second, @@ -124,12 +123,16 @@ func newHarness(t *testing.T, mutate func(*Config)) *harness { return h } -// flushed triggers a flush and returns once it has run. The tick channel is -// unbuffered, so the second send returns only after the first flush finished. +// flushed runs one flush of the streamer's session to completion. Sessions +// flush only here and when they end: one flushing on its own could take in, +// between a test's say and buffered, the words the test is waiting for. func (h *harness) flushed() { h.t.Helper() - h.tick <- time.Time{} - h.tick <- time.Time{} + h.w.mu.Lock() + s := h.w.sessions[streamer] + h.w.mu.Unlock() + require.NotNil(h.t, s, "no session to flush") + s.flush(context.Background(), false) } // buffered waits until the session has taken in n words, so a flush sees them. @@ -546,12 +549,13 @@ func TestWriterGivesUpAtSessionEndWithoutHanging(t *testing.T) { } func TestWriterNeverBlocksTheHub(t *testing.T) { - h := newHarness(t, nil) + tick := make(chan time.Time) + h := newHarness(t, func(c *Config) { c.Tick = tick }) h.repos.block = make(chan struct{}) h.w.StartSession(context.Background(), streamer, t0) say(h, autoTrack, "c0", 0, "x") h.buffered(streamer, 1) - go func() { h.tick <- time.Time{} }() // the flush now hangs inside the publisher + go func() { tick <- time.Time{} }() // the session's flush now hangs inside the publisher done := make(chan struct{}) go func() { -- knotmirror 0.0.1