diff --git a/internal/delivery/dispatcher.go b/internal/delivery/dispatcher.go index f3ad35d..22b7724 100644 --- a/internal/delivery/dispatcher.go +++ b/internal/delivery/dispatcher.go @@ -166,7 +166,14 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim continue } attemptID := d.beginOutbox(ctx, "nudge", c.Rule.Name, 0, ch, messageForChannel(s), now) - if err := sink.Send(ctx, s); err != nil { + if err := safeSend(ctx, sink, s); err != nil { + if errors.Is(err, ErrSinkPanicked) { + // one broken sink must not eat the other channels for this + // nudge (sev4 present is voice + ntfy). the attempt is closed + // as failed and we move on. + d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + continue + } if errors.Is(err, ErrVoiceNoSession) { // voice was assumed reachable (presence=present) but no live // session exists — the presence guess was wrong. reroute through @@ -232,7 +239,11 @@ func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, n continue } attemptID := d.beginOutbox(ctx, "reminder", "", rd.Reminder.ID, ch, messageForChannel(s), now) - if err := sink.Send(ctx, s); err != nil { + if err := safeSend(ctx, sink, s); err != nil { + if errors.Is(err, ErrSinkPanicked) { + d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + continue + } if errors.Is(err, ErrVoiceNoSession) { // presence guess was wrong — reroute reminder to the away // channel (ntfy). voice is the only present channel, so nothing @@ -319,8 +330,11 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. } s = minimalForAway(s) attemptID := d.beginOutbox(ctx, "nudge", key, 0, ChannelTelegram, messageForChannel(s), now) - if err := d.cfg.Telegram.Send(ctx, s); err != nil { + if err := safeSend(ctx, d.cfg.Telegram, s); err != nil { d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + if errors.Is(err, ErrSinkPanicked) { + continue + } return out, fmt.Errorf("repeat send telegram %s: %w", key, err) } d.completeOutbox(ctx, attemptID, store.DeliverySent, now) @@ -332,6 +346,24 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. return out, nil } +// ErrSinkPanicked — a sink panicked mid-send. the send did not happen, so the +// attempt is recorded failed and never silently retried as if it had. +var ErrSinkPanicked = errors.New("delivery: sink panicked mid-send") + +// safeSend calls a sink and turns a panic into an error. without this a +// panicking sink unwinds past completeOutbox and leaves the delivery_attempts +// row pending forever — reconciliation only runs at daemon startup, and core +// is long-lived, so the row would sit there for weeks. +func safeSend(ctx context.Context, sink Sink, s Sendable) (err error) { + defer func() { + if r := recover(); r != nil { + log.Printf("dispatcher: PANIC in %s sink (this is a bug, fix the sink): %v", s.Channel, r) + err = fmt.Errorf("%w: %s: %v", ErrSinkPanicked, s.Channel, r) + } + }() + return sink.Send(ctx, s) +} + func (d *Dispatcher) sinkFor(ch Channel) Sink { switch ch { case ChannelVoice: