console: the log's event times are Unix seconds like the rest of the self-serve view; only a permanent delivery failure (the account or app cannot be resolved) counts toward the dead-letter, a transient one is retried uncounted; the waiting event is stamped with the tick's clock
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
4 files changed,
+60
−21
internal/console/selfserve_events.go+9 −6internal/console/selfserve_events_test.go+37 −4internal/console/selfserve_job.go+2 −2internal/console/selfserve_store.go+12 −9
diff --git a/internal/console/selfserve_events.go b/internal/console/selfserve_events.go| index 068cd0cd7..b2c8a1d3d 100644 |
| --- a/internal/console/selfserve_events.go |
| +++ b/internal/console/selfserve_events.go |
| @@ -8,8 +8,8 @@ package console |
| // Phase is the phase entered, any other event's is "<phase>:<kind>" |
| // (reports:lane, health:probe, prechecks:facts, ready:refusal, |
| // stopping:timing). The pass delivers each one once as an audit row |
| -// (drainSelfServeEvents), and an event it cannot deliver five times in a |
| -// row (its account deleted) is dead-lettered so it never holds recovery |
| +// (drainSelfServeEvents), and an event whose account or app cannot be |
| +// resolved five times (its account deleted) is dead-lettered so it never holds recovery |
| // mode on for ever. |
| import ( |
| @@ -82,8 +82,9 @@ func ssRowEventTx(id, kind string, at time.Time, reason, detail string) ssAlso { |
| } |
| } |
| -// selfServeDeliverAttempts is how many failed deliveries dead-letter an |
| -// event (Task 4's ruling for an event whose account was deleted). |
| +// selfServeDeliverAttempts is how many permanent delivery failures |
| +// dead-letter an event (Task 4's ruling for an event whose account was |
| +// deleted). |
| const selfServeDeliverAttempts = 5 |
| // errSelfServeEventUndeliverable is a delivery that dead-lettered its |
| @@ -149,7 +150,9 @@ FROM self_serve_events WHERE id = ? ORDER BY seq`, id) |
| } |
| // selfServeEventView is one event in the operator's move log (GET |
| -// .../move/offline?log=1). At is Unix nanoseconds, as the journal keeps it. |
| +// .../move/offline?log=1). At is Unix seconds, as every time in the |
| +// self-serve view (unixOrZero); the events are in seq order, so events |
| +// within one second keep theirs. |
| type selfServeEventView struct { |
| Phase string `json:"phase"` |
| At int64 `json:"at"` |
| @@ -187,7 +190,7 @@ func (s *Server) offlineSelfServeOf(v *offlineView, d offlineDoor, withLog bool) |
| } |
| v.Events = make([]selfServeEventView, 0, len(evs)) |
| for _, e := range evs { |
| - v.Events = append(v.Events, selfServeEventView{Phase: e.Phase, At: ssNanos(e.At), Reason: e.Reason, Detail: e.Detail}) |
| + v.Events = append(v.Events, selfServeEventView{Phase: e.Phase, At: unixOrZero(e.At), Reason: e.Reason, Detail: e.Detail}) |
| } |
| } |
diff --git a/internal/console/selfserve_events_test.go b/internal/console/selfserve_events_test.go| index bfea686de..31f569a0e 100644 |
| --- a/internal/console/selfserve_events_test.go |
| +++ b/internal/console/selfserve_events_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "database/sql" |
| "encoding/json" |
| "errors" |
| + "fmt" |
| "net/http" |
| "strings" |
| "testing" |
| @@ -65,7 +66,7 @@ func TestEventsAreSequencedAndDeliveredOnce(t *testing.T) { |
| t.Fatal(err) |
| } |
| for i := 0; i < 2; i++ { |
| - if err := st.SetSelfServeReason(ssID, msgSelfServeWaitingEarlier); err != nil { |
| + if err := st.SetSelfServeReason(ssID, msgSelfServeWaitingEarlier, now.Add(time.Minute)); err != nil { |
| t.Fatal(err) |
| } |
| } |
| @@ -86,6 +87,9 @@ func TestEventsAreSequencedAndDeliveredOnce(t *testing.T) { |
| if e := evs[4]; e.Reason != "" || e.Detail != "failed: the probe was answered 503" { |
| t.Fatalf("the probe's event %+v", e) |
| } |
| + if e := evs[5]; !e.At.Equal(now.Add(time.Minute)) { |
| + t.Fatalf("the waiting event is stamped %v, want the tick's %v", e.At, now.Add(time.Minute)) |
| + } |
| // A delivery failure leaves the event undelivered and counts it. |
| pending, err := st.UnemittedSelfServeEvents(100) |
| @@ -102,9 +106,32 @@ func TestEventsAreSequencedAndDeliveredOnce(t *testing.T) { |
| if left, _ := st.UnemittedSelfServeEvents(100); len(left) != len(evs) { |
| t.Fatal("a crashed delivery marked its event emitted") |
| } |
| - var attempts int |
| - if err := st.db.QueryRow(`SELECT attempts FROM self_serve_events WHERE id = ? AND seq = ?`, first.ID, first.Seq).Scan(&attempts); err != nil || attempts != 1 { |
| - t.Fatalf("attempts = %d, %v; want 1", attempts, err) |
| + attempts := func() int { |
| + t.Helper() |
| + var n int |
| + if err := st.db.QueryRow(`SELECT attempts FROM self_serve_events WHERE id = ? AND seq = ?`, first.ID, first.Seq).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| + } |
| + // A transient failure (the crash above, a busy database) counts |
| + // nothing, however often it happens. |
| + for i := 0; i < 2*selfServeDeliverAttempts; i++ { |
| + if err := st.DeliverSelfServeEvent(context.Background(), first, func(*sql.Tx) error { return errors.New("database is locked (SQLITE_BUSY)") }); err == nil { |
| + t.Fatal("a failed delivery succeeded") |
| + } |
| + } |
| + if n := attempts(); n != 0 { |
| + t.Fatalf("transient failures counted %d attempts, want none", n) |
| + } |
| + // A permanent one (the account or app cannot be resolved) counts. |
| + if err := st.DeliverSelfServeEvent(context.Background(), first, func(*sql.Tx) error { |
| + return fmt.Errorf("console: audit for account %q: %w", "gone", sql.ErrNoRows) |
| + }); err == nil || errors.Is(err, errSelfServeEventUndeliverable) { |
| + t.Fatalf("a permanent failure = %v, want counted and not yet dead-lettered", err) |
| + } |
| + if n := attempts(); n != 1 { |
| + t.Fatalf("a permanent failure counted %d attempts, want 1", n) |
| } |
| // Delivered again, then a third time from a stale snapshot: one audit |
| // row. |
| @@ -227,7 +254,13 @@ func TestOfflineStatusShowsSelfServeAndLog(t *testing.T) { |
| if err := ssDeliver(w.fx.st, evs[0], false); err != nil { |
| t.Fatal(err) |
| } |
| + before := time.Now().Add(-time.Minute).Unix() |
| v, out = get(t, w, "?log=1") |
| + for _, e := range v.Events { |
| + if e.At < before || e.At > time.Now().Unix() { |
| + t.Fatalf("event at %d is not Unix seconds (now %d): %s", e.At, time.Now().Unix(), out) |
| + } |
| + } |
| if len(v.Events) != 2 || v.Events[0].Phase != ssRequested || v.Events[1].Phase != "requested:timing" || v.Events[1].Reason != "offline_at" || v.Events[0].At > v.Events[1].At { |
| t.Fatalf("events %+v in %s", v.Events, out) |
| } |
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go| index 40e0c8e63..3f68e6431 100644 |
| --- a/internal/console/selfserve_job.go |
| +++ b/internal/console/selfserve_job.go |
| @@ -863,7 +863,7 @@ func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *Sel |
| } |
| if selfServeWaitsForEarlier(open, row) { |
| if row.Reason != msgSelfServeWaitingEarlier { |
| - return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingEarlier) |
| + return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingEarlier, now) |
| } |
| return nil |
| } |
| @@ -891,7 +891,7 @@ func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *Sel |
| return s.selfServeAbort(row, ref.msg, now) |
| default: |
| if row.Reason != msgSelfServeWaitingAnother { |
| - return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingAnother) |
| + return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingAnother, now) |
| } |
| return nil |
| } |
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go| index 51499f066..739b81b68 100644 |
| --- a/internal/console/selfserve_store.go |
| +++ b/internal/console/selfserve_store.go |
| @@ -1123,8 +1123,8 @@ WHERE id = ? AND outcome = '' AND abort_requested_at = 0`, ssNanos(now), reason, |
| // SetSelfServeReason records why an open row waits (the view's reason). |
| // A new non-empty reason is a waiting event (<phase>:waiting, Task 9) in |
| // the same transaction; the same reason again is none, so a wait that |
| -// lasts many ticks logs once. |
| -func (s *Store) SetSelfServeReason(id, reason string) error { |
| +// lasts many ticks logs once. The event is stamped now, the tick's clock. |
| +func (s *Store) SetSelfServeReason(id, reason string, now time.Time) error { |
| tx, err := s.db.Begin() |
| if err != nil { |
| return err |
| @@ -1142,7 +1142,7 @@ func (s *Store) SetSelfServeReason(id, reason string) error { |
| return err |
| } |
| if reason != "" && reason != was { |
| - if err := ssRowEventTx(id, ssEvWaiting, time.Now().UTC(), reason, "")(tx); err != nil { |
| + if err := ssRowEventTx(id, ssEvWaiting, now, reason, "")(tx); err != nil { |
| return err |
| } |
| } |
| @@ -1243,11 +1243,14 @@ FROM self_serve_events WHERE emitted = 0 ORDER BY at, id, seq LIMIT ?`, limit) |
| // first marks the event emitted (WHERE id = ? AND seq = ? AND emitted = 0); |
| // matching nothing commits nothing and returns (another worker delivered |
| // it); otherwise deliver inserts the audit row inside tx. A failure rolls |
| -// both back, leaving the event undelivered, and counts one attempt |
| -// against it (deliver's own failure only: a cancelled context or a busy |
| -// database is not the event's fault); the selfServeDeliverAttempts-th |
| -// failed attempt dead-letters it (selfServeEventDeliveryFailed), and the |
| -// error then wraps errSelfServeEventUndeliverable. |
| +// both back, leaving the event undelivered. Only a permanent failure |
| +// counts an attempt against it: deliver could not resolve the event's |
| +// account or app (sql.ErrNoRows, auditActivityTx's account lookup). A |
| +// transient one (a busy database, a cancelled context, a crash) counts |
| +// nothing and is retried for as long as it lasts. The |
| +// selfServeDeliverAttempts-th permanent failure dead-letters the event |
| +// (selfServeEventDeliveryFailed), and the error then wraps |
| +// errSelfServeEventUndeliverable. |
| func (s *Store) DeliverSelfServeEvent(ctx context.Context, e SelfServeEvent, deliver func(tx *sql.Tx) error) error { |
| tx, err := s.db.BeginTx(ctx, nil) |
| if err != nil { |
| @@ -1261,7 +1264,7 @@ func (s *Store) DeliverSelfServeEvent(ctx context.Context, e SelfServeEvent, del |
| } |
| if derr := deliver(tx); derr != nil { |
| _ = tx.Rollback() |
| - if ctx.Err() != nil { |
| + if ctx.Err() != nil || !errors.Is(derr, sql.ErrNoRows) { |
| return derr |
| } |
| return s.selfServeEventDeliveryFailed(e, derr) |