Broadcast a session's QUIT once when its clients leave together (closes #154)

Logout deleted the client, counted the session's clients and ran
BroadcastQuit as separate statements, and BroadcastQuit read the peers,
parted, deleted empty channels and deleted the session separately too.
Two logouts or QUITs of one session at once both sent its QUIT, and a
JOIN landing between a part and the channel's deletion got into a +k
channel without the key.

db.QuitSession now does all of that in one transaction, parting each
channel as PartChannelAndDeleteIfEmpty does, and stores and queues the
QUIT in it only when it deleted the session, so the QUIT is queued
before the nick is free. db.LogoutClient deletes the client and counts
the rest in the same transaction.

Model: opus-5-5

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

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(

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)
 	}
 }
 

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,

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,
@@ -325,6 +335,7 @@ func newTestServerFx(
 	cfg *config.Config,
 	mware *middleware.Middleware,
 	hdlr *handlers.Handlers,
+	shutdowner fx.Shutdowner,
 ) (*server.Server, error) {
 	srv, err := server.New(lifecycle, server.Params{ //nolint:exhaustruct
 		Logger:     log,
@@ -332,6 +343,7 @@ func newTestServerFx(
 		Config:     cfg,
 		Middleware: mware,
 		Handlers:   hdlr,
+		Shutdowner: shutdowner,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("test server: %w", err)
@@ -1393,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/server/export_test.go

@@ -1,6 +1,10 @@
 package server
 
-import "github.com/go-chi/chi/v5"
+import (
+	"net"
+
+	"github.com/go-chi/chi/v5"
+)
 
 // NewTestSPAServer creates a Server whose router serves only
 // the embedded web client.
@@ -10,3 +14,9 @@ func NewTestSPAServer() *Server {
 
 	return srv
 }
+
+// ServeUntilShutdown runs serveUntilShutdown, which serves the
+// HTTP server on listener, and returns when it does.
+func (srv *Server) ServeUntilShutdown(listener net.Listener) {
+	srv.serveUntilShutdown(listener)
+}

internal/server/server.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
@@ -39,6 +40,7 @@ type Params struct {
 	Config     *config.Config
 	Middleware *middleware.Middleware
 	Handlers   *handlers.Handlers
+	Shutdowner fx.Shutdowner
 }
 
 // Server is the main HTTP server.
@@ -66,7 +68,7 @@ func New(
 	}
 
 	lifecycle.Append(fx.Hook{
-		OnStart: func(_ context.Context) error {
+		OnStart: func(ctx context.Context) error {
 			srv.startupTime = time.Now()
 
 			// Build the router before OnStart returns, so srv can
@@ -78,9 +80,7 @@ func New(
 			// A closed stdout or stderr must not end the process.
 			signal.Ignore(syscall.SIGPIPE)
 
-			srv.startHTTP()
-
-			return nil
+			return srv.startHTTP(ctx)
 		},
 		OnStop: func(ctx context.Context) error {
 			srv.cleanShutdown(ctx)
@@ -162,15 +162,23 @@ func (srv *Server) configure() {
 	// Server configuration placeholder.
 }
 
-// startHTTP creates the HTTP server and serves it from a new
-// goroutine. srv.httpServer is set before that goroutine
-// starts and never changes, so it and cleanShutdown read it
-// without a lock.
-func (srv *Server) startHTTP() {
+// startHTTP binds the HTTP listen address, so that a port
+// that is taken fails startup, then serves the HTTP server
+// from a new goroutine. srv.httpServer is set before that
+// goroutine starts and never changes, so it and cleanShutdown
+// read it without a lock.
+func (srv *Server) startHTTP(ctx context.Context) error {
 	listenAddr := fmt.Sprintf(
 		":%d", srv.params.Config.Port,
 	)
 
+	var listenConfig net.ListenConfig
+
+	listener, err := listenConfig.Listen(ctx, "tcp", listenAddr)
+	if err != nil {
+		return fmt.Errorf("http listen: %w", err)
+	}
+
 	srv.httpServer = &http.Server{ //nolint:exhaustruct // optional fields
 		Addr:           listenAddr,
 		ReadTimeout:    httpReadTimeout,
@@ -179,16 +187,30 @@ func (srv *Server) startHTTP() {
 		Handler:        srv,
 	}
 
-	go srv.serveUntilShutdown()
+	go func() {
+		srv.serveUntilShutdown(listener)
+	}()
+
+	return nil
 }
 
-func (srv *Server) serveUntilShutdown() {
+// serveUntilShutdown serves the HTTP server on listener until
+// cleanShutdown stops it. If serving fails before then, it
+// stops the app with exit code 1.
+func (srv *Server) serveUntilShutdown(listener net.Listener) {
 	srv.log.Info(
 		"http begin listen", "listenaddr", srv.httpServer.Addr,
 	)
 
-	err := srv.httpServer.ListenAndServe()
-	if err != nil && !errors.Is(err, http.ErrServerClosed) {
-		srv.log.Error("listen error", "error", err)
+	err := srv.httpServer.Serve(listener)
+	if errors.Is(err, http.ErrServerClosed) {
+		return
+	}
+
+	srv.log.Error("listen error", "error", err)
+
+	err = srv.params.Shutdowner.Shutdown(fx.ExitCode(1))
+	if err != nil {
+		srv.log.Error("shutdown failed", "error", err)
 	}
 }

internal/server/server_test.go

@@ -1,3 +1,7 @@
+// Tests use a global viper instance for configuration,
+// making parallel execution unsafe.
+//
+//nolint:paralleltest
 package server_test
 
 import (
@@ -6,6 +10,7 @@ import (
 	"net"
 	"net/http"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 
@@ -40,8 +45,6 @@ const (
 // for a long poll the HTTP server is serving to finish, and
 // that the poll gets its response.
 func TestStopWaitsForRequestInProgress(t *testing.T) {
-	t.Parallel()
-
 	port := freePort(t)
 	addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
 
@@ -103,12 +106,114 @@ func TestStopWaitsForRequestInProgress(t *testing.T) {
 	}
 }
 
-// startApp starts the server with the dependencies
-// cmd/neoircd gives it, on a fresh in-memory database, with
-// the HTTP server listening on port.
+// TestStartFailsWhenHTTPPortIsTaken checks that starting the
+// app, which runs the start hook server.New registers,
+// returns an error naming the HTTP listen address when
+// something else is already listening on its port.
+func TestStartFailsWhenHTTPPortIsTaken(t *testing.T) {
+	var listenConfig net.ListenConfig
+
+	taken, err := listenConfig.Listen(t.Context(), "tcp", ":0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+
+	t.Cleanup(func() { _ = taken.Close() })
+
+	tcpAddr, ok := taken.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("listener address %v is not TCP", taken.Addr())
+	}
+
+	app, _, _ := newApp(t, tcpAddr.Port)
+
+	err = app.Start(t.Context())
+	if err == nil {
+		app.RequireStop()
+		t.Fatal("the app started while its HTTP port was taken")
+	}
+
+	listenAddr := fmt.Sprintf(":%d", tcpAddr.Port)
+	if !strings.Contains(err.Error(), listenAddr) {
+		t.Errorf("start error %q does not name %s", err, listenAddr)
+	}
+}
+
+// TestServeFailureStopsApp checks that when serving HTTP fails
+// after the start hook has bound the listen address, the
+// server stops the app with exit code 1.
+func TestServeFailureStopsApp(t *testing.T) {
+	var srv *server.Server
+
+	app, _, _ := startApp(t, freePort(t), fx.Populate(&srv))
+
+	shutdowns := app.Wait()
+
+	srv.ServeUntilShutdown(closedListener(t))
+
+	select {
+	case shutdown := <-shutdowns:
+		if shutdown.ExitCode != 1 {
+			t.Errorf("exit code %d, want 1", shutdown.ExitCode)
+		}
+	default:
+		t.Error("the server did not stop the app")
+	}
+}
+
+// TestServeAfterStopDoesNotStopApp checks that serving HTTP
+// after the app has stopped, which ends at once because the
+// stop hook has shut the HTTP server down, does not stop the
+// app. Serving ends the same way on every normal stop.
+func TestServeAfterStopDoesNotStopApp(t *testing.T) {
+	var srv *server.Server
+
+	app, _, _ := startApp(t, freePort(t), fx.Populate(&srv))
+
+	shutdowns := app.Wait()
+
+	app.RequireStop()
+
+	srv.ServeUntilShutdown(closedListener(t))
+
+	select {
+	case shutdown := <-shutdowns:
+		t.Errorf(
+			"the server stopped the app with exit code %d",
+			shutdown.ExitCode,
+		)
+	default:
+	}
+}
+
+// startApp starts the app newApp builds, with the HTTP server
+// listening on port.
 func startApp(
 	t *testing.T,
 	port int,
+	options ...fx.Option,
+) (*fxtest.App, *broker.Broker, *db.Database) {
+	t.Helper()
+
+	app, brk, database := newApp(t, port, options...)
+
+	app.RequireStart()
+
+	// Stops the app if the test ends before stopping it.
+	t.Cleanup(app.RequireStop)
+
+	return app, brk, database
+}
+
+// newApp builds, without starting it, an app that runs the
+// server with the dependencies cmd/neoircd gives it, on a
+// fresh in-memory database, with the HTTP port set to port.
+// options are added to the app, for a test that needs more
+// from it than the broker and the database.
+func newApp(
+	t *testing.T,
+	port int,
+	options ...fx.Option,
 ) (*fxtest.App, *broker.Broker, *db.Database) {
 	t.Helper()
 
@@ -146,13 +251,9 @@ func startApp(
 		}),
 		fx.Invoke(func(*server.Server) {}),
 		fx.Populate(&brk, &database),
+		fx.Options(options...),
 	)
 
-	app.RequireStart()
-
-	// Stops the app if the test ends before stopping it.
-	t.Cleanup(app.RequireStop)
-
 	return app, brk, database
 }
 
@@ -161,6 +262,21 @@ func startApp(
 func freePort(t *testing.T) int {
 	t.Helper()
 
+	listener := closedListener(t)
+
+	tcpAddr, ok := listener.Addr().(*net.TCPAddr)
+	if !ok {
+		t.Fatalf("listener address %v is not TCP", listener.Addr())
+	}
+
+	return tcpAddr.Port
+}
+
+// closedListener returns a TCP listener that has already been
+// closed, so that accepting a connection on it fails.
+func closedListener(t *testing.T) net.Listener {
+	t.Helper()
+
 	var listenConfig net.ListenConfig
 
 	listener, err := listenConfig.Listen(
@@ -170,17 +286,12 @@ func freePort(t *testing.T) int {
 		t.Fatalf("listen: %v", err)
 	}
 
-	tcpAddr, ok := listener.Addr().(*net.TCPAddr)
-	if !ok {
-		t.Fatalf("listener address %v is not TCP", listener.Addr())
-	}
-
 	err = listener.Close()
 	if err != nil {
 		t.Fatalf("close listener: %v", err)
 	}
 
-	return tcpAddr.Port
+	return listener
 }
 
 // canConnect reports whether addr accepts a TCP connection.

internal/service/service.go

@@ -826,61 +826,55 @@ 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, 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(
-		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.
+// If that fails, nothing changes and Logout returns the
+// error.
+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

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.