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
+  session types in the cleanup path: the session is parted from its channels,
-  `DeleteSession` unconditionally. Idle clients are automatically removed after
+  deleted, and its QUIT queued to its channel peers in one database transaction,
-  `SESSION_IDLE_TIMEOUT` (default 30 days) without an authenticated HTTP API
+  so when several QUITs or logouts end one session at once only the one that
-  request or, on the IRC listener, a command — the server runs a background
+  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

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.
+// QuitSession ends a session in one transaction: it parts
-func (database *Database) DeleteSession(
+// 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 {
+	nick string,
-	_, err := database.conn.ExecContext(
+	body json.RawMessage,
-		ctx,
+) ([]int64, error) {
-		"DELETE FROM sessions WHERE id = ?",
+	transaction, err := database.conn.BeginTx(ctx, nil)
-		sessionID,
-	)
 	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.
+// LogoutClient deletes a client of a session and, if that
-func (database *Database) DeleteClient(
+// 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,
+	sessionID, clientID int64,
-) error {
+	nick string,
-	_, err := database.conn.ExecContext(
+	body json.RawMessage,
-		ctx,
+) ([]int64, error) {
-		"DELETE FROM clients WHERE id = ?",
+	transaction, err := database.conn.BeginTx(ctx, nil)
-		clientID,
-	)
 	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(

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)
+	err = database.PartChannelAndDeleteIfEmpty(ctx, chID, sids[1])
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = database.DeleteChannelIfEmpty(ctx, chID)
 	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)
+	// The session is the channel's only member, so quitting
-	if err != nil {
+	// deletes the channel too.
-		t.Fatal(err)
+	_, err = database.QuitSession(ctx, oldSID, "reused", nil)
-	}
-
-	err = database.DeleteChannelIfEmpty(ctx, oldChID)
 	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(
+	sid, sharedID, _ := createSessionWithChannels(
-		ctx, "deleteme", "", "", "",
+		t, database, "deleteme", "#shared", "#alone",
+	)
+
+	peerSID, peerClientID, _, err := database.CreateSession(
+		ctx, "peer", "", "", "",
 	)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	chID, err := database.GetOrCreateChannel(
+	err = database.JoinChannel(ctx, sharedID, peerSID)
-		ctx, "#delchan",
-	)
 	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 !slices.Equal(peerIDs, []int64{peerSID}) {
-	if err != nil {
+		t.Fatalf("got peers %v, want [%d]", peerIDs, peerSID)
-		t.Fatal(err)
 	}
 
 	_, 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)
+	ids, _ := database.GetChannelMemberIDs(ctx, sharedID)
-	if len(ids) != 0 {
+	if !slices.Equal(ids, []int64{peerSID}) {
-		t.Fatal("expected no members after deletion")
+		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)
 	}
 }
 

internal/handlers/api.go

@@ -3395,15 +3395,11 @@ func (hdlr *Handlers) HandleLogout() http.HandlerFunc {
 			return
 		}
 
-		ctx := request.Context()
+		err := hdlr.svc.Logout(
-
+			request.Context(), sessionID, clientID, nick,
-		err := hdlr.params.Database.DeleteClient(
-			ctx, clientID,
 		)
 		if err != nil {
-			hdlr.log.Error(
+			hdlr.log.Error("logout failed", "error", err)
-				"delete client 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,

internal/handlers/api_test.go

@@ -126,6 +126,7 @@ const (
 	privmsgCmd     = "PRIVMSG"
 	joinCmd        = "JOIN"
 	apiMessages    = "/api/v1/messages"
+	apiLogout      = "/api/v1/logout"
 	apiSession     = "/api/v1/session"
 	apiState       = "/api/v1/state"
 	authCookieName = "neoirc_auth"
@@ -136,6 +137,8 @@ type testServer struct {
 	httpServer *httptest.Server
 	t          *testing.T
 	fxApp      *fxtest.App
+	broker     *broker.Broker
+	database   *db.Database
 }
 
 func newTestServer(
@@ -162,7 +165,11 @@ func newTestServerWith(
 	)
 	copyTemplateDB(t, dbURL)
 
-	var srv *server.Server
+	var (
+		srv      *server.Server
+		brk      *broker.Broker
+		database *db.Database
+	)
 
 	app := fxtest.New(t,
 		fx.Provide(
@@ -195,13 +202,14 @@ func newTestServerWith(
 				return cfg, nil
 			},
 			newTestDB,
+			broker.New,
 			stats.New,
 			newTestHealthcheck,
 			newTestMiddleware,
 			newTestHandlers,
 			newTestServerFx,
 		),
-		fx.Populate(&srv),
+		fx.Populate(&srv, &brk, &database),
 	)
 
 	app.RequireStart()
@@ -217,6 +225,8 @@ func newTestServerWith(
 		httpServer: httpSrv,
 		t:          t,
 		fxApp:      app,
+		broker:     brk,
+		database:   database,
 	}
 }
 
@@ -291,8 +301,8 @@ func newTestHandlers(
 	database *db.Database,
 	hcheck *healthcheck.Healthcheck,
 	tracker *stats.Tracker,
+	brk *broker.Broker,
 ) (*handlers.Handlers, error) {
-	brk := broker.New()
 	svc := service.New(service.Params{ //nolint:exhaustruct
 		Logger:   log,
 		Config:   cfg,
@@ -1395,6 +1405,486 @@ 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(apiLogout, 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)
+	}
+
+	names := channelNames(state)
+	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)
+	}
+}
+
+// TestFailedLogoutChangesNothing checks that a logout whose
+// session cannot be deleted answers 500 and changes
+// nothing: the client's token still works, the session is
+// still on its channel and its channel peer has no QUIT.
+func TestFailedLogoutChangesNothing(t *testing.T) {
+	const (
+		nick    = "stuckleaver"
+		channel = "#stuck"
+	)
+
+	tserver := newTestServer(t)
+	token := tserver.createSession(nick)
+	observerToken := tserver.createSession("stuckobserver")
+
+	for _, joiner := range []string{token, observerToken} {
+		tserver.sendCommand(joiner, map[string]any{
+			commandKey: joinCmd, toKey: channel,
+		})
+	}
+
+	_, lastID := tserver.pollMessages(observerToken, 0)
+
+	_, err := tserver.database.GetDB().ExecContext(t.Context(),
+		`CREATE TRIGGER keep_sessions
+		 BEFORE DELETE ON sessions
+		 BEGIN
+		   SELECT RAISE(ABORT, 'sessions are kept');
+		 END`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := doRequestAuth(
+		t, http.MethodPost, tserver.url(apiLogout), token, nil,
+	)
+	if err != nil {
+		t.Fatalf("logout: %v", err)
+	}
+
+	_ = resp.Body.Close()
+
+	if resp.StatusCode != http.StatusInternalServerError {
+		t.Errorf("logout: got %d, want 500", resp.StatusCode)
+	}
+
+	status, state := tserver.getState(token)
+	if status != http.StatusOK {
+		t.Fatalf("state after the failed logout: got %d, want 200", status)
+	}
+
+	names := channelNames(state)
+	if !slices.Equal(names, []string{channel}) {
+		t.Errorf("session's channels: got %v, want [%s]", names, channel)
+	}
+
+	msgs, _ := tserver.pollMessages(observerToken, lastID)
+
+	quits := countQuits(msgs, nick)
+	if quits != 0 {
+		t.Errorf("%d QUITs, want 0: %v", quits, msgs)
+	}
+}
+
+const (
+	// quitWakePollTimeout is the timeout, in seconds, of the
+	// long poll in checkQuitWakesPoll.
+	quitWakePollTimeout = 10
+
+	// quitWakeDeadline is how long checkQuitWakesPoll waits
+	// for that long poll to start waiting, and then to
+	// return once the session has left.
+	quitWakeDeadline = 5 * time.Second
+
+	// pollWaitStep is how often checkQuitWakesPoll checks
+	// whether the long poll is waiting yet.
+	pollWaitStep = 10 * time.Millisecond
+)
+
+// TestQuitWakesPeerLongPoll checks that a session's QUIT
+// wakes the long poll of a user who shares a channel with
+// it.
+func TestQuitWakesPeerLongPoll(t *testing.T) {
+	const reason = "gone fishing"
+
+	checkQuitWakesPoll(t, reason, func(tserver *testServer, token string) {
+		tserver.sendCommand(token, map[string]any{
+			commandKey: irc.CmdQuit, bodyKey: []string{reason},
+		})
+	})
+}
+
+// TestLastLogoutWakesPeerLongPoll checks that the logout of
+// a session's last client wakes the long poll of a user who
+// shares a channel with it.
+func TestLastLogoutWakesPeerLongPoll(t *testing.T) {
+	checkQuitWakesPoll(t, "Connection closed", func(tserver *testServer, token string) {
+		tserver.logout(token)
+	})
+}
+
+// checkQuitWakesPoll puts a session and a peer on a
+// channel, and calls leave with the session's token while
+// the peer long-polls with a timeout of quitWakePollTimeout
+// seconds. It checks that the poll returns within
+// quitWakeDeadline with the session's QUIT, whose body is
+// reason.
+func checkQuitWakesPoll(
+	t *testing.T,
+	reason string,
+	leave func(tserver *testServer, token string),
+) {
+	t.Helper()
+
+	const (
+		nick     = "wakequitter"
+		peerNick = "wakepeer"
+	)
+
+	tserver := newTestServer(t)
+	token := tserver.createSession(nick)
+	peerToken := tserver.createSession(peerNick)
+
+	for _, joiner := range []string{token, peerToken} {
+		tserver.sendCommand(joiner, map[string]any{
+			commandKey: joinCmd, toKey: "#wake",
+		})
+	}
+
+	_, lastID := tserver.pollMessages(peerToken, 0)
+
+	peerID, err := tserver.database.GetSessionByNick(t.Context(), peerNick)
+	if err != nil {
+		t.Fatalf("get session of %s: %v", peerNick, err)
+	}
+
+	var (
+		msgs    []map[string]any
+		pollErr error
+	)
+
+	polled := make(chan struct{})
+
+	go func() {
+		defer close(polled)
+
+		msgs, pollErr = tserver.longPoll(peerToken, lastID, quitWakePollTimeout)
+	}()
+
+	deadline := time.Now().Add(quitWakeDeadline)
+
+	for tserver.broker.ListenerCount(peerID) == 0 {
+		if time.Now().After(deadline) {
+			t.Fatal("the long poll did not start waiting")
+		}
+
+		time.Sleep(pollWaitStep)
+	}
+
+	leave(tserver, token)
+
+	select {
+	case <-polled:
+	case <-time.After(quitWakeDeadline):
+		t.Fatalf("the long poll did not return within %v", quitWakeDeadline)
+	}
+
+	if pollErr != nil {
+		t.Fatalf("long poll: %v", pollErr)
+	}
+
+	if len(msgs) != 1 || countQuits(msgs, nick) != 1 {
+		t.Fatalf("long poll: got %v, want one QUIT from %s", msgs, nick)
+	}
+
+	lines, _ := msgs[0][bodyKey].([]any)
+	if len(lines) != 1 || lines[0] != reason {
+		t.Errorf("QUIT body: got %v, want [%s]", lines, reason)
+	}
+}
+
+// channelNames returns the names of the channels in state,
+// a response to GET /api/v1/state.
+func channelNames(state map[string]any) []string {
+	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)
+	}
+
+	return names
+}
+
+// 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(apiLogout), 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)
+	}
+}
+
+// longPoll returns the messages after afterID that the
+// client token has been sent, waiting up to timeout seconds
+// for one if there are none yet. It returns errors instead
+// of failing the test, so a goroutine can call it.
+func (tserver *testServer) longPoll(
+	token string, afterID int64, timeout int,
+) ([]map[string]any, error) {
+	resp, err := doRequestAuth(
+		tserver.t, http.MethodGet,
+		tserver.url(fmt.Sprintf(
+			"%s?timeout=%d&after=%d", apiMessages, timeout, afterID,
+		)),
+		token, nil,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	defer func() { _ = resp.Body.Close() }()
+
+	var result struct {
+		Messages []map[string]any `json:"messages"`
+	}
+
+	err = json.NewDecoder(resp.Body).Decode(&result)
+	if err != nil {
+		return nil, fmt.Errorf("decode poll: %w", err)
+	}
+
+	return result.Messages, nil
+}
+
+// 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")

internal/service/service.go

@@ -826,61 +826,55 @@ func (s *Service) ChangeNick(
 	return nil
 }
 
-// BroadcastQuit broadcasts a QUIT to all channel peers,
+// BroadcastQuit parts the session from all its channels,
-// parts all channels, and deletes the session. Uses the
+// deletes it, and queues a QUIT to the users who shared a
-// FanOut pattern: one message row fanned out to all unique
+// channel with it, in one transaction, so that the QUIT is
-// peer sessions.
+// queued before anyone else can take the nick, and then
+// wakes those users' long polls. 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(
+	body, _ := json.Marshal([]string{reason}) //nolint:errchkjson
-		ctx, sessionID,
+
-	)
+	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 {
+// Logout deletes a client of the session and, if that was
-		memberIDs, memErr := s.db.GetChannelMemberIDs(
+// the session's last client, does what BroadcastQuit does.
-			ctx, ch.ID,
+// If that fails, nothing changes and Logout returns the
-		)
+// error.
-		if memErr != nil {
+func (s *Service) Logout(
-			continue
+	ctx context.Context,
-		}
+	sessionID, clientID int64,
+	nick string,
+) error {
+	body, _ := json.Marshal([]string{"Connection closed"}) //nolint:errchkjson
 
-		for _, mid := range memberIDs {
+	peerIDs, err := s.db.LogoutClient(
-			if mid == sessionID || notified[mid] {
+		ctx, sessionID, clientID, nick, body,
-				continue
+	)
-			}
+	if err != nil {
-
+		return fmt.Errorf("logout: %w", err)
-			notified[mid] = true
-		}
 	}
 
-	if len(notified) > 0 {
+	for _, peerID := range peerIDs {
-		recipients := make([]int64, 0, len(notified))
+		s.broker.Notify(peerID)
-		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 _, ch := range channels {
+	return nil
-		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
 }
 
 // SetAway sets or clears the away message. Returns true

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.