amadan

carlos / platform Public

region HA 2b: joining mode, the gate and the hosting flag

RunAgent asks hostingGate before runAgent: with claim mode or region HA
off it hosts at once; otherwise it selects the mode, waits out an unknown
one (any facts error included), and on joining runs joining until the box
can host. Joining publishes its identity record, answers only the load
balancer's health check (503, naming hosting), touches the join marker
once a minute while invited and unarmed so a failed carlos system join is
retried, and publishes its guard-ready report once armed.

Before the gate, a box carrying the group marker resolves its group
reference with backoff, whatever the switches say, so runAgent's loops
never keep an empty label.

ClaimWiring.startRegional sets the health answer's hosting condition from
the gate's decision before the claim worker starts and clears it when the
worker ends.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude a8822d6b75d248de61a0a26bba264cfaaa3f9775 parent fb7dd5e
7 files changed, +770 −19
  • internal/host/agent.go +14 −1
  • internal/host/claimwiring.go +12 −7
  • internal/host/claimwiring_hosting_test.go +33 −10
  • internal/host/joining.go +296 −0
  • internal/host/joining_test.go +412 −0
  • internal/provision/testdata/plan14-root-reads-region-ha-2b.tsv +2 −1
  • region_ha_boundary_test.go +1 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index dd9441505..67b0fa7c8 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -210,7 +210,20 @@ func dataDir() string {
// edge.Run's (-registry, -dev, -http, -https, ...): the args pass through to
// edge.RunWithActivator, which owns the serving loop and blocks until
// SIGTERM/SIGINT.
-func RunAgent(args []string) { runAgent(args, productionClaimFactory) }
+//
+// With CARLOS_HOST_CLAIM and CARLOS_REGION_HA on, a box that is not yet
+// enrolled with its guard armed runs region HA's joining mode instead, and
+// RunAgent returns once it can host, so systemd starts it again as the
+// claim agent (hostingGate, joining.go).
+//
+// Before either, a box its region's group launched resolves its group
+// reference, whatever the switches say: the loops runAgent starts read
+// the box's label once and keep it (agentEntry, joining.go).
+func RunAgent(args []string) {
+ agentEntry(os.Getenv, resolveGroupBoxReference, groupReferenceBackoff, log.Printf,
+ func() bool { return hostingGate(args) },
+ func() { runAgent(args, productionClaimFactory) })
+}
func runAgent(args []string, buildClaims ClaimFactory) {
agentCtx, cancelAgent := context.WithCancel(context.Background())
diff --git a/internal/host/claimwiring.go b/internal/host/claimwiring.go
index 94388fd7f..aef973f15 100644
--- a/internal/host/claimwiring.go
+++ b/internal/host/claimwiring.go
@@ -94,13 +94,18 @@ func (w *ClaimWiring) startRegional(ctx context.Context, serial <-chan struct{},
workerCtx, cancel := context.WithCancel(ctx)
w.cancel = cancel
w.joined = make(chan struct{})
- go func() { defer close(w.joined); w.Providers.Regional.Loop(workerCtx, serial, report) }()
- // The claim agent is running: it refused to start without its
- // host-actor enrollment and a matching roster row, so this box is
- // hosting (region HA spec 2 §5, §6). Plan 2b's mode selection owns
- // this flag from then on: joining mode leaves it false, and only an
- // active roster entry with an armed guard sets it.
- lbhealth.Default.SetHosting(true)
+ // The load balancer's hosting condition (region HA spec 2 §5, §6) is
+ // this process's mode, as RunAgent's gate chose it (agentHosting):
+ // the claim agent of an enrolled box with an armed guard. It is set
+ // before the worker starts and cleared when the worker ends, so a box
+ // whose claim worker has stopped never answers healthy.
+ hosting := agentHosting.Load()
+ lbhealth.Default.SetHosting(hosting)
+ go func() {
+ defer close(w.joined)
+ defer lbhealth.Default.SetHosting(false)
+ w.Providers.Regional.Loop(workerCtx, serial, report)
+ }()
return nil
}
func (w *ClaimWiring) close(ctx context.Context) error {
diff --git a/internal/host/claimwiring_hosting_test.go b/internal/host/claimwiring_hosting_test.go
index 0c296aaeb..cb5d95029 100644
--- a/internal/host/claimwiring_hosting_test.go
+++ b/internal/host/claimwiring_hosting_test.go
@@ -7,12 +7,16 @@ import (
"github.com/carlosframework/platform/internal/lbhealth"
)
-// Region HA spec 2 §5, §6: a claim agent that is running is enrolled, so
-// it is hosting; an agent with claim mode off never says so. Plan 2b's
-// joining mode is what leaves this false.
-func TestClaimWiringStartMarksTheBoxHosting(t *testing.T) {
+// Region HA spec 2 §5, §6: the health answer's hosting condition is the
+// mode RunAgent's gate chose (agentHosting), held only while the claim
+// worker runs. Claim mode off never says hosting, a joining process never
+// does, and a hosting box's answer turns 503 when its worker ends.
+func TestClaimWiringHostingFollowsTheModeAndTheWorker(t *testing.T) {
lbhealth.Default.SetHosting(false)
- t.Cleanup(func() { lbhealth.Default.SetHosting(false) })
+ was := agentHosting.Load()
+ t.Cleanup(func() { lbhealth.Default.SetHosting(false); agentHosting.Store(was) })
+
+ agentHosting.Store(true)
off := &ClaimWiring{}
if err := off.startRegional(t.Context(), nil, nil); err != nil {
t.Fatal(err)
@@ -20,16 +24,35 @@ func TestClaimWiringStartMarksTheBoxHosting(t *testing.T) {
if lbhealth.Default.Hosting() {
t.Fatal("claim mode off marked the box hosting")
}
- loop := testRegionalLoop{started: make(chan struct{})}
- w := &ClaimWiring{Writes: hostwrite.Config{Enabled: true}, Providers: ClaimProviders{Regional: loop}}
- if err := w.startRegional(t.Context(), nil, nil); err != nil {
+
+ start := func() *ClaimWiring {
+ loop := testRegionalLoop{started: make(chan struct{})}
+ w := &ClaimWiring{Writes: hostwrite.Config{Enabled: true}, Providers: ClaimProviders{Regional: loop}}
+ if err := w.startRegional(t.Context(), nil, nil); err != nil {
+ t.Fatal(err)
+ }
+ <-loop.started
+ return w
+ }
+
+ agentHosting.Store(false)
+ w := start()
+ if lbhealth.Default.Hosting() {
+ t.Fatal("a process that is not hosting marked the box hosting")
+ }
+ if err := w.close(t.Context()); err != nil {
t.Fatal(err)
}
- <-loop.started
+
+ agentHosting.Store(true)
+ w = start()
if !lbhealth.Default.Hosting() {
- t.Fatal("a running claim agent is not hosting")
+ t.Fatal("the claim agent of an enrolled box is not hosting")
}
if err := w.close(t.Context()); err != nil {
t.Fatal(err)
}
+ if lbhealth.Default.Hosting() {
+ t.Fatal("the box still says hosting after its claim worker ended")
+ }
}
diff --git a/internal/host/joining.go b/internal/host/joining.go
new file mode 100644
index 000000000..76176312a
--- /dev/null
+++ b/internal/host/joining.go
@@ -0,0 +1,296 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "flag"
+ "io"
+ "log"
+ "net/http"
+ "os"
+ "os/signal"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "syscall"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lbhealth"
+ "github.com/carlosframework/platform/internal/status"
+ "github.com/carlosframework/platform/internal/version"
+)
+
+// The agent's joining mode (region HA spec 2 §5 step 4) and the gate that
+// chooses it (joinmode.go).
+//
+// A joining box publishes its signed identity record (boxRecordLoop) and,
+// once its guard is armed, its guard-ready report: the private status
+// report every hosting box publishes, carrying the guard's facts
+// (claimReport). It never publishes an edges file: a box an edges file
+// names is referenced, and automatic enrollment refuses a referenced box
+// (hostactors references.go, enroll.go). It opens no registry, starts no
+// activator and serves no instance. Its only listener is the load
+// balancer's health check on the edge's HTTP address, which answers 503:
+// lbhealth.Default is never marked hosting here.
+//
+// When the box is invited it asks root to arm the guard by touching
+// joinRequestMarker (carlos-system-join.path). Once a start check finds it
+// enrolled and armed, it returns, and RunAgent ends the process so systemd
+// starts the agent again, now as the claim agent through the guard.
+
+// joinEvery is how often joining checks its mode.
+const joinEvery = 15 * time.Second
+
+// joinNudgeEvery is how often joining re-publishes its report or re-asks
+// root, at most: the report stays well inside reportFreshness, and root
+// is not asked again while it is still arming. While the box stays
+// invited with its guard not armed, root is asked again at this interval,
+// once a minute: carlos-system-join.path fires only when the marker
+// changes, so a carlos system join that failed (an IMDS or store read,
+// say) is retried by the next touch, and by nothing else.
+const joinNudgeEvery = time.Minute
+
+// joinRequestMarker is the file carlos-system-join.path watches, under the
+// data dir.
+const joinRequestMarker = "system-join-requested"
+
+// joining is one joining run's steps; production's are productionJoining.
+type joining struct {
+ facts modeFacts
+ every time.Duration
+ // serve answers the load balancer until ctx ends.
+ serve func(ctx context.Context) error
+ // record publishes the box's identity record, retrying until it is in
+ // place.
+ record func(ctx context.Context)
+ // report publishes the guard-ready report.
+ report func(ctx context.Context) error
+ // requestArm asks root to arm the guard.
+ requestArm func() error
+ logf func(format string, args ...any)
+}
+
+// run joins until the box can host (the decision is returned) or ctx
+// ends (the last decision, and ctx's error).
+func (j joining) run(ctx context.Context) (modeDecision, error) {
+ lbhealth.Default.SetHosting(false)
+ go func() {
+ if err := j.serve(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
+ j.logf("agent: joining: the load balancer health check is not answered: %v", err)
+ }
+ }()
+ go j.record(ctx)
+ var last modeDecision
+ var lastErr string
+ var nudged time.Time
+ for {
+ d, err := selectMode(ctx, j.facts)
+ now := j.facts.now()
+ switch {
+ case err != nil:
+ if err.Error() != lastErr {
+ j.logf("agent: joining: %v; checking again", err)
+ lastErr = err.Error()
+ }
+ case d.Mode == modeHosting:
+ j.logf("agent: joining: %s is enrolled and its guard is armed; restarting as the claim agent", d.Box)
+ return d, nil
+ default:
+ lastErr = ""
+ if d.Stage != last.Stage || d.Why != last.Why {
+ j.logf("agent: joining: %s: %s %s", d.Box, d.Stage, d.Why)
+ nudged = time.Time{}
+ }
+ last = d
+ if now.Sub(nudged) >= joinNudgeEvery {
+ switch d.Stage {
+ case stageInvited:
+ if err := j.requestArm(); err != nil {
+ j.logf("agent: joining: could not ask root to arm the guard: %v", err)
+ }
+ case stageGuardInstalled, stageGuardReady:
+ if err := j.report(ctx); err != nil {
+ j.logf("agent: joining: the guard-ready report was not published: %v", err)
+ }
+ }
+ nudged = now
+ }
+ }
+ select {
+ case <-ctx.Done():
+ return last, ctx.Err()
+ case <-time.After(j.every):
+ }
+ }
+}
+
+// productionJoining is joining on a real box. args are the agent's own
+// (edge.Run's flags): the health check listens on -http, as the edge will.
+func productionJoining(args []string, facts modeFacts, getenv func(string) string) joining {
+ addr := joiningHTTPAddr(args)
+ return joining{
+ facts: facts,
+ every: joinEvery,
+ serve: func(ctx context.Context) error {
+ srv := &http.Server{Addr: addr, ReadHeaderTimeout: 10 * time.Second, Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if !lbhealth.Serve(w, r, lbhealth.Default, time.Now()) {
+ http.NotFound(w, r)
+ }
+ })}
+ go func() { <-ctx.Done(); _ = srv.Close() }()
+ return srv.ListenAndServe()
+ },
+ record: func(context.Context) { boxRecordLoop() },
+ report: func(ctx context.Context) error {
+ store, _, err := openStatusStore(ctx, getenv)
+ if err != nil {
+ return err
+ }
+ p := status.Probes{MachineID: readMachineID(), InstanceID: cachedInstanceID(), Region: strings.TrimSpace(getenv("CARLOS_REGION")),
+ Release: version.Build, Platform: runtime.GOOS + "-" + runtime.GOARCH, SystemChannel: boxSystemChannel(getenv)}
+ claimReport(&p)
+ box := status.CollectBox(nil, p, time.Now())
+ if groupReferenced(getenv) {
+ // Under the group reference the console derives (census
+ // ruling region-ha-2b), as statusLoop publishes it.
+ box.Label = facts.box
+ }
+ return status.WriteBox(ctx, store, box)
+ },
+ requestArm: func() error {
+ return os.WriteFile(filepath.Join(dataDir(), joinRequestMarker), []byte(time.Now().UTC().Format(time.RFC3339Nano)+"\n"), 0o644)
+ },
+ logf: log.Printf,
+ }
+}
+
+// joiningHTTPAddr is the agent's -http flag, ":80" when it is not given.
+func joiningHTTPAddr(args []string) string {
+ fs := flag.NewFlagSet("agent", flag.ContinueOnError)
+ fs.SetOutput(io.Discard)
+ addr := fs.String("http", ":80", "")
+ fs.String("registry", "", "")
+ fs.String("acme-cache", "", "")
+ fs.String("https", "", "")
+ fs.String("fallback", "", "")
+ fs.Bool("dev", false, "")
+ _ = fs.Parse(args)
+ return *addr
+}
+
+// decideAgentMode is RunAgent's gate. With claim mode or region HA off
+// there is no joining mode, and the agent hosts as on main. Otherwise it
+// selects the mode, waiting out an unknown one, and on joining runs join
+// until the box can host or ctx ends; it reports whether this process
+// should go on to start the claim agent.
+func decideAgentMode(ctx context.Context, getenv func(string) string, open func(context.Context, func(string) string) (modeFacts, error), join func(context.Context, modeFacts, modeDecision) error, wait time.Duration, logf func(string, ...any)) bool {
+ if !hostwrite.Enabled(getenv) || !regionHAEnabled(getenv) {
+ return true
+ }
+ var last string
+ note := func(err error) bool {
+ if err.Error() != last {
+ logf("agent: mode: %v; checking again", err)
+ last = err.Error()
+ }
+ select {
+ case <-ctx.Done():
+ return false
+ case <-time.After(wait):
+ return true
+ }
+ }
+ for {
+ facts, err := open(ctx, getenv)
+ if err != nil {
+ if !note(modeUnknown("the deployment's stores and this box's guard", err)) {
+ return false
+ }
+ continue
+ }
+ d, err := selectMode(ctx, facts)
+ if err != nil {
+ if !note(err) {
+ return false
+ }
+ continue
+ }
+ if d.Mode == modeHosting {
+ return true
+ }
+ logf("agent: %s is joining (%s): it hosts nothing until it is enrolled", d.Box, d.Stage)
+ if err := join(ctx, facts, d); err != nil && !errors.Is(err, context.Canceled) {
+ logf("agent: joining: %v", err)
+ }
+ return false
+ }
+}
+
+// hostingGate decides, for RunAgent, whether this process starts the claim
+// agent, and records the answer in agentHosting. SIGTERM or SIGINT while
+// joining ends the process cleanly.
+func hostingGate(args []string) bool {
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)
+ defer stop()
+ hosting := decideAgentMode(ctx, os.Getenv, productionModeFacts, func(ctx context.Context, facts modeFacts, _ modeDecision) error {
+ _, err := productionJoining(args, facts, os.Getenv).run(ctx)
+ return err
+ }, joinEvery, log.Printf)
+ agentHosting.Store(hosting)
+ return hosting
+}
+
+// groupReferenceBackoff is how long agentEntry waits after the n-th
+// failed read of a group box's reference (n from 1): a second, doubling,
+// at most 30 seconds.
+func groupReferenceBackoff(n int) time.Duration {
+ d := time.Second
+ for ; n > 1 && d < 30*time.Second; n-- {
+ d *= 2
+ }
+ return min(d, 30*time.Second)
+}
+
+// agentEntry is RunAgent's order. A box its region's group launched (the
+// CARLOS_BOX_REFERENCE=group marker) first resolves its group reference,
+// whatever the claim and region HA switches say: metricsBoxLabel answers
+// "" until it resolves, and the loops runAgent starts (the edges file,
+// the registry snapshot, metrics, usage, the layout box, the moves pass)
+// read the label once and keep it for the process. It asks IMDS again
+// with backoff, logging each failure's cause, until it answers or the
+// process is told to stop. Any other box resolves nothing. Then the gate
+// (hostingGate) decides whether this process starts the agent (run).
+func agentEntry(getenv func(string) string, resolve func() (string, error), backoff func(int) time.Duration, logf func(string, ...any), gate func() bool, run func()) {
+ if groupReferenced(getenv) {
+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)
+ ok := awaitGroupReference(ctx, resolve, backoff, logf)
+ stop()
+ if !ok {
+ return
+ }
+ }
+ if !gate() {
+ return
+ }
+ run()
+}
+
+// awaitGroupReference asks resolve until it answers, waiting backoff(n)
+// after the n-th failure; false when ctx ends first.
+func awaitGroupReference(ctx context.Context, resolve func() (string, error), backoff func(int) time.Duration, logf func(string, ...any)) bool {
+ for n := 1; ; n++ {
+ ref, err := resolve()
+ if err == nil {
+ logf("agent: this box's group reference is %s", ref)
+ return true
+ }
+ wait := backoff(n)
+ logf("agent: this box's group reference could not be read from its identity document (attempt %d): %v; the agent starts once it is, asking again in %s", n, err, wait)
+ select {
+ case <-ctx.Done():
+ return false
+ case <-time.After(wait):
+ }
+ }
+}
diff --git a/internal/host/joining_test.go b/internal/host/joining_test.go
new file mode 100644
index 000000000..03c1df772
--- /dev/null
+++ b/internal/host/joining_test.go
@@ -0,0 +1,412 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostenv"
+ "github.com/carlosframework/platform/internal/hostfloor"
+ "github.com/carlosframework/platform/internal/lbhealth"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// stagedFacts walks a box through the joining stages, one per selectMode
+// call: no invitation, invited, guard installed, guard-ready reported,
+// then hosting.
+type stagedFacts struct {
+ mu sync.Mutex
+ calls int
+ steps []joinFacts
+}
+
+func (s *stagedFacts) facts() modeFacts {
+ pick := func() joinFacts {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ i := min(s.calls/5, len(s.steps)-1) // each stage for five checks
+ s.calls++
+ return s.steps[i]
+ }
+ cur := joinFacts{}
+ f := modeFacts{box: "ada-hiking", now: time.Now}
+ f.enrollment = func() (hostfloor.Enrollment, error) { cur = pick(); return cur.facts().enrollment() }
+ f.roster = func(ctx context.Context) (hostactors.Roster, error) { return cur.facts().roster(ctx) }
+ f.guard = func() (hostfloor.Running, error) { return cur.facts().guard() }
+ f.invited = func(ctx context.Context, b string) (bool, error) { return cur.facts().invited(ctx, b) }
+ f.report = func(ctx context.Context, b string) (status.Box, error) { return cur.facts().report(ctx, b) }
+ return f
+}
+
+func freePort(t *testing.T) string {
+ t.Helper()
+ l, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Skipf("no TCP listener in this sandbox: %v", err)
+ }
+ addr := l.Addr().String()
+ l.Close()
+ return addr
+}
+
+// Joining publishes no edges file and admits nothing: it answers the load
+// balancer 503 throughout, asks root to arm the guard once invited,
+// publishes its report once armed, and returns as soon as it can host.
+func TestJoiningServes503UntilItCanHostAndPublishesNoEdgesFile(t *testing.T) {
+ deploy, data := t.TempDir(), t.TempDir()
+ t.Setenv("CARLOS_DEPLOYMENT_DIR", deploy)
+ t.Setenv("CARLOS_DATA_DIR", data)
+ t.Cleanup(func() { lbhealth.Default.SetHosting(false) })
+ addr := freePort(t)
+ f := enrolledFacts()
+ f.report = freshReport(time.Now())
+ steps := &stagedFacts{steps: []joinFacts{{}, {invited: true}, guardInstalled(), func() joinFacts { g := guardInstalled(); g.report = freshReport(time.Now()); return g }(), f}}
+ j := productionJoining([]string{"-http", addr, "-https", ":0"}, steps.facts(), os.Getenv)
+ j.every = 20 * time.Millisecond
+ var mu sync.Mutex
+ var reports int
+ records := make(chan struct{}, 1)
+ j.record = func(context.Context) { records <- struct{}{} }
+ j.report = func(context.Context) error { mu.Lock(); reports++; mu.Unlock(); return nil }
+ j.logf = t.Logf
+ lbhealth.Default.SetHosting(true) // a stale flag from an earlier process state is cleared
+ done := make(chan modeDecision, 1)
+ go func() { d, _ := j.run(t.Context()); done <- d }()
+ // The health check answers 503, naming hosting, while joining.
+ var body string
+ for i := 0; i < 100; i++ {
+ resp, err := http.Get("http://" + addr + lbhealth.Path)
+ if err == nil {
+ b, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("joining box answered %d", resp.StatusCode)
+ }
+ body = string(b)
+ break
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ if !strings.Contains(body, lbhealth.CondHosting) {
+ t.Fatalf("health body %q", body)
+ }
+ if resp, err := http.Get("http://" + addr + "/"); err != nil || resp.StatusCode != http.StatusNotFound {
+ t.Fatalf("a joining box served something other than its health check: %v %v", resp, err)
+ }
+ select {
+ case d := <-done:
+ if d.Mode != modeHosting {
+ t.Fatalf("returned %+v", d)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("joining never returned")
+ }
+ <-records
+ if _, err := os.Stat(filepath.Join(data, joinRequestMarker)); err != nil {
+ t.Fatalf("root was never asked to arm the guard: %v", err)
+ }
+ mu.Lock()
+ if reports == 0 {
+ t.Fatal("the guard-ready report was never published")
+ }
+ mu.Unlock()
+ if lbhealth.Default.Hosting() {
+ t.Fatal("joining marked the box hosting")
+ }
+ err := filepath.WalkDir(deploy, func(p string, d fs.DirEntry, err error) error {
+ if err == nil && strings.Contains(filepath.ToSlash(p), "control/edges") {
+ t.Errorf("joining wrote %s", p)
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestDecideAgentMode(t *testing.T) {
+ on := func(k string) string {
+ switch k {
+ case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA":
+ return "1"
+ }
+ return ""
+ }
+ noHA := func(k string) string {
+ if k == "CARLOS_HOST_CLAIM" {
+ return "1"
+ }
+ return ""
+ }
+ never := func(context.Context, func(string) string) (modeFacts, error) {
+ t.Fatal("facts read with joining mode off")
+ return modeFacts{}, nil
+ }
+ noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil }
+ quiet := func(string, ...any) {}
+ ctx := context.Background()
+ for name, env := range map[string]func(string) string{"claim off": func(string) string { return "" }, "region HA off": noHA} {
+ if !decideAgentMode(ctx, env, never, noJoin, time.Millisecond, quiet) {
+ t.Fatalf("%s: not hosting", name)
+ }
+ }
+ open := func(f joinFacts) func(context.Context, func(string) string) (modeFacts, error) {
+ return func(context.Context, func(string) string) (modeFacts, error) { return f.facts(), nil }
+ }
+ if !decideAgentMode(ctx, on, open(enrolledFacts()), noJoin, time.Millisecond, quiet) {
+ t.Fatal("an enrolled, armed box does not host")
+ }
+ var joined modeDecision
+ join := func(_ context.Context, _ modeFacts, d modeDecision) error { joined = d; return nil }
+ if decideAgentMode(ctx, on, open(joinFacts{invited: true}), join, time.Millisecond, quiet) || joined.Stage != stageInvited {
+ t.Fatalf("an invited box: joined %+v", joined)
+ }
+ // Unknown, then known: the gate waits, asks again, and logs the
+ // reason once.
+ reads, logs := 0, 0
+ flaky := func(context.Context, func(string) string) (modeFacts, error) {
+ reads++
+ f := enrolledFacts()
+ if reads < 3 {
+ f.rosterErr = errors.New("connection reset")
+ }
+ return f.facts(), nil
+ }
+ if !decideAgentMode(ctx, on, flaky, noJoin, time.Millisecond, func(string, ...any) { logs++ }) || reads != 3 || logs != 1 {
+ t.Fatalf("flaky roster: %d reads, %d log lines", reads, logs)
+ }
+ // Stopped while unknown: no claim agent.
+ stopped, cancel := context.WithCancel(ctx)
+ cancel()
+ failing := func(context.Context, func(string) string) (modeFacts, error) {
+ return modeFacts{}, errors.New("no bucket")
+ }
+ if decideAgentMode(stopped, on, failing, noJoin, time.Hour, quiet) {
+ t.Fatal("a stopped gate started the claim agent")
+ }
+}
+
+// A failed carlos system join is retried (Task 7's review): the path unit
+// fires only when the marker changes, so joining touches it again once a
+// minute for as long as the box stays invited with its guard not armed,
+// and never more often. Once the guard is armed it stops asking.
+func TestJoiningAsksRootAgainEachMinuteWhileInvited(t *testing.T) {
+ t.Cleanup(func() { lbhealth.Default.SetHosting(false) })
+ var mu sync.Mutex
+ clock, checks := joinNow, 0
+ var touches []time.Time
+ f := joinFacts{invited: true}.facts()
+ armedAfter := 40 // checks; then the guard is armed and not yet enrolled
+ f.enrollment = func() (hostfloor.Enrollment, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ if checks >= armedAfter {
+ return guardInstalled().facts().enrollment()
+ }
+ return joinFacts{}.facts().enrollment()
+ }
+ f.roster = guardInstalled().facts().roster
+ f.guard = guardInstalled().facts().guard
+ f.now = func() time.Time {
+ mu.Lock()
+ defer mu.Unlock()
+ checks++
+ clock = clock.Add(20 * time.Second) // each check, 20 seconds on
+ return clock
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ reported := make(chan struct{}, 1)
+ j := joining{
+ facts: f,
+ every: time.Millisecond,
+ serve: func(ctx context.Context) error { <-ctx.Done(); return nil },
+ record: func(context.Context) {},
+ report: func(context.Context) error {
+ select {
+ case reported <- struct{}{}:
+ default:
+ }
+ return nil
+ },
+ requestArm: func() error {
+ mu.Lock()
+ touches = append(touches, clock)
+ mu.Unlock()
+ return nil
+ },
+ logf: t.Logf,
+ }
+ done := make(chan struct{})
+ go func() { defer close(done); _, _ = j.run(ctx) }()
+ select {
+ case <-reported:
+ case <-time.After(10 * time.Second):
+ t.Fatal("the guard was armed and joining never published its report")
+ }
+ // A few more checks with the guard armed: no more touches.
+ for {
+ mu.Lock()
+ n := checks
+ mu.Unlock()
+ if n >= armedAfter+10 {
+ break
+ }
+ time.Sleep(time.Millisecond)
+ }
+ cancel()
+ <-done
+ mu.Lock()
+ defer mu.Unlock()
+ // 40 checks 20 seconds apart while invited: about 13 minutes, so a
+ // touch at the first check and then one each minute.
+ if len(touches) < 12 || len(touches) > 14 {
+ t.Fatalf("%d touches over %d invited checks: %v", len(touches), armedAfter, touches)
+ }
+ for i := 1; i < len(touches); i++ {
+ if gap := touches[i].Sub(touches[i-1]); gap < joinNudgeEvery || gap >= joinNudgeEvery+20*time.Second {
+ t.Fatalf("touch %d came %s after the one before; want once a minute", i, gap)
+ }
+ }
+ if last := touches[len(touches)-1]; last.After(joinNow.Add(time.Duration(armedAfter) * 20 * time.Second)) {
+ t.Fatalf("root was asked again at %s, after the guard was armed", last)
+ }
+}
+
+// Any error opening the facts (an empty group reference, say) is unknown:
+// the gate waits and asks again, and neither joins nor hosts until a
+// read answers (Task 6's review).
+func TestDecideAgentModeWaitsOutAFactsError(t *testing.T) {
+ on := func(k string) string {
+ switch k {
+ case "CARLOS_HOST_CLAIM", "CARLOS_REGION_HA":
+ return "1"
+ }
+ return ""
+ }
+ for name, then := range map[string]joinFacts{"enrolled": enrolledFacts(), "invited": {invited: true}} {
+ t.Run(name, func(t *testing.T) {
+ reads, logs, joins := 0, 0, 0
+ open := func(context.Context, func(string) string) (modeFacts, error) {
+ reads++
+ if reads < 3 {
+ return modeFacts{}, errors.New("this box's group reference could not be read")
+ }
+ return then.facts(), nil
+ }
+ join := func(_ context.Context, _ modeFacts, d modeDecision) error {
+ if reads < 3 {
+ t.Fatalf("joined on read %d: %+v", reads, d)
+ }
+ joins++
+ return nil
+ }
+ hosting := decideAgentMode(context.Background(), on, open, join, time.Millisecond, func(string, ...any) { logs++ })
+ if reads != 3 || logs < 1 {
+ t.Fatalf("%d reads, %d log lines", reads, logs)
+ }
+ wantHosting, wantJoins := name == "enrolled", 1
+ if wantHosting {
+ wantJoins = 0
+ }
+ if hosting != wantHosting || joins != wantJoins {
+ t.Fatalf("hosting %t, %d joins; want %t, %d", hosting, joins, wantHosting, wantJoins)
+ }
+ })
+ }
+}
+
+// Task 6's review (R-A): a group box resolves its reference before the
+// gate and before runAgent arms any loop, whatever the switches say, since
+// those loops read metricsBoxLabel once and keep it. IMDS that fails twice
+// holds everything back until the third read; every failure is logged with
+// its cause. Any other box asks IMDS nothing and starts at once.
+func TestAGroupBoxResolvesItsReferenceBeforeTheAgentStarts(t *testing.T) {
+ quick := func(int) time.Duration { return time.Millisecond }
+ never := func(context.Context, func(string) string) (modeFacts, error) {
+ t.Fatal("facts read with joining mode off")
+ return modeFacts{}, nil
+ }
+ enrolled := func(context.Context, func(string) string) (modeFacts, error) { return enrolledFacts().facts(), nil }
+ noJoin := func(context.Context, modeFacts, modeDecision) error { t.Fatal("joined"); return nil }
+ for name, tc := range map[string]struct {
+ env map[string]string
+ facts func(context.Context, func(string) string) (modeFacts, error)
+ marked bool
+ }{
+ "group box, switches off": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference}, facts: never, marked: true},
+ "group box, switches on": {env: map[string]string{hostenv.BoxReference: hostenv.GroupReference, "CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1", "CARLOS_REGION": "eu-west-1"},
+ facts: enrolled, marked: true},
+ "other box, switches off": {env: map[string]string{}, facts: never},
+ "other box, switches on": {env: map[string]string{"CARLOS_HOST_CLAIM": "1", "CARLOS_REGION_HA": "1"}, facts: enrolled},
+ } {
+ t.Run(name, func(t *testing.T) {
+ withGroupIdentity(t, "", "", nil)
+ asked := 0
+ groupIdentity = func() (string, string, error) {
+ asked++
+ if asked < 3 {
+ return "", "", errors.New("IMDS answered 500")
+ }
+ return "123456789012", "i-0a61f580df1acfa65", nil
+ }
+ getenv := groupEnv(tc.env)
+ var lines []string
+ logf := func(format string, args ...any) { lines = append(lines, fmt.Sprintf(format, args...)) }
+ gatedAt, ranAt, label := -1, -1, ""
+ gate := func() bool {
+ gatedAt = asked
+ return decideAgentMode(context.Background(), getenv, tc.facts, noJoin, time.Millisecond, logf)
+ }
+ run := func() { ranAt = asked; label = metricsBoxLabel(getenv, "machine-1") }
+ agentEntry(getenv, resolveGroupBoxReference, quick, logf, gate, run)
+ want, ref := 0, status.Label("machine-1")
+ if tc.marked {
+ want, ref = 3, hostactors.GroupBoxReference("123456789012", "i-0a61f580df1acfa65")
+ }
+ if gatedAt != want || ranAt != want {
+ t.Fatalf("gate after %d IMDS reads, agent after %d; want %d", gatedAt, ranAt, want)
+ }
+ if label != ref {
+ t.Fatalf("the agent started with label %q, want %q", label, ref)
+ }
+ failures := 0
+ for _, l := range lines {
+ if strings.Contains(l, "IMDS answered 500") {
+ failures++
+ }
+ }
+ if tc.marked && failures != 2 {
+ t.Fatalf("logged %d failures with their cause, want 2: %q", failures, lines)
+ }
+ })
+ }
+}
+
+func TestAwaitGroupReferenceStopsWithItsContext(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ failing := func() (string, error) { return "", errors.New("IMDS answered 500") }
+ if awaitGroupReference(ctx, failing, func(int) time.Duration { return time.Hour }, func(string, ...any) {}) {
+ t.Fatal("a stopped wait reported the reference resolved")
+ }
+ var got []time.Duration
+ for n := 1; n <= 8; n++ {
+ got = append(got, groupReferenceBackoff(n))
+ }
+ want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, 30 * time.Second, 30 * time.Second, 30 * time.Second}
+ if fmt.Sprint(got) != fmt.Sprint(want) {
+ t.Fatalf("backoff %v, want %v", got, want)
+ }
+}
diff --git a/internal/provision/testdata/plan14-root-reads-region-ha-2b.tsv b/internal/provision/testdata/plan14-root-reads-region-ha-2b.tsv
index c77f5e7d0..b391766ee 100644
--- a/internal/provision/testdata/plan14-root-reads-region-ha-2b.tsv
+++ b/internal/provision/testdata/plan14-root-reads-region-ha-2b.tsv
@@ -5,12 +5,13 @@
# Task 5: the console's automatic enrollment pass (internal/hostactors autoenroll.go: AutoEnroller: an enrolling write the deadline left unknown is settled by the next pass's roster read; internal/console enrollpass.go, started in StartClaimRecoveryPass with region HA on, and awsmembership.go, its membership reader). One new root-read unit, AutoEnroller.inputs, whose two read sites are the signed roster (Directory.Read) and the host floor (hostfloor.Read), in the console. Ruled accepted-legacy (J4 root-input). Invite (only on the bound store, after a membership read), enrollmentNonce (the read-only invitation check), BeginEnrollmentAttempt and checkMembership, accepted-legacy, are called and not changed. Fix round 1: CheckConfirmed is unexported (enrollment_confirmation.go), the pass binds the invitation to the confirmation checkConfirmed returns, and an invited box is found with enrollmentNonce, never Invite. The seal moves.
# Task 6: mode selection and group references (internal/host joinmode.go: selectMode, productionModeFacts, relaunched and crossCheck: a group box stopped and started since its enrollment, or whose host.env disagrees with the roster, is held; boxreference.go, metrics_loop.go's metricsBoxLabel and boxrecord.go's boxRecordLoop: a group box takes its group reference; internal/status fleet.go: the status listing drops reports no box has rewritten in a day before the key cap). statusLoop, accepted-legacy, is re-sealed as ruled: its tick is groupStatusTick, which publishes a group box's report only under its group reference, resolved afresh each tick, and nothing until it resolves; every other box's tick is statusTick's, unchanged. No other root-read unit is added or changed. The seal moves.
# Task 7: root's half of joining (internal/host joinroot.go: carlos system join, which runs guard-arm in process; units carlos-system-join.path and .service; bootstrap's enabledUnits arms the path). joinroot.go adds no root-read unit: RunSystemGuardArm, authority F,J4, is called and not changed. The two unit files are new root inputs, ruled authority F,J4 on the condition that the path unit's file is only a trigger: root's join reads nothing from it. The legacy seal stands.
+# Task 8: joining mode and the gate (internal/host joining.go; RunAgent calls hostingGate before runAgent; ClaimWiring.startRegional sets the health answer's hosting condition from the gate's decision and clears it when the worker ends). runAgent, authority B,C,D,F,J4, is not changed. No root-read unit changed or was added. Envelope-only, and the seal stands. Before the gate, RunAgent has a box carrying the group marker resolve its group reference (agentEntry, through Task 6's resolveGroupBoxReference, with backoff), whatever the switches say, so runAgent's loops never keep an empty label; joining touches the join marker again once a minute while the box stays invited, so a failed carlos system join is retried.
@coverage . 3678cfe22582d35fe0d8e93d3fed898cdffa6fb2ce2d81e0252c8a8da77993b3
@coverage internal/hostenv 45d9d4b90a5d441020f5ba4573821ca92701246c80b32d6f1823665fb0f65208
internal/hostenv/cli.go runHostEnv authority A,J4 rev3=census-ruling-region-ha-2b;basis=carlos system host-env: write runs as root at a box's start-up and writes /etc/carlos/host.env, 0600, whole (temporary file and rename), from allowlisted parameters with safe values only, refusing a missing required value; its one read site is seed's --from file, an operator's host.env whose values go only to the parameter store; write's path is fixed and reached only through writeHostEnv file:ReadFile@1500 df4470176b229b311927ff0ac4f268d354c3a2357cac4300ac0992d2ebb90899
internal/hostenv/write_linux.go checkedDir authority A,J4 rev3=census-ruling-region-ha-2b;basis=carlos system host-env write (root, Linux): opens one directory under one already held, O_NOFOLLOW|O_DIRECTORY, creating it 0755 first only when asked and missing, and refuses it unless fstat shows a directory owned by this process's user (root, as write requires) that no group or other can write file:Fstat@576,file:Openat@168,file:Openat@372 012257ac0866f3385044785d8abae9c31e61134928c394e22a2aeaef39ec9f3c
internal/hostenv/write_linux.go writeHostEnv authority A,J4 rev3=census-ruling-region-ha-2b;basis=carlos system host-env write (root, Linux): writes /etc/carlos/host.env only through /etc and /etc/carlos held by descriptor and checked by checkedDir; the new file is created 0600, O_EXCL|O_NOFOLLOW, in that directory, synced, renamed over host.env within it, and the directory synced; its read sites open the root and the temporary file file:Open@68,file:Openat@509 1245fc0f819ba73fb45545ffdcf92d5d4941802175e73bac209ab4d4238adff1
-@coverage internal/host a4e90f7c06e670f4fb4103bc96eb83ac2c1d85d0f3d3428625992ec12e8f8ecb
+@coverage internal/host b9f6f09b3cb083f333097b893208d3fb82ccd9f537c94d18de31eda52f31f4ad
internal/host/installchannel.go channelInstall.run authority F,J4 rev3=census-ruling-region-ha-2b;basis=carlos system install-channel (root, at a fresh box's start-up): its one read site opens root's apply lock; the install goes through verifyPointer, hostfloor.CheckImage and stageCandidate into both protected slots, and refuses a box whose guard is armed or that already runs another release file:OpenFile@196 f0cdb6dd28e83da282c3b0d7bc2012dcd80db181a8cecbf1a2db6bcc41d1d5e3
internal/host/installchannel_linux.go slotState authority F,J4 rev3=census-ruling-region-ha-2b;basis=carlos system install-channel (root, Linux): opens one protected slot with O_NOFOLLOW (a symlink is never followed), fstats the descriptor and hashes from it, to compare with the sha256 of the signed, verified artifact; a slot holding anything else, a symlink or a non-regular file is refused and left as it is, and nothing in it is executed file:Open@66,file:Stat@388 bc533a29f92ae95d31123b551354ab9f8960576e5ffef1bd6a74a06df0b57a24
internal/host/installchannel_linux.go var openat2 authority F,J4 rev3=census-ruling-region-ha-2b;basis=carlos system install-channel (root, Linux): unix.Openat2 held in a variable so a test can make it fail; removeBeneath is its only user and has no path-based fallback when it fails file:Openat2@14 6081e86f758f4ae3948ec2d4f4c037888e1362f122a410c110e31290ba914816
diff --git a/region_ha_boundary_test.go b/region_ha_boundary_test.go
index 45153ac23..e5bab36b5 100644
--- a/region_ha_boundary_test.go
+++ b/region_ha_boundary_test.go
@@ -43,6 +43,7 @@ var regionHACoreFiles = []string{
"internal/host/boxreference.go",
"internal/status/fleet.go",
"internal/host/joinroot.go",
+ "internal/host/joining.go",
}
// regionHAAdapterFiles are spec 2's AWS adapters. Each is named for AWS so