amadan

carlos / platform Public

host: the landing's generation requests run under the lease's cancellation context (Task 10 fix round 2)

leaseContext derives the build's context from the held lease's own, as
hostbox's renewingHeld cancels it when the lease is lost or released, so a
generation request in flight when the lease lapses, and the S3 client's
retries of it, stop with the lease instead of landing after it. Both build
paths use it; leasedStore's check before each request stands. The test's
fake lease carries the same Context method, and a store that blocks like
a retrying client proves the request is cancelled and nothing recorded.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude d50652c330c1f52c38d3eb66f43677866a864ea5 parent 0035ce9
4 files changed, +122 −11
  • internal/host/movesland.go +4 −3
  • internal/host/movesland_test.go +97 −4
  • internal/host/moveslandinstall.go +1 −1
  • internal/host/moveslandpark.go +20 −3
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
index 16329dfe6..4ec659e2f 100644
--- a/internal/host/movesland.go
+++ b/internal/host/movesland.go
@@ -706,9 +706,10 @@ func (l *landingRun) building(ctx context.Context) (bool, error) {
}
// Every generation request runs under live lease authority (fix round
// 1, item 3): the store checks the lease before each call, as the cut's
- // guarded writes check theirs, and the lease is checked again before
- // the build's completion is recorded.
- m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ // guarded writes check theirs, the requests run under a context the
+ // lease's loss cancels (leaseContext, fix round 2), and the lease is
+ // checked again before the build's completion is recorded.
+ m, err := moves.BuildGeneration(leaseContext(ctx, held), moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
GenerationPrefix: moves.GenerationPrefix(holder.Location, id), Op: id, Host: host, WorkDir: l.buildDir(), Now: cfg.Now})
if err != nil {
return false, fmt.Errorf("build the generation: %w", err)
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
index 0a13762e4..60b729d10 100644
--- a/internal/host/movesland_test.go
+++ b/internal/host/movesland_test.go
@@ -71,18 +71,25 @@ type landingHarness struct {
parkDone bool
parkView func(context.Context, hostclaim.Observed) (park.View, error)
landFault func(point string) error
+ store lease.VersionedStore // when set, the target's versioned store
}
// fakeParkLeases refuses a second acquisition of a held location.
type fakeParkLeases struct {
mu sync.Mutex
held map[string]bool
+ leases map[string]*fakeHeldLease
acquired []string
}
+// fakeHeldLease is a park.HeldLease with the Context method hostbox's
+// renewingHeld has: a context derived from it is cancelled when the lease
+// is lost or released.
type fakeHeldLease struct {
- l *fakeParkLeases
- key string
+ l *fakeParkLeases
+ key string
+ ctx context.Context
+ cancel context.CancelFunc
}
func (l *fakeParkLeases) Acquire(_ context.Context, bucket, scope string) (park.HeldLease, error) {
@@ -97,7 +104,21 @@ func (l *fakeParkLeases) Acquire(_ context.Context, bucket, scope string) (park.
}
l.held[key] = true
l.acquired = append(l.acquired, key)
- return &fakeHeldLease{l: l, key: key}, nil
+ ctx, cancel := context.WithCancel(context.Background())
+ h := &fakeHeldLease{l: l, key: key, ctx: ctx, cancel: cancel}
+ if l.leases == nil {
+ l.leases = map[string]*fakeHeldLease{}
+ }
+ l.leases[key] = h
+ return h, nil
+}
+
+// Context is renewingHeld.Context: ctx, cancelled too when the lease is.
+func (h *fakeHeldLease) Context(ctx context.Context) context.Context {
+ c, cancel := context.WithCancel(ctx)
+ stop := context.AfterFunc(h.ctx, cancel)
+ context.AfterFunc(c, func() { stop() })
+ return c
}
func (l *fakeParkLeases) holding() []string {
@@ -125,6 +146,7 @@ func (h *fakeHeldLease) Release(context.Context) error {
h.l.mu.Lock()
defer h.l.mu.Unlock()
delete(h.l.held, h.key)
+ h.cancel()
return nil
}
func (h *fakeHeldLease) Close() { _ = h.Release(context.Background()) }
@@ -307,6 +329,9 @@ func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
c.ParkView = nil
}
c.OwnLease = h.leases.Acquire
+ if h.store != nil {
+ c.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return h.store, nil }
+ }
c.RenderHeld = func(context.Context, release.Store, registry.Route, string) (appconfig.HeldRender, error) {
h.renders++
return h.render()
@@ -385,11 +410,15 @@ func (h *landingHarness) freshTarget() {
h.leases.expire()
}
-// expire lapses every held lease, as the TTL does for a dead holder.
+// expire lapses every held lease, as the TTL does for a dead holder, and
+// cancels every context derived from them.
func (l *fakeParkLeases) expire() {
l.mu.Lock()
defer l.mu.Unlock()
l.held = map[string]bool{}
+ for _, h := range l.leases {
+ h.cancel()
+ }
}
func (h *landingHarness) landJournal(t *testing.T) *moves.Journal {
@@ -1434,3 +1463,67 @@ func TestLandingDeadlineBeforeTheInstallTransition(t *testing.T) {
})
}
}
+
+// retryingStore is a versioned store whose first generation write blocks
+// like a client retrying a request: the lease lapses while it is in flight,
+// and if nothing cancels the request its retry lands after the loss.
+type retryingStore struct {
+ lease.VersionedStore
+ h *landingHarness
+ prefix string
+ once sync.Once
+ landed bool
+ blocked bool
+}
+
+func (r *retryingStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ block := false
+ if strings.HasPrefix(key, r.prefix) {
+ r.once.Do(func() { block = true })
+ }
+ if !block {
+ return r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
+ }
+ r.blocked = true
+ r.h.leases.expire() // the holder goes dark while the request is in flight
+ select {
+ case <-ctx.Done():
+ return "", ctx.Err()
+ case <-time.After(1500 * time.Millisecond):
+ r.landed = true // the client's retry succeeds after the lease was lost
+ return r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
+ }
+}
+
+// Fix round 2: every generation request runs under a context derived from
+// the lease's, so a lease lost while a request is in flight cancels it and
+// its retries; nothing is recorded. The replay builds afresh.
+func TestLandingBuildIsCancelledWhenTheLeaseIsLostMidRequest(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ rs := &retryingStore{VersionedStore: h.vs, h: h, prefix: prefix}
+ h.store = rs
+ h.land(t)
+ h.store = nil
+ if !rs.blocked {
+ t.Fatal("the build never wrote under the generation")
+ }
+ if rs.landed {
+ t.Fatal("the request landed after the lease was lost: it was not cancelled")
+ }
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
+ t.Fatalf("record: %+v", m)
+ }
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ t.Fatalf("written on a lost lease: %s", k)
+ }
+ }
+ if j := h.landJournal(t); j.Done(landStepInstall) {
+ t.Fatal("the build was recorded on a lost lease")
+ }
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+}
diff --git a/internal/host/moveslandinstall.go b/internal/host/moveslandinstall.go
index 7ed2bdc58..a492fb29b 100644
--- a/internal/host/moveslandinstall.go
+++ b/internal/host/moveslandinstall.go
@@ -291,7 +291,7 @@ func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves
return moves.InstallManifest{}, waiting, err
}
defer l.releaseLease()
- m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ m, err := moves.BuildGeneration(leaseContext(ctx, held), moves.BuildInputs{Store: leasedStore{VersionedStore: vs, check: held.Check}, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
GenerationPrefix: moves.GenerationPrefix(holder.Location, l.g.ID), Op: l.g.ID, Host: l.rec.Host, WorkDir: l.buildDir(), Now: cfg.Now})
if err != nil {
return moves.InstallManifest{}, false, fmt.Errorf("rebuild the generation: %w", err)
diff --git a/internal/host/moveslandpark.go b/internal/host/moveslandpark.go
index 58e0d34cd..b57c0ef15 100644
--- a/internal/host/moveslandpark.go
+++ b/internal/host/moveslandpark.go
@@ -176,12 +176,29 @@ func (l *landingRun) settleLease() {
}
}
+// leaseContext derives the context every generation request runs under from
+// the lease's own (fix round 2): hostbox's renewingHeld, the production
+// park.HeldLease, cancels it when the lease is lost or released, so a
+// request in flight, and the S3 client's retries of it, stop with the lease
+// rather than landing after it. A lease without a Context method (a plain
+// fake) leaves ctx as it is; leasedStore's check before each request still
+// stands.
+func leaseContext(ctx context.Context, held park.HeldLease) context.Context {
+ if bound, ok := held.(interface {
+ Context(context.Context) context.Context
+ }); ok {
+ return bound.Context(ctx)
+ }
+ return ctx
+}
+
// leasedStore is the own bucket by version under live lease authority: check
// (the held lease's Check) runs before every request, so a generation write
// after the lease was lost is refused before it is sent, as the cut's
-// guarded staging writes are (fix round 1, item 3). The build's reads run
-// under it too: a holder that has lost the lease should not read ahead
-// either.
+// guarded staging writes are (fix round 1, item 3); the requests themselves
+// run under leaseContext, so one already in flight when the lease is lost is
+// cancelled with its retries (fix round 2). The build's reads run under both
+// too: a holder that has lost the lease should not read ahead either.
type leasedStore struct {
lease.VersionedStore
check func(context.Context) error