Merge branch 'worktree-agent-afe3f2ec18b2b8497' into overnight-jul31
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user