diff --git a/concurrent_test.go b/concurrent_test.go index f6c8dc1..4f0e691 100644 --- a/concurrent_test.go +++ b/concurrent_test.go @@ -5,10 +5,12 @@ import ( "log/slog" "net/http" "runtime" + "slices" "strings" "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/coder/chat" @@ -110,67 +112,59 @@ func TestConcurrentRunsSameThreadHandlersInParallelWithoutLock(t *testing.T) { func TestConcurrentBoundedByMaxConcurrent(t *testing.T) { t.Parallel() - state := newFakeState() - adapter := newFakeAdapter("fake") - var logs syncBuffer - bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { - o.MaxConcurrent = 1 - }) + synctest.Test(t, func(t *testing.T) { + state := newFakeState() + adapter := newFakeAdapter("fake") + var logs syncBuffer + const detachTimeout = time.Hour + bot := newConcurrentRuntime(t, state, adapter, &logs, func(o *chat.RuntimeOptions) { + o.MaxConcurrent = 1 + o.DetachTimeout = detachTimeout + }) - firstStarted := make(chan struct{}) - releaseFirst := make(chan struct{}) - var mu sync.Mutex - var handled []string - bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { - mu.Lock() - first := len(handled) == 0 - handled = append(handled, ev.Event.ID) - mu.Unlock() - if first { - close(firstStarted) - <-releaseFirst + releaseFirst := make(chan struct{}) + release := sync.OnceFunc(func() { close(releaseFirst) }) + defer release() + var mu sync.Mutex + var handled []string + bot.OnNewMention(func(ctx context.Context, ev *chat.MessageEvent) error { + mu.Lock() + handled = append(handled, ev.Event.ID) + mu.Unlock() + if ev.Event.ID == "first" { + <-releaseFirst + } + return nil + }) + handledIDs := func() []string { + mu.Lock() + defer mu.Unlock() + return slices.Clone(handled) } - return nil - }) - if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { - t.Fatalf("first status = %d", status) - } - select { - case <-firstStarted: - case <-time.After(2 * time.Second): - t.Fatal("first handler did not start") - } - - // The second event is acked promptly but must wait for the single slot. - if status := postEvent(t, bot, "fake", mentionEvent("second", "fake:v1:thread-2")); status != http.StatusOK { - t.Fatalf("second status = %d", status) - } - time.Sleep(80 * time.Millisecond) - mu.Lock() - if len(handled) != 1 { - got := append([]string(nil), handled...) - mu.Unlock() - t.Fatalf("MaxConcurrent=1 did not bound execution, handled = %v", got) - } - mu.Unlock() - - close(releaseFirst) + if status := postEvent(t, bot, "fake", mentionEvent("first", "fake:v1:thread-1")); status != http.StatusOK { + t.Fatalf("first status = %d", status) + } + synctest.Wait() + if got := handledIDs(); !slices.Equal(got, []string{"first"}) { + t.Fatalf("first handler did not start alone, handled = %v", got) + } - deadline := time.After(2 * time.Second) - for { - mu.Lock() - n := len(handled) - mu.Unlock() - if n >= 2 { - break + if status := postEvent(t, bot, "fake", mentionEvent("second", "fake:v1:thread-2")); status != http.StatusOK { + t.Fatalf("second status = %d", status) } - select { - case <-deadline: - t.Fatal("second handler did not run after the slot freed") - case <-time.After(5 * time.Millisecond): + justBeforeSlotWaitDeadline := detachTimeout - time.Nanosecond + time.Sleep(justBeforeSlotWaitDeadline) + if got := handledIDs(); !slices.Equal(got, []string{"first"}) { + t.Fatalf("MaxConcurrent=1 did not bound execution, handled = %v", got) } - } + + release() + synctest.Wait() + if got := handledIDs(); !slices.Equal(got, []string{"first", "second"}) { + t.Fatalf("second handler did not run after the slot freed, handled = %v", got) + } + }) } func TestConcurrentSlotWaiterDrainedByShutdown(t *testing.T) {