diff --git a/service.go b/service.go index 6a11f8c..b16d1b1 100644 --- a/service.go +++ b/service.go @@ -164,7 +164,14 @@ func newSubscriber(conn coreapi.Stream) *subscriber { return &subscriber{conn: conn} } -func (s *subscriber) Close() error { return s.conn.Close() } +// Close releases the underlying stream. Tolerates a subscriber with no +// stream behind it, matching remote(). +func (s *subscriber) Close() error { + if s.conn == nil { + return nil + } + return s.conn.Close() +} func (s *subscriber) remote() string { if s.conn == nil { @@ -436,7 +443,14 @@ func (b *broker) publishWith(evt *Event, sender *subscriber, write eventWriter) } wg.Wait() for _, s := range dead { + // Full teardown, not just deregistration. The peer is by + // definition not draining, so its handleConn goroutine is parked + // in ReadEvent and any writer still blocked on the same stream + // stays blocked; closing the stream is what returns both, along + // with the descriptor and the rate-bucket entry. b.removeSub(s) + b.forgetPublisher(s) + _ = s.Close() } slog.Info("eventstream published", "topic", evt.Topic, "bytes", len(evt.Payload), "from", sender.remote(), "targets", len(targets)) if b.events != nil { diff --git a/zz_evict_teardown_test.go b/zz_evict_teardown_test.go new file mode 100644 index 0000000..22dbc6f --- /dev/null +++ b/zz_evict_teardown_test.go @@ -0,0 +1,93 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +//go:build !no_eventstream +// +build !no_eventstream + +package eventstream + +import ( + "testing" + "time" +) + +func (b *broker) rateEntries() int { + b.rateMu.Lock() + defer b.rateMu.Unlock() + return len(b.rate) +} + +func (b *broker) subCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + n := 0 + for _, subs := range b.subs { + n += len(subs) + } + return n +} + +// A subscriber dropped for repeated publish failures is, by definition, +// a peer that isn't draining. Its handleConn goroutine is parked in +// ReadEvent and will stay parked — along with the stream it holds — +// unless eviction closes the stream. Deregistering alone leaks both. +func TestEvictedSubscriberIsTornDown(t *testing.T) { + b := newBroker(nil, defaultAllowPolicy{}) + + subStream, peerStream := newPipeStreamPair() + sub := newSubscriber(subStream) + + handled := make(chan struct{}) + go func() { + defer close(handled) + b.handleConn(sub) + }() + + // The peer subscribes, then goes quiet: handleConn parks on its next + // ReadEvent, exactly like a peer that has stopped talking. + if err := WriteEvent(peerStream, &Event{Topic: "topic-x", Payload: []byte("sub")}); err != nil { + t.Fatalf("subscribe write: %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for b.subCount() == 0 { + if time.Now().After(deadline) { + t.Fatal("subscriber never registered") + } + time.Sleep(time.Millisecond) + } + + // Give the subscriber a rate bucket, as any publisher would have. + if !b.takeToken(sub) { + t.Fatal("takeToken denied on a fresh bucket") + } + if b.rateEntries() != 1 { + t.Fatalf("rate entries = %d, want 1", b.rateEntries()) + } + + // Fail past the tolerance so the subscriber is dropped. + write, _ := alwaysFailWriter() + for i := 0; i < maxConsecutivePublishFailures; i++ { + b.publishWith(&Event{Topic: "topic-x", Payload: []byte("p")}, stubSubscriber(), write) + } + + if b.subCount() != 0 { + t.Fatalf("subscriber count = %d after eviction, want 0", b.subCount()) + } + + select { + case <-handled: + case <-time.After(3 * time.Second): + t.Fatal("handleConn still parked after eviction: the stream was never closed") + } + + if got := b.rateEntries(); got != 0 { + t.Fatalf("rate entries = %d after eviction, want 0", got) + } + + // The stream is closed, so the peer side sees the connection go away + // rather than hanging on a descriptor nobody will ever read. + _ = peerStream.Close() + if err := WriteEvent(subStream, &Event{Topic: "topic-x"}); err == nil { + t.Fatal("evicted subscriber's stream is still open") + } +}