From 215aa331c57af7368f92d5bad8d9147e088a4df1 Mon Sep 17 00:00:00 2001
From: kami <kami@kvmx.ru>
Date: Fri, 31 Jul 2026 02:35:54 +0400
Subject: [PATCH] Recover from a panicking sink so the attempt is always closed
 (#369)

A panic in Send used to unwind past completeOutbox and leave the
delivery_attempts row pending forever, since reconciliation only runs at
startup. safeSend turns the panic into an error, logs it loudly, records the
attempt failed, and lets the other channels for the same nudge still go out.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ
---
 internal/delivery/dispatcher.go | 38 ++++++++++++++++++++++++++++++---
 1 file changed, 35 insertions(+), 3 deletions(-)

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: