Compare commits

...

2 Commits

Author SHA1 Message Date
kami 7d8b0af99d Test voice fallthrough and delivery durability
Fallthrough is checked per severity through the outbox trail, so sev3/sev4
reroute and sev1/sev2 still drop. Durability uses a real store on a temp file:
a crash between Begin and Complete becomes unknown, is not resent, is not
dropped, and a late Complete cannot overwrite it. One skipped test marks a real
gap: a panic mid-send leaves a permanent pending row.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ
2026-07-31 02:27:33 +04:00
kami a2835bbdf6 Cover every cell of the delivery routing table
Table-driven tests for all four severity bands crossed with present and away,
both as the pure table and end to end through the dispatcher. Three tests are
written to DESIGN.md and skipped because the code does not keep the claim: the
care-away drop is recorded nowhere, and the minimal body is enforced per-sink
rather than by the dispatcher. Also gofmt'd dispatcher_test.go.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ
2026-07-31 02:24:43 +04:00
3 changed files with 516 additions and 5 deletions
+5 -5
View File
@@ -639,11 +639,11 @@ func TestDispatchRecurringReminderReschedules(t *testing.T) {
// ----------------------------- durable outbox -------------------------------- // ----------------------------- durable outbox --------------------------------
type outboxAttempt struct { type outboxAttempt struct {
kind, rule string kind, rule string
reminderID int64 reminderID int64
channel, hash string channel, hash string
status string status string
begunAt, doneAt time.Time begunAt, doneAt time.Time
} }
// fakeOutbox — an in-memory Outbox that also lets a test simulate a crash // fakeOutbox — an in-memory Outbox that also lets a test simulate a crash
+247
View File
@@ -0,0 +1,247 @@
package delivery
import (
"context"
"errors"
"path/filepath"
"testing"
"time"
"github.com/kami/maven/internal/loop"
"github.com/kami/maven/internal/store"
)
// panicSink — a sink that dies mid-send. Models the ugly case: the process is
// still alive, so startup reconciliation will not run, but the attempt row was
// already begun.
type panicSink struct{ calls int }
func (p *panicSink) Send(_ context.Context, _ Sendable) error {
p.calls++
panic("sink exploded mid-send")
}
// ------------------------- voice fallthrough, per severity -------------------
// TestVoiceNoSessionFallthroughLeavesOutboxTrail — the fallthrough must be
// visible in the ledger too: the voice attempt closes as failed and the away
// attempt is a separate row, so an operator can see the reroute happened.
func TestVoiceNoSessionFallthroughLeavesOutboxTrail(t *testing.T) {
cases := []struct {
name string
sev loop.Severity
wantAt []string // channel per outbox attempt, in order
wantEnd []string // status per attempt, in order
}{
{"sev3 falls through to ntfy", loop.Sev3,
[]string{"voice", "ntfy"}, []string{store.DeliveryFailed, store.DeliverySent}},
{"sev4 falls through to telegram", loop.Sev4,
[]string{"voice", "telegram"}, []string{store.DeliveryFailed, store.DeliverySent}},
{"sev1 does not fall through", loop.Sev1,
[]string{"voice"}, []string{store.DeliveryFailed}},
{"sev2 does not fall through", loop.Sev2,
[]string{"voice"}, []string{store.DeliveryFailed}},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
voice := &fakeSink{err: ErrVoiceNoSession}
ntfy, telegram := &fakeSink{}, &fakeSink{}
ob := &fakeOutbox{}
d := NewDispatcher(Config{
Voice: voice, Ntfy: ntfy, Telegram: telegram,
Ack: newFakeAck(), Outbox: ob,
})
if _, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("some_rule", c.sev, store.Present),
Body: "detail", Summary: "short",
}, refNow()); err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(ob.attempts) != len(c.wantAt) {
t.Fatalf("want %d outbox attempts, got %d (%+v)", len(c.wantAt), len(ob.attempts), ob.attempts)
}
for i, a := range ob.attempts {
if a.channel != c.wantAt[i] || a.status != c.wantEnd[i] {
t.Fatalf("attempt %d: want %s/%s, got %s/%s", i, c.wantAt[i], c.wantEnd[i], a.channel, a.status)
}
}
// care severities must not reach an away channel — that would
// defeat the drop rule.
if c.sev <= loop.Sev2 && (len(ntfy.sends) != 0 || len(telegram.sends) != 0) {
t.Fatalf("care nudge escaped to an away channel: ntfy=%d telegram=%d",
len(ntfy.sends), len(telegram.sends))
}
})
}
}
// ------------------------- crash between Begin and Complete ------------------
// openTestStore — a real store on a temp file. The reconciliation promise is a
// SQL promise, so a fake would only test the fake.
func openTestStore(t *testing.T) *store.Store {
t.Helper()
st, err := store.Open(context.Background(), filepath.Join(t.TempDir(), "maven.db"))
if err != nil {
t.Fatalf("open store: %v", err)
}
t.Cleanup(func() { _ = st.Close() })
return st
}
// attemptStatus reads one attempt row back. Returns ok=false when the row is
// gone, which would itself be a broken promise (a dropped attempt).
func attemptStatus(t *testing.T, st *store.Store, id int64) (status string, completed bool, ok bool) {
t.Helper()
tx, err := st.DB(context.Background())
if err != nil {
t.Fatalf("read tx: %v", err)
}
defer func() { _ = tx.Rollback() }()
var completedTS *int64
err = tx.QueryRowContext(context.Background(),
`SELECT status, completed_ts FROM delivery_attempts WHERE id = ?`, id).Scan(&status, &completedTS)
if err != nil {
return "", false, false
}
return status, completedTS != nil, true
}
// TestCrashBetweenBeginAndCompleteBecomesUnknown — simulate the crash window:
// Begin lands, the process dies before Complete. Startup reconciliation must
// turn that row into "unknown" — neither resent nor dropped, because Maven
// cannot know whether the message left the box.
func TestCrashBetweenBeginAndCompleteBecomesUnknown(t *testing.T) {
st := openTestStore(t)
ctx := context.Background()
sink := &fakeSink{}
// the crash: intent recorded, no completion.
id, err := st.BeginDeliveryAttempt(ctx, "nudge", "disk_low", 0, "telegram", "hash", refNow())
if err != nil {
t.Fatalf("begin: %v", err)
}
if s, _, ok := attemptStatus(t, st, id); !ok || s != store.DeliveryPending {
t.Fatalf("before reconcile: want pending, got %q ok=%v", s, ok)
}
// restart.
n, err := st.ReconcileStaleDeliveryAttempts(ctx, refNow().Add(time.Minute))
if err != nil {
t.Fatalf("reconcile: %v", err)
}
if n != 1 {
t.Fatalf("want 1 row reconciled, got %d", n)
}
s, completed, ok := attemptStatus(t, st, id)
if !ok {
t.Fatal("reconciliation dropped the row; the promise is it is never dropped")
}
if s != store.DeliveryUnknown {
t.Fatalf("want status unknown, got %q", s)
}
if !completed {
t.Fatal("reconciled row should carry a completed_ts")
}
// not resent: reconciliation is bookkeeping only, it must never push.
if len(sink.sends) != 0 {
t.Fatalf("reconciliation must not resend, got %d sends", len(sink.sends))
}
// idempotent: a second restart must not churn the row again.
n2, err := st.ReconcileStaleDeliveryAttempts(ctx, refNow().Add(2*time.Minute))
if err != nil {
t.Fatalf("reconcile again: %v", err)
}
if n2 != 0 {
t.Fatalf("second reconcile should find nothing, got %d", n2)
}
if s2, _, _ := attemptStatus(t, st, id); s2 != store.DeliveryUnknown {
t.Fatalf("unknown must stay unknown, got %q", s2)
}
}
// TestUnknownIsNeverResolvedToSentOrFailed — the "never guess" half of the
// promise: nothing may quietly turn an unknown into a definite outcome.
func TestUnknownIsNeverResolvedToSentOrFailed(t *testing.T) {
st := openTestStore(t)
ctx := context.Background()
id, err := st.BeginDeliveryAttempt(ctx, "nudge", "disk_low", 0, "telegram", "hash", refNow())
if err != nil {
t.Fatalf("begin: %v", err)
}
if _, err := st.ReconcileStaleDeliveryAttempts(ctx, refNow()); err != nil {
t.Fatalf("reconcile: %v", err)
}
// a late Complete from the old in-flight send must not win.
if err := st.CompleteDeliveryAttempt(ctx, id, store.DeliverySent, refNow().Add(time.Minute)); err != nil {
t.Fatalf("late complete: %v", err)
}
if s, _, _ := attemptStatus(t, st, id); s != store.DeliveryUnknown {
t.Fatalf("late complete overwrote an unknown outcome: %q", s)
}
}
// ------------------------- the boring failure modes --------------------------
// TestSendTimeoutResolvesTheAttempt — a send that times out is a definite
// failure from Maven's side, so the row must not be left pending.
func TestSendTimeoutResolvesTheAttempt(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel() // the deadline already blew
ob := &fakeOutbox{}
d := NewDispatcher(Config{Ntfy: &fakeSink{err: context.DeadlineExceeded}, Outbox: ob})
if _, err := d.DispatchNudge(ctx, PhrasedNudge{
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
Body: "detail", Summary: "short",
}, refNow()); err == nil {
t.Fatal("want a timeout error to propagate")
}
if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryFailed {
t.Fatalf("timed-out send must close the attempt as failed, got %+v", ob.attempts)
}
}
// TestCompleteFailureLeavesRowPendingForReconciliation — if Complete itself
// fails, the row stays pending on purpose. That is the correct ambiguous state
// and startup reconciliation is what resolves it.
func TestCompleteFailureLeavesRowPendingForReconciliation(t *testing.T) {
ob := &fakeOutbox{completeErr: errors.New("db busy")}
d := NewDispatcher(Config{Ntfy: &fakeSink{}, Outbox: ob})
if _, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
Body: "detail", Summary: "short",
}, refNow()); err != nil {
t.Fatalf("a failed outbox complete must not fail the dispatch: %v", err)
}
if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryPending {
t.Fatalf("want the row left pending, got %+v", ob.attempts)
}
}
// TestPanicMidSendResolvesTheAttempt — a sink that panics leaves the attempt
// pending forever while the process keeps running: the dispatcher has no
// recover, and reconciliation only runs at startup. Written to the promise
// ("never silently resent or dropped" implies every attempt gets resolved),
// skipped because the code does not keep it.
func TestPanicMidSendResolvesTheAttempt(t *testing.T) {
t.Skip("real gap: dispatcher.go:168 has no recover around Send, so a panicking sink leaves a permanent pending row (reconciliation only runs at startup, cmd/mavend/main.go:330)")
ob := &fakeOutbox{}
d := NewDispatcher(Config{Ntfy: &panicSink{}, Outbox: ob})
func() {
defer func() { _ = recover() }()
_, _ = d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
Body: "detail", Summary: "short",
}, refNow())
}()
if len(ob.attempts) != 1 || ob.attempts[0].status == store.DeliveryPending {
t.Fatalf("a panic mid-send must still resolve the attempt, got %+v", ob.attempts)
}
}
+264
View File
@@ -0,0 +1,264 @@
package delivery
import (
"context"
"strings"
"testing"
"time"
"github.com/kami/maven/internal/loop"
"github.com/kami/maven/internal/store"
)
// This file walks every cell of the DESIGN.md § "Delivery / channel routing"
// table, once as the pure table and once through the dispatcher, so a change
// to either side has to break a named cell.
//
// present away
// sev1-2 (care) voice drop
// sev3 (soft) voice ntfy, once
// sev4 (hard) voice + ntfy telegram, repeat til ack
type tableCell struct {
name string
sev loop.Severity
presence store.Bucket
want []Channel
}
func allTableCells() []tableCell {
return []tableCell{
{"sev1 present", loop.Sev1, store.Present, []Channel{ChannelVoice}},
{"sev2 present", loop.Sev2, store.Present, []Channel{ChannelVoice}},
{"sev3 present", loop.Sev3, store.Present, []Channel{ChannelVoice}},
{"sev4 present", loop.Sev4, store.Present, []Channel{ChannelVoice, ChannelNtfy}},
{"sev1 away", loop.Sev1, store.Away, []Channel{ChannelDrop}},
{"sev2 away", loop.Sev2, store.Away, []Channel{ChannelDrop}},
{"sev3 away", loop.Sev3, store.Away, []Channel{ChannelNtfy}},
{"sev4 away", loop.Sev4, store.Away, []Channel{ChannelTelegram}},
}
}
func sameChannels(got, want []Channel) bool {
if len(got) != len(want) {
return false
}
for i := range got {
if got[i] != want[i] {
return false
}
}
return true
}
func TestChannelsForEveryTableCell(t *testing.T) {
for _, c := range allTableCells() {
t.Run(c.name, func(t *testing.T) {
got := ChannelsFor(c.sev, c.presence)
if !sameChannels(got, c.want) {
t.Fatalf("%s: want %v, got %v", c.name, c.want, got)
}
})
}
}
// TestDispatchNudgeEveryTableCell — the same eight cells end to end: exactly
// the wanted channels get a send, and every other channel gets none.
func TestDispatchNudgeEveryTableCell(t *testing.T) {
for _, c := range allTableCells() {
t.Run(c.name, func(t *testing.T) {
voice, ntfy, telegram := &fakeSink{}, &fakeSink{}, &fakeSink{}
rec := &fakeNudgeRecorder{}
d := NewDispatcher(Config{
Voice: voice, Ntfy: ntfy, Telegram: telegram,
Ack: newFakeAck(), Nudges: rec,
})
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("some_rule", c.sev, c.presence),
Body: "full detail body",
Summary: "short form",
}, refNow())
if err != nil {
t.Fatalf("dispatch: %v", err)
}
sent := map[Channel]int{
ChannelVoice: len(voice.sends),
ChannelNtfy: len(ntfy.sends),
ChannelTelegram: len(telegram.sends),
}
for ch, n := range sent {
want := 0
for _, w := range c.want {
if w == ch {
want = 1
}
}
if n != want {
t.Fatalf("%s: channel %s got %d sends, want %d", c.name, ch, n, want)
}
}
// one dispatch and one nudge row per real (non-drop) channel.
wantDispatches := 0
for _, w := range c.want {
if w != ChannelDrop {
wantDispatches++
}
}
if len(out) != wantDispatches {
t.Fatalf("%s: want %d dispatches, got %d", c.name, wantDispatches, len(out))
}
if len(rec.rows) != wantDispatches {
t.Fatalf("%s: want %d nudge rows, got %d", c.name, wantDispatches, len(rec.rows))
}
})
}
}
// TestSev3AwayIsNtfyExactlyOnce — "ntfy, once": one send, and nothing on the
// sendable asks for a repeat, so the daemon's repeat driver has no reason to
// pick it up.
func TestSev3AwayIsNtfyExactlyOnce(t *testing.T) {
ntfy := &fakeSink{}
ack := newFakeAck()
d := NewDispatcher(Config{Ntfy: ntfy, Telegram: &fakeSink{}, Ack: ack})
out, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("cert_expiring", loop.Sev3, store.Away),
Body: "cert detail", Summary: "cert expiring",
}, refNow())
if err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(ntfy.sends) != 1 {
t.Fatalf("sev3 away: want exactly 1 ntfy send, got %d", len(ntfy.sends))
}
if out[0].Sendable.RepeatUntilAck {
t.Fatalf("sev3 away must not repeat til ack")
}
if _, ok := ack.lastSent["cert_expiring"]; ok {
t.Fatalf("sev3 away must not enter the ack/repeat tracker")
}
}
// TestSev4AwayRepeatsUntilAcked — "telegram, repeat til ack": the initial send
// arms the ack clock, the repeat driver re-sends while un-acked, and an ack
// stops it.
func TestSev4AwayRepeatsUntilAcked(t *testing.T) {
telegram := &fakeSink{}
ack := newFakeAck()
d := NewDispatcher(Config{Telegram: telegram, Ack: ack})
ctx := context.Background()
if _, err := d.DispatchNudge(ctx, PhrasedNudge{
Candidate: candidate("disk_low", loop.Sev4, store.Away),
Body: "disk detail", Summary: "disk low on homesrv",
}, refNow()); err != nil {
t.Fatalf("dispatch: %v", err)
}
// two intervals pass, still un-acked → two more sends.
for i := 1; i <= 2; i++ {
at := refNow().Add(time.Duration(i) * 10 * time.Minute)
if _, err := d.RepeatUnacked(ctx, []string{"disk_low"}, at, 5*time.Minute, "disk detail", "disk low on homesrv"); err != nil {
t.Fatalf("repeat %d: %v", i, err)
}
}
if len(telegram.sends) != 3 {
t.Fatalf("want 1 initial + 2 repeats = 3 telegram sends, got %d", len(telegram.sends))
}
// acked → no further sends, however long we wait.
_ = ack.MarkAcked(ctx, "disk_low")
if _, err := d.RepeatUnacked(ctx, []string{"disk_low"}, refNow().Add(time.Hour), 5*time.Minute, "b", "s"); err != nil {
t.Fatalf("repeat after ack: %v", err)
}
if len(telegram.sends) != 3 {
t.Fatalf("ack must stop the repeat; got %d sends", len(telegram.sends))
}
}
// TestAwayChannelsGetMinimalBody — what leaves the box is the short form, for
// every away cell of the table. messageForChannel is the last-mile choice both
// away sinks make too.
func TestAwayChannelsGetMinimalBody(t *testing.T) {
detail := "disk /mnt/hdd1 on homesrv at 97% — 12GB free, biggest offender /var/lib/docker"
short := "disk low on homesrv"
for _, ch := range []Channel{ChannelNtfy, ChannelTelegram} {
t.Run(string(ch), func(t *testing.T) {
msg := messageForChannel(Sendable{Channel: ch, Body: detail, Summary: short})
if msg != short {
t.Fatalf("%s message: want %q, got %q", ch, short, msg)
}
})
}
if got := messageForChannel(Sendable{Channel: ChannelVoice, Body: detail, Summary: short}); got != detail {
t.Fatalf("voice is local and gets the full body, got %q", got)
}
}
// TestSev4AwaySendableCarriesNoDetail — DESIGN.md § Delivery: away channels
// leave the box, so a sev4-away message must not carry detail beyond the short
// form. Today the dispatcher hands the away sink the FULL Body as well as the
// Summary (dispatcher.go:153-162 copies pn.Body into every Sendable) and
// trusts each sink to pick Summary. That works for the two sinks in-tree, but
// the minimal body is not enforced at the dispatcher, so a new away sink that
// reads Body exfils by default.
func TestSev4AwaySendableCarriesNoDetail(t *testing.T) {
t.Skip("not enforced: dispatcher.go:159 puts the full Body on away sendables; minimal body is only enforced per-sink (ntfysink.go:77, telegramsink.go:148)")
telegram := &fakeSink{}
d := NewDispatcher(Config{Telegram: telegram, Ack: newFakeAck()})
detail := "disk /mnt/hdd1 at 97%, biggest offender /var/lib/docker"
if _, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("disk_low", loop.Sev4, store.Away),
Body: detail, Summary: "disk low on homesrv",
}, refNow()); err != nil {
t.Fatalf("dispatch: %v", err)
}
if strings.Contains(telegram.sends[0].Body, "/var/lib/docker") {
t.Fatalf("away sendable carries detail: %q", telegram.sends[0].Body)
}
}
// TestAwayFallsBackToFullBodyWhenSummaryEmpty — the other half of the same
// gap: with no Summary, the full body leaves the box. The code chooses that on
// purpose ("a terse full message is better than no message",
// dispatcher.go:345-357), which contradicts the spec's minimal-body rule.
// Written to the spec, skipped because the code disagrees.
func TestAwayFallsBackToFullBodyWhenSummaryEmpty(t *testing.T) {
t.Skip("by design today: dispatcher.go:356 and ntfysink.go:79 fall back to the full Body when Summary is empty, so detail can leave the box")
msg := messageForChannel(Sendable{
Channel: ChannelNtfy,
Body: "internal detail that should never leave the box",
})
if msg != "" {
t.Fatalf("empty summary must not fall back to body, got %q", msg)
}
}
// TestCareAwayDropIsRecorded — DESIGN.md's drop is a decision ("a missed water
// nudge is noise, a missed backup failure isn't"), so it should be visible
// rather than vanish. Today drop is a bare `continue`: no nudge row, no outbox
// attempt, no log — nothing an operator can see afterwards.
func TestCareAwayDropIsRecorded(t *testing.T) {
t.Skip("not implemented: dispatcher.go:149-151 skips a Drop channel with no record; there is no 'dropped' outcome in store/delivery.go:16-21")
ob := &fakeOutbox{}
d := NewDispatcher(Config{Voice: &fakeSink{}, Nudges: &fakeNudgeRecorder{}, Outbox: ob})
if _, err := d.DispatchNudge(context.Background(), PhrasedNudge{
Candidate: candidate("water", loop.Sev1, store.Away),
Body: "drink water", Summary: "water",
}, refNow()); err != nil {
t.Fatalf("dispatch: %v", err)
}
if len(ob.attempts) != 1 || ob.attempts[0].channel != string(ChannelDrop) {
t.Fatalf("care-away drop should leave a visible record, got %+v", ob.attempts)
}
}