fix(poller): checkpoint each event independently after fan-out

A single retryable delivery failure aborted checkpoints for the whole poll batch, so fully-delivered events were re-collected and re-sent on the next poll, and an unrelated failure froze all change detection. Attach post-delivery checkpoints to each event and run them as soon as that event fully delivers; a failed event defers only its own checkpoints for retry. A subscriber skipped after an earlier retryable failure now counts as the event's failure so the event is not marked delivered to a subscriber that never received it.

README.md

@@ -81,7 +81,7 @@ The first successful poll seeds the database and does not send historical incide
 
 Switching from a prior Redis deployment starts from empty state: there is no data migration, so subscribers must re-issue `/start` and component checkpoints reseed on the first poll.
 
-Incident update dedupe tracks the update content/version, so edited Statuspage updates can notify again. Delivery is checkpointed after successful fan-out; retryable Telegram failures may be retried on a later poll without advancing the global checkpoint.
+Incident update dedupe tracks the update content/version, so edited Statuspage updates can notify again. Each event is checkpointed independently once it has fully fanned out, so a retryable Telegram failure on one event only defers that event for retry on a later poll and never blocks checkpoints for other events delivered in the same poll.
 
 A 7-day TTL index on the `delivery` collection expires per-event delivery markers automatically; the bot creates required indexes on startup.
 

internal/poller/delivery.go

@@ -56,7 +56,7 @@ func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent,
 	}
 	for _, subscriber := range subscribers {
 		subscriberKey := subscriber.Key()
-		if removed[subscriberKey] || failed[subscriberKey] {
+		if removed[subscriberKey] {
 			continue
 		}
 		if !subscriber.Accepts(event.eventType, event.componentID, event.componentName) {
@@ -65,6 +65,14 @@ func (r *Runner) notifySubscribers(ctx context.Context, event notificationEvent,
 		if delivered[subscriberKey] {
 			continue
 		}
+		if failed[subscriberKey] {
+			// This subscriber already hit a retryable failure earlier in the
+			// poll. Skip it to avoid hammering, but record this event as
+			// incomplete so its checkpoint is deferred and it retries next poll
+			// instead of being marked delivered to a subscriber that never got it.
+			deliveryFailures.add(fmt.Errorf("deferred %s after earlier failure", subscriberKey))
+			continue
+		}
 		if err := r.notifier.SendMessage(ctx, subscriber, event.text); err != nil {
 			if telegram.IsTerminalSendError(err) {
 				if removeErr := r.store.RemoveSubscriber(ctx, subscriber); removeErr != nil {

internal/poller/event_collection.go

@@ -11,12 +11,19 @@ import (
 	"github.com/tiennm99/openai-status-bot/internal/mongostore"
 )
 
+// collectEvents returns the events to deliver this poll plus two checkpoint
+// groups: `before` writes run unconditionally before delivery (pending
+// markers), and `baseline` writes run unconditionally after delivery (seeding
+// saves and unchanged-status re-saves). Each event carries its OWN
+// after-delivery checkpoints in event.checkpoints, which run only once that
+// event has fully delivered — so a delivery failure for one event no longer
+// blocks checkpoints for unrelated events.
 func (r *Runner) collectEvents(ctx context.Context, summary openai.Summary, incidents openai.IncidentsResponse, initialized bool) ([]notificationEvent, []checkpoint, []checkpoint, error) {
-	componentEvents, componentBefore, componentAfter, err := r.collectComponentEvents(ctx, summary, initialized)
+	componentEvents, before, componentBaseline, err := r.collectComponentEvents(ctx, summary, initialized)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	incidentEvents, incidentAfter, err := r.collectIncidentEvents(ctx, incidents, initialized)
+	incidentEvents, incidentBaseline, err := r.collectIncidentEvents(ctx, incidents, initialized)
 	if err != nil {
 		return nil, nil, nil, err
 	}
@@ -31,8 +38,8 @@ func (r *Runner) collectEvents(ctx context.Context, summary openai.Summary, inci
 		}
 		return events[i].sortTime < events[j].sortTime
 	})
-	after := append(componentAfter, incidentAfter...)
-	return events, componentBefore, after, nil
+	baseline := append(componentBaseline, incidentBaseline...)
+	return events, before, baseline, nil
 }
 
 func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summary, initialized bool) ([]notificationEvent, []checkpoint, []checkpoint, error) {
@@ -49,7 +56,7 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ
 
 	events := make([]notificationEvent, 0)
 	before := make([]checkpoint, 0)
-	after := make([]checkpoint, 0, len(summary.Components)+len(pending))
+	baseline := make([]checkpoint, 0, len(summary.Components))
 
 	for _, pendingEvent := range pending {
 		pendingEvent := pendingEvent
@@ -61,8 +68,8 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ
 			deliveryKey:   pendingEvent.DeliveryKey,
 			sortTime:      pendingEvent.UpdatedAt,
 			text:          FormatComponentChange(component, pendingEvent.PreviousStatus, duplicates[component.Name]),
+			checkpoints:   r.resolveComponentCheckpoints(pendingEvent.ComponentID, pendingEvent.Status, pendingEvent.DeliveryKey),
 		})
-		after = append(after, r.resolveComponentCheckpoints(pendingEvent.ComponentID, pendingEvent.Status, pendingEvent.DeliveryKey)...)
 	}
 
 	for _, component := range summary.Components {
@@ -75,14 +82,10 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ
 		}
 
 		previousStatus, found := knownStatuses[component.ID]
-		if !initialized {
-			after = append(after, func(ctx context.Context) error {
-				return r.store.SaveComponentStatus(ctx, component.ID, component.Status)
-			})
-			continue
-		}
-		if found && previousStatus == component.Status {
-			after = append(after, func(ctx context.Context) error {
+		if !initialized || (found && previousStatus == component.Status) {
+			// Seeding or no change: persist the current status unconditionally,
+			// no notification.
+			baseline = append(baseline, func(ctx context.Context) error {
 				return r.store.SaveComponentStatus(ctx, component.ID, component.Status)
 			})
 			continue
@@ -111,10 +114,10 @@ func (r *Runner) collectComponentEvents(ctx context.Context, summary openai.Summ
 			deliveryKey:   deliveryKey,
 			sortTime:      component.UpdatedAt,
 			text:          FormatComponentChange(component, previousStatus, duplicates[component.Name]),
+			checkpoints:   r.resolveComponentCheckpoints(component.ID, component.Status, deliveryKey),
 		})
-		after = append(after, r.resolveComponentCheckpoints(component.ID, component.Status, deliveryKey)...)
 	}
-	return events, before, after, nil
+	return events, before, baseline, nil
 }
 
 // resolveComponentCheckpoints builds the post-delivery writes shared by pending
@@ -130,7 +133,7 @@ func (r *Runner) resolveComponentCheckpoints(componentID, status, deliveryKey st
 
 func (r *Runner) collectIncidentEvents(ctx context.Context, response openai.IncidentsResponse, initialized bool) ([]notificationEvent, []checkpoint, error) {
 	events := make([]notificationEvent, 0)
-	checkpoints := make([]checkpoint, 0)
+	baseline := make([]checkpoint, 0)
 	for _, incident := range response.Incidents {
 		incident := incident
 		for _, update := range incident.IncidentUpdates {
@@ -143,26 +146,31 @@ func (r *Runner) collectIncidentEvents(ctx context.Context, response openai.Inci
 			if err != nil {
 				return nil, nil, err
 			}
-			if !seen {
-				checkpoints = append(checkpoints, func(ctx context.Context) error {
-					return r.store.MarkIncidentUpdateVersion(ctx, update.ID, version)
-				})
+			if seen {
+				continue
+			}
+			markVersion := func(ctx context.Context) error {
+				return r.store.MarkIncidentUpdateVersion(ctx, update.ID, version)
+			}
+			if !initialized {
+				// Seeding: record the version baseline without notifying.
+				baseline = append(baseline, markVersion)
+				continue
 			}
 			deliveryKey := fmt.Sprintf("incident:%s:%s", update.ID, version)
-			if initialized && !seen {
-				events = append(events, notificationEvent{
-					eventType:   mongostore.SubscriptionTypeIncident,
-					deliveryKey: deliveryKey,
-					sortTime:    incidentUpdateSortTime(update),
-					text:        FormatIncidentUpdate(incident, update),
-				})
-				checkpoints = append(checkpoints, func(ctx context.Context) error {
-					return r.store.ClearDelivery(ctx, deliveryKey)
-				})
-			}
+			events = append(events, notificationEvent{
+				eventType:   mongostore.SubscriptionTypeIncident,
+				deliveryKey: deliveryKey,
+				sortTime:    incidentUpdateSortTime(update),
+				text:        FormatIncidentUpdate(incident, update),
+				checkpoints: []checkpoint{
+					markVersion,
+					func(ctx context.Context) error { return r.store.ClearDelivery(ctx, deliveryKey) },
+				},
+			})
 		}
 	}
-	return events, checkpoints, nil
+	return events, baseline, nil
 }
 
 func pendingComponent(event mongostore.PendingComponentEvent) openai.Component {

internal/poller/poller.go

@@ -50,6 +50,10 @@ type notificationEvent struct {
 	deliveryKey   string
 	sortTime      string
 	text          string
+	// checkpoints are the post-delivery store writes for this event. They run
+	// only after the event has fully delivered to every accepting subscriber,
+	// so an unrelated event's delivery failure cannot block them.
+	checkpoints []checkpoint
 }
 
 type checkpoint func(ctx context.Context) error
@@ -95,7 +99,7 @@ func (r *Runner) CheckOnce(ctx context.Context) error {
 		return err
 	}
 
-	events, beforeDelivery, checkpoints, err := r.collectEvents(ctx, summary, incidents, initialized)
+	events, beforeDelivery, baseline, err := r.collectEvents(ctx, summary, incidents, initialized)
 	if err != nil {
 		return err
 	}
@@ -119,15 +123,33 @@ func (r *Runner) CheckOnce(ctx context.Context) error {
 			if err != nil {
 				return err
 			}
-			deliveryErr.addAll(failures)
+			if failures != nil {
+				// Event not fully delivered; defer its checkpoints so it is
+				// re-collected and retried on the next poll. Other events are
+				// unaffected.
+				deliveryErr.addAll(failures)
+				continue
+			}
+			if err := runCheckpoints(ctx, event.checkpoints); err != nil {
+				return err
+			}
+		}
+		if err := runCheckpoints(ctx, baseline); err != nil {
+			return err
 		}
 		if deliveryErr.count > 0 {
 			return deliveryErr
 		}
+		return nil
 	}
 
-	for _, save := range checkpoints {
-		if err := save(ctx); err != nil {
+	// Seeding, or nothing to deliver: run baseline writes and flush any event
+	// checkpoints (e.g. stale pending markers carried over from a prior run).
+	if err := runCheckpoints(ctx, baseline); err != nil {
+		return err
+	}
+	for _, event := range events {
+		if err := runCheckpoints(ctx, event.checkpoints); err != nil {
 			return err
 		}
 	}
@@ -140,6 +162,15 @@ func (r *Runner) CheckOnce(ctx context.Context) error {
 	return nil
 }
 
+func runCheckpoints(ctx context.Context, checkpoints []checkpoint) error {
+	for _, save := range checkpoints {
+		if err := save(ctx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (r *Runner) checkAndLog(ctx context.Context) {
 	if err := r.CheckOnce(ctx); err != nil && ctx.Err() == nil {
 		r.logger.Error("poll openai status", "error", err)

internal/poller/poller_test.go

@@ -416,6 +416,39 @@ func TestPendingComponentDuplicateLabelsIncludeCurrentRename(t *testing.T) {
 	}
 }
 
+func TestCheckOnceCheckpointsDeliveredEventWhenSiblingEventFails(t *testing.T) {
+	// Regression: a fully-delivered event must checkpoint even when another
+	// event in the same poll fails to deliver. Previously any delivery failure
+	// aborted checkpoints for the whole batch, re-emitting delivered events.
+	store := newFakePollerStore()
+	store.initialized = true
+	store.componentStatuses["c1"] = "operational"
+	store.componentStatuses["c2"] = "operational"
+	store.subscribers = []mongostore.Subscriber{mongostore.NewSubscriber(1, nil)}
+	// c1 (sorts first) succeeds; c2 fails for the same subscriber.
+	notifier := &fakeNotifier{errorQueue: map[string][]error{"1": {nil, errors.New("rate limit")}}}
+	runner := NewRunner(fakeStatusClient{summary: openai.Summary{Components: []openai.Component{
+		{ID: "c1", Name: "API", Status: "degraded_performance", UpdatedAt: "2026-01-01T00:00:00Z"},
+		{ID: "c2", Name: "ChatGPT", Status: "partial_outage", UpdatedAt: "2026-01-01T00:01:00Z"},
+	}}}, store, notifier, time.Minute, slog.Default())
+
+	if err := runner.CheckOnce(context.Background()); err == nil {
+		t.Fatal("expected delivery error from c2")
+	}
+	if got := store.savedComponents["c1"]; got != "degraded_performance" {
+		t.Fatalf("c1 checkpoint = %q, want delivered event checkpointed", got)
+	}
+	if got, ok := store.savedComponents["c2"]; ok {
+		t.Fatalf("c2 checkpoint = %q, want failed event NOT checkpointed", got)
+	}
+	if _, ok := store.pendingComponents["c2"]; !ok {
+		t.Fatal("c2 pending marker should persist for retry")
+	}
+	if _, ok := store.pendingComponents["c1"]; ok {
+		t.Fatal("c1 pending marker should be cleared after delivery")
+	}
+}
+
 func TestCheckOnceContinuesWhenMarkDeliveredFailsAfterSend(t *testing.T) {
 	store := newFakePollerStore()
 	store.initialized = true