| 1 | package calls |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "log/slog" |
| 7 | "sync" |
| 8 | "sync/atomic" |
| 9 | "time" |
| 10 | |
| 11 | "amadan.net/rastrillo/rastrillo/mail" |
| 12 | ) |
| 13 | |
| 14 | // mailQueue is the sign-in mailer the auth plugin is given: Send queues |
| 15 | // the link and returns at once, and one background worker decides |
| 16 | // whether the address may have it and only then talks to SMTP. |
| 17 | // |
| 18 | // Both halves are the no-oracle guarantee, measured rather than |
| 19 | // reasoned about. When the send ran inside the request, a permitted |
| 20 | // address waited for the SMTP round trip before its `303 ?sent=1` and a |
| 21 | // denied one did not: 252 ms against 0.35 ms (Astra on oficina/slides |
| 22 | // #20). A response time is an answer like any other. So POST /signin |
| 23 | // does the same synchronous work for every address — the auth plugin's |
| 24 | // budgets, its link row, this enqueue — and the one admission-dependent |
| 25 | // step, the send, happens after the response, where nobody can time it. |
| 26 | // |
| 27 | // The request path is Send, and Send does no I/O and never blocks: not |
| 28 | // a send, not an admission check, not even a log line. Every address's |
| 29 | // link is queued, permitted or not, so how full the queue is never |
| 30 | // depends on admission either; the worker decides afterwards. A full |
| 31 | // queue drops the link and counts it in an atomic; the worker logs the |
| 32 | // count when it next runs. The person is answered exactly as when the |
| 33 | // queue is empty, because a different answer for a full queue would be |
| 34 | // one more thing to read. (Astra on #20, round 2.) |
| 35 | // |
| 36 | // The worker survives a panic in the mailer: it recovers per message, |
| 37 | // logs, and keeps draining. Delivery failures are logged, never |
| 38 | // surfaced: a failure only a permitted address can reach is the oracle |
| 39 | // again. |
| 40 | // |
| 41 | // Nothing but the log pump ever writes to the logger. The worker and |
| 42 | // close hand their lines to logf, which never blocks: a buffered |
| 43 | // channel, and a counted drop when it is full. A stalled log sink can |
| 44 | // therefore hold only the pump's one write — never a request, never the |
| 45 | // worker, never close (Astra on #20, round 3). The pump ends when the |
| 46 | // worker does. |
| 47 | // |
| 48 | // close drains what is queued within drainTimeout. Past it, the send |
| 49 | // in flight is cancelled through its context, everything still queued |
| 50 | // is abandoned and counted, and close waits a short grace for the |
| 51 | // worker to stop. A mailer that ignores its context is the only way a |
| 52 | // worker outlives close; that is counted in stuck and logged, best |
| 53 | // effort, too. close may be called more than once (the app's context |
| 54 | // ending, then main on its way out): a later call waits for the first |
| 55 | // one to finish, so whoever returns from close knows the drain is over. |
| 56 | type mailQueue struct { |
| 57 | next mail.Sender |
| 58 | admits func(to string) bool |
| 59 | // gate, when set, is read off the request's context at enqueue time |
| 60 | // and travels with the link; the worker sends only if it said yes. |
| 61 | gate func(context.Context) bool |
| 62 | logger *slog.Logger |
| 63 | |
| 64 | drainTimeout time.Duration |
| 65 | sendTimeout time.Duration |
| 66 | |
| 67 | mu sync.RWMutex // held for reading by Send, for writing by close |
| 68 | closed bool |
| 69 | jobs chan mailJob |
| 70 | dropped atomic.Int64 |
| 71 | pending sync.WaitGroup |
| 72 | |
| 73 | ctx context.Context // cancelled when close gives up waiting |
| 74 | cancel context.CancelFunc |
| 75 | done chan struct{} |
| 76 | shut chan struct{} // closed when the first close returns |
| 77 | |
| 78 | logMu sync.RWMutex // logf reads logsClosed; the worker's exit sets it |
| 79 | logsClosed bool |
| 80 | logs chan logLine // to the pump; logf never blocks on it |
| 81 | unlogged atomic.Int64 // lines handed over and not yet written |
| 82 | lostLogs atomic.Int64 // lines logf could not hand over |
| 83 | stuck atomic.Bool // close gave up on a worker that ignored cancellation |
| 84 | |
| 85 | // refusedLog, when set, is the line the worker logs (through logf, |
| 86 | // with a running count and never the address) for each link it |
| 87 | // refuses. |
| 88 | refusedLog string |
| 89 | refusals atomic.Int64 |
| 90 | } |
| 91 | |
| 92 | type logLine struct { |
| 93 | msg string |
| 94 | args []any |
| 95 | } |
| 96 | |
| 97 | type mailJob struct { |
| 98 | to, subject, body string |
| 99 | gated bool |
| 100 | } |
| 101 | |
| 102 | const ( |
| 103 | mailQueueDepth = 64 |
| 104 | mailSendTimeout = time.Minute |
| 105 | mailDrainTimeout = 30 * time.Second |
| 106 | mailAbandonGrace = 5 * time.Second |
| 107 | mailLogDepth = 32 |
| 108 | ) |
| 109 | |
| 110 | func newMailQueue(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger) *mailQueue { |
| 111 | return newMailQueueLogging(next, admits, gate, logger, "") |
| 112 | } |
| 113 | |
| 114 | // newMailQueueLogging is newMailQueue with refusedLog set (see the |
| 115 | // field). |
| 116 | func newMailQueueLogging(next mail.Sender, admits func(string) bool, gate func(context.Context) bool, logger *slog.Logger, refusedLog string) *mailQueue { |
| 117 | if logger == nil { |
| 118 | logger = slog.Default() |
| 119 | } |
| 120 | ctx, cancel := context.WithCancel(context.Background()) |
| 121 | q := &mailQueue{next: next, admits: admits, gate: gate, logger: logger, |
| 122 | drainTimeout: mailDrainTimeout, sendTimeout: mailSendTimeout, |
| 123 | jobs: make(chan mailJob, mailQueueDepth), ctx: ctx, cancel: cancel, |
| 124 | done: make(chan struct{}), shut: make(chan struct{}), logs: make(chan logLine, mailLogDepth), refusedLog: refusedLog} |
| 125 | go q.pump() |
| 126 | go q.run() |
| 127 | return q |
| 128 | } |
| 129 | |
| 130 | // Send queues one link. It never blocks, never does I/O and never |
| 131 | // fails: the request that called it answers the same way whatever |
| 132 | // happens next. |
| 133 | func (q *mailQueue) Send(ctx context.Context, to, subject, body string) error { |
| 134 | job := mailJob{to: to, subject: subject, body: body, gated: true} |
| 135 | if q.gate != nil { |
| 136 | job.gated = q.gate(ctx) |
| 137 | } |
| 138 | q.mu.RLock() |
| 139 | defer q.mu.RUnlock() |
| 140 | if q.closed { |
| 141 | q.dropped.Add(1) |
| 142 | return nil |
| 143 | } |
| 144 | q.pending.Add(1) |
| 145 | select { |
| 146 | case q.jobs <- job: |
| 147 | default: |
| 148 | q.pending.Done() |
| 149 | q.dropped.Add(1) |
| 150 | } |
| 151 | return nil |
| 152 | } |
| 153 | |
| 154 | // logf hands one error line to the pump. It never blocks: when the |
| 155 | // pump is behind, the line is counted in lostLogs instead. |
| 156 | func (q *mailQueue) logf(msg string, args ...any) { |
| 157 | q.logMu.RLock() |
| 158 | defer q.logMu.RUnlock() |
| 159 | if q.logsClosed { |
| 160 | q.lostLogs.Add(1) |
| 161 | return |
| 162 | } |
| 163 | q.unlogged.Add(1) |
| 164 | select { |
| 165 | case q.logs <- logLine{msg: msg, args: args}: |
| 166 | default: |
| 167 | q.unlogged.Add(-1) |
| 168 | q.lostLogs.Add(1) |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // pump is the only goroutine that writes to the logger. It ends when |
| 173 | // the worker closes logs on its way out. |
| 174 | // |
| 175 | // Its limit, stated rather than engineered around: a write blocked |
| 176 | // inside the logger (a stalled sink) cannot be interrupted from Go. It |
| 177 | // can only finish, or die with the process. Closing logs does not reach |
| 178 | // it mid-write, so a stalled sink leaves this one goroutine behind |
| 179 | // until the sink moves. Nothing else waits on it: not Send, not the |
| 180 | // worker, not close. |
| 181 | func (q *mailQueue) pump() { |
| 182 | for l := range q.logs { |
| 183 | if n := q.lostLogs.Swap(0); n > 0 { |
| 184 | q.logger.Error("calls: sign-in mail log lines lost to a slow log sink", "count", n) |
| 185 | } |
| 186 | q.logger.Error(l.msg, l.args...) |
| 187 | q.unlogged.Add(-1) |
| 188 | } |
| 189 | } |
| 190 | |
| 191 | func (q *mailQueue) run() { |
| 192 | defer close(q.done) |
| 193 | defer func() { // after the last logf below; ends the pump |
| 194 | q.logMu.Lock() |
| 195 | q.logsClosed = true |
| 196 | close(q.logs) |
| 197 | q.logMu.Unlock() |
| 198 | }() |
| 199 | abandoned := 0 |
| 200 | for job := range q.jobs { |
| 201 | q.reportDrops() |
| 202 | if q.ctx.Err() != nil { |
| 203 | abandoned++ |
| 204 | } else { |
| 205 | q.deliver(job) |
| 206 | } |
| 207 | q.pending.Done() |
| 208 | } |
| 209 | q.reportDrops() |
| 210 | if abandoned > 0 { |
| 211 | q.logf("calls: sign-in links abandoned at shutdown", "count", abandoned) |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | // reportDrops logs the links Send had to drop since the last report. |
| 216 | // Here, on the worker, so the request path never writes a log line. |
| 217 | func (q *mailQueue) reportDrops() { |
| 218 | if n := q.dropped.Swap(0); n > 0 { |
| 219 | q.logf("calls: sign-in mail queue full; links dropped", "count", n) |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | // deliver sends one link if it may go. A panic in the admission check |
| 224 | // or the mailer is this message's failure, not the process's. |
| 225 | func (q *mailQueue) deliver(job mailJob) { |
| 226 | defer func() { |
| 227 | if p := recover(); p != nil { |
| 228 | q.logf("calls: sign-in mail worker recovered from a panic", "panic", fmt.Sprint(p)) |
| 229 | } |
| 230 | }() |
| 231 | if !job.gated || !q.admits(job.to) { |
| 232 | // Not admitted: the link the auth plugin stored is never |
| 233 | // delivered, so it is never redeemed. No address in the log. |
| 234 | if q.refusedLog != "" { |
| 235 | q.logf(q.refusedLog, "stage", "signin.gate", "refusals_since_boot", q.refusals.Add(1)) |
| 236 | } |
| 237 | return |
| 238 | } |
| 239 | ctx, cancel := context.WithTimeout(q.ctx, q.sendTimeout) |
| 240 | defer cancel() |
| 241 | if err := q.next.Send(ctx, job.to, job.subject, job.body); err != nil { |
| 242 | q.logf("calls: send sign-in link", "err", err) |
| 243 | } |
| 244 | } |
| 245 | |
| 246 | // wait blocks until every link queued so far has been sent or refused, |
| 247 | // and every line logged about them has been written. Tests use it to |
| 248 | // read the mailer, and the log, afterwards. |
| 249 | func (q *mailQueue) wait() { |
| 250 | q.pending.Wait() |
| 251 | for q.unlogged.Load() > 0 { |
| 252 | time.Sleep(time.Millisecond) |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | // close stops taking links and drains what is queued, within |
| 257 | // drainTimeout; then it cancels, abandons the rest, and stops the |
| 258 | // worker (see the type's comment). It never waits on the log pump: |
| 259 | // shutdown deliberately does not wait for a write blocked inside the |
| 260 | // logger, which can only finish or die with the process (see pump). |
| 261 | // |
| 262 | // A second close joins the first: it returns when that one does, never |
| 263 | // earlier, so main can wait for a drain the app's context began. |
| 264 | func (q *mailQueue) close() { |
| 265 | q.mu.Lock() |
| 266 | if q.closed { |
| 267 | q.mu.Unlock() |
| 268 | <-q.shut |
| 269 | return |
| 270 | } |
| 271 | q.closed = true |
| 272 | close(q.jobs) |
| 273 | q.mu.Unlock() |
| 274 | defer close(q.shut) |
| 275 | defer q.cancel() |
| 276 | select { |
| 277 | case <-q.done: |
| 278 | return |
| 279 | case <-time.After(q.drainTimeout): |
| 280 | } |
| 281 | q.cancel() |
| 282 | select { |
| 283 | case <-q.done: |
| 284 | case <-time.After(mailAbandonGrace): |
| 285 | // Counted, not written: logging here could hang close on the |
| 286 | // very sink that may be the trouble. The worker is still running, |
| 287 | // so logf is still safe to call, and never blocks. |
| 288 | q.stuck.Store(true) |
| 289 | q.logf("calls: sign-in mail worker ignored cancellation; left running") |
| 290 | } |
| 291 | } |
| 292 | |
| 293 | // mayMail is the queue's gate: gateBegin's mayRequestLink answer for |
| 294 | // the request that asked for the link (ruling 12's budgets), carried |
| 295 | // through Begin on the request context. A context lookup, no I/O. |
| 296 | func mayMail(ctx context.Context) bool { |
| 297 | may, _ := ctx.Value(mayMailKey{}).(bool) |
| 298 | return may |
| 299 | } |
| 300 | |
| 301 | // queueAttacher is a test mailer (another package's, so the method is |
| 302 | // exported) that wants to wait for the queue in front of it before it |
| 303 | // is read (callstest's recordingMailer). |
| 304 | type queueAttacher interface{ AttachSigninQueue(wait func()) } |
| 305 | |
amadan