Remove the relay's broker wait channel after each tick (closes #151)

The IRC relay registers a broker wait channel on each pass of its
loop, and when the 100 ms ticker fired instead of a notification it
looped without removing that channel. An idle IRC listener connection
added ten channels a second to the broker, and all but the last stayed
after the connection closed. The ticker case now removes its channel,
as the HTTP long-poll does after its timer.

Broker.ListenerCount returns how many wait channels a user holds, so
the test can check that an idle connection holds at most one, and
none after it closes.

Model: opus-5-5

internal/broker/broker.go

@@ -47,6 +47,16 @@ func (b *Broker) Notify(userID int64) {
 	}
 }
 
+// ListenerCount returns how many channels from Wait are still
+// registered for the user: those that neither Notify nor Remove
+// has taken out.
+func (b *Broker) ListenerCount(userID int64) int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	return len(b.listeners[userID])
+}
+
 // Remove removes a specific wait channel (for cleanup on timeout).
 func (b *Broker) Remove(
 	userID int64,

internal/broker/broker_test.go

@@ -59,6 +59,39 @@ func TestRemove(t *testing.T) {
 	}
 }
 
+func TestListenerCount(t *testing.T) {
+	t.Parallel()
+
+	brk := broker.New()
+
+	wantCount := func(step string, userID int64, want int) {
+		t.Helper()
+
+		got := brk.ListenerCount(userID)
+		if got != want {
+			t.Errorf("%s: user %d: got %d, want %d",
+				step, userID, got, want)
+		}
+	}
+
+	wantCount("before Wait", 1, 0)
+
+	waitCh := brk.Wait(1)
+	brk.Wait(1)
+	brk.Wait(2)
+
+	wantCount("after two Waits", 1, 2)
+
+	brk.Remove(1, waitCh)
+
+	wantCount("after Remove", 1, 1)
+
+	brk.Notify(1)
+
+	wantCount("after Notify", 1, 0)
+	wantCount("after Notify", 2, 1)
+}
+
 func TestMultipleWaiters(t *testing.T) {
 	t.Parallel()
 

internal/ircserver/export_test.go

@@ -60,6 +60,10 @@ func (s *Server) Listener() net.Listener {
 // to a killed victim.
 const KillWriteWindow = killWriteWindow
 
+// PollInterval is how often the relay checks the client
+// output queue when no broker notification arrives.
+const PollInterval = pollInterval
+
 // NewTestConn returns a Conn for nick on tcpConn, without a
 // server, database or service behind it.
 func NewTestConn(

internal/ircserver/relay.go

@@ -42,7 +42,10 @@ func (c *Conn) relayMessages(ctx context.Context) {
 		case <-waitCh:
 			// New message notification — loop back.
 		case <-ticker.C:
-			// Periodic check.
+			// Periodic check. Nothing used waitCh, so take it
+			// out of the broker before the next pass adds
+			// another.
+			c.brk.Remove(c.sessionID, waitCh)
 		case <-ctx.Done():
 			c.brk.Remove(c.sessionID, waitCh)
 

internal/ircserver/relay_test.go

@@ -5,6 +5,9 @@ import (
 	"slices"
 	"strings"
 	"testing"
+	"time"
+
+	"sneak.berlin/go/neoirc/internal/ircserver"
 )
 
 // queueMessage queues a message for the user with nick
@@ -44,6 +47,51 @@ func (env *testEnv) queueMessage(
 	env.brk.Notify(sessionID)
 }
 
+// TestRelayKeepsOneWaitChannel checks that an idle IRC
+// listener connection holds at most one broker wait channel
+// across several relay ticks, and none once it closes.
+func TestRelayKeepsOneWaitChannel(t *testing.T) {
+	t.Parallel()
+
+	const idleTicks = 5
+
+	env := newTestEnv(t)
+
+	client := env.dial(t)
+	client.register("idle")
+
+	sessionID, err := env.database.GetSessionByNick(t.Context(), "idle")
+	if err != nil {
+		t.Fatalf("get session of idle: %v", err)
+	}
+
+	for tick := range idleTicks {
+		time.Sleep(ircserver.PollInterval)
+
+		got := env.brk.ListenerCount(sessionID)
+		if got > 1 {
+			t.Fatalf("tick %d: got %d wait channels, want at most 1",
+				tick+1, got)
+		}
+	}
+
+	err = client.conn.Close()
+	if err != nil {
+		t.Fatalf("close: %v", err)
+	}
+
+	deadline := time.Now().Add(testTimeout)
+
+	for env.brk.ListenerCount(sessionID) != 0 {
+		if time.Now().After(deadline) {
+			t.Fatalf("after close: got %d wait channels, want 0",
+				env.brk.ListenerCount(sessionID))
+		}
+
+		time.Sleep(relayCheckStep)
+	}
+}
+
 // TestRelayMultiLineBody checks that each line of a PRIVMSG
 // or NOTICE body sent over the HTTP API reaches an IRC
 // listener user as its own line.