From baaf54499d7f0e804615c67c9b7f60f5970aa553 Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 10 Nov 2026 20:37:33 +0000 Subject: [PATCH] Stop the slow test packages waiting (closes #263) - Tests hash passwords at bcrypt.MinCost, set through the PasswordCost field of db.Params. TestPasswordCost checks the cost with and without it. - Test IRC servers check the queue only on broker notification; one test keeps the 100 ms fallback poll. - The stop-timeout test shortens the timeout through a test-only setter. - service and handlers tests build config.Config directly instead of through config.New, so they run in parallel. config exports DefaultTrustedProxies and ParseTrustedProxies for the handlers test config. - Repeated race trials run as parallel subtests. - Fixed sleeps become waits on the broker. - A new test covers the relay's failed queue read, which only the 100 ms poll had reached. Model: opus-5-5 --- internal/config/config.go | 46 ++-- internal/db/auth.go | 2 +- internal/db/db.go | 23 +- internal/db/db_test.go | 80 +++++++ internal/db/export_test.go | 9 +- internal/handlers/api_test.go | 290 +++++++++++++++---------- internal/handlers/export_test.go | 4 + internal/ircserver/conn.go | 7 + internal/ircserver/export_test.go | 22 +- internal/ircserver/integration_test.go | 4 - internal/ircserver/relay.go | 2 +- internal/ircserver/relay_test.go | 64 +++++- internal/ircserver/server.go | 13 +- internal/ircserver/server_test.go | 39 ++-- internal/service/service_test.go | 218 +++++++++++-------- 15 files changed, 560 insertions(+), 263 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 01c47a5..a9af4d9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -35,9 +35,9 @@ Type /help for available commands.` const DefaultDBURL = "file:///var/lib/neoirc/state.db" + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)" -// defaultTrustedProxies is TRUSTED_PROXIES when it is not set: the +// DefaultTrustedProxies is TRUSTED_PROXIES when it is not set: the // private address ranges of RFC 1918. -const defaultTrustedProxies = "10.0.0.0/8,172.16.0.0/12,192.168.0.0/16" +const DefaultTrustedProxies = "10.0.0.0/8,172.16.0.0/12,192.168.0.0/16" var errOutOfRange = errors.New("setting out of range") @@ -115,7 +115,7 @@ func New( viper.SetDefault("LOGIN_RATE_LIMIT", "1") viper.SetDefault("LOGIN_RATE_BURST", "5") viper.SetDefault("IRC_LISTEN_ADDR", ":6667") - viper.SetDefault("TRUSTED_PROXIES", defaultTrustedProxies) + viper.SetDefault("TRUSTED_PROXIES", DefaultTrustedProxies) err := viper.ReadInConfig() if err != nil { @@ -185,11 +185,9 @@ func parseSettings(cfg *Config) error { ) } -// parseTrustedProxies sets *value to the addresses in TRUSTED_PROXIES, a -// comma-separated list of networks such as 10.0.0.0/8 and single -// addresses such as 192.0.2.1. The empty string trusts nothing. Viper -// counts an empty environment variable as unset, so the environment is -// read directly as well. +// parseTrustedProxies sets *value to the addresses in TRUSTED_PROXIES. +// Viper counts an empty environment variable as unset, so the +// environment is read directly as well. func parseTrustedProxies(value **netipx.IPSet) error { setting := viper.Get("TRUSTED_PROXIES") @@ -206,18 +204,7 @@ func parseTrustedProxies(value **netipx.IPSet) error { setting, errNotString) } - var builder netipx.IPSetBuilder - - if strings.TrimSpace(raw) != "" { - for entry := range strings.SplitSeq(raw, ",") { - err := addTrustedProxy(&builder, strings.TrimSpace(entry)) - if err != nil { - return fmt.Errorf("cannot parse TRUSTED_PROXIES: %w", err) - } - } - } - - ipSet, err := builder.IPSet() + ipSet, err := ParseTrustedProxies(raw) if err != nil { return fmt.Errorf("cannot parse TRUSTED_PROXIES: %w", err) } @@ -227,6 +214,25 @@ func parseTrustedProxies(value **netipx.IPSet) error { return nil } +// ParseTrustedProxies returns the addresses in list, which has the form +// of TRUSTED_PROXIES: a comma-separated list of networks such as +// 10.0.0.0/8 and single addresses such as 192.0.2.1. The empty string +// trusts nothing. +func ParseTrustedProxies(list string) (*netipx.IPSet, error) { + var builder netipx.IPSetBuilder + + if strings.TrimSpace(list) != "" { + for entry := range strings.SplitSeq(list, ",") { + err := addTrustedProxy(&builder, strings.TrimSpace(entry)) + if err != nil { + return nil, err + } + } + } + + return builder.IPSet() +} + // addTrustedProxy adds entry, a network with a slash or a single // address, to builder. An IPv4 address written in IPv6 form, such as // ::ffff:192.0.2.1, is added as IPv4. A network in that form and an diff --git a/internal/db/auth.go b/internal/db/auth.go index 03e8999..2a92dd5 100644 --- a/internal/db/auth.go +++ b/internal/db/auth.go @@ -22,7 +22,7 @@ func (database *Database) SetPassword( password string, ) error { hash, err := bcrypt.GenerateFromPassword( - []byte(password), bcrypt.DefaultCost, + []byte(password), database.passwordCost, ) if err != nil { return fmt.Errorf("hash password: %w", err) diff --git a/internal/db/db.go b/internal/db/db.go index 922ee2e..ed70496 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -15,6 +15,7 @@ import ( "strings" "go.uber.org/fx" + "golang.org/x/crypto/bcrypt" "sneak.berlin/go/neoirc/internal/config" "sneak.berlin/go/neoirc/internal/logger" @@ -39,13 +40,19 @@ type Params struct { Logger *logger.Logger Config *config.Config + + // PasswordCost is the bcrypt cost at which SetPassword hashes + // a password. The server leaves it unset, which means + // bcrypt.DefaultCost. Tests set a lower cost to save time. + PasswordCost int `optional:"true"` } // Database manages the SQLite connection and migrations. type Database struct { - conn *sql.DB - log *slog.Logger - params *Params + conn *sql.DB + log *slog.Logger + params *Params + passwordCost int } // New creates a new Database and registers lifecycle hooks. @@ -53,9 +60,15 @@ func New( lifecycle fx.Lifecycle, params Params, ) (*Database, error) { + passwordCost := params.PasswordCost + if passwordCost == 0 { + passwordCost = bcrypt.DefaultCost + } + database := &Database{ //nolint:exhaustruct // conn set in OnStart - params: ¶ms, - log: params.Logger.Get(), + params: ¶ms, + log: params.Logger.Get(), + passwordCost: passwordCost, } database.log.Info("Database instantiated") diff --git a/internal/db/db_test.go b/internal/db/db_test.go index 6d10283..9ac327d 100644 --- a/internal/db/db_test.go +++ b/internal/db/db_test.go @@ -10,6 +10,7 @@ import ( "time" "go.uber.org/fx/fxtest" + "golang.org/x/crypto/bcrypt" "sneak.berlin/go/neoirc/internal/config" "sneak.berlin/go/neoirc/internal/db" "sneak.berlin/go/neoirc/internal/dbtest" @@ -236,6 +237,85 @@ func TestDBURLSettingForeignKeysFailsStartup(t *testing.T) { } } +// TestPasswordCost checks the bcrypt cost of the hash that SetPassword +// stores: bcrypt.DefaultCost when db.Params leaves PasswordCost unset, +// as the server does, and PasswordCost when it is set. +func TestPasswordCost(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + passwordCost int + wantCost int + }{ + {name: "unset", wantCost: bcrypt.DefaultCost}, + { + name: "bcrypt.MinCost", + passwordCost: bcrypt.MinCost, + wantCost: bcrypt.MinCost, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + lifecycle := fxtest.NewLifecycle(t) + + log, err := logger.New(lifecycle, logger.Params{ + Globals: &globals.Globals{}, + }) + if err != nil { + t.Fatal(err) + } + + database, err := db.New(lifecycle, db.Params{ + Logger: log, + Config: &config.Config{DBURL: dbtest.NewURL(t)}, + PasswordCost: test.passwordCost, + }) + if err != nil { + t.Fatal(err) + } + + lifecycle.RequireStart() + t.Cleanup(lifecycle.RequireStop) + + sessionID, _, _, err := database.CreateSession( + t.Context(), "costuser", "", "", "", + ) + if err != nil { + t.Fatal(err) + } + + err = database.SetPassword(t.Context(), sessionID, "password123") + if err != nil { + t.Fatal(err) + } + + var hash string + + err = database.GetDB().QueryRowContext( + t.Context(), + "SELECT password_hash FROM sessions WHERE id = ?", + sessionID, + ).Scan(&hash) + if err != nil { + t.Fatal(err) + } + + cost, err := bcrypt.Cost([]byte(hash)) + if err != nil { + t.Fatal(err) + } + + if cost != test.wantCost { + t.Errorf("cost = %d, want %d", cost, test.wantCost) + } + }) + } +} + // startDatabase starts a Database for dbURL the way the server does. // It returns the error that stopped it from starting, if any. func startDatabase(t *testing.T, dbURL string) (*db.Database, error) { diff --git a/internal/db/export_test.go b/internal/db/export_test.go index f24537c..5ecda2c 100644 --- a/internal/db/export_test.go +++ b/internal/db/export_test.go @@ -8,10 +8,12 @@ import ( "testing" "time" + "golang.org/x/crypto/bcrypt" "sneak.berlin/go/neoirc/internal/dbtest" ) -// NewTestDatabase creates an in-memory database for testing. +// NewTestDatabase creates an in-memory database for testing. It +// hashes passwords at bcrypt.MinCost. func NewTestDatabase(tb testing.TB) (*Database, error) { tb.Helper() @@ -23,8 +25,9 @@ func NewTestDatabase(tb testing.TB) (*Database, error) { } database := &Database{ //nolint:exhaustruct // test helper, params not needed - conn: conn, - log: slog.Default(), + conn: conn, + log: slog.Default(), + passwordCost: bcrypt.MinCost, } err = database.runMigrations(context.Background()) diff --git a/internal/handlers/api_test.go b/internal/handlers/api_test.go index 18137e2..bc9ce2a 100644 --- a/internal/handlers/api_test.go +++ b/internal/handlers/api_test.go @@ -145,34 +145,46 @@ type testServer struct { handlers *handlers.Handlers } -// viperMu makes tests start their servers one at a time, -// because config.New writes the global viper instance and -// the server reads it as it starts. -// -//nolint:gochecknoglobals // guards a global, so must be one -var viperMu sync.Mutex +const ( + // maxMessageSize is MAX_MESSAGE_SIZE on every test server, the + // setting's default. + maxMessageSize = 4096 + + // loginRateLimit is LOGIN_RATE_LIMIT on every test server, the + // setting's default. + loginRateLimit = 1 + + // loginRateBurst is LOGIN_RATE_BURST on every test server, the + // setting's default. + loginRateBurst = 5 +) func newTestServer( t *testing.T, ) *testServer { t.Helper() - return newTestServerWith(t, 0, "", "") + return newTestServerWith(t, 0, "", "", config.DefaultTrustedProxies) } func newTestServerWith( t *testing.T, hashcashBits int, - operName, operPassword string, + operName, operPassword, trustedProxies string, ) *testServer { t.Helper() - viperMu.Lock() - defer viperMu.Unlock() - dbURL := dbtest.NewURL(t) copyTemplateDB(t, dbURL) + trusted, err := config.ParseTrustedProxies(trustedProxies) + if err != nil { + t.Fatalf("parse trusted proxies: %v", err) + } + + // Each duration setting defaults to 30 days. + const thirtyDays = "720h" + var ( srv *server.Server brk *broker.Broker @@ -184,31 +196,25 @@ func newTestServerWith( fx.Provide( newTestGlobals, logger.New, - func( - lifecycle fx.Lifecycle, - globs *globals.Globals, - log *logger.Logger, - ) (*config.Config, error) { - cfg, err := config.New( - lifecycle, config.Params{ //nolint:exhaustruct - Globals: globs, Logger: log, - }, - ) - if err != nil { - return nil, fmt.Errorf( - "test config: %w", err, - ) + // Built here rather than by config.New, which writes + // the global viper instance, so servers start in + // parallel. + func() *config.Config { + return &config.Config{ + DBURL: dbURL, + // One line: each line is a database write per new session. + MOTD: "neoirc test server", + MaxMessageSize: maxMessageSize, + HashcashBits: hashcashBits, + OperName: operName, + OperPassword: operPassword, + MessageMaxAge: thirtyDays, + QueueMaxAge: thirtyDays, + SessionIdleTimeout: thirtyDays, + LoginRateLimit: loginRateLimit, + LoginRateBurst: loginRateBurst, + TrustedProxies: trusted, } - - cfg.DBURL = dbURL - cfg.Port = 0 - // One line: each line is a database write per new session. - cfg.MOTD = "neoirc test server" - cfg.HashcashBits = hashcashBits - cfg.OperName = operName - cfg.OperPassword = operPassword - - return cfg, nil }, newTestDB, broker.New, @@ -255,7 +261,7 @@ func newTestDB( cfg *config.Config, ) (*db.Database, error) { database, err := db.New(lifecycle, db.Params{ //nolint:exhaustruct - Logger: log, Config: cfg, + Logger: log, Config: cfg, PasswordCost: bcrypt.MinCost, }) if err != nil { return nil, fmt.Errorf("test db: %w", err) @@ -830,10 +836,7 @@ func (tserver *testServer) postBody( func TestRequestBodyLimit(t *testing.T) { t.Parallel() - const ( - maxMessageSize = 4096 // the default MAX_MESSAGE_SIZE - nick = "alice_bodylimit" - ) + const nick = "alice_bodylimit" endpoints := []struct { name, path, value string @@ -1750,52 +1753,67 @@ func TestConcurrentQuitsSendOneQuit(t *testing.T) { // 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. +// quitRaceTrials times, at once, 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() + for trial := range quitRaceTrials { + t.Run(fmt.Sprintf("trial %d", trial), func(t *testing.T) { + t.Parallel() + + checkQuitRaceOnce(t, newTestServer(t), leave) + }) + } +} + +// checkQuitRaceOnce runs one trial of checkQuitRace on +// tserver. +func checkQuitRaceOnce( + t *testing.T, + tserver *testServer, + 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") + firstToken := tserver.createSession(nick) + observerToken := tserver.createSession("raceobserver") - tserver.setPassword(nick) + tserver.setPassword(nick) - tokens := []string{firstToken, tserver.login(nick)} + 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) + 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("%d QUITs, want 1: %v", quits, msgs) + } + + for _, token := range tokens { + status, _ := tserver.getState(token) + if status != http.StatusUnauthorized { + t.Errorf("state got %d, want 401", status) + } + } + + // createSession fails the test if the nick is taken. + tserver.createSession(nick) } // TestLogoutOfOneClientKeepsSession checks that when one of @@ -1933,11 +1951,28 @@ const ( // return once the session has left. quitWakeDeadline = 5 * time.Second - // pollWaitStep is how often checkQuitWakesPoll checks + // pollWaitStep is how often waitForLongPoll checks // whether the long poll is waiting yet. pollWaitStep = 10 * time.Millisecond ) +// waitForLongPoll waits until a long poll of the session +// sessionID is waiting, and fails the test if none is within +// quitWakeDeadline. +func (tserver *testServer) waitForLongPoll(sessionID int64) { + tserver.t.Helper() + + deadline := time.Now().Add(quitWakeDeadline) + + for tserver.broker.ListenerCount(sessionID) == 0 { + if time.Now().After(deadline) { + tserver.t.Fatal("the long poll did not start waiting") + } + + time.Sleep(pollWaitStep) + } +} + // TestQuitWakesPeerLongPoll checks that a session's QUIT // wakes the long poll of a user who shares a channel with // it. @@ -2012,15 +2047,7 @@ func checkQuitWakesPoll( 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) - } + tserver.waitForLongPoll(peerID) leave(tserver, token) @@ -2850,7 +2877,12 @@ func TestLongPoll(t *testing.T) { pollMsgs = result.Messages }() - time.Sleep(200 * time.Millisecond) + bobID, err := tserver.database.GetSessionByNick(t.Context(), "lp_bob") + if err != nil { + t.Fatalf("get session of lp_bob: %v", err) + } + + tserver.waitForLongPoll(bobID) tserver.sendCommand(aliceToken, map[string]any{ commandKey: privmsgCmd, @@ -3009,18 +3041,18 @@ func TestGetMessagesRemovesWaitChannel(t *testing.T) { longTimeout = 10 // shortTimeout, in seconds, is the timeout of the - // requests that wait it out. + // request that waits it out. shortTimeout = 1 ) // queued is whether a message is queued before the request, - // whose timeout is in seconds. failingRead is as for - // getMessagesFailingRead. wantReads is how many times the - // handler reads the queue: once if it answers at once, + // whose timeout is in seconds. failingRead and notified are + // as for getMessagesFailingRead. wantReads is how many times + // the handler reads the queue: once if it answers at once, // twice if it waits first. tests := []struct { name string - queued bool + queued, notified bool timeout, failingRead int wantReads, wantStatus int }{ @@ -3041,9 +3073,9 @@ func TestGetMessagesRemovesWaitChannel(t *testing.T) { wantReads: 1, wantStatus: http.StatusInternalServerError, }, { - name: "read after the wait failed", timeout: shortTimeout, - failingRead: 2, wantReads: 2, - wantStatus: http.StatusInternalServerError, + name: "read after the wait failed", notified: true, + timeout: handlers.MaxLongPollTimeout, failingRead: 2, + wantReads: 2, wantStatus: http.StatusInternalServerError, }, } @@ -3075,7 +3107,7 @@ func TestGetMessagesRemovesWaitChannel(t *testing.T) { } status, reads := tserver.getMessagesFailingRead( - token, lastID, tc.timeout, tc.failingRead, + token, sessionID, lastID, tc.timeout, tc.failingRead, tc.notified, ) if status != tc.wantStatus { t.Errorf("status %d, want %d", status, tc.wantStatus) @@ -3092,14 +3124,17 @@ func TestGetMessagesRemovesWaitChannel(t *testing.T) { } // getMessagesFailingRead sends GET /api/v1/messages for the -// client token with timeout and after afterID. The handler's -// read of the queue numbered failingRead, counting from 1, -// fails; with 0, none does. It returns the response status -// and how many times the handler read the queue. +// client token of the session sessionID, with timeout and +// after afterID. The handler's read of the queue numbered +// failingRead, counting from 1, fails; with 0, none does. +// With notified, it notifies the broker once the request +// waits. It returns the response status and how many times +// the handler read the queue. func (tserver *testServer) getMessagesFailingRead( token string, - afterID int64, + sessionID, afterID int64, timeout, failingRead int, + notified bool, ) (int, int) { tserver.t.Helper() @@ -3121,20 +3156,45 @@ func (tserver *testServer) getMessagesFailingRead( ) }) - resp, err := doRequestAuth( - tserver.t, http.MethodGet, - tserver.url(fmt.Sprintf( - "%s?timeout=%d&after=%d", apiMessages, timeout, afterID, - )), - token, nil, + var ( + status int + err error ) + + answered := make(chan struct{}) + + go func() { + defer close(answered) + + var resp *http.Response + + 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 + } + + status = resp.StatusCode + _ = resp.Body.Close() + }() + + if notified { + tserver.waitForLongPoll(sessionID) + tserver.broker.Notify(sessionID) + } + + <-answered + if err != nil { tserver.t.Fatalf("get messages: %v", err) } - _ = resp.Body.Close() - - return resp.StatusCode, reads + return status, reads } // TestLongPollRemovesWaitChannelWhenClientGivesUp checks that @@ -4343,12 +4403,11 @@ func TestLoginRateLimitIgnoresForwardedHeaders(t *testing.T) { } func TestLoginRateLimitPerClientBehindTrustedProxy(t *testing.T) { - // The test client connects from 127.0.0.1. Trusted as a proxy, it - // passes each client's address in X-Forwarded-For. t.Setenv rules - // out t.Parallel. - t.Setenv("TRUSTED_PROXIES", "127.0.0.1") + t.Parallel() - tserver := newTestServer(t) + // The test client connects from 127.0.0.1. Trusted as a proxy, it + // passes each client's address in X-Forwarded-For. + tserver := newTestServerWith(t, 0, "", "", "127.0.0.1") // An empty JSON object is refused after the rate limit counts the // attempt, so no attempt checks a password or looks up a hostname. @@ -5079,7 +5138,9 @@ func TestRefusedSessionLeavesStampUnspent(t *testing.T) { badNick = "bad nick" ) - tserver := newTestServerWith(t, bits, "", "") + tserver := newTestServerWith( + t, bits, "", "", config.DefaultTrustedProxies, + ) // A server with no SERVER_NAME takes stamps for "neoirc". status := postSessionWithStamp( @@ -5285,6 +5346,7 @@ func newTestServerWithOper( return newTestServerWith( t, 0, testOperName, testOperPassword, + config.DefaultTrustedProxies, ) } diff --git a/internal/handlers/export_test.go b/internal/handlers/export_test.go index 67d1b65..31bd141 100644 --- a/internal/handlers/export_test.go +++ b/internal/handlers/export_test.go @@ -6,6 +6,10 @@ import ( "sneak.berlin/go/neoirc/internal/db" ) +// MaxLongPollTimeout is the longest HandleGetMessages waits, in +// seconds. +const MaxLongPollTimeout = maxLongPollTimeout + // SetPollMessages makes HandleGetMessages read the client // output queue with poll in place of Database.PollMessages. func (hdlr *Handlers) SetPollMessages( diff --git a/internal/ircserver/conn.go b/internal/ircserver/conn.go index 0584e19..3976972 100644 --- a/internal/ircserver/conn.go +++ b/internal/ircserver/conn.go @@ -70,6 +70,10 @@ type Conn struct { closed bool killed bool + // pollInterval is how often the relay checks the client + // output queue when no broker notification arrives. + pollInterval time.Duration + // quitReason is the reason in the QUIT that peers are // sent when the connection ends. The client's QUIT // replaces it. @@ -83,6 +87,7 @@ func newConn( brk *broker.Broker, cfg *config.Config, svc *service.Service, + poll time.Duration, ) *Conn { host, _, _ := net.SplitHostPort(tcpConn.RemoteAddr().String()) @@ -101,6 +106,8 @@ func newConn( serverSfx: srvName, remoteIP: host, quitReason: "Connection closed", + + pollInterval: poll, } conn.commands = conn.buildCommandMap() diff --git a/internal/ircserver/export_test.go b/internal/ircserver/export_test.go index 5357b7e..19e1824 100644 --- a/internal/ircserver/export_test.go +++ b/internal/ircserver/export_test.go @@ -14,16 +14,16 @@ import ( ) // NewTestServer creates a Server suitable for testing, whose -// connections use interval and deadline in place of -// pingInterval and pongDeadline, and resolve in place of -// resolveHost. +// connections use interval, deadline and poll in place of +// pingInterval, pongDeadline and pollInterval, and resolve in +// place of resolveHost. // The caller must call Stop() when finished. func NewTestServer( log *slog.Logger, cfg *config.Config, database *db.Database, brk *broker.Broker, - interval, deadline time.Duration, + interval, deadline, poll time.Duration, resolve func(ctx context.Context, addr string) string, ) *Server { globs := &globals.Globals{ @@ -46,10 +46,18 @@ func NewTestServer( pingInterval: interval, pongDeadline: deadline, + pollInterval: poll, + stopTimeout: stopTimeout, resolveHost: resolve, } } +// SetStopTimeout makes Stop wait at most timeout for the +// connections' cleanup. +func (s *Server) SetStopTimeout(timeout time.Duration) { + s.stopTimeout = timeout +} + // ResolveHost is the reverse DNS lookup that a server made by // New does for each client. func ResolveHost(ctx context.Context, addr string) string { @@ -99,10 +107,6 @@ 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( @@ -111,7 +115,7 @@ func NewTestConn( tcpConn net.Conn, nick string, ) *Conn { - conn := newConn(tcpConn, log, nil, nil, cfg, nil) + conn := newConn(tcpConn, log, nil, nil, cfg, nil, pollInterval) conn.nick = nick return conn diff --git a/internal/ircserver/integration_test.go b/internal/ircserver/integration_test.go index db4b4f2..3be1525 100644 --- a/internal/ircserver/integration_test.go +++ b/internal/ircserver/integration_test.go @@ -3,7 +3,6 @@ package ircserver_test import ( "strings" "testing" - "time" "sneak.berlin/go/neoirc/internal/config" ) @@ -1508,9 +1507,6 @@ func TestIntegrationThirdClientObserver(t *testing.T) { bob.joinAndDrain("#trio") carol.joinAndDrain("#trio") - // Drain join notifications. - time.Sleep(100 * time.Millisecond) - // alice sends; both bob and carol should receive. alice.send("PRIVMSG #trio :hello trio") diff --git a/internal/ircserver/relay.go b/internal/ircserver/relay.go index 2933410..62d49be 100644 --- a/internal/ircserver/relay.go +++ b/internal/ircserver/relay.go @@ -17,7 +17,7 @@ import ( func (c *Conn) relayMessages(ctx context.Context) { // Use a ticker as a fallback; primary wakeup is via // broker notification. - ticker := time.NewTicker(pollInterval) + ticker := time.NewTicker(c.pollInterval) defer ticker.Stop() for { diff --git a/internal/ircserver/relay_test.go b/internal/ircserver/relay_test.go index 5db0982..3bd4e2f 100644 --- a/internal/ircserver/relay_test.go +++ b/internal/ircserver/relay_test.go @@ -9,6 +9,7 @@ import ( "time" "golang.org/x/crypto/bcrypt" + "sneak.berlin/go/neoirc/internal/config" "sneak.berlin/go/neoirc/internal/dbtest" "sneak.berlin/go/neoirc/internal/ircserver" "sneak.berlin/go/neoirc/internal/service" @@ -58,7 +59,11 @@ func (env *testEnv) queueMessage( func TestRelayKeepsOneWaitChannel(t *testing.T) { t.Parallel() - env := newTestEnv(t) + env := newTestEnvWithResolver( + t, &config.Config{ServerName: testServerName}, + ircserver.PingInterval, ircserver.PongDeadline, + ircserver.PollInterval, ircserver.ResolveHost, + ) client := env.dial(t) client.register("idle") @@ -119,6 +124,63 @@ func (env *testEnv) checkOneWaitChannel( } } +// TestRelayDeliversAfterFailedRead checks that when the relay +// cannot read the client output queue, it waits for the next +// notification, and delivers once the queue can be read again. +func TestRelayDeliversAfterFailedRead(t *testing.T) { + t.Parallel() + + env := newTestEnv(t) + client := env.dial(t) + client.register("reader") + + sessionID, err := env.database.GetSessionByNick(t.Context(), "reader") + if err != nil { + t.Fatalf("get session of reader: %v", err) + } + + dbtest.Exec(t, env.dbURL, + "ALTER TABLE client_queues RENAME TO hidden_queues") + + // Each pass of the relay takes a wait channel and then reads + // the queue. Once it holds a channel after the second + // notification, the read after the first one has failed. + for range 2 { + env.waitForWaitChannel(t, sessionID) + env.brk.Notify(sessionID) + } + + env.waitForWaitChannel(t, sessionID) + + dbtest.Exec(t, env.dbURL, + "ALTER TABLE hidden_queues RENAME TO client_queues") + + env.queueMessage( + t, "reader", "PRIVMSG", "carol", "reader", []string{"readable"}, + ) + + client.readUntil(func(l string) bool { + return strings.Contains(l, "readable") + }) +} + +// waitForWaitChannel waits until the session sessionID holds a +// broker wait channel, and fails the test if it does not +// within testTimeout. +func (env *testEnv) waitForWaitChannel(t *testing.T, sessionID int64) { + t.Helper() + + deadline := time.Now().Add(testTimeout) + + for env.brk.ListenerCount(sessionID) == 0 { + if time.Now().After(deadline) { + t.Fatal("the relay is not waiting for a notification") + } + + time.Sleep(relayCheckStep) + } +} + // TestRelayMultiLineBody checks that each line of a PRIVMSG // or NOTICE body sent over the HTTP API reaches an IRC // listener user as its own line. diff --git a/internal/ircserver/server.go b/internal/ircserver/server.go index 8269934..7d6950c 100644 --- a/internal/ircserver/server.go +++ b/internal/ircserver/server.go @@ -53,6 +53,11 @@ type Server struct { pingInterval time.Duration pongDeadline time.Duration + // pollInterval and stopTimeout are the constants of the + // same names, except in tests, which change them. + pollInterval time.Duration + stopTimeout time.Duration + // resolveHost is the function of the same name, except in // tests, which replace it with one that blocks. resolveHost func(ctx context.Context, addr string) string @@ -79,6 +84,8 @@ func New( pingInterval: pingInterval, pongDeadline: pongDeadline, + pollInterval: pollInterval, + stopTimeout: stopTimeout, resolveHost: resolveHost, } @@ -154,10 +161,10 @@ func (s *Server) stop() { select { case <-done: - case <-time.After(stopTimeout): + case <-time.After(s.stopTimeout): s.log.Error( "irc connections not cleaned up before stop timeout", - "timeout", stopTimeout, + "timeout", s.stopTimeout, ) } } @@ -181,7 +188,7 @@ func (s *Server) acceptLoop(ctx context.Context) { client := newConn( tcpConn, s.log, - s.database, s.brk, s.cfg, s.svc, + s.database, s.brk, s.cfg, s.svc, s.pollInterval, ) s.mu.Lock() diff --git a/internal/ircserver/server_test.go b/internal/ircserver/server_test.go index 2ca7d02..ea7f2f9 100644 --- a/internal/ircserver/server_test.go +++ b/internal/ircserver/server_test.go @@ -18,6 +18,7 @@ import ( "time" "go.uber.org/fx/fxtest" + "golang.org/x/crypto/bcrypt" "sneak.berlin/go/neoirc/internal/broker" "sneak.berlin/go/neoirc/internal/config" "sneak.berlin/go/neoirc/internal/db" @@ -32,6 +33,12 @@ import ( const testTimeout = 5 * time.Second +// testPollInterval is how often the relay of a test server +// checks the client output queue without a broker +// notification. It is so long that the relay delivers only on +// notification, and a test fails when one is missing. +const testPollInterval = time.Hour + // testEnv holds the shared test infrastructure. type testEnv struct { dbURL string @@ -91,16 +98,19 @@ func newTestEnvWithPingTimes( t.Helper() return newTestEnvWithResolver( - t, cfg, interval, deadline, ircserver.ResolveHost, + t, cfg, interval, deadline, testPollInterval, + ircserver.ResolveHost, ) } // newTestEnvWithResolver is newTestEnvWithPingTimes, except -// that the server looks up a client's hostname with resolve. +// that the relay checks the client output queue every poll +// without a broker notification, and the server looks up a +// client's hostname with resolve. func newTestEnvWithResolver( t *testing.T, cfg *config.Config, - interval, deadline time.Duration, + interval, deadline, poll time.Duration, resolve func(ctx context.Context, addr string) string, ) *testEnv { t.Helper() @@ -116,8 +126,9 @@ func newTestEnvWithResolver( } database, err := db.New(lifecycle, db.Params{ - Logger: dbLog, - Config: &config.Config{DBURL: dbURL}, + Logger: dbLog, + Config: &config.Config{DBURL: dbURL}, + PasswordCost: bcrypt.MinCost, }) if err != nil { t.Fatalf("new database: %v", err) @@ -134,7 +145,7 @@ func newTestEnvWithResolver( )) srv := ircserver.NewTestServer( - log, cfg, database, brk, interval, deadline, resolve, + log, cfg, database, brk, interval, deadline, poll, resolve, ) // Port 0 makes the system pick a free port. dial reads it @@ -430,7 +441,8 @@ func TestRegistrationWhileAnotherLookupIsPending(t *testing.T) { env := newTestEnvWithResolver( t, &config.Config{ServerName: testServerName}, - ircserver.PingInterval, ircserver.PongDeadline, resolve, + ircserver.PingInterval, ircserver.PongDeadline, + testPollInterval, resolve, ) env.dial(t) @@ -1039,12 +1051,16 @@ func TestStopEndsSessions(t *testing.T) { } // TestStopReturnsWhileCleanupIsStuck checks that Stop stops -// waiting for a connection's cleanup after StopTimeout, so a -// cleanup that cannot finish does not hold up stopping. +// waiting for a connection's cleanup after its stop timeout, +// so a cleanup that cannot finish does not hold up stopping. func TestStopReturnsWhileCleanupIsStuck(t *testing.T) { t.Parallel() + const stopTimeout = 100 * time.Millisecond + env := newTestEnv(t) + env.srv.SetStopTimeout(stopTimeout) + client := env.dial(t) client.register("stuck") @@ -1070,7 +1086,7 @@ func TestStopReturnsWhileCleanupIsStuck(t *testing.T) { select { case <-stopped: - case <-time.After(ircserver.StopTimeout + testTimeout): + case <-time.After(stopTimeout + testTimeout): t.Error("Stop did not return while a cleanup was stuck") } @@ -1122,9 +1138,6 @@ func TestDirectMessage(t *testing.T) { receiver := env.dial(t) receiver.register("dmreceiver") - // Give relay goroutines time to start. - time.Sleep(100 * time.Millisecond) - sender.send("PRIVMSG dmreceiver :hello privately") lines := receiver.readUntil(func(line string) bool { diff --git a/internal/service/service_test.go b/internal/service/service_test.go index f27eff8..4a67183 100644 --- a/internal/service/service_test.go +++ b/internal/service/service_test.go @@ -1,7 +1,3 @@ -// Tests use a global viper instance for configuration, -// making parallel execution unsafe. -// -//nolint:paralleltest package service_test import ( @@ -59,28 +55,12 @@ func newTestEnv(t *testing.T) *testEnv { } }, logger.New, - func( - lifecycle fx.Lifecycle, - globs *globals.Globals, - log *logger.Logger, - ) (*config.Config, error) { - cfg, err := config.New( - lifecycle, config.Params{ //nolint:exhaustruct - Globals: globs, Logger: log, - }, - ) - if err != nil { - return nil, fmt.Errorf( - "test config: %w", err, - ) + func() *config.Config { + return &config.Config{ + DBURL: dbURL, + OperName: "admin", + OperPassword: "secret", } - - cfg.DBURL = dbURL - cfg.Port = 0 - cfg.OperName = "admin" - cfg.OperPassword = "secret" - - return cfg, nil }, func( lifecycle fx.Lifecycle, @@ -136,6 +116,8 @@ func createSession( // environment starts on an empty database while another one // is still open. func TestEachTestEnvHasItsOwnDatabase(t *testing.T) { + t.Parallel() + ctx := t.Context() first := newTestEnv(t) @@ -146,6 +128,8 @@ func TestEachTestEnvHasItsOwnDatabase(t *testing.T) { } func TestFanOut(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -173,6 +157,8 @@ func TestFanOut(t *testing.T) { } func TestJoinChannel(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -213,6 +199,8 @@ func TestJoinChannel(t *testing.T) { } func TestPartChannel(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -313,6 +301,8 @@ func joinConcurrently( // concurrent JOINs of a +l 2 channel with one member, one // gets in and the other gets ERR_CHANNELISFULL. func TestConcurrentJoinsRespectUserLimit(t *testing.T) { + t.Parallel() + const userLimit = 2 env := newTestEnv(t) @@ -375,6 +365,8 @@ func TestConcurrentJoinsRespectUserLimit(t *testing.T) { // two concurrent JOINs that create a channel, exactly one // makes its session the operator. func TestConcurrentFirstJoinsMakeOneOperator(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -426,6 +418,8 @@ func TestConcurrentFirstJoinsMakeOneOperator(t *testing.T) { // either joins a channel that still exists or gets // ERR_NOSUCHCHANNEL, and never fails otherwise. func TestJoinConcurrentWithLastPart(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -491,6 +485,8 @@ func TestJoinConcurrentWithLastPart(t *testing.T) { // quits, gets in only once the channel is deleted, into a // new channel of that name with no key. func TestJoinConcurrentWithLastQuit(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -569,81 +565,95 @@ const quitCrowdSize = 10 // 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. +// last. Each trial runs on a database of its own. func TestQuitQueuedAheadOfNewNickHoldersJoin(t *testing.T) { - env := newTestEnv(t) - ctx := t.Context() + t.Parallel() 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) + t.Run(fmt.Sprintf("trial %d", trial), func(t *testing.T) { + t.Parallel() - for _, channel := range []string{crowd, shared} { - _, err := env.svc.JoinChannel(ctx, quitterSID, 0, nick, channel, "") - if err != nil { - t.Fatalf("join %s: %v", channel, err) - } + checkQuitQueuedAheadOfJoin(t, newTestEnv(t)) + }) + } +} + +// checkQuitQueuedAheadOfJoin runs one trial of +// TestQuitQueuedAheadOfNewNickHoldersJoin on env. +func checkQuitQueuedAheadOfJoin(t *testing.T, env *testEnv) { + t.Helper() + + const ( + nick = "reused" + crowd = "#crowd" + shared = "#shared" + watcherNick = "watcher" + ) + + ctx := t.Context() + quitterSID := createSession(ctx, t, env.db, nick) + + for _, channel := range []string{crowd, shared} { + _, err := env.svc.JoinChannel(ctx, quitterSID, 0, 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) + for member := range quitCrowdSize { + memberNick := fmt.Sprintf("crowd%d", member) + memberSID := createSession(ctx, t, env.db, memberNick) - _, err := env.svc.JoinChannel(ctx, memberSID, 0, memberNick, crowd, "") - if err != nil { - t.Fatalf("join %s: %v", memberNick, err) - } + _, err := env.svc.JoinChannel(ctx, memberSID, 0, 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) + } - watcherSID, watcherClientID, _, err := env.db.CreateSession( - ctx, watcherNick, watcherNick, "localhost", "127.0.0.1", + _, err = env.svc.JoinChannel(ctx, watcherSID, 0, 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("new session: %v", 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( + "watcher got %v from %s, want QUIT then JOIN", + commands, nick, ) - if err != nil { - t.Fatalf("create session %s: %v", watcherNick, err) - } - - _, err = env.svc.JoinChannel(ctx, watcherSID, 0, 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, - ) - } } } @@ -691,6 +701,8 @@ func commandsFrom(msgs []db.IRCMessage, nick string) []string { // into a +i channel uses the invite up, so that a JOIN // after leaving gets ERR_INVITEONLYCHAN. func TestJoinUsesUpInvite(t *testing.T) { + t.Parallel() + const channel = "#club" env := newTestEnv(t) @@ -736,6 +748,8 @@ func TestJoinUsesUpInvite(t *testing.T) { // session's nick, username and hostname refuses its JOIN // with ERR_BANNEDFROMCHAN. func TestJoinRefusedByHostmaskBan(t *testing.T) { + t.Parallel() + const channel = "#club" env := newTestEnv(t) @@ -774,6 +788,8 @@ func TestJoinRefusedByHostmaskBan(t *testing.T) { // keeps a channel other members are still on and deletes // the channel when its last member parts. func TestPartDeletesChannelOnlyWhenEmpty(t *testing.T) { + t.Parallel() + const channel = "#room" env := newTestEnv(t) @@ -821,6 +837,8 @@ func TestPartDeletesChannelOnlyWhenEmpty(t *testing.T) { // TestKickOfLastMemberDeletesChannel checks that a KICK // that leaves a channel with no members deletes it. func TestKickOfLastMemberDeletesChannel(t *testing.T) { + t.Parallel() + const channel = "#room" env := newTestEnv(t) @@ -845,6 +863,8 @@ func TestKickOfLastMemberDeletesChannel(t *testing.T) { } func TestSendChannelMessage(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -892,6 +912,8 @@ func TestSendChannelMessage(t *testing.T) { } func TestBroadcastQuit(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -929,6 +951,8 @@ func TestBroadcastQuit(t *testing.T) { // error as a DM to an unknown nick. The trigger deletes the // target's session just before the DM is stored. func TestSendDirectMessageTargetQuits(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -960,6 +984,8 @@ func TestSendDirectMessageTargetQuits(t *testing.T) { // trigger deletes the channel just before the message is // stored. func TestSendChannelMessageChannelDeleted(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -991,6 +1017,8 @@ func TestSendChannelMessageChannelDeleted(t *testing.T) { } func TestSendChannelMessage_Moderated(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -1042,6 +1070,8 @@ func TestSendChannelMessage_Moderated(t *testing.T) { // not spend its hashcash stamp: the operator can then send // the same message with it. func TestRefusedChannelMessageLeavesStampUnspent(t *testing.T) { + t.Parallel() + const ( channel = "#paid" bits = 2 @@ -1067,6 +1097,8 @@ func TestRefusedChannelMessageLeavesStampUnspent(t *testing.T) { for _, refusal := range refusals { t.Run(refusal.name, func(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -1119,6 +1151,8 @@ func TestRefusedChannelMessageLeavesStampUnspent(t *testing.T) { } func TestQueryUserMode(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -1169,6 +1203,8 @@ func TestQueryUserMode(t *testing.T) { } func TestUserhostReply(t *testing.T) { + t.Parallel() + env := newTestEnv(t) ctx := t.Context() @@ -1274,10 +1310,14 @@ func applyUserModeTests() []applyUserModeTest { } func TestApplyUserMode(t *testing.T) { + t.Parallel() + env := newTestEnv(t) for i, test := range applyUserModeTests() { t.Run(test.modeStr, func(t *testing.T) { + t.Parallel() + ctx := t.Context() sid := createSession( ctx, t, env.db, fmt.Sprintf("user%d", i),