amadan

carlos / platform Public

host: the moves owners read the claim's bucket through hostbox.RequestStore (Task 8 fix round 1, item 1)

openProductionClaimMoveWake now hands the registered moves source the
claim's data bucket as the serving owners see it: hostbox.RequestStore
over out.Data, checked by the box's bound identity
(claimBoundIdentity(out.Identity.Verify, actor)), so every bucket request
is bounded by the claim's request limit, runs under the operation's own
context, and a refused identity issues no later bucket call.

Regression: TestMovesOwnersReadTheBucketThroughTheRequestStore (the
deadline, the identity refusal before a later call, and an in-flight
request cancelled by the operation). The integrated harness opens the
owners over the same wrapper.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e462f931195402881285fed7e902b84cea1ea5d1 parent 7afef72
5 files changed, +110 −7
  • internal/host/claimmoveparks.go +13 −2
  • internal/host/claimmoveparks_test.go +3 −3
  • internal/host/claimproduction.go +1 −1
  • internal/host/moveswakesource_linux_test.go +4 −1
  • internal/host/moveswakesource_test.go +89 −0
diff --git a/internal/host/claimmoveparks.go b/internal/host/claimmoveparks.go
index 00d37106b..26b056aed 100644
--- a/internal/host/claimmoveparks.go
+++ b/internal/host/claimmoveparks.go
@@ -40,12 +40,23 @@ func RegisterProductionClaimMoveParks(view func(context.Context, hostclaim.Obser
return nil
}
-func openProductionClaimMoveWake(ctx context.Context, store release.Store, actor hostbox.Actor, external ClaimProductionMoveSources, out *ClaimOwnerBindings) {
+// openProductionClaimMoveWake opens the registered moves source over the
+// claim's data bucket as the serving owners see it (moves Task 8 fix round
+// 1): through hostbox.RequestStore, so every bucket request the moves
+// owners make is bounded by the claim's request limit, runs under the
+// operation's own context (its cancellation still reaches the request), and
+// is preceded by check, the box's bound identity (claimBoundIdentity), so a
+// refused identity issues no later bucket call.
+func openProductionClaimMoveWake(ctx context.Context, store release.Store, actor hostbox.Actor, external ClaimProductionMoveSources, out *ClaimOwnerBindings, check func(context.Context) error) {
if external == nil {
out.wakeUnavailable = &ClaimDependencyError{Contracts: []string{"moves wake owners and C8 origin-qualified files manifest provider"}}
return
}
- owners, files, err := external.Open(ctx, store, out.Data, actor, out.moveRead)
+ var data lease.VersionedStore
+ if out.Data != nil && check != nil {
+ data = hostbox.RequestStore{Store: out.Data, Check: check}
+ }
+ owners, files, err := external.Open(ctx, store, data, actor, out.moveRead)
if err != nil {
out.wakeUnavailable = errors.Join(&ClaimDependencyError{Contracts: []string{"moves wake owners and C8 origin-qualified files manifest provider"}}, err)
return
diff --git a/internal/host/claimmoveparks_test.go b/internal/host/claimmoveparks_test.go
index 96ffc1e5a..729657ca4 100644
--- a/internal/host/claimmoveparks_test.go
+++ b/internal/host/claimmoveparks_test.go
@@ -91,7 +91,7 @@ func TestMoveParksWakeFailureRemainsWakeRefusal(t *testing.T) {
boom := errors.New("wake failed")
source := &parkRegistrationWake{failure: boom}
out := ClaimOwnerBindings{}
- openProductionClaimMoveWake(t.Context(), nil, hostbox.Actor{}, source, &out)
+ openProductionClaimMoveWake(t.Context(), nil, hostbox.Actor{}, source, &out, func(context.Context) error { return nil })
if source.opens != 1 || !errors.Is(out.wakeUnavailable, boom) {
t.Fatalf("wake failure lost: %v", out.wakeUnavailable)
}
@@ -99,7 +99,7 @@ func TestMoveParksWakeFailureRemainsWakeRefusal(t *testing.T) {
if err := w.Prerequisites(t.Context(), registry.Route{}, hostclaim.Observed{}, activator.StartPolicy{}); !errors.Is(err, boom) {
t.Fatalf("wake opened: %v", err)
}
- openProductionClaimMoveWake(t.Context(), nil, hostbox.Actor{}, nil, &out)
+ openProductionClaimMoveWake(t.Context(), nil, hostbox.Actor{}, nil, &out, func(context.Context) error { return nil })
if out.wakeUnavailable == nil || !strings.Contains(out.wakeUnavailable.Error(), "C8") {
t.Fatal("missing wake not named")
}
@@ -167,7 +167,7 @@ func TestMoveParksBuildWithFailedWakeUsesRegionalEngine(t *testing.T) {
boom := errors.New("configured wake failed")
wakeSource := &parkRegistrationWake{failure: boom}
owners := ClaimOwnerBindings{}
- openProductionClaimMoveWake(ctx, store, actor, wakeSource, &owners)
+ openProductionClaimMoveWake(ctx, store, actor, wakeSource, &owners, claimBoundIdentity(identity.Verify, actor))
viewCalls := 0
a := ClaimAssembly{
wakeUnavailable: owners.wakeUnavailable, Claims: store, Guard: hostguard.Guard{DataDir: base}, Identity: identity,
diff --git a/internal/host/claimproduction.go b/internal/host/claimproduction.go
index d00fc6416..b44965df5 100644
--- a/internal/host/claimproduction.go
+++ b/internal/host/claimproduction.go
@@ -188,7 +188,7 @@ func (ProductionClaimOwnerProvider) OpenClaimOwners(ctx context.Context, store r
parks := claimMoveSources.Parks
claimMoveSources.Unlock()
if hostwrite.Enabled(claimEntryEnv(ctx)) && moves.Enabled(claimEntryEnv(ctx)) {
- openProductionClaimMoveWake(ctx, store, actor, external, &out)
+ openProductionClaimMoveWake(ctx, store, actor, external, &out, claimBoundIdentity(out.Identity.Verify, actor))
} else if external != nil {
out.Moves, out.Serving.Files, err = external.Open(ctx, store, out.Data, actor, out.moveRead)
if err != nil {
diff --git a/internal/host/moveswakesource_linux_test.go b/internal/host/moveswakesource_linux_test.go
index 1621b6577..dbbb00e5d 100644
--- a/internal/host/moveswakesource_linux_test.go
+++ b/internal/host/moveswakesource_linux_test.go
@@ -142,7 +142,10 @@ func mwRegister(t *testing.T, f *poFixture, box, region string, files ClaimFiles
actor := hostbox.Actor{Box: box, Region: region, Bucket: "data", Enrollment: "enrollment-1", Cloud: hostbox.CloudIdentity{Region: "eu-west-1"}}
out := ClaimOwnerBindings{Data: f.bucket}
var opened ClaimFilesManifest
- out.Moves, opened, err = claimMoveSources.Source.Open(ctx, f.claims, f.bucket, actor, mwAccess{f})
+ // The claim hands Open its bucket through the request store
+ // (openProductionClaimMoveWake), with the box's identity check.
+ data := hostbox.RequestStore{Store: f.bucket, Check: func(context.Context) error { return nil }}
+ out.Moves, opened, err = claimMoveSources.Source.Open(ctx, f.claims, data, actor, mwAccess{f})
if err != nil || opened != nil {
t.Fatalf("the registered source's Open: %v (files %v)", err, opened)
}
diff --git a/internal/host/moveswakesource_test.go b/internal/host/moveswakesource_test.go
index 55c205688..fed015a73 100644
--- a/internal/host/moveswakesource_test.go
+++ b/internal/host/moveswakesource_test.go
@@ -365,3 +365,92 @@ func TestRegisteredSourceOpensTheMovesOwnersWithTheClaimsFiles(t *testing.T) {
t.Fatalf("a bare actor: %v", err)
}
}
+
+// mwDataSource is a moves source that records the bucket the claim hands
+// its Open.
+type mwDataSource struct {
+ parkRegistrationWake
+ data lease.VersionedStore
+}
+
+func (s *mwDataSource) Open(_ context.Context, _ release.Store, data lease.VersionedStore, _ hostbox.Actor, _ ClaimMoveReadAccess) (ClaimMovesOwners, ClaimFilesManifest, error) {
+ s.data = data
+ return ClaimMovesOwners{}, nil, nil
+}
+
+// mwBucket is the claim's data bucket, recording each listing's context
+// and, when block is set, holding the listing until its context ends.
+type mwBucket struct {
+ *leasetest.VersionedBucket
+ lists int
+ deadlines []time.Time
+ block bool
+}
+
+func (b *mwBucket) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ b.lists++
+ d, _ := ctx.Deadline()
+ b.deadlines = append(b.deadlines, d)
+ if b.block {
+ <-ctx.Done()
+ return nil, ctx.Err()
+ }
+ return b.VersionedBucket.ListVersions(ctx, bucket, prefix)
+}
+
+// TestMovesOwnersReadTheBucketThroughTheRequestStore (Task 8 fix round 1,
+// Astra and platform-1e): the moves owners get the claim's data bucket as
+// the serving owners do, through hostbox.RequestStore: every request is
+// bounded by the claim's request limit, the box's identity is checked
+// before each one (a refused identity issues no later bucket call), and
+// the operation's own context still cancels a request in flight.
+func TestMovesOwnersReadTheBucketThroughTheRequestStore(t *testing.T) {
+ inner := &mwBucket{VersionedBucket: leasetest.NewVersionedBucket()}
+ inner.AddBucket("data")
+ var refuse error
+ checks := 0
+ check := func(context.Context) error { checks++; return refuse }
+ src := &mwDataSource{}
+ out := ClaimOwnerBindings{Data: inner}
+ openProductionClaimMoveWake(t.Context(), release.NewDirStore(t.TempDir()), hostbox.Actor{Box: "ie-box"}, src, &out, check)
+ if out.wakeUnavailable != nil || src.data == nil {
+ t.Fatalf("open: %v", out.wakeUnavailable)
+ }
+ if _, ok := src.data.(hostbox.RequestStore); !ok {
+ t.Fatalf("the moves owners' bucket is %T, not the claim's request store", src.data)
+ }
+ before := time.Now()
+ if _, err := src.data.ListVersions(t.Context(), "data", "instances/"); err != nil {
+ t.Fatal(err)
+ }
+ if inner.lists != 1 || checks != 1 || inner.deadlines[0].IsZero() || inner.deadlines[0].After(time.Now().Add(hostbox.RequestLimit)) || inner.deadlines[0].Before(before.Add(hostbox.RequestLimit-time.Minute)) {
+ t.Fatalf("the request's deadline %v (lists %d, checks %d), want the request limit", inner.deadlines, inner.lists, checks)
+ }
+ refuse = hostbox.ErrChanged
+ if _, err := src.data.ListVersions(t.Context(), "data", "instances/"); !errors.Is(err, hostbox.ErrChanged) || inner.lists != 1 {
+ t.Fatalf("a refused identity: %v, %d bucket calls", err, inner.lists)
+ }
+ if _, err := src.data.GetVersion(t.Context(), "data", "k", "v"); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("a refused identity's read: %v", err)
+ }
+ refuse = nil
+ inner.block = true
+ ctx, cancel := context.WithCancel(t.Context())
+ done := make(chan error, 1)
+ go func() { _, err := src.data.ListVersions(ctx, "data", "instances/"); done <- err }()
+ for {
+ if l := func() int { return inner.lists }(); l == 2 {
+ break
+ }
+ time.Sleep(time.Millisecond)
+ }
+ cancel()
+ select {
+ case err := <-done:
+ if !errors.Is(err, context.Canceled) {
+ t.Fatalf("a cancelled operation's request: %v", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("the operation's cancellation did not reach the request")
+ }
+}