host, activator: the probe's pass and admissions survive a crash (Plan 3 Task 6 fix round 1)
1. A pass is journalled (probe_passed_<k>: session, stamp, generation) before the probing to probed record write. A pass that finds it does not probe again: it checks the claim's holder and the activator's live probe entry (new Service.LiveProbe, moveCfg.ProbeLive) and retries only the phase writes; a writer that is gone consumes the attempt. 2. One cold admission per budgeted attempt (§7.7): an attempt with Task 5's session binding is consumed on restart, failed unless it passed; only a counted attempt with no binding is made again. Minor: the fake probe wake admits on the claim, and the test asserts the admitted session, stamp and generation survive the probed mirroring.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
6 files changed,
+308
−30
internal/activator/probewake.go+20 −2internal/activator/probewake_test.go+24 −0internal/host/movesland_probe_test.go+140 −12internal/host/movesland_test.go+5 −0internal/host/moveslandprobe.go+116 −16internal/host/movespass.go+3 −0
diff --git a/internal/activator/probewake.go b/internal/activator/probewake.go| index 6181ce680..a301deba2 100644 |
| --- a/internal/activator/probewake.go |
| +++ b/internal/activator/probewake.go |
| @@ -71,9 +71,28 @@ type ProbeConfig struct { |
| // now holds, which serves after the open on the same session (H3). |
| type ProbeResult struct { |
| Session uint64 |
| + Stamp uint64 |
| Generation string |
| } |
| +// LiveProbe reports the awake entry ProbeWake left for host: the probe's op |
| +// and its admitted session. ok is false for no entry, a waking or stopping |
| +// one, or an entry that is not a passed probe. |
| +func (a *Service) LiveProbe(host string) (op string, res ProbeResult, ok bool) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[host] |
| + if e == nil || e.state != actAwake || e.probe == nil || e.probe.admitted == nil || e.probe.admitted.Claim == nil || e.probe.admitted.Claim.Holder == nil { |
| + return "", ProbeResult{}, false |
| + } |
| + return e.probe.op, probeResultOf(e.probe.admitted), true |
| +} |
| + |
| +func probeResultOf(o *hostclaim.Observed) ProbeResult { |
| + h := o.Claim.Holder |
| + return ProbeResult{Session: h.Session, Stamp: h.Stamp, Generation: h.Generation} |
| +} |
| + |
| // Test seams: retireBetweenLocks runs in RetireExcept between the probe |
| // snapshot and the retirement pass; probeWakePublished runs in ProbeWake |
| // after the entry is published and before its wake starts. |
| @@ -135,8 +154,7 @@ func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult, |
| if e.err != nil { |
| return ProbeResult{}, e.err |
| } |
| - h := e.probe.admitted.Claim.Holder |
| - return ProbeResult{Session: h.Session, Generation: h.Generation}, nil |
| + return probeResultOf(e.probe.admitted), nil |
| } |
| var ( |
diff --git a/internal/activator/probewake_test.go b/internal/activator/probewake_test.go| index 17eff56aa..62822a543 100644 |
| --- a/internal/activator/probewake_test.go |
| +++ b/internal/activator/probewake_test.go |
| @@ -364,3 +364,27 @@ func TestProbeWakeCancelBeforeTheWakeStarts(t *testing.T) { |
| t.Fatal("the cancelled probe left its entry") |
| } |
| } |
| + |
| +// Task 6 fix round 1: LiveProbe answers the awake entry a passing probe left, |
| +// with its op and admitted session; nothing for a waking entry, an ordinary |
| +// wake, or a probe that never passed. |
| +func TestLiveProbeAnswersThePassingProbeEntry(t *testing.T) { |
| + a, _, _, _ := testActivator(t, time.Hour) |
| + admitted := hostclaim.Observed{Host: pwHost, Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Session: 20, Stamp: 18, Generation: pwOp}}} |
| + probe := &probeEntry{op: pwOp, holdOp: pwOp, admitted: &admitted} |
| + a.mu.Lock() |
| + a.entries[pwHost] = &actEntry{host: pwHost, state: actAwake, probe: probe} |
| + a.entries["plain.example.com"] = &actEntry{host: "plain.example.com", state: actAwake} |
| + a.entries["waking.example.com"] = &actEntry{host: "waking.example.com", state: actWaking, probe: &probeEntry{op: pwOp, admitted: &admitted}} |
| + a.entries["unpassed.example.com"] = &actEntry{host: "unpassed.example.com", state: actAwake, probe: &probeEntry{op: pwOp}} |
| + a.mu.Unlock() |
| + op, res, ok := a.LiveProbe(pwHost) |
| + if !ok || op != pwOp || res != (ProbeResult{Session: 20, Stamp: 18, Generation: pwOp}) { |
| + t.Fatalf("the passing probe: %q %+v %v", op, res, ok) |
| + } |
| + for _, host := range []string{"plain.example.com", "waking.example.com", "unpassed.example.com", "absent.example.com"} { |
| + if _, _, ok := a.LiveProbe(host); ok { |
| + t.Fatalf("%s answered", host) |
| + } |
| + } |
| +} |
diff --git a/internal/host/movesland_probe_test.go b/internal/host/movesland_probe_test.go| index fa2a2e6b2..45b44d76e 100644 |
| --- a/internal/host/movesland_probe_test.go |
| +++ b/internal/host/movesland_probe_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "os" |
| "path/filepath" |
| "reflect" |
| + "strconv" |
| "strings" |
| "testing" |
| "time" |
| @@ -18,24 +19,84 @@ import ( |
| "github.com/carlosframework/platform/internal/provision" |
| ) |
| -// probeLog is a fake probe wake that answers from a script and records |
| -// every configuration it was given. |
| +// probeLog is a fake probe wake. Each call admits a writer on the claim as |
| +// the claim's runtime would (a cold admission into the move id, or a fresh |
| +// generation once the move id is admitted), journals the session's binding |
| +// to the record's attempt as Task 5's attempts do (and, when openWindow is |
| +// set, the attempt's opened time), then answers from its script. A passing |
| +// call leaves a live entry for the admitted session. |
| type probeLog struct { |
| - configs []activator.ProbeConfig |
| - answers []error |
| - before func(n int) |
| + t *testing.T |
| + h *landingHarness |
| + configs []activator.ProbeConfig |
| + answers []error |
| + before func(n int) |
| + admissions []probeAdmission |
| + openWindow bool |
| + last activator.ProbeResult |
| + live bool |
| } |
| -func (p *probeLog) wake(_ context.Context, cfg activator.ProbeConfig) (activator.ProbeResult, error) { |
| +// probeAdmission is one admission the fake made: the record attempt it was |
| +// made under and the session it admitted. |
| +type probeAdmission struct { |
| + attempt int |
| + session uint64 |
| +} |
| + |
| +func (p *probeLog) wake(ctx 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) |
| } |
| + k := p.h.read(p.t, p1Host).Move.Landing.Attempts |
| + obs, err := hostclaim.Read(ctx, p.h.s, p1Host) |
| + if err != nil { |
| + return activator.ProbeResult{}, err |
| + } |
| + h := obs.Claim.Holder |
| + gen := p1ID |
| + if h.WriterBox == "uk-box" { |
| + gen = "fresh" + strconv.Itoa(n) |
| + } |
| + from := "ie" |
| + if len(h.Admissions) > 0 { |
| + from = h.Admissions[len(h.Admissions)-1].Region |
| + } |
| + admitted, err := hostclaim.Update(ctx, p.h.s, obs, hostclaim.AdmitWriter(hostclaim.Admission{ObservedOpID: p1ID, ObservedPhase: provision.LandProbing, |
| + ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "uk-box", Mode: "cold", NewGeneration: gen, KeepStamp: h.Stamp > h.Session, Region: "uk", PredecessorRegion: from})) |
| + if err != nil { |
| + return activator.ProbeResult{}, err |
| + } |
| + ah := admitted.Claim.Holder |
| + p.admissions = append(p.admissions, probeAdmission{attempt: k, session: ah.Session}) |
| + j, err := moves.ReadJournal(p.h.tdir, p1Host, p1ID) |
| + if err != nil { |
| + return activator.ProbeResult{}, err |
| + } |
| + j.Mark(probeStepSession+strconv.FormatUint(ah.Session, 10), strconv.Itoa(k)) |
| + if p.openWindow { |
| + j.Mark(probeStepOpened+strconv.Itoa(k), p.h.clock.Now().UTC().Format(time.RFC3339Nano)) |
| + } |
| + if err := j.Write(p.h.tdir); err != nil { |
| + return activator.ProbeResult{}, err |
| + } |
| + p.last = activator.ProbeResult{Session: ah.Session, Stamp: ah.Stamp, Generation: ah.Generation} |
| if n <= len(p.answers) && p.answers[n-1] != nil { |
| + p.live = false |
| return activator.ProbeResult{}, p.answers[n-1] |
| } |
| - return activator.ProbeResult{Session: 20, Generation: p1ID}, nil |
| + p.live = true |
| + return p.last, nil |
| +} |
| + |
| +// liveProbe is the activator's LiveProbe over the fake's last passing call. |
| +func (p *probeLog) liveProbe(host string) (string, activator.ProbeResult, bool) { |
| + if host != p1Host || !p.live { |
| + return "", activator.ProbeResult{}, false |
| + } |
| + return p1ID, p.last, true |
| } |
| func failedAttempt(why string) error { |
| @@ -56,8 +117,8 @@ func installedLanding(t *testing.T, attempts int) (*landingHarness, *probeLog) { |
| }); err != nil { |
| t.Fatal(err) |
| } |
| - p := &probeLog{} |
| - h.probe = p.wake |
| + p := &probeLog{t: t, h: h} |
| + h.probe, h.probeLive = p.wake, p.liveProbe |
| return h, p |
| } |
| @@ -81,7 +142,6 @@ func (h *landingHarness) atPhase(t *testing.T, phase string) *provision.MoveStat |
| 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 { |
| @@ -94,8 +154,10 @@ func TestLandingProbesWithinTheBudget(t *testing.T) { |
| 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) |
| + // The passing attempt's admitted writer (fix round 1, Minor): its |
| + // session, stamp and generation survive the probed mirroring. |
| + if c := h.claim(t); len(p.admissions) != 2 || c.Holder.Session != p.last.Session || c.Holder.Stamp != p.last.Stamp || c.Holder.Generation != p.last.Generation || c.Holder.WriterBox != "uk-box" { |
| + t.Fatalf("the probed step did not keep the admitted writer %+v: holder %+v", p.last, c.Holder) |
| } |
| // probed is the probe's end: another pass probes nothing. |
| h.land(t) |
| @@ -284,3 +346,69 @@ func TestMovesRecordAdmittedAttempts(t *testing.T) { |
| } |
| } |
| } |
| + |
| +// Fix round 1, ruling 1: a passing probe whose probing to probed record |
| +// write fails is not probed again. probe_passed_<k> names the admitted |
| +// writer; the next pass validates it against the claim and the activator's |
| +// live entry and retries only the phase writes, record then claim. |
| +func TestLandingProbePassReplaysTheRecordWrite(t *testing.T) { |
| + h, p := installedLanding(t, 2) |
| + h.landAt = "before-record-probed" |
| + h.crashLand(t) |
| + if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || len(p.configs) != 1 { |
| + t.Fatalf("the crash point: %d calls, %+v", len(p.configs), m) |
| + } |
| + if !h.landJournal(t).Done("probe_passed_1") { |
| + t.Fatalf("the pass was not journalled before the record write: %v", h.landJournal(t).Steps) |
| + } |
| + h.land(t) |
| + h.atPhase(t, provision.LandProbed) |
| + if len(p.configs) != 1 || len(p.admissions) != 1 { |
| + t.Fatalf("the replay admitted again: %d calls, %d admissions", len(p.configs), len(p.admissions)) |
| + } |
| + if c := h.claim(t); c.Holder.Session != p.last.Session || c.Holder.Generation != p1ID { |
| + t.Fatalf("the probed writer: %+v", c.Holder) |
| + } |
| + |
| + // The session is gone (the entry did not survive): the attempt is |
| + // consumed and failed, and the next one is made. |
| + h2, p2 := installedLanding(t, 2) |
| + h2.landAt = "before-record-probed" |
| + h2.crashLand(t) |
| + p2.live = false |
| + h2.land(t) |
| + if m := h2.atPhase(t, provision.LandProbed); len(p2.configs) != 2 || m.Landing.Attempts != 2 || p2.admissions[1].attempt != 2 { |
| + t.Fatalf("a lost passing session: %d calls, record %+v, admissions %+v", len(p2.configs), m.Landing, p2.admissions) |
| + } |
| +} |
| + |
| +// Fix round 1, ruling 2 (§7.7): one cold admission per budgeted attempt. A |
| +// crash after the attempt's admission, with its window still open and no |
| +// failure journalled, consumes the attempt: the restart never admits again |
| +// under the same index. |
| +func TestLandingProbeAdmittedAttemptIsConsumedOnRestart(t *testing.T) { |
| + h, p := installedLanding(t, 2) |
| + p.openWindow = true |
| + p.answers = []error{errors.New("the controller died during the attempt")} |
| + h.land(t) |
| + if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandProbing || len(p.admissions) != 1 || p.admissions[0].attempt != 1 || h.landJournal(t).Done("probe_failed_1") { |
| + t.Fatalf("the crash: %+v admissions %+v", m, p.admissions) |
| + } |
| + h.freshTarget() |
| + h.land(t) |
| + h.atPhase(t, provision.LandProbed) |
| + if len(p.admissions) != 2 || p.admissions[1].attempt != 2 { |
| + t.Fatalf("an index was admitted twice: %+v", p.admissions) |
| + } |
| + |
| + // With one attempt admitted, the restart has none left. |
| + h1, p1 := installedLanding(t, 1) |
| + p1.openWindow = true |
| + p1.answers = []error{errors.New("the controller died during the attempt")} |
| + h1.land(t) |
| + h1.freshTarget() |
| + h1.land(t) |
| + if m := h1.read(t, p1Host).Move; m.Reason != provision.RollbackProbeAttempts || len(p1.admissions) != 1 { |
| + t.Fatalf("the only attempt was admitted again: %+v admissions %+v", m, p1.admissions) |
| + } |
| +} |
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go| index 63b0faa72..efad7f361 100644 |
| --- a/internal/host/movesland_test.go |
| +++ b/internal/host/movesland_test.go |
| @@ -75,6 +75,8 @@ type landingHarness struct { |
| 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) |
| + // probeLive, when set, is the activator's live probe entry. |
| + probeLive func(string) (string, activator.ProbeResult, bool) |
| } |
| // fakeParkLeases refuses a second acquisition of a held location. |
| @@ -351,6 +353,9 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg { |
| if h.probe != nil { |
| c.Probe = h.probe |
| } |
| + if h.probeLive != nil { |
| + c.ProbeLive = h.probeLive |
| + } |
| 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| index fc673b4b7..be7c1ae91 100644 |
| --- a/internal/host/moveslandprobe.go |
| +++ b/internal/host/moveslandprobe.go |
| @@ -7,9 +7,11 @@ import ( |
| "fmt" |
| "log" |
| "strconv" |
| + "strings" |
| "time" |
| "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| "github.com/carlosframework/platform/internal/moves" |
| "github.com/carlosframework/platform/internal/provision" |
| ) |
| @@ -39,15 +41,36 @@ import ( |
| // 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. |
| +// again; a fresh worker finds probing and the counted attempt. One cold |
| +// admission per budgeted attempt (§7.7, fix round 1, ruling 2): an attempt |
| +// is admitted once Task 5's journalled session binding (probe_session_<s> = |
| +// k) exists for it, and an admitted attempt is consumed on restart, failed |
| +// unless it passed. Only a counted attempt with no binding is made again. |
| +// |
| +// A pass is journalled (probe_passed_<k>: the admitted session, stamp and |
| +// generation) before the probing to probed record write (ruling 1). A pass |
| +// that finds it does not probe again: it checks the claim's holder still |
| +// carries that writer and the activator's live probe entry is this move's, |
| +// then retries only the phase writes, record then claim. A writer that is |
| +// gone consumes the attempt as failed. |
| // |
| // 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_" |
| +// probeStepFailed marks attempt k over in the op journal; probeStepPassed |
| +// names attempt k's passing writer (probePassed). |
| +const ( |
| + probeStepFailed = "probe_failed_" |
| + probeStepPassed = "probe_passed_" |
| +) |
| + |
| +// probePassed is a probe_passed_<k> step's value: the admitted writer the |
| +// passing attempt left serving. |
| +type probePassed struct { |
| + Session uint64 `json:"session"` |
| + Stamp uint64 `json:"stamp"` |
| + Generation string `json:"generation"` |
| +} |
| // probeWindowLen is the attempt's window, counted from its first start. |
| var probeWindowLen = time.Duration(moves.ProbeSeconds * float64(time.Second)) |
| @@ -62,9 +85,10 @@ func MovesRecordAdmittedAttempts(_ context.Context, rec *provision.Record) (int, |
| } |
| // attemptSpent reports whether counted attempt k is over: its failure is |
| -// journalled, or its window opened and has passed. |
| +// journalled, a session was admitted under it (Task 5's binding), or its |
| +// window opened and has passed. |
| func (l *landingRun) attemptSpent(j *moves.Journal, k int) bool { |
| - if j.Done(probeStepFailed + strconv.Itoa(k)) { |
| + if j.Done(probeStepFailed+strconv.Itoa(k)) || attemptAdmitted(j, k) { |
| return true |
| } |
| raw, ok := j.Steps[probeStepOpened+strconv.Itoa(k)] |
| @@ -75,6 +99,58 @@ func (l *landingRun) attemptSpent(j *moves.Journal, k int) bool { |
| return err == nil && !l.cfg().Now().Before(opened.Add(probeWindowLen)) |
| } |
| +// attemptAdmitted reports whether a session is journalled as bound to |
| +// record attempt k (movesProbeAttempts.bindSession). |
| +func attemptAdmitted(j *moves.Journal, k int) bool { |
| + want := strconv.Itoa(k) |
| + for step, v := range j.Steps { |
| + if strings.HasPrefix(step, probeStepSession) && v == want { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +// failAttempt journals attempt k over. |
| +func (l *landingRun) failAttempt(k int) error { |
| + j, err := l.journal() |
| + if err != nil { |
| + return err |
| + } |
| + if j.Done(probeStepFailed + strconv.Itoa(k)) { |
| + return nil |
| + } |
| + j.Mark(probeStepFailed+strconv.Itoa(k), l.cfg().Now().UTC().Format(time.RFC3339Nano)) |
| + return j.Write(l.cfg().DataDir) |
| +} |
| + |
| +// passed writes the phase steps of a passing attempt: record, then claim. |
| +func (l *landingRun) passed(ctx context.Context) (bool, error) { |
| + if err := l.fault("before-record-probed"); err != nil { |
| + return false, err |
| + } |
| + return l.advance(ctx, provision.LandProbing, provision.LandProbed, nil) |
| +} |
| + |
| +// passedWriterLives checks a journalled pass against the claim's holder and |
| +// the activator's live probe entry. |
| +func (l *landingRun) passedWriterLives(obs hostclaim.Observed, raw string) bool { |
| + var p probePassed |
| + if err := json.Unmarshal([]byte(raw), &p); err != nil || p.Session == 0 { |
| + return false |
| + } |
| + h := obs.Claim.Holder |
| + if h == nil || h.Session != p.Session || h.Stamp != p.Stamp || h.Generation != p.Generation || h.WriterBox != l.cfg().Box { |
| + return false |
| + } |
| + live := l.cfg().ProbeLive |
| + if live == nil { |
| + return false |
| + } |
| + op, res, ok := live(l.rec.Host) |
| + return ok && op == l.g.ID && res.Session == p.Session && res.Stamp == p.Stamp && res.Generation == p.Generation |
| +} |
| + |
| // probing makes the landing's probe attempts (see the file comment). |
| func (l *landingRun) probing(ctx context.Context) (bool, error) { |
| cfg := l.cfg() |
| @@ -111,8 +187,25 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) { |
| counted = m.Landing.Attempts |
| } |
| k := counted |
| - if k == 0 || l.attemptSpent(j, k) { |
| - k++ |
| + if k > 0 { |
| + if raw, ok := j.Steps[probeStepPassed+strconv.Itoa(k)]; ok { |
| + if l.passedWriterLives(obs, raw) { |
| + log.Printf("moves: %s: landing %s attempt %d passed; its phase writes are retried", l.rec.Host, l.g.ID, k) |
| + return l.passed(ctx) |
| + } |
| + log.Printf("moves: %s: landing %s attempt %d passed, but its writer is gone; the attempt is consumed", l.rec.Host, l.g.ID, k) |
| + if err := l.failAttempt(k); err != nil { |
| + return false, err |
| + } |
| + k++ |
| + } else if l.attemptSpent(j, k) { |
| + if err := l.failAttempt(k); err != nil { |
| + return false, err |
| + } |
| + k++ |
| + } |
| + } else { |
| + k = 1 |
| } |
| for ; k <= n; k++ { |
| if l.pastDeadline() { |
| @@ -139,22 +232,29 @@ func (l *landingRun) probing(ctx context.Context) (bool, error) { |
| 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}) |
| + res, 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) |
| + raw, err := json.Marshal(probePassed{Session: res.Session, Stamp: res.Stamp, Generation: res.Generation}) |
| + if err != nil { |
| + return false, err |
| + } |
| + if j, err = l.journal(); err != nil { |
| + return false, err |
| + } |
| + j.Mark(probeStepPassed+strconv.Itoa(k), string(raw)) |
| + if err := j.Write(cfg.DataDir); err != nil { |
| + return false, err |
| + } |
| + return l.passed(ctx) |
| 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 { |
| + if err := l.failAttempt(k); err != nil { |
| return false, err |
| } |
| default: |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| index f4e19d5f6..050161923 100644 |
| --- a/internal/host/movespass.go |
| +++ b/internal/host/movespass.go |
| @@ -563,6 +563,9 @@ type moveCfg struct { |
| // 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) |
| + // ProbeLive is the activator's LiveProbe: the awake probe entry for a |
| + // host, its op and admitted session. Nil in production with Probe. |
| + ProbeLive func(host string) (op string, res activator.ProbeResult, ok bool) |
| // 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 |