amadan

carlos / platform Public

hostclaim, park: staging scopes moves/<host>/<op id> and moves/<host>, evidence_at as the original bound (claim rev 6.23, joint round 15 R15-1)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude bd84260e4c839ee41d05ab8a37c5d99e39733200 parent edc16d9
6 files changed, +454 −20
  • internal/hostclaim/claim.go +18 −5
  • internal/hostclaim/park_test.go +87 −0
  • internal/park/moves_test.go +204 −0
  • internal/park/remote.go +24 −14
  • internal/park/sweep.go +7 −1
  • internal/park/sweep_test.go +114 −0
diff --git a/internal/hostclaim/claim.go b/internal/hostclaim/claim.go
index 7280b2ea5..825556541 100644
--- a/internal/hostclaim/claim.go
+++ b/internal/hostclaim/claim.go
@@ -11,6 +11,7 @@ import (
"encoding/hex"
"errors"
"fmt"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/controlname"
@@ -112,6 +113,20 @@ type ParkScope struct {
RequestLimit time.Duration `json:"request_limit_ns"`
}
+func (p *ParkScope) Validate(host string) error {
+ if p == nil || !isHex64(p.Incarnation) || p.Incarnation != strings.ToLower(p.Incarnation) || p.EvidenceAt.IsZero() || p.RequestLimit <= 0 || p.RequestLimit > time.Duration(1<<63-1)-15*time.Minute {
+ return fmt.Errorf("%w: park scope", ErrMalformed)
+ }
+ base, staging := "instances/"+host, "moves/"+host
+ if p.Location == base || p.Location == base+"/"+p.Incarnation || p.Location == staging {
+ return nil
+ }
+ if op, ok := strings.CutPrefix(p.Location, staging+"/"); ok && op != "" && op != "." && op != ".." && !strings.Contains(op, "/") {
+ return nil
+ }
+ return fmt.Errorf("%w: park location", ErrMalformed)
+}
+
type Obligation struct {
Park *ParkScope `json:"park,omitempty"`
ID string `json:"id"`
@@ -247,13 +262,11 @@ func (c *Claim) Validate() error {
return fmt.Errorf("%w: obligation ids", ErrMalformed)
}
if o.Park != nil {
- p := o.Park
- if (o.Kind != "park-replica" && o.Kind != "park-generation") || !isHex64(p.Incarnation) || p.EvidenceAt.IsZero() || p.RequestLimit <= 0 || p.RequestLimit > time.Duration(1<<63-1)-15*time.Minute {
+ if o.Kind != "park-replica" && o.Kind != "park-generation" {
return fmt.Errorf("%w: park scope", ErrMalformed)
}
- base := "instances/" + c.Host
- if p.Location != base && p.Location != base+"/"+p.Incarnation {
- return fmt.Errorf("%w: park location", ErrMalformed)
+ if err := o.Park.Validate(c.Host); err != nil {
+ return err
}
}
seen[o.ID] = true
diff --git a/internal/hostclaim/park_test.go b/internal/hostclaim/park_test.go
index 74a6f914b..3ca5fc2b9 100644
--- a/internal/hostclaim/park_test.go
+++ b/internal/hostclaim/park_test.go
@@ -1,8 +1,12 @@
package hostclaim
import (
+ "errors"
+ "strings"
"testing"
"time"
+
+ "github.com/carlosframework/platform/internal/release"
)
func TestParkScopeAndHolderlessSweep(t *testing.T) {
@@ -30,3 +34,86 @@ func TestParkScopeAndHolderlessSweep(t *testing.T) {
}
}
}
+
+func TestParkScopeLocationGrammar(t *testing.T) {
+ host := "shop.bab.example.com"
+ inc := strings.Repeat("a", 64)
+ instances, moves := "instances/"+host, "moves/"+host
+ for _, tc := range []struct {
+ location string
+ valid bool
+ }{
+ {instances, true},
+ {instances + "/" + inc, true},
+ {moves, true},
+ {moves + "/move-1", true},
+ {moves + "/" + strings.Repeat("b", 64), true},
+ {instances + "/", false},
+ {instances + "/" + inc + "/", false},
+ {instances + "/" + inc + "/x", false},
+ {instances + "/" + strings.Repeat("b", 64), false},
+ {moves + "/", false},
+ {moves + "/move-1/", false},
+ {moves + "/move-1/x", false},
+ {moves + "/.", false},
+ {moves + "/..", false},
+ {moves + "//move-1", false},
+ {"moves/other.example/move-1", false},
+ {"instances/other.example", false},
+ {"other/" + host, false},
+ } {
+ t.Run(tc.location, func(t *testing.T) {
+ c := &Claim{V: 2, Host: host, Gen: 1, Op: Op{Kind: KindParking, ID: "sweep", Intent: map[string]string{"park_sweep": "v1"}}}
+ c.Obligations = []Obligation{{ID: "ob", Op: "sweep", Kind: "park-replica", Region: "ie", Retirement: "rt", Owner: "region:ie", Park: &ParkScope{Location: tc.location, Incarnation: inc, EvidenceAt: time.Now(), RequestLimit: time.Second}}}
+ err := c.Validate()
+ if tc.valid && err != nil {
+ t.Fatalf("valid scope rejected: %v", err)
+ }
+ if !tc.valid && !errors.Is(err, ErrMalformed) {
+ t.Fatalf("invalid scope: %v", err)
+ }
+ })
+ }
+ for _, location := range []string{instances, instances + "/" + strings.ToUpper(inc), moves, moves + "/move-1"} {
+ t.Run("uppercase-archive/"+location, func(t *testing.T) {
+ c := &Claim{V: 2, Host: host, Gen: 1, Op: Op{Kind: KindParking, ID: "sweep", Intent: map[string]string{"park_sweep": "v1"}}}
+ c.Obligations = []Obligation{{ID: "ob", Op: "sweep", Kind: "park-replica", Region: "ie", Retirement: "rt", Owner: "region:ie", Park: &ParkScope{Location: location, Incarnation: strings.ToUpper(inc), EvidenceAt: time.Now(), RequestLimit: time.Second}}}
+ if err := c.Validate(); !errors.Is(err, ErrMalformed) {
+ t.Fatalf("uppercase archive incarnation accepted: %v", err)
+ }
+ })
+ }
+}
+
+func TestHolderlessSweepJoinsRegion(t *testing.T) {
+ host := "shop.bab.example.com"
+ st := release.NewDirStore(t.TempDir())
+ chore := Obligation{ID: "ie-ob", Op: "sweep", Kind: "park-replica", Region: "ie", Retirement: "ie-rt", Owner: "region:ie", Park: &ParkScope{Location: "moves/" + host, Incarnation: strings.Repeat("a", 64), EvidenceAt: time.Now(), RequestLimit: time.Second}}
+ obs, err := Update(t.Context(), st, Observed{Host: host, Absent: true}, func(c *Claim) error {
+ c.Op = Op{Kind: KindParking, ID: "sweep", Intent: map[string]string{"park_sweep": "v1"}}
+ c.Obligations = []Obligation{chore}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ joined := chore
+ joined.ID, joined.Region, joined.Retirement, joined.Owner = "uk-ob", "uk", "uk-rt", "region:uk"
+ obs, err = Update(t.Context(), st, obs, func(c *Claim) error {
+ c.Obligations = append(c.Obligations, joined)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if obs.Claim.Free() || obs.Claim.Holder != nil || obs.Claim.Op.ID != "sweep" || len(obs.Claim.Obligations) != 2 {
+ t.Fatalf("invalid joined sweep: %+v", obs.Claim)
+ }
+ _, err = Update(t.Context(), st, obs, func(c *Claim) error {
+ c.Obligations[1].Op = "other-sweep"
+ return nil
+ })
+ if !errors.Is(err, ErrMalformed) {
+ t.Fatalf("joined region with a different op accepted: %v", err)
+ }
+}
diff --git a/internal/park/moves_test.go b/internal/park/moves_test.go
new file mode 100644
index 000000000..92ba9d2ea
--- /dev/null
+++ b/internal/park/moves_test.go
@@ -0,0 +1,204 @@
+package park
+
+import (
+ "context"
+ "errors"
+ "net/url"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+func movesFixture(t *testing.T, op string) *fixture {
+ t.Helper()
+ f := newFixture(t)
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.ob.Park.Location = "moves/" + f.w.Host
+ if op != "" {
+ f.ob.Park.Location += "/" + op
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
+ c.Holder.Generation = "live"
+ c.Obligations = []hostclaim.Obligation{f.ob}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: "unrelated", GenerationOf: func(Ref) (string, bool) { return "live", true }}, nil
+ }
+ return f
+}
+
+func TestMovesScopeParksExactPrefixAndResumes(t *testing.T) {
+ for _, op := range []string{"move-1", strings.Repeat("b", 64), ""} {
+ t.Run("op="+op, func(t *testing.T) {
+ f := movesFixture(t, op)
+ p := f.ob.Park.Location + "/"
+ expected := map[Ref]string{}
+ for _, name := range []string{"return/data", "other-op/data"} {
+ for _, body := range []string{"old", "new"} {
+ id := f.b.Put("data", p+name, []byte(body))
+ expected[Ref{p + name, id}] = body
+ }
+ }
+ f.b.BareDelete("data", p+"return/data")
+ lock := Ref{p + "lock.json", f.b.Put("data", p+"lock.json", []byte("lease"))}
+ untouched := map[Ref]string{lock: "lease"}
+ keys := []string{"instances/" + f.w.Host + "/0000/data", "instances/" + f.w.Host + "/" + f.ob.Park.Incarnation + "/live/data", "moves/other.example/move-1/data", "moves/" + f.w.Host + "-other/move-1/data"}
+ if op != "" {
+ keys = append(keys, f.ob.Park.Location+"-sibling/data", "moves/"+f.w.Host+"/sibling/data")
+ }
+ for _, key := range keys {
+ untouched[Ref{key, f.b.Put("data", key, []byte("untouched"))}] = "untouched"
+ }
+ listings := 0
+ f.b.FailOn = func(action, key string) error {
+ if action == "list" && !strings.HasPrefix(key, "parked/") {
+ listings++
+ if key != p {
+ t.Errorf("listing %q, want only %q", key, p)
+ return ErrUnsafe
+ }
+ }
+ return nil
+ }
+ f.e.After = func(step string) error {
+ if step == "delete" {
+ return ErrPending
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
+ t.Fatalf("interrupted retirement: %v", err)
+ }
+ f.e.After = nil
+ checkListing := f.b.FailOn
+ f.b.FailOn = func(action, key string) error {
+ if action == "copy" {
+ t.Error("manifest retry copied a source")
+ return ErrUnsafe
+ }
+ return checkListing(action, key)
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatalf("resumed retirement: %v", err)
+ }
+ if listings != 2 || f.held.released != 1 {
+ t.Fatalf("listings=%d releases=%d", listings, f.held.released)
+ }
+ f.done(t)
+ m, err := f.e.manifest(t.Context(), f.w, f.ob)
+ if err != nil || m == nil || len(m.Archive) != len(expected) || len(m.Delete) != len(expected)+1 {
+ t.Fatalf("manifest: %+v %v", m, err)
+ }
+ for _, copied := range m.Archive {
+ body, ok := expected[ref(copied.Source)]
+ wantKey := "parked/" + f.ob.Park.Incarnation + "/" + f.w.Host + "/" + f.ob.Retirement + "/moves/" + strings.TrimPrefix(copied.Source.Key, p) + "@" + url.QueryEscape(copied.Source.VersionID)
+ if !ok || copied.Destination.Key != wantKey {
+ t.Fatalf("unexpected archive copy: %+v", copied)
+ }
+ got, err := f.b.GetVersion(t.Context(), "data", copied.Destination.Key, copied.Destination.VersionID)
+ if err != nil || string(got) != body {
+ t.Fatalf("archive body=%q, want %q: %v", got, body, err)
+ }
+ }
+ for _, source := range m.Delete {
+ if _, err := f.b.GetVersion(t.Context(), "data", source.Key, source.VersionID); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("source remains: %+v %v", source, err)
+ }
+ }
+ for r, body := range untouched {
+ got, err := f.b.GetVersion(t.Context(), "data", r.Key, r.VersionID)
+ if err != nil || string(got) != body {
+ t.Fatalf("unrelated version changed: %+v %q %v", r, got, err)
+ }
+ }
+ })
+ }
+}
+
+func TestMovesScopeUsesPersistedEvidenceHorizon(t *testing.T) {
+ for _, op := range []string{"move-1", ""} {
+ for _, offset := range []time.Duration{-time.Minute, 0, time.Hour} {
+ t.Run("op="+op+"/offset="+offset.String(), func(t *testing.T) {
+ f := movesFixture(t, op)
+ f.b.Put("data", f.ob.Park.Location+"/staging/data", []byte("staged"))
+ deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + f.ob.Park.RequestLimit)
+ f.now = deadline.Add(offset)
+ f.e.After = func(step string) error {
+ if step == "verify" {
+ f.now = deadline.Add(2 * time.Hour)
+ }
+ return nil
+ }
+ err := f.e.Run(t.Context(), f.w)
+ if offset > 0 {
+ if err != nil {
+ t.Fatalf("delayed cleanup did not acknowledge immediately: %v", err)
+ }
+ } else {
+ if !errors.Is(err, ErrPending) {
+ t.Fatalf("early verification: %v", err)
+ }
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(obs.Claim.Obligations) != 1 || len(obs.Claim.QuietAfter) != 0 || f.held.released != 0 {
+ t.Fatal("acknowledged before a listing after the horizon")
+ }
+ if !reflect.DeepEqual(obs.Claim.Obligations[0].Park, f.ob.Park) {
+ t.Fatal("persisted evidence changed")
+ }
+ f.e.After = nil
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatalf("later verification: %v", err)
+ }
+ }
+ f.done(t)
+ })
+ }
+ }
+}
+
+func TestBindMovesScope(t *testing.T) {
+ f := newFixture(t)
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations[0].Park = nil
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ p := *f.ob.Park
+ p.Location = "moves/" + f.w.Host + "/move-1"
+ authorized := false
+ f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
+ authorized = reflect.DeepEqual(o.Park, &p)
+ return nil
+ }
+ if err := f.e.Bind(t.Context(), f.w, p); err != nil {
+ t.Fatal(err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !authorized || !reflect.DeepEqual(obs.Claim.Obligations[0].Park, &p) {
+ t.Fatal("staging scope not authorized and persisted")
+ }
+}
diff --git a/internal/park/remote.go b/internal/park/remote.go
index edf4b88bd..225596dac 100644
--- a/internal/park/remote.go
+++ b/internal/park/remote.go
@@ -92,11 +92,7 @@ func find(c *hostclaim.Claim, w Work) (hostclaim.Obligation, bool) {
}
func validateScope(host string, o hostclaim.Obligation) error {
p := o.Park
- if p == nil || !hex64(p.Incarnation) || p.EvidenceAt.IsZero() || p.RequestLimit <= 0 || p.RequestLimit > time.Duration(1<<63-1)-SignatureAge || !token(o.Retirement) || o.Region == "" || o.Owner != "region:"+o.Region {
- return ErrUnsafe
- }
- base := "instances/" + host
- if p.Location != base && p.Location != base+"/"+p.Incarnation {
+ if p.Validate(host) != nil || !token(o.Retirement) || o.Region == "" || o.Owner != "region:"+o.Region {
return ErrUnsafe
}
if o.Kind != "park-replica" && o.Kind != "park-generation" {
@@ -126,7 +122,7 @@ func (e Engine) Bind(ctx context.Context, w Work, p hostclaim.ParkScope) error {
}
return ErrChanged
}
- if obs.Claim.Holder == nil || p.Location != obs.Claim.Holder.Location || p.Incarnation != obs.Claim.Holder.Incarnation {
+ if obs.Claim.Holder == nil || (p.Location != obs.Claim.Holder.Location && !isMovesScope(w.Host, p.Location)) || p.Incarnation != obs.Claim.Holder.Incarnation {
return ErrUnsafe
}
o.Park = &p
@@ -150,7 +146,14 @@ func (e Engine) Bind(ctx context.Context, w Work, p hostclaim.ParkScope) error {
})
return err
}
+func isMovesScope(host, location string) bool {
+ base := "moves/" + host
+ return location == base || strings.HasPrefix(location, base+"/")
+}
func prefixes(host, location string) []string {
+ if isMovesScope(host, location) {
+ return []string{location + "/"}
+ }
suffix := strings.TrimPrefix(location, "instances/"+host)
return []string{location + "/", "moves/" + host + suffix + "/"}
}
@@ -175,6 +178,13 @@ func locationOf(host, key string) (string, error) {
}
return "", ErrUnsafe
}
+func scopeContains(host, location, key string) (bool, error) {
+ if isMovesScope(host, location) {
+ return strings.HasPrefix(key, location+"/"), nil
+ }
+ loc, err := locationOf(host, key)
+ return loc == location, err
+}
func isLock(host, location, key string) bool {
for _, p := range prefixes(host, location) {
if key == p+"lock.json" {
@@ -195,12 +205,12 @@ func (e Engine) list(ctx context.Context, host, location string) ([]lease.Object
seen := map[Ref]bool{}
filtered := out[:0]
for _, v := range out {
- l, err := locationOf(host, v.Key)
+ inScope, err := scopeContains(host, location, v.Key)
if err != nil || v.VersionID == "" || seen[ref(v)] {
return nil, ErrUnsafe
}
seen[ref(v)] = true
- if l == location && !isLock(host, location, v.Key) {
+ if inScope && !isLock(host, location, v.Key) {
filtered = append(filtered, v)
}
}
@@ -268,8 +278,8 @@ func (e Engine) manifest(ctx context.Context, w Work, o hostclaim.Obligation) (*
}
deletes := map[Ref]lease.ObjectVersion{}
for _, v := range m.Delete {
- l, err := locationOf(w.Host, v.Key)
- if err != nil || l != o.Park.Location || v.VersionID == "" || isLock(w.Host, l, v.Key) {
+ inScope, err := scopeContains(w.Host, o.Park.Location, v.Key)
+ if err != nil || !inScope || v.VersionID == "" || isLock(w.Host, o.Park.Location, v.Key) {
return nil, ErrUnsafe
}
if _, ok := deletes[ref(v)]; ok {
@@ -327,14 +337,14 @@ func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, he
}
out := make([]lease.ObjectVersion, 0, len(vs))
for _, x := range vs {
- loc, err := locationOf(w.Host, x.Key)
- if err != nil || loc != o.Park.Location || x.VersionID == "" {
+ inScope, err := scopeContains(w.Host, o.Park.Location, x.Key)
+ if err != nil || !inScope || x.VersionID == "" {
return obs, nil, ErrUnsafe
}
- if isLock(w.Host, loc, x.Key) {
+ if isLock(w.Host, o.Park.Location, x.Key) {
return obs, nil, ErrUnsafe
}
- if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && loc == obs.Claim.Holder.Location && v.Region == v.AuthoritativeRegion {
+ if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && o.Park.Location == obs.Claim.Holder.Location && v.Region == v.AuthoritativeRegion {
if x.Key == v.Marker.Key {
if o.Kind == "park-generation" {
continue
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index 2f468b2ae..da3886601 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -113,7 +113,8 @@ func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, lega
if err := c.Validate(); err != nil {
return d, err
}
- if !c.Free() && c.Op.Kind != hostclaim.KindServing {
+ joining := c.Holder == nil && c.Op.Kind == hostclaim.KindParking && c.Op.Intent["park_sweep"] == "v1"
+ if !c.Free() && c.Op.Kind != hostclaim.KindServing && !joining {
return d, nil
}
for _, o := range c.Obligations {
@@ -132,6 +133,9 @@ func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, lega
if err != nil || x.VersionID == "" || seen[ref(x)] {
return Decision{}, ErrUnsafe
}
+ if strings.HasPrefix(x.Key, "moves/"+c.Host+"/") {
+ loc = "moves/" + c.Host
+ }
seen[ref(x)] = true
if isLock(c.Host, loc, x.Key) {
continue
@@ -220,6 +224,8 @@ func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evide
var op string
if obs.Claim.Holder != nil {
op = fmt.Sprintf("session:%d", obs.Claim.Holder.Session)
+ } else if obs.Claim.Op.Kind == hostclaim.KindParking {
+ op = obs.Claim.Op.ID
} else {
op = e.NewID()
if !token(op) {
diff --git a/internal/park/sweep_test.go b/internal/park/sweep_test.go
index b6fdf9fdc..23389ffa1 100644
--- a/internal/park/sweep_test.go
+++ b/internal/park/sweep_test.go
@@ -3,12 +3,14 @@ package park
import (
"context"
"errors"
+ "reflect"
"strings"
"testing"
"time"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
)
func sweepFixture(t *testing.T) (*fixture, hostclaim.Observed, View, []lease.ObjectVersion) {
@@ -51,6 +53,118 @@ func sweepFixture(t *testing.T) (*fixture, hostclaim.Observed, View, []lease.Obj
}
return f, obs, v, vs
}
+
+func TestSweepMovesNamespaceIsSeparateFromLiveInstance(t *testing.T) {
+ f := newFixture(t)
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations = nil
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ liveKey := f.ob.Park.Location + "/0000/live"
+ liveID := f.b.Put("data", liveKey, []byte("live"))
+ for _, op := range []string{"move-1", f.ob.Park.Incarnation} {
+ f.b.Put("data", "moves/"+f.w.Host+"/"+op+"/staging", []byte(op))
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
+ }
+ d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []Candidate{{Location: "moves/" + f.w.Host, Incarnation: f.ob.Park.Incarnation}}
+ if !reflect.DeepEqual(d.Candidates, want) {
+ t.Fatalf("staging candidates=%+v, want %+v", d.Candidates, want)
+ }
+ vs, err := f.b.ListVersions(t.Context(), "data", "moves/"+f.w.Host+"/")
+ if err != nil || len(vs) != 0 {
+ t.Fatalf("staging residue: %+v %v", vs, err)
+ }
+ body, err := f.b.GetVersion(t.Context(), "data", liveKey, liveID)
+ if err != nil || string(body) != "live" {
+ t.Fatalf("live instance changed: %q %v", body, err)
+ }
+}
+
+func TestSweepJoinsRunningHolderlessOperation(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations = nil
+ return hostclaim.FreeAfterPark("op")(c)
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.held.failed = true
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
+ t.Fatalf("first region queue: %v", err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ first := obs.Claim.Obligations[0]
+ second := f.e
+ second.Region = "uk"
+ b := leasetest.NewVersionedBucket()
+ b.AddBucket("data")
+ b.Put("data", f.ob.Park.Location+"/0000/data", []byte("second region"))
+ second.Store = b
+ second.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "uk"}, nil
+ }
+ second.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
+ if o.Op != first.Op || o.Region != "uk" {
+ return ErrUnsafe
+ }
+ return nil
+ }
+ if _, err := second.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, f.ob.Park.RequestLimit); !errors.Is(err, ErrChanged) {
+ t.Fatalf("second region did not join the running sweep: %v", err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if obs.Claim.Holder != nil || obs.Claim.Op.ID != first.Op || len(obs.Claim.Obligations) != 2 || !reflect.DeepEqual(obs.Claim.Obligations[0], first) {
+ t.Fatalf("joined sweep: %+v", obs.Claim)
+ }
+ joined := obs.Claim.Obligations[1]
+ f.held.failed = false
+ if err := f.e.Run(t.Context(), Work{f.w.Host, first.ID, first.Op}); err != nil {
+ t.Fatal(err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || obs.Claim.Free() || len(obs.Claim.Obligations) != 1 {
+ t.Fatalf("first completion freed an outstanding sweep: %+v %v", obs.Claim, err)
+ }
+ if err := second.Run(t.Context(), Work{f.w.Host, joined.ID, joined.Op}); err != nil {
+ t.Fatal(err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || !obs.Claim.Free() {
+ t.Fatalf("last completion did not free the sweep: %+v %v", obs.Claim, err)
+ }
+ deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + f.ob.Park.RequestLimit)
+ for _, region := range []string{"ie", "uk"} {
+ if !obs.Claim.QuietAfter[region].Equal(deadline) {
+ t.Fatalf("quiet_after[%s]=%v, want %v", region, obs.Claim.QuietAfter[region], deadline)
+ }
+ }
+}
func TestSweepProtectsUnfinishedRunAndPins(t *testing.T) {
f, obs, v, vs := sweepFixture(t)
d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)