amadan
email events push: a failing app ends its turn; queued bodies do not travel
From re-review of the review fixes. An app's turn ended only on an attempt with no HTTP answer, so a handler that answered 500 after nine seconds held one of the eight slots for five timeouts a pass while every other app waited on the pass. Any attempt that does not land now ends the app's turn. A row given up on because the app is not live, or the push is off, now gives up on all of that app's rows in one UPDATE, rather than one row per pass. A rename or a move carried queued deliveries whose bodies still name the old account or app, which an app checking its host correctly refuses — eight retries and an abandon, or, for an app that does not check, a stale id. They are dropped instead.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
3 files changed,
+62
−7
internal/console/mailevents.go+10 −0internal/console/mailpush.go+24 −7internal/console/mailpush_test.go+28 −0
CI passed — run details
diff --git a/internal/console/mailevents.go b/internal/console/mailevents.go| index 2895377fb..46874d64f 100644 |
| --- a/internal/console/mailevents.go |
| +++ b/internal/console/mailevents.go |
| @@ -121,6 +121,16 @@ func carryMailApp(ctx context.Context, tx execer, fromSqid, fromApp, toSqid, toA |
| return err |
| } |
| } |
| + if table == "mail_event_deliveries" { |
| + // Not carried: a queued body names the account and app it was |
| + // cut for, so after a move or a rename it names neither, and an |
| + // app checking its host is right to refuse it. The final DELETE |
| + // below drops them. |
| + if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE account = ? AND app = ?`, fromSqid, fromApp); err != nil { |
| + return err |
| + } |
| + continue |
| + } |
| if _, err := tx.ExecContext(ctx, `UPDATE `+table+` SET account = ?, app = ? WHERE account = ? AND app = ? AND `+ |
| mailAppTimeColumn(table)+` >= ?`, toSqid, toApp, fromSqid, fromApp, createdAt); err != nil { |
| return err |
diff --git a/internal/console/mailpush.go b/internal/console/mailpush.go| index b314899c6..2a8160153 100644 |
| --- a/internal/console/mailpush.go |
| +++ b/internal/console/mailpush.go |
| @@ -391,6 +391,14 @@ SET attempts = ?, next_at = ?, last_attempt_at = ?, last_status = ?, last_error |
| return err |
| } |
| +// abandonMailEventDeliveries gives up on everything still queued for one |
| +// app, in one statement. |
| +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 = ? |
| +WHERE account = ? AND app = ? AND delivered_at = 0 AND abandoned_at = 0`, now.Unix(), reason, acct, app) |
| + return err |
| +} |
| + |
| // pruneMailEventDeliveries drops every queued row past the retention window, |
| // delivered or not: the payload is a copy of the log's addresses and lives |
| // no longer than they do. |
| @@ -553,21 +561,30 @@ func (s *Server) mailPushPass(ctx context.Context, clock func() time.Time) error |
| 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. |
| +// deliverAppMailEvents sends one app's due rows in order. Any attempt that |
| +// 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. |
| +// |
| +// 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) { |
| for _, d := range rows { |
| if ctx.Err() != nil { |
| return |
| } |
| status, failure, giveUp := s.deliverMailEvent(ctx, d, key, clock) |
| - if err := s.store.recordMailEventAttempt(ctx, d, status, failure, giveUp, clock()); err != nil { |
| + if giveUp { |
| + if err := s.store.abandonMailEventDeliveries(ctx, d.Account, d.App, failure, 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 { |
| log.Printf("console: mail push: recording delivery %d for %s/%s: %v", d.ID, d.Account, d.App, err) |
| } |
| - if failure != "" && status == 0 { |
| + if failure != "" { |
| return |
| } |
| } |
diff --git a/internal/console/mailpush_test.go b/internal/console/mailpush_test.go| index cae75d897..00133fd84 100644 |
| --- a/internal/console/mailpush_test.go |
| +++ b/internal/console/mailpush_test.go |
| @@ -426,6 +426,9 @@ func TestThePushFollowsTheAppAndNotItsName(t *testing.T) { |
| 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 n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 { |
| + t.Errorf("a rename carried %d queued deliveries whose bodies name the old app", n) |
| + } |
| if err := fx.st.TrashAppRow(OpsAccountID, "jb-web", time.Now()); err != nil { |
| t.Fatal(err) |
| @@ -440,3 +443,28 @@ func TestThePushFollowsTheAppAndNotItsName(t *testing.T) { |
| t.Errorf("a purged app's queue survived: %d rows", n) |
| } |
| } |
| + |
| +// An app that answers, but not with a 2xx, ends its turn at the first |
| +// failure: offering it the next row only holds a slot others wait on. |
| +func TestAFailingAppEndsItsTurnAtTheFirstFailure(t *testing.T) { |
| + var mu sync.Mutex |
| + hits := 0 |
| + fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| + mu.Lock() |
| + hits++ |
| + mu.Unlock() |
| + w.WriteHeader(http.StatusInternalServerError) |
| + }) |
| + ctx := context.Background() |
| + now := time.Now() |
| + for i := 0; i < mailPushPerApp; i++ { |
| + id := "m-" + strconv.Itoa(i) |
| + fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id) |
| + } |
| + fx.srv.mailPushPass(ctx, time.Now) |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if hits != 1 { |
| + t.Errorf("a failing app was offered %d rows in one pass, want 1", hits) |
| + } |
| +} |