This is an automated email from the ASF dual-hosted git repository. sruehl pushed a commit to branch develop in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 69f65244fa33bbfa4cbf90873166aed6c9a33713 Author: Sebastian Rühl <[email protected]> AuthorDate: Mon Jul 20 13:58:10 2026 +0200 fix(plc4go): defaultCodec.Connect started workers before running=true - worker could die at startup A worker goroutine that got scheduled between startWorkers() and running.Store(true) observed running==false twice - in its mainLoop condition and in its restart defer - and terminated permanently while Connect still returned success. A codec without its expire worker never times out expectations, so SendRequest callers waiting on handleError hang forever. This is what intermittently wedged TestReader_sendMessageOverTheWire (internal/cbus) under -race: the timeout dump shows a live ReceiveWork but no ExpireWork goroutine. Store running=true before starting the workers; the goroutine-creation happens-before edge then guarantees both workers observe true. Falsification test probes worker liveness after Connect via a blocking send on the unbuffered notify channels (received iff a worker sits in its select). Verified RED first: expire worker died within ~5 of 1000 iterations pre-fix; 5000 iterations green post-fix, and the cbus test survives 30 -race iterations (previously hung within 10). --- plc4go/spi/default/DefaultCodec.go | 6 ++++- plc4go/spi/default/DefaultCodec_test.go | 43 +++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/plc4go/spi/default/DefaultCodec.go b/plc4go/spi/default/DefaultCodec.go index be59bcb655..e1d7ad4cb6 100644 --- a/plc4go/spi/default/DefaultCodec.go +++ b/plc4go/spi/default/DefaultCodec.go @@ -171,8 +171,12 @@ func (m *defaultCodec) Connect(ctx context.Context) error { } m.log.Debug().Msg("Message codec currently not running, starting worker now") - m.startWorkers() + // running must be true BEFORE the workers start: a worker that observes + // running==false in its loop condition AND in its restart defer terminates + // permanently, leaving a "connected" codec whose expectations never expire. + // The goroutine-creation happens-before edge guarantees the workers see true. m.running.Store(true) + m.startWorkers() m.log.Trace().Msg("connected") return nil } diff --git a/plc4go/spi/default/DefaultCodec_test.go b/plc4go/spi/default/DefaultCodec_test.go index d5906085c4..e85ff31b0a 100644 --- a/plc4go/spi/default/DefaultCodec_test.go +++ b/plc4go/spi/default/DefaultCodec_test.go @@ -438,6 +438,49 @@ func Test_defaultCodec_Connect(t *testing.T) { } } +// Test_defaultCodec_Connect_workersAliveAfterConnect is a falsification test +// for the worker-startup race: Connect used to start the workers BEFORE +// running.Store(true), so a worker observing running==false in both its loop +// condition and its restart defer terminated permanently — leaving a +// successfully "connected" codec whose expectations never expire and whose +// messages are never received (observed as an infinite hang in the cbus +// Reader tests). An idle worker blocks in its select receiving from its +// (unbuffered) notify channel, so a blocking send succeeds iff the worker is +// alive. +func Test_defaultCodec_Connect_workersAliveAfterConnect(t *testing.T) { + for i := 0; i < 1000; i++ { + requirements := NewMockDefaultCodecRequirements(t) + requirements.EXPECT().Receive(mock.Anything).Return(nil, nil).Maybe() + instance := NewMockTransportInstance(t) + instance.EXPECT().IsConnected().Return(true) + instance.EXPECT().Close().Return(nil) + // A receive cycle in flight during Disconnect classifies the synthetic + // "not running" error before the worker notices the shutdown. + instance.EXPECT().ClassifyError(mock.Anything).Return(transports.TransportErrorTransient).Maybe() + codec := buildDefaultCodec(requirements, instance, options.WithCustomLogger(testutils.ProduceTestingLogger(t))).(*defaultCodec) + require.NoError(t, codec.Connect(testutils.TestContext(t))) + t.Cleanup(func() { + if codec.IsRunning() { + _ = codec.Disconnect() + } + }) + for _, probe := range []struct { + worker string + notify chan struct{} + }{ + {"expire", codec.notifyExpireWorker}, + {"receive", codec.notifyReceiveWorker}, + } { + select { + case probe.notify <- struct{}{}: + case <-time.After(5 * time.Second): + t.Fatalf("iteration %d: %s worker died during Connect", i, probe.worker) + } + } + require.NoError(t, codec.Disconnect()) + } +} + func Test_defaultCodec_Disconnect(t *testing.T) { type fields struct { DefaultCodecRequirements DefaultCodecRequirements
