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")