fix(poller): stop infinite re-enqueue of unsendable notifications

In shadow/dry-run/kill-switch modes, when the target has no token/recipient/
chat, or when not in the canary cohort, the handlers returned without marking
the notification, so the poller (pendingFilter on notified:{$ne:true})
re-selected the same batch every cycle forever (observed: fcm:500 every 10s).

Add a service-owned per-channel marker processedBy.<channel>, set in every
terminal 'won't/can't send' branch, and exclude marked docs in pendingFilter.
Unlike the shared notified/emailNotified/telegramNotified flags, processedBy is
invisible to the old system, so it never suppresses a real send in prod shadow.

- types.ts: NotificationDoc.processedBy?: Partial<Record<Channel, Date>>
- poller.ts: pendingFilter excludes processedBy.<channel>
- handlers.ts: markProcessed() in no-token/no-recipient/no-chat, decideSend=false,
  dead-token and telegram-blocked branches
- fake-repo: dotted-path support in matching and $set/$unset
- tests: regression test + processedBy assertions (86 passing)

src/handlers.ts

@@ -14,7 +14,7 @@ import { getEmailOf } from './recipient.js';
 import { decideSend } from './mode.js';
 import { underPerUserLimit } from './limits.js';
 import { markResult, release, reserve } from './sends.js';
-import type { NotificationDoc, UserDoc } from './types.js';
+import type { Channel, NotificationDoc, UserDoc } from './types.js';
 
 export interface Deps {
   cfg: Config;
@@ -43,6 +43,25 @@ async function loadPair(
   return { notif, user };
 }
 
+/**
+ * Mark a notification as terminally processed for a channel WITHOUT sending it.
+ * Sets a service-owned marker (`processedBy.<channel>`) that the poller excludes,
+ * so a doc we will never send (shadow/dry-run/kill-switch, not in the canary
+ * cohort, no token/recipient/chat) is not re-enqueued every cycle. It never
+ * touches the shared notified/emailNotified/telegramNotified flags, so the old
+ * system is unaffected.
+ */
+async function markProcessed(
+  repos: Repos,
+  id: NotificationDoc['_id'],
+  channel: Channel,
+): Promise<void> {
+  await repos.notifications.updateOne(
+    { _id: id },
+    { $set: { [`processedBy.${channel}`]: new Date() } },
+  );
+}
+
 /**
  * FCM push handler. Mirrors the old processNotif mobile branch, with all
  * safeguards. Throws RetryableError on transient failures so BullMQ retries.
@@ -60,7 +79,8 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise
 
   if (!user?.fireBaseToken) {
     log.warn({ notificationId, userId: notif.userId }, 'fcm: user has no fireBaseToken');
-    return; // matches old behavior: warn, no send, no mark
+    await markProcessed(repos, notif._id, 'fcm'); // terminal: no token, don't re-enqueue
+    return;
   }
 
   const decision = decideSend(cfg, notif);
@@ -69,6 +89,7 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise
       { notificationId, userId: notif.userId, mode: cfg.mode, reason: decision.reason, title: payload.title, body: payload.body },
       `fcm: not sending (${decision.reason})`,
     );
+    await markProcessed(repos, notif._id, 'fcm'); // won't send in this mode; don't re-enqueue
     return;
   }
 
@@ -104,6 +125,7 @@ export async function processFcmJob(deps: Deps, notificationId: string): Promise
       { _id: notif.userId },
       { $set: { fireBaseToken: null, fireBaseTokenDeadAt: new Date() } },
     );
+    await markProcessed(repos, notif._id, 'fcm'); // dead token: terminal, don't re-enqueue
     log.warn({ notificationId, userId: notif.userId, error: result.error }, 'fcm: dead token purged');
     return;
   }
@@ -126,6 +148,7 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi
   if (notif.type !== 'web' || notif.emailNotified === true) return;
   if (!user) {
     log.warn({ notificationId, userId: notif.userId }, 'email: user not found');
+    await markProcessed(repos, notif._id, 'email'); // terminal: no user, don't re-enqueue
     return;
   }
 
@@ -135,6 +158,8 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi
     if (cfg.mode === 'full' || cfg.mode === 'canary') {
       await repos.notifications.deleteOne({ _id: notif._id });
       await markResult(repos, notif._id, 'email', 'no-recipient');
+    } else {
+      await markProcessed(repos, notif._id, 'email'); // shadow/dry-run: keep doc but don't re-enqueue
     }
     log.info({ notificationId, userId: notif.userId }, 'email: no verified recipient');
     return;
@@ -153,6 +178,7 @@ export async function processEmailJob(deps: Deps, notificationId: string): Promi
       { notificationId, userId: notif.userId, mode: cfg.mode, reason: decision.reason, to: emailAddress, subject: content.subject },
       `email: not sending (${decision.reason})`,
     );
+    await markProcessed(repos, notif._id, 'email'); // won't send in this mode; don't re-enqueue
     return;
   }
 
@@ -197,12 +223,14 @@ export async function processTelegramJob(deps: Deps, notificationId: string): Pr
   if (notif.type !== 'telegram' || notif.telegramNotified === true) return;
   if (!notif.subsId) {
     log.warn({ notificationId }, 'telegram: notification has no subsId');
+    await markProcessed(repos, notif._id, 'telegram'); // terminal: no subsId, don't re-enqueue
     return;
   }
 
   const sub = await repos.subscriptions.findOne({ _id: notif.subsId });
   if (!sub || typeof sub.chatId !== 'number') {
     log.warn({ notificationId, subsId: String(notif.subsId) }, 'telegram: subscription/chatId not found');
+    await markProcessed(repos, notif._id, 'telegram'); // terminal: no chatId, don't re-enqueue
     return;
   }
   const chatId = sub.chatId;
@@ -212,6 +240,7 @@ export async function processTelegramJob(deps: Deps, notificationId: string): Pr
   const decision = decideSend(cfg, notif);
   if (!decision.send) {
     log.info({ notificationId, chatId, mode: cfg.mode, reason: decision.reason }, `telegram: not sending (${decision.reason})`);
+    await markProcessed(repos, notif._id, 'telegram'); // won't send in this mode; don't re-enqueue
     return;
   }
 
@@ -256,6 +285,7 @@ async function handleTgFailure(
       { _id: subId },
       { $set: { active: false, blockedAt: new Date() } } as never,
     );
+    await markProcessed(repos, notif._id, 'telegram'); // blocked: terminal, don't re-enqueue
     log.warn({ chatId, error: res.error }, 'telegram: bot blocked, sub marked inactive');
     return;
   }

src/poller.ts

@@ -11,11 +11,11 @@ import type { Channel, NotificationDoc, SendJob } from './types.js';
 export function pendingFilter(channel: Channel): FilterQuery<NotificationDoc> {
   switch (channel) {
     case 'fcm':
-      return { type: 'mobile', notified: { $ne: true } };
+      return { type: 'mobile', notified: { $ne: true }, 'processedBy.fcm': { $exists: false } };
     case 'email':
-      return { type: 'web', emailNotified: { $ne: true } };
+      return { type: 'web', emailNotified: { $ne: true }, 'processedBy.email': { $exists: false } };
     case 'telegram':
-      return { type: 'telegram', telegramNotified: { $ne: true } };
+      return { type: 'telegram', telegramNotified: { $ne: true }, 'processedBy.telegram': { $exists: false } };
   }
 }
 

src/types.ts

@@ -21,6 +21,14 @@ export interface NotificationDoc {
   // provenance set by our poller (invisible to the old system)
   claimedBy?: string;
   claimedAt?: Date;
+  /**
+   * Per-channel terminal marker set by our sender when a notification will NOT
+   * be sent and must not be re-enqueued (shadow/dry-run/kill-switch, not in the
+   * canary cohort, no token/recipient/chat). Service-owned and invisible to the
+   * old system: unlike `notified`, setting it never makes the old sender skip a
+   * real send. Stops the poller from re-selecting the same batch forever.
+   */
+  processedBy?: Partial<Record<Channel, Date>>;
 }
 
 export interface UserDoc {

test/fake-repo.ts

@@ -3,14 +3,51 @@
  * MongoDB 3.2, for which no mongod binary is downloadable/runnable on modern
  * dev/CI hosts (glibc/libssl). So the DB integration tests run against this
  * fake, which faithfully emulates the subset of behaviour the code relies on:
- * the query operators used ($ne/$in/$gte/$exists/equality), $set/$unset updates,
+ * the query operators used ($ne/$in/$gte/$exists/equality), $set/$unset updates
- * projection/limit, and — crucially — unique-index duplicate-key errors (11000)
+ * (including dotted paths like `processedBy.fcm`), projection/limit, and —
- * that back the idempotency guarantee.
+ * crucially — unique-index duplicate-key errors (11000) that back the
+ * idempotency guarantee.
  */
 import type { Repos } from '../src/db.js';
 
 type Doc = Record<string, any>;
 
+/** Read a possibly-dotted path (`a.b.c`) from a doc, like MongoDB does. */
+function getPath(doc: Doc, key: string): any {
+  if (!key.includes('.')) return doc[key];
+  return key.split('.').reduce<any>((cur, part) => (cur == null ? undefined : cur[part]), doc);
+}
+
+/** Set a possibly-dotted path, creating intermediate objects as MongoDB would. */
+function setPath(doc: Doc, key: string, value: any): void {
+  if (!key.includes('.')) {
+    doc[key] = value;
+    return;
+  }
+  const parts = key.split('.');
+  let cur = doc;
+  for (const part of parts.slice(0, -1)) {
+    if (cur[part] == null || typeof cur[part] !== 'object') cur[part] = {};
+    cur = cur[part];
+  }
+  cur[parts[parts.length - 1]!] = value;
+}
+
+/** Delete a possibly-dotted path. */
+function delPath(doc: Doc, key: string): void {
+  if (!key.includes('.')) {
+    delete doc[key];
+    return;
+  }
+  const parts = key.split('.');
+  let cur = doc;
+  for (const part of parts.slice(0, -1)) {
+    if (cur[part] == null) return;
+    cur = cur[part];
+  }
+  delete cur[parts[parts.length - 1]!];
+}
+
 function isObjectId(v: any): boolean {
   return v != null && typeof v === 'object' && typeof v.equals === 'function' && v._bsontype === 'ObjectID';
 }
@@ -64,12 +101,12 @@ function matchField(docVal: any, cond: any): boolean {
 }
 
 function matches(doc: Doc, filter: Doc): boolean {
-  return Object.entries(filter).every(([k, cond]) => matchField(doc[k], cond));
+  return Object.entries(filter).every(([k, cond]) => matchField(getPath(doc, k), cond));
 }
 
 function applyUpdate(doc: Doc, update: Doc): void {
-  if (update.$set) Object.assign(doc, update.$set);
+  if (update.$set) for (const [k, v] of Object.entries(update.$set)) setPath(doc, k, v);
-  if (update.$unset) for (const k of Object.keys(update.$unset)) delete doc[k];
+  if (update.$unset) for (const k of Object.keys(update.$unset)) delPath(doc, k);
 }
 
 class FakeCollection {

test/handlers.test.ts

@@ -2,6 +2,7 @@ import { beforeEach, describe, expect, it } from 'vitest';
 import type { Repos } from '../src/db.js';
 import type { Config } from '../src/config.js';
 import { processEmailJob, processFcmJob, type Deps } from '../src/handlers.js';
+import { pendingFilter } from '../src/poller.js';
 import { aNotif, aUser, fakeFcm, fakeMailer, freshRepos, silentLog, testCfg } from './helpers.js';
 
 let repos: Repos;
@@ -60,8 +61,31 @@ describe('FCM handler', () => {
 
     expect(fcm.calls).toHaveLength(0);
     const after = await repos.notifications.findOne({ _id: notif._id });
-    expect(after?.notified).toBeUndefined();
+    expect(after?.notified).toBeUndefined(); // never touches the old system's flag
-    expect(await repos.sends.countDocuments({})).toBe(0); // no reservation in dry-run
+    expect(after?.processedBy?.fcm).toBeInstanceOf(Date); // but marks it processed…
+    expect(await repos.sends.countDocuments({})).toBe(0); // …with no reservation in dry-run
+  });
+
+  it('does not re-enqueue a doc already processed in shadow mode', async () => {
+    // Regression for the infinite fcm:500/cycle re-enqueue: in shadow the handler
+    // marks processedBy so the poller filter no longer selects the doc.
+    const notif = aNotif();
+    await repos.notifications.insertOne(notif);
+    await repos.users.insertOne(aUser());
+    const fcm = fakeFcm({ ok: true, messageId: 'x' });
+    const shadow = { ...baseCfg, mode: 'shadow' as const };
+
+    // Before: the poller would select this pending mobile doc.
+    expect(await repos.notifications.countDocuments(pendingFilter('fcm'))).toBe(1);
+
+    await processFcmJob(deps({ cfg: shadow, fcm }), notif._id.toHexString());
+
+    expect(fcm.calls).toHaveLength(0); // nothing sent in shadow
+    const after = await repos.notifications.findOne({ _id: notif._id });
+    expect(after?.notified).toBeUndefined(); // old system's flag untouched
+    expect(after?.processedBy?.fcm).toBeInstanceOf(Date);
+    // After: the poller no longer selects it -> no re-enqueue.
+    expect(await repos.notifications.countDocuments(pendingFilter('fcm'))).toBe(0);
   });
 
   it('kill switch stops sending', async () => {
@@ -133,6 +157,7 @@ describe('FCM handler', () => {
     expect(fcm.calls).toHaveLength(0);
     const after = await repos.notifications.findOne({ _id: notif._id });
     expect(after?.notified).toBeUndefined();
+    expect(after?.processedBy?.fcm).toBeInstanceOf(Date); // terminal: won't be re-enqueued
   });
 
   it('enforces the per-user/day limit', async () => {

test/poller.test.ts

@@ -28,8 +28,8 @@ function fakeQueues() {
 
 describe('pendingFilter', () => {
   it('uses the CORRECT field names (old cron had a typo)', () => {
-    expect(pendingFilter('fcm')).toEqual({ type: 'mobile', notified: { $ne: true } });
+    expect(pendingFilter('fcm')).toEqual({ type: 'mobile', notified: { $ne: true }, 'processedBy.fcm': { $exists: false } });
-    expect(pendingFilter('email')).toEqual({ type: 'web', emailNotified: { $ne: true } });
+    expect(pendingFilter('email')).toEqual({ type: 'web', emailNotified: { $ne: true }, 'processedBy.email': { $exists: false } });
   });
 });