diff --git a/README.md b/README.md index 48b20c2..2885ab2 100644 --- a/README.md +++ b/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. diff --git a/internal/poller/delivery.go b/internal/poller/delivery.go index 1beb98e..8ed4cc5 100644 --- a/internal/poller/delivery.go +++ b/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 { diff --git a/internal/poller/event_collection.go b/internal/poller/event_collection.go index 40cf21b..2ebd7c0 100644 --- a/internal/poller/event_collection.go +++ b/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 { diff --git a/internal/poller/poller.go b/internal/poller/poller.go index cbed597..c726f01 100644 --- a/internal/poller/poller.go +++ b/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) diff --git a/internal/poller/poller_test.go b/internal/poller/poller_test.go index e14ac29..87367ca 100644 --- a/internal/poller/poller_test.go +++ b/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