carlos / platform Public

console: every refusal, fact decision, lane and health probe and new waiting reason is an event in its change's transaction; an event undelivered five times is dead-lettered (attempts, detail undeliverable:, logged once); the operator status shows self_serve and, with ?log=1, the move's events

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 1 day ago 109a0b6a8b3bc45b4b6c594ce5b48c6974b13ee7 parent 2fb1738

5 files changed, +232 −32

internal/console/apiinstancemove_offline.go +10 −1
internal/console/selfserve_events.go +134 −1
internal/console/selfserve_job.go +13 −6
internal/console/selfserve_pass.go +4 −10
internal/console/selfserve_store.go +71 −14

diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go

index be6a7db0b..58e93d930 100644

--- a/internal/console/apiinstancemove_offline.go

+++ b/internal/console/apiinstancemove_offline.go

@@ -1739,6 +1739,13 @@ type offlineView struct {

Reports []offlineReportView `json:"reports,omitempty"`

Items []offlineItemView `json:"items,omitempty"`

Last *provision.MoveOutcome `json:"last,omitempty"`

+ // SelfServe is the self-serve move's own view when the instance has

+ // one to show (selfServeViewOf's rules), and Events, with ?log=1, its

+ // events oldest first, delivered or not (self-serve moves plan 2 Task

+ // 9). The status door alone fills them; a legacy or operator move

+ // shows neither.

+ SelfServe *selfServeMoveView `json:"self_serve,omitempty"`

+ Events []selfServeEventView `json:"events,omitempty"`

}

// offlineStageView is the pre-copy's progress, from the latest stager

@@ -1896,5 +1903,7 @@ func (s *Server) handleAPIInstanceOfflineStatus(w http.ResponseWriter, r *http.R

if !ok {

return

}

- writeJSON(w, http.StatusOK, s.offlineViewOf(r.Context(), d.rec))

+ v := s.offlineViewOf(r.Context(), d.rec)

+ s.offlineSelfServeOf(&v, d, r.URL.Query().Get("log") == "1")

+ writeJSON(w, http.StatusOK, v)

}

diff --git a/internal/console/selfserve_events.go b/internal/console/selfserve_events.go

index b7aedda4b..068cd0cd7 100644

--- a/internal/console/selfserve_events.go

+++ b/internal/console/selfserve_events.go

@@ -13,14 +13,22 @@ package console

// mode on for ever.

import (

+ "context"

"database/sql"

+ "errors"

"fmt"

+ "log"

"time"

)

// The kinds of event that are not a phase change.

const (

- ssEvTiming = "timing" // a timing mark (reason the mark)

+ ssEvRefusal = "refusal" // the job's own abort decision, with its reason

+ ssEvWaiting = "waiting" // a new reason the row waits for

+ ssEvFacts = "facts" // a fact decision that passed (a failed one is a refusal)

+ ssEvLane = "lane" // the staging lane's probe

+ ssEvProbe = "probe" // a health probe

+ ssEvTiming = "timing" // a timing mark (reason the mark)

)

// ssAlso is a write a journal change makes in its own transaction, run

@@ -73,3 +81,128 @@ func ssRowEventTx(id, kind string, at time.Time, reason, detail string) ssAlso {

return ssEventTx(id, phase+":"+kind, at, reason, detail)(tx)

}

+// selfServeDeliverAttempts is how many failed deliveries 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

+// event.

+var errSelfServeEventUndeliverable = errors.New("undeliverable")

+// selfServeEventDeliveryFailed counts a failed delivery of e (cause) and,

+// at selfServeDeliverAttempts, marks it emitted with its detail prefixed

+// "undeliverable:", in one statement:

+//

+// WHERE id = ? AND seq = ? AND emitted = 0

+//

+// It returns cause, wrapped with errSelfServeEventUndeliverable when this

+// failure dead-lettered the event.

+func (s *Store) selfServeEventDeliveryFailed(e SelfServeEvent, cause error) error {

+ tx, err := s.db.Begin()

+ if err != nil {

+ return errors.Join(cause, err)

+ }

+ defer tx.Rollback()

+ res, err := tx.Exec(`UPDATE self_serve_events SET attempts = attempts + 1,

+ emitted = CASE WHEN attempts + 1 >= ? THEN 1 ELSE 0 END,

+ detail = CASE WHEN attempts + 1 >= ? THEN 'undeliverable: ' || detail ELSE detail END

+WHERE id = ? AND seq = ? AND emitted = 0`, selfServeDeliverAttempts, selfServeDeliverAttempts, e.ID, e.Seq)

+ won, err := ssWon(res, err)

+ if err != nil {

+ return errors.Join(cause, err)

+ }

+ var emitted int64

+ if won {

+ if err := tx.QueryRow(`SELECT emitted FROM self_serve_events WHERE id = ? AND seq = ?`, e.ID, e.Seq).Scan(&emitted); err != nil {

+ return errors.Join(cause, err)

+ }

+ if err := tx.Commit(); err != nil {

+ return errors.Join(cause, err)

+ }

+ if emitted != 0 {

+ return fmt.Errorf("%w after %d attempts: %w", errSelfServeEventUndeliverable, selfServeDeliverAttempts, cause)

+ }

+ return cause

+// SelfServeEvents are the row's events, delivered or not, oldest first.

+func (s *Store) SelfServeEvents(id string) ([]SelfServeEvent, error) {

+ rows, err := s.db.Query(`SELECT id, seq, phase, account, app, host, at, reason, detail, emitted

+FROM self_serve_events WHERE id = ? ORDER BY seq`, id)

+ if err != nil {

+ return nil, err

+ }

+ defer rows.Close()

+ var out []SelfServeEvent

+ for rows.Next() {

+ var e SelfServeEvent

+ var at, emitted int64

+ if err := rows.Scan(&e.ID, &e.Seq, &e.Phase, &e.Account, &e.App, &e.Host, &at, &e.Reason, &e.Detail, &emitted); err != nil {

+ return nil, err

+ }

+ e.At, e.Emitted = ssTime(at), emitted != 0

+ out = append(out, e)

+ }

+ return out, rows.Err()

+// selfServeEventView is one event in the operator's move log (GET

+// .../move/offline?log=1). At is Unix nanoseconds, as the journal keeps it.

+type selfServeEventView struct {

+ Phase string `json:"phase"`

+ At int64 `json:"at"`

+ Reason string `json:"reason,omitempty"`

+ Detail string `json:"detail,omitempty"`

+// offlineSelfServeOf joins the instance's self-serve rows into the

+// operator's status view: the self-serve move's own view when one shows

+// (selfServeViewOf's rules), and with log its events, oldest first. A

+// journal that cannot be read leaves both out and is logged; it never

+// fails the answer.

+func (s *Server) offlineSelfServeOf(v *offlineView, d offlineDoor, withLog bool) {

+ open, terminal, err := s.store.SelfServeMoveForHost(d.acct.Sqid, d.app, d.host)

+ if err != nil {

+ log.Printf("console: offline move %s: the self-serve rows: %v", d.host, err)

+ return

+ }

+ sv, ok := s.selfServeViewOf(d.rec, open, terminal, time.Now()).(selfServeMoveView)

+ if !ok {

+ return

+ }

+ v.SelfServe = &sv

+ if !withLog {

+ return

+ }

+ id := sv.ID

+ if id == "" && sv.Last != nil {

+ id = sv.Last.ID

+ }

+ evs, err := s.store.SelfServeEvents(id)

+ if err != nil {

+ log.Printf("console: offline move %s: the self-serve events of %s: %v", d.host, id, err)

+ return

+ }

+ 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})

+ }

+// drainSelfServeEvent delivers one event as the pass does: one audit row

+// in the move's app's activity, at most once, under ctx. A dead-lettered

+// event is logged here, once (it is emitted from then on).

+func (s *Server) drainSelfServeEvent(ctx context.Context, e SelfServeEvent) {

+ err := s.store.DeliverSelfServeEvent(ctx, e, func(tx *sql.Tx) error {

+ return auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{

+ Action: "instance-move", TargetType: "instance", Target: e.Host, Message: e.Phase, Detail: e.Reason})

+ })

+ switch {

+ case errors.Is(err, errSelfServeEventUndeliverable):

+ log.Printf("console: self-serve move %s: event %d (%s) is %v; marked emitted and kept in the log", e.ID, e.Seq, e.Phase, err)

+ case err != nil:

+ log.Printf("console: self-serve move %s: delivering event %d: %v (retrying next tick)", e.ID, e.Seq, err)

+ }

diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go

index f7e4e11d7..40e0c8e63 100644

--- a/internal/console/selfserve_job.go

+++ b/internal/console/selfserve_job.go

@@ -326,10 +326,11 @@ func (s *Server) selfServeDeadlines(row *SelfServeMove, now time.Time) (bool, er

return s.store.SetSelfServeExpiry(row.ID, row.Phase, kind, saw, true)

}

-// selfServeAbort records the job's own decision to abort with reason.

-func (s *Server) selfServeAbort(row *SelfServeMove, reason string, now time.Time) error {

+// selfServeAbort records the job's own decision to abort with reason (its

+// refusal event goes with it, and also's events before it).

+func (s *Server) selfServeAbort(row *SelfServeMove, reason string, now time.Time, also ...ssAlso) error {

TestHooks.point("abort:" + row.Phase)

- _, err := s.store.AbortSelfServe(row.ID, reason, now)

+ _, err := s.store.AbortSelfServe(row.ID, reason, now, also...)

log.Printf("console: self-serve move %s (%s): aborting: %s", row.ID, row.Host, reason)

return err

}

@@ -1053,6 +1054,7 @@ func (s *Server) selfServeReports(ctx context.Context, t *leaseToken, row *SelfS

case v.fail != "":

if v.fail == s.selfServeNoLane(row.From, row.To).msg {

s.selfServePageRow(ctx, row, ssPageLane, ssPageWhyLane(s.offlineName(row.From), s.offlineName(row.To)), now)

+ return s.selfServeAbort(row, v.fail, now, ssEventTx(row.ID, ssReports+":"+ssEvLane, now, "", "failed"))

}

return s.selfServeAbort(row, v.fail, now)

case !v.pass:

@@ -1063,7 +1065,10 @@ func (s *Server) selfServeReports(ctx context.Context, t *leaseToken, row *SelfS

return s.selfServePreOfflineRefusal(row, err, now)

}

TestHooks.point("advance:reports->staging")

- _, err := s.store.AdvanceSelfServe(row.ID, ssReports, ssStaging, "", now.Add(selfServeRoundAllowance(row)), time.Time{}, "", now)

+ // The reports' decisions go with the advance they allowed: the lane's

+ // probe and the facts.

+ _, err := s.store.AdvanceSelfServe(row.ID, ssReports, ssStaging, "", now.Add(selfServeRoundAllowance(row)), time.Time{}, "", now,

+ ssEventTx(row.ID, ssReports+":"+ssEvLane, now, "", "ok"), ssEventTx(row.ID, ssReports+":"+ssEvFacts, now, "", "passed"))

return err

}

@@ -1264,7 +1269,8 @@ func (s *Server) selfServePrechecks(ctx context.Context, t *leaseToken, row *Sel

return err

}

TestHooks.point("precheck:consume")

- if _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepEstimate, row.StepDeadline); err != nil {

+ if _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepEstimate, row.StepDeadline,

+ ssEventTx(row.ID, ssPrechecks+":"+ssEvFacts, now, "", "passed")); err != nil {

return err

}

return nil

@@ -1671,7 +1677,8 @@ func (s *Server) selfServeBarrier(ctx context.Context, t *leaseToken, row *SelfS

return nil

}

TestHooks.point("advance:barrier->stopping")

- _, err = s.store.AdvanceSelfServe(row.ID, ssBarrier, ssStopping, "", selfServePhaseDeadline(row, ssStopping, now), time.Time{}, row.Request.Nonce, now)

+ _, err = s.store.AdvanceSelfServe(row.ID, ssBarrier, ssStopping, "", selfServePhaseDeadline(row, ssStopping, now), time.Time{}, row.Request.Nonce, now,

+ ssEventTx(row.ID, ssBarrier+":"+ssEvFacts, now, "", "passed"))

return err

}

diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go

index 19b0901d1..8a345f78c 100644

--- a/internal/console/selfserve_pass.go

+++ b/internal/console/selfserve_pass.go

@@ -5,7 +5,6 @@ package console

import (

"context"

- "database/sql"

"errors"

"fmt"

"log"

@@ -205,9 +204,10 @@ func (s *Server) keepSelfServeLease(ctx context.Context, t *leaseToken) (stop fu

}

// drainSelfServeEvents delivers the undelivered events, each as one audit

-// row in the move's app's activity, at most once (DeliverSelfServeEvent),

+// row in the move's app's activity, at most once (drainSelfServeEvent),

// under a context bounded by the lease. A failed delivery stays undelivered

-// for the next tick; the others go on.

+// for the next tick, and the fifth dead-letters it (Task 9); the others go

+// on.

func (s *Server) drainSelfServeEvents(ctx context.Context, t *leaseToken) {

evs, err := s.store.UnemittedSelfServeEvents(selfServeDrainBatch)

if err != nil {

@@ -219,13 +219,7 @@ func (s *Server) drainSelfServeEvents(ctx context.Context, t *leaseToken) {

return

}

dctx, cancel := context.WithDeadline(ctx, t.deadline())

- err := s.store.DeliverSelfServeEvent(dctx, e, func(tx *sql.Tx) error {

- return auditActivityTx(tx, e.Account, e.App, "self-serve move", Activity{

- Action: "instance-move", TargetType: "instance", Target: e.Host, Message: e.Phase, Detail: e.Reason})

- })

+ s.drainSelfServeEvent(dctx, e)

cancel()

- if err != nil {

- log.Printf("console: self-serve move %s: delivering event %d: %v (retrying next tick)", e.ID, e.Seq, err)

- }

}

diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go

index 9a38f5957..51499f066 100644

--- a/internal/console/selfserve_store.go

+++ b/internal/console/selfserve_store.go

@@ -270,6 +270,11 @@ func (s *Store) migrateSelfServe() error {

return err

}

+ // Task 9: an event's failed deliveries, so one that can never be

+ // delivered is dead-lettered (DeliverSelfServeEvent). Additive.

+ if _, err := s.db.Exec(`ALTER TABLE self_serve_events ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0`); err != nil && !strings.Contains(err.Error(), "duplicate column") {

+ return err

+ }

return nil

}

@@ -706,11 +711,16 @@ WHERE id = ? AND request_nonce = ? AND request_consumed = 0 AND request_voided =

// statement moves to the next substep and gives its wait a fresh deadline:

// WHERE request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0

-func (s *Store) ConsumeSelfServeRequest(id, nonce, nextStep string, nextDeadline time.Time) (bool, error) {

- return ssWon(s.db.Exec(`UPDATE self_serve_moves SET

+//

+// also (the decision's events, Task 9) runs in the same transaction, only

+// when the consume wins.

+func (s *Store) ConsumeSelfServeRequest(id, nonce, nextStep string, nextDeadline time.Time, also ...ssAlso) (bool, error) {

+ return s.ssWithAlso(also, func(tx *sql.Tx) (bool, error) {

+ return ssWon(tx.Exec(`UPDATE self_serve_moves SET

request_consumed = 1, phase_step = ?, step_deadline = ?, updated_at = ?

WHERE id = ? AND request_nonce = ? AND request_confirmed <> 0 AND request_consumed = 0`,

- nextStep, ssNanos(nextDeadline), time.Now().UnixNano(), id, nonce))

+ nextStep, ssNanos(nextDeadline), time.Now().UnixNano(), id, nonce))

+ })

}

// AbandonSelfServeRequest gives up a confirmed request whose wait will

@@ -902,13 +912,23 @@ func (s *Store) SetSelfServeEstimate(id string, seconds int64, terms string) err

// RecordSelfServeHealth counts one health probe: attempts + 1, the

// consecutive successes (reset by a failure), the first success kept once

-// set, the last probe's time and error.

+// set, the last probe's time and error; and the probe's event

+// (health:probe, detail "ok" or "failed: <error>", Task 9) in the same

+// transaction.

func (s *Store) RecordSelfServeHealth(id string, probeAt time.Time, ok bool, errText string) error {

- return s.setSelfServe("health", id, `health_attempts = health_attempts + 1,

+ detail := "ok"

+ if !ok {

+ detail = "failed: " + errText

+ }

+ _, err := s.ssWithAlso([]ssAlso{ssEventTx(id, ssHealth+":"+ssEvProbe, probeAt, "", detail)}, func(tx *sql.Tx) (bool, error) {

+ res, err := tx.Exec(`UPDATE self_serve_moves SET health_attempts = health_attempts + 1,

health_consecutive_ok = CASE WHEN ? THEN health_consecutive_ok + 1 ELSE 0 END,

health_first_ok_at = CASE WHEN ? AND health_first_ok_at = 0 THEN ? ELSE health_first_ok_at END,

- health_last_probe_at = ?, health_last_error = ?`,

- ssBool(ok), ssBool(ok), ssNanos(probeAt), ssNanos(probeAt), errText)

+ health_last_probe_at = ?, health_last_error = ?, updated_at = ? WHERE id = ?`,

+ ssBool(ok), ssBool(ok), ssNanos(probeAt), ssNanos(probeAt), errText, time.Now().UnixNano(), id)

+ return true, ssOne(res, err, "health", id)

+ })

+ return err

}

// SetSelfServeFinishGate records which finishing gate the move is at.

@@ -1086,18 +1106,47 @@ WHERE id = ? AND request_nonce = ? AND request_confirmed <> 0 AND request_consum

// reason is also the row's reason, so the view says why the move is going

// back. False: an abort (a cancellation, an earlier decision) was already

-// requested, and it keeps its own time and reason.

-func (s *Store) AbortSelfServe(id, reason string, now time.Time) (bool, error) {

+// requested, and it keeps its own time and reason. The decision is a

+// refusal event (<phase>:refusal, the reason) in the same transaction,

+// after also (Task 9).

+func (s *Store) AbortSelfServe(id, reason string, now time.Time, also ...ssAlso) (bool, error) {

if ssNanos(now) <= 0 {

return false, fmt.Errorf("console: self-serve move %s: an abort needs a time", id)

}

- return ssWon(s.db.Exec(`UPDATE self_serve_moves SET abort_requested_at = ?, outcome_reason = ?, reason = ?, updated_at = ?

+ also = append(also, ssRowEventTx(id, ssEvRefusal, now, reason, ""))

+ return s.ssWithAlso(also, func(tx *sql.Tx) (bool, error) {

+ return ssWon(tx.Exec(`UPDATE self_serve_moves SET abort_requested_at = ?, outcome_reason = ?, reason = ?, updated_at = ?

WHERE id = ? AND outcome = '' AND abort_requested_at = 0`, ssNanos(now), reason, reason, ssNanos(now), id))

+ })

}

// 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 {

- return s.setSelfServe("reason", id, `reason = ?`, reason)

+ tx, err := s.db.Begin()

+ if err != nil {

+ return err

+ }

+ defer tx.Rollback()

+ var was string

+ if err := tx.QueryRow(`SELECT reason FROM self_serve_moves WHERE id = ?`, id).Scan(&was); err != nil {

+ if errors.Is(err, sql.ErrNoRows) {

+ return fmt.Errorf("console: reason: no self-serve move %q", id)

+ }

+ return err

+ }

+ res, err := tx.Exec(`UPDATE self_serve_moves SET reason = ?, updated_at = ? WHERE id = ?`, reason, time.Now().UnixNano(), id)

+ if err := ssOne(res, err, "reason", id); err != nil {

+ return err

+ }

+ if reason != "" && reason != was {

+ if err := ssRowEventTx(id, ssEvWaiting, time.Now().UTC(), reason, "")(tx); err != nil {

+ return err

+ }

+ return tx.Commit()

}

// SetSelfServeStep moves an open row from one substep of its phase to

@@ -1194,7 +1243,11 @@ 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.

+// 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.

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 {

@@ -1206,8 +1259,12 @@ func (s *Store) DeliverSelfServeEvent(ctx context.Context, e SelfServeEvent, del

if err != nil || !won {

return err

}

- if err := deliver(tx); err != nil {

- return err

+ if derr := deliver(tx); derr != nil {

+ _ = tx.Rollback()

+ if ctx.Err() != nil {

+ return derr

+ }

+ return s.selfServeEventDeliveryFailed(e, derr)

}

return tx.Commit()

}