amadan

carlos / platform Public

console: the self-serve job from requested to ready, reconciling forward with persisted requests, a conditional fenced abandonment and at-least-once notification

selfServeStep: the row's deadlines before any evidence, the judged
record, the outstanding request resolved (an expired one fenced and
classified, never replayed), the reconciliation table's compatible sets
(forward only), the expiry action, the pre-offline abort (no row is
finalised while a request is outstanding), then the phase: requested,
reserving (per-box FIFO, reserveBoxes outside the locks, the record and
start validation again before the queue, the queue's 30 s fence), queued,
reports (the lane, the source precheck and the target-ready decisions),
staging (the ready rule and its bounds), prechecks (the facts again and
the staged book, then the estimate), ready (maintenance rounds with their
own allowance and the cut handoff), restage up to the barrier, and the
minimal reverting step that finishes a pre-offline abort. A minimal
finaliser (outcome, effects, release) Task 8 extends.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude b0df7b197a10c2fe856c6d8a4bdc9138402f1653 parent 9ef6e32
3 files changed, +1631 −9
  • internal/console/selfserve_checks.go +233 −0
  • internal/console/selfserve_job.go +1398 −0
  • internal/console/selfserve_pass.go +0 −9
diff --git a/internal/console/selfserve_checks.go b/internal/console/selfserve_checks.go
new file mode 100644
index 000000000..82bea2d04
--- /dev/null
+++ b/internal/console/selfserve_checks.go
@@ -0,0 +1,233 @@
+package console
+
+// The self-serve job's decisions over the boxes' reports (self-serve moves
+// plan 2 Task 7; spec §7.3). The boxes report facts; the console decides.
+// Every decision answers "" when the facts pass, or the sentence the move
+// aborts with. The console enumerates the path set and the fact set
+// itself, so a fact the box left out is "could not determine <fact> on
+// <label>", exactly as a fact the box could not read (Known false).
+
+import (
+ "fmt"
+ "path/filepath"
+ "sort"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The facts' names, as "could not determine <fact> on <label>" says them.
+const (
+ factRoute = "the route"
+ factTimers = "the timers"
+ factLiveLink = "the live link"
+ factProgramsFile = "the programs file"
+ factParkDir = "the park directory"
+ factMovesPrefix = "the staging prefix"
+ factInstances = "the instances prefix"
+ factLock = "the lock"
+ factArchive = "the archive"
+ factStrays = "the stray entries"
+ factLane = "the staging lane"
+ factSourceFacts = "the source's facts"
+ factTargetFacts = "the target's facts"
+ factStagedCopy = "the staged copy"
+ selfServeReadyMiB = 16 << 20
+)
+
+// couldNotDetermine is the sentence for a fact that is missing or unknown.
+func couldNotDetermine(fact, label string) string {
+ return fmt.Sprintf("could not determine %s on %s", fact, label)
+}
+
+// The decisions' other sentences (copy record, self-serve moves plan 2
+// Task 7, approved by Paul 2026-10-22).
+func ssRouteHeld(host string) string {
+ return fmt.Sprintf("the target box already has a route for %s", host)
+}
+func ssTimersHeld(host string) string {
+ return fmt.Sprintf("the target box has timers for %s", host)
+}
+func ssStaleLiveLink(host, build string) string {
+ return fmt.Sprintf("the target box has a stale live link for %s naming %s", host, build)
+}
+func ssProgramsFile(build string) string {
+ return fmt.Sprintf("the target box's programs file names %s", build)
+}
+func ssDataHeld(host, path string) string {
+ return fmt.Sprintf("the target box already holds data for %s: %s", host, path)
+}
+
+const msgSelfServeParkDirty = "the target box's park directory is not clean"
+
+func ssDiskLow(host string) string {
+ return fmt.Sprintf("the target box has too little free disk for %s", host)
+}
+func ssBucketHeld(host string) string {
+ return fmt.Sprintf("the target bucket already holds objects for %s", host)
+}
+func ssBucketLocked(host, owner string) string {
+ return fmt.Sprintf("the target bucket holds a lock for %s (%s)", host, owner)
+}
+func ssStray(host, path string) string {
+ return fmt.Sprintf("%s has a stray entry beside its files directory: %s", host, path)
+}
+func ssUnarchivable(host, detail string) string {
+ return fmt.Sprintf("%s's files cannot be archived: %s", host, detail)
+}
+func ssStagedMismatch(host string) string {
+ return fmt.Sprintf("the staging prefix for %s does not match what this move staged", host)
+}
+func ssReplaced(label string) string { return fmt.Sprintf("box %s was replaced", label) }
+
+// The reasons the job's own outcomes carry (Task 7's copy rows).
+const (
+ msgSelfServeNotStarted = "the move was not started in time"
+ msgSelfServeConfirmLate = "the move was not confirmed in time"
+ msgSelfServeCouldNotStart = "the move could not be started"
+ msgSelfServeAnotherMove = "another move is in flight"
+ msgSelfServeRecreated = "the instance was recreated"
+ msgSelfServeWaitingAnother = "waiting for another move on this box"
+)
+
+// liveLinkBuild is the build a live link names: the version directory its
+// target sits in (release.VersionPath: <versions>/<build>/[<acct>/]<app>),
+// or the target itself when it does not have that shape.
+func liveLinkBuild(target string) string {
+ parts := strings.Split(filepath.ToSlash(filepath.Clean(target)), "/")
+ for i, p := range parts {
+ if p == "versions" && i+1 < len(parts) {
+ return parts[i+1]
+ }
+ }
+ return target
+}
+
+// targetDataPaths are the paths of host the target must not hold, under the
+// data directory the report's paths name (provision.InstanceDir, the raw
+// nested directory, <dd>/<host>, its env files and its database), keyed by
+// absolute path as the box reports them. The data directory is read from
+// the report's own <host>.db entry, else /data.
+func targetDataPaths(host string, reported map[string]moves.Fact) []string {
+ dd := "/data"
+ for p := range reported {
+ if filepath.Base(p) == host+".db" {
+ dd = filepath.Dir(p)
+ break
+ }
+ }
+ return []string{
+ provision.InstanceDir(dd, host),
+ filepath.Join(dd, ".instances", provision.InstanceID(host)),
+ filepath.Join(dd, host),
+ filepath.Join(dd, host+".env"),
+ filepath.Join(dd, host+".config.env"),
+ filepath.Join(dd, host+".db"),
+ }
+}
+
+// targetReadyDecision decides the target box's ready report for host. Each
+// fact must be known (a missing one cannot be determined) and pass:
+// no route and no timers for the host, no live link naming another build,
+// no entry in the programs file, none of the host's data paths, a clean
+// park directory, free disk for twice the replica, and no lock in the
+// staging bucket. beforeStaging also requires both bucket prefixes known
+// and empty; after staging the instances prefix must be (the staging
+// prefix is the staged copy, which the book comparison decides).
+func targetReadyDecision(host, label string, r *moves.TargetReadyReport, replicaBytes int64, beforeStaging bool) string {
+ if r == nil {
+ return couldNotDetermine(factTargetFacts, label)
+ }
+ switch {
+ case !r.RouteAbsent.Known:
+ return couldNotDetermine(factRoute, label)
+ case !r.RouteAbsent.OK:
+ return ssRouteHeld(host)
+ case !r.TimersAbsent.Known:
+ return couldNotDetermine(factTimers, label)
+ case !r.TimersAbsent.OK:
+ return ssTimersHeld(host)
+ case !r.LiveLink.Known:
+ return couldNotDetermine(factLiveLink, label)
+ case !r.LiveLink.OK:
+ return ssStaleLiveLink(host, liveLinkBuild(r.LiveLink.Detail))
+ case !r.ProgramsFile.Known:
+ return couldNotDetermine(factProgramsFile, label)
+ case !r.ProgramsFile.OK:
+ return ssProgramsFile(r.ProgramsFile.Detail)
+ }
+ for _, p := range targetDataPaths(host, r.DataPaths) {
+ f, ok := r.DataPaths[p]
+ switch {
+ case !ok || !f.Known:
+ return couldNotDetermine(p, label)
+ case !f.OK:
+ return ssDataHeld(host, p)
+ }
+ }
+ switch {
+ case !r.ParkDir.Known:
+ return couldNotDetermine(factParkDir, label)
+ case !r.ParkDir.OK:
+ return msgSelfServeParkDirty
+ case !r.DiskFree.Known || r.DiskFreeBytes < 2*replicaBytes:
+ return ssDiskLow(host)
+ }
+ if beforeStaging {
+ if !r.MovesPrefix.Known {
+ return couldNotDetermine(factMovesPrefix, label)
+ }
+ if len(r.MovesPrefix.Objects) > 0 {
+ return ssBucketHeld(host)
+ }
+ }
+ switch {
+ case !r.InstancesPrefix.Known:
+ return couldNotDetermine(factInstances, label)
+ case len(r.InstancesPrefix.Objects) > 0:
+ return ssBucketHeld(host)
+ case !r.Lock.Known:
+ return couldNotDetermine(factLock, label)
+ case !r.Lock.OK:
+ return ssBucketLocked(host, r.Lock.Detail)
+ }
+ return ""
+}
+
+// sourcePrecheckDecision decides a source box's precheck for host: the
+// archive must take the files, and every stray entry beside the files
+// directory must be an empty directory (the source removes those at the
+// start, after the stop).
+func sourcePrecheckDecision(host, label string, p *moves.SourcePrecheck) string {
+ if p == nil {
+ return couldNotDetermine(factSourceFacts, label)
+ }
+ switch {
+ case !p.ArchiveEligible.Known:
+ return couldNotDetermine(factArchive, label)
+ case !p.ArchiveEligible.OK:
+ return ssUnarchivable(host, p.ArchiveEligible.Detail)
+ case !p.Strays.Known:
+ return couldNotDetermine(factStrays, label)
+ }
+ entries := append([]moves.StrayEntry(nil), p.Strays.Entries...)
+ sort.Slice(entries, func(i, j int) bool { return entries[i].Path < entries[j].Path })
+ for _, e := range entries {
+ switch {
+ case e.Kind == moves.StrayDir && e.Known && e.Entries == 0:
+ // Empty: removed at the start.
+ case e.Kind == moves.StrayDir && !e.Known:
+ return couldNotDetermine(e.Path, label)
+ default:
+ return ssStray(host, e.Path)
+ }
+ }
+ return ""
+}
+
+// stagingReady is the ready rule (spec §6.2, the legacy move's): at least
+// two rounds, and the last one copied and deferred under 16 MiB.
+func stagingReady(r moves.OfflineReport) bool {
+ return r.Rounds >= 2 && r.RoundCopiedBytes+r.RoundDeferredBytes < selfServeReadyMiB
+}
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
new file mode 100644
index 000000000..8bf9344ff
--- /dev/null
+++ b/internal/console/selfserve_job.go
@@ -0,0 +1,1398 @@
+package console
+
+// The self-serve job (self-serve moves plan 2 Task 7; spec §7): one step
+// per open row per tick, from requested to ready, and the reconciliation
+// that lets any crash resume. The record is the truth: a step reads the
+// row's deadlines first, then the judged record, reconciles the row forward
+// to the phases the record admits (never backwards), runs a pending expiry
+// or abort, and only then the phase's own step. Every record write is a
+// persisted request (selfServeApply). A job tick never calls an HTTP door.
+//
+// Task 8 owns the phases from the cut's barrier on (stopping, sealing,
+// landing, landed, open-uncertain, opened, health, finishing), the full
+// reverting rules and the finaliser's timings; the minimal reverting step
+// here carries the pre-offline abort to its finish.
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The waits' allowances and bounds (spec §6.2, §7.2).
+const (
+ selfServeReportsWait = 5 * time.Minute
+ selfServePrechecksWait = 5 * time.Minute
+ selfServeRoundFloor = 10 * time.Minute
+ selfServeStagingBound = 30 * time.Minute
+ selfServeStagingRounds = 10
+ selfServeCopyFloor = 5 * time.Minute
+ selfServeRevertWait = 10 * time.Minute
+ // selfServeQueueFence is how long a queue request may stay unconfirmed
+ // on a record with no move before the job fences the record and gives
+ // the admission up (the reconciliation table's reserving row).
+ selfServeQueueFence = 30 * time.Second
+)
+
+// The substeps (phase_step) of the phases that write more than once.
+const (
+ ssStepRound = "round"
+ ssStepPrecheck = "precheck"
+ ssStepEstimate = "estimate"
+ ssStepAbort = "abort"
+ ssStepFinish = "finish"
+ ssStepAdopted = "adopted"
+ ssStepIssue = "issue"
+ ssStepSeal = "seal"
+ ssStepGate2 = "gate2"
+)
+
+// maxDuration is the larger of a and b.
+func maxDuration(a, b time.Duration) time.Duration {
+ if a > b {
+ return a
+ }
+ return b
+}
+
+// selfServeRoundAllowance is one staging round's: max(10 min, 3 × the
+// copy's estimate).
+func selfServeRoundAllowance(row *SelfServeMove) time.Duration {
+ return maxDuration(selfServeRoundFloor, 3*estimateTermsOf(row.EstimateTerms).Copy)
+}
+
+// selfServeCopyAllowance is a maintenance or cut round's: max(5 min, 2 ×
+// the copy's estimate), as the cut door computes it.
+func selfServeCopyAllowance(row *SelfServeMove) time.Duration {
+ return maxDuration(selfServeCopyFloor, 2*estimateTermsOf(row.EstimateTerms).Copy)
+}
+
+// selfServePhaseDeadline is the step deadline a phase entered at now gets
+// (spec §7.2's table): the reconciliation advance's, and the job's own.
+func selfServePhaseDeadline(row *SelfServeMove, phase string, now time.Time) time.Time {
+ est := estimateTermsOf(row.EstimateTerms)
+ switch phase {
+ case ssQueued, ssReports:
+ return now.Add(selfServeReportsWait)
+ case ssStaging:
+ return now.Add(selfServeRoundAllowance(row))
+ case ssPrechecks:
+ return now.Add(selfServePrechecksWait)
+ case ssRestage, ssBarrier:
+ return now.Add(selfServeCopyAllowance(row))
+ case ssStopping:
+ return now.Add(maxDuration(5*time.Minute, 2*est.Stop))
+ case ssSealing:
+ return now.Add(maxDuration(10*time.Minute, 3*est.Copy))
+ case ssLanding:
+ return now.Add(maxDuration(10*time.Minute, 3*est.Land))
+ case ssLanded, ssOpened:
+ return now.Add(maxDuration(3*time.Minute, 2*est.Open))
+ case ssHealth:
+ return now.Add(maxDuration(3*time.Minute, 2*est.Promote) + 30*time.Second)
+ case ssFinishing, ssReverting:
+ return now.Add(selfServeRevertWait)
+ }
+ return time.Time{}
+}
+
+// selfServeOurs is the record's move when it is this row's.
+func selfServeOurs(rec *provision.Record, row *SelfServeMove) *provision.MoveState {
+ if rec != nil && rec.Move != nil && rec.Move.ID == row.ID {
+ return rec.Move
+ }
+ return nil
+}
+
+// selfServePreOpen reports whether a move state is at or before
+// offline-landed: the states an abort can still take back losslessly.
+func selfServePreOpen(state string) bool {
+ switch state {
+ case provision.OfflineQueued, provision.OfflineStaging, provision.OfflineStopping, provision.OfflineSealing,
+ provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:
+ return true
+ }
+ return false
+}
+
+// selfServeSets is the reconciliation table's compatible sets: for each
+// record state, the row phases that state admits, the set's first phase
+// and the substep reconciliation enters it at. A row in the set stays (a
+// phase whose write has not been issued yet, or whose write is pending,
+// shares its predecessor's record state); any other row is advanced to the
+// first phase. reverting is in every pre-open set: an abort's own write
+// may still be pending.
+var selfServeSets = map[string]struct {
+ phases []string
+ first, substep string
+}{
+ provision.OfflineQueued: {[]string{ssQueued, ssReports, ssStaging, ssReverting}, ssQueued, ""},
+ provision.OfflineStaging: {[]string{ssStaging, ssPrechecks, ssReady, ssRestage, ssBarrier, ssStopping, ssReverting}, ssStaging, ssStepRound},
+ provision.OfflineStopping: {[]string{ssStopping, ssReverting}, ssStopping, ssStepSeal},
+ provision.OfflineSealing: {[]string{ssSealing, ssLanding, ssReverting}, ssSealing, ""},
+ provision.OfflineSealed: {[]string{ssSealing, ssLanding, ssReverting}, ssSealing, ""},
+ provision.OfflineLanding: {[]string{ssLanding, ssReverting}, ssLanding, ""},
+ provision.OfflineLanded: {[]string{ssLanded, ssOpenUncertain, ssReverting}, ssLanded, ""},
+ provision.OfflineOpen: {[]string{ssOpened, ssHealth, ssFinishing}, ssOpened, ""},
+ provision.OfflineFinishing: {[]string{ssFinishing}, ssFinishing, ssStepGate2},
+ provision.OfflineReverting: {[]string{ssReverting}, ssReverting, ssStepFinish},
+}
+
+func containsPhase(phases []string, p string) bool {
+ for _, x := range phases {
+ if x == p {
+ return true
+ }
+ }
+ return false
+}
+
+// selfServeOutstanding reports whether the row has an issued request that
+// is not consumed, and whether it is confirmed.
+func selfServeOutstanding(row *SelfServeMove) (outstanding, confirmed bool) {
+ if row.RequestIssued.IsZero() || row.Request.Consumed {
+ return false, false
+ }
+ return true, row.Request.Confirmed
+}
+
+// selfServeWaiting reports whether the row's outstanding request is a
+// confirmed one of kind: the wait the phase is in.
+func selfServeWaiting(row *SelfServeMove, kinds ...string) bool {
+ out, conf := selfServeOutstanding(row)
+ if !out || !conf {
+ return false
+ }
+ for _, k := range kinds {
+ if row.Request.Kind == k {
+ return true
+ }
+ }
+ return false
+}
+
+// selfServeTransient reports whether a check's refusal is one that clears
+// on its own (a read that failed, a report not yet current), which the job
+// waits out rather than aborting on.
+func selfServeTransient(ref *offlineRefusal) bool {
+ return ref.status >= http.StatusInternalServerError || strings.Contains(ref.msg, "try again") ||
+ strings.HasSuffix(ref.msg, "has not reported recently")
+}
+
+// selfServeStep advances one row by at most one step, in the plan's order:
+// (1) the row's deadlines, decided before any evidence is read; (2) the
+// judged record; (3) the outstanding request resolved (an expired one
+// classified and never replayed) and the row reconciled forward by the
+// record; (4) a pending expiry's action; (5) a requested abort; (6) the
+// phase's own step. A finished row resumes the finaliser.
+func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServeMove) error {
+ if row.Outcome != "" {
+ return s.selfServeFinaliseRest(ctx, t, row.ID)
+ }
+ now := s.selfServeNow()
+
+ // (1) The deadlines.
+ if changed, err := s.selfServeDeadlines(row, now); err != nil {
+ return err
+ } else if changed {
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ }
+
+ // (2) The judged record.
+ rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ gone := errors.Is(err, errRecordGone)
+ if err != nil && !gone {
+ return err
+ }
+
+ // (3) The outstanding request first.
+ if out, conf := selfServeOutstanding(row); (out && !conf) || row.Request.Expired && !row.Request.Consumed {
+ _, rerr := s.selfServeResume(ctx, t, row)
+ switch {
+ case errors.Is(rerr, errRequestExpired):
+ return s.selfServeExpired(ctx, t, row)
+ case rerr == nil, errors.Is(rerr, errRequestVoided), errors.Is(rerr, errRecordGone), errors.Is(rerr, errRequestRefused):
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ rec, err = selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ gone = errors.Is(err, errRecordGone)
+ if err != nil && !gone {
+ return err
+ }
+ case errors.Is(rerr, errReplayNeeded):
+ // The write has not landed and the record is unchanged: the
+ // phase function replays it (below), unless the reserving
+ // row's queue has waited past the fence.
+ default:
+ return rerr
+ }
+ }
+ if !gone {
+ s.selfServeCatchUp(ctx, row, rec)
+ }
+ if done, err := s.selfServeReconcile(ctx, t, row, rec, gone, now); done || err != nil {
+ return err
+ }
+
+ // (4) A pending expiry of this phase.
+ if row.ExpiryPending && row.ExpiryPhase == row.Phase {
+ if done, err := s.selfServeExpiryAction(ctx, t, row, rec, now); done || err != nil {
+ return err
+ }
+ }
+
+ // (5) A requested abort.
+ if !row.AbortRequestedAt.IsZero() {
+ if done, err := s.selfServeAbortStep(ctx, t, row, rec, now); done || err != nil {
+ return err
+ }
+ }
+
+ // (6) The phase.
+ err = s.selfServePhase(ctx, t, row, rec, now)
+ switch {
+ case errors.Is(err, errRequestExpired):
+ return s.selfServeExpired(ctx, t, row)
+ case errors.Is(err, errRequestForeign), errors.Is(err, errRequestRefused), errors.Is(err, errRequestVoided), errors.Is(err, errRecordGone):
+ // Reconciliation decides next tick.
+ return nil
+ }
+ return err
+}
+
+// selfServeReload reads the row again.
+func (s *Server) selfServeReload(id string) (*SelfServeMove, error) {
+ return s.store.SelfServeMove(id)
+}
+
+// selfServeDeadlines is step (1): an expired step deadline, or an expired
+// window while no cut is requested, is recorded as the phase's pending
+// expiry against the exact deadline read (no action yet). A pending expiry
+// of this phase is left as it is.
+func (s *Server) selfServeDeadlines(row *SelfServeMove, now time.Time) (bool, error) {
+ if row.ExpiryPending && row.ExpiryPhase == row.Phase {
+ return false, nil
+ }
+ var kind string
+ var saw time.Time
+ switch {
+ case !row.WindowDeadline.IsZero() && !now.Before(row.WindowDeadline) && row.CutRequestedAt.IsZero():
+ kind, saw = "window", row.WindowDeadline
+ case !row.StepDeadline.IsZero() && !now.Before(row.StepDeadline):
+ kind, saw = "step", row.StepDeadline
+ default:
+ return false, nil
+ }
+ TestHooks.point("expiry:" + row.Phase)
+ 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 {
+ TestHooks.point("abort:" + row.Phase)
+ _, err := s.store.AbortSelfServe(row.ID, reason, now)
+ log.Printf("console: self-serve move %s (%s): aborting: %s", row.ID, row.Host, reason)
+ return err
+}
+
+// ---- reconciliation ------------------------------------------------------------
+
+// selfServeReconcile is step (3)'s second half, the reconciliation table:
+// the record decides, and the row only ever moves forward. It reports
+// whether the step ended (a finalisation, an advance, the queue's fence).
+func (s *Server) selfServeReconcile(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, gone bool, now time.Time) (bool, error) {
+ if gone {
+ if out, conf := selfServeOutstanding(row); out && !conf {
+ // selfServeResume voids it; a request still outstanding here
+ // is resolved next tick before the row is finalised.
+ return true, nil
+ }
+ return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, msgSelfServeRecreated, "")
+ }
+ mv := rec.Move
+ switch {
+ case mv == nil && rec.LastMove != nil && rec.LastMove.ID == row.ID:
+ if out, conf := selfServeOutstanding(row); out && !conf {
+ return true, nil // resolved first
+ }
+ return true, s.selfServeFinalise(ctx, t, row, rec.LastMove.Outcome, s.selfServeOutcomeReason(row, rec.LastMove.Outcome), "")
+ case mv == nil:
+ switch row.Phase {
+ case ssRequested, ssReserving:
+ return s.selfServeQueueFence(ctx, t, row, rec, now)
+ }
+ if out, conf := selfServeOutstanding(row); out && !conf {
+ return true, nil
+ }
+ return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, msgSelfServeCouldNotStart, "")
+ case mv.ID != row.ID:
+ if out, conf := selfServeOutstanding(row); out && !conf {
+ return true, nil
+ }
+ return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, msgSelfServeAnotherMove, "")
+ }
+ set, ok := selfServeSets[mv.State]
+ if !ok || containsPhase(set.phases, row.Phase) {
+ return false, nil
+ }
+ // The record carries this move and the row is behind it: the queue
+ // landed (reserving, requested), or a write the row has not caught up
+ // with did (a crash between a confirmation and its advance, an
+ // operator's transition). Forward to the set's first phase, consuming
+ // a confirmed request whose wait the record has passed.
+ nonce := ""
+ if selfServeWaiting(row, "queue") && set.first == ssQueued {
+ nonce = row.Request.Nonce
+ }
+ if out, conf := selfServeOutstanding(row); out && conf && nonce == "" && set.first != ssStaging {
+ // A confirmed wait the record has passed (an operator went past
+ // it): it is left, never retained across phases.
+ if _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline); err != nil {
+ return true, err
+ }
+ }
+ window := time.Time{}
+ 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)
+ if err != nil {
+ return true, err
+ }
+ if !won {
+ // An abort is requested (the advance's predicate), or the row
+ // moved: the abort step or the next tick decides.
+ return false, nil
+ }
+ return true, nil
+}
+
+// selfServeOutcomeReason is a finished move's reason: none for done; for an
+// abort, the reason the job recorded when it decided it, else a
+// cancellation's.
+func (s *Server) selfServeOutcomeReason(row *SelfServeMove, outcome string) string {
+ if outcome != provision.OfflineOutcomeAborted {
+ return ""
+ }
+ if row.OutcomeReason != "" {
+ return row.OutcomeReason
+ }
+ if !row.AbortRequestedAt.IsZero() {
+ return msgSelfServeCancelled
+ }
+ return msgSelfServeCouldNotStart
+}
+
+// selfServeQueueFence is the reconciliation table's reserving row for a
+// record with no move: an unconfirmed queue request older than 30 s is
+// given up only after the fence (fenceRecord under the app and hostname
+// locks and admission), so its write can never land later; a fenced
+// observation carrying the receipt is the queue landing after all
+// (confirmed, the row continues as queued); otherwise the request is voided
+// and the outcome persisted before the reservations are released.
+// Anything younger, or no queue request, continues the phase.
+func (s *Server) selfServeQueueFence(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) (bool, error) {
+ out, conf := selfServeOutstanding(row)
+ if !out || conf || row.Request.Kind != "queue" || now.Sub(row.RequestIssued) <= selfServeQueueFence {
+ return false, nil
+ }
+ TestHooks.point("queue:fence")
+ frec, err := s.selfServeFence(ctx, t, row)
+ if err != nil {
+ log.Printf("console: self-serve move %s: the queue's fence: %v (pending)", row.ID, err)
+ return true, nil
+ }
+ if frec != nil && frec.Incarnation == row.Incarnation && selfServeHasReceipt(frec, row.Request.Nonce) {
+ return true, s.selfServeConfirmFenced(ctx, row, frec)
+ }
+ TestHooks.point("queue:void")
+ if err := s.store.VoidSelfServeRequest(row.ID, row.Request.Nonce); err != nil {
+ return true, err
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return true, err
+ }
+ return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, msgSelfServeCouldNotStart, "")
+}
+
+// selfServeFence fences the row's record: plan 1's fenceRecord under the
+// app and hostname locks and admission, its write under the lease. It
+// answers the fenced observation's record, nil when the record is absent.
+func (s *Server) selfServeFence(ctx context.Context, t *leaseToken, row *SelfServeMove) (*provision.Record, error) {
+ if !t.valid(s.selfServeNow()) {
+ return nil, errSelfServeLeaseLost
+ }
+ wctx, cancel := s.selfServeWriteCtx(ctx, t)
+ defer cancel()
+ var (
+ res fenceResult
+ obs provision.Observation
+ err error
+ )
+ func() {
+ defer s.lockAppHost(row.Account, row.App, row.Host)()
+ if err = s.admit(row.Host, ""); err != nil {
+ return
+ }
+ res, obs, err = s.fenceRecord(wctx, row.Account, row.App, row.Host, fenceOwner{})
+ }()
+ if err != nil {
+ return nil, err
+ }
+ if res == fenceAbsent || obs.Rec == nil {
+ return nil, nil
+ }
+ return obs.Rec, nil
+}
+
+// selfServeConfirmFenced confirms the row's request from a fenced
+// observation that carries its receipt, and announces it.
+func (s *Server) selfServeConfirmFenced(ctx context.Context, row *SelfServeMove, rec *provision.Record) error {
+ TestHooks.point(row.Request.Kind + ":confirm")
+ ok, err := s.store.ConfirmSelfServeRequest(row.ID, row.Request.Nonce, s.selfServeNow())
+ if err != nil || !ok {
+ return err
+ }
+ s.selfServeNotify(ctx, row.ID, rec.RecordSeq, row.Request.Kind)
+ return nil
+}
+
+// selfServeExpired is the fence-and-classify path of a won expiry (the
+// only place expiry is decided is selfServeApply): the record observed
+// once; its receipt confirms the request (an expiry never undoes a write
+// that landed); absent, the record is fenced so the outstanding write can
+// never land, and the fenced observation is classified the same way;
+// absent again, the request is voided and the abort it obliges recorded in
+// one statement. The abort then runs on a record at or before
+// offline-landed; a record past it was an operator's progress and the row
+// reconciles forward. Never a replay.
+func (s *Server) selfServeExpired(ctx context.Context, t *leaseToken, row *SelfServeMove) error {
+ pr, err := s.store.selfServeRequest(row.ID)
+ if err != nil || pr == nil || !pr.Expired || pr.Consumed {
+ return err
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ if err != nil && !errors.Is(err, errRecordGone) {
+ return err
+ }
+ if err == nil && selfServeHasReceipt(rec, pr.Nonce) {
+ return s.selfServeConfirmFenced(ctx, row, rec)
+ }
+ TestHooks.point(pr.Kind + ":fence")
+ frec, err := s.selfServeFence(ctx, t, row)
+ if err != nil {
+ log.Printf("console: self-serve move %s: fencing the expired %s request: %v (pending)", row.ID, pr.Kind, err)
+ return nil
+ }
+ if frec != nil && frec.Incarnation == row.Incarnation && selfServeHasReceipt(frec, pr.Nonce) {
+ return s.selfServeConfirmFenced(ctx, row, frec)
+ }
+ TestHooks.point(pr.Kind + ":void-expired")
+ if _, err := s.store.VoidExpiredSelfServeRequest(row.ID, pr.Nonce, s.selfServeNow(), msgSelfServeCouldNotStart); err != nil {
+ return err
+ }
+ return nil
+}
+
+// ---- the expiry action and the abort ---------------------------------------------
+
+// selfServeExpiryAction is step (4) for an expiry bound to this phase:
+// while the record is at or before offline-landed, abort, with the reason
+// the phase's wait names. The phases past the no-return boundary and the
+// abort's own (reverting) page and retain their wait (Task 8).
+func (s *Server) selfServeExpiryAction(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) (bool, error) {
+ switch row.Phase {
+ case ssOpened, ssHealth, ssFinishing, ssOpenUncertain, ssReverting:
+ return false, nil // Task 8: page and retain, or warn and finish
+ }
+ if mv := selfServeOurs(rec, row); mv != nil && !selfServePreOpen(mv.State) {
+ return false, nil // a committed open: never aborted (Task 8)
+ }
+ if !row.AbortRequestedAt.IsZero() {
+ return false, nil
+ }
+ return true, s.selfServeAbort(row, s.selfServeExpiryReason(ctx, row, rec, now), now)
+}
+
+// selfServeExpiryReason is what an expired wait names: the window's
+// sentence, or the report the phase was still waiting for.
+func (s *Server) selfServeExpiryReason(ctx context.Context, row *SelfServeMove, rec *provision.Record, now time.Time) string {
+ stager := ""
+ if len(row.Participants.Source) > 0 {
+ stager = row.Participants.Source[0].Label
+ }
+ switch row.Phase {
+ case ssRequested, ssReserving:
+ return msgSelfServeNotStarted
+ case ssReady:
+ if row.ExpiryKind == "window" {
+ return msgSelfServeConfirmLate
+ }
+ return couldNotDetermine(factStagedCopy, stager)
+ case ssStaging, ssRestage:
+ return couldNotDetermine(factStagedCopy, stager)
+ case ssQueued, ssReports:
+ if mv := selfServeOurs(rec, row); mv != nil {
+ if ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv); err == nil {
+ if fleet, ok := s.readOfflineFleet(ctx); ok {
+ if v := s.selfServeReportsVerdict(row, mv, ev, fleet); v.waiting != "" {
+ return v.waiting
+ }
+ }
+ }
+ }
+ return couldNotDetermine(factLane, stager)
+ case ssPrechecks:
+ if mv := selfServeOurs(rec, row); mv != nil {
+ if ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv); err == nil {
+ if v := s.selfServePrechecksVerdict(row, mv, ev); v.waiting != "" {
+ return v.waiting
+ }
+ }
+ }
+ return couldNotDetermine(factTargetFacts, row.Participants.Target.Label)
+ }
+ return msgSelfServeCouldNotStart
+}
+
+// selfServeAbortStep is step (5), the abort before the no-return boundary
+// (the record at or before offline-landed, or carrying no move of this
+// row). No row is finalised while a request is outstanding: an unconfirmed
+// one is fenced and classified first (its receipt confirms it, and the
+// record then carries the move; absent, it is voided); a confirmed wait is
+// abandoned (leaving it). Then no move on the record is the finaliser
+// (aborted; the source never stopped, so it reopened when the abort was
+// asked for), and a move goes to reverting, whose step writes the abort.
+func (s *Server) selfServeAbortStep(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) (bool, error) {
+ mv := selfServeOurs(rec, row)
+ if mv != nil && !selfServePreOpen(mv.State) && mv.State != provision.OfflineReverting {
+ return false, nil // past the boundary: the row reconciles forward (Task 8)
+ }
+ if row.Phase == ssReverting {
+ return false, nil // the abort's own writes
+ }
+ out, conf := selfServeOutstanding(row)
+ switch {
+ case out && !conf:
+ TestHooks.point(row.Request.Kind + ":fence")
+ frec, err := s.selfServeFence(ctx, t, row)
+ if err != nil {
+ log.Printf("console: self-serve move %s: fencing the %s request before the abort: %v (pending)", row.ID, row.Request.Kind, err)
+ return true, nil
+ }
+ if frec != nil && frec.Incarnation == row.Incarnation && selfServeHasReceipt(frec, row.Request.Nonce) {
+ return true, s.selfServeConfirmFenced(ctx, row, frec)
+ }
+ TestHooks.point(row.Request.Kind + ":void")
+ if err := s.store.VoidSelfServeRequest(row.ID, row.Request.Nonce); err != nil {
+ return true, err
+ }
+ return true, nil // the next tick decides from the record
+ case out && conf && row.Request.Kind != "abort" && row.Request.Kind != "finish":
+ TestHooks.point(row.Request.Kind + ":abandon")
+ if _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline); err != nil {
+ return true, err
+ }
+ return true, nil
+ }
+ if mv == nil {
+ if row.SourceReopenedAt.IsZero() {
+ if err := s.store.SetSelfServeReopened(row.ID, row.AbortRequestedAt); err != nil {
+ return true, err
+ }
+ row.SourceReopenedAt = row.AbortRequestedAt
+ }
+ return true, s.selfServeFinalise(ctx, t, row, provision.OfflineOutcomeAborted, s.selfServeOutcomeReason(row, provision.OfflineOutcomeAborted), "")
+ }
+ if (mv.State == provision.OfflineQueued || mv.State == provision.OfflineStaging) && row.SourceReopenedAt.IsZero() {
+ // Never offline: the source served throughout.
+ if err := s.store.SetSelfServeReopened(row.ID, row.AbortRequestedAt); err != nil {
+ return true, err
+ }
+ }
+ TestHooks.point("advance:" + row.Phase + "->" + ssReverting)
+ _, err := s.store.AdvanceSelfServe(row.ID, row.Phase, ssReverting, row.Reason, now.Add(selfServeRevertWait), time.Time{}, "", now)
+ return true, err
+}
+
+// ---- the finaliser -----------------------------------------------------------
+
+// selfServeFinalise ends a row: (1) the outcome (FinishSelfServeMove, which
+// consumes a confirmed clearing request and inserts the terminal event in
+// one transaction); (2) the terminal effects (the serial bump; Task 9's
+// timings); (3) the reservations released; (4) the release recorded. A
+// crash between any two leaves the row in OpenSelfServeMoves and the next
+// tick resumes at the first unfinished step. A row with an unconfirmed
+// request is never finalised: the request is resolved first.
+func (s *Server) selfServeFinalise(ctx context.Context, t *leaseToken, row *SelfServeMove, outcome, reason, warning string) error {
+ if out, conf := selfServeOutstanding(row); out && !conf {
+ return fmt.Errorf("console: self-serve move %s: a %s request is outstanding; it is resolved before the move is finished", row.ID, row.Request.Kind)
+ }
+ if row.Outcome == "" {
+ now := s.selfServeNow()
+ if outcome == provision.OfflineOutcomeAborted && row.AbortFinishedAt.IsZero() && !row.AbortRequestedAt.IsZero() {
+ if err := s.store.SetSelfServeAbortFinished(row.ID, now); err != nil {
+ return err
+ }
+ }
+ TestHooks.point("finalise:outcome")
+ if err := s.store.FinishSelfServeMove(row.ID, outcome, reason, warning, now); err != nil {
+ return err
+ }
+ log.Printf("console: self-serve move %s (%s): %s %s", row.ID, row.Host, outcome, reason)
+ }
+ return s.selfServeFinaliseRest(ctx, t, row.ID)
+}
+
+// selfServeFinaliseRest is the finaliser from step (2) on, for a row that
+// has its outcome.
+func (s *Server) selfServeFinaliseRest(ctx context.Context, t *leaseToken, id string) error {
+ row, err := s.selfServeReload(id)
+ if err != nil || row == nil || row.Outcome == "" {
+ return err
+ }
+ if !row.EffectsDone {
+ TestHooks.point("finalise:effects")
+ if err := s.bumpSerialErr(ctx); err != nil {
+ return fmt.Errorf("console: self-serve move %s: the terminal bump: %w (retried next tick)", id, err)
+ }
+ if err := s.store.SetSelfServeEffectsDone(id); err != nil {
+ return err
+ }
+ }
+ if row.ReleasePending {
+ if !t.valid(s.selfServeNow()) {
+ return errSelfServeLeaseLost
+ }
+ TestHooks.point("finalise:release")
+ wctx, cancel := s.selfServeWriteCtx(ctx, t)
+ err := s.releaseBoxes(wctx, moves.OfflineKindMove, id)
+ cancel()
+ if err != nil {
+ return err
+ }
+ TestHooks.point("finalise:released")
+ if err := s.store.ClearSelfServeRelease(id); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// ---- the phases --------------------------------------------------------------
+
+// selfServePhase is step (6): the phase's own step.
+func (s *Server) selfServePhase(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ switch row.Phase {
+ case ssRequested:
+ TestHooks.point("advance:requested->reserving")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssRequested, ssReserving, row.Reason, time.Time{}, row.WindowDeadline, "", now)
+ return err
+ case ssReserving:
+ return s.selfServeReserving(ctx, t, row, now)
+ case ssQueued:
+ TestHooks.point("advance:queued->reports")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssQueued, ssReports, "", now.Add(selfServeReportsWait), time.Time{}, "", now)
+ return err
+ case ssReports:
+ return s.selfServeReports(ctx, t, row, rec, now)
+ case ssStaging:
+ return s.selfServeStaging(ctx, t, row, rec, now)
+ case ssPrechecks:
+ return s.selfServePrechecks(ctx, t, row, rec, now)
+ case ssReady:
+ return s.selfServeReady(ctx, t, row, rec, now)
+ case ssRestage:
+ return s.selfServeRestage(ctx, t, row, rec, now)
+ case ssReverting:
+ return s.selfServeReverting(ctx, t, row, rec, now)
+ }
+ // barrier through finishing: Task 8.
+ return nil
+}
+
+// selfServeReplaced is the identity check at a gate: the label of a pinned
+// participant whose current status report names another box id or
+// instance (moves.ParticipantStillPresent), or no report at all; a report
+// too old or too far ahead to judge (moves.OfflineIdentityFresh) decides
+// nothing yet.
+func selfServeReplaced(row *SelfServeMove, f offlineFleet, now time.Time) string {
+ for _, p := range append(append([]provision.Participant(nil), row.Participants.Source...), row.Participants.Target) {
+ m := f.metrics[p.Label]
+ if m == nil {
+ if _, reported := f.generated[p.Label]; !reported {
+ return p.Label
+ }
+ continue
+ }
+ age := now.Sub(f.generated[p.Label])
+ if age > moves.OfflineIdentityFresh || age < -moves.OfflineRequestSkew {
+ continue
+ }
+ if !moves.ParticipantStillPresent(p, m) {
+ return p.Label
+ }
+ }
+ return ""
+}
+
+// selfServeSelectionMoved names a pinned participant the start validation
+// no longer selects, by box id and instance.
+func selfServeSelectionMoved(row *SelfServeMove, sel offlineSelection) string {
+ have := map[string]string{}
+ for _, p := range append(append([]provision.Participant(nil), sel.Source...), sel.Target) {
+ have[p.BoxID] = p.InstanceID
+ }
+ for _, p := range append(append([]provision.Participant(nil), row.Participants.Source...), row.Participants.Target) {
+ if inst, ok := have[p.BoxID]; !ok || (p.InstanceID != "" && inst != p.InstanceID) {
+ return p.Label
+ }
+ }
+ return ""
+}
+
+// selfServeReserving reserves the participants (per-box FIFO first), then
+// checks the fresh judged record again in full (its class, an open run, a
+// move, the record checks and the start validation with the stage rules and
+// the participants' identities) and queues the move.
+func (s *Server) selfServeReserving(ctx context.Context, t *leaseToken, row *SelfServeMove, now time.Time) error {
+ open, err := s.store.OpenSelfServeMoves()
+ if err != nil {
+ return err
+ }
+ if selfServeWaitsForEarlier(open, row) {
+ if row.Reason != msgSelfServeWaitingEarlier {
+ return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingEarlier)
+ }
+ return nil
+ }
+ fleet, ok := s.readOfflineFleet(ctx)
+ if !ok {
+ return nil // the fleet is read again next tick
+ }
+ if label := selfServeReplaced(row, fleet, now); label != "" {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ if !t.valid(s.selfServeNow()) {
+ return errSelfServeLeaseLost
+ }
+ sel := offlineSelection{From: row.From, To: row.To, Source: row.Participants.Source, Target: row.Participants.Target}
+ wctx, cancel := s.selfServeWriteCtx(ctx, t)
+ TestHooks.point("reserve")
+ ref := s.reserveBoxes(wctx, sel, moves.OfflineReservation{Kind: moves.OfflineKindMove, ID: row.ID, Host: row.Host,
+ Account: row.Account, App: row.App, ExpectIncarnation: row.Incarnation})
+ cancel()
+ switch {
+ case ref == nil:
+ case ref == offlineReserveUnreadable:
+ return fmt.Errorf("console: self-serve move %s: the reservations could not be read (retried next tick)", row.ID)
+ case ref == offlineReserveMismatch:
+ return s.selfServeAbort(row, ref.msg, now)
+ default:
+ if row.Reason != msgSelfServeWaitingAnother {
+ return s.store.SetSelfServeReason(row.ID, msgSelfServeWaitingAnother)
+ }
+ return nil
+ }
+
+ // Reserved: the fresh judged record, checked again before the queue.
+ rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
+ if err != nil {
+ return err
+ }
+ if out, conf := selfServeOutstanding(row); !(out && !conf && row.Request.Kind == "queue") {
+ // A queue already issued is replayed as it was decided; a new one
+ // is decided on this record.
+ if ref := selfServeBusy(rec); ref != nil {
+ return s.selfServeRefused(row, ref, now)
+ }
+ if ref := s.offlineRecordChecks(ctx, row.Account, row.App, row.Host, rec); ref != nil {
+ return s.selfServeRefused(row, ref, now)
+ }
+ }
+ sel, ref = s.offlineRegionChecks(ctx, row.Account, row.App, row.Host, rec, s.homeRegionOut(row.To), true, true, "")
+ if ref != nil {
+ return s.selfServeRefused(row, ref, now)
+ }
+ if label := selfServeSelectionMoved(row, sel); label != "" {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ ss := provision.SelfServeMove{ID: row.ID, Incarnation: row.Incarnation, DisplayName: row.DisplayName,
+ RequestedBy: row.RequestedBy, EstimateSeconds: int(row.EstimateSeconds)}
+ _, err = s.selfServeApply(ctx, t, row, request{Kind: "queue", Phase: ssReserving}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineQueue(row.ID, row.From, row.To, row.RequestedBy, sel.Stage, row.Participants, ss, at, rc)
+ })
+ if err != nil {
+ var refusal *requestRefusal
+ if errors.As(err, &refusal) {
+ // The record refuses the queue: the request stays outstanding
+ // until the queue's fence gives the admission up.
+ log.Printf("console: self-serve move %s: %v", row.ID, err)
+ return nil
+ }
+ return err
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ TestHooks.point("advance:reserving->queued")
+ _, err = s.store.AdvanceSelfServe(row.ID, ssReserving, ssQueued, "", now.Add(selfServeReportsWait), time.Time{}, row.Request.Nonce, now)
+ return err
+}
+
+// selfServeRefused is a check's refusal at a gate: a transient one is
+// waited out (the row's window or step deadline bounds it); any other
+// aborts with its sentence.
+func (s *Server) selfServeRefused(row *SelfServeMove, ref *offlineRefusal, now time.Time) error {
+ if selfServeTransient(ref) {
+ log.Printf("console: self-serve move %s: waiting: %s", row.ID, ref.msg)
+ return nil
+ }
+ return s.selfServeAbort(row, ref.msg, now)
+}
+
+// ssVerdict is a wait's decision over the reports: pass, fail with the
+// sentence the move aborts with, or still waiting for what waiting names.
+type ssVerdict struct {
+ pass bool
+ fail, waiting string
+}
+
+// selfServeReplicaBytes is the replica's size as the source box last
+// measured it for the host.
+func selfServeReplicaBytes(row *SelfServeMove, f offlineFleet) int64 {
+ if len(row.Participants.Source) == 0 {
+ return 0
+ }
+ if m := f.metrics[row.Participants.Source[0].Label]; m != nil {
+ return m.Hosts[row.Host].ReplicaBytes
+ }
+ return 0
+}
+
+// selfServeReportsVerdict is the reports phase's decision: the lane from
+// the stager, each source's precheck and the target's ready report, each
+// from its pinned participant and answering the record's precheck stamp
+// (Request == PrecheckAt), read before anything later overwrites their
+// keys; then the facts.
+func (s *Server) selfServeReportsVerdict(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence, f offlineFleet) ssVerdict {
+ if len(row.Participants.Source) == 0 || mv.PrecheckAt == nil {
+ return ssVerdict{fail: msgSelfServeCouldNotStart}
+ }
+ host, src, tgt, at := row.Host, row.Participants.Source[0], row.Participants.Target, *mv.PrecheckAt
+ lane, ok := ev.get(moves.OfflineRoleLane, src.Label)
+ if !ok || lane.Lane == nil {
+ return ssVerdict{waiting: couldNotDetermine(factLane, src.Label)}
+ }
+ if !lane.Lane.OK {
+ return ssVerdict{fail: s.selfServeNoLane(row.From, row.To).msg}
+ }
+ for _, p := range row.Participants.Source {
+ pre, ok := ev.get(moves.OfflineRoleSource, p.Label)
+ if !ok || pre.Phase != moves.OfflinePhasePrecheck || !pre.Request.Equal(at) {
+ return ssVerdict{waiting: couldNotDetermine(factSourceFacts, p.Label)}
+ }
+ if why := sourcePrecheckDecision(host, p.Label, pre.Precheck); why != "" {
+ return ssVerdict{fail: why}
+ }
+ }
+ rd, ok := ev.get(moves.OfflineRoleTarget, tgt.Label)
+ if !ok || rd.Phase != moves.OfflinePhaseReady || !rd.Request.Equal(at) {
+ return ssVerdict{waiting: couldNotDetermine(factTargetFacts, tgt.Label)}
+ }
+ if why := targetReadyDecision(host, tgt.Label, rd.Ready, selfServeReplicaBytes(row, f), true); why != "" {
+ return ssVerdict{fail: why}
+ }
+ return ssVerdict{pass: true}
+}
+
+// selfServeReports waits for the queued move's reports and starts the
+// pre-copy: OfflineQueuedToStaging, then staging, which waits for the
+// stager's round.
+func (s *Server) selfServeReports(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineQueued {
+ return nil
+ }
+ stage := func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineQueuedToStaging(row.ID, at, rc)
+ }
+ if out, conf := selfServeOutstanding(row); !(out && !conf && row.Request.Kind == "stage") {
+ fleet, ok := s.readOfflineFleet(ctx)
+ if !ok {
+ return nil
+ }
+ if label := selfServeReplaced(row, fleet, now); label != "" {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ v := s.selfServeReportsVerdict(row, mv, ev, fleet)
+ switch {
+ case v.fail != "":
+ return s.selfServeAbort(row, v.fail, now)
+ case !v.pass:
+ return nil
+ }
+ }
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "stage", Phase: ssReports}, nil, stage); err != nil {
+ 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)
+ return err
+}
+
+// selfServePreOfflineRefusal is a pre-offline write's mutation refusing the
+// record it was bound to: the move could not be started, and the abort
+// fences and voids the request. Any other error is returned.
+func (s *Server) selfServePreOfflineRefusal(row *SelfServeMove, err error, now time.Time) error {
+ var refusal *requestRefusal
+ if errors.As(err, &refusal) {
+ log.Printf("console: self-serve move %s: %v", row.ID, err)
+ return s.selfServeAbort(row, msgSelfServeCouldNotStart, now)
+ }
+ return err
+}
+
+// selfServeRestageMut is the restage the staging, ready and restage
+// phases write: staged_at moves to the proposed stamp.
+func selfServeRestageMut(id string) func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineRestage(id, at, rc)
+ }
+}
+
+// selfServeStagerReport is the stager's report for the round the row
+// waits on (StageRequest == the confirmed request's stamp), if it is in.
+func selfServeStagerReport(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence) (offlineReport, bool) {
+ rep, ok := ev.get(moves.OfflineRoleStager, mv.Stager)
+ if !ok || !rep.StageRequest.Equal(row.StageRequest) {
+ return offlineReport{}, false
+ }
+ return rep, true
+}
+
+// selfServeSuperseded is supersession by stamp: the record's stamp for the
+// awaited field (staged_at for a round, precheck_at for a precheck) is
+// later than the confirmed request's proposed stamp, so an operator round
+// moved it without a receipt. The job abandons its wait (one UPDATE) and
+// issues a fresh bound round on the next step; it never adopts the
+// operator's round.
+func (s *Server) selfServeSuperseded(row *SelfServeMove, mv *provision.MoveState) (bool, error) {
+ var stamp *time.Time
+ switch row.Request.Kind {
+ case "stage", "maintain":
+ stamp = mv.StagedAt
+ case "precheck":
+ stamp = mv.PrecheckAt
+ default:
+ return false, nil
+ }
+ if stamp == nil || !stamp.After(row.Request.Proposed) {
+ return false, nil
+ }
+ TestHooks.point(row.Request.Kind + ":abandon")
+ _, err := s.store.AbandonSelfServeRequest(row.ID, row.Request.Nonce, row.StepDeadline)
+ return true, err
+}
+
+// selfServeStaging is the pre-copy, round after round: a round's report
+// (the stager's, for the confirmed request's stamp) either meets the ready
+// rule (or the bound: 30 minutes or 10 rounds) and the row goes to the
+// prechecks, consuming it in the advance, or is consumed and the next
+// round is issued. A failed round aborts with its detail; a round past its
+// allowance is the step deadline.
+func (s *Server) selfServeStaging(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineStaging {
+ return nil
+ }
+ if selfServeWaiting(row, "stage", "maintain") {
+ if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ return err
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ rep, ok := selfServeStagerReport(row, mv, ev)
+ switch {
+ case !ok:
+ return nil
+ case rep.Phase == moves.OfflinePhaseFailed:
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ case rep.Phase != moves.OfflinePhaseStaged:
+ return nil
+ }
+ if stagingReady(rep.OfflineReport) || now.Sub(row.StagingStartedAt) > selfServeStagingBound || rep.Rounds >= selfServeStagingRounds {
+ TestHooks.point("advance:staging->prechecks")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssStaging, ssPrechecks, "", now.Add(selfServePrechecksWait), time.Time{}, row.Request.Nonce, now)
+ return err
+ }
+ TestHooks.point("stage:consume")
+ if _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepRound, now.Add(selfServeRoundAllowance(row))); err != nil {
+ return err
+ }
+ var rerr error
+ if row, rerr = s.selfServeReload(row.ID); rerr != nil || row == nil {
+ return rerr
+ }
+ }
+ if out, _ := selfServeOutstanding(row); out && row.Request.Kind != "stage" {
+ return nil // another request: reconciliation decides
+ }
+ if !row.StagingStartedAt.IsZero() && now.Sub(row.StagingStartedAt) > selfServeStagingBound && !row.StageRequest.IsZero() {
+ // Past the bound with no round outstanding: the prechecks run from
+ // the last completed round.
+ if out, _ := selfServeOutstanding(row); !out {
+ TestHooks.point("advance:staging->prechecks")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssStaging, ssPrechecks, "", now.Add(selfServePrechecksWait), time.Time{}, "", now)
+ return err
+ }
+ }
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "stage", Phase: ssStaging}, nil, selfServeRestageMut(row.ID)); err != nil {
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ return nil
+}
+
+// selfServePrechecksVerdict is the prechecks' decision: each source's
+// precheck and the target's ready report answering the confirmed precheck
+// (Request == the row's precheck request), the facts again, the target's
+// instances prefix known and empty, and the stager's book of the last
+// round equal to the target's staging prefix (moves.StagedMatches).
+func (s *Server) selfServePrechecksVerdict(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence) ssVerdict {
+ host, tgt, at := row.Host, row.Participants.Target, row.PrecheckRequest
+ for _, p := range row.Participants.Source {
+ pre, ok := ev.get(moves.OfflineRoleSource, p.Label)
+ if !ok || pre.Phase != moves.OfflinePhasePrecheck || !pre.Request.Equal(at) {
+ return ssVerdict{waiting: couldNotDetermine(factSourceFacts, p.Label)}
+ }
+ if why := sourcePrecheckDecision(host, p.Label, pre.Precheck); why != "" {
+ return ssVerdict{fail: why}
+ }
+ }
+ rd, ok := ev.get(moves.OfflineRoleTarget, tgt.Label)
+ if !ok || rd.Phase != moves.OfflinePhaseReady || !rd.Request.Equal(at) {
+ return ssVerdict{waiting: couldNotDetermine(factTargetFacts, tgt.Label)}
+ }
+ stager, ok := ev.get(moves.OfflineRoleStager, mv.Stager)
+ if !ok || stager.Phase != moves.OfflinePhaseStaged {
+ return ssVerdict{waiting: couldNotDetermine(factStagedCopy, mv.Stager)}
+ }
+ if why := targetReadyDecision(host, tgt.Label, rd.Ready, stager.ReplicaBytes, false); why != "" {
+ return ssVerdict{fail: why}
+ }
+ if err := moves.StagedMatches(stager.OfflineReport, mv.StageBucket, mv.StageRegion, row.StageRequest, rd.Ready.MovesPrefix, false); err != nil {
+ log.Printf("console: self-serve move %s: %v", row.ID, err)
+ return ssVerdict{fail: ssStagedMismatch(host)}
+ }
+ return ssVerdict{pass: true}
+}
+
+// selfServePrechecks asks for a precheck (OfflinePrecheck), decides its
+// reports, then refines the estimate on the row and writes it to the
+// record (OfflineSetEstimate; the row is the truth), and goes ready with
+// the 24-hour confirmation window.
+func (s *Server) selfServePrechecks(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineStaging {
+ return nil
+ }
+ if row.PhaseStep == ssStepPrecheck {
+ if selfServeWaiting(row, "precheck") {
+ if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ return err
+ }
+ fleet, ok := s.readOfflineFleet(ctx)
+ if !ok {
+ return nil
+ }
+ if label := selfServeReplaced(row, fleet, now); label != "" {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ v := s.selfServePrechecksVerdict(row, mv, ev)
+ switch {
+ case v.fail != "":
+ return s.selfServeAbort(row, v.fail, now)
+ case !v.pass:
+ return nil
+ }
+ if err := s.selfServeRefineEstimate(row, mv, ev); err != nil {
+ return err
+ }
+ TestHooks.point("precheck:consume")
+ if _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepEstimate, row.StepDeadline); err != nil {
+ return err
+ }
+ return nil
+ }
+ if out, _ := selfServeOutstanding(row); out && row.Request.Kind != "precheck" {
+ return nil
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "precheck", Phase: ssPrechecks}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflinePrecheck(row.ID, at, rc)
+ })
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ // The estimate.
+ if !selfServeWaiting(row, "estimate") {
+ if out, _ := selfServeOutstanding(row); out && row.Request.Kind != "estimate" {
+ return nil
+ }
+ seconds := int(row.EstimateSeconds)
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "estimate", Phase: ssPrechecks}, nil,
+ func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineSetEstimate(row.ID, seconds, rc)
+ }); err != nil {
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ var err error
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ }
+ TestHooks.point("advance:prechecks->ready")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssPrechecks, ssReady, "", time.Time{}, now.Add(selfServeConfirmWindow), row.Request.Nonce, now)
+ return err
+}
+
+// selfServeRefineEstimate stores the estimate made from what staging
+// measured: the last round's copied and deferred bytes still to carry,
+// the source's files and the replica.
+func (s *Server) selfServeRefineEstimate(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence) error {
+ stager, ok := ev.get(moves.OfflineRoleStager, mv.Stager)
+ if !ok {
+ return nil
+ }
+ var files int64
+ for _, p := range row.Participants.Source {
+ if pre, ok := ev.get(moves.OfflineRoleSource, p.Label); ok && pre.Precheck != nil {
+ files += pre.Precheck.FilesBytes
+ }
+ }
+ terms, seconds := s.estimate(row.From, row.To, stager.RoundCopiedBytes+stager.RoundDeferredBytes, files, stager.ReplicaBytes)
+ raw, err := json.Marshal(terms)
+ if err != nil {
+ return err
+ }
+ return s.store.SetSelfServeEstimate(row.ID, int64(seconds), string(raw))
+}
+
+// selfServeReady is the ready phase: idle between rounds (no step
+// deadline; the only deadline is the 24-hour window), a maintenance
+// restage every 10 minutes with its own allowance, consumed back to idle
+// only while no cut is requested; and the cut handoff: an outstanding
+// maintenance round becomes the cut's first round (restage, adopted), and
+// with none restage issues its own (restage, issue).
+func (s *Server) selfServeReady(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineStaging {
+ return nil
+ }
+ out, conf := selfServeOutstanding(row)
+ maintain := out && row.Request.Kind == "maintain"
+ if !row.CutRequestedAt.IsZero() {
+ if !maintain {
+ TestHooks.point("advance:ready->restage")
+ _, err := s.store.AdvanceSelfServeToStep(row.ID, ssReady, ssRestage, "", ssStepIssue, now.Add(selfServeCopyAllowance(row)), time.Time{}, "", now)
+ return err
+ }
+ if !conf {
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "maintain", Phase: ssReady}, nil, selfServeRestageMut(row.ID))
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ rep, ok := selfServeStagerReport(row, mv, ev)
+ switch {
+ case !ok:
+ return nil
+ case rep.Phase == moves.OfflinePhaseFailed:
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ case rep.Phase != moves.OfflinePhaseStaged:
+ return nil
+ }
+ TestHooks.point("advance:ready->restage")
+ _, err = s.store.AdvanceSelfServeToStep(row.ID, ssReady, ssRestage, "", ssStepAdopted, now.Add(selfServeCopyAllowance(row)), time.Time{}, row.Request.Nonce, now)
+ return err
+ }
+ if maintain {
+ if !conf {
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "maintain", Phase: ssReady}, nil, selfServeRestageMut(row.ID))
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ return err
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ rep, ok := selfServeStagerReport(row, mv, ev)
+ switch {
+ case !ok:
+ return nil
+ case rep.Phase == moves.OfflinePhaseFailed:
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ case rep.Phase != moves.OfflinePhaseStaged:
+ return nil
+ }
+ TestHooks.point("maintain:consume")
+ won, err := s.store.ConsumeSelfServeMaintenance(row.ID, row.Request.Nonce, now.Add(selfServeMaintainEvery))
+ if err != nil || won {
+ return err
+ }
+ // A cut won meanwhile: the next step re-reads the row and adopts
+ // this round as the cut's first.
+ return nil
+ }
+ if out {
+ return nil // another request: reconciliation decides
+ }
+ if now.Before(row.NextMaintenanceAt) {
+ return nil
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "maintain", Phase: ssReady, Deadline: now.Add(selfServeCopyAllowance(row))}, nil, selfServeRestageMut(row.ID))
+ return s.selfServePreOfflineRefusal(row, err, now)
+}
+
+// selfServeRestage is the cut's restage: substep issue writes its own
+// round and waits for the stager's report of it; substep adopted (the
+// maintenance round the cut found outstanding, already reported) has
+// nothing to write. Both go to the barrier (Task 8), consuming the round.
+func (s *Server) selfServeRestage(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineStaging {
+ return nil
+ }
+ barrier := now.Add(selfServeCopyAllowance(row))
+ if row.PhaseStep == ssStepAdopted {
+ TestHooks.point("advance:restage->barrier")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssRestage, ssBarrier, "", barrier, time.Time{}, "", now)
+ return err
+ }
+ if selfServeWaiting(row, "stage") && row.Request.Phase == ssRestage {
+ if sup, err := s.selfServeSuperseded(row, mv); sup || err != nil {
+ return err
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ rep, ok := selfServeStagerReport(row, mv, ev)
+ switch {
+ case !ok:
+ return nil
+ case rep.Phase == moves.OfflinePhaseFailed:
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ case rep.Phase != moves.OfflinePhaseStaged:
+ return nil
+ }
+ TestHooks.point("advance:restage->barrier")
+ _, err = s.store.AdvanceSelfServe(row.ID, ssRestage, ssBarrier, "", barrier, time.Time{}, row.Request.Nonce, now)
+ return err
+ }
+ if out, _ := selfServeOutstanding(row); out && !(row.Request.Kind == "stage" && row.Request.Phase == ssRestage) {
+ return nil
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "stage", Phase: ssRestage}, nil, selfServeRestageMut(row.ID))
+ return s.selfServePreOfflineRefusal(row, err, now)
+}
+
+// selfServeReverting carries an abort to its finish: substep abort writes
+// it (OfflineAbortAt for a queued move, which clears it and is finalised
+// from last_move; OfflineAbort otherwise, a repeated one on a reverting
+// record included), then substep finish waits for the abort's reports (the
+// operator door's own finish check at offline-reverting) and writes
+// OfflineFinish, which the finaliser consumes with the outcome. Task 8
+// adds the source-reopened and abort-finished facts after a start.
+func (s *Server) selfServeReverting(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil {
+ return nil // reconciliation finalises from last_move
+ }
+ if row.PhaseStep != ssStepFinish {
+ if !selfServeWaiting(row, "abort") {
+ if out, _ := selfServeOutstanding(row); out && row.Request.Kind != "abort" {
+ return nil
+ }
+ queued := mv.State == provision.OfflineQueued
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "abort", Phase: ssReverting}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ if queued {
+ return provision.OfflineAbortAt(row.ID, at, rc)
+ }
+ return provision.OfflineAbort(row.ID, rc)
+ })
+ if err != nil {
+ return err
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ if queued {
+ return nil // cleared: the next step finalises from last_move
+ }
+ }
+ TestHooks.point("abort:consume")
+ _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepFinish, now.Add(selfServeRevertWait))
+ return err
+ }
+ if mv.State != provision.OfflineReverting {
+ return nil
+ }
+ if !selfServeWaiting(row, "finish") {
+ if out, _ := selfServeOutstanding(row); out && row.Request.Kind != "finish" {
+ return nil
+ }
+ if out, _ := selfServeOutstanding(row); !out {
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ d := offlineDoor{app: row.App, host: row.Host, rec: rec, job: true}
+ if _, ref := s.offlineStepFor(ctx, d, mv, ev, now); ref != nil {
+ return nil // the abort's reports are not all in
+ }
+ if row.AbortFinishedAt.IsZero() {
+ if err := s.store.SetSelfServeAbortFinished(row.ID, now); err != nil {
+ return err
+ }
+ }
+ }
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "finish", Phase: ssReverting}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineFinish(row.ID, at, rc)
+ }); err != nil {
+ return err
+ }
+ }
+ // Cleared: the next step finalises from last_move, consuming the
+ // confirmed finish with the outcome.
+ return nil
+}
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index 927d651c3..26e587c22 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -224,12 +224,3 @@ func (s *Server) drainSelfServeEvents(ctx context.Context, t *leaseToken) {
}
}
}
-
-// selfServeStep advances one row by at most one step. Task 7 writes it: the
-// row's deadlines, the judged record, reconciliation by the compatible
-// sets, the expiry action, abort and the phase function, every record write
-// through selfServeApply under selfServeWriteCtx. Until then it leaves
-// every row exactly as it found it.
-func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServeMove) error {
- return nil
-}