amadan

carlos / platform Public

moves build, fix round 1: every recorded or replay-chosen version id must pin an immutable object (task-8-review)

A fresh PutIfAbsent's returned id and the current version a replay chooses
must both pass lease.UsableVersionID before they are read or recorded; ""
and "null" fail the build with ErrBuildUnpinned and no install.json is
written. Tests: both ids on creation (intent, snapshot, anchor, manifest)
and on replay (intent, snapshot); the replay tests advance the clock
between the attempt and the replay; the sealed-client refusals each
isolate one property with header-sized fixtures.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1686431e76f80de063a096517ba45e2de87c1921 parent ca9ccc9
2 files changed, +160 −13
  • internal/moves/build.go +32 −3
  • internal/moves/build_test.go +128 −10
diff --git a/internal/moves/build.go b/internal/moves/build.go
index d6dc9caa0..f50022234 100644
--- a/internal/moves/build.go
+++ b/internal/moves/build.go
@@ -32,8 +32,24 @@ var (
// snapshot or anchor bytes, another manifest). The generation is new
// and the build is deterministic, so this is never a replay.
ErrBuildDiverged = errors.New("moves: the prepared generation holds an object this build did not make")
+ // ErrBuildUnpinned: a version id the build would record or read back
+ // pins nothing ("" is no version; "null" is what a bucket whose
+ // versioning is suspended answers, and the next write to that key
+ // replaces it). Every artifact identity the manifest names, and every
+ // current version a replay chooses, must be immutable, so the build
+ // fails before reading or recording such an id and no install.json is
+ // written (G27 on the output side).
+ ErrBuildUnpinned = errors.New("moves: a prepared generation object has no usable version id")
)
+// pinned is the check every recorded or chosen version id passes.
+func pinned(key, versionID string) error {
+ if !lease.UsableVersionID(versionID) {
+ return fmt.Errorf("%w: %s as %q", ErrBuildUnpinned, key, versionID)
+ }
+ return nil
+}
+
// buildContractVersion is the V of the intent and the manifest.
const buildContractVersion = 1
@@ -297,7 +313,8 @@ func verifySealedFile(ctx context.Context, client *SealedClient, info *ltx.FileI
}
// currentVersion is the object version currently under key, found by an
-// exact-key listing (a store names versions, never a bare key).
+// exact-key listing (a store names versions, never a bare key). Its id
+// must pin an immutable object before anything is read by it.
func currentVersion(ctx context.Context, in BuildInputs, key string) (lease.ObjectVersion, error) {
vs, err := in.Store.ListVersions(ctx, in.Bucket, key)
if err != nil {
@@ -305,6 +322,9 @@ func currentVersion(ctx context.Context, in BuildInputs, key string) (lease.Obje
}
for _, v := range vs {
if v.Key == key && v.IsLatest && !v.DeleteMarker {
+ if err := pinned(key, v.VersionID); err != nil {
+ return lease.ObjectVersion{}, err
+ }
return v, nil
}
}
@@ -318,6 +338,9 @@ func currentVersion(ctx context.Context, in BuildInputs, key string) (lease.Obje
func putOrReuse(ctx context.Context, in BuildInputs, key string, body []byte) (string, error) {
v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body)
if err == nil {
+ if err := pinned(key, v); err != nil {
+ return "", err
+ }
return v, nil
}
if !errors.Is(err, lease.ErrExists) {
@@ -346,7 +369,10 @@ func putOrReadIntent(ctx context.Context, in BuildInputs, want BuildIntent) (Bui
if err != nil {
return BuildIntent{}, err
}
- if _, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
+ if v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
+ if err := pinned(key, v); err != nil {
+ return BuildIntent{}, err
+ }
return want, nil
} else if !errors.Is(err, lease.ErrExists) {
return BuildIntent{}, fmt.Errorf("moves: put %s: %w", key, err)
@@ -379,7 +405,10 @@ func putOrReadManifest(ctx context.Context, in BuildInputs, want InstallManifest
if err != nil {
return InstallManifest{}, err
}
- if _, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
+ if v, err := in.Store.PutIfAbsent(ctx, in.Bucket, key, body); err == nil {
+ if err := pinned(key, v); err != nil {
+ return InstallManifest{}, err
+ }
return want, nil
} else if !errors.Is(err, lease.ErrExists) {
return InstallManifest{}, fmt.Errorf("moves: put %s: %w", key, err)
diff --git a/internal/moves/build_test.go b/internal/moves/build_test.go
index 8958eb385..fa0096f87 100644
--- a/internal/moves/build_test.go
+++ b/internal/moves/build_test.go
@@ -264,13 +264,50 @@ func sameManifest(a, b InstallManifest) bool {
return a == b
}
+var testClock = time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)
+
func inputs(b *leasetest.VersionedBucket, src []SealedObject, n uint64, work string) BuildInputs {
+ return inputsAt(b, src, n, work, testClock)
+}
+
+// inputsAt is inputs with the build's clock at now: a replay runs later
+// than the attempt it replays.
+func inputsAt(store lease.VersionedStore, src []SealedObject, n uint64, work string, now time.Time) BuildInputs {
return BuildInputs{
- Store: b, Bucket: testBucket, Source: src, N: n,
+ Store: store, Bucket: testBucket, Source: src, N: n,
GenerationPrefix: testGenPrefix(), Op: testOp, Host: testHost,
WorkDir: work,
- Now: func() time.Time { return time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC) },
+ Now: func() time.Time { return now },
+ }
+}
+
+// versionRewriter is a VersionedStore whose answers name a version id the
+// builder must refuse: what a bucket whose versioning was suspended after
+// the seal returns for a new write ("null"), or a provider that returns
+// none (""). putAs rewrites the id a fresh PutIfAbsent returns for a key;
+// listAs rewrites the current version a replay finds for a key.
+type versionRewriter struct {
+ lease.VersionedStore
+ putAs map[string]string
+ listAs map[string]string
+}
+
+func (r *versionRewriter) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ v, err := r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
+ if as, ok := r.putAs[key]; ok && err == nil {
+ return as, nil
+ }
+ return v, err
+}
+
+func (r *versionRewriter) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ vs, err := r.VersionedStore.ListVersions(ctx, bucket, prefix)
+ for i := range vs {
+ if as, ok := r.listAs[vs[i].Key]; ok && vs[i].IsLatest {
+ vs[i].VersionID = as
+ }
}
+ return vs, err
}
// restoreGeneration restores the built generation through litestream's own
@@ -409,23 +446,98 @@ func TestSealedClientServesOnlySealedVersions(t *testing.T) {
t.Fatalf("the sealed key has %d versions after the refused writes, want 2", n)
}
+ // Each refusal below isolates one property: every other field of the
+ // fixture is valid (a usable version, an LTX key, a header-sized object).
+ okSize := int64(len(sealed))
// G27: a version id that pins nothing is refused before any read.
for _, bad := range []string{"", "null"} {
- if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: bad, Size: 10}}); err == nil {
+ if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: bad, Size: okSize}}); err == nil {
t.Fatalf("version %q accepted", bad)
}
}
// A key that is not <level>/<min>-<max>.ltx has no position.
- if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: testLane + "lock.json", VersionID: v1, Size: 10}}); err == nil {
+ if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: testLane + "lock.json", VersionID: v1, Size: okSize}}); err == nil {
t.Fatal("a non-LTX key accepted")
}
// Two objects at one position would make the plan ambiguous.
if _, err := NewSealedClient(b, testBucket, []SealedObject{
- {Key: key, VersionID: v1, Size: 10},
- {Key: "other/" + "0000/" + ltx.FormatFilename(1, 1), VersionID: v1, Size: 10},
+ {Key: key, VersionID: v1, Size: okSize},
+ {Key: "other/" + "0000/" + ltx.FormatFilename(1, 1), VersionID: v1, Size: okSize},
}); err == nil {
t.Fatal("two objects at one position accepted")
}
+ // A size below an LTX header is refused on its own.
+ if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize - 1}}); err == nil {
+ t.Fatal("an object smaller than an LTX header accepted")
+ }
+ if _, err := NewSealedClient(b, testBucket, []SealedObject{{Key: key, VersionID: v1, Size: ltx.HeaderSize}}); err != nil {
+ t.Fatalf("a header-sized object refused: %v", err)
+ }
+}
+
+// Every artifact identity the build records, and every current version a
+// replay chooses, must pin an immutable object: "" and "null" fail the
+// build before they are read or recorded, and no install.json is written.
+func TestBuildRefusesUnusableArtifactVersions(t *testing.T) {
+ src, n := sourceReplica(t, 1, true)
+ prefix := testGenPrefix()
+ ctx := context.Background()
+ keys := map[string]string{
+ "intent": BuildIntentKey(prefix),
+ "snapshot": SnapshotKey(prefix, n),
+ "anchor": AnchorKey(prefix, n),
+ "manifest": InstallManifestKey(prefix),
+ }
+ for name, key := range keys {
+ for _, bad := range []string{"", "null"} {
+ t.Run("created "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
+ b := newBucket()
+ seal := sealReplica(t, b, src, testLane)
+ store := &versionRewriter{VersionedStore: b, putAs: map[string]string{key: bad}}
+ _, err := BuildGeneration(ctx, inputsAt(store, seal, n, filepath.Join(t.TempDir(), "build"), testClock))
+ if !errors.Is(err, ErrBuildUnpinned) {
+ t.Fatalf("err = %v, want ErrBuildUnpinned", err)
+ }
+ if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok && name != "manifest" {
+ t.Fatal("install.json was written")
+ }
+ })
+ }
+ }
+ // On replay: the attempt crashed after the snapshot PUT; the replay's
+ // listing names an unusable current version for the intent or the
+ // snapshot, as a bucket whose versioning was suspended would.
+ for _, name := range []string{"intent", "snapshot"} {
+ for _, bad := range []string{"", "null"} {
+ t.Run("replayed "+name+" as "+fmt.Sprintf("%q", bad), func(t *testing.T) {
+ b := newBucket()
+ seal := sealReplica(t, b, src, testLane)
+ work := filepath.Join(t.TempDir(), "build")
+ boom := errors.New("the box died")
+ b.FailOn = func(op, key string) error {
+ if op == "put" && key == keys["anchor"] {
+ return boom
+ }
+ return nil
+ }
+ if _, err := BuildGeneration(ctx, inputs(b, seal, n, work)); !errors.Is(err, boom) {
+ t.Fatalf("first run = %v, want the injected failure", err)
+ }
+ b.FailOn = nil
+ store := &versionRewriter{VersionedStore: b, listAs: map[string]string{keys[name]: bad}}
+ _, err := BuildGeneration(ctx, inputsAt(store, seal, n, work, testClock.Add(time.Hour)))
+ if !errors.Is(err, ErrBuildUnpinned) {
+ t.Fatalf("replay = %v, want ErrBuildUnpinned", err)
+ }
+ if _, ok := b.Current(testBucket, InstallManifestKey(prefix)); ok {
+ t.Fatal("install.json was written on the replay")
+ }
+ if vs := versionsOf(t, b, keys["snapshot"]); len(vs) != 1 {
+ t.Fatalf("the snapshot has %d versions, want the attempt's one", len(vs))
+ }
+ })
+ }
+ }
}
func TestBuildGenerationReachesNAndWritesSnapshotAndAnchor(t *testing.T) {
@@ -609,11 +721,16 @@ func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) {
}
first, _ := b.GetVersion(ctx, testBucket, SnapshotKey(prefix, n), snaps[0].VersionID)
+ // The replay runs later: a build that encoded with its own clock
+ // instead of the intent's would write other bytes.
b.FailOn = nil
- m, err := BuildGeneration(ctx, inputs(b, seal, n, work))
+ m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
if err != nil {
t.Fatalf("replay: %v", err)
}
+ if !m.BuiltAt.Equal(testClock.Add(time.Hour)) {
+ t.Fatalf("BuiltAt = %v, want the replay's clock", m.BuiltAt)
+ }
snaps = versionsOf(t, b, SnapshotKey(prefix, n))
if len(snaps) != 1 {
t.Fatalf("after the replay the snapshot has %d versions, want 1", len(snaps))
@@ -631,8 +748,9 @@ func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) {
if n := len(versionsOf(t, b, BuildIntentKey(prefix))); n != 1 {
t.Fatalf("intent has %d versions, want 1", n)
}
- // A third run is a pure replay: same manifest, no new versions.
- m2, err := BuildGeneration(ctx, inputs(b, seal, n, work))
+ // A third run, later again, is a pure replay: the stored manifest,
+ // its BuiltAt included, and no new versions.
+ m2, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(2*time.Hour)))
if err != nil {
t.Fatalf("second replay: %v", err)
}
@@ -694,7 +812,7 @@ func TestBuildReplayReusesIdenticalSnapshot(t *testing.T) {
}
}
b.FailOn = nil
- m, err := BuildGeneration(ctx, inputs(b, seal, n, work))
+ m, err := BuildGeneration(ctx, inputsAt(b, seal, n, work, testClock.Add(time.Hour)))
if err != nil {
t.Fatalf("replay: %v", err)
}