diff --git a/README.md b/README.md index 6f983ab..5927f7f 100644 --- a/README.md +++ b/README.md @@ -2569,12 +2569,14 @@ issues) and simpler than UUIDs (integer comparison vs. string comparison). - **Sessions**: Both anonymous and password-protected sessions are deleted on `QUIT` or when the last client logs out (`POST /api/v1/logout` with no remaining clients triggers session cleanup). There is no distinction between - session types in the cleanup path — `handleQuit` and `cleanupUser` both call - `DeleteSession` unconditionally. Idle clients are automatically removed after - `SESSION_IDLE_TIMEOUT` (default 30 days) without an authenticated HTTP API - request or, on the IRC listener, a command — the server runs a background - cleanup loop that, for a session left with no clients, parts the user from all - channels, broadcasts QUIT, and releases the nick. + session types in the cleanup path: the session is parted from its channels and + deleted in one database transaction, and when several QUITs or logouts end one + session at once, only the one that deletes it broadcasts QUIT. Idle clients + are automatically removed after `SESSION_IDLE_TIMEOUT` (default 30 days) + without an authenticated HTTP API request or, on the IRC listener, a command — + the server runs a background cleanup loop that, for a session left with no + clients, parts the user from all channels, broadcasts QUIT, and releases the + nick. - **Clients**: Individual client auth cookies are invalidated on `POST /api/v1/logout`. A session can have multiple clients; removing one doesn't affect others. However, when the last client is removed (via logout), diff --git a/internal/db/queries.go b/internal/db/queries.go index 3db2e63..7b2a242 100644 --- a/internal/db/queries.go +++ b/internal/db/queries.go @@ -905,7 +905,29 @@ func (database *Database) PartChannelAndDeleteIfEmpty( defer func() { _ = transaction.Rollback() }() - _, err = transaction.ExecContext(ctx, + err = partChannelAndDeleteIfEmptyInTx( + ctx, transaction, channelID, sessionID, + ) + if err != nil { + return err + } + + err = transaction.Commit() + if err != nil { + return fmt.Errorf("commit part: %w", err) + } + + return nil +} + +// partChannelAndDeleteIfEmptyInTx is +// PartChannelAndDeleteIfEmpty, writing through transaction. +func partChannelAndDeleteIfEmptyInTx( + ctx context.Context, + transaction *sql.Tx, + channelID, sessionID int64, +) error { + _, err := transaction.ExecContext(ctx, `DELETE FROM channel_members WHERE channel_id = ? AND session_id = ?`, channelID, sessionID) @@ -925,11 +947,6 @@ func (database *Database) PartChannelAndDeleteIfEmpty( ) } - err = transaction.Commit() - if err != nil { - return fmt.Errorf("commit part: %w", err) - } - return nil } @@ -1487,21 +1504,150 @@ func (database *Database) DeleteSession( return nil } -// DeleteClient removes a single client record by ID. -func (database *Database) DeleteClient( +// QuitSession parts a session from its channels, deleting +// each channel that this leaves with no members, and +// deletes the session with its clients, in one +// transaction. It returns the other sessions that were on +// those channels, and whether this call deleted the +// session: of several calls for one session at once, only +// one finds it still there, and the others change nothing. +func (database *Database) QuitSession( ctx context.Context, - clientID int64, -) error { - _, err := database.conn.ExecContext( - ctx, - "DELETE FROM clients WHERE id = ?", - clientID, - ) + sessionID int64, +) ([]int64, bool, error) { + transaction, err := database.conn.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("delete client: %w", err) + return nil, false, fmt.Errorf("begin quit: %w", err) } - return nil + defer func() { _ = transaction.Rollback() }() + + peerIDs, deleted, err := quitSessionInTx( + ctx, transaction, sessionID, + ) + if err != nil { + return nil, false, err + } + + err = transaction.Commit() + if err != nil { + return nil, false, fmt.Errorf("commit quit: %w", err) + } + + return peerIDs, deleted, nil +} + +// LogoutClient deletes a client of a session and, if that +// leaves the session with no clients, does what +// QuitSession does, in one transaction. It returns what +// QuitSession returns, or nothing and false if the session +// keeps a client. +func (database *Database) LogoutClient( + ctx context.Context, + sessionID, clientID int64, +) ([]int64, bool, error) { + transaction, err := database.conn.BeginTx(ctx, nil) + if err != nil { + return nil, false, fmt.Errorf("begin logout: %w", err) + } + + defer func() { _ = transaction.Rollback() }() + + _, err = transaction.ExecContext(ctx, + "DELETE FROM clients WHERE id = ?", clientID) + if err != nil { + return nil, false, fmt.Errorf("delete client: %w", err) + } + + var remaining int64 + + err = transaction.QueryRowContext(ctx, + "SELECT COUNT(*) FROM clients WHERE session_id = ?", + sessionID, + ).Scan(&remaining) + if err != nil { + return nil, false, fmt.Errorf("count clients: %w", err) + } + + var ( + peerIDs []int64 + deleted bool + ) + + if remaining == 0 { + peerIDs, deleted, err = quitSessionInTx( + ctx, transaction, sessionID, + ) + if err != nil { + return nil, false, err + } + } + + err = transaction.Commit() + if err != nil { + return nil, false, fmt.Errorf("commit logout: %w", err) + } + + return peerIDs, deleted, nil +} + +// quitSessionInTx is QuitSession, writing through +// transaction. +func quitSessionInTx( + ctx context.Context, + transaction *sql.Tx, + sessionID int64, +) ([]int64, bool, error) { + rows, err := transaction.QueryContext(ctx, + `SELECT DISTINCT session_id FROM channel_members + WHERE session_id != ? AND channel_id IN + (SELECT channel_id FROM channel_members + WHERE session_id = ?)`, + sessionID, sessionID) + if err != nil { + return nil, false, fmt.Errorf("get channel peers: %w", err) + } + + peerIDs, err := scanInt64s(rows) + if err != nil { + return nil, false, err + } + + rows, err = transaction.QueryContext(ctx, + `SELECT channel_id FROM channel_members + WHERE session_id = ?`, sessionID) + if err != nil { + return nil, false, fmt.Errorf( + "get session channel ids: %w", err, + ) + } + + channelIDs, err := scanInt64s(rows) + if err != nil { + return nil, false, err + } + + for _, channelID := range channelIDs { + err = partChannelAndDeleteIfEmptyInTx( + ctx, transaction, channelID, sessionID, + ) + if err != nil { + return nil, false, err + } + } + + result, err := transaction.ExecContext(ctx, + "DELETE FROM sessions WHERE id = ?", sessionID) + if err != nil { + return nil, false, fmt.Errorf("delete session: %w", err) + } + + deletedRows, err := result.RowsAffected() + if err != nil { + return nil, false, fmt.Errorf("delete session: %w", err) + } + + return peerIDs, deletedRows > 0, nil } // GetUserCount returns the number of active users. @@ -1543,29 +1689,6 @@ func (database *Database) GetOperCount( return count, nil } -// ClientCountForSession returns the number of clients -// belonging to a session. -func (database *Database) ClientCountForSession( - ctx context.Context, - sessionID int64, -) (int64, error) { - var count int64 - - err := database.conn.QueryRowContext( - ctx, - `SELECT COUNT(*) FROM clients - WHERE session_id = ?`, - sessionID, - ).Scan(&count) - if err != nil { - return 0, fmt.Errorf( - "client count for session: %w", err, - ) - } - - return count, nil -} - // DeleteStaleUsers removes clients not seen since the // cutoff and cleans up orphaned users (sessions). func (database *Database) DeleteStaleUsers( diff --git a/internal/handlers/api.go b/internal/handlers/api.go index 9e10b46..5f73c49 100644 --- a/internal/handlers/api.go +++ b/internal/handlers/api.go @@ -3395,15 +3395,11 @@ func (hdlr *Handlers) HandleLogout() http.HandlerFunc { return } - ctx := request.Context() - - err := hdlr.params.Database.DeleteClient( - ctx, clientID, + err := hdlr.svc.Logout( + request.Context(), sessionID, clientID, nick, ) if err != nil { - hdlr.log.Error( - "delete client failed", "error", err, - ) + hdlr.log.Error("logout failed", "error", err) hdlr.respondError( writer, request, "internal error", @@ -3413,23 +3409,6 @@ func (hdlr *Handlers) HandleLogout() http.HandlerFunc { return } - // If no clients remain, clean up the user fully: - // part all channels (notifying members) and - // delete the session. - remaining, err := hdlr.params.Database. - ClientCountForSession(ctx, sessionID) - if err != nil { - hdlr.log.Error( - "client count check failed", "error", err, - ) - } - - if remaining == 0 { - hdlr.cleanupUser( - ctx, sessionID, nick, - ) - } - hdlr.clearAuthCookie(writer) hdlr.respondJSON(writer, request, diff --git a/internal/handlers/api_test.go b/internal/handlers/api_test.go index fca94c9..3fdbad3 100644 --- a/internal/handlers/api_test.go +++ b/internal/handlers/api_test.go @@ -1393,6 +1393,170 @@ func TestQuit(t *testing.T) { } } +// quitRaceTrials is how many times checkQuitRace repeats +// its race. +const quitRaceTrials = 5 + +// TestConcurrentLogoutsSendOneQuit checks that when both +// clients of a session log out at once, the session's QUIT +// is sent once. +func TestConcurrentLogoutsSendOneQuit(t *testing.T) { + checkQuitRace(t, func(tserver *testServer, tokens []string) { + tserver.postConcurrently("/api/v1/logout", nil, tokens...) + }) +} + +// TestConcurrentQuitsSendOneQuit checks that when both +// clients of a session send QUIT at once, the session's +// QUIT is sent once. +func TestConcurrentQuitsSendOneQuit(t *testing.T) { + quit, err := json.Marshal(map[string]any{commandKey: irc.CmdQuit}) + if err != nil { + t.Fatal(err) + } + + checkQuitRace(t, func(tserver *testServer, tokens []string) { + tserver.postConcurrently(apiMessages, quit, tokens...) + }) +} + +// checkQuitRace gives a session a password and two +// clients, puts it on a channel with a second session, and +// calls leave with the tokens of the two clients. It checks +// that the second session then holds one QUIT from the +// first, and that the first session is deleted: its tokens +// are refused and its nick is free. It does all this +// quitRaceTrials times, each on a new server, because each +// trial logs in and logins are rate limited. +func checkQuitRace( + t *testing.T, + leave func(tserver *testServer, tokens []string), +) { + t.Helper() + + const nick = "racingquitter" + + for trial := range quitRaceTrials { + tserver := newTestServer(t) + firstToken := tserver.createSession(nick) + observerToken := tserver.createSession("raceobserver") + + tserver.sendCommand(firstToken, map[string]any{ + commandKey: irc.CmdPass, bodyKey: []string{testPassword}, + }) + + tokens := []string{firstToken, tserver.login(nick)} + + for _, token := range []string{firstToken, observerToken} { + tserver.sendCommand(token, map[string]any{ + commandKey: joinCmd, toKey: "#quitrace", + }) + } + + _, lastID := tserver.pollMessages(observerToken, 0) + + leave(tserver, tokens) + + msgs, _ := tserver.pollMessages(observerToken, lastID) + + quits := 0 + + for _, msg := range msgs { + if msg[commandKey] == irc.CmdQuit && msg["from"] == nick { + quits++ + } + } + + if quits != 1 { + t.Errorf("trial %d: %d QUITs, want 1: %v", trial, quits, msgs) + } + + for _, token := range tokens { + status, _ := tserver.getState(token) + if status != http.StatusUnauthorized { + t.Errorf("trial %d: state got %d, want 401", trial, status) + } + } + + // createSession fails the test if the nick is taken. + tserver.createSession(nick) + } +} + +// login logs in to the session nick with testPassword and +// returns the new client's token. +func (tserver *testServer) login(nick string) string { + tserver.t.Helper() + + body, err := json.Marshal(map[string]string{ + nickKey: nick, passwordKey: testPassword, + }) + if err != nil { + tserver.t.Fatalf("marshal login: %v", err) + } + + resp, err := doRequest( + tserver.t, http.MethodPost, + tserver.url("/api/v1/login"), bytes.NewReader(body), + ) + if err != nil { + tserver.t.Fatalf("login: %v", err) + } + + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + tserver.t.Fatalf("login: status %d", resp.StatusCode) + } + + for _, cookie := range resp.Cookies() { + if cookie.Name == authCookieName { + return cookie.Value + } + } + + tserver.t.Fatal("no auth cookie in login response") + + return "" +} + +// postConcurrently POSTs body to path once with each of +// tokens, all at once. +func (tserver *testServer) postConcurrently( + path string, body []byte, tokens ...string, +) { + tserver.t.Helper() + + start := make(chan struct{}) + + var waitGroup sync.WaitGroup + + for _, token := range tokens { + waitGroup.Add(1) + + go func() { + defer waitGroup.Done() + + <-start + + resp, err := doRequestAuth( + tserver.t, http.MethodPost, tserver.url(path), + token, bytes.NewReader(body), + ) + if err != nil { + tserver.t.Error(err) + + return + } + + _ = resp.Body.Close() + }() + } + + close(start) + waitGroup.Wait() +} + func TestUnknownCommand(t *testing.T) { tserver := newTestServer(t) token := tserver.createSession("cmdtest") diff --git a/internal/service/service.go b/internal/service/service.go index dabe74f..9124f27 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -826,61 +826,42 @@ func (s *Service) ChangeNick( return nil } -// BroadcastQuit broadcasts a QUIT to all channel peers, -// parts all channels, and deletes the session. Uses the -// FanOut pattern: one message row fanned out to all unique -// peer sessions. +// BroadcastQuit parts the session from all its channels, +// deletes it, and sends a QUIT to the users who shared a +// channel with it. Of several calls for one session at +// once, only the one that deletes it sends the QUIT. func (s *Service) BroadcastQuit( ctx context.Context, sessionID int64, nick, reason string, ) { - channels, err := s.db.GetSessionChannels( - ctx, sessionID, - ) - if err != nil { + peerIDs, deleted, err := s.db.QuitSession(ctx, sessionID) + if err != nil || !deleted { return } - notified := make(map[int64]bool) + s.sendQuit(ctx, nick, reason, peerIDs) +} - for _, ch := range channels { - memberIDs, memErr := s.db.GetChannelMemberIDs( - ctx, ch.ID, - ) - if memErr != nil { - continue - } - - for _, mid := range memberIDs { - if mid == sessionID || notified[mid] { - continue - } - - notified[mid] = true - } +// Logout deletes a client of the session and, if that was +// the session's last client, does what BroadcastQuit does. +func (s *Service) Logout( + ctx context.Context, + sessionID, clientID int64, + nick string, +) error { + peerIDs, deleted, err := s.db.LogoutClient( + ctx, sessionID, clientID, + ) + if err != nil { + return fmt.Errorf("logout: %w", err) } - if len(notified) > 0 { - recipients := make([]int64, 0, len(notified)) - for sid := range notified { - recipients = append(recipients, sid) - } - - body, _ := json.Marshal([]string{reason}) //nolint:errchkjson - - _, _, _ = s.FanOut( - ctx, irc.CmdQuit, nick, "", 0, 0, - nil, body, nil, recipients, - ) + if deleted { + s.sendQuit(ctx, nick, "Connection closed", peerIDs) } - for _, ch := range channels { - s.db.PartChannel(ctx, ch.ID, sessionID) //nolint:errcheck,gosec - s.db.DeleteChannelIfEmpty(ctx, ch.ID) //nolint:errcheck,gosec - } - - s.db.DeleteSession(ctx, sessionID) //nolint:errcheck,gosec + return nil } // SetAway sets or clears the away message. Returns true @@ -1362,6 +1343,26 @@ func (s *Service) broadcastNickChange( } } +// sendQuit sends the QUIT of nick, with reason, to the +// sessions peerIDs, as one message row fanned out to all of +// them. +func (s *Service) sendQuit( + ctx context.Context, + nick, reason string, + peerIDs []int64, +) { + if len(peerIDs) == 0 { + return + } + + body, _ := json.Marshal([]string{reason}) //nolint:errchkjson + + _, _, _ = s.FanOut( //nolint:dogsled // fire-and-forget broadcast + ctx, irc.CmdQuit, nick, "", 0, 0, + nil, body, nil, peerIDs, + ) +} + // joinError returns the IRC error for a JOIN of channel // that db.CreateOrJoinChannel refused with err, or err // wrapped if it is not a refusal. diff --git a/internal/service/service_test.go b/internal/service/service_test.go index 6170f54..54c52d5 100644 --- a/internal/service/service_test.go +++ b/internal/service/service_test.go @@ -495,6 +495,78 @@ func TestJoinConcurrentWithLastPart(t *testing.T) { } } +// TestJoinConcurrentWithLastQuit checks that a user who +// sends JOIN without the key to a +k channel, again after +// each ERR_BADCHANNELKEY, while the channel's only member +// quits, gets in only once the channel is deleted, into a +// new channel of that name with no key. +func TestJoinConcurrentWithLastQuit(t *testing.T) { + env := newTestEnv(t) + ctx := t.Context() + + for trial := range concurrentJoinTrials { + channel := fmt.Sprintf("#keyed%d", trial) + leaverNick := fmt.Sprintf("leaver%d", trial) + leaverSID := createSession(ctx, t, env.db, leaverNick) + joinerNick := fmt.Sprintf("joiner%d", trial) + joinerSID := createSession(ctx, t, env.db, joinerNick) + + created, err := env.svc.JoinChannel( + ctx, leaverSID, leaverNick, channel, "", + ) + if err != nil { + t.Fatalf("join %s: %v", leaverNick, err) + } + + err = env.db.SetChannelKey(ctx, created.ChannelID, "secret") + if err != nil { + t.Fatalf("set +k: %v", err) + } + + var joined *service.JoinResult + + errs := runConcurrently( + func() error { + env.svc.BroadcastQuit(ctx, leaverSID, leaverNick, "") + + return nil + }, + // The database has one connection, so retried + // JOINs take turns with the QUIT's statements + // and land between any two that are not in one + // transaction. A JOIN after the QUIT gets in, so + // the loop ends. + func() error { + for { + var ( + joinErr error + ircErr *service.IRCError + ) + + joined, joinErr = env.svc.JoinChannel( + ctx, joinerSID, joinerNick, channel, "", + ) + if !errors.As(joinErr, &ircErr) || + ircErr.Code != irc.ErrBadChannelKey { + return joinErr + } + } + }, + ) + if errs[1] != nil { + t.Fatalf("trial %d: JOIN: %v", trial, errs[1]) + } + + key, _ := env.db.GetChannelKey(ctx, joined.ChannelID) + if key != "" { + t.Errorf( + "trial %d: %s got into +k %s without the key", + trial, joinerNick, channel, + ) + } + } +} + // TestJoinUsesUpInvite checks that the JOIN an invite let // into a +i channel uses the invite up, so that a JOIN // after leaving gets ERR_INVITEONLYCHAN.