amadan

carlos / platform Public

console: the self-serve cut from the barrier to finishing, the no-return boundary at the committed open, open-uncertain decided by the record, the health check, the abort facts without the target, and durable paging

barrier (precheck, the book with nothing deferred), stopping (the start
validation in full, then the seal over the pinned sources), sealing,
landing (the flip), landed (the open record, then the open at its stamp),
open-uncertain (back to landed before the open's deadline, fenced and
classified after it), opened, health (probeTarget, two passes, the window's
warning), finishing (both gates, never bypassed); every expiry decided by
the record: abort at or before offline-landed, page and retain after,
warning and finishing for the health window. reverting records the source
reopened only on the sealer's lift and every source reverted, and names a
target that did not report. Pages are selfserve:<id>:<condition>, the
attempt recorded first. The bridge renders the source's held config as
config sync would (BoxHandle.RenderHeldConfig).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 2c9ec59386b5055c227ae9b8899e83b98d607720 parent 017e777
12 files changed, +1124 −21
  • internal/console/apiinstancemove_offline.go +2 −2
  • internal/console/selfserve_apply.go +5 −1
  • internal/console/selfserve_checks.go +4 −0
  • internal/console/selfserve_health.go +168 −0
  • internal/console/selfserve_job.go +696 −14
  • internal/console/selfserve_pages.go +166 −0
  • internal/console/selfserve_pass.go +5 −0
  • internal/console/testhooks.go +16 −0
  • internal/e2e/selfserve/job_boundary_test.go +2 −1
  • internal/e2e/selfserve/job_reconcile_test.go +9 −3
  • internal/e2e/selfserve/job_world_test.go +28 −0
  • internal/host/e2e_bridge.go +23 −0
diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go
index 5bc8e0b22..c076f84bf 100644
--- a/internal/console/apiinstancemove_offline.go
+++ b/internal/console/apiinstancemove_offline.go
@@ -127,7 +127,7 @@ func offlineStateRefusal(state, host string) *offlineRefusal {
// offlineWaitRefusal is M-C10 with <from> filled in: a source box has not
// confirmed the stop.
func (s *Server) offlineWaitRefusal(box, from, host string) *offlineRefusal {
- return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("Waiting for %s in %s to confirm it has stopped %s. %s is offline.", box, from, host, host)}
+ return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("Waiting for %s in %s to confirm it has stopped %s. %s is offline.", box, from, host, host), transient: true}
}
// offlineStaleRefusal is MV-O178: box's report is there but older than
@@ -1366,7 +1366,7 @@ func (s *Server) offlineFlipCheck(ev offlineEvidence, mv *provision.MoveState, h
}
func processRefusal(box, host string) *offlineRefusal {
- return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s still runs a process for %s. wait for it to stop, then flip", box, host)}
+ return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s still runs a process for %s. wait for it to stop, then flip", box, host), transient: true}
}
// offlineOpenCheck is the re-check of the stop: the target's landing report
diff --git a/internal/console/selfserve_apply.go b/internal/console/selfserve_apply.go
index 433306d1b..d3a105ee6 100644
--- a/internal/console/selfserve_apply.go
+++ b/internal/console/selfserve_apply.go
@@ -57,6 +57,10 @@ var (
// land and the record is unchanged; the phase function replays it
// with its mutation.
errReplayNeeded = errors.New("console: self-serve move: the request's write has not landed")
+ // errRecordUnreadable wraps a judged read that could not tell what the
+ // record says (unreadable or corrupt): nothing is decided from it, and
+ // an open whose write may have landed goes to open-uncertain (Task 8).
+ errRecordUnreadable = errors.New("console: self-serve move: the record could not be read")
)
// requestRefusal is the bound write's mutation refusing the very record it
@@ -96,7 +100,7 @@ func selfServeRecordOf(obs provision.Observation, row *SelfServeMove) (*provisio
if err == nil {
err = provision.ErrLifetimeUnreadable
}
- return nil, fmt.Errorf("console: self-serve move %s: the record: %w", row.ID, err)
+ return nil, fmt.Errorf("console: self-serve move %s: the record: %w: %w", row.ID, errRecordUnreadable, err)
case !obs.Present || obs.Verdict == provision.LifetimeStale || obs.Rec == nil || obs.Rec.Retired:
return nil, errRecordGone
case obs.Rec.Incarnation != row.Incarnation:
diff --git a/internal/console/selfserve_checks.go b/internal/console/selfserve_checks.go
index 41bf3fc02..c535ea969 100644
--- a/internal/console/selfserve_checks.go
+++ b/internal/console/selfserve_checks.go
@@ -35,6 +35,10 @@ const (
factTargetFacts = "the target's facts"
factStagedCopy = "the staged copy"
factReplica = "the replica's size"
+ factStop = "the stop"
+ factSeal = "the seal"
+ factLanding = "the landing"
+ factOpen = "the open"
selfServeReadyMiB = 16 << 20
)
diff --git a/internal/console/selfserve_health.go b/internal/console/selfserve_health.go
new file mode 100644
index 000000000..89124c8d2
--- /dev/null
+++ b/internal/console/selfserve_health.go
@@ -0,0 +1,168 @@
+package console
+
+// The health check after open (self-serve moves plan 2 Task 8; spec §7.4):
+// the app answers on the target. A probe at most every 30 seconds, from
+// the console to the target box directly (its address from its edges file,
+// the host as SNI and Host), GET <the manifest's ready path>?carlos-probe=
+// <random> with Cache-Control: no-cache, no redirects followed, 30 seconds
+// per attempt; it passes on a 2xx whose x-carlos-version is the build the
+// seal recorded. Two consecutive passes finish the move; the window
+// running out finishes it with a warning and a page. Plan 3 adds the
+// edge's probe mode (X-Carlos-Probe, X-Carlos-Served-By).
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/tls"
+ "encoding/hex"
+ "fmt"
+ "math"
+ "net"
+ "net/http"
+ "net/url"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+const (
+ selfServeProbeEvery = 30 * time.Second
+ selfServeProbeTimeout = 30 * time.Second
+ selfServeHealthPasses = 2
+)
+
+// ssNoAnswer is the warning a move that never answered finishes with.
+func ssNoAnswer(region string, minutes int) string {
+ return fmt.Sprintf("the account did not answer on %s within %d minutes", region, minutes)
+}
+
+// selfServeHealthWindow is the health phase's allowance:
+// max(3 min, 2 × the promotion's estimate) + 30 s.
+func selfServeHealthWindow(row *SelfServeMove) time.Duration {
+ return maxDuration(3*time.Minute, 2*estimateTermsOf(row.EstimateTerms).Promote) + 30*time.Second
+}
+
+// selfServeHealth probes the account on the target, and with two passes in
+// a row goes to finishing. Every probe is recorded on the row before the
+// next decision (RecordSelfServeHealth), so a restart keeps the count; a
+// failed one pages.
+func (s *Server) selfServeHealth(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineOpen {
+ return nil
+ }
+ if row.HealthConsecutiveOK >= selfServeHealthPasses {
+ return s.selfServeHealthDone(row, now)
+ }
+ if !row.HealthLastProbeAt.IsZero() && now.Sub(row.HealthLastProbeAt) < selfServeProbeEvery {
+ return nil
+ }
+ ok, detail := s.selfServeProbe(ctx, row, rec, mv)
+ TestHooks.point("health:probe")
+ if err := s.store.RecordSelfServeHealth(row.ID, now, ok, detail); err != nil {
+ return err
+ }
+ if !ok {
+ s.selfServePageRow(ctx, row, ssPageHealth, ssPageWhyHealth(row.ID, s.offlineName(row.To)), now)
+ return nil
+ }
+ if row.HealthFirstOKAt.IsZero() {
+ s.selfServeMark(row, "first_ok_at", now, 0)
+ }
+ if row.HealthConsecutiveOK+1 >= selfServeHealthPasses {
+ return s.selfServeHealthDone(row, now)
+ }
+ return nil
+}
+
+// selfServeHealthDone goes to finishing at gate 1.
+func (s *Server) selfServeHealthDone(row *SelfServeMove, now time.Time) error {
+ TestHooks.point("advance:health->finishing")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssHealth, ssFinishing, "", selfServePhaseDeadline(row, ssFinishing, now), time.Time{}, "", now)
+ return err
+}
+
+// selfServeHealthExpired is the health phase's expiry, which takes
+// precedence over the page rule of the other post-open phases: the
+// warning recorded, a page, and finishing.
+func (s *Server) selfServeHealthExpired(ctx context.Context, row *SelfServeMove, now time.Time) error {
+ minutes := int(math.Ceil(selfServeHealthWindow(row).Minutes()))
+ if row.Warning == "" {
+ if err := s.store.SetSelfServeWarning(row.ID, ssNoAnswer(s.offlineName(row.To), minutes)); err != nil {
+ return err
+ }
+ }
+ s.selfServePageRow(ctx, row, ssPageHealth, ssPageWhyHealth(row.ID, s.offlineName(row.To)), now)
+ return s.selfServeHealthDone(row, now)
+}
+
+// selfServeProbe is one health probe: the witnesses' seam when set,
+// probeTarget otherwise.
+func (s *Server) selfServeProbe(ctx context.Context, row *SelfServeMove, rec *provision.Record, mv *provision.MoveState) (bool, string) {
+ if f := TestHooks.probeFunc(); f != nil {
+ return f(row.Host)
+ }
+ return s.probeTarget(ctx, row, mv)
+}
+
+// probeTarget asks the target box directly whether the account answers
+// with the build the seal recorded.
+func (s *Server) probeTarget(ctx context.Context, row *SelfServeMove, mv *provision.MoveState) (bool, string) {
+ snap, err := s.liveEdges(ctx)
+ if err != nil {
+ return false, "the edges files could not be read"
+ }
+ eip := ""
+ for _, f := range snap.Files {
+ if f.Box == row.Participants.Target.Label && (row.Participants.Target.BoxID == "" || f.BoxID == "" || f.BoxID == row.Participants.Target.BoxID) {
+ eip = f.EIP
+ }
+ }
+ if eip == "" {
+ return false, "the target box publishes no address"
+ }
+ m, err := provision.GetManifest(ctx, s.cfg.Releases, row.Account, row.App)
+ if err != nil {
+ return false, "the app's manifest could not be read"
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil || ev.seal == nil {
+ return false, "the seal could not be read"
+ }
+ want := ev.seal.Release
+ var nonce [8]byte
+ if _, err := rand.Read(nonce[:]); err != nil {
+ return false, "no probe value"
+ }
+ u := url.URL{Scheme: "https", Host: row.Host, Path: m.ReadyPathOrDefault(), RawQuery: "carlos-probe=" + hex.EncodeToString(nonce[:])}
+ pctx, cancel := context.WithTimeout(ctx, selfServeProbeTimeout)
+ defer cancel()
+ req, err := http.NewRequestWithContext(pctx, http.MethodGet, u.String(), nil)
+ if err != nil {
+ return false, "the probe could not be built"
+ }
+ req.Header.Set("Cache-Control", "no-cache")
+ dialer := &net.Dialer{Timeout: selfServeProbeTimeout}
+ client := &http.Client{
+ Timeout: selfServeProbeTimeout,
+ Transport: &http.Transport{
+ TLSClientConfig: &tls.Config{ServerName: row.Host},
+ DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+ return dialer.DialContext(ctx, network, net.JoinHostPort(eip, "443"))
+ },
+ },
+ CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+ }
+ resp, err := client.Do(req)
+ if err != nil {
+ return false, "the probe did not get an answer"
+ }
+ resp.Body.Close()
+ switch got := resp.Header.Get("x-carlos-version"); {
+ case resp.StatusCode < 200 || resp.StatusCode > 299:
+ return false, fmt.Sprintf("the probe was answered %d", resp.StatusCode)
+ case got != want:
+ return false, fmt.Sprintf("the probe was answered by build %q, not %q", got, want)
+ }
+ return true, ""
+}
diff --git a/internal/console/selfserve_job.go b/internal/console/selfserve_job.go
index bbed8cd81..b75674431 100644
--- a/internal/console/selfserve_job.go
+++ b/internal/console/selfserve_job.go
@@ -19,6 +19,7 @@ import (
"errors"
"fmt"
"log"
+ "slices"
"time"
"github.com/carlosframework/platform/internal/moves"
@@ -51,6 +52,8 @@ const (
ssStepAdopted = "adopted"
ssStepIssue = "issue"
ssStepSeal = "seal"
+ ssStepStart = "start"
+ ssStepGate1 = "gate1"
ssStepGate2 = "gate2"
)
@@ -220,9 +223,18 @@ func (s *Server) selfServeStep(ctx context.Context, t *leaseToken, row *SelfServ
// (2) The judged record.
rec, err := selfServeRecordOf(s.observeRecord(ctx, row.Account, row.App, row.Host), row)
gone := errors.Is(err, errRecordGone)
+ if row.Phase == ssLanded || row.Phase == ssOpenUncertain {
+ // Only the record decides an open that may have landed.
+ if done, oerr := s.selfServeOpenUncertain(ctx, t, row, rec, err, now); done {
+ return oerr
+ }
+ }
if err != nil && !gone {
return err
}
+ if row.OpenRecordPending && now.Sub(row.PhaseStartedAt) >= selfServeUncertainPage {
+ s.selfServePageRow(ctx, row, ssPageOpenRecord, ssPageWhyOpenRecord(row.ID), now)
+ }
// (3) The outstanding request first.
if out, conf := selfServeOutstanding(row); (out && !conf) || row.Request.Expired && !row.Request.Consumed {
@@ -561,17 +573,29 @@ func (s *Server) selfServeExpired(ctx context.Context, t *leaseToken, row *SelfS
// ---- 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).
+// selfServeExpiryAction is step (4) for an expiry bound to this phase,
+// decided by the record: while the record is at or before offline-landed,
+// abort, with the reason the phase's wait names; past the no-return
+// boundary (opened, finishing) and in the abort's own wait (reverting),
+// page and keep waiting, retried every tick; the health window's expiry
+// takes precedence: a warning, a page, and finishing. open-uncertain has
+// no step deadline: its open request's deadline is decided by the record
+// (selfServeOpenUncertain).
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
+ case ssHealth:
+ if mv := selfServeOurs(rec, row); mv != nil && mv.State == provision.OfflineOpen {
+ return true, s.selfServeHealthExpired(ctx, row, now)
+ }
+ return false, nil
+ case ssOpened, ssFinishing, ssReverting:
+ s.selfServePageBlocked(ctx, row, now)
+ return false, nil
+ case ssOpenUncertain:
+ return false, nil
}
if mv := selfServeOurs(rec, row); mv != nil && !selfServePreOpen(mv.State) {
- return false, nil // a committed open: never aborted (Task 8)
+ return false, nil // a committed open is never aborted: the row reconciles forward
}
if !row.AbortRequestedAt.IsZero() {
return false, nil
@@ -607,19 +631,35 @@ func (s *Server) selfServeExpiryReason(ctx context.Context, row *SelfServeMove,
}
}
return couldNotDetermine(factLane, stager)
- case ssPrechecks:
+ case ssPrechecks, ssBarrier:
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 != "" {
+ if v := s.selfServePrechecksVerdict(row, mv, ev, row.Phase == ssBarrier); v.waiting != "" {
return v.waiting
}
}
}
return couldNotDetermine(factTargetFacts, row.Participants.Target.Label)
+ case ssStopping:
+ return couldNotDetermine(factStop, selfServeSourceLabel(row))
+ case ssSealing:
+ return couldNotDetermine(factSeal, selfServeSourceLabel(row))
+ case ssLanding:
+ return couldNotDetermine(factLanding, row.Participants.Target.Label)
+ case ssLanded, ssOpenUncertain:
+ return couldNotDetermine(factOpen, row.Participants.Target.Label)
}
return msgSelfServeCouldNotStart
}
+// selfServeSourceLabel is the row's first pinned source (the stager).
+func selfServeSourceLabel(row *SelfServeMove) string {
+ if len(row.Participants.Source) > 0 {
+ return row.Participants.Source[0].Label
+ }
+ return ""
+}
+
// 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
@@ -693,11 +733,23 @@ func (s *Server) selfServePhase(ctx context.Context, t *leaseToken, row *SelfSer
return s.selfServeReady(ctx, t, row, rec, now)
case ssRestage:
return s.selfServeRestage(ctx, t, row, rec, now)
+ case ssBarrier:
+ return s.selfServeBarrier(ctx, t, row, rec, now)
+ case ssStopping:
+ return s.selfServeStopping(ctx, t, row, rec, now)
+ case ssSealing:
+ return s.selfServeSealing(ctx, t, row, rec, now)
+ case ssLanding:
+ return s.selfServeLanding(ctx, t, row, rec, now)
+ case ssLanded:
+ return s.selfServeLanded(ctx, t, row, rec, now)
case ssReverting:
return s.selfServeReverting(ctx, t, row, rec, now)
}
- // barrier through finishing: Task 8.
- return nil
+ // open-uncertain is decided before the phase (selfServeOpenUncertain);
+ // opened, health and finishing are the open's (selfserve_health.go and
+ // below).
+ return s.selfServeAfterOpen(ctx, t, row, rec, now)
}
// selfServeReplaced is the identity check at a gate: the label of a pinned
@@ -940,6 +992,9 @@ func (s *Server) selfServeReports(ctx context.Context, t *leaseToken, row *SelfS
v := s.selfServeReportsVerdict(row, mv, ev, fleet, now)
switch {
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)
case !v.pass:
return nil
@@ -1072,7 +1127,7 @@ func (s *Server) selfServeStaging(ctx context.Context, t *leaseToken, row *SelfS
// (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 {
+func (s *Server) selfServePrechecksVerdict(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence, barrier bool) 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)
@@ -1094,7 +1149,7 @@ func (s *Server) selfServePrechecksVerdict(row *SelfServeMove, mv *provision.Mov
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 {
+ if err := moves.StagedMatches(stager.OfflineReport, mv.StageBucket, mv.StageRegion, row.StageRequest, rd.Ready.MovesPrefix, barrier); err != nil {
log.Printf("console: self-serve move %s: %v", row.ID, err)
return ssVerdict{fail: ssStagedMismatch(host)}
}
@@ -1126,7 +1181,7 @@ func (s *Server) selfServePrechecks(ctx context.Context, t *leaseToken, row *Sel
if err != nil {
return err
}
- v := s.selfServePrechecksVerdict(row, mv, ev)
+ v := s.selfServePrechecksVerdict(row, mv, ev, false)
switch {
case v.fail != "":
return s.selfServeAbort(row, v.fail, now)
@@ -1324,6 +1379,46 @@ func (s *Server) selfServeRestage(ctx context.Context, t *leaseToken, row *SelfS
return s.selfServePreOfflineRefusal(row, err, now)
}
+// ssTargetLeft is the abort's detail for a target that did not report its
+// revert.
+func ssTargetLeft(label string) string {
+ return fmt.Sprintf("target %s did not report; its staged copy is left for cleanup", label)
+}
+
+// selfServeSourceReopened is the abort's second fact: no tombstone was ever
+// taken (no sealer named), or the sealer reports reverting or reverted with
+// the tombstone lifted; and every pinned source reports reverted.
+func selfServeSourceReopened(row *SelfServeMove, mv *provision.MoveState, ev offlineEvidence) bool {
+ if mv.Sealer != "" {
+ rep, ok := ev.get(moves.OfflineRoleSealer, mv.Sealer)
+ if !ok || !rep.TombstoneLifted || (rep.Phase != moves.OfflinePhaseReverting && rep.Phase != moves.OfflinePhaseReverted) {
+ return false
+ }
+ }
+ boxes := selfServeSourceLabels(row)
+ if len(boxes) == 0 {
+ boxes = mv.Boxes
+ }
+ for _, b := range boxes {
+ if rep, ok := ev.get(moves.OfflineRoleSource, b); !ok || rep.Phase != moves.OfflinePhaseReverted {
+ return false
+ }
+ }
+ return len(boxes) > 0
+}
+
+// selfServeTargetLeft is the abort's detail when the pinned target took
+// part (it has a report of this move) and has not reported its revert:
+// what it extracted is left on it for cleanup (plan 4's). "" otherwise.
+func selfServeTargetLeft(row *SelfServeMove, ev offlineEvidence) string {
+ label := row.Participants.Target.Label
+ rep, ok := ev.get(moves.OfflineRoleTarget, label)
+ if label == "" || !ok || rep.Phase == moves.OfflinePhaseReverted {
+ return ""
+ }
+ return ssTargetLeft(label)
+}
+
// 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
@@ -1375,10 +1470,32 @@ func (s *Server) selfServeReverting(ctx context.Context, t *leaseToken, row *Sel
if err != nil {
return err
}
+ // The second fact: the source reopened (the tombstone lifted,
+ // when one was taken, and every pinned source reverted). A
+ // sealer that reverted without lifting (another lease is live)
+ // is waited on, never taken for the lift.
+ if row.SourceReopenedAt.IsZero() {
+ if !selfServeSourceReopened(row, mv, ev) {
+ return nil
+ }
+ TestHooks.point("abort:reopened")
+ if err := s.store.SetSelfServeReopened(row.ID, now); err != nil {
+ return err
+ }
+ }
+ // The third: every abort report in (the operator's own finish
+ // check at offline-reverting). The target is never asked: one
+ // that stopped reporting after the landing does not hold the
+ // source, and what it holds is left for cleanup, named.
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 left := selfServeTargetLeft(row, ev); left != "" && row.Warning == "" {
+ if err := s.store.SetSelfServeWarning(row.ID, left); err != nil {
+ return err
+ }
+ }
if row.AbortFinishedAt.IsZero() {
if err := s.store.SetSelfServeAbortFinished(row.ID, now); err != nil {
return err
@@ -1396,3 +1513,568 @@ func (s *Server) selfServeReverting(ctx context.Context, t *leaseToken, row *Sel
// confirmed finish with the outcome.
return nil
}
+
+// ---- the cut, from the barrier to the open (Task 8) ------------------------------
+
+// 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) {}
+
+// selfServeBarrier is the cut's last check before anything goes offline: a
+// precheck asked for (OfflinePrecheck), then each source's precheck and the
+// target's ready report for it, the identities, the decisions and the book
+// comparison with nothing deferred (moves.StagedMatches at the barrier). A
+// failed fact aborts; all pass, the row goes to stopping, consuming the
+// precheck.
+func (s *Server) selfServeBarrier(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, "precheck") {
+ if out, _ := selfServeOutstanding(row); out && !(row.Request.Kind == "precheck" && row.Request.Phase == ssBarrier) {
+ return nil
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "precheck", Phase: ssBarrier}, 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)
+ }
+ 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, true)
+ switch {
+ case v.fail != "":
+ return s.selfServeAbort(row, v.fail, now)
+ case !v.pass:
+ 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)
+ return err
+}
+
+// selfServeSourceLabels are the pinned sources' labels, in pin order.
+func selfServeSourceLabels(row *SelfServeMove) []string {
+ var out []string
+ for _, p := range row.Participants.Source {
+ out = append(out, p.Label)
+ }
+ return out
+}
+
+// selfServeStopping takes the instance offline and seals it. Substep start:
+// the start validation in full on the fresh record (offlineRecordChecks,
+// then offlineRegionChecks with the start's rules and the participants'
+// identities, the selection still pinning the row's participants), then
+// OfflineStartStaged; substep seal, once every pinned source reports the
+// stop: the seal check (offlineSealCheck) over exactly the pinned sources,
+// then OfflineSeal naming the sealer, and the row goes to sealing. A
+// refusal that clears on its own (a box that has not stopped yet, a report
+// not current) is waited out until the phase's deadline; any other aborts.
+func (s *Server) selfServeStopping(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil {
+ return nil
+ }
+ out, conf := selfServeOutstanding(row)
+ switch mv.State {
+ case provision.OfflineStaging:
+ if row.PhaseStep != ssStepStart || (out && row.Request.Kind != "start") {
+ return nil
+ }
+ if !(out && !conf) {
+ 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), false, true, "")
+ if ref != nil {
+ return s.selfServeRefused(row, ref, now)
+ }
+ if label := selfServeSelectionMoved(row, sel); label != "" {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "start", Phase: ssStopping}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineStartStaged(row.ID, row.RequestedBy, at, rc)
+ })
+ if err != nil {
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ s.selfServeMark(row, "offline_at", now, 0)
+ TestHooks.point("start:consume")
+ _, err = s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepSeal, row.StepDeadline)
+ return err
+ case provision.OfflineStopping:
+ default:
+ return nil
+ }
+ if selfServeWaiting(row, "start") {
+ // The start confirmed and not consumed (a crash between the two).
+ TestHooks.point("start:consume")
+ _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepSeal, row.StepDeadline)
+ return err
+ }
+ if row.PhaseStep == ssStepStart {
+ // An operator's start: the seal is the job's next write.
+ _, err := s.store.SetSelfServeStep(row.ID, ssStopping, ssStepStart, ssStepSeal, row.StepDeadline)
+ return err
+ }
+ if out && row.Request.Kind != "seal" {
+ return nil
+ }
+ fleet, ok := s.readOfflineFleet(ctx)
+ if !ok {
+ return nil
+ }
+ if label := selfServeReplaced(row, fleet, now); label != "" && !(out && !conf) {
+ return s.selfServeAbort(row, ssReplaced(label), now)
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ sealer, boxes, ref := s.offlineSealCheck(fleet, ev, mv, row.Host, s.offlineName(mv.From), now)
+ if ref != nil {
+ if out && !conf {
+ // The seal already asked for waits for its own confirmation;
+ // the deadline bounds it.
+ return nil
+ }
+ return s.selfServeCutRefused(row, mv, ref, now)
+ }
+ if want := selfServeSourceLabels(row); !slices.Equal(sortedCopy(boxes), sortedCopy(want)) {
+ for _, b := range append(sortedCopy(want), boxes...) {
+ if !slices.Contains(boxes, b) || !slices.Contains(want, b) {
+ return s.selfServeAbort(row, ssReplaced(b), now)
+ }
+ }
+ }
+ s.selfServeMark(row, "stopped_at", now, 0)
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "seal", Phase: ssStopping}, nil,
+ func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineSeal(row.ID, sealer, boxes, rc)
+ }); err != nil {
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ s.selfServeMark(row, "seal_at", now, 0)
+ TestHooks.point("advance:stopping->sealing")
+ _, err = s.store.AdvanceSelfServe(row.ID, ssStopping, ssSealing, "", selfServePhaseDeadline(row, ssSealing, now), time.Time{}, row.Request.Nonce, now)
+ return err
+}
+
+// selfServeCutRefused is a cut check's refusal (the seal, flip and open
+// checks): one that says the evidence is not in yet (a box that has not
+// stopped, a report not current, the state's own MV-O25: the seal, the
+// landing or the tombstone not reported yet) is waited for, the phase's
+// deadline bounding it, whose expiry aborts naming the step; any other
+// (a sidecar, local data on two boxes, a stager gone) aborts with its
+// sentence.
+func (s *Server) selfServeCutRefused(row *SelfServeMove, mv *provision.MoveState, ref *offlineRefusal, now time.Time) error {
+ if ref.transient || ref.msg == offlineStateRefusal(mv.State, row.Host).msg {
+ log.Printf("console: self-serve move %s: waiting: %s", row.ID, ref.msg)
+ return nil
+ }
+ return s.selfServeAbort(row, ref.msg, now)
+}
+
+// sortedCopy is a sorted copy of ss.
+func sortedCopy(ss []string) []string {
+ out := append([]string(nil), ss...)
+ slices.Sort(out)
+ return out
+}
+
+// selfServeSealing waits for the sealer's seal: the record sealed
+// (OfflineSealedBy, the sealer's own write) moves the row to landing; a
+// sealer that reports it failed aborts with its detail.
+func (s *Server) selfServeSealing(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil {
+ return nil
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ switch mv.State {
+ case provision.OfflineSealed:
+ var copied int64
+ if ev.seal != nil {
+ copied = ev.seal.FinalCopiedBytes
+ }
+ s.selfServeMark(row, "sealed_at", now, copied)
+ TestHooks.point("advance:sealing->landing")
+ _, err := s.store.AdvanceSelfServe(row.ID, ssSealing, ssLanding, "", selfServePhaseDeadline(row, ssLanding, now), time.Time{}, "", now)
+ return err
+ case provision.OfflineSealing:
+ if rep, ok := ev.get(moves.OfflineRoleSealer, mv.Sealer); ok && rep.Phase == moves.OfflinePhaseFailed {
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ }
+ }
+ return nil
+}
+
+// selfServeLanding flips the region to the target (OfflineFlip, after the
+// flip check: the seal the record names, every pinned source still stopped
+// with no process) and waits for the pinned target's landing: the record
+// landed (OfflineLandedBy, the target's own write) is reconciliation's
+// advance to landed. A target that reports it failed aborts with its
+// detail.
+func (s *Server) selfServeLanding(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil {
+ return nil
+ }
+ out, conf := selfServeOutstanding(row)
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ switch mv.State {
+ case provision.OfflineSealed:
+ if out && row.Request.Kind != "flip" {
+ return nil
+ }
+ if ref := s.offlineFlipCheck(ev, mv, row.Host, s.offlineName(mv.From), now); ref != nil {
+ if out && !conf {
+ return nil
+ }
+ return s.selfServeCutRefused(row, mv, ref, now)
+ }
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "flip", Phase: ssLanding}, nil,
+ func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineFlip(row.ID, rc)
+ }); err != nil {
+ return s.selfServePreOfflineRefusal(row, err, now)
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ s.selfServeMark(row, "flip_at", now, 0)
+ TestHooks.point("flip:consume")
+ _, err = s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, "", row.StepDeadline)
+ return err
+ case provision.OfflineLanding:
+ if selfServeWaiting(row, "flip") {
+ TestHooks.point("flip:consume")
+ _, err := s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, "", row.StepDeadline)
+ return err
+ }
+ if rep, ok := ev.get(moves.OfflineRoleTarget, row.Participants.Target.Label); ok && rep.Phase == moves.OfflinePhaseFailed {
+ return s.selfServeAbort(row, moves.FailedDetail(rep.OfflineReport), now)
+ }
+ }
+ return nil
+}
+
+// selfServeLanded opens the target: the open check (offlineOpenCheck: the
+// landing report with every item OK, the seal, the sealer's tombstone held,
+// every source still stopped), then the open record written create-only
+// (open_record_pending until it exists; this move's own adopted), then
+// OfflineOpenTransition at the open record's stamp. A refusal, or the
+// deadline while the record is offline-landed, aborts; an open whose
+// confirmation could not read the record goes to open-uncertain, where only
+// the record decides.
+func (s *Server) selfServeLanded(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineLanded {
+ return nil
+ }
+ out, conf := selfServeOutstanding(row)
+ if out && row.Request.Kind != "open" {
+ return nil
+ }
+ var proposed *time.Time
+ if !(out && !conf) {
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ if ref := s.offlineOpenCheck(ev, mv, row.Host, s.offlineName(mv.From), now); ref != nil {
+ return s.selfServeCutRefused(row, mv, ref, now)
+ }
+ if !row.OpenRecordPending {
+ if err := s.store.SetSelfServeOpenRecordPending(row.ID, true); err != nil {
+ return err
+ }
+ }
+ TestHooks.point("open:record")
+ open, err := s.offlineOpenRecord(ctx, rec, mv, ev, provision.ProposeStamp(rec, now))
+ if errors.Is(err, errOpenRecordInTheWay) {
+ return s.selfServeAbort(row, msgOpenRecordInTheWay, now)
+ }
+ if err != nil {
+ return fmt.Errorf("console: self-serve move %s: the open record: %w (retried next tick)", row.ID, err)
+ }
+ TestHooks.point("open:recorded")
+ if err := s.store.SetSelfServeOpenRecordPending(row.ID, false); err != nil {
+ return err
+ }
+ proposed = &open.OpenedAt
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "open", Phase: ssLanded}, proposed,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineOpenTransition(row.ID, at, rc)
+ })
+ var refusal *requestRefusal
+ switch {
+ case err == nil:
+ case errors.As(err, &refusal):
+ return s.selfServePreOfflineRefusal(row, err, now)
+ case errors.Is(err, errRecordUnreadable):
+ return s.selfServeOpenUnconfirmable(row, now)
+ default:
+ return err
+ }
+ if row, err = s.selfServeReload(row.ID); err != nil || row == nil {
+ return err
+ }
+ s.selfServeMark(row, "open_at", now, 0)
+ TestHooks.point("advance:landed->opened")
+ _, err = s.store.AdvanceSelfServe(row.ID, ssLanded, ssOpened, "", selfServePhaseDeadline(row, ssOpened, now), time.Time{}, row.Request.Nonce, now)
+ return err
+}
+
+// selfServeOpenUnconfirmable is an open whose write may have landed and
+// whose record cannot be read: the row goes to open-uncertain with the
+// request outstanding, where it waits for the record to decide.
+func (s *Server) selfServeOpenUnconfirmable(row *SelfServeMove, now time.Time) error {
+ row, err := s.selfServeReload(row.ID)
+ if err != nil || row == nil {
+ return err
+ }
+ if out, conf := selfServeOutstanding(row); !out || conf || row.Request.Kind != "open" || row.Phase != ssLanded {
+ return nil
+ }
+ TestHooks.point("advance:landed->open-uncertain")
+ _, err = s.store.AdvanceSelfServeToStep(row.ID, ssLanded, ssOpenUncertain, "", "", time.Time{}, time.Time{}, "", now)
+ return err
+}
+
+// selfServeOpenUncertain decides an open whose write may have landed, from
+// the record alone, before any other step (step (2) for landed and
+// open-uncertain rows). It reports whether the step ended.
+//
+// - The record unreadable: a landed row with its open outstanding goes to
+// open-uncertain; an open-uncertain row stays, and pages once it has
+// been unreadable for 5 minutes.
+// - The record readable and still offline-landed without the open's
+// receipt, the open outstanding: before the open request's deadline the
+// row goes back to landed, whose phase replays the same bound write;
+// after it, the no-replay rule: the expiry decided durably, then
+// fenced and classified (the receipt confirms; absent, voided and
+// aborted). Never a replay past the deadline.
+// - An open-uncertain row with nothing outstanding on a landed record goes
+// back to landed to decide again (an abort requested is the abort
+// step's).
+//
+// Anything else (the receipt present, the record opened, reverting or gone)
+// is the ordinary step's: confirmation and reconciliation.
+func (s *Server) selfServeOpenUncertain(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, recErr error, now time.Time) (bool, error) {
+ out, conf := selfServeOutstanding(row)
+ open := out && !conf && row.Request.Kind == "open"
+ if recErr != nil && errors.Is(recErr, errRecordUnreadable) {
+ switch {
+ case row.Phase == ssLanded && open:
+ TestHooks.point("advance:landed->open-uncertain")
+ _, err := s.store.AdvanceSelfServeToStep(row.ID, ssLanded, ssOpenUncertain, "", "", time.Time{}, time.Time{}, "", now)
+ return true, err
+ case row.Phase == ssOpenUncertain:
+ if now.Sub(row.PhaseStartedAt) >= selfServeUncertainPage {
+ s.selfServePageRow(ctx, row, ssPageOpenUncertain, ssPageWhyUncertain(row.ID), now)
+ }
+ return true, recErr
+ }
+ return false, nil
+ }
+ if row.Phase != ssOpenUncertain || recErr != nil {
+ return false, nil
+ }
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineLanded {
+ return false, nil
+ }
+ switch {
+ case open && selfServeHasReceipt(rec, row.Request.Nonce):
+ return false, nil
+ case open && !row.Request.Expired && !row.Request.Deadline.IsZero() && now.Before(row.Request.Deadline):
+ TestHooks.point("advance:open-uncertain->landed")
+ _, err := s.store.AdvanceSelfServeToStep(row.ID, ssOpenUncertain, ssLanded, "", "", row.Request.Deadline, time.Time{}, "", now)
+ return true, err
+ case open:
+ if !row.Request.Expired {
+ TestHooks.point("open:expire")
+ if _, err := s.store.ExpireSelfServeRequest(row.ID, row.Request.Nonce, row.Request.Deadline, row.Request.CutRequestedAt); err != nil {
+ return true, err
+ }
+ }
+ return true, s.selfServeExpired(ctx, t, row)
+ case !out && row.AbortRequestedAt.IsZero():
+ TestHooks.point("advance:open-uncertain->landed")
+ _, err := s.store.AdvanceSelfServeToStep(row.ID, ssOpenUncertain, ssLanded, "", "", selfServePhaseDeadline(row, ssLanded, now), time.Time{}, "", now)
+ return true, err
+ }
+ return false, nil
+}
+
+// selfServeAfterOpen is the phases past the no-return boundary: opened,
+// health and finishing. Nothing here aborts: a wait past its deadline
+// pages and retries (the expiry action), and the health window finishes
+// with a warning.
+func (s *Server) selfServeAfterOpen(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ switch row.Phase {
+ case ssOpened:
+ return s.selfServeOpened(ctx, row, rec, now)
+ case ssHealth:
+ return s.selfServeHealth(ctx, t, row, rec, now)
+ case ssFinishing:
+ return s.selfServeFinishing(ctx, t, row, rec, now)
+ }
+ return nil
+}
+
+// selfServeOpened waits for the pinned target's opened report (the first
+// finish gate needs it; a ready row alone is not enough), then probes.
+func (s *Server) selfServeOpened(ctx context.Context, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil || mv.State != provision.OfflineOpen {
+ return nil
+ }
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ if rep, ok := ev.get(moves.OfflineRoleTarget, row.Participants.Target.Label); !ok || rep.Phase != moves.OfflinePhaseOpened {
+ return nil
+ }
+ s.selfServeMark(row, "opened_at", now, 0)
+ TestHooks.point("advance:opened->health")
+ _, err = s.store.AdvanceSelfServe(row.ID, ssOpened, ssHealth, "", selfServePhaseDeadline(row, ssHealth, now), time.Time{}, "", now)
+ return err
+}
+
+// selfServeFinishing is the existing two finish gates in turn, never
+// bypassed. Gate 1, from offline-open: the target's opened report and
+// every pinned source parked (offlineFinishCheck), then
+// OfflineFinishingTransition, the gate recorded as lifted. Gate 2, from
+// offline-finishing: the sealer's lifted report and every pinned source
+// gone, then OfflineFinish, which clears the move; reconciliation then
+// finalises the row done, keeping any warning. Each gate waits 10 minutes,
+// then pages and keeps waiting.
+func (s *Server) selfServeFinishing(ctx context.Context, t *leaseToken, row *SelfServeMove, rec *provision.Record, now time.Time) error {
+ mv := selfServeOurs(rec, row)
+ if mv == nil {
+ return nil
+ }
+ out, conf := selfServeOutstanding(row)
+ boxes := mv.Boxes
+ if len(boxes) == 0 {
+ boxes = selfServeSourceLabels(row)
+ }
+ switch mv.State {
+ case provision.OfflineOpen:
+ if out && row.Request.Kind != "finish1" {
+ return nil
+ }
+ if !(out && !conf) {
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ if ref := offlineFinishCheck(ev, mv, row.Host, moves.OfflinePhaseOpened, "", moves.OfflinePhaseParked, boxes); ref != nil {
+ return nil
+ }
+ }
+ if _, err := s.selfServeApply(ctx, t, row, request{Kind: "finish1", Phase: ssFinishing}, nil,
+ func(_ time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineFinishingTransition(row.ID, rc)
+ }); err != nil {
+ return s.selfServePostOpenRefusal(row, err)
+ }
+ return s.selfServeGate2(row, now)
+ case provision.OfflineFinishing:
+ default:
+ return nil
+ }
+ if selfServeWaiting(row, "finish1") || row.PhaseStep != ssStepGate2 && !out {
+ // Gate 1 written (by the job, consumed here after a crash, or by an
+ // operator's door): gate 2 next.
+ return s.selfServeGate2(row, now)
+ }
+ if out && row.Request.Kind != "finish" {
+ return nil
+ }
+ if !(out && !conf) {
+ ev, err := s.readOfflineEvidence(ctx, row.Host, row.ID, mv)
+ if err != nil {
+ return err
+ }
+ if ref := offlineFinishCheck(ev, mv, row.Host, "", moves.OfflinePhaseLifted, moves.OfflinePhaseGone, boxes); ref != nil {
+ return nil
+ }
+ }
+ _, err := s.selfServeApply(ctx, t, row, request{Kind: "finish", Phase: ssFinishing}, nil,
+ func(at time.Time, rc provision.Receipt) func(*provision.Record) error {
+ return provision.OfflineFinish(row.ID, at, rc)
+ })
+ // Cleared: the next step finalises from last_move, consuming the
+ // confirmed finish with the outcome.
+ return s.selfServePostOpenRefusal(row, err)
+}
+
+// selfServeGate2 records gate 1 as lifted and moves the row to gate 2 with
+// its own 10 minutes, consuming gate 1's confirmed request when it is the
+// row's.
+func (s *Server) selfServeGate2(row *SelfServeMove, now time.Time) error {
+ row, err := s.selfServeReload(row.ID)
+ if err != nil || row == nil {
+ return err
+ }
+ if row.FinishGate != "lifted" {
+ if err := s.store.SetSelfServeFinishGate(row.ID, "lifted"); err != nil {
+ return err
+ }
+ }
+ TestHooks.point("finish1:consume")
+ if selfServeWaiting(row, "finish1") {
+ _, err = s.store.ConsumeSelfServeRequest(row.ID, row.Request.Nonce, ssStepGate2, now.Add(selfServeRevertWait))
+ return err
+ }
+ _, err = s.store.SetSelfServeStep(row.ID, ssFinishing, row.PhaseStep, ssStepGate2, now.Add(selfServeRevertWait))
+ return err
+}
+
+// selfServePostOpenRefusal is a post-open write that the record refused:
+// nothing is aborted after the open; the refusal is logged and the row
+// decides again next tick (reconciliation follows the record).
+func (s *Server) selfServePostOpenRefusal(row *SelfServeMove, err error) error {
+ var refusal *requestRefusal
+ if errors.As(err, &refusal) {
+ log.Printf("console: self-serve move %s: %v (decided again next tick)", row.ID, err)
+ return nil
+ }
+ return err
+}
diff --git a/internal/console/selfserve_pages.go b/internal/console/selfserve_pages.go
new file mode 100644
index 000000000..5129f97ae
--- /dev/null
+++ b/internal/console/selfserve_pages.go
@@ -0,0 +1,166 @@
+package console
+
+// The self-serve job's pages (self-serve moves plan 2 Task 8; spec §8):
+// through the existing alert sink (alerting.Page) and the move pass's own
+// durable throttle (MovePageAttempt and SetMovePageAttempt, to the second),
+// one rule per move and condition, selfserve:<id>:<condition>, at most once
+// every 10 minutes. The attempt is recorded before the page is sent, as the
+// move pass's pages are: a crash between the two loses one page, never
+// sends two.
+
+import (
+ "context"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// selfServePageEvery is how often one condition of one move pages again.
+const selfServePageEvery = 10 * time.Minute
+
+// The conditions a self-serve move pages for.
+const (
+ ssPageOpened = "opened" // the target's opened report past its deadline
+ ssPageHealth = "health" // a health probe failed, or the health window ran out
+ ssPageFinishing = "finishing" // a finishing gate past its 10 minutes
+ ssPageReverting = "reverting" // the abort's reports past their 10 minutes
+ ssPageOpenUncertain = "open-uncertain" // the record unreadable for 5 minutes after an open
+ ssPageOpenRecord = "open-record" // the open record still not written after 5 minutes
+ ssPageReservation = "reservation" // a reservation no move owns, past the grace period
+ ssPageLane = "lane" // a lane probe failed on a configured pair
+)
+
+// selfServeUncertainPage is how long open-uncertain and a pending open
+// record wait before they page.
+const selfServeUncertainPage = 5 * time.Minute
+
+// selfServePageRule is a page's throttle key.
+func selfServePageRule(id, condition string) string {
+ return "selfserve:" + id + ":" + condition
+}
+
+// selfServePage pages for (id, condition) unless the same page was
+// attempted less than selfServePageEvery ago, recording the attempt first.
+func (s *Server) selfServePage(ctx context.Context, id, condition, subject, body string, now time.Time) {
+ if s.store == nil {
+ return
+ }
+ rule := selfServePageRule(id, condition)
+ last, paged, err := s.store.MovePageAttempt(rule)
+ if err != nil {
+ log.Printf("console: self-serve move %s: the page record for %s: %v", id, rule, err)
+ return
+ }
+ if paged && now.Sub(last) < selfServePageEvery {
+ return
+ }
+ if err := s.store.SetMovePageAttempt(rule, now); err != nil {
+ log.Printf("console: self-serve move %s: record the page %s: %v", id, rule, err)
+ return
+ }
+ TestHooks.point("page:" + condition)
+ if err := alerting.Page(ctx, alerting.LevelWarning, subject, body); err != nil {
+ log.Printf("console: self-serve move %s: page %s: %v", id, rule, err)
+ }
+}
+
+// selfServePageSubject is every self-serve page's subject.
+func selfServePageSubject(host string) string {
+ return fmt.Sprintf("carlos self-serve move needs an operator: %s", host)
+}
+
+// selfServePageBody is a page's body: what the move waits for, then where
+// to look.
+func selfServePageBody(why, host string) string {
+ return fmt.Sprintf("%s. run carlos move offline status --host %s to see what it waits for", why, host)
+}
+
+// The pages' reasons.
+func ssPageWhyOpened(id, region string) string {
+ return fmt.Sprintf("the move %s is open in %s and its target has not reported opened", id, region)
+}
+func ssPageWhyHealth(id, region string) string {
+ return fmt.Sprintf("the move %s is open in %s and the account did not answer its health check", id, region)
+}
+func ssPageWhyFinishing(id string) string {
+ return fmt.Sprintf("the move %s has been finishing for more than 10 minutes", id)
+}
+func ssPageWhyReverting(id string) string {
+ return fmt.Sprintf("the move %s has been reverting for more than 10 minutes", id)
+}
+func ssPageWhyUncertain(id string) string {
+ return fmt.Sprintf("the move %s cannot read its record to confirm its open", id)
+}
+func ssPageWhyOpenRecord(id string) string {
+ return fmt.Sprintf("the move %s has not written its open record", id)
+}
+func ssPageWhyReservation(id string) string {
+ return fmt.Sprintf("a reservation names %s, which no move owns", id)
+}
+func ssPageWhyLane(from, to string) string {
+ return fmt.Sprintf("the staging lane from %s to %s failed its probe", from, to)
+}
+
+// selfServePageRow pages condition for row with why.
+func (s *Server) selfServePageRow(ctx context.Context, row *SelfServeMove, condition, why string, now time.Time) {
+ s.selfServePage(ctx, row.ID, condition, selfServePageSubject(row.Host), selfServePageBody(why, row.Host), now)
+}
+
+// selfServePageBlocked is the page of a wait past its deadline that pages
+// and retains (the expiry action after the no-return boundary): opened,
+// finishing and reverting, while the current deadline has passed.
+func (s *Server) selfServePageBlocked(ctx context.Context, row *SelfServeMove, now time.Time) {
+ if row.StepDeadline.IsZero() || now.Before(row.StepDeadline) {
+ return
+ }
+ switch row.Phase {
+ case ssOpened:
+ s.selfServePageRow(ctx, row, ssPageOpened, ssPageWhyOpened(row.ID, s.offlineName(row.To)), now)
+ case ssFinishing:
+ s.selfServePageRow(ctx, row, ssPageFinishing, ssPageWhyFinishing(row.ID), now)
+ case ssReverting:
+ s.selfServePageRow(ctx, row, ssPageReverting, ssPageWhyReverting(row.ID), now)
+ }
+}
+
+// selfServeOrphanScan is the per-tick reservation scan: a move reservation
+// older than the grace period that no self-serve row and no record names
+// (one the reclaim rule cannot reclaim, since nothing says it is over) is
+// paged. Read failures page nothing and are retried next tick.
+func (s *Server) selfServeOrphanScan(ctx context.Context, now time.Time) {
+ raw, _, err := s.cfg.Releases.Get(ctx, moves.OfflineActiveKey)
+ if err != nil {
+ return
+ }
+ table, err := moves.ParseOfflineActive(raw)
+ if err != nil {
+ return
+ }
+ seen := map[string]bool{}
+ for _, e := range table {
+ if e.Kind != moves.OfflineKindMove || seen[e.ID] || now.Sub(e.ReservedAt) <= offlineLockGrace {
+ continue
+ }
+ seen[e.ID] = true
+ row, err := s.store.SelfServeMove(e.ID)
+ if err != nil || row != nil {
+ continue
+ }
+ if e.Account == "" || e.App == "" || e.Host == "" {
+ s.selfServePage(ctx, e.ID, ssPageReservation, selfServePageSubject(e.Host), selfServePageBody(ssPageWhyReservation(e.ID), e.Host), now)
+ continue
+ }
+ obs := s.observeRecord(ctx, e.Account, e.App, e.Host)
+ if obs.Verdict == provision.LifetimeUnreadable || obs.Verdict == provision.LifetimeCorrupt {
+ continue
+ }
+ if obs.Present && obs.Rec != nil && obs.Rec.Move != nil && obs.Rec.Move.ID == e.ID {
+ continue // an operator move in flight owns it
+ }
+ s.selfServePage(ctx, e.ID, ssPageReservation, selfServePageSubject(e.Host), selfServePageBody(ssPageWhyReservation(e.ID), e.Host), now)
+ }
+}
diff --git a/internal/console/selfserve_pass.go b/internal/console/selfserve_pass.go
index 26e587c22..19b0901d1 100644
--- a/internal/console/selfserve_pass.go
+++ b/internal/console/selfserve_pass.go
@@ -56,6 +56,10 @@ func (s *Server) selfServeScanAtStart(ctx context.Context) {
log.Printf("console: self-serve moves: reading at start: %v (recovery mode on; the pass retries each tick)", err)
}
s.selfServeRecovery.Store(found)
+ if found && err == nil && s.cfg.Releases != nil {
+ // A reservation no move owns pages from the start, as every tick.
+ s.selfServeOrphanScan(ctx, time.Now())
+ }
}
// selfServeReevaluate is the tick's re-evaluation of recovery mode: a read
@@ -121,6 +125,7 @@ func (s *Server) selfServeTick(ctx context.Context) {
stop := s.keepSelfServeLease(ctx, t)
defer stop()
s.drainSelfServeEvents(ctx, t)
+ s.selfServeOrphanScan(ctx, s.selfServeNow())
rows, err := s.store.OpenSelfServeMoves()
if err != nil {
log.Printf("console: self-serve moves: the open rows: %v (retrying next tick)", err)
diff --git a/internal/console/testhooks.go b/internal/console/testhooks.go
index b68bcbc19..97c9b6400 100644
--- a/internal/console/testhooks.go
+++ b/internal/console/testhooks.go
@@ -17,6 +17,22 @@ type testHooks struct {
mu sync.Mutex
fault func(point string)
now time.Time // the ticking pass's clock (Tick, Advance)
+ probe func(host string) (ok bool, detail string)
+}
+
+// Probe answers the job's health probes from now on (nil clears it, and
+// the real probe of the target box runs): the witnesses' boxes serve no
+// HTTP.
+func (h *testHooks) Probe(f func(host string) (ok bool, detail string)) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ h.probe = f
+}
+
+func (h *testHooks) probeFunc() func(host string) (bool, string) {
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ return h.probe
}
// testCrash is what Crash panics with: the console process dying at a
diff --git a/internal/e2e/selfserve/job_boundary_test.go b/internal/e2e/selfserve/job_boundary_test.go
index 117769a62..bf4cc5428 100644
--- a/internal/e2e/selfserve/job_boundary_test.go
+++ b/internal/e2e/selfserve/job_boundary_test.go
@@ -14,7 +14,8 @@ import (
// jobPhaseOrder is the order the job's phases move in; reverting and done
// may follow any of them, and nothing moves backwards.
var jobPhaseOrder = map[string]int{"requested": 0, "reserving": 1, "queued": 2, "reports": 3, "staging": 4, "prechecks": 5,
- "ready": 6, "restage": 7, "barrier": 8, "reverting": 100}
+ "ready": 6, "restage": 7, "barrier": 8, "stopping": 9, "sealing": 10, "landing": 11, "landed": 12, "open-uncertain": 12,
+ "opened": 14, "health": 15, "finishing": 16, "reverting": 100}
// jobBoundary is one record write the job makes before the cut's barrier:
// the request's kind and the phase that writes it, how the witness brings
diff --git a/internal/e2e/selfserve/job_reconcile_test.go b/internal/e2e/selfserve/job_reconcile_test.go
index 16398c105..4006c2f13 100644
--- a/internal/e2e/selfserve/job_reconcile_test.go
+++ b/internal/e2e/selfserve/job_reconcile_test.go
@@ -74,7 +74,10 @@ func TestReconcileForwardNeverRegresses(t *testing.T) {
for i := 0; i < 3; i++ {
w.tick()
}
- if r := w.row(row.ID); r.Phase != c.phase {
+ // The phases that share the record's state may follow (Task 8:
+ // sealing goes on to landing once the record is sealed, its
+ // flip still to come); the boxes are down, so nothing else.
+ if r := w.row(row.ID); r.Phase != c.phase && !(c.state == provision.OfflineSealed && r.Phase == "landing") {
t.Fatalf("after more ticks the row is %s", describe(r))
}
})
@@ -109,8 +112,11 @@ func TestReconcileForwardNeverRegresses(t *testing.T) {
request_expect_inc = ?, request_expect_seq = ?, request_proposed = ?, request_confirmed = 0, request_consumed = 0, request_voided = 0, request_deadline = 0
WHERE id = ?`, time.Now().UnixNano(), rec.Incarnation, rec.RecordSeq, time.Now().UnixNano(), row.ID)
stays(t, w, row.ID, "stopping")
- if r := w.row(row.ID); r.Request.Nonce != "rq_pendingstart" || r.Request.Voided || r.Request.Consumed {
- t.Fatalf("the pending start was resolved: %s", describe(r))
+ // Task 8: the stopping phase replays the pending start as it was
+ // issued (its own nonce, bound to the record it read) and confirms
+ // it; it never issues another.
+ if r := w.row(row.ID); r.Request.Nonce != "rq_pendingstart" || r.Request.Voided || !r.Request.Confirmed {
+ t.Fatalf("the pending start was not replayed as issued: %s", describe(r))
}
})
t.Run("landing before its flip stays on offline-sealed", func(t *testing.T) {
diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go
index c71bbe9ed..f896f484a 100644
--- a/internal/e2e/selfserve/job_world_test.go
+++ b/internal/e2e/selfserve/job_world_test.go
@@ -181,6 +181,12 @@ type jobWorld struct {
// any other write still runs).
failKinds map[string]bool
failing bool
+ // rendered are the moves whose held config the source has rendered.
+ rendered map[string]bool
+ // probe answers the job's health probes when set (a pass otherwise);
+ // probes counts them.
+ probe func() (bool, string)
+ probes int
}
const jobConsole = "console-e2e-job"
@@ -195,7 +201,17 @@ func newJobWorld(t *testing.T) *jobWorld {
failKinds: map[string]bool{}}
w.fs.failing = func() bool { return w.failing }
w.open()
+ // The boxes serve no HTTP: the health probe is the witness's answer,
+ // a pass unless a witness says otherwise.
+ console.TestHooks.Probe(func(string) (bool, string) {
+ w.probes++
+ if w.probe != nil {
+ return w.probe()
+ }
+ return true, ""
+ })
t.Cleanup(func() {
+ console.TestHooks.Probe(nil)
console.TestHooks.Fault(nil)
if w.cst != nil {
w.cst.Close()
@@ -298,6 +314,18 @@ func (w *jobWorld) boxes() {
if err := w.e.Status(w.ctx); err != nil {
w.t.Logf("status: %v", err)
}
+ // Config sync's render on the source once a move holds its row, as
+ // the operator end-to-end test stands in for it (renderSourceConfig).
+ if rec, _, err := provision.GetRecordETag(w.ctx, w.raw, w.acct, host.E2EApp, host.E2EHost); err == nil && rec.Move != nil && !w.rendered[rec.Move.ID] {
+ if ok, err := w.e.Box(host.E2ESourceBox).RenderHeldConfig(w.ctx, rec.Move.ID); err != nil {
+ w.t.Logf("render: %v", err)
+ } else if ok {
+ if w.rendered == nil {
+ w.rendered = map[string]bool{}
+ }
+ w.rendered[rec.Move.ID] = true
+ }
+ }
}
// tick runs one console pass at the console's clock and answers the crash
diff --git a/internal/host/e2e_bridge.go b/internal/host/e2e_bridge.go
index a3273e267..ba21f4c12 100644
--- a/internal/host/e2e_bridge.go
+++ b/internal/host/e2e_bridge.go
@@ -1396,3 +1396,26 @@ func (h BoxHandle) SetDiskFree(f func(dir string) (total, avail uint64, err erro
defer h.x.knobs.Unlock()
h.x.disk = f
}
+
+// RenderHeldConfig stands in for config sync's render on a source box whose
+// row for E2EHost the move moveID holds (the operator end-to-end test's
+// renderSourceConfig): the held config.env rendered from the bundle (the
+// store binding and its secret), at the flat path the seal reads. It
+// answers false, rendering nothing, when no row is held by moveID.
+func (h BoxHandle) RenderHeldConfig(ctx context.Context, moveID string) (bool, error) {
+ b := h.x.b
+ rt, err := b.reg.Get(offHost)
+ if err != nil || rt == nil || rt.HoldOp != moveID || b.w.cfg.RenderHeld == nil {
+ return false, err
+ }
+ hr, err := b.w.cfg.RenderHeld(ctx, b.wd.s, *rt, moveID, nil, func(string) error { return nil })
+ if err != nil {
+ return false, err
+ }
+ if want := filepath.Join(b.dd, offHost+".config.env"); hr.Path != want {
+ if err := os.Rename(hr.Path, want); err != nil {
+ return false, err
+ }
+ }
+ return true, nil
+}