diff --git a/README.md b/README.md index 6f983ab..36caadd 100644 --- a/README.md +++ b/README.md @@ -2569,10 +2569,13 @@ 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 + session types in the cleanup path: the session is parted from its channels, + deleted, and its QUIT queued to its channel peers in one database transaction, + so when several QUITs or logouts end one session at once only the one that + deletes it broadcasts QUIT, and whatever a new holder of the nick sends + reaches those peers after the 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 diff --git a/internal/db/queries.go b/internal/db/queries.go index 3db2e63..8b216a1 100644 --- a/internal/db/queries.go +++ b/internal/db/queries.go @@ -854,42 +854,6 @@ func (database *Database) SetChannelTopicLocked( return nil } -// PartChannel removes a session from a channel. -func (database *Database) PartChannel( - ctx context.Context, - channelID, sessionID int64, -) error { - _, err := database.conn.ExecContext(ctx, - `DELETE FROM channel_members - WHERE channel_id = ? AND session_id = ?`, - channelID, sessionID) - if err != nil { - return fmt.Errorf("part channel: %w", err) - } - - return nil -} - -// DeleteChannelIfEmpty removes a channel with no members. -func (database *Database) DeleteChannelIfEmpty( - ctx context.Context, - channelID int64, -) error { - _, err := database.conn.ExecContext(ctx, - `DELETE FROM channels WHERE id = ? - AND NOT EXISTS - (SELECT 1 FROM channel_members - WHERE channel_id = ?)`, - channelID, channelID) - if err != nil { - return fmt.Errorf( - "delete channel if empty: %w", err, - ) - } - - return nil -} - // PartChannelAndDeleteIfEmpty removes a session from a // channel and deletes the channel if that left it with no // members, in one transaction, so that no JOIN lands @@ -905,7 +869,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 +911,6 @@ func (database *Database) PartChannelAndDeleteIfEmpty( ) } - err = transaction.Commit() - if err != nil { - return fmt.Errorf("commit part: %w", err) - } - return nil } @@ -1156,6 +1137,40 @@ func (database *Database) InsertMessage( params json.RawMessage, body json.RawMessage, meta json.RawMessage, +) (int64, string, error) { + transaction, err := database.conn.BeginTx(ctx, nil) + if err != nil { + return 0, "", fmt.Errorf("begin insert message: %w", err) + } + + defer func() { _ = transaction.Rollback() }() + + dbID, msgUUID, err := insertMessageInTx( + ctx, transaction, command, from, target, + toSessionID, toChannelID, params, body, meta, + ) + if err != nil { + return 0, "", err + } + + err = transaction.Commit() + if err != nil { + return 0, "", fmt.Errorf("commit insert message: %w", err) + } + + return dbID, msgUUID, nil +} + +// insertMessageInTx is InsertMessage, writing through +// transaction. +func insertMessageInTx( + ctx context.Context, + transaction *sql.Tx, + command, from, target string, + toSessionID, toChannelID int64, + params json.RawMessage, + body json.RawMessage, + meta json.RawMessage, ) (int64, string, error) { msgUUID := uuid.New().String() now := time.Now().UTC() @@ -1172,7 +1187,7 @@ func (database *Database) InsertMessage( meta = json.RawMessage("{}") } - res, err := database.conn.ExecContext(ctx, + res, err := transaction.ExecContext(ctx, `INSERT INTO messages (uuid, command, msg_from, msg_to, to_session_id, to_channel_id, @@ -1199,7 +1214,34 @@ func (database *Database) EnqueueToSession( ctx context.Context, sessionID, messageID int64, ) error { - _, err := database.conn.ExecContext(ctx, + transaction, err := database.conn.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin enqueue: %w", err) + } + + defer func() { _ = transaction.Rollback() }() + + err = enqueueToSessionInTx(ctx, transaction, sessionID, messageID) + if err != nil { + return err + } + + err = transaction.Commit() + if err != nil { + return fmt.Errorf("commit enqueue: %w", err) + } + + return nil +} + +// enqueueToSessionInTx is EnqueueToSession, writing through +// transaction. +func enqueueToSessionInTx( + ctx context.Context, + transaction *sql.Tx, + sessionID, messageID int64, +) error { + _, err := transaction.ExecContext(ctx, `INSERT OR IGNORE INTO client_queues (client_id, message_id, created_at) SELECT c.id, ?, ? @@ -1470,38 +1512,175 @@ func (database *Database) SetTopic( return nil } -// DeleteSession removes a session and all its data. -func (database *Database) DeleteSession( +// QuitSession ends a session in one transaction: it parts +// the session from its channels, deleting each channel this +// leaves with no members, deletes the session with its +// clients, and stores a QUIT from nick with body and queues +// it to the other sessions that were on those channels. It +// returns those sessions, for the caller to wake once the +// QUIT is committed. Of several calls for one session at +// once, only the first finds the session; the others change +// nothing and return none. +func (database *Database) QuitSession( ctx context.Context, sessionID int64, -) error { - _, err := database.conn.ExecContext( - ctx, - "DELETE FROM sessions WHERE id = ?", - sessionID, - ) + nick string, + body json.RawMessage, +) ([]int64, error) { + transaction, err := database.conn.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("delete session: %w", err) + return nil, fmt.Errorf("begin quit: %w", err) } - return nil + defer func() { _ = transaction.Rollback() }() + + peerIDs, err := quitSessionInTx( + ctx, transaction, sessionID, nick, body, + ) + if err != nil { + return nil, err + } + + err = transaction.Commit() + if err != nil { + return nil, fmt.Errorf("commit quit: %w", err) + } + + return peerIDs, nil } -// DeleteClient removes a single client record by ID. -func (database *Database) DeleteClient( +// 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 none if the session keeps a client. +func (database *Database) LogoutClient( ctx context.Context, - clientID int64, -) error { - _, err := database.conn.ExecContext( - ctx, - "DELETE FROM clients WHERE id = ?", - clientID, - ) + sessionID, clientID int64, + nick string, + body json.RawMessage, +) ([]int64, error) { + transaction, err := database.conn.BeginTx(ctx, nil) if err != nil { - return fmt.Errorf("delete client: %w", err) + return nil, fmt.Errorf("begin logout: %w", err) } - return nil + defer func() { _ = transaction.Rollback() }() + + _, err = transaction.ExecContext(ctx, + "DELETE FROM clients WHERE id = ?", clientID) + if err != nil { + return nil, 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, fmt.Errorf("count clients: %w", err) + } + + var peerIDs []int64 + + if remaining == 0 { + peerIDs, err = quitSessionInTx( + ctx, transaction, sessionID, nick, body, + ) + if err != nil { + return nil, err + } + } + + err = transaction.Commit() + if err != nil { + return nil, fmt.Errorf("commit logout: %w", err) + } + + return peerIDs, nil +} + +// quitSessionInTx is QuitSession, writing through +// transaction. +func quitSessionInTx( + ctx context.Context, + transaction *sql.Tx, + sessionID int64, + nick string, + body json.RawMessage, +) ([]int64, 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, fmt.Errorf("get channel peers: %w", err) + } + + peerIDs, err := scanInt64s(rows) + if err != nil { + return nil, err + } + + rows, err = transaction.QueryContext(ctx, + `SELECT channel_id FROM channel_members + WHERE session_id = ?`, sessionID) + if err != nil { + return nil, fmt.Errorf( + "get session channel ids: %w", err, + ) + } + + channelIDs, err := scanInt64s(rows) + if err != nil { + return nil, err + } + + for _, channelID := range channelIDs { + err = partChannelAndDeleteIfEmptyInTx( + ctx, transaction, channelID, sessionID, + ) + if err != nil { + return nil, err + } + } + + result, err := transaction.ExecContext(ctx, + "DELETE FROM sessions WHERE id = ?", sessionID) + if err != nil { + return nil, fmt.Errorf("delete session: %w", err) + } + + deletedRows, err := result.RowsAffected() + if err != nil { + return nil, fmt.Errorf("delete session: %w", err) + } + + if deletedRows == 0 || len(peerIDs) == 0 { + return nil, nil + } + + messageID, _, err := insertMessageInTx( + ctx, transaction, irc.CmdQuit, nick, "", 0, 0, + nil, body, nil, + ) + if err != nil { + return nil, err + } + + for _, peerID := range peerIDs { + err = enqueueToSessionInTx( + ctx, transaction, peerID, messageID, + ) + if err != nil { + return nil, err + } + } + + return peerIDs, nil } // GetUserCount returns the number of active users. @@ -1543,29 +1722,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/db/queries_test.go b/internal/db/queries_test.go index 99fad5f..29f9928 100644 --- a/internal/db/queries_test.go +++ b/internal/db/queries_test.go @@ -2,6 +2,7 @@ package db_test import ( "encoding/json" + "slices" "testing" "sneak.berlin/go/neoirc/internal/db" @@ -411,7 +412,7 @@ func TestJoinAndPart(t *testing.T) { t.Fatal(err) } - err = database.PartChannel(ctx, chID, sid) + err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sid) if err != nil { t.Fatal(err) } @@ -422,7 +423,10 @@ func TestJoinAndPart(t *testing.T) { } } -func TestDeleteChannelIfEmpty(t *testing.T) { +// TestPartChannelAndDeleteIfEmpty checks that a part keeps +// a channel another session is still on and deletes the +// channel when its last member parts. +func TestPartChannelAndDeleteIfEmpty(t *testing.T) { t.Parallel() database := setupTestDB(t) @@ -435,22 +439,34 @@ func TestDeleteChannelIfEmpty(t *testing.T) { t.Fatal(err) } - sid, _, _, err := database.CreateSession(ctx, "temp", "", "", "") + nicks := []string{"temp1", "temp2"} + sids := make([]int64, 0, len(nicks)) + + for _, nick := range nicks { + sid, _, _, createErr := database.CreateSession(ctx, nick, "", "", "") + if createErr != nil { + t.Fatal(createErr) + } + + err = database.JoinChannel(ctx, chID, sid) + if err != nil { + t.Fatal(err) + } + + sids = append(sids, sid) + } + + err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sids[0]) if err != nil { t.Fatal(err) } - err = database.JoinChannel(ctx, chID, sid) + _, err = database.GetChannelByName(ctx, "#empty") if err != nil { - t.Fatal(err) + t.Fatalf("channel deleted while a member is on it: %v", err) } - err = database.PartChannel(ctx, chID, sid) - if err != nil { - t.Fatal(err) - } - - err = database.DeleteChannelIfEmpty(ctx, chID) + err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sids[1]) if err != nil { t.Fatal(err) } @@ -777,6 +793,11 @@ func TestGetHistoryReusedNames(t *testing.T) { t.Fatal(err) } + err = database.JoinChannel(ctx, oldChID, oldSID) + if err != nil { + t.Fatal(err) + } + body := json.RawMessage(`["hi"]`) _, _, err = database.InsertMessage( @@ -799,12 +820,9 @@ func TestGetHistoryReusedNames(t *testing.T) { t.Fatal("expected one DM and one channel message") } - err = database.DeleteSession(ctx, oldSID) - if err != nil { - t.Fatal(err) - } - - err = database.DeleteChannelIfEmpty(ctx, oldChID) + // The session is the channel's only member, so quitting + // deletes the channel too. + _, err = database.QuitSession(ctx, oldSID, "reused", nil) if err != nil { t.Fatal(err) } @@ -826,34 +844,42 @@ func TestGetHistoryReusedNames(t *testing.T) { } } -func TestDeleteSession(t *testing.T) { +// TestQuitSession checks that QuitSession deletes the +// session, keeps a channel another session is still on, +// deletes a channel it leaves with no members, and queues +// one QUIT to the other session; and that a second call for +// the same session changes nothing. +func TestQuitSession(t *testing.T) { t.Parallel() database := setupTestDB(t) ctx := t.Context() - sid, _, _, err := database.CreateSession( - ctx, "deleteme", "", "", "", + sid, sharedID, _ := createSessionWithChannels( + t, database, "deleteme", "#shared", "#alone", + ) + + peerSID, peerClientID, _, err := database.CreateSession( + ctx, "peer", "", "", "", ) if err != nil { t.Fatal(err) } - chID, err := database.GetOrCreateChannel( - ctx, "#delchan", - ) + err = database.JoinChannel(ctx, sharedID, peerSID) if err != nil { t.Fatal(err) } - err = database.JoinChannel(ctx, chID, sid) + body := json.RawMessage(`["bye"]`) + + peerIDs, err := database.QuitSession(ctx, sid, "deleteme", body) if err != nil { t.Fatal(err) } - err = database.DeleteSession(ctx, sid) - if err != nil { - t.Fatal(err) + if !slices.Equal(peerIDs, []int64{peerSID}) { + t.Fatalf("got peers %v, want [%d]", peerIDs, peerSID) } _, err = database.GetSessionByNick(ctx, "deleteme") @@ -861,9 +887,80 @@ func TestDeleteSession(t *testing.T) { t.Fatal("session should be deleted") } - ids, _ := database.GetChannelMemberIDs(ctx, chID) - if len(ids) != 0 { - t.Fatal("expected no members after deletion") + ids, _ := database.GetChannelMemberIDs(ctx, sharedID) + if !slices.Equal(ids, []int64{peerSID}) { + t.Fatalf("#shared members: got %v, want [%d]", ids, peerSID) + } + + _, err = database.GetChannelByName(ctx, "#alone") + if err == nil { + t.Fatal("expected the emptied channel to be deleted") + } + + checkOneQuitQueued(t, database, peerClientID, "deleteme") + + peerIDs, err = database.QuitSession(ctx, sid, "deleteme", body) + if err != nil || len(peerIDs) != 0 { + t.Fatalf("second quit: got %v, %v; want none", peerIDs, err) + } + + checkOneQuitQueued(t, database, peerClientID, "deleteme") +} + +// checkOneQuitQueued fails the test unless the queue of the +// client clientID holds one message, a QUIT from nick. +func checkOneQuitQueued( + t *testing.T, + database *db.Database, + clientID int64, + nick string, +) { + t.Helper() + + msgs, _, err := database.PollMessages(t.Context(), clientID, 0, 0) + if err != nil { + t.Fatal(err) + } + + if len(msgs) != 1 || msgs[0].Command != "QUIT" || msgs[0].From != nick { + t.Fatalf("queue: got %v, want one QUIT from %s", msgs, nick) + } +} + +// TestLogoutClientFailureKeepsClient checks that when the +// session of a logout's last client cannot be deleted, +// LogoutClient fails and the client stays logged in. +func TestLogoutClientFailureKeepsClient(t *testing.T) { + t.Parallel() + + database := setupTestDB(t) + ctx := t.Context() + + sid, clientID, token, err := database.CreateSession( + ctx, "stuck", "", "", "", + ) + if err != nil { + t.Fatal(err) + } + + _, err = database.GetDB().ExecContext(ctx, + `CREATE TRIGGER keep_sessions + BEFORE DELETE ON sessions + BEGIN + SELECT RAISE(ABORT, 'sessions are kept'); + END`) + if err != nil { + t.Fatal(err) + } + + _, err = database.LogoutClient(ctx, sid, clientID, "stuck", nil) + if err == nil { + t.Fatal("logout succeeded although the session was kept") + } + + _, _, nick, err := database.GetSessionByToken(ctx, token) + if err != nil || nick != "stuck" { + t.Fatalf("client gone after a failed logout: %q, %v", nick, err) } } 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..c898b2f 100644 --- a/internal/handlers/api_test.go +++ b/internal/handlers/api_test.go @@ -1393,6 +1393,267 @@ 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 := countQuits(msgs, nick) + 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) + } +} + +// TestLogoutOfOneClientKeepsSession checks that when one of +// a session's two clients logs out, the other client's +// token still works, the session stays on its channel and +// its channel peer gets no QUIT, and that the other +// client's logout then sends one QUIT. +func TestLogoutOfOneClientKeepsSession(t *testing.T) { + const ( + nick = "twoclients" + channel = "#logouts" + ) + + tserver := newTestServer(t) + firstToken := tserver.createSession(nick) + observerToken := tserver.createSession("logoutobserver") + + tserver.sendCommand(firstToken, map[string]any{ + commandKey: irc.CmdPass, bodyKey: []string{testPassword}, + }) + + secondToken := tserver.login(nick) + + for _, token := range []string{firstToken, observerToken} { + tserver.sendCommand(token, map[string]any{ + commandKey: joinCmd, toKey: channel, + }) + } + + _, lastID := tserver.pollMessages(observerToken, 0) + + tserver.logout(firstToken) + + status, _ := tserver.getState(firstToken) + if status != http.StatusUnauthorized { + t.Errorf("logged-out client's state: got %d, want 401", status) + } + + status, state := tserver.getState(secondToken) + if status != http.StatusOK { + t.Fatalf("other client's state: got %d, want 200", status) + } + + channels, _ := state["channels"].([]any) + names := make([]string, 0, len(channels)) + + for _, entry := range channels { + info, _ := entry.(map[string]any) + name, _ := info["name"].(string) + names = append(names, name) + } + + if !slices.Equal(names, []string{channel}) { + t.Errorf("session's channels: got %v, want [%s]", names, channel) + } + + msgs, lastID := tserver.pollMessages(observerToken, lastID) + + quits := countQuits(msgs, nick) + if quits != 0 { + t.Errorf("after the first logout: %d QUITs, want 0: %v", quits, msgs) + } + + tserver.logout(secondToken) + + msgs, _ = tserver.pollMessages(observerToken, lastID) + + quits = countQuits(msgs, nick) + if quits != 1 { + t.Errorf("after the second logout: %d QUITs, want 1: %v", quits, msgs) + } +} + +// countQuits returns how many of msgs are a QUIT from nick. +func countQuits(msgs []map[string]any, nick string) int { + quits := 0 + + for _, msg := range msgs { + if msg[commandKey] == irc.CmdQuit && msg["from"] == nick { + quits++ + } + } + + return quits +} + +// logout logs out the client token and fails the test +// unless the server answers 200. +func (tserver *testServer) logout(token string) { + tserver.t.Helper() + + resp, err := doRequestAuth( + tserver.t, http.MethodPost, + tserver.url("/api/v1/logout"), token, nil, + ) + if err != nil { + tserver.t.Fatalf("logout: %v", err) + } + + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + tserver.t.Fatalf("logout: status %d", resp.StatusCode) + } +} + +// 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..372eef5 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -826,61 +826,52 @@ 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 queues a QUIT to the users who shared a +// channel with it, in one transaction, so that the QUIT is +// queued before anyone else can take the nick. 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, - ) + body, _ := json.Marshal([]string{reason}) //nolint:errchkjson + + peerIDs, err := s.db.QuitSession(ctx, sessionID, nick, body) if err != nil { + s.log.Error("quit failed", "error", err) + return } - notified := make(map[int64]bool) + for _, peerID := range peerIDs { + s.broker.Notify(peerID) + } +} - for _, ch := range channels { - memberIDs, memErr := s.db.GetChannelMemberIDs( - ctx, ch.ID, - ) - if memErr != nil { - continue - } +// 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 { + body, _ := json.Marshal([]string{"Connection closed"}) //nolint:errchkjson - for _, mid := range memberIDs { - if mid == sessionID || notified[mid] { - continue - } - - notified[mid] = true - } + peerIDs, err := s.db.LogoutClient( + ctx, sessionID, clientID, nick, body, + ) + 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, - ) + for _, peerID := range peerIDs { + s.broker.Notify(peerID) } - 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 diff --git a/internal/service/service_test.go b/internal/service/service_test.go index 6170f54..f3e3d0b 100644 --- a/internal/service/service_test.go +++ b/internal/service/service_test.go @@ -495,6 +495,208 @@ 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, + ) + } + } +} + +// quitCrowdSize is how many sessions share a second channel +// with the quitting session in +// TestQuitQueuedAheadOfNewNickHoldersJoin. +const quitCrowdSize = 10 + +// TestQuitQueuedAheadOfNewNickHoldersJoin checks that when +// a new session takes a quitting session's nick and joins a +// channel the quitting session shared with a watcher, while +// the QUIT is going out, the watcher has the QUIT queued +// ahead of that JOIN. The quitting session is also on a +// channel with quitCrowdSize other sessions, made before the +// watcher's channel, so that its QUIT reaches the watcher +// last. +func TestQuitQueuedAheadOfNewNickHoldersJoin(t *testing.T) { + env := newTestEnv(t) + ctx := t.Context() + + for trial := range concurrentJoinTrials { + nick := fmt.Sprintf("reused%d", trial) + crowd := fmt.Sprintf("#crowd%d", trial) + shared := fmt.Sprintf("#shared%d", trial) + quitterSID := createSession(ctx, t, env.db, nick) + + for _, channel := range []string{crowd, shared} { + _, err := env.svc.JoinChannel(ctx, quitterSID, nick, channel, "") + if err != nil { + t.Fatalf("join %s: %v", channel, err) + } + } + + for member := range quitCrowdSize { + memberNick := fmt.Sprintf("crowd%dx%d", trial, member) + memberSID := createSession(ctx, t, env.db, memberNick) + + _, err := env.svc.JoinChannel(ctx, memberSID, memberNick, crowd, "") + if err != nil { + t.Fatalf("join %s: %v", memberNick, err) + } + } + + watcherNick := fmt.Sprintf("watcher%d", trial) + + watcherSID, watcherClientID, _, err := env.db.CreateSession( + ctx, watcherNick, watcherNick, "localhost", "127.0.0.1", + ) + if err != nil { + t.Fatalf("create session %s: %v", watcherNick, err) + } + + _, err = env.svc.JoinChannel(ctx, watcherSID, watcherNick, shared, "") + if err != nil { + t.Fatalf("join %s: %v", watcherNick, err) + } + + _, lastQueueID, err := env.db.PollMessages(ctx, watcherClientID, 0, 0) + if err != nil { + t.Fatalf("poll: %v", err) + } + + errs := runConcurrently( + func() error { + env.svc.BroadcastQuit(ctx, quitterSID, nick, "") + + return nil + }, + func() error { + return takeNickAndJoin(ctx, env, nick, shared) + }, + ) + if errs[1] != nil { + t.Fatalf("trial %d: new session: %v", trial, errs[1]) + } + + msgs, _, err := env.db.PollMessages( + ctx, watcherClientID, lastQueueID, 0, + ) + if err != nil { + t.Fatalf("poll: %v", err) + } + + commands := commandsFrom(msgs, nick) + if !slices.Equal(commands, []string{irc.CmdQuit, irc.CmdJoin}) { + t.Errorf( + "trial %d: watcher got %v from %s, want QUIT then JOIN", + trial, commands, nick, + ) + } + } +} + +// takeNickAndJoin creates a session with nick, trying again +// while another session holds the nick, and joins it to +// channel. +func takeNickAndJoin( + ctx context.Context, + env *testEnv, + nick, channel string, +) error { + for { + sessionID, _, _, err := env.db.CreateSession( + ctx, nick, nick, "localhost", "127.0.0.1", + ) + if db.IsUniqueConstraintError(err) { + continue + } + + if err != nil { + return err + } + + _, err = env.svc.JoinChannel(ctx, sessionID, nick, channel, "") + + return err + } +} + +// commandsFrom returns the commands of the messages in msgs +// that nick sent, in order. +func commandsFrom(msgs []db.IRCMessage, nick string) []string { + commands := make([]string, 0, len(msgs)) + + for _, msg := range msgs { + if msg.From == nick { + commands = append(commands, msg.Command) + } + } + + return commands +} + // 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.