amadan

oficina / meet Public

Keep taking sign-in links until the HTTP server has drained

Astra on #25, round 2: SIGTERM cancelled the app context, which closed
the mail queue at once, while Server.Shutdown was still letting accepted
requests finish. A request already in flight then answered ?sent=1 and
its link was refused; main delivered it.

- The queue is no longer closed when App's ctx ends. ctx still ends
  streams, the sweep and the tick at SIGTERM. main's deferred
  Runtime.Close, which runs after rastrillo.Serve returns and before the
  database closes, is now the production close, still bounded and
  joining.
- build closes the queue on its own error paths. NewForTest keeps the
  ctx-driven close for its TestHandle. Callers of App in callstest and
  of build in internal/calls (via buildForTest) close it in cleanup.
- TestShutdownStopsTheMailWorker now asserts the new contract: the ctx
  ending leaves the queue open, and close stops the worker.

Test, failing on 493ece6: TestSIGTERMDeliversAnInFlightSigninsLink
sends a forced POST /signin minus its last body byte, sends SIGTERM,
waits for "rastrillo: shutting down", then finishes the body 300 ms
later. It requires 303 ?sent=1, a clean exit, and the relay to have the
message. Before: answered ?sent=1, nothing delivered.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude pushed by paul@keymail.dev cf50d5ca79148b6473c6ac621fb6c27d645816f1 parent 493ece6
14 files changed, +233 −83
  • cmd/calls/main.go +4 −4
  • cmd/calls/shutdown_test.go +145 −48
  • internal/calls/app.go +24 −8
  • internal/calls/app_relay_test.go +1 −1
  • internal/calls/auth_test.go +1 −1
  • internal/calls/home_router_test.go +1 −1
  • internal/calls/keymail_signin_test.go +3 −2
  • internal/calls/signin_equal_work_test.go +1 −1
  • internal/calls/signin_mail_test.go +36 −10
  • internal/calls/signin_timing_test.go +1 −1
  • internal/callstest/config_test.go +9 −3
  • internal/callstest/main_test.go +2 −1
  • internal/callstest/schema_test.go +3 −1
  • internal/callstest/signin_test.go +2 −1

CI passed — run details

CI log 
=== vet ===
go vet ./...
=== fmt ===
=== test ===
go test ./...
ok  	amadan.net/oficina/calls/callscore	(cached)
ok  	amadan.net/oficina/calls/cmd/calls	1.628s
ok  	amadan.net/oficina/calls/cmd/quality-summary	(cached)
ok  	amadan.net/oficina/calls/internal/calls	3.173s
ok  	amadan.net/oficina/calls/internal/calls/apns	(cached)
ok  	amadan.net/oficina/calls/internal/calls/hub	(cached)
?   	amadan.net/oficina/calls/internal/calls/icons	[no test files]
?   	amadan.net/oficina/calls/internal/calls/keymailtest	[no test files]
ok  	amadan.net/oficina/calls/internal/calls/relay	(cached)
ok  	amadan.net/oficina/calls/internal/calls/seal	(cached)
ok  	amadan.net/oficina/calls/internal/callstest	10.727s
?   	amadan.net/oficina/calls/protocol	[no test files]
?   	amadan.net/oficina/calls/scripts/writeestilo	[no test files]
=== migration-check ===
go run amadan.net/rastrillo/rastrillo/cmd/rastrillo migration check
rastrillo migration check: models and migrations agree
=== race ===
CGO_ENABLED=1 go test -race ./internal/...
ok  	amadan.net/oficina/calls/internal/calls	12.836s
ok  	amadan.net/oficina/calls/internal/calls/apns	(cached)
ok  	amadan.net/oficina/calls/internal/calls/hub	(cached)
?   	amadan.net/oficina/calls/internal/calls/icons	[no test files]
?   	amadan.net/oficina/calls/internal/calls/keymailtest	[no test files]
ok  	amadan.net/oficina/calls/internal/calls/relay	(cached)
ok  	amadan.net/oficina/calls/internal/calls/seal	(cached)
ok  	amadan.net/oficina/calls/internal/callstest	31.626s
diff --git a/cmd/calls/main.go b/cmd/calls/main.go
index 1a9c2d6..42b768c 100644
--- a/cmd/calls/main.go
+++ b/cmd/calls/main.go
@@ -93,10 +93,10 @@ func main() {
os.Exit(1)
}
// After rastrillo.Serve returns and before d.Close() above (defers
- // run last-first): wait for the sign-in mail queue's bounded drain.
- // SIGTERM ends ctx, which starts that drain; this joins it, so a link
- // already answered ?sent=1 reaches the relay rather than dying with
- // the process.
+ // run last-first): close the sign-in mail queue and wait for its
+ // bounded drain. Not at SIGTERM: Serve's Shutdown is still letting
+ // accepted requests finish then, and a link one of them queues must
+ // reach the relay rather than be refused or die with the process.
defer rt.Close()
opts.Mux = mux
diff --git a/cmd/calls/shutdown_test.go b/cmd/calls/shutdown_test.go
index be82d3d..a368e36 100644
--- a/cmd/calls/shutdown_test.go
+++ b/cmd/calls/shutdown_test.go
@@ -9,6 +9,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "strconv"
"strings"
"sync"
"syscall"
@@ -38,7 +39,118 @@ func TestMain(m *testing.M) {
// cleanly, and only after the relay has the message.
func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) {
relay := newHeldRelay(t)
+ p := startCalls(t, relay.port)
+ // force=1: an emailed link without asking anyone whether
+ // example.test runs Keymail.
+ form := url.Values{"address": {"paul@example.test"}, "force": {"1"}}
+ req, _ := http.NewRequest(http.MethodPost, p.origin+"/signin", strings.NewReader(form.Encode()))
+ req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+ req.Header.Set("Origin", p.origin)
+ req.Header.Set("Sec-Fetch-Site", "same-origin")
+ res, err := p.client.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") {
+ t.Fatalf("POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location"))
+ }
+
+ select {
+ case <-relay.atData:
+ case <-time.After(10 * time.Second):
+ t.Fatal("the queued link never reached the relay")
+ }
+ if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+ t.Fatal(err)
+ }
+ // Long enough that a process which does not wait has gone (Astra's
+ // probe saw it exit in 4 ms), short against the 30 s drain window.
+ select {
+ case err := <-p.exited:
+ t.Fatalf("process exited (%v) while its acknowledged sign-in link was still with the relay", err)
+ case <-time.After(500 * time.Millisecond):
+ }
+ relay.release()
+ p.waitClean(t)
+ if !relay.accepted() {
+ t.Fatal("process exited without the relay accepting the sign-in link")
+ }
+}
+
+// TestSIGTERMDeliversAnInFlightSigninsLink: Server.Shutdown lets a
+// request it had already accepted finish, so that request's link must
+// still be queued and sent, not refused by a queue that SIGTERM closed
+// while the handler was still running (Astra on oficina/meet #25, round
+// 2: the request answered ?sent=1 and no mail went out).
+//
+// The POST's headers and all but the last byte of its body are sent
+// before SIGTERM, so the handler is in flight, still reading the form.
+// The last byte follows once the process has logged its shutdown.
+func TestSIGTERMDeliversAnInFlightSigninsLink(t *testing.T) {
+ relay := newHeldRelay(t)
+ relay.release() // an ordinary relay: this test is about intake
+ p := startCalls(t, relay.port)
+
+ body := url.Values{"address": {"paul@example.test"}, "force": {"1"}}.Encode()
+ conn, err := net.Dial("tcp", p.addr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+ head := "POST /signin HTTP/1.1\r\nHost: " + p.addr + "\r\nOrigin: " + p.origin +
+ "\r\nSec-Fetch-Site: same-origin\r\nContent-Type: application/x-www-form-urlencoded\r\n" +
+ "Content-Length: " + strconv.Itoa(len(body)) + "\r\nConnection: close\r\n\r\n"
+ if _, err := conn.Write([]byte(head + body[:len(body)-1])); err != nil {
+ t.Fatal(err)
+ }
+ // Let the server start the handler on what it has.
+ time.Sleep(200 * time.Millisecond)
+
+ if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {
+ t.Fatal(err)
+ }
+ until := time.Now().Add(10 * time.Second)
+ for !strings.Contains(p.out.String(), "rastrillo: shutting down") {
+ if time.Now().After(until) {
+ t.Fatal("the process never began shutting down")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ time.Sleep(300 * time.Millisecond)
+ if _, err := conn.Write([]byte(body[len(body)-1:])); err != nil {
+ t.Fatal(err)
+ }
+ _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+ res, err := http.ReadResponse(bufio.NewReader(conn), nil)
+ if err != nil {
+ t.Fatalf("the in-flight POST got no answer: %v", err)
+ }
+ res.Body.Close()
+ if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") {
+ t.Fatalf("in-flight POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location"))
+ }
+
+ p.waitClean(t)
+ if !relay.accepted() {
+ t.Fatal("the in-flight request was answered ?sent=1, but the process exited without sending its link")
+ }
+}
+
+// proc is a Calls process started by startCalls.
+type proc struct {
+ addr, origin string
+ cmd *exec.Cmd
+ exited chan error
+ out *lockedBuffer
+ client *http.Client
+}
+
+// startCalls runs this test binary as main() in development, mailing
+// through the relay on relayPort, and waits until it serves.
+func startCalls(t *testing.T, relayPort string) *proc {
+ t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
@@ -47,7 +159,7 @@ func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) {
ln.Close()
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
- defer cancel()
+ t.Cleanup(cancel)
dir := t.TempDir()
cmd := exec.CommandContext(ctx, os.Args[0], "-env", "development", "-addr", addr, "-db", filepath.Join(dir, "calls.db"))
cmd.Dir = dir
@@ -60,17 +172,18 @@ func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) {
cmd.Env = append(cmd.Env,
"CALLS_TEST_RUN_MAIN=1",
"CALLS_SMTP_HOST=127.0.0.1",
- "CALLS_SMTP_PORT="+relay.port,
+ "CALLS_SMTP_PORT="+relayPort,
"CALLS_SMTP_FROM=calls@example.test",
"CALLS_ALLOWED_ADDRESSES=paul@example.test",
)
- var out strings.Builder
- cmd.Stdout, cmd.Stderr = &out, &out
+ out := &lockedBuffer{}
+ cmd.Stdout, cmd.Stderr = out, out
if err := cmd.Start(); err != nil {
t.Fatal(err)
}
- exited := make(chan error, 1)
- go func() { exited <- cmd.Wait() }()
+ p := &proc{addr: addr, origin: "http://" + addr, cmd: cmd, exited: make(chan error, 1), out: out,
+ client: &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}}
+ go func() { p.exited <- cmd.Wait() }()
t.Cleanup(func() {
_ = cmd.Process.Kill()
if t.Failed() {
@@ -78,67 +191,51 @@ func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) {
}
})
- origin := "http://" + addr
- client := &http.Client{Timeout: 5 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
up := time.Now().Add(20 * time.Second)
for {
- res, err := client.Get(origin + "/healthz")
+ res, err := p.client.Get(p.origin + "/healthz")
if err == nil {
res.Body.Close()
- break
+ return p
}
if time.Now().After(up) {
t.Fatalf("server never came up: %v", err)
}
time.Sleep(20 * time.Millisecond)
}
+}
- // force=1: an emailed link without asking anyone whether
- // example.test runs Keymail.
- form := url.Values{"address": {"paul@example.test"}, "force": {"1"}}
- req, _ := http.NewRequest(http.MethodPost, origin+"/signin", strings.NewReader(form.Encode()))
- req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
- req.Header.Set("Origin", origin)
- req.Header.Set("Sec-Fetch-Site", "same-origin")
- res, err := client.Do(req)
- if err != nil {
- t.Fatal(err)
- }
- res.Body.Close()
- if res.StatusCode != http.StatusSeeOther || !strings.Contains(res.Header.Get("Location"), "sent=1") {
- t.Fatalf("POST /signin: %d %q, want 303 ?sent=1", res.StatusCode, res.Header.Get("Location"))
- }
-
- select {
- case <-relay.atData:
- case <-time.After(10 * time.Second):
- t.Fatal("the queued link never reached the relay")
- }
- if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
- t.Fatal(err)
- }
- // Long enough that a process which does not wait has gone (Astra's
- // probe saw it exit in 4 ms), short against the 30 s drain window.
- select {
- case err := <-exited:
- t.Fatalf("process exited (%v) while its acknowledged sign-in link was still with the relay", err)
- case <-time.After(500 * time.Millisecond):
- }
- relay.release()
-
+// waitClean waits for the process to exit, and requires it to exit 0.
+func (p *proc) waitClean(t *testing.T) {
+ t.Helper()
select {
- case err := <-exited:
+ case err := <-p.exited:
if err != nil {
t.Fatalf("process exit: %v", err)
}
case <-time.After(20 * time.Second):
- t.Fatal("process did not exit after the relay accepted the message")
- }
- if !relay.accepted() {
- t.Fatal("process exited without the relay accepting the sign-in link")
+ t.Fatal("process did not exit")
}
}
+// lockedBuffer is the process's output, readable while it is written.
+type lockedBuffer struct {
+ mu sync.Mutex
+ b strings.Builder
+}
+
+func (l *lockedBuffer) Write(p []byte) (int, error) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.b.Write(p)
+}
+
+func (l *lockedBuffer) String() string {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.b.String()
+}
+
// heldRelay is a plaintext SMTP server on loopback (no STARTTLS, no
// AUTH, which mail.SMTP accepts when no credentials are set) that holds
// the first message at DATA until release.
diff --git a/internal/calls/app.go b/internal/calls/app.go
index e70f4cc..1c1cf4c 100644
--- a/internal/calls/app.go
+++ b/internal/calls/app.go
@@ -439,22 +439,30 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err
return nil, nil, err
}
// Sign-in links go out after the response, from a worker that asks
- // authorize first (signin_mail.go); it drains when ctx ends.
+ // authorize first (signin_mail.go). It is NOT closed when ctx ends:
+ // ctx ends at SIGTERM, while Server.Shutdown is still letting
+ // accepted requests finish, and a request that reaches the queue
+ // after that must still have its link sent (Astra on #25, round 2).
+ // Whoever built the app closes it once no request can arrive:
+ // Runtime.Close (main, after rastrillo.Serve returns) or the
+ // TestHandle's stop. Only a failed build closes it here.
a.signinMail = newMailQueue(a.cfg.mailer, a.authorize, mayMail, a.logger)
if w, ok := cfg.Mailer.(queueAttacher); ok {
w.AttachSigninQueue(a.signinMail.wait)
}
- go func() { <-ctx.Done(); a.signinMail.close() }()
authPlugin, err := a.newAuthPlugin(writer)
if err != nil {
+ a.signinMail.close()
return nil, nil, err
}
a.auth = authPlugin
if err := a.wireHome(); err != nil {
+ a.signinMail.close()
return nil, nil, err
}
if err := a.wireNativeAuth(); err != nil {
+ a.signinMail.close()
return nil, nil, err
}
@@ -594,12 +602,15 @@ type Runtime struct {
// ErrorPage answers rastrillo.Options.ErrorPage: the app's own error
// page for a panic outside the router.
ErrorPage func(w http.ResponseWriter, r *http.Request, status int, ref string)
- // Close drains the sign-in mail queue, within its bounded deadline,
- // and returns when that is done. main calls it after rastrillo.Serve
- // returns and before the database closes: SMTP runs after the
- // response, so Server.Shutdown does not wait for it, and a link
- // already answered ?sent=1 would otherwise die with the process. The
- // app's context ending starts the same drain; Close joins it.
+ // Close stops the sign-in mail queue taking links and drains it,
+ // within its bounded deadline, returning when that is done. The
+ // queue is NOT closed when App's ctx ends: that happens at SIGTERM,
+ // while Server.Shutdown still lets accepted requests finish, and
+ // their links must go out. So main calls Close after rastrillo.Serve
+ // returns (no request can arrive any more) and before the database
+ // closes; SMTP runs after the response, so Shutdown does not wait
+ // for it. Every caller of App must call Close, or the queue's worker
+ // outlives the app. A second call joins the first.
Close func()
}
@@ -653,6 +664,8 @@ func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, e
}
go a.sweepForever(ctx)
go a.tickForever(ctx)
+ // ctx's end stops streams, the sweep and the tick; the sign-in mail
+ // queue keeps taking links until Runtime.Close (see build).
return mux, &Runtime{NextDue: a.nextDue, ErrorPage: a.ErrorPage, Close: a.signinMail.close}, nil
}
@@ -666,5 +679,8 @@ func NewForTest(d *db.DB, cfg Config) (*http.ServeMux, *TestHandle, error) {
cancel()
return nil, nil, err
}
+ // A test's stop (ShutdownForTest) ends the app outright, mail queue
+ // included: no server is draining requests in front of it.
+ go func() { <-ctx.Done(); a.signinMail.close() }()
return mux, &TestHandle{a: a, stop: cancel}, nil
}
diff --git a/internal/calls/app_relay_test.go b/internal/calls/app_relay_test.go
index b8d0402..5266d34 100644
--- a/internal/calls/app_relay_test.go
+++ b/internal/calls/app_relay_test.go
@@ -63,7 +63,7 @@ func TestBuildLogsOneINFOLineWithTheRegionCountAndNotTheKey(t *testing.T) {
cfg.Logger = logger
cfg.Relay = relayVector(t)
d := openRelayTestDB(t, logger)
- _, a, err := build(t.Context(), d, cfg)
+ _, a, err := buildForTest(t, d, cfg)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/calls/auth_test.go b/internal/calls/auth_test.go
index 4375e98..77b02de 100644
--- a/internal/calls/auth_test.go
+++ b/internal/calls/auth_test.go
@@ -23,7 +23,7 @@ func testApp(t *testing.T, members ...string) *app {
t.Fatal(err)
}
t.Cleanup(func() { d.Close() })
- _, a, err := build(t.Context(), d, Config{
+ _, a, err := buildForTest(t, d, Config{
Env: EnvDevelopment, Origin: "http://app.test",
AllowedAddresses: members, Logger: logger,
})
diff --git a/internal/calls/home_router_test.go b/internal/calls/home_router_test.go
index 47e976e..d8d15e3 100644
--- a/internal/calls/home_router_test.go
+++ b/internal/calls/home_router_test.go
@@ -66,7 +66,7 @@ func homeKeysBase64(key *assertioncrypto.Keypair) string {
func newHomeRouterFixtureWith(t *testing.T, d *db.DB, logger *slog.Logger, key *assertioncrypto.Keypair, creds *connect.Credentials) *homeRouterFixture {
t.Helper()
mail := &homeTestMail{}
- mux, a, err := build(t.Context(), d, Config{Origin: homeTestOrigin, InstanceKey: "test-only-instance", DEK: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)), AllowedAddresses: []string{"alice@example.test"}, Mailer: mail, Logger: logger, Home: creds})
+ mux, a, err := buildForTest(t, d, Config{Origin: homeTestOrigin, InstanceKey: "test-only-instance", DEK: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)), AllowedAddresses: []string{"alice@example.test"}, Mailer: mail, Logger: logger, Home: creds})
if err != nil {
t.Fatal(err)
}
diff --git a/internal/calls/keymail_signin_test.go b/internal/calls/keymail_signin_test.go
index 9ba5be4..f14a34a 100644
--- a/internal/calls/keymail_signin_test.go
+++ b/internal/calls/keymail_signin_test.go
@@ -36,7 +36,8 @@ func TestMain(m *testing.M) {
}
code := m.Run()
// No test may leave a sign-in mail worker running: every app a test
- // builds ends with its context.
+ // builds has its queue closed in cleanup (Runtime.Close, a
+ // TestHandle's stop, or buildForTest).
if code == 0 {
deadline := time.Now().Add(5 * time.Second)
for mailWorkers() > 0 && time.Now().Before(deadline) {
@@ -118,7 +119,7 @@ func newKeymailFixture(t *testing.T, home bool) *keymailFixture {
}
cfg.Home = &connect.Credentials{Issuer: homeTestIssuer, ConsumerID: "calls-test", KeysBase64: homeKeysBase64(key), DEKBase64: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32))}
}
- mux, _, err := build(t.Context(), d, cfg)
+ mux, _, err := buildForTest(t, d, cfg)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/calls/signin_equal_work_test.go b/internal/calls/signin_equal_work_test.go
index d4fd9ca..20b6296 100644
--- a/internal/calls/signin_equal_work_test.go
+++ b/internal/calls/signin_equal_work_test.go
@@ -49,7 +49,7 @@ func TestMemberAndOutsiderDoTheSameWorkOverBudget(t *testing.T) {
}
t.Cleanup(func() { d.Close() })
mail := &countingMailer{}
- mux, a, err := build(t.Context(), d, Config{Origin: keymailOrigin, InstanceKey: "test-only-instance",
+ mux, a, err := buildForTest(t, d, Config{Origin: keymailOrigin, InstanceKey: "test-only-instance",
DEK: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)), AllowedAddresses: []string{plainMember}, Mailer: mail, Logger: logger})
if err != nil {
t.Fatal(err)
diff --git a/internal/calls/signin_mail_test.go b/internal/calls/signin_mail_test.go
index 143462f..999a8ad 100644
--- a/internal/calls/signin_mail_test.go
+++ b/internal/calls/signin_mail_test.go
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
+ "net/http"
"path/filepath"
"runtime"
"strings"
@@ -189,12 +190,12 @@ func TestMailQueueCloseIsBoundedAndStopsTheWorker(t *testing.T) {
}
}
-// TestMailQueueCloseJoinsAnInProgressClose: the app closes the queue
-// when its context ends, and main closes it again before it closes the
-// database and exits. The second close must wait for the first one's
-// drain, not return because the queue is already marked closed, or the
-// process exits with an acknowledged link still at the relay (Astra on
-// oficina/meet #25).
+// TestMailQueueCloseJoinsAnInProgressClose: close can be reached twice
+// (a TestHandle's stop and a cleanup, say). The second close must wait
+// for the first one's drain, not return because the queue is already
+// marked closed, or a caller about to close the database or exit would
+// leave an acknowledged link still at the relay (Astra on oficina/meet
+// #25).
func TestMailQueueCloseJoinsAnInProgressClose(t *testing.T) {
release, sending := make(chan struct{}), make(chan struct{})
var sent atomic.Bool
@@ -319,9 +320,25 @@ func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) {
}
}
-// TestShutdownStopsTheMailWorker: the queue lives as long as the app's
-// context, and ending it stops the worker, so no test that cancels its
-// app leaves one running.
+// buildForTest is build on the test's context, with the sign-in mail
+// queue closed in cleanup: the context ending no longer closes it (see
+// build), so whoever builds an app closes its queue.
+func buildForTest(t *testing.T, d *db.DB, cfg Config) (*http.ServeMux, *app, error) {
+ t.Helper()
+ mux, a, err := build(t.Context(), d, cfg)
+ if err == nil {
+ t.Cleanup(a.signinMail.close)
+ }
+ return mux, a, err
+}
+
+// TestShutdownStopsTheMailWorker: the app's context ending (SIGTERM)
+// stops streams and background work but NOT mail intake, because
+// Server.Shutdown is still letting accepted requests finish and their
+// links must still go out (Astra on #25, round 2). The queue takes links
+// until its owner closes it (Runtime.Close, or a TestHandle's stop), and
+// that close stops the worker, so nothing that closes its app leaves
+// one running.
func TestShutdownStopsTheMailWorker(t *testing.T) {
logger := slog.New(slog.DiscardHandler)
d, err := db.Open(filepath.Join(t.TempDir(), "calls.db"), logger)
@@ -335,10 +352,19 @@ func TestShutdownStopsTheMailWorker(t *testing.T) {
t.Fatal(err)
}
q := a.signinMail
+ t.Cleanup(q.close)
cancel()
+ time.Sleep(100 * time.Millisecond)
+ q.mu.RLock()
+ closed := q.closed
+ q.mu.RUnlock()
+ if closed {
+ t.Fatal("the app's context ending closed the sign-in mail queue while requests may still be finishing")
+ }
+ q.close()
select {
case <-q.done:
case <-time.After(2 * time.Second):
- t.Fatal("the sign-in mail worker was still running after the app's context ended")
+ t.Fatal("the sign-in mail worker was still running after close")
}
}
diff --git a/internal/calls/signin_timing_test.go b/internal/calls/signin_timing_test.go
index e17a88f..707a18d 100644
--- a/internal/calls/signin_timing_test.go
+++ b/internal/calls/signin_timing_test.go
@@ -55,7 +55,7 @@ func TestSigninNeverWaitsForTheMailer(t *testing.T) {
var once sync.Once
free := func() { once.Do(func() { close(bm.release) }) }
t.Cleanup(free)
- mux, _, err := build(t.Context(), d, Config{Origin: keymailOrigin, InstanceKey: "test-only-instance",
+ mux, _, err := buildForTest(t, d, Config{Origin: keymailOrigin, InstanceKey: "test-only-instance",
DEK: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)), AllowedAddresses: []string{plainMember}, Mailer: bm, Logger: logger})
if err != nil {
t.Fatal(err)
diff --git a/internal/callstest/config_test.go b/internal/callstest/config_test.go
index 7846e15..98828e1 100644
--- a/internal/callstest/config_test.go
+++ b/internal/callstest/config_test.go
@@ -47,20 +47,26 @@ func TestProductionRefusesToStartWithoutSecrets(t *testing.T) {
"no allowlist": {Env: "production", Origin: "https://x.test", InstanceKey: "k", DEK: validDEK, Mailer: nopMailer{}},
}
for name, cfg := range cases {
- if _, _, err := calls.App(testCtx(t), openDB(t), cfg); err == nil {
+ if _, rt, err := calls.App(testCtx(t), openDB(t), cfg); err == nil {
+ rt.Close()
t.Errorf("%s: production started without it", name)
}
}
}
func TestDevelopmentStartsWithNothing(t *testing.T) {
- if _, _, err := calls.App(testCtx(t), openDB(t), calls.Config{Env: calls.EnvDevelopment}); err != nil {
+ _, rt, err := calls.App(testCtx(t), openDB(t), calls.Config{Env: calls.EnvDevelopment})
+ if err != nil {
t.Fatalf("development refused: %v", err)
}
+ t.Cleanup(rt.Close)
}
func TestBadDEKIsRefusedEvenInDevelopment(t *testing.T) {
- _, _, err := calls.App(testCtx(t), openDB(t), calls.Config{Env: calls.EnvDevelopment, DEK: "not base64!"})
+ _, rt, err := calls.App(testCtx(t), openDB(t), calls.Config{Env: calls.EnvDevelopment, DEK: "not base64!"})
+ if err == nil {
+ rt.Close()
+ }
if err == nil || !strings.Contains(err.Error(), "base64") {
t.Fatalf("want a base64 refusal, got %v", err)
}
diff --git a/internal/callstest/main_test.go b/internal/callstest/main_test.go
index c339a28..8abb9fc 100644
--- a/internal/callstest/main_test.go
+++ b/internal/callstest/main_test.go
@@ -24,7 +24,8 @@ func TestMain(m *testing.M) {
}
code := m.Run()
// No test may leave a sign-in mail worker running: every app a test
- // builds ends with its context.
+ // builds has its queue closed in cleanup (Runtime.Close, a
+ // TestHandle's stop, or buildForTest).
if code == 0 {
deadline := time.Now().Add(5 * time.Second)
for mailWorkers() > 0 && time.Now().Before(deadline) {
diff --git a/internal/callstest/schema_test.go b/internal/callstest/schema_test.go
index 35416ac..5cb4bd1 100644
--- a/internal/callstest/schema_test.go
+++ b/internal/callstest/schema_test.go
@@ -8,9 +8,11 @@ import (
func TestBootCreatesEveryTable(t *testing.T) {
d := openDB(t)
- if _, _, err := calls.App(testCtx(t), d, calls.Config{Env: calls.EnvDevelopment}); err != nil {
+ _, rt, err := calls.App(testCtx(t), d, calls.Config{Env: calls.EnvDevelopment})
+ if err != nil {
t.Fatal(err)
}
+ t.Cleanup(rt.Close)
for _, table := range []string{"persons", "meetings", "meeting_links", "guest_grants", "pending_joins", "sessions", "auth_links"} {
var n int
if err := d.G.Raw("SELECT count(*) FROM sqlite_master WHERE type='table' AND name=?", table).Scan(&n).Error; err != nil {
diff --git a/internal/callstest/signin_test.go b/internal/callstest/signin_test.go
index b3caa4e..f4281c1 100644
--- a/internal/callstest/signin_test.go
+++ b/internal/callstest/signin_test.go
@@ -89,13 +89,14 @@ func TestSigninOverAFailingMailerLeavesNoAddressInTheLog(t *testing.T) {
t.Cleanup(func() { d.Close() })
failing := &failingMailer{err: fmt.Errorf("rastrillo/mail: send to %s: %w", address, errFakeSMTPFailure)}
- mux, _, err := calls.App(testCtx(t), d, calls.Config{
+ mux, rt, err := calls.App(testCtx(t), d, calls.Config{
Env: calls.EnvDevelopment, Origin: testOrigin, AllowedAddresses: []string{address},
Mailer: failing, Logger: logger,
})
if err != nil {
t.Fatal(err)
}
+ t.Cleanup(rt.Close) // App's queue outlives its ctx until Close
rec := post(t, mux, "/signin", url.Values{"address": {address}})
if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/signin?sent=1" {