amadan

carlos / platform Public

host: the landing's probing and probed phases (Plan 3 Task 6, §7.6, §7.7, §7.13, H8, H10)

restored steps to probing when this box has a probe wake (moveCfg.Probe,
Task 5's ProbeWake; nil and waiting in production). probing makes the
attempts the cut admitted (CutAttempts, AttemptsFor), each counted on the
record and journalled before its call, with the landing's deadline: a
pass steps to probed (the claim's step keeps the writer session); Task
5's ProbeInputsChanged and exhausted attempts are rollback marks; a crash
restarts only the probe, retrying a counted attempt unless it is spent.
Every phase now checks the landing deadline at entry. Task 5's Admitted
hook reads CutAttempts (MovesRecordAdmittedAttempts).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e03233cec1f78277771f2406ee247036e60a1731 parent 131f1d8
7 files changed, +497 −10
  • approved_moves_copy_test.go +1 −0
  • internal/host/movesland.go +29 −7
  • internal/host/movesland_probe_test.go +286 −0
  • internal/host/movesland_test.go +5 −0
  • internal/host/moveslandprobe.go +168 −0
  • internal/host/movespass.go +5 −0
  • internal/host/movesprobeowners_test.go +3 −3
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index f4591b472..43d039478 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -582,6 +582,7 @@ var movesScanFiles = []string{
"internal/host/movesland.go",
"internal/host/moveslandinstall.go",
"internal/host/moveslandpark.go",
+ "internal/host/moveslandprobe.go",
"internal/host/movesmeasure.go",
"internal/host/movesmeasure_linux.go",
"internal/host/movesmeasure_other.go",
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
index 874a252da..3ed35df5c 100644
--- a/internal/host/movesland.go
+++ b/internal/host/movesland.go
@@ -202,7 +202,10 @@ func (l *landingRun) step(ctx context.Context) (bool, error) {
// it back; markRollback nudged the serial for it.
return false, nil
}
- if landingBefore(m.LandingPhase, provision.LandRestored) && l.pastDeadline() {
+ // Every phase checks the landing's deadline at entry: the install's
+ // phases (P2-deadline), and restored, probing and probed (H8: the
+ // probe works to the same deadline).
+ if l.pastDeadline() {
return l.markRollback(ctx, provision.RollbackLandingDeadline)
}
switch m.LandingPhase {
@@ -213,14 +216,27 @@ func (l *landingRun) step(ctx context.Context) (bool, error) {
advanced, _, err := installGeneration(ctx, l.s, l.w, l.rec, l.installSource())
return advanced, err
case provision.LandRestored:
- // The install is complete and the probe's (Plan 3). The one thing
- // still this pass's is the claim half of the last CAS: a crash after
- // the record's restored, or a mirror that failed, leaves the claim
- // behind, and nothing later would walk it (fix round 1, item 1).
+ // The install is complete. The claim half of the last CAS first: a
+ // crash after the record's restored, or a mirror that failed, leaves
+ // the claim behind (fix round 1, item 1). Then the probe begins,
+ // when this box has a probe wake (unarmed in production).
+ if _, err := l.syncClaimPhase(ctx); err != nil {
+ return false, err
+ }
+ if l.cfg().Probe == nil {
+ l.w.notice(l.rec.Host+"#probe", "moves: %s: landing %s waits at restored: this box has no probe wake (claim Plan 11's launcher and Plan 5 Task 9's arming are needed)", l.rec.Host, l.g.ID)
+ return false, nil
+ }
+ return l.advance(ctx, provision.LandRestored, provision.LandProbing, nil)
+ case provision.LandProbing:
+ return l.probing(ctx)
+ case provision.LandProbed:
+ // The probe passed. The commit is Task 7's; this phase ends at the
+ // claim's probed step.
_, err := l.syncClaimPhase(ctx)
return false, err
}
- return false, nil // probing and later: the probe's (Plan 3)
+ return false, nil
}
// landingDeadlineOf is the deadline a landing before restored works to:
@@ -304,7 +320,13 @@ func (l *landingRun) mirrorPhase(ctx context.Context, from, to string) error {
if err != nil {
return err
}
- _, err = hostclaim.Update(ctx, l.s, obs, hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, from, to))
+ step := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, from, to)
+ if hostclaim.SessionPhase(hostclaim.KindLanding, from) && hostclaim.SessionPhase(hostclaim.KindLanding, to) {
+ // probing to probed (and Task 7's committed): the admitted probe
+ // session is the serving session, so the CAS must keep it (H3).
+ step = hostclaim.KeepWriterSession(l.g.ID, step)
+ }
+ _, err = hostclaim.Update(ctx, l.s, obs, step)
return err
}
diff --git a/internal/host/movesland_probe_test.go b/internal/host/movesland_probe_test.go
new file mode 100644
index 000000000..fa2a2e6b2
--- /dev/null
+++ b/internal/host/movesland_probe_test.go
@@ -0,0 +1,286 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// probeLog is a fake probe wake that answers from a script and records
+// every configuration it was given.
+type probeLog struct {
+ configs []activator.ProbeConfig
+ answers []error
+ before func(n int)
+}
+
+func (p *probeLog) wake(_ context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) {
+ p.configs = append(p.configs, cfg)
+ n := len(p.configs)
+ if p.before != nil {
+ p.before(n)
+ }
+ if n <= len(p.answers) && p.answers[n-1] != nil {
+ return activator.ProbeResult{}, p.answers[n-1]
+ }
+ return activator.ProbeResult{Session: 20, Generation: p1ID}, nil
+}
+
+func failedAttempt(why string) error {
+ return errors.Join(activator.ErrProbeFailed, errors.New(why))
+}
+
+// installedLanding is a landing at restored, with the cut's admitted probe
+// attempts set to attempts (0 clears them) through the record's writer.
+func installedLanding(t *testing.T, attempts int) (*landingHarness, *probeLog) {
+ t.Helper()
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.land(t)
+ h.installed(t)
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
+ rec.Move.CutAttempts = attempts
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ p := &probeLog{}
+ h.probe = p.wake
+ return h, p
+}
+
+func (h *landingHarness) atPhase(t *testing.T, phase string) *provision.MoveState {
+ t.Helper()
+ m := h.read(t, p1Host).Move
+ if m.State != provision.MoveLanding || m.LandingPhase != phase || m.Reason != "" {
+ t.Fatalf("the record is not landing/%s: %+v", phase, m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != phase {
+ t.Fatalf("the claim is not landing/%s: %+v", phase, c.Op)
+ }
+ return m
+}
+
+// TestLandingProbesWithinTheBudget (brief): two admitted attempts, the
+// first fails and the second passes, so the landing is probed, record and
+// claim; a third is never attempted. Each attempt is counted on the record
+// and journalled before its call, and the probe is given the landing's
+// deadline (H8), the build's manifest and the pins.
+func TestLandingProbesWithinTheBudget(t *testing.T) {
+ h, p := installedLanding(t, 2)
+ p.answers = []error{failedAttempt("not ready")}
+ session := h.claim(t).Holder.Session
+ h.land(t)
+ m := h.atPhase(t, provision.LandProbed)
+ if len(p.configs) != 2 || m.Landing == nil || m.Landing.Attempts != 2 || h.landJournal(t).Attempt != 2 {
+ t.Fatalf("attempts: calls %d record %+v journal %d", len(p.configs), m.Landing, h.landJournal(t).Attempt)
+ }
+ deadline, _ := landingDeadlineOf(m)
+ want := activator.ProbeConfig{Host: p1Host, Op: p1ID, HoldOp: p1ID, Deadline: *deadline, Prepared: h.preparedManifest(t), Inputs: m.Inputs}
+ for i, got := range p.configs {
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("probe %d was given %+v, want %+v", i+1, got, want)
+ }
+ }
+ if c := h.claim(t); c.Holder.Session != session {
+ t.Fatalf("the probed step changed the writer session: %d, was %d", c.Holder.Session, session)
+ }
+ // probed is the probe's end: another pass probes nothing.
+ h.land(t)
+ if len(p.configs) != 2 {
+ t.Fatalf("a pass at probed probed again: %d", len(p.configs))
+ }
+ h.atPhase(t, provision.LandProbed)
+}
+
+// preparedManifest is the build's install manifest from the op journal.
+func (h *landingHarness) preparedManifest(t *testing.T) *moves.InstallManifest {
+ t.Helper()
+ m := h.installedManifest(t)
+ return &m
+}
+
+func (h *landingHarness) installedManifest(t *testing.T) moves.InstallManifest {
+ t.Helper()
+ j := h.landJournal(t)
+ var m moves.InstallManifest
+ if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &m); err != nil {
+ t.Fatal(err)
+ }
+ return m
+}
+
+// TestLandingExhaustedAttemptsMarkRollback (brief): every admitted attempt
+// fails: the landing records the attempts-exhausted rollback sentence,
+// which the next pass's rollback decision turns into the rollback.
+func TestLandingExhaustedAttemptsMarkRollback(t *testing.T) {
+ h, p := installedLanding(t, 2)
+ p.answers = []error{failedAttempt("one"), errors.Join(activator.ErrProbeFailed, activator.ErrAttemptOver)}
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackProbeAttempts || len(p.configs) != 2 {
+ t.Fatalf("exhausted: %d calls, %+v", len(p.configs), m)
+ }
+ h.land(t)
+ rolledBack(t, h, provision.RollbackProbeAttempts, provision.LandProbing)
+ if len(p.configs) != 2 {
+ t.Fatalf("a rolled-back landing probed again: %d", len(p.configs))
+ }
+}
+
+// A changed configuration found by the probe (Task 5's ProbeInputsChanged,
+// before or after the admission) is the configuration rollback, at once.
+func TestLandingConfigChangedDuringTheProbeRollsBack(t *testing.T) {
+ for name, answer := range map[string]error{
+ "before the admission": &ProbeInputsChanged{Reason: provision.RollbackConfigChanged},
+ "after the admission": errors.Join(activator.ErrProbeFailed, &ProbeInputsChanged{Reason: provision.RollbackConfigChanged}),
+ } {
+ t.Run(name, func(t *testing.T) {
+ h, p := installedLanding(t, 3)
+ p.answers = []error{answer}
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackConfigChanged || len(p.configs) != 1 {
+ t.Fatalf("%d calls, %+v", len(p.configs), m)
+ }
+ h.land(t)
+ rolledBack(t, h, provision.RollbackConfigChanged, provision.LandProbing)
+ })
+ }
+}
+
+// H10: a cut that recorded no probe attempts refuses the probe with its
+// recorded sentence; nothing is probed.
+func TestLandingProbeNeedsTheCutsAttempts(t *testing.T) {
+ h, p := installedLanding(t, 0)
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackProbeUnpriced || len(p.configs) != 0 {
+ t.Fatalf("%d calls, %+v", len(p.configs), m)
+ }
+ h.land(t)
+ rolledBack(t, h, provision.RollbackProbeUnpriced, provision.LandProbing)
+}
+
+// H8: probing and probed check the landing's deadline at entry. A deadline
+// that passes during an attempt ends the probing with the deadline's
+// sentence, and no further attempt is made; a probed landing past its
+// deadline rolls back.
+func TestLandingProbingChecksTheDeadline(t *testing.T) {
+ h, p := installedLanding(t, 3)
+ p.before = func(int) { h.clock.Advance(cutLandingWindow + time.Minute) }
+ p.answers = []error{errors.Join(activator.ErrProbeFailed, activator.ErrAttemptOver)}
+ h.land(t)
+ if len(p.configs) != 1 {
+ t.Fatalf("probed past the deadline: %d calls", len(p.configs))
+ }
+ h.land(t)
+ rolledBack(t, h, provision.RollbackLandingDeadline, provision.LandProbing)
+
+ h2, _ := installedLanding(t, 3)
+ h2.land(t)
+ h2.atPhase(t, provision.LandProbed)
+ h2.clock.Advance(cutLandingWindow + time.Minute)
+ h2.land(t)
+ rolledBack(t, h2, provision.RollbackLandingDeadline, provision.LandProbed)
+}
+
+// TestLandingProbeCrashRestartsOnlyTheProbe (brief, §7.13): a crash after
+// an attempt is counted and before its probe returns restarts only the
+// probe: a fresh worker probes the same attempt again (its window never
+// opened), and local and restored are not rerun, their journal steps and
+// .files-gen untouched. An attempt whose window had opened and passed
+// counts as failed, and the next one is made.
+func TestLandingProbeCrashRestartsOnlyTheProbe(t *testing.T) {
+ h, p := installedLanding(t, 2)
+ steps := map[string]string{}
+ for k, v := range h.landJournal(t).Steps {
+ steps[k] = v
+ }
+ marker := filepath.Join(h.tdir, p1Host, archive.MarkerName)
+ before, err := os.Stat(marker)
+ if err != nil {
+ t.Fatal(err)
+ }
+ renders, timers := h.renders, len(h.replaced)
+ h.landAt = "before-probe-1"
+ h.crashLand(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || m.Landing.Attempts != 1 || h.landJournal(t).Attempt != 1 || len(p.configs) != 0 {
+ t.Fatalf("the crash point: %+v journal %d calls %d", m, h.landJournal(t).Attempt, len(p.configs))
+ }
+ h.freshTarget()
+ h.land(t)
+ m := h.atPhase(t, provision.LandProbed)
+ if len(p.configs) != 1 || m.Landing.Attempts != 1 {
+ t.Fatalf("after the restart: %d calls, record %+v", len(p.configs), m.Landing)
+ }
+ after, err := os.Stat(marker)
+ if err != nil || !after.ModTime().Equal(before.ModTime()) {
+ t.Fatalf(".files-gen was written again: %v", err)
+ }
+ j := h.landJournal(t)
+ for k, v := range steps {
+ if j.Steps[k] != v {
+ t.Fatalf("the journal step %s changed", k)
+ }
+ }
+ for k := range j.Steps {
+ if _, ok := steps[k]; !ok && !strings.HasPrefix(k, "probe_") {
+ t.Fatalf("a non-probe step %s was added", k)
+ }
+ }
+ if h.renders != renders || len(h.replaced) != timers {
+ t.Fatalf("configured or local ran again: renders %d->%d timers %d->%d", renders, h.renders, timers, len(h.replaced))
+ }
+
+ // An attempt whose window opened and passed before the crash counts as
+ // failed: the restarted worker makes the next one.
+ h2, p2 := installedLanding(t, 2)
+ h2.landAt = "before-probe-1"
+ h2.crashLand(t)
+ j2 := h2.landJournal(t)
+ j2.Mark("probe_opened_1", h2.clock.Now().Add(-31*time.Second).UTC().Format(time.RFC3339Nano))
+ if err := j2.Write(h2.tdir); err != nil {
+ t.Fatal(err)
+ }
+ h2.freshTarget()
+ h2.land(t)
+ if m := h2.atPhase(t, provision.LandProbed); len(p2.configs) != 1 || m.Landing.Attempts != 2 {
+ t.Fatalf("the spent attempt was retried: %d calls, record %+v", len(p2.configs), m.Landing)
+ }
+}
+
+// Unarmed (production until claim Plan 11's launcher and Plan 5 Task 9):
+// with no probe wake the landing waits at restored and writes nothing.
+func TestLandingUnarmedProbeWaitsAtRestored(t *testing.T) {
+ h, _ := installedLanding(t, 3)
+ h.probe = nil
+ before := h.read(t, p1Host).Move.UpdatedAt
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandRestored || !m.UpdatedAt.Equal(before) {
+ t.Fatalf("an unarmed probe moved the landing: %+v", m)
+ }
+}
+
+// H10: Task 5's Admitted hook reads the admission's recorded count, and
+// refuses a record that has none.
+func TestMovesRecordAdmittedAttempts(t *testing.T) {
+ ctx := context.Background()
+ if n, err := MovesRecordAdmittedAttempts(ctx, &provision.Record{Move: &provision.MoveState{CutAttempts: 2}}); err != nil || n != 2 {
+ t.Fatalf("recorded: %d %v", n, err)
+ }
+ for _, rec := range []*provision.Record{nil, {}, {Move: &provision.MoveState{}}} {
+ if _, err := MovesRecordAdmittedAttempts(ctx, rec); err == nil {
+ t.Fatalf("no recorded count was admitted: %+v", rec)
+ }
+ }
+}
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
index 48e6ba338..63b0faa72 100644
--- a/internal/host/movesland_test.go
+++ b/internal/host/movesland_test.go
@@ -73,6 +73,8 @@ type landingHarness struct {
parkView func(context.Context, hostclaim.Observed) (park.View, error)
landFault func(point string) error
store lease.VersionedStore // when set, the target's versioned store
+ // probe, when set, is the target's probe wake (Plan 3 Task 6).
+ probe func(context.Context, activator.ProbeConfig) (activator.ProbeResult, error)
}
// fakeParkLeases refuses a second acquisition of a held location.
@@ -346,6 +348,9 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
c.SetNextDue = func(host string, due int64) error { h.nextDue = append(h.nextDue, due); return nil }
c.RemoveVerifyEntry = dbverify.RemoveHost
c.WriteGen = hostbox.WriteGenerationMarker
+ if h.probe != nil {
+ c.Probe = h.probe
+ }
c.landFault = func(point string) error {
h.landHits = append(h.landHits, point)
if h.landFault != nil {
diff --git a/internal/host/moveslandprobe.go b/internal/host/moveslandprobe.go
new file mode 100644
index 000000000..fc673b4b7
--- /dev/null
+++ b/internal/host/moveslandprobe.go
@@ -0,0 +1,168 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// The landing's probing phase (moves spec §7.6, §7.7, §7.13; Plan 3 Task 6).
+//
+// restored to probing is the landing's step (movesland.go). At probing the
+// pass makes the probe attempts T11 admitted (H10: the record's
+// CutAttempts, written by MoveAdmit), at most moves.AttemptsFor("landing",
+// n). Each attempt is counted on the record (provision.MoveProbeAttempt)
+// and journalled (Journal.Attempt) before its call, then made through the
+// probe wake (Task 5's activator.Service.ProbeWake behind moveCfg.Probe):
+// one cold admission into the prepared generation (or a fresh one for a
+// later attempt), restored, rendered, published and started by the claim
+// (Group D), up to five starts inside the attempt's 30 s window, bounded by
+// the landing's deadline (H8). The attempt's own journal steps
+// (probe_opened_<k>, probe_start_<k>) are Task 5's movesProbeAttempts';
+// probe_failed_<k> is this file's, written when the attempt is over.
+//
+// Outcomes. A pass: probing to probed, record then claim (the claim's step
+// keeps the admitted writer session, H3); the commit is Task 7's. Task 5's
+// ProbeInputsChanged: its recorded sentence, the configuration rollback. An
+// attempt that is over (ErrProbeFailed, ErrAttemptOver): the next one, if
+// any; none left: the attempts-exhausted rollback; the deadline passed: the
+// deadline's. Rollbacks are marks (MoveMarkRollback); the one conversion to
+// MoveRollBack is the pass's rollback decision (movesrollback.go). Any other
+// error ends the pass and the same attempt is made again on the next.
+//
+// A crash restarts only the probe (§7.13): local and restored are never run
+// again; a fresh worker finds probing and the counted attempt, and makes it
+// again unless it is spent (its failure journalled, or its window opened and
+// passed), in which case the next one is counted.
+//
+// Unarmed in production: moveCfg.Probe is nil until claim Plan 11's launcher
+// (R10) and Plan 5 Task 9's arming, so the landing waits at restored.
+
+// probeStepFailed marks attempt k over in the op journal.
+const probeStepFailed = "probe_failed_"
+
+// probeWindowLen is the attempt's window, counted from its first start.
+var probeWindowLen = time.Duration(moves.ProbeSeconds * float64(time.Second))
+
+// MovesRecordAdmittedAttempts is Task 5's Admitted hook (H10): the probe
+// attempts the record's admission priced, refused when it recorded none.
+func MovesRecordAdmittedAttempts(_ context.Context, rec *provision.Record) (int, error) {
+ if rec == nil || rec.Move == nil || rec.Move.CutAttempts < 1 {
+ return 0, errors.New("moves: " + provision.RollbackProbeUnpriced)
+ }
+ return rec.Move.CutAttempts, nil
+}
+
+// attemptSpent reports whether counted attempt k is over: its failure is
+// journalled, or its window opened and has passed.
+func (l *landingRun) attemptSpent(j *moves.Journal, k int) bool {
+ if j.Done(probeStepFailed + strconv.Itoa(k)) {
+ return true
+ }
+ raw, ok := j.Steps[probeStepOpened+strconv.Itoa(k)]
+ if !ok {
+ return false
+ }
+ opened, err := time.Parse(time.RFC3339Nano, raw)
+ return err == nil && !l.cfg().Now().Before(opened.Add(probeWindowLen))
+}
+
+// probing makes the landing's probe attempts (see the file comment).
+func (l *landingRun) probing(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandProbing); err != nil {
+ return false, err
+ }
+ if cfg.Probe == nil {
+ l.w.notice(l.rec.Host+"#probe", "moves: %s: landing %s waits at probing: this box has no probe wake", l.rec.Host, l.g.ID)
+ return false, nil
+ }
+ m := l.rec.Move
+ if m.CutAttempts < 1 {
+ return l.markRollback(ctx, provision.RollbackProbeUnpriced)
+ }
+ n := moves.AttemptsFor(moves.ProbeLanding, m.CutAttempts)
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ raw, ok := j.Steps[landStepInstall]
+ if !ok {
+ return false, fmt.Errorf("move %s: probing without an install manifest in the journal", l.g.ID)
+ }
+ var manifest moves.InstallManifest
+ if err := json.Unmarshal([]byte(raw), &manifest); err != nil {
+ return false, fmt.Errorf("move %s: the journalled install manifest: %w", l.g.ID, err)
+ }
+ counted := 0
+ if m.Landing != nil {
+ counted = m.Landing.Attempts
+ }
+ k := counted
+ if k == 0 || l.attemptSpent(j, k) {
+ k++
+ }
+ for ; k <= n; k++ {
+ if l.pastDeadline() {
+ return l.markRollback(ctx, provision.RollbackLandingDeadline)
+ }
+ if k > counted {
+ ok, err := l.transition(ctx, "probe attempt "+strconv.Itoa(k), provision.MoveProbeAttempt(l.g, k))
+ if !ok {
+ return false, err
+ }
+ counted = k
+ }
+ if j, err = l.journal(); err != nil {
+ return false, err
+ }
+ j.Phase, j.Attempt = provision.LandProbing, k
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ if err := l.fault("before-probe-" + strconv.Itoa(k)); err != nil {
+ return false, err
+ }
+ deadline, _ := landingDeadlineOf(m)
+ if deadline == nil {
+ return false, fmt.Errorf("move %s: the landing has no deadline", l.g.ID)
+ }
+ _, err := cfg.Probe(ctx, activator.ProbeConfig{Host: l.rec.Host, Op: l.g.ID, HoldOp: l.g.ID, Deadline: *deadline, Prepared: &manifest, Inputs: m.Inputs})
+ var changed *ProbeInputsChanged
+ switch {
+ case err == nil:
+ log.Printf("moves: %s: landing %s passed its probe on attempt %d of %d", l.rec.Host, l.g.ID, k, n)
+ return l.advance(ctx, provision.LandProbing, provision.LandProbed, nil)
+ case errors.As(err, &changed):
+ log.Printf("moves: %s: landing %s probe attempt %d: %v", l.rec.Host, l.g.ID, k, err)
+ return l.markRollback(ctx, changed.Reason)
+ case errors.Is(err, activator.ErrProbeFailed) || errors.Is(err, activator.ErrAttemptOver):
+ log.Printf("moves: %s: landing %s probe attempt %d of %d is over: %v", l.rec.Host, l.g.ID, k, n, err)
+ if j, err = l.journal(); err != nil {
+ return false, err
+ }
+ j.Mark(probeStepFailed+strconv.Itoa(k), cfg.Now().UTC().Format(time.RFC3339Nano))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ default:
+ return false, fmt.Errorf("probe attempt %d: %w", k, err)
+ }
+ }
+ if l.pastDeadline() {
+ return l.markRollback(ctx, provision.RollbackLandingDeadline)
+ }
+ return l.markRollback(ctx, provision.RollbackProbeAttempts)
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 5285cc85c..f4e19d5f6 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -558,6 +558,11 @@ type moveCfg struct {
SetNextDue func(host string, due int64) error
RemoveVerifyEntry func(ctx context.Context, s release.Store, host string) error
WriteGen func(path, generation string) error
+ // Probe is the landing's probe wake (Plan 3 Task 6): Task 5's
+ // activator.Service.ProbeWake. Nil in production, with a standing
+ // notice, until claim Plan 11's launcher (R10) and Plan 5 Task 9's
+ // arming exist; the landing then waits at restored.
+ Probe func(ctx context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error)
// landFault and parkFault are test seams: the landing's crash points
// and the park engine's (park.Engine.After). Nil in production.
landFault func(point string) error
diff --git a/internal/host/movesprobeowners_test.go b/internal/host/movesprobeowners_test.go
index 0e187443c..129c084f6 100644
--- a/internal/host/movesprobeowners_test.go
+++ b/internal/host/movesprobeowners_test.go
@@ -650,7 +650,7 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
pins.EnvReport = envDigest(poEnvKey, []string{"A=1"})
deadline := f.now.Add(time.Hour)
rec := &provision.Record{Host: recoveryHost, Account: "bab", App: "jam", Region: "ie",
- Move: &provision.MoveState{ID: poOp, Incarnation: h.Incarnation, From: "uk", To: "ie", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, CutDeadline: &deadline,
+ Move: &provision.MoveState{ID: poOp, Incarnation: h.Incarnation, From: "uk", To: "ie", State: provision.MoveLanding, LandingPhase: provision.LandProbing, Inputs: pins, CutDeadline: &deadline, CutAttempts: 3,
Landing: &provision.MoveLandingJournal{Attempts: 1, InstalledDBSHA256: "sha256:" + manifest.DBSHA256}}}
records := func(context.Context, string, string, string) (*provision.Record, error) { return rec, nil }
var order []string
@@ -663,7 +663,7 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
return ClaimServingLaunch{Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), Env: []string{"A=1"}, Version: pins.Release}, nil
},
EnvKey: poEnvKey,
- Admitted: func(context.Context, *provision.Record) (int, error) { return 3, nil },
+ Admitted: MovesRecordAdmittedAttempts,
Works: func(_ context.Context, _, action string, _ hostguard.Captured) (hostwrite.Work, error) {
order = append(order, "select:"+action)
return plan9D5Work{}, nil
@@ -681,7 +681,7 @@ func TestMovesProbeOwnersThroughTheClaimRuntime(t *testing.T) {
},
NewID: f.newID, Now: func() time.Time { return f.now }}
attempts := &movesProbeAttempts{Claims: f.claims, Records: records, DataDir: journalDir, NewID: f.newID, Now: func() time.Time { return f.now },
- Admitted: func(context.Context, *provision.Record) (int, error) { return 3, nil }}
+ Admitted: MovesRecordAdmittedAttempts}
c := f.controller("boot-1", "controller-1", 4101, "start-1")
bound := owners.MovesOwners(attempts)
c.wake.Inputs = ClaimOwnerDispatch{Moves: &bound.ClaimKindOwners}