kami
9ff726ee26
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
85 lines
3.6 KiB
Go
85 lines
3.6 KiB
Go
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
|
|
}
|