kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Recover from a panicking sink so the attempt is always closed (#369)

Browse Source 
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
This commit is contained in:
kami
2026-07-31 02:35:54 +04:00
parent 859bbf750f
commit 215aa331c5
1 changed files with 35 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/delivery/dispatcher.go
+35 -3
View File
@@ -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: