Stop the slow test packages waiting (closes #263)
check / check (push) Failing after 1m9s

The slowest test packages spent their time waiting rather than testing.

db.Params gains an optional PasswordCost (zero means bcrypt.DefaultCost), and the db, ircserver and handlers tests hash at bcrypt.MinCost. A test IRC server's relay checks its queue on broker notification instead of polling. The service and handlers tests build config.Config directly and run in parallel; config exports DefaultTrustedProxies and ParseTrustedProxies for them. Fixed sleeps give way to waits for the event. Production defaults are unchanged.

What still takes time is CPU work under -race or a timeout a test is about.

Model: opus-5-5
This commit was merged in pull request #267.
This commit is contained in:
clawbot committed 2026-11-11 01:15:43 +01:00
1 parent 847f5c6fec
commit ed5125606a
15 files changed
+558 -261

No files matched your search

+24 -18
View File
@@ -36,9 +36,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")
@@ -117,7 +117,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 {
@@ -219,9 +219,7 @@ func parseStringAllowEmptyEnv(key string, value *string) error {
return parseString(key, value)
}
// 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.
// parseTrustedProxies sets *value to the addresses in TRUSTED_PROXIES.
func parseTrustedProxies(value **netipx.IPSet) error {
var raw string
@@ -230,18 +228,7 @@ func parseTrustedProxies(value **netipx.IPSet) error {
return err
}
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)
}
@@ -251,6 +238,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
+1 -1
View File
@@ -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)
+18 -5
View File
@@ -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: &params,
log: params.Logger.Get(),
params: &params,
log: params.Logger.Get(),
passwordCost: passwordCost,
}
database.log.Info("Database instantiated")
+80
View File
@@ -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"
@@ -240,6 +241,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) {
+6 -3
View File
@@ -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())
+176 -114
View File
@@ -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,
)
}
+4
View File
@@ -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(
+7
View File
@@ -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()
+13 -9
View File
@@ -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
-4
View File
@@ -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")
+1 -1
View File
@@ -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 {
+63 -1
View File
@@ -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.
+10 -3
View File
@@ -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()
+26 -13
View File
@@ -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"
@@ -35,6 +36,12 @@ import (
// enough for a heavily loaded host.
const testTimeout = 30 * 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
@@ -94,16 +101,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()
@@ -119,8 +129,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)
@@ -137,7 +148,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
@@ -433,7 +444,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)
@@ -1064,12 +1076,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")
@@ -1095,7 +1111,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")
}
@@ -1147,9 +1163,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 {
+129 -89
View File
@@ -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),