amadan

carlos / platform Public

email events push: fair between apps, and only for a live app (review fixes)

From an Opus review of this branch.

One app could switch the push off for every app. The queue was one line,
worked one delivery at a time, 20s a POST and a minute a pass, so an app
whose handler hung — or just woke slowly — sending ten messages a minute
grew everyone's delay without bound. Now a pass takes at most five rows per
app (a window function, not a LIMIT across the whole queue), sends to up to
eight apps side by side, and ends an app's turn at the first attempt that
got no answer at all, so a hung handler costs its own app one timeout per
pass and nobody else anything. The timeout is 10s.

A trashed app kept being sent addresses. Nothing checked the app was still
live, and while an app sits in the trash its apex is not reliably its own —
the instances door treats a trashed sibling's apex as free. The sender now
gives up on any row for an app that is not live, and the ingest queues
nothing for one. mail_event_hooks and mail_event_deliveries join mail_events
in the tables a purge forgets and a rename or a move carries, so a new app
of the same name inherits no push and no queue.

The prune scanned the whole queue every 15s with no index on created_at,
behind a single-connection store. It has the index now, and runs at most
hourly, as the log's does.

A delivery landing after the push was turned off mid-flight marked an
abandoned row delivered and counted it twice; every attempt now leaves a
given-up row alone. DeleteMailEventHook is one transaction, so the push is
never off with its queue still live.

Docs: the host an app should check the body against is named — the key is
the deployment's — and a split paragraph is joined.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Bill Horsman pushed by bill@keymail.dev c2dc6132ad5d1e706d93a5cb2a34053b17cd27ec parent c65c8dc
5 files changed, +279 −54
  • docs/site/cli.md +7 −5
  • internal/console/mailevents.go +5 −2
  • internal/console/mailpush.go +127 −47
  • internal/console/mailpush_test.go +137 −0
  • internal/console/store.go +3 −0
diff --git a/docs/site/cli.md b/docs/site/cli.md
index 1884875d8..0615dbcd7 100644
--- a/docs/site/cli.md
+++ b/docs/site/cli.md
@@ -904,11 +904,13 @@ Every delivery carries a `Carlos-Signature: t=<unix seconds>,ed25519=<base64>`
header over the bytes `carlos-mail-event`, a newline, the `t` value, a newline,
and the request body exactly as received. An app should refuse a signature
that does not verify, a `t` more than five minutes from its own clock, and a
-body whose `account` or `app` is not its own. 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.
- `credentials create` mints a
-standalone SMTP credential for something not running on CARLOS — a laptop, a
+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.
+
+`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`
mints a fresh delivered credential and leaves the old one live until you run
it again with `--finish`. `pause` and `resume` are operator verbs. None of it
diff --git a/internal/console/mailevents.go b/internal/console/mailevents.go
index 26c8629bf..2895377fb 100644
--- a/internal/console/mailevents.go
+++ b/internal/console/mailevents.go
@@ -91,7 +91,7 @@ func mailNoticeKey(notice string, ev mail.Event, now time.Time) string {
// mailAppTables are the tables keyed by an app's (account sqid, name) that
// belong to that app alone: a name that is freed must not hand them on.
-var mailAppTables = []string{"mail_events"}
+var mailAppTables = []string{"mail_events", "mail_event_hooks", "mail_event_deliveries"}
// execer is what carryMailApp and forgetMailApp need from a transaction.
type execer interface {
@@ -158,8 +158,11 @@ func forgetMailApp(ctx context.Context, tx execer, sqid, app string) error {
// mailAppTimeColumn is the column that says when a row came to be.
func mailAppTimeColumn(table string) string {
- if table == "mail_events" {
+ switch table {
+ case "mail_events":
return "received_at"
+ case "mail_event_hooks":
+ return "set_at"
}
return "created_at"
}
diff --git a/internal/console/mailpush.go b/internal/console/mailpush.go
index ac5d490d0..b314899c6 100644
--- a/internal/console/mailpush.go
+++ b/internal/console/mailpush.go
@@ -53,6 +53,7 @@ import (
"net/http"
"strconv"
"strings"
+ "sync"
"time"
"github.com/carlosframework/platform/internal/mail"
@@ -70,17 +71,24 @@ const (
// long for its first attempt; notifications themselves arrive seconds
// after SES hears back from the receiving server.
mailPushEvery = 15 * time.Second
- // mailPushTimeout bounds one POST. Long enough for the edge to wake a
- // hibernating app and for it to answer; short enough that one wedged
- // app does not hold up everyone else's deliveries for long.
- mailPushTimeout = 20 * time.Second
- // mailPushBatch and mailPushPassBudget bound one pass: so many rows,
- // and no new row started after the budget is spent.
- mailPushBatch = 50
- mailPushPassBudget = time.Minute
+ // mailPushPerApp is the most one app is sent in a pass, and
+ // mailPushApps how many apps are sent to at once. Together they are what
+ // keeps one app from being everyone's problem: an app whose handler
+ // hangs occupies one of mailPushApps for at most one timeout (see
+ // deliverAppMailEvents), and its backlog waits behind itself, not in
+ // front of anybody else's.
+ mailPushPerApp = 5
+ mailPushApps = 8
+ mailPushBatch = 400
+ // mailPushPruneEvery bounds how often the queue is trimmed, as the log is.
+ mailPushPruneEvery = time.Hour
mailPushMaxBody = 64 << 10
)
+// mailPushTimeout bounds one POST: enough for the edge to wake a
+// hibernating app and for it to answer. A var so a test can shorten it.
+var mailPushTimeout = 10 * time.Second
+
// mailPushBackoff is when each retry is due after a failed attempt, counted
// from that attempt. After the last, the delivery is given up on — a little
// under a day of trying, which is past any deploy an app could be halfway
@@ -89,9 +97,9 @@ var mailPushBackoff = []time.Duration{
time.Minute, 5 * time.Minute, 15 * time.Minute, time.Hour, 3 * time.Hour, 6 * time.Hour, 12 * time.Hour,
}
-// mailPushClient refuses redirects: see the file comment.
+// mailPushClient refuses redirects: see the file comment. The per-request
+// timeout is the context's (deliverMailEvent), so a test can shorten it.
var mailPushClient = &http.Client{
- Timeout: mailPushTimeout,
CheckRedirect: func(*http.Request, []*http.Request) error {
return http.ErrUseLastResponse
},
@@ -285,18 +293,24 @@ func (s *Store) MailEventHookFor(ctx context.Context, acct, app string) (*MailEv
// DeleteMailEventHook turns the push off, and gives up on everything still
// queued for it: an app that has stopped asking must not keep being sent
-// addresses from a queue it has forgotten about.
+// addresses from a queue it has forgotten about. One transaction, so the
+// push is never off with its queue still live.
func (s *Store) DeleteMailEventHook(ctx context.Context, acct, app string, now time.Time) (bool, error) {
- res, err := s.db.ExecContext(ctx, `DELETE FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app)
+ tx, err := s.db.BeginTx(ctx, nil)
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ res, err := tx.ExecContext(ctx, `DELETE FROM mail_event_hooks WHERE account = ? AND app = ?`, acct, app)
if err != nil {
return false, err
}
n, _ := res.RowsAffected()
- if _, err := s.db.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 = ?, last_error = 'push turned off'
WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), acct, app); err != nil {
- return n > 0, err
+ return false, err
}
- return n > 0, nil
+ return n > 0, tx.Commit()
}
// ---- the queue --------------------------------------------------------------------
@@ -317,9 +331,16 @@ type mailEventDelivery struct {
Attempts int
}
-func (s *Store) dueMailEventDeliveries(ctx context.Context, now time.Time, limit int) ([]mailEventDelivery, error) {
- rows, err := s.db.QueryContext(ctx, `SELECT id, account, app, payload, attempts FROM mail_event_deliveries
-WHERE delivered_at = 0 AND abandoned_at = 0 AND next_at <= ? ORDER BY next_at, id LIMIT ?`, now.Unix(), limit)
+// 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)
if err != nil {
return nil, err
}
@@ -338,26 +359,34 @@ WHERE delivered_at = 0 AND abandoned_at = 0 AND next_at <= ? ORDER BY next_at, i
}
// recordMailEventAttempt writes one attempt's outcome: delivered, due again
-// after the backoff, or given up on once the backoff is spent.
-func (s *Store) recordMailEventAttempt(ctx context.Context, d mailEventDelivery, status int, failure string, now time.Time) error {
+// 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 {
attempts := d.Attempts + 1
if failure == "" {
_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
-SET attempts = ?, delivered_at = ?, last_attempt_at = ?, last_status = ?, last_error = '' WHERE id = ?`,
+SET attempts = ?, delivered_at = ?, last_attempt_at = ?, last_status = ?, last_error = '' WHERE id = ? AND abandoned_at = 0`,
attempts, now.Unix(), now.Unix(), status, d.ID)
return err
}
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 = ?`,
+SET attempts = ?, abandoned_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND abandoned_at = 0`,
attempts, now.Unix(), now.Unix(), status, failure, d.ID)
return err
}
_, err := s.db.ExecContext(ctx, `UPDATE mail_event_deliveries
-SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ?`,
+SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error = ? WHERE id = ? AND abandoned_at = 0`,
attempts, now.Add(mailPushBackoff[attempts-1]).Unix(), now.Unix(), status, failure, d.ID)
return err
}
@@ -366,6 +395,13 @@ SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error
// delivered or not: the payload is a copy of the log's addresses and lives
// no longer than they do.
func (s *Store) pruneMailEventDeliveries(ctx context.Context, now time.Time) error {
+ last := s.mailPushPruned.Load()
+ if last != 0 && now.Unix()-last < int64(mailPushPruneEvery/time.Second) {
+ return nil
+ }
+ 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
}
@@ -473,56 +509,100 @@ func (s *Server) StartMailEventPush(ctx context.Context) {
}
}
-// mailPushPass sends what is due, one delivery at a time, and trims the
-// queue. clock is read per delivery so a slow pass records honest times.
+// mailPushPass sends what is due and trims the queue. 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)
}
- due, err := s.store.dueMailEventDeliveries(ctx, start, mailPushBatch)
+ due, err := s.store.dueMailEventDeliveries(ctx, start, mailPushPerApp, mailPushBatch)
if err != nil {
return fmt.Errorf("reading the queue: %w", err)
}
- var key ed25519.PrivateKey
+ if len(due) == 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 {
- if clock().Sub(start) > mailPushPassBudget || ctx.Err() != nil {
- break
+ k := d.Account + "/" + d.App
+ if _, seen := byApp[k]; !seen {
+ order = append(order, k)
}
- if key == nil {
- if key, err = s.store.MailEventsSigningKey(); err != nil {
- return fmt.Errorf("reading the signing key: %w", err)
- }
+ byApp[k] = append(byApp[k], d)
+ }
+ sem := make(chan struct{}, mailPushApps)
+ var wg sync.WaitGroup
+ for _, k := range order {
+ rows := byApp[k]
+ wg.Add(1)
+ sem <- struct{}{}
+ go func() {
+ defer func() { <-sem; wg.Done() }()
+ s.deliverAppMailEvents(ctx, rows, key, clock)
+ }()
+ }
+ wg.Wait()
+ return nil
+}
+
+// deliverAppMailEvents sends one app's due rows in order. An attempt that got
+// no answer at all — a hang, a refused connection, a wake that never came —
+// ends the app's turn for this pass: the rest are no more likely to be
+// answered in the next ten seconds, and they keep their place for the next
+// pass rather than each burning a timeout now.
+func (s *Server) deliverAppMailEvents(ctx context.Context, rows []mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) {
+ for _, d := range rows {
+ if ctx.Err() != nil {
+ return
}
- status, failure := s.deliverMailEvent(ctx, d, key, clock)
- if err := s.store.recordMailEventAttempt(ctx, d, status, failure, clock()); err != nil {
+ status, failure, giveUp := s.deliverMailEvent(ctx, d, key, clock)
+ if err := s.store.recordMailEventAttempt(ctx, d, status, failure, giveUp, clock()); err != nil {
log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err)
}
+ if failure != "" && status == 0 {
+ return
+ }
}
- return nil
}
-// deliverMailEvent makes one attempt, and answers the HTTP status and what
-// went wrong, "" when it landed.
-func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, key ed25519.PrivateKey, clock func() time.Time) (int, string) {
+// 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
+ }
hook, err := s.store.MailEventHookFor(ctx, d.Account, d.App)
if err != nil {
- return 0, "could not read the hook: " + err.Error()
+ 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.
- return 0, "push turned off"
+ return 0, "push turned off", true
}
target, err := s.mailPushURL(d.Account, d.App, hook.Path)
if err != nil {
- return 0, err.Error()
+ return 0, err.Error(), false
}
rctx, cancel := context.WithTimeout(ctx, mailPushTimeout)
defer cancel()
req, err := http.NewRequestWithContext(rctx, http.MethodPost, target, bytes.NewReader(d.Payload))
if err != nil {
- return 0, "could not build the request: " + err.Error()
+ return 0, "could not build the request: " + err.Error(), false
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("User-Agent", "carlos-mail-events/1")
@@ -531,15 +611,15 @@ func (s *Server) deliverMailEvent(ctx context.Context, d mailEventDelivery, key
if err != nil {
// The URL is the app's own and not a secret, but the error can be
// long; the reason is what a member needs.
- return 0, "no answer: " + strings.TrimPrefix(err.Error(), "Post \""+target+"\": ")
+ return 0, "no answer: " + strings.TrimPrefix(err.Error(), "Post \""+target+"\": "), false
}
defer resp.Body.Close()
io.Copy(io.Discard, io.LimitReader(resp.Body, mailPushMaxBody))
if resp.StatusCode >= 200 && resp.StatusCode < 300 {
- return resp.StatusCode, ""
+ return resp.StatusCode, "", false
}
if resp.StatusCode >= 300 && resp.StatusCode < 400 {
- return resp.StatusCode, "answered with a redirect, which is not followed — POST to the path itself"
+ return resp.StatusCode, "answered with a redirect, which is not followed — POST to the path itself", false
}
- return resp.StatusCode, fmt.Sprintf("answered %d", resp.StatusCode)
+ return resp.StatusCode, fmt.Sprintf("answered %d", resp.StatusCode), false
}
diff --git a/internal/console/mailpush_test.go b/internal/console/mailpush_test.go
index f9b15c8d4..cae75d897 100644
--- a/internal/console/mailpush_test.go
+++ b/internal/console/mailpush_test.go
@@ -6,6 +6,7 @@ import (
"io"
"net/http"
"net/http/httptest"
+ "strconv"
"strings"
"sync"
"testing"
@@ -303,3 +304,139 @@ func TestThePushGoesOnlyToTheAppsOwnHost(t *testing.T) {
t.Error("a deployment with no apps domain should have nowhere to send, not a guess")
}
}
+
+// One app's handler hanging must not be every app's problem. A pass sends to
+// apps side by side, and an app that does not answer loses its turn after
+// one timeout rather than spending the whole pass.
+func TestAHungAppDoesNotHoldUpAnotherApp(t *testing.T) {
+ defer func(d time.Duration) { mailPushTimeout = d }(mailPushTimeout)
+ mailPushTimeout = 300 * time.Millisecond
+ var mu sync.Mutex
+ hungHits, otherHits := 0, 0
+ unblock := make(chan struct{})
+ defer close(unblock)
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {
+ mu.Lock()
+ if r.URL.Query().Get("app") == "jb-book" {
+ hungHits++
+ mu.Unlock()
+ select {
+ case <-unblock:
+ case <-time.After(5 * time.Second):
+ }
+ return
+ }
+ otherHits++
+ mu.Unlock()
+ w.WriteHeader(http.StatusNoContent)
+ })
+ ctx := context.Background()
+ now := time.Now()
+ for i := 0; i < 5; i++ {
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-hung-"+string(rune('a'+i)), now)), "sns-hung-"+string(rune('a'+i)))
+ }
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-site", now)), "sns-site")
+
+ started := time.Now()
+ if err := fx.srv.mailPushPass(ctx, time.Now); err != nil {
+ t.Fatal(err)
+ }
+ if took := time.Since(started); took > 2*time.Second {
+ t.Errorf("the pass took %v — a hung app spent it", took)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if otherHits != 1 {
+ t.Errorf("the other app was sent %d, want its one delivery", otherHits)
+ }
+ 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")
+ if st.Pending != 5 {
+ t.Errorf("the hung app's deliveries should all still be waiting: %+v", st)
+ }
+}
+
+// A backlog waits behind itself. An app with far more due than a pass sends
+// does not push another app's one delivery out of the pass.
+func TestABacklogIsItsOwnAppsToWaitBehind(t *testing.T) {
+ var mu sync.Mutex
+ sent := map[string]int{}
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {
+ mu.Lock()
+ sent[r.URL.Query().Get("app")]++
+ mu.Unlock()
+ w.WriteHeader(http.StatusNoContent)
+ })
+ ctx := context.Background()
+ now := time.Now()
+ for i := 0; i < 3*mailPushPerApp; i++ {
+ id := "m-busy-" + strconv.Itoa(i)
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id)
+ }
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-quiet", now)), "sns-quiet")
+ if err := fx.srv.mailPushPass(ctx, time.Now); err != nil {
+ t.Fatal(err)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if sent["jb-site"] != 1 || sent["jb-book"] != mailPushPerApp {
+ t.Errorf("sent %v, want jb-site 1 and jb-book %d", sent, mailPushPerApp)
+ }
+}
+
+// Nothing is sent for an app that is not live, and a trashed app's queue is
+// given up on rather than kept waiting for a host it may no longer own.
+func TestNothingIsPushedForATrashedApp(t *testing.T) {
+ hits := 0
+ fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { hits++ })
+ ctx := context.Background()
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1")
+ if err := fx.st.TrashAppRow(OpsAccountID, "jb-book", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.mailPushPass(ctx, time.Now)
+ if hits != 0 {
+ t.Error("a delivery went to a trashed app's host")
+ }
+ st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book")
+ if st.Abandoned != 1 || st.Pending != 0 {
+ t.Errorf("a trashed app's delivery should be given up on: %+v", st)
+ }
+ fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-bob", time.Now())), "sns-2")
+ if n := queued(t, fx, "jb-book"); n != 1 {
+ t.Errorf("a trashed app was queued something new")
+ }
+}
+
+// The push is the app's, like its log: a rename takes it along, and a purge
+// leaves nothing for the next app of that name to inherit.
+func TestThePushFollowsTheAppAndNotItsName(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-ann", time.Now())), "sns-1")
+
+ if err := fx.st.RenameAppIn(OpsAccountID, "jb-site", "jb-web"); err != nil {
+ t.Fatal(err)
+ }
+ if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook == nil {
+ t.Error("the renamed app lost its push")
+ }
+ if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-site"); hook != nil {
+ t.Error("the old name kept a push for whoever claims it next")
+ }
+
+ if err := fx.st.TrashAppRow(OpsAccountID, "jb-web", time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if err := fx.st.PurgeAppRow(OpsAccountID, "jb-web"); err != nil {
+ t.Fatal(err)
+ }
+ if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook != nil {
+ t.Error("a purged app's push survived for its name")
+ }
+ if n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 {
+ t.Errorf("a purged app's queue survived: %d rows", n)
+ }
+}
diff --git a/internal/console/store.go b/internal/console/store.go
index f6145048d..ec5ca86fe 100644
--- a/internal/console/store.go
+++ b/internal/console/store.go
@@ -45,6 +45,8 @@ type Store struct {
// this process, which is what makes the first event after a restart
// prune.
mailEventsPruned atomic.Int64
+ // mailPushPruned is the same stamp for the push queue (mailpush.go).
+ mailPushPruned atomic.Int64
}
// Open opens (creating if needed) the console DB at path.
@@ -275,6 +277,7 @@ func (s *Store) migrate() error {
UNIQUE (account, app, notice)
);`,
`CREATE INDEX IF NOT EXISTS mail_event_deliveries_due ON mail_event_deliveries (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 {
return err