Remove the relay's broker wait channel after each tick (closes #151)
check / check (push) Waiting to run
check / check (push) Waiting to run
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 Co-authored-by: clawbot <sneak+clawbot@sneak.cloud>
This commit was merged in pull request #160.
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user