amadan
Contain a panicking sender, close never waits on the log, the refusal is the door itself
Astra's round 3 on #28.
- P2, the bounded sender runs the real send on its own goroutine,
outside the mail queue's recover; a panicking mailer took the process
down (Astra's production-composition probe). That goroutine now
recovers and answers ErrMailRefused. TestAPanickingMailerDoesNotStop
SigninMail builds the app as New does, panics the first send and
requires the next sign-in's link; it crashed the test binary before.
- 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. The
refusal count moves into the queue (newMailQueueLogging) so it, too,
goes through logf rather than blocking the worker.
TestMailQueueCloseIsBoundedWithAStalledLogger fails on the previous
queue ("close hung on a stalled log sink"). TestMain fails the run if
a mailQueue worker or pump outlives the tests.
- P2, ?err=refused could be typed by anybody. doorOnRefusal renders the
door in the refusal's own 403, marked through the request context;
TestKeymailRefusalIsTheDoor requires it in both modes, and that the
query says nothing; it fails on b9b6c91.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9 files changed,
+310
−59
internal/calendar/app.go+3 −1internal/calendar/auth_keymail.go+42 −7internal/calendar/keymail_signin_test.go+24 −6internal/calendar/mail_bounded.go+9 −0internal/calendar/mail_bounded_panic_test.go+61 −0internal/calendar/routes.go+2 −6internal/calendar/signin.go+6 −4internal/calendar/signin_mail.go+94 −29internal/calendar/signin_mail_test.go+69 −6
CI passed — run details
CI log
=== vet === go vet ./... go vet -tags browser ./internal/calendartest/ === fmt === === test === go test ./... ? amadan.net/oficina/calendar/cmd/calendar [no test files] ok amadan.net/oficina/calendar/internal/calendar 5.100s ? amadan.net/oficina/calendar/internal/calendar/icons [no test files] ? amadan.net/oficina/calendar/internal/calendar/keymailtest [no test files] ? amadan.net/oficina/calendar/internal/calendar/seal [no test files] ok amadan.net/oficina/calendar/internal/calendartest 2.414s ok amadan.net/oficina/calendar/internal/guardstest 0.058s ? amadan.net/oficina/calendar/scripts/writeestilo [no test files] === migration-check === go tool rastrillo migration check rastrillo migration check: models and migrations agree
diff --git a/internal/calendar/app.go b/internal/calendar/app.go| index b20ba8e..0b44f6d 100644 |
| --- a/internal/calendar/app.go |
| +++ b/internal/calendar/app.go |
| @@ -144,7 +144,9 @@ func New(cfg Config) (*App, error) { |
| mailer = newBoundedSender(mailer) |
| // Sign-in links alone go through a queue: sent after the response, |
| // to permitted addresses only (signin_mail.go). Closed by Close. |
| - signinMail := newMailQueue(mailer, allowlistAdmits(list, logger), nil, logger) |
| + // Refusals are counted in the log without the address (AGENTS.md |
| + // §Logs) — by the queue's worker, through its non-blocking logf. |
| + signinMail := newMailQueueLogging(mailer, list.allows, nil, logger, "calendar: signin refused by allowlist") |
| authPlugin, err := newAuthPlugin(origin, writer, instanceKey, signinMail, list, logger, homeSettings != nil) |
| if err != nil { |
| signinMail.close() |
diff --git a/internal/calendar/auth_keymail.go b/internal/calendar/auth_keymail.go| index 5e04311..334261a 100644 |
| --- a/internal/calendar/auth_keymail.go |
| +++ b/internal/calendar/auth_keymail.go |
| @@ -2,6 +2,7 @@ package calendar |
| import ( |
| "bytes" |
| + "context" |
| "net/http" |
| "strings" |
| @@ -51,12 +52,17 @@ func beginSignin(origin string, begin http.HandlerFunc, render func(http.Respons |
| const authRefusal = "This address is verified but not admitted here." |
| // doorOnRefusal wraps rauth'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 { |
| +// in admission, and turns its bare-text refusal into the door, rendered |
| +// in the refusal's own 403 by door (the sign-in page) 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 it is not admitted" (Astra, round 3 on #28). 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) |
| @@ -65,7 +71,8 @@ func doorOnRefusal(next http.HandlerFunc, form string) http.HandlerFunc { |
| 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 { |
| @@ -81,6 +88,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/calendar/keymail_signin_test.go b/internal/calendar/keymail_signin_test.go| index 4c94bb7..64241e0 100644 |
| --- a/internal/calendar/keymail_signin_test.go |
| +++ b/internal/calendar/keymail_signin_test.go |
| @@ -18,6 +18,7 @@ import ( |
| "strings" |
| "sync" |
| "testing" |
| + "time" |
| "amadan.net/oficina/calendar/internal/calendar/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 is closed in its cleanup. |
| + 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) |
| } |
| @@ -221,9 +234,10 @@ func TestKeymailFailureFallsBackToAnEmailLink(t *testing.T) { |
| } |
| // TestKeymailRefusalIsTheDoor: a non-permitted address Keymail verified |
| -// is refused after verification on the door, with the form for a |
| -// different address, not in rastrillo's bare text; an admitted one is |
| -// signed in. Both modes. |
| +// is refused after verification with the door, rendered in the |
| +// refusal's own 403 with the form for a different address — not |
| +// rastrillo's bare text, and not a redirect to a query anybody could |
| +// type. An admitted one 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) { |
| @@ -241,15 +255,19 @@ func TestKeymailRefusalIsTheDoor(t *testing.T) { |
| t.Fatalf("permitted 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("non-permitted 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("non-permitted 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 Calendar"} { |
| 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 Calendar") { |
| + t.Fatalf("anybody can show the refusal by visiting %s?err=refused", form) |
| + } |
| }) |
| } |
| } |
diff --git a/internal/calendar/mail_bounded.go b/internal/calendar/mail_bounded.go| index b2930eb..817e226 100644 |
| --- a/internal/calendar/mail_bounded.go |
| +++ b/internal/calendar/mail_bounded.go |
| @@ -82,6 +82,15 @@ func (b *boundedSender) Send(ctx context.Context, to, subject, body string) erro |
| // however long after this call gave up on it. That is the |
| // bound on abandoned sends: at most maxSendsInFlight of them. |
| defer func() { <-b.slots }() |
| + // A panicking sender is this send's failure, not the process's: |
| + // this goroutine is outside every caller's recover (Astra's |
| + // production-composition probe on #28 took the process down |
| + // through it). done is buffered, so this never blocks. |
| + defer func() { |
| + if p := recover(); p != nil { |
| + done <- ErrMailRefused |
| + } |
| + }() |
| done <- b.next.Send(ctx, to, subject, body) |
| }() |
diff --git a/internal/calendar/mail_bounded_panic_test.go b/internal/calendar/mail_bounded_panic_test.go| new file mode 100644 |
| index 0000000..a50f83d |
| --- /dev/null |
| +++ b/internal/calendar/mail_bounded_panic_test.go |
| @@ -0,0 +1,61 @@ |
| +package calendar |
| + |
| +import ( |
| + "context" |
| + "io" |
| + "log/slog" |
| + "net/http" |
| + "net/http/httptest" |
| + "net/url" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "testing" |
| +) |
| + |
| +// panicOnceMailer panics on its first send and records the rest. |
| +type panicOnceMailer struct { |
| + mu sync.Mutex |
| + calls int |
| + to []string |
| +} |
| + |
| +func (m *panicOnceMailer) Send(_ context.Context, to, _, _ string) error { |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + m.calls++ |
| + if m.calls == 1 { |
| + panic("relay exploded") |
| + } |
| + m.to = append(m.to, to) |
| + return nil |
| +} |
| + |
| +// TestAPanickingMailerDoesNotStopSigninMail is Astra's production- |
| +// composition probe (round 3 on #28): the sign-in queue's worker hands |
| +// each link to the bounded sender, which runs the real send on a |
| +// goroutine of its own — outside the worker's recover. A mailer that |
| +// panicked there took the whole process down. Built exactly as New |
| +// builds it, the first link's panic must be contained and the next |
| +// sign-in's link delivered. |
| +func TestAPanickingMailerDoesNotStopSigninMail(t *testing.T) { |
| + m := &panicOnceMailer{} |
| + a, err := New(Config{DBPath: filepath.Join(t.TempDir(), "calendar.db"), Env: "development", Origin: TestOrigin, |
| + AllowedAddresses: "one@example.test,two@example.test", Mailer: m, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { a.Close() }) |
| + for _, address := range []string{"one@example.test", "two@example.test"} { |
| + r := httptest.NewRequest(http.MethodPost, TestOrigin+"/signin", strings.NewReader(url.Values{"address": {address}}.Encode())) |
| + r.Header.Set("Content-Type", "application/x-www-form-urlencoded") |
| + r.Header.Set("Origin", TestOrigin) |
| + a.mux.ServeHTTP(httptest.NewRecorder(), r) |
| + a.FlushSigninMailForTest() |
| + } |
| + m.mu.Lock() |
| + defer m.mu.Unlock() |
| + if len(m.to) != 1 || m.to[0] != "two@example.test" { |
| + t.Fatalf("delivered %v, want the second link after the first send panicked", m.to) |
| + } |
| +} |
diff --git a/internal/calendar/routes.go b/internal/calendar/routes.go| index 5f0cf92..8058d10 100644 |
| --- a/internal/calendar/routes.go |
| +++ b/internal/calendar/routes.go |
| @@ -57,13 +57,9 @@ func (a *App) routes(h *handlers, authPlugin *rauth.Auth, list allowlist, origin |
| r.Get("/signin", h.signin) |
| r.Get("/signed-out", h.signedOut) |
| } |
| - signinPath := "/signin" |
| - if a.home != nil { |
| - signinPath = "/signin/local" |
| - } |
| begin := beginSignin(origin, authPlugin.Begin, h.renderKeymail) |
| - callback := doorOnRefusal(authPlugin.Callback, signinPath) |
| - verify := doorOnRefusal(authPlugin.Verify, signinPath) |
| + callback := doorOnRefusal(authPlugin.Callback, h.signin) |
| + verify := doorOnRefusal(authPlugin.Verify, h.signin) |
| if a.home != nil { |
| r.Post("/signin", a.home.routes.LocalGuard(begin)) |
| r.Get("/auth/callback", a.home.routes.LocalGuard(callback)) |
diff --git a/internal/calendar/signin.go b/internal/calendar/signin.go| index 09261ea..5665274 100644 |
| --- a/internal/calendar/signin.go |
| +++ b/internal/calendar/signin.go |
| @@ -54,13 +54,15 @@ func (h *handlers) signin(w http.ResponseWriter, r *http.Request) { |
| form.Error = "That link has expired. Ask for a new one." |
| case "keymail": |
| form.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 the allowlist does |
| - // not hold it (doorOnRefusal). The form is the way on. |
| - form.Error = "That address can’t sign in to Calendar. Use a different address, or ask whoever runs this team to add it." |
| default: |
| form.Error = "That link did not work. Ask for a new one." |
| } |
| + if refusedHere(r) { |
| + // Keymail or a link verified the address and the allowlist does |
| + // not hold it, in this very response (doorOnRefusal). Never from |
| + // the query. The form is the way on. |
| + form.Error = "That address can’t sign in to Calendar. Use a different address, or ask whoever runs this team to add it." |
| + } |
| sent := q.Get("sent") == "1" |
| failed := form.Error != "" || form.AddressError != "" |
| switch { |
diff --git a/internal/calendar/signin_mail.go b/internal/calendar/signin_mail.go| index 6882643..3dcc230 100644 |
| --- a/internal/calendar/signin_mail.go |
| +++ b/internal/calendar/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,44 @@ 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. |
| +func (q *mailQueue) pump() { |
| + for l := range q.logs { |
| + if n := q.lostLogs.Swap(0); n > 0 { |
| + q.logger.Error("calendar: 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 +198,7 @@ func (q *mailQueue) run() { |
| } |
| q.reportDrops() |
| if abandoned > 0 { |
| - q.logger.Error("calendar: sign-in links abandoned at shutdown", "count", abandoned) |
| + q.logf("calendar: sign-in links abandoned at shutdown", "count", abandoned) |
| } |
| } |
| @@ -136,7 +206,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("calendar: sign-in mail queue full; links dropped", "count", n) |
| + q.logf("calendar: sign-in mail queue full; links dropped", "count", n) |
| } |
| } |
| @@ -145,24 +215,33 @@ func (q *mailQueue) reportDrops() { |
| func (q *mailQueue) deliver(job mailJob) { |
| defer func() { |
| if p := recover(); p != nil { |
| - q.logger.Error("calendar: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| + q.logf("calendar: 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("calendar: send sign-in link", "err", err) |
| + q.logf("calendar: 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 |
| @@ -186,24 +265,10 @@ func (q *mailQueue) close() { |
| select { |
| case <-q.done: |
| case <-time.After(mailAbandonGrace): |
| - q.logger.Error("calendar: sign-in mail worker ignored cancellation; left running") |
| - } |
| -} |
| - |
| -// allowlistAdmits is the queue's admission check: list.allows, with |
| -// each refusal COUNTED in the log, without the address. The response |
| -// has to stay silent; the log does not (AGENTS.md §Logs). Without this |
| -// line an instance whose allowlist is missing the only address anybody |
| -// types is indistinguishable from a working one. It runs on the worker, |
| -// never on the request path. |
| -func allowlistAdmits(list allowlist, logger *slog.Logger) func(string) bool { |
| - var refusals atomic.Int64 |
| - return func(to string) bool { |
| - if list.allows(to) { |
| - return true |
| - } |
| - logger.Warn("calendar: signin refused by allowlist", |
| - "stage", "signin.gate", "refusals_since_boot", refusals.Add(1)) |
| - return false |
| + // 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("calendar: sign-in mail worker ignored cancellation; left running") |
| } |
| } |
diff --git a/internal/calendar/signin_mail_test.go b/internal/calendar/signin_mail_test.go| index 8f36591..e6e7ff0 100644 |
| --- a/internal/calendar/signin_mail_test.go |
| +++ b/internal/calendar/signin_mail_test.go |
| @@ -75,15 +75,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) |
| } |
| @@ -179,6 +185,63 @@ 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. With the log sink stalled — every write hangs |
| +// — and a relay that never answers but honours cancellation, close |
| +// still returns within its bound and the worker has exited. Once the |
| +// sink is released, the goroutine count is back to its baseline: the |
| +// only thing a stalled sink can hold is the log pump's one write, never |
| +// close or the worker (Astra, round 3, P2). |
| +func TestMailQueueCloseIsBoundedWithAStalledLogger(t *testing.T) { |
| + baseline := runtime.NumGoroutine() |
| + 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") |
| + } |
| + select { |
| + case <-q.done: |
| + default: |
| + t.Fatal("the worker was still running after close, held by the stalled log sink") |
| + } |
| + free() |
| + deadline := time.Now().Add(2 * 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 and release, %d before", n, baseline) |
| + } |
| +} |
| + |
| // TestAppCloseStopsTheMailWorker: the app owns its queue, and Close |
| // stops its worker, so no test that builds and closes an app leaves |
| // one running. |