Wire durable delivery outbox: migration, store, and dispatcher config
Companion to the dispatcher-side outbox change: adds the delivery_attempts table migration, Store.BeginDeliveryAttempt/ CompleteDeliveryAttempt/ReconcileStaleDeliveryAttempts, and wires ReconcileStaleDeliveryAttempts + Config.Outbox into mavend startup before the tick loop resumes. Vikunja #270. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA
This commit is contained in:
@@ -319,6 +319,13 @@ func run(args []string) error {
|
|||||||
if voiceW != nil {
|
if voiceW != nil {
|
||||||
voiceSink = voiceW.voiceSink
|
voiceSink = voiceW.voiceSink
|
||||||
}
|
}
|
||||||
|
// A crashed prior run may have left "pending" delivery attempts (send
|
||||||
|
// may have landed externally, then the process died before recording
|
||||||
|
// it) — reconcile them to "unknown" before the tick loop resumes
|
||||||
|
// sending, so nothing auto-resends into that ambiguity.
|
||||||
|
if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil {
|
||||||
|
log.Printf("delivery outbox reconcile: %v", err)
|
||||||
|
}
|
||||||
dispatcher = delivery.NewDispatcher(delivery.Config{
|
dispatcher = delivery.NewDispatcher(delivery.Config{
|
||||||
Ntfy: ntfy,
|
Ntfy: ntfy,
|
||||||
Telegram: telegram,
|
Telegram: telegram,
|
||||||
@@ -326,6 +333,7 @@ func run(args []string) error {
|
|||||||
Ack: st,
|
Ack: st,
|
||||||
Nudges: st,
|
Nudges: st,
|
||||||
Reminders: st,
|
Reminders: st,
|
||||||
|
Outbox: st,
|
||||||
})
|
})
|
||||||
|
|
||||||
// tick loop
|
// tick loop
|
||||||
@@ -475,6 +483,9 @@ func run(args []string) error {
|
|||||||
if voiceW != nil {
|
if voiceW != nil {
|
||||||
voiceSink = voiceW.voiceSink
|
voiceSink = voiceW.voiceSink
|
||||||
}
|
}
|
||||||
|
if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil {
|
||||||
|
log.Printf("delivery outbox reconcile: %v", err)
|
||||||
|
}
|
||||||
dispatcher = delivery.NewDispatcher(delivery.Config{
|
dispatcher = delivery.NewDispatcher(delivery.Config{
|
||||||
Ntfy: ntfy,
|
Ntfy: ntfy,
|
||||||
Telegram: telegram,
|
Telegram: telegram,
|
||||||
@@ -482,6 +493,7 @@ func run(args []string) error {
|
|||||||
Ack: st,
|
Ack: st,
|
||||||
Nudges: st,
|
Nudges: st,
|
||||||
Reminders: st,
|
Reminders: st,
|
||||||
|
Outbox: st,
|
||||||
})
|
})
|
||||||
|
|
||||||
tickInterval := time.Duration(cfg.TickInterval)
|
tickInterval := time.Duration(cfg.TickInterval)
|
||||||
|
|||||||
@@ -0,0 +1,84 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"log"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// DeliveryAttempt statuses. pending = Begin recorded, no Complete yet — either
|
||||||
|
// still in flight or the process died mid-send (crash window the outbox
|
||||||
|
// exists to close). sent/failed = Complete recorded the sink's outcome.
|
||||||
|
// unknown = a pending row found stale at startup: the process that started it
|
||||||
|
// is gone, and the send may or may not have reached the external channel.
|
||||||
|
// Never auto-resolved into sent or failed — that would be guessing.
|
||||||
|
const (
|
||||||
|
DeliveryPending = "pending"
|
||||||
|
DeliverySent = "sent"
|
||||||
|
DeliveryFailed = "failed"
|
||||||
|
DeliveryUnknown = "unknown"
|
||||||
|
)
|
||||||
|
|
||||||
|
// BeginDeliveryAttempt durably records intent to send BEFORE the external
|
||||||
|
// send happens, so a crash between "sent externally" and "recorded" leaves a
|
||||||
|
// trace instead of silence. kind is "nudge" or "reminder"; rule is set for
|
||||||
|
// nudges, reminderID for reminders (the other left at its zero value).
|
||||||
|
// bodyHash is an opaque caller-computed key (e.g. sha256 of channel+body) —
|
||||||
|
// stored for post-crash operator triage, not enforced as a uniqueness
|
||||||
|
// constraint (a rule/reminder legitimately re-sends across ticks).
|
||||||
|
func (s *Store) BeginDeliveryAttempt(ctx context.Context, kind, rule string, reminderID int64, channel, bodyHash string, now time.Time) (int64, error) {
|
||||||
|
res, err := s.db.ExecContext(ctx,
|
||||||
|
`INSERT INTO delivery_attempts (kind, rule, reminder_id, channel, body_hash, status, created_ts)
|
||||||
|
VALUES (?, ?, ?, ?, ?, 'pending', ?)`,
|
||||||
|
kind, rule, reminderID, channel, bodyHash, now.UnixMilli())
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("begin delivery attempt: %w", err)
|
||||||
|
}
|
||||||
|
id, err := res.LastInsertId()
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("begin delivery attempt: last insert id: %w", err)
|
||||||
|
}
|
||||||
|
return id, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// CompleteDeliveryAttempt records the sink's outcome for a prior
|
||||||
|
// BeginDeliveryAttempt. status is "sent" or "failed" — never "pending" or
|
||||||
|
// "unknown" (those are set only by Begin and reconciliation respectively).
|
||||||
|
func (s *Store) CompleteDeliveryAttempt(ctx context.Context, id int64, status string, now time.Time) error {
|
||||||
|
if status != DeliverySent && status != DeliveryFailed {
|
||||||
|
return fmt.Errorf("store: invalid delivery completion status %q", status)
|
||||||
|
}
|
||||||
|
_, err := s.db.ExecContext(ctx,
|
||||||
|
`UPDATE delivery_attempts SET status = ?, completed_ts = ? WHERE id = ? AND status = 'pending'`,
|
||||||
|
status, now.UnixMilli(), id)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("complete delivery attempt %d: %w", id, err)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReconcileStaleDeliveryAttempts runs once at daemon startup, before the tick
|
||||||
|
// loop resumes sending. Any attempt still "pending" from a previous process
|
||||||
|
// life is the exact crash window the outbox exists to close: the external
|
||||||
|
// send may have landed and the process died before recording the outcome.
|
||||||
|
// Marking it "unknown" (rather than silently resending, and rather than
|
||||||
|
// silently dropping it) preserves the same never-guess-an-ambiguous-outcome
|
||||||
|
// rule as the IPC client and Hexis execution engine. Returns the count
|
||||||
|
// reconciled, for startup logging.
|
||||||
|
func (s *Store) ReconcileStaleDeliveryAttempts(ctx context.Context, now time.Time) (int, error) {
|
||||||
|
res, err := s.db.ExecContext(ctx,
|
||||||
|
`UPDATE delivery_attempts SET status = 'unknown', completed_ts = ? WHERE status = 'pending'`,
|
||||||
|
now.UnixMilli())
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("reconcile stale delivery attempts: %w", err)
|
||||||
|
}
|
||||||
|
n, err := res.RowsAffected()
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("reconcile stale delivery attempts: rows affected: %w", err)
|
||||||
|
}
|
||||||
|
if n > 0 {
|
||||||
|
log.Printf("store: reconciled %d stale delivery attempt(s) from a prior run as outcome=unknown", n)
|
||||||
|
}
|
||||||
|
return int(n), nil
|
||||||
|
}
|
||||||
@@ -50,6 +50,19 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
|
|||||||
sent_at INTEGER NOT NULL
|
sent_at INTEGER NOT NULL
|
||||||
);
|
);
|
||||||
CREATE INDEX IF NOT EXISTS idx_ack_sends_rule ON ack_sends (rule, sent_at DESC);`, // #5 — sev4 telegram repeat-til-ack tracking
|
CREATE INDEX IF NOT EXISTS idx_ack_sends_rule ON ack_sends (rule, sent_at DESC);`, // #5 — sev4 telegram repeat-til-ack tracking
|
||||||
|
|
||||||
|
`CREATE TABLE IF NOT EXISTS delivery_attempts (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
kind TEXT NOT NULL CHECK (kind IN ('nudge','reminder')),
|
||||||
|
rule TEXT NOT NULL DEFAULT '',
|
||||||
|
reminder_id INTEGER NOT NULL DEFAULT 0,
|
||||||
|
channel TEXT NOT NULL,
|
||||||
|
body_hash TEXT NOT NULL,
|
||||||
|
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','sent','failed','unknown')),
|
||||||
|
created_ts INTEGER NOT NULL,
|
||||||
|
completed_ts INTEGER
|
||||||
|
);
|
||||||
|
CREATE INDEX IF NOT EXISTS idx_delivery_attempts_status ON delivery_attempts (status);`, // #6 — durable delivery outbox
|
||||||
}
|
}
|
||||||
|
|
||||||
// migrate applies every migration with a number greater than the DB's current
|
// migrate applies every migration with a number greater than the DB's current
|
||||||
|
|||||||
Reference in New Issue
Block a user