fix(eventstream): M9 broker leaks fd + goroutine when a subscriber is evicted mid-publish

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 {

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")
+	}
+}