| index 10289f532..5f2de290a 100644 |
| --- a/internal/host/movessources_test.go |
| +++ b/internal/host/movessources_test.go |
| @@ -4,7 +4,9 @@ import ( |
| "context" |
| "encoding/json" |
| "errors" |
| + "sort" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| |
| @@ -20,8 +22,9 @@ import ( |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| |
| -// The moves park providers (Plan 5 Task 9a): the view, the lease set, the |
| -// claim Plan 12 guard, the registration and the arming. |
| +// The moves park providers (Plan 5 Task 9a): the view, the claim's lease set |
| +// over moves' scopes, the wake-source gate (Q1), the registration and the |
| +// arming from the claim's MoveParks. |
| |
| const ( |
| vwHost = "vw.example" |
| @@ -280,19 +283,29 @@ func TestMoveParkViewReadsTheMarkerByVersion(t *testing.T) { |
| }) |
| } |
| |
| -// TestMovesLeaseSetCoversTheScope (M6): the scope's own lock, for the four |
| -// scope shapes moves parks, in this box's own bucket only; acquired through |
| -// the real AggregateLeases it takes exactly that lock. |
| -func TestMovesLeaseSetCoversTheScope(t *testing.T) { |
| +// TestMovesScopesResolveThroughTheClaimsLeaseSet (M6): the claim's lease set |
| +// (hostbox.ReplicaParkLeaseSet, the resolver of the MoveParks engine's |
| +// AggregateLeases) answers every scope a moves park names with that scope's |
| +// own lock, and acquired through the real AggregateLeases it takes exactly |
| +// that lock. The scopes are moves' own: the holder location, the residue |
| +// park's, the lane and the op's lane. |
| +func TestMovesScopesResolveThroughTheClaimsLeaseSet(t *testing.T) { |
| inc := strings.Repeat("a", 64) |
| - set := movesLeaseSet("data-ie") |
| - for _, scope := range []string{"instances/" + vwHost, "instances/" + vwHost + "/" + inc, "moves/" + vwHost, "moves/" + vwHost + "/" + vwMove} { |
| - names, err := set(t.Context(), "data-ie", scope) |
| + holder := hostclaim.Holder{Location: "instances/" + vwHost + "/" + inc} |
| + scopes := []string{ |
| + "instances/" + vwHost, |
| + strings.TrimSuffix(holder.Location, "/"), |
| + moves.ResidueScope(holder, vwCreated).Location, |
| + "moves/" + vwHost, |
| + moves.ScopeLocation(vwHost, vwMove), |
| + } |
| + for _, scope := range scopes { |
| + names, err := hostbox.ReplicaParkLeaseSet(t.Context(), "data-ie", scope) |
| if err != nil || len(names) != 1 || names[0] != scope { |
| t.Fatalf("%s: %v %v", scope, names, err) |
| } |
| leases := &fakeParkLeases{} |
| - held, err := hostbox.AggregateLeases{Leases: leases, Resolve: set}.Acquire(t.Context(), "data-ie", scope) |
| + held, err := hostbox.AggregateLeases{Leases: leases, Resolve: hostbox.ReplicaParkLeaseSet}.Acquire(t.Context(), "data-ie", scope) |
| if err != nil { |
| t.Fatalf("%s: %v", scope, err) |
| } |
| @@ -301,47 +314,128 @@ func TestMovesLeaseSetCoversTheScope(t *testing.T) { |
| } |
| held.Close() |
| } |
| - if _, err := set(t.Context(), "data-uk", "instances/"+vwHost); !errors.Is(err, hostbox.ErrChanged) { |
| - t.Fatalf("another bucket: %v", err) |
| + for _, scope := range []string{"", "instances", "instances/" + vwHost + "/g-cur", "instances/" + vwHost + "/" + strings.ToUpper(inc), |
| + "moves/" + vwHost + "/x/y", "parked/" + vwHost, "instances/../" + vwHost} { |
| + if names, err := hostbox.ReplicaParkLeaseSet(t.Context(), "data-ie", scope); err == nil { |
| + t.Fatalf("%q answered %v", scope, names) |
| + } |
| } |
| - for _, scope := range []string{"", "instances", "instances/" + vwHost + "/g-cur", "instances/" + vwHost + "/" + strings.ToUpper(inc), "moves/" + vwHost + "/x/y", |
| - "moves/" + vwHost + "/not an op", "parked/" + vwHost, "instances/../" + vwHost} { |
| - if names, err := set(t.Context(), "data-ie", scope); !errors.Is(err, park.ErrUnsafe) { |
| - t.Fatalf("%q answered %v %v", scope, names, err) |
| +} |
| + |
| +// withMovesWakeSource empties the claim's move-source slots and the moves |
| +// gate's own record for the test, restoring all three after it. register |
| +// then registers a stand-in source through registerMovesWakeSource, which |
| +// opens the gate (movesWakeSourceRegistered). |
| +func withMovesWakeSource(t *testing.T, register bool) { |
| + t.Helper() |
| + claimMoveSources.Lock() |
| + oldSource, oldParks := claimMoveSources.Source, claimMoveSources.Parks |
| + claimMoveSources.Source, claimMoveSources.Parks = nil, nil |
| + claimMoveSources.Unlock() |
| + movesWakeSource.Lock() |
| + oldReg := movesWakeSource.registered |
| + movesWakeSource.registered = nil |
| + movesWakeSource.Unlock() |
| + t.Cleanup(func() { |
| + claimMoveSources.Lock() |
| + claimMoveSources.Source, claimMoveSources.Parks = oldSource, oldParks |
| + claimMoveSources.Unlock() |
| + movesWakeSource.Lock() |
| + movesWakeSource.registered = oldReg |
| + movesWakeSource.Unlock() |
| + }) |
| + if register { |
| + if err := registerMovesWakeSource(&parkRegistrationWake{}); err != nil { |
| + t.Fatal(err) |
| } |
| } |
| } |
| |
| -// fakePlan12 stands in for claim Plan 12's pieces in the arming tests. |
| -type fakePlan12 struct{ runs, locals int } |
| +// renewTable is a lease table behind hostbox.RenewingLeases' Open: one |
| +// holder per bucket/scope, a second open of a held one refused as |
| +// lease.Acquire refuses it. Everything above Open (RenewingLeases, |
| +// AggregateLeases, AcquireHeld, LeaseScope) is the claim's own. |
| +type renewTable struct { |
| + mu sync.Mutex |
| + held map[string]bool |
| + opens []string |
| +} |
| + |
| +type renewHeld struct { |
| + t *renewTable |
| + key string |
| + lost chan struct{} |
| +} |
| + |
| +func (r *renewTable) open(_ context.Context, bucket, scope, _ string) (hostbox.Renewable, error) { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + key := bucket + "/" + scope |
| + if r.held[key] { |
| + return nil, &lease.HeldError{Owner: "another holder"} |
| + } |
| + if r.held == nil { |
| + r.held = map[string]bool{} |
| + } |
| + r.held[key] = true |
| + r.opens = append(r.opens, key) |
| + return &renewHeld{t: r, key: key, lost: make(chan struct{})}, nil |
| +} |
| + |
| +func (r *renewTable) holding() []string { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + var out []string |
| + for k, on := range r.held { |
| + if on { |
| + out = append(out, k) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out |
| +} |
| |
| -func (p *fakePlan12) RunHeld(ctx context.Context, e park.Engine, w park.Work, held park.HeldLease) error { |
| - p.runs++ |
| - return testRunHeld(ctx, e, w, held) |
| +func (h *renewHeld) Err() error { |
| + h.t.mu.Lock() |
| + defer h.t.mu.Unlock() |
| + if !h.t.held[h.key] { |
| + return errors.New("renew table: released") |
| + } |
| + return nil |
| } |
| -func (p *fakePlan12) AuthorizeLocal(context.Context, *hostclaim.Claim, hostclaim.Obligation) error { |
| - p.locals++ |
| +func (h *renewHeld) Lost() <-chan struct{} { return h.lost } |
| +func (h *renewHeld) Release(context.Context) error { |
| + h.t.mu.Lock() |
| + defer h.t.mu.Unlock() |
| + delete(h.t.held, h.key) |
| return nil |
| } |
| |
| +// leases is the claim's park leases over the table, as Build assembles |
| +// them: RenewingLeases under AggregateLeases with the replica park lease set. |
| +func (r *renewTable) leases() hostbox.AggregateLeases { |
| + return hostbox.AggregateLeases{Leases: hostbox.RenewingLeases{Region: "eu-west-1", Open: r.open}, Resolve: hostbox.ReplicaParkLeaseSet} |
| +} |
| + |
| +// ownLease is the build's own-location lease over the table, as |
| +// armPhase1's OwnLease takes it (RenewingLeases.Acquire). |
| +func (r *renewTable) ownLease(ctx context.Context, bucket, location string) (park.HeldLease, error) { |
| + return hostbox.RenewingLeases{Region: "eu-west-1", Open: r.open}.Acquire(ctx, bucket, location) |
| +} |
| + |
| // TestArmMovesPassRefusesAMismatchedBucketOrRegion (M17): the arms take the |
| -// claim's engine only when its bucket and region are this box's own, and |
| -// only with claim Plan 12's pieces; any refusal sets none of the hooks. |
| +// claim's engine only when its bucket and region are this box's own; any |
| +// refusal sets none of the hooks. |
| func TestArmMovesPassRefusesAMismatchedBucketOrRegion(t *testing.T) { |
| + withMovesWakeSource(t, true) |
| view := func(context.Context, hostclaim.Observed) (park.View, error) { return park.View{}, nil } |
| auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil } |
| - locals := 0 |
| providers := func(bucket, region string) ClaimProviders { |
| return ClaimProviders{moveParks: &claimMoveParks{ |
| - local: func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error { |
| - locals++ |
| - return nil |
| - }, |
| - engine: park.Engine{Bucket: bucket, Region: region, Leases: &fakeParkLeases{}, View: view, Authorize: auth}, |
| - leases: movesLeaseSet(bucket), check: func(context.Context) error { return nil }}} |
| - } |
| - plan12 := &fakePlan12{} |
| - ok := func() (movesClaimPlan12, error) { return plan12, nil } |
| + local: claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth), |
| + engine: park.Engine{Bucket: bucket, Region: region, Leases: (&renewTable{}).leases(), View: view, Authorize: auth}, |
| + leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}} |
| + } |
| unarmed := func(t *testing.T, c moveCfg) { |
| t.Helper() |
| if c.ParkLeases != nil || c.ParkView != nil || c.ParkAuthorize != nil || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil { |
| @@ -351,7 +445,7 @@ func TestArmMovesPassRefusesAMismatchedBucketOrRegion(t *testing.T) { |
| for name, p := range map[string]ClaimProviders{"another bucket": providers("data-uk", "ie"), "another region": providers("data-ie", "uk")} { |
| t.Run(name, func(t *testing.T) { |
| c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| - if err := armMovesParks(t.Context(), &c, p, ok); !errors.Is(err, hostbox.ErrChanged) { |
| + if err := armMovesParks(t.Context(), &c, p); !errors.Is(err, hostbox.ErrChanged) { |
| t.Fatalf("armed: %v", err) |
| } |
| unarmed(t, c) |
| @@ -359,96 +453,231 @@ func TestArmMovesPassRefusesAMismatchedBucketOrRegion(t *testing.T) { |
| } |
| t.Run("no MoveParks", func(t *testing.T) { |
| c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| - if err := armMovesParks(t.Context(), &c, ClaimProviders{}, ok); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + if err := armMovesParks(t.Context(), &c, ClaimProviders{}); !errors.Is(err, hostwrite.ErrUnavailable) { |
| t.Fatalf("armed: %v", err) |
| } |
| unarmed(t, c) |
| }) |
| - t.Run("without claim Plan 12", func(t *testing.T) { |
| - c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| - err := armMovesParks(t.Context(), &c, providers("data-ie", "ie"), movesPlan12Parks) |
| - var dep *ClaimDependencyError |
| - if !errors.As(err, &dep) { |
| - t.Fatalf("armed: %v", err) |
| - } |
| - for _, piece := range []string{"RunHeld", "lease-set contract", "local authorizer"} { |
| - if !strings.Contains(err.Error(), piece) || !strings.Contains(err.Error(), "claim Plan 12") { |
| - t.Fatalf("the refusal does not name %s: %v", piece, err) |
| - } |
| - } |
| - unarmed(t, c) |
| - }) |
| t.Run("this box's own engine", func(t *testing.T) { |
| c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| - if err := armMovesParks(t.Context(), &c, providers("data-ie", "ie"), ok); err != nil { |
| + if err := armMovesParks(t.Context(), &c, providers("data-ie", "ie")); err != nil { |
| t.Fatal(err) |
| } |
| if c.ParkLeases == nil || c.ParkView == nil || c.ParkAuthorize == nil || c.ParkRunHeld == nil || c.ParkLocalAuthorize == nil { |
| t.Fatalf("hooks unset: %+v", c) |
| } |
| - if err := c.ParkLocalAuthorize(t.Context(), nil, hostclaim.Obligation{}); err != nil || locals != 1 || plan12.locals != 0 { |
| - t.Fatal("the local authorizer is not claim Plan 12's") |
| - } |
| }) |
| } |
| |
| -// TestMovesParksRefuseUntilClaimPlan12: the registered view and lease set |
| -// refuse by name while claim Plan 12 is missing, so the claim's regional |
| -// worker queues and runs nothing through them; Task 5's evidence is |
| -// registered as it is, reachable as the sweep verifier. |
| -func TestMovesParksRefuseUntilClaimPlan12(t *testing.T) { |
| - f := newVwFixture(t) |
| - evidence := movesRetirementEvidence{Records: f.records(), Now: time.Now} |
| - src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence, movesPlan12Parks) |
| - if err != nil { |
| +// TestArmMovesParksTakesRunHeldAndTheLocalAuthorizerFromTheClaim (M5, M9): |
| +// the arm's RunHeld is park.Engine.RunHeld over the claim's own leases, so a |
| +// residue park borrows the build's own-location lease through the claim's |
| +// AggregateLeases (AcquireHeld, LeaseScope) where Engine.Run would wait on |
| +// it; and its local authorizer is the claim's MoveParks one, which asks the |
| +// claim's identity check and owner and refuses anything but a park-local. A |
| +// claim engine whose leases cannot lend a held lease does not arm. |
| +func TestArmMovesParksTakesRunHeldAndTheLocalAuthorizerFromTheClaim(t *testing.T) { |
| + withMovesWakeSource(t, true) |
| + ctx := t.Context() |
| + inc := strings.Repeat("a", 64) |
| + loc := "instances/" + vwHost + "/" + inc |
| + now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) |
| + claims := release.NewDirStore(t.TempDir()) |
| + if _, err := hostclaim.Update(ctx, claims, hostclaim.Observed{Host: vwHost, Absent: true}, func(c *hostclaim.Claim) error { |
| + c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op", Intent: map[string]string{"park_sweep": "v1"}} |
| + c.Obligations = []hostclaim.Obligation{{ID: "chore", Op: "op", Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: "ret", |
| + Park: &hostclaim.ParkScope{Location: loc, Incarnation: inc, EvidenceAt: now.Add(-time.Hour), RequestLimit: hostbox.RequestLimit}}} |
| + return nil |
| + }); err != nil { |
| t.Fatal(err) |
| } |
| - obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}) |
| - var dep *ClaimDependencyError |
| - if _, err := src.View(t.Context(), obs); !errors.As(err, &dep) || !strings.Contains(err.Error(), "claim Plan 12") { |
| - t.Fatalf("the view answered without claim Plan 12: %v", err) |
| + data := leasetest.NewVersionedBucket() |
| + data.AddBucket("data-ie") |
| + data.Put("data-ie", loc+"/db", []byte("residue")) |
| + table := &renewTable{} |
| + view := func(_ context.Context, o hostclaim.Observed) (park.View, error) { |
| + return park.View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: movesNoLiveRegion}, nil |
| + } |
| + authorized := 0 |
| + auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { authorized++; return nil } |
| + owned, identity := 0, error(nil) |
| + owner := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { owned++; return nil } |
| + providers := func(leases park.Leases) ClaimProviders { |
| + return ClaimProviders{moveParks: &claimMoveParks{ |
| + local: claimMoveLocalAuthorizer(func(context.Context) error { return identity }, owner), |
| + engine: park.Engine{Bucket: "data-ie", Region: "ie", Leases: leases, View: view, Authorize: auth}, |
| + leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}} |
| } |
| - if _, err := src.Leases(t.Context(), "data", "instances/"+vwHost); !errors.As(err, &dep) || !strings.Contains(err.Error(), "claim Plan 12") { |
| - t.Fatalf("the lease set answered without claim Plan 12: %v", err) |
| + |
| + t.Run("leases that cannot lend a held lease", func(t *testing.T) { |
| + c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| + if err := armMovesParks(ctx, &c, providers(&fakeParkLeases{})); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("armed over leases with no AcquireHeld: %v", err) |
| + } |
| + if parkHooksArmed(c) || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil { |
| + t.Fatal("a refused arming set a hook") |
| + } |
| + }) |
| + |
| + c := moveCfg{OwnBucket: "data-ie", Region: "ie"} |
| + if err := armMovesParks(ctx, &c, providers(table.leases())); err != nil { |
| + t.Fatal(err) |
| } |
| - if f.reads != 0 || f.lists != 0 { |
| - t.Fatal("a refused view read the record or the bucket") |
| + own, err := table.ownLease(ctx, c.OwnBucket, loc) |
| + if err != nil { |
| + t.Fatal(err) |
| } |
| - if _, ok := src.Evidence.(park.MoveSweepVerifier); !ok { |
| - t.Fatal("the registered evidence is not the sweep verifier") |
| + t.Cleanup(own.Close) |
| + // The engine as an install builds it (installRun.engine): this box's |
| + // store and bucket, the armed leases, view and authorizer. |
| + eng := park.Engine{Store: data, Claims: claims, Bucket: c.OwnBucket, Region: c.parkRegion(), Leases: c.ParkLeases, |
| + Now: func() time.Time { return now }, NewID: func() string { return "next" }, View: c.ParkView, Authorize: c.ParkAuthorize} |
| + w := park.Work{Host: vwHost, ID: "chore", Op: "op"} |
| + if err := eng.Run(ctx, w); err == nil { |
| + t.Fatal("Engine.Run took the location's lock while the build holds it") |
| } |
| - // With claim Plan 12 the same sources answer. |
| - src, err = newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence, func() (movesClaimPlan12, error) { return &fakePlan12{}, nil }) |
| - if err != nil { |
| + if err := c.ParkRunHeld(ctx, eng, w, own); err != nil && !errors.Is(err, park.ErrPending) { |
| t.Fatal(err) |
| } |
| - if v, err := src.View(t.Context(), obs); err != nil || v.AuthoritativeRegion != "ie" { |
| - t.Fatalf("view %+v, %v", v, err) |
| + if versions, err := data.ListVersions(ctx, "data-ie", loc+"/"); err != nil || len(versions) != 0 { |
| + t.Fatalf("the residue is still there: %v %v", versions, err) |
| } |
| - if names, err := src.Leases(t.Context(), "data", "instances/"+vwHost); err != nil || len(names) != 1 { |
| - t.Fatalf("leases %v, %v", names, err) |
| + if len(table.opens) != 1 || authorized == 0 { |
| + t.Fatalf("lease opens %v, authorized %d: the park did not borrow the build's lease under the claim's authorizer", table.opens, authorized) |
| } |
| - // Registration is the claim's one-shot slot. |
| - claimMoveSources.Lock() |
| - oldSource, oldParks := claimMoveSources.Source, claimMoveSources.Parks |
| - claimMoveSources.Source, claimMoveSources.Parks = nil, nil |
| - claimMoveSources.Unlock() |
| - t.Cleanup(func() { |
| - claimMoveSources.Lock() |
| - claimMoveSources.Source, claimMoveSources.Parks = oldSource, oldParks |
| - claimMoveSources.Unlock() |
| - }) |
| - if err := registerMovesParks(src); err != nil { |
| - t.Fatal(err) |
| + if err := own.Check(ctx); err != nil { |
| + t.Fatalf("the run released the build's own lease: %v", err) |
| } |
| - if p := claimMoveSources.Parks; p == nil || p.View == nil || p.Evidence == nil { |
| - t.Fatalf("registered %+v", p) |
| + |
| + local := hostclaim.Obligation{ID: "local", Op: "op", Kind: "park-local", Box: "ie-box"} |
| + if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, local); err != nil || owned != 1 { |
| + t.Fatalf("the claim's owner was not asked: %v (%d)", err, owned) |
| } |
| - if err := registerMovesParks(src); !errors.Is(err, hostwrite.ErrUnavailable) { |
| - t.Fatalf("a second registration: %v", err) |
| + if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, hostclaim.Obligation{ID: "chore", Op: "op", Kind: "park-replica"}); err == nil || owned != 1 { |
| + t.Fatalf("a park-replica passed the local authorizer: %v", err) |
| + } |
| + identity = hostbox.ErrChanged |
| + if err := c.ParkLocalAuthorize(ctx, &hostclaim.Claim{Host: vwHost}, local); !errors.Is(err, hostbox.ErrChanged) || owned != 1 { |
| + t.Fatalf("the local authorizer skipped the claim's identity check: %v", err) |
| } |
| } |
| |
| +// TestMovesParksRefuseUntilTheWakeSourceRegisters (Q1): one gate, |
| +// movesWakeSourceRegistered, stands between the moves parks and production. |
| +// Until this box's moves source is registered with the claim |
| +// (registerMovesWakeSource, RegisterProductionClaimMoveSources), the |
| +// registered view refuses with a named dependency, reading nothing, so the |
| +// claim's sweep queues and runs nothing through it, and armMovesParks sets no |
| +// hook. Registering the park sources alone, or another source, does not |
| +// open it. Task 5's evidence is registered as it is, reachable as the |
| +// sweep verifier. |
| +func TestMovesParksRefuseUntilTheWakeSourceRegisters(t *testing.T) { |
| + const contract = "moves wake owners and C8 files manifest provider (Plan 5 Task 9b)" |
| + f := newVwFixture(t) |
| + evidence := movesRetirementEvidence{Records: f.records(), Now: time.Now} |
| + obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}) |
| + auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil } |
| + providers := ClaimProviders{moveParks: &claimMoveParks{ |
| + local: claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth), |
| + engine: park.Engine{Bucket: "data", Region: "ie", Leases: (&renewTable{}).leases(), View: f.view("ie"), Authorize: auth}, |
| + leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}} |
| + shut := func(t *testing.T, src movesParkSources) { |
| + t.Helper() |
| + var dep *ClaimDependencyError |
| + if _, err := src.View(t.Context(), obs); !errors.As(err, &dep) || !strings.Contains(err.Error(), contract) { |
| + t.Fatalf("the view answered before the wake source registered: %v", err) |
| + } |
| + if f.reads != 0 || f.lists != 0 { |
| + t.Fatal("a refused view read the record or the bucket") |
| + } |
| + c := moveCfg{OwnBucket: "data", Region: "ie"} |
| + if err := armMovesParks(t.Context(), &c, providers); !errors.As(err, &dep) || !strings.Contains(err.Error(), contract) { |
| + t.Fatalf("armed before the wake source registered: %v", err) |
| + } |
| + if parkHooksArmed(c) || c.ParkRunHeld != nil || c.ParkLocalAuthorize != nil { |
| + t.Fatal("a refused arming set a hook") |
| + } |
| + } |
| + |
| + t.Run("the park sources alone", func(t *testing.T) { |
| + withMovesWakeSource(t, false) |
| + src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := src.Evidence.(park.MoveSweepVerifier); !ok { |
| + t.Fatal("the registered evidence is not the sweep verifier") |
| + } |
| + shut(t, src) |
| + if err := registerMovesParks(src); err != nil { |
| + t.Fatal(err) |
| + } |
| + if p := claimMoveSources.Parks; p == nil || p.View == nil || p.Evidence == nil { |
| + t.Fatalf("registered %+v", p) |
| + } |
| + if err := registerMovesParks(src); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("a second registration: %v", err) |
| + } |
| + shut(t, src) |
| + if _, err := claimMoveSources.Parks.View(t.Context(), obs); err == nil { |
| + t.Fatal("the claim's registered park view answered before the wake source registered") |
| + } |
| + }) |
| + |
| + t.Run("another source registered with the claim", func(t *testing.T) { |
| + withMovesWakeSource(t, false) |
| + src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := RegisterProductionClaimMoveSources(&parkRegistrationWake{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + shut(t, src) |
| + if err := registerMovesWakeSource(&parkRegistrationWake{}); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("registered over another source: %v", err) |
| + } |
| + shut(t, src) |
| + }) |
| + |
| + t.Run("this box's moves source", func(t *testing.T) { |
| + withMovesWakeSource(t, false) |
| + src, err := newMovesParkSources(f.records(), f.bucket, "data", fixedRegion("ie"), evidence) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := registerMovesParks(src); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := registerMovesWakeSource(nil); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("a nil source: %v", err) |
| + } |
| + shut(t, src) |
| + if err := registerMovesWakeSource(&parkRegistrationWake{}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesWakeSourceRegistered(); err != nil { |
| + t.Fatalf("the gate: %v", err) |
| + } |
| + if v, err := src.View(t.Context(), obs); err != nil || v.AuthoritativeRegion != "ie" { |
| + t.Fatalf("view %+v, %v", v, err) |
| + } |
| + if v, err := claimMoveSources.Parks.View(t.Context(), obs); err != nil || v.AuthoritativeRegion != "ie" { |
| + t.Fatalf("the claim's park view is not the moves view: %+v, %v", v, err) |
| + } |
| + c := moveCfg{OwnBucket: "data", Region: "ie"} |
| + if err := armMovesParks(t.Context(), &c, providers); err != nil || !parkHooksArmed(c) { |
| + t.Fatalf("not armed: %v", err) |
| + } |
| + if err := registerMovesWakeSource(&parkRegistrationWake{}); !errors.Is(err, hostwrite.ErrUnavailable) { |
| + t.Fatalf("a second registration: %v", err) |
| + } |
| + if err := movesWakeSourceRegistered(); err != nil { |
| + t.Fatalf("a refused second registration shut the gate: %v", err) |
| + } |
| + }) |
| +} |
| + |
| // testRunHeld is the agreed shape of claim Plan 12's Engine.RunHeld (M5): |
| // the chore runs under the lease the caller holds, which must be the |
| // scope's own lock (the lease set, M6); it is neither acquired again, nor |
| @@ -523,17 +752,12 @@ func TestCommittedLaneViewNamesTheTargetWithItsMarker(t *testing.T) { |
| // TestRegisterProductionMovesParksNeedsBothSwitches (ruling H1): with either |
| // switch off the agent registers nothing. With both on it registers without |
| // CARLOS_REGION: the view's region is the claim's verified roster row's |
| -// (fix round 1, region source), resolved when the view is first asked. |
| +// (fix round 1, region source), resolved when the view is first asked. What |
| +// it registers stays shut (Q1): nothing in production registers the moves |
| +// wake source yet, so the claim's park view refuses every host before it |
| +// reads anything, and the claim's sweep is not widened. |
| func TestRegisterProductionMovesParksNeedsBothSwitches(t *testing.T) { |
| - claimMoveSources.Lock() |
| - oldSource, oldParks := claimMoveSources.Source, claimMoveSources.Parks |
| - claimMoveSources.Source, claimMoveSources.Parks = nil, nil |
| - claimMoveSources.Unlock() |
| - t.Cleanup(func() { |
| - claimMoveSources.Lock() |
| - claimMoveSources.Source, claimMoveSources.Parks = oldSource, oldParks |
| - claimMoveSources.Unlock() |
| - }) |
| + withMovesWakeSource(t, false) |
| env := func(vals map[string]string) func(string) string { return func(k string) string { return vals[k] } } |
| hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1"} |
| store := release.NewDirStore(t.TempDir()) |
| @@ -556,6 +780,12 @@ func TestRegisterProductionMovesParksNeedsBothSwitches(t *testing.T) { |
| if claimMoveSources.Parks == nil || asked != 0 { |
| t.Fatalf("registered %v, roster asked %d times at registration", claimMoveSources.Parks != nil, asked) |
| } |
| + f := newVwFixture(t) |
| + obs := vwObs(&hostclaim.Claim{Holder: f.holder("cur", "cur@ie"), Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}}) |
| + var dep *ClaimDependencyError |
| + if _, err := claimMoveSources.Parks.View(t.Context(), obs); !errors.As(err, &dep) || asked != 0 || claimMoveSources.Source != nil { |
| + t.Fatalf("the production park view answered (%v), roster asked %d times", err, asked) |
| + } |
| } |
| |
| // TestHomeBoxArms (fix round 1, region source; home-region parks, part 1): |
| @@ -564,6 +794,7 @@ func TestRegisterProductionMovesParksNeedsBothSwitches(t *testing.T) { |
| // of "" (unset) never arms, and neither does a box whose code is no |
| // placement, even when the engine's region spells the same string. |
| func TestHomeBoxArms(t *testing.T) { |
| + withMovesWakeSource(t, true) |
| f := newVwFixture(t) |
| f.rec.Region = "" |
| view := movesParkViewAt(f.records(), f.bucket, "data", fixedRegion(park.HomeRegion)) |
| @@ -576,11 +807,11 @@ func TestHomeBoxArms(t *testing.T) { |
| auth := func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil } |
| arm := func(engineRegion, placement string) (moveCfg, error) { |
| providers := ClaimProviders{moveParks: &claimMoveParks{ |
| - local: func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error { return nil }, |
| - engine: park.Engine{Bucket: "data-home", Region: engineRegion, Leases: &fakeParkLeases{}, View: view, Authorize: auth}, |
| - leases: movesLeaseSet("data-home"), check: func(context.Context) error { return nil }}} |
| + local: claimMoveLocalAuthorizer(func(context.Context) error { return nil }, auth), |
| + engine: park.Engine{Bucket: "data-home", Region: engineRegion, Leases: (&renewTable{}).leases(), View: view, Authorize: auth}, |
| + leases: hostbox.ReplicaParkLeaseSet, check: func(context.Context) error { return nil }}} |
| c := moveCfg{OwnBucket: "data-home", Region: placement} |
| - err := armMovesParks(t.Context(), &c, providers, func() (movesClaimPlan12, error) { return &fakePlan12{}, nil }) |
| + err := armMovesParks(t.Context(), &c, providers) |
| return c, err |
| } |
| c, err := arm(park.HomeRegion, "") |
| @@ -644,7 +875,7 @@ func TestLandingResidueParkUnderItsOwnLease(t *testing.T) { |
| h := newLandingHarness(t) |
| h.flip(t) |
| keys := h.residue(t) |
| - h.parkLeases = hostbox.AggregateLeases{Leases: h.leases, Resolve: movesLeaseSet("data-uk")} |
| + h.parkLeases = hostbox.AggregateLeases{Leases: h.leases, Resolve: hostbox.ReplicaParkLeaseSet} |
| runs := 0 |
| h.runHeld = func(ctx context.Context, e park.Engine, w park.Work, held park.HeldLease) error { |
| runs++ |
| @@ -668,7 +899,7 @@ func TestLandingResidueParkUnderItsOwnLease(t *testing.T) { |
| h := newLandingHarness(t) |
| h.flip(t) |
| keys := h.residue(t) |
| - h.parkLeases = hostbox.AggregateLeases{Leases: h.leases, Resolve: movesLeaseSet("data-uk")} |
| + h.parkLeases = hostbox.AggregateLeases{Leases: h.leases, Resolve: hostbox.ReplicaParkLeaseSet} |
| h.runHeld = func(ctx context.Context, e park.Engine, w park.Work, _ park.HeldLease) error { return e.Run(ctx, w) } |
| for i := 0; i < 3; i++ { |
| h.land(t) |