amadan
Equal work over budget, close never waits on the log, the refusal is the door itself
Astra's round 3 on #25.
- P1, a timing oracle in mayRequestLink: with a live invitation and the
outsider budgets spent, a member wrote a pending row while an outsider
wrote a log line and returned (0.4-0.65 ms against 77-78 ms). Now a
live cookie costs the same whoever the address names: the link
lookup and the pending_joins row (written for an over-budget outsider
too, whose answer still says no link). Nothing is logged on the
response path; over-budget attempts are counted and the hourly sweep
reports them. TestMemberAndOutsiderDoTheSameWorkOverBudget re-runs the
probe and counts statements and log lines: 3 and 0 against 2 and 1 on
c066739, equal and zero now; medians 272 µs and 272 µs.
- P2, close could hang on a stalled log sink. Only a log pump writes to
the logger; the worker and close use logf, which never blocks. With
the sink still stalled, close returns in its bound, the worker is
gone, and at most the pump remains; released, that goes too. The pump
and close say the limit: a blocked write cannot be interrupted, and
shutdown does not wait for it. Fails on c066739 ("close hung").
- TestMain in both packages fails the run if a queue worker or pump
outlives the tests; the test apps built on context.Background now use
t.Context or shut down in cleanup.
- P2, ?err=refused could be typed by anybody. doorOnRefusal renders the
door in the refusal's own 403, marked through the request context.
TestKeymailRefusalIsTheDoor and the two join tests require it, and
that the query says nothing.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
12 files changed,
+477
−57
AGENTS.md+4 −1internal/calls/app.go+11 −5internal/calls/app_relay_test.go+4 −1internal/calls/auth.go+31 −14internal/calls/auth_refusal.go+42 −8internal/calls/auth_test.go+1 −1internal/calls/keymail_signin_test.go+20 −3internal/calls/signin_equal_work_test.go+139 −0internal/calls/signin_mail.go+104 −12internal/calls/signin_mail_test.go+92 −6internal/callstest/join_test.go+6 −6internal/callstest/main_test.go+23 −0
CI failed — run details
CI log
=== vet ===
go vet ./...
=== fmt ===
=== test ===
go test ./...
ok amadan.net/oficina/calls/callscore (cached)
? amadan.net/oficina/calls/cmd/calls [no test files]
ok amadan.net/oficina/calls/cmd/quality-summary (cached)
ok amadan.net/oficina/calls/internal/calls 1.937s
ok amadan.net/oficina/calls/internal/calls/apns (cached)
ok amadan.net/oficina/calls/internal/calls/hub (cached)
? amadan.net/oficina/calls/internal/calls/icons [no test files]
? amadan.net/oficina/calls/internal/calls/keymailtest [no test files]
ok amadan.net/oficina/calls/internal/calls/relay (cached)
ok amadan.net/oficina/calls/internal/calls/seal (cached)
ok amadan.net/oficina/calls/internal/callstest 9.965s
? amadan.net/oficina/calls/protocol [no test files]
? amadan.net/oficina/calls/scripts/writeestilo [no test files]
=== migration-check ===
go run amadan.net/rastrillo/rastrillo/cmd/rastrillo migration check
rastrillo migration check: models and migrations agree
=== race ===
CGO_ENABLED=1 go test -race ./internal/...
2026/10/06 20:31:47 INFO migrate applied="[sessions/0001_init auth/0001_init auth/0002_sessions_backfill calls/0001_init calls/0002_agents calls/0003_alter_meeting_links calls/0004_alter_persons calls/0005_create_device_tokens calls/0006_alter_meetings]" skipped=0 adopted=false
2026/10/06 20:31:47 INFO migrate applied="[sessions/0001_init auth/0001_init auth/0002_sessions_backfill calls/0001_init calls/0002_agents calls/0003_alter_meeting_links calls/0004_alter_persons calls/0005_create_device_tokens calls/0006_alter_meetings]" skipped=0 adopted=false
2026/10/06 20:31:47 INFO migrate applied="[sessions/0001_init auth/0001_init auth/0002_sessions_backfill calls/0001_init calls/0002_agents calls/0003_alter_meeting_links calls/0004_alter_persons calls/0005_create_device_tokens calls/0006_alter_meetings]" skipped=0 adopted=false
--- FAIL: TestMemberAndOutsiderDoTheSameWorkOverBudget (0.10s)
signin_equal_work_test.go:115: member did 6 statements and 0 log lines, outsider 3 and 0: want the same work and no log line on the response path
FAIL
FAIL amadan.net/oficina/calls/internal/calls 11.185s
ok amadan.net/oficina/calls/internal/calls/apns (cached)
ok amadan.net/oficina/calls/internal/calls/hub (cached)
? amadan.net/oficina/calls/internal/calls/icons [no test files]
? amadan.net/oficina/calls/internal/calls/keymailtest [no test files]
ok amadan.net/oficina/calls/internal/calls/relay (cached)
ok amadan.net/oficina/calls/internal/calls/seal (cached)
ok amadan.net/oficina/calls/internal/callstest 29.728s
FAIL
make: *** [Makefile:79: race] Error 1
amadan: step race failed: exit status 2
amadan: job failed at step race — its output is above, not at the tail of this log
diff --git a/AGENTS.md b/AGENTS.md| index 3049b48..a256bde 100644 |
| --- a/AGENTS.md |
| +++ b/AGENTS.md |
| @@ -40,7 +40,10 @@ mechanically. |
| `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 with `?err=refused`. |
| + 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/internal/calls/app.go b/internal/calls/app.go| index 80a12b8..2c0c95d 100644 |
| --- a/internal/calls/app.go |
| +++ b/internal/calls/app.go |
| @@ -41,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 |
| @@ -253,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) |
| @@ -494,14 +500,14 @@ func build(ctx context.Context, d *db.DB, cfg Config) (*http.ServeMux, *app, err |
| r.Get(signinPath, a.signin) |
| if a.home == nil { |
| r.Post(signinPath, a.newBegin(authPlugin.Begin)) |
| - r.Get("/auth/callback", doorOnRefusal(authPlugin.Callback, signinPath)) |
| - r.Get("/auth/verify", doorOnRefusal(authPlugin.Verify, signinPath)) |
| + 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.newBegin(authPlugin.Begin))) |
| - r.Get("/auth/callback", a.home.routes.LocalGuard(doorOnRefusal(authPlugin.Callback, "/signin/local"))) |
| - r.Get("/auth/verify", a.home.routes.LocalGuard(doorOnRefusal(authPlugin.Verify, "/signin/local"))) |
| + 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) |
diff --git a/internal/calls/app_relay_test.go b/internal/calls/app_relay_test.go| index d0a7cbc..b8d0402 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 := build(t.Context(), 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 78bd7ac..d90cdcc 100644 |
| --- a/internal/calls/auth.go |
| +++ b/internal/calls/auth.go |
| @@ -124,12 +124,26 @@ func (a *app) renderKeymail(w http.ResponseWriter, r *http.Request, destination |
| // 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 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). |
| +// 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) |
| @@ -145,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 == "" { |
| @@ -161,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 |
| @@ -259,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 |
| @@ -316,14 +331,16 @@ func (a *app) localSignin(w http.ResponseWriter, r *http.Request) { |
| 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 "refused": |
| - // Keymail or a link verified the address and Calls does not |
| - // admit it (doorOnRefusal). 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." |
| 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| index 88e3ff6..d668f1e 100644 |
| --- a/internal/calls/auth_refusal.go |
| +++ b/internal/calls/auth_refusal.go |
| @@ -2,6 +2,7 @@ package calls |
| import ( |
| "bytes" |
| + "context" |
| "net/http" |
| "strings" |
| ) |
| @@ -14,12 +15,17 @@ import ( |
| 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: a 303 |
| -// to the local form with ?err=refused, which says the address cannot |
| -// sign in here and offers the form for a different one. Every other |
| -// answer, cookies included, passes through untouched. Remove it when |
| -// rastrillo has a refusal hook and the pin is bumped. |
| -func doorOnRefusal(next http.HandlerFunc, form string) http.HandlerFunc { |
| +// 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) |
| @@ -27,8 +33,8 @@ func doorOnRefusal(next http.HandlerFunc, form string) http.HandlerFunc { |
| for _, c := range rec.header.Values("Set-Cookie") { |
| w.Header().Add("Set-Cookie", c) |
| } |
| - w.Header().Set("Cache-Control", "no-store") |
| - http.Redirect(w, r, form+"?err=refused", http.StatusSeeOther) |
| + door(&forcedStatus{ResponseWriter: w, status: http.StatusForbidden}, |
| + r.WithContext(context.WithValue(r.Context(), refusedKey{}, true))) |
| return |
| } |
| for name, values := range rec.header { |
| @@ -44,6 +50,34 @@ func doorOnRefusal(next http.HandlerFunc, form string) http.HandlerFunc { |
| } |
| } |
| +// 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. |
diff --git a/internal/calls/auth_test.go b/internal/calls/auth_test.go| index 629dd8c..4375e98 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 := build(t.Context(), d, Config{ |
| Env: EnvDevelopment, Origin: "http://app.test", |
| AllowedAddresses: members, Logger: logger, |
| }) |
diff --git a/internal/calls/keymail_signin_test.go b/internal/calls/keymail_signin_test.go| index 72da2bf..9ba5be4 100644 |
| --- a/internal/calls/keymail_signin_test.go |
| +++ b/internal/calls/keymail_signin_test.go |
| @@ -17,6 +17,7 @@ import ( |
| "strings" |
| "sync" |
| "testing" |
| + "time" |
| "amadan.net/oficina/calls/internal/calls/keymailtest" |
| "amadan.net/oficina/home/connect" |
| @@ -34,6 +35,18 @@ func TestMain(m *testing.M) { |
| os.Exit(1) |
| } |
| code := m.Run() |
| + // No test may leave a sign-in mail worker running: every app a test |
| + // builds ends with its context. |
| + 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) |
| } |
| @@ -249,15 +262,19 @@ func TestKeymailRefusalIsTheDoor(t *testing.T) { |
| t.Fatalf("member callback: %d %q", cb.Code, cb.Header().Get("Location")) |
| } |
| cb := roundTrip(keymailStranger) |
| - if cb.Code != http.StatusSeeOther || cb.Header().Get("Location") != form+"?err=refused" { |
| - t.Fatalf("stranger callback: %d %q, want 303 %s?err=refused (bare text is the bug)\n%s", cb.Code, cb.Header().Get("Location"), form, cb.Body.String()) |
| + 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 := f.get(form + "?err=refused").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/signin_equal_work_test.go b/internal/calls/signin_equal_work_test.go| new file mode 100644 |
| index 0000000..1d58b1d |
| --- /dev/null |
| +++ b/internal/calls/signin_equal_work_test.go |
| @@ -0,0 +1,139 @@ |
| +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" |
| +) |
| + |
| +// 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 := build(t.Context(), 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 |
| + count := func(*gorm.DB) { 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" |
| + 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| index 717eaac..4340e88 100644 |
| --- a/internal/calls/signin_mail.go |
| +++ b/internal/calls/signin_mail.go |
| @@ -38,11 +38,19 @@ import ( |
| // 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 with a log line, and close waits a short grace for the |
| +// 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, and that is logged too. |
| +// worker outlives close; that is counted in stuck and logged, best |
| +// effort, too. |
| type mailQueue struct { |
| next mail.Sender |
| admits func(to string) bool |
| @@ -63,6 +71,24 @@ type mailQueue struct { |
| ctx context.Context // cancelled when close gives up waiting |
| cancel context.CancelFunc |
| done chan struct{} |
| + |
| + 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 { |
| @@ -75,9 +101,16 @@ const ( |
| 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() |
| } |
| @@ -85,7 +118,8 @@ func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context. |
| 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{})} |
| + done: make(chan struct{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog} |
| + go q.pump() |
| go q.run() |
| return q |
| } |
| @@ -114,8 +148,51 @@ func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error { |
| 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() |
| @@ -128,7 +205,7 @@ func (q *mailQueue) run() { |
| } |
| q.reportDrops() |
| if abandoned > 0 { |
| - q.logger.Error("calls: sign-in links abandoned at shutdown", "count", abandoned) |
| + q.logf("calls: sign-in links abandoned at shutdown", "count", abandoned) |
| } |
| } |
| @@ -136,7 +213,7 @@ func (q *mailQueue) run() { |
| // Here, on the worker, so the request path never writes a log line. |
| func (q *mailQueue) reportDrops() { |
| if n := q.dropped.Swap(0); n > 0 { |
| - q.logger.Error("calls: sign-in mail queue full; links dropped", "count", n) |
| + q.logf("calls: sign-in mail queue full; links dropped", "count", n) |
| } |
| } |
| @@ -145,28 +222,39 @@ func (q *mailQueue) reportDrops() { |
| func (q *mailQueue) deliver(job mailJob) { |
| defer func() { |
| if p := recover(); p != nil { |
| - q.logger.Error("calls: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| + 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.logger.Error("calls: send sign-in link", "err", err) |
| + q.logf("calls: send sign-in link", "err", err) |
| } |
| } |
| -// wait blocks until every link queued so far has been sent or refused. |
| -// Tests use it to read the mailer afterwards. |
| -func (q *mailQueue) wait() { q.pending.Wait() } |
| +// 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). |
| +// 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). |
| func (q *mailQueue) close() { |
| q.mu.Lock() |
| if q.closed { |
| @@ -186,7 +274,11 @@ func (q *mailQueue) close() { |
| select { |
| case <-q.done: |
| case <-time.After(mailAbandonGrace): |
| - q.logger.Error("calls: sign-in mail worker ignored cancellation; left running") |
| + // 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") |
| } |
| } |
diff --git a/internal/calls/signin_mail_test.go b/internal/calls/signin_mail_test.go| index 0502e96..923dfff 100644 |
| --- a/internal/calls/signin_mail_test.go |
| +++ b/internal/calls/signin_mail_test.go |
| @@ -78,15 +78,21 @@ func TestMailQueueSendNeverBlocksWhenFull(t *testing.T) { |
| <-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, once it is free to. |
| + // The drops are the worker's to log, through the pump, once the |
| + // sink is free. |
| free() |
| q.close() |
| - mu.Lock() |
| - defer mu.Unlock() |
| - for _, m := range msgs { |
| - if strings.Contains(m, "links dropped") { |
| - return |
| + 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) |
| } |
| @@ -182,6 +188,86 @@ func TestMailQueueCloseIsBoundedAndStopsTheWorker(t *testing.T) { |
| } |
| } |
| +// 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) |
| + } |
| +} |
| + |
| // TestShutdownStopsTheMailWorker: the queue lives as long as the app's |
| // context, and ending it stops the worker, so no test that cancels its |
| // app leaves one running. |
diff --git a/internal/callstest/join_test.go b/internal/callstest/join_test.go| index 347e28e..9a5efba 100644 |
| --- a/internal/callstest/join_test.go |
| +++ b/internal/callstest/join_test.go |
| @@ -103,11 +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 (doorOnRefusal), |
| - // not rastrillo's bare-text 403. |
| + // 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.StatusSeeOther || rec.Header().Get("Location") != "/signin?err=refused" { |
| - t.Fatalf("outsider's stale click: %d %q, want Authorize's refusal as 303 /signin?err=refused", rec.Code, rec.Header().Get("Location")) |
| + 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") |
| @@ -397,8 +397,8 @@ func TestExpiredPendingRowAdmitsNobody(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| - if rec := g.get(u.RequestURI()); rec.Code != http.StatusSeeOther || rec.Header().Get("Location") != "/signin?err=refused" { |
| - t.Fatalf("expired pending row click: %d %q, want Authorize's refusal as 303 /signin?err=refused", rec.Code, rec.Header().Get("Location")) |
| + 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| index cce6da6..c339a28 100644 |
| --- a/internal/callstest/main_test.go |
| +++ b/internal/callstest/main_test.go |
| @@ -3,7 +3,10 @@ package callstest |
| import ( |
| "fmt" |
| "os" |
| + "runtime" |
| + "strings" |
| "testing" |
| + "time" |
| "amadan.net/oficina/calls/internal/calls/keymailtest" |
| ) |
| @@ -20,6 +23,26 @@ func TestMain(m *testing.M) { |
| os.Exit(1) |
| } |
| code := m.Run() |
| + // No test may leave a sign-in mail worker running: every app a test |
| + // builds ends with its context. |
| + 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(") |
| +} |