amadan
Close never waits on the log, the refusal is the door itself, no test leaks a worker
Astra's round 3 on oficina/slides #20 and on #45.
- P2, close could hang on a stalled log sink. Only a log pump writes to
the logger now; the worker and close hand lines to logf, which never
blocks. close returns within its bound and the worker exits whatever
the sink does. TestMailQueueCloseIsBoundedWithAStalledLogger fails on
the previous queue ("close hung on a stalled log sink").
- TestMain fails the run if any mailQueue worker or pump outlives the
tests (every plugin a test builds is closed in its cleanup).
- P2, ?err=refused could be typed by anybody. doorOnRefusal renders the
door in the refusal's own 403, marked through the request context;
the query no longer selects the sentence. TestKeymailRefusalIsTheDoor
requires that, and that /signin?err=refused says nothing of the sort;
it fails on e54dae7.
- The registry of queues moves to mailqueue_registry.go (300-line cap).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+285
−78
internal/sheets/begin.go+42 −8internal/sheets/home_routes.go+5 −5internal/sheets/keymail_signin_test.go+24 −5internal/sheets/mailqueue.go+94 −48internal/sheets/mailqueue_registry.go+43 −0internal/sheets/mailqueue_test.go+69 −6internal/sheets/signin.go+8 −6
CI passed — run details
CI log
amadan: .amadan/timeout: this job's budget is 15m0s (this runner allows up to 30m0s) === vet === CGO_ENABLED=0 go vet ./... CGO_ENABLED=0 go vet -tags browser ./internal/sheets/ === fmt === === build === CGO_ENABLED=0 go build ./... === migrations === go tool rastrillo migration check rastrillo migration check: models and migrations agree === go-test === CGO_ENABLED=0 go test -count=1 ./... ? amadan.net/oficina/sheets/cmd/sheets [no test files] ok amadan.net/oficina/sheets/internal/guardstest 0.087s ok amadan.net/oficina/sheets/internal/sheets 3.976s ok amadan.net/oficina/sheets/internal/sheets/calc 0.015s ok amadan.net/oficina/sheets/internal/sheets/format 0.002s ? amadan.net/oficina/sheets/internal/sheets/keymailtest [no test files] ok amadan.net/oficina/sheets/internal/sheets/live 0.053s ok amadan.net/oficina/sheets/internal/sheets/seal 0.002s ok amadan.net/oficina/sheets/internal/sheets/sealtest 0.006s ok amadan.net/oficina/sheets/internal/sheets/store 2.940s ok amadan.net/oficina/sheets/internal/sheets/xlsx 0.302s ? amadan.net/oficina/sheets/tools/gencorpus [no test files] === race === go test -race -count=1 ./internal/sheets/store/... ./internal/sheets/live/... ok amadan.net/oficina/sheets/internal/sheets/store 46.678s ok amadan.net/oficina/sheets/internal/sheets/live 1.063s
diff --git a/internal/sheets/begin.go b/internal/sheets/begin.go| index 1befcf4..a0ded16 100644 |
| --- a/internal/sheets/begin.go |
| +++ b/internal/sheets/begin.go |
| @@ -2,6 +2,7 @@ package sheets |
| import ( |
| "bytes" |
| + "context" |
| "net/http" |
| "strings" |
| @@ -58,12 +59,17 @@ func (a *App) handleBegin(w http.ResponseWriter, r *http.Request) { |
| const authRefusal = "This address is verified but not admitted here." |
| // doorOnRefusal wraps auth's Callback and Verify, the handlers that end |
| -// in admission, and turns its bare-text refusal into the sign-in 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 sign-in door, |
| +// rendered in the refusal's own 403 by door (handleSignin) 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 on oficina/slides #20, 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) |
| @@ -71,8 +77,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 { |
| @@ -88,6 +94,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/sheets/home_routes.go b/internal/sheets/home_routes.go| index ce8c46a..057cd83 100644 |
| --- a/internal/sheets/home_routes.go |
| +++ b/internal/sheets/home_routes.go |
| @@ -29,14 +29,14 @@ func (a *App) mountAuth(r chi.Router) { |
| // links already in people's inboxes. |
| if a.homeRoutes != nil { |
| r.Post("/signin", a.homeRoutes.LocalGuard(a.handleBegin)) |
| - r.Get("/auth/callback", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Callback, "/signin/local"))) |
| - r.Get("/auth/verify", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Verify, "/signin/local"))) |
| - r.Post("/auth/verify", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Verify, "/signin/local"))) |
| + r.Get("/auth/callback", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Callback, a.handleSignin))) |
| + r.Get("/auth/verify", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Verify, a.handleSignin))) |
| + r.Post("/auth/verify", a.homeRoutes.LocalGuard(doorOnRefusal(a.auth.Verify, a.handleSignin))) |
| r.Post("/signout", a.homeRoutes.Signout) |
| } else { |
| r.Post("/signin", a.handleBegin) |
| - r.Get("/auth/callback", doorOnRefusal(a.auth.Callback, "/signin")) |
| - r.Get("/auth/verify", doorOnRefusal(a.auth.Verify, "/signin")) |
| + r.Get("/auth/callback", doorOnRefusal(a.auth.Callback, a.handleSignin)) |
| + r.Get("/auth/verify", doorOnRefusal(a.auth.Verify, a.handleSignin)) |
| r.Post("/signout", a.signOutToPage) |
| } |
| } |
diff --git a/internal/sheets/keymail_signin_test.go b/internal/sheets/keymail_signin_test.go| index 381ea8b..edeb63e 100644 |
| --- a/internal/sheets/keymail_signin_test.go |
| +++ b/internal/sheets/keymail_signin_test.go |
| @@ -10,6 +10,7 @@ import ( |
| "regexp" |
| "strings" |
| "testing" |
| + "time" |
| "amadan.net/oficina/sheets/internal/sheets/keymailtest" |
| ) |
| @@ -25,6 +26,18 @@ func TestMain(m *testing.M) { |
| os.Exit(1) |
| } |
| code := m.Run() |
| + // No test may leave a sign-in mail worker running: every plugin a |
| + // test builds is closed in its cleanup (CloseSigninMail). |
| + 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) |
| } |
| @@ -175,8 +188,9 @@ func TestKeymailFailureFallsBackToAnEmailLink(t *testing.T) { |
| } |
| // TestKeymailRefusalIsTheDoor: a non-admitted address Keymail verified |
| -// is refused after verification on the sign-in door, not in |
| -// rastrillo's bare text, and no session is minted. An admitted one is |
| +// is refused after verification with the sign-in door, rendered in the |
| +// refusal's own 403 — not rastrillo's bare text, and not a redirect to a |
| +// query anybody could type. No session is minted. An admitted one is |
| // signed in. |
| func TestKeymailRefusalIsTheDoor(t *testing.T) { |
| app, _, km := keymailApp(t) |
| @@ -190,18 +204,23 @@ func TestKeymailRefusalIsTheDoor(t *testing.T) { |
| t.Fatalf("admitted callback: %d %q, want 303 /sheets", cb.Code, cb.Header().Get("Location")) |
| } |
| cb := roundTrip(keymailStranger) |
| - if cb.Code != http.StatusSeeOther || cb.Header().Get("Location") != "/signin?err=refused" { |
| - t.Fatalf("non-admitted callback: %d %q, want 303 /signin?err=refused (bare text is the bug)\n%s", cb.Code, cb.Header().Get("Location"), cb.Body.String()) |
| + if cb.Code != http.StatusForbidden || cb.Header().Get("Location") != "" { |
| + t.Fatalf("non-admitted 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()) |
| } |
| for _, c := range cb.Result().Cookies() { |
| if strings.Contains(c.Name, "session") && c.MaxAge >= 0 && c.Value != "" { |
| t.Fatalf("a refused sign-in set a session cookie %q", c.Name) |
| } |
| } |
| - door := keymailGet(t, h, "/signin?err=refused").Body.String() |
| + door := cb.Body.String() |
| for _, want := range []string{"estilo-door", `name="address"`, "can’t sign in to Sheets"} { |
| if !strings.Contains(door, want) { |
| t.Errorf("refusal door lacks %q:\n%s", want, door) |
| } |
| } |
| + // The sentence means "you proved this address and it is not |
| + // admitted"; nobody may summon it by typing a URL (Astra, round 3). |
| + if spoof := keymailGet(t, h, "/signin?err=refused").Body.String(); strings.Contains(spoof, "can’t sign in to Sheets") { |
| + t.Fatal("anybody can show the refusal by visiting /signin?err=refused") |
| + } |
| } |
diff --git a/internal/sheets/mailqueue.go b/internal/sheets/mailqueue.go| index 5ac37c8..27d01f6 100644 |
| --- a/internal/sheets/mailqueue.go |
| +++ b/internal/sheets/mailqueue.go |
| @@ -8,7 +8,6 @@ import ( |
| "sync/atomic" |
| "time" |
| - "amadan.net/rastrillo/rastrillo/auth" |
| "amadan.net/rastrillo/rastrillo/mail" |
| ) |
| @@ -39,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 |
| @@ -64,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 { |
| @@ -76,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() |
| } |
| @@ -86,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 |
| } |
| @@ -115,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("sheets: 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() |
| @@ -129,7 +198,7 @@ func (q *mailQueue) run() { |
| } |
| q.reportDrops() |
| if abandoned > 0 { |
| - q.logger.Error("sheets: sign-in links abandoned at shutdown", "count", abandoned) |
| + q.logf("sheets: sign-in links abandoned at shutdown", "count", abandoned) |
| } |
| } |
| @@ -137,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("sheets: sign-in mail queue full; links dropped", "count", n) |
| + q.logf("sheets: sign-in mail queue full; links dropped", "count", n) |
| } |
| } |
| @@ -146,24 +215,33 @@ func (q *mailQueue) reportDrops() { |
| func (q *mailQueue) deliver(job mailJob) { |
| defer func() { |
| if p := recover(); p != nil { |
| - q.logger.Error("sheets: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| + q.logf("sheets: 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("sheets: send sign-in link", "err", err) |
| + q.logf("sheets: 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 |
| @@ -187,42 +265,10 @@ func (q *mailQueue) close() { |
| select { |
| case <-q.done: |
| case <-time.After(mailAbandonGrace): |
| - q.logger.Error("sheets: 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("sheets: sign-in mail worker ignored cancellation; left running") |
| } |
| } |
| - |
| -// signinQueues is every queue NewAuth built, by the plugin it feeds, so |
| -// CloseSigninMail can stop one and DrainSigninMail all of them. |
| -var ( |
| - signinQueuesMu sync.Mutex |
| - signinQueues = map[*auth.Auth]*mailQueue{} |
| -) |
| - |
| -// CloseSigninMail stops the sign-in mail queue behind x, sending what |
| -// is still in it within the queue's deadline. Tests close every plugin |
| -// they build with it. |
| -func CloseSigninMail(x *auth.Auth) { |
| - signinQueuesMu.Lock() |
| - q := signinQueues[x] |
| - delete(signinQueues, x) |
| - signinQueuesMu.Unlock() |
| - if q != nil { |
| - q.close() |
| - } |
| -} |
| - |
| -// DrainSigninMail stops every sign-in mail queue and sends what is |
| -// still in them. main calls it once Serve returns. |
| -func DrainSigninMail() { |
| - signinQueuesMu.Lock() |
| - qs := signinQueues |
| - signinQueues = map[*auth.Auth]*mailQueue{} |
| - signinQueuesMu.Unlock() |
| - for _, q := range qs { |
| - q.close() |
| - } |
| -} |
| - |
| -// queueAttacher is a test mailer that wants to wait for the queue in |
| -// front of it before it is read (begin_test.go's recorder). |
| -type queueAttacher interface{ attachQueue(wait func()) } |
diff --git a/internal/sheets/mailqueue_registry.go b/internal/sheets/mailqueue_registry.go| new file mode 100644 |
| index 0000000..cb63a65 |
| --- /dev/null |
| +++ b/internal/sheets/mailqueue_registry.go |
| @@ -0,0 +1,43 @@ |
| +package sheets |
| + |
| +import ( |
| + "sync" |
| + |
| + "amadan.net/rastrillo/rastrillo/auth" |
| +) |
| + |
| +// signinQueues is every queue NewAuth built, by the plugin it feeds, so |
| +// CloseSigninMail can stop one and DrainSigninMail all of them. |
| +var ( |
| + signinQueuesMu sync.Mutex |
| + signinQueues = map[*auth.Auth]*mailQueue{} |
| +) |
| + |
| +// CloseSigninMail stops the sign-in mail queue behind x, sending what |
| +// is still in it within the queue's deadline. Tests close every plugin |
| +// they build with it. |
| +func CloseSigninMail(x *auth.Auth) { |
| + signinQueuesMu.Lock() |
| + q := signinQueues[x] |
| + delete(signinQueues, x) |
| + signinQueuesMu.Unlock() |
| + if q != nil { |
| + q.close() |
| + } |
| +} |
| + |
| +// DrainSigninMail stops every sign-in mail queue and sends what is |
| +// still in them. main calls it once Serve returns. |
| +func DrainSigninMail() { |
| + signinQueuesMu.Lock() |
| + qs := signinQueues |
| + signinQueues = map[*auth.Auth]*mailQueue{} |
| + signinQueuesMu.Unlock() |
| + for _, q := range qs { |
| + q.close() |
| + } |
| +} |
| + |
| +// queueAttacher is a test mailer that wants to wait for the queue in |
| +// front of it before it is read (begin_test.go's recorder). |
| +type queueAttacher interface{ attachQueue(wait func()) } |
diff --git a/internal/sheets/mailqueue_test.go b/internal/sheets/mailqueue_test.go| index eb4c471..9120e65 100644 |
| --- a/internal/sheets/mailqueue_test.go |
| +++ b/internal/sheets/mailqueue_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) |
| + } |
| +} |
| + |
| // TestCloseSigninMailStopsTheWorker: every plugin NewAuth builds owns a |
| // queue, and CloseSigninMail stops its worker — which is what every |
| // test that builds one calls in its cleanup, so none leaves a worker |
diff --git a/internal/sheets/signin.go b/internal/sheets/signin.go| index eb71542..df02aad 100644 |
| --- a/internal/sheets/signin.go |
| +++ b/internal/sheets/signin.go |
| @@ -45,11 +45,6 @@ func signinError(code string) (text string, field bool) { |
| // A refused (unadmitted) address is never told so at the form: |
| // handleBegin answers it exactly as it answers an admitted one. |
| return "That doesn’t look like an email address.", true |
| - case "refused": |
| - // Keymail or a link verified the address and Admits said no |
| - // (doorOnRefusal). Told only to someone who just proved they |
| - // hold the address; the form below is the way on. |
| - return "That address can’t sign in to Sheets. Use a different address, or ask someone to share a sheet with it.", false |
| case "expired": |
| return "That link has expired or was already used. Enter your email to get a new one.", false |
| case "keymail": |
| @@ -86,7 +81,7 @@ func (a *App) signinFormPath() string { |
| // the skip in that mode. |
| func (a *App) handleSignin(w http.ResponseWriter, r *http.Request) { |
| q := r.URL.Query() |
| - if a.home == nil && q.Get("sent") == "" && q.Get("err") == "" && q.Get("force") == "" { |
| + if a.home == nil && !refusedHere(r) && q.Get("sent") == "" && q.Get("err") == "" && q.Get("force") == "" { |
| if _, ok := a.identity(r); ok { |
| http.Redirect(w, r, signedInLanding(r), http.StatusSeeOther) |
| return |
| @@ -101,6 +96,13 @@ func (a *App) handleSignin(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| text, field := signinError(q.Get("err")) |
| + if refusedHere(r) { |
| + // Keymail or a link verified the address and Admits said no |
| + // (doorOnRefusal), in this response. Never from the query: told |
| + // only to someone who just proved they hold the address; the |
| + // form below is the way on. |
| + text, field = "That address can’t sign in to Sheets. Use a different address, or ask someone to share a sheet with it.", false |
| + } |
| form := chrome.SignIn{ |
| Icon: brand.Path("sheets"), |
| App: "Sheets", |