amadan

carlos / platform Public

activator: ProbeWake, the probe as glue over the claim's admission path (Plan 3 Task 5a, H2, H18)

ProbeWake makes a probe entry and runs the ordinary claim wake with the
probe's policy (five starts, the 30 s window, the landing deadline, the
manifest and pins). doClaimWake adds the probe's refusals (row held by
the hold op, claim at the op's probing phase) and maps a failure after
admission to ErrProbeFailed. RetireExcept keeps a probe entry while its
row is held by the probe's hold op. No admission or restore logic.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d5407d02df7e816fdb4c8f312ea94ff5693bf9f4 parent 0f47cb5
4 files changed, +520 −6
  • internal/activator/activator.go +8 −1
  • internal/activator/claimwake.go +25 −5
  • internal/activator/probewake.go +188 −0
  • internal/activator/probewake_test.go +299 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 7dea2b1cf..7e5bdeb1e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -338,6 +338,10 @@ const (
type actEntry struct {
claimResult *wokenInstance
+ // probe is set for an entry ProbeWake made (moves Plan 3 Task 5): the
+ // claim path runs with the probe's policy, and the entry is held for
+ // the move while its registry row is.
+ probe *probeEntry
// bounceAfterWake is a Bounce that arrived while this INSTANCE was still
// waking. The wake had already read its config and execed its build, so
// the bounce's reason has not reached the child; wake() spends it the
@@ -2939,10 +2943,13 @@ func (a *Service) sweepOnce() {
// a retired child are cut off as a deploy restart would cut them — the flip is
// a deliberate operator action.
func (a *Service) RetireExcept(keep map[string]bool) {
+ // A probe's entry (ProbeWake) is held for its move while the registry
+ // row is: the edge manages ready rows only, so it is never in keep.
+ held := a.heldProbes(keep)
a.mu.Lock()
var waking []*actEntry
for _, e := range a.entries {
- if keep[e.host] {
+ if keep[e.host] || held[e] {
continue
}
// Workers are outside this reconcile's world entirely. `keep` is the
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index f6d3aa125..462c97ded 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -97,13 +97,19 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
}
if observed.Claim == nil || executor != r.Box {
stopErr := r.StopForeign(ctx, e.host)
- return nil, errors.Join(hostbox.ErrChanged, stopErr)
+ return nil, probeRefused(e, errors.Join(hostbox.ErrChanged, stopErr))
}
if e.route.WakeIncarnation != "" && (observed.Claim == nil || observed.Claim.Holder == nil || observed.Claim.Holder.Incarnation != e.route.WakeIncarnation) {
- return nil, hostbox.ErrChanged
+ return nil, probeRefused(e, hostbox.ErrChanged)
}
- if err := hostbox.WakePhase(observed.Claim, r.Box, false); err != nil {
- return nil, err
+ // A probe (ProbeWake, moves Plan 3 Task 5a) is admitted only at its
+ // op's probing phase (K6's probe row) and runs under its own policy.
+ probe := e.probe
+ if err := hostbox.WakePhase(observed.Claim, r.Box, probe != nil); err != nil {
+ return nil, probeRefused(e, err)
+ }
+ if probe != nil && observed.Claim.Op.ID != probe.op {
+ return nil, probeRefused(e, errProbeOp)
}
if r.Prerequisites == nil || r.Admit == nil || r.Prepare == nil || r.FailedAdmission == nil {
return nil, hostwrite.ErrUnavailable
@@ -120,6 +126,13 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
rt = *fresh
}
policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second}
+ if probe != nil {
+ // K1: the row is still held by the probe's hold op.
+ if !probeHeld(rt, probe.holdOp) {
+ return nil, probeRefused(e, errProbeRow)
+ }
+ policy = probe.policy
+ }
if err := r.Prerequisites(ctx, rt, observed, policy); err != nil {
return nil, err
}
@@ -134,7 +147,14 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
if err := a.startAdmitted(ctx, e, admitted, policy); err != nil {
recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), 10*time.Second)
defer cancelRecovery()
- return nil, errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err))
+ err = errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err))
+ if probe != nil {
+ err = errors.Join(ErrProbeFailed, err)
+ }
+ return nil, err
+ }
+ if probe != nil {
+ probe.admitted = &admitted
}
return e.claimResult, nil
}
diff --git a/internal/activator/probewake.go b/internal/activator/probewake.go
new file mode 100644
index 000000000..51a6714dc
--- /dev/null
+++ b/internal/activator/probewake.go
@@ -0,0 +1,188 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// The landing's probe wake (moves Phase 1 Plan 3 Task 5a; spec §7.7, §7.14;
+// ruling H2). ProbeWake is glue over the claim's probe admission path (claim
+// Plan 9 Group D): the same entry, wake and doClaimWake a request's wake
+// runs, with the probe's start policy and the probe's two refusals (the
+// registry row held by HoldOp, the claim this box's landing op at probing).
+// The claim runtime does everything else (Prerequisites, Admit, Prepare and
+// startAdmitted): the moves owners it dispatches to choose and evidence the
+// cold admission into the prepared generation, render and materialize the
+// held inputs, run the start check and publish; the moves attempts journal
+// the window and the starts; the claim's probe readiness checks the peer.
+// ProbeWake contains no admission, restore or readiness logic of its own.
+//
+// Production: unarmed. Nothing calls ProbeWake until the landing's probing
+// phase (Plan 3 Task 6) is armed with Plan 5 Task 9's parks, and on main
+// every claim start is refused (ClaimRoleAssembly.Preflight and
+// RoleLaunch.Start answer ErrClaimIsolationUnavailable) until the claim's
+// isolated root launcher lands.
+
+var (
+ // ErrProbeRefused is a probe refused before any admission: the row is not
+ // held by the probe's hold op, the claim does not name this box the
+ // executor of the probe's op at probing, or the host already has a live
+ // entry. Nothing was admitted and nothing ran.
+ ErrProbeRefused = errors.New("activator: the probe was refused before admission")
+ // ErrProbeFailed is an admitted probe attempt that did not pass: the
+ // claim settled the admitted tree (FailedAdmission) and the attempt is
+ // over. The cause is joined to it (ErrAttemptOver, a refusal from the
+ // moves owners, a readiness failure).
+ ErrProbeFailed = errors.New("activator: the probe attempt failed")
+)
+
+// The probe's start policy (§7.7): at most five starts, the 30 s window
+// counted from the first app start.
+const (
+ probeMaxStarts = 5
+ probeWindow30 = time.Duration(moves.ProbeSeconds * float64(time.Second))
+)
+
+// ProbeConfig is one probe attempt.
+type ProbeConfig struct {
+ Host string
+ // Op is the claim's op id: the move id.
+ Op string
+ // HoldOp is the registry row's hold_op: the move id for a landing.
+ HoldOp string
+ // Deadline bounds the whole attempt: the landing's deadline, the earlier
+ // of cut_deadline and landing_deadline (ruling H8).
+ Deadline time.Time
+ // Prepared and Inputs are the build's install manifest and the move's
+ // pins; the moves owners check them against the journal and the record.
+ Prepared *moves.InstallManifest
+ Inputs *provision.MoveInputs
+}
+
+// ProbeResult names the admitted session that passed: the process the entry
+// now holds, which serves after the open on the same session (H3).
+type ProbeResult struct {
+ Session uint64
+ Generation string
+}
+
+// probeEntry marks an actEntry ProbeWake made.
+type probeEntry struct {
+ policy startPolicy
+ op, holdOp string
+ admitted *hostclaim.Observed
+}
+
+// ProbeWake runs one probe attempt for cfg.Host and waits for it. On
+// success the host's entry is awake, holding the admitted session's process
+// and replication, and stays held while the row is (RetireExcept).
+func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult, error) {
+ if ctx == nil || cfg.Host == "" || cfg.Op == "" || cfg.HoldOp == "" || cfg.Deadline.IsZero() {
+ return ProbeResult{}, hostwrite.ErrUnavailable
+ }
+ if !a.cfg.HostWrites.Enabled || a.cfg.Claims == nil || a.Routes == nil {
+ return ProbeResult{}, hostwrite.ErrUnavailable
+ }
+ // The cut deadline alone, before any admission: the window opens only
+ // at the first app start (brief step 4).
+ if !a.now().Before(cfg.Deadline) {
+ return ProbeResult{}, errors.Join(ErrProbeRefused, ErrAttemptOver)
+ }
+ rt, err := a.Routes.Get(cfg.Host)
+ if err != nil {
+ return ProbeResult{}, err
+ }
+ if rt == nil || !probeHeld(*rt, cfg.HoldOp) {
+ return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeRow)
+ }
+ e := &actEntry{host: cfg.Host, route: *rt, state: actWaking, ready: make(chan struct{}),
+ probe: &probeEntry{op: cfg.Op, holdOp: cfg.HoldOp,
+ policy: startPolicy{MaxStarts: probeMaxStarts, Window: probeWindow30, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}}}
+ a.mu.Lock()
+ switch {
+ case a.fences[cfg.Host] != 0:
+ a.mu.Unlock()
+ return ProbeResult{}, errors.Join(ErrProbeRefused, ErrFenced)
+ case a.entries[cfg.Host] != nil:
+ a.mu.Unlock()
+ return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeLive)
+ }
+ a.entries[cfg.Host] = e
+ a.mu.Unlock()
+ go a.wake(e)
+ select {
+ case <-e.ready:
+ case <-ctx.Done():
+ a.mu.Lock()
+ if e.wakeCancel != nil {
+ e.wakeCancel()
+ }
+ a.mu.Unlock()
+ <-e.ready
+ }
+ // e.err and e.probe are written before e.ready closes.
+ if e.err != nil {
+ return ProbeResult{}, e.err
+ }
+ h := e.probe.admitted.Claim.Holder
+ return ProbeResult{Session: h.Session, Generation: h.Generation}, nil
+}
+
+var (
+ errProbeRow = errors.New("activator: the registry row is not held by the probe's hold op")
+ errProbeOp = errors.New("activator: the claim is not the probe's op")
+ errProbeLive = errors.New("activator: the host already has a live entry")
+)
+
+func probeHeld(rt registry.Route, holdOp string) bool {
+ return registry.IsHold(rt.Status) && rt.HoldOp == holdOp
+}
+
+// probeRefused marks a probe's refusal before admission; other wakes'
+// errors are unchanged.
+func probeRefused(e *actEntry, err error) error {
+ if e.probe == nil || err == nil {
+ return err
+ }
+ return errors.Join(ErrProbeRefused, err)
+}
+
+// heldProbes is the set of entries RetireExcept keeps beyond keep: probe
+// entries whose registry row is still held by the probe's hold op. The edge
+// manages only ready rows (edge.Reload), so a held probe is never in keep.
+// The rows are read outside a.mu. A row that cannot be read keeps its entry
+// for this reload: the next reload reads it again.
+func (a *Service) heldProbes(keep map[string]bool) map[*actEntry]bool {
+ a.mu.Lock()
+ var probes []*actEntry
+ for _, e := range a.entries {
+ if e.probe != nil && !keep[e.host] {
+ probes = append(probes, e)
+ }
+ }
+ a.mu.Unlock()
+ held := map[*actEntry]bool{}
+ if a.Routes == nil {
+ return held
+ }
+ for _, e := range probes {
+ rt, err := a.Routes.Get(e.host)
+ if err != nil {
+ log.Printf("activator: %s: the probe's row could not be read (%v); kept for this reload", e.host, err)
+ held[e] = true
+ continue
+ }
+ if rt != nil && probeHeld(*rt, e.probe.holdOp) {
+ held[e] = true
+ }
+ }
+ return held
+}
diff --git a/internal/activator/probewake_test.go b/internal/activator/probewake_test.go
new file mode 100644
index 000000000..1c8a0a866
--- /dev/null
+++ b/internal/activator/probewake_test.go
@@ -0,0 +1,299 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os/exec"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+const (
+ pwHost = "shop.example.com"
+ pwOp = "mv_probewake"
+)
+
+type pwRoutes struct {
+ mu sync.Mutex
+ rt registry.Route
+}
+
+func (r *pwRoutes) Get(string) (*registry.Route, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ rt := r.rt
+ return &rt, nil
+}
+
+// pwRuntime is a claim runtime that records what the activator asked of it.
+type pwRuntime struct {
+ mu sync.Mutex
+ observed hostclaim.Observed
+ calls []string
+ policies []startPolicy
+ failedFor []hostclaim.Observed
+ stop error
+ stopped int
+}
+
+func (r *pwRuntime) note(call string, p startPolicy) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.calls = append(r.calls, call)
+ r.policies = append(r.policies, p)
+}
+
+func (r *pwRuntime) admitted() hostclaim.Observed {
+ o := r.observed
+ c := *o.Claim
+ h := *c.Holder
+ h.Session, h.Generation, h.WriterBox = h.Session+5, pwOp, "box"
+ c.Holder = &h
+ o.Claim, o.ETag = &c, "admitted"
+ return o
+}
+
+func (r *pwRuntime) runtime(t *testing.T) *ClaimRuntime {
+ work, _ := (hostwrite.Config{}).Select(t.Context(), pwHost, "spawn")
+ return &ClaimRuntime{Box: "box",
+ Read: func(_ context.Context, host string) (hostclaim.Observed, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if len(r.calls) > 1 {
+ return r.admitted(), nil
+ }
+ return r.observed, nil
+ },
+ StopForeign: func(context.Context, string) error { r.mu.Lock(); r.stopped++; r.mu.Unlock(); return nil },
+ Prerequisites: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) error {
+ r.note("prerequisites", p)
+ return nil
+ },
+ Admit: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
+ r.note("admit", p)
+ return r.admitted(), nil
+ },
+ Prepare: func(_ context.Context, _ registry.Route, got hostclaim.Observed, p startPolicy) (ClaimStart, error) {
+ r.note("prepare", p)
+ return ClaimStart{RoleLauncher: &RoleLaunch{}, CurrentAttempt: func() hostclaim.WakeRetryAttempt { return hostclaim.WakeRetryAttempt{} }, Host: pwHost, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"},
+ OpenProbe: func(context.Context, time.Time) (time.Time, error) { return time.Now(), nil },
+ Restore: func(context.Context) error { r.note("restore", p); return nil },
+ Materialize: func(context.Context) error { r.note("materialize", p); return r.stop },
+ Build: func() *exec.Cmd { return nil }, Ready: func(context.Context, *exec.Cmd) error { return nil },
+ Reserve: func(_ context.Context, n int) error { r.note("reserve", p); return nil },
+ Failed: func(context.Context, int) error { r.note("failed", p); return nil },
+ Healthy: func(context.Context, int, *exec.Cmd) error { return nil },
+ Replicate: func(context.Context, string) (Rep, error) { return nil, nil },
+ Publish: func(_ context.Context, o hostclaim.Observed) (hostclaim.Observed, error) { return o, nil },
+ Retain: func(Proc, Rep, error) {}}, nil
+ },
+ FailedAdmission: func(_ context.Context, o hostclaim.Observed, _ error) error {
+ r.mu.Lock()
+ r.failedFor = append(r.failedFor, o)
+ r.mu.Unlock()
+ return nil
+ }}
+}
+
+func pwFlip() hostclaim.Observed {
+ return hostclaim.Observed{Host: pwHost, ETag: "flip", Claim: &hostclaim.Claim{Host: pwHost,
+ Holder: &hostclaim.Holder{Incarnation: "inc", Location: "instances/" + pwHost, Stamp: 18, Session: 15, Generation: "src", WriterBox: "ie-box"},
+ Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: pwOp, Phase: hostclaim.PhaseLandingProbing, Exec: "box"}}}
+}
+
+func pwService(t *testing.T, r *pwRuntime) (*Service, *pwRoutes) {
+ t.Helper()
+ routes := &pwRoutes{rt: registry.Route{Host: pwHost, Kind: "instance", Status: "landing", HoldOp: pwOp, Socket: "/run/x.sock", DBPath: "/data/x.db"}}
+ a := New(Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: r.runtime(t)}, &fakeBackend{}, nil)
+ a.Routes = routes
+ return a, routes
+}
+
+func pwConfig() ProbeConfig {
+ return ProbeConfig{Host: pwHost, Op: pwOp, HoldOp: pwOp, Deadline: time.Now().Add(time.Minute),
+ Prepared: &moves.InstallManifest{Op: pwOp, Host: pwHost, N: 42}, Inputs: &provision.MoveInputs{Release: "v1"}}
+}
+
+// ProbeWake is the claim path with the probe's policy (Plan 3 Task 5a, H2):
+// the claim's own Prerequisites, Admit and admitted start sequence, each
+// given the probe policy (five starts, the 30 s window, the deadline, the
+// prepared manifest and the pins). It adds no admission or restore of its
+// own. A failure after the admission is ErrProbeFailed, the admitted tree is
+// settled through FailedAdmission, and the entry is gone.
+func TestProbeWakeRunsTheClaimPathWithTheProbePolicy(t *testing.T) {
+ stop := errors.New("materialize failed")
+ r := &pwRuntime{observed: pwFlip(), stop: stop}
+ a, _ := pwService(t, r)
+ cfg := pwConfig()
+ _, err := a.ProbeWake(t.Context(), cfg)
+ if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, stop) || errors.Is(err, ErrProbeRefused) {
+ t.Fatalf("err = %v", err)
+ }
+ if got := strings.Join(r.calls, ","); got != "prerequisites,admit,prepare,restore,reserve,materialize,failed" {
+ t.Fatalf("calls: %s", got)
+ }
+ want := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}
+ for i, p := range r.policies {
+ if p != want {
+ t.Fatalf("%s ran under %+v", r.calls[i], p)
+ }
+ }
+ if len(r.failedFor) != 1 || r.failedFor[0].ETag != "admitted" {
+ t.Fatalf("FailedAdmission: %+v", r.failedFor)
+ }
+ if a.Owns(pwHost) {
+ t.Fatal("a failed probe left its entry")
+ }
+}
+
+// TestProbeWakeRefusesUnheldRowOrForeignExec (the activator's half; the
+// owners' is TestProbeRefusesUnheldRowOrForeignExec in internal/host): the
+// row must be held with hold_op == HoldOp and the claim must name this box
+// op.exec under op Op at probing, else ErrProbeRefused before any admission.
+func TestProbeWakeRefusesUnheldRowOrForeignExec(t *testing.T) {
+ for name, tc := range map[string]struct {
+ route func(*registry.Route)
+ claim func(*hostclaim.Claim)
+ cfg func(*ProbeConfig)
+ }{
+ "a ready row": {route: func(r *registry.Route) { r.Status, r.HoldOp = "ready", "" }},
+ "held by another op": {route: func(r *registry.Route) { r.HoldOp = "mv_other" }},
+ "a foreign executor": {claim: func(c *hostclaim.Claim) { c.Op.Exec = "other-box" }},
+ "another op": {claim: func(c *hostclaim.Claim) { c.Op.ID = "mv_other" }},
+ "not probing": {claim: func(c *hostclaim.Claim) { c.Op.Phase = hostclaim.PhaseLandingCommitted }},
+ "a serving claim": {claim: func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} }},
+ "another hold op": {cfg: func(c *ProbeConfig) { c.HoldOp = "rb_other" }},
+ } {
+ t.Run(name, func(t *testing.T) {
+ r := &pwRuntime{observed: pwFlip()}
+ if tc.claim != nil {
+ tc.claim(r.observed.Claim)
+ }
+ a, routes := pwService(t, r)
+ if tc.route != nil {
+ tc.route(&routes.rt)
+ }
+ cfg := pwConfig()
+ if tc.cfg != nil {
+ tc.cfg(&cfg)
+ }
+ _, err := a.ProbeWake(t.Context(), cfg)
+ if !errors.Is(err, ErrProbeRefused) {
+ t.Fatalf("err = %v", err)
+ }
+ if len(r.calls) != 0 {
+ t.Fatalf("the claim path ran: %v", r.calls)
+ }
+ if a.Owns(pwHost) {
+ t.Fatal("a refused probe left an entry")
+ }
+ })
+ }
+ // A live entry for the host (an earlier probe's process, or a wake) is
+ // never replaced by a probe.
+ r := &pwRuntime{observed: pwFlip()}
+ a, _ := pwService(t, r)
+ a.mu.Lock()
+ a.entries[pwHost] = &actEntry{host: pwHost, state: actWaking, ready: make(chan struct{})}
+ a.mu.Unlock()
+ if _, err := a.ProbeWake(t.Context(), pwConfig()); !errors.Is(err, ErrProbeRefused) || len(r.calls) != 0 {
+ t.Fatalf("over a live entry: %v %v", err, r.calls)
+ }
+ // Malformed configurations never reach the claim.
+ for _, mutate := range []func(*ProbeConfig){
+ func(c *ProbeConfig) { c.Host = "" }, func(c *ProbeConfig) { c.Op = "" }, func(c *ProbeConfig) { c.HoldOp = "" },
+ func(c *ProbeConfig) { c.Deadline = time.Time{} },
+ } {
+ cfg := pwConfig()
+ mutate(&cfg)
+ if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, hostwrite.ErrUnavailable) {
+ t.Fatalf("malformed config %+v: %v", cfg, err)
+ }
+ }
+}
+
+// The cut deadline is checked before the admission, with the window not yet
+// open (brief step 4): a deadline already past is ErrAttemptOver and nothing
+// reaches the claim. A deadline that passes during the admitted sequence
+// ends it with ErrAttemptOver through ErrProbeFailed (startAdmitted's loop,
+// claimloop_linux_test.go). The 30 s window itself opens only inside
+// RoleLaunch.Start, which main still refuses
+// (TestStartAdmittedLoopWindowAndReadinessAtTheBoundary).
+func TestProbeWakeEndsAtTheCutDeadline(t *testing.T) {
+ r := &pwRuntime{observed: pwFlip()}
+ a, _ := pwService(t, r)
+ cfg := pwConfig()
+ cfg.Deadline = time.Now().Add(-time.Second)
+ if _, err := a.ProbeWake(t.Context(), cfg); !errors.Is(err, ErrAttemptOver) || errors.Is(err, ErrProbeFailed) || len(r.calls) != 0 {
+ t.Fatalf("err = %v calls = %v", err, r.calls)
+ }
+ // Past the deadline once admitted: the admitted sequence ends at once.
+ cfg.Deadline = time.Now().Add(100 * time.Millisecond)
+ r.stop = errors.New("unused")
+ slow := r.runtime(t)
+ admit := slow.Admit
+ slow.Admit = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p startPolicy) (hostclaim.Observed, error) {
+ time.Sleep(150 * time.Millisecond)
+ return admit(ctx, rt, o, p)
+ }
+ a.cfg.Claims = slow
+ _, err := a.ProbeWake(t.Context(), cfg)
+ if !errors.Is(err, ErrProbeFailed) || !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("err = %v", err)
+ }
+ for _, call := range r.calls {
+ if call == "reserve" {
+ t.Fatal("a start was reserved past the deadline")
+ }
+ }
+}
+
+// TestProbeAttemptEndsAtTheWindow (brief): blocked on the claim's isolated
+// root launcher, as the loop test it would extend is.
+func TestProbeAttemptEndsAtTheWindow(t *testing.T) {
+ t.Skip("blocked: RoleLaunch.Start and ClaimRoleAssembly.Preflight return ErrClaimIsolationUnavailable on main " +
+ "(claim Plan 9 Group D did not replace them), so OpenProbe and Ready never run through startAdmitted. " +
+ "When they do: OpenProbe answering a time 31 s ago, readiness answering after it, gives ErrProbeFailed and ErrAttemptOver.")
+}
+
+// The probe's entry stays held: an edge reload (RetireExcept) never stops a
+// probe entry while the registry row is still held by the probe's hold op,
+// and does once it is not (H18).
+func TestRetireExceptKeepsTheHeldProbeEntry(t *testing.T) {
+ a, _, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ held := registry.Route{Host: "one.example.com", Kind: "instance", Socket: "/tmp/nope.sock", DBPath: "/tmp/one.example.com.db", Hibernate: true}
+ if err := reg.Upsert(held); err != nil {
+ t.Fatal(err)
+ }
+ release, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ release()
+ if err := reg.SetHold("one.example.com", "landing", pwOp); err != nil {
+ t.Fatal(err)
+ }
+ a.mu.Lock()
+ a.entries["one.example.com"].probe = &probeEntry{policy: startPolicy{Probe: true}, op: pwOp, holdOp: pwOp}
+ a.mu.Unlock()
+
+ a.RetireExcept(map[string]bool{})
+ time.Sleep(50 * time.Millisecond)
+ if !a.Owns("one.example.com") {
+ t.Fatal("an edge reload retired the held probe's entry")
+ }
+ if err := reg.SetHold("one.example.com", "landing", "mv_other"); err != nil {
+ t.Fatal(err)
+ }
+ a.RetireExcept(map[string]bool{})
+ waitFor(t, "the entry no longer held for its probe to retire", func() bool { return !a.Owns("one.example.com") })
+}