amadan

carlos / platform Public

settings-4 task 5d: claimed wakes and probes record the build they hashed, before every exec

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 8bb475a00149c8bfe5cba841d939e276accabf23 parent 10336e1
14 files changed, +589 −26
  • internal/activator/activator.go +7 −0
  • internal/activator/beforeexec.go +18 −3
  • internal/activator/claim_snapshot.go +19 −4
  • internal/activator/claimrecord_linux_test.go +342 −0
  • internal/activator/claimwake.go +55 −1
  • internal/host/buildstate.go +10 −3
  • internal/host/buildstate_test.go +45 −0
  • internal/host/claimmoveaccess.go +1 −1
  • internal/host/claimowners.go +7 −2
  • internal/host/claimproduction.go +1 −1
  • internal/host/claimstaged_test.go +56 −0
  • internal/host/claimwakeassembly.go +9 −4
  • internal/host/movesprobeowners.go +2 −2
  • internal/release/staged.go +17 −5
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 7e1288a21..6dc7b13b6 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1496,7 +1496,14 @@ func (a *Service) wake(e *actEntry) {
// own.
e.gen = w.gen
if e.gen.N == 0 && !e.worker {
+ ran := e.gen
e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
+ if ran.StagedPath != "" && ran.Build == w.spawnVersion {
+ // A claimed start with no number names the file it ran
+ // (startAdmitted): the handle keeps it, so a legacy
+ // sidecar restart runs that file too (Task 5d).
+ e.gen.StagedPath, e.gen.Digest, e.gen.Demand = ran.StagedPath, ran.Digest, ran.Demand
+ }
}
if a.gens && e.worker && e.gen.N == 0 {
// A worker woken some way that carried no number (doWake gives
diff --git a/internal/activator/beforeexec.go b/internal/activator/beforeexec.go
index 30c2533b6..15879fdb3 100644
--- a/internal/activator/beforeexec.go
+++ b/internal/activator/beforeexec.go
@@ -35,9 +35,9 @@ var errBuildStateUnwritten = errors.New("the build state could not be written, s
//
// With host claims off the lease is the hibernate lease, so a context that
// names no location gets that one (Config.Bucket, HibernatePrefix), and the
-// agent writes beside its lock.json. A claimed wake attaches its own
-// (Task 5c); with claims on and none attached, the agent locates the host
-// through its claim.
+// agent writes beside its lock.json. A claimed start attaches its own
+// (startAdmitted); with claims on and none attached (a claimed instance's
+// legacy sidecar restart), the agent locates the host through its claim.
func (a *Service) recordBuild(ctx context.Context, host, build string) error {
if a.BeforeExec == nil {
return nil
@@ -59,6 +59,21 @@ func (a *Service) recordBuild(ctx context.Context, host, build string) error {
return nil
}
+// claimedBuild is the claimed path's version check before its record
+// (ruling R-EXEC-CLAIMVERSION). The record must name the bytes that run:
+// s.Staged is the file the snapshot hashed, and the launch verifies the
+// file it opens against that digest. s.Version is a separate read of the
+// live link, so a repoint between the hash and that read leaves the start
+// naming a build other than the one it would run, and it is refused. With
+// no hook nothing is recorded, and the start is today's.
+func (a *Service) claimedBuild(host string, s ClaimStart) error {
+ if a.BeforeExec == nil || s.Staged.Version == s.Version {
+ return nil
+ }
+ log.Printf("activator: %s not started: the executable it hashed (%s) is build %q, but the start names build %q (the live link moved between the two reads); the next wake reads it again", host, s.Staged.StagedPath, s.Staged.Version, s.Version)
+ return fmt.Errorf("%w: the executable hashed for this start is build %q, but the start names build %q", errBuildStateUnwritten, s.Staged.Version, s.Version)
+}
+
// settleWakeIdentity is doWake's step 2b (settings spec §4.2, Astra I1
// #5): the one file this wake runs, its seed included, chosen and recorded
// under the wake's lease after the restore and before anything execs.
diff --git a/internal/activator/claim_snapshot.go b/internal/activator/claim_snapshot.go
index 976b3ecbc..ee227a6ad 100644
--- a/internal/activator/claim_snapshot.go
+++ b/internal/activator/claim_snapshot.go
@@ -12,6 +12,7 @@ import (
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// ClaimLaunchSnapshot is read-only launch selection. The actual launch still
@@ -19,9 +20,14 @@ import (
type ClaimLaunchSnapshot struct {
Identity hostwrite.Identity
Executable, SHA256, Version string
- Args, Env []string
- Roles []string
- Limit cgroup.Limit
+ // Staged is the file SHA256 hashes, from one resolution of the live
+ // link (release.ResolveLiveLink): the version it is staged under, its path
+ // and that digest. Version is a separate read of the link; the claimed
+ // start refuses when the two disagree (ClaimStart.Staged).
+ Staged release.StagedIdentity
+ Args, Env []string
+ Roles []string
+ Limit cgroup.Limit
}
// claimLaunchRoute is the route shape the claim launch path serves.
@@ -85,7 +91,15 @@ func (b *ExecBackend) SnapshotClaimLaunch(ctx context.Context, rt registry.Route
return out, hostwrite.ErrUnavailable
}
out.Executable = b.liveLink(rt.Host)
- f, err := os.Open(out.Executable)
+ // One resolution of the live link names both the file hashed below and
+ // the version it is staged under (settings spec §4.2, ruling
+ // R-EXEC-CLAIMVERSION): opening the link itself would hash whatever it
+ // named at that moment, with no way to say which build that was.
+ out.Staged, err = release.ResolveLiveLink(b.OptDir, rt.Host)
+ if err != nil {
+ return out, err
+ }
+ f, err := os.Open(out.Staged.StagedPath)
if err != nil {
return out, err
}
@@ -104,6 +118,7 @@ func (b *ExecBackend) SnapshotClaimLaunch(ctx context.Context, rt registry.Route
return out, closeErr
}
out.SHA256 = hex.EncodeToString(digest.Sum(nil))
+ out.Staged.Digest = out.SHA256
out.Version, err = b.LiveVersion(rt.Host)
if err != nil || out.Version == "" {
return out, hostwrite.ErrUnavailable
diff --git a/internal/activator/claimrecord_linux_test.go b/internal/activator/claimrecord_linux_test.go
new file mode 100644
index 000000000..bab6806ac
--- /dev/null
+++ b/internal/activator/claimrecord_linux_test.go
@@ -0,0 +1,342 @@
+//go:build linux
+
+package activator
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The claimed (CARLOS_HOST_CLAIM) path records its build before every exec
+// (settings spec §4.2, settings plan 4 Task 5d). These tests launch through
+// the root launch fixture (newRootLaunchFixture), so startAdmitted's
+// beforeStart really runs inside RoleLaunch.Start, immediately before root's
+// Launch: f.root.launches is the exec count.
+
+// claimedBuildOf is the staged identity of the fixture's executable, as one
+// resolution of the live link names it.
+func claimedBuildOf(t *testing.T, f *rootLaunchFixture, ver string) release.StagedIdentity {
+ t.Helper()
+ raw, err := os.ReadFile(f.exe)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sum := sha256.Sum256(raw)
+ return release.StagedIdentity{Version: ver, StagedPath: f.exe, Digest: hex.EncodeToString(sum[:])}
+}
+
+// rootLaunched is s launching through f: the launcher, its attempt and its
+// command, the staged identity the snapshot hashed (v1) and the version it
+// read (v1).
+func rootLaunched(t *testing.T, f *rootLaunchFixture, s ClaimStart) ClaimStart {
+ t.Helper()
+ s.RoleLauncher, s.PrepareLaunch = &f.launch, nil
+ s.CurrentAttempt = func() hostclaim.WakeRetryAttempt { return f.attempt }
+ s.Build = f.build
+ s.Staged, s.Version = claimedBuildOf(t, f, "v1"), "v1"
+ if s.OpenProbe != nil {
+ // A probe's window opens at the start the journal was offered.
+ s.OpenProbe = func(_ context.Context, at time.Time) (time.Time, error) { return at, nil }
+ }
+ return s
+}
+
+// launchedChild grants a launch on a real child process (sleep), so a start
+// that root granted can be stopped and its exit watched; the child is
+// killed when the test ends.
+func launchedChild(t *testing.T) func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ return func(q hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ child := exec.Command("sleep", "30")
+ if err := child.Start(); err != nil {
+ return hostbox.RoleLaunchGrant{}, err
+ }
+ t.Cleanup(func() { _ = child.Process.Kill(); _ = child.Wait() })
+ start, err := roleProcessStart(child.Process.Pid)
+ if err != nil {
+ return hostbox.RoleLaunchGrant{}, err
+ }
+ return hostbox.RoleLaunchGrant{ReservationID: q.Reservation.ID, Unit: "carlos-instance@tenant.test.service", UnitInvocationID: "unit-1", PID: child.Process.Pid, ProcessStart: start,
+ Group: hostbox.LaunchGroup{RelativePath: q.Reservation.Group.RelativePath, Device: 3, Inode: 4, InvocationID: "unit-1"}}, nil
+ }
+}
+
+// claimedRecord is a claims-on Service whose Prepare hands startAdmitted a
+// start launching through the root fixture.
+type claimedRecord struct {
+ a *Service
+ f *rootLaunchFixture
+ obs hostclaim.Observed
+ log *execLog
+ mu sync.Mutex
+ reserved int
+ failed int
+ restored int
+}
+
+func newClaimedRecord(t *testing.T, mutate func(*ClaimStart)) *claimedRecord {
+ t.Helper()
+ f := newRootLaunchFixture(t)
+ c := &claimedRecord{f: f, obs: claimedObs("tenant.test")}
+ c.log = &execLog{spawns: func() int { f.root.mu.Lock(); defer f.root.mu.Unlock(); return len(f.root.launches) }}
+ work, _ := (hostwrite.Config{}).Select(t.Context(), c.obs.Host, "spawn")
+ c.a = &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Bucket: "data",
+ Read: func(context.Context, string) (hostclaim.Observed, error) { return c.obs, nil },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ s := rootLaunched(t, f, claimedStartFor(c.obs, work, "v1"))
+ s.Reserve = func(context.Context, int) error { c.mu.Lock(); c.reserved++; c.mu.Unlock(); return nil }
+ s.Failed = func(context.Context, int) error { c.mu.Lock(); c.failed++; c.mu.Unlock(); return nil }
+ s.Restore = func(context.Context) error { c.mu.Lock(); c.restored++; c.mu.Unlock(); return nil }
+ s.Replicate = func(context.Context, string) (Rep, error) { return &fakeRep{}, nil }
+ if mutate != nil {
+ mutate(&s)
+ }
+ return s, nil
+ }}}}
+ c.a.BeforeExec = c.log.hook
+ return c
+}
+
+func (c *claimedRecord) counts() (reserved, failed, restored, launches int) {
+ c.mu.Lock()
+ reserved, failed, restored = c.reserved, c.failed, c.restored
+ c.mu.Unlock()
+ c.f.root.mu.Lock()
+ defer c.f.root.mu.Unlock()
+ return reserved, failed, restored, len(c.f.root.launches)
+}
+
+func (c *claimedRecord) start(t *testing.T, e *actEntry, maxStarts int) error {
+ t.Helper()
+ return c.a.startAdmitted(t.Context(), e, c.obs, startPolicy{MaxStarts: maxStarts, Window: 30 * time.Second})
+}
+
+// One record of the claimed build, at the claim holder's Location in the
+// claim actor's bucket, before each exec. A start that failed for any other
+// reason is retried, and the retry is a new exec: it records again.
+func TestAClaimedStartRecordsBeforeEachExec(t *testing.T) {
+ c := newClaimedRecord(t, nil)
+ launches := 0
+ granted := launchedChild(t)
+ c.f.root.grant = func(q hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ if launches++; launches == 1 {
+ return hostbox.RoleLaunchGrant{}, errors.New("connection reset")
+ }
+ return granted(q)
+ }
+ e := &actEntry{host: c.obs.Host}
+ if err := c.start(t, e, 2); err != nil {
+ t.Fatalf("startAdmitted = %v", err)
+ }
+ calls, at, locs := c.log.seen()
+ if !slices.Equal(calls, []string{"tenant.test@v1", "tenant.test@v1"}) || !slices.Equal(at, []int{0, 1}) {
+ t.Fatalf("records %v at exec counts %v, want v1 before each of the two execs", calls, at)
+ }
+ want := BuildStateLocation{Bucket: "data", Prefix: c.obs.Claim.Holder.Location}
+ for _, l := range locs {
+ if l != want {
+ t.Fatalf("recorded at %+v, want beside the claimed lease %+v (never a g-<gen>/ prefix)", l, want)
+ }
+ }
+ if _, failed, _, n := c.counts(); failed != 1 || n != 2 {
+ t.Fatalf("failed=%d launches=%d, want the first start failed and retried", failed, n)
+ }
+ // The identity it ran rides on the generation, so a legacy sidecar
+ // restart runs the same file (Task 5d, C).
+ g := e.claimResult.gen
+ id := claimedBuildOf(t, c.f, "v1")
+ if g.Build != "v1" || g.StagedPath != id.StagedPath || g.Digest != id.Digest {
+ t.Fatalf("generation %+v, want the hashed v1 by path and digest", g)
+ }
+}
+
+// A refused record is no failed attempt of the build (ruling
+// R-EXEC-CLAIMRETRY): nothing execs, the claim's WakeRetry is not written
+// (s.Failed), and the loop does not retry; the wake fails as fenced.
+func TestAClaimedRecordRefusalIsTerminal(t *testing.T) {
+ c := newClaimedRecord(t, nil)
+ c.log.fail = failAll
+ began := time.Now()
+ err := c.start(t, &actEntry{host: c.obs.Host}, 5)
+ if !errors.Is(err, errBuildStateUnwritten) || errors.Is(err, errClaimsAdmission) {
+ t.Fatalf("startAdmitted = %v, want the record's refusal, not the admission's", err)
+ }
+ if got := memberSafeWakeReason(err); got != memberSafeWakeReason(ErrFenced) {
+ t.Fatalf("member reads %q, want the fence's wording", got)
+ }
+ reserved, failed, _, launches := c.counts()
+ if launches != 0 || failed != 0 || reserved != 1 {
+ t.Fatalf("launches=%d failed=%d reserved=%d, want one attempt, nothing execed, no failed attempt", launches, failed, reserved)
+ }
+ if calls, _, _ := c.log.seen(); len(calls) != 1 {
+ t.Fatalf("records %v, want one refused record and no retry", calls)
+ }
+ if took := time.Since(began); took >= time.Second {
+ t.Fatalf("took %v: the refusal waited out a retry backoff", took)
+ }
+}
+
+// Ruling R-EXEC-CLAIMVERSION: the record names the version of the bytes the
+// snapshot hashed. A start whose hashed path names v1 while its version
+// read says v2 (a repoint between the hash and the read) is refused before
+// anything touches the database or execs, and says why.
+func TestAClaimedVersionMismatchIsRefused(t *testing.T) {
+ c := newClaimedRecord(t, func(s *ClaimStart) { s.Version = "v2" })
+ err := c.start(t, &actEntry{host: c.obs.Host}, 5)
+ if !errors.Is(err, errBuildStateUnwritten) || !strings.Contains(err.Error(), "v1") || !strings.Contains(err.Error(), "v2") {
+ t.Fatalf("startAdmitted = %v, want a refusal naming v1 and v2", err)
+ }
+ if _, failed, restored, launches := c.counts(); launches != 0 || failed != 0 || restored != 0 {
+ t.Fatalf("launches=%d failed=%d restored=%d, want nothing run", launches, failed, restored)
+ }
+ if calls, _, _ := c.log.seen(); len(calls) != 0 {
+ t.Fatalf("records %v, want none", calls)
+ }
+ // With no hook (a box with no replica bucket) the start is today's.
+ c = newClaimedRecord(t, func(s *ClaimStart) { s.Version = "v2" })
+ c.a.BeforeExec = nil
+ c.f.root.grant = func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ return hostbox.RoleLaunchGrant{}, ErrAttemptOver
+ }
+ if err := c.start(t, &actEntry{host: c.obs.Host}, 1); !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("no hook: startAdmitted = %v, want today's launch", err)
+ }
+}
+
+// Ruling R-EXEC-PROBES: a moves probe runs a build against the moved
+// database through the same beforeStart, so it records too, beside the
+// holder's lease. Both entries: ProbeWake (a landing) and RecoveryProbe.
+func TestAClaimedProbeRecords(t *testing.T) {
+ refused := func(hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ return hostbox.RoleLaunchGrant{}, ErrAttemptOver
+ }
+ t.Run("ProbeWake", func(t *testing.T) {
+ r := &pwRuntime{observed: pwFlip()}
+ a, _ := pwService(t, r)
+ f := newRootLaunchFixture(t)
+ f.launch.Target.Host = pwHost
+ f.root.grant = refused
+ a.cfg.Claims.Bucket = "data"
+ prepare := a.cfg.Claims.Prepare
+ a.cfg.Claims.Prepare = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p startPolicy) (ClaimStart, error) {
+ s, err := prepare(ctx, rt, o, p)
+ return rootLaunched(t, f, s), err
+ }
+ l := &execLog{spawns: func() int { f.root.mu.Lock(); defer f.root.mu.Unlock(); return len(f.root.launches) }}
+ a.BeforeExec = l.hook
+ if _, err := a.ProbeWake(t.Context(), pwConfig()); !errors.Is(err, ErrProbeFailed) || !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("err = %v", err)
+ }
+ calls, at, locs := l.seen()
+ want := BuildStateLocation{Bucket: "data", Prefix: r.admitted().Claim.Holder.Location}
+ if !slices.Equal(calls, []string{pwHost + "@v1"}) || at[0] != 0 || locs[0] != want {
+ t.Fatalf("records %v at exec counts %v, %+v; want v1 before the exec at %+v", calls, at, locs, want)
+ }
+ })
+ t.Run("RecoveryProbe", func(t *testing.T) {
+ obs := probeObserved(probeClaim(hostclaim.KindRestoring, hostclaim.PhaseProbing, "box"))
+ a, r, _ := probeService(t, obs)
+ f := newRootLaunchFixture(t)
+ f.launch.Target.Host = probeHost
+ f.root.grant = refused
+ r.Bucket = "data"
+ prepare := r.Prepare
+ r.Prepare = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p startPolicy) (ClaimStart, error) {
+ s, err := prepare(ctx, rt, o, p)
+ return rootLaunched(t, f, s), err
+ }
+ l := &execLog{spawns: func() int { f.root.mu.Lock(); defer f.root.mu.Unlock(); return len(f.root.launches) }}
+ a.BeforeExec = l.hook
+ if err := a.RecoveryProbe(t.Context(), registry.Route{Host: probeHost}, probePolicy(probeInputs())); !errors.Is(err, ErrAttemptOver) {
+ t.Fatalf("err = %v", err)
+ }
+ calls, at, locs := l.seen()
+ want := BuildStateLocation{Bucket: "data", Prefix: obs.Claim.Holder.Location}
+ if !slices.Equal(calls, []string{probeHost + "@v1"}) || at[0] != 0 || locs[0] != want {
+ t.Fatalf("records %v at exec counts %v, %+v; want v1 before the exec at %+v", calls, at, locs, want)
+ }
+ })
+}
+
+// A claimed instance's legacy sidecar restart (Task 5d, C) runs the file its
+// primary ran, by path, and records that build first. It names no location
+// on its context: with claims on the agent locates the host through its
+// claim's Holder.Location (internal/host buildStates.locate, ruling
+// R-CLAIMPREFIX-2), which is the claimed lease's.
+func TestAClaimedSidecarRestartRecordsTheBuildItsPrimaryRan(t *testing.T) {
+ f := newRootLaunchFixture(t)
+ f.root.grant = launchedChild(t)
+ host := f.launch.Target.Host
+ obs := claimedObs(host)
+ obs.Claim.Holder.WriterBox = "box"
+ admitted := probeAdmitted(obs)
+ work, _ := (hostwrite.Config{}).Select(t.Context(), host, "spawn")
+ var mu sync.Mutex
+ reads := 0
+ b := newPlanGenBackend()
+ b.sidecarStaged = true
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := New(Config{Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", Bucket: "data",
+ Read: func(context.Context, string) (hostclaim.Observed, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ if reads++; reads > 1 {
+ return admitted, nil
+ }
+ return obs, nil
+ },
+ StopForeign: func(context.Context, string) error { return nil },
+ Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil },
+ Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
+ return admitted, nil
+ },
+ FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil },
+ Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
+ s := rootLaunched(t, f, claimedStartFor(admitted, work, "v1"))
+ s.Replicate = func(context.Context, string) (Rep, error) { return &fakeRep{}, nil }
+ return s, nil
+ }}}, b, reg)
+ l := &execLog{}
+ a.BeforeExec = l.hook
+ rt := registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: filepath.Join(f.runDir, "app.sock"), DBPath: filepath.Join(f.launchDir, "app.db")}
+ done, err := a.beginRequest(t.Context(), rt)
+ if err != nil {
+ t.Fatalf("the claimed wake: %v", err)
+ }
+ done()
+ g, ok := a.entryGen(host)
+ id := claimedBuildOf(t, f, "v1")
+ if !ok || g.Build != "v1" || g.StagedPath != id.StagedPath || g.Digest != id.Digest {
+ t.Fatalf("serving %+v (%v), want the hashed v1 by path and digest", g, ok)
+ }
+ a.sweepOnce()
+ waitFor(t, "the sidecar start", func() bool { return len(b.sidecarsOn()) == 1 && a.sidecarSettled(host) })
+ if sc := b.sidecarsOn(); sc[0].StagedPath != id.StagedPath || sc[0].Digest != id.Digest || sc[0].Build != "v1" || b.sidecarStartCount() != 0 {
+ t.Fatalf("sidecar on %+v (%d live-link starts), want the primary's v1 by path", sc[0], b.sidecarStartCount())
+ }
+ calls, _, locs := l.seen()
+ if !slices.Equal(calls, []string{host + "@v1", host + "@v1"}) {
+ t.Fatalf("records %v, want the primary's v1, then the sidecar's v1", calls)
+ }
+ if locs[0] != (BuildStateLocation{Bucket: "data", Prefix: admitted.Claim.Holder.Location}) || locs[1] != (BuildStateLocation{}) {
+ t.Fatalf("locations %+v: want the primary's at the claimed lease, the sidecar's left to the claim read", locs)
+ }
+}
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index f7c439c71..7a30b7b7e 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -18,6 +18,7 @@ import (
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
var ErrAttemptOver = moves.ErrAttemptOver
@@ -44,7 +45,9 @@ type StartPolicy struct {
// Fallback is a first wake's good build (zero-downtime spec §8.2), set
// only beside Pin, by plan 5's planner. It is honoured once the isolated
// launch opens; on this base the launch refuses before anything runs
- // (stepStale), so there is nothing to fall back from.
+ // (stepStale), so there is nothing to fall back from. Any fallback arm
+ // added later must record its build first (recordBuild, settings spec
+ // §4.2; ruling R-EXEC-CLAIMFALLBACK).
Fallback *Gen
// HealthBy, when set, caps the pinned start's Ready window: a first
// wake's health check ends at the earlier of Window and HealthBy (spec
@@ -120,6 +123,14 @@ type ClaimStart struct {
// Retain owns uncertain process or replica cleanup until it can be proved.
Retain func(Proc, Rep, error)
Version string
+ // Staged is the file the launch hashed, from one resolution of the
+ // host's live link (SnapshotClaimLaunch): the version it is staged
+ // under, its path and its digest (ExecutableSHA256). Version is a
+ // separate read of the link, so a repoint between the two can make them
+ // disagree; startAdmitted records Staged.Version, and refuses a start
+ // whose Staged.Version is not Version (settings spec §4.2, ruling
+ // R-EXEC-CLAIMVERSION).
+ Staged release.StagedIdentity
// Gen is what the provider bound. startAdmitted refuses a start whose
// number, socket or control socket is not StartPolicy.Gen's.
Gen Gen
@@ -370,6 +381,12 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
// refused before Restore touches the database.
return hostwrite.ErrUnavailable
}
+ if err := a.claimedBuild(e.host, s); err != nil {
+ // The bytes the snapshot hashed are not the build the start names:
+ // nothing could record the build that would run. Refused before
+ // Restore touches the database, and no failed attempt of the build.
+ return err
+ }
if policy.Gen.N != 0 && !genMatches(policy.Gen, s.Gen) {
// The provider bound something other than the generation the
// activator chose: its handle would name a socket nobody serves.
@@ -429,6 +446,7 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
// This hook runs after strict preparation and immediately before Start.
// Restore and materialization consume only the separate cut deadline.
reached := false // the exec hook ran: an error before it is the admission's (claimAdmission)
+ unrecorded := false
beforeStart := func() error {
reached = true
if policy.Gate != nil {
@@ -445,6 +463,19 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
return err
}
}
+ // The build state names the build before it runs against the
+ // database (settings spec §4.2), beside the claimed lease's
+ // lock.json: the claim actor's bucket at Holder.Location, never
+ // the session's g-<gen>/ prefix (ruling R-CLAIMPREFIX). The file
+ // is already open and verified against its digest here
+ // (RoleLaunch.Start). Every exec records, a retry's included;
+ // the same build makes the write idempotent. Moves probes come
+ // through here too (ruling R-EXEC-PROBES).
+ at := WithBuildStateLocation(ctx, BuildStateLocation{Bucket: r.Bucket, Prefix: admitted.Claim.Holder.Location})
+ if err := a.recordBuild(at, e.host, s.Staged.Version); err != nil {
+ unrecorded = true
+ return err
+ }
if err := budget.start(ctx, s.OpenProbe); err != nil {
return err
}
@@ -464,6 +495,21 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
p := newClaimProcessFor(cmd, prepared, s.RoleLauncher.Root != nil)
p.settle = settled
p.bindProducer(producer)
+ if unrecorded {
+ // A refused record ran nothing of the build (ruling
+ // R-EXEC-CLAIMRETRY): no failed attempt is written (s.Failed),
+ // and the loop does not retry it. The wake fails as fenced
+ // (memberSafeWakeReason).
+ cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second)
+ cleanErr := p.Stop(cleanup)
+ cancelCleanup()
+ if cleanErr != nil {
+ s.Retain(p, rep, errors.Join(startErr, cleanErr))
+ retained = true
+ return errors.Join(startErr, cleanErr)
+ }
+ return startErr
+ }
if startErr == nil {
deadline := time.Now().Add(policy.Window)
if policy.Probe {
@@ -516,6 +562,14 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
g.StagedPath, g.Digest, g.Demand = policy.Pin.StagedPath, policy.Pin.Digest, policy.Pin.Demand
}
}
+ if policy.Pin == nil && s.Staged.StagedPath != "" && s.Staged.Digest != "" && s.Staged.Version == s.Version {
+ // The file this start ran (the hashed resolution of the
+ // live link), so a legacy sidecar restart runs that file
+ // by path, never whatever the link names by then (Task
+ // 5d; ruling R-EXEC-SIDECAR). With no number, wake()
+ // keeps it on the canonical handle it gives this start.
+ g.Build, g.StagedPath, g.Digest = s.Version, s.Staged.StagedPath, s.Staged.Digest
+ }
e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, gen: g, stopOutcome: &stopObservation{}}
return nil
}
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go
index f81ff3453..d3e00e8d1 100644
--- a/internal/host/buildstate.go
+++ b/internal/host/buildstate.go
@@ -248,11 +248,18 @@ func (b *buildStates) refresh(ctx context.Context, host string) error {
// Hibernate flag), exec-backed and not tunnel-served. Never a named unit or
// the platform's own row. Settings apply only to Kind instance; that is
// root's concern, not this store's.
-func keepsBuildState(rt registry.Route) bool {
+//
+// With host claims on (claims) a worker never leases: the claim runtime
+// refuses sidecar rows before any lease (claimwakeassembly.go
+// Prerequisites), so it has no build state and no claim to locate one by.
+func keepsBuildState(rt registry.Route, claims bool) bool {
if rt.Unit != "" || rt.Platform {
return false
}
- return activator.ActivatorManaged(rt) || (rt.IsSidecar() && rt.Backing == "" && rt.Tunnel == "")
+ if rt.IsSidecar() {
+ return !claims && rt.Backing == "" && rt.Tunnel == ""
+ }
+ return activator.ActivatorManaged(rt)
}
// refreshBudget bounds one refreshAll: it runs inline in the reconcile legs,
@@ -278,7 +285,7 @@ func (b *buildStates) refreshAll(ctx context.Context) int {
}
var hosts []string
for _, rt := range routes {
- if keepsBuildState(rt) {
+ if keepsBuildState(rt, b.claims != nil) {
hosts = append(hosts, rt.Host)
}
}
diff --git a/internal/host/buildstate_test.go b/internal/host/buildstate_test.go
index 53e52edc6..adce25a47 100644
--- a/internal/host/buildstate_test.go
+++ b/internal/host/buildstate_test.go
@@ -213,6 +213,51 @@ func TestRefreshAllCoversExecInstancesAndWorkers(t *testing.T) {
}
}
+// With host claims on (settings plan 4 Task 5d, carried from Task 4) a
+// worker row never leases: the claim runtime refuses sidecar rows before
+// any lease (claimwakeassembly.go Prerequisites). refreshAll skips them, so
+// no claim is read for them and no copy is written; the claimed instance is
+// still refreshed through its claim.
+func TestRefreshAllSkipsWorkerRowsWithClaimsOn(t *testing.T) {
+ b, hib, reg := testBuildStates(t)
+ data := release.NewDirStore(t.TempDir())
+ b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": data})
+ b.claimBucket = "data"
+ var read []string
+ b.claims = func(_ context.Context, host string) (hostclaim.Observed, error) {
+ read = append(read, host)
+ return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host, Generation: "7", Session: 3}}}, nil
+ }
+ for _, rt := range []registry.Route{
+ {Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
+ {Host: "always.example.com", Kind: "sidecar", App: "km"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if skipped := b.refreshAll(context.Background()); skipped != 0 {
+ t.Fatalf("skipped %d", skipped)
+ }
+ if c := copyOf(t, reg); !c.Known || c.ReadErr != "" {
+ t.Fatalf("the claimed instance was not refreshed: %+v", c)
+ }
+ if len(read) != 1 || read[0] != "a.example.com" {
+ t.Fatalf("claims read for %v, want the instance alone", read)
+ }
+ for _, h := range []string{"w.example.com", "always.example.com"} {
+ if rt, err := reg.Get(h); err != nil || rt == nil || rt.BuildState != "" {
+ t.Fatalf("%s: a worker row got a build state copy with claims on: %+v %v", h, rt, err)
+ }
+ }
+ // With claims off the same rows are refreshed (ruling R-WORKER-ROWS).
+ b.claims = nil
+ b.refreshAll(context.Background())
+ if rt, _ := reg.Get("w.example.com"); rt.BuildState == "" {
+ t.Fatal("claims off: the worker row was not refreshed")
+ }
+}
+
// Ruling R-BS-REFRESH: refreshAll stops at its budget; the hosts it did not
// reach keep their last copy and are counted.
func TestRefreshAllStopsAtItsBudget(t *testing.T) {
diff --git a/internal/host/claimmoveaccess.go b/internal/host/claimmoveaccess.go
index 1177759a2..4bc41a314 100644
--- a/internal/host/claimmoveaccess.go
+++ b/internal/host/claimmoveaccess.go
@@ -182,7 +182,7 @@ func claimMoveLaunchSource(backend *activator.ExecBackend) func(context.Context,
}
digest := sha256.Sum256(raw)
snapshot.Identity.Config = hex.EncodeToString(digest[:])
- return ClaimServingLaunch{Identity: snapshot.Identity, Executable: snapshot.Executable, ExecutableSHA256: snapshot.SHA256, Args: snapshot.Args, Env: entries, Version: snapshot.Version, Limit: snapshot.Limit}, moves.ReportEnv(key, env), nil
+ return ClaimServingLaunch{Identity: snapshot.Identity, Executable: snapshot.Executable, ExecutableSHA256: snapshot.SHA256, Args: snapshot.Args, Env: entries, Version: snapshot.Version, Staged: snapshot.Staged, Limit: snapshot.Limit}, moves.ReportEnv(key, env), nil
}
}
diff --git a/internal/host/claimowners.go b/internal/host/claimowners.go
index 81f797b73..08ba77a50 100644
--- a/internal/host/claimowners.go
+++ b/internal/host/claimowners.go
@@ -19,6 +19,7 @@ import (
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// ClaimKindOwners is one family of op kinds' input, admission and
@@ -104,7 +105,11 @@ type ClaimServingLaunch struct {
ExecutableSHA256 string
Args, Env []string
Version string
- Limit cgroup.Limit
+ // Staged is the file ExecutableSHA256 hashes and the version it is
+ // staged under, from one resolution of the live link
+ // (activator.ClaimLaunchSnapshot.Staged).
+ Staged release.StagedIdentity
+ Limit cgroup.Limit
// Roles are the processes the real start spawns: "main", plus "seed"
// and "sidecar" when the route carries them. Nil means main alone.
Roles []string
@@ -396,7 +401,7 @@ func (s *claimServingOwners) Select(ctx context.Context, rt registry.Route, o ho
if len(roles) == 0 {
roles = []string{"main"}
}
- in := ClaimLaunchInputs{Roles: roles, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...), DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit,
+ in := ClaimLaunchInputs{Roles: roles, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...), DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit,
Admission: hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: s.Box, Mode: "cold", NewGeneration: generation, Region: s.Region}}
if claimEmptyBootstrap(h) {
in.Empty = true
diff --git a/internal/host/claimproduction.go b/internal/host/claimproduction.go
index 9f903b913..0aa27de03 100644
--- a/internal/host/claimproduction.go
+++ b/internal/host/claimproduction.go
@@ -449,6 +449,6 @@ func claimServingLaunchSource(backend *activator.ExecBackend, tokens claimAdminT
return ClaimServingLaunch{}, err
}
snapshot, err := backend.SnapshotClaimLaunch(ctx, rt)
- return ClaimServingLaunch{Identity: snapshot.Identity, Executable: snapshot.Executable, ExecutableSHA256: snapshot.SHA256, Args: snapshot.Args, Env: snapshot.Env, Roles: snapshot.Roles, Version: snapshot.Version, Limit: snapshot.Limit}, err
+ return ClaimServingLaunch{Identity: snapshot.Identity, Executable: snapshot.Executable, ExecutableSHA256: snapshot.SHA256, Args: snapshot.Args, Env: snapshot.Env, Roles: snapshot.Roles, Version: snapshot.Version, Staged: snapshot.Staged, Limit: snapshot.Limit}, err
}
}
diff --git a/internal/host/claimstaged_test.go b/internal/host/claimstaged_test.go
new file mode 100644
index 000000000..250d55bf5
--- /dev/null
+++ b/internal/host/claimstaged_test.go
@@ -0,0 +1,56 @@
+package host
+
+import (
+ "encoding/base64"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Settings plan 4 Task 5d (ruling R-EXEC-CLAIMVERSION): the claimed launch
+// names the file it hashed by one resolution of the live link, with the
+// version that file is staged under, so the record before the exec names
+// the bytes that run. Executable stays the live link (the launch digest and
+// root's binding are unchanged), and Version is still its own read.
+func TestTheClaimedLaunchNamesTheBuildItHashed(t *testing.T) {
+ f := newAdminTokenFixture(t, "box-1")
+ bin := filepath.Join(f.backend.OptDir, "versions", "v1", "bab", "jam")
+ sum, err := release.FileSHA256(bin)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := release.StagedIdentity{Version: "v1", StagedPath: bin, Digest: sum}
+ l, err := claimServingLaunchSource(f.backend, f.tokens)(t.Context(), f.rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if l.Staged != want || l.ExecutableSHA256 != sum || l.Version != "v1" || l.Executable != filepath.Join(f.backend.OptDir, "live", adminTokenHost) {
+ t.Fatalf("serving launch of %s (sha %s, version %q) staged %+v, want the live link hashed as %+v", l.Executable, l.ExecutableSHA256, l.Version, l.Staged, want)
+ }
+ t.Setenv("CARLOS_SECRETS_KEY", base64.StdEncoding.EncodeToString(make([]byte, 32)))
+ m, _, err := claimMoveLaunchSource(f.backend)(t.Context(), f.rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if m.Staged != want || m.ExecutableSHA256 != sum {
+ t.Fatalf("moves launch staged %+v, want %+v", m.Staged, want)
+ }
+ // A live link outside the version store hashes a file no build names.
+ other := filepath.Join(t.TempDir(), "app")
+ if err := os.WriteFile(other, []byte("#!/bin/sh\n"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ link := filepath.Join(f.backend.OptDir, "live", adminTokenHost)
+ if err := os.Remove(link); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(other, link); err != nil {
+ t.Fatal(err)
+ }
+ snap, err := f.backend.SnapshotClaimLaunch(t.Context(), f.rt)
+ if err == nil {
+ t.Fatalf("an unversioned live link was snapshotted: %+v", snap)
+ }
+}
diff --git a/internal/host/claimwakeassembly.go b/internal/host/claimwakeassembly.go
index 6c64291ce..af61adb58 100644
--- a/internal/host/claimwakeassembly.go
+++ b/internal/host/claimwakeassembly.go
@@ -20,6 +20,7 @@ import (
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/stopevidence"
)
@@ -34,9 +35,13 @@ type ClaimLaunchInputs struct {
ExecutableSHA256 string
Args, Env []string
DBPath, Version string
- Limit cgroup.Limit
- Empty bool
- Prepared bool
+ // Staged is the file ExecutableSHA256 hashes and the version it is
+ // staged under (ClaimServingLaunch.Staged); Prepare hands it to the
+ // start, which records it before the exec (settings spec §4.2).
+ Staged release.StagedIdentity
+ Limit cgroup.Limit
+ Empty bool
+ Prepared bool
// ReadyPath is the probe's declared readiness path (moves §7.7,
// moves.ready_path). Serving and committed wakes ignore it.
ReadyPath string
@@ -353,7 +358,7 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
_, in := a.snapshot()
selected = claimLeaseWork{Work: selected, Alive: a.ctx, Check: a.leaseErr}
restore := hostbox.ColdRestore{Reader: writer.Data, Bucket: writer.Bucket, Generation: o.Claim.Holder.Generation, Work: selected, Identity: in.Identity, RetireArtifacts: func(ctx context.Context, g string) error { return w.RetireArtifacts(ctx, o, in, g) }}
- s := activator.ClaimStart{Live: a.ctx, Work: selected, Identity: in.Identity, Manager: w.Manager, Host: rt.Host, Limit: in.Limit, Version: in.Version}
+ s := activator.ClaimStart{Live: a.ctx, Work: selected, Identity: in.Identity, Manager: w.Manager, Host: rt.Host, Limit: in.Limit, Version: in.Version, Staged: in.Staged}
s.Restore = func(call context.Context) error {
if err := a.admitsWriters(); err != nil {
return err
diff --git a/internal/host/movesprobeowners.go b/internal/host/movesprobeowners.go
index 3ea0cf25e..916ec8212 100644
--- a/internal/host/movesprobeowners.go
+++ b/internal/host/movesprobeowners.go
@@ -562,7 +562,7 @@ func (m *movesProbeOwners) Select(ctx context.Context, rt registry.Route, o host
}
return ClaimLaunchInputs{Roles: []string{"main"}, Identity: launchIdentity(pins), Baseline: plan, Admission: admission,
Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: probeEnv(l.Env),
- DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit, Prepared: prepared, ReadyPath: ready}, nil
+ DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit, Prepared: prepared, ReadyPath: ready}, nil
}
// AdmissionEvidence re-derives Select's admission against the observation
@@ -1291,7 +1291,7 @@ func (m *movesProbeOwners) selectCommitted(ctx context.Context, rt registry.Rout
}
return ClaimLaunchInputs{Roles: []string{"main"}, Identity: id, Baseline: plan, Admission: m.committedAdmission(o, v, gen, plan),
Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: withoutProbeVars(l.Env),
- DBPath: rt.DBPath, Version: l.Version, Limit: l.Limit}, nil
+ DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit}, nil
}
func (m *movesProbeOwners) committedEvidence(ctx context.Context, o hostclaim.Observed, v probeLanding, in hostclaim.Admission, baseline *genselect.Plan) error {
diff --git a/internal/release/staged.go b/internal/release/staged.go
index 5c2760aef..a3883961a 100644
--- a/internal/release/staged.go
+++ b/internal/release/staged.go
@@ -116,16 +116,28 @@ func linkChain(link string) (raw string, hops, aliases []string, err error) {
// outside the version store; the path and digest identify the bytes either
// way. Demand is left empty: only CanonicalDemand can vouch for it.
func ResolveLive(optDir, host string) (StagedIdentity, error) {
- raw, hops, _, err := linkChain(LiveLink(optDir, host))
+ id, err := ResolveLiveLink(optDir, host)
if err != nil {
return StagedIdentity{}, err
}
- path := hops[len(hops)-1]
- sum, err := FileSHA256(path)
+ sum, err := FileSHA256(id.StagedPath)
+ if err != nil {
+ return StagedIdentity{}, fmt.Errorf("hash %s: %w", id.StagedPath, err)
+ }
+ id.Digest = sum
+ return id, nil
+}
+
+// ResolveLiveLink is ResolveLive without the hash: the file at the end of
+// the live link's chain and the version that same reading of the link names,
+// for a caller that opens and hashes the file itself (the claimed launch
+// snapshot). Digest is left empty.
+func ResolveLiveLink(optDir, host string) (StagedIdentity, error) {
+ raw, hops, _, err := linkChain(LiveLink(optDir, host))
if err != nil {
- return StagedIdentity{}, fmt.Errorf("hash %s: %w", path, err)
+ return StagedIdentity{}, err
}
- return StagedIdentity{Version: versionOfTarget(optDir, raw), StagedPath: path, Digest: sum}, nil
+ return StagedIdentity{Version: versionOfTarget(optDir, raw), StagedPath: hops[len(hops)-1]}, nil
}
// VerifyStaged checks the file at id.StagedPath against id.Digest, the digest