| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "io" |
| 7 | "net/http" |
| 8 | "net/http/httptest" |
| 9 | "strconv" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/mail" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // The scheme has one executable definition, and both ends are held to it: a |
| 20 | // body signed by the console verifies with the key the app is given, and |
| 21 | // nothing about it can be changed on the way. |
| 22 | func TestAMailEventSignatureVerifiesAndNothingElseDoes(t *testing.T) { |
| 23 | fx := newConsoleFixture(t) |
| 24 | key, err := fx.st.MailEventsSigningKey() |
| 25 | if err != nil { |
| 26 | t.Fatal(err) |
| 27 | } |
| 28 | again, _ := fx.st.MailEventsSigningKey() |
| 29 | if mailEventsPublicKey(again) != mailEventsPublicKey(key) { |
| 30 | t.Fatal("the signing key changed between reads — every app's delivered key would stop verifying") |
| 31 | } |
| 32 | pub := mailEventsPublicKey(key) |
| 33 | now := time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC) |
| 34 | body := []byte(`{"id":"n1","account":"beb","app":"jb-book","type":"Bounce"}`) |
| 35 | header := signMailEvent(key, now, body) |
| 36 | |
| 37 | if err := VerifyMailEvent(pub, header, body, now.Add(time.Minute), 5*time.Minute); err != nil { |
| 38 | t.Fatalf("a genuine delivery did not verify: %v", err) |
| 39 | } |
| 40 | other, _ := newConsoleFixture(t).st.MailEventsSigningKey() |
| 41 | for name, c := range map[string]struct { |
| 42 | key, header string |
| 43 | body []byte |
| 44 | at time.Time |
| 45 | }{ |
| 46 | "a changed body": {pub, header, []byte(strings.Replace(string(body), "jb-book", "jb-site", 1)), now}, |
| 47 | "another deployment's key": {mailEventsPublicKey(other), header, body, now}, |
| 48 | "a replay long after": {pub, header, body, now.Add(10 * time.Minute)}, |
| 49 | "a timestamp from the future": {pub, header, body, now.Add(-10 * time.Minute)}, |
| 50 | "a moved timestamp": {pub, strings.Replace(header, "t=", "t=1", 1), body, now}, |
| 51 | "no signature at all": {pub, "t=1726390800", body, now}, |
| 52 | } { |
| 53 | if err := VerifyMailEvent(c.key, c.header, c.body, c.at, 5*time.Minute); err == nil { |
| 54 | t.Errorf("%s verified", name) |
| 55 | } |
| 56 | } |
| 57 | } |
| 58 | |
| 59 | // pushFixture is one app on one domain, asking to be told, with the push |
| 60 | // pointed at a test server instead of the app's platform host. |
| 61 | func pushFixture(t *testing.T, handler http.HandlerFunc) (*consoleFixture, *httptest.Server) { |
| 62 | t.Helper() |
| 63 | fx := newConsoleFixture(t) |
| 64 | ctx := context.Background() |
| 65 | now := time.Now() |
| 66 | for _, app := range []string{"jb-book", "jb-site"} { |
| 67 | if _, err := fx.st.CreateAppIn(OpsAccountID, app, "bill@tito.io", now); err != nil { |
| 68 | t.Fatal(err) |
| 69 | } |
| 70 | } |
| 71 | rec := &mail.DomainRecord{ |
| 72 | Domain: "lgc.li", Kind: mail.KindCustom, Apps: []string{"jb-book", "jb-site"}, |
| 73 | Regions: map[string]*mail.RegionState{"eu-west-1": { |
| 74 | IdentityARN: "arn:aws:ses:eu-west-1:1:identity/lgc.li", MailFrom: "mail-eu-west-1.lgc.li", Status: mail.StatusVerified, |
| 75 | }}, |
| 76 | CreatedAt: now, |
| 77 | } |
| 78 | if _, err := mail.PutDomain(ctx, fx.rel, release.OpsAccount, rec, ""); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | for _, app := range []string{"jb-book", "jb-site"} { |
| 82 | if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, app, "/mail/events", "bill@tito.io", now); err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | } |
| 86 | srv := httptest.NewServer(handler) |
| 87 | t.Cleanup(srv.Close) |
| 88 | fx.srv.mailPushTarget = func(acct, app, path string) string { return srv.URL + path + "?app=" + app } |
| 89 | return fx, srv |
| 90 | } |
| 91 | |
| 92 | func sentBy(app string, ev mail.Event) mail.Event { |
| 93 | ev.Caller = mail.UserName(release.OpsAccount, app) |
| 94 | return ev |
| 95 | } |
| 96 | |
| 97 | func queued(t *testing.T, fx *consoleFixture, app string) int { |
| 98 | t.Helper() |
| 99 | var n int |
| 100 | if err := fx.st.db.QueryRow(`SELECT count(*) FROM mail_event_deliveries WHERE app = ?`, app).Scan(&n); err != nil { |
| 101 | t.Fatal(err) |
| 102 | } |
| 103 | return n |
| 104 | } |
| 105 | |
| 106 | // Only an event SES attributed to the app itself is ever queued for it, and |
| 107 | // only a type worth telling. A notification SNS delivers twice is queued |
| 108 | // once. |
| 109 | func TestOnlyTheSendersOwnOutcomesAreQueued(t *testing.T) { |
| 110 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| 111 | ctx := context.Background() |
| 112 | now := time.Now() |
| 113 | |
| 114 | bounce := sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)) |
| 115 | fx.srv.recordAndEnforceMail(ctx, bounce, "sns-1") |
| 116 | fx.srv.recordAndEnforceMail(ctx, bounce, "sns-1") |
| 117 | if n := queued(t, fx, "jb-book"); n != 1 { |
| 118 | t.Errorf("jb-book queued %d, want 1 — a redelivered notification is one delivery", n) |
| 119 | } |
| 120 | if n := queued(t, fx, "jb-site"); n != 0 { |
| 121 | t.Errorf("jb-site was queued another app's recipient: %d", n) |
| 122 | } |
| 123 | |
| 124 | // Attributed only by the shared domain: kept in both logs, pushed to |
| 125 | // neither. |
| 126 | fx.srv.recordAndEnforceMail(ctx, bounceTo("bob@example.com", "m-bob", now), "sns-2") |
| 127 | if n := queued(t, fx, "jb-book") + queued(t, fx, "jb-site"); n != 1 { |
| 128 | t.Errorf("a domain-attributed event was queued (%d rows in all)", n-1) |
| 129 | } |
| 130 | |
| 131 | // A Send is not news to the app that sent it. |
| 132 | send := sentBy("jb-book", mail.Event{Type: mail.EventSend, Domain: "lgc.li", MessageID: "m-cat", At: now, Destination: []string{"cat@example.com"}}) |
| 133 | fx.srv.recordAndEnforceMail(ctx, send, "sns-3") |
| 134 | if n := queued(t, fx, "jb-book"); n != 1 { |
| 135 | t.Errorf("a Send was queued") |
| 136 | } |
| 137 | |
| 138 | // And an app that has not asked is not queued anything. |
| 139 | if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", now); err != nil { |
| 140 | t.Fatal(err) |
| 141 | } |
| 142 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("dan@example.com", "m-dan", now)), "sns-4") |
| 143 | if n := queued(t, fx, "jb-book"); n != 1 { |
| 144 | t.Errorf("an app with no hook was queued") |
| 145 | } |
| 146 | } |
| 147 | |
| 148 | // A delivery lands at the app's path, signed with the key the app was |
| 149 | // given, carrying who it was about and what the receiving server said. |
| 150 | func TestADeliveryReachesTheAppSigned(t *testing.T) { |
| 151 | var mu sync.Mutex |
| 152 | var gotPath, gotSig string |
| 153 | var gotBody []byte |
| 154 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 155 | mu.Lock() |
| 156 | defer mu.Unlock() |
| 157 | gotPath = r.URL.Path + "?" + r.URL.RawQuery |
| 158 | gotSig = r.Header.Get("Carlos-Signature") |
| 159 | gotBody, _ = io.ReadAll(r.Body) |
| 160 | w.WriteHeader(http.StatusNoContent) |
| 161 | }) |
| 162 | ctx := context.Background() |
| 163 | now := time.Now() |
| 164 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)), "sns-1") |
| 165 | |
| 166 | if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| 167 | t.Fatal(err) |
| 168 | } |
| 169 | mu.Lock() |
| 170 | defer mu.Unlock() |
| 171 | if gotPath != "/mail/events?app=jb-book" { |
| 172 | t.Fatalf("delivered to %q", gotPath) |
| 173 | } |
| 174 | key, _ := fx.st.MailEventsSigningKey() |
| 175 | if err := VerifyMailEvent(mailEventsPublicKey(key), gotSig, gotBody, time.Now(), 5*time.Minute); err != nil { |
| 176 | t.Fatalf("the app could not verify what it was sent: %v", err) |
| 177 | } |
| 178 | var p MailEventPayload |
| 179 | if err := json.Unmarshal(gotBody, &p); err != nil { |
| 180 | t.Fatal(err) |
| 181 | } |
| 182 | if p.ID != "sns-1" || p.Account != release.OpsAccount || p.App != "jb-book" || p.Type != mail.EventBounce || |
| 183 | p.MessageID != "m-ann" || p.Detail != "Permanent/NoEmail" { |
| 184 | t.Errorf("payload = %+v", p) |
| 185 | } |
| 186 | if len(p.Recipients) != 1 || p.Recipients[0].Address != "ann@example.com" || p.Recipients[0].Status != "5.1.1" { |
| 187 | t.Errorf("recipients = %+v", p.Recipients) |
| 188 | } |
| 189 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 190 | if st.Delivered != 1 || st.Pending != 0 || st.LastDelivered.IsZero() { |
| 191 | t.Errorf("status after delivery = %+v", st) |
| 192 | } |
| 193 | |
| 194 | // Delivered is delivered: the next pass sends nothing. |
| 195 | gotPath = "" |
| 196 | mu.Unlock() |
| 197 | fx.srv.mailPushPass(ctx, time.Now) |
| 198 | mu.Lock() |
| 199 | if gotPath != "" { |
| 200 | t.Error("a delivered event was sent again") |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // A delivery that does not land is tried again on a widening schedule, and |
| 205 | // given up on when the schedule is spent — never retried for ever, and never |
| 206 | // kept past the log it came from. |
| 207 | func TestAFailedDeliveryBacksOffAndIsGivenUpOn(t *testing.T) { |
| 208 | var mu sync.Mutex |
| 209 | hits := 0 |
| 210 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 211 | mu.Lock() |
| 212 | hits++ |
| 213 | mu.Unlock() |
| 214 | w.WriteHeader(http.StatusServiceUnavailable) |
| 215 | }) |
| 216 | ctx := context.Background() |
| 217 | now := time.Date(2026, 9, 15, 9, 0, 0, 0, time.UTC) |
| 218 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", now)), "sns-1") |
| 219 | // Queued by the real clock; start the fake one there. |
| 220 | var created int64 |
| 221 | fx.st.db.QueryRow(`SELECT created_at FROM mail_event_deliveries`).Scan(&created) |
| 222 | clock := time.Unix(created, 0) |
| 223 | tick := func() time.Time { return clock } |
| 224 | |
| 225 | fx.srv.mailPushPass(ctx, tick) |
| 226 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 227 | if st.Pending != 1 || st.LastStatus != 503 || !strings.Contains(st.LastError, "503") || !st.LastFailedAt.Equal(clock) { |
| 228 | t.Fatalf("after one failure: %+v — the failure should be dated when it happened, %s", st, clock.UTC()) |
| 229 | } |
| 230 | // Not due again until the first backoff has passed. |
| 231 | clock = clock.Add(30 * time.Second) |
| 232 | fx.srv.mailPushPass(ctx, tick) |
| 233 | if hits != 1 { |
| 234 | t.Fatalf("retried before its backoff: %d attempts", hits) |
| 235 | } |
| 236 | for _, wait := range mailPushBackoff { |
| 237 | clock = clock.Add(wait) |
| 238 | fx.srv.mailPushPass(ctx, tick) |
| 239 | } |
| 240 | if hits != len(mailPushBackoff)+1 { |
| 241 | t.Errorf("attempts = %d, want %d", hits, len(mailPushBackoff)+1) |
| 242 | } |
| 243 | st, _ = fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 244 | if st.Abandoned != 1 || st.Pending != 0 { |
| 245 | t.Errorf("after the schedule is spent: %+v", st) |
| 246 | } |
| 247 | clock = clock.Add(24 * time.Hour) |
| 248 | fx.srv.mailPushPass(ctx, tick) |
| 249 | if hits != len(mailPushBackoff)+1 { |
| 250 | t.Error("an abandoned delivery was tried again") |
| 251 | } |
| 252 | |
| 253 | // And the row is gone once the retention window has passed. |
| 254 | clock = time.Unix(created, 0).Add(mailEventsRetention + time.Hour) |
| 255 | fx.srv.mailPushPass(ctx, tick) |
| 256 | if n := queued(t, fx, "jb-book"); n != 0 { |
| 257 | t.Errorf("a queued payload outlived the retention window") |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | // A redirect is where a delivery would become a POST of somebody's address |
| 262 | // at a host the far end chooses. It is not followed. |
| 263 | func TestADeliveryDoesNotFollowARedirect(t *testing.T) { |
| 264 | elsewhere := notReached(t) |
| 265 | defer elsewhere.Close() |
| 266 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 267 | http.Redirect(w, r, elsewhere.URL+"/steal", http.StatusTemporaryRedirect) |
| 268 | }) |
| 269 | ctx := context.Background() |
| 270 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| 271 | fx.srv.mailPushPass(ctx, time.Now) |
| 272 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 273 | if st.Delivered != 0 || !strings.Contains(st.LastError, "redirect") { |
| 274 | t.Errorf("a redirect counted as delivered, or went unexplained: %+v", st) |
| 275 | } |
| 276 | } |
| 277 | |
| 278 | // Turning the push off stops what is already queued, not only what comes |
| 279 | // next. |
| 280 | func TestTurningThePushOffStopsTheQueue(t *testing.T) { |
| 281 | hits := 0 |
| 282 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { hits++ }) |
| 283 | ctx := context.Background() |
| 284 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| 285 | if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", time.Now()); err != nil { |
| 286 | t.Fatal(err) |
| 287 | } |
| 288 | fx.srv.mailPushPass(ctx, time.Now) |
| 289 | if hits != 0 { |
| 290 | t.Error("a delivery went out after the push was turned off") |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | // The production target is the app's own platform host and nothing else. |
| 295 | func TestThePushGoesOnlyToTheAppsOwnHost(t *testing.T) { |
| 296 | fx := newConsoleFixture(t) |
| 297 | fx.srv.cfg.AppsDomain = "OnCarlos.com" |
| 298 | got, err := fx.srv.mailPushURL("beb", "jb-book", "/mail/events") |
| 299 | if err != nil || got != "https://jb-book.beb.oncarlos.com/mail/events" { |
| 300 | t.Errorf("target = %q, %v", got, err) |
| 301 | } |
| 302 | fx.srv.cfg.AppsDomain = "" |
| 303 | if _, err := fx.srv.mailPushURL("beb", "jb-book", "/mail/events"); err == nil { |
| 304 | t.Error("a deployment with no apps domain should have nowhere to send, not a guess") |
| 305 | } |
| 306 | } |
| 307 | |
| 308 | // One app's handler hanging must not be every app's problem. A pass sends to |
| 309 | // apps side by side, and an app that does not answer loses its turn after |
| 310 | // one timeout rather than spending the whole pass. |
| 311 | func TestAHungAppDoesNotHoldUpAnotherApp(t *testing.T) { |
| 312 | defer func(d time.Duration) { mailPushTimeout = d }(mailPushTimeout) |
| 313 | mailPushTimeout = 300 * time.Millisecond |
| 314 | var mu sync.Mutex |
| 315 | hungHits, otherHits := 0, 0 |
| 316 | unblock := make(chan struct{}) |
| 317 | defer close(unblock) |
| 318 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 319 | mu.Lock() |
| 320 | if r.URL.Query().Get("app") == "jb-book" { |
| 321 | hungHits++ |
| 322 | mu.Unlock() |
| 323 | select { |
| 324 | case <-unblock: |
| 325 | case <-time.After(5 * time.Second): |
| 326 | } |
| 327 | return |
| 328 | } |
| 329 | otherHits++ |
| 330 | mu.Unlock() |
| 331 | w.WriteHeader(http.StatusNoContent) |
| 332 | }) |
| 333 | ctx := context.Background() |
| 334 | now := time.Now() |
| 335 | for i := 0; i < 5; i++ { |
| 336 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-hung-"+string(rune('a'+i)), now)), "sns-hung-"+string(rune('a'+i))) |
| 337 | } |
| 338 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-site", now)), "sns-site") |
| 339 | |
| 340 | started := time.Now() |
| 341 | if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| 342 | t.Fatal(err) |
| 343 | } |
| 344 | if took := time.Since(started); took > 2*time.Second { |
| 345 | t.Errorf("the pass took %v — a hung app spent it", took) |
| 346 | } |
| 347 | mu.Lock() |
| 348 | defer mu.Unlock() |
| 349 | if otherHits != 1 { |
| 350 | t.Errorf("the other app was sent %d, want its one delivery", otherHits) |
| 351 | } |
| 352 | if hungHits != 1 { |
| 353 | t.Errorf("the hung app was tried %d times in one pass, want 1 — its turn ends at the first silence", hungHits) |
| 354 | } |
| 355 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 356 | if st.Pending != 5 { |
| 357 | t.Errorf("the hung app's deliveries should all still be waiting: %+v", st) |
| 358 | } |
| 359 | } |
| 360 | |
| 361 | // A backlog waits behind itself. An app with far more due than a pass sends |
| 362 | // does not push another app's one delivery out of the pass. |
| 363 | func TestABacklogIsItsOwnAppsToWaitBehind(t *testing.T) { |
| 364 | var mu sync.Mutex |
| 365 | sent := map[string]int{} |
| 366 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 367 | mu.Lock() |
| 368 | sent[r.URL.Query().Get("app")]++ |
| 369 | mu.Unlock() |
| 370 | w.WriteHeader(http.StatusNoContent) |
| 371 | }) |
| 372 | ctx := context.Background() |
| 373 | now := time.Now() |
| 374 | for i := 0; i < 3*mailPushPerApp; i++ { |
| 375 | id := "m-busy-" + strconv.Itoa(i) |
| 376 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id) |
| 377 | } |
| 378 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("bob@example.com", "m-quiet", now)), "sns-quiet") |
| 379 | if err := fx.srv.mailPushPass(ctx, time.Now); err != nil { |
| 380 | t.Fatal(err) |
| 381 | } |
| 382 | mu.Lock() |
| 383 | defer mu.Unlock() |
| 384 | if sent["jb-site"] != 1 || sent["jb-book"] != mailPushPerApp { |
| 385 | t.Errorf("sent %v, want jb-site 1 and jb-book %d", sent, mailPushPerApp) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | // Nothing is sent for an app that is not live, and a trashed app's queue is |
| 390 | // given up on rather than kept waiting for a host it may no longer own. |
| 391 | func TestNothingIsPushedForATrashedApp(t *testing.T) { |
| 392 | hits := 0 |
| 393 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { hits++ }) |
| 394 | ctx := context.Background() |
| 395 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| 396 | if err := fx.st.TrashAppRow(OpsAccountID, "jb-book", time.Now()); err != nil { |
| 397 | t.Fatal(err) |
| 398 | } |
| 399 | fx.srv.mailPushPass(ctx, time.Now) |
| 400 | if hits != 0 { |
| 401 | t.Error("a delivery went to a trashed app's host") |
| 402 | } |
| 403 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}) |
| 404 | if st.Abandoned != 1 || st.Pending != 0 { |
| 405 | t.Errorf("a trashed app's delivery should be given up on: %+v", st) |
| 406 | } |
| 407 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-bob", time.Now())), "sns-2") |
| 408 | if n := queued(t, fx, "jb-book"); n != 1 { |
| 409 | t.Errorf("a trashed app was queued something new") |
| 410 | } |
| 411 | } |
| 412 | |
| 413 | // The push is the app's, like its log: a rename takes it along, and a purge |
| 414 | // leaves nothing for the next app of that name to inherit. |
| 415 | func TestThePushFollowsTheAppAndNotItsName(t *testing.T) { |
| 416 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| 417 | ctx := context.Background() |
| 418 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("ann@example.com", "m-ann", time.Now())), "sns-1") |
| 419 | |
| 420 | if err := fx.st.RenameAppIn(OpsAccountID, "jb-site", "jb-web"); err != nil { |
| 421 | t.Fatal(err) |
| 422 | } |
| 423 | if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook == nil { |
| 424 | t.Error("the renamed app lost its push") |
| 425 | } |
| 426 | if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-site"); hook != nil { |
| 427 | t.Error("the old name kept a push for whoever claims it next") |
| 428 | } |
| 429 | if n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 { |
| 430 | t.Errorf("a rename carried %d queued deliveries whose bodies name the old app", n) |
| 431 | } |
| 432 | |
| 433 | if err := fx.st.TrashAppRow(OpsAccountID, "jb-web", time.Now()); err != nil { |
| 434 | t.Fatal(err) |
| 435 | } |
| 436 | if err := fx.st.PurgeAppRow(OpsAccountID, "jb-web"); err != nil { |
| 437 | t.Fatal(err) |
| 438 | } |
| 439 | if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-web"); hook != nil { |
| 440 | t.Error("a purged app's push survived for its name") |
| 441 | } |
| 442 | if n := queued(t, fx, "jb-web") + queued(t, fx, "jb-site"); n != 0 { |
| 443 | t.Errorf("a purged app's queue survived: %d rows", n) |
| 444 | } |
| 445 | } |
| 446 | |
| 447 | // An app that answers, but not with a 2xx, ends its turn at the first |
| 448 | // failure: offering it the next row only holds a slot others wait on. |
| 449 | func TestAFailingAppEndsItsTurnAtTheFirstFailure(t *testing.T) { |
| 450 | var mu sync.Mutex |
| 451 | hits := 0 |
| 452 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 453 | mu.Lock() |
| 454 | hits++ |
| 455 | mu.Unlock() |
| 456 | w.WriteHeader(http.StatusInternalServerError) |
| 457 | }) |
| 458 | ctx := context.Background() |
| 459 | now := time.Now() |
| 460 | for i := 0; i < mailPushPerApp; i++ { |
| 461 | id := "m-" + strconv.Itoa(i) |
| 462 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, now)), "sns-"+id) |
| 463 | } |
| 464 | fx.srv.mailPushPass(ctx, time.Now) |
| 465 | mu.Lock() |
| 466 | defer mu.Unlock() |
| 467 | if hits != 1 { |
| 468 | t.Errorf("a failing app was offered %d rows in one pass, want 1", hits) |
| 469 | } |
| 470 | } |
| 471 | |
| 472 | // An app that accepts nothing for a day is suspended: what is waiting is |
| 473 | // given up on, nothing new is queued, and turning the push on again lifts |
| 474 | // it. Without this, a failing app's backlog grows for the thirty days the |
| 475 | // queue is kept. |
| 476 | func TestADayOfFailuresSuspendsThePush(t *testing.T) { |
| 477 | hits := 0 |
| 478 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { |
| 479 | hits++ |
| 480 | w.WriteHeader(http.StatusNotFound) |
| 481 | }) |
| 482 | ctx := context.Background() |
| 483 | send := func(id string, at time.Time) { |
| 484 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", id, at)), "sns-"+id) |
| 485 | } |
| 486 | clock := time.Now() |
| 487 | tick := func() time.Time { return clock } |
| 488 | for i := 0; i < 20; i++ { |
| 489 | send("m-"+strconv.Itoa(i), clock) |
| 490 | } |
| 491 | fx.srv.mailPushPass(ctx, tick) |
| 492 | hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| 493 | if hook.FailingSince.IsZero() || hook.Suspended() { |
| 494 | t.Fatalf("after the first failure: %+v", hook) |
| 495 | } |
| 496 | |
| 497 | // Most of a day of failing is not a day. |
| 498 | clock = clock.Add(23 * time.Hour) |
| 499 | fx.srv.mailPushPass(ctx, tick) |
| 500 | if hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() { |
| 501 | t.Fatal("suspended before a day had passed") |
| 502 | } |
| 503 | |
| 504 | clock = clock.Add(2 * time.Hour) |
| 505 | fx.srv.mailPushPass(ctx, tick) |
| 506 | hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| 507 | if !hook.Suspended() || !strings.Contains(hook.SuspendedWhy, "404") { |
| 508 | t.Fatalf("a day of failures should suspend, saying what the app answered: %+v", hook) |
| 509 | } |
| 510 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", hook.SetAt) |
| 511 | if st.Pending != 0 || st.Abandoned != 20 { |
| 512 | t.Errorf("a suspension should give up on the whole queue: %+v", st) |
| 513 | } |
| 514 | |
| 515 | // Suspended: nothing new is queued, and a pass sends nothing. |
| 516 | send("m-late", clock) |
| 517 | if n := queued(t, fx, "jb-book"); n != 20 { |
| 518 | t.Errorf("a suspended push queued more (%d rows)", n) |
| 519 | } |
| 520 | before := hits |
| 521 | clock = clock.Add(time.Hour) |
| 522 | fx.srv.mailPushPass(ctx, tick) |
| 523 | if hits != before { |
| 524 | t.Error("a suspended push was sent to") |
| 525 | } |
| 526 | |
| 527 | // Turning it on again lifts the suspension and starts clean. |
| 528 | if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", clock); err != nil { |
| 529 | t.Fatal(err) |
| 530 | } |
| 531 | hook, _ = fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book") |
| 532 | if hook.Suspended() || !hook.FailingSince.IsZero() { |
| 533 | t.Errorf("turning it on again should lift the suspension: %+v", hook) |
| 534 | } |
| 535 | send("m-after", clock) |
| 536 | if n := queued(t, fx, "jb-book"); n != 21 { |
| 537 | t.Errorf("after turning it on again, events should queue: %d rows", n) |
| 538 | } |
| 539 | } |
| 540 | |
| 541 | // A run of failures belongs to a backlog, not to the app's history: an app |
| 542 | // whose queue emptied — here by its only delivery running out of retries, |
| 543 | // with nothing ever landing — is not suspended by its next miss days later. |
| 544 | func TestAnOldFailureWithNothingWaitingDoesNotSuspend(t *testing.T) { |
| 545 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) }) |
| 546 | ctx := context.Background() |
| 547 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| 548 | var created int64 |
| 549 | fx.st.db.QueryRow(`SELECT created_at FROM mail_event_deliveries`).Scan(&created) |
| 550 | clock := time.Unix(created, 0) |
| 551 | tick := func() time.Time { return clock } |
| 552 | fx.srv.mailPushPass(ctx, tick) |
| 553 | for _, wait := range mailPushBackoff { |
| 554 | clock = clock.Add(wait) |
| 555 | fx.srv.mailPushPass(ctx, tick) |
| 556 | } |
| 557 | if st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", time.Time{}); st.Pending != 0 || st.Abandoned != 1 { |
| 558 | t.Fatalf("the one delivery should have run out of retries by now: %+v", st) |
| 559 | } |
| 560 | clock = clock.Add(3 * 24 * time.Hour) |
| 561 | fx.srv.mailPushPass(ctx, tick) // nothing waiting: the run ends here |
| 562 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("bob@example.com", "m-2", clock)), "sns-2") |
| 563 | fx.st.db.Exec(`UPDATE mail_event_deliveries SET next_at = ?, created_at = ? WHERE notice = 'sns-2'`, clock.Unix(), clock.Unix()) |
| 564 | fx.srv.mailPushPass(ctx, tick) |
| 565 | if hook, _ := fx.st.MailEventHookFor(ctx, release.OpsAccount, "jb-book"); hook.Suspended() { |
| 566 | t.Errorf("one miss after days of quiet suspended the push: %+v", hook) |
| 567 | } |
| 568 | } |
| 569 | |
| 570 | // A pass seeks each app's oldest due rows through an index rather than |
| 571 | // sorting everything due: its cost follows the number of apps asking, not |
| 572 | // the size of anyone's backlog. |
| 573 | func TestAPassDoesNotSortAnAppsBacklog(t *testing.T) { |
| 574 | fx := newConsoleFixture(t) |
| 575 | rows, err := fx.st.db.Query(`EXPLAIN QUERY PLAN `+mailPushDueSQL, "beb", "jb-book", time.Now().Unix(), mailPushPerApp) |
| 576 | if err != nil { |
| 577 | t.Fatal(err) |
| 578 | } |
| 579 | defer rows.Close() |
| 580 | var plan []string |
| 581 | for rows.Next() { |
| 582 | var id, parent, notused int |
| 583 | var detail string |
| 584 | rows.Scan(&id, &parent, ¬used, &detail) |
| 585 | plan = append(plan, detail) |
| 586 | } |
| 587 | joined := strings.Join(plan, " | ") |
| 588 | if !strings.Contains(joined, "mail_event_deliveries_app_due") || strings.Contains(joined, "TEMP B-TREE") { |
| 589 | t.Errorf("the due read should seek the per-app index without sorting: %s", joined) |
| 590 | } |
| 591 | |
| 592 | // And a large backlog costs a pass one app's worth of rows. |
| 593 | sent := 0 |
| 594 | fx2, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { sent++; w.WriteHeader(http.StatusNoContent) }) |
| 595 | ctx := context.Background() |
| 596 | now := time.Now() |
| 597 | for i := 0; i < 2000; i++ { |
| 598 | if err := fx2.st.EnqueueMailEventDelivery(ctx, release.OpsAccount, "jb-book", "bulk-"+strconv.Itoa(i), []byte(`{}`), now); err != nil { |
| 599 | t.Fatal(err) |
| 600 | } |
| 601 | } |
| 602 | fx2.srv.mailPushPass(ctx, time.Now) |
| 603 | if sent != mailPushPerApp { |
| 604 | t.Errorf("one pass sent %d, want %d", sent, mailPushPerApp) |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | // Turning the push off and on again, or lifting a suspension, starts the |
| 609 | // status afresh: an old failure is not reported as the last attempt. |
| 610 | func TestTheStatusForgetsFailuresFromBeforeThePushWasTurnedOn(t *testing.T) { |
| 611 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusServiceUnavailable) }) |
| 612 | ctx := context.Background() |
| 613 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-book", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| 614 | fx.srv.mailPushPass(ctx, time.Now) |
| 615 | if _, err := fx.st.DeleteMailEventHook(ctx, release.OpsAccount, "jb-book", time.Now()); err != nil { |
| 616 | t.Fatal(err) |
| 617 | } |
| 618 | later := time.Now().Add(time.Minute) |
| 619 | if err := fx.st.SetMailEventHook(ctx, release.OpsAccount, "jb-book", "/mail/events", "bill@tito.io", later); err != nil { |
| 620 | t.Fatal(err) |
| 621 | } |
| 622 | st, _ := fx.st.MailEventPushStatusFor(ctx, release.OpsAccount, "jb-book", later) |
| 623 | if st.LastError != "" || st.LastStatus != 0 { |
| 624 | t.Errorf("a failure from before the push was turned on again was reported: %+v", st) |
| 625 | } |
| 626 | var why, lastErr string |
| 627 | fx.st.db.QueryRow(`SELECT abandoned_why, last_error FROM mail_event_deliveries`).Scan(&why, &lastErr) |
| 628 | if why != "push turned off" || !strings.Contains(lastErr, "503") { |
| 629 | t.Errorf("giving up should record why without overwriting what the app answered: why=%q last_error=%q", why, lastErr) |
| 630 | } |
| 631 | } |
| 632 | |
| 633 | // A move to another account drops the queue, like a rename: the bodies name |
| 634 | // the old account. |
| 635 | func TestAMoveDropsTheQueue(t *testing.T) { |
| 636 | fx, _ := pushFixture(t, func(w http.ResponseWriter, r *http.Request) {}) |
| 637 | ctx := context.Background() |
| 638 | fx.srv.recordAndEnforceMail(ctx, sentBy("jb-site", bounceTo("ann@example.com", "m-1", time.Now())), "sns-1") |
| 639 | if err := fx.st.CreateAccount(99, "dest", "Destination", "bill@tito.io", time.Now()); err != nil { |
| 640 | t.Fatal(err) |
| 641 | } |
| 642 | if err := fx.st.MoveApp("jb-site", OpsAccountID, 99); err != nil { |
| 643 | t.Fatal(err) |
| 644 | } |
| 645 | var n int |
| 646 | fx.st.db.QueryRow(`SELECT count(*) FROM mail_event_deliveries WHERE app = 'jb-site'`).Scan(&n) |
| 647 | if n != 0 { |
| 648 | t.Errorf("a move carried %d queued deliveries naming the old account", n) |
| 649 | } |
| 650 | if hook, _ := fx.st.MailEventHookFor(ctx, "dest", "jb-site"); hook == nil { |
| 651 | t.Error("the push itself should move with the app") |
| 652 | } |
| 653 | } |
| 654 | |
| 655 | // A prune that fails is not stamped, and does not cost the pass. |
| 656 | func TestAFailedQueuePruneIsTriedAgain(t *testing.T) { |
| 657 | fx := newConsoleFixture(t) |
| 658 | ctx, cancel := context.WithCancel(context.Background()) |
| 659 | cancel() |
| 660 | if err := fx.st.pruneMailEventDeliveries(ctx, time.Now()); err == nil { |
| 661 | t.Fatal("a prune with a cancelled context should fail") |
| 662 | } |
| 663 | if got := fx.st.mailPushPruned.Load(); got != 0 { |
| 664 | t.Errorf("the stamp was left at %d after a failed prune", got) |
| 665 | } |
| 666 | } |
| 667 | |
amadan