test(pluginhost,handlers): fix flaky races in plugin load and stream cancel tests

- Add a loadCleanupDoneHook callback in pluginhost.Host so canceled plugin load tests wait until the in-flight load token is cleared after client shutdown.
- Drain dataChan after canceling the context in TestHandlerPluginExecutorPublishesStreamCancellation before closing the chunk channel to avoid racing ctx.Done() against normal stream EOF.

Closes: #6885
This commit is contained in:
Luis Pater
2026-10-23 01:14:55 +08:00
parent b90190dbaa
commit a014ba0e49
3 changed files with 20 additions and 9 deletions

View File

@@ -92,6 +92,8 @@ type Host struct {
modelStreams *modelStreamBridge
callbackContexts *callbackContextRegistry
snapshot atomic.Value
// loadCleanupDoneHook, set by tests, runs after a late load cleanup dropped its load token.
loadCleanupDoneHook func()
}
func New() *Host {
@@ -578,6 +580,9 @@ func (h *Host) finishPluginLoadCleanup(id string, request *pluginLoadRequest, lo
go func() {
h.discardLoadedPlugin(loaded)
h.clearLoadingRequest(id, request)
if h.loadCleanupDoneHook != nil {
h.loadCleanupDoneHook()
}
}()
}

View File

@@ -1695,6 +1695,8 @@ func TestHostCanceledInitializationDiscardsBlockedClient(t *testing.T) {
registration: validTestPlugin("alpha"),
}
h := NewForTest(&blockingHostCallLoader{client: client})
cleanupDone := make(chan struct{})
h.loadCleanupDoneHook = func() { close(cleanupDone) }
cfg := &config.Config{Plugins: config.PluginsConfig{
Enabled: true,
Dir: makePluginDir(t, "alpha"),
@@ -1714,10 +1716,9 @@ func TestHostCanceledInitializationDiscardsBlockedClient(t *testing.T) {
}
close(client.release)
deadline := time.Now().Add(time.Second)
for client.shutdown.Load() == 0 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
// Late cleanup shuts the client down first and drops the load token after;
// wait for the end of the cleanup, not for the shutdown alone.
waitForHostTestSignal(t, cleanupDone, "late load cleanup")
if got := client.shutdown.Load(); got != 1 {
t.Fatalf("blocked initialization client shutdown calls = %d, want 1", got)
}
@@ -1734,6 +1735,8 @@ func TestHostCancellationUnderMutationLockDoesNotInsertLoadedPlugin(t *testing.T
registration: validTestPlugin("alpha"),
}
h := NewForTest(&blockingHostCallLoader{client: client})
cleanupDone := make(chan struct{})
h.loadCleanupDoneHook = func() { close(cleanupDone) }
cfg := &config.Config{Plugins: config.PluginsConfig{
Enabled: true,
Dir: makePluginDir(t, "alpha"),
@@ -1754,10 +1757,9 @@ func TestHostCancellationUnderMutationLockDoesNotInsertLoadedPlugin(t *testing.T
h.mu.Unlock()
waitForHostTestSignal(t, applyDone, "canceled plugin apply")
deadline := time.Now().Add(time.Second)
for client.shutdown.Load() == 0 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
// Late cleanup shuts the client down first and drops the load token after;
// wait for the end of the cleanup, not for the shutdown alone.
waitForHostTestSignal(t, cleanupDone, "late load cleanup")
if got := client.shutdown.Load(); got != 1 {
t.Fatalf("late client shutdown calls = %d, want 1", got)
}

View File

@@ -550,9 +550,13 @@ func TestHandlerPluginExecutorPublishesStreamCancellation(t *testing.T) {
handler.SetModelRouterHost(mockHost)
dataChan, _, _ := handler.ExecuteStreamWithAuthManager(ctx, "openai", originalModel, []byte(fmt.Sprintf(`{"model":%q,"stream":true}`, originalModel)), "")
// Receive first chunk then cancel context
// Receive the first chunk, then cancel. The chunk channel stays open until the stream
// goroutine has exited: closed early, it races ctx.Done() in nextStreamChunk's select and the
// stream can end as a normal completion (about 1 in 1000 runs under -race).
<-dataChan
cancel()
for range dataChan {
}
close(chunks)
record := plugin.waitRecord(t)