通知を outbox で冪等に届ける — 検知と送信を分け、送信直前に再確認する

通知を「条件を検知した瞬間にそのまま送る」設計にすると、検知から送信までの短い間に状況が変わって誤報になる。在庫が切れる、値が元に戻る、ユーザーがウォッチを外す——検知した数十秒〜数分後には、その通知はもう当てはまらないことがある。

個人でヤスゴロという価格ウォッチャーを Cloudflare Workers + D1 で作って公開した。安くなった消耗品を通知するものなので、「もう安くない通知」を送るのは一番やってはいけない失敗だ。そこで検知と送信を分け、検知は outbox に積むだけにして、送信直前にもう一度確かめてから送る形にした。この記事は、その outbox を D1 の状態機械としてどう組んだか——冪等な投入・送信直前の再確認・再送——に絞る。何を検知するか（買い時の判定）の中身には踏み込まない。

検知は outbox に「積むだけ」にする

検知側の仕事は、条件が成立したことを outbox に1行 INSERT するところで終わりにする。実際に送るのは別の工程（毎分のディスパッチ）に任せる。こうして検知と送信を時間的に切り離す。

投入は冪等にしておく。同じウォッチの同じトリガーが1日に何度検知されても、通知は1件に収束してほしい。D1（SQLite）の式インデックスと ON CONFLICT DO NOTHING で吸収する。

-- migration: 重複を1件に収束させる式インデックス
CREATE UNIQUE INDEX uq_outbox_dedup
  ON outbox (watch_id, trigger, date(detected_at), channel);

// 投入は冪等。ON CONFLICT で無視されたら false が返る
export async function enqueueOutbox(db: D1Database, input: EnqueueInput): Promise<boolean> {
  const res = await db.prepare(
    `INSERT INTO outbox
       (watch_id, user_id, channel, trigger, status, detected_at, detected_payload_json, retry_count, created_at)
     VALUES (?, ?, ?, ?, 'pending', ?, ?, 0, ?)
     ON CONFLICT (watch_id, trigger, date(detected_at), channel) DO NOTHING`
  ).bind(/* ... */).run();
  return (res.meta?.changes ?? 0) > 0; // 実挿入なら changes=1、dedup 無視なら 0
}

date(detected_at) を鍵に含めるのがポイントで、「同じ日の同じ検知」を1件に畳みつつ、翌日の再検知は別行として通す。検知が多重に走っても outbox が重複で膨らまない。

状態は5つ、遷移は一方向

outbox の各行は5状態を一方向に進む。積まれた pending が、送信直前の再確認を通れば sending → sent、崩れていれば dropped。送信が一時失敗したら retry、上限で failed。

flowchart LR
  pending -->|再照会で成立| sending
  pending -->|条件が崩れた| dropped
  sending -->|配信成功| sent
  sending -->|一時失敗| retry
  retry -->|再確認へ| pending
  retry -->|上限到達| failed

sent と dropped と failed が終端。dropped は異常ではなく正常系の終端で、「送るべきでないと分かったので送らなかった」を表す。エラーと区別しておくと、あとで「何件を正しく見送ったか」を数えられる。

送信直前に再確認して、崩れていたら送らない

ディスパッチは1件ずつ「再照会 → 検知条件が今も成立するか再判定 → 成立したものだけ送る」を回す。誤報を止めているのはこの工程だ。

// outbox 1件の処理（骨子）
async function processOutbox(o, now, deps) {
  // (0) 事前ガード: ウォッチ解除 / 対象停止 → dropped
  if (watchRemoved || listingPaused) { await markDropped(db, o.id, ...); return 'dropped'; }

  // (1) 送信直前に現在値を取り直す（取得失敗の種類で drop / retry を分ける）
  const r = await adapter.getOne(itemCode);
  if (!r.ok) {
    if (r.error.kind === 'empty') { await markDropped(db, o.id, 'out_of_stock'); return 'dropped'; }
    await scheduleRetry(db, o, `refetch_${r.error.kind}`); return 'retry';
  }

  // (2) 検知時の条件が「今も成立」か再判定（中身はドメイン依存）
  const recheck = recheckTrigger(/* 現在値 */);
  if (!recheck.ok) { await markDropped(db, o.id, 'condition_lost'); return 'dropped'; }

  // (3) 成立したものだけ、最新値で文面を作って配信
  await markSending(db, o.id, o.channel);
  const result = await deps.deliver(o.userId, buildPayload(/* 再照会後の最新値 */));
  // ... sent / retry / dropped
}

再判定の中身（どのトリガーが「まだ成立」か）は扱うサービスによって変わる。共通するのは、検知時に控えた値で送らず、送信直前に取り直した最新値で断定すること。文面に出す数字と、送っていいと判断した根拠を、同じ最新値にそろえる。ズレると「本文の価格と実際の価格が違う通知」になる。

「通知済み」は送ったときだけ進める

冪等性でもう一つ効くのが、last_notified（このウォッチに最後に通知した基準）を sent のときだけ更新するというルールだ。dropped や retry では触らない。

理由は、drop や retry で基準を動かすと、次の正当な検知を「もう通知済み」と誤って弾いてしまうから。送っていないのに送った印を付けない。送れたときだけ前に進める。これで「送信は一度きり、でも送れなかったものは正しく次につながる」が両立する。

一時失敗は指数バックオフで retry、上限で failed

再照会のレート制限・ネットワーク・配信の一時失敗は、失敗として捨てず retry に落とす。retry_count を増やし、指数バックオフで次の実行まで待たせ、上限で failed にする。

export function backoffMs(retryCount: number): number {
  return Math.min(2 ** retryCount * 30_000, 30 * 60_000); // 30s, 60s, 120s … 最大30分
}

ここで実装上の妥協が1つ。スキーマに next_attempt_at / last_error の列を足さず、再送予約時刻はペイロード JSON の中の _retry キーに内包した。検知スナップショット本体（listingId 等）は不変のまま、予約情報だけ相乗りさせる形だ。取得キューは SQL で粗く pending/retry を引き、アプリ側で _retry.nextAttemptAt が未到来のものを外す。列を増やさずにバックオフを尊重できる。

export async function pendingOutbox(db, limit, now) {
  const rows = await db.prepare(
    `SELECT * FROM outbox WHERE status IN ('pending','retry') ORDER BY created_at ASC LIMIT ?`
  ).bind(limit * 2).all();
  const ready = rows.results.filter((r) => {
    if (r.status === 'pending') return true;
    const meta = parseDetectedPayload(r.detected_payload_json)._retry;
    return !meta?.nextAttemptAt || meta.nextAttemptAt <= now; // バックオフ未到来は除外
  });
  return ready.slice(0, limit);
}

まとめ

通知を確実に届けたいときほど、「検知した瞬間に送る」をやめて outbox を挟むのが効いた。積むのは冪等に、送るのは送信直前の再確認を通ったものだけ、送れたときだけ通知済みを前に進める。状態を pending → sending → sent / dropped / retry → failed の一方向に倒すと、「送るべきでないから送らなかった（dropped）」と「送りたいがまだ送れていない（retry）」がコードでも運用でも区別できる。

この outbox を毎分どう回しているかはWorkers の cron 1本で複数ジョブを回す話に、送信チャネル（ログイン不要の Web Push）側は匿名 device_token で Web Push を届ける話に書いた。動いているものはヤスゴロで確かめてほしい。

よくある質問

なぜ検知した瞬間に送らず outbox を挟むの？

検知から送信までの間に状況が変わり、誤報になるからです。「条件を満たした」と検知しても、実際に送るまでの数十秒〜数分で在庫が消えたり条件が戻ったりし得ます。そのまま送ると「もう当てはまらない通知」が飛びます。検知は outbox に1行積むだけにして、送信直前に再照会・再判定する分離にすると、古くなった通知を送らずに落とせます。

同じ通知が二重に飛ばないのはなぜ？

投入が冪等だからです。enqueue は UNIQUE(watch_id, trigger, date(detected_at), channel) の式インデックスに対して INSERT … ON CONFLICT DO NOTHING で入れるので、同じ日・同じ watch・同じトリガー・同じチャネルの重複は1件に収束します。加えて「通知済み」の基準（last_notified）は sent のときだけ進めるので、dropped や retry は次の正当な検知を妨げません。

送信直前に条件が崩れていたらどうなる？

送らずに dropped にします（正常系の終端）。送信直前の再照会で在庫が消えていたり検知時の条件が今は成立しなければ、condition_lost として落とし、通知済みの基準も更新しません。文面は検知時に控えた値ではなく、再照会で確定した最新の値で作ります。断定する数字と実際に送る根拠をそろえるためです。

送信の一時失敗はどう扱う？

retry です。再照会のレート制限やネットワーク、配信の一時失敗は、retry_count を増やして指数バックオフ（30秒→60秒→…最大30分）で待ち、上限（OUTBOX_MAX_RETRY）に達したら failed で終端します。スキーマに next_attempt_at 列を足さず、再送予約時刻はペイロード JSON の _retry に内包しました。取得キューはこの予約時刻を見て、まだ来ていない retry を除外します。