Mail queue: nothing on the request path, a worker that survives, a close that ends
Astra's second round on oficina/slides #20, applied here: the same queue. - Send does no I/O: a drop is counted in an atomic and logged by the worker. Every address's link is queued, admitted or not, so the fill level does not depend on admission. - The worker recovers per message from a panic and keeps draining. - Sends run under the queue's context; past the drain deadline close cancels the one in flight, abandons the rest with a log line, and the worker exits. - NewAuth registers its queue under the plugin; CloseSigninMail stops one and every test that builds a plugin calls it in cleanup, so no test leaves a worker running. DrainSigninMail still drains them all for main. Tests (each of the first four fails with its defect put back): TestMailQueueSendNeverBlocksWhenFull, TestMailQueueSurvivesAPanic, TestMailQueueCloseIsBoundedAndStopsTheWorker (goroutine count), TestCloseSigninMailStopsTheWorker, TestMailQueueQueuesEveryAddress.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+325
−54
internal/sheets/auth.go+10 −5internal/sheets/begin_test.go+1 −0internal/sheets/home_test.go+1 −0internal/sheets/mailqueue.go+130 −49internal/sheets/mailqueue_test.go+181 −0internal/sheets/signin_test.go+1 −0internal/sheets/signin_timing_test.go+1 −0
CI passed — run details
CI log
amadan: .amadan/timeout: this job's budget is 15m0s (this runner allows up to 30m0s) === vet === CGO_ENABLED=0 go vet ./... CGO_ENABLED=0 go vet -tags browser ./internal/sheets/ === fmt === === build === CGO_ENABLED=0 go build ./... === migrations === go tool rastrillo migration check rastrillo migration check: models and migrations agree === go-test === CGO_ENABLED=0 go test -count=1 ./... ? amadan.net/oficina/sheets/cmd/sheets [no test files] ok amadan.net/oficina/sheets/internal/guardstest 0.098s ok amadan.net/oficina/sheets/internal/sheets 4.087s ok amadan.net/oficina/sheets/internal/sheets/calc 0.010s ok amadan.net/oficina/sheets/internal/sheets/format 0.003s ? amadan.net/oficina/sheets/internal/sheets/keymailtest [no test files] ok amadan.net/oficina/sheets/internal/sheets/live 0.053s ok amadan.net/oficina/sheets/internal/sheets/seal 0.003s ok amadan.net/oficina/sheets/internal/sheets/sealtest 0.005s ok amadan.net/oficina/sheets/internal/sheets/store 3.060s ok amadan.net/oficina/sheets/internal/sheets/xlsx 0.315s ? amadan.net/oficina/sheets/tools/gencorpus [no test files] === race === go test -race -count=1 ./internal/sheets/store/... ./internal/sheets/live/... ok amadan.net/oficina/sheets/internal/sheets/store 54.822s ok amadan.net/oficina/sheets/internal/sheets/live 1.061s
diff --git a/internal/sheets/auth.go b/internal/sheets/auth.go| index 6e52b09..681460c 100644 |
| --- a/internal/sheets/auth.go |
| +++ b/internal/sheets/auth.go |
| @@ -151,18 +151,15 @@ func NewAuth(db *sql.DB, env AuthEnv, logger *slog.Logger, authorize func(string |
| // is the version that says so. |
| return nil, errors.New("sign-in: refusing to wire auth with no admission gate") |
| } |
| - queue := newMailQueue(env.Mailer, authorize, logger) |
| + queue := newMailQueue(env.Mailer, authorize, nil, logger) |
| if a, ok := env.Mailer.(queueAttacher); ok { |
| a.attachQueue(queue.wait) |
| } |
| - signinQueuesMu.Lock() |
| - signinQueues = append(signinQueues, queue) |
| - signinQueuesMu.Unlock() |
| signedInPath := "/sheets" |
| if env.Home != nil { |
| signedInPath = connect.LocalCompletePath |
| } |
| - return auth.New(auth.Config{ |
| + plugin, err := auth.New(auth.Config{ |
| DB: db, |
| Origin: env.Origin, |
| InstanceKey: env.InstanceKey, |
| @@ -179,6 +176,14 @@ func NewAuth(db *sql.DB, env AuthEnv, logger *slog.Logger, authorize func(string |
| Subject: chrome.EmailSubject("Sheets"), |
| Logger: logger, |
| }) |
| + if err != nil { |
| + queue.close() |
| + return nil, err |
| + } |
| + signinQueuesMu.Lock() |
| + signinQueues[plugin] = queue |
| + signinQueuesMu.Unlock() |
| + return plugin, nil |
| } |
| // admits builds the admission gate. |
diff --git a/internal/sheets/begin_test.go b/internal/sheets/begin_test.go| index f767cf2..4169d8a 100644 |
| --- a/internal/sheets/begin_test.go |
| +++ b/internal/sheets/begin_test.go |
| @@ -123,6 +123,7 @@ func appWithOwners(t *testing.T, origin string, owners ...string) (*App, *record |
| if err != nil { |
| t.Fatalf("NewAuth: %v", err) |
| } |
| + t.Cleanup(func() { CloseSigninMail(a) }) |
| app.UseAuth(a) |
| rememberMailer(app, rec) |
| return app, rec |
diff --git a/internal/sheets/home_test.go b/internal/sheets/home_test.go| index aa3dff7..16254ce 100644 |
| --- a/internal/sheets/home_test.go |
| +++ b/internal/sheets/home_test.go |
| @@ -49,6 +49,7 @@ func newHomeFixtureWith(t *testing.T, directory string) *homeFixture { |
| if err != nil { |
| t.Fatal(err) |
| } |
| + t.Cleanup(func() { CloseSigninMail(localAuth) }) |
| app.UseAuth(localAuth) |
| if err := app.ConfigureHome(&cfg); err != nil { |
| t.Fatal(err) |
diff --git a/internal/sheets/mailqueue.go b/internal/sheets/mailqueue.go| index 86ae141..5ac37c8 100644 |
| --- a/internal/sheets/mailqueue.go |
| +++ b/internal/sheets/mailqueue.go |
| @@ -2,111 +2,172 @@ package sheets |
| import ( |
| "context" |
| + "fmt" |
| "log/slog" |
| "sync" |
| + "sync/atomic" |
| "time" |
| + "amadan.net/rastrillo/rastrillo/auth" |
| "amadan.net/rastrillo/rastrillo/mail" |
| ) |
| -// mailQueue is the sign-in mailer rauth is given: Send queues the link |
| -// and returns at once, and one background worker asks App.Admits and |
| -// only then talks to SMTP. |
| +// mailQueue is the sign-in mailer the auth plugin is given: Send queues |
| +// the link and returns at once, and one background worker decides |
| +// whether the address may have it and only then talks to SMTP. |
| // |
| -// Both halves are the no-oracle guarantee (handleBegin), measured rather |
| -// than reasoned about. When the send ran inside the request, a permitted |
| +// Both halves are the no-oracle guarantee, measured rather than |
| +// reasoned about. When the send ran inside the request, a permitted |
| // address waited for the SMTP round trip before its `303 ?sent=1` and a |
| -// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides #20; Sheets had the same split). A response |
| -// time is an answer like any other. So POST /signin does the same |
| -// synchronous work for every address — rauth's budgets, its link row, |
| -// this enqueue — and the one admission-dependent step, the send, happens |
| -// after the response, where nobody can time it. The admission check |
| -// moved into the worker for the same reason: a check before responding |
| -// would be a branch an attacker could time, however cheap it is today. |
| +// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides |
| +// #20). A response time is an answer like any other. So POST /signin |
| +// does the same synchronous work for every address — the auth plugin's |
| +// budgets, its link row, this enqueue — and the one admission-dependent |
| +// step, the send, happens after the response, where nobody can time it. |
| // |
| -// Delivery failures are logged, never surfaced. A failure only a |
| -// permitted address can reach is the oracle again, and the person is |
| -// told what every address is told: check your email, or ask again. |
| +// The request path is Send, and Send does no I/O and never blocks: not |
| +// a send, not an admission check, not even a log line. Every address's |
| +// link is queued, permitted or not, so how full the queue is never |
| +// depends on admission either; the worker decides afterwards. A full |
| +// queue drops the link and counts it in an atomic; the worker logs the |
| +// count when it next runs. The person is answered exactly as when the |
| +// queue is empty, because a different answer for a full queue would be |
| +// one more thing to read. (Astra on #20, round 2.) |
| // |
| -// Bounded: a queue of mailQueueDepth links. A full queue drops the link |
| -// with a log line instead of blocking a request — the per-IP and |
| -// per-address budgets inside Begin keep that from being an ordinary |
| -// event. Close drains what is queued, within mailDrainTimeout. |
| +// The worker survives a panic in the mailer: it recovers per message, |
| +// logs, and keeps draining. Delivery failures are logged, never |
| +// surfaced: a failure only a permitted address can reach is the oracle |
| +// again. |
| +// |
| +// close drains what is queued within drainTimeout. Past it, the send |
| +// in flight is cancelled through its context, everything still queued |
| +// is abandoned with a log line, and close waits a short grace for the |
| +// worker to stop. A mailer that ignores its context is the only way a |
| +// worker outlives close, and that is logged too. |
| type mailQueue struct { |
| next mail.Sender |
| - admits func(string) bool |
| + admits func(to string) bool |
| + // gate, when set, is read off the request's context at enqueue time |
| + // and travels with the link; the worker sends only if it said yes. |
| + gate func(context.Context) bool |
| logger *slog.Logger |
| - mu sync.Mutex |
| + drainTimeout time.Duration |
| + sendTimeout time.Duration |
| + |
| + mu sync.RWMutex // held for reading by Send, for writing by close |
| closed bool |
| jobs chan mailJob |
| + dropped atomic.Int64 |
| pending sync.WaitGroup |
| - done chan struct{} |
| + |
| + ctx context.Context // cancelled when close gives up waiting |
| + cancel context.CancelFunc |
| + done chan struct{} |
| } |
| -type mailJob struct{ to, subject, body string } |
| +type mailJob struct { |
| + to, subject, body string |
| + gated bool |
| +} |
| const ( |
| mailQueueDepth = 64 |
| mailSendTimeout = time.Minute |
| mailDrainTimeout = 30 * time.Second |
| + mailAbandonGrace = 5 * time.Second |
| ) |
| -func newMailQueue(next mail.Sender, admits func(string) bool, logger *slog.Logger) *mailQueue { |
| +func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue { |
| if logger == nil { |
| logger = slog.Default() |
| } |
| - q := &mailQueue{next: next, admits: admits, logger: logger, |
| - jobs: make(chan mailJob, mailQueueDepth), done: make(chan struct{})} |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + q := &mailQueue{next: next, admits: admits, gate: gate, logger: logger, |
| + drainTimeout: mailDrainTimeout, sendTimeout: mailSendTimeout, |
| + jobs: make(chan mailJob, mailQueueDepth), ctx: ctx, cancel: cancel, |
| + done: make(chan struct{})} |
| go q.run() |
| return q |
| } |
| -// Send queues one link. It never blocks and never fails: the request |
| -// that called it answers the same way whatever happens next. |
| -func (q *mailQueue) Send(_ context.Context, to, subject, body string) error { |
| - q.mu.Lock() |
| - defer q.mu.Unlock() |
| +// Send queues one link. It never blocks, never does I/O and never |
| +// fails: the request that called it answers the same way whatever |
| +// happens next. |
| +func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error { |
| + job := mailJob{to: to, subject: subject, body: body, gated: true} |
| + if q.gate != nil { |
| + job.gated = q.gate(ctx) |
| + } |
| + q.mu.RLock() |
| + defer q.mu.RUnlock() |
| if q.closed { |
| - q.logger.Error("sheets: sign-in mail after shutdown; dropped") |
| + q.dropped.Add(1) |
| return nil |
| } |
| q.pending.Add(1) |
| select { |
| - case q.jobs <- mailJob{to: to, subject: subject, body: body}: |
| + case q.jobs <- job: |
| default: |
| q.pending.Done() |
| - q.logger.Error("sheets: sign-in mail queue full; link dropped") |
| + q.dropped.Add(1) |
| } |
| return nil |
| } |
| func (q *mailQueue) run() { |
| defer close(q.done) |
| + abandoned := 0 |
| for job := range q.jobs { |
| - q.deliver(job) |
| + q.reportDrops() |
| + if q.ctx.Err() != nil { |
| + abandoned++ |
| + } else { |
| + q.deliver(job) |
| + } |
| q.pending.Done() |
| } |
| + q.reportDrops() |
| + if abandoned > 0 { |
| + q.logger.Error("sheets: sign-in links abandoned at shutdown", "count", abandoned) |
| + } |
| +} |
| + |
| +// reportDrops logs the links Send had to drop since the last report. |
| +// Here, on the worker, so the request path never writes a log line. |
| +func (q *mailQueue) reportDrops() { |
| + if n := q.dropped.Swap(0); n > 0 { |
| + q.logger.Error("sheets: sign-in mail queue full; links dropped", "count", n) |
| + } |
| } |
| +// deliver sends one link if it may go. A panic in the admission check |
| +// or the mailer is this message's failure, not the process's. |
| func (q *mailQueue) deliver(job mailJob) { |
| - if !q.admits(job.to) { |
| - // Not admitted: the link auth stored is never |
| + defer func() { |
| + if p := recover(); p != nil { |
| + q.logger.Error("sheets: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| + } |
| + }() |
| + if !job.gated || !q.admits(job.to) { |
| + // Not admitted: the link the auth plugin stored is never |
| // delivered, so it is never redeemed. No address in the log. |
| return |
| } |
| - ctx, cancel := context.WithTimeout(context.Background(), mailSendTimeout) |
| + ctx, cancel := context.WithTimeout(q.ctx, q.sendTimeout) |
| defer cancel() |
| if err := q.next.Send(ctx, job.to, job.subject, job.body); err != nil { |
| q.logger.Error("sheets: send sign-in link", "err", err) |
| } |
| } |
| -// wait blocks until every link queued so far has been delivered or |
| -// dropped. Tests use it (attachQueue) to read the mailer afterwards. |
| +// wait blocks until every link queued so far has been sent or refused. |
| +// Tests use it to read the mailer afterwards. |
| func (q *mailQueue) wait() { q.pending.Wait() } |
| -// close stops taking links and drains what is queued. |
| +// close stops taking links and drains what is queued, within |
| +// drainTimeout; then it cancels, abandons the rest, and stops the |
| +// worker (see the type's comment). |
| func (q *mailQueue) close() { |
| q.mu.Lock() |
| if q.closed { |
| @@ -116,26 +177,46 @@ func (q *mailQueue) close() { |
| q.closed = true |
| close(q.jobs) |
| q.mu.Unlock() |
| + defer q.cancel() |
| select { |
| case <-q.done: |
| - case <-time.After(mailDrainTimeout): |
| - q.logger.Error("sheets: sign-in mail still sending at shutdown; abandoned") |
| + return |
| + case <-time.After(q.drainTimeout): |
| + } |
| + q.cancel() |
| + select { |
| + case <-q.done: |
| + case <-time.After(mailAbandonGrace): |
| + q.logger.Error("sheets: sign-in mail worker ignored cancellation; left running") |
| } |
| } |
| -// signinQueues is every queue NewAuth built, so DrainSigninMail can |
| -// send what is still queued when the process stops. |
| +// signinQueues is every queue NewAuth built, by the plugin it feeds, so |
| +// CloseSigninMail can stop one and DrainSigninMail all of them. |
| var ( |
| signinQueuesMu sync.Mutex |
| - signinQueues []*mailQueue |
| + signinQueues = map[*auth.Auth]*mailQueue{} |
| ) |
| +// CloseSigninMail stops the sign-in mail queue behind x, sending what |
| +// is still in it within the queue's deadline. Tests close every plugin |
| +// they build with it. |
| +func CloseSigninMail(x *auth.Auth) { |
| + signinQueuesMu.Lock() |
| + q := signinQueues[x] |
| + delete(signinQueues, x) |
| + signinQueuesMu.Unlock() |
| + if q != nil { |
| + q.close() |
| + } |
| +} |
| + |
| // DrainSigninMail stops every sign-in mail queue and sends what is |
| -// still in it. main calls it once Serve returns. |
| +// still in them. main calls it once Serve returns. |
| func DrainSigninMail() { |
| signinQueuesMu.Lock() |
| qs := signinQueues |
| - signinQueues = nil |
| + signinQueues = map[*auth.Auth]*mailQueue{} |
| signinQueuesMu.Unlock() |
| for _, q := range qs { |
| q.close() |
| @@ -143,5 +224,5 @@ func DrainSigninMail() { |
| } |
| // queueAttacher is a test mailer that wants to wait for the queue in |
| -// front of it before it is read (helpers_test.go's recorder). |
| +// front of it before it is read (begin_test.go's recorder). |
| type queueAttacher interface{ attachQueue(wait func()) } |
diff --git a/internal/sheets/mailqueue_test.go b/internal/sheets/mailqueue_test.go| new file mode 100644 |
| index 0000000..0a8ab71 |
| --- /dev/null |
| +++ b/internal/sheets/mailqueue_test.go |
| @@ -0,0 +1,181 @@ |
| +package sheets |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "log/slog" |
| + "runtime" |
| + "sync" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// stuckHandler is a slog handler whose every write blocks until release |
| +// is closed: a log sink that has stalled, as a full pipe does. |
| +type stuckHandler struct{ release chan struct{} } |
| + |
| +func (h stuckHandler) Enabled(context.Context, slog.Level) bool { return true } |
| +func (h stuckHandler) Handle(context.Context, slog.Record) error { |
| + <-h.release |
| + return nil |
| +} |
| +func (h stuckHandler) WithAttrs([]slog.Attr) slog.Handler { return h } |
| +func (h stuckHandler) WithGroup(string) slog.Handler { return h } |
| + |
| +// funcSender adapts a function to mail.Sender. |
| +type funcSender func(ctx context.Context, to string) error |
| + |
| +func (f funcSender) Send(ctx context.Context, to, _, _ string) error { return f(ctx, to) } |
| + |
| +func everyone(string) bool { return true } |
| + |
| +// TestMailQueueSendNeverBlocksWhenFull: Send is the request path, so it |
| +// may not block or do I/O even when the queue is full — not even to log |
| +// the drop (Astra on oficina/slides #20, round 2, P1: a "queue full" log |
| +// line on a stalled sink held the request, and saturation is |
| +// admission-dependent). The worker is stuck on its first link and the |
| +// log sink is stuck too; twice the queue's depth of Sends must still |
| +// return at once, and the drops are logged by the worker afterwards. |
| +func TestMailQueueSendNeverBlocksWhenFull(t *testing.T) { |
| + release := make(chan struct{}) |
| + var once sync.Once |
| + free := func() { once.Do(func() { close(release) }) } |
| + stuck := funcSender(func(ctx context.Context, _ string) error { |
| + select { |
| + case <-release: |
| + case <-ctx.Done(): |
| + } |
| + return nil |
| + }) |
| + q := newMailQueue(stuck, everyone, nil, slog.New(stuckHandler{release})) |
| + t.Cleanup(func() { free(); q.close() }) |
| + |
| + done := make(chan struct{}) |
| + go func() { |
| + for i := 0; i < 2*mailQueueDepth; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b") |
| + } |
| + close(done) |
| + }() |
| + select { |
| + case <-done: |
| + case <-time.After(2 * time.Second): |
| + free() |
| + <-done |
| + t.Fatal("Send blocked with the queue full: the request path waited on the queue or the log") |
| + } |
| + if n := q.dropped.Load(); n == 0 { |
| + t.Fatal("no drop was counted with twice the queue's depth sent") |
| + } |
| +} |
| + |
| +// TestMailQueueQueuesEveryAddress: the queue's fill level must not |
| +// depend on admission, so Send queues a refused address's link exactly |
| +// as it queues a permitted one; only the worker asks. |
| +func TestMailQueueQueuesEveryAddress(t *testing.T) { |
| + release := make(chan struct{}) |
| + stuck := funcSender(func(ctx context.Context, _ string) error { |
| + select { |
| + case <-release: |
| + case <-ctx.Done(): |
| + } |
| + return nil |
| + }) |
| + var asked sync.Map |
| + admits := func(to string) bool { asked.Store(to, true); return to == "first@example.test" } |
| + q := newMailQueue(stuck, admits, nil, slog.New(slog.DiscardHandler)) |
| + t.Cleanup(func() { close(release); q.close() }) |
| + _ = q.Send(context.Background(), "first@example.test", "s", "b") |
| + deadline := time.Now().Add(2 * time.Second) |
| + for _, ok := asked.Load("first@example.test"); !ok && time.Now().Before(deadline); _, ok = asked.Load("first@example.test") { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + for i := 0; i < 5; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("refused%d@example.test", i), "s", "b") |
| + } |
| + if got := len(q.jobs); got != 5 { |
| + t.Fatalf("queued %d refused links, want all 5: fill level must not depend on admission", got) |
| + } |
| +} |
| + |
| +// TestMailQueueSurvivesAPanic: a panic in the mailer is that message's |
| +// failure; the worker recovers, logs, and sends the next one (Astra, |
| +// round 2, P2). |
| +func TestMailQueueSurvivesAPanic(t *testing.T) { |
| + var mu sync.Mutex |
| + var sent []string |
| + calls := 0 |
| + s := funcSender(func(_ context.Context, to string) error { |
| + mu.Lock() |
| + defer mu.Unlock() |
| + calls++ |
| + if calls == 1 { |
| + panic("mailer exploded") |
| + } |
| + sent = append(sent, to) |
| + return nil |
| + }) |
| + q := newMailQueue(s, everyone, nil, slog.New(slog.DiscardHandler)) |
| + t.Cleanup(q.close) |
| + _ = q.Send(context.Background(), "first@example.test", "s", "b") |
| + _ = q.Send(context.Background(), "second@example.test", "s", "b") |
| + q.wait() |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if len(sent) != 1 || sent[0] != "second@example.test" { |
| + t.Fatalf("sent = %v, want the second link after the first panicked", sent) |
| + } |
| +} |
| + |
| +// TestMailQueueCloseIsBoundedAndStopsTheWorker: close drains within its |
| +// deadline, then cancels the send in flight, abandons the rest, and the |
| +// worker is gone — no goroutine outlives close (Astra, round 2, P2). |
| +func TestMailQueueCloseIsBoundedAndStopsTheWorker(t *testing.T) { |
| + baseline := runtime.NumGoroutine() |
| + honours := funcSender(func(ctx context.Context, _ string) error { |
| + <-ctx.Done() // a relay that never answers, but obeys cancellation |
| + return ctx.Err() |
| + }) |
| + q := newMailQueue(honours, everyone, nil, slog.New(slog.DiscardHandler)) |
| + q.drainTimeout = 50 * time.Millisecond |
| + for i := 0; i < 3; i++ { |
| + _ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b") |
| + } |
| + start := time.Now() |
| + q.close() |
| + if took := time.Since(start); took > 2*time.Second { |
| + t.Fatalf("close took %v, want about its 50ms deadline", took) |
| + } |
| + select { |
| + case <-q.done: |
| + default: |
| + t.Fatal("the worker was still running after close") |
| + } |
| + deadline := time.Now().Add(time.Second) |
| + for runtime.NumGoroutine() > baseline && time.Now().Before(deadline) { |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + if n := runtime.NumGoroutine(); n > baseline { |
| + t.Fatalf("%d goroutines after close, %d before: the queue leaked", n, baseline) |
| + } |
| +} |
| + |
| +// TestCloseSigninMailStopsTheWorker: every plugin NewAuth builds owns a |
| +// queue, and CloseSigninMail stops its worker — which is what every |
| +// test that builds one calls in its cleanup, so none leaves a worker |
| +// behind. |
| +func TestCloseSigninMailStopsTheWorker(t *testing.T) { |
| + app, _ := appWithOrigin(t, testOrigin) |
| + signinQueuesMu.Lock() |
| + q := signinQueues[app.auth] |
| + signinQueuesMu.Unlock() |
| + if q == nil { |
| + t.Fatal("NewAuth registered no mail queue for its plugin") |
| + } |
| + CloseSigninMail(app.auth) |
| + select { |
| + case <-q.done: |
| + case <-time.After(2 * time.Second): |
| + t.Fatal("the sign-in mail worker was still running after CloseSigninMail") |
| + } |
| +} |
diff --git a/internal/sheets/signin_test.go b/internal/sheets/signin_test.go| index 77a0993..3acb42e 100644 |
| --- a/internal/sheets/signin_test.go |
| +++ b/internal/sheets/signin_test.go |
| @@ -90,6 +90,7 @@ func testAuth(t *testing.T) *App { |
| if err != nil { |
| t.Fatalf("NewAuth: %v", err) |
| } |
| + t.Cleanup(func() { CloseSigninMail(a) }) |
| return New(Config{DB: d.G, Auth: a, Admit: Admits([]string{"paul@example.com"})}) |
| } |
diff --git a/internal/sheets/signin_timing_test.go b/internal/sheets/signin_timing_test.go| index abb85af..3e2a104 100644 |
| --- a/internal/sheets/signin_timing_test.go |
| +++ b/internal/sheets/signin_timing_test.go |
| @@ -62,6 +62,7 @@ func TestSigninNeverWaitsForTheMailer(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| + t.Cleanup(func() { CloseSigninMail(a) }) |
| app.UseAuth(a) |
| var once sync.Once |
| free := func() { once.Do(func() { close(bm.release) }) } |