diff --git a/internal/handlers/api.go b/internal/handlers/api.go index b9bf7e2..e46a6a8 100644 --- a/internal/handlers/api.go +++ b/internal/handlers/api.go @@ -813,7 +813,14 @@ func (hdlr *Handlers) HandleGetMessages() http.HandlerFunc { timeout = maxLongPollTimeout } - msgs, lastQID, err := hdlr.params.Database.PollMessages( + // Register with the broker before reading the queue. + // Notify wakes only the channels registered at that + // moment, so a message queued after the read below + // must find this one there already. + waitCh := hdlr.broker.Wait(sessionID) + defer hdlr.broker.Remove(sessionID, waitCh) + + msgs, lastQID, err := hdlr.pollMessages( request.Context(), clientID, afterID, pollMessageLimit, ) @@ -841,19 +848,21 @@ func (hdlr *Handlers) HandleGetMessages() http.HandlerFunc { hdlr.longPoll( writer, request, - sessionID, clientID, afterID, timeout, + waitCh, clientID, afterID, timeout, ) } } +// longPoll waits up to timeout seconds for the broker to +// signal waitCh, then answers with the messages queued for +// clientID after afterID. func (hdlr *Handlers) longPoll( writer http.ResponseWriter, request *http.Request, - sessionID, clientID, afterID int64, + waitCh <-chan struct{}, + clientID, afterID int64, timeout int, ) { - waitCh := hdlr.broker.Wait(sessionID) - timer := time.NewTimer( time.Duration(timeout) * time.Second, ) @@ -864,14 +873,10 @@ func (hdlr *Handlers) longPoll( case <-waitCh: case <-timer.C: case <-request.Context().Done(): - hdlr.broker.Remove(sessionID, waitCh) - return } - hdlr.broker.Remove(sessionID, waitCh) - - msgs, lastQID, err := hdlr.params.Database.PollMessages( + msgs, lastQID, err := hdlr.pollMessages( request.Context(), clientID, afterID, pollMessageLimit, ) diff --git a/internal/handlers/api_test.go b/internal/handlers/api_test.go index 05b8603..9395e63 100644 --- a/internal/handlers/api_test.go +++ b/internal/handlers/api_test.go @@ -140,6 +140,7 @@ type testServer struct { fxApp *fxtest.App broker *broker.Broker database *db.Database + handlers *handlers.Handlers } func newTestServer( @@ -170,6 +171,7 @@ func newTestServerWith( srv *server.Server brk *broker.Broker database *db.Database + hdlr *handlers.Handlers ) app := fxtest.New(t, @@ -210,7 +212,7 @@ func newTestServerWith( newTestHandlers, newTestServerFx, ), - fx.Populate(&srv, &brk, &database), + fx.Populate(&srv, &brk, &database, &hdlr), ) app.RequireStart() @@ -228,6 +230,7 @@ func newTestServerWith( fxApp: app, broker: brk, database: database, + handlers: hdlr, } } @@ -2554,6 +2557,110 @@ func TestLongPollTimeout(t *testing.T) { } } +// TestLongPollReturnsMessageQueuedAfterFirstRead checks that +// a message queued, and the broker notified, after the long +// poll's first read of the queue found nothing but before the +// poll waits, is returned at once and not when the poll times +// out. +func TestLongPollReturnsMessageQueuedAfterFirstRead(t *testing.T) { + const ( + nick = "lp_late" + sender = "lp_late_sender" + + // pollTimeout is the long poll's timeout, in seconds. + pollTimeout = 10 + + // atOnce is the longest the long poll may take to + // return the message. + atOnce = 5 * time.Second + ) + + tserver := newTestServer(t) + token := tserver.createSession(nick) + + _, lastID := tserver.pollMessages(token, 0) + + sessionID, err := tserver.database.GetSessionByNick(t.Context(), nick) + if err != nil { + t.Fatalf("get session of %s: %v", nick, err) + } + + queued := make(chan error, 1) + + var once sync.Once + + tserver.handlers.SetPollMessages(func( + ctx context.Context, + clientID, afterQueueID int64, + limit int, + ) ([]db.IRCMessage, int64, error) { + msgs, lastQID, pollErr := tserver.database.PollMessages( + ctx, clientID, afterQueueID, limit, + ) + + once.Do(func() { + queued <- tserver.queueMessage(ctx, sessionID, sender, nick) + }) + + return msgs, lastQID, pollErr + }) + + start := time.Now() + + msgs, err := tserver.longPoll(token, lastID, pollTimeout) + if err != nil { + t.Fatalf("long poll: %v", err) + } + + elapsed := time.Since(start) + + select { + case queueErr := <-queued: + if queueErr != nil { + t.Fatalf("queue message: %v", queueErr) + } + default: + t.Fatal("the long poll did not read the queue") + } + + if elapsed > atOnce { + t.Fatalf("the long poll returned after %v, want within %v", + elapsed, atOnce) + } + + if !findMessage(msgs, privmsgCmd, sender) { + t.Fatalf("long poll: got %v, want the PRIVMSG from %s", + msgs, sender) + } +} + +// queueMessage stores a PRIVMSG from sender to nick, queues +// it to the session sessionID and notifies the broker, as the +// server does for a DM. It returns errors instead of failing +// the test, so a handler's goroutine can call it. +func (tserver *testServer) queueMessage( + ctx context.Context, + sessionID int64, + sender, nick string, +) error { + msgID, _, err := tserver.database.InsertMessage( + ctx, privmsgCmd, sender, nick, sessionID, 0, + nil, json.RawMessage(`["hello"]`), nil, + ) + if err != nil { + return fmt.Errorf("insert message: %w", err) + } + + err = tserver.database.EnqueueToSession(ctx, sessionID, msgID) + if err != nil { + return fmt.Errorf("enqueue: %w", err) + } + + tserver.broker.Notify(sessionID) + + return nil +} + func TestEphemeralChannelCleanup(t *testing.T) { const channel = "#ephemeral" diff --git a/internal/handlers/export_test.go b/internal/handlers/export_test.go new file mode 100644 index 0000000..67d1b65 --- /dev/null +++ b/internal/handlers/export_test.go @@ -0,0 +1,19 @@ +package handlers + +import ( + "context" + + "sneak.berlin/go/neoirc/internal/db" +) + +// SetPollMessages makes HandleGetMessages read the client +// output queue with poll in place of Database.PollMessages. +func (hdlr *Handlers) SetPollMessages( + poll func( + ctx context.Context, + clientID, afterQueueID int64, + limit int, + ) ([]db.IRCMessage, int64, error), +) { + hdlr.pollMessages = poll +} diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 690b045..9a61699 100644 --- a/internal/handlers/handlers.go +++ b/internal/handlers/handlers.go @@ -64,6 +64,15 @@ type Handlers struct { loginLimiter *ratelimit.Limiter stats *stats.Tracker cancelCleanup context.CancelFunc + + // pollMessages is Database.PollMessages, with which + // HandleGetMessages reads the client output queue. Tests + // replace it to queue a message just after a read. + pollMessages func( + ctx context.Context, + clientID, afterQueueID int64, + limit int, + ) ([]db.IRCMessage, int64, error) } // New creates a new Handlers instance. @@ -95,6 +104,7 @@ func New( hashcashVal: hashcash.NewValidator(resource), loginLimiter: ratelimit.New(loginRate, loginBurst), stats: params.Stats, + pollMessages: params.Database.PollMessages, } lifecycle.Append(fx.Hook{ diff --git a/internal/ircserver/relay.go b/internal/ircserver/relay.go index bbb5309..c568133 100644 --- a/internal/ircserver/relay.go +++ b/internal/ircserver/relay.go @@ -27,17 +27,22 @@ func (c *Conn) relayMessages(ctx context.Context) { default: } + // Register with the broker before draining, so that a + // message queued after the drain has read the queue + // still signals waitCh. + waitCh := c.brk.Wait(c.sessionID) + // Drain any available messages. delivered := c.drainQueue(ctx) if delivered { + c.brk.Remove(c.sessionID, waitCh) + // Tight loop while there are messages. continue } // Wait for notification or timeout. - waitCh := c.brk.Wait(c.sessionID) - select { case <-waitCh: // New message notification — loop back.