amadan

carlos / platform Public

settings-4 task 4: the agent's build-state store and registry copy

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 0d6e5d775c7bf2c78befeac5eaaa7c7315822935 parent aca319e
7 files changed, +684 −9
  • internal/activator/buildstatelocation.go +29 −0
  • internal/activator/buildstatelocation_test.go +29 −0
  • internal/activator/claimwake.go +10 −5
  • internal/host/agent.go +24 −4
  • internal/host/buildstate.go +249 −0
  • internal/host/buildstate_test.go +340 −0
  • internal/host/claimassembly.go +3 −0
diff --git a/internal/activator/buildstatelocation.go b/internal/activator/buildstatelocation.go
new file mode 100644
index 000000000..1b0e493be
--- /dev/null
+++ b/internal/activator/buildstatelocation.go
@@ -0,0 +1,29 @@
+package activator
+
+import "context"
+
+// BuildStateLocation is where an instance's build-state.json lives (settings
+// spec §4.2): the bucket and the prefix of the lease's lock.json. With host
+// claims off that is Config.Bucket and HibernatePrefix(Config.Prefix, host);
+// with them on, the claim actor's bucket and Holder.Location (never a
+// g-<gen>/ generation prefix). The wake knows which lease it holds, so it
+// puts the location on the context it hands BeforeExec, and the agent's hook
+// writes there rather than guessing.
+type BuildStateLocation struct{ Bucket, Prefix string }
+
+type buildStateLocationKey struct{}
+
+// WithBuildStateLocation returns ctx carrying l. A location with either part
+// empty reads back as none, so a caller can mask a location it inherited.
+func WithBuildStateLocation(ctx context.Context, l BuildStateLocation) context.Context {
+ return context.WithValue(ctx, buildStateLocationKey{}, l)
+}
+
+// BuildStateLocationFrom is the location ctx carries, if a whole one.
+func BuildStateLocationFrom(ctx context.Context) (BuildStateLocation, bool) {
+ l, ok := ctx.Value(buildStateLocationKey{}).(BuildStateLocation)
+ if !ok || l.Bucket == "" || l.Prefix == "" {
+ return BuildStateLocation{}, false
+ }
+ return l, true
+}
diff --git a/internal/activator/buildstatelocation_test.go b/internal/activator/buildstatelocation_test.go
new file mode 100644
index 000000000..c38020377
--- /dev/null
+++ b/internal/activator/buildstatelocation_test.go
@@ -0,0 +1,29 @@
+package activator
+
+import (
+ "context"
+ "testing"
+)
+
+func TestBuildStateLocationRidesTheContext(t *testing.T) {
+ ctx := context.Background()
+ if l, ok := BuildStateLocationFrom(ctx); ok {
+ t.Fatalf("a bare context carries %+v", l)
+ }
+ want := BuildStateLocation{Bucket: "data-eu", Prefix: "instances/a.example.com/inc1"}
+ got, ok := BuildStateLocationFrom(WithBuildStateLocation(ctx, want))
+ if !ok || got != want {
+ t.Fatalf("got %+v, %v; want %+v", got, ok, want)
+ }
+}
+
+// A zero location is no location: it is how a caller that serves many
+// hosts (the agent's refreshAll) masks one a wake left on its context.
+func TestAnEmptyBuildStateLocationIsNone(t *testing.T) {
+ ctx := WithBuildStateLocation(context.Background(), BuildStateLocation{Bucket: "b", Prefix: "p"})
+ for _, l := range []BuildStateLocation{{}, {Bucket: "b"}, {Prefix: "p"}} {
+ if got, ok := BuildStateLocationFrom(WithBuildStateLocation(ctx, l)); ok {
+ t.Fatalf("%+v read back as %+v", l, got)
+ }
+ }
+}
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index 8c344f5fc..f7c439c71 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -69,11 +69,16 @@ type ClaimRuntime struct {
RecoveryAttemptsReady func() bool
FailedAdmission func(context.Context, hostclaim.Observed, error) error
Box string
- Read func(context.Context, string) (hostclaim.Observed, error)
- StopForeign func(context.Context, string) error
- Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
- Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
- Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
+ // Bucket and LeaseRegion are the claim actor's bucket and the region its
+ // leases sign for: a claimed wake leases Holder.Location there, and the
+ // instance's build-state.json sits beside that lease (settings spec
+ // §4.2, ruling R-CLAIMPREFIX).
+ Bucket, LeaseRegion string
+ Read func(context.Context, string) (hostclaim.Observed, error)
+ StopForeign func(context.Context, string) error
+ Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error
+ Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error)
+ Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error)
// Pins reports that Prepare honours StartPolicy.Pin. A hold swap is
// refused on a claims-on box whose runtime does not, before anything
// stops.
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 77bfafda0..f15ff61fd 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -294,6 +294,15 @@ func runAgent(args []string, buildClaims ClaimFactory) {
if err := configureWakeClaims(claims, cfg); err != nil {
log.Fatalf("agent: host claims: %v", err)
}
+ // The build-state store (settings spec §4.2), built once after the claim
+ // wiring so it locates claimed hosts through cfg.Claims. nil, nil on a box
+ // with no replica bucket. Settings plan 4 Task 5 installs the BeforeExec
+ // hook from bs, and a refusing hook when bsErr is set; until then a
+ // failure here is one operator line and the agent keeps running.
+ bs, bsErr := newBuildStates(cfg, reg, release.OptDir())
+ if bsErr != nil {
+ log.Printf("agent: build state store unavailable: %v. this box does not keep build states", bsErr)
+ }
// Fleet box identity (customer-fleets plan, Task 6 fix round — controller
// ruling): resolved ONCE, here, before the activator is even constructed.
@@ -609,7 +618,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}()
go adoptLoop(reg, svc, sd, claims.Writes)
- go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, claims.Providers.Provision)
+ go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)
if logSink != nil {
go logshipLoop(logSink, status.Label(readMachineID()))
}
@@ -621,7 +630,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// by CARLOS_HOST_METRICS — one log line and nothing else where unset.
go memwatchLoop()
go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)
- go instancesLoop(reg, claims.Writes, claims.Providers.Provision)
+ go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)
// The deployment-global election, armed BEFORE the loops it gates so a
// single-box deployment is holding well before their first tick (a
// minute away) — see housekeeper.go. Not a `go` of its own: it publishes
@@ -2253,7 +2262,7 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
//
// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
// whole decision lives in fastTick/fastMembers, which are tested.
-func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, configure ...func(*provision.ReconcileConfig)) {
+func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -2350,6 +2359,11 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
if err := reconcileSerialized(ctx, store, reg, configuredClaimProvision(provisionTickConfig(os.Getenv), configure)); err != nil {
log.Printf("agent: fast converge reconcile: %v", err)
}
+ // The registry's build-state copies, refreshed at every reconcile
+ // (settings spec §4.2).
+ if bs != nil {
+ bs.refreshAll(ctx)
+ }
mark("reconcile")
telemetryPass(ctx, store, reg)
mark("telemetry")
@@ -2898,7 +2912,7 @@ func domainsDuty(pass func(context.Context) error) housekeeperDuty {
// already had a store+reg in scope) — that coupled binding materialization
// to PLATFORM_APPS_DOMAIN being set, which has nothing to do with fleets;
// this loop has no such gate and shares the same store already opened above.
-func instancesLoop(reg *registry.Registry, writes hostwrite.Config, configure ...func(*provision.ReconcileConfig)) {
+func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
ctx := context.Background()
store, err := openDeployStore(ctx)
if err != nil {
@@ -2926,6 +2940,12 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, configure ..
if err := reconcileSerialized(ctx, store, reg, cfg); err != nil {
log.Printf("agent: instances pass: %v", err)
}
+ // Every reconcile refreshes the build-state copies (settings spec
+ // §4.2), this 5-minute one too: the fast leg above runs only on a
+ // serial move.
+ if bs != nil {
+ bs.refreshAll(ctx)
+ }
telemetryPass(ctx, store, reg)
runMovesPass(ctx, store)
deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go
new file mode 100644
index 000000000..099b47044
--- /dev/null
+++ b/internal/host/buildstate.go
@@ -0,0 +1,249 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// buildStates is the agent's side of settings spec §4.2: the build-state
+// object beside each instance's replica, under its lease, and its registry
+// copy root reads.
+//
+// The object sits beside the lease's lock.json (ruling R-CLAIMPREFIX). With
+// host claims off that is the hibernate bucket at HibernatePrefix; with them
+// on, the claim actor's bucket at the claim holder's Location (never the
+// session's g-<gen>/ prefix). A wake that holds the lease says which on its
+// context (activator.WithBuildStateLocation), and that wins.
+type buildStates struct {
+ hibBucket string // CARLOS_HIBERNATE_BUCKET
+ prefix func(host string) string // HibernatePrefix, host claims off
+ // claims reads a host's claim; nil means host claims are off. With them
+ // on, every exec instance wakes through its claim (doWake ->
+ // doClaimWake), so every host is located through it.
+ claims func(context.Context, string) (hostclaim.Observed, error)
+ claimBucket string // the claim actor's bucket, where claimed wakes lease
+ storeFor func(ctx context.Context, bucket string) (release.Store, error)
+ reg *registry.Registry
+ optDir string
+ now func() time.Time
+
+ mu sync.Mutex
+ hosts map[string]*sync.Mutex // per-host: a read and its column write are one step
+}
+
+// newBuildStates is nil, nil on a box with no replica bucket: it runs no
+// hibernating instances, so it has no build state to keep. cfg is the
+// activator's, after configureWakeClaims: cfg.HostWrites.Enabled is the
+// CARLOS_HOST_CLAIM switch and cfg.Claims its wake runtime.
+func newBuildStates(cfg *activator.Config, reg *registry.Registry, optDir string) (*buildStates, error) {
+ if cfg == nil || cfg.Bucket == "" {
+ return nil, nil
+ }
+ regions := map[string]string{cfg.Bucket: cfg.Region}
+ b := &buildStates{hibBucket: cfg.Bucket, prefix: func(h string) string { return activator.HibernatePrefix(cfg.Prefix, h) },
+ reg: reg, optDir: optDir, now: time.Now}
+ if cfg.HostWrites.Enabled {
+ if cfg.Claims == nil || cfg.Claims.Read == nil || cfg.Claims.Bucket == "" {
+ return nil, errors.New("host claims are on but the claim reader or the claim bucket is missing")
+ }
+ if r, ok := regions[cfg.Claims.Bucket]; ok && r != cfg.Claims.LeaseRegion {
+ return nil, fmt.Errorf("bucket %s is leased in two regions (%q, %q)", cfg.Claims.Bucket, r, cfg.Claims.LeaseRegion)
+ }
+ regions[cfg.Claims.Bucket] = cfg.Claims.LeaseRegion
+ b.claims, b.claimBucket = cfg.Claims.Read, cfg.Claims.Bucket
+ }
+ b.storeFor = cachedStores(regions, lease.ExecutionStore)
+ return b, nil
+}
+
+// cachedStores opens one store per bucket the box leases in, with the
+// lease's own S3 configuration (lease.ExecutionStore: the region the lock
+// was signed for), and keeps it. Any other bucket is refused.
+func cachedStores(regions map[string]string, open func(ctx context.Context, bucket, region string) (release.Store, error)) func(context.Context, string) (release.Store, error) {
+ var mu sync.Mutex
+ stores := map[string]release.Store{}
+ return func(ctx context.Context, bucket string) (release.Store, error) {
+ region, ok := regions[bucket]
+ if !ok {
+ return nil, fmt.Errorf("bucket %q is not one this box leases in", bucket)
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ if s := stores[bucket]; s != nil {
+ return s, nil
+ }
+ s, err := open(ctx, bucket, region)
+ if err != nil {
+ return nil, err
+ }
+ stores[bucket] = s
+ return s, nil
+ }
+}
+
+// locate is the store and prefix of host's build state: the context's
+// location when a wake set one; otherwise, with host claims on, the claim
+// holder's Location in the claim bucket (an unreadable claim is an error,
+// never the hibernate prefix); otherwise the hibernate bucket and prefix.
+func (b *buildStates) locate(ctx context.Context, host string) (release.Store, string, error) {
+ bucket, prefix := b.hibBucket, ""
+ if l, ok := activator.BuildStateLocationFrom(ctx); ok {
+ bucket, prefix = l.Bucket, l.Prefix
+ } else if b.claims != nil {
+ obs, err := b.claims(ctx, host)
+ if err != nil {
+ return nil, "", fmt.Errorf("read claim: %w", err)
+ }
+ if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Holder.Location == "" {
+ return nil, "", fmt.Errorf("claim of %s has no holder location", host)
+ }
+ bucket, prefix = b.claimBucket, obs.Claim.Holder.Location
+ } else {
+ prefix = b.prefix(host)
+ }
+ prefix = strings.TrimSuffix(prefix, "/")
+ s, err := b.storeFor(ctx, bucket)
+ if err != nil {
+ return nil, "", err
+ }
+ return s, prefix, nil
+}
+
+func (b *buildStates) hostLock(host string) *sync.Mutex {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if b.hosts == nil {
+ b.hosts = map[string]*sync.Mutex{}
+ }
+ m := b.hosts[host]
+ if m == nil {
+ m = &sync.Mutex{}
+ b.hosts[host] = m
+ }
+ return m
+}
+
+// read returns host's build state (nil when there is none) and whether its
+// replica has any object besides the lease's lock.json and build-state.json.
+func (b *buildStates) read(ctx context.Context, host string) (*release.BuildState, bool, error) {
+ store, prefix, err := b.locate(ctx, host)
+ if err != nil {
+ return nil, false, err
+ }
+ raw, _, err := store.Get(ctx, release.BuildStateKey(prefix))
+ var st *release.BuildState
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ case err != nil:
+ return nil, false, err
+ default:
+ var s release.BuildState
+ if err := json.Unmarshal(raw, &s); err != nil || s.Host != host {
+ return nil, false, fmt.Errorf("build state for %s does not parse or names another host", host)
+ }
+ st = &s
+ }
+ keys, err := store.List(ctx, prefix+"/")
+ if err != nil {
+ return nil, false, err
+ }
+ replica := false
+ for _, k := range keys {
+ rel := strings.TrimPrefix(k, prefix+"/")
+ if rel != "lock.json" && rel != "build-state.json" {
+ replica = true
+ break
+ }
+ }
+ return st, replica, nil
+}
+
+// refresh stores what read found. A failed read keeps the last known state
+// for root (a known copy is not wiped by a blip; an unread one stays
+// unknown) and records the failure in ReadErr, so the box's report says this
+// host's build state is incomplete and the housekeeper prunes nothing for its
+// app (Astra I1 #6). A good read clears ReadErr.
+func (b *buildStates) refresh(ctx context.Context, host string) error {
+ m := b.hostLock(host)
+ m.Lock()
+ defer m.Unlock()
+ st, replica, err := b.read(ctx, host)
+ if err != nil {
+ rt, gerr := b.reg.Get(host)
+ prev := release.BuildStateCopy{}
+ if gerr == nil && rt != nil {
+ prev, _ = release.ParseBuildStateCopy(rt.BuildState)
+ }
+ prev.ReadErr = err.Error()
+ if serr := b.reg.SetBuildState(host, prev.Column()); serr != nil {
+ return errors.Join(err, serr)
+ }
+ return err
+ }
+ c := release.BuildStateCopy{Known: true, Replica: replica, State: st, ReadAt: b.now().UTC().Truncate(time.Second)}
+ return b.reg.SetBuildState(host, c.Column())
+}
+
+// refreshAll runs after every reconcile, for every activator-managed exec
+// instance: Kind instance, Hibernate, no Unit, not the platform's own.
+// Each host is located on its own: a location a caller's context carries
+// names one wake's lease, so it is masked here.
+func (b *buildStates) refreshAll(ctx context.Context) {
+ ctx = activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{})
+ routes, err := b.reg.List()
+ if err != nil {
+ log.Printf("agent: build state refresh: %v", err)
+ return
+ }
+ for _, rt := range routes {
+ if rt.Kind != "instance" || !rt.Hibernate || rt.Unit != "" || rt.Platform {
+ continue
+ }
+ hctx, cancel := context.WithTimeout(ctx, buildStateRefreshLimit)
+ if err := b.refresh(hctx, rt.Host); err != nil {
+ log.Printf("agent: build state of %s: %v", rt.Host, err)
+ }
+ cancel()
+ }
+}
+
+// buildStateRefreshLimit bounds one host's refresh inside refreshAll, so a
+// hung bucket call cannot stall the reconcile leg that runs it. A timeout is
+// a read error like any other: the copy keeps its state and records it.
+const buildStateRefreshLimit = 30 * time.Second
+
+// write puts the object. Callers hold the instance's lease.
+func (b *buildStates) write(ctx context.Context, host string, s release.BuildState) error {
+ store, prefix, err := b.locate(ctx, host)
+ if err != nil {
+ return err
+ }
+ s.Schema, s.Host = 1, host
+ s.At = b.now().UTC().Truncate(time.Second)
+ raw, err := json.MarshalIndent(s, "", " ")
+ if err != nil {
+ return err
+ }
+ key := release.BuildStateKey(prefix)
+ _, etag, err := store.Get(ctx, key)
+ if errors.Is(err, os.ErrNotExist) {
+ etag = ""
+ } else if err != nil {
+ return err
+ }
+ _, err = store.Put(ctx, key, raw, etag)
+ return err
+}
diff --git a/internal/host/buildstate_test.go b/internal/host/buildstate_test.go
new file mode 100644
index 000000000..281966eb5
--- /dev/null
+++ b/internal/host/buildstate_test.go
@@ -0,0 +1,340 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// storesOf is a storeFor over fixed stores, one per bucket name.
+func storesOf(m map[string]release.Store) func(context.Context, string) (release.Store, error) {
+ return func(_ context.Context, bucket string) (release.Store, error) {
+ if s, ok := m[bucket]; ok {
+ return s, nil
+ }
+ return nil, fmt.Errorf("no store for bucket %q", bucket)
+ }
+}
+
+func testBuildStates(t *testing.T) (*buildStates, release.Store, *registry.Registry) {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "r.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "km", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ s := release.NewDirStore(t.TempDir())
+ return &buildStates{hibBucket: "hib", prefix: func(h string) string { return "inst/" + h },
+ storeFor: storesOf(map[string]release.Store{"hib": s}), reg: reg, optDir: t.TempDir(), now: time.Now}, s, reg
+}
+
+func copyOf(t *testing.T, reg *registry.Registry) release.BuildStateCopy {
+ t.Helper()
+ rt, _ := reg.Get("a.example.com")
+ c, err := release.ParseBuildStateCopy(rt.BuildState)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return c
+}
+
+func TestRefreshRecordsNoStateAndNoReplica(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ if err := b.refresh(context.Background(), "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if c := copyOf(t, reg); !c.Known || c.Replica || c.State != nil || c.ReadErr != "" {
+ t.Fatalf("copy = %+v", c)
+ }
+}
+
+func TestRefreshSeesAReplicaAndAState(t *testing.T) {
+ b, s, reg := testBuildStates(t)
+ ctx := context.Background()
+ if _, err := s.Put(ctx, "inst/a.example.com/generations/0000/x.ltx", []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 2}); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ c := copyOf(t, reg)
+ if !c.Known || !c.Replica || c.State == nil || c.State.LastExecuted != "v1" || c.State.Completed != 2 {
+ t.Fatalf("copy = %+v", c)
+ }
+}
+
+// Astra I1 #6: a failed read keeps the last state and is recorded, and the
+// next good read clears the record.
+func TestAFailedRefreshIsRecorded(t *testing.T) {
+ b, s, reg := testBuildStates(t)
+ ctx := context.Background()
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ key := "inst/a.example.com/build-state.json"
+ if _, err := s.Put(ctx, key, []byte("{"), mustEtag(t, s, key)); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err == nil {
+ t.Fatal("an unparseable build state refreshed cleanly")
+ }
+ c := copyOf(t, reg)
+ if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
+ t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
+ }
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if c := copyOf(t, reg); c.ReadErr != "" || c.State == nil || c.State.LastExecuted != "v2" {
+ t.Fatalf("copy = %+v; want a good read to clear the failure", c)
+ }
+}
+
+// A host whose copy was never read stays unknown when its first read fails.
+func TestAFailedFirstRefreshStaysUnknown(t *testing.T) {
+ b, s, reg := testBuildStates(t)
+ ctx := context.Background()
+ if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte("{"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err == nil {
+ t.Fatal("an unparseable build state refreshed cleanly")
+ }
+ if c := copyOf(t, reg); c.Known || c.ReadErr == "" || c.State != nil {
+ t.Fatalf("copy = %+v; want unknown with the failure recorded", c)
+ }
+}
+
+func mustEtag(t *testing.T, s release.Store, key string) string {
+ t.Helper()
+ _, etag, err := s.Get(context.Background(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return etag
+}
+
+func TestALockAloneIsNotAReplica(t *testing.T) {
+ b, s, reg := testBuildStates(t)
+ ctx := context.Background()
+ if _, err := s.Put(ctx, "inst/a.example.com/lock.json", []byte("{}"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if copyOf(t, reg).Replica {
+ t.Fatal("a lease's lock was taken for a replica")
+ }
+}
+
+// A state naming another host is refused, not adopted.
+func TestAStateForAnotherHostIsAReadError(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ ctx := context.Background()
+ if _, err := s.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"b.example.com","last_executed":"v1"}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := b.read(ctx, "a.example.com"); err == nil {
+ t.Fatal("a state naming another host was read as this one's")
+ }
+}
+
+// refreshAll covers exactly the activator-managed exec instances.
+func TestRefreshAllCoversOnlyExecInstances(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ for _, rt := range []registry.Route{
+ {Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Unit: "km-unit"},
+ {Host: "awake.example.com", Kind: "instance", App: "km"},
+ {Host: "plat.example.com", Kind: "instance", App: "km", Hibernate: true, Platform: true},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ b.refreshAll(context.Background())
+ if c := copyOf(t, reg); !c.Known {
+ t.Fatalf("the exec instance was not refreshed: %+v", c)
+ }
+ for _, h := range []string{"unit.example.com", "awake.example.com", "plat.example.com"} {
+ rt, err := reg.Get(h)
+ if err != nil || rt == nil {
+ t.Fatalf("%s: %v", h, err)
+ }
+ if rt.BuildState != "" {
+ t.Fatalf("%s got a build state copy: %s", h, rt.BuildState)
+ }
+ }
+}
+
+// --- where the object lives (ruling R-CLAIMPREFIX) ---
+
+// Host claims off: the hibernate bucket, at the hibernate prefix.
+func TestClaimsOffUsesTheHibernatePrefix(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ ctx := context.Background()
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
+ t.Fatalf("the build state is not beside the hibernate lease: %v", err)
+ }
+}
+
+// A location on the context (the wake's own lease) wins over both modes.
+func TestAContextLocationWins(t *testing.T) {
+ b, hib, reg := testBuildStates(t)
+ other := release.NewDirStore(t.TempDir())
+ b.storeFor = storesOf(map[string]release.Store{"hib": hib, "data": other})
+ b.claimBucket = "data"
+ b.claims = func(context.Context, string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{}, errors.New("the claim must not be read when the wake names its lease")
+ }
+ ctx := activator.WithBuildStateLocation(context.Background(), activator.BuildStateLocation{Bucket: "data", Prefix: "instances/a.example.com/inc1"})
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := other.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
+ t.Fatalf("the build state is not at the context's location: %v", err)
+ }
+ if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the hibernate prefix was written too: %v", err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
+ t.Fatalf("copy = %+v", c)
+ }
+}
+
+func claimedAt(location string) func(context.Context, string) (hostclaim.Observed, error) {
+ return func(_ context.Context, host string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: location, Generation: "7", Session: 3}}}, nil
+ }
+}
+
+// Host claims on: the claim actor's bucket, at the holder's Location, never
+// under the session's g-<gen>/ prefix; the generation's objects are the
+// replica.
+func TestClaimsOnUsesTheHolderLocationInTheActorBucket(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"
+ b.claims = claimedAt("instances/a.example.com/inc1")
+ ctx := context.Background()
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := data.Get(ctx, "instances/a.example.com/inc1/build-state.json"); err != nil {
+ t.Fatalf("the build state is not beside the claimed lease: %v", err)
+ }
+ if _, _, err := hib.Get(ctx, "inst/a.example.com/build-state.json"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a claimed host's build state went to the hibernate prefix: %v", err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if c := copyOf(t, reg); !c.Known || c.Replica || c.State == nil || c.State.LastExecuted != "v1" {
+ t.Fatalf("copy = %+v", c)
+ }
+ if _, err := data.Put(ctx, "instances/a.example.com/inc1/g-7/0000.ltx", []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if c := copyOf(t, reg); !c.Replica {
+ t.Fatalf("copy = %+v; want the generation's objects seen as the replica", c)
+ }
+}
+
+// A claimed host whose claim cannot be read (or has no holder) records the
+// failure and keeps its last state; it never falls back to the hibernate
+// prefix, even when an object sits there.
+func TestAnUnreadableClaimIsAReadError(t *testing.T) {
+ for name, read := range map[string]func(context.Context, string) (hostclaim.Observed, error){
+ "unreadable": func(context.Context, string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{}, errors.New("bucket blip")
+ },
+ "absent": func(_ context.Context, h string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{Host: h, Absent: true}, nil
+ },
+ "no holder": func(_ context.Context, h string) (hostclaim.Observed, error) {
+ return hostclaim.Observed{Host: h, Claim: &hostclaim.Claim{Host: h}}, nil
+ },
+ } {
+ t.Run(name, func(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"
+ b.claims = claimedAt("instances/a.example.com")
+ ctx := context.Background()
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := b.refresh(ctx, "a.example.com"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hib.Put(ctx, "inst/a.example.com/build-state.json", []byte(`{"schema":1,"host":"a.example.com","last_executed":"hib"}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ b.claims = read
+ if err := b.refresh(ctx, "a.example.com"); err == nil {
+ t.Fatal("a claimed host refreshed without its claim")
+ }
+ c := copyOf(t, reg)
+ if c.ReadErr == "" || !c.Known || c.State == nil || c.State.LastExecuted != "v1" {
+ t.Fatalf("copy = %+v; want the last state kept and the failure recorded", c)
+ }
+ if err := b.write(ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
+ t.Fatal("a claimed host's build state was written without its claim")
+ }
+ })
+ }
+}
+
+// The production store opener: one store per known bucket, opened once,
+// each with its lease's region; an unknown bucket is refused.
+func TestStoresAreOpenedPerBucketAndCached(t *testing.T) {
+ opened := map[string]int{}
+ open := func(_ context.Context, bucket, region string) (release.Store, error) {
+ opened[bucket+"@"+region]++
+ return release.NewDirStore(t.TempDir()), nil
+ }
+ storeFor := cachedStores(map[string]string{"hib": "eu-west-1", "data": "us-east-1"}, open)
+ ctx := context.Background()
+ for _, bucket := range []string{"hib", "data", "hib", "data"} {
+ if _, err := storeFor(ctx, bucket); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if len(opened) != 2 || opened["hib@eu-west-1"] != 1 || opened["data@us-east-1"] != 1 {
+ t.Fatalf("opened = %v", opened)
+ }
+ if _, err := storeFor(ctx, "elsewhere"); err == nil {
+ t.Fatal("a bucket the box does not lease in was opened")
+ }
+}
diff --git a/internal/host/claimassembly.go b/internal/host/claimassembly.go
index e07a49179..29b91b004 100644
--- a/internal/host/claimassembly.go
+++ b/internal/host/claimassembly.go
@@ -204,6 +204,9 @@ func (a ClaimAssembly) Build(ctx context.Context) (ClaimProviders, error) {
if wake == nil || wake.Read == nil || wake.Prerequisites == nil || wake.Admit == nil || wake.Prepare == nil || wake.StopForeign == nil || wake.FailedAdmission == nil {
return ClaimProviders{}, hostwrite.ErrUnavailable
}
+ // Where claimed wakes lease (writer.Bucket, a.Wake.LeaseRegion above):
+ // the agent's build-state store reads and writes beside that lease.
+ wake.Bucket, wake.LeaseRegion = actor.Bucket, actor.Cloud.Region
enrollment := claimRegionalEnrollment{identity: a.Identity, actor: actor}
authorize := func(ctx context.Context, o hostclaim.Observed, ob hostclaim.Obligation) error {
if err := enrollment.VerifyRegional(ctx, actor.Region, actor.Bucket, hostbox.RequestLimit); err != nil {