From c68fcfa3c1cd0f76eeff95ac390c9f8b00657675 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Tue, 28 Jul 2026 19:24:26 +0200 Subject: [PATCH] test: stop the e2e stack broker deadlocking on close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit go/cmd/sim-ferroamp/e2e_test.go carries the same race, and the fix here is the same one. mochi-mqtt v2.7.9 Clients.GetByListener takes the read lock and then calls Clients.Len, which takes it again. sync.RWMutex is not reentrant, so a Clients.Delete landing between the two read locks blocks the second one while the writer waits for the first. Both goroutines stop for good. Server.Close walks GetByListener, and the broker runs Clients.Delete the moment a client drops. stack.Close calls reg.ShutdownAll, which disconnects the Ferroamp driver's MQTT client, and then closes the broker — exactly that race. Nothing bounds it but `make e2e`'s 180s timeout, so it would burn three minutes and fail an unrelated pull request. Close the broker through closeBroker, which waits for the client count to fall back to the inline client before closing. Both writers on Clients are then quiet: the disconnect path has already run its Delete, and the clearExpiredClients sweeper skips the inline client because its StopTime is zero. A 5-second deadline on Close is the backstop — should the broker still wedge, the test says so in seconds rather than burning the whole timeout. v2.7.9 is the newest release, so there is nothing to upgrade to. Co-Authored-By: Claude Opus 5 --- go/test/e2e/stack_test.go | 56 +++++++++++++++++++++++++++++++++++---- 1 file changed, 51 insertions(+), 5 deletions(-) diff --git a/go/test/e2e/stack_test.go b/go/test/e2e/stack_test.go index 3c3fa3d3..5c91e320 100644 --- a/go/test/e2e/stack_test.go +++ b/go/test/e2e/stack_test.go @@ -85,10 +85,14 @@ type stack struct { // Simulators mqttBroker *mqttserver.Server mqttPort int - ferroSim *ferroamp.Simulator - modbusSrv *sv.ModbusServer - modbusPort int - sungSim *sungrow.Simulator + // mqttBaseline is the broker's client count with no driver attached — the + // inline client, which stays for the life of the server. Close waits for + // the count to fall back to it; see closeBroker. + mqttBaseline int + ferroSim *ferroamp.Simulator + modbusSrv *sv.ModbusServer + modbusPort int + sungSim *sungrow.Simulator // Core pieces reg *drivers.Registry @@ -144,6 +148,9 @@ func setupStack(t *testing.T) *stack { } go func() { _ = mb.Serve() }() s.mqttBroker = mb + // New registers the inline client, so the count is already settled here and + // does not race Serve. + s.mqttBaseline = mb.Clients.Len() ferroCfg := ferroamp.Default() ferroCfg.ResponseTauS = 0.3 @@ -407,13 +414,52 @@ func (s *stack) Close() { s.modbusSrv.Stop() } if s.mqttBroker != nil { - _ = s.mqttBroker.Close() + closeBroker(s.t, s.mqttBroker, s.mqttBaseline) } if s.st != nil { s.st.Close() } } +// closeBroker stops the broker without tripping the recursive read-lock in +// mochi-mqtt v2.7.9: Clients.GetByListener takes the read lock and then calls +// Clients.Len, which takes it again. sync.RWMutex is not reentrant, so a +// Clients.Delete — what the broker runs once a client drops — landing between +// the two read locks wedges both goroutines for good. Server.Close walks +// GetByListener, so closing while a client is still tearing down is the race. +// ShutdownAll above disconnects the Ferroamp driver, which is exactly that. +// +// Waiting for the client count to fall back to baseline closes the window: no +// disconnect is in flight, so no writer can queue between the two read locks. +// The deadline is the backstop — if the broker still wedges, the test says so +// in seconds instead of burning the whole `make e2e` timeout on an unrelated +// pull request. +// +// go/cmd/sim-ferroamp/e2e_test.go carries the same helper for the same reason. +// v2.7.9 is the newest release, so there is nothing to upgrade to. +func closeBroker(t *testing.T, mb *mqttserver.Server, baseline int) { + t.Helper() + + settle := time.Now().Add(2 * time.Second) + for mb.Clients.Len() > baseline && time.Now().Before(settle) { + time.Sleep(5 * time.Millisecond) + } + if n := mb.Clients.Len(); n > baseline { + t.Logf("broker still holds %d clients (baseline %d) at close", n, baseline) + } + + done := make(chan struct{}) + go func() { + defer close(done) + _ = mb.Close() + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Errorf("broker Close blocked for 5s: mochi-mqtt Clients read lock is held by GetByListener while a client disconnect waits for the write lock") + } +} + func (s *stack) baseURL() string { return fmt.Sprintf("http://127.0.0.1:%d", s.apiPort) } func (s *stack) getJSON(path string, v any) {