diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 6974110..2b6b976 100644 --- a/internal/broker/broker.go +++ b/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, diff --git a/internal/broker/broker_test.go b/internal/broker/broker_test.go index f2443c0..d439a50 100644 --- a/internal/broker/broker_test.go +++ b/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() diff --git a/internal/ircserver/export_test.go b/internal/ircserver/export_test.go index 4e665b2..60d039b 100644 --- a/internal/ircserver/export_test.go +++ b/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( diff --git a/internal/ircserver/relay.go b/internal/ircserver/relay.go index 0dee85b..bbb5309 100644 --- a/internal/ircserver/relay.go +++ b/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) diff --git a/internal/ircserver/relay_test.go b/internal/ircserver/relay_test.go index caf0cc8..8ab07a4 100644 --- a/internal/ircserver/relay_test.go +++ b/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.