From 7ebc802d0a269218f05b3b51eb10ac66eacb4d1f Mon Sep 17 00:00:00 2001 From: Jeff Halter <868228+jhalter@users.noreply.github.com> Date: Fri, 12 Jun 2026 08:11:55 -0700 Subject: Replace shared outbox with per-client send queues The outbox channel spawned one goroutine per outbound transaction, so concurrent sends to the same client could interleave bytes within the transaction framing, per-client message ordering was not guaranteed, and a slow client accumulated unbounded goroutines. Each ClientConn now has a bounded send queue drained by a single writer goroutine, which serializes writes and preserves enqueue order. Send never blocks: if a client's queue overflows, its connection is closed and the read loop performs the usual disconnect cleanup. Server.Send routes transactions to the target client's queue, replacing processOutbox and sendTransaction. Handler signatures are unchanged. Disconnect now removes the client from the manager before notifying peers so no new transactions are routed to a departing client, then idempotently closes its send queue. New tests cover write ordering, framing integrity under concurrent senders, the slow-client disconnect policy, and a Send/Disconnect race exercise (run with -race). --- internal/mobius/api_test.go | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) (limited to 'internal') diff --git a/internal/mobius/api_test.go b/internal/mobius/api_test.go index ae9916b..ca1e207 100644 --- a/internal/mobius/api_test.go +++ b/internal/mobius/api_test.go @@ -628,10 +628,9 @@ func TestShutdownHandler(t *testing.T) { srv, _, _, _ := newTestAPIServer(t, "") // Shutdown calls srv.hlServer.Shutdown in a goroutine, which calls - // SendAll on the server. We need outbox to be non-nil to avoid a panic. - // Since Shutdown runs in a goroutine, we just verify the HTTP response. - // The goroutine may panic but that's acceptable in test since we're only - // testing the HTTP layer. + // SendAll on the server to enqueue a disconnect message for each + // connected client. Since Shutdown runs in a goroutine, we just verify + // the HTTP response here. srv.hlServer.Logger = slog.Default() req := httptest.NewRequest(http.MethodPost, "/api/v1/shutdown", strings.NewReader("Server is going down")) -- cgit