Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
44143a9734 | ||
|
|
497514d3df |
@@ -2569,14 +2569,15 @@ 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: the session is parted from its channels and
|
||||
deleted in one database transaction, and when several QUITs or logouts end one
|
||||
session at once, only the one that deletes it broadcasts QUIT. Idle clients
|
||||
are automatically removed after `SESSION_IDLE_TIMEOUT` (default 30 days)
|
||||
without an authenticated HTTP API request or, on the IRC listener, a command —
|
||||
the server runs a background cleanup loop that, for a session left with no
|
||||
clients, parts the user from all channels, broadcasts QUIT, and releases the
|
||||
nick.
|
||||
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
|
||||
`POST /api/v1/logout`. A session can have multiple clients; removing one
|
||||
doesn't affect others. However, when the last client is removed (via logout),
|
||||
|
||||
+128
-95
@@ -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
|
||||
@@ -1173,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()
|
||||
@@ -1189,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,
|
||||
@@ -1216,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, ?, ?
|
||||
@@ -1487,68 +1512,56 @@ func (database *Database) SetTopic(
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteSession removes a session and all its data.
|
||||
func (database *Database) DeleteSession(
|
||||
ctx context.Context,
|
||||
sessionID int64,
|
||||
) error {
|
||||
_, err := database.conn.ExecContext(
|
||||
ctx,
|
||||
"DELETE FROM sessions WHERE id = ?",
|
||||
sessionID,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("delete session: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// QuitSession parts a session from its channels, deleting
|
||||
// each channel that this leaves with no members, and
|
||||
// deletes the session with its clients, in one
|
||||
// transaction. It returns the other sessions that were on
|
||||
// those channels, and whether this call deleted the
|
||||
// session: of several calls for one session at once, only
|
||||
// one finds it still there, and the others change nothing.
|
||||
// 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,
|
||||
) ([]int64, bool, error) {
|
||||
nick string,
|
||||
body json.RawMessage,
|
||||
) ([]int64, error) {
|
||||
transaction, err := database.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("begin quit: %w", err)
|
||||
return nil, fmt.Errorf("begin quit: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = transaction.Rollback() }()
|
||||
|
||||
peerIDs, deleted, err := quitSessionInTx(
|
||||
ctx, transaction, sessionID,
|
||||
peerIDs, err := quitSessionInTx(
|
||||
ctx, transaction, sessionID, nick, body,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
err = transaction.Commit()
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("commit quit: %w", err)
|
||||
return nil, fmt.Errorf("commit quit: %w", err)
|
||||
}
|
||||
|
||||
return peerIDs, deleted, nil
|
||||
return peerIDs, nil
|
||||
}
|
||||
|
||||
// LogoutClient deletes a client of a session and, if that
|
||||
// leaves the session with no clients, does what
|
||||
// QuitSession does, in one transaction. It returns what
|
||||
// QuitSession returns, or nothing and false if the session
|
||||
// keeps a client.
|
||||
// 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,
|
||||
sessionID, clientID int64,
|
||||
) ([]int64, bool, error) {
|
||||
nick string,
|
||||
body json.RawMessage,
|
||||
) ([]int64, error) {
|
||||
transaction, err := database.conn.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("begin logout: %w", err)
|
||||
return nil, fmt.Errorf("begin logout: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = transaction.Rollback() }()
|
||||
@@ -1556,7 +1569,7 @@ func (database *Database) LogoutClient(
|
||||
_, err = transaction.ExecContext(ctx,
|
||||
"DELETE FROM clients WHERE id = ?", clientID)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("delete client: %w", err)
|
||||
return nil, fmt.Errorf("delete client: %w", err)
|
||||
}
|
||||
|
||||
var remaining int64
|
||||
@@ -1566,29 +1579,26 @@ func (database *Database) LogoutClient(
|
||||
sessionID,
|
||||
).Scan(&remaining)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("count clients: %w", err)
|
||||
return nil, fmt.Errorf("count clients: %w", err)
|
||||
}
|
||||
|
||||
var (
|
||||
peerIDs []int64
|
||||
deleted bool
|
||||
)
|
||||
var peerIDs []int64
|
||||
|
||||
if remaining == 0 {
|
||||
peerIDs, deleted, err = quitSessionInTx(
|
||||
ctx, transaction, sessionID,
|
||||
peerIDs, err = quitSessionInTx(
|
||||
ctx, transaction, sessionID, nick, body,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
err = transaction.Commit()
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("commit logout: %w", err)
|
||||
return nil, fmt.Errorf("commit logout: %w", err)
|
||||
}
|
||||
|
||||
return peerIDs, deleted, nil
|
||||
return peerIDs, nil
|
||||
}
|
||||
|
||||
// quitSessionInTx is QuitSession, writing through
|
||||
@@ -1597,7 +1607,9 @@ func quitSessionInTx(
|
||||
ctx context.Context,
|
||||
transaction *sql.Tx,
|
||||
sessionID int64,
|
||||
) ([]int64, bool, error) {
|
||||
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
|
||||
@@ -1605,26 +1617,26 @@ func quitSessionInTx(
|
||||
WHERE session_id = ?)`,
|
||||
sessionID, sessionID)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("get channel peers: %w", err)
|
||||
return nil, fmt.Errorf("get channel peers: %w", err)
|
||||
}
|
||||
|
||||
peerIDs, err := scanInt64s(rows)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
rows, err = transaction.QueryContext(ctx,
|
||||
`SELECT channel_id FROM channel_members
|
||||
WHERE session_id = ?`, sessionID)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf(
|
||||
return nil, fmt.Errorf(
|
||||
"get session channel ids: %w", err,
|
||||
)
|
||||
}
|
||||
|
||||
channelIDs, err := scanInt64s(rows)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, channelID := range channelIDs {
|
||||
@@ -1632,22 +1644,43 @@ func quitSessionInTx(
|
||||
ctx, transaction, channelID, sessionID,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, false, err
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
result, err := transaction.ExecContext(ctx,
|
||||
"DELETE FROM sessions WHERE id = ?", sessionID)
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("delete session: %w", err)
|
||||
return nil, fmt.Errorf("delete session: %w", err)
|
||||
}
|
||||
|
||||
deletedRows, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
return nil, false, fmt.Errorf("delete session: %w", err)
|
||||
return nil, fmt.Errorf("delete session: %w", err)
|
||||
}
|
||||
|
||||
return peerIDs, deletedRows > 0, nil
|
||||
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.
|
||||
|
||||
+127
-30
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1459,14 +1459,7 @@ func checkQuitRace(
|
||||
|
||||
msgs, _ := tserver.pollMessages(observerToken, lastID)
|
||||
|
||||
quits := 0
|
||||
|
||||
for _, msg := range msgs {
|
||||
if msg[commandKey] == irc.CmdQuit && msg["from"] == nick {
|
||||
quits++
|
||||
}
|
||||
}
|
||||
|
||||
quits := countQuits(msgs, nick)
|
||||
if quits != 1 {
|
||||
t.Errorf("trial %d: %d QUITs, want 1: %v", trial, quits, msgs)
|
||||
}
|
||||
@@ -1483,6 +1476,110 @@ func checkQuitRace(
|
||||
}
|
||||
}
|
||||
|
||||
// TestLogoutOfOneClientKeepsSession checks that when one of
|
||||
// a session's two clients logs out, the other client's
|
||||
// token still works, the session stays on its channel and
|
||||
// its channel peer gets no QUIT, and that the other
|
||||
// client's logout then sends one QUIT.
|
||||
func TestLogoutOfOneClientKeepsSession(t *testing.T) {
|
||||
const (
|
||||
nick = "twoclients"
|
||||
channel = "#logouts"
|
||||
)
|
||||
|
||||
tserver := newTestServer(t)
|
||||
firstToken := tserver.createSession(nick)
|
||||
observerToken := tserver.createSession("logoutobserver")
|
||||
|
||||
tserver.sendCommand(firstToken, map[string]any{
|
||||
commandKey: irc.CmdPass, bodyKey: []string{testPassword},
|
||||
})
|
||||
|
||||
secondToken := tserver.login(nick)
|
||||
|
||||
for _, token := range []string{firstToken, observerToken} {
|
||||
tserver.sendCommand(token, map[string]any{
|
||||
commandKey: joinCmd, toKey: channel,
|
||||
})
|
||||
}
|
||||
|
||||
_, lastID := tserver.pollMessages(observerToken, 0)
|
||||
|
||||
tserver.logout(firstToken)
|
||||
|
||||
status, _ := tserver.getState(firstToken)
|
||||
if status != http.StatusUnauthorized {
|
||||
t.Errorf("logged-out client's state: got %d, want 401", status)
|
||||
}
|
||||
|
||||
status, state := tserver.getState(secondToken)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("other client's state: got %d, want 200", status)
|
||||
}
|
||||
|
||||
channels, _ := state["channels"].([]any)
|
||||
names := make([]string, 0, len(channels))
|
||||
|
||||
for _, entry := range channels {
|
||||
info, _ := entry.(map[string]any)
|
||||
name, _ := info["name"].(string)
|
||||
names = append(names, name)
|
||||
}
|
||||
|
||||
if !slices.Equal(names, []string{channel}) {
|
||||
t.Errorf("session's channels: got %v, want [%s]", names, channel)
|
||||
}
|
||||
|
||||
msgs, lastID := tserver.pollMessages(observerToken, lastID)
|
||||
|
||||
quits := countQuits(msgs, nick)
|
||||
if quits != 0 {
|
||||
t.Errorf("after the first logout: %d QUITs, want 0: %v", quits, msgs)
|
||||
}
|
||||
|
||||
tserver.logout(secondToken)
|
||||
|
||||
msgs, _ = tserver.pollMessages(observerToken, lastID)
|
||||
|
||||
quits = countQuits(msgs, nick)
|
||||
if quits != 1 {
|
||||
t.Errorf("after the second logout: %d QUITs, want 1: %v", quits, msgs)
|
||||
}
|
||||
}
|
||||
|
||||
// countQuits returns how many of msgs are a QUIT from nick.
|
||||
func countQuits(msgs []map[string]any, nick string) int {
|
||||
quits := 0
|
||||
|
||||
for _, msg := range msgs {
|
||||
if msg[commandKey] == irc.CmdQuit && msg["from"] == nick {
|
||||
quits++
|
||||
}
|
||||
}
|
||||
|
||||
return quits
|
||||
}
|
||||
|
||||
// logout logs out the client token and fails the test
|
||||
// unless the server answers 200.
|
||||
func (tserver *testServer) logout(token string) {
|
||||
tserver.t.Helper()
|
||||
|
||||
resp, err := doRequestAuth(
|
||||
tserver.t, http.MethodPost,
|
||||
tserver.url("/api/v1/logout"), token, nil,
|
||||
)
|
||||
if err != nil {
|
||||
tserver.t.Fatalf("logout: %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
tserver.t.Fatalf("logout: status %d", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
// login logs in to the session nick with testPassword and
|
||||
// returns the new client's token.
|
||||
func (tserver *testServer) login(nick string) string {
|
||||
|
||||
@@ -222,7 +222,9 @@ func (c *Conn) serve(
|
||||
) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
defer c.cleanup(ctx)
|
||||
// Stopping the server cancels ctx, and the cleanup must
|
||||
// still end the session then.
|
||||
defer c.cleanup(context.WithoutCancel(ctx))
|
||||
|
||||
ping := time.AfterFunc(interval, func() {
|
||||
c.send("PING :" + c.serverSfx)
|
||||
|
||||
@@ -85,6 +85,10 @@ const PingInterval = pingInterval
|
||||
// for a line before it closes the connection.
|
||||
const PongDeadline = pongDeadline
|
||||
|
||||
// StopTimeout is the longest Stop waits for the connections'
|
||||
// cleanup.
|
||||
const StopTimeout = stopTimeout
|
||||
|
||||
// NewTestConn returns a Conn for nick on tcpConn, without a
|
||||
// server, database or service behind it.
|
||||
func NewTestConn(
|
||||
|
||||
@@ -16,6 +16,10 @@ import (
|
||||
"sneak.berlin/go/neoirc/internal/service"
|
||||
)
|
||||
|
||||
// stopTimeout is the longest stop waits for the connections'
|
||||
// cleanup, which ends their sessions.
|
||||
const stopTimeout = 5 * time.Second
|
||||
|
||||
// Params defines the dependencies for creating an IRC
|
||||
// Server.
|
||||
type Params struct {
|
||||
@@ -40,6 +44,10 @@ type Server struct {
|
||||
conns map[*Conn]struct{}
|
||||
cancel context.CancelFunc
|
||||
|
||||
// serving counts the connections whose serve, cleanup
|
||||
// included, has not returned yet.
|
||||
serving sync.WaitGroup
|
||||
|
||||
// pingInterval and pongDeadline are the constants of the
|
||||
// same names, except in tests, which shorten them.
|
||||
pingInterval time.Duration
|
||||
@@ -63,6 +71,7 @@ func New(
|
||||
listener: nil,
|
||||
cancel: nil,
|
||||
mu: sync.Mutex{},
|
||||
serving: sync.WaitGroup{},
|
||||
|
||||
pingInterval: pingInterval,
|
||||
pongDeadline: pongDeadline,
|
||||
@@ -112,7 +121,10 @@ func (s *Server) start(ctx context.Context, addr string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// stop shuts down the listener and all connections.
|
||||
// stop shuts down the listener and all connections, then
|
||||
// waits up to stopTimeout for every connection's cleanup to
|
||||
// finish, so that their sessions are ended before the
|
||||
// database closes.
|
||||
func (s *Server) stop() {
|
||||
if s.cancel != nil {
|
||||
s.cancel()
|
||||
@@ -127,6 +139,22 @@ func (s *Server) stop() {
|
||||
c.conn.Close() //nolint:errcheck,gosec
|
||||
}
|
||||
s.mu.Unlock()
|
||||
|
||||
done := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
s.serving.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(stopTimeout):
|
||||
s.log.Error(
|
||||
"irc connections not cleaned up before stop timeout",
|
||||
"timeout", stopTimeout,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// acceptLoop accepts new connections.
|
||||
@@ -152,10 +180,24 @@ func (s *Server) acceptLoop(ctx context.Context) {
|
||||
)
|
||||
|
||||
s.mu.Lock()
|
||||
|
||||
// stop cancels ctx before it takes s.mu to close the
|
||||
// connections, so a connection accepted while stop
|
||||
// runs is either closed and waited for by stop or
|
||||
// closed here.
|
||||
if ctx.Err() != nil {
|
||||
s.mu.Unlock()
|
||||
tcpConn.Close() //nolint:errcheck,gosec
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
s.conns[client] = struct{}{}
|
||||
s.serving.Add(1)
|
||||
s.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer s.serving.Done()
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
delete(s.conns, client)
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"bufio"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
@@ -812,6 +813,120 @@ func TestQuit(t *testing.T) {
|
||||
assertContains(t, lines, "goodbye", "QUIT reason")
|
||||
}
|
||||
|
||||
// TestStopEndsSessions checks that stopping the server ends
|
||||
// a registered connection's session before Stop returns, as
|
||||
// closing the connection does: the session, its client and
|
||||
// its channel membership are gone from the database, and a
|
||||
// channel peer has been sent the QUIT.
|
||||
func TestStopEndsSessions(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := newTestEnv(t)
|
||||
client := env.dial(t)
|
||||
|
||||
client.register("shutirc")
|
||||
client.joinAndDrain("#shut")
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// A peer on the HTTP API, whose session outlives the stop.
|
||||
peerID, peerClientID, _, err := env.database.CreateSession(
|
||||
ctx, "peer", "peer", "", "",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("create peer session: %v", err)
|
||||
}
|
||||
|
||||
channelID, err := env.database.GetChannelByName(ctx, "#shut")
|
||||
if err != nil {
|
||||
t.Fatalf("get channel: %v", err)
|
||||
}
|
||||
|
||||
err = env.database.JoinChannel(ctx, channelID, peerID)
|
||||
if err != nil {
|
||||
t.Fatalf("join peer: %v", err)
|
||||
}
|
||||
|
||||
env.srv.Stop()
|
||||
|
||||
_, err = env.database.GetSessionByNick(ctx, "shutirc")
|
||||
if !errors.Is(err, sql.ErrNoRows) {
|
||||
t.Errorf("look up shutirc after stop: %v, want no rows", err)
|
||||
}
|
||||
|
||||
clients, err := env.database.GetClientCount(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("count clients: %v", err)
|
||||
}
|
||||
|
||||
if clients != 1 {
|
||||
t.Errorf("%d clients after stop, want only the peer's", clients)
|
||||
}
|
||||
|
||||
members, err := env.database.GetChannelMemberIDs(ctx, channelID)
|
||||
if err != nil {
|
||||
t.Fatalf("get members: %v", err)
|
||||
}
|
||||
|
||||
if len(members) != 1 || members[0] != peerID {
|
||||
t.Errorf(
|
||||
"members of #shut after stop: %v, want only the peer %d",
|
||||
members, peerID,
|
||||
)
|
||||
}
|
||||
|
||||
msgs, _, err := env.database.PollMessages(ctx, peerClientID, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatalf("poll peer: %v", err)
|
||||
}
|
||||
|
||||
for _, msg := range msgs {
|
||||
if msg.Command == "QUIT" && msg.From == "shutirc" {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
t.Errorf("peer was not sent the QUIT of shutirc: %v", msgs)
|
||||
}
|
||||
|
||||
// TestStopReturnsWhileCleanupIsStuck checks that Stop stops
|
||||
// waiting for a connection's cleanup after StopTimeout, so a
|
||||
// cleanup that cannot finish does not hold up stopping.
|
||||
func TestStopReturnsWhileCleanupIsStuck(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
env := newTestEnv(t)
|
||||
client := env.dial(t)
|
||||
|
||||
client.register("stuck")
|
||||
|
||||
// The test database has one connection. While the test
|
||||
// holds it, the cleanup's first query waits for it.
|
||||
held, err := env.database.GetDB().Conn(t.Context())
|
||||
if err != nil {
|
||||
t.Fatalf("take the database connection: %v", err)
|
||||
}
|
||||
|
||||
stopped := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
env.srv.Stop()
|
||||
close(stopped)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
case <-time.After(ircserver.StopTimeout + testTimeout):
|
||||
t.Error("Stop did not return while a cleanup was stuck")
|
||||
}
|
||||
|
||||
// Lets the cleanup finish before the database is closed.
|
||||
err = held.Close()
|
||||
if err != nil {
|
||||
t.Errorf("release the database connection: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTopicSetAndGet(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
+21
-46
@@ -45,11 +45,8 @@ type Params struct {
|
||||
// It manages routing, middleware, and lifecycle.
|
||||
type Server struct {
|
||||
startupTime time.Time
|
||||
exitCode int
|
||||
sentryEnabled bool
|
||||
log *slog.Logger
|
||||
ctx context.Context //nolint:containedctx // signal handling pattern
|
||||
cancelFunc context.CancelFunc
|
||||
httpServer *http.Server
|
||||
router *chi.Mux
|
||||
params Params
|
||||
@@ -78,13 +75,16 @@ func New(
|
||||
srv.enableSentry()
|
||||
srv.SetupRoutes()
|
||||
|
||||
// The start hook's context ends when the hook
|
||||
// returns; serving must outlive it.
|
||||
go srv.serve() //nolint:contextcheck,gosec // G118
|
||||
// A closed stdout or stderr must not end the process.
|
||||
signal.Ignore(syscall.SIGPIPE)
|
||||
|
||||
srv.startHTTP()
|
||||
|
||||
return nil
|
||||
},
|
||||
OnStop: func(_ context.Context) error {
|
||||
OnStop: func(ctx context.Context) error {
|
||||
srv.cleanShutdown(ctx)
|
||||
|
||||
return nil
|
||||
},
|
||||
})
|
||||
@@ -129,44 +129,15 @@ func (srv *Server) enableSentry() {
|
||||
srv.sentryEnabled = true
|
||||
}
|
||||
|
||||
func (srv *Server) serve() int {
|
||||
srv.ctx, srv.cancelFunc = context.WithCancel(
|
||||
context.Background(),
|
||||
)
|
||||
|
||||
go func() {
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
|
||||
signal.Ignore(syscall.SIGPIPE)
|
||||
signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
|
||||
|
||||
sig := <-sigCh
|
||||
|
||||
srv.log.Info("signal received", "signal", sig)
|
||||
|
||||
if srv.cancelFunc != nil {
|
||||
srv.cancelFunc()
|
||||
}
|
||||
}()
|
||||
|
||||
go srv.serveUntilShutdown()
|
||||
|
||||
<-srv.ctx.Done()
|
||||
|
||||
srv.cleanShutdown()
|
||||
|
||||
return srv.exitCode
|
||||
}
|
||||
|
||||
func (srv *Server) cleanupForExit() {
|
||||
srv.log.Info("cleaning up")
|
||||
}
|
||||
|
||||
func (srv *Server) cleanShutdown() {
|
||||
srv.exitCode = 0
|
||||
|
||||
// cleanShutdown stops the HTTP server, waiting up to
|
||||
// shutdownTimeout for the requests in progress to finish.
|
||||
func (srv *Server) cleanShutdown(ctx context.Context) {
|
||||
ctxShutdown, shutdownCancel := context.WithTimeout(
|
||||
context.Background(), shutdownTimeout,
|
||||
ctx, shutdownTimeout,
|
||||
)
|
||||
|
||||
err := srv.httpServer.Shutdown(ctxShutdown)
|
||||
@@ -191,7 +162,11 @@ func (srv *Server) configure() {
|
||||
// Server configuration placeholder.
|
||||
}
|
||||
|
||||
func (srv *Server) serveUntilShutdown() {
|
||||
// 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() {
|
||||
listenAddr := fmt.Sprintf(
|
||||
":%d", srv.params.Config.Port,
|
||||
)
|
||||
@@ -204,16 +179,16 @@ func (srv *Server) serveUntilShutdown() {
|
||||
Handler: srv,
|
||||
}
|
||||
|
||||
go srv.serveUntilShutdown()
|
||||
}
|
||||
|
||||
func (srv *Server) serveUntilShutdown() {
|
||||
srv.log.Info(
|
||||
"http begin listen", "listenaddr", listenAddr,
|
||||
"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)
|
||||
|
||||
if srv.cancelFunc != nil {
|
||||
srv.cancelFunc()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,247 @@
|
||||
package server_test
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/fx"
|
||||
"go.uber.org/fx/fxtest"
|
||||
"sneak.berlin/go/neoirc/internal/broker"
|
||||
"sneak.berlin/go/neoirc/internal/config"
|
||||
"sneak.berlin/go/neoirc/internal/db"
|
||||
"sneak.berlin/go/neoirc/internal/globals"
|
||||
"sneak.berlin/go/neoirc/internal/handlers"
|
||||
"sneak.berlin/go/neoirc/internal/healthcheck"
|
||||
"sneak.berlin/go/neoirc/internal/logger"
|
||||
"sneak.berlin/go/neoirc/internal/middleware"
|
||||
"sneak.berlin/go/neoirc/internal/server"
|
||||
"sneak.berlin/go/neoirc/internal/service"
|
||||
"sneak.berlin/go/neoirc/internal/stats"
|
||||
)
|
||||
|
||||
const (
|
||||
testTimeout = 5 * time.Second
|
||||
|
||||
// checkStep is how often waitUntil checks again.
|
||||
checkStep = 10 * time.Millisecond
|
||||
|
||||
// stopHoldTime is how long a test waits to see that the
|
||||
// app has not stopped.
|
||||
stopHoldTime = 200 * time.Millisecond
|
||||
)
|
||||
|
||||
// TestStopWaitsForRequestInProgress checks that stopping the
|
||||
// app, which runs the stop hook server.New registers, waits
|
||||
// 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))
|
||||
|
||||
app, brk, database := startApp(t, port)
|
||||
|
||||
sessionID, _, token, err := database.CreateSession(
|
||||
t.Context(), "poller", "poller", "", "",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("create session: %v", err)
|
||||
}
|
||||
|
||||
waitUntil(t, "the HTTP server to listen", func() bool {
|
||||
return canConnect(t, addr)
|
||||
})
|
||||
|
||||
statuses := make(chan int, 1)
|
||||
|
||||
go func() {
|
||||
statuses <- getAuthenticated(
|
||||
t, "http://"+addr+"/api/v1/messages?timeout=30", token,
|
||||
)
|
||||
}()
|
||||
|
||||
waitUntil(t, "the long poll to wait for a message", func() bool {
|
||||
return brk.ListenerCount(sessionID) == 1
|
||||
})
|
||||
|
||||
var stopErr error
|
||||
|
||||
stopped := make(chan struct{})
|
||||
|
||||
go func() {
|
||||
stopErr = app.Stop(t.Context())
|
||||
|
||||
close(stopped)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
t.Error("the app stopped while a request was in progress")
|
||||
case <-time.After(stopHoldTime):
|
||||
}
|
||||
|
||||
brk.Notify(sessionID)
|
||||
|
||||
status := <-statuses
|
||||
if status != http.StatusOK {
|
||||
t.Errorf("status %d, want %d", status, http.StatusOK)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-stopped:
|
||||
if stopErr != nil {
|
||||
t.Errorf("stop: %v", stopErr)
|
||||
}
|
||||
case <-time.After(testTimeout):
|
||||
t.Fatal("the app did not stop after the request finished")
|
||||
}
|
||||
}
|
||||
|
||||
// startApp starts the server with the dependencies
|
||||
// cmd/neoircd gives it, on a fresh in-memory database, with
|
||||
// the HTTP server listening on port.
|
||||
func startApp(
|
||||
t *testing.T,
|
||||
port int,
|
||||
) (*fxtest.App, *broker.Broker, *db.Database) {
|
||||
t.Helper()
|
||||
|
||||
var (
|
||||
brk *broker.Broker
|
||||
database *db.Database
|
||||
)
|
||||
|
||||
app := fxtest.New(t,
|
||||
fx.Supply(&globals.Globals{
|
||||
Appname: "neoirc-test",
|
||||
Version: "test",
|
||||
StartTime: time.Now(),
|
||||
}),
|
||||
fx.Provide(
|
||||
broker.New,
|
||||
config.New,
|
||||
db.New,
|
||||
handlers.New,
|
||||
healthcheck.New,
|
||||
logger.New,
|
||||
middleware.New,
|
||||
server.New,
|
||||
service.New,
|
||||
stats.New,
|
||||
),
|
||||
fx.Decorate(func(cfg *config.Config) *config.Config {
|
||||
// A random name, so that no other test can share it.
|
||||
cfg.DBURL = fmt.Sprintf(
|
||||
"file:%s?mode=memory&cache=shared", rand.Text(),
|
||||
)
|
||||
cfg.Port = port
|
||||
|
||||
return cfg
|
||||
}),
|
||||
fx.Invoke(func(*server.Server) {}),
|
||||
fx.Populate(&brk, &database),
|
||||
)
|
||||
|
||||
app.RequireStart()
|
||||
|
||||
// Stops the app if the test ends before stopping it.
|
||||
t.Cleanup(app.RequireStop)
|
||||
|
||||
return app, brk, database
|
||||
}
|
||||
|
||||
// freePort returns a TCP port that nothing was listening on a
|
||||
// moment ago.
|
||||
func freePort(t *testing.T) int {
|
||||
t.Helper()
|
||||
|
||||
var listenConfig net.ListenConfig
|
||||
|
||||
listener, err := listenConfig.Listen(
|
||||
t.Context(), "tcp", "127.0.0.1:0",
|
||||
)
|
||||
if err != nil {
|
||||
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
|
||||
}
|
||||
|
||||
// canConnect reports whether addr accepts a TCP connection.
|
||||
func canConnect(t *testing.T, addr string) bool {
|
||||
t.Helper()
|
||||
|
||||
var dialer net.Dialer
|
||||
|
||||
conn, err := dialer.DialContext(t.Context(), "tcp", addr)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
_ = conn.Close()
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// waitUntil calls done every checkStep until it returns true,
|
||||
// and fails the test if it does not within testTimeout.
|
||||
func waitUntil(t *testing.T, what string, done func() bool) {
|
||||
t.Helper()
|
||||
|
||||
deadline := time.Now().Add(testTimeout)
|
||||
|
||||
for !done() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("timed out waiting for %s", what)
|
||||
}
|
||||
|
||||
time.Sleep(checkStep)
|
||||
}
|
||||
}
|
||||
|
||||
// getAuthenticated sends a GET request to url as the session
|
||||
// whose auth token is token, and returns the response status,
|
||||
// or 0 if there was no response.
|
||||
func getAuthenticated(t *testing.T, url, token string) int {
|
||||
t.Helper()
|
||||
|
||||
request, err := http.NewRequestWithContext(
|
||||
t.Context(), http.MethodGet, url, nil,
|
||||
)
|
||||
if err != nil {
|
||||
t.Errorf("new request: %v", err)
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
request.Header.Set("Cookie", "neoirc_auth="+token)
|
||||
|
||||
client := http.Client{Timeout: testTimeout}
|
||||
|
||||
response, err := client.Do(request)
|
||||
if err != nil {
|
||||
t.Errorf("get %s: %v", url, err)
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
_ = response.Body.Close()
|
||||
|
||||
return response.StatusCode
|
||||
}
|
||||
+20
-30
@@ -827,20 +827,28 @@ func (s *Service) ChangeNick(
|
||||
}
|
||||
|
||||
// BroadcastQuit parts the session from all its channels,
|
||||
// deletes it, and sends a QUIT to the users who shared a
|
||||
// channel with it. Of several calls for one session at
|
||||
// once, only the one that deletes it sends the QUIT.
|
||||
// deletes it, and queues a QUIT to the users who shared a
|
||||
// channel with it, in one transaction, so that the QUIT is
|
||||
// queued before anyone else can take the nick. Of several
|
||||
// calls for one session at once, only the one that deletes
|
||||
// it sends the QUIT.
|
||||
func (s *Service) BroadcastQuit(
|
||||
ctx context.Context,
|
||||
sessionID int64,
|
||||
nick, reason string,
|
||||
) {
|
||||
peerIDs, deleted, err := s.db.QuitSession(ctx, sessionID)
|
||||
if err != nil || !deleted {
|
||||
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
|
||||
}
|
||||
|
||||
s.sendQuit(ctx, nick, reason, peerIDs)
|
||||
for _, peerID := range peerIDs {
|
||||
s.broker.Notify(peerID)
|
||||
}
|
||||
}
|
||||
|
||||
// Logout deletes a client of the session and, if that was
|
||||
@@ -850,15 +858,17 @@ func (s *Service) Logout(
|
||||
sessionID, clientID int64,
|
||||
nick string,
|
||||
) error {
|
||||
peerIDs, deleted, err := s.db.LogoutClient(
|
||||
ctx, sessionID, clientID,
|
||||
body, _ := json.Marshal([]string{"Connection closed"}) //nolint:errchkjson
|
||||
|
||||
peerIDs, err := s.db.LogoutClient(
|
||||
ctx, sessionID, clientID, nick, body,
|
||||
)
|
||||
if err != nil {
|
||||
return fmt.Errorf("logout: %w", err)
|
||||
}
|
||||
|
||||
if deleted {
|
||||
s.sendQuit(ctx, nick, "Connection closed", peerIDs)
|
||||
for _, peerID := range peerIDs {
|
||||
s.broker.Notify(peerID)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -1343,26 +1353,6 @@ func (s *Service) broadcastNickChange(
|
||||
}
|
||||
}
|
||||
|
||||
// sendQuit sends the QUIT of nick, with reason, to the
|
||||
// sessions peerIDs, as one message row fanned out to all of
|
||||
// them.
|
||||
func (s *Service) sendQuit(
|
||||
ctx context.Context,
|
||||
nick, reason string,
|
||||
peerIDs []int64,
|
||||
) {
|
||||
if len(peerIDs) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
body, _ := json.Marshal([]string{reason}) //nolint:errchkjson
|
||||
|
||||
_, _, _ = s.FanOut( //nolint:dogsled // fire-and-forget broadcast
|
||||
ctx, irc.CmdQuit, nick, "", 0, 0,
|
||||
nil, body, nil, peerIDs,
|
||||
)
|
||||
}
|
||||
|
||||
// joinError returns the IRC error for a JOIN of channel
|
||||
// that db.CreateOrJoinChannel refused with err, or err
|
||||
// wrapped if it is not a refusal.
|
||||
|
||||
@@ -567,6 +567,136 @@ func TestJoinConcurrentWithLastQuit(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
|
||||
Reference in New Issue
Block a user