amadan

carlos / platform Public

settings-4 task 4: fix round 1: update merges against the read etag; bounded refresh; worker rows

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1cd62de810d00a2801a0bff2be45890bcb1b388b parent 0d6e5d7
3 files changed, +448 −82
  • internal/host/agent.go +4 −2
  • internal/host/buildstate.go +157 −62
  • internal/host/buildstate_test.go +287 −18
diff --git a/internal/host/agent.go b/internal/host/agent.go
index f15ff61fd..0ffa1623a 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -2359,12 +2359,14 @@ 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)
}
+ mark("reconcile")
// The registry's build-state copies, refreshed at every reconcile
- // (settings spec §4.2).
+ // (settings spec §4.2), before adopt reads them; refreshAll keeps
+ // to its own budget, and the stage is timed apart from reconcile.
if bs != nil {
bs.refreshAll(ctx)
+ mark("build states")
}
- mark("reconcile")
telemetryPass(ctx, store, reg)
mark("telemetry")
runMovesPass(ctx, store)
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go
index 099b47044..86caab1e8 100644
--- a/internal/host/buildstate.go
+++ b/internal/host/buildstate.go
@@ -39,9 +39,14 @@ type buildStates struct {
reg *registry.Registry
optDir string
now func() time.Time
+ // budget bounds one refreshAll; 0 means refreshBudget.
+ budget time.Duration
+ // beforePut, when set (tests only), runs in update between its read and
+ // its conditional Put.
+ beforePut func(ctx context.Context, store release.Store, key string)
mu sync.Mutex
- hosts map[string]*sync.Mutex // per-host: a read and its column write are one step
+ hosts map[string]*sync.Mutex // per-host: a read or update and its column write are one step
}
// newBuildStates is nil, nil on a box with no replica bucket: it runs no
@@ -52,23 +57,34 @@ func newBuildStates(cfg *activator.Config, reg *registry.Registry, optDir string
if cfg == nil || cfg.Bucket == "" {
return nil, nil
}
- regions := map[string]string{cfg.Bucket: cfg.Region}
+ regions, err := buildStateRegions(cfg)
+ if err != nil {
+ return nil, err
+ }
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
}
+// buildStateRegions is the one bucket the box leases in, with the region its
+// leases sign for. With host claims on every wake is a claimed wake
+// (doWake -> doClaimWake) and leases in the claim actor's bucket, so only
+// that bucket is registered: the hibernate bucket is never located there,
+// even when it is the same bucket under another region setting.
+func buildStateRegions(cfg *activator.Config) (map[string]string, error) {
+ if !cfg.HostWrites.Enabled {
+ return map[string]string{cfg.Bucket: cfg.Region}, nil
+ }
+ 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")
+ }
+ return map[string]string{cfg.Claims.Bucket: cfg.Claims.LeaseRegion}, 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.
@@ -136,6 +152,47 @@ func (b *buildStates) hostLock(host string) *sync.Mutex {
return m
}
+// getState reads the object at key: nil and "" when there is none, an error
+// when it does not parse or names another host.
+func getState(ctx context.Context, store release.Store, key, host string) (*release.BuildState, string, error) {
+ raw, etag, err := store.Get(ctx, key)
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ return nil, "", nil
+ case err != nil:
+ return nil, "", err
+ }
+ var s release.BuildState
+ if err := json.Unmarshal(raw, &s); err != nil || s.Host != host {
+ return nil, "", fmt.Errorf("build state for %s does not parse or names another host", host)
+ }
+ return &s, etag, nil
+}
+
+// hasReplica is whether prefix holds any object besides the lease's
+// lock.json and build-state.json. Three keys decide it, so a store with a
+// bounded listing (release.PageLister) is asked for three; any other lists
+// the prefix whole.
+func hasReplica(ctx context.Context, store release.Store, prefix string) (bool, error) {
+ var keys []string
+ var err error
+ if pl, ok := store.(release.PageLister); ok {
+ keys, err = pl.ListPage(ctx, prefix+"/", "", 3)
+ } else {
+ keys, err = store.List(ctx, prefix+"/")
+ }
+ if err != nil {
+ return false, err
+ }
+ for _, k := range keys {
+ rel := strings.TrimPrefix(k, prefix+"/")
+ if rel != "lock.json" && rel != "build-state.json" {
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
// 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) {
@@ -143,31 +200,14 @@ func (b *buildStates) read(ctx context.Context, host string) (*release.BuildStat
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:
+ st, _, err := getState(ctx, store, release.BuildStateKey(prefix), host)
+ if 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+"/")
+ replica, err := hasReplica(ctx, store, 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
}
@@ -197,53 +237,108 @@ func (b *buildStates) refresh(ctx context.Context, host string) error {
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{})
+// keepsBuildState is a row whose spawns run under a lease in the replica
+// bucket, so the store keeps a build state beside it (ruling R-WORKER-ROWS):
+// an activator-managed exec instance (activator.ActivatorManaged: Kind
+// instance, Hibernate, a socket, no Backing, no Tunnel), or a standalone
+// worker (Kind "sidecar"; every worker wake takes the lease at
+// HibernatePrefix(its host), activator.go doWake step 1, whatever its
+// 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 {
+ if rt.Unit != "" || rt.Platform {
+ return false
+ }
+ return activator.ActivatorManaged(rt) || (rt.IsSidecar() && rt.Backing == "" && rt.Tunnel == "")
+}
+
+// refreshBudget bounds one refreshAll: it runs inline in the reconcile legs,
+// before the adopt pass. A host it does not reach keeps its last copy; a
+// host whose read the deadline cuts records the failure like any other.
+const refreshBudget = 20 * time.Second
+
+// refreshAll runs after every reconcile, for every row keepsBuildState
+// names, within the budget. Each host is located on its own: a location a
+// caller's context carries names one wake's lease, so it is masked here. It
+// returns how many hosts it did not reach.
+func (b *buildStates) refreshAll(ctx context.Context) int {
+ budget := b.budget
+ if budget <= 0 {
+ budget = refreshBudget
+ }
+ ctx, cancel := context.WithTimeout(activator.WithBuildStateLocation(ctx, activator.BuildStateLocation{}), budget)
+ defer cancel()
routes, err := b.reg.List()
if err != nil {
log.Printf("agent: build state refresh: %v", err)
- return
+ return 0
}
+ var hosts []string
for _, rt := range routes {
- if rt.Kind != "instance" || !rt.Hibernate || rt.Unit != "" || rt.Platform {
- continue
+ if keepsBuildState(rt) {
+ hosts = append(hosts, rt.Host)
}
- 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)
+ }
+ for i, host := range hosts {
+ if ctx.Err() != nil {
+ skipped := len(hosts) - i
+ log.Printf("agent: build state refresh: the %s budget ran out; %d of %d hosts not reached keep their last copy", budget, skipped, len(hosts))
+ return skipped
+ }
+ if err := b.refresh(ctx, host); err != nil {
+ log.Printf("agent: build state of %s: %v", host, err)
}
- cancel()
}
+ return 0
}
-// 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
+// updateTries bounds update's compare-and-swap loop.
+const updateTries = 3
-// write puts the object. Callers hold the instance's lease.
-func (b *buildStates) write(ctx context.Context, host string, s release.BuildState) error {
+// update is the only writer of the object (ruling R-BS-UPDATE). Callers hold
+// the host's lease. Under the host's lock it reads the object (absent is a
+// nil cur and a create-only Put), merges, and Puts against the etag it read,
+// so a write that landed in between is never overwritten: a CAS conflict
+// re-reads and merges again, up to updateTries times. The registry copy is
+// set under the same lock, so a concurrent refresh cannot put older state
+// over it. A failed update writes nothing the caller may run on.
+func (b *buildStates) update(ctx context.Context, host string, merge func(cur *release.BuildState) release.BuildState) (release.BuildState, error) {
+ m := b.hostLock(host)
+ m.Lock()
+ defer m.Unlock()
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
+ return release.BuildState{}, err
}
key := release.BuildStateKey(prefix)
- _, etag, err := store.Get(ctx, key)
- if errors.Is(err, os.ErrNotExist) {
- etag = ""
- } else if err != nil {
- return err
+ for try := 0; try < updateTries; try++ {
+ cur, etag, err := getState(ctx, store, key, host)
+ if err != nil {
+ return release.BuildState{}, err
+ }
+ next := merge(cur)
+ next.Schema, next.Host = 1, host
+ next.At = b.now().UTC().Truncate(time.Second)
+ raw, err := json.MarshalIndent(next, "", " ")
+ if err != nil {
+ return release.BuildState{}, err
+ }
+ if b.beforePut != nil {
+ b.beforePut(ctx, store, key)
+ }
+ if _, err := store.Put(ctx, key, raw, etag); errors.Is(err, release.ErrCASConflict) {
+ continue
+ } else if err != nil {
+ return release.BuildState{}, err
+ }
+ // A process is about to run against the database, so it has (or is
+ // about to have) a replica.
+ c := release.BuildStateCopy{Known: true, Replica: true, State: &next, ReadAt: b.now().UTC().Truncate(time.Second)}
+ if err := b.reg.SetBuildState(host, c.Column()); err != nil {
+ return next, fmt.Errorf("build state of %s written, registry copy not: %w", host, err)
+ }
+ return next, nil
}
- _, err = store.Put(ctx, key, raw, etag)
- return err
+ return release.BuildState{}, fmt.Errorf("build state of %s: %d tries lost the race: %w", host, updateTries, release.ErrCASConflict)
}
diff --git a/internal/host/buildstate_test.go b/internal/host/buildstate_test.go
index 281966eb5..2e7c0f1f6 100644
--- a/internal/host/buildstate_test.go
+++ b/internal/host/buildstate_test.go
@@ -11,6 +11,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -31,7 +32,7 @@ func testBuildStates(t *testing.T) (*buildStates, release.Store, *registry.Regis
if err != nil {
t.Fatal(err)
}
- if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "km", Hibernate: true}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/carlos/a.sock"}); err != nil {
t.Fatal(err)
}
s := release.NewDirStore(t.TempDir())
@@ -65,7 +66,7 @@ func TestRefreshSeesAReplicaAndAState(t *testing.T) {
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 {
+ if err := put(t, b, 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 {
@@ -82,7 +83,7 @@ func TestRefreshSeesAReplicaAndAState(t *testing.T) {
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 {
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
t.Fatal(err)
}
if err := b.refresh(ctx, "a.example.com"); err != nil {
@@ -99,7 +100,16 @@ func TestAFailedRefreshIsRecorded(t *testing.T) {
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 {
+ // update never writes over an object it cannot read: the merge would
+ // drop what that object recorded.
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
+ t.Fatal("an update wrote over an unparseable build state")
+ }
+ if c := copyOf(t, reg); c.State == nil || c.State.LastExecuted != "v1" {
+ t.Fatalf("copy = %+v; a refused update changed it", c)
+ }
+ // Repaired by hand, the next good read clears the failure.
+ if _, err := s.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","last_executed":"v2"}`), mustEtag(t, s, key)); err != nil {
t.Fatal(err)
}
if err := b.refresh(ctx, "a.example.com"); err != nil {
@@ -160,23 +170,38 @@ func TestAStateForAnotherHostIsAReadError(t *testing.T) {
}
}
-// refreshAll covers exactly the activator-managed exec instances.
-func TestRefreshAllCoversOnlyExecInstances(t *testing.T) {
+// refreshAll covers the activator-managed exec instances and the standalone
+// workers (ruling R-WORKER-ROWS): every row whose spawns run under a lease
+// in the replica bucket.
+func TestRefreshAllCoversExecInstancesAndWorkers(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},
+ {Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true},
+ {Host: "always.example.com", Kind: "sidecar", App: "km"},
+ {Host: "unit.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/u.sock", Unit: "km-unit"},
+ {Host: "backed.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/b.sock", Backing: "unit"},
+ {Host: "backedw.example.com", Kind: "sidecar", App: "km", Hibernate: true, Backing: "unit"},
+ {Host: "awake.example.com", Kind: "instance", App: "km", Socket: "/run/w.sock"},
+ {Host: "plat.example.com", Kind: "instance", App: "km", Hibernate: true, Socket: "/run/p.sock", Platform: true},
+ {Host: "svc.example.com", Kind: "service", App: "km", Hibernate: true, Socket: "/run/s.sock"},
} {
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)
+ if skipped := b.refreshAll(context.Background()); skipped != 0 {
+ t.Fatalf("skipped %d", skipped)
+ }
+ for _, h := range []string{"a.example.com", "w.example.com", "always.example.com"} {
+ rt, err := reg.Get(h)
+ if err != nil || rt == nil {
+ t.Fatalf("%s: %v", h, err)
+ }
+ if c, _ := release.ParseBuildStateCopy(rt.BuildState); !c.Known {
+ t.Fatalf("%s was not refreshed: %+v", h, c)
+ }
}
- for _, h := range []string{"unit.example.com", "awake.example.com", "plat.example.com"} {
+ for _, h := range []string{"unit.example.com", "backed.example.com", "backedw.example.com", "awake.example.com", "plat.example.com", "svc.example.com"} {
rt, err := reg.Get(h)
if err != nil || rt == nil {
t.Fatalf("%s: %v", h, err)
@@ -187,13 +212,90 @@ func TestRefreshAllCoversOnlyExecInstances(t *testing.T) {
}
}
+// 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) {
+ b, _, reg := testBuildStates(t)
+ ctx := context.Background()
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ before := copyOf(t, reg)
+ if err := reg.Upsert(registry.Route{Host: "w.example.com", Kind: "sidecar", App: "km", Hibernate: true}); err != nil {
+ t.Fatal(err)
+ }
+ b.budget = time.Nanosecond
+ if skipped := b.refreshAll(ctx); skipped != 2 {
+ t.Fatalf("skipped %d, want 2", skipped)
+ }
+ if after := copyOf(t, reg); after.Column() != before.Column() {
+ t.Fatalf("a host not reached changed its copy: %+v -> %+v", before, after)
+ }
+ if rt, _ := reg.Get("w.example.com"); rt.BuildState != "" {
+ t.Fatalf("a host not reached got a copy: %s", rt.BuildState)
+ }
+}
+
+// listCounter records how the replica was listed; hidePages hides the
+// bounded listing so the fallback is exercised.
+type listCounter struct {
+ release.Store
+ pages []int
+ full int
+}
+
+func (l *listCounter) List(ctx context.Context, prefix string) ([]string, error) {
+ l.full++
+ return l.Store.List(ctx, prefix)
+}
+
+type pagedCounter struct{ *listCounter }
+
+func (p pagedCounter) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
+ p.pages = append(p.pages, limit)
+ return p.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit)
+}
+
+// Ruling R-BS-REFRESH: three keys decide the replica, so a store with a
+// bounded listing is asked for three, never for the whole prefix; one
+// without it still answers through List.
+func TestTheReplicaIsDecidedByABoundedListing(t *testing.T) {
+ ctx := context.Background()
+ dir := release.NewDirStore(t.TempDir())
+ for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json", "inst/a/generations/0000/a.ltx", "inst/a/generations/0000/b.ltx", "inst/a/generations/0001/c.ltx"} {
+ if _, err := dir.Put(ctx, k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ lc := &listCounter{Store: dir}
+ paged := pagedCounter{lc}
+ if ok, err := hasReplica(ctx, paged, "inst/a"); err != nil || !ok {
+ t.Fatalf("paged: %v, %v", ok, err)
+ }
+ if lc.full != 0 || len(lc.pages) != 1 || lc.pages[0] != 3 {
+ t.Fatalf("the whole prefix was listed %d times; pages asked %v, want [3]", lc.full, lc.pages)
+ }
+ if ok, err := hasReplica(ctx, lc, "inst/a"); err != nil || !ok || lc.full != 1 {
+ t.Fatalf("fallback: %v, %v, %d lists", ok, err, lc.full)
+ }
+ lockOnly := release.NewDirStore(t.TempDir())
+ for _, k := range []string{"inst/a/build-state.json", "inst/a/lock.json"} {
+ if _, err := lockOnly.Put(ctx, k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if ok, err := hasReplica(ctx, lockOnly, "inst/a"); err != nil || ok {
+ t.Fatalf("lock and state alone: %v, %v", ok, err)
+ }
+}
+
// --- 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 {
+ if err := put(t, b, 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 {
@@ -211,7 +313,7 @@ func TestAContextLocationWins(t *testing.T) {
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 {
+ if err := put(t, b, 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 {
@@ -244,7 +346,7 @@ func TestClaimsOnUsesTheHolderLocationInTheActorBucket(t *testing.T) {
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 {
+ if err := put(t, b, 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 {
@@ -292,7 +394,7 @@ func TestAnUnreadableClaimIsAReadError(t *testing.T) {
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 {
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
t.Fatal(err)
}
if err := b.refresh(ctx, "a.example.com"); err != nil {
@@ -309,7 +411,7 @@ func TestAnUnreadableClaimIsAReadError(t *testing.T) {
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 {
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v2"}); err == nil {
t.Fatal("a claimed host's build state was written without its claim")
}
})
@@ -338,3 +440,170 @@ func TestStoresAreOpenedPerBucketAndCached(t *testing.T) {
t.Fatal("a bucket the box does not lease in was opened")
}
}
+
+func put(t *testing.T, b *buildStates, ctx context.Context, host string, s release.BuildState) error {
+ t.Helper()
+ _, err := b.update(ctx, host, func(*release.BuildState) release.BuildState { return s })
+ return err
+}
+
+// --- update (ruling R-BS-UPDATE) ---
+
+// bump is a merge that keeps what it read and raises Completed by one.
+func bump(cur *release.BuildState) release.BuildState {
+ next := release.BuildState{}
+ if cur != nil {
+ next = *cur
+ }
+ next.Completed++
+ return next
+}
+
+// Concurrent updates for one host (an instance spawn and its worker or
+// sidecar spawns, under the same lease) never lose a merge.
+func TestConcurrentUpdatesKeepEveryMerge(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ ctx := context.Background()
+ const n = 8
+ errs := make(chan error, n)
+ for i := 0; i < n; i++ {
+ go func() {
+ _, err := b.update(ctx, "a.example.com", bump)
+ errs <- err
+ }()
+ }
+ for i := 0; i < n; i++ {
+ if err := <-errs; err != nil {
+ t.Fatal(err)
+ }
+ }
+ st, _, err := b.read(ctx, "a.example.com")
+ if err != nil || st == nil || st.Completed != n {
+ t.Fatalf("state = %+v, %v; want Completed %d", st, err, n)
+ }
+ if c := copyOf(t, reg); !c.Known || !c.Replica || c.State == nil || c.State.Completed != n {
+ t.Fatalf("copy = %+v", c)
+ }
+}
+
+// A write that lands between update's read and its Put (another process)
+// is not overwritten: update re-reads and merges again.
+func TestAnUpdateRacedBetweenReadAndPutMergesAgain(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ ctx := context.Background()
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1", Completed: 1}); err != nil {
+ t.Fatal(err)
+ }
+ raced := false
+ b.beforePut = func(ctx context.Context, s release.Store, key string) {
+ if raced {
+ return
+ }
+ raced = true
+ other := []byte(`{"schema":1,"host":"a.example.com","last_executed":"v1","resolved":{"id":7,"target":"v3"},"completed":6}`)
+ if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ merges := 0
+ got, err := b.update(ctx, "a.example.com", func(cur *release.BuildState) release.BuildState {
+ merges++
+ next := *cur
+ next.LastExecuted = "v2"
+ return next
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if merges != 2 || got.LastExecuted != "v2" || got.Completed != 6 || got.Resolved == nil || got.Resolved.ID != 7 {
+ t.Fatalf("merges %d, got %+v; want the raced write's Completed and Resolved kept", merges, got)
+ }
+ st, _, err := b.read(ctx, "a.example.com")
+ if err != nil || st == nil || st.Completed != 6 || st.LastExecuted != "v2" {
+ t.Fatalf("stored %+v, %v", st, err)
+ }
+ if c := copyOf(t, reg); c.State == nil || c.State.Completed != 6 || c.State.LastExecuted != "v2" {
+ t.Fatalf("copy = %+v", c)
+ }
+}
+
+// A conflict on every try is an error, and the copy keeps what it had.
+func TestAPersistentConflictIsAnError(t *testing.T) {
+ b, _, reg := testBuildStates(t)
+ ctx := context.Background()
+ if err := put(t, b, ctx, "a.example.com", release.BuildState{LastExecuted: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ before := copyOf(t, reg)
+ n := 0
+ b.beforePut = func(ctx context.Context, s release.Store, key string) {
+ n++
+ other := []byte(fmt.Sprintf(`{"schema":1,"host":"a.example.com","last_executed":"v1","completed":%d}`, n))
+ if _, err := s.Put(ctx, key, other, mustEtag(t, s, key)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if _, err := b.update(ctx, "a.example.com", bump); !errors.Is(err, release.ErrCASConflict) {
+ t.Fatalf("err = %v; want a CAS conflict", err)
+ }
+ if n != updateTries {
+ t.Fatalf("tried %d times, want %d", n, updateTries)
+ }
+ if after := copyOf(t, reg); after.Column() != before.Column() {
+ t.Fatalf("a lost update changed the copy: %+v -> %+v", before, after)
+ }
+}
+
+// The first update on a host with no object is a create-only write.
+func TestTheFirstUpdateCreates(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ ctx := context.Background()
+ b.beforePut = func(ctx context.Context, st release.Store, key string) {
+ if _, _, err := st.Get(ctx, key); err == nil {
+ return
+ }
+ if _, err := st.Put(ctx, key, []byte(`{"schema":1,"host":"a.example.com","completed":4}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ got, err := b.update(ctx, "a.example.com", bump)
+ if err != nil || got.Completed != 5 {
+ t.Fatalf("got %+v, %v; want the created object merged (Completed 5)", got, err)
+ }
+ if _, _, err := s.Get(ctx, "inst/a.example.com/build-state.json"); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// Minor (fix round 1): with host claims on only the claim bucket is
+// registered, so a hibernate bucket that is the same bucket with no region
+// set cannot fail the box.
+func TestClaimsOnRegistersOnlyTheClaimBucket(t *testing.T) {
+ read := func(context.Context, string) (hostclaim.Observed, error) { return hostclaim.Observed{}, nil }
+ cfg := &activator.Config{Bucket: "data", Region: "", Prefix: "hib",
+ HostWrites: hostwrite.Config{Enabled: true},
+ Claims: &activator.ClaimRuntime{Bucket: "data", LeaseRegion: "eu-west-1", Read: read}}
+ regions, err := buildStateRegions(cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(regions) != 1 || regions["data"] != "eu-west-1" {
+ t.Fatalf("regions = %v", regions)
+ }
+ b, err := newBuildStates(cfg, nil, "")
+ if err != nil || b == nil || b.claimBucket != "data" || b.claims == nil {
+ t.Fatalf("newBuildStates = %+v, %v", b, err)
+ }
+ cfg.Claims.Bucket = "other"
+ if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["other"] != "eu-west-1" {
+ t.Fatalf("regions = %v, %v; want only the claim bucket", regions, err)
+ }
+ cfg.HostWrites.Enabled = false
+ if regions, err := buildStateRegions(cfg); err != nil || len(regions) != 1 || regions["data"] != "" {
+ t.Fatalf("claims off: regions = %v, %v", regions, err)
+ }
+ cfg.HostWrites.Enabled, cfg.Claims = true, nil
+ if _, err := buildStateRegions(cfg); err == nil {
+ t.Fatal("claims on with no claim runtime was accepted")
+ }
+}