From 2e635af5417901da1308a3158641c1fa1332955c Mon Sep 17 00:00:00 2001 From: Sahil Sojitra Date: Tue, 16 Jun 2026 11:30:26 +0530 Subject: [PATCH 1/8] fix: resolve concurrency bugs in eventbus and webtransport --- p2p/host/eventbus/basic.go | 27 ++------- p2p/host/eventbus/basic_test.go | 65 ++++++++++++++++++++++ p2p/transport/webtransport/cert_manager.go | 12 +++- 3 files changed, 81 insertions(+), 23 deletions(-) diff --git a/p2p/host/eventbus/basic.go b/p2p/host/eventbus/basic.go index a9db7b1e9a..b361a25402 100644 --- a/p2p/host/eventbus/basic.go +++ b/p2p/host/eventbus/basic.go @@ -334,8 +334,6 @@ type wildcardNode struct { nSinks atomic.Int32 sinks []*namedSink metricsTracer MetricsTracer - - slowConsumerTimer *time.Timer } func (n *wildcardNode) addSink(sink *namedSink) { @@ -385,12 +383,7 @@ func (n *wildcardNode) emit(evt any) { select { case sink.ch <- evt: default: - slowConsumerTimer := emitAndLogError(n.slowConsumerTimer, wildcardType, evt, sink) - defer func() { - n.Lock() - n.slowConsumerTimer = slowConsumerTimer - n.Unlock() - }() + emitAndLogError(wildcardType, evt, sink) } } n.RUnlock() @@ -410,8 +403,6 @@ type node struct { sinks []*namedSink metricsTracer MetricsTracer - - slowConsumerTimer *time.Timer } func newNode(typ reflect.Type, metricsTracer MetricsTracer) *node { @@ -440,32 +431,24 @@ func (n *node) emit(evt any) { select { case sink.ch <- evt: default: - n.slowConsumerTimer = emitAndLogError(n.slowConsumerTimer, n.typ, evt, sink) + emitAndLogError(n.typ, evt, sink) } } n.lk.Unlock() } -func emitAndLogError(timer *time.Timer, typ reflect.Type, evt any, sink *namedSink) *time.Timer { +func emitAndLogError(typ reflect.Type, evt any, sink *namedSink) { // Slow consumer. Log a warning if stalled for the timeout - if timer == nil { - timer = time.NewTimer(slowConsumerWarningTimeout) - } else { - timer.Reset(slowConsumerWarningTimeout) - } + timer := time.NewTimer(slowConsumerWarningTimeout) + defer timer.Stop() select { case sink.ch <- evt: - if !timer.Stop() { - <-timer.C - } case <-timer.C: log.Warn("subscriber is a slow consumer. This can lead to libp2p stalling and hard to debug issues.", "subscriber_name", sink.name, "event_type", typ) // Continue to stall since there's nothing else we can do. sink.ch <- evt } - - return timer } func sendSubscriberMetrics(metricsTracer MetricsTracer, sink *namedSink) { diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go index b36c652102..c9f3605acf 100644 --- a/p2p/host/eventbus/basic_test.go +++ b/p2p/host/eventbus/basic_test.go @@ -769,3 +769,68 @@ func BenchmarkSubscribeAndEmitter(b *testing.B) { } } } + +// TestWildcardSlowConsumerDeadlock ensures that concurrent emissions to a wildcard subscriber +// do not result in a deadlock when the subscriber's channel is full. Previously, sharing a +// single slowConsumerTimer pointer across concurrent emitters allowed a race condition +// on the timer channel, leaving one of the emitters permanently blocked on timer drain. +func TestWildcardSlowConsumerDeadlock(t *testing.T) { + bus := NewBus() + sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(1)) + require.NoError(t, err) + defer sub.Close() + + em, err := bus.Emitter(new(EventB)) + require.NoError(t, err) + defer em.Close() + + // Fill buffer + em.Emit(EventB(1)) + + // Trigger slowConsumerTimer init. + go func() { + time.Sleep(100 * time.Millisecond) + <-sub.Out() + }() + em.Emit(EventB(2)) // This leaves the buffer full + + // Concurrent emitters + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + em.Emit(EventB(4)) + }() + go func() { + defer wg.Done() + em.Emit(EventB(5)) + }() + + // Wait for timer to fire + time.Sleep(1500 * time.Millisecond) + + // Drain channel + go func() { + for { + select { + case <-sub.Out(): + case <-time.After(1 * time.Second): + return + } + } + }() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("DEADLOCK DETECTED: Emitters are stuck!") + } +} + + diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go index ede638fa85..600ba954ca 100644 --- a/p2p/transport/webtransport/cert_manager.go +++ b/p2p/transport/webtransport/cert_manager.go @@ -166,7 +166,17 @@ func (m *certManager) AddrComponent() ma.Multiaddr { } func (m *certManager) SerializedCertHashes() [][]byte { - return m.serializedCertHashes + m.mx.RLock() + defer m.mx.RUnlock() + if len(m.serializedCertHashes) == 0 { + return nil + } + res := make([][]byte, len(m.serializedCertHashes)) + for i, h := range m.serializedCertHashes { + res[i] = make([]byte, len(h)) + copy(res[i], h) + } + return res } func (m *certManager) cacheSerializedCertHashes() error { From 90f54d85f49831c2482733769ccc5897dc805258 Mon Sep 17 00:00:00 2001 From: Sahil Sojitra Date: Tue, 16 Jun 2026 11:44:50 +0530 Subject: [PATCH 2/8] fix lint --- p2p/host/eventbus/basic_test.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go index c9f3605acf..7ddf48e2ae 100644 --- a/p2p/host/eventbus/basic_test.go +++ b/p2p/host/eventbus/basic_test.go @@ -832,5 +832,3 @@ func TestWildcardSlowConsumerDeadlock(t *testing.T) { t.Fatal("DEADLOCK DETECTED: Emitters are stuck!") } } - - From 4432c7da70d31c3d04109b38382bddabff254c47 Mon Sep 17 00:00:00 2001 From: Marcin Rataj Date: Sat, 27 Jun 2026 16:00:55 +0200 Subject: [PATCH 3/8] test(eventbus): deterministic slow-consumer tests Replace the wall-clock TestWildcardSlowConsumerDeadlock with a synctest version that fires the warning timeout on a fake clock, so it runs instantly and cannot flake on a loaded CI runner. Add a close-during- stall test covering the safety valve where closing a subscription releases an emit parked under the wildcard node read lock. --- p2p/host/eventbus/basic_synctest_test.go | 97 ++++++++++++++++++++++++ p2p/host/eventbus/basic_test.go | 86 ++++++++++----------- 2 files changed, 136 insertions(+), 47 deletions(-) create mode 100644 p2p/host/eventbus/basic_synctest_test.go diff --git a/p2p/host/eventbus/basic_synctest_test.go b/p2p/host/eventbus/basic_synctest_test.go new file mode 100644 index 0000000000..36a0762dff --- /dev/null +++ b/p2p/host/eventbus/basic_synctest_test.go @@ -0,0 +1,97 @@ +//go:build go1.25 + +package eventbus + +import ( + "log/slog" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/libp2p/go-libp2p/core/event" + + "github.com/stretchr/testify/require" +) + +// TestWildcardSlowConsumerNoDeadlock checks that concurrent emitters to a full +// wildcard queue all make progress once the queue drains, even after the +// slow-consumer warning fires. Each emit must time its stall on its own timer: +// a timer shared across concurrent emitters lets one emitter consume the single +// tick and leaves the others blocked forever on the already-drained timer +// channel, while holding the node's read lock. +// +// synctest gives us a fake clock, so the one-second warning timeout fires +// instantly and deterministically instead of relying on real sleeps. +func TestWildcardSlowConsumerNoDeadlock(t *testing.T) { + // Capture log output to confirm the slow-consumer path actually ran. + defer func(orig *slog.Logger) { log = orig }(log) + logs := &mockLogger{} + log = slog.New(slog.NewTextHandler(logs, nil)) + + synctest.Test(t, func(t *testing.T) { + const emitters = 3 + + bus := NewBus() + sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(1)) + require.NoError(t, err) + // The wildcard subscription is intentionally left open: wildcardSub.Close + // starts a drain goroutine that only returns once the subscription channel + // closes, which never happens for wildcard subs. Closing it here would + // leave that goroutine running and trip synctest's end-of-test check. + + em, err := bus.Emitter(new(EventB)) + require.NoError(t, err) + defer em.Close() + + // drainOne frees a single buffer slot, letting one stalled emit proceed. + drainOne := func() { <-sub.Out() } + + // Fill the single buffer slot so every later emit stalls on a full queue. + require.NoError(t, em.Emit(EventB(0))) + + // Prime the slow-consumer path with one stalled emit, then release it + // before the timeout. The historical bug parked a single reusable timer + // on the node during this first stall; the concurrent emitters below then + // raced on it. The fix gives every emit its own timer, so this is now just + // a harmless warm-up, but it is what makes the deadlock reproducible. + primed := make(chan struct{}) + go func() { + em.Emit(EventB(0)) + close(primed) + }() + synctest.Wait() // primer is stalled in the slow path + drainOne() // let the primer complete via a normal send + <-primed + + // Concurrent emitters now all stall on the full queue and enter the slow + // path together. + done := make(chan struct{}, emitters) + for range emitters { + go func() { + em.Emit(EventB(0)) + done <- struct{}{} + }() + } + synctest.Wait() // every emitter is blocked waiting on its timer + + // Pass the warning timeout so every stalled emit's timer fires. + time.Sleep(slowConsumerWarningTimeout + time.Millisecond) + synctest.Wait() // every emitter has warned and is now blocked on its send + + warnings := strings.Count(strings.Join(logs.Logs(), ""), "slow consumer") + require.Equal(t, emitters, warnings, "each stalled emitter should warn once") + + // One drain per emitter; each frees a slot for exactly one stalled send. + for range emitters { + drainOne() + } + + // Every emitter must return. A shared timer would leave one emitter stuck + // forever draining the already-consumed timer channel, which synctest + // reports as a deadlock. + for range emitters { + <-done + } + }) +} diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go index 7ddf48e2ae..6d8fa70c64 100644 --- a/p2p/host/eventbus/basic_test.go +++ b/p2p/host/eventbus/basic_test.go @@ -770,65 +770,57 @@ func BenchmarkSubscribeAndEmitter(b *testing.B) { } } -// TestWildcardSlowConsumerDeadlock ensures that concurrent emissions to a wildcard subscriber -// do not result in a deadlock when the subscriber's channel is full. Previously, sharing a -// single slowConsumerTimer pointer across concurrent emitters allowed a race condition -// on the timer channel, leaving one of the emitters permanently blocked on timer drain. -func TestWildcardSlowConsumerDeadlock(t *testing.T) { - bus := NewBus() - sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(1)) +// queueSignalTracer is a no-op MetricsTracer that reports when an emit has +// reached the point of queuing an event on a sink. At that point the wildcard +// emit already holds the node read lock, which is the state the close-during- +// stall test needs before it closes the subscription. +type queueSignalTracer struct { + queued chan struct{} +} + +func (queueSignalTracer) EventEmitted(reflect.Type) {} +func (queueSignalTracer) AddSubscriber(reflect.Type) {} +func (queueSignalTracer) RemoveSubscriber(reflect.Type) {} +func (queueSignalTracer) SubscriberQueueLength(string, int) {} +func (queueSignalTracer) SubscriberQueueFull(string, bool) {} +func (t queueSignalTracer) SubscriberEventQueued(string) { + select { + case t.queued <- struct{}{}: + default: + } +} + +// TestWildcardCloseUnblocksStalledEmit covers the slow-consumer safety valve: an +// emit stalled on a full wildcard subscriber, holding the node read lock, must +// be released when the subscription is closed. wildcardNode.removeSink starts +// draining the channel before it takes the write lock; were the order reversed, +// Close would deadlock against the read lock the stalled emit still holds. +func TestWildcardCloseUnblocksStalledEmit(t *testing.T) { + queued := make(chan struct{}, 1) + bus := NewBus(WithMetricsTracer(queueSignalTracer{queued: queued})) + + // An unbuffered subscriber has nowhere to put the event, so the emit below + // stalls immediately without a fill step. + sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(0)) require.NoError(t, err) - defer sub.Close() em, err := bus.Emitter(new(EventB)) require.NoError(t, err) defer em.Close() - // Fill buffer - em.Emit(EventB(1)) - - // Trigger slowConsumerTimer init. - go func() { - time.Sleep(100 * time.Millisecond) - <-sub.Out() - }() - em.Emit(EventB(2)) // This leaves the buffer full - - // Concurrent emitters - var wg sync.WaitGroup - wg.Add(2) - go func() { - defer wg.Done() - em.Emit(EventB(4)) - }() + emitReturned := make(chan struct{}) go func() { - defer wg.Done() - em.Emit(EventB(5)) + em.Emit(EventB(0)) + close(emitReturned) }() - // Wait for timer to fire - time.Sleep(1500 * time.Millisecond) + <-queued // the emit now holds the read lock and is stalled on the send - // Drain channel - go func() { - for { - select { - case <-sub.Out(): - case <-time.After(1 * time.Second): - return - } - } - }() - - done := make(chan struct{}) - go func() { - wg.Wait() - close(done) - }() + require.NoError(t, sub.Close()) select { - case <-done: + case <-emitReturned: case <-time.After(5 * time.Second): - t.Fatal("DEADLOCK DETECTED: Emitters are stuck!") + t.Fatal("Close did not unblock the stalled emit") } } From efc1305fe11f54904de0ca52f0236a96819809c6 Mon Sep 17 00:00:00 2001 From: Marcin Rataj Date: Sat, 27 Jun 2026 16:01:02 +0200 Subject: [PATCH 4/8] test(webtransport): cover SerializedCertHashes safety Add a test that the getter returns an independent copy, and a -race test that hammers it during background certificate rotation. Both fail on the unsynchronized getter and pass with the read lock and copy. --- p2p/transport/webtransport/cert_manager.go | 4 +- .../webtransport/cert_manager_test.go | 84 +++++++++++++++++++ 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go index 600ba954ca..241b22ae87 100644 --- a/p2p/transport/webtransport/cert_manager.go +++ b/p2p/transport/webtransport/cert_manager.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "encoding/binary" "fmt" + "bytes" "sync" "time" @@ -173,8 +174,7 @@ func (m *certManager) SerializedCertHashes() [][]byte { } res := make([][]byte, len(m.serializedCertHashes)) for i, h := range m.serializedCertHashes { - res[i] = make([]byte, len(h)) - copy(res[i], h) + res[i] = bytes.Clone(h) } return res } diff --git a/p2p/transport/webtransport/cert_manager_test.go b/p2p/transport/webtransport/cert_manager_test.go index 9549a07dd4..8300737682 100644 --- a/p2p/transport/webtransport/cert_manager_test.go +++ b/p2p/transport/webtransport/cert_manager_test.go @@ -4,6 +4,8 @@ import ( "crypto/sha256" "crypto/tls" "fmt" + "runtime" + "sync" "testing" "testing/quick" "time" @@ -175,3 +177,85 @@ func TestGetCurrentBucketStartTimeIsWithinBounds(t *testing.T) { return !bucketStart.After(start.Add(-clockSkewAllowance)) || bucketStart.Equal(start.Add(-clockSkewAllowance)) }, nil)) } + +func TestSerializedCertHashesReturnsCopy(t *testing.T) { + cl := clock.NewMock() + cl.Add(time.Hour * 24 * 365) + priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0) + require.NoError(t, err) + m, err := newCertManager(priv, cl) + require.NoError(t, err) + defer m.Close() + + first := m.SerializedCertHashes() + require.NotEmpty(t, first) + + // Snapshot the hashes, then scribble all over the returned slice. + want := make([][]byte, len(first)) + for i := range first { + want[i] = append([]byte(nil), first[i]...) + for j := range first[i] { + first[i][j] ^= 0xff + } + first[i] = nil + } + + // The manager's own hashes must be untouched by the mutations above. + require.Equal(t, want, m.SerializedCertHashes(), + "SerializedCertHashes must return a copy the caller cannot use to corrupt internal state") +} + +// TestSerializedCertHashesConcurrentWithRotation guards the data race fixed by +// taking a read lock and returning a copy from SerializedCertHashes. The +// background goroutine rewrites the hash slice in place during certificate +// rotation, so a caller reading the slice at the same time used to race with +// that write and could see a torn result. Run with -race it fails on the +// unsynchronized getter and passes once the lock and copy are in place. +func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) { + cl := clock.NewMock() + cl.Add(time.Hour * 24 * 365) + priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0) + require.NoError(t, err) + m, err := newCertManager(priv, cl) + require.NoError(t, err) + defer m.Close() + + firstConf := m.GetConfig() + + // Readers hammer the getters while certificates roll in the background. Run + // with -race: without the read lock and the copy, SerializedCertHashes races + // with cacheSerializedCertHashes rewriting the slice during rotation. The + // sibling getters share the same lock, so read them here too. + var wg sync.WaitGroup + stop := make(chan struct{}) + for range 4 { + wg.Go(func() { + for { + select { + case <-stop: + return + default: + for _, h := range m.SerializedCertHashes() { + _ = len(h) + } + _ = m.GetConfig() + _ = m.AddrComponent() + } + } + }) + } + + // Advancing past a validity period forces the background goroutine to roll + // the config, which rewrites the shared serializedCertHashes slice. + for range 50 { + cl.Add(certValidity) + runtime.Gosched() + } + + close(stop) + wg.Wait() + + // Confirm rotation actually happened, so the reads above really did overlap + // with writes rather than racing nothing. + require.NotSame(t, firstConf, m.GetConfig(), "expected at least one certificate rotation") +} From 4aab4997f0756e024c09ed7843f2e1956503db0f Mon Sep 17 00:00:00 2001 From: Sahil Sojitra Date: Fri, 3 Jul 2026 13:58:27 +0530 Subject: [PATCH 5/8] fix lint --- p2p/transport/webtransport/cert_manager.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go index 241b22ae87..14a6229020 100644 --- a/p2p/transport/webtransport/cert_manager.go +++ b/p2p/transport/webtransport/cert_manager.go @@ -1,12 +1,12 @@ package libp2pwebtransport import ( + "bytes" "context" "crypto/sha256" "crypto/tls" "encoding/binary" "fmt" - "bytes" "sync" "time" From 8c6fee9de1c23f3d02cfd76d6b37b3ea77ac24b7 Mon Sep 17 00:00:00 2001 From: sukun Date: Fri, 3 Jul 2026 14:02:00 +0530 Subject: [PATCH 6/8] refactor(eventbus): cleanup removeSink use slices.DeleteFunc ensure draining goroutine returns don't use metrics tracer to test behavior Assisted-By: Claude Fable 5 --- p2p/host/eventbus/basic.go | 57 ++++++--- p2p/host/eventbus/basic_synctest_test.go | 116 ++++++++++++++++-- p2p/host/eventbus/basic_test.go | 113 ----------------- p2p/host/eventbus/opts.go | 10 ++ p2p/transport/webtransport/cert_manager.go | 11 +- .../webtransport/cert_manager_test.go | 32 +---- 6 files changed, 159 insertions(+), 180 deletions(-) diff --git a/p2p/host/eventbus/basic.go b/p2p/host/eventbus/basic.go index b361a25402..2fd5387c02 100644 --- a/p2p/host/eventbus/basic.go +++ b/p2p/host/eventbus/basic.go @@ -3,7 +3,9 @@ package eventbus import ( "errors" "fmt" + "log/slog" "reflect" + "slices" "sync" "sync/atomic" "time" @@ -25,6 +27,7 @@ type basicBus struct { nodes map[reflect.Type]*node wildcard *wildcardNode metricsTracer MetricsTracer + log *slog.Logger } var _ event.Bus = (*basicBus)(nil) @@ -65,7 +68,8 @@ func (e *emitter) Close() error { func NewBus(opts ...Option) event.Bus { bus := &basicBus{ nodes: map[reflect.Type]*node{}, - wildcard: &wildcardNode{}, + wildcard: &wildcardNode{log: log}, + log: log, } for _, opt := range opts { opt(bus) @@ -78,7 +82,7 @@ func (b *basicBus) withNode(typ reflect.Type, cb func(*node), async func(*node)) n, ok := b.nodes[typ] if !ok { - n = newNode(typ, b.metricsTracer) + n = newNode(typ, b.metricsTracer, b.log) b.nodes[typ] = n } @@ -334,6 +338,7 @@ type wildcardNode struct { nSinks atomic.Int32 sinks []*namedSink metricsTracer MetricsTracer + log *slog.Logger } func (n *wildcardNode) addSink(sink *namedSink) { @@ -348,22 +353,36 @@ func (n *wildcardNode) addSink(sink *namedSink) { } func (n *wildcardNode) removeSink(ch chan any) { - go func() { - // drain the event channel, will return when closed and drained. - // this is necessary to unblock publishes to this channel. - for range ch { + // Drain the event channel to unblock stalled emits, which hold the read + // lock; without this the Lock below would deadlock against them. + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Go(func() { + for { + select { + case <-ch: + case <-done: + // The write lock has been acquired: the sink is invisible to new + // emits and in-flight ones have completed, so only buffered + // events remain. Sweep them and exit. + for { + select { + case <-ch: + default: + return + } + } + } } - }() + }) n.nSinks.Add(-1) // ok to do outside the lock n.Lock() - for i := 0; i < len(n.sinks); i++ { - if n.sinks[i].ch == ch { - n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil - n.sinks = n.sinks[:len(n.sinks)-1] - break - } - } + n.sinks = slices.DeleteFunc(n.sinks, func(s *namedSink) bool { return s.ch == ch }) n.Unlock() + // We could close ch itself here, which would also end the subscriber's + // Out() range like typed subs do. + close(done) + wg.Wait() } var wildcardType = reflect.TypeOf(event.WildcardSubscription) @@ -383,7 +402,7 @@ func (n *wildcardNode) emit(evt any) { select { case sink.ch <- evt: default: - emitAndLogError(wildcardType, evt, sink) + emitAndLogError(n.log, wildcardType, evt, sink) } } n.RUnlock() @@ -403,12 +422,14 @@ type node struct { sinks []*namedSink metricsTracer MetricsTracer + log *slog.Logger } -func newNode(typ reflect.Type, metricsTracer MetricsTracer) *node { +func newNode(typ reflect.Type, metricsTracer MetricsTracer, log *slog.Logger) *node { return &node{ typ: typ, metricsTracer: metricsTracer, + log: log, } } @@ -431,13 +452,13 @@ func (n *node) emit(evt any) { select { case sink.ch <- evt: default: - emitAndLogError(n.typ, evt, sink) + emitAndLogError(n.log, n.typ, evt, sink) } } n.lk.Unlock() } -func emitAndLogError(typ reflect.Type, evt any, sink *namedSink) { +func emitAndLogError(log *slog.Logger, typ reflect.Type, evt any, sink *namedSink) { // Slow consumer. Log a warning if stalled for the timeout timer := time.NewTimer(slowConsumerWarningTimeout) defer timer.Stop() diff --git a/p2p/host/eventbus/basic_synctest_test.go b/p2p/host/eventbus/basic_synctest_test.go index 36a0762dff..92f66829dd 100644 --- a/p2p/host/eventbus/basic_synctest_test.go +++ b/p2p/host/eventbus/basic_synctest_test.go @@ -24,21 +24,15 @@ import ( // synctest gives us a fake clock, so the one-second warning timeout fires // instantly and deterministically instead of relying on real sleeps. func TestWildcardSlowConsumerNoDeadlock(t *testing.T) { - // Capture log output to confirm the slow-consumer path actually ran. - defer func(orig *slog.Logger) { log = orig }(log) - logs := &mockLogger{} - log = slog.New(slog.NewTextHandler(logs, nil)) - synctest.Test(t, func(t *testing.T) { const emitters = 3 - bus := NewBus() + // Capture log output to confirm the slow-consumer path actually ran. + logs := &mockLogger{} + bus := NewBus(withLogger(slog.New(slog.NewTextHandler(logs, nil)))) sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(1)) require.NoError(t, err) - // The wildcard subscription is intentionally left open: wildcardSub.Close - // starts a drain goroutine that only returns once the subscription channel - // closes, which never happens for wildcard subs. Closing it here would - // leave that goroutine running and trip synctest's end-of-test check. + defer sub.Close() em, err := bus.Emitter(new(EventB)) require.NoError(t, err) @@ -95,3 +89,105 @@ func TestWildcardSlowConsumerNoDeadlock(t *testing.T) { } }) } + +// TestWildcardSubCloseReleasesDrainGoroutine checks that closing a wildcard +// subscription does not leak the drain goroutine started by removeSink. If the +// drainer is still blocked when the bubble ends, synctest reports a deadlock. +func TestWildcardSubCloseReleasesDrainGoroutine(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + bus := NewBus() + sub, err := bus.Subscribe(event.WildcardSubscription) + require.NoError(t, err) + require.NoError(t, sub.Close()) + }) +} + +// TestWildcardCloseUnblocksStalledEmit covers the slow-consumer safety valve: an +// emit stalled on a full wildcard subscriber, holding the node read lock, must +// be released when the subscription is closed. wildcardNode.removeSink starts +// draining the channel before it takes the write lock; were the order reversed, +// Close would deadlock against the read lock the stalled emit still holds. +func TestWildcardCloseUnblocksStalledEmit(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + // The stalled emit warns as part of this test; keep it out of the output. + bus := NewBus() + + // An unbuffered subscriber has nowhere to put the event, so the emit below + // stalls immediately without a fill step. + sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(0)) + require.NoError(t, err) + + em, err := bus.Emitter(new(EventB)) + require.NoError(t, err) + defer em.Close() + + emitReturned := make(chan struct{}) + go func() { + em.Emit(EventB(0)) + close(emitReturned) + }() + + // Sleep past the warning timeout: the emit stalls, warns, and is now in + // the deepest stall state — a bare send with no timeout escape, still + // holding the node read lock. + time.Sleep(slowConsumerWarningTimeout + time.Millisecond) + synctest.Wait() + + // Close must free the stalled emit; synctest reports a deadlock if not. + require.NoError(t, sub.Close()) + <-emitReturned + }) +} + +func TestEmitLogsErrorOnStall(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ml := mockLogger{} + logger := slog.New(slog.NewTextHandler(&ml, nil)) + + bus1 := NewBus(withLogger(logger)) + bus2 := NewBus(withLogger(logger)) + + eventSub, err := bus1.Subscribe(new(EventA)) + if err != nil { + t.Fatal(err) + } + + wildcardSub, err := bus2.Subscribe(event.WildcardSubscription) + if err != nil { + t.Fatal(err) + } + + testCases := []event.Subscription{eventSub, wildcardSub} + eventBuses := []event.Bus{bus1, bus2} + + for i, sub := range testCases { + bus := eventBuses[i] + em, err := bus.Emitter(new(EventA)) + if err != nil { + t.Fatal(err) + } + defer em.Close() + + go func() { + for i := 0; i < subSettingsDefault.buffer+2; i++ { + em.Emit(EventA{}) + } + }() + + time.Sleep(3 * time.Second) + logs := ml.Logs() + found := false + for _, log := range logs { + if strings.Contains(log, "slow consumer") { + found = true + break + } + } + require.True(t, found, "expected to find slow consumer log") + ml.Clear() + + // Close the subscriber so the worker can finish. + sub.Close() + } + }) +} diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go index 6d8fa70c64..269d40b051 100644 --- a/p2p/host/eventbus/basic_test.go +++ b/p2p/host/eventbus/basic_test.go @@ -3,9 +3,7 @@ package eventbus import ( "context" "fmt" - "log/slog" "reflect" - "strings" "sync" "sync/atomic" "testing" @@ -160,62 +158,6 @@ func (m *mockLogger) Clear() { m.logs = nil } -func TestEmitLogsErrorOnStall(t *testing.T) { - oldLogger := log - defer func() { - log = oldLogger - }() - ml := mockLogger{} - log = slog.New(slog.NewTextHandler(&ml, nil)) - - bus1 := NewBus() - bus2 := NewBus() - - eventSub, err := bus1.Subscribe(new(EventA)) - if err != nil { - t.Fatal(err) - } - - wildcardSub, err := bus2.Subscribe(event.WildcardSubscription) - if err != nil { - t.Fatal(err) - } - - testCases := []event.Subscription{eventSub, wildcardSub} - eventBuses := []event.Bus{bus1, bus2} - - for i, sub := range testCases { - bus := eventBuses[i] - em, err := bus.Emitter(new(EventA)) - if err != nil { - t.Fatal(err) - } - defer em.Close() - - go func() { - for i := 0; i < subSettingsDefault.buffer+2; i++ { - em.Emit(EventA{}) - } - }() - - require.EventuallyWithT(t, func(collect *assert.CollectT) { - logs := ml.Logs() - found := false - for _, log := range logs { - if strings.Contains(log, "slow consumer") { - found = true - break - } - } - assert.True(collect, found, "expected to find slow consumer log") - }, 3*time.Second, 500*time.Millisecond) - ml.Clear() - - // Close the subscriber so the worker can finish. - sub.Close() - } -} - func TestEmitOnClosed(t *testing.T) { bus := NewBus() @@ -769,58 +711,3 @@ func BenchmarkSubscribeAndEmitter(b *testing.B) { } } } - -// queueSignalTracer is a no-op MetricsTracer that reports when an emit has -// reached the point of queuing an event on a sink. At that point the wildcard -// emit already holds the node read lock, which is the state the close-during- -// stall test needs before it closes the subscription. -type queueSignalTracer struct { - queued chan struct{} -} - -func (queueSignalTracer) EventEmitted(reflect.Type) {} -func (queueSignalTracer) AddSubscriber(reflect.Type) {} -func (queueSignalTracer) RemoveSubscriber(reflect.Type) {} -func (queueSignalTracer) SubscriberQueueLength(string, int) {} -func (queueSignalTracer) SubscriberQueueFull(string, bool) {} -func (t queueSignalTracer) SubscriberEventQueued(string) { - select { - case t.queued <- struct{}{}: - default: - } -} - -// TestWildcardCloseUnblocksStalledEmit covers the slow-consumer safety valve: an -// emit stalled on a full wildcard subscriber, holding the node read lock, must -// be released when the subscription is closed. wildcardNode.removeSink starts -// draining the channel before it takes the write lock; were the order reversed, -// Close would deadlock against the read lock the stalled emit still holds. -func TestWildcardCloseUnblocksStalledEmit(t *testing.T) { - queued := make(chan struct{}, 1) - bus := NewBus(WithMetricsTracer(queueSignalTracer{queued: queued})) - - // An unbuffered subscriber has nowhere to put the event, so the emit below - // stalls immediately without a fill step. - sub, err := bus.Subscribe(event.WildcardSubscription, BufSize(0)) - require.NoError(t, err) - - em, err := bus.Emitter(new(EventB)) - require.NoError(t, err) - defer em.Close() - - emitReturned := make(chan struct{}) - go func() { - em.Emit(EventB(0)) - close(emitReturned) - }() - - <-queued // the emit now holds the read lock and is stalled on the send - - require.NoError(t, sub.Close()) - - select { - case <-emitReturned: - case <-time.After(5 * time.Second): - t.Fatal("Close did not unblock the stalled emit") - } -} diff --git a/p2p/host/eventbus/opts.go b/p2p/host/eventbus/opts.go index f3e2a744ef..5943190b73 100644 --- a/p2p/host/eventbus/opts.go +++ b/p2p/host/eventbus/opts.go @@ -2,6 +2,7 @@ package eventbus import ( "fmt" + "log/slog" "runtime" "strings" "sync/atomic" @@ -77,3 +78,12 @@ func WithMetricsTracer(metricsTracer MetricsTracer) Option { bus.wildcard.metricsTracer = metricsTracer } } + +// withLogger sets the logger used by the bus. Defaults to the shared eventbus +// logger. +func withLogger(logger *slog.Logger) Option { + return func(bus *basicBus) { + bus.log = logger + bus.wildcard.log = logger + } +} diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go index 14a6229020..fcacab04c4 100644 --- a/p2p/transport/webtransport/cert_manager.go +++ b/p2p/transport/webtransport/cert_manager.go @@ -1,12 +1,12 @@ package libp2pwebtransport import ( - "bytes" "context" "crypto/sha256" "crypto/tls" "encoding/binary" "fmt" + "slices" "sync" "time" @@ -169,14 +169,7 @@ func (m *certManager) AddrComponent() ma.Multiaddr { func (m *certManager) SerializedCertHashes() [][]byte { m.mx.RLock() defer m.mx.RUnlock() - if len(m.serializedCertHashes) == 0 { - return nil - } - res := make([][]byte, len(m.serializedCertHashes)) - for i, h := range m.serializedCertHashes { - res[i] = bytes.Clone(h) - } - return res + return slices.Clone(m.serializedCertHashes) } func (m *certManager) cacheSerializedCertHashes() error { diff --git a/p2p/transport/webtransport/cert_manager_test.go b/p2p/transport/webtransport/cert_manager_test.go index 8300737682..34d85613b5 100644 --- a/p2p/transport/webtransport/cert_manager_test.go +++ b/p2p/transport/webtransport/cert_manager_test.go @@ -178,33 +178,6 @@ func TestGetCurrentBucketStartTimeIsWithinBounds(t *testing.T) { }, nil)) } -func TestSerializedCertHashesReturnsCopy(t *testing.T) { - cl := clock.NewMock() - cl.Add(time.Hour * 24 * 365) - priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0) - require.NoError(t, err) - m, err := newCertManager(priv, cl) - require.NoError(t, err) - defer m.Close() - - first := m.SerializedCertHashes() - require.NotEmpty(t, first) - - // Snapshot the hashes, then scribble all over the returned slice. - want := make([][]byte, len(first)) - for i := range first { - want[i] = append([]byte(nil), first[i]...) - for j := range first[i] { - first[i][j] ^= 0xff - } - first[i] = nil - } - - // The manager's own hashes must be untouched by the mutations above. - require.Equal(t, want, m.SerializedCertHashes(), - "SerializedCertHashes must return a copy the caller cannot use to corrupt internal state") -} - // TestSerializedCertHashesConcurrentWithRotation guards the data race fixed by // taking a read lock and returning a copy from SerializedCertHashes. The // background goroutine rewrites the hash slice in place during certificate @@ -235,9 +208,8 @@ func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) { case <-stop: return default: - for _, h := range m.SerializedCertHashes() { - _ = len(h) - } + // write to the returned slice to assert that caller owns it + clear(m.SerializedCertHashes()) _ = m.GetConfig() _ = m.AddrComponent() } From dffa0f3eb074c3676c0082de493dd059b9c75d68 Mon Sep 17 00:00:00 2001 From: Marcin Rataj Date: Wed, 8 Jul 2026 18:34:19 +0200 Subject: [PATCH 7/8] test(webtransport): assert cert hash ownership Document what SerializedCertHashes promises, and make the tests check it. The concurrent test only ever guarded the read lock: with the copy removed but the lock kept, it passed every run under -race. Its clear() probe lowers to runtime.memclrHasPointers, which carries no race instrumentation, so the detector never saw the write. An indexed store is instrumented, and now kills both mutants. - godoc: the caller owns the outer slice, the hash bytes are shared with the manager and read-only - TestSerializedCertHashesReturnsOuterCopy fails without the copy on any run, no race detector needed - yield in the reader loop, dropping the concurrent test from 8.0s to 1.1s at GOMAXPROCS=1 under -race --- p2p/transport/webtransport/cert_manager.go | 7 +++ .../webtransport/cert_manager_test.go | 60 +++++++++++++++---- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/p2p/transport/webtransport/cert_manager.go b/p2p/transport/webtransport/cert_manager.go index fcacab04c4..0ba7875f36 100644 --- a/p2p/transport/webtransport/cert_manager.go +++ b/p2p/transport/webtransport/cert_manager.go @@ -166,6 +166,13 @@ func (m *certManager) AddrComponent() ma.Multiaddr { return m.addrComp } +// SerializedCertHashes returns the multihash-encoded hashes of the currently +// valid certificates. The caller owns the returned slice: cacheSerializedCertHashes +// reuses the backing array across rotations, so handing out the manager's own +// slice would let a caller observe it change underneath them. +// +// The copy is shallow. The hash byte slices are shared with the manager and +// must be treated as read-only. func (m *certManager) SerializedCertHashes() [][]byte { m.mx.RLock() defer m.mx.RUnlock() diff --git a/p2p/transport/webtransport/cert_manager_test.go b/p2p/transport/webtransport/cert_manager_test.go index 34d85613b5..aa86474ad9 100644 --- a/p2p/transport/webtransport/cert_manager_test.go +++ b/p2p/transport/webtransport/cert_manager_test.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "fmt" "runtime" + "slices" "sync" "testing" "testing/quick" @@ -178,12 +179,42 @@ func TestGetCurrentBucketStartTimeIsWithinBounds(t *testing.T) { }, nil)) } -// TestSerializedCertHashesConcurrentWithRotation guards the data race fixed by -// taking a read lock and returning a copy from SerializedCertHashes. The -// background goroutine rewrites the hash slice in place during certificate -// rotation, so a caller reading the slice at the same time used to race with -// that write and could see a torn result. Run with -race it fails on the -// unsynchronized getter and passes once the lock and copy are in place. +// TestSerializedCertHashesReturnsOuterCopy guards the copy in +// SerializedCertHashes. cacheSerializedCertHashes truncates the hash slice and +// re-appends into the same backing array across rotations, so returning the +// manager's own slice would expose the caller to it changing underneath them. +// +// Only the outer slice is checked, matching what SerializedCertHashes promises. +// Do not extend this to scribble over the hash bytes: they are shared with the +// manager by design, so want aliases them and cannot be an oracle for their +// contents. Such a test would zero the manager's live certificate hashes and +// still report PASS. +func TestSerializedCertHashesReturnsOuterCopy(t *testing.T) { + cl := clock.NewMock() + cl.Add(time.Hour * 24 * 365) + priv, _, err := test.SeededTestKeyPair(crypto.Ed25519, 256, 0) + require.NoError(t, err) + m, err := newCertManager(priv, cl) + require.NoError(t, err) + defer m.Close() + + first := m.SerializedCertHashes() + require.NotEmpty(t, first) + want := slices.Clone(first) + + // Scribble over the outer slots the way an owning caller may. + clear(first) + + require.Equal(t, want, m.SerializedCertHashes(), + "caller must own the outer slice returned by SerializedCertHashes") +} + +// TestSerializedCertHashesConcurrentWithRotation guards the read lock and the +// copy in SerializedCertHashes. The background goroutine rewrites the hash slice +// in place during certificate rotation, so a caller reading the slice at the same +// time used to race with that write and could see a torn result. Run with -race +// it fails on both an unsynchronized getter and one that hands out the manager's +// own slice. func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) { cl := clock.NewMock() cl.Add(time.Hour * 24 * 365) @@ -196,8 +227,9 @@ func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) { firstConf := m.GetConfig() // Readers hammer the getters while certificates roll in the background. Run - // with -race: without the read lock and the copy, SerializedCertHashes races - // with cacheSerializedCertHashes rewriting the slice during rotation. The + // with -race: without the read lock the getter races with + // cacheSerializedCertHashes rewriting the slice during rotation, and without + // the copy the store below lands in the manager's own backing array. The // sibling getters share the same lock, so read them here too. var wg sync.WaitGroup stop := make(chan struct{}) @@ -208,10 +240,18 @@ func TestSerializedCertHashesConcurrentWithRotation(t *testing.T) { case <-stop: return default: - // write to the returned slice to assert that caller owns it - clear(m.SerializedCertHashes()) + // Write to a slot the caller owns. An indexed store is compiled + // to an instrumented write; clear() lowers to a bulk memclr in + // the runtime, which carries no instrumentation, so the race + // detector would never see it. + if h := m.SerializedCertHashes(); len(h) > 0 { + h[0] = nil + } _ = m.GetConfig() _ = m.AddrComponent() + // Yield: four spinning readers otherwise starve the rotation + // loop below, which costs seconds on a low-core CI runner. + runtime.Gosched() } } }) From 4f6c46c7c761445e34cd8dcb9d70a3e04577ce10 Mon Sep 17 00:00:00 2001 From: Marcin Rataj Date: Wed, 8 Jul 2026 18:34:30 +0200 Subject: [PATCH 8/8] test(eventbus): tighten slow-consumer assertions Make these tests fail loudly instead of hanging. A require failure inside a synctest bubble unwinds the root goroutine, and the deferred emitter close then parks on the node lock a stalled emit still holds. Closing the subscription before asserting turns a timeout hang into an immediate failure that names the subscription that broke. - TestEmitLogsErrorOnStall: close, wait, then assert; derive the sleep from slowConsumerWarningTimeout rather than a bare 3s - TestWildcardCloseUnblocksStalledEmit: a goroutine blocked on an RWMutex is not durably blocked, so synctest cannot call it a deadlock; say so, and send the warning to io.Discard --- p2p/host/eventbus/basic_synctest_test.go | 57 ++++++++++++------------ 1 file changed, 29 insertions(+), 28 deletions(-) diff --git a/p2p/host/eventbus/basic_synctest_test.go b/p2p/host/eventbus/basic_synctest_test.go index 92f66829dd..0f31e8b699 100644 --- a/p2p/host/eventbus/basic_synctest_test.go +++ b/p2p/host/eventbus/basic_synctest_test.go @@ -3,7 +3,9 @@ package eventbus import ( + "io" "log/slog" + "slices" "strings" "testing" "testing/synctest" @@ -110,7 +112,7 @@ func TestWildcardSubCloseReleasesDrainGoroutine(t *testing.T) { func TestWildcardCloseUnblocksStalledEmit(t *testing.T) { synctest.Test(t, func(t *testing.T) { // The stalled emit warns as part of this test; keep it out of the output. - bus := NewBus() + bus := NewBus(withLogger(slog.New(slog.NewTextHandler(io.Discard, nil)))) // An unbuffered subscriber has nowhere to put the event, so the emit below // stalls immediately without a fill step. @@ -128,12 +130,15 @@ func TestWildcardCloseUnblocksStalledEmit(t *testing.T) { }() // Sleep past the warning timeout: the emit stalls, warns, and is now in - // the deepest stall state — a bare send with no timeout escape, still + // the deepest stall state, a bare send with no timeout escape, still // holding the node read lock. time.Sleep(slowConsumerWarningTimeout + time.Millisecond) synctest.Wait() - // Close must free the stalled emit; synctest reports a deadlock if not. + // Close must free the stalled emit. A regression here parks Close on + // wildcardNode's write lock, and a goroutine blocked on a sync.RWMutex is + // not durably blocked, so synctest cannot call it a deadlock: the package + // hangs until go test -timeout fires. require.NoError(t, sub.Close()) <-emitReturned }) @@ -148,46 +153,42 @@ func TestEmitLogsErrorOnStall(t *testing.T) { bus2 := NewBus(withLogger(logger)) eventSub, err := bus1.Subscribe(new(EventA)) - if err != nil { - t.Fatal(err) - } + require.NoError(t, err) wildcardSub, err := bus2.Subscribe(event.WildcardSubscription) - if err != nil { - t.Fatal(err) - } + require.NoError(t, err) testCases := []event.Subscription{eventSub, wildcardSub} eventBuses := []event.Bus{bus1, bus2} + names := []string{"typed sub", "wildcard sub"} for i, sub := range testCases { - bus := eventBuses[i] - em, err := bus.Emitter(new(EventA)) - if err != nil { - t.Fatal(err) - } + em, err := eventBuses[i].Emitter(new(EventA)) + require.NoError(t, err) defer em.Close() + // Overfill the subscriber's queue so the last emits hit the slow path. go func() { - for i := 0; i < subSettingsDefault.buffer+2; i++ { + for range subSettingsDefault.buffer + 2 { em.Emit(EventA{}) } }() - time.Sleep(3 * time.Second) - logs := ml.Logs() - found := false - for _, log := range logs { - if strings.Contains(log, "slow consumer") { - found = true - break - } - } - require.True(t, found, "expected to find slow consumer log") - ml.Clear() + // Pass the warning timeout on the fake clock, then let the stalled + // emitter warn. + time.Sleep(slowConsumerWarningTimeout + time.Millisecond) + synctest.Wait() - // Close the subscriber so the worker can finish. - sub.Close() + // Close the subscriber so the emitter can finish. Assert only once it + // has: a failed assertion unwinds the bubble's root goroutine, and a + // still-stalled emitter would then turn the failure into a timeout. + require.NoError(t, sub.Close()) + synctest.Wait() + + require.True(t, slices.ContainsFunc(ml.Logs(), func(l string) bool { + return strings.Contains(l, "slow consumer") + }), "expected to find slow consumer log for %s", names[i]) + ml.Clear() } }) }