amadan

carlos / platform Public

host: the landing catches the claim up at restored, a refused install is never an install, every generation write runs under the lease, and the deadline is checked before the install transition (Task 10 fix round 1)

The restored step syncs the claim before the pass returns, so a crash
after the record's last CAS, or a refused mirror, converges through the
pass. restore returns installed, marked or waiting: a build that does not
verify marks the rollback and leaves the phase at configured with no
digest, no database and no marker. BuildGeneration and the rebuild run
over a store that checks the location's lease before every request, the
lease is checked again before the result is recorded, and the rebuild
takes the lease first and waits, writing nothing, while another holder
has it. landing_deadline is checked after the release gate and again
right before the transition to restored.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 0035ce9e8c888a66a6a3e9cc969c840091411a2c parent ba73b15
4 files changed, +360 −42
  • internal/host/movesland.go +40 −5
  • internal/host/movesland_test.go +193 −0
  • internal/host/moveslandinstall.go +73 −37
  • internal/host/moveslandpark.go +54 −0
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
index 746de3755..16329dfe6 100644
--- a/internal/host/movesland.go
+++ b/internal/host/movesland.go
@@ -197,8 +197,21 @@ func (l *landingRun) step(ctx context.Context) (bool, error) {
return l.local_(ctx)
case provision.LandConfigured:
return l.configured(ctx)
+ case provision.LandRestored:
+ // The install is complete and the probe's (Plan 3). The one thing
+ // still this pass's is the claim half of the last CAS: a crash after
+ // the record's restored, or a mirror that failed, leaves the claim
+ // behind, and nothing later would walk it (fix round 1, item 1).
+ _, err := l.syncClaimPhase(ctx)
+ return false, err
}
- return false, nil // restored and later: the probe's (Plan 3)
+ return false, nil // probing and later: the probe's (Plan 3)
+}
+
+// pastDeadline reports whether landing_deadline has passed now.
+func (l *landingRun) pastDeadline() bool {
+ d := l.rec.Move.LandingDeadline
+ return d != nil && l.cfg().Now().After(*d)
}
// landingBefore reports whether phase comes before other in LandingOrder.
@@ -691,11 +704,18 @@ func (l *landingRun) building(ctx context.Context) (bool, error) {
if err := held.Check(ctx); err != nil {
return false, fmt.Errorf("the location's lease: %w", err)
}
- m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ // 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,
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)
}
+ if err := held.Check(ctx); err != nil {
+ return false, fmt.Errorf("the location's lease after the build: %w", err)
+ }
body, err := json.Marshal(m)
if err != nil {
return false, err
@@ -980,9 +1000,24 @@ func (l *landingRun) configured(ctx context.Context) (bool, error) {
if err := l.fault("after-release-ready"); err != nil {
return false, err
}
- sha, installed, err := l.restore(ctx, obs)
- if err != nil || !installed {
- return installed, err
+ // The render, the release gate and the install can outlast the deadline
+ // (fix round 1, item 4): it is checked again before the install and
+ // right before the transition that names it complete.
+ if l.pastDeadline() {
+ return l.markRollback(ctx, provision.RollbackLandingDeadline)
+ }
+ sha, outcome, err := l.restore(ctx, obs)
+ if err != nil {
+ return false, err
+ }
+ switch outcome {
+ case restoreWaiting:
+ return false, nil
+ case restoreMarked:
+ return true, nil // the record changed: the rollback marker, not an install
+ }
+ if l.pastDeadline() {
+ return l.markRollback(ctx, provision.RollbackLandingDeadline)
}
return l.advance(ctx, provision.LandConfigured, provision.LandRestored, func(m *provision.MoveState) {
if m.Landing == nil {
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
index 66c2bbe50..0a13762e4 100644
--- a/internal/host/movesland_test.go
+++ b/internal/host/movesland_test.go
@@ -1241,3 +1241,196 @@ func TestTargetLiveSetCoversTheCutAndTheLanding(t *testing.T) {
t.Fatal("a finished move's verdict was kept")
}
}
+
+// Fix round 1, item 1: the pass catches the claim up at restored before it
+// returns. A crash after the record's restored CAS, or a following claim
+// CAS that fails, leaves the claim behind; the next pass (a fresh worker)
+// converges through movesPass alone.
+func TestLandingRestoredCatchesTheClaimUp(t *testing.T) {
+ t.Run("crash after the record's restored", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landAt = "after-record-restored"
+ h.crashLand(t)
+ if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
+ t.Fatalf("fixture: claim %+v", c.Op)
+ }
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+ })
+ t.Run("the following claim CAS fails", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ // At the boundary the claim is moved two steps back, so the mirror's
+ // SetMovePhase(configured, restored) is refused and the claim stays
+ // behind; the pass must not fail, and the next pass walks it up.
+ h.landFault = func(point string) error {
+ if point == "after-record-restored" {
+ rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandBuilding })
+ }
+ return nil
+ }
+ h.land(t)
+ h.landFault = nil
+ if !containsString(h.landHits, "after-record-restored") {
+ t.Fatalf("the boundary was never reached: %v", h.landHits)
+ }
+ // The refused mirror is logged, and the same pass's restored step
+ // walks the claim from building to restored before it returns.
+ h.installed(t)
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+ })
+}
+
+// Fix round 1, item 2: a build that does not verify marks the rollback and
+// leaves the phase at configured: no transition to restored, no digest, no
+// database installed, no marker.
+func TestLandingCorruptBuildMarksRollback(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landFault = func(point string) error {
+ if point == "before-db-verify" {
+ f, err := os.OpenFile(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build", p1Host+".db"), os.O_APPEND|os.O_WRONLY, 0o600)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := f.Write([]byte("corrupt")); err != nil {
+ t.Fatal(err)
+ }
+ f.Close()
+ }
+ return nil
+ }
+ h.land(t)
+ if !containsString(h.landHits, "before-db-verify") {
+ t.Fatalf("the verify point was never reached: %v", h.landHits)
+ }
+ m := h.read(t, p1Host).Move
+ if m.Reason != provision.RollbackInstallMismatch || m.LandingPhase != provision.LandConfigured || m.State != provision.MoveLanding {
+ t.Fatalf("after the corrupt build: %+v", m)
+ }
+ if m.Landing != nil && m.Landing.InstalledDBSHA256 != "" {
+ t.Fatalf("a digest was recorded: %+v", m.Landing)
+ }
+ dbPath := provision.DefaultDBPath(h.tdir, p1Host)
+ for _, p := range []string{dbPath, dbPath + ".gen"} {
+ if _, err := os.Stat(p); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("%s exists after a refused install: %v", p, err)
+ }
+ }
+ if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
+ t.Fatalf("claim: %+v", c.Op)
+ }
+}
+
+// Fix round 1, item 3: the generation is written only under live lease
+// authority. A rebuild (the scratch lost after published released the
+// lease) takes the location's lease first, and waits while another holder
+// has it, writing nothing; a lease lost during the build stops the build
+// at the next generation write, and the replay completes.
+func TestLandingRebuildNeedsTheLease(t *testing.T) {
+ ctx := context.Background()
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landAt = "after-record-configured"
+ h.crashLand(t)
+ if err := os.RemoveAll(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build")); err != nil {
+ t.Fatal(err)
+ }
+ h.freshTarget()
+ other, err := h.leases.Acquire(ctx, "data-uk", cutRoot)
+ if err != nil {
+ t.Fatal(err)
+ }
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ before := len(h.vs.callsOn(prefix))
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
+ t.Fatalf("installed without the lease: %+v", m)
+ }
+ if got := h.vs.callsOn(prefix); len(got) != before {
+ t.Fatalf("generation requests while another holder had the lease: %v", got[before:])
+ }
+ if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a database was installed without the lease: %v", err)
+ }
+ if err := other.Release(ctx); err != nil {
+ t.Fatal(err)
+ }
+ h.land(t)
+ h.installed(t)
+}
+
+func TestLandingBuildStopsWhenTheLeaseIsLost(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ lost := false
+ h.vs.deny = func(op, bucket, key string) error {
+ if op == "put-if-absent" && strings.HasPrefix(key, prefix) && !lost {
+ lost = true
+ h.leases.expire() // the holder went dark between two generation writes
+ }
+ return nil
+ }
+ h.land(t)
+ h.vs.deny = nil
+ if !lost {
+ t.Fatal("the build never wrote")
+ }
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
+ t.Fatalf("the build completed on a lost lease: %+v", m)
+ }
+ if _, ok := h.f.buckets["data-uk"][moves.InstallManifestKey(prefix)]; ok {
+ t.Fatal("install.json was written on a lost lease")
+ }
+ var written []string
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ written = append(written, strings.TrimPrefix(k, prefix))
+ }
+ }
+ if len(written) > 1 {
+ t.Fatalf("writes after the lease was lost: %v", written)
+ }
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+}
+
+// Fix round 1, item 4: landing_deadline is checked again right before the
+// install transition. Past it after the release gate, the rollback is marked
+// and nothing is installed; past it after the install itself, the rollback
+// is marked and restored is never written.
+func TestLandingDeadlineBeforeTheInstallTransition(t *testing.T) {
+ for _, point := range []string{"after-release-ready", "after-db-rename"} {
+ t.Run(point, func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landFault = func(p string) error {
+ if p == point {
+ h.clock.Advance(cutLandingWindow + time.Minute)
+ }
+ return nil
+ }
+ h.land(t)
+ m := h.read(t, p1Host).Move
+ if m.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandConfigured {
+ t.Fatalf("past the deadline at %s: %+v", point, m)
+ }
+ if m.Landing != nil && m.Landing.InstalledDBSHA256 != "" {
+ t.Fatalf("a digest was recorded: %+v", m.Landing)
+ }
+ _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host))
+ if point == "after-release-ready" && !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("installed past the deadline: %v", err)
+ }
+ if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
+ t.Fatalf("claim: %+v", c.Op)
+ }
+ })
+ }
+}
diff --git a/internal/host/moveslandinstall.go b/internal/host/moveslandinstall.go
index 509343b9c..7ed2bdc58 100644
--- a/internal/host/moveslandinstall.go
+++ b/internal/host/moveslandinstall.go
@@ -156,123 +156,159 @@ func sameBundle(a, b map[string]string) bool {
return true
}
+// restoreOutcome is what restore ended with: the database installed and its
+// markers written; the record changed instead (the rollback marker, fix
+// round 1, item 2: a marker written is never read as an install); or the
+// phase waiting for the next pass (the location's lease held elsewhere).
+type restoreOutcome int
+
+const (
+ restoreInstalled restoreOutcome = iota
+ restoreMarked
+ restoreWaiting
+)
+
// restore is the install of the built database (§7.6 restored, R6-3):
// its sha256 must be the manifest's and CheckDB must pass, else the
// rollback marker; then the rename into DBPath and the generation marker.
-// installed is false when the phase ended with the marker written. The
-// returned sha carries the record's "sha256:" prefix, as Source.DBSHA256
+// The returned sha carries the record's "sha256:" prefix, as Source.DBSHA256
// does.
-func (l *landingRun) restore(ctx context.Context, obs hostclaim.Observed) (sha string, installed bool, err error) {
+func (l *landingRun) restore(ctx context.Context, obs hostclaim.Observed) (sha string, outcome restoreOutcome, err error) {
cfg := l.cfg()
host, id := l.rec.Host, l.g.ID
- m, err := l.installManifest(ctx, obs)
- if err != nil {
- return "", false, err
+ m, waiting, err := l.installManifest(ctx, obs)
+ if err != nil || waiting {
+ return "", restoreWaiting, err
}
want := "sha256:" + m.DBSHA256
dbPath := l.dbPath()
present, err := regularFile(dbPath)
if err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
if present {
got, err := fileSHA256(dbPath)
if err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
if got != want {
- return "", false, fmt.Errorf("%s holds a database that is not this landing's build (%s, want %s)", dbPath, got, want)
+ return "", restoreWaiting, fmt.Errorf("%s holds a database that is not this landing's build (%s, want %s)", dbPath, got, want)
}
} else {
built := filepath.Join(l.buildDir(), host+".db")
if ok, err := regularFile(built); err != nil {
- return "", false, err
+ return "", restoreWaiting, err
} else if !ok {
// The scratch was lost (a rebuilt data directory): build again
- // from the pinned set, which reproduces the same bytes.
- if _, err := l.rebuild(ctx, obs); err != nil {
- return "", false, err
+ // from the pinned set, which reproduces the same bytes, under
+ // the location's lease.
+ if _, waiting, err := l.rebuild(ctx, obs); err != nil || waiting {
+ return "", restoreWaiting, err
}
}
+ if err := l.fault("before-db-verify"); err != nil {
+ return "", restoreWaiting, err
+ }
got, err := fileSHA256(built)
if err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
if got != want {
log.Printf("moves: %s: the built database is %s, the manifest says %s", host, got, want)
- ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
- return "", ok, err
+ return l.refuseInstall(ctx)
}
if err := lease.CheckDB(built); err != nil {
log.Printf("moves: %s: the built database does not pass its check: %v", host, err)
- ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
- return "", ok, err
+ return l.refuseInstall(ctx)
}
for _, side := range []string{"-wal", "-shm"} {
os.Remove(built + side)
}
if err := os.Rename(built, dbPath); err != nil {
- return "", false, fmt.Errorf("install the database: %w", err)
+ return "", restoreWaiting, fmt.Errorf("install the database: %w", err)
}
if err := syncPath(filepath.Dir(dbPath)); err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
}
if err := l.fault("after-db-rename"); err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
if err := cfg.WriteGen(dbPath+".gen", id); err != nil {
- return "", false, fmt.Errorf("write the generation marker: %w", err)
+ return "", restoreWaiting, fmt.Errorf("write the generation marker: %w", err)
}
if err := syncPath(filepath.Dir(dbPath)); err != nil {
- return "", false, err
+ return "", restoreWaiting, err
}
log.Printf("moves: %s: database installed at %s (%s), generation %s", host, dbPath, want, id)
- return want, true, nil
+ return want, restoreInstalled, nil
+}
+
+// refuseInstall marks the rollback for a build that does not verify. A
+// marker that landed is restoreMarked; one refused (the record moved on) is
+// restoreWaiting, and nothing is installed either way.
+func (l *landingRun) refuseInstall(ctx context.Context) (string, restoreOutcome, error) {
+ ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
+ if err != nil || !ok {
+ return "", restoreWaiting, err
+ }
+ return "", restoreMarked, nil
}
// installManifest is the journal's, or the build's again when the journal
-// lost it.
-func (l *landingRun) installManifest(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+// lost it. waiting is true when the rebuild could not take the lease.
+func (l *landingRun) installManifest(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, bool, error) {
j, err := l.journal()
if err != nil {
- return moves.InstallManifest{}, err
+ return moves.InstallManifest{}, false, err
}
var m moves.InstallManifest
if raw, ok := j.Steps[landStepInstall]; ok && json.Unmarshal([]byte(raw), &m) == nil && m.Op == l.g.ID && m.Host == l.rec.Host && m.DBSHA256 != "" {
- return m, nil
+ return m, false, nil
}
return l.rebuild(ctx, obs)
}
// rebuild runs BuildGeneration again over the seal: a replay that reuses
-// every object it finds and restores the same bytes into the scratch.
-func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+// every object it finds and restores the same bytes into the scratch. It
+// takes the location's lease first (published released it), runs every
+// generation request under it, checks it again before recording the result
+// and releases it after (fix round 1, item 3). waiting is true when another
+// holder has the lease; nothing is written then.
+func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, bool, error) {
cfg := l.cfg()
seal, err := journalledSeal(obs.Claim)
if err != nil {
- return moves.InstallManifest{}, err
+ return moves.InstallManifest{}, false, err
}
vs, err := l.versioned(ctx)
if err != nil {
- return moves.InstallManifest{}, err
+ return moves.InstallManifest{}, false, err
}
holder := *obs.Claim.Holder
- m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ held, waiting, err := l.lease(ctx, holder)
+ if waiting || err != nil {
+ 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,
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{}, fmt.Errorf("rebuild the generation: %w", err)
+ return moves.InstallManifest{}, false, fmt.Errorf("rebuild the generation: %w", err)
+ }
+ if err := held.Check(ctx); err != nil {
+ return moves.InstallManifest{}, false, fmt.Errorf("the location's lease after the rebuild: %w", err)
}
j, err := l.journal()
if err != nil {
- return m, err
+ return m, false, err
}
body, err := json.Marshal(m)
if err != nil {
- return m, err
+ return m, false, err
}
j.Mark(landStepInstall, string(body))
- return m, j.Write(cfg.DataDir)
+ return m, false, j.Write(cfg.DataDir)
}
// writeFileAtomic writes body to path through a temporary file in the same
diff --git a/internal/host/moveslandpark.go b/internal/host/moveslandpark.go
index eeb262e27..58e0d34cd 100644
--- a/internal/host/moveslandpark.go
+++ b/internal/host/moveslandpark.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
+ "io"
"log"
"strings"
"time"
@@ -174,3 +175,56 @@ func (l *landingRun) settleLease() {
l.releaseLease()
}
}
+
+// 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.
+type leasedStore struct {
+ lease.VersionedStore
+ check func(context.Context) error
+}
+
+func (s leasedStore) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ if err := s.check(ctx); err != nil {
+ return nil, fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.ListVersions(ctx, bucket, prefix)
+}
+
+func (s leasedStore) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) {
+ if err := s.check(ctx); err != nil {
+ return nil, fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.OpenVersion(ctx, bucket, key, versionID)
+}
+
+func (s leasedStore) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) {
+ if err := s.check(ctx); err != nil {
+ return nil, fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.GetVersion(ctx, bucket, key, versionID)
+}
+
+func (s leasedStore) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if err := s.check(ctx); err != nil {
+ return "", "", fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.CopyVersion(ctx, srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly)
+}
+
+func (s leasedStore) DeleteVersion(ctx context.Context, bucket, key, versionID string) error {
+ if err := s.check(ctx); err != nil {
+ return fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.DeleteVersion(ctx, bucket, key, versionID)
+}
+
+func (s leasedStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ if err := s.check(ctx); err != nil {
+ return "", fmt.Errorf("the location's lease: %w", err)
+ }
+ return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
+}