| index 31a12d3..6882643 100644 |
| --- a/internal/calendar/signin_mail.go |
| +++ b/internal/calendar/signin_mail.go |
| @@ -2,6 +2,7 @@ package calendar |
| |
| import ( |
| "context" |
| + "fmt" |
| "log/slog" |
| "sync" |
| "sync/atomic" |
| @@ -10,115 +11,162 @@ import ( |
| "amadan.net/rastrillo/rastrillo/mail" |
| ) |
| |
| -// mailQueue is the sign-in mailer rauth is given: Send queues the link |
| -// and returns at once, and one background worker asks the allowlist and |
| -// only then talks to SMTP. |
| +// mailQueue is the sign-in mailer the auth plugin is given: Send queues |
| +// the link and returns at once, and one background worker decides |
| +// whether the address may have it and only then talks to SMTP. |
| // |
| -// Both halves are the no-oracle guarantee (beginSignin), measured rather |
| -// than reasoned about. When the send ran inside the request, a permitted |
| +// Both halves are the no-oracle guarantee, measured rather than |
| +// reasoned about. When the send ran inside the request, a permitted |
| // address waited for the SMTP round trip before its `303 ?sent=1` and a |
| -// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides #20; Calendar had the same split). A response |
| -// time is an answer like any other. So POST /signin does the same |
| -// synchronous work for every address — rauth's budgets, its link row, |
| -// this enqueue — and the one admission-dependent step, the send, happens |
| -// after the response, where nobody can time it. The admission check |
| -// moved into the worker for the same reason: a check before responding |
| -// would be a branch an attacker could time, however cheap it is today. |
| +// denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides |
| +// #20). A response time is an answer like any other. So POST /signin |
| +// does the same synchronous work for every address — the auth plugin's |
| +// budgets, its link row, this enqueue — and the one admission-dependent |
| +// step, the send, happens after the response, where nobody can time it. |
| // |
| -// It sits in front of the bounded sender, so an ErrMailBusy is a log |
| -// line here rather than a ?err=1 only a permitted address could see. |
| +// The request path is Send, and Send does no I/O and never blocks: not |
| +// a send, not an admission check, not even a log line. Every address's |
| +// link is queued, permitted or not, so how full the queue is never |
| +// depends on admission either; the worker decides afterwards. A full |
| +// queue drops the link and counts it in an atomic; the worker logs the |
| +// count when it next runs. The person is answered exactly as when the |
| +// queue is empty, because a different answer for a full queue would be |
| +// one more thing to read. (Astra on #20, round 2.) |
| // |
| -// Delivery failures are logged, never surfaced. A failure only a |
| -// permitted address can reach is the oracle again, and the person is |
| -// told what every address is told: check your email, or ask again. |
| +// The worker survives a panic in the mailer: it recovers per message, |
| +// logs, and keeps draining. Delivery failures are logged, never |
| +// surfaced: a failure only a permitted address can reach is the oracle |
| +// again. |
| // |
| -// Bounded: a queue of mailQueueDepth links. A full queue drops the link |
| -// with a log line instead of blocking a request — the per-IP and |
| -// per-address budgets inside Begin keep that from being an ordinary |
| -// event. Close drains what is queued, within mailDrainTimeout. |
| +// 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 |
| +// worker to stop. A mailer that ignores its context is the only way a |
| +// worker outlives close, and that is logged too. |
| type mailQueue struct { |
| next mail.Sender |
| - admits func(string) bool |
| + admits func(to string) bool |
| + // gate, when set, is read off the request's context at enqueue time |
| + // and travels with the link; the worker sends only if it said yes. |
| + gate func(context.Context) bool |
| logger *slog.Logger |
| |
| - refusals atomic.Int64 |
| + drainTimeout time.Duration |
| + sendTimeout time.Duration |
| |
| - mu sync.Mutex |
| + mu sync.RWMutex // held for reading by Send, for writing by close |
| closed bool |
| jobs chan mailJob |
| + dropped atomic.Int64 |
| pending sync.WaitGroup |
| - done chan struct{} |
| + |
| + ctx context.Context // cancelled when close gives up waiting |
| + cancel context.CancelFunc |
| + done chan struct{} |
| } |
| |
| -type mailJob struct{ to, subject, body string } |
| +type mailJob struct { |
| + to, subject, body string |
| + gated bool |
| +} |
| |
| const ( |
| mailQueueDepth = 64 |
| mailSendTimeout = time.Minute |
| mailDrainTimeout = 30 * time.Second |
| + mailAbandonGrace = 5 * time.Second |
| ) |
| |
| -func newMailQueue(next mail.Sender, admits func(string) bool, logger *slog.Logger) *mailQueue { |
| +func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue { |
| if logger == nil { |
| logger = slog.Default() |
| } |
| - q := &mailQueue{next: next, admits: admits, logger: logger, |
| - jobs: make(chan mailJob, mailQueueDepth), done: make(chan struct{})} |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + q := &mailQueue{next: next, admits: admits, gate: gate, logger: logger, |
| + drainTimeout: mailDrainTimeout, sendTimeout: mailSendTimeout, |
| + jobs: make(chan mailJob, mailQueueDepth), ctx: ctx, cancel: cancel, |
| + done: make(chan struct{})} |
| go q.run() |
| return q |
| } |
| |
| -// Send queues one link. It never blocks and never fails: the request |
| -// that called it answers the same way whatever happens next. |
| -func (q *mailQueue) Send(_ context.Context, to, subject, body string) error { |
| - q.mu.Lock() |
| - defer q.mu.Unlock() |
| +// Send queues one link. It never blocks, never does I/O and never |
| +// fails: the request that called it answers the same way whatever |
| +// happens next. |
| +func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error { |
| + job := mailJob{to: to, subject: subject, body: body, gated: true} |
| + if q.gate != nil { |
| + job.gated = q.gate(ctx) |
| + } |
| + q.mu.RLock() |
| + defer q.mu.RUnlock() |
| if q.closed { |
| - q.logger.Error("calendar: sign-in mail after shutdown; dropped") |
| + q.dropped.Add(1) |
| return nil |
| } |
| q.pending.Add(1) |
| select { |
| - case q.jobs <- mailJob{to: to, subject: subject, body: body}: |
| + case q.jobs <- job: |
| default: |
| q.pending.Done() |
| - q.logger.Error("calendar: sign-in mail queue full; link dropped") |
| + q.dropped.Add(1) |
| } |
| return nil |
| } |
| |
| func (q *mailQueue) run() { |
| defer close(q.done) |
| + abandoned := 0 |
| for job := range q.jobs { |
| - q.deliver(job) |
| + q.reportDrops() |
| + if q.ctx.Err() != nil { |
| + abandoned++ |
| + } else { |
| + q.deliver(job) |
| + } |
| q.pending.Done() |
| } |
| + q.reportDrops() |
| + if abandoned > 0 { |
| + q.logger.Error("calendar: sign-in links abandoned at shutdown", "count", abandoned) |
| + } |
| +} |
| + |
| +// reportDrops logs the links Send had to drop since the last report. |
| +// Here, on the worker, so the request path never writes a log line. |
| +func (q *mailQueue) reportDrops() { |
| + if n := q.dropped.Swap(0); n > 0 { |
| + q.logger.Error("calendar: sign-in mail queue full; links dropped", "count", n) |
| + } |
| } |
| |
| +// deliver sends one link if it may go. A panic in the admission check |
| +// or the mailer is this message's failure, not the process's. |
| func (q *mailQueue) deliver(job mailJob) { |
| - if !q.admits(job.to) { |
| - // Not on the allowlist: the link rauth stored is never |
| - // delivered, so it is never redeemed. The refusal is 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. |
| - q.logger.Warn("calendar: signin refused by allowlist", |
| - "stage", "signin.gate", "refusals_since_boot", q.refusals.Add(1)) |
| + defer func() { |
| + if p := recover(); p != nil { |
| + q.logger.Error("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. |
| return |
| } |
| - ctx, cancel := context.WithTimeout(context.Background(), mailSendTimeout) |
| + 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) |
| } |
| } |
| |
| -// wait blocks until every link queued so far has been delivered or |
| -// dropped. SignInForTest uses it to read the recorder afterwards. |
| +// 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() } |
| |
| -// close stops taking links and drains what is queued. |
| +// 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). |
| func (q *mailQueue) close() { |
| q.mu.Lock() |
| if q.closed { |
| @@ -128,9 +176,34 @@ func (q *mailQueue) close() { |
| q.closed = true |
| close(q.jobs) |
| q.mu.Unlock() |
| + defer q.cancel() |
| select { |
| case <-q.done: |
| - case <-time.After(mailDrainTimeout): |
| - q.logger.Error("calendar: sign-in mail still sending at shutdown; abandoned") |
| + return |
| + case <-time.After(q.drainTimeout): |
| + } |
| + q.cancel() |
| + 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 |
| } |
| } |
amadan