amadan

carlos / platform Public

email events push: a failing app is suspended after a day, and a pass costs per app (Fable review fixes)

From a Fable review of this branch.

A failing app could slow the whole console for a month. Ending an app's
turn at its first failure (the last round's fix) leaves a failing app one
attempt a pass — about 5,760 a day — while each row takes eight attempts
to give up on, so it can shed at most ~720 rows a day against 10–20k events
a day at a pro cap. Rows never attempted stay due until the 30-day prune,
and every 15s pass sorted all of them through a ROW_NUMBER window on the
store's one connection, which SNS ingest, cap enforcement and every console
request queue behind. Measured at ~180ms a pass for 50,000 rows. The easiest
way in was the one the CLI suggested: turn the push on before deploying the
handler.

Two changes. A pass now starts from mail_event_hooks and reads each app's
oldest due rows through a new (account, app, delivered_at, abandoned_at,
next_at) index — no sort, cost following the number of apps asking, not
anyone's backlog; a test holds the query plan to it. And the hook keeps
the run of failures: failing_since starts at an app's first failed turn,
ends when a delivery lands or its queue empties, and at a day old the push
is suspended in one transaction — suspended_at and why, the queue given up
on, nothing new queued. Turning the push on again lifts it. The CLI says
SUSPENDED, why, and the command that turns it back on.

The status reported a give-up as the last attempt. abandonMailEventDeliveries
and DeleteMailEventHook wrote their reason into last_error, over what the
app had actually answered, so after off→on or trash→restore a live app read
"last attempt failed: the app is not live" with a stale status. Give-ups now
go in abandoned_why, and the last failure is read only from attempts since
the push was last turned on.

--off took the key back out only when a push was recorded, so a turn-on
that delivered the key and then failed to record the hook left the key for
good. It is unset whether or not a push was on.

A failed queue prune returned from the pass before anything was sent and
kept its stamp for an hour; it is logged, un-stamped, and the pass goes on.
recordMailEventAttempt's giveUp parameter had no caller left; it is gone.
The file comment now says what the fairness bound really is — per
deployment, a slot per timeout for a hung app, up to five for a slow 2xx
one — and the docs say a trashed app's events are dropped, not held.

Tests for each, each checked to fail with its fix undone, plus the gaps the
review named: status after off→on, --off after a half-failed turn-on, a move
to another account dropping the queue.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Bill Horsman pushed by bill@keymail.dev 6c78d716326842af631a132b01223311f47842a7 parent 53aafa8
8 files changed, +552 −116
  • docs/site/cli.md +5 −1
  • internal/console/apimail.go +19 −8
  • internal/console/apimail_test.go +22 −0
  • internal/console/mailpush.go +250 −86
  • internal/console/mailpush_test.go +202 −6
  • internal/console/store.go +18 −6
  • mail_cli.go +20 −9
  • mail_cli_test.go +16 −0

CI passed — run details

diff --git a/docs/site/cli.md b/docs/site/cli.md
index 0615dbcd7..618e6eab3 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -908,7 +908,11 @@ body whose `account` and `app` do not name the host the request arrived on —
`<app>.<account>.` — since the key is the deployment's, not the app's. The
body's `id` is the same on every retry of one event. Bare, `events push` shows
where events go and how delivery has been going; `--off` stops it and takes the
-key back out. Nothing is sent to an app while it is in the trash.
+key back out. While an app is in the trash, what was queued for it is given
+up on and new events are dropped, not held for a restore. An app that accepts
+nothing for a day — every attempt answered with anything but 2xx — has its push
+suspended: what is waiting is given up on, nothing new is queued, and `events
+push` says so. Running `events push --path` again turns it back on.
`credentials create` mints a standalone SMTP credential for something not running on CARLOS — a laptop, a
cron box — printed once on that command's output and nowhere else. `rotate`
diff --git a/internal/console/apimail.go b/internal/console/apimail.go
index f3683eeed..773418753 100644
--- a/internal/console/apimail.go
+++ b/internal/console/apimail.go
@@ -734,7 +734,11 @@ func (s *Server) emailPushReply(ctx context.Context, acct *Account, app string)
if err != nil {
return nil, err
}
- st, err := s.store.MailEventPushStatusFor(ctx, acct.Sqid, app)
+ var since time.Time
+ if hook != nil {
+ since = hook.SetAt
+ }
+ st, err := s.store.MailEventPushStatusFor(ctx, acct.Sqid, app, since)
if err != nil {
return nil, err
}
@@ -742,6 +746,10 @@ func (s *Server) emailPushReply(ctx context.Context, acct *Account, app string)
if hook == nil {
return out, nil
}
+ if hook.Suspended() {
+ out["suspended_at"] = hook.SuspendedAt
+ out["suspended_why"] = hook.SuspendedWhy
+ }
key, err := s.store.MailEventsSigningKey()
if err != nil {
return nil, err
@@ -862,14 +870,17 @@ func (s *Server) handleAPIEmailPushOff(w http.ResponseWriter, r *http.Request) {
jsonError(w, http.StatusBadGateway, "could not turn the email events push off — try again")
return
}
+ // The key is taken out whether or not a push was on: turning it on
+ // delivers the key before recording the hook, so a failure between the
+ // two leaves a key with no push, and --off is how that is cleaned up.
+ if removed, err := appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, nil, []string{MailEventsKeyEnv}); err != nil {
+ // A key left in the environment verifies nothing that will ever
+ // arrive. Said, not failed.
+ log.Printf("console: api email push off %s/%s: unsetting %s: %v", acct.Sqid, app, MailEventsKeyEnv, err)
+ } else if removed > 0 {
+ s.bumpSerial(ctx)
+ }
if was {
- if _, err := appconfig.Apply(ctx, s.cfg.Releases, acct.Sqid, app, nil, []string{MailEventsKeyEnv}); err != nil {
- // The push is already off; a key left in the environment verifies
- // nothing that will ever arrive. Said, not failed.
- log.Printf("console: api email push off %s/%s: unsetting %s: %v", acct.Sqid, app, MailEventsKeyEnv, err)
- } else {
- s.bumpSerial(ctx)
- }
s.audit(acct, app, addr, "email-events-push-off", "stopped pushing email events (cli)")
}
out, err := s.emailPushReply(ctx, acct, app)
diff --git a/internal/console/apimail_test.go b/internal/console/apimail_test.go
index 184258aca..82f57ed3d 100644
--- a/internal/console/apimail_test.go
+++ b/internal/console/apimail_test.go
@@ -19,6 +19,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/appconfig"
"github.com/carlosframework/platform/internal/dnsrec"
"github.com/carlosframework/platform/internal/mail"
"github.com/carlosframework/platform/internal/release"
@@ -1234,3 +1235,24 @@ func TestAPIEmailPushNeedsSomewhereToSend(t *testing.T) {
t.Error("a push was recorded with nowhere to send it")
}
}
+
+// TestAPIEmailPushOffTakesTheKeyOutEvenWithNoPush: turning the push on
+// delivers the key before it records the push, so a failure between the two
+// leaves a key and no push — and --off must still take the key out.
+func TestAPIEmailPushOffTakesTheKeyOutEvenWithNoPush(t *testing.T) {
+ fx, _, _, token := emailArmed(t, "code-email-push-halfway")
+ fx.srv.cfg.AppsDomain = "oncarlos.com"
+ app := "titogo"
+ key, _ := fx.st.MailEventsSigningKey()
+ if _, err := appconfig.Apply(context.Background(), fx.rel, fx.sqid, app,
+ map[string]appconfig.Var{MailEventsKeyEnv: {Kind: "env", Value: mailEventsPublicKey(key)}}, nil); err != nil {
+ t.Fatal(err)
+ }
+ status, _, body := rawReq(t, http.MethodDelete, emailURL(fx, app)+"/events/push", token, "")
+ if status != http.StatusOK {
+ t.Fatalf("off: %d %s", status, body)
+ }
+ if _, ok := appVars(t, fx, app)[MailEventsKeyEnv]; ok {
+ t.Errorf("%s was left behind by --off with no push recorded", MailEventsKeyEnv)
+ }
+}
diff --git a/internal/console/mailpush.go b/internal/console/mailpush.go
index 2a8160153..5179f496b 100644
--- a/internal/console/mailpush.go
+++ b/internal/console/mailpush.go
@@ -36,6 +36,18 @@ package console
// its own clock, and a body whose account or app is not its own — the key is
// the deployment's, so the last check is what stops one app's event,
// replayed at another, from being believed there.
+//
+// WHAT ONE APP CAN COST OTHERS. A pass starts from the apps that asked, and
+// reads each one's oldest due rows through an index, so its cost follows how
+// many apps ask, not how much any of them has queued. Apps are sent to eight
+// at a time across the whole deployment, not per account, and an app's turn
+// ends at its first failure — so a hung app holds one slot for one timeout a
+// pass, and an account with many hung apps costs everyone about ten seconds
+// a pass per eight of them. An app that answers 2xx slowly keeps its turn,
+// and can hold a slot for up to five timeouts. And an app that has accepted
+// nothing for a day is suspended: its queue is given up on and nothing more
+// is queued until the push is turned on again (mailPushSuspendAfter), so a
+// failing app cannot grow a backlog that a pass has to carry for a month.
import (
"bytes"
@@ -79,7 +91,12 @@ const (
// front of anybody else's.
mailPushPerApp = 5
mailPushApps = 8
- mailPushBatch = 400
+ // mailPushSuspendAfter is how long an app may go on failing — every
+ // attempt at every row answered with anything but a 2xx — before its
+ // push is suspended. A day is past any deploy an app could be halfway
+ // through, and it bounds what a failing app can have queued to a day of
+ // its own mail.
+ mailPushSuspendAfter = 24 * time.Hour
// mailPushPruneEvery bounds how often the queue is trimmed, as the log is.
mailPushPruneEvery = time.Hour
mailPushMaxBody = 64 << 10
@@ -266,29 +283,140 @@ type MailEventHook struct {
Path string
SetBy string
SetAt time.Time
+ // FailingSince is when the app's current run of failed deliveries began;
+ // zero when it has none. SuspendedAt is when the push was suspended for
+ // failing for a day, and SuspendedWhy says what the last attempt got.
+ FailingSince time.Time
+ SuspendedAt time.Time
+ SuspendedWhy string
}
+// Suspended reports whether the push is suspended.
+func (h *MailEventHook) Suspended() bool { return h != nil && !h.SuspendedAt.IsZero() }
+
+// SetMailEventHook turns the push on, or moves its path — and, either way,
+// lifts a suspension and forgets the run of failures behind it: turning it
+// on again is how a member says the handler is there now.
func (s *Store) SetMailEventHook(ctx context.Context, acct, app, path, by string, now time.Time) error {
_, err := s.db.ExecContext(ctx, `INSERT INTO mail_event_hooks (account, app, path, set_by, set_at) VALUES (?, ?, ?, ?, ?)
-ON CONFLICT(account, app) DO UPDATE SET path = excluded.path, set_by = excluded.set_by, set_at = excluded.set_at`,
+ON CONFLICT(account, app) DO UPDATE SET path = excluded.path, set_by = excluded.set_by, set_at = excluded.set_at,
+ failing_since = 0, suspended_at = 0, suspended_why = ''`,
acct, app, path, by, now.Unix())
return err
}
+const mailEventHookColumns = `account, app, path, set_by, set_at, failing_since, suspended_at, suspended_why`
+
+type mailEventHookRow struct {
+ Account, App string
+ MailEventHook
+}
+
+func scanMailEventHook(scan func(...any) error) (mailEventHookRow, error) {
+ var h mailEventHookRow
+ var setAt, failing, suspended int64
+ if err := scan(&h.Account, &h.App, &h.Path, &h.SetBy, &setAt, &failing, &suspended, &h.SuspendedWhy); err != nil {
+ return h, err
+ }
+ h.SetAt = time.Unix(setAt, 0).UTC()
+ if failing > 0 {
+ h.FailingSince = time.Unix(failing, 0).UTC()
+ }
+ if suspended > 0 {
+ h.SuspendedAt = time.Unix(suspended, 0).UTC()
+ }
+ return h, nil
+}
+
// MailEventHookFor answers nil when the app has not asked.
func (s *Store) MailEventHookFor(ctx context.Context, acct, app string) (*MailEventHook, error) {
- var h MailEventHook
- var at int64
- err := s.db.QueryRowContext(ctx, `SELECT path, set_by, set_at FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app).
- Scan(&h.Path, &h.SetBy, &at)
+ h, err := scanMailEventHook(s.db.QueryRowContext(ctx,
+ `SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app).Scan)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
- h.SetAt = time.Unix(at, 0).UTC()
- return &h, nil
+ return &h.MailEventHook, nil
+}
+
+// activeMailEventHooks is every push that is on and not suspended: where a
+// pass starts.
+func (s *Store) activeMailEventHooks(ctx context.Context) ([]mailEventHookRow, error) {
+ rows, err := s.db.QueryContext(ctx, `SELECT `+mailEventHookColumns+` FROM mail_event_hooks WHERE suspended_at = 0 ORDER BY account, app`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []mailEventHookRow
+ for rows.Next() {
+ h, err := scanMailEventHook(rows.Scan)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, h)
+ }
+ return out, rows.Err()
+}
+
+// noteMailEventTurn records how an app's turn in a pass went: a delivery
+// that landed ends any run of failures; a failure starts one if none is
+// running; and a run that has lasted mailPushSuspendAfter suspends the push
+// and gives up on its queue, in one transaction. It answers whether the push
+// is now suspended.
+func (s *Store) noteMailEventTurn(ctx context.Context, h mailEventHookRow, landed bool, failure string, now time.Time) (bool, error) {
+ switch {
+ case landed:
+ if h.FailingSince.IsZero() {
+ return false, nil
+ }
+ _, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0 WHERE account = ? AND app = ?`, h.Account, h.App)
+ return false, err
+ case failure == "":
+ return false, nil
+ case h.FailingSince.IsZero():
+ _, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = ? WHERE account = ? AND app = ? AND failing_since = 0`,
+ now.Unix(), h.Account, h.App)
+ return false, err
+ case now.Sub(h.FailingSince) < mailPushSuspendAfter:
+ return false, nil
+ }
+ why := "nothing was accepted for a day — the last attempt " + strings.TrimPrefix(failure, "no answer: ")
+ if strings.HasPrefix(failure, "no answer") {
+ why = "nothing was accepted for a day — the last attempt got no answer"
+ }
+ if len(why) > 300 {
+ why = why[:300]
+ }
+ tx, err := s.db.BeginTx(ctx, nil)
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ if _, err := tx.ExecContext(ctx, `UPDATE mail_event_hooks SET suspended_at = ?, suspended_why = ? WHERE account = ? AND app = ? AND set_at = ?`,
+ now.Unix(), why, h.Account, h.App, h.SetAt.Unix()); err != nil {
+ return false, err
+ }
+ if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push suspended'
+WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), h.Account, h.App); err != nil {
+ return false, err
+ }
+ return true, tx.Commit()
+}
+
+// clearMailEventFailing ends a run of failures that has no queue left behind
+// it: failing_since measures one backlog's failing, not the app's history,
+// so an app that failed once last week is not suspended by its next miss.
+func (s *Store) clearMailEventFailing(ctx context.Context, h mailEventHookRow) error {
+ if h.FailingSince.IsZero() {
+ return nil
+ }
+ _, err := s.db.ExecContext(ctx, `UPDATE mail_event_hooks SET failing_since = 0
+WHERE account = ? AND app = ? AND NOT EXISTS (
+ SELECT 1 FROM mail_event_deliveries WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0)`,
+ h.Account, h.App, h.Account, h.App)
+ return err
}
// DeleteMailEventHook turns the push off, and gives up on everything still
@@ -306,7 +434,7 @@ func (s *Store) DeleteMailEventHook(ctx context.Context, acct, app string, now t
return false, err
}
n, _ := res.RowsAffected()
- if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, last_error = 'push turned off'
+ if _, err := tx.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = 'push turned off'
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), acct, app); err != nil {
return false, err
}
@@ -331,16 +459,19 @@ type mailEventDelivery struct {
Attempts int
}
-// dueMailEventDeliveries is what a pass may send: at most perApp rows for
-// any one app, oldest first within it, so a backlog is its own app's to wait
-// behind.
-func (s *Store) dueMailEventDeliveries(ctx context.Context, now time.Time, perApp, limit int) ([]mailEventDelivery, error) {
- rows, err := s.db.QueryContext(ctx, `SELECT id, account, app, payload, attempts FROM (
- SELECT id, account, app, payload, attempts, next_at,
- ROW_NUMBER() OVER (PARTITION BY account, app ORDER BY next_at, id) AS nth
- FROM mail_event_deliveries
- WHERE delivered_at = 0 AND abandoned_at = 0 AND next_at <= ?
-) WHERE nth <= ? ORDER BY next_at, id LIMIT ?`, now.Unix(), perApp, limit)
+// mailPushDueSQL is the read a pass makes per app. A test holds its query
+// plan to the index.
+const mailPushDueSQL = `SELECT id, account, app, payload, attempts FROM mail_event_deliveries
+INDEXED BY mail_event_deliveries_app_due
+WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0 AND next_at <= ?
+ORDER BY next_at LIMIT ?`
+
+// dueMailEventDeliveriesFor is one app's oldest due rows, at most limit of
+// them, sought through mail_event_deliveries_app_due: the four equalities
+// and the range on next_at are all in the index, in its order, so the rows
+// come back without sorting the app's backlog.
+func (s *Store) dueMailEventDeliveriesFor(ctx context.Context, acct, app string, now time.Time, limit int) ([]mailEventDelivery, error) {
+ rows, err := s.db.QueryContext(ctx, mailPushDueSQL, acct, app, now.Unix(), limit)
if err != nil {
return nil, err
}
@@ -359,11 +490,10 @@ func (s *Store) dueMailEventDeliveries(ctx context.Context, now time.Time, perAp
}
// recordMailEventAttempt writes one attempt's outcome: delivered, due again
-// after the backoff, or given up on — once the backoff is spent, or at once
-// when giveUp says retrying cannot help. A row already given up on (the push
-// was turned off while this attempt was in flight) stays given up on: it is
-// not counted twice, and it is not revived.
-func (s *Store) recordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, giveUp bool, now time.Time) error {
+// after the backoff, or given up on once the backoff is spent. A row already
+// given up on (the push was turned off while this attempt was in flight)
+// stays given up on: it is not counted twice, and it is not revived.
+func (s *Store) recordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error {
attempts := d.Attempts + 1
if failure == "" {
_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
@@ -374,14 +504,9 @@ SET attempts = ?, delivered_at = ?, last_attempt_at = ?, last_status = ?, last_e
if len(failure) > 300 {
failure = failure[:300]
}
- if giveUp {
- _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
-SET abandoned_at = ?, last_error = ? WHERE id = ? AND abandoned_at = 0`, now.Unix(), failure, d.ID)
- return err
- }
if attempts > len(mailPushBackoff) {
_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
-SET attempts = ?, abandoned_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND abandoned_at = 0`,
+SET attempts = ?, abandoned_at = ?, abandoned_why = 'retries spent', last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND abandoned_at = 0`,
attempts, now.Unix(), now.Unix(), status, failure, d.ID)
return err
}
@@ -392,9 +517,10 @@ SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error
}
// abandonMailEventDeliveries gives up on everything still queued for one
-// app, in one statement.
+// app, in one statement. The reason goes in abandoned_why, never over
+// last_error: what the app last answered stays what it last answered.
func (s *Store) abandonMailEventDeliveries(ctx context.Context, acct, app, reason string, now time.Time) error {
- _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, last_error = ?
+ _, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries SET abandoned_at = ?, abandoned_why = ?
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), reason, acct, app)
return err
}
@@ -410,8 +536,12 @@ func (s *Store) pruneMailEventDeliveries(ctx context.Context, now time.Time) err
if !s.mailPushPruned.CompareAndSwap(last, now.Unix()) {
return nil
}
- _, err := s.db.ExecContext(ctx, `DELETE FROM mail_event_deliveries WHERE created_at < ?`, now.Add(-mailEventsRetention).Unix())
- return err
+ if _, err := s.db.ExecContext(ctx, `DELETE FROM mail_event_deliveries WHERE created_at < ?`, now.Add(-mailEventsRetention).Unix()); err != nil {
+ // Not pruned, so not stamped: the next pass tries again.
+ s.mailPushPruned.CompareAndSwap(now.Unix(), last)
+ return err
+ }
+ return nil
}
// MailEventPushStatus is what `carlos email events push` shows.
@@ -425,7 +555,11 @@ type MailEventPushStatus struct {
LastError string `json:"last_error,omitempty"`
}
-func (s *Store) MailEventPushStatusFor(ctx context.Context, acct, app string) (MailEventPushStatus, error) {
+// MailEventPushStatusFor reads the counts over the retention window, and the
+// last failure since `since` — the moment the push was last turned on — so a
+// failure from before a member turned it off and on again, or before a
+// suspension was lifted, is not reported as today's news.
+func (s *Store) MailEventPushStatusFor(ctx context.Context, acct, app string, since time.Time) (MailEventPushStatus, error) {
var st MailEventPushStatus
var lastDelivered sql.NullInt64
if err := s.db.QueryRowContext(ctx, `SELECT
@@ -439,14 +573,14 @@ FROM mail_event_deliveries WHERE account = ? AND app = ?`, acct, app).Scan(&st.P
if lastDelivered.Valid {
st.LastDelivered = time.Unix(lastDelivered.Int64, 0).UTC()
}
- // The most recent attempt that did not land. The reader decides whether
- // it is news: one older than the last delivery has been recovered from.
- // A row given up on because the push was turned off was never attempted
- // and is not a failure of the app's.
+ // The most recent attempt that did not land, since the push was turned
+ // on. The reader decides whether it is news: one older than the last
+ // delivery has been recovered from. Giving up on a row is not an attempt
+ // and never writes these columns.
var failedAt int64
err := s.db.QueryRowContext(ctx, `SELECT last_attempt_at, last_status, last_error FROM mail_event_deliveries
-WHERE account = ? AND app = ? AND delivered_at = 0 AND last_attempt_at > 0
-ORDER BY last_attempt_at DESC, id DESC LIMIT 1`, acct, app).
+WHERE account = ? AND app = ? AND delivered_at = 0 AND last_attempt_at > 0 AND last_attempt_at >= ?
+ORDER BY last_attempt_at DESC, id DESC LIMIT 1`, acct, app, since.Unix()).
Scan(&failedAt, &st.LastStatus, &st.LastError)
switch {
case err == sql.ErrNoRows:
@@ -474,7 +608,7 @@ func (s *Server) queueMailEventPush(ctx context.Context, acct, app, notice strin
log.Printf("console: mail push: reading the hook for %s/%s: %v", acct, app, err)
return
}
- if hook == nil {
+ if hook == nil || hook.Suspended() {
return
}
body, err := json.Marshal(mailEventPayload(acct, app, notice, ev, now))
@@ -517,44 +651,55 @@ func (s *Server) StartMailEventPush(ctx context.Context) {
}
}
-// mailPushPass sends what is due and trims the queue. Apps are sent to in
+// mailPushPass sends what is due and trims the queue. It starts from the
+// apps that asked — a short list — and seeks each one's oldest due rows, so
+// its cost does not grow with anyone's backlog. Apps are sent to in
// parallel, mailPushApps at a time, and each app's own rows in order; see
// deliverAppMailEvents for why one app cannot hold up another. clock is read
// per delivery so a slow pass records honest times.
func (s *Server) mailPushPass(ctx context.Context, clock func() time.Time) error {
start := clock()
if err := s.store.pruneMailEventDeliveries(ctx, start); err != nil {
- return fmt.Errorf("pruning the queue: %w", err)
+ // A prune that fails costs an hour's slack on a thirty-day window,
+ // not the pass: what is due still goes out.
+ log.Printf("console: mail push: pruning the queue: %v", err)
}
- due, err := s.store.dueMailEventDeliveries(ctx, start, mailPushPerApp, mailPushBatch)
+ hooks, err := s.store.activeMailEventHooks(ctx)
if err != nil {
- return fmt.Errorf("reading the queue: %w", err)
+ return fmt.Errorf("reading the hooks: %w", err)
}
- if len(due) == 0 {
+ if len(hooks) == 0 {
return nil
}
- key, err := s.store.MailEventsSigningKey()
- if err != nil {
- return fmt.Errorf("reading the signing key: %w", err)
- }
- byApp := map[string][]mailEventDelivery{}
- var order []string
- for _, d := range due {
- k := d.Account + "/" + d.App
- if _, seen := byApp[k]; !seen {
- order = append(order, k)
- }
- byApp[k] = append(byApp[k], d)
- }
+ var key ed25519.PrivateKey
sem := make(chan struct{}, mailPushApps)
var wg sync.WaitGroup
- for _, k := range order {
- rows := byApp[k]
+ for _, h := range hooks {
+ if ctx.Err() != nil {
+ break
+ }
+ rows, err := s.store.dueMailEventDeliveriesFor(ctx, h.Account, h.App, start, mailPushPerApp)
+ if err != nil {
+ log.Printf("console: mail push: reading the queue for %s/%s: %v", h.Account, h.App, err)
+ continue
+ }
+ if len(rows) == 0 {
+ if err := s.store.clearMailEventFailing(ctx, h); err != nil {
+ log.Printf("console: mail push: %s/%s: %v", h.Account, h.App, err)
+ }
+ continue
+ }
+ if key == nil {
+ if key, err = s.store.MailEventsSigningKey(); err != nil {
+ wg.Wait()
+ return fmt.Errorf("reading the signing key: %w", err)
+ }
+ }
wg.Add(1)
sem <- struct{}{}
go func() {
defer func() { <-sem; wg.Done() }()
- s.deliverAppMailEvents(ctx, rows, key, clock)
+ s.deliverAppMailEvents(ctx, h, rows, key, clock)
}()
}
wg.Wait()
@@ -565,53 +710,72 @@ func (s *Server) mailPushPass(ctx context.Context, clock func() time.Time) error
// does not land ends the app's turn for this pass: an app that answers 500
// after nine seconds is no more likely to take the next row than the last,
// and each row it is offered holds one of mailPushApps for a timeout that
-// every other app waits on. The rest keep their place for the next pass.
+// every other app waits on. The rest keep their place for the next pass, and
+// how the turn went is noted on the hook — which is where a day of nothing
+// but failures suspends the push.
//
-// A row given up on because the app is no longer live, or the push is off,
-// gives up on all of that app's rows at once — there is nothing to wait for.
-func (s *Server) deliverAppMailEvents(ctx context.Context, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) {
+// An app that is no longer live, or whose push was turned off since the pass
+// began, has all of its rows given up on at once — there is nothing to wait
+// for.
+func (s *Server) deliverAppMailEvents(ctx context.Context, h mailEventHookRow, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) {
+ if !s.mailAppLive(h.Account, h.App) {
+ // A trashed app's name is no longer a member's live app, and while it
+ // sits in the trash its platform host is not reliably its own (the
+ // instances door treats a trashed sibling's apex as free), so nothing
+ // is sent to it; a purged, renamed or moved app's rows have already
+ // been forgotten, carried or dropped.
+ if err := s.store.abandonMailEventDeliveries(ctx, h.Account, h.App, "the app is not live", clock()); err != nil {
+ log.Printf("console: mail push: giving up on %s/%s: %v", h.Account, h.App, err)
+ }
+ return
+ }
+ landed, failure := false, ""
for _, d := range rows {
if ctx.Err() != nil {
- return
+ break
}
- status, failure, giveUp := s.deliverMailEvent(ctx, d, key, clock)
+ status, why, giveUp := s.deliverMailEvent(ctx, d, h.Path, key, clock)
if giveUp {
- if err := s.store.abandonMailEventDeliveries(ctx, d.Account, d.App, failure, clock()); err != nil {
+ if err := s.store.abandonMailEventDeliveries(ctx, d.Account, d.App, why, clock()); err != nil {
log.Printf("console: mail push: giving up on %s/%s: %v", d.Account, d.App, err)
}
return
}
- if err := s.store.recordMailEventAttempt(ctx, d, status, failure, false, clock()); err != nil {
+ if err := s.store.recordMailEventAttempt(ctx, d, status, why, clock()); err != nil {
log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err)
}
- if failure != "" {
- return
+ if why != "" {
+ failure = why
+ break
}
+ landed = true
+ }
+ suspended, err := s.store.noteMailEventTurn(ctx, h, landed, failure, clock())
+ if err != nil {
+ log.Printf("console: mail push: noting the turn for %s/%s: %v", h.Account, h.App, err)
+ }
+ if suspended {
+ log.Printf("console: mail push: suspended %s/%s — nothing accepted for %s", h.Account, h.App, mailPushSuspendAfter)
}
}
// deliverMailEvent makes one attempt, and answers the HTTP status, what went
// wrong ("" when it landed), and whether to give up on the row now rather
// than retry it.
-func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) (int, string, bool) {
- // The app must still be the app that asked. A trashed app's name is no
- // longer a member's live app, and while it sits in the trash its platform
- // host is not reliably its own (the instances door treats a trashed
- // sibling's apex as free), so nothing is sent to it; a purged, renamed or
- // moved app's rows have already been forgotten or carried.
- if !s.mailAppLive(d.Account, d.App) {
- return 0, "the app is not live", true
- }
+func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, path string, key ed25519.PrivateKey, clock func() time.Time) (int, string, bool) {
hook, err := s.store.MailEventHookFor(ctx, d.Account, d.App)
if err != nil {
return 0, "could not read the hook: " + err.Error(), false
}
- if hook == nil {
- // Turned off between queueing and sending. DeleteMailEventHook gives
- // up on the queue itself; this covers the race.
+ if hook == nil || hook.Suspended() {
+ // Turned off, or suspended, since the pass began. Both give up on the
+ // queue themselves; this covers the race.
return 0, "push turned off", true
}
- target, err := s.mailPushURL(d.Account, d.App, hook.Path)
+ if hook.Path != path {
+ path = hook.Path // moved since the pass began; the new path is the one asked for
+ }
+ target, err := s.mailPushURL(d.Account, d.App, path)
if err != nil {
return 0, err.Error(), false
}
diff --git a/internal/console/mailpush_test.go b/internal/console/mailpush_test.go
index 00133fd84..19c965c0a 100644
--- a/internal/console/mailpush_test.go
+++ b/internal/console/mailpush_test.go
@@ -186,7 +186,7 @@ func TestADeliveryReachesTheAppSigned(t *testing.T) {
if len(p.Recipients) != 1 || p.Recipients[0].Address != "ann@example.com" || p.Recipients[0].Status != "5.1.1" {
t.Errorf("recipients = %+v", p.Recipients)
}
- st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Delivered != 1 || st.Pending != 0 || st.LastDelivered.IsZero() {
t.Errorf("status after delivery = %+v", st)
}
@@ -223,7 +223,7 @@ func TestAFailedDeliveryBacksOffAndIsGivenUpOn(t *testing.T) {
tick := func() time.Time { return clock }
fx.srv.mailPushPass(ctx, tick)
- st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Pending != 1 || st.LastStatus != 503 || !strings.Contains(st.LastError, "503") || !st.LastFailedAt.Equal(clock) {
t.Fatalf("after one failure: %+v — the failure should be dated when it happened, %s", st, clock.UTC())
}
@@ -240,7 +240,7 @@ func TestAFailedDeliveryBacksOffAndIsGivenUpOn(t *testing.T) {
if hits != len(mailPushBackoff)+1 {
t.Errorf("attempts = %d, want %d", hits, len(mailPushBackoff)+1)
}
- st, _ = fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ = fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Abandoned != 1 || st.Pending != 0 {
t.Errorf("after the schedule is spent: %+v", st)
}
@@ -269,7 +269,7 @@ func TestADeliveryDoesNotFollowARedirect(t *testing.T) {
ctx := context.Background()
fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1")
fx.srv.mailPushPass(ctx, time.Now)
- st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Delivered != 0 || !strings.Contains(st.LastError, "redirect") {
t.Errorf("a redirect counted as delivered, or went unexplained: %+v", st)
}
@@ -352,7 +352,7 @@ func TestAHungAppDoesNotHoldUpAnotherApp(t *testing.T) {
if hungHits != 1 {
t.Errorf("the hung app was tried %d times in one pass, want 1 — its turn ends at the first silence", hungHits)
}
- st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Pending != 5 {
t.Errorf("the hung app's deliveries should all still be waiting: %+v", st)
}
@@ -400,7 +400,7 @@ func TestNothingIsPushedForATrashedApp(t *testing.T) {
if hits != 0 {
t.Error("a delivery went to a trashed app's host")
}
- st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{})
if st.Abandoned != 1 || st.Pending != 0 {
t.Errorf("a trashed app's delivery should be given up on: %+v", st)
}
@@ -468,3 +468,199 @@ func TestAFailingAppEndsItsTurnAtTheFirstFailure(t *testing.T) {
t.Errorf("a failing app was offered %d rows in one pass, want 1", hits)
}
}
+
+// An app that accepts nothing for a day is suspended: what is waiting is
+// given up on, nothing new is queued, and turning the push on again lifts
+// it. Without this, a failing app's backlog grows for the thirty days the
+// queue is kept.
+func TestADayOfFailuresSuspendsThePush(t *testing.T) {
+ hits := 0
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {
+ hits++
+ w.WriteHeader(http.StatusNotFound)
+ })
+ ctx := context.Background()
+ send := func(id string, at time.Time) {
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, at)), "sns-"+id)
+ }
+ clock := time.Now()
+ tick := func() time.Time { return clock }
+ for i := 0; i < 20; i++ {
+ send("m-"+strconv.Itoa(i), clock)
+ }
+ fx.srv.mailPushPass(ctx, tick)
+ hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book")
+ if hook.FailingSince.IsZero() || hook.Suspended() {
+ t.Fatalf("after the first failure: %+v", hook)
+ }
+
+ // Most of a day of failing is not a day.
+ clock = clock.Add(23 * time.Hour)
+ fx.srv.mailPushPass(ctx, tick)
+ if hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() {
+ t.Fatal("suspended before a day had passed")
+ }
+
+ clock = clock.Add(2 * time.Hour)
+ fx.srv.mailPushPass(ctx, tick)
+ hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book")
+ if !hook.Suspended() || !strings.Contains(hook.SuspendedWhy, "404") {
+ t.Fatalf("a day of failures should suspend, saying what the app answered: %+v", hook)
+ }
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", hook.SetAt)
+ if st.Pending != 0 || st.Abandoned != 20 {
+ t.Errorf("a suspension should give up on the whole queue: %+v", st)
+ }
+
+ // Suspended: nothing new is queued, and a pass sends nothing.
+ send("m-late", clock)
+ if n := queued(t, fx, "jb-book"); n != 20 {
+ t.Errorf("a suspended push queued more (%d rows)", n)
+ }
+ before := hits
+ clock = clock.Add(time.Hour)
+ fx.srv.mailPushPass(ctx, tick)
+ if hits != before {
+ t.Error("a suspended push was sent to")
+ }
+
+ // Turning it on again lifts the suspension and starts clean.
+ if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", clock); err != nil {
+ t.Fatal(err)
+ }
+ hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book")
+ if hook.Suspended() || !hook.FailingSince.IsZero() {
+ t.Errorf("turning it on again should lift the suspension: %+v", hook)
+ }
+ send("m-after", clock)
+ if n := queued(t, fx, "jb-book"); n != 21 {
+ t.Errorf("after turning it on again, events should queue: %d rows", n)
+ }
+}
+
+// A run of failures belongs to a backlog, not to the app's history: an app
+// whose queue emptied — here by its only delivery running out of retries,
+// with nothing ever landing — is not suspended by its next miss days later.
+func TestAnOldFailureWithNothingWaitingDoesNotSuspend(t *testing.T) {
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) })
+ ctx := context.Background()
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1")
+ var created int64
+ fx.st.db.QueryRow(`SELECT created_at FROM mail_event_deliveries`).Scan(&created)
+ clock := time.Unix(created, 0)
+ tick := func() time.Time { return clock }
+ fx.srv.mailPushPass(ctx, tick)
+ for _, wait := range mailPushBackoff {
+ clock = clock.Add(wait)
+ fx.srv.mailPushPass(ctx, tick)
+ }
+ if st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}); st.Pending != 0 || st.Abandoned != 1 {
+ t.Fatalf("the one delivery should have run out of retries by now: %+v", st)
+ }
+ clock = clock.Add(3 * 24 * time.Hour)
+ fx.srv.mailPushPass(ctx, tick) // nothing waiting: the run ends here
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-2", clock)), "sns-2")
+ fx.st.db.Exec(`UPDATE mail_event_deliveries SET next_at = ?, created_at = ? WHERE notice = 'sns-2'`, clock.Unix(), clock.Unix())
+ fx.srv.mailPushPass(ctx, tick)
+ if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() {
+ t.Errorf("one miss after days of quiet suspended the push: %+v", hook)
+ }
+}
+
+// A pass seeks each app's oldest due rows through an index rather than
+// sorting everything due: its cost follows the number of apps asking, not
+// the size of anyone's backlog.
+func TestAPassDoesNotSortAnAppsBacklog(t *testing.T) {
+ fx := newConsoleFixture(t)
+ rows, err := fx.st.db.Query(`EXPLAIN QUERY PLAN `+mailPushDueSQL, "beb", "jb-book", time.Now().Unix(), mailPushPerApp)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer rows.Close()
+ var plan []string
+ for rows.Next() {
+ var id, parent, notused int
+ var detail string
+ rows.Scan(&id, &parent, &notused, &detail)
+ plan = append(plan, detail)
+ }
+ joined := strings.Join(plan, " | ")
+ if !strings.Contains(joined, "mail_event_deliveries_app_due") || strings.Contains(joined, "TEMP B-TREE") {
+ t.Errorf("the due read should seek the per-app index without sorting: %s", joined)
+ }
+
+ // And a large backlog costs a pass one app's worth of rows.
+ sent := 0
+ fx2, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { sent++; w.WriteHeader(http.StatusNoContent) })
+ ctx := context.Background()
+ now := time.Now()
+ for i := 0; i < 2000; i++ {
+ if err := fx2.st.EnqueueMailEventDelivery(ctx, release.OpsAccount, "jb-book", "bulk-"+strconv.Itoa(i), []byte(`{}`), now); err != nil {
+ t.Fatal(err)
+ }
+ }
+ fx2.srv.mailPushPass(ctx, time.Now)
+ if sent != mailPushPerApp {
+ t.Errorf("one pass sent %d, want %d", sent, mailPushPerApp)
+ }
+}
+
+// Turning the push off and on again, or lifting a suspension, starts the
+// status afresh: an old failure is not reported as the last attempt.
+func TestTheStatusForgetsFailuresFromBeforeThePushWasTurnedOn(t *testing.T) {
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) })
+ ctx := context.Background()
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1")
+ fx.srv.mailPushPass(ctx, time.Now)
+ if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ later := time.Now().Add(time.Minute)
+ if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", later); err != nil {
+ t.Fatal(err)
+ }
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", later)
+ if st.LastError != "" || st.LastStatus != 0 {
+ t.Errorf("a failure from before the push was turned on again was reported: %+v", st)
+ }
+ var why, lastErr string
+ fx.st.db.QueryRow(`SELECT abandoned_why, last_error FROM mail_event_deliveries`).Scan(&why, &lastErr)
+ if why != "push turned off" || !strings.Contains(lastErr, "503") {
+ t.Errorf("giving up should record why without overwriting what the app answered: why=%q last_error=%q", why, lastErr)
+ }
+}
+
+// A move to another account drops the queue, like a rename: the bodies name
+// the old account.
+func TestAMoveDropsTheQueue(t *testing.T) {
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {})
+ ctx := context.Background()
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1")
+ if err := fx.st.CreateAccount(99, "dest", "Destination", "bill@tito.io", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.st.MoveApp("jb-site", OpsAccountID, 99); err != nil {
+ t.Fatal(err)
+ }
+ var n int
+ fx.st.db.QueryRow(`SELECT count(*) FROM mail_event_deliveries WHERE app = 'jb-site'`).Scan(&n)
+ if n != 0 {
+ t.Errorf("a move carried %d queued deliveries naming the old account", n)
+ }
+ if hook, _ := fx.st.MailEventHookFor(ctx, "dest", "jb-site"); hook == nil {
+ t.Error("the push itself should move with the app")
+ }
+}
+
+// A prune that fails is not stamped, and does not cost the pass.
+func TestAFailedQueuePruneIsTriedAgain(t *testing.T) {
+ fx := newConsoleFixture(t)
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ if err := fx.st.pruneMailEventDeliveries(ctx, time.Now()); err == nil {
+ t.Fatal("a prune with a cancelled context should fail")
+ }
+ if got := fx.st.mailPushPruned.Load(); got != 0 {
+ t.Errorf("the stamp was left at %d after a failed prune", got)
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index ec5ca86fe..fd47e4e41 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -246,12 +246,20 @@ func (s *Store) migrate() error {
// Where an app wants to be told about its email (mailpush.go): one
// path per app, POSTed on the app's own platform host. Absent row =
// the app is not told, which is every app until it asks.
+ //
+ // failing_since is when the app's current run of failed deliveries
+ // began (0: none), and suspended_at when the push was suspended for
+ // a day of them (0: not suspended) — see mailPushSuspendAfter. Both
+ // clear when the push is turned on again.
`CREATE TABLE IF NOT EXISTS mail_event_hooks (
- account TEXT NOT NULL,
- app TEXT NOT NULL,
- path TEXT NOT NULL,
- set_by TEXT NOT NULL,
- set_at INTEGER NOT NULL,
+ account TEXT NOT NULL,
+ app TEXT NOT NULL,
+ path TEXT NOT NULL,
+ set_by TEXT NOT NULL,
+ set_at INTEGER NOT NULL,
+ failing_since INTEGER NOT NULL DEFAULT 0,
+ suspended_at INTEGER NOT NULL DEFAULT 0,
+ suspended_why TEXT NOT NULL DEFAULT '',
PRIMARY KEY (account, app)
);`,
// The queue of those notifications, one row per SES notification per
@@ -274,9 +282,13 @@ func (s *Store) migrate() error {
last_attempt_at INTEGER NOT NULL DEFAULT 0,
last_status INTEGER NOT NULL DEFAULT 0,
last_error TEXT NOT NULL DEFAULT '',
+ abandoned_why TEXT NOT NULL DEFAULT '',
UNIQUE (account, app, notice)
);`,
- `CREATE INDEX IF NOT EXISTS mail_event_deliveries_due ON mail_event_deliveries (delivered_at, abandoned_at, next_at);`,
+ // A pass seeks each app's oldest due rows through this index, so its
+ // cost follows the number of apps asking, not the size of anyone's
+ // backlog.
+ `CREATE INDEX IF NOT EXISTS mail_event_deliveries_app_due ON mail_event_deliveries (account, app, delivered_at, abandoned_at, next_at);`,
`CREATE INDEX IF NOT EXISTS mail_event_deliveries_by_age ON mail_event_deliveries (created_at);`,
} {
if _, err := s.db.Exec(stmt); err != nil {
diff --git a/mail_cli.go b/mail_cli.go
index e8a256410..2f9110ff2 100644
--- a/mail_cli.go
+++ b/mail_cli.go
@@ -942,14 +942,18 @@ func runEmailEvents(args []string) {
// apiEmailPushReply is what the three push doors answer.
type apiEmailPushReply struct {
- On bool `json:"on"`
- Path string `json:"path,omitempty"`
- URL string `json:"url,omitempty"`
- KeyEnv string `json:"key_env"`
- PublicKey string `json:"public_key,omitempty"`
- SetBy string `json:"set_by,omitempty"`
- SetAt time.Time `json:"set_at,omitempty"`
- Status apiEmailPushDelivery `json:"status"`
+ On bool `json:"on"`
+ Path string `json:"path,omitempty"`
+ URL string `json:"url,omitempty"`
+ KeyEnv string `json:"key_env"`
+ PublicKey string `json:"public_key,omitempty"`
+ SetBy string `json:"set_by,omitempty"`
+ SetAt time.Time `json:"set_at,omitempty"`
+ // SuspendedAt is set when the push was suspended after a day in which
+ // the app accepted nothing; SuspendedWhy says what the last attempt got.
+ SuspendedAt time.Time `json:"suspended_at,omitempty"`
+ SuspendedWhy string `json:"suspended_why,omitempty"`
+ Status apiEmailPushDelivery `json:"status"`
}
type apiEmailPushDelivery struct {
@@ -1039,6 +1043,13 @@ func printEmailPush(w io.Writer, app string, res apiEmailPushReply, justSet bool
mutedf(w, "turn it on: carlos email events push --app %s --path /mail/events", app)
return
}
+ if !res.SuspendedAt.IsZero() {
+ fmt.Fprintf(w, "%s's email events push is SUSPENDED since %s\n", app, res.SuspendedAt.Local().Format("2006-01-02 15:04"))
+ fmt.Fprintf(w, " %s\n", res.SuspendedWhy)
+ fmt.Fprintf(w, " nothing is queued or sent while it is suspended; what was waiting was given up on\n")
+ mutedf(w, " when %s answers 2xx again, turn it back on: carlos email events push --app %s --path %s", res.URL, app, res.Path)
+ return
+ }
fmt.Fprintf(w, "%s is told about its email at %s\n", app, res.URL)
fmt.Fprintf(w, " verify with %s: an Ed25519 public key, standard base64\n", res.KeyEnv)
if justSet {
@@ -1053,7 +1064,7 @@ func printEmailPush(w io.Writer, app string, res apiEmailPushReply, justSet bool
fmt.Fprintf(w, " last attempt failed: %s\n", st.LastError)
switch {
case st.LastStatus == http.StatusNotFound:
- mutedf(w, " the app has nothing at %s yet — deploy the handler; waiting deliveries are retried", res.Path)
+ mutedf(w, " the app has nothing at %s yet — deploy the handler; waiting deliveries are retried, and a day with none accepted suspends the push", res.Path)
case st.LastStatus == http.StatusUnauthorized:
mutedf(w, " a 401 is usually carlos gate — a gated app cannot receive email events")
}
diff --git a/mail_cli_test.go b/mail_cli_test.go
index ba995aada..7c0dcafa5 100644
--- a/mail_cli_test.go
+++ b/mail_cli_test.go
@@ -1209,3 +1209,19 @@ func TestPrintEmailPushPointsAtTheFix(t *testing.T) {
t.Errorf("an old failure was reported after a later delivery landed:\n%s", buf.String())
}
}
+
+// A suspended push says so first, says why, and says how to turn it back on.
+func TestPrintEmailPushSaysItIsSuspended(t *testing.T) {
+ var buf bytes.Buffer
+ printEmailPush(&buf, "jb-book", apiEmailPushReply{
+ On: true, Path: "/mail/events", URL: "https://jb-book.beb.oncarlos.com/mail/events", KeyEnv: "CARLOS_MAIL_EVENTS_KEY",
+ SuspendedAt: time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC),
+ SuspendedWhy: "nothing was accepted for a day — the last attempt answered 404",
+ }, false)
+ out := buf.String()
+ for _, want := range []string{"SUSPENDED", "answered 404", "given up on", "carlos email events push --app jb-book --path /mail/events"} {
+ if !strings.Contains(out, want) {
+ t.Errorf("missing %q:\n%s", want, out)
+ }
+ }
+}