amadan
host: the relocation pass says a claim it cannot read rather than pass it over, a full pass that could not be built at startup is tried again every pass behind the completion-only runtime rather than read as no relocation on the box, and the agent says why it starts no completion pass in claim mode; hostrelocate: a converted landing's wait with no readable fence observation is refused with the chooser's answer rather than left unbounded
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+290
−23
internal/host/relocatepass.go+81 −18internal/host/relocatepass_absence_test.go+140 −0internal/hostrelocate/worker.go+20 −5internal/hostrelocate/worker_bounds_test.go+49 −0
diff --git a/internal/host/relocatepass.go b/internal/host/relocatepass.go| index 5ab01a894..eda18ecd5 100644 |
| --- a/internal/host/relocatepass.go |
| +++ b/internal/host/relocatepass.go |
| @@ -58,6 +58,12 @@ type RelocationRuntime struct { |
| Self func(context.Context) (string, error) |
| DataDir string // the managed data root: relocate/ journals (N9) |
| alerted sync.Map // host\x00cause: one alert per host and cause (rulings 35, 36) |
| + // upgrade, when set, builds the full pass this completion-only runtime |
| + // stands in for: region HA is on, but the build failed (a roster, an |
| + // enrollment or a key that could not be read). Each pass tries it |
| + // again until it builds, or answers that the owners are not bound. A |
| + // read that failed once is never taken as "no relocation on this box". |
| + upgrade func(context.Context) (*RelocationRuntime, error) |
| } |
| // relocationStepper is hostrelocate.Worker's and Participant's Step. |
| @@ -385,6 +391,7 @@ func (r *RelocationRuntime) Run(ctx context.Context, store release.Store, serial |
| // found from the claim's record (ruling 80). A landing whose target is |
| // fenced is converted when its move committed (J3). |
| func (r *RelocationRuntime) once(ctx context.Context, store release.Store) { |
| + r.tryUpgrade(ctx) |
| hosts, err := r.hosts(ctx, store) |
| if err != nil { |
| log.Printf("relocate: list hosts: %v", err) |
| @@ -400,7 +407,13 @@ func (r *RelocationRuntime) once(ctx context.Context, store release.Store) { |
| } |
| for _, host := range hosts { |
| obs, err := hostclaim.Read(ctx, store, host) |
| - if err != nil || obs.Absent { |
| + if err != nil { |
| + // A claim that cannot be read is not "nothing to do": said once, |
| + // and read again next pass. |
| + r.alert(host, "read", err) |
| + continue |
| + } |
| + if obs.Absent { |
| continue |
| } |
| switch obs.Claim.Op.Kind { |
| @@ -434,6 +447,27 @@ func (r *RelocationRuntime) once(ctx context.Context, store release.Store) { |
| } |
| } |
| +// tryUpgrade replaces a completion-only stand-in with the full pass once |
| +// it builds (upgrade). A build that answers no relocation owners ends the |
| +// tries: completions are then all this box runs, as NewRelocationRuntime |
| +// would have answered at startup. |
| +func (r *RelocationRuntime) tryUpgrade(ctx context.Context) { |
| + if r.upgrade == nil { |
| + return |
| + } |
| + full, err := r.upgrade(ctx) |
| + if err != nil { |
| + r.alert("", "region HA: relocation pass not built, finishing completions only", err) |
| + return |
| + } |
| + r.upgrade = nil |
| + if full == nil || full.Worker == nil { |
| + return |
| + } |
| + r.Worker, r.Participant, r.Box, r.Convert, r.Finish, r.Hosts, r.Self, r.DataDir = full.Worker, full.Participant, full.Box, full.Convert, full.Finish, full.Hosts, full.Self, full.DataDir |
| + log.Printf("agent: region HA: relocation pass started") |
| +} |
| + |
| // hosts is every claim in the bucket, or Hosts when it is set. |
| func (r *RelocationRuntime) hosts(ctx context.Context, store release.Store) ([]string, error) { |
| if r.Hosts != nil { |
| @@ -519,6 +553,10 @@ func convertFencedLanding(w hostrelocate.Worker, store release.Store) func(conte |
| func (r *RelocationRuntime) alert(host, who string, err error) { |
| if _, said := r.alerted.LoadOrStore(host+"\x00"+err.Error(), struct{}{}); !said { |
| + if host == "" { |
| + log.Printf("relocate: %s: %v", who, err) |
| + return |
| + } |
| log.Printf("relocate %s: %s: %v", host, who, err) |
| } |
| } |
| @@ -549,11 +587,14 @@ func registerRelocationSource(claimCtx context.Context, dataDir string) error { |
| // activator, or when not even the completion pass can be built (logged). |
| func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring, reg *registry.Registry, svc *activator.Service) *RelocationRuntime { |
| getenv := claimEntryEnv(claimCtx) |
| - if !hostwrite.Enabled(getenv) || claims == nil || svc == nil { |
| + if !hostwrite.Enabled(getenv) { |
| return nil |
| } |
| entry, _ := claimCtx.Value(claimEntryKey{}).(claimEntry) |
| - if entry.Assembly == nil { |
| + // Claim mode is on, so a completion recorded on this box must be |
| + // finished here (ruling 82): every way of not starting the pass is said. |
| + if claims == nil || svc == nil || entry.Assembly == nil { |
| + log.Printf("agent: relocation completion pass not started: claim mode is on but the claim wiring, the activator or the claim assembly is missing") |
| return nil |
| } |
| local, ok := entry.Assembly.Local.(*productionClaimLocal) |
| @@ -561,25 +602,47 @@ func startRelocationPass(agentCtx, claimCtx context.Context, claims *ClaimWiring |
| log.Printf("agent: region HA: relocation pass not started: the assembly has no production participation owner") |
| return nil |
| } |
| - var runtime *RelocationRuntime |
| - var err error |
| + var full func(context.Context) (*RelocationRuntime, error) |
| if regionHAEnabled(getenv) { |
| - var in relocationRuntimeInputs |
| - if in, err = productionRelocationRuntimeInputs(os.Getenv, release.OptDir()); err == nil { |
| - runtime, err = NewRelocationRuntime(agentCtx, getenv, entry.Store, local, claims.Providers, reg, svc, in) |
| - } |
| - if err != nil { |
| - log.Printf("agent: region HA: relocation pass not started, finishing completions only: %v", err) |
| + full = func(ctx context.Context) (*RelocationRuntime, error) { |
| + in, err := productionRelocationRuntimeInputs(os.Getenv, release.OptDir()) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return NewRelocationRuntime(ctx, getenv, entry.Store, local, claims.Providers, reg, svc, in) |
| } |
| } |
| - if runtime == nil { |
| - // Ruling 82: with claim mode on, a relocation that completed here |
| - // is finished whatever CARLOS_REGION_HA says. |
| - if runtime, err = newCompletionRuntime(local, reg, svc); err != nil { |
| - log.Printf("agent: relocation completion pass not started: %v", err) |
| - return nil |
| - } |
| + runtime, err := relocationPassFor(agentCtx, local, reg, svc, full) |
| + if err != nil { |
| + log.Printf("agent: relocation completion pass not started: %v", err) |
| + return nil |
| } |
| go runtime.Run(agentCtx, entry.Store, nil) |
| return runtime |
| } |
| + |
| +// relocationPassFor is the runtime the agent runs in claim mode: the full |
| +// pass when full (region HA on) builds it, otherwise the completion-only |
| +// one (ruling 82). A full build that fails is not "no relocation on this |
| +// box": the completion-only runtime stands in and tries the build again |
| +// every pass (upgrade) until it builds or answers that the owners are not |
| +// bound. |
| +func relocationPassFor(ctx context.Context, local *productionClaimLocal, reg *registry.Registry, act RelocationActivator, full func(context.Context) (*RelocationRuntime, error)) (*RelocationRuntime, error) { |
| + var failed error |
| + if full != nil { |
| + r, err := full(ctx) |
| + if err == nil && r != nil { |
| + return r, nil |
| + } |
| + failed = err |
| + } |
| + r, err := newCompletionRuntime(local, reg, act) |
| + if err != nil { |
| + return nil, err |
| + } |
| + if failed != nil { |
| + r.alert("", "region HA: relocation pass not built, finishing completions only", failed) |
| + r.upgrade = full |
| + } |
| + return r, nil |
| +} |
diff --git a/internal/host/relocatepass_absence_test.go b/internal/host/relocatepass_absence_test.go| new file mode 100644 |
| index 000000000..b94ebc7e6 |
| --- /dev/null |
| +++ b/internal/host/relocatepass_absence_test.go |
| @@ -0,0 +1,140 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "path/filepath" |
| + "slices" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// unreadableClaim is a control bucket whose reads of one claim key fail. |
| +type unreadableClaim struct { |
| + release.Store |
| + key string |
| +} |
| + |
| +func (s unreadableClaim) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == s.key { |
| + return nil, "", errors.New("the claim could not be read") |
| + } |
| + return s.Store.Get(ctx, key) |
| +} |
| + |
| +// Absence is never evidence: a claim the pass cannot read is said, once, |
| +// and read again next pass; it is never passed over as nothing to do. The |
| +// other hosts' steps run as before. |
| +func TestRelocatePassSaysAClaimItCannotRead(t *testing.T) { |
| + ctx := t.Context() |
| + dir := release.NewDirStore(t.TempDir()) |
| + passClaim(t, dir, "a.example", relocatingOp("rel-a")) |
| + passClaim(t, dir, "b.example", relocatingOp("rel-b")) |
| + key, _ := hostclaim.Key("b.example") |
| + store := unreadableClaim{Store: dir, key: key} |
| + buf := captureRacyLog(t) |
| + var order []string |
| + r := &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order}, Participant: stepRecorder{name: "participant", order: &order}, Box: "dst"} |
| + r.once(ctx, store) |
| + r.once(ctx, store) |
| + if want := []string{"participant:a.example", "worker:a.example", "participant:a.example", "worker:a.example"}; !slices.Equal(order, want) { |
| + t.Fatalf("stepped %v, want %v", order, want) |
| + } |
| + if n := strings.Count(buf.String(), "relocate b.example: read: the claim could not be read"); n != 1 { |
| + t.Fatalf("an unreadable claim was said %d times, want once:\n%s", n, buf.String()) |
| + } |
| +} |
| + |
| +// Absence is never evidence: with region HA on, a full pass that could not |
| +// be built at startup (the roster, the enrollment or the release key not |
| +// readable then) is not "no relocation on this box". The completion-only |
| +// runtime stands in, finishing completions, and tries the build again each |
| +// pass until it builds; then the same Run is the full pass. A build that |
| +// answers the owners are not bound ends the tries. |
| +func TestRelocationPassBuildsTheFullPassOnceItCan(t *testing.T) { |
| + ctx := t.Context() |
| + store := release.NewDirStore(t.TempDir()) |
| + passClaim(t, store, "a.example", relocatingOp("rel-a")) |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer reg.Close() |
| + local := &productionClaimLocal{assembly: &ClaimAssembly{Claims: store, Guard: hostguard.Guard{DataDir: t.TempDir()}, stopLocal: func(context.Context) (claimStopLocal, error) { |
| + return claimStopLocal{Actor: hostbox.Actor{Box: "dst"}}, nil |
| + }}} |
| + buf := captureRacyLog(t) |
| + var order []string |
| + builds := 0 |
| + unreadable := errors.New("the roster could not be read") |
| + full := func(context.Context) (*RelocationRuntime, error) { |
| + builds++ |
| + if builds <= 2 { |
| + return nil, unreadable |
| + } |
| + return &RelocationRuntime{Worker: stepRecorder{name: "worker", order: &order}, Participant: stepRecorder{name: "participant", order: &order}, Box: "dst", |
| + Finish: func(context.Context, string, string) error { return nil }}, nil |
| + } |
| + r, err := relocationPassFor(ctx, local, reg, &activator.Service{}, full) |
| + if err != nil || r == nil || r.Worker != nil || r.Finish == nil || r.Hosts == nil { |
| + t.Fatalf("not the completion-only stand-in: %+v %v", r, err) |
| + } |
| + r.once(ctx, store) // the second build fails too |
| + if len(order) != 0 || builds != 2 { |
| + t.Fatalf("stepped %v after %d builds", order, builds) |
| + } |
| + r.once(ctx, store) // the third builds |
| + if want := []string{"participant:a.example", "worker:a.example"}; !slices.Equal(order, want) || r.Box != "dst" || r.Hosts != nil { |
| + t.Fatalf("not upgraded: stepped %v, %+v", order, r) |
| + } |
| + r.once(ctx, store) |
| + if builds != 3 { |
| + t.Fatalf("built %d times, want no more once built", builds) |
| + } |
| + if n := strings.Count(buf.String(), unreadable.Error()); n != 1 { |
| + t.Fatalf("the failed build was said %d times, want once:\n%s", n, buf.String()) |
| + } |
| + // Owners not bound: the build answers the completion-only runtime, and |
| + // the tries end. |
| + builds = 0 |
| + r, err = relocationPassFor(ctx, local, reg, &activator.Service{}, func(context.Context) (*RelocationRuntime, error) { |
| + builds++ |
| + if builds == 1 { |
| + return nil, unreadable |
| + } |
| + return newCompletionRuntime(local, reg, &activator.Service{}) |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + r.once(ctx, store) |
| + r.once(ctx, store) |
| + if builds != 2 || r.Worker != nil { |
| + t.Fatalf("built %d times (want 2), worker %v", builds, r.Worker) |
| + } |
| +} |
| + |
| +// Claim mode is on, so a completion recorded here must be finished here: |
| +// the agent never leaves the completion pass unstarted without saying so. |
| +func TestStartRelocationPassSaysWhyItDidNotStart(t *testing.T) { |
| + buf := captureRacyLog(t) |
| + claimCtx := context.WithValue(t.Context(), claimEntryKey{}, claimEntry{Enabled: true, FlagRead: true}) |
| + if r := startRelocationPass(t.Context(), claimCtx, &ClaimWiring{}, nil, &activator.Service{}); r != nil { |
| + t.Fatalf("started over no assembly: %+v", r) |
| + } |
| + if !strings.Contains(buf.String(), "relocation completion pass not started") { |
| + t.Fatalf("not said:\n%s", buf.String()) |
| + } |
| + off := context.WithValue(t.Context(), claimEntryKey{}, claimEntry{FlagRead: true}) |
| + before := buf.String() |
| + if r := startRelocationPass(t.Context(), off, &ClaimWiring{}, nil, &activator.Service{}); r != nil || buf.String() != before { |
| + t.Fatalf("claim mode off: %+v %q", r, strings.TrimPrefix(buf.String(), before)) |
| + } |
| +} |
diff --git a/internal/hostrelocate/worker.go b/internal/hostrelocate/worker.go| index 97718804a..669261f7b 100644 |
| --- a/internal/hostrelocate/worker.go |
| +++ b/internal/hostrelocate/worker.go |
| @@ -231,18 +231,29 @@ func (w Worker) ConvertLanding(ctx context.Context, obs hostclaim.Observed, guar |
| } |
| // newestFenceObservation is the latest fence.<box>.observed_at the op |
| -// journals. |
| -func newestFenceObservation(c *hostclaim.Claim) (time.Time, bool) { |
| +// journals. An observation that does not parse, or none at all, is an |
| +// error, never "no bound" (absence is never evidence): the conversion |
| +// journals its target's fence observation in the same CAS that leaves the |
| +// op waiting, so a wait with nothing to bound it is a defect, refused as |
| +// an unreadable installed_at is. |
| +func newestFenceObservation(c *hostclaim.Claim) (time.Time, error) { |
| var newest time.Time |
| for k, v := range c.Op.Journal { |
| if !strings.HasPrefix(k, "fence.") || !strings.HasSuffix(k, ".observed_at") { |
| continue |
| } |
| - if at, err := time.Parse(time.RFC3339Nano, v); err == nil && at.After(newest) { |
| + at, err := time.Parse(time.RFC3339Nano, v) |
| + if err != nil { |
| + return time.Time{}, fmt.Errorf("%w: %s of %s: %v", ErrRefused, k, c.Op.ID, err) |
| + } |
| + if at.After(newest) { |
| newest = at |
| } |
| } |
| - return newest, !newest.IsZero() |
| + if newest.IsZero() { |
| + return time.Time{}, fmt.Errorf("%w: %s journals no fence observation to bound its wait for a target", ErrRefused, c.Op.ID) |
| + } |
| + return newest, nil |
| } |
| // Start writes the start CAS for req (§5.8 "The start CAS", §5.10). It |
| @@ -398,7 +409,11 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro |
| // target is stuck ReleaseDeadline after its target's |
| // fence was observed; before that the chooser's refusal |
| // is returned and the pass alerts once. |
| - if at, ok := newestFenceObservation(c); ok && w.Now().Sub(at) > ReleaseDeadline { |
| + at, ferr := newestFenceObservation(c) |
| + if ferr != nil { |
| + return obs, errors.Join(err, ferr) |
| + } |
| + if w.Now().Sub(at) > ReleaseDeadline { |
| return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("no eligible box for the converted landing after %s: %v", ReleaseDeadline, err))) |
| } |
| return obs, err |
diff --git a/internal/hostrelocate/worker_bounds_test.go b/internal/hostrelocate/worker_bounds_test.go| index 8dc09b658..848005eb6 100644 |
| --- a/internal/hostrelocate/worker_bounds_test.go |
| +++ b/internal/hostrelocate/worker_bounds_test.go |
| @@ -215,3 +215,52 @@ func TestWorkerBoundsAConversionWaitingForATarget(t *testing.T) { |
| t.Fatalf("not stuck for an operator: %+v", o.Claim.Op) |
| } |
| } |
| + |
| +// Absence is never evidence: a converted landing whose fence observation |
| +// does not parse, or is gone, has nothing to bound its wait. The chooser's |
| +// refusal comes back with that defect named (ErrRefused), long after the |
| +// bound would have passed, and nothing is written: it is neither waited on |
| +// silently for ever nor stuck on a guess. |
| +func TestWorkerRefusesAConversionWaitWithNoReadableFenceObservation(t *testing.T) { |
| + for name, change := range map[string]func(map[string]string){ |
| + "unparseable": func(j map[string]string) { |
| + for k := range j { |
| + if strings.HasPrefix(k, "fence.") && strings.HasSuffix(k, ".observed_at") { |
| + j[k] = "not a time" |
| + } |
| + } |
| + }, |
| + "unparseable beside a readable one": func(j map[string]string) { |
| + for k := range j { |
| + if strings.HasPrefix(k, "fence.") && strings.HasSuffix(k, ".observed_at") { |
| + j[k] = "not a time" |
| + } |
| + } |
| + j["fence.spa.observed_at"] = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Format(time.RFC3339Nano) |
| + }, |
| + "missing": func(j map[string]string) { |
| + for k := range j { |
| + if strings.HasPrefix(k, "fence.") && strings.HasSuffix(k, ".observed_at") { |
| + delete(j, k) |
| + } |
| + } |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + f := newFixture(t) |
| + converted := f.step(f.convertedLanding(), func(c *hostclaim.Claim) error { change(c.Op.Journal); return nil }) |
| + none := errors.New("no eligible box") |
| + f.w.Choose = func(context.Context, hostclaim.Observed, []string) (hostclaim.Target, error) { |
| + return hostclaim.Target{}, none |
| + } |
| + f.now = f.now.Add(10 * ReleaseDeadline) |
| + o, err := f.w.Step(context.Background(), host) |
| + if !errors.Is(err, none) || !errors.Is(err, ErrRefused) { |
| + t.Fatalf("step: %v, want the chooser's refusal and the defect", err) |
| + } |
| + if o.ETag != converted.ETag || f.claim().ETag != converted.ETag { |
| + t.Fatalf("the op was written: %+v", f.claim().Claim.Op) |
| + } |
| + }) |
| + } |
| +} |