38ad0b7 · oficina/meet

diff --git a/AGENTS.md b/AGENTS.md

index 5420d63..a256bde 100644

--- a/AGENTS.md

+++ b/AGENTS.md

@@ -32,8 +32,18 @@ mechanically.

- `internal/calls/store.go` — every database access. `standing.go` is

the entry authority (spec §2.3), computed per request, never stored.

- Sign-in: `auth.go` is Docs' shape with the pending-join exception of

- spec §3.2 and Slides' same-answer wrapper. Every answer from

- `POST /signin` is `303 /signin?sent=1`.

+ spec §3.2 and Slides' same-answer wrapper. Keymail or an emailed link

+ is chosen by the address's domain alone, never by membership (Paul,

+ 2026-10-06). A Keymail address, member or not, gets the continuation

+ page to Keymail (never a 303 out of the form POST: Safari applies

+ `form-action` to it); every other answer from `POST /signin` is

+ `303 /signin?sent=1`, and the link is mailed after the response by

+ `signin_mail.go`'s worker, only if this request may have one, so

+ neither the answer nor its timing says who is a member. A refusal

+ after verification is the door, rendered in the refusal's own 403

+ (never a query anyone could type). `mayRequestLink` does the same

+ database work for a member and an outsider and logs nothing on the

+ response path.

## Rules the family holds hard

diff --git a/cmd/calls/main.go b/cmd/calls/main.go

index 1cd452d..21b2f23 100644

--- a/cmd/calls/main.go

+++ b/cmd/calls/main.go

@@ -30,6 +30,20 @@ func main() {

// is a JSON object.

reportLogger := slog.New(slog.NewJSONHandler(stderr, nil))

+ // The only os.Exit after startup. run owns every resource and

+ // returns its error only after its own defers have run: the sign-in

+ // mail queue's bounded drain, then the database close. An os.Exit

+ // inside it would skip them, and a failed HTTP shutdown (rastrillo's

+ // ten-second deadline passing with a request still running) used to

+ // cost a link already answered ?sent=1 (Astra on #25, round 3).

+ if err := run(logger, reportLogger); err != nil {

+ logger.Error("calls: exiting", "err", err)

+ os.Exit(1)

+ }

+}

+

+// run is main's body. Every failure returns, never exits; see main.

+func run(logger, reportLogger *slog.Logger) error {

// -env is stripped before Resolve sees argv, because Resolve parses

// os.Args itself for -socket, -addr and -db and a second flag set

// would refuse the platform's own activation arguments. Only the

@@ -37,22 +51,19 @@ func main() {

// misconfigured box relax every refusal (config.go).

envFlag, rest, err := extractEnvFlag(os.Args[1:])

if err != nil {

- logger.Error("read flags", "err", err)

- os.Exit(1)

+ return fmt.Errorf("read flags: %w", err)

}

os.Args = append(os.Args[:1], rest...)

env := &envFlag

opts, err := rastrillo.Resolve(rastrillo.Options{DBPath: "calls.db", Logger: logger})

if err != nil {

- logger.Error("resolve activation", "err", err)

- os.Exit(1)

+ return fmt.Errorf("resolve activation: %w", err)

}

d, err := db.Open(opts.DBPath, logger)

if err != nil {

- logger.Error("open database", "err", err)

- os.Exit(1)

+ return fmt.Errorf("open database: %w", err)

}

defer d.Close()

@@ -89,9 +100,20 @@ func main() {

mux, rt, err := calls.App(ctx, d, cfg)

if err != nil {

- logger.Error("build app", "err", err)

- os.Exit(1)

+ return fmt.Errorf("build app: %w", err)

}

+ // After rastrillo.Serve returns, however it returns, and before

+ // d.Close() above (defers 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.

+ //

+ // The cutoff, when Shutdown times out: a link queued before this

+ // close is drained; a handler still running past rastrillo's

+ // deadline that reaches the queue afterwards is refused, and dies

+ // with the process like the rest of its request.

+ defer rt.Close()

opts.Mux = mux

opts.DBPath = ""

@@ -104,9 +126,9 @@ func main() {

// rastrillo's plain "Something went wrong." (shared chrome §3.3).

opts.ErrorPage = rt.ErrorPage

if err := rastrillo.Serve(opts); err != nil {

- logger.Error("serve failed", "err", err)

- os.Exit(1)

+ return fmt.Errorf("serve failed: %w", err)

}

+ return nil

}

// smtpSetting reads one SMTP setting, this app's own variable first and

diff --git a/cmd/calls/shutdown_test.go b/cmd/calls/shutdown_test.go

new file mode 100644

index 0000000..3d51390

--- /dev/null

+++ b/cmd/calls/shutdown_test.go

@@ -0,0 +1,403 @@

+package main

+

+import (

+ "bufio"

+ "context"

+ "errors"

+ "net"

+ "net/http"

+ "net/url"

+ "os"

+ "os/exec"

+ "path/filepath"

+ "strconv"

+ "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)

+ 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")

+ }

+}

+

+// TestSIGTERMDrainsMailWhenHTTPShutdownTimesOut: rastrillo.Serve

+// returns an error when its ten-second Server.Shutdown deadline passes

+// with a request still running, and main used to os.Exit(1) on it,

+// skipping the deferred drain. An unrelated slow request then cost an

+// already-acknowledged link (Astra on oficina/meet #25, round 3). The

+// drain, and the database close after it, must run on that path too.

+//

+// One sign-in is answered ?sent=1 and held at the relay's DATA; a second

+// POST never finishes its body, so HTTP shutdown times out. The relay

+// recovers after the HTTP deadline but well inside the mail drain

+// window. The process must still be alive then, deliver the link, and

+// exit 1 for the failed shutdown.

+func TestSIGTERMDrainsMailWhenHTTPShutdownTimesOut(t *testing.T) {

+ if testing.Short() {

+ t.Skip("waits out rastrillo's ten-second shutdown deadline")

+ }

+ relay := newHeldRelay(t)

+ p := startCalls(t, relay.port)

+

+ 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")

+ }

+

+ // A request that never finishes: HTTP shutdown cannot complete.

+ stalled, err := net.Dial("tcp", p.addr)

+ if err != nil {

+ t.Fatal(err)

+ }

+ defer stalled.Close()

+ body := url.Values{"address": {"other@example.test"}, "force": {"1"}}.Encode()

+ if _, err := stalled.Write([]byte("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\n\r\n" + body[:len(body)-1])); err != nil {

+ t.Fatal(err)

+ }

+ time.Sleep(200 * time.Millisecond)

+

+ if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil {

+ t.Fatal(err)

+ }

+ signalled := time.Now()

+ select {

+ case err := <-p.exited:

+ t.Fatalf("process exited (%v) %v after SIGTERM, with an acknowledged link still at the relay", err, time.Since(signalled).Round(time.Millisecond))

+ case <-time.After(11500 * time.Millisecond):

+ }

+ relay.release()

+

+ select {

+ case err := <-p.exited:

+ var exit *exec.ExitError

+ if !errors.As(err, &exit) || exit.ExitCode() != 1 {

+ t.Fatalf("process exit: %v, want status 1 for the failed HTTP shutdown", 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 acknowledged sign-in link")

+ }

+ if !strings.Contains(p.out.String(), "context deadline exceeded") {

+ t.Fatal("HTTP shutdown did not time out; the test did not exercise the failed-shutdown path")

+ }

+}

+

+// 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)

+ }

+ addr := ln.Addr().String()

+ ln.Close()

+

+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)

+ 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

+ 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="+relayPort,

+ "CALLS_SMTP_FROM=calls@example.test",

+ "CALLS_ALLOWED_ADDRESSES=paul@example.test",

+ )

+ out := &lockedBuffer{}

+ cmd.Stdout, cmd.Stderr = out, out

+ if err := cmd.Start(); err != nil {

+ t.Fatal(err)

+ }

+ 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() {

+ t.Logf("process output:\n%s", out.String())

+ }

+ })

+

+ up := time.Now().Add(20 * time.Second)

+ for {

+ res, err := p.client.Get(p.origin + "/healthz")

+ if err == nil {

+ res.Body.Close()

+ return p

+ }

+ if time.Now().After(up) {

+ t.Fatalf("server never came up: %v", err)

+ }

+ time.Sleep(20 * time.Millisecond)

+ }

+}

+

+// 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 := <-p.exited:

+ if err != nil {

+ t.Fatalf("process exit: %v", err)

+ }

+ case <-time.After(20 * time.Second):

+ 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.

+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 0f1de31..1c1cf4c 100644

--- a/internal/calls/app.go

+++ b/internal/calls/app.go

@@ -29,6 +29,8 @@ import (

// app holds what every handler needs.

type app struct {

+ // signinMail is the queue sign-in links go through (signin_mail.go).

+ signinMail *mailQueue

db *db.DB

cfg resolved

sealer *seal.Sealer

@@ -39,7 +41,10 @@ type app struct {

nativeAuth *nativeAuthorization

linkBudget *budget

outsiderBudget *budget

- hub *hub.Hub

+ // overBudget counts outsider sign-in attempts the budgets refused,

+ // for sweepHourly to report: never logged on the response path.

+ overBudget atomic.Int64

+ hub *hub.Hub

// push is the APNs sender, or nil where no key is configured — which

// is every test and every deployment that has not been given one.

// Nil is not a degraded mode: knocks still reach every member who is

@@ -251,6 +256,9 @@ func (a *app) sweepOnce(ctx context.Context, now time.Time) {

// sweep and SweepForTest deliberately do not do, because a deadline-

// bound or test-driven caller wants one page of work, not "until done".

func (a *app) sweepHourly(ctx context.Context) {

+ if n := a.overBudget.Swap(0); n > 0 {

+ a.logger.Warn("calls: outsider sign-in attempts over budget", "count", n)

+ }

now := a.store.now()

for {

n, err := a.store.sweep(ctx, now)

@@ -430,16 +438,31 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err

if err != nil {

return nil, nil, err

}

+ // Sign-in links go out after the response, from a worker that asks

+ // 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)

+ }

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

}

@@ -484,15 +507,15 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err

r.Post("/appearance", chrome.HandleAppearance)

r.Get(signinPath, a.signin)

if a.home == nil {

- r.Post(signinPath, a.gateBegin(authPlugin.Begin))

- r.Get("/auth/callback", authPlugin.Callback)

- r.Get("/auth/verify", authPlugin.Verify)

+ r.Post(signinPath, a.newBegin(authPlugin.Begin))

+ r.Get("/auth/callback", doorOnRefusal(authPlugin.Callback, a.localSignin))

+ r.Get("/auth/verify", doorOnRefusal(authPlugin.Verify, a.localSignin))

r.Post("/signout", signedOutAfter(authPlugin.Signout))

r.Get("/signed-out", a.signedOut)

} else {

- r.Post(signinPath, a.home.routes.LocalGuard(a.gateBegin(authPlugin.Begin)))

- r.Get("/auth/callback", a.home.routes.LocalGuard(authPlugin.Callback))

- r.Get("/auth/verify", a.home.routes.LocalGuard(authPlugin.Verify))

+ r.Post(signinPath, a.home.routes.LocalGuard(a.newBegin(authPlugin.Begin)))

+ r.Get("/auth/callback", a.home.routes.LocalGuard(doorOnRefusal(authPlugin.Callback, a.localSignin)))

+ r.Get("/auth/verify", a.home.routes.LocalGuard(doorOnRefusal(authPlugin.Verify, a.localSignin)))

r.Post("/signout", a.home.routes.Signout)

r.Get("/signin/local", a.localSignin)

r.Get("/signed-out", a.home.routes.SignedOut)

@@ -569,9 +592,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.

@@ -579,6 +602,16 @@ 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 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()

}

// nextDue is Runtime.NextDue's implementation: the earliest of the next

@@ -631,7 +664,9 @@ 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

+ // 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

}

// NewForTest builds the app and returns the handler and its test

@@ -644,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 d0a7cbc..5266d34 100644

--- a/internal/calls/app_relay_test.go

+++ b/internal/calls/app_relay_test.go

@@ -38,6 +38,7 @@ func TestSetRelayFlipsRelayConfigured(t *testing.T) {

if err != nil {

t.Fatal(err)

}

+ t.Cleanup(ctl.ShutdownForTest) // stops the sign-in mail worker

if ctl.a.relayConfigured() {

t.Fatal("expected no relay at boot")

}

@@ -62,7 +63,7 @@ func TestBuildLogsOneINFOLineWithTheRegionCountAndNotTheKey(t *testing.T) {

cfg.Logger = logger

cfg.Relay = relayVector(t)

d := openRelayTestDB(t, logger)

- _, a, err := build(context.Background(), d, cfg)

+ _, a, err := buildForTest(t, d, cfg)

if err != nil {

t.Fatal(err)

}

@@ -236,6 +237,7 @@ func TestSetRelayIsTypeConsistentWithBuildsInitialStore(t *testing.T) {

if err != nil {

t.Fatal(err)

}

+ t.Cleanup(ctl.ShutdownForTest) // stops the sign-in mail worker

if err := ctl.SetRelay(relayVector(t)); err != nil {

t.Fatal(err)

}

@@ -262,6 +264,7 @@ func TestSetRelayOfAMalformedValueLeavesRegionsUnchanged(t *testing.T) {

if err != nil {

t.Fatal(err)

}

+ t.Cleanup(ctl.ShutdownForTest) // stops the sign-in mail worker

if err := ctl.SetRelay(relayVector(t)); err != nil {

t.Fatal(err)

}

diff --git a/internal/calls/auth.go b/internal/calls/auth.go

index 8a29ba1..d90cdcc 100644

--- a/internal/calls/auth.go

+++ b/internal/calls/auth.go

@@ -10,6 +10,7 @@ import (

"amadan.net/oficina/estilo.git/chrome"

"amadan.net/oficina/estilo.git/sky"

+ "amadan.net/oficina/home/connect"

rauth "amadan.net/rastrillo/rastrillo/auth"

"amadan.net/rastrillo/rastrillo/sessions"

"github.com/keymaildev/signin"

@@ -32,7 +33,7 @@ func (a *app) newAuthPlugin(writer *sql.DB) (*rauth.Auth, error) {

DB: writer,

Origin: a.cfg.Origin,

InstanceKey: a.cfg.instanceKey,

- Mailer: a.cfg.mailer,

+ Mailer: a.signinMail,

Subject: chrome.EmailSubject("Calls"),

Authorize: a.authorize,

SigninPath: signinPath,

@@ -53,12 +54,32 @@ func (a *app) authorize(address string) bool {

return live

}

-// gateBegin sits in front of rastrillo's Begin. Calls' own refusals are

-// byte-identical to success; rastrillo's address and rate outcomes are

-// shown; membership of a specific address is not a secret Calls keeps at

-// this form, per spec §3.2, and the timing of a sent mail is a further

-// channel the response cannot hide. Task 6 adds the pending-join

-// exception in front of the allowlist check.

+// gateBegin sits in front of rastrillo's Begin, which it reaches for

+// EVERY well-formed address, so whether an address signs in at Keymail

+// or by emailed link depends on its domain alone, never on membership.

+//

+// - A Keymail address (keymail.dev itself, or a domain delegating to

+// Keymail with a `_keymail` TXT record) goes to Keymail, member or

+// not. Begin answers with a 303 to Keymail's authorize URL, which

+// newBegin's continuation turns into a page: Safari applies

+// form-action 'self' to a form's whole redirect chain. Admission is

+// rastrillo's Authorize after Keymail has verified the address (a

+// member, or a guest holding a live pending-join row), and its

+// refusal is the door saying so (doorOnRefusal).

+// - Any other address is answered ?sent=1, and a link is mailed only

+// when mayRequestLink allowed it for this request (ruling 12's

+// budgets included) and authorize admits the address. mailQueue

+// decides that after the response, in a worker, so not even the

+// response time differs (signin_mail.go).

+//

+// Calls' own refusals are byte-identical to success; rastrillo's

+// address and rate outcomes are shown. Membership of a specific address

+// is not a secret Calls keeps at this form (spec §3.2), but the choice

+// of ceremony must still not depend on it: it used to set force=1 for a

+// permitted address only, which kept Keymail out of Calls altogether

+// (Paul, 2026-10-06: "I do want it to use keymail"). force now arrives

+// only from the form rastrillo lands a failed Keymail exchange on

+// (?force=1&err=keymail), and asks for an emailed link instead.

func (a *app) gateBegin(next http.HandlerFunc) http.HandlerFunc {

return func(w http.ResponseWriter, r *http.Request) {

_ = r.ParseForm()

@@ -73,29 +94,56 @@ func (a *app) gateBegin(next http.HandlerFunc) http.HandlerFunc {

http.Redirect(w, r, a.signinAgain(r, "err=address"), http.StatusSeeOther)

return

}

- if !a.mayRequestLink(w, r, address) {

- http.Redirect(w, r, sentLocation, http.StatusSeeOther)

- return

- }

- if r.Form == nil {

- r.Form = url.Values{}

- }

- // force makes rastrillo skip its address classifier, so Begin

- // never takes the keymail branch (a 303 to an external host

- // chosen by the address's DNS, which would be a distinguishable

- // outcome and an oracle); it also means keymail sign-in is

- // disabled app-wide, deliberately.

- r.Form.Set("force", "1")

+ may := a.mayRequestLink(w, r, address)

+ r = r.WithContext(context.WithValue(r.Context(), mayMailKey{}, may))

next(&sameAnswer{ResponseWriter: w, app: a}, r)

}

-// mayRequestLink decides whether Begin is reached (ruling 12). A member

-// address always may. Any other address may only when the browser

-// carries a valid pending-join cookie for a live link, within two

-// budgets counted before Begin: five attempts per link and ten across

-// all links per fifteen minutes. When it may, a pending_joins row is

-// written first, for the member too (it carries their destination).

+// newBegin is POST /signin in both modes: gateBegin under

+// connect.OAuthContinuation, which ends the POST with a page

+// (renderKeymail) when Begin's answer is Keymail's authorize URL.

+func (a *app) newBegin(begin http.HandlerFunc) http.HandlerFunc {

+ return connect.OAuthContinuation(a.cfg.Origin, a.gateBegin(begin), a.renderKeymail)

+}

+

+// renderKeymail is the page between the form and Keymail: the door's

+// card saying where the person is going, with the meta refresh that

+// goes there (home-oauth-head) and a link for when it is blocked.

+func (a *app) renderKeymail(w http.ResponseWriter, r *http.Request, destination string) {

+ p := connect.OAuthPage(destination)

+ notice := chrome.SignInNotice{

+ Icon: assets.Path("static/brand/icon.svg"), App: appName, Sky: sky.For(r.Host),

+ Heading: p.Title, Body: "Taking you to Keymail to finish signing in.",

+ Link: &chrome.Link{Href: p.URL, Label: p.ActionLabel},

+ }

+ render(w, "signin", a.page(r, frame{parts: []string{p.Title}, signinPage: true}, map[string]any{"Sent": true, "Notice": notice, "Continue": p}))

+}

+

+// mayMailKey carries gateBegin's mayRequestLink answer to mailQueue,

+// through Begin, on the request context.

+type mayMailKey struct{}

+

+// mayRequestLink decides whether a link may be mailed (ruling 12). A

+// member address always may. Any other address may only when the

+// browser carries a valid pending-join cookie for a live link, within

+// two budgets: five attempts per link and ten across all links per

+// fifteen minutes.

+//

+// It runs on the response path, so its WORK must not depend on who the

+// address is (Astra on #25, round 3, P1: over budget, a member wrote a

+// pending row while an outsider wrote a log line and returned —

+// 0.4–0.65 ms against 77–78 ms). Whoever the address names, a live

+// cookie costs the same: the link lookup, and the pending_joins row,

+// written for an over-budget outsider too. Only the in-memory budget

+// counters, charged for outsiders as before, differ. Nothing is logged

+// here: an over-budget attempt is counted (overBudget) and the hourly

+// sweep reports the count.

+//

+// The row an over-budget outsider gets admits nothing new by mail: the

+// answer this returns still says no, so no link is sent (the budgets are

+// about mail). It does let a Keymail sign-in through that link's guest

+// path, which the budgets never governed.

func (a *app) mayRequestLink(w http.ResponseWriter, r *http.Request, address string) bool {

address = normalizeAddress(address)

member := a.allowNow().allows(address)

@@ -111,13 +159,14 @@ func (a *app) mayRequestLink(w http.ResponseWriter, r *http.Request, address str

if !live {

return member

}

+ may := true

if !member {

now := a.store.now()

// The per-link budget is consulted first: a link already over

// its own budget must not keep charging the shared pool.

if !a.linkBudget.allow(ck.TokenHash, now) || !a.outsiderBudget.allow("all", now) {

- a.logger.Warn("calls: outsider sign-in attempts over budget", "link", ck.TokenHash[:8])

- return false

+ a.overBudget.Add(1)

+ may = false

}

if address == "" {

@@ -127,7 +176,7 @@ func (a *app) mayRequestLink(w http.ResponseWriter, r *http.Request, address str

a.logger.Error("calls: write pending join", "err", err)

return member

}

- return true

+ return may

}

// rastrillo's own Begin also same-origin-checks the request and

@@ -225,7 +274,7 @@ func (a *app) requireMember(next http.Handler) http.Handler {

func (a *app) localSignin(w http.ResponseWriter, r *http.Request) {

q := r.URL.Query()

ret, hasReturn := safeReturn(q)

- if v := a.viewerOf(r); v.signedIn() && q.Get("sent") == "" && q.Get("err") == "" && r.URL.Path == signinPath {

+ if v := a.viewerOf(r); !refusedHere(r) && v.signedIn() && q.Get("sent") == "" && q.Get("err") == "" && r.URL.Path == signinPath {

dest := "/"

if hasReturn {

dest = ret

@@ -264,6 +313,12 @@ func (a *app) localSignin(w http.ResponseWriter, r *http.Request) {

if hasReturn {

door.Return = ret

}

+ if q.Get("force") != "" {

+ // A failed Keymail exchange lands here as ?force=1&err=keymail.

+ // Posting force back asks for an emailed link instead, so the

+ // retry does not return to the Keymail that just failed.

+ door.Hidden = []chrome.Hidden{{Name: "force", Value: "1"}}

+ }

switch q.Get("err") {

case "address":

// An address error belongs to the field, not a banner.

@@ -274,10 +329,18 @@ func (a *app) localSignin(w http.ResponseWriter, r *http.Request) {

door.Error = "That link has expired or was already used. Enter your email to get a new one."

case "removed":

door.Error = "Your access has been removed, so you’ve been signed out. Ask whoever runs this team if that’s wrong."

+ case "keymail":

+ door.Error = "Your Keymail sign-in couldn’t be completed. Enter your email to get a link instead."

case "":

default:

door.Error = "We couldn’t sign you in. Enter your email to get a new link."

}

+ if refusedHere(r) {

+ // Keymail or a link verified the address and Calls does not

+ // admit it, in this very response (doorOnRefusal). Never from the

+ // query. The form is the way on.

+ door.Error = "That address can’t sign in to Calls. Use a different address, or ask whoever runs this team to add it."

+ }

render(w, "signin", a.page(r, frame{parts: []string{door.Page()}, signinPage: true}, map[string]any{"Door": door}))

}

diff --git a/internal/calls/auth_refusal.go b/internal/calls/auth_refusal.go

new file mode 100644

index 0000000..d668f1e

--- /dev/null

+++ b/internal/calls/auth_refusal.go

@@ -0,0 +1,101 @@

+package calls

+

+import (

+ "bytes"

+ "context"

+ "net/http"

+ "strings"

+)

+

+// authRefusal is the bare-text body rastrillo/auth's admit() writes,

+// with a 403, when a verified address is not admitted. rastrillo at the

+// version pinned here has no hook to render anything else, so

+// doorOnRefusal recognises this exact answer; TestKeymailRefusalIsTheDoor

+// pins it, so a bump that changes the sentence fails loudly.

+const authRefusal = "This address is verified but not admitted here."

+

+// doorOnRefusal wraps rastrillo's Callback and Verify, the handlers that

+// end in admission, and turns its bare-text refusal into the door,

+// rendered in the refusal's own 403 by door (localSignin) with the

+// refusal marked on the request context (refusedHere). Every other

+// answer, cookies included, passes through untouched.

+//

+// Rendered, not redirected to ?err=refused: a query parameter is

+// something anybody can type, and the sentence means "you proved this

+// address and Calls does not admit it" (Astra on #25, round 3). Remove

+// it when rastrillo has a refusal hook (feat/auth-refused-hook) and the

+// pin is bumped.

+func doorOnRefusal(next, door http.HandlerFunc) http.HandlerFunc {

+ return func(w http.ResponseWriter, r *http.Request) {

+ rec := &bufferedResponse{header: make(http.Header)}

+ next(rec, r)

+ if rec.status == http.StatusForbidden && strings.TrimSpace(rec.body.String()) == authRefusal {

+ for _, c := range rec.header.Values("Set-Cookie") {

+ w.Header().Add("Set-Cookie", c)

+ }

+ door(&forcedStatus{ResponseWriter: w, status: http.StatusForbidden},

+ r.WithContext(context.WithValue(r.Context(), refusedKey{}, true)))

+ return

+ }

+ for name, values := range rec.header {

+ for _, v := range values {

+ w.Header().Add(name, v)

+ }

+ if rec.status == 0 {

+ rec.status = http.StatusOK

+ }

+ w.WriteHeader(rec.status)

+ _, _ = w.Write(rec.body.Bytes())

+ }

+}

+

+// refusedKey marks a request doorOnRefusal is answering with the door.

+type refusedKey struct{}

+

+// refusedHere reports whether r is a refusal doorOnRefusal is rendering.

+func refusedHere(r *http.Request) bool {

+ v, _ := r.Context().Value(refusedKey{}).(bool)

+ return v

+}

+

+// forcedStatus answers with status whatever status the page writes.

+type forcedStatus struct {

+ http.ResponseWriter

+ status int

+ wrote bool

+}

+

+func (f *forcedStatus) WriteHeader(int) {

+ if !f.wrote {

+ f.wrote = true

+ f.ResponseWriter.WriteHeader(f.status)

+ }

+}

+

+func (f *forcedStatus) Write(p []byte) (int, error) {

+ f.WriteHeader(f.status)

+ return f.ResponseWriter.Write(p)

+}

+

+// bufferedResponse holds a whole answer so doorOnRefusal can decide what

+// to send after seeing it. Callback and Verify answer with a redirect

+// or a line of text, never anything large.

+type bufferedResponse struct {

+ header http.Header

+ status int

+ body bytes.Buffer

+}

+

+func (b *bufferedResponse) Header() http.Header { return b.header }

+func (b *bufferedResponse) WriteHeader(status int) {

+ if b.status == 0 {

+ b.status = status

+ }

+}

+func (b *bufferedResponse) Write(p []byte) (int, error) {

+ if b.status == 0 {

+ b.status = http.StatusOK

+ }

+ return b.body.Write(p)

+}

diff --git a/internal/calls/auth_test.go b/internal/calls/auth_test.go

index 629dd8c..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(context.Background(), 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 c1eb08f..d8d15e3 100644

--- a/internal/calls/home_router_test.go

+++ b/internal/calls/home_router_test.go

@@ -66,13 +66,20 @@ 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)

}

return &homeRouterFixture{a, mux, key, mail, map[string]*http.Cookie{}}

}

+// mailBody is the last sign-in mail, once the queue has sent it

+// (links go out after the response: signin_mail.go).

+func (f *homeRouterFixture) mailBody() string {

+ f.app.signinMail.wait()

+ return f.mail.body

+}

+

func homeKeyID(key *assertioncrypto.Keypair) string {

sum := sha256.Sum256(key.SignPub())

return hex.EncodeToString(sum[:])

@@ -111,7 +118,7 @@ func homeRedirect(t *testing.T, w *httptest.ResponseRecorder) string {

func (f *homeRouterFixture) local(t *testing.T) {

t.Helper()

homeRedirect(t, f.do("POST", "/signin", "address=alice%40example.test&force=1"))

- link := regexp.MustCompile(`https://[^\s<>\"]+/auth/verify\?[^\s<>\"]+`).FindString(f.mail.body)

+ link := regexp.MustCompile(`https://[^\s<>\"]+/auth/verify\?[^\s<>\"]+`).FindString(f.mailBody())

u, err := url.Parse(link)

if err != nil || link == "" {

t.Fatal("no captured local verification link")

@@ -240,7 +247,7 @@ func TestHomeEnabledPreservesInvitedGuestFlow(t *testing.T) {

t.Fatal("guest redirected to Home", w.Code)

}

homeRedirect(t, f.do("POST", "/signin", "address=guest%40example.test"))

- link := regexp.MustCompile(`https://[^\s<>\"]+/auth/verify\?[^\s<>\"]+`).FindString(f.mail.body)

+ link := regexp.MustCompile(`https://[^\s<>\"]+/auth/verify\?[^\s<>\"]+`).FindString(f.mailBody())

u, _ := url.Parse(link)

landing := homeRedirect(t, f.do("GET", u.RequestURI(), ""))

if landing != "/auth/local/complete" {

diff --git a/internal/calls/keymail_signin_test.go b/internal/calls/keymail_signin_test.go

new file mode 100644

index 0000000..f14a34a

--- /dev/null

+++ b/internal/calls/keymail_signin_test.go

@@ -0,0 +1,281 @@

+package calls

+

+import (

+ "bytes"

+ "context"

+ "encoding/base64"

+ "fmt"

+ "html"

+ "io"

+ "log/slog"

+ "net/http"

+ "net/http/httptest"

+ "net/url"

+ "os"

+ "path/filepath"

+ "regexp"

+ "strings"

+ "sync"

+ "testing"

+ "time"

+

+ "amadan.net/oficina/calls/internal/calls/keymailtest"

+ "amadan.net/oficina/home/connect"

+ assertioncrypto "amadan.net/rastrillo/rastrillo/crypto"

+ "amadan.net/rastrillo/rastrillo/db"

+)

+

+// TestMain keeps Keymail classification inside the process: every DNS

+// query is answered NXDOMAIN and only the Keymail fixtures a test

+// registers answer HTTPS (keymailtest).

+func TestMain(m *testing.M) {

+ restore, err := keymailtest.Install()

+ if err != nil {

+ fmt.Fprintln(os.Stderr, "keymailtest.Install:", err)

+ os.Exit(1)

+ }

+ code := m.Run()

+ // No test may leave a sign-in mail worker running: every app a test

+ // 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) {

+ time.Sleep(10 * time.Millisecond)

+ }

+ if n := mailWorkers(); n > 0 {

+ fmt.Fprintf(os.Stderr, "FAIL: %d sign-in mail queue goroutines still running after the tests\n", n)

+ code = 1

+ }

+ restore()

+ os.Exit(code)

+}

+

+const (

+ keymailHost = "keymail.calls.test"

+ keymailMember = "paul@" + keymailHost

+ keymailStranger = "stranger@" + keymailHost

+ plainMember = "paul@example.test"

+ keymailOrigin = "https://calls.example.test"

+)

+

+type countingMailer struct {

+ mu sync.Mutex

+ to []string

+ waits []func() // the sign-in mail queues in front of it

+}

+

+func (m *countingMailer) AttachSigninQueue(wait func()) {

+ m.mu.Lock()

+ m.waits = append(m.waits, wait)

+ m.mu.Unlock()

+}

+

+func (m *countingMailer) Send(_ context.Context, to, _, _ string) error {

+ m.mu.Lock()

+ defer m.mu.Unlock()

+ m.to = append(m.to, to)

+ return nil

+}

+

+func (m *countingMailer) sent() []string {

+ m.mu.Lock()

+ waits := append([]func(){}, m.waits...)

+ m.mu.Unlock()

+ for _, w := range waits {

+ w()

+ }

+ m.mu.Lock()

+ defer m.mu.Unlock()

+ return append([]string(nil), m.to...)

+}

+

+type keymailFixture struct {

+ handler http.Handler

+ mail *countingMailer

+ km *keymailtest.Server

+}

+

+// newKeymailFixture builds Calls with plainMember and keymailMember on

+// the allowlist, beside a Keymail deployment holding keymailMember and

+// keymailStranger; home configures Home sign-in.

+func newKeymailFixture(t *testing.T, home bool) *keymailFixture {

+ t.Helper()

+ km := keymailtest.Serve(t, keymailHost, keymailMember, keymailStranger)

+ logger := slog.New(slog.NewTextHandler(io.Discard, nil))

+ d, err := db.Open(filepath.Join(t.TempDir(), "calls.db"), logger)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { d.Close() })

+ mail := &countingMailer{}

+ cfg := Config{Origin: keymailOrigin, InstanceKey: "test-only-instance", DEK: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{3}, 32)),

+ AllowedAddresses: []string{plainMember, keymailMember}, Mailer: mail, Logger: logger}

+ if home {

+ key, err := assertioncrypto.Generate()

+ if err != nil {

+ t.Fatal(err)

+ }

+ cfg.Home = &connect.Credentials{Issuer: homeTestIssuer, ConsumerID: "calls-test", KeysBase64: homeKeysBase64(key), DEKBase64: base64.StdEncoding.EncodeToString(bytes.Repeat([]byte{7}, 32))}

+ }

+ mux, _, err := buildForTest(t, d, cfg)

+ if err != nil {

+ t.Fatal(err)

+ }

+ return &keymailFixture{handler: mux, mail: mail, km: km}

+}

+

+func (f *keymailFixture) post(form url.Values, cookies ...*http.Cookie) *httptest.ResponseRecorder {

+ r := httptest.NewRequest(http.MethodPost, keymailOrigin+"/signin", strings.NewReader(form.Encode()))

+ r.Header.Set("Content-Type", "application/x-www-form-urlencoded")

+ r.Header.Set("Origin", keymailOrigin)

+ r.Header.Set("Sec-Fetch-Site", "same-origin")

+ for _, c := range cookies {

+ r.AddCookie(c)

+ }

+ w := httptest.NewRecorder()

+ f.handler.ServeHTTP(w, r)

+ return w

+}

+

+func (f *keymailFixture) get(path string, cookies ...*http.Cookie) *httptest.ResponseRecorder {

+ r := httptest.NewRequest(http.MethodGet, keymailOrigin+path, nil)

+ for _, c := range cookies {

+ r.AddCookie(c)

+ }

+ w := httptest.NewRecorder()

+ f.handler.ServeHTTP(w, r)

+ return w

+}

+

+var authorizeHref = regexp.MustCompile(`href="(https://` + regexp.QuoteMeta(keymailHost) + `/oauth/authorize\?[^"]+)"`)

+

+// continuation asserts rec is the page between the form and Keymail —

+// a 200 document with a meta refresh, never a 303 out of the form POST

+// (Safari applies form-action to that redirect) — and returns the

+// authorize URL it carries.

+func continuation(t *testing.T, label string, rec *httptest.ResponseRecorder) *url.URL {

+ t.Helper()

+ if rec.Code != http.StatusOK || rec.Header().Get("Location") != "" {

+ t.Fatalf("%s: %d %q, want 200 and the continuation page to Keymail", label, rec.Code, rec.Header().Get("Location"))

+ }

+ body := rec.Body.String()

+ m := authorizeHref.FindStringSubmatch(body)

+ if m == nil || !strings.Contains(body, `http-equiv="refresh"`) {

+ t.Fatalf("%s: no Keymail continuation on the page:\n%s", label, body)

+ }

+ u, err := url.Parse(html.UnescapeString(m[1]))

+ if err != nil {

+ t.Fatal(err)

+ }

+ return u

+}

+

+// TestKeymailAddressGoesToKeymailWhetherPermittedOrNot: in both modes,

+// whether an address goes to Keymail depends on its domain alone. A

+// member and a stranger at a Keymail domain get the same first answer,

+// the continuation page, and neither is mailed. Before, gateBegin set

+// force=1 on every member's POST, so Keymail was never used.

+func TestKeymailAddressGoesToKeymailWhetherPermittedOrNot(t *testing.T) {

+ for _, home := range []bool{false, true} {

+ t.Run(fmt.Sprintf("home=%v", home), func(t *testing.T) {

+ f := newKeymailFixture(t, home)

+ member := f.post(url.Values{"address": {keymailMember}})

+ stranger := f.post(url.Values{"address": {keymailStranger}})

+ mu := continuation(t, "member Keymail address", member)

+ su := continuation(t, "stranger Keymail address", stranger)

+ mask := func(rec *httptest.ResponseRecorder, u *url.URL) string {

+ return strings.ReplaceAll(rec.Body.String(), html.EscapeString(u.String()), "AUTHORIZE")

+ }

+ if a, b := mask(member, mu), mask(stranger, su); a != b {

+ t.Errorf("the two Keymail answers differ beyond the authorize URL:\n%s\n----\n%s", a, b)

+ }

+ if got := f.mail.sent(); len(got) != 0 {

+ t.Errorf("a Keymail address was mailed: %v", got)

+ }

+ if n := f.km.Lookups(); n != 2 {

+ t.Errorf("Keymail saw %d lookups, want 2", n)

+ }

+ })

+ }

+}

+

+// TestNonPermittedPlainAddressIsAnsweredSentAndNotMailed: an address

+// with no Keymail gets ?sent=1, member or not; only the member is

+// mailed.

+func TestNonPermittedPlainAddressIsAnsweredSentAndNotMailed(t *testing.T) {

+ f := newKeymailFixture(t, false)

+ for _, address := range []string{"someone@else.test", plainMember} {

+ rec := f.post(url.Values{"address": {address}})

+ if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/signin?sent=1" {

+ t.Fatalf("POST /signin %q: %d %q, want 303 /signin?sent=1", address, rec.Code, rec.Header().Get("Location"))

+ }

+ if got := f.mail.sent(); len(got) != 1 || got[0] != plainMember {

+ t.Fatalf("mail = %v, want one, to %s", got, plainMember)

+ }

+}

+

+// TestKeymailFailureFallsBackToAnEmailLink: the page rastrillo lands a

+// failed Keymail exchange on (?force=1&err=keymail) carries force, so

+// the retry mails a link instead of returning to Keymail; the ordinary

+// form carries none.

+func TestKeymailFailureFallsBackToAnEmailLink(t *testing.T) {

+ f := newKeymailFixture(t, false)

+ forced := `name="force" value="1"`

+ if strings.Contains(f.get("/signin").Body.String(), forced) {

+ t.Fatal("the ordinary form forces an email link")

+ }

+ if !strings.Contains(f.get("/signin?force=1&err=keymail").Body.String(), forced) {

+ t.Fatal("the err=keymail form does not carry force; a failed Keymail sign-in would loop back to Keymail")

+ }

+ rec := f.post(url.Values{"address": {keymailMember}, "force": {"1"}})

+ if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/signin?sent=1" {

+ t.Fatalf("forced POST: %d %q", rec.Code, rec.Header().Get("Location"))

+ }

+ if got := f.mail.sent(); len(got) != 1 || got[0] != keymailMember {

+ t.Fatalf("mail = %v, want one link to %s", got, keymailMember)

+ }

+ if f.km.Lookups() != 0 {

+ t.Error("a forced sign-in classified the address")

+ }

+}

+

+// TestKeymailRefusalIsTheDoor: a stranger Keymail verified is refused

+// after verification on the door, with the form for a different

+// address, not in rastrillo's bare text; a member is signed in. Both

+// modes.

+func TestKeymailRefusalIsTheDoor(t *testing.T) {

+ for _, home := range []bool{false, true} {

+ t.Run(fmt.Sprintf("home=%v", home), func(t *testing.T) {

+ f := newKeymailFixture(t, home)

+ form := "/signin"

+ if home {

+ form = "/signin/local"

+ }

+ roundTrip := func(address string) *httptest.ResponseRecorder {

+ begin := f.post(url.Values{"address": {address}})

+ u := continuation(t, address, begin)

+ return f.get("/auth/callback?"+url.Values{"state": {u.Query().Get("state")}, "code": {f.km.Approve(address)}}.Encode(), begin.Result().Cookies()...)

+ }

+ if cb := roundTrip(keymailMember); cb.Code != http.StatusSeeOther || strings.Contains(cb.Header().Get("Location"), "err=") {

+ t.Fatalf("member callback: %d %q", cb.Code, cb.Header().Get("Location"))

+ }

+ cb := roundTrip(keymailStranger)

+ if cb.Code != http.StatusForbidden || cb.Header().Get("Location") != "" {

+ t.Fatalf("stranger callback: %d %q, want 403 and the door itself (bare text is the bug)\n%s", cb.Code, cb.Header().Get("Location"), cb.Body.String())

+ }

+ door := cb.Body.String()

+ for _, want := range []string{"estilo-door", `name="address"`, "can’t sign in to Calls"} {

+ if !strings.Contains(door, want) {

+ t.Errorf("refusal door lacks %q:\n%s", want, door)

+ }

+ // Nobody may summon the sentence by typing a URL (Astra, round 3).

+ if spoof := f.get(form + "?err=refused").Body.String(); strings.Contains(spoof, "can’t sign in to Calls") {

+ t.Fatalf("anybody can show the refusal by visiting %s?err=refused", form)

+ }

+ })

+ }

+}

diff --git a/internal/calls/keymailtest/keymailtest.go b/internal/calls/keymailtest/keymailtest.go

new file mode 100644

index 0000000..92ad056

--- /dev/null

+++ b/internal/calls/keymailtest/keymailtest.go

@@ -0,0 +1,190 @@

+// Package keymailtest keeps sign-in's Keymail classification inside the

+// test process. It is imported by tests only.

+//

+// rastrillo/auth (at the version this app pins) builds its classifier

+// with no seams: TXT lookups go through net.DefaultResolver and the

+// well-known probe, the federation lookup and the token exchange go

+// through http.DefaultTransport. So those two globals are the hooks,

+// exactly as oficina/docs' auth_keymail_test.go uses them. Install

+// replaces both for the life of the test binary:

+//

+// - every DNS query is answered NXDOMAIN by a resolver on loopback,

+// so no DNS packet leaves the process, and a domain with no

+// `_keymail` record is probed directly — keymail.dev's own shape;

+// - an https request to a host registered with Serve is answered

+// in-process by that fixture; anything else falls through to the

+// real transport, where the resolver above makes a name

+// unreachable (IP literals, which httptest servers use, still work).

+//

+// Every address at a domain nobody registered therefore classifies as

+// "not Keymail", which is what every pre-existing sign-in test assumes.

+package keymailtest

+

+import (

+ "context"

+ "crypto/rand"

+ "encoding/hex"

+ "encoding/json"

+ "net"

+ "net/http"

+ "net/http/httptest"

+ "strings"

+ "sync"

+ "time"

+)

+

+var (

+ mu sync.Mutex

+ servers = map[string]*Server{}

+)

+

+// Install swaps net.DefaultResolver and http.DefaultTransport for the

+// stubs described in the package comment and returns a function that

+// puts them back. Call it once, from TestMain.

+func Install() (restore func(), err error) {

+ dns, err := net.ListenPacket("udp", "127.0.0.1:0")

+ if err != nil {

+ return nil, err

+ }

+ go answerNXDOMAIN(dns)

+ oldResolver, oldTransport := net.DefaultResolver, http.DefaultTransport

+ net.DefaultResolver = &net.Resolver{PreferGo: true, Dial: func(ctx context.Context, _, _ string) (net.Conn, error) {

+ return (&net.Dialer{}).DialContext(ctx, "udp", dns.LocalAddr().String())

+ }}

+ http.DefaultTransport = transport{next: oldTransport}

+ return func() {

+ net.DefaultResolver, http.DefaultTransport = oldResolver, oldTransport

+ _ = dns.Close()

+ }, nil

+}

+

+// answerNXDOMAIN echoes each query back as a "no such name" answer.

+func answerNXDOMAIN(conn net.PacketConn) {

+ b := make([]byte, 4096)

+ for {

+ n, addr, err := conn.ReadFrom(b)

+ if err != nil {

+ return

+ }

+ if n < 12 {

+ continue

+ }

+ reply := append([]byte(nil), b[:n]...)

+ reply[2], reply[3] = 0x81, 0x83 // response, recursion, NXDOMAIN

+ for i := 6; i < 12; i++ {

+ reply[i] = 0 // no answer, authority or additional records

+ }

+ _, _ = conn.WriteTo(reply, addr)

+ }

+}

+

+type transport struct{ next http.RoundTripper }

+

+func (t transport) RoundTrip(r *http.Request) (*http.Response, error) {

+ mu.Lock()

+ s := servers[r.URL.Host]

+ mu.Unlock()

+ if s == nil || r.URL.Scheme != "https" {

+ return t.next.RoundTrip(r)

+ }

+ rec := httptest.NewRecorder()

+ s.ServeHTTP(rec, r)

+ res := rec.Result()

+ res.Request = r

+ return res, nil

+}

+

+// Server is one Keymail deployment at one host: the well-known

+// document, the federation lookup for the accounts it holds, and the

+// token exchange for the codes Approve hands out.

+type Server struct {

+ host string

+

+ mu sync.Mutex

+ accounts map[string]bool

+ codes map[string]string

+ lookups int

+ tokens int

+}

+

+// Serve registers a Keymail deployment at host (which is also the

+// address domain it answers for, since nothing publishes a `_keymail`

+// record) holding the given accounts. It is unregistered when the test

+// ends.

+func Serve(t interface{ Cleanup(func()) }, host string, accounts ...string) *Server {

+ s := &Server{host: host, accounts: map[string]bool{}, codes: map[string]string{}}

+ for _, a := range accounts {

+ s.accounts[strings.ToLower(a)] = true

+ }

+ mu.Lock()

+ servers[host] = s

+ mu.Unlock()

+ t.Cleanup(func() {

+ mu.Lock()

+ delete(servers, host)

+ mu.Unlock()

+ })

+ return s

+}

+

+// Approve is Keymail's consent screen saying yes for address: it

+// returns the code the browser would carry back to /auth/callback.

+func (s *Server) Approve(address string) string {

+ var b [12]byte

+ _, _ = rand.Read(b[:])

+ code := hex.EncodeToString(b[:])

+ s.mu.Lock()

+ s.codes[code] = address

+ s.mu.Unlock()

+ return code

+}

+

+// Lookups is how many federation lookups reached this deployment.

+func (s *Server) Lookups() int {

+ s.mu.Lock()

+ defer s.mu.Unlock()

+ return s.lookups

+}

+

+// Tokens is how many token exchanges reached this deployment.

+func (s *Server) Tokens() int {

+ s.mu.Lock()

+ defer s.mu.Unlock()

+ return s.tokens

+}

+

+func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {

+ w.Header().Set("Content-Type", "application/json")

+ switch r.URL.Path {

+ case "/.well-known/keymail":

+ _ = json.NewEncoder(w).Encode(map[string]any{"version": "1", "reauth": true})

+ case "/api/federation/lookup":

+ s.mu.Lock()

+ s.lookups++

+ ok := s.accounts[strings.ToLower(r.URL.Query().Get("addr"))]

+ s.mu.Unlock()

+ if !ok {

+ w.WriteHeader(http.StatusNotFound)

+ }

+ _ = json.NewEncoder(w).Encode(map[string]bool{"ok": ok})

+ case "/api/oauth/token":

+ var in map[string]string

+ _ = json.NewDecoder(r.Body).Decode(&in)

+ s.mu.Lock()

+ s.tokens++

+ address, ok := s.codes[in["code"]]

+ delete(s.codes, in["code"])

+ s.mu.Unlock()

+ if r.Method != http.MethodPost || !ok {

+ w.WriteHeader(http.StatusBadRequest)

+ _ = json.NewEncoder(w).Encode(map[string]string{"error": "invalid_grant"})

+ return

+ }

+ _ = json.NewEncoder(w).Encode(map[string]any{

+ "access_token": "synthetic", "token_type": "bearer",

+ "address": address, "auth_time": time.Now().Unix(),

+ })

+ default:

+ w.WriteHeader(http.StatusNotFound)

+ }

+}

diff --git a/internal/calls/render.go b/internal/calls/render.go

index 2c625a2..9c294b1 100644

--- a/internal/calls/render.go

+++ b/internal/calls/render.go

@@ -67,6 +67,10 @@ var pages = map[string]*template.Template{}

func init() {

for _, name := range pageNames {

pages[name] = template.Must(buildPage(name).ParseFS(credits.Templates(), "templates/credits.html"))

+ if name == "signin" {

+ // The Keymail continuation's meta refresh (home-oauth-head).

+ pages[name] = template.Must(pages[name].ParseFS(connect.Templates(), "templates/oauth.html"))

+ }

if name == "home" {

pages[name] = template.Must(pages[name].ParseFS(connect.Templates(), "templates/home_link.html"))

}

diff --git a/internal/calls/signin_equal_work_test.go b/internal/calls/signin_equal_work_test.go

new file mode 100644

index 0000000..20b6296

--- /dev/null

+++ b/internal/calls/signin_equal_work_test.go

@@ -0,0 +1,149 @@

+package calls

+

+import (

+ "bytes"

+ "context"

+ "encoding/base64"

+ "fmt"

+ "log/slog"

+ "net/http"

+ "net/http/httptest"

+ "net/url"

+ "path/filepath"

+ "sort"

+ "strconv"

+ "strings"

+ "sync/atomic"

+ "testing"

+ "time"

+

+ "amadan.net/rastrillo/rastrillo/db"

+ "gorm.io/gorm"

+)

+

+// measuredKey marks the context of a request whose statements count.

+type measuredKey struct{}

+

+// countingHandler counts log records, and nothing else.

+type countingHandler struct{ n *atomic.Int64 }

+

+func (h countingHandler) Enabled(context.Context, slog.Level) bool { return true }

+func (h countingHandler) Handle(context.Context, slog.Record) error { h.n.Add(1); return nil }

+func (h countingHandler) WithAttrs([]slog.Attr) slog.Handler { return h }

+func (h countingHandler) WithGroup(string) slog.Handler { return h }

+

+// TestMemberAndOutsiderDoTheSameWorkOverBudget is Astra's round-3 probe

+// on #25 (P1). With a live invitation in the browser and the outsider

+// budgets spent, membership used to decide what POST /signin did on the

+// response path: a member wrote a pending row, an outsider wrote a log

+// line and returned — 0.38–0.65 ms against 76.9–77.9 ms. Every POST now

+// does the same synchronous work whoever it names: the same database

+// statements, and no log line at all. The budgets still hold: the

+// outsider is mailed nothing.

+func TestMemberAndOutsiderDoTheSameWorkOverBudget(t *testing.T) {

+ var logs atomic.Int64

+ logger := slog.New(countingHandler{&logs})

+ d, err := db.Open(filepath.Join(t.TempDir(), "calls.db"), slog.New(slog.DiscardHandler))

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { d.Close() })

+ mail := &countingMailer{}

+ 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)

+ }

+ _, tok, err := a.store.createMeeting(t.Context(), "x", nil, "person:1")

+ if err != nil {

+ t.Fatal(err)

+ }

+ join := httptest.NewRecorder()

+ mux.ServeHTTP(join, httptest.NewRequest(http.MethodGet, keymailOrigin+"/j/"+tok, nil))

+ cookies := join.Result().Cookies()

+ if len(cookies) == 0 {

+ t.Fatal("following the invitation set no join cookie")

+ }

+

+ var statements atomic.Int64

+ // Only the request's own statements: the mail worker runs beside it

+ // (authorize, after the response) on a context of its own.

+ count := func(tx *gorm.DB) {

+ if tx.Statement.Context != nil && tx.Statement.Context.Value(measuredKey{}) != nil {

+ statements.Add(1)

+ }

+ cb := a.db.G.Callback()

+ for _, err := range []error{

+ cb.Query().After("gorm:query").Register("equalwork:query", count),

+ cb.Create().After("gorm:create").Register("equalwork:create", count),

+ cb.Update().After("gorm:update").Register("equalwork:update", count),

+ cb.Delete().After("gorm:delete").Register("equalwork:delete", count),

+ cb.Raw().After("gorm:raw").Register("equalwork:raw", count),

+ cb.Row().After("gorm:row").Register("equalwork:row", count),

+ } {

+ if err != nil {

+ t.Fatal(err)

+ }

+ ip := 0

+ post := func(form url.Values) *httptest.ResponseRecorder {

+ ip++ // a fresh client address each time, so rastrillo's per-IP budget stays out of it

+ r := httptest.NewRequest(http.MethodPost, keymailOrigin+"/signin", strings.NewReader(form.Encode()))

+ r.Header.Set("Content-Type", "application/x-www-form-urlencoded")

+ r.Header.Set("Origin", keymailOrigin)

+ r.Header.Set("Sec-Fetch-Site", "same-origin")

+ r.RemoteAddr = "198.51.100." + strconv.Itoa(ip) + ":1234"

+ r = r.WithContext(context.WithValue(r.Context(), measuredKey{}, true))

+ for _, c := range cookies {

+ r.AddCookie(c)

+ }

+ w := httptest.NewRecorder()

+ mux.ServeHTTP(w, r)

+ return w

+ }

+ // Spend the per-link budget (five) with outsiders, each its own

+ // address so rastrillo's per-address budget stays out of it.

+ for i := 0; i < 6; i++ {

+ post(url.Values{"address": {"spender" + string(rune('a'+i)) + "@else.test"}})

+ }

+

+ measure := func(address string) (stmts, lines int64, took time.Duration) {

+ statements.Store(0)

+ logs.Store(0)

+ start := time.Now()

+ rec := post(url.Values{"address": {address}})

+ took = time.Since(start)

+ if rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/signin?sent=1" {

+ t.Fatalf("POST %s: %d %q", address, rec.Code, rec.Header().Get("Location"))

+ }

+ return statements.Load(), logs.Load(), took

+ }

+ ms, ml, _ := measure(plainMember)

+ os, ol, _ := measure("outsider@else.test")

+ if ms != os || ml != ol || ml != 0 {

+ t.Fatalf("member did %d statements and %d log lines, outsider %d and %d: want the same work and no log line on the response path", ms, ml, os, ol)

+ }

+

+ // Astra's measurement, re-run for the record (not asserted: a

+ // stopwatch in a test is a flake, the statement count is the proof).

+ // Four more of each: rastrillo's per-address budget is five.

+ var mt, ot []time.Duration

+ for i := 0; i < 4; i++ {

+ _, _, d1 := measure(plainMember)

+ _, _, d2 := measure(fmt.Sprintf("outsider%d@else.test", i))

+ mt, ot = append(mt, d1), append(ot, d2)

+ }

+ median := func(xs []time.Duration) time.Duration {

+ sort.Slice(xs, func(i, j int) bool { return xs[i] < xs[j] })

+ return xs[len(xs)/2]

+ }

+ t.Logf("median POST /signin over budget: member %v, outsider %v", median(mt), median(ot))

+

+ a.signinMail.wait()

+ for _, to := range mail.sent() {

+ if strings.HasPrefix(to, "outsider") {

+ t.Fatalf("an over-budget outsider was mailed (%s): the budgets must still hold", to)

+ }

+}

diff --git a/internal/calls/signin_mail.go b/internal/calls/signin_mail.go

new file mode 100644

index 0000000..27fba4d

--- /dev/null

+++ b/internal/calls/signin_mail.go

@@ -0,0 +1,304 @@

+package calls

+

+import (

+ "context"

+ "fmt"

+ "log/slog"

+ "sync"

+ "sync/atomic"

+ "time"

+

+ "amadan.net/rastrillo/rastrillo/mail"

+)

+

+// 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, 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). 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.

+//

+// 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.)

+//

+// 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.

+//

+// Nothing but the log pump ever writes to the logger. The worker and

+// close hand their lines to logf, which never blocks: a buffered

+// channel, and a counted drop when it is full. A stalled log sink can

+// therefore hold only the pump's one write — never a request, never the

+// worker, never close (Astra on #20, round 3). The pump ends when the

+// worker does.

+//

+// close drains what is queued within drainTimeout. Past it, the send

+// in flight is cancelled through its context, everything still queued

+// 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. 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

+ // 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

+

+ 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

+

+ 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

+ logs chan logLine // to the pump; logf never blocks on it

+ unlogged atomic.Int64 // lines handed over and not yet written

+ lostLogs atomic.Int64 // lines logf could not hand over

+ stuck atomic.Bool // close gave up on a worker that ignored cancellation

+

+ // refusedLog, when set, is the line the worker logs (through logf,

+ // with a running count and never the address) for each link it

+ // refuses.

+ refusedLog string

+ refusals atomic.Int64

+}

+

+type logLine struct {

+ msg string

+ args []any

+}

+

+type mailJob struct {

+ to, subject, body string

+ gated bool

+}

+

+const (

+ mailQueueDepth = 64

+ mailSendTimeout = time.Minute

+ mailDrainTimeout = 30 * time.Second

+ mailAbandonGrace = 5 * time.Second

+ mailLogDepth = 32

+)

+

+func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue {

+ return newMailQueueLogging(next, admits, gate, logger, "")

+}

+

+// newMailQueueLogging is newMailQueue with refusedLog set (see the

+// field).

+func newMailQueueLogging(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger, refusedLog string) *mailQueue {

+ if logger == nil {

+ logger = slog.Default()

+ }

+ 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{}), shut: make(chan struct{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog}

+ go q.pump()

+ go q.run()

+ return q

+}

+

+// 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.dropped.Add(1)

+ return nil

+ }

+ q.pending.Add(1)

+ select {

+ case q.jobs <- job:

+ default:

+ q.pending.Done()

+ q.dropped.Add(1)

+ }

+ return nil

+}

+

+// logf hands one error line to the pump. It never blocks: when the

+// pump is behind, the line is counted in lostLogs instead.

+func (q *mailQueue) logf(msg string, args ...any) {

+ q.logMu.RLock()

+ defer q.logMu.RUnlock()

+ if q.logsClosed {

+ q.lostLogs.Add(1)

+ return

+ }

+ q.unlogged.Add(1)

+ select {

+ case q.logs <- logLine{msg: msg, args: args}:

+ default:

+ q.unlogged.Add(-1)

+ q.lostLogs.Add(1)

+ }

+}

+

+// pump is the only goroutine that writes to the logger. It ends when

+// the worker closes logs on its way out.

+//

+// Its limit, stated rather than engineered around: a write blocked

+// inside the logger (a stalled sink) cannot be interrupted from Go. It

+// can only finish, or die with the process. Closing logs does not reach

+// it mid-write, so a stalled sink leaves this one goroutine behind

+// until the sink moves. Nothing else waits on it: not Send, not the

+// worker, not close.

+func (q *mailQueue) pump() {

+ for l := range q.logs {

+ if n := q.lostLogs.Swap(0); n > 0 {

+ q.logger.Error("calls: sign-in mail log lines lost to a slow log sink", "count", n)

+ }

+ q.logger.Error(l.msg, l.args...)

+ q.unlogged.Add(-1)

+ }

+}

+

+func (q *mailQueue) run() {

+ defer close(q.done)

+ defer func() { // after the last logf below; ends the pump

+ q.logMu.Lock()

+ q.logsClosed = true

+ close(q.logs)

+ q.logMu.Unlock()

+ }()

+ abandoned := 0

+ for job := range q.jobs {

+ q.reportDrops()

+ if q.ctx.Err() != nil {

+ abandoned++

+ } else {

+ q.deliver(job)

+ }

+ q.pending.Done()

+ }

+ q.reportDrops()

+ if abandoned > 0 {

+ q.logf("calls: 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.logf("calls: 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) {

+ defer func() {

+ if p := recover(); p != nil {

+ q.logf("calls: 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.

+ if q.refusedLog != "" {

+ q.logf(q.refusedLog, "stage", "signin.gate", "refusals_since_boot", q.refusals.Add(1))

+ }

+ return

+ }

+ 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.logf("calls: send sign-in link", "err", err)

+ }

+}

+

+// wait blocks until every link queued so far has been sent or refused,

+// and every line logged about them has been written. Tests use it to

+// read the mailer, and the log, afterwards.

+func (q *mailQueue) wait() {

+ q.pending.Wait()

+ for q.unlogged.Load() > 0 {

+ time.Sleep(time.Millisecond)

+ }

+}

+

+// 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). 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:

+ return

+ case <-time.After(q.drainTimeout):

+ }

+ q.cancel()

+ select {

+ case <-q.done:

+ case <-time.After(mailAbandonGrace):

+ // Counted, not written: logging here could hang close on the

+ // very sink that may be the trouble. The worker is still running,

+ // so logf is still safe to call, and never blocks.

+ q.stuck.Store(true)

+ q.logf("calls: sign-in mail worker ignored cancellation; left running")

+ }

+}

+

+// mayMail is the queue's gate: gateBegin's mayRequestLink answer for

+// the request that asked for the link (ruling 12's budgets), carried

+// through Begin on the request context. A context lookup, no I/O.

+func mayMail(ctx context.Context) bool {

+ may, _ := ctx.Value(mayMailKey{}).(bool)

+ return may

+}

+

+// queueAttacher is a test mailer (another package's, so the method is

+// exported) that wants to wait for the queue in front of it before it

+// is read (callstest's recordingMailer).

+type queueAttacher interface{ AttachSigninQueue(wait func()) }

diff --git a/internal/calls/signin_mail_test.go b/internal/calls/signin_mail_test.go

new file mode 100644

index 0000000..999a8ad

--- /dev/null

+++ b/internal/calls/signin_mail_test.go

@@ -0,0 +1,370 @@

+package calls

+

+import (

+ "context"

+ "fmt"

+ "log/slog"

+ "net/http"

+ "path/filepath"

+ "runtime"

+ "strings"

+ "sync"

+ "sync/atomic"

+ "testing"

+ "time"

+

+ "amadan.net/rastrillo/rastrillo/db"

+)

+

+// stuckHandler is a slog handler whose every write blocks until release

+// is closed: a log sink that has stalled, as a full pipe does. It

+// remembers what it was finally allowed to write.

+type stuckHandler struct {

+ release chan struct{}

+ mu *sync.Mutex

+ msgs *[]string

+}

+

+func (h stuckHandler) Enabled(context.Context, slog.Level) bool { return true }

+func (h stuckHandler) Handle(_ context.Context, r slog.Record) error {

+ <-h.release

+ h.mu.Lock()

+ *h.msgs = append(*h.msgs, r.Message)

+ h.mu.Unlock()

+ 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

+ })

+ var mu sync.Mutex

+ var msgs []string

+ q := newMailQueue(stuck, everyone, nil, slog.New(stuckHandler{release, &mu, &msgs}))

+ 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")

+ }

+ // The drops are the worker's to log, through the pump, once the

+ // sink is free.

+ free()

+ q.close()

+ deadline := time.Now().Add(2 * time.Second)

+ for time.Now().Before(deadline) {

+ mu.Lock()

+ for _, m := range msgs {

+ if strings.Contains(m, "links dropped") {

+ mu.Unlock()

+ return

+ }

+ mu.Unlock()

+ time.Sleep(5 * time.Millisecond)

+ }

+ t.Fatalf("no drop was logged with twice the queue's depth sent; the log has %q", msgs)

+}

+

+// 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)

+ }

+}

+

+// 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

+ 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.

+func mailWorkers() int {

+ buf := make([]byte, 1<<22)

+ stacks := string(buf[:runtime.Stack(buf, true)])

+ return strings.Count(stacks, "(*mailQueue).run(") + strings.Count(stacks, "(*mailQueue).pump(")

+}

+

+// TestMailQueueCloseIsBoundedWithAStalledLogger: close reports through

+// nothing that can block, and never waits on the log pump. With the log

+// sink stalled — every write hangs — and a relay that never answers but

+// honours cancellation, and WHILE THE SINK IS STILL STALLED:

+//

+// - close returns within its bound;

+// - the mail worker has exited;

+// - at most the one log pump remains, blocked in its write.

+//

+// A write blocked inside the logger cannot be interrupted from Go; it

+// can only finish or die with the process, and shutdown deliberately

+// does not wait for it. So only once the sink is released must the pump

+// exit and the goroutine count return to its baseline (Astra on #20,

+// rounds 3 and 4). The test fails if close waited on the pump, or if

+// the pump could hold the worker.

+func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) {

+ waitFor := func(cond func() bool) bool {

+ deadline := time.Now().Add(2 * time.Second)

+ for !cond() && time.Now().Before(deadline) {

+ time.Sleep(5 * time.Millisecond)

+ }

+ return cond()

+ }

+ // Baselines, not zero: another test's queue may still be shutting

+ // down beside this one.

+ baseline, workers := runtime.NumGoroutine(), mailWorkers()

+ release := make(chan struct{})

+ var once sync.Once

+ free := func() { once.Do(func() { close(release) }) }

+ t.Cleanup(free)

+ var mu sync.Mutex

+ var msgs []string

+ honours := funcSender(func(ctx context.Context, _ string) error {

+ <-ctx.Done()

+ return ctx.Err()

+ })

+ q := newMailQueue(honours, everyone, nil, slog.New(stuckHandler{release, &mu, &msgs}))

+ q.drainTimeout = 50 * time.Millisecond

+ for i := 0; i < 2*mailQueueDepth; i++ { // drops too, so there is plenty to log

+ _ = q.Send(context.Background(), fmt.Sprintf("p%d@example.test", i), "s", "b")

+ }

+ closed := make(chan struct{})

+ go func() { q.close(); close(closed) }()

+ select {

+ case <-closed:

+ case <-time.After(2 * time.Second):

+ free()

+ <-closed

+ t.Fatal("close hung on a stalled log sink")

+ }

+

+ // The sink is still stalled.

+ select {

+ case <-q.done:

+ default:

+ t.Fatal("the worker was still running after close, held by the stalled log sink")

+ }

+ if !waitFor(func() bool { return runtime.NumGoroutine() <= baseline+1 }) {

+ t.Fatalf("%d goroutines with the sink stalled, %d before: more than the one blocked log pump remains", runtime.NumGoroutine(), baseline)

+ }

+ if n := mailWorkers(); n > workers+1 {

+ t.Fatalf("%d mail queue goroutines with the sink stalled, %d before: want at most the log pump more", n, workers)

+ }

+

+ // Released, the pump finishes its write and exits.

+ free()

+ if !waitFor(func() bool { return mailWorkers() <= workers && runtime.NumGoroutine() <= baseline }) {

+ t.Fatalf("after release: %d mail queue goroutines, %d goroutines against %d before", mailWorkers(), runtime.NumGoroutine(), baseline)

+ }

+}

+

+// 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)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { d.Close() })

+ ctx, cancel := context.WithCancel(context.Background())

+ _, a, err := build(ctx, d, Config{Env: EnvDevelopment, Origin: "http://app.test", Logger: logger})

+ if err != nil {

+ 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 close")

+ }

+}

diff --git a/internal/calls/signin_timing_test.go b/internal/calls/signin_timing_test.go

new file mode 100644

index 0000000..707a18d

--- /dev/null

+++ b/internal/calls/signin_timing_test.go

@@ -0,0 +1,94 @@

+package calls

+

+import (

+ "bytes"

+ "context"

+ "encoding/base64"

+ "io"

+ "log/slog"

+ "net/http/httptest"

+ "net/url"

+ "path/filepath"

+ "sync"

+ "testing"

+ "time"

+

+ "amadan.net/rastrillo/rastrillo/db"

+)

+

+// blockingMailer holds every Send until release is closed — an SMTP

+// relay that takes as long as the test likes.

+type blockingMailer struct {

+ release chan struct{}

+ mu sync.Mutex

+ to []string

+}

+

+func (m *blockingMailer) Send(_ context.Context, to, _, _ string) error {

+ <-m.release

+ m.mu.Lock()

+ m.to = append(m.to, to)

+ m.mu.Unlock()

+ return nil

+}

+

+func (m *blockingMailer) sent() []string {

+ m.mu.Lock()

+ defer m.mu.Unlock()

+ return append([]string(nil), m.to...)

+}

+

+// TestSigninNeverWaitsForTheMailer: when the send ran inside the

+// request, a member waited for SMTP before its ?sent=1 and a stranger

+// did not (Astra measured 252 ms against 0.35 ms on Slides, which had

+// the same shape). With the mailer blocked, every POST — member,

+// stranger, and both with force=1 — must answer, and the member's

+// links must then go out.

+func TestSigninNeverWaitsForTheMailer(t *testing.T) {

+ logger := slog.New(slog.NewTextHandler(io.Discard, nil))

+ d, err := db.Open(filepath.Join(t.TempDir(), "calls.db"), logger)

+ if err != nil {

+ t.Fatal(err)

+ }

+ t.Cleanup(func() { d.Close() })

+ bm := &blockingMailer{release: make(chan struct{})}

+ var once sync.Once

+ free := func() { once.Do(func() { close(bm.release) }) }

+ t.Cleanup(free)

+ 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)

+ }

+ f := &keymailFixture{handler: mux}

+

+ for _, form := range []url.Values{

+ {"address": {plainMember}},

+ {"address": {"someone@else.test"}},

+ {"address": {plainMember}, "force": {"1"}},

+ {"address": {"someone@else.test"}, "force": {"1"}},

+ } {

+ done := make(chan *httptest.ResponseRecorder, 1)

+ go func() { done <- f.post(form) }()

+ select {

+ case rec := <-done:

+ if rec.Code != 303 || rec.Header().Get("Location") != "/signin?sent=1" {

+ t.Errorf("POST %v: %d %q, want 303 /signin?sent=1", form, rec.Code, rec.Header().Get("Location"))

+ }

+ case <-time.After(2 * time.Second):

+ t.Errorf("POST %v waited on the mailer: the response time says the address is a member", form)

+ free()

+ <-done

+ }

+

+ free()

+ deadline := time.Now().Add(5 * time.Second)

+ for len(bm.sent()) < 2 && time.Now().Before(deadline) {

+ time.Sleep(10 * time.Millisecond)

+ }

+ time.Sleep(50 * time.Millisecond)

+ if got := bm.sent(); len(got) != 2 || got[0] != plainMember || got[1] != plainMember {

+ t.Fatalf("mail sent = %v, want two links, both to %s", got, plainMember)

+ }

+}

diff --git a/internal/calls/templates/signin.html b/internal/calls/templates/signin.html

index c5f7a3b..06fca6b 100644

--- a/internal/calls/templates/signin.html

+++ b/internal/calls/templates/signin.html

@@ -1,6 +1,9 @@

{{/* The Oficina door (estilo chrome): one card on the suite's sky, Calls'

mark and name at its head. The form, or once a link has been asked

- for, the check-your-email notice. auth.go's localSignin builds both. */}}

+ for, the check-your-email notice — or, on the way to Keymail, the

+ door's notice with .Continue's meta refresh in head (renderKeymail).

+ auth.go's localSignin builds the first two. */}}

+{{define "head"}}{{with .Continue}}{{template "home-oauth-head" .}}{{end}}{{end}}

diff --git a/internal/callstest/browser_test.go b/internal/callstest/browser_test.go

index 974da52..20acda9 100644

--- a/internal/callstest/browser_test.go

+++ b/internal/callstest/browser_test.go

@@ -419,6 +419,7 @@ func (rig *browserRig) serve(t *testing.T, ln net.Listener) {

if err != nil {

t.Fatalf("calls.NewForTest: %v", err)

}

+ t.Cleanup(ctl.ShutdownForTest) // stops the sign-in mail worker too

srv := &http.Server{Handler: mux}

go func() { _ = srv.Serve(ln) }()

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/harness_test.go b/internal/callstest/harness_test.go

index a72510f..85e9734 100644

--- a/internal/callstest/harness_test.go

+++ b/internal/callstest/harness_test.go

@@ -163,6 +163,25 @@ type recordingMailer struct {

to []string

subjects []string

links []string

+ // waits are the sign-in mail queues in front of this recorder:

+ // links go out after the response (signin_mail.go), so every read

+ // waits for them first.

+ waits []func()

+}

+

+func (m *recordingMailer) AttachSigninQueue(wait func()) {

+ m.mu.Lock()

+ m.waits = append(m.waits, wait)

+ m.mu.Unlock()

+}

+

+func (m *recordingMailer) flush() {

+ m.mu.Lock()

+ waits := append([]func(){}, m.waits...)

+ m.mu.Unlock()

+ for _, w := range waits {

+ w()

+ }

}

var magicLink = regexp.MustCompile(`https?://\S+/auth/verify\?token=\S+`)

@@ -177,6 +196,7 @@ func (m *recordingMailer) Send(ctx context.Context, to, subject, body string) er

}

func (m *recordingMailer) last() (string, string) {

+ m.flush()

m.mu.Lock()

defer m.mu.Unlock()

if len(m.to) == 0 {

@@ -187,6 +207,7 @@ func (m *recordingMailer) last() (string, string) {

func (m *recordingMailer) lastSubject(t *testing.T) string {

t.Helper()

+ m.flush()

m.mu.Lock()

defer m.mu.Unlock()

if len(m.subjects) == 0 {

@@ -196,6 +217,7 @@ func (m *recordingMailer) lastSubject(t *testing.T) string {

}

func (m *recordingMailer) count() int {

+ m.flush()

m.mu.Lock()

defer m.mu.Unlock()

return len(m.to)

@@ -248,6 +270,9 @@ func newHarnessWith(t *testing.T, env string, logger *slog.Logger, adjust func(*

if err != nil {

t.Fatal(err)

}

+ // Ends the app's context, which stops its sign-in mail worker

+ // (signin_mail.go): no test leaves one running.

+ t.Cleanup(ctl.ShutdownForTest)

// The base is now, not a fixed date. Calls' clock is the only one a

// test can move; rastrillo stamps its auth_links rows from the real

// one. A fixed base put the two most of a day apart, so nothing that

diff --git a/internal/callstest/join_test.go b/internal/callstest/join_test.go

index a129127..9a5efba 100644

--- a/internal/callstest/join_test.go

+++ b/internal/callstest/join_test.go

@@ -103,9 +103,11 @@ func TestLinkReplacedBetweenMailAndClickAdmitsNobody(t *testing.T) {

member.signIn("paul@example.com")

member.post("/meetings/"+sqid+"/links", nil)

u, _ := url.Parse(link)

+ // Authorize refuses it, and the refusal is the door rendered in a

+ // 403 (doorOnRefusal), not rastrillo's bare text.

rec := g.get(u.RequestURI())

- if rec.Code != http.StatusForbidden {

- t.Fatalf("outsider's stale click: %d, want 403 from Authorize", rec.Code)

+ if rec.Code != http.StatusForbidden || !strings.Contains(rec.Body.String(), "can’t sign in to Calls") {

+ t.Fatalf("outsider's stale click: %d, want Authorize's refusal as the door in a 403", rec.Code)

}

if rec := g.get("/m/" + sqid + "/join"); rec.Code == http.StatusOK {

t.Fatal("a grant was minted through a replaced link")

@@ -395,8 +397,8 @@ func TestExpiredPendingRowAdmitsNobody(t *testing.T) {

if err != nil {

t.Fatal(err)

}

- if rec := g.get(u.RequestURI()); rec.Code != http.StatusForbidden {

- t.Fatalf("expired pending row click: %d, want 403 from Authorize", rec.Code)

+ if rec := g.get(u.RequestURI()); rec.Code != http.StatusForbidden || !strings.Contains(rec.Body.String(), "can’t sign in to Calls") {

+ t.Fatalf("expired pending row click: %d, want Authorize's refusal as the door in a 403", rec.Code)

}

if got := x.mail.count(); got != before {

t.Fatalf("mail count changed after the expired click: %d, want %d", got, before)

diff --git a/internal/callstest/main_test.go b/internal/callstest/main_test.go

new file mode 100644

index 0000000..8abb9fc

--- /dev/null

+++ b/internal/callstest/main_test.go

@@ -0,0 +1,49 @@

+package callstest

+

+import (

+ "fmt"

+ "os"

+ "runtime"

+ "strings"

+ "testing"

+ "time"

+

+ "amadan.net/oficina/calls/internal/calls/keymailtest"

+)

+

+// TestMain keeps Keymail classification inside the process: every POST

+// /signin now classifies its address (auth.go's gateBegin), and without

+// this each one would ask the real network whether example.com runs

+// Keymail. Every DNS query is answered NXDOMAIN; no fixture is

+// registered here, so every address is plain email.

+func TestMain(m *testing.M) {

+ restore, err := keymailtest.Install()

+ if err != nil {

+ fmt.Fprintln(os.Stderr, "keymailtest.Install:", err)

+ os.Exit(1)

+ }

+ code := m.Run()

+ // No test may leave a sign-in mail worker running: every app a test

+ // 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) {

+ time.Sleep(10 * time.Millisecond)

+ }

+ if n := mailWorkers(); n > 0 {

+ fmt.Fprintf(os.Stderr, "FAIL: %d sign-in mail queue goroutines still running after the tests\n", n)

+ code = 1

+ }

+ restore()

+ os.Exit(code)

+}

+

+// mailWorkers counts goroutines running a sign-in mail queue's worker

+// or log pump (internal/calls/signin_mail.go).

+func mailWorkers() int {

+ buf := make([]byte, 1<<22)

+ stacks := string(buf[:runtime.Stack(buf, true)])

+ return strings.Count(stacks, "(*mailQueue).run(") + strings.Count(stacks, "(*mailQueue).pump(")

+}

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 9d67806..f4281c1 100644

--- a/internal/callstest/signin_test.go

+++ b/internal/callstest/signin_test.go

@@ -88,14 +88,15 @@ 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{

+ failing := &failingMailer{err: fmt.Errorf("rastrillo/mail: send to %s: %w", address, errFakeSMTPFailure)}

+ 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" {

@@ -103,6 +104,7 @@ func TestSigninOverAFailingMailerLeavesNoAddressInTheLog(t *testing.T) {

rec.Code, rec.Header().Get("Location"))

}

+ failing.flush() // the send, and its failure, happen after the response

log := buf.String()

if strings.Contains(strings.ToLower(log), strings.ToLower(address)) {

t.Fatalf("the log carries the address after a failing send:\n%s", log)

@@ -114,9 +116,20 @@ func TestSigninOverAFailingMailerLeavesNoAddressInTheLog(t *testing.T) {

var errFakeSMTPFailure = errors.New("550 mailbox unavailable")

-type failingMailer struct{ err error }

+type failingMailer struct {

+ err error

+ waits []func() // the sign-in mail queue in front of it

+}

+

+func (f *failingMailer) Send(ctx context.Context, to, subject, body string) error { return f.err }

+

+func (f *failingMailer) AttachSigninQueue(wait func()) { f.waits = append(f.waits, wait) }

-func (f failingMailer) Send(ctx context.Context, to, subject, body string) error { return f.err }

+func (f *failingMailer) flush() {

+ for _, w := range f.waits {

+ w()

+ }

+}

// Re-review of 9e067e4: a plain member sign-in — no meeting-link cookie,

// so no pending_joins row at all — still leaves rastrillo's own

oficina / meet Public

Calls signs in through Keymail, without revealing who is admitted