amadan

carlos / platform Public

console: timing marks at the console's first observation, insert-only, with their events; landed_at with the advance into landed; the finaliser derives a done move's five move_timings rows

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude cf2cb871f72a29cf2b81b03aed0f87a65f137730 parent a569797
5 files changed, +297 −12
  • internal/console/selfserve_estimate.go +151 −0
  • internal/console/selfserve_events.go +75 −0
  • internal/console/selfserve_finalise.go +23 −4
  • internal/console/selfserve_job.go +39 −4
  • internal/console/selfserve_store.go +9 −4
diff --git a/internal/console/selfserve_estimate.go b/internal/console/selfserve_estimate.go
index af9739b97..450a1545c 100644
--- a/internal/console/selfserve_estimate.go
+++ b/internal/console/selfserve_estimate.go
@@ -1,8 +1,13 @@
package console
import (
+ "database/sql"
"encoding/json"
+ "errors"
+ "fmt"
"time"
+
+ "github.com/carlosframework/platform/internal/provision"
)
// The self-serve move's estimate (self-serve moves plan 2; spec §6.2). The
@@ -80,3 +85,149 @@ func estimateTermsOf(raw string) estimateTerms {
}
return t
}
+
+// ---- the timings ----------------------------------------------------------------
+
+// timingMark is one mark: when the console first observed it, and the
+// bytes it carries (sealed_at: the final copy; landed_at: the files and the
+// replica; landed_replica: the replica).
+type timingMark struct {
+ At time.Time
+ Bytes int64
+}
+
+// MoveTiming is one move_timings row: a finished move's phase, how long it
+// took and how many bytes it carried, on the region pair From -> To. At is
+// the phase's end, which orders the window.
+type MoveTiming struct {
+ ID, Phase, From, To string
+ Bytes int64
+ Seconds float64
+ At time.Time
+}
+
+// ssMarkTx is a timing mark of row id, insert-only: a mark the row already
+// has keeps its first time and bytes, and only a mark actually inserted
+// carries its timing event (<phase>:timing, reason the mark).
+func ssMarkTx(id, mark string, at time.Time, bytes int64) ssAlso {
+ return func(tx *sql.Tx) error {
+ res, err := tx.Exec(`INSERT INTO self_serve_timings (id, mark, account, app, host, at, bytes)
+SELECT id, ?, account, app, host, ?, ? FROM self_serve_moves WHERE id = ?
+ON CONFLICT (id, mark) DO NOTHING`, mark, ssNanos(at), bytes, id)
+ inserted, err := ssWon(res, err)
+ if err != nil || !inserted {
+ return err
+ }
+ detail := ""
+ if bytes > 0 {
+ detail = fmt.Sprintf("%d bytes", bytes)
+ }
+ return ssRowEventTx(id, ssEvTiming, at, mark, detail)(tx)
+ }
+}
+
+// MarkSelfServeTiming records mark at (with bytes) for row id unless the
+// row has it already, and its timing event, in one transaction. A row that
+// does not exist is an error.
+func (s *Store) MarkSelfServeTiming(id, mark string, at time.Time, bytes int64) error {
+ _, err := s.ssWithAlso([]ssAlso{ssMarkTx(id, mark, at, bytes)}, func(tx *sql.Tx) (bool, error) {
+ var n int
+ if err := tx.QueryRow(`SELECT COUNT(*) FROM self_serve_moves WHERE id = ?`, id).Scan(&n); err != nil {
+ return false, err
+ }
+ if n != 1 {
+ return false, fmt.Errorf("console: timing %s: no self-serve move %q", mark, id)
+ }
+ return true, nil
+ })
+ return err
+}
+
+// SelfServeTimings are row id's marks by name.
+func (s *Store) SelfServeTimings(id string) (map[string]timingMark, error) {
+ rows, err := s.db.Query(`SELECT mark, at, bytes FROM self_serve_timings WHERE id = ?`, id)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := map[string]timingMark{}
+ for rows.Next() {
+ var mark string
+ var at, bytes int64
+ if err := rows.Scan(&mark, &at, &bytes); err != nil {
+ return nil, err
+ }
+ out[mark] = timingMark{At: ssTime(at), Bytes: bytes}
+ }
+ return out, rows.Err()
+}
+
+// InsertMoveTiming inserts m unless its (id, phase) row exists: inserted
+// is false for a row already there, so deriving a move's timings again
+// (the finaliser resumed after a crash) changes nothing.
+func (s *Store) InsertMoveTiming(m MoveTiming) (inserted bool, err error) {
+ if m.ID == "" || m.Phase == "" {
+ return false, errors.New("console: a move timing needs its move id and phase")
+ }
+ return ssWon(s.db.Exec(`INSERT INTO move_timings (id, phase, from_region, to_region, bytes, seconds, at)
+VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (id, phase) DO NOTHING`,
+ m.ID, m.Phase, m.From, m.To, m.Bytes, m.Seconds, ssNanos(m.At)))
+}
+
+// MoveTimings are the region pair's newest rows for phase, at most limit,
+// newest first; withBytes keeps only the rows that carried bytes (a rate's
+// rows).
+func (s *Store) MoveTimings(from, to, phase string, withBytes bool, limit int) ([]MoveTiming, error) {
+ rows, err := s.db.Query(`SELECT id, phase, from_region, to_region, bytes, seconds, at FROM move_timings
+WHERE from_region = ? AND to_region = ? AND phase = ? AND seconds >= 0 AND (? = 0 OR (bytes > 0 AND seconds > 0))
+ORDER BY at DESC, id DESC LIMIT ?`, from, to, phase, ssBool(withBytes), limit)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []MoveTiming
+ for rows.Next() {
+ var m MoveTiming
+ var at int64
+ if err := rows.Scan(&m.ID, &m.Phase, &m.From, &m.To, &m.Bytes, &m.Seconds, &at); err != nil {
+ return nil, err
+ }
+ m.At = ssTime(at)
+ out = append(out, m)
+ }
+ return out, rows.Err()
+}
+
+// selfServeTimingPhases are the five rows a finished move derives: each
+// phase from one mark to another, with the bytes of a third (none for a
+// fixed term).
+var selfServeTimingPhases = []struct{ phase, from, to, bytes string }{
+ {"stop", "offline_at", "stopped_at", ""},
+ {"copy", "seal_at", "sealed_at", "sealed_at"},
+ {"land", "flip_at", "landed_at", "landed_at"},
+ {"open", "open_at", "opened_at", ""},
+ {"promote", "open_at", "first_ok_at", "landed_replica"},
+}
+
+// selfServeMoveTimings are the rows row's marks give: none unless the move
+// finished done; a phase whose start or end mark is missing, or ends
+// before it starts, gives none.
+func selfServeMoveTimings(row *SelfServeMove, marks map[string]timingMark) []MoveTiming {
+ if row.Outcome != provision.OfflineOutcomeDone {
+ return nil
+ }
+ var out []MoveTiming
+ for _, p := range selfServeTimingPhases {
+ start, ok1 := marks[p.from]
+ end, ok2 := marks[p.to]
+ if !ok1 || !ok2 || end.At.Before(start.At) {
+ continue
+ }
+ m := MoveTiming{ID: row.ID, Phase: p.phase, From: row.From, To: row.To, Seconds: end.At.Sub(start.At).Seconds(), At: end.At}
+ if p.bytes != "" {
+ m.Bytes = marks[p.bytes].Bytes
+ }
+ out = append(out, m)
+ }
+ return out
+}
diff --git a/internal/console/selfserve_events.go b/internal/console/selfserve_events.go
new file mode 100644
index 000000000..b7aedda4b
--- /dev/null
+++ b/internal/console/selfserve_events.go
@@ -0,0 +1,75 @@
+package console
+
+// The self-serve move's events (self-serve moves plan 2 Task 9; spec §8):
+// the claims-off move log and the account's member stream. Every phase
+// change, refusal, fact decision, lane probe, health probe and timing mark
+// inserts one event in the transaction of the change that caused it
+// (InsertSelfServeEventTx allocates its per-row seq there); a phase change's
+// 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
+// mode on for ever.
+
+import (
+ "database/sql"
+ "fmt"
+ "time"
+)
+
+// The kinds of event that are not a phase change.
+const (
+ ssEvTiming = "timing" // a timing mark (reason the mark)
+)
+
+// ssAlso is a write a journal change makes in its own transaction, run
+// only when the change wins: an event, a timing mark.
+type ssAlso func(tx *sql.Tx) error
+
+// ssRunAlso runs every write in also inside tx, in order.
+func ssRunAlso(tx *sql.Tx, also []ssAlso) error {
+ for _, a := range also {
+ if err := a(tx); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// ssWithAlso runs stmt in one transaction and, when it wins, also; then
+// commits. A lost stmt writes nothing.
+func (s *Store) ssWithAlso(also []ssAlso, stmt func(tx *sql.Tx) (bool, error)) (bool, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return false, err
+ }
+ defer tx.Rollback()
+ won, err := stmt(tx)
+ if err != nil || !won {
+ return false, err
+ }
+ if err := ssRunAlso(tx, also); err != nil {
+ return false, err
+ }
+ return true, tx.Commit()
+}
+
+// ssEventTx is the event phase (reason, detail) of row id at at.
+func ssEventTx(id, phase string, at time.Time, reason, detail string) ssAlso {
+ return func(tx *sql.Tx) error {
+ return InsertSelfServeEventTx(tx, SelfServeEvent{ID: id, Phase: phase, At: at, Reason: reason, Detail: detail})
+ }
+}
+
+// ssRowEventTx is an event of kind in the row's phase as the transaction
+// reads it ("<phase>:<kind>").
+func ssRowEventTx(id, kind string, at time.Time, reason, detail string) ssAlso {
+ return func(tx *sql.Tx) error {
+ var phase string
+ if err := tx.QueryRow(`SELECT phase FROM self_serve_moves WHERE id = ?`, id).Scan(&phase); err != nil {
+ return fmt.Errorf("console: self-serve move %s: the event's phase: %w", id, err)
+ }
+ return ssEventTx(id, phase+":"+kind, at, reason, detail)(tx)
+ }
+}
diff --git a/internal/console/selfserve_finalise.go b/internal/console/selfserve_finalise.go
index 181e6acea..3f0166621 100644
--- a/internal/console/selfserve_finalise.go
+++ b/internal/console/selfserve_finalise.go
@@ -96,7 +96,26 @@ func (s *Server) selfServeFinaliseRest(ctx context.Context, t *leaseToken, id st
return nil
}
-// selfServeTimingsOnFinish is the finaliser's timings effect: Task 9
-// derives the move_timings rows from the marks here. A no-op until then;
-// it must be idempotent (a crash before effects_done runs it again).
-func (s *Server) selfServeTimingsOnFinish(ctx context.Context, row *SelfServeMove) error { return nil }
+// selfServeTimingsOnFinish is the finaliser's timings effect (Task 9): the
+// move_timings rows the row's marks give (selfServeMoveTimings: five for a
+// move that finished done with every mark, none for an aborted one), each
+// inserted unless it exists, so the effect run again after a crash before
+// effects_done changes nothing.
+func (s *Server) selfServeTimingsOnFinish(ctx context.Context, row *SelfServeMove) error {
+ if row.Outcome != provision.OfflineOutcomeDone {
+ return nil
+ }
+ marks, err := s.store.SelfServeTimings(row.ID)
+ if err != nil {
+ return err
+ }
+ for _, m := range selfServeMoveTimings(row, marks) {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if _, err := s.store.InsertMoveTiming(m); err != nil {
+ return err
+ }
+ }
+ return nil
+}
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index 6c3b905bb..f7e4e11d7 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -424,9 +424,15 @@ func (s *Server) selfServeReconcile(ctx context.Context, t *leaseToken, row *Sel
}
}
window := time.Time{}
+ var also []ssAlso
+ if set.first == ssLanded && selfServeRank[row.Phase] < selfServeRank[ssLanded] {
+ // The row's first sight of the target's landing: the landed marks
+ // go with the advance (Task 9).
+ also = s.selfServeLandedMarks(ctx, row, mv, now)
+ }
TestHooks.point("advance:" + row.Phase + "->" + set.first)
won, err := s.store.AdvanceSelfServeToStep(row.ID, row.Phase, set.first, "", set.substep,
- selfServePhaseDeadline(row, set.first, now), window, nonce, now)
+ selfServePhaseDeadline(row, set.first, now), window, nonce, now, also...)
if err != nil {
return true, err
}
@@ -1589,9 +1595,38 @@ func (s *Server) selfServeReverting(ctx context.Context, t *leaseToken, row *Sel
// selfServeMark is a timing mark: the console's observation time of one of
// the cut's events (offline_at, stopped_at, seal_at, sealed_at, flip_at,
-// landed_at, open_at, opened_at, first_ok_at), insert-only. A no-op until
-// Task 9's timings fill it.
-func (s *Server) selfServeMark(row *SelfServeMove, mark string, at time.Time, bytes int64) {}
+// open_at, opened_at, first_ok_at; landed_at goes with reconciliation's
+// advance, selfServeLandedMarks), insert-only (MarkSelfServeTiming: a
+// second observation keeps the first). A mark that cannot be written is
+// logged and the move goes on: it only costs the estimate that move's row
+// for the phase.
+func (s *Server) selfServeMark(row *SelfServeMove, mark string, at time.Time, bytes int64) {
+ if err := s.store.MarkSelfServeTiming(row.ID, mark, at, bytes); err != nil {
+ log.Printf("console: self-serve move %s (%s): the %s mark: %v", row.ID, row.Host, mark, err)
+ }
+}
+
+// selfServeLandedMarks are the marks of the row's first sight of the
+// target's landing (reconciliation's advance into landed), written in the
+// advancing transaction: landed_at with the files and the replica the
+// landing carried, and landed_replica with the replica alone (the
+// promotion's bytes). The sizes are the seal's (its files archive and its
+// staged replica); a seal that cannot be read gives the marks no bytes, so
+// the move's landing and promotion count for no rate.
+func (s *Server) selfServeLandedMarks(ctx context.Context, row *SelfServeMove, mv *provision.MoveState, now time.Time) []ssAlso {
+ var files, replica int64
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ switch {
+ case err != nil:
+ log.Printf("console: self-serve move %s (%s): the landed marks' sizes: %v (marked without bytes)", row.ID, row.Host, err)
+ case ev.seal != nil:
+ files = ev.seal.FilesBytes
+ for _, o := range ev.seal.Staged {
+ replica += o.Size
+ }
+ }
+ return []ssAlso{ssMarkTx(row.ID, "landed_at", now, files+replica), ssMarkTx(row.ID, "landed_replica", now, replica)}
+}
// selfServeBarrier is the cut's last check before anything goes offline: a
// precheck asked for (OfflinePrecheck), then each source's precheck and the
diff --git a/internal/console/selfserve_store.go b/internal/console/selfserve_store.go
index f308161d4..9a38f5957 100644
--- a/internal/console/selfserve_store.go
+++ b/internal/console/selfserve_store.go
@@ -744,8 +744,8 @@ WHERE id = ? AND request_nonce = ? AND request_consumed = 0 AND request_confirme
// the phase event in the same transaction, entering the new phase at its
// first substep (the job's own progression). It is
// AdvanceSelfServeToStep with selfServeFirstSubstep[to].
-func (s *Store) AdvanceSelfServe(id, from, to, reason string, step, window time.Time, nonce string, now time.Time) (bool, error) {
- return s.AdvanceSelfServeToStep(id, from, to, reason, selfServeFirstSubstep[to], step, window, nonce, now)
+func (s *Store) AdvanceSelfServe(id, from, to, reason string, step, window time.Time, nonce string, now time.Time, also ...ssAlso) (bool, error) {
+ return s.AdvanceSelfServeToStep(id, from, to, reason, selfServeFirstSubstep[to], step, window, nonce, now, also...)
}
// selfServeMaintainEvery is the ready phase's idle interval: a maintenance
@@ -770,8 +770,10 @@ const selfServeMaintainEvery = 10 * time.Minute
// after a void, cannot advance. Entering staging stamps staging_started_at
// (once), and entering ready starts the idle interval
// (next_maintenance_at = now + 10 min), in the same statement. The phase
-// event goes in the same transaction.
-func (s *Store) AdvanceSelfServeToStep(id, from, to, reason, substep string, step, window time.Time, nonce string, now time.Time) (bool, error) {
+// event goes in the same transaction, after also (the events and timing
+// marks of the decision the advance records, Task 9), which run only when
+// the advance wins.
+func (s *Store) AdvanceSelfServeToStep(id, from, to, reason, substep string, step, window time.Time, nonce string, now time.Time, also ...ssAlso) (bool, error) {
var nextMaint int64
if to == ssReady {
nextMaint = ssNanos(now.Add(selfServeMaintainEvery))
@@ -804,6 +806,9 @@ WHERE id = ? AND phase = ? AND outcome = '' AND (abort_requested_at = 0 OR ? IN
if err != nil || !won {
return false, err
}
+ if err := ssRunAlso(tx, also); err != nil {
+ return false, err
+ }
if err := InsertSelfServeEventTx(tx, SelfServeEvent{ID: id, Phase: to, At: now, Reason: reason}); err != nil {
return false, err
}