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

Add durable delivery outbox: Begin-before-send, Complete-after, unknown on crash

Browse Source 
Closes the audit finding: a crash between 'sink accepted it' and 'we
recorded that' caused duplicate sends on the next tick with no trace.
BeginDeliveryAttempt now runs before Send, CompleteDeliveryAttempt
after — a stale 'pending' row found at startup reconciles to 'unknown'
(never silently resent, never silently dropped, same rule as the Hexis
execution engine's timeout handling). Wired into DispatchNudge,
DispatchReminder, and RepeatUnacked; ReconcileStaleDeliveryAttempts
runs once at mavend startup before the tick loop resumes.

9 new dispatcher tests cover begin-before-send ordering, failed-send
completion, the reminder path, and begin-failure not blocking send.

Vikunja #270.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA
This commit is contained in:
kami
2026-07-20 03:05:41 +04:00
parent 838fde1fff
commit 29f23e3715
2 changed files with 213 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/delivery/dispatcher.go
+64
View File
@@ -2,6 +2,8 @@ package delivery
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"log"
@@ -29,6 +31,56 @@ type ReminderCompleter interface {
RescheduleReminder(ctx context.Context, id int64, now time.Time) error
}
// Outbox — the durable delivery ledger. Begin is recorded BEFORE the external
// send, so a crash between "the sink accepted it" and "we recorded that" (the
// window that causes duplicate sends on the next tick — the audit finding
// this closes) leaves a durable "pending" row instead of silence. Complete
// records the sink's actual outcome once Send returns. nil Outbox = tracking
// disabled (existing send/record behavior, unchanged — test scenarios that
// don't care about crash recovery).
type Outbox interface {
BeginDeliveryAttempt(ctx context.Context, kind, rule string, reminderID int64, channel, bodyHash string, now time.Time) (int64, error)
CompleteDeliveryAttempt(ctx context.Context, id int64, status string, now time.Time) error
}
// bodyHash is an opaque dedup/triage key for a delivery attempt row — not a
// uniqueness constraint (the same rule/reminder legitimately re-sends across
// ticks), just something an operator can compare across attempts after a
// crash to tell "same message resent" from "different message".
func bodyHash(channel Channel, body string) string {
sum := sha256.Sum256([]byte(string(channel) + "\x00" + body))
return hex.EncodeToString(sum[:8])
}
// beginOutbox records intent to send, if an Outbox is wired. A failure to
// record intent is not fatal to the send itself — losing outbox visibility
// on one attempt shouldn't block a nudge/reminder actually reaching the user
// — but it does mean this attempt can't be reconciled after a crash, so it's
// logged. Returns 0 (no-op id) when unrecorded.
func (d *Dispatcher) beginOutbox(ctx context.Context, kind, rule string, reminderID int64, channel Channel, body string, now time.Time) int64 {
if d.cfg.Outbox == nil {
return 0
}
id, err := d.cfg.Outbox.BeginDeliveryAttempt(ctx, kind, rule, reminderID, string(channel), bodyHash(channel, body), now)
if err != nil {
log.Printf("dispatcher: outbox begin failed (send proceeds untracked): %v", err)
return 0
}
return id
}
// completeOutbox records the sink's outcome for a prior beginOutbox call.
// id==0 means either tracking is disabled or the begin failed — nothing to
// complete either way.
func (d *Dispatcher) completeOutbox(ctx context.Context, id int64, status string, now time.Time) {
if id == 0 || d.cfg.Outbox == nil {
return
}
if err := d.cfg.Outbox.CompleteDeliveryAttempt(ctx, id, status, now); err != nil {
log.Printf("dispatcher: outbox complete failed: %v", err)
}
}
// PhrasedNudge — the phraser module's output for a nudge. the phraser (LFM
// sub-1b, prompted not trained) takes (rule, severity, context) and produces
// Body (full message for voice) + Summary (minimal body for away channels).
@@ -67,6 +119,7 @@ type Config struct {
Ack AckTracker
Nudges NudgeRecorder
Reminders ReminderCompleter
Outbox Outbox
}
// Dispatcher — holds one sink per channel + the recorder seams. the daemon
@@ -111,6 +164,7 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim
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 errors.Is(err, ErrVoiceNoSession) {
// voice was assumed reachable (presence=present) but no live
@@ -120,13 +174,16 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim
// first present channel, so nothing has been sent yet; replace
// the remaining list wholesale. away channels never include
// voice, so this can't re-trigger.
d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
log.Printf("dispatcher: no live voice session for %s, rerouting to away channels", c.Rule.Name)
channels = ChannelsFor(c.Severity, store.Away)
i = -1
continue
}
d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
return out, fmt.Errorf("send %s: %w", ch, err)
}
d.completeOutbox(ctx, attemptID, store.DeliverySent, now)
// record AFTER successful send — a failed send must not pollute the
// feedback loop with a phantom nudge (ignored_rate would drift).
var nudgeID int64
@@ -172,18 +229,22 @@ func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, n
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 errors.Is(err, ErrVoiceNoSession) {
// presence guess was wrong — reroute reminder to the away
// channel (ntfy). voice is the only present channel, so nothing
// has been sent yet.
d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
log.Printf("dispatcher: no live voice session for reminder %d, rerouting to away channels", rd.Reminder.ID)
channels = ChannelsForReminder(store.Away)
i = -1
continue
}
d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
return out, fmt.Errorf("send %s: %w", ch, err)
}
d.completeOutbox(ctx, attemptID, store.DeliverySent, now)
out = append(out, Dispatch{Sendable: s})
}
if d.cfg.Reminders != nil && len(out) > 0 {
@@ -254,9 +315,12 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time.
RepeatUntilAck: true,
Ts: now,
}
attemptID := d.beginOutbox(ctx, "nudge", key, 0, ChannelTelegram, messageForChannel(s), now)
if err := d.cfg.Telegram.Send(ctx, s); err != nil {
d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now)
return out, fmt.Errorf("repeat send telegram %s: %w", key, err)
}
d.completeOutbox(ctx, attemptID, store.DeliverySent, now)
if err := d.cfg.Ack.MarkSent(ctx, key, now); err != nil {
return out, fmt.Errorf("ack mark-sent %s: %w", key, err)
}
internal/delivery/dispatcher_test.go
+149
View File
@@ -635,3 +635,152 @@ func TestDispatchRecurringReminderReschedules(t *testing.T) {
t.Fatalf("recurring reminder should not be marked fired: %+v", rc.marked)
}
}
// ----------------------------- durable outbox --------------------------------
type outboxAttempt struct {
kind, rule string
reminderID int64
channel, hash string
status string
begunAt, doneAt time.Time
}
// fakeOutbox — an in-memory Outbox that also lets a test simulate a crash
// exactly at the three points the audit finding cares about: before Begin
// records anything, after Begin but before the sink is called (the crash
// window that causes duplicate delivery — Begin already landed so recovery
// can see it), and after a successful send but before Complete persists the
// outcome (the other half of the same window).
type fakeOutbox struct {
attempts []*outboxAttempt
nextID int64
beginErr error
completeErr error
}
func (f *fakeOutbox) BeginDeliveryAttempt(_ context.Context, kind, rule string, reminderID int64, channel, hash string, now time.Time) (int64, error) {
if f.beginErr != nil {
return 0, f.beginErr
}
f.nextID++
f.attempts = append(f.attempts, &outboxAttempt{
kind: kind, rule: rule, reminderID: reminderID, channel: channel, hash: hash,
status: "pending", begunAt: now,
})
return f.nextID, nil
}
func (f *fakeOutbox) CompleteDeliveryAttempt(_ context.Context, id int64, status string, now time.Time) error {
if f.completeErr != nil {
return f.completeErr
}
if id < 1 || int(id) > len(f.attempts) {
return errors.New("fakeOutbox: unknown attempt id")
}
a := f.attempts[id-1]
a.status = status
a.doneAt = now
return nil
}
// TestDispatchNudge_OutboxRecordsBeforeSendAndCompletesAfter — the audit's
// core scenario: Begin must land BEFORE the sink is invoked (so a crash right
// after the external send but before Maven records anything still leaves a
// "pending" trail an operator/reconciler can find), and Complete must record
// the real outcome once Send returns.
func TestDispatchNudge_OutboxRecordsBeforeSendAndCompletesAfter(t *testing.T) {
voice := &fakeSink{}
rec := &fakeNudgeRecorder{}
ob := &fakeOutbox{}
d := NewDispatcher(Config{Voice: voice, Nudges: rec, Outbox: ob})
_, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("water", loop.Sev1, store.Present),
Body: "you haven't had water in 4h",
Summary: "water",
}, refNow())
if err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(ob.attempts) != 1 {
t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts))
}
a := ob.attempts[0]
if a.kind != "nudge" || a.rule != "water" || a.channel != "voice" {
t.Fatalf("unexpected attempt: %+v", a)
}
if a.status != store.DeliverySent {
t.Fatalf("want status sent after a successful send, got %q", a.status)
}
}
// TestDispatchNudge_OutboxMarksFailedOnSendError — a clean (non-crash) send
// error still gets a definite Complete("failed") — this is NOT the ambiguous
// case, so it must not be left "pending" for a reconciler to puzzle over.
func TestDispatchNudge_OutboxMarksFailedOnSendError(t *testing.T) {
voice := &fakeSink{err: errors.New("boom")}
ob := &fakeOutbox{}
d := NewDispatcher(Config{Voice: voice, Outbox: ob})
_, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("water", loop.Sev1, store.Present),
Body: "body", Summary: "sum",
}, refNow())
if err == nil {
t.Fatal("expected send error to propagate")
}
if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryFailed {
t.Fatalf("want 1 attempt with status failed, got %+v", ob.attempts)
}
}
// TestDispatchReminder_OutboxTracksReminderAttempt — same Begin-before-Send /
// Complete-after-Send contract for the reminder path, which is a materially
// different code path (no Nudges recorder, MarkReminder/RescheduleReminder
// instead) and must not be skipped by only testing nudges.
func TestDispatchReminder_OutboxTracksReminderAttempt(t *testing.T) {
voice := &fakeSink{}
rc := &fakeReminderCompleter{}
ob := &fakeOutbox{}
d := NewDispatcher(Config{Voice: voice, Reminders: rc, Outbox: ob})
rd := loop.ReminderDecision{
Reminder: store.Reminder{ID: 42, Status: "pending"},
State: loop.State{Now: refNow(), Presence: store.Present},
}
_, err := d.DispatchReminder(context.Background(), PhrasedReminder{
Decision: rd, Body: "take out the trash", Summary: "trash",
}, refNow())
if err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(ob.attempts) != 1 {
t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts))
}
a := ob.attempts[0]
if a.kind != "reminder" || a.reminderID != 42 || a.status != store.DeliverySent {
t.Fatalf("unexpected attempt: %+v", a)
}
}
// TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend — losing outbox
// visibility on one attempt (e.g. a transient store error recording intent)
// must not itself prevent the nudge from reaching the user; the send still
// proceeds, just untracked for crash recovery on this one attempt.
func TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend(t *testing.T) {
voice := &fakeSink{}
ob := &fakeOutbox{beginErr: errors.New("db busy")}
d := NewDispatcher(Config{Voice: voice, Outbox: ob})
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("water", loop.Sev1, store.Present),
Body: "body", Summary: "sum",
}, refNow())
if err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(voice.sends) != 1 || len(out) != 1 {
t.Fatalf("send should still happen despite outbox begin failure: sends=%v out=%v", voice.sends, out)
}
}