amadan

oficina / meet Public

Wait for the sign-in mail drain before the process exits

Astra on #25: SMTP runs after the response, so Server.Shutdown no
longer waits for it, and main returned (closing the database) while the
app's context goroutine was still draining the queue. A link already
answered ?sent=1 died with the process on an ordinary SIGTERM.

- mailQueue.close now joins a close already in progress: a second call
  returns when the first does, never earlier.
- Runtime.Close exposes it; main defers it after rastrillo.Serve
  returns and before d.Close(), bounded by the queue's drain deadline.

Tests, both failing on 6b18d00:
- TestSIGTERMDrainsQueuedSigninMail runs main() as a real process
  against a loopback SMTP relay that holds the message at DATA, sends
  SIGTERM, then lets the relay accept it. Before: the process exited in
  ~10 ms with nothing delivered.
- TestMailQueueCloseJoinsAnInProgressClose: a second close returned
  while the first was still draining.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude pushed by paul@keymail.dev 493ece6f807b1546675a2da072c934d6d968b6a0 parent 6b18d00
5 files changed, +303 −6
  • cmd/calls/main.go +6 −0
  • cmd/calls/shutdown_test.go +225 −0
  • internal/calls/app.go +11 −4
  • internal/calls/signin_mail.go +10 −2
  • internal/calls/signin_mail_test.go +51 −0

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	0.619s
ok  	amadan.net/oficina/calls/cmd/quality-summary	(cached)
ok  	amadan.net/oficina/calls/internal/calls	2.000s
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.465s
?   	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.587s
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.545s
diff --git a/cmd/calls/main.go b/cmd/calls/main.go
index 1cd452d..1a9c2d6 100644
--- a/cmd/calls/main.go
+++ b/cmd/calls/main.go
@@ -92,6 +92,12 @@ func main() {
logger.Error("build app", "err", err)
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.
+ defer rt.Close()
opts.Mux = mux
opts.DBPath = ""
diff --git a/cmd/calls/shutdown_test.go b/cmd/calls/shutdown_test.go
new file mode 100644
index 0000000..be82d3d
--- /dev/null
+++ b/cmd/calls/shutdown_test.go
@@ -0,0 +1,225 @@
+package main
+
+import (
+ "bufio"
+ "context"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+)
+
+// TestMain lets TestSIGTERMDrainsQueuedSigninMail run this binary as
+// the real process: with CALLS_TEST_RUN_MAIN=1 it is main() and nothing
+// else, so os.Args are the activation flags the test passes.
+func TestMain(m *testing.M) {
+ if os.Getenv("CALLS_TEST_RUN_MAIN") == "1" {
+ main()
+ os.Exit(0)
+ }
+ os.Exit(m.Run())
+}
+
+// TestSIGTERMDrainsQueuedSigninMail: SMTP runs after the response, in
+// the sign-in mail queue, so Server.Shutdown no longer waits for it.
+// A link that was already answered ?sent=1 must still go out when the
+// process is stopped: main has to wait for the queue's bounded drain
+// before it closes the database and exits (Astra on oficina/meet #25).
+//
+// The relay holds the message at DATA until after SIGTERM has arrived,
+// then accepts it well inside the drain window. The process must exit
+// cleanly, and only after the relay has the message.
+func TestSIGTERMDrainsQueuedSigninMail(t *testing.T) {
+ relay := newHeldRelay(t)
+
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ addr := ln.Addr().String()
+ ln.Close()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer cancel()
+ dir := t.TempDir()
+ cmd := exec.CommandContext(ctx, os.Args[0], "-env", "development", "-addr", addr, "-db", filepath.Join(dir, "calls.db"))
+ cmd.Dir = dir
+ for _, e := range os.Environ() {
+ if strings.HasPrefix(e, "CALLS_") || strings.HasPrefix(e, "CARLOS_") || strings.HasPrefix(e, "HOME_") {
+ continue
+ }
+ cmd.Env = append(cmd.Env, e)
+ }
+ cmd.Env = append(cmd.Env,
+ "CALLS_TEST_RUN_MAIN=1",
+ "CALLS_SMTP_HOST=127.0.0.1",
+ "CALLS_SMTP_PORT="+relay.port,
+ "CALLS_SMTP_FROM=calls@example.test",
+ "CALLS_ALLOWED_ADDRESSES=paul@example.test",
+ )
+ var out strings.Builder
+ 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() }()
+ t.Cleanup(func() {
+ _ = cmd.Process.Kill()
+ if t.Failed() {
+ t.Logf("process output:\n%s", out.String())
+ }
+ })
+
+ 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")
+ if err == nil {
+ res.Body.Close()
+ break
+ }
+ 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()
+
+ select {
+ case err := <-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")
+ }
+}
+
+// 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.
+type heldRelay struct {
+ port string
+ atData chan struct{}
+ hold chan struct{}
+ once sync.Once
+ mu sync.Mutex
+ gotMail bool
+}
+
+func newHeldRelay(t *testing.T) *heldRelay {
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { ln.Close() })
+ _, port, _ := net.SplitHostPort(ln.Addr().String())
+ r := &heldRelay{port: port, atData: make(chan struct{}), hold: make(chan struct{})}
+ t.Cleanup(r.release)
+ go func() {
+ var signalled sync.Once
+ for {
+ c, err := ln.Accept()
+ if err != nil {
+ return
+ }
+ go r.serve(c, &signalled)
+ }
+ }()
+ return r
+}
+
+func (r *heldRelay) release() { r.once.Do(func() { close(r.hold) }) }
+
+func (r *heldRelay) accepted() bool {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return r.gotMail
+}
+
+func (r *heldRelay) serve(c net.Conn, signalled *sync.Once) {
+ defer c.Close()
+ br := bufio.NewReader(c)
+ say := func(s string) { _, _ = c.Write([]byte(s + "\r\n")) }
+ say("220 relay.test ESMTP")
+ for {
+ line, err := br.ReadString('\n')
+ if err != nil {
+ return
+ }
+ cmd := strings.ToUpper(strings.TrimSpace(line))
+ switch {
+ case strings.HasPrefix(cmd, "EHLO"):
+ say("250-relay.test")
+ say("250 8BITMIME")
+ case strings.HasPrefix(cmd, "HELO"), strings.HasPrefix(cmd, "MAIL"), strings.HasPrefix(cmd, "RCPT"), strings.HasPrefix(cmd, "RSET"), strings.HasPrefix(cmd, "NOOP"):
+ say("250 OK")
+ case cmd == "DATA":
+ signalled.Do(func() { close(r.atData) })
+ <-r.hold
+ say("354 go ahead")
+ for {
+ l, err := br.ReadString('\n')
+ if err != nil {
+ return
+ }
+ if l == ".\r\n" || l == ".\n" {
+ break
+ }
+ }
+ r.mu.Lock()
+ r.gotMail = true
+ r.mu.Unlock()
+ say("250 queued")
+ case cmd == "QUIT":
+ say("221 bye")
+ return
+ default:
+ say("502 unrecognised")
+ }
+ }
+}
diff --git a/internal/calls/app.go b/internal/calls/app.go
index 2c0c95d..e70f4cc 100644
--- a/internal/calls/app.go
+++ b/internal/calls/app.go
@@ -584,9 +584,9 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err
}
// Runtime is what App hands main beyond the mux: the platform's
-// scheduled-wake function (review C2-2). Nothing else lives here yet;
-// it is a struct rather than a bare func so a later addition (the
-// sidecar pass, a shutdown func) does not change App's signature again.
+// scheduled-wake function (review C2-2), the error page, and the
+// shutdown drain. It is a struct rather than a bare func so a later
+// addition does not change App's signature again.
type Runtime struct {
// NextDue answers rastrillo.Options.NextDue: the next moment this
// app has work, or the zero time when nothing is scheduled.
@@ -594,6 +594,13 @@ 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 func()
}
// nextDue is Runtime.NextDue's implementation: the earliest of the next
@@ -646,7 +653,7 @@ func App(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *Runtime, e
}
go a.sweepForever(ctx)
go a.tickForever(ctx)
- return mux, &Runtime{NextDue: a.nextDue, ErrorPage: a.ErrorPage}, nil
+ return mux, &Runtime{NextDue: a.nextDue, ErrorPage: a.ErrorPage, Close: a.signinMail.close}, nil
}
// NewForTest builds the app and returns the handler and its test
diff --git a/internal/calls/signin_mail.go b/internal/calls/signin_mail.go
index 4340e88..27fba4d 100644
--- a/internal/calls/signin_mail.go
+++ b/internal/calls/signin_mail.go
@@ -50,7 +50,9 @@ import (
// is abandoned and counted, 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; that is counted in stuck and logged, best
-// effort, too.
+// effort, too. close may be called more than once (the app's context
+// ending, then main on its way out): a later call waits for the first
+// one to finish, so whoever returns from close knows the drain is over.
type mailQueue struct {
next mail.Sender
admits func(to string) bool
@@ -71,6 +73,7 @@ type mailQueue struct {
ctx context.Context // cancelled when close gives up waiting
cancel context.CancelFunc
done chan struct{}
+ shut chan struct{} // closed when the first close returns
logMu sync.RWMutex // logf reads logsClosed; the worker's exit sets it
logsClosed bool
@@ -118,7 +121,7 @@ func newMailQueueLogging(next mail.Sender, admits func(string) bool, gate func(c
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{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog}
+ done: make(chan struct{}), shut: make(chan struct{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog}
go q.pump()
go q.run()
return q
@@ -255,15 +258,20 @@ func (q *mailQueue) wait() {
// worker (see the type's comment). It never waits on the log pump:
// shutdown deliberately does not wait for a write blocked inside the
// logger, which can only finish or die with the process (see pump).
+//
+// A second close joins the first: it returns when that one does, never
+// earlier, so main can wait for a drain the app's context began.
func (q *mailQueue) close() {
q.mu.Lock()
if q.closed {
q.mu.Unlock()
+ <-q.shut
return
}
q.closed = true
close(q.jobs)
q.mu.Unlock()
+ defer close(q.shut)
defer q.cancel()
select {
case <-q.done:
diff --git a/internal/calls/signin_mail_test.go b/internal/calls/signin_mail_test.go
index 923dfff..143462f 100644
--- a/internal/calls/signin_mail_test.go
+++ b/internal/calls/signin_mail_test.go
@@ -8,6 +8,7 @@ import (
"runtime"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -188,6 +189,56 @@ 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).
+func TestMailQueueCloseJoinsAnInProgressClose(t *testing.T) {
+ release, sending := make(chan struct{}), make(chan struct{})
+ var sent atomic.Bool
+ slow := funcSender(func(context.Context, string) error {
+ close(sending)
+ <-release
+ sent.Store(true)
+ return nil
+ })
+ q := newMailQueue(slow, everyone, nil, slog.New(slog.DiscardHandler))
+ _ = q.Send(context.Background(), "paul@example.test", "s", "b")
+ <-sending
+
+ first := make(chan struct{})
+ go func() { q.close(); close(first) }()
+ for {
+ q.mu.RLock()
+ closed := q.closed
+ q.mu.RUnlock()
+ if closed {
+ break
+ }
+ time.Sleep(time.Millisecond)
+ }
+ second := make(chan struct{})
+ go func() { q.close(); close(second) }()
+ select {
+ case <-second:
+ t.Fatal("a second close returned while the first was still draining")
+ case <-time.After(100 * time.Millisecond):
+ }
+ close(release)
+ for _, c := range []chan struct{}{first, second} {
+ select {
+ case <-c:
+ case <-time.After(5 * time.Second):
+ t.Fatal("close did not return after the send finished")
+ }
+ }
+ if !sent.Load() {
+ t.Fatal("the queued link was not sent before close returned")
+ }
+}
+
// mailWorkers counts goroutines running a sign-in mail queue's worker
// or log pump — the targeted leak check (goleak's idea, without the
// dependency) TestMain runs after the package's tests.