fix(notifications): drain digest window fully and re-check kill switch under claim

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 <noreply@anthropic.com>

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
 	}

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)
+	}
+}