| 1 | package calendar |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "net/http" |
| 8 | "time" |
| 9 | |
| 10 | "amadan.net/rastrillo/rastrillo/carlos" |
| 11 | ) |
| 12 | |
| 13 | // drainReminders is the platform's scheduled-work handler: POST |
| 14 | // /jobs/reminders, mounted outside CSRF (see mountTick) and guarded by |
| 15 | // carlos.Tick. |
| 16 | // |
| 17 | // The work happens INSIDE the request, synchronously. The platform |
| 18 | // holds the instance awake for exactly as long as this handler is open |
| 19 | // and starts the idle clock when it returns, so answering 202 and |
| 20 | // continuing in a goroutine is a reliable way to be hibernated |
| 21 | // mid-send. 2xx means done, 5xx means retry, 4xx means do not. |
| 22 | func (h *handlers) drainReminders(w http.ResponseWriter, r *http.Request) { |
| 23 | if !carlos.Tick(r) { |
| 24 | // Constant-time bearer check, false when there is no token in |
| 25 | // the environment — so this fails closed off-platform rather |
| 26 | // than exposing a send-everything endpoint on the internet. |
| 27 | http.Error(w, "forbidden", http.StatusForbidden) |
| 28 | return |
| 29 | } |
| 30 | started := time.Now() |
| 31 | sent, progress, err := h.drain(r.Context(), started) |
| 32 | registerErr := h.scheduled.finishDrain(started, progress, err) |
| 33 | if err == nil { |
| 34 | err = registerErr |
| 35 | } |
| 36 | if err != nil { |
| 37 | // 5xx so the platform retries. The occurrence key makes a |
| 38 | // retry safe. |
| 39 | h.logger.Error("calendar: drain reminders", "stage", "reminders.drain", "err", scrubErr(err)) |
| 40 | http.Error(w, "retry", http.StatusInternalServerError) |
| 41 | return |
| 42 | } |
| 43 | h.logger.Info("calendar: reminders drained", "sent", sent) |
| 44 | w.WriteHeader(http.StatusNoContent) |
| 45 | } |
| 46 | |
| 47 | // drainDeadline bounds one pass, so a very long queue cannot hold the |
| 48 | // instance awake indefinitely. What is left is still due, and the next |
| 49 | // wake takes it. |
| 50 | const drainDeadline = 45 * time.Second |
| 51 | |
| 52 | // drain is one pass of everything scheduled, in order, inside one |
| 53 | // deadline: the cheap booking sweeps first, then reminders until half |
| 54 | // the budget is gone, then booking mail with what is left (booking |
| 55 | // spec §8). Reminders get the first half so a queue of booking mail |
| 56 | // behind a slow relay cannot starve them. |
| 57 | func (h *handlers) drain(ctx context.Context, now time.Time) (int, int, error) { |
| 58 | progress := 0 |
| 59 | deadline := now.Add(drainDeadline) |
| 60 | ctx, cancel := context.WithDeadline(ctx, deadline) |
| 61 | defer cancel() |
| 62 | if n, err := h.store.ExpireHolds(ctx, now, 200); err != nil { |
| 63 | return 0, progress, err |
| 64 | } else if n > 0 { |
| 65 | progress += n |
| 66 | h.logger.Info("calendar: holds expired", "n", n) |
| 67 | } |
| 68 | if n, err := h.store.ForgetOldIPs(ctx, now, 500); err != nil { |
| 69 | return 0, progress, err |
| 70 | } else if n > 0 { |
| 71 | progress += n |
| 72 | h.logger.Info("calendar: ip fingerprints forgotten", "n", n) |
| 73 | } |
| 74 | // Each phase's deadline is a CONTEXT deadline, not only a loop |
| 75 | // condition: the bounded sender waits no longer than what is left, |
| 76 | // so a send started at second 44 cannot run to second 64. Sends |
| 77 | // stop keepReserve before each phase's end, which is where the |
| 78 | // bookkeeping of the last send runs. |
| 79 | half := now.Add(drainDeadline / 2).Add(-keepReserve) |
| 80 | rctx, rcancel := context.WithDeadline(ctx, half) |
| 81 | sent, err := h.drainReminderQueue(rctx, now, half) |
| 82 | rcancel() |
| 83 | if err != nil && !errors.Is(err, context.DeadlineExceeded) { |
| 84 | return sent, progress + sent, err |
| 85 | } |
| 86 | // The reminder phase running out is a handover, not a failure: |
| 87 | // booking mail still gets the second half. |
| 88 | mctx, mcancel := context.WithDeadline(ctx, deadline.Add(-keepReserve)) |
| 89 | defer mcancel() |
| 90 | mailed, err := h.drainBookingMail(mctx, now, deadline.Add(-keepReserve)) |
| 91 | return sent + mailed, progress + sent + mailed, err |
| 92 | } |
| 93 | |
| 94 | // drainReminderQueue sends every reminder that has come due, until the |
| 95 | // deadline. |
| 96 | // |
| 97 | // Delivery is AT LEAST ONCE, deliberately. A reminder that never |
| 98 | // arrives is worse than one that arrives twice, so every ambiguous case |
| 99 | // resolves towards sending: a claim that expired, a process killed |
| 100 | // between the send and the mark, a tick delivered twice. The |
| 101 | // occurrence key is what keeps "twice" rare rather than routine. |
| 102 | func (h *handlers) drainReminderQueue(ctx context.Context, now, deadline time.Time) (int, error) { |
| 103 | sent := 0 |
| 104 | seen := map[string]bool{} |
| 105 | for time.Now().Before(deadline) { |
| 106 | due, ok, err := h.store.ClaimDueReminder(ctx, now.Unix()) |
| 107 | if err != nil { |
| 108 | return sent, err |
| 109 | } |
| 110 | if !ok { |
| 111 | return sent, nil |
| 112 | } |
| 113 | // Belt to the reschedule's braces, checked BEFORE sending: if a |
| 114 | // pass claims the same occurrence twice, the reschedule is |
| 115 | // wrong again and this loop would spend its whole deadline |
| 116 | // re-sending one reminder. Stop, and name it — a bounded bug |
| 117 | // that says what it is beats an unbounded one that does not. |
| 118 | if seen[due.Queue.OccurrenceKey] { |
| 119 | h.logger.Error("calendar: the same occurrence was claimed twice in one drain; stopping the pass") |
| 120 | return sent, errors.New("calendar: the same occurrence was claimed twice in one drain; stopping the pass") |
| 121 | } |
| 122 | seen[due.Queue.OccurrenceKey] = true |
| 123 | |
| 124 | if err := h.sendReminder(ctx, due, now); err != nil { |
| 125 | if errors.Is(err, ErrMailBusy) { |
| 126 | // Nothing was tried: the sender is full. Back in the |
| 127 | // queue in thirty seconds, attempts untouched. |
| 128 | if _, err := h.store.DeferReminder(keep(ctx), due.Queue.ID, due.ClaimID, now.Unix()); err != nil { |
| 129 | return sent, err |
| 130 | } |
| 131 | continue |
| 132 | } |
| 133 | // Leave it claimed; the lease expires and a later pass |
| 134 | // retries. Recording the attempt is what stops a |
| 135 | // permanently undeliverable address being retried forever. |
| 136 | h.logger.Error("calendar: send reminder", "stage", "reminders.send", |
| 137 | "why", mailCode(err), "attempts", due.Queue.Attempts) |
| 138 | if err := h.store.FailReminder(keep(ctx), due.Queue.ID, due.ClaimID, due.StartUTC); err != nil { |
| 139 | return sent, err |
| 140 | } |
| 141 | continue |
| 142 | } |
| 143 | // Mark and advance in one transaction, conditional on the |
| 144 | // claim this worker holds: a reschedule that overtook it finds |
| 145 | // this a no-op, which is the whole point of the claim id. |
| 146 | if err := h.store.CompleteReminder(keep(ctx), due, now.Unix()); err != nil { |
| 147 | return sent, err |
| 148 | } |
| 149 | sent++ |
| 150 | } |
| 151 | return sent, nil |
| 152 | } |
| 153 | |
| 154 | // sendReminder delivers one. |
| 155 | // |
| 156 | // The subject carries the title, which means the title leaves the |
| 157 | // sealed database and travels through the mail relay into somebody's |
| 158 | // mailbox. That is delivery rather than storage, and it is the product |
| 159 | // working — but it is a real leak and the README says so rather than |
| 160 | // letting the one rule be read as covering it. |
| 161 | func (h *handlers) sendReminder(ctx context.Context, due DueReminder, now time.Time) error { |
| 162 | if h.mailer == nil { |
| 163 | return fmt.Errorf("calendar: no mailer configured") |
| 164 | } |
| 165 | when := hhmm(due.StartUTC, due.Zone) |
| 166 | subject := fmt.Sprintf("%s at %s", due.Title, when) |
| 167 | |
| 168 | body := fmt.Sprintf("%s\n\n%s at %s", due.Title, due.DayWords, when) |
| 169 | if due.Location != "" { |
| 170 | body += "\n" + due.Location |
| 171 | } |
| 172 | body += "\n\n" + h.origin + fmt.Sprintf("/events/%d", due.EventID) |
| 173 | |
| 174 | // Lateness is stated rather than hidden. An instance that was |
| 175 | // asleep fires late, and a reminder that arrives at 09:05 for a |
| 176 | // 09:00 meeting should say so instead of implying it was on time. |
| 177 | if late := now.Unix() - due.Queue.DueAt; late > 120 { |
| 178 | body += fmt.Sprintf("\n\nThis reminder is %s late.", humanDuration(int(late/60))) |
| 179 | } |
| 180 | return h.mailer.Send(ctx, due.Address, subject, body) |
| 181 | } |
| 182 | |
| 183 | // drainOpportunistically is the third entrance, as middleware on the |
| 184 | // signed-in group. |
| 185 | // |
| 186 | // GET and HEAD only. A drain inside a POST would put an email send |
| 187 | // between somebody's form submission and its redirect, which is the one |
| 188 | // moment they are already waiting. |
| 189 | // |
| 190 | // The cost is honest: when a reminder IS due, one page load carries one |
| 191 | // SMTP send. The claim query is a single indexed read and returns |
| 192 | // immediately when nothing is due, which is almost always, so the |
| 193 | // common case pays nothing. A goroutine would avoid even that, and |
| 194 | // would also be a reliable way to be hibernated mid-send — the same |
| 195 | // reason the tick handler works inside its request. |
| 196 | func (h *handlers) drainOpportunistically(next http.Handler) http.Handler { |
| 197 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 198 | if r.Method == http.MethodGet || r.Method == http.MethodHead { |
| 199 | h.DrainOnRequest(r.Context()) |
| 200 | } |
| 201 | next.ServeHTTP(w, r) |
| 202 | }) |
| 203 | } |
| 204 | |
| 205 | // DrainOnRequest does ONE thing: expires one lapsed hold if there is |
| 206 | // one, else sends one overdue reminder, else one owed booking mail. |
| 207 | // |
| 208 | // Three ways scheduled work gets done, and the redundancy is the point: |
| 209 | // the platform's own wake from NextDue, the declared schedule's tick, |
| 210 | // and this. If the first two are misconfigured the app still reminds |
| 211 | // people, just less punctually, rather than silently never reminding |
| 212 | // anybody. |
| 213 | // |
| 214 | // Bounded to a single item and five seconds so that loading a page |
| 215 | // never becomes slow because a queue got long. |
| 216 | func (h *handlers) DrainOnRequest(ctx context.Context) { |
| 217 | ctx, cancel := context.WithTimeout(ctx, 5*time.Second) |
| 218 | defer cancel() |
| 219 | sctx, scancel := sendBudget(ctx) |
| 220 | defer scancel() |
| 221 | now := time.Now() |
| 222 | if n, err := h.store.ExpireHolds(ctx, now, 1); err != nil || n > 0 { |
| 223 | return |
| 224 | } |
| 225 | due, ok, err := h.store.ClaimDueReminder(ctx, now.Unix()) |
| 226 | if err != nil { |
| 227 | return |
| 228 | } |
| 229 | if !ok { |
| 230 | if row, ok, err := h.store.ClaimBookingMail(ctx, now.Unix()); err == nil && ok { |
| 231 | _ = h.sendBookingMail(sctx, row, now) |
| 232 | } |
| 233 | return |
| 234 | } |
| 235 | if err := h.sendReminder(sctx, due, now); err != nil { |
| 236 | if errors.Is(err, ErrMailBusy) { |
| 237 | _, _ = h.store.DeferReminder(keep(ctx), due.Queue.ID, due.ClaimID, now.Unix()) |
| 238 | return |
| 239 | } |
| 240 | h.logger.Error("calendar: opportunistic reminder", "stage", "reminders.send", |
| 241 | "why", mailCode(err)) |
| 242 | _ = h.store.FailReminder(keep(ctx), due.Queue.ID, due.ClaimID, due.StartUTC) |
| 243 | return |
| 244 | } |
| 245 | if err := h.store.CompleteReminder(keep(ctx), due, now.Unix()); err != nil { |
| 246 | h.logger.Error("calendar: mark reminder sent", "stage", "reminders.mark", "err", scrubErr(err)) |
| 247 | } |
| 248 | } |
| 249 | |
| 250 | // drainBookingMail sends every owed booking mail that is claimable, |
| 251 | // until the deadline. Each send is bounded by the sender; each outcome |
| 252 | // — sent, deferred, failed — is the row's own. |
| 253 | func (h *handlers) drainBookingMail(ctx context.Context, now, deadline time.Time) (int, error) { |
| 254 | sent := 0 |
| 255 | for time.Now().Before(deadline) { |
| 256 | row, ok, err := h.store.ClaimBookingMail(ctx, time.Now().Unix()) |
| 257 | if err != nil { |
| 258 | return sent, err |
| 259 | } |
| 260 | if !ok { |
| 261 | return sent, nil |
| 262 | } |
| 263 | if err := h.sendBookingMail(ctx, row, time.Now()); err == nil { |
| 264 | sent++ // deferred and retired rows report themselves as errors, and are not sends |
| 265 | } |
| 266 | } |
| 267 | return sent, nil |
| 268 | } |
| 269 | |
amadan