diff --git a/internal/delivery/dispatcher.go b/internal/delivery/dispatcher.go index 93fc47d..22b7724 100644 --- a/internal/delivery/dispatcher.go +++ b/internal/delivery/dispatcher.go @@ -160,12 +160,20 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim RepeatUntilAck: ch == ChannelTelegram && c.Severity >= loop.Sev4, Ts: now, } + s = minimalForAway(s) sink := d.sinkFor(ch) if sink == nil { 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 @@ -225,12 +233,17 @@ func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, n Summary: pr.Summary, Ts: now, } + s = minimalForAway(s) sink := d.sinkFor(ch) if sink == nil { 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 @@ -315,9 +328,13 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. RepeatUntilAck: true, Ts: now, } + 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) @@ -329,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: @@ -342,20 +377,44 @@ func (d *Dispatcher) sinkFor(ch Channel) Sink { } } +// GenericAwayMessage — what an away channel gets when the phraser gave us no +// summary. no gendered forms, so it stays right whoever reads it. +const GenericAwayMessage = "что-то требует внимания" + +// isAway — this channel leaves the box, so it only ever gets a minimal body. +func isAway(ch Channel) bool { + return ch == ChannelNtfy || ch == ChannelTelegram +} + // messageForChannel — away channels get the minimal summary (no shoulder-surf // exfil — "disk low on homesrv," not detail); voice gets the full body (local). -// a missing summary falls back to body — a terse full message is better than -// no message, and the phraser should have produced a summary for away-bound -// severities. this is the "minimal body" rule from the spec, enforced at the -// last mile so a phraser bug can't accidentally exfil via the relay. +// an empty summary must NOT fall back to the body: the resident model is small +// and drops fields often, and the away path crosses the "never phones home" +// boundary. so we send a fixed generic line plus the rule name instead. voice +// is local, so it keeps the full body. func messageForChannel(s Sendable) string { - switch s.Channel { - case ChannelNtfy, ChannelTelegram: - if s.Summary != "" { - return s.Summary - } - return s.Body - default: + if !isAway(s.Channel) { return s.Body } + if s.Summary != "" { + return s.Summary + } + if s.RuleName != "" { + return GenericAwayMessage + ": " + s.RuleName + } + return GenericAwayMessage +} + +// minimalForAway — strips detail from a Sendable bound for an away channel +// before any sink sees it. the sinks pick Summary themselves too, but this is +// where the boundary actually is: a sink added later must not be able to leak +// the full body just by reading the wrong field. +func minimalForAway(s Sendable) Sendable { + if !isAway(s.Channel) { + return s + } + msg := messageForChannel(s) + s.Body = msg + s.Summary = msg + return s } diff --git a/internal/delivery/dispatcher_test.go b/internal/delivery/dispatcher_test.go index 8901248..e3d79a0 100644 --- a/internal/delivery/dispatcher_test.go +++ b/internal/delivery/dispatcher_test.go @@ -784,3 +784,222 @@ func TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend(t *testing.T) { t.Fatalf("send should still happen despite outbox begin failure: sends=%v out=%v", voice.sends, out) } } + +// ----------------------- away channels carry no detail ----------------------- + +// panicSink — a broken sink. models the #369 case: the sink blows up mid-send. +type panicSink struct{ calls int } + +func (p *panicSink) Send(_ context.Context, _ Sendable) error { + p.calls++ + panic("sink is broken") +} + +// TestAwaySendsGenericLineWhenSummaryEmpty — #368. The phraser is a small +// model and drops fields often. An empty Summary must NOT put the full body +// on a channel that leaves the box; the away sendable gets a fixed generic +// line plus the rule name instead. +func TestAwaySendsGenericLineWhenSummaryEmpty(t *testing.T) { + ntfy := &fakeSink{} + rec := &fakeNudgeRecorder{} + d := NewDispatcher(Config{Ntfy: ntfy, Nudges: rec}) + + body := "disk /dev/sda1 at 96%, 2.1G free, largest offender /var/lib/docker" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("disk-low", loop.Sev3, store.Away), + Body: body, + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(ntfy.sends) != 1 { + t.Fatalf("want 1 ntfy send, got %d", len(ntfy.sends)) + } + want := GenericAwayMessage + ": disk-low" + got := messageForChannel(ntfy.sends[0]) + if got != want { + t.Fatalf("away message: want %q, got %q", want, got) + } + if ntfy.sends[0].Body == body { + t.Fatal("away sendable still carries the full body") + } + if len(rec.rows) != 1 || rec.rows[0].message != want { + t.Fatalf("recorded message: want %q, got %+v", want, rec.rows) + } +} + +// TestSev4AwaySendableCarriesNoDetail — the rule is enforced at the +// dispatcher, not in each sink. A sink added later must not be able to leak +// the body just by reading the wrong field, so neither field may hold detail. +func TestSev4AwaySendableCarriesNoDetail(t *testing.T) { + tg := &fakeSink{} + d := NewDispatcher(Config{Telegram: tg, Ack: newFakeAck()}) + + body := "backup job failed: rsync exit 23 on /home/kami, see /var/log/backup.log" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("backup-failed", loop.Sev4, store.Away), + Body: body, + Summary: "бэкап не прошёл", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(tg.sends) != 1 { + t.Fatalf("want 1 telegram send, got %d", len(tg.sends)) + } + s := tg.sends[0] + if s.Body != "бэкап не прошёл" || s.Summary != "бэкап не прошёл" { + t.Fatalf("away sendable should hold only the summary, got body=%q summary=%q", s.Body, s.Summary) + } +} + +// TestAwayKeepsANonEmptySummary — the normal path is untouched. +func TestAwayKeepsANonEmptySummary(t *testing.T) { + ntfy := &fakeSink{} + d := NewDispatcher(Config{Ntfy: ntfy}) + + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("cert", loop.Sev3, store.Away), + Body: "cert for maven.local expires in 3 days, issuer letsencrypt", + Summary: "сертификат истекает", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if got := messageForChannel(ntfy.sends[0]); got != "сертификат истекает" { + t.Fatalf("want the summary unchanged, got %q", got) + } +} + +// TestVoiceStillGetsTheFullBody — voice never leaves the box, so it keeps the +// full phrased message even when Summary is empty. +func TestVoiceStillGetsTheFullBody(t *testing.T) { + voice := &fakeSink{} + d := NewDispatcher(Config{Voice: voice}) + + body := "ты не пил воду четыре часа" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("water", loop.Sev1, store.Present), + Body: body, + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(voice.sends) != 1 || voice.sends[0].Body != body { + t.Fatalf("voice should get the full body, got %+v", voice.sends) + } + if got := messageForChannel(voice.sends[0]); got != body { + t.Fatalf("voice message: want %q, got %q", body, got) + } +} + +// TestReminderAwaySendsGenericLineWhenSummaryEmpty — the reminder path crosses +// the same boundary and has its own Sendable construction. +func TestReminderAwaySendsGenericLineWhenSummaryEmpty(t *testing.T) { + ntfy := &fakeSink{} + d := NewDispatcher(Config{Ntfy: ntfy}) + + _, err := d.DispatchReminder(context.Background(), PhrasedReminder{ + Decision: loop.ReminderDecision{ + Reminder: store.Reminder{ID: 5}, + State: loop.State{Now: refNow(), Presence: store.Away}, + }, + Body: "позвонить в клинику по поводу анализов", + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(ntfy.sends) != 1 { + t.Fatalf("want 1 ntfy send, got %d", len(ntfy.sends)) + } + if got := messageForChannel(ntfy.sends[0]); got != GenericAwayMessage { + t.Fatalf("away reminder message: want %q, got %q", GenericAwayMessage, got) + } +} + +// --------------------------- a panicking sink ------------------------------- + +// TestPanicMidSendResolvesTheAttempt — #369. A panic used to unwind past +// completeOutbox and leave the row pending forever, because reconciliation +// only runs at daemon startup and core is long-lived. +func TestPanicMidSendResolvesTheAttempt(t *testing.T) { + ob := &fakeOutbox{} + d := NewDispatcher(Config{Voice: &panicSink{}, Outbox: ob}) + + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("water", loop.Sev1, store.Present), + Body: "body", Summary: "sum", + }, refNow()) + if err != nil { + t.Fatalf("a panicking sink must not fail the dispatch: %v", err) + } + if len(ob.attempts) != 1 { + t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts)) + } + if ob.attempts[0].status != store.DeliveryFailed { + t.Fatalf("want status failed after a panic, got %q", ob.attempts[0].status) + } +} + +// TestPanicInOneSinkStillDeliversTheOther — sev4 present is voice + ntfy. One +// broken sink must not eat the other channel for the same nudge. +func TestPanicInOneSinkStillDeliversTheOther(t *testing.T) { + bad := &panicSink{} + ntfy := &fakeSink{} + ob := &fakeOutbox{} + d := NewDispatcher(Config{Voice: bad, Ntfy: ntfy, Outbox: ob}) + + out, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("backup-failed", loop.Sev4, store.Present), + Body: "long detailed body", Summary: "бэкап не прошёл", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if bad.calls != 1 { + t.Fatalf("want the voice sink called once, got %d", bad.calls) + } + if len(ntfy.sends) != 1 { + t.Fatalf("ntfy should still get the nudge, got %d sends", len(ntfy.sends)) + } + if len(out) != 1 || out[0].Sendable.Channel != ChannelNtfy { + t.Fatalf("want only the ntfy dispatch reported, got %+v", out) + } + if len(ob.attempts) != 2 { + t.Fatalf("want 2 outbox attempts, got %d", len(ob.attempts)) + } + if ob.attempts[0].status != store.DeliveryFailed || ob.attempts[1].status != store.DeliverySent { + t.Fatalf("want [failed, sent], got %q %q", ob.attempts[0].status, ob.attempts[1].status) + } +} + +// TestPanicInReminderSinkResolvesTheAttempt — the reminder path has its own +// send call, and a panic there must not leave the reminder marked fired. +func TestPanicInReminderSinkResolvesTheAttempt(t *testing.T) { + ob := &fakeOutbox{} + rc := &fakeReminderCompleter{} + d := NewDispatcher(Config{Voice: &panicSink{}, Reminders: rc, Outbox: ob}) + + out, err := d.DispatchReminder(context.Background(), PhrasedReminder{ + Decision: loop.ReminderDecision{ + Reminder: store.Reminder{ID: 9}, + State: loop.State{Now: refNow(), Presence: store.Present}, + }, + Body: "звонок", Summary: "звонок", + }, refNow()) + if err != nil { + t.Fatalf("a panicking sink must not fail the dispatch: %v", err) + } + if len(out) != 0 { + t.Fatalf("nothing was delivered, want no dispatches, got %+v", out) + } + if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryFailed { + t.Fatalf("want 1 attempt with status failed, got %+v", ob.attempts) + } + if len(rc.marked) != 0 { + t.Fatalf("reminder must stay pending after a panic, got %+v", rc.marked) + } +}