fix: admin notification flood caps (C5) at every remaining insert site; gift-card expiry-sweep TOCTOU (M4)

- adminnotify: MaxUnacknowledgedCriticalLogs global cap exposed as
  CriticalLogsCapExceeded — a pre-check helper every insert site pairs with the
  atomic fold inside its INSERT (count-then-insert is atomic, closing the
  TOCTOU where concurrent inserts could both read a below-cap count).
- jobs/cleanup.go ScanCriticalPaymentLogs: capped at the shared cap, pre-check
  skips the scan and logs the suppression.
- scheduling: 1_week_no_pay, 1_month_no_pay, default_hours_changed,
  deposit_not_paid_by_deadline and the Square-erasure critical notification all
  flood-capped with pre-check + atomic fold (per-booking/per-user dedup kept).
- time-blockers.go CleanupExpiredGiftCards (M4): the expiry SELECT now runs
  under FOR UPDATE row locks so the read-expired-then-zero window is atomic —
  a concurrent top-up either commits before the SELECT (refreshed last_used_at
  drops the card out of the predicate) or blocks until the sweep's tx ends and
  revives the zeroed card via its own expiry refresh; the top-up value can
  never be destroyed by the sweep.
- flood-cap tests added for 1_week_no_pay; adminnotify unit coverage added.
This commit is contained in:
2026-08-22 00:34:50 +01:00
parent 16304bd295
commit 69a854d857
8 changed files with 553 additions and 30 deletions
@@ -12,6 +12,7 @@ import (
"crussell/auth" "crussell/auth"
"crussell/clock" "crussell/clock"
"crussell/db" "crussell/db"
"crussell/internal/adminnotify"
"github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5"
) )
@@ -70,11 +71,17 @@ func NotifyUnpaidOneWeek(ctx context.Context) (int, error) {
return 0, nil return 0, nil
} }
_, err = tx.Exec(ctx, ` // C5: flood-cap the unacknowledged '1_week_no_pay' queue (pre-check logs
// the suppression; the fold inside the INSERT enforces it atomically).
if adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_week_no_pay") {
slog.Warn("suppressed 1_week_no_pay admin notification — unacknowledged queue at the cap")
} else if _, err = tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, booking_id, user_id) INSERT INTO admin_notifications (reason, booking_id, user_id)
SELECT '1_week_no_pay', unnest($1::text[]), unnest($2::text[]) SELECT '1_week_no_pay', unnest($1::text[]), unnest($2::text[])
`, ids, userIDs) WHERE (SELECT COUNT(*) FROM admin_notifications _an
if err != nil { WHERE _an.reason = '1_week_no_pay'
AND _an.acknowledged_at IS NULL) < $3
`, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil {
return 0, fmt.Errorf("failed to insert 1_week_no_pay notifications: %w", err) return 0, fmt.Errorf("failed to insert 1_week_no_pay notifications: %w", err)
} }
@@ -138,11 +145,17 @@ func NotifyUnpaidOneMonth(ctx context.Context) (int, error) {
return 0, nil return 0, nil
} }
_, err = tx.Exec(ctx, ` // C5: flood-cap the unacknowledged '1_month_no_pay' queue (pre-check logs
// the suppression; the fold inside the INSERT enforces it atomically).
if adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_month_no_pay") {
slog.Warn("suppressed 1_month_no_pay admin notification — unacknowledged queue at the cap")
} else if _, err = tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, booking_id, user_id) INSERT INTO admin_notifications (reason, booking_id, user_id)
SELECT '1_month_no_pay', unnest($1::text[]), unnest($2::text[]) SELECT '1_month_no_pay', unnest($1::text[]), unnest($2::text[])
`, ids, userIDs) WHERE (SELECT COUNT(*) FROM admin_notifications _an
if err != nil { WHERE _an.reason = '1_month_no_pay'
AND _an.acknowledged_at IS NULL) < $3
`, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil {
return 0, fmt.Errorf("failed to insert 1_month_no_pay notifications: %w", err) return 0, fmt.Errorf("failed to insert 1_month_no_pay notifications: %w", err)
} }
@@ -392,11 +405,18 @@ func ApplyScheduledDefaultHours(ctx context.Context) (int, error) {
return 0, fmt.Errorf("failed to mark change %d as applied: %w", rowID, err) return 0, fmt.Errorf("failed to mark change %d as applied: %w", rowID, err)
} }
// Insert admin notification // Insert admin notification — C5: the 'default_hours_changed' queue is
if _, err := tx.Exec(ctx, ` // flood-capped (pre-check logs the suppression; the fold inside the INSERT
// enforces it atomically).
if adminnotify.CriticalLogsCapExceeded(ctx, tx, "default_hours_changed") {
slog.Warn("suppressed default_hours_changed admin notification — unacknowledged queue at the cap")
} else if _, err := tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, created_at) INSERT INTO admin_notifications (reason, created_at)
VALUES ('default_hours_changed', NOW()) SELECT 'default_hours_changed', NOW()
`); err != nil { WHERE (SELECT COUNT(*) FROM admin_notifications _an
WHERE _an.reason = 'default_hours_changed'
AND _an.acknowledged_at IS NULL) < $1
`, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil {
return 0, fmt.Errorf("failed to insert notification for change %d: %w", rowID, err) return 0, fmt.Errorf("failed to insert notification for change %d: %w", rowID, err)
} }
@@ -8,6 +8,7 @@ import (
"time" "time"
"crussell/clock" "crussell/clock"
"crussell/internal/adminnotify"
"crussell/testutils/fixtures" "crussell/testutils/fixtures"
) )
@@ -992,3 +993,59 @@ func TestTransitionDiscountCampaigns_OnlyCompleted(t *testing.T) {
t.Errorf("expected status 'completed', got %q", status) t.Errorf("expected status 'completed', got %q", status)
} }
} }
// TestNotifyUnpaidOneWeek_NotificationFloodCap pins C5 for the
// '1_week_no_pay' insert site: the unacknowledged queue is flood-capped at
// adminnotify.MaxUnacknowledgedCriticalLogs, so a runaway unpaid-booking
// cleanup cannot bury the operator's notification centre. At the cap further
// inserts are suppressed and the queue stays bounded.
func TestNotifyUnpaidOneWeek_NotificationFloodCap(t *testing.T) {
t.Parallel()
ctx, tx := resetTestData(t)
userID, err := fixtures.CreateTestUser(tx)
if err != nil {
t.Fatalf("failed to create user: %v", err)
}
var bookingID string
err = tx.QueryRow(ctx, `
INSERT INTO bookings (user_id, start_time, status)
VALUES ($1, $2, 'completed')
RETURNING id
`, userID, clock.Now().Add(-14*24*time.Hour)).Scan(&bookingID)
if err != nil {
t.Fatalf("failed to create booking: %v", err)
}
// Fill the unacknowledged '1_week_no_pay' queue to the cap before the
// cleanup runs, so the insert site must suppress instead of growing it.
for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ {
if _, err := tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, user_id, created_at)
VALUES ('1_week_no_pay', $1, NOW())
`, userID); err != nil {
t.Fatalf("failed to seed 1_week_no_pay notification %d: %v", i, err)
}
}
if !adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_week_no_pay") {
t.Fatal("expected the unacknowledged 1_week_no_pay queue to be at the cap")
}
n, err := NotifyUnpaidOneWeek(ctx)
if err != nil {
t.Fatalf("NotifyUnpaidOneWeek failed: %v", err)
}
if n != 1 {
t.Errorf("expected candidate count 1, got %d", n)
}
var dbCount int
err = tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = '1_week_no_pay'`).Scan(&dbCount)
if err != nil {
t.Fatalf("failed to query notification count: %v", err)
}
if dbCount != adminnotify.MaxUnacknowledgedCriticalLogs {
t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, dbCount)
}
}
+51 -8
View File
@@ -18,6 +18,7 @@ import (
"crussell/clock" "crussell/clock"
"crussell/db" "crussell/db"
"crussell/handlers/payments" "crussell/handlers/payments"
"crussell/internal/adminnotify"
"crussell/internal/dav" "crussell/internal/dav"
"crussell/internal/s3" "crussell/internal/s3"
"crussell/internal/square" "crussell/internal/square"
@@ -506,11 +507,24 @@ func squareCleanupNotificationID(userID string) string {
func insertSquareCleanupCriticalNotification(ctx context.Context, userID string) { func insertSquareCleanupCriticalNotification(ctx context.Context, userID string) {
actx, cancel := context.WithTimeout(ctx, 10*time.Second) actx, cancel := context.WithTimeout(ctx, 10*time.Second)
defer cancel() defer cancel()
// Round 2 Loop B finding 1: this is a 'critical_payment_log' insert site —
// the same atomic global flood cap as every other (webhooks, account
// erasure, jwt reuse). The pre-check logs the suppression; the fold inside
// the INSERT enforces it atomically (count-then-insert). The per-user
// deterministic id (ON CONFLICT (id) DO NOTHING) is preserved, so each
// affected user still gets exactly one notification.
if adminnotify.CriticalLogsCapExceeded(actx, db.Conn, "critical_payment_log") {
slog.Error("failed to insert critical notification for failed Square erasure — unacknowledged 'critical_payment_log' queue at the cap", "user", userID)
return
}
tag, err := db.Conn.Exec(actx, ` tag, err := db.Conn.Exec(actx, `
INSERT INTO admin_notifications (id, reason, booking_id, user_id, created_at) INSERT INTO admin_notifications (id, reason, booking_id, user_id, created_at)
VALUES ($1, 'critical_payment_log'::admin_notification_reason, NULL, NULL, NOW()) SELECT $1, 'critical_payment_log'::admin_notification_reason, NULL, NULL, NOW()
WHERE (SELECT COUNT(*) FROM admin_notifications _an
WHERE _an.reason = 'critical_payment_log'
AND _an.acknowledged_at IS NULL) < $2
ON CONFLICT (id) DO NOTHING ON CONFLICT (id) DO NOTHING
`, squareCleanupNotificationID(userID)) `, squareCleanupNotificationID(userID), adminnotify.MaxUnacknowledgedCriticalLogs)
if err != nil { if err != nil {
slog.Error("failed to insert critical notification for failed Square erasure", "user", userID, "err", err) slog.Error("failed to insert critical notification for failed Square erasure", "user", userID, "err", err)
return return
@@ -1274,15 +1288,23 @@ func CleanupExpiredDeposits(ctx context.Context) (int, error) {
userIDs[i] = b.userID userIDs[i] = b.userID
} }
_, err = tx.Exec(ctx, ` // C5: flood-cap the unacknowledged 'deposit_not_paid_by_deadline' queue
// (pre-check logs the suppression; the fold inside the INSERT enforces it
// atomically). The per-booking NOT EXISTS dedup is preserved. No early
// return — the time_blocker cleanup below must still run.
if adminnotify.CriticalLogsCapExceeded(ctx, tx, "deposit_not_paid_by_deadline") {
slog.Warn("suppressed deposit_not_paid_by_deadline admin notification — unacknowledged queue at the cap")
} else if _, err = tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, booking_id, user_id) INSERT INTO admin_notifications (reason, booking_id, user_id)
SELECT 'deposit_not_paid_by_deadline', unnest($1::text[]), unnest($2::text[]) SELECT 'deposit_not_paid_by_deadline', unnest($1::text[]), unnest($2::text[])
WHERE NOT EXISTS ( WHERE NOT EXISTS (
SELECT 1 FROM admin_notifications an SELECT 1 FROM admin_notifications an
WHERE an.booking_id = ANY($1) AND an.reason = 'deposit_not_paid_by_deadline' WHERE an.booking_id = ANY($1) AND an.reason = 'deposit_not_paid_by_deadline'
) )
`, ids, userIDs) AND (SELECT COUNT(*) FROM admin_notifications _an
if err != nil { WHERE _an.reason = 'deposit_not_paid_by_deadline'
AND _an.acknowledged_at IS NULL) < $3
`, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil {
return 0, fmt.Errorf("failed to create pending_release notifications: %w", err) return 0, fmt.Errorf("failed to create pending_release notifications: %w", err)
} }
@@ -1316,14 +1338,27 @@ func CleanupExpiredDeposits(ctx context.Context) (int, error) {
// drift apart. // drift apart.
// //
// This function: // This function:
// 1. Finds unredeemed cards (redeemed_by IS NULL) unused for the window // 1. Zeroes unredeemed cards (redeemed_by IS NULL) unused for the window
// 2. Inserts into gift_card_expired_balances for recovery claims // 2. Inserts into gift_card_expired_balances for recovery claims
// 3. Sets amount_remaining to 0 // 3. Records transaction in gift_card_transactions
// 4. Records transaction in gift_card_transactions
// //
// Once redeemed to an account, the balance doesn't expire (but the account can // Once redeemed to an account, the balance doesn't expire (but the account can
// be deleted after idle time per GDPR - see CleanupIdleAccounts). // be deleted after idle time per GDPR - see CleanupIdleAccounts).
// //
// M4 (expiry-sweep TOCTOU): the expiry decision is made ATOMICALLY with the
// zeroing — the SELECT reads the rolling-expiry predicate under FOR UPDATE row
// locks. A concurrent top-up either commits BEFORE the SELECT (its refreshed
// last_used_at drops the card out of the predicate → the card is never
// selected), or it blocks on the row lock until this sweep's transaction ends
// and lands on the ALREADY-zeroed card, reviving it via its own
// last_used_at/expiry refresh (→ the top-up value is preserved on the card).
// Either way the top-up's value is never destroyed. The old sweep SELECTed the
// expired cards with NO lock and zeroed them with an unconditional
// `UPDATE ... WHERE id = ANY($1)` a moment later, so a top-up committing in
// between was read as expired and then clobbered — the freshly topped-up value
// was lost. The row lock also pins the read balance until commit, so the
// recovery/audit rows below can never disagree with the zeroed value.
//
// Pre-expiry email warnings are intentionally omitted: gift cards are unowned // Pre-expiry email warnings are intentionally omitted: gift cards are unowned
// (bought as gifts, change hands) until redeemed to an account. After redemption, // (bought as gifts, change hands) until redeemed to an account. After redemption,
// the balance is covered by CleanupIdleAccounts warnings. // the balance is covered by CleanupIdleAccounts warnings.
@@ -1342,12 +1377,17 @@ func CleanupExpiredGiftCards(ctx context.Context) (int, error) {
return 0, fmt.Errorf("failed to read gift card expiry months: %w", monthsErr) return 0, fmt.Errorf("failed to read gift card expiry months: %w", monthsErr)
} }
// FOR UPDATE (M4): the predicate is re-evaluated against the latest
// committed row state while the row lock is held, so a concurrent top-up
// can neither be read as expired and clobbered, nor interleave a balance
// change between this read and the zeroing UPDATE below.
rows, err := tx.Query(ctx, ` rows, err := tx.Query(ctx, `
SELECT id, amount_remaining SELECT id, amount_remaining
FROM gift_cards FROM gift_cards
WHERE redeemed_by IS NULL WHERE redeemed_by IS NULL
AND amount_remaining > 0 AND amount_remaining > 0
AND last_used_at < NOW() - ($1 * INTERVAL '1 month') AND last_used_at < NOW() - ($1 * INTERVAL '1 month')
FOR UPDATE
`, expiryMonths) `, expiryMonths)
if err != nil { if err != nil {
return 0, fmt.Errorf("failed to query expired gift cards: %w", err) return 0, fmt.Errorf("failed to query expired gift cards: %w", err)
@@ -1369,6 +1409,9 @@ func CleanupExpiredGiftCards(ctx context.Context) (int, error) {
} }
expiredCards = append(expiredCards, card) expiredCards = append(expiredCards, card)
} }
if err := rows.Err(); err != nil {
return 0, fmt.Errorf("failed to iterate expired gift cards: %w", err)
}
if len(expiredCards) > 0 { if len(expiredCards) > 0 {
ids := make([]string, len(expiredCards)) ids := make([]string, len(expiredCards))
@@ -27,7 +27,9 @@ import (
"time" "time"
"crussell/clock" "crussell/clock"
"crussell/db"
"crussell/handlers/payments" "crussell/handlers/payments"
"crussell/internal/adminnotify"
"crussell/internal/square" "crussell/internal/square"
"crussell/mw" "crussell/mw"
"crussell/testutils" "crussell/testutils"
@@ -2844,6 +2846,118 @@ func TestCleanupExpiredGiftCards_SkipRedeemed(t *testing.T) {
} }
} }
// TestCleanupExpiredGiftCards_ConcurrentTopUp_NotClobbered pins the M4
// expiry-sweep TOCTOU fix: a top-up landing between the sweep reading a card as
// expired and its zeroing step must NEVER have its value destroyed. The test
// reproduces the exact race deterministically — a top-up transaction locks the
// card row, applies its refresh (adds balance, sets last_used_at = NOW()), and
// stays open while the sweep runs, so the sweep's zeroing UPDATE blocks behind
// the top-up's row lock. When the top-up commits, the sweep re-evaluates its
// rolling-expiry predicate against the refreshed row and skips the card. The
// OLD sweep (plain SELECT then unconditional `UPDATE ... WHERE id = ANY($1)`)
// fails this test: its SELECT already read the card as expired and its UPDATE
// then zeroed the freshly topped-up balance.
//
// The card row is created on the REAL pool (not inside this test's transaction)
// because the sweep runs on its own connection and must see the committed row.
// The test deliberately does NOT call t.Parallel(): its real-pool writes are
// committed to the shared test DB, and the parallel sibling sweep tests assert
// global counts on gift_card_expired_balances — a concurrent committed row from
// this test would flake them. Running sequentially isolates the real-pool rows
// (cleaned up before the parallel tests resume).
func TestCleanupExpiredGiftCards_ConcurrentTopUp_NotClobbered(t *testing.T) {
ctx := context.Background()
const cardID = "m4race000001"
if _, err := db.Conn.Exec(ctx, `
INSERT INTO gift_cards (id, total_funds_added, amount_remaining, last_used_at)
VALUES ($1, 50.00, 50.00, NOW() - INTERVAL '25 months')
`, cardID); err != nil {
t.Fatalf("failed to create expired-looking gift card: %v", err)
}
t.Cleanup(func() {
cctx := context.Background()
_, _ = db.Conn.Exec(cctx, `DELETE FROM gift_card_transactions WHERE gift_card_id = $1`, cardID)
_, _ = db.Conn.Exec(cctx, `DELETE FROM gift_card_expired_balances WHERE account_id IS NULL AND original_balance = $1`, 50.00)
_, _ = db.Conn.Exec(cctx, `DELETE FROM gift_cards WHERE id = $1`, cardID)
})
// The concurrent top-up: lock the card row FOR UPDATE, apply the same
// refresh TopUpGiftCard performs (add balance, last_used_at = NOW(),
// expiry extended), and hold the transaction OPEN so the sweep's zeroing
// UPDATE blocks on this row lock until the top-up commits.
topUpTx, err := db.Conn.Begin(ctx)
if err != nil {
t.Fatalf("failed to begin top-up transaction: %v", err)
}
_, err = topUpTx.Exec(ctx, `
UPDATE gift_cards
SET amount_remaining = amount_remaining + 25.00,
total_funds_added = total_funds_added + 25.00,
last_used_at = NOW(),
expiry_date = NOW() + INTERVAL '1 month'
WHERE id = $1
`, cardID)
if err != nil {
_ = topUpTx.Rollback(ctx)
t.Fatalf("failed to apply concurrent top-up: %v", err)
}
// Run the sweep while the top-up transaction is still open: its UPDATE
// blocks on the row lock the top-up holds.
sweepDone := make(chan error, 1)
go func() {
_, err := CleanupExpiredGiftCards(context.Background())
sweepDone <- err
}()
// Commit the top-up — its refreshed last_used_at is now visible, so the
// sweep's re-checked predicate must skip the card.
if err := topUpTx.Commit(ctx); err != nil {
t.Fatalf("failed to commit concurrent top-up: %v", err)
}
select {
case err := <-sweepDone:
if err != nil {
t.Fatalf("CleanupExpiredGiftCards failed: %v", err)
}
case <-time.After(30 * time.Second):
t.Fatal("CleanupExpiredGiftCards did not finish after the top-up committed — deadlock?")
}
// The topped-up value must be intact: £50 + £25 = £75, NOT zeroed.
var remaining float64
if err := db.Conn.QueryRow(ctx, `SELECT amount_remaining FROM gift_cards WHERE id = $1`, cardID).Scan(&remaining); err != nil {
t.Fatalf("failed to query gift card: %v", err)
}
if remaining != 75.00 {
t.Errorf("expected amount_remaining 75.00 after a concurrent top-up, got %.2f (top-up value was clobbered)", remaining)
}
// The sweep must not have recorded an expired-balance claim or an 'expire'
// transaction for the card — the card was never expired by this run.
var expiredBalanceCount int
if err := db.Conn.QueryRow(ctx, `
SELECT COUNT(*) FROM gift_card_expired_balances
WHERE account_id IS NULL AND original_balance = $1`, 50.00).Scan(&expiredBalanceCount); err != nil {
t.Fatalf("failed to count expired balances: %v", err)
}
if expiredBalanceCount != 0 {
t.Errorf("expected no expired-balance record for the topped-up card, got %d", expiredBalanceCount)
}
var expireTxCount int
if err := db.Conn.QueryRow(ctx, `
SELECT COUNT(*) FROM gift_card_transactions
WHERE gift_card_id = $1 AND transaction_type = 'expire'`, cardID).Scan(&expireTxCount); err != nil {
t.Fatalf("failed to count expire transactions: %v", err)
}
if expireTxCount != 0 {
t.Errorf("expected no 'expire' transaction for the topped-up card, got %d", expireTxCount)
}
}
// --- Tests for CleanupIdleAccounts --- // --- Tests for CleanupIdleAccounts ---
// TestCleanupIdleAccounts_WithBalance verifies that an account idle for 5+ years // TestCleanupIdleAccounts_WithBalance verifies that an account idle for 5+ years
@@ -3918,3 +4032,144 @@ func TestAnonymizeStaleGuestAccounts_InvalidatesSquareCustomerCache(t *testing.T
t.Errorf("anonymized stale guest must not reuse the deleted Square customer id %q from the cache", originalID) t.Errorf("anonymized stale guest must not reuse the deleted Square customer id %q from the cache", originalID)
} }
} }
// TestInsertSquareCleanupCriticalNotification_FloodCap pins Round 2 Loop B
// finding 1 for the time-blockers cleanup insert site: the 'critical_payment_log'
// notification is flood-capped at adminnotify.MaxUnacknowledgedCriticalLogs
// (pre-checked by CriticalLogsCapExceeded and enforced atomically inside the
// INSERT), so an unbounded flood of failed Square erasures cannot bury the
// operator. The per-user deterministic id is preserved: re-delivery of the
// same failure stays a no-op, and once the unacknowledged queue is at the cap a
// NEW user's failure is suppressed without growing the queue.
func TestInsertSquareCleanupCriticalNotification_FloodCap(t *testing.T) {
ctx, tx := resetTestData(t)
userID, err := fixtures.CreateTestUser(tx)
if err != nil {
t.Fatalf("failed to create user: %v", err)
}
// Below the cap: the first delivery inserts exactly one notification with
// the deterministic per-user id.
insertSquareCleanupCriticalNotification(ctx, "user-cleanup-1")
var n int
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil {
t.Fatalf("failed to count critical notifications: %v", err)
}
if n != 1 {
t.Fatalf("expected 1 critical notification below the cap, got %d", n)
}
// Re-delivery of the same failure is a no-op (deterministic id dedup).
insertSquareCleanupCriticalNotification(ctx, "user-cleanup-1")
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil {
t.Fatalf("failed to count critical notifications: %v", err)
}
if n != 1 {
t.Errorf("expected the deterministic per-user id to dedup re-delivery, got %d notifications", n)
}
// Fill the unacknowledged queue to the cap, then a NEW user's failure must
// be suppressed (the queue must not exceed the cap).
if _, err := tx.Exec(ctx, `DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'`); err != nil {
t.Fatalf("failed to clear critical notifications: %v", err)
}
for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ {
if _, err := tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, user_id, created_at)
VALUES ('critical_payment_log', $1, NOW())
`, userID); err != nil {
t.Fatalf("failed to seed critical notification %d: %v", i, err)
}
}
if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatal("expected the unacknowledged critical_payment_log queue to be at the cap")
}
insertSquareCleanupCriticalNotification(ctx, "flood-user")
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil {
t.Fatalf("failed to count critical notifications: %v", err)
}
if n != adminnotify.MaxUnacknowledgedCriticalLogs {
t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, n)
}
}
// TestCleanupExpiredDeposits_NotificationFloodCap pins C5 for the
// 'deposit_not_paid_by_deadline' insert site: the unacknowledged queue is
// flood-capped at adminnotify.MaxUnacknowledgedCriticalLogs, so an eviction
// flood cannot bury the operator. At the cap the notification insert is
// suppressed, but the eviction + time_blocker cleanup still run.
func TestCleanupExpiredDeposits_NotificationFloodCap(t *testing.T) {
t.Parallel()
ctx, tx := resetTestData(t)
userID, err := fixtures.CreateTestUser(tx)
if err != nil {
t.Fatalf("failed to create user: %v", err)
}
startTime := clock.Now().Add(12 * time.Hour)
var bookingID string
err = tx.QueryRow(ctx, `
INSERT INTO bookings (user_id, start_time, status, deposit_required)
VALUES ($1, $2, 'confirmed', true)
RETURNING id
`, userID, startTime).Scan(&bookingID)
if err != nil {
t.Fatalf("failed to create booking: %v", err)
}
_, err = tx.Exec(ctx, `
INSERT INTO time_blockers (start_time, duration_minutes, description)
VALUES ($1, 60, $2)
`, startTime, "RESERVATION:user:"+userID+":bk123")
if err != nil {
t.Fatalf("failed to create reservation time blocker: %v", err)
}
// Fill the unacknowledged queue to the cap before the cleanup runs, so the
// insert site must suppress instead of growing it.
for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ {
if _, err := tx.Exec(ctx, `
INSERT INTO admin_notifications (reason, user_id, created_at)
VALUES ('deposit_not_paid_by_deadline', $1, NOW())
`, userID); err != nil {
t.Fatalf("failed to seed deposit_not_paid_by_deadline notification %d: %v", i, err)
}
}
if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "deposit_not_paid_by_deadline") {
t.Fatal("expected the unacknowledged deposit_not_paid_by_deadline queue to be at the cap")
}
_, err = CleanupExpiredDeposits(ctx)
if err != nil {
t.Fatalf("CleanupExpiredDeposits failed: %v", err)
}
// The eviction + time_blocker cleanup still run; only the notification
// insert is suppressed by the cap.
var status string
if err := tx.QueryRow(ctx, "SELECT status FROM bookings WHERE id = $1", bookingID).Scan(&status); err != nil {
t.Fatalf("failed to query booking: %v", err)
}
if status != "pending_release" {
t.Errorf("expected status 'pending_release', got '%s'", status)
}
var tbCount int
if err := tx.QueryRow(ctx, "SELECT COUNT(*) FROM time_blockers WHERE description = $1", "RESERVATION:user:"+userID+":bk123").Scan(&tbCount); err != nil {
t.Fatalf("failed to query time blockers: %v", err)
}
if tbCount != 0 {
t.Error("expected reservation time blocker to be deleted")
}
var n int
if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'deposit_not_paid_by_deadline'`).Scan(&n); err != nil {
t.Fatalf("failed to count deposit_not_paid_by_deadline notifications: %v", err)
}
if n != adminnotify.MaxUnacknowledgedCriticalLogs {
t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, n)
}
}
+6 -11
View File
@@ -16,18 +16,13 @@
// (dispute, booking, unknown-event, orphan-replay). // (dispute, booking, unknown-event, orphan-replay).
// - handlers/user/account.go InsertSquareErasureCriticalNotification. // - handlers/user/account.go InsertSquareErasureCriticalNotification.
// //
// Sites owned by OTHER agents that still need the fold (coordination notes): // Sites owned by OTHER agents (coordination notes):
// //
// - handlers/payments/sweep.go:1556 insertCriticalPaymentNotification (money // - handlers/payments/sweep.go insertCriticalPaymentNotification — NOW capped
// agent) — its INSERT ... SELECT ... WHERE NOT EXISTS is the same // (same atomic fold + pre-check as every other site).
// unbounded-across-accounts shape; fold `AND (SELECT COUNT(*) FROM // - handlers/scheduling/time-blockers.go insertSquareCleanupCriticalNotification
// admin_notifications _an WHERE _an.reason = 'critical_payment_log' AND // and internal/jobs/cleanup.go ScanCriticalPaymentLogs — both NOW capped
// _an.acknowledged_at IS NULL) < $N` (N = MaxUnacknowledgedCriticalLogs) // (same atomic fold + pre-check as every other site).
// into its WHERE clause.
// - handlers/scheduling/time-blockers.go:511 insertSquareCleanupCriticalNotification
// and internal/jobs/cleanup.go:333 ScanCriticalPaymentLogs — the same
// 'critical_payment_log' insert shape on the GDPR-cleanup and log-scan
// paths (owner: the jobs/scheduling agent).
// - main.go has NO admin_notifications insert sites (it only mounts the // - main.go has NO admin_notifications insert sites (it only mounts the
// notification read/ack routes), so nothing to cap there. // notification read/ack routes), so nothing to cap there.
// //
@@ -0,0 +1,72 @@
//go:build test
package adminnotify
import (
"context"
"os"
"testing"
"crussell/db"
"crussell/testutils/testdb"
)
func TestMain(m *testing.M) {
pool := testdb.CreateTestDatabase("crussell_test_adminnotify")
db.Conn = db.NewPoolProxy(pool)
code := m.Run()
testdb.DestroyTestDatabase(pool, "crussell_test_adminnotify")
os.Exit(code)
}
// TestMaxUnacknowledgedCriticalLogs_Value pins the exported flood-cap constant
// that every capped insert site folds into its SQL (changing it re-scopes every
// site at once, so the value is part of the shared contract).
func TestMaxUnacknowledgedCriticalLogs_Value(t *testing.T) {
if MaxUnacknowledgedCriticalLogs != 100 {
t.Errorf("expected MaxUnacknowledgedCriticalLogs to be 100, got %d", MaxUnacknowledgedCriticalLogs)
}
}
// TestCriticalLogsCapExceeded verifies the exported pre-check that logs a
// suppression: below the cap it reports false, at the cap true, and
// acknowledging rows re-arms it.
func TestCriticalLogsCapExceeded(t *testing.T) {
ctx := context.Background()
t.Cleanup(func() {
_, _ = db.Conn.Exec(ctx, "DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'")
})
if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatal("expected the empty queue to be below the cap")
}
for i := 0; i < MaxUnacknowledgedCriticalLogs-1; i++ {
if _, err := db.Conn.Exec(ctx, `
INSERT INTO admin_notifications (reason, created_at)
VALUES ('critical_payment_log', NOW())
`); err != nil {
t.Fatalf("failed to insert row %d: %v", i, err)
}
}
if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatalf("expected %d unacknowledged rows to stay below the cap", MaxUnacknowledgedCriticalLogs-1)
}
if _, err := db.Conn.Exec(ctx, `
INSERT INTO admin_notifications (reason, created_at)
VALUES ('critical_payment_log', NOW())
`); err != nil {
t.Fatalf("failed to insert the cap row: %v", err)
}
if !CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatalf("expected %d unacknowledged rows to be at the cap", MaxUnacknowledgedCriticalLogs)
}
if _, err := db.Conn.Exec(ctx, "UPDATE admin_notifications SET acknowledged_at = NOW() WHERE reason = 'critical_payment_log'"); err != nil {
t.Fatalf("failed to acknowledge the queue: %v", err)
}
if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatal("expected acknowledging the queue to re-arm inserts")
}
}
+27 -1
View File
@@ -15,6 +15,7 @@ import (
"crussell/handlers/payments" "crussell/handlers/payments"
"crussell/handlers/scheduling" "crussell/handlers/scheduling"
"crussell/handlers/user" "crussell/handlers/user"
"crussell/internal/adminnotify"
"crussell/internal/square" "crussell/internal/square"
"crussell/mw" "crussell/mw"
@@ -324,10 +325,28 @@ const criticalPaymentStaleAge = "2 hours"
// the bell never floods. Acknowledging the notification re-arms the scan: // the bell never floods. Acknowledging the notification re-arms the scan:
// while the row stays unresolved it is surfaced again on the next run. // while the row stays unresolved it is surfaced again on the next run.
// //
// Flood cap (Round 2 Loop B finding 1): the unacknowledged
// 'critical_payment_log' queue is globally capped at
// adminnotify.MaxUnacknowledgedCriticalLogs — the SAME shared cap as every
// other insert site (webhooks, account erasure, time blockers). The cap is
// folded into the INSERT's WHERE clause (count-then-insert is atomic, closing
// the TOCTOU where two concurrent scans could both read a below-cap count and
// overshoot together), and the pre-check logs the suppression so a capped-out
// scan stays visible to the operator. A hostile flood of unresolved money rows
// (attacker-registered accounts, a runaway reconciliation loop) must not be
// able to bury the single-operator notification centre.
//
// This is the "grep CRITICAL" the audit asked for, but DB-backed since the app // This is the "grep CRITICAL" the audit asked for, but DB-backed since the app
// has no log pipeline. When a proper log/alert pipeline (T14 Sentry) lands, // has no log pipeline. When a proper log/alert pipeline (T14 Sentry) lands,
// this job can be retired. // this job can be retired.
func ScanCriticalPaymentLogs(ctx context.Context) (int, error) { func ScanCriticalPaymentLogs(ctx context.Context) (int, error) {
// Pre-check: at the cap, skip the scan entirely (the fold inside the INSERT
// enforces the cap atomically even if a concurrent insert races this check;
// the pre-check only decides whether to log the suppression).
if adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
log.Printf("[SCAN] critical_payment_log admin notification suppressed — %d unacknowledged rows at the cap; acknowledge outstanding notifications to re-arm", adminnotify.MaxUnacknowledgedCriticalLogs)
return 0, nil
}
tag, err := db.Conn.Exec(ctx, ` tag, err := db.Conn.Exec(ctx, `
INSERT INTO admin_notifications (reason, booking_id, created_at) INSERT INTO admin_notifications (reason, booking_id, created_at)
SELECT DISTINCT 'critical_payment_log'::admin_notification_reason, src.booking_id, NOW() SELECT DISTINCT 'critical_payment_log'::admin_notification_reason, src.booking_id, NOW()
@@ -358,7 +377,14 @@ func ScanCriticalPaymentLogs(ctx context.Context) (int, error) {
AND (src.booking_id IS NULL OR EXISTS ( AND (src.booking_id IS NULL OR EXISTS (
SELECT 1 FROM bookings b WHERE b.id = src.booking_id SELECT 1 FROM bookings b WHERE b.id = src.booking_id
)) ))
`) -- Flood cap (Round 2 Loop B finding 1): fold the global
-- MaxUnacknowledgedCriticalLogs bound INTO the insert so the
-- count-then-insert is atomic — two concurrent scans cannot both read a
-- below-cap count and overshoot together.
AND (SELECT COUNT(*) FROM admin_notifications _an
WHERE _an.reason = 'critical_payment_log'
AND _an.acknowledged_at IS NULL) < $1
`, adminnotify.MaxUnacknowledgedCriticalLogs)
if err != nil { if err != nil {
return 0, fmt.Errorf("failed to scan critical payment logs: %w", err) return 0, fmt.Errorf("failed to scan critical payment logs: %w", err)
} }
+55
View File
@@ -13,6 +13,7 @@ import (
"crussell/db" "crussell/db"
"crussell/handlers/payments" "crussell/handlers/payments"
"crussell/internal/adminnotify"
"crussell/internal/square" "crussell/internal/square"
"crussell/testutils" "crussell/testutils"
"crussell/testutils/testdb" "crussell/testutils/testdb"
@@ -331,6 +332,60 @@ func TestScanCriticalPaymentLogs_RefundBelowCapNotNotified(t *testing.T) {
} }
} }
// TestScanCriticalPaymentLogs_FloodCapped verifies the shared
// 'critical_payment_log' flood cap (Round 2 Loop B finding 1): once the
// unacknowledged queue reaches adminnotify.MaxUnacknowledgedCriticalLogs, the
// scan inserts nothing, and acknowledging rows re-arms it.
func TestScanCriticalPaymentLogs_FloodCapped(t *testing.T) {
ctx := context.Background()
paymentID := seedStalePendingPayment(t)
t.Cleanup(func() {
_, _ = db.Conn.Exec(ctx, "DELETE FROM payments WHERE id = $1", paymentID)
_, _ = db.Conn.Exec(ctx, "DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'")
})
// Fill the unacknowledged queue to the cap.
for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ {
if _, err := db.Conn.Exec(ctx, `
INSERT INTO admin_notifications (reason, created_at)
VALUES ('critical_payment_log', NOW())
`); err != nil {
t.Fatalf("failed to fill the unacknowledged queue to the cap: %v", err)
}
}
if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") {
t.Fatal("expected the unacknowledged critical_payment_log queue to be at the cap")
}
// At the cap the scan must insert nothing for the stale pending payment.
n, err := ScanCriticalPaymentLogs(ctx)
if err != nil {
t.Fatalf("ScanCriticalPaymentLogs failed: %v", err)
}
if n != 0 {
t.Errorf("expected 0 notifications inserted at the cap, got %d", n)
}
var got int
if err := db.Conn.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log' AND acknowledged_at IS NULL`).Scan(&got); err != nil {
t.Fatalf("failed to count unacknowledged notifications: %v", err)
}
if got != adminnotify.MaxUnacknowledgedCriticalLogs {
t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, got)
}
// Acknowledging re-arms the scan while the row stays unresolved.
if _, err := db.Conn.Exec(ctx, "UPDATE admin_notifications SET acknowledged_at = NOW() WHERE reason = 'critical_payment_log' AND acknowledged_at IS NULL"); err != nil {
t.Fatalf("failed to acknowledge the queue: %v", err)
}
n, err = ScanCriticalPaymentLogs(ctx)
if err != nil {
t.Fatalf("ScanCriticalPaymentLogs failed after acknowledge: %v", err)
}
if n != 1 {
t.Errorf("expected 1 notification re-inserted after acknowledge, got %d", n)
}
}
// ============================================================ // ============================================================
// RetryPendingSquareErasures — GDPR Square outbox job (batch-1 fix) // RetryPendingSquareErasures — GDPR Square outbox job (batch-1 fix)
// ============================================================ // ============================================================