From b83df366369a0ef0329d6d64b0abb3dbfc7c2686 Mon Sep 17 00:00:00 2001 From: Quick <31828688+Quick104@users.noreply.github.com> Date: Thu, 11 Jun 2026 22:05:57 -0400 Subject: [PATCH] fix(notifications): drain digest window fully and re-check kill switch under claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Digest sends read one channelFetchLimit page and then stamped last_digest_at, permanently dropping overflow rows from combined-mode recaps and slipping digest-only overflow by a day per page. Digest legs now page listSince until the window is empty before stamping; renderers already cap displayed items, so large drains stay deliverable. Per-episode sends keep single-page reads — their watermark-advance semantics were already correct. Also re-check the channel's enabled() under the claim lock so flipping the admin kill switch stops an in-flight pass immediately instead of after it completes; the existing errChannelUnavailable path aborts the pass without penalizing the recipient. Co-Authored-By: Claude Fable 5 --- .../notifications/account_channel_engine.go | 44 +++++- .../account_channel_engine_test.go | 125 ++++++++++++++++++ 2 files changed, 166 insertions(+), 3 deletions(-) create mode 100644 internal/notifications/account_channel_engine_test.go diff --git a/internal/notifications/account_channel_engine.go b/internal/notifications/account_channel_engine.go index 0dd9c5f4..f8b2fbd2 100644 --- a/internal/notifications/account_channel_engine.go +++ b/internal/notifications/account_channel_engine.go @@ -46,8 +46,9 @@ const ( // channelNudgeDelay coalesces the per-row dispatch nudges of one fanout // batch (all rows commit before the first nudge fires) into one pass. channelNudgeDelay = 2 * time.Second - // channelFetchLimit bounds one send's worth of watermark progress; the - // next pass drains the remainder. + // channelFetchLimit is the delivery read page size. Per-episode sends + // stop after one page (the next pass drains the remainder); digest sends + // page until the window is empty before stamping last_digest_at. channelFetchLimit = 200 // channelMaxFailuresPerPass stops a pass early when sends keep failing — // transport trouble is almost always global, not per-recipient. @@ -90,6 +91,24 @@ func channelDigestDue(now time.Time, digestHour int, lastDigestAt *time.Time) bo return lastDigestAt == nil || lastDigestAt.Before(todaySend) } +// drainSince pages fetch from the given cursor until a short read, returning +// every row in the window in delivery order. +func drainSince(fetch func(since Cursor, limit int) ([]DeliveryRow, error), from Cursor) ([]DeliveryRow, error) { + var all []DeliveryRow + for { + batch, err := fetch(from, channelFetchLimit) + if err != nil { + return nil, err + } + all = append(all, batch...) + if len(batch) < channelFetchLimit { + return all, nil + } + last := batch[len(batch)-1] + from = Cursor{CreatedAt: last.CreatedAt, ID: last.ID} + } +} + // channelRetryEligible applies exponential backoff after failed sends: // 1m, 2m, 4m, ... capped at channelFailureBackoffMax. func channelRetryEligible(now time.Time, lastAttemptAt *time.Time, consecutiveFailures int) bool { @@ -298,6 +317,13 @@ func (w *accountChannelWorker[K]) processRecipient(ctx context.Context, rec acco return nil // another node is handling this recipient } + // Re-check the admin kill switch under the lock: a pass over many + // recipients can outlive a settings flip, and disabling the channel must + // stop in-flight sends, not just future passes. + if !w.channel.enabled(ctx) { + return fmt.Errorf("channel disabled: %w", errChannelUnavailable) + } + // Re-derive eligibility from the locked row: the pre-scan snapshot may // predate a user mode flip or another node's digest stamp. mode := effectiveChannelMode(claimed.Mode, w.channel.allowPerEpisode(ctx)) @@ -340,7 +366,19 @@ func (w *accountChannelWorker[K]) processRecipient(ctx context.Context, rec acco return nil } - rows, err := w.channel.listSince(ctx, tx, rec.Key, fetchFrom, channelFetchLimit) + fetch := func(since Cursor, limit int) ([]DeliveryRow, error) { + return w.channel.listSince(ctx, tx, rec.Key, since, limit) + } + var rows []DeliveryRow + if digestAt != nil { + // Stamping last_digest_at closes the digest window — permanently for + // the combined mode, until tomorrow for digest-only — so the digest + // must drain the whole window, not stop at one page. Renderers cap + // how many items they show, so a large drain stays deliverable. + rows, err = drainSince(fetch, fetchFrom) + } else { + rows, err = fetch(fetchFrom, channelFetchLimit) + } if err != nil { return err } diff --git a/internal/notifications/account_channel_engine_test.go b/internal/notifications/account_channel_engine_test.go new file mode 100644 index 00000000..e6078113 --- /dev/null +++ b/internal/notifications/account_channel_engine_test.go @@ -0,0 +1,125 @@ +package notifications + +import ( + "errors" + "fmt" + "testing" + "time" +) + +// fakeDeliveryWindow serves listSince-style paged reads over a fixed, +// (created_at, id)-ordered dataset, mirroring the exclusive-cursor semantics +// of the real delivery queries. +type fakeDeliveryWindow struct { + rows []DeliveryRow + fetches int +} + +func (f *fakeDeliveryWindow) fetch(since Cursor, limit int) ([]DeliveryRow, error) { + f.fetches++ + out := make([]DeliveryRow, 0, limit) + for _, row := range f.rows { + if !cursorLess(since, Cursor{CreatedAt: row.CreatedAt, ID: row.ID}) { + continue + } + out = append(out, row) + if len(out) == limit { + break + } + } + return out, nil +} + +func makeDeliveryRows(n int) []DeliveryRow { + base := time.Date(2026, 6, 11, 8, 0, 0, 0, time.UTC) + rows := make([]DeliveryRow, n) + for i := range rows { + rows[i].ID = fmt.Sprintf("d%06d", i) + rows[i].CreatedAt = base.Add(time.Duration(i) * time.Second) + } + return rows +} + +func TestDrainSinceShortWindow(t *testing.T) { + window := &fakeDeliveryWindow{rows: makeDeliveryRows(3)} + got, err := drainSince(window.fetch, Cursor{}) + if err != nil { + t.Fatalf("drainSince: %v", err) + } + if len(got) != 3 { + t.Fatalf("expected 3 rows, got %d", len(got)) + } + if window.fetches != 1 { + t.Fatalf("expected 1 fetch for a short window, got %d", window.fetches) + } +} + +func TestDrainSinceMultiplePages(t *testing.T) { + // 2.5 pages: a single-page read would drop 300 rows from the digest. + total := channelFetchLimit*2 + channelFetchLimit/2 + window := &fakeDeliveryWindow{rows: makeDeliveryRows(total)} + + got, err := drainSince(window.fetch, Cursor{}) + if err != nil { + t.Fatalf("drainSince: %v", err) + } + if len(got) != total { + t.Fatalf("expected %d rows, got %d", total, len(got)) + } + if window.fetches != 3 { + t.Fatalf("expected 3 fetches, got %d", window.fetches) + } + for i, row := range got { + if want := fmt.Sprintf("d%06d", i); row.ID != want { + t.Fatalf("row %d out of order: got %s, want %s", i, row.ID, want) + } + } +} + +func TestDrainSinceExactPageBoundary(t *testing.T) { + window := &fakeDeliveryWindow{rows: makeDeliveryRows(channelFetchLimit)} + got, err := drainSince(window.fetch, Cursor{}) + if err != nil { + t.Fatalf("drainSince: %v", err) + } + if len(got) != channelFetchLimit { + t.Fatalf("expected %d rows, got %d", channelFetchLimit, len(got)) + } + // A full first page can't prove the window is empty; the confirming + // second fetch is expected. + if window.fetches != 2 { + t.Fatalf("expected 2 fetches, got %d", window.fetches) + } +} + +func TestDrainSinceRespectsCursor(t *testing.T) { + rows := makeDeliveryRows(10) + window := &fakeDeliveryWindow{rows: rows} + from := Cursor{CreatedAt: rows[6].CreatedAt, ID: rows[6].ID} + + got, err := drainSince(window.fetch, from) + if err != nil { + t.Fatalf("drainSince: %v", err) + } + if len(got) != 3 { + t.Fatalf("expected 3 rows past the cursor, got %d", len(got)) + } + if got[0].ID != rows[7].ID { + t.Fatalf("expected first row %s, got %s", rows[7].ID, got[0].ID) + } +} + +func TestDrainSincePropagatesError(t *testing.T) { + window := &fakeDeliveryWindow{rows: makeDeliveryRows(channelFetchLimit + 1)} + wantErr := errors.New("boom") + fetch := func(since Cursor, limit int) ([]DeliveryRow, error) { + if window.fetches >= 1 { + return nil, wantErr + } + return window.fetch(since, limit) + } + + if _, err := drainSince(fetch, Cursor{}); !errors.Is(err, wantErr) { + t.Fatalf("expected fetch error to propagate, got %v", err) + } +}