From 7b66df6651c61713ade7df64e2750b60f35ab4b9 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Sun, 6 Sep 2026 05:06:47 +0000 Subject: [PATCH] test: fix the three data races that made -race runs noisy None was introduced by this branch and all three are in test code, which is what made them survive: every suite passed on every run and only the race detector said otherwise. A known-failing -race run is worse than a noisy one, because a real race raised by production code lands in the same report and is read as one of these. galleryop: gatedModelManager guarded the recorded names and not the gate channel itself. A spec frees the parked worker by closing the gate and installing a fresh one, on the spec goroutine, while the worker goroutine reads the field to park on it. The channel is now read and replaced under the same mutex, and cleanup closes idempotently. pkg/model: two specs swapped xlog's package logger to capture output and swapped it back on cleanup. xlog.SetLogger writes an unsynchronised global, so the restore raced with the backend process watcher, which logs while a process is stopping; the captured bytes.Buffer was written by that goroutine and read by an Eventually at the same time. SetLogger is now called once for the whole test binary, from init, before a goroutine exists to race with, and a spec swaps the DESTINATION under a mutex through a routing slog.Handler. Per-spec level filtering is preserved deliberately: one of these specs asserts that a debug emission is filtered OUT and would pass vacuously against a handler that recorded everything. openai: fakeTransport appended to its event and audio logs from the response and turn coordinators' goroutines while a spec ranged over them. Both are behind a mutex and are read through snapshot accessors; the fields are renamed so a raw read from another spec file does not compile. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- .../openai/realtime_classifier_test.go | 10 +- .../endpoints/openai/realtime_doubles_test.go | 39 ++++++- .../openai/realtime_semantic_vad_test.go | 6 +- .../openai/realtime_sound_detection_test.go | 4 +- .../endpoints/openai/realtime_stream_test.go | 12 +- .../realtime_voicegate_integration_test.go | 2 +- .../galleryop/cancellable_phase_test.go | 45 +++++++- pkg/model/initializers_load_logging_test.go | 14 +-- pkg/model/log_capture_test.go | 104 ++++++++++++++++++ pkg/model/process_exit_test.go | 10 +- 10 files changed, 199 insertions(+), 47 deletions(-) create mode 100644 pkg/model/log_capture_test.go diff --git a/core/http/endpoints/openai/realtime_classifier_test.go b/core/http/endpoints/openai/realtime_classifier_test.go index b8630ca81..757c91dec 100644 --- a/core/http/endpoints/openai/realtime_classifier_test.go +++ b/core/http/endpoints/openai/realtime_classifier_test.go @@ -45,7 +45,7 @@ var classifierTestHistory = schema.Messages{ func classifierResultEvents(t *fakeTransport) []types.ClassifierResultEvent { var out []types.ClassifierResultEvent - for _, e := range t.events { + for _, e := range t.recordedEvents() { if ev, ok := e.(types.ClassifierResultEvent); ok { out = append(out, ev) } @@ -57,7 +57,7 @@ func classifierResultEvents(t *fakeTransport) []types.ClassifierResultEvent { // item — what a classifier response actually "spoke". func replyTexts(t *fakeTransport) []string { var out []string - for _, e := range t.events { + for _, e := range t.recordedEvents() { if ev, ok := e.(types.ResponseOutputTextDoneEvent); ok { out = append(out, ev.Text) } @@ -277,7 +277,7 @@ var _ = Describe("classifierRespond", func() { Expect(t.countEvents(types.ServerEventTypeResponseOutputTextDone)).To(Equal(1)) Expect(t.countEvents(types.ServerEventTypeResponseFunctionCallArgumentsDone)).To(Equal(1)) var fcArgs string - for _, e := range t.events { + for _, e := range t.recordedEvents() { if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok { fcArgs = done.Arguments } @@ -656,7 +656,7 @@ var _ = Describe("classifierRespond slot filling", func() { Expect(results[0].Arguments).To(MatchJSON(`{"direction":"up","distance":3,"units":"meters"}`)) var fcArgs string - for _, e := range t.events { + for _, e := range t.recordedEvents() { if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok { fcArgs = done.Arguments } @@ -695,7 +695,7 @@ var _ = Describe("classifierRespond slot filling", func() { Expect(handled).To(BeTrue()) var fcArgs string - for _, e := range t.events { + for _, e := range t.recordedEvents() { if done, ok := e.(types.ResponseFunctionCallArgumentsDoneEvent); ok { fcArgs = done.Arguments } diff --git a/core/http/endpoints/openai/realtime_doubles_test.go b/core/http/endpoints/openai/realtime_doubles_test.go index a2c104b3c..0d7cc3fa6 100644 --- a/core/http/endpoints/openai/realtime_doubles_test.go +++ b/core/http/endpoints/openai/realtime_doubles_test.go @@ -16,9 +16,17 @@ import ( // fakeTransport records the server events and audio sent to a realtime client // so streaming behaviour can be asserted without a real WebSocket/WebRTC peer. // It is not a *WebRTCTransport, so handler code takes the WebSocket path. +// +// Every field is behind the mutex, and the recorded slices are read only +// through recordedEvents and recordedAudio. A real transport is written to by +// the response and turn coordinators' goroutines while the spec goroutine +// reads what has arrived so far, so a double that appended without a lock could +// not be driven the way production drives it. Both fields are named with a +// `Log` suffix so a raw read from another spec file does not compile. type fakeTransport struct { - events []types.ServerEvent - audio []fakeAudioChunk + mu sync.Mutex + eventLog []types.ServerEvent + audioLog []fakeAudioChunk } type fakeAudioChunk struct { @@ -27,23 +35,42 @@ type fakeAudioChunk struct { } func (f *fakeTransport) SendEvent(e types.ServerEvent) error { - f.events = append(f.events, e) + f.mu.Lock() + defer f.mu.Unlock() + f.eventLog = append(f.eventLog, e) return nil } func (f *fakeTransport) ReadEvent() ([]byte, error) { return nil, nil } func (f *fakeTransport) SendAudio(_ context.Context, pcm []byte, sampleRate int) error { - f.audio = append(f.audio, fakeAudioChunk{pcm: pcm, sampleRate: sampleRate}) + f.mu.Lock() + defer f.mu.Unlock() + f.audioLog = append(f.audioLog, fakeAudioChunk{pcm: pcm, sampleRate: sampleRate}) return nil } func (f *fakeTransport) Close() error { return nil } +// recordedEvents returns a snapshot of the events sent so far. A COPY, because +// the caller ranges over it while the coordinators may still be sending. +func (f *fakeTransport) recordedEvents() []types.ServerEvent { + f.mu.Lock() + defer f.mu.Unlock() + return append([]types.ServerEvent(nil), f.eventLog...) +} + +// recordedAudio returns a snapshot of the audio chunks sent so far. +func (f *fakeTransport) recordedAudio() []fakeAudioChunk { + f.mu.Lock() + defer f.mu.Unlock() + return append([]fakeAudioChunk(nil), f.audioLog...) +} + // countEvents returns how many recorded events have the given type. func (f *fakeTransport) countEvents(et types.ServerEventType) int { n := 0 - for _, e := range f.events { + for _, e := range f.recordedEvents() { if e.ServerEventType() == et { n++ } @@ -55,7 +82,7 @@ func (f *fakeTransport) countEvents(et types.ServerEventType) int { // delta event — i.e. the text streamed to the client as it is generated. func (f *fakeTransport) transcriptDeltaText() string { var b strings.Builder - for _, e := range f.events { + for _, e := range f.recordedEvents() { if d, ok := e.(types.ResponseOutputAudioTranscriptDeltaEvent); ok { b.WriteString(d.Delta) } diff --git a/core/http/endpoints/openai/realtime_semantic_vad_test.go b/core/http/endpoints/openai/realtime_semantic_vad_test.go index c3f5d7ef8..52daf9cf2 100644 --- a/core/http/endpoints/openai/realtime_semantic_vad_test.go +++ b/core/http/endpoints/openai/realtime_semantic_vad_test.go @@ -269,7 +269,7 @@ var _ = Describe("liveTurnState", func() { lts.drainEvents(1.0) var got []types.ConversationItemInputAudioTranscriptionDeltaEvent - for _, e := range ftr.events { + for _, e := range ftr.recordedEvents() { if d, ok := e.(types.ConversationItemInputAudioTranscriptionDeltaEvent); ok { got = append(got, d) } @@ -335,7 +335,7 @@ var _ = Describe("commitUtteranceWithTranscript", func() { Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) var completed types.ConversationItemInputAudioTranscriptionCompletedEvent - for _, e := range tr.events { + for _, e := range tr.recordedEvents() { if c, ok := e.(types.ConversationItemInputAudioTranscriptionCompletedEvent); ok { completed = c } @@ -394,7 +394,7 @@ var _ = Describe("emitPrecomputedTranscription", func() { Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionDelta)).To(Equal(2), "empty deltas skipped") Expect(tr.countEvents(types.ServerEventTypeConversationItemInputAudioTranscriptionCompleted)).To(Equal(1)) - for _, e := range tr.events { + for _, e := range tr.recordedEvents() { switch ev := e.(type) { case types.ConversationItemInputAudioTranscriptionDeltaEvent: Expect(ev.ItemID).To(Equal("item42")) diff --git a/core/http/endpoints/openai/realtime_sound_detection_test.go b/core/http/endpoints/openai/realtime_sound_detection_test.go index e440e80c3..02ced312a 100644 --- a/core/http/endpoints/openai/realtime_sound_detection_test.go +++ b/core/http/endpoints/openai/realtime_sound_detection_test.go @@ -38,7 +38,7 @@ var _ = Describe("emitSoundDetection", func() { Expect(err).ToNot(HaveOccurred()) Expect(t.countEvents(types.ServerEventTypeConversationItemSoundDetection)).To(Equal(1)) - ev, ok := t.events[0].(types.ConversationItemSoundDetectionEvent) + ev, ok := t.recordedEvents()[0].(types.ConversationItemSoundDetectionEvent) Expect(ok).To(BeTrue()) Expect(ev.ItemID).To(Equal("item1")) Expect(ev.ContentIndex).To(Equal(0)) @@ -62,7 +62,7 @@ var _ = Describe("emitSoundDetection", func() { Expect(err).ToNot(HaveOccurred()) Expect(t.countEvents(types.ServerEventTypeConversationItemSoundDetection)).To(Equal(1)) - ev, ok := t.events[0].(types.ConversationItemSoundDetectionEvent) + ev, ok := t.recordedEvents()[0].(types.ConversationItemSoundDetectionEvent) Expect(ok).To(BeTrue()) Expect(ev.Detections).To(BeEmpty()) }) diff --git a/core/http/endpoints/openai/realtime_stream_test.go b/core/http/endpoints/openai/realtime_stream_test.go index 2d5d7d7a1..d6962c087 100644 --- a/core/http/endpoints/openai/realtime_stream_test.go +++ b/core/http/endpoints/openai/realtime_stream_test.go @@ -250,8 +250,8 @@ var _ = Describe("triggerResponse", func() { // The single terminal carries the produced output item and the usage — // both empty in the legacy code. var done *types.ResponseDoneEvent - for i := range t.events { - if d, ok := t.events[i].(types.ResponseDoneEvent); ok { + for _, e := range t.recordedEvents() { + if d, ok := e.(types.ResponseDoneEvent); ok { done = &d } } @@ -287,8 +287,8 @@ var _ = Describe("triggerResponse", func() { var created *types.ResponseCreatedEvent var done *types.ResponseDoneEvent - for i := range t.events { - switch e := t.events[i].(type) { + for _, sent := range t.recordedEvents() { + switch e := sent.(type) { case types.ResponseCreatedEvent: created = &e case types.ResponseDoneEvent: @@ -317,8 +317,8 @@ var _ = Describe("triggerResponse", func() { triggerResponse(context.Background(), session, &Conversation{}, t, nil) - for i := range t.events { - if d, ok := t.events[i].(types.ResponseDoneEvent); ok { + for _, e := range t.recordedEvents() { + if d, ok := e.(types.ResponseDoneEvent); ok { Expect(d.Response.Metadata).To(BeEmpty()) } } diff --git a/core/http/endpoints/openai/realtime_voicegate_integration_test.go b/core/http/endpoints/openai/realtime_voicegate_integration_test.go index b0f7f0b49..caacab877 100644 --- a/core/http/endpoints/openai/realtime_voicegate_integration_test.go +++ b/core/http/endpoints/openai/realtime_voicegate_integration_test.go @@ -67,7 +67,7 @@ func itSession(gate *voiceGate) (*Session, *fakeModel) { // hasSpeakerNotAuthorized reports whether a speaker_not_authorized error event // was emitted to the client. func hasSpeakerNotAuthorized(tr *fakeTransport) bool { - for _, e := range tr.events { + for _, e := range tr.recordedEvents() { if ev, ok := e.(types.ErrorEvent); ok && ev.Error.Code == "speaker_not_authorized" { return true } diff --git a/core/services/galleryop/cancellable_phase_test.go b/core/services/galleryop/cancellable_phase_test.go index 47c98c660..5ef05ecc9 100644 --- a/core/services/galleryop/cancellable_phase_test.go +++ b/core/services/galleryop/cancellable_phase_test.go @@ -17,6 +17,13 @@ import ( // Parking is what makes the running phase observable at all: without it the // handler-entry status is overwritten by the terminal write before a spec can // read it. +// +// The gate CHANNEL is under the mutex, not only the recorded names. A spec that +// frees the worker replaces the channel so the next operation parks on a fresh +// one, and the worker goroutine reads the field to park on it; the two happen +// on different goroutines and raced. It was a race in this double and not in +// anything it stands for, which is exactly why it survived: the suite passed +// every run and only -race said so. type gatedModelManager struct { mu sync.Mutex started []string @@ -39,15 +46,44 @@ func (m *gatedModelManager) Started() []string { return append([]string(nil), m.started...) } +// gateCh reads the current gate, for a handler about to park on it. +func (m *gatedModelManager) gateCh() chan struct{} { + m.mu.Lock() + defer m.mu.Unlock() + return m.gate +} + +// releaseGate frees every handler parked on the current gate and arms a fresh +// one, so the next operation parks again. +func (m *gatedModelManager) releaseGate() { + m.mu.Lock() + defer m.mu.Unlock() + close(m.gate) + m.gate = make(chan struct{}) +} + +// closeGate frees every parked handler without arming another. It is what +// cleanup does, and it is idempotent so a spec that already released can be +// cleaned up after. +func (m *gatedModelManager) closeGate() { + m.mu.Lock() + defer m.mu.Unlock() + select { + case <-m.gate: + default: + close(m.gate) + } +} + func (m *gatedModelManager) InstallModel(_ context.Context, op *galleryop.ManagementOp[gallery.GalleryModel, gallery.ModelConfig], _ galleryop.ProgressCallback) error { m.record(op.GalleryElementName) - <-m.gate + <-m.gateCh() return nil } func (m *gatedModelManager) DeleteModel(name string) error { m.record(name) - <-m.gate + <-m.gateCh() return nil } @@ -120,7 +156,7 @@ var _ = Describe("operation cancellability by phase", func() { BeforeEach(func() { manager = newGatedModelManager() - DeferCleanup(func() { close(manager.gate) }) + DeferCleanup(manager.closeGate) svc.SetModelManager(manager) ctx, cancel := context.WithCancel(context.Background()) @@ -188,8 +224,7 @@ var _ = Describe("operation cancellability by phase", func() { // Free the worker: it loops back to an empty channel because the // delivery goroutine gave up on the send. - close(manager.gate) - manager.gate = make(chan struct{}) + manager.releaseGate() Consistently(manager.Started, "500ms", "20ms").ShouldNot( ContainElement("localai@doomed-removal"), diff --git a/pkg/model/initializers_load_logging_test.go b/pkg/model/initializers_load_logging_test.go index e785b03db..0a863f192 100644 --- a/pkg/model/initializers_load_logging_test.go +++ b/pkg/model/initializers_load_logging_test.go @@ -1,12 +1,10 @@ package model import ( - "bytes" "log/slog" pb "github.com/mudler/LocalAI/pkg/grpc/proto" "github.com/mudler/LocalAI/pkg/system" - "github.com/mudler/xlog" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -23,7 +21,7 @@ import ( var _ = Describe("backendLoader load logging", func() { var ( ml *ModelLoader - captured *bytes.Buffer + captured *syncBuffer ) BeforeEach(func() { @@ -33,16 +31,10 @@ var _ = Describe("backendLoader load logging", func() { // Capture at info level so a debug-level emission is filtered out: the // assertions then fail on severity, not merely on wording. - captured = &bytes.Buffer{} - handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelInfo}) - xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelInfo)) + captured = captureLogs(slog.LevelInfo) }) - AfterEach(func() { - // xlog exposes no getter for the package logger, so restore the same - // default the suite entrypoint installs rather than the prior value. - xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text")) - }) + AfterEach(stopCapturingLogs) Context("when the model is already resident", func() { BeforeEach(func() { diff --git a/pkg/model/log_capture_test.go b/pkg/model/log_capture_test.go new file mode 100644 index 000000000..3057eac3f --- /dev/null +++ b/pkg/model/log_capture_test.go @@ -0,0 +1,104 @@ +package model + +import ( + "bytes" + "context" + "log/slog" + "sync" + + "github.com/mudler/xlog" +) + +// xlog keeps its logger in a package-level variable that SetLogger writes +// without a lock, so every xlog.Info in the process reads that variable +// concurrently with the write. A spec that swapped the logger to capture output +// and swapped it back on cleanup therefore raced with any goroutine this +// package had left logging: the backend process watcher, which logs while a +// process is stopping, is one of those on every run. +// +// So SetLogger is called exactly ONCE for this whole test binary, from init, +// before a goroutine exists to race with. What a spec swaps afterwards is the +// destination, under a mutex, through the routing handler below. That keeps the +// per-spec level filtering intact, which matters: at least one spec asserts +// that a debug emission is filtered OUT, and would pass vacuously against a +// handler that simply recorded everything. +type routingLogHandler struct { + mu sync.Mutex + to slog.Handler +} + +// sharedLogHandler is the one handler xlog is given for this binary. +var sharedLogHandler = &routingLogHandler{} + +func init() { + xlog.SetLogger(xlog.NewLoggerWithHandler(sharedLogHandler, xlog.LogLevelInfo)) +} + +func (h *routingLogHandler) current() slog.Handler { + h.mu.Lock() + defer h.mu.Unlock() + return h.to +} + +// arm points the shared handler at inner, or discards everything when inner is +// nil. Returns nothing: a spec restores by arming nil, because xlog exposes no +// getter and there is no previous value to hand back. +func (h *routingLogHandler) arm(inner slog.Handler) { + h.mu.Lock() + defer h.mu.Unlock() + h.to = inner +} + +func (h *routingLogHandler) Enabled(ctx context.Context, level slog.Level) bool { + inner := h.current() + return inner != nil && inner.Enabled(ctx, level) +} + +func (h *routingLogHandler) Handle(ctx context.Context, r slog.Record) error { + inner := h.current() + if inner == nil { + return nil + } + return inner.Handle(ctx, r) +} + +// WithAttrs and WithGroup hand back the router itself. xlog never calls either +// (it has no With), and a copy would be a second handler holding the same +// mutex by value. +func (h *routingLogHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *routingLogHandler) WithGroup(string) slog.Handler { return h } + +var _ slog.Handler = (*routingLogHandler)(nil) + +// syncBuffer is a bytes.Buffer that a log handler and a poller may share. +// +// The diagnostic under test is written by a goroutine and read by an Eventually +// on the spec goroutine, which is a plain concurrent use of a bytes.Buffer. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// captureLogs sends everything logged through xlog at or above level into a +// fresh buffer, until stopCapturingLogs is called. +func captureLogs(level slog.Level) *syncBuffer { + captured := &syncBuffer{} + sharedLogHandler.arm(slog.NewTextHandler(captured, &slog.HandlerOptions{Level: level})) + return captured +} + +// stopCapturingLogs sends everything logged afterwards nowhere, which is what +// a test binary wants of a package whose goroutines outlive their spec. +func stopCapturingLogs() { sharedLogHandler.arm(nil) } diff --git a/pkg/model/process_exit_test.go b/pkg/model/process_exit_test.go index cc5554bcf..d0bff14e8 100644 --- a/pkg/model/process_exit_test.go +++ b/pkg/model/process_exit_test.go @@ -1,13 +1,11 @@ package model import ( - "bytes" "log/slog" "os" "path/filepath" "github.com/mudler/LocalAI/pkg/system" - "github.com/mudler/xlog" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" ) @@ -18,12 +16,8 @@ var _ = Describe("backend process exit diagnostics", func() { backendPath := filepath.Join(tmpDir, "failing-backend") Expect(os.WriteFile(backendPath, []byte("#!/bin/sh\necho 'first diagnostic' >&2\necho 'fatal metal pipeline error' >&2\nexit 42\n"), 0o700)).To(Succeed()) - captured := &bytes.Buffer{} - handler := slog.NewTextHandler(captured, &slog.HandlerOptions{Level: slog.LevelWarn}) - xlog.SetLogger(xlog.NewLoggerWithHandler(handler, xlog.LogLevelWarn)) - DeferCleanup(func() { - xlog.SetLogger(xlog.NewLogger(xlog.LogLevel("info"), "text")) - }) + captured := captureLogs(slog.LevelWarn) + DeferCleanup(stopCapturingLogs) loader := NewModelLoader(&system.SystemState{Model: system.Model{ModelsPath: tmpDir}}) process, err := loader.startProcess(backendPath, "test-model", "127.0.0.1:65535")