From 53639c4df399506743b565d2fd0e17253fe8dca5 Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Mon, 31 Aug 2026 10:42:49 +0000 Subject: [PATCH] test(distributed): scope the log-subscriber wait and mark the race it works around MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three corrections from review of the previous commit. The lock-order comment on SubscriberCount claimed no path takes s.mu and a buffer lock together. Subscribe does exactly that, holding s.mu.RLock across replica registrations that take buf.mu. State the rule that is actually true — s.mu precedes any buffer lock, so counting after releasing it preserves the order — and say what follows from it: the total is a sample, not a snapshot. waitForLogSubscriber read as general-purpose but unblocks on the first registered subscription. Subscribe attaches the exact-key buffer and each replica buffer one at a time, so for a replicated model the count goes positive while later replicas are still unattached and the race survives. Rename it waitForSingleLogSubscriber, document that it holds only where Subscribe resolves to one buffer, and assert on exactly 1: misuse then fails loudly on the count rather than going quietly back to being flaky. Taking the expected count as a parameter was the alternative, but that makes callers predict a store-internal number and an under-count fails the same silent way as the original bug. The snapshot-then-subscribe race had no artifact outside a report, and review found a second site carrying it. Mark both handlers identically, including the point that swapping the two calls duplicates rather than drops and so is not the fix. The race itself is left alone; this branch stays test infrastructure. Assisted-by: Claude Opus 5 [claude-code] Signed-off-by: Ettore Di Giacinto --- core/http/endpoints/localai/backend_logs.go | 7 ++++++ core/services/nodes/file_transfer_server.go | 7 ++++++ pkg/model/backend_log_store.go | 7 ++++-- tests/e2e/distributed/backend_logs_test.go | 24 +++++++++++++++------ 4 files changed, 36 insertions(+), 9 deletions(-) diff --git a/core/http/endpoints/localai/backend_logs.go b/core/http/endpoints/localai/backend_logs.go index 6072b8483..8b5f99d66 100644 --- a/core/http/endpoints/localai/backend_logs.go +++ b/core/http/endpoints/localai/backend_logs.go @@ -121,6 +121,13 @@ func BackendLogsWebSocketEndpoint(ml *model.ModelLoader) echo.HandlerFunc { conn := &backendLogsConn{Conn: ws} + // KNOWN RACE: the snapshot is sent before the subscription is registered, so + // a line appended in that window is never streamed. A viewer attaching while + // a model loads (when a backend is at its noisiest) can silently miss lines; + // they stay in the buffer, so a reload shows them. Fixing it needs an atomic + // snapshot-plus-subscribe under the store lock, not a reorder of these two + // calls, which would duplicate instead of drop. + // Send existing lines as initial batch existingLines := ml.BackendLogs().GetLines(modelID) initialMsg := map[string]any{ diff --git a/core/services/nodes/file_transfer_server.go b/core/services/nodes/file_transfer_server.go index 0fc5ac634..2d4bc03ba 100644 --- a/core/services/nodes/file_transfer_server.go +++ b/core/services/nodes/file_transfer_server.go @@ -839,6 +839,13 @@ func handleBackendLogsWS(w http.ResponseWriter, r *http.Request, logStore *model conn := &backendLogsWSConn{Conn: ws} + // KNOWN RACE: the snapshot is sent before the subscription is registered, so + // a line appended in that window is never streamed. A viewer attaching while + // a model loads (when a backend is at its noisiest) can silently miss lines; + // they stay in the buffer, so a reload shows them. Fixing it needs an atomic + // snapshot-plus-subscribe under the store lock, not a reorder of these two + // calls, which would duplicate instead of drop. + // Send existing lines as initial batch existingLines := logStore.GetLines(modelID) initialMsg := map[string]any{ diff --git a/pkg/model/backend_log_store.go b/pkg/model/backend_log_store.go index 30268fde4..3c60f34a3 100644 --- a/pkg/model/backend_log_store.go +++ b/pkg/model/backend_log_store.go @@ -366,8 +366,11 @@ func (s *BackendLogStore) SubscriberCount(modelID string) int { } s.mu.RUnlock() - // Counted after releasing s.mu: no other path takes s.mu and a buffer lock - // together, and keeping it that way costs nothing here. + // Lock order in this type is always s.mu before any buffer lock — Subscribe + // holds s.mu.RLock across its replica registrations, which take buf.mu — so + // counting after releasing s.mu keeps that order rather than inverting it. + // The total is therefore a sample, not a snapshot: a concurrent Subscribe + // can register a further buffer while this loop runs. count := func(buf *backendLogBuffer) int { buf.mu.Lock() defer buf.mu.Unlock() diff --git a/tests/e2e/distributed/backend_logs_test.go b/tests/e2e/distributed/backend_logs_test.go index 473dad54f..82e8ac156 100644 --- a/tests/e2e/distributed/backend_logs_test.go +++ b/tests/e2e/distributed/backend_logs_test.go @@ -25,7 +25,7 @@ import ( "gorm.io/gorm/logger" ) -// waitForLogSubscriber blocks until the worker's WebSocket log handler has +// waitForSingleLogSubscriber blocks until the worker's WebSocket log handler has // registered its subscription on the store. // // The handler writes the "initial" batch first and subscribes only afterwards, @@ -33,11 +33,21 @@ import ( // streamed, and the spec then waits out its full read deadline. Measured at // roughly one run in seventeen with `--repeat`, which is far too often for CI. // Waiting on the subscription removes the race from the spec; the handler's own -// snapshot/subscribe window is a separate production question. -func waitForLogSubscriber(logStore *model.BackendLogStore, modelID string) { +// snapshot/subscribe window is a separate production question, marked at both +// production sites. +// +// Only valid where BackendLogStore.Subscribe resolves modelID to exactly ONE +// buffer: a bare model ID with no "#N" replica buffers in the store, or +// a full process key. Subscribe registers the exact-key buffer and each replica +// buffer one at a time, so for a model that does have replicas the count goes +// positive while later replicas are still unattached and the race survives. +// Hence the assertion is on exactly 1 rather than "at least 1": a spec that +// misapplies this to a replicated model fails loudly on the count instead of +// going quietly back to being flaky. +func waitForSingleLogSubscriber(logStore *model.BackendLogStore, modelID string) { GinkgoHelper() Eventually(func() int { return logStore.SubscriberCount(modelID) }, "10s", "5ms"). - Should(BeNumerically(">", 0), "the WebSocket handler never subscribed to %q", modelID) + Should(Equal(1), "the WebSocket handler never subscribed to %q exactly once", modelID) } var _ = Describe("Distributed Backend Log Streaming", Label("Distributed"), func() { @@ -227,7 +237,7 @@ var _ = Describe("Distributed Backend Log Streaming", Label("Distributed"), func Expect(initialLines[1].Text).To(Equal("line-2")) // Now append a new line and verify it arrives via WebSocket - waitForLogSubscriber(logStore, "ws-model") + waitForSingleLogSubscriber(logStore, "ws-model") logStore.AppendLine("ws-model", "stdout", "line-3-realtime") conn.SetReadDeadline(time.Now().Add(5 * time.Second)) @@ -296,7 +306,7 @@ var _ = Describe("Distributed Backend Log Streaming", Label("Distributed"), func Expect(conn.ReadJSON(&initialMsg)).To(Succeed()) // Append line to a different model - waitForLogSubscriber(logStore, "ws-model") + waitForSingleLogSubscriber(logStore, "ws-model") logStore.AppendLine("other-model", "stdout", "should not appear") // Append line to our model logStore.AppendLine("ws-model", "stdout", "should appear") @@ -492,7 +502,7 @@ var _ = Describe("Distributed Backend Log Streaming", Label("Distributed"), func Expect(initialLines[0].Text).To(Equal("initial line from worker")) // Append a new line on the worker's log store - waitForLogSubscriber(logStore, "proxy-model") + waitForSingleLogSubscriber(logStore, "proxy-model") logStore.AppendLine("proxy-model", "stderr", "realtime via proxy") // Read the streamed line through the proxy