Notification Pipeline Poll vs Delivery

Notification Pipeline — Poll vs Delivery

Date: 2026-07-07

Audience: developers and agents implementing or reviewing the unified notification pipeline.

Related:

• unified-notification-pipeline-plan.md — §4.1, §27 (normative spec)
• notification-pipeline-usage.md — how to add feeds

---

The concern

BENECO monitors poll frequently (e.g. every 60 seconds). Each poll can update feed tables (power_outages, scheduled_outages) with newer area text, status, cause, or PayloadHash.

Delivery — fan-out to many subscriptions, ntfy HTTP calls, retries — can take much longer than one poll interval.

During that window, without explicit design rules, you risk:

Risk	Example
Spam	Same outage triggers a new push on every poll because area/status changed
Stale content	Message body built at minute 1 but sent at minute 5 with outdated area text
Duplicate sends	Slow fan-out retries or overlapping poll cycles deliver the same logical event twice

This document explains how the pipeline mitigates those risks and where in code each rule lives.

---

Design principle: separate poll, signal, and snapshot

┌─────────────────────────────────────────────────────────────────┐
│  POLL (every ~60s)                                            │
│  - Fetch BENECO feed                                            │
│  - Upsert power_outages / scheduled_outages  ← analytics + truth │
│  - Detect signals (cheap, no ntfy)                              │
└────────────────────────────┬────────────────────────────────────┘
                             │ NotificationEvent = signal only
                             â–¼
┌─────────────────────────────────────────────────────────────────┐
│  DELIVER (may take minutes)                                     │
│  - Ledger check → skip if already sent                          │
│  - Re-fetch latest row from DB  ← fresh snapshot                │
│  - Render title/body                                            │
│  - ntfy / email → write ledger on success only                  │
└─────────────────────────────────────────────────────────────────┘

Concept	What it is	When it is created
Poll data	Rows in power_outages / scheduled_outages	Every poll
Signal	NotificationEvent (FeedId, EventType, SourceKey)	Only on real state transitions
Snapshot	Rendered title/body sent to subscribers	Immediately before each delivery

Rule: Never queue or cache rendered message text between detect and send. Signals may be passed to the pipeline in memory; snapshots are always built at delivery time.

---

Timeline example

T+0:00  Poll — new outage source_id=42, signal NewOutage emitted
T+0:00  Pipeline starts (500 subscriptions; slow ntfy)
T+1:00  Poll — area text corrected in power_outages (PayloadHash changed)
T+2:00  Poll — status refreshed in power_outages
T+3:00  Subscription #1 delivered — body uses area from T+2:00 ✓
T+4:00  Poll — no new signal (outage already notified) ✓
T+5:00  Subscription #50 delivered — body uses latest row at T+5:00 ✓
T+5:01  Subscription #1 attempted again — ledger hit → skip ✓

Without mitigation, T+1:00 and T+2:00 polls could emit more NewOutage signals or send bodies frozen at T+0:00.

---

Three layers of mitigation (all required)

Layer 1 — Source change rules (anti-spam at detect)

Where: PowerOutageChangeDetector, ScheduledOutageChangeDetector, ScheduledOutageDigestPolicy, ScheduledOutageReminderPolicy in Core Called from: SyncPowerOutagesAndPublishEvents, SyncScheduledOutagesAndPublishEvents
When: After feed state is persisted, before ProcessNotificationEvents

Source-owned Core policies decide whether a real source change exists. Application then maps those source changes into NotificationEvent signals.

Emit a signal only when today’s production logic would call Publish*Async — not on every poll.

Power outages — emit once:

Signal	Emit when
NewOutage	New ongoing row, or existing ongoing with OutageNotifiedAt / ledger miss
Restoration	time_restored set, restoration path
RestoredAnnouncement	First-seen-already-restored path

Do not emit when an ongoing outage is merely updated:

DB change on existing source_id	New signal?
area, cause, status text changed	No
PayloadHash changed	No
last_seen_at refreshed	No
time_restored newly set	Yes (once per event type)

UpdateFromFeedAsync may run every minute; the detector must not treat that as a new outage.

Order in monitor:

await PersistFeedFromPollAsync(...);           // always
var changes = PowerOutageChangeDetector.Detect(...);      // source-owned rules
var signals = PowerOutageNotificationEvents.FromChanges(...);  // notification mapping
if (signals.Count > 0)
    await processNotificationEvents.ProcessAsync(signals, ct);

---

Layer 2 — Ledger dedup (anti-spam at delivery)

Where: notification_delivery_ledger + INotificationDeliveryLedgerStore
Checked in: ProcessNotificationEvents before render

Unique key:

(subscription_id, feed_id, source_key, event_type)

Behavior	Detail
Ledger row exists	Skip delivery for that subscription — even if polls ran again
Delivery fails	No ledger write → safe retry on next poll if signal still valid
Delivery succeeds	Write ledger → never send same logical event twice to same sub

This protects against slow fan-out, overlapping poll cycles, and retries.

---

Layer 3 — Fresh render (anti-stale at delivery)

Where: INotificationEventSourceResolver
Called from: ProcessNotificationEvents, immediately before INotificationRenderer.Render

For row-based events (source_key = BENECO numeric id), reload the latest feed row from the store:

// ProcessNotificationEvents (simplified)
if (await _ledgerStore.ExistsAsync(ledgerKey, ct))
    continue;

var freshEvent = await _sourceResolver.ResolveAsync(signal, ct);
if (freshEvent is null)
    continue; // row deleted — no send, no ledger

var rendered = renderer.Render(freshEvent, subscription);
await _deliveryRouter.DeliverAsync(rendered, subscription.Target, ct);

Power outage resolver (PowerOutageNotificationEventSourceResolver):

var row = await _powerOutageStore.GetBySourceIdAsync(sourceId, ct);
if (row is null) return null;

return signal with
{
    Fields = PowerOutageFields.From(row),
    SourcePayload = row
};

Digest events (weekly-2026-07-07, etc.) use the signal payload as-is — no single source_id row.

Why: Polls between signal and send update the DB. Subscribers get the latest area/status at their delivery instant, not text from when the signal was first detected.

---

ntfy SequenceId (secondary guard)

Where: *NotificationRenderer classes

Use stable ids tied to logical event, not poll count:

power-outage-{sourceId}
power-restored-{sourceId}
power-restored-announced-{sourceId}

Do not include poll number, PayloadHash, or timestamp in SequenceId.

ntfy may treat duplicate sequence publishes as updates; the ledger remains the primary dedup source.

---

What developers must not do

Anti-pattern	Why it breaks
Build message body in Detect and pass strings to pipeline	Stale text when delivery lags
Emit NewOutage when only PayloadHash changed	Spam every poll
Write ledger before confirmed delivery	Skips legitimate retries
Check ledger after render	Wastes work; race under load
Let clients set ntfy topic	Security; unrelated but same pipeline
Use SourcePayload from signal without resolver	Same as caching body at detect time

---

Component map

Component	Layer	Project
PowerOutageChangeDetector	Power outage source change rules	Obaki.Core
ScheduledOutageChangeDetector	Scheduled outage source change rules	Obaki.Core
ScheduledOutageDigestPolicy / ScheduledOutageReminderPolicy	Scheduled digest and reminder timing rules	Obaki.Core
*NotificationEvents	Map source changes to NotificationEvent signals	Obaki.Application
SyncPowerOutagesAndPublishEvents	Persist → detect → pipeline	Obaki.Application
ProcessNotificationEvents	Orchestration	Obaki.Application
INotificationEventSourceResolver	Fresh render	Obaki.Core contract
PowerOutageNotificationEventSourceResolver	Fresh render	Obaki.Application
INotificationDeliveryLedgerStore	Ledger dedup	Obaki.Core / Infrastructure
DatabaseNotificationDeliveryLedgerStore	Ledger persistence	Obaki.Infrastructure
*NotificationRenderer	Snapshot build	Obaki.Application
NtfyNotificationDeliveryRouter	Send	Obaki.Infrastructure

---

Tests developers should add or maintain

Test	Proves
PowerOutageChangeDetector_does_not_emit_when_only_payload_hash_changes	Layer 1 — no spam from polls
ProcessNotificationEvents_skips_when_ledger_exists	Layer 2 — no duplicate delivery
ProcessNotificationEvents_uses_latest_row_at_render	Layer 3 — area updated in DB → body matches new text
Resolver_returns_null_when_row_deleted	No send, no ledger for missing source
Failed_delivery_does_not_write_ledger	Retry still possible on next poll

Example for Layer 3:

// Arrange: signal with area "Old"
// Update power_outages row to area "New" before pipeline runs
// Act: ProcessNotificationEvents
// Assert: ntfy message contains "New", not "Old"

---

Optional future: batched delivery interval

v1 runs detect + pipeline on the same poll cycle. Fan-out slowness is handled by fresh render per subscription, not by delaying detect.

If you later add DeliveryIntervalSeconds (e.g. poll 60s, deliver every 300s):

• Still persist every poll
• On delivery tick, re-scan DB for candidates (ledger miss), do not rely only on in-memory signals from one poll
• Layers 1–3 still apply

See unified-notification-pipeline-plan.md §27.4.

---

Quick checklist (PR review)

- [ ] Monitor persists feed state before source change detection
- [ ] `PowerOutageChangeDetector` does not re-emit NewOutage on PayloadHash / text-only updates
- [ ] ProcessNotificationEvents checks ledger before Render
- [ ] INotificationEventSourceResolver called before every Render for row-based events
- [ ] Ledger written only after successful delivery
- [ ] SequenceId stable per source + event type
- [ ] Tests cover stale-data and duplicate-send scenarios

---

Summary

Concern	Mitigation	Code hook
Spam from frequent polls	Emit signals only on source-owned state transitions	*ChangeDetector / *NotificationEvents
Duplicate pushes	Ledger per subscription + event	INotificationDeliveryLedgerStore
Stale area/status in message	Re-read DB before render	INotificationEventSourceResolver
Failed send retry storm	No ledger on failure	ProcessNotificationEvents

Polling every minute keeps data fresh. Delivery can be slow. Signals are stable; snapshots are always current.