fix(snapshot): enforce checkpoint parent chain
- commit
- 305b83a3149a08f70f2eea1f17440008f4909122
- author
- brice · · 2026-07-25 23:12 UTC
- parent
- f1d1822
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
18 files · +223 -36
M internal/adapters/snapshot/load_bounds_test.go +2 -2
| @@ -49,7 +49,7 @@ func TestRepositoryLoadStopsAtGenerationTraversalBudget(t *testing.T) { | |||
| 49 | 49 | func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *testing.T) { | |
| 50 | 50 | repo := NewRepository(privateDir(t)) | |
| 51 | 51 | for generation := uint64(1); generation <= 3; generation++ { | |
| 52 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | |
| 52 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | |
| 53 | 53 | t.Fatal(err) | |
| 54 | 54 | } | |
| 55 | 55 | } | |
| @@ -71,7 +71,7 @@ func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *te | |||
| 71 | 71 | func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing.T) { | |
| 72 | 72 | repo := NewRepository(privateDir(t)) | |
| 73 | 73 | for generation := uint64(1); generation <= uint64(maxMaintenanceMarkedGenerations+1); generation++ { | |
| 74 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | |
| 74 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | |
| 75 | 75 | t.Fatal(err) | |
| 76 | 76 | } | |
| 77 | 77 | } | |
| @@ -49,7 +49,7 @@ func TestRepositoryLoadStopsAtGenerationTraversalBudget(t *testing.T) { | |||||
| 49 | func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *testing.T) { | 49 | func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *testing.T) { | ||
| 50 | repo := NewRepository(privateDir(t)) | 50 | repo := NewRepository(privateDir(t)) | ||
| 51 | for generation := uint64(1); generation <= 3; generation++ { | 51 | for generation := uint64(1); generation <= 3; generation++ { | ||
| 52 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | 52 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { |
| 53 | t.Fatal(err) | 53 | t.Fatal(err) | ||
| 54 | } | 54 | } | ||
| 55 | } | 55 | } | ||
| @@ -71,7 +71,7 @@ func TestRepositoryLoadFallsBackNewestToOldestWithoutGenerationEnumeration(t *te | |||||
| 71 | func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing.T) { | 71 | func TestRepositoryMaintainDoesNotCollectBeyondRetainedMetadataBudget(t *testing.T) { | ||
| 72 | repo := NewRepository(privateDir(t)) | 72 | repo := NewRepository(privateDir(t)) | ||
| 73 | for generation := uint64(1); generation <= uint64(maxMaintenanceMarkedGenerations+1); generation++ { | 73 | for generation := uint64(1); generation <= uint64(maxMaintenanceMarkedGenerations+1); generation++ { | ||
| 74 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { | 74 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation)))); err != nil { |
| 75 | t.Fatal(err) | 75 | t.Fatal(err) | ||
| 76 | } | 76 | } | ||
| 77 | } | 77 | } | ||
M internal/adapters/snapshot/load_major_test.go +1 -1
| @@ -42,7 +42,7 @@ func TestRepositoryLoadFallsBackFromInvalidHead(t *testing.T) { | |||
| 42 | 42 | func TestRepositoryLoadFallsBackAcrossMultipleCorruptCandidates(t *testing.T) { | |
| 43 | 43 | repo := NewRepository(privateDir(t)) | |
| 44 | 44 | for generation := uint64(1); generation <= 4; generation++ { | |
| 45 | - | require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)}))) | |
| 45 | + | require.NoError(t, repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)}))) | |
| 46 | 46 | } | |
| 47 | 47 | key := legacyIncarnationID("named").String() | |
| 48 | 48 | for _, generation := range []uint64{4, 3} { | |
| @@ -42,7 +42,7 @@ func TestRepositoryLoadFallsBackFromInvalidHead(t *testing.T) { | |||||
| 42 | func TestRepositoryLoadFallsBackAcrossMultipleCorruptCandidates(t *testing.T) { | 42 | func TestRepositoryLoadFallsBackAcrossMultipleCorruptCandidates(t *testing.T) { | ||
| 43 | repo := NewRepository(privateDir(t)) | 43 | repo := NewRepository(privateDir(t)) | ||
| 44 | for generation := uint64(1); generation <= 4; generation++ { | 44 | for generation := uint64(1); generation <= 4; generation++ { | ||
| 45 | - | require.NoError(t, repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)}))) | 45 | + | require.NoError(t, repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)}))) |
| 46 | } | 46 | } | ||
| 47 | key := legacyIncarnationID("named").String() | 47 | key := legacyIncarnationID("named").String() | ||
| 48 | for _, generation := range []uint64{4, 3} { | 48 | for _, generation := range []uint64{4, 3} { | ||
M internal/adapters/snapshot/maintenance_test.go +6 -6
| @@ -348,7 +348,7 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |||
| 348 | 348 | } | |
| 349 | 349 | publishDone := make(chan error, 1) | |
| 350 | 350 | go func() { | |
| 351 | - | publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second"))) | |
| 351 | + | publishDone <- repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", 2, []byte("second"))) | |
| 352 | 352 | }() | |
| 353 | 353 | <-entered | |
| 354 | 354 | deleteReached := make(chan struct{}) | |
| @@ -374,7 +374,7 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |||
| 374 | 374 | func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |
| 375 | 375 | repo := NewRepository(privateDir(t)) | |
| 376 | 376 | for generation := uint64(1); generation <= 3; generation++ { | |
| 377 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |
| 377 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)})); err != nil { | |
| 378 | 378 | t.Fatal(err) | |
| 379 | 379 | } | |
| 380 | 380 | } | |
| @@ -412,7 +412,7 @@ func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |||
| 412 | 412 | func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) { | |
| 413 | 413 | repo := NewRepository(privateDir(t)) | |
| 414 | 414 | for generation := uint64(1); generation <= 3; generation++ { | |
| 415 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | |
| 415 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)})); err != nil { | |
| 416 | 416 | t.Fatal(err) | |
| 417 | 417 | } | |
| 418 | 418 | } | |
| @@ -690,7 +690,7 @@ func TestRepositoryMaintainMarksLargeGenerationSetBeforeSweeping(t *testing.T) { | |||
| 690 | 690 | repo := NewRepository(privateDir(t)) | |
| 691 | 691 | var newest ports.SnapshotPublication | |
| 692 | 692 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | |
| 693 | - | publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | |
| 693 | + | publication := repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | |
| 694 | 694 | if err := repo.Publish(context.Background(), publication); err != nil { | |
| 695 | 695 | t.Fatal(err) | |
| 696 | 696 | } | |
| @@ -872,7 +872,7 @@ func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testin | |||
| 872 | 872 | key := legacyIncarnationID("named").String() | |
| 873 | 873 | publications := make([]ports.SnapshotPublication, 0, maintenanceBatch+2) | |
| 874 | 874 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | |
| 875 | - | publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | |
| 875 | + | publication := repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | |
| 876 | 876 | if err := repo.Publish(context.Background(), publication); err != nil { | |
| 877 | 877 | t.Fatal(err) | |
| 878 | 878 | } | |
| @@ -955,7 +955,7 @@ func TestRepositoryMaintainRestartsMarkAfterFailedPublication(t *testing.T) { | |||
| 955 | 955 | t.Fatal("sweep did not leave a pending stale object") | |
| 956 | 956 | } | |
| 957 | 957 | ||
| 958 | - | second := publicationWithTailShard(t, "named", 2, "ff") | |
| 958 | + | second := publicationWithCurrentParent(t, repo, publicationWithTailShard(t, "named", 2, "ff")) | |
| 959 | 959 | injected := errors.New("before HEAD") | |
| 960 | 960 | repo.hooks.beforeHeadWrite = func(string) error { return injected } | |
| 961 | 961 | if err := repo.Publish(context.Background(), second); !errors.Is(err, injected) { | |
| @@ -348,7 +348,7 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |||||
| 348 | } | 348 | } | ||
| 349 | publishDone := make(chan error, 1) | 349 | publishDone := make(chan error, 1) | ||
| 350 | go func() { | 350 | go func() { | ||
| 351 | - | publishDone <- repo.Publish(context.Background(), repositoryPublication(t, "named", 2, []byte("second"))) | 351 | + | publishDone <- repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", 2, []byte("second"))) |
| 352 | }() | 352 | }() | ||
| 353 | <-entered | 353 | <-entered | ||
| 354 | deleteReached := make(chan struct{}) | 354 | deleteReached := make(chan struct{}) | ||
| @@ -374,7 +374,7 @@ func TestRepositoryDeleteWaitsForPublication(t *testing.T) { | |||||
| 374 | func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | 374 | func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | ||
| 375 | repo := NewRepository(privateDir(t)) | 375 | repo := NewRepository(privateDir(t)) | ||
| 376 | for generation := uint64(1); generation <= 3; generation++ { | 376 | for generation := uint64(1); generation <= 3; generation++ { | ||
| 377 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | 377 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)})); err != nil { |
| 378 | t.Fatal(err) | 378 | t.Fatal(err) | ||
| 379 | } | 379 | } | ||
| 380 | } | 380 | } | ||
| @@ -412,7 +412,7 @@ func TestRepositoryMaintainRetainsNewestTwoCompleteGenerations(t *testing.T) { | |||||
| 412 | func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) { | 412 | func TestRepositoryMaintainPreservesIncompleteManifestReferences(t *testing.T) { | ||
| 413 | repo := NewRepository(privateDir(t)) | 413 | repo := NewRepository(privateDir(t)) | ||
| 414 | for generation := uint64(1); generation <= 3; generation++ { | 414 | for generation := uint64(1); generation <= 3; generation++ { | ||
| 415 | - | if err := repo.Publish(context.Background(), repositoryPublication(t, "named", generation, []byte{byte(generation)})); err != nil { | 415 | + | if err := repo.Publish(context.Background(), repositoryPublicationAfter(t, repo, "named", generation, []byte{byte(generation)})); err != nil { |
| 416 | t.Fatal(err) | 416 | t.Fatal(err) | ||
| 417 | } | 417 | } | ||
| 418 | } | 418 | } | ||
| @@ -690,7 +690,7 @@ func TestRepositoryMaintainMarksLargeGenerationSetBeforeSweeping(t *testing.T) { | |||||
| 690 | repo := NewRepository(privateDir(t)) | 690 | repo := NewRepository(privateDir(t)) | ||
| 691 | var newest ports.SnapshotPublication | 691 | var newest ports.SnapshotPublication | ||
| 692 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | 692 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | ||
| 693 | - | publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | 693 | + | publication := repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) |
| 694 | if err := repo.Publish(context.Background(), publication); err != nil { | 694 | if err := repo.Publish(context.Background(), publication); err != nil { | ||
| 695 | t.Fatal(err) | 695 | t.Fatal(err) | ||
| 696 | } | 696 | } | ||
| @@ -872,7 +872,7 @@ func TestRepositoryMaintainClassifiesUnpublishedGenerationsBeforeSweep(t *testin | |||||
| 872 | key := legacyIncarnationID("named").String() | 872 | key := legacyIncarnationID("named").String() | ||
| 873 | publications := make([]ports.SnapshotPublication, 0, maintenanceBatch+2) | 873 | publications := make([]ports.SnapshotPublication, 0, maintenanceBatch+2) | ||
| 874 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | 874 | for generation := uint64(1); generation <= maintenanceBatch+2; generation++ { | ||
| 875 | - | publication := repositoryPublication(t, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) | 875 | + | publication := repositoryPublicationAfter(t, repo, "named", generation, []byte(fmt.Sprintf("state-%d", generation))) |
| 876 | if err := repo.Publish(context.Background(), publication); err != nil { | 876 | if err := repo.Publish(context.Background(), publication); err != nil { | ||
| 877 | t.Fatal(err) | 877 | t.Fatal(err) | ||
| 878 | } | 878 | } | ||
| @@ -955,7 +955,7 @@ func TestRepositoryMaintainRestartsMarkAfterFailedPublication(t *testing.T) { | |||||
| 955 | t.Fatal("sweep did not leave a pending stale object") | 955 | t.Fatal("sweep did not leave a pending stale object") | ||
| 956 | } | 956 | } | ||
| 957 | 957 | ||||
| 958 | - | second := publicationWithTailShard(t, "named", 2, "ff") | 958 | + | second := publicationWithCurrentParent(t, repo, publicationWithTailShard(t, "named", 2, "ff")) |
| 959 | injected := errors.New("before HEAD") | 959 | injected := errors.New("before HEAD") | ||
| 960 | repo.hooks.beforeHeadWrite = func(string) error { return injected } | 960 | repo.hooks.beforeHeadWrite = func(string) error { return injected } | ||
| 961 | if err := repo.Publish(context.Background(), second); !errors.Is(err, injected) { | 961 | if err := repo.Publish(context.Background(), second); !errors.Is(err, injected) { | ||
M internal/adapters/snapshot/migration_v0.go +3 -0
| @@ -212,6 +212,9 @@ func uncertainLegacyError(operation string, cause error) error { | |||
| 212 | 212 | return fmt.Errorf("%w: %s: %v", ports.ErrLegacySnapshotUncertain, operation, cause) | |
| 213 | 213 | } | |
| 214 | 214 | ||
| 215 | + | // MigrateV1Checkpoint is the migration-only admission path for an authoritative | |
| 216 | + | // legacy checkpoint whose v1 format has no parent link. Runtime snapshot | |
| 217 | + | // publication must use Publish and cannot bypass its parent-chain validation. | |
| 215 | 218 | func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) { | |
| 216 | 219 | if err := ctx.Err(); err != nil { | |
| 217 | 220 | return domain.CheckpointRef{}, err | |
| @@ -212,6 +212,9 @@ func uncertainLegacyError(operation string, cause error) error { | |||||
| 212 | return fmt.Errorf("%w: %s: %v", ports.ErrLegacySnapshotUncertain, operation, cause) | 212 | return fmt.Errorf("%w: %s: %v", ports.ErrLegacySnapshotUncertain, operation, cause) | ||
| 213 | } | 213 | } | ||
| 214 | 214 | ||||
| 215 | + | // MigrateV1Checkpoint is the migration-only admission path for an authoritative | |||
| 216 | + | // legacy checkpoint whose v1 format has no parent link. Runtime snapshot | |||
| 217 | + | // publication must use Publish and cannot bypass its parent-chain validation. | |||
| 215 | func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) { | 218 | func (r *Repository) MigrateV1Checkpoint(ctx context.Context, req ports.SnapshotMigrationRequest) (domain.CheckpointRef, error) { | ||
| 216 | if err := ctx.Err(); err != nil { | 219 | if err := ctx.Err(); err != nil { | ||
| 217 | return domain.CheckpointRef{}, err | 220 | return domain.CheckpointRef{}, err | ||
M internal/adapters/snapshot/migration_v0_test.go +1 -1
| @@ -128,7 +128,7 @@ func TestMigrateV1CheckpointValidatesObjectEnvelopeAndKind(t *testing.T) { | |||
| 128 | 128 | } | |
| 129 | 129 | } | |
| 130 | 130 | ||
| 131 | - | func TestMigrationOver4096(t *testing.T) { | |
| 131 | + | func TestMigrateV1CheckpointAcceptsLegacyNilParentBeyondGenerationOne(t *testing.T) { | |
| 132 | 132 | dir := t.TempDir() | |
| 133 | 133 | require.NoError(t, os.Chmod(dir, 0o700)) | |
| 134 | 134 | repo := NewRepository(dir) | |
| @@ -128,7 +128,7 @@ func TestMigrateV1CheckpointValidatesObjectEnvelopeAndKind(t *testing.T) { | |||||
| 128 | } | 128 | } | ||
| 129 | } | 129 | } | ||
| 130 | 130 | ||||
| 131 | - | func TestMigrationOver4096(t *testing.T) { | 131 | + | func TestMigrateV1CheckpointAcceptsLegacyNilParentBeyondGenerationOne(t *testing.T) { |
| 132 | dir := t.TempDir() | 132 | dir := t.TempDir() | ||
| 133 | require.NoError(t, os.Chmod(dir, 0o700)) | 133 | require.NoError(t, os.Chmod(dir, 0o700)) | ||
| 134 | repo := NewRepository(dir) | 134 | repo := NewRepository(dir) | ||
M internal/adapters/snapshot/publish.go +13 -10
| @@ -34,12 +34,6 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl | |||
| 34 | 34 | if err := ctx.Err(); err != nil { | |
| 35 | 35 | return err | |
| 36 | 36 | } | |
| 37 | - | // A failed publication can still leave immutable blobs or a manifest behind. | |
| 38 | - | // Invalidate an in-progress GC mark before creating any publication storage. | |
| 39 | - | r.invalidateStorageEpoch(key) | |
| 40 | - | if err := r.ensureSession(publication.IncarnationID); err != nil { | |
| 41 | - | return err | |
| 42 | - | } | |
| 43 | 37 | // The common replay path needs neither a manifest decode nor a reference | |
| 44 | 38 | // map. Checking the immutable bytes directly also keeps its descriptor | |
| 45 | 39 | // reads beneath the pinned repository root. | |
| @@ -61,6 +55,14 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl | |||
| 61 | 55 | return nil | |
| 62 | 56 | } | |
| 63 | 57 | ||
| 58 | + | // Parent validation and all other publication checks above happen before | |
| 59 | + | // creating or invalidating repository state. A rejected child must leave the | |
| 60 | + | // authoritative checkpoint and its storage byte-for-byte unchanged. | |
| 61 | + | r.invalidateStorageEpoch(key) | |
| 62 | + | if err := r.ensureSession(publication.IncarnationID); err != nil { | |
| 63 | + | return err | |
| 64 | + | } | |
| 65 | + | ||
| 64 | 66 | // Retained refs from the authoritative generation were verified when their | |
| 65 | 67 | // immutable files were installed. They are intentionally not re-read when a | |
| 66 | 68 | // producer supplies them again. Every other referenced object is checked | |
| @@ -194,11 +196,12 @@ func (r *Repository) currentIncarnationPublication(ctx context.Context, publicat | |||
| 194 | 196 | if manifest.IncarnationID != publication.IncarnationID || manifest.Generation != generation { | |
| 195 | 197 | return 0, nil, nil, fmt.Errorf("current manifest identity mismatch") | |
| 196 | 198 | } | |
| 197 | - | // Parent checkpoints are optional in manifest v2. When a publisher supplies | |
| 198 | - | // one, it must bind the next generation to the authoritative current HEAD. | |
| 199 | - | if publication.ParentCheckpoint != nil { | |
| 199 | + | // Every normal child checkpoint is bound to the exact authoritative HEAD. | |
| 200 | + | // Nil-parent checkpoints beyond generation one are admitted only through the | |
| 201 | + | // explicitly migration-only MigrateV1Checkpoint path. | |
| 202 | + | if publication.Generation == generation+1 { | |
| 200 | 203 | wantParent := &domain.CheckpointRef{Generation: generation, ManifestDigest: digest} | |
| 201 | - | if publication.Generation == generation+1 && !checkpointRefEqual(publication.ParentCheckpoint, wantParent) { | |
| 204 | + | if !checkpointRefEqual(publication.ParentCheckpoint, wantParent) { | |
| 202 | 205 | return 0, nil, nil, fmt.Errorf("publication parent does not match current checkpoint") | |
| 203 | 206 | } | |
| 204 | 207 | } | |
| @@ -34,12 +34,6 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl | |||||
| 34 | if err := ctx.Err(); err != nil { | 34 | if err := ctx.Err(); err != nil { | ||
| 35 | return err | 35 | return err | ||
| 36 | } | 36 | } | ||
| 37 | - | // A failed publication can still leave immutable blobs or a manifest behind. | |||
| 38 | - | // Invalidate an in-progress GC mark before creating any publication storage. | |||
| 39 | - | r.invalidateStorageEpoch(key) | |||
| 40 | - | if err := r.ensureSession(publication.IncarnationID); err != nil { | |||
| 41 | - | return err | |||
| 42 | - | } | |||
| 43 | // The common replay path needs neither a manifest decode nor a reference | 37 | // The common replay path needs neither a manifest decode nor a reference | ||
| 44 | // map. Checking the immutable bytes directly also keeps its descriptor | 38 | // map. Checking the immutable bytes directly also keeps its descriptor | ||
| 45 | // reads beneath the pinned repository root. | 39 | // reads beneath the pinned repository root. | ||
| @@ -61,6 +55,14 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl | |||||
| 61 | return nil | 55 | return nil | ||
| 62 | } | 56 | } | ||
| 63 | 57 | ||||
| 58 | + | // Parent validation and all other publication checks above happen before | |||
| 59 | + | // creating or invalidating repository state. A rejected child must leave the | |||
| 60 | + | // authoritative checkpoint and its storage byte-for-byte unchanged. | |||
| 61 | + | r.invalidateStorageEpoch(key) | |||
| 62 | + | if err := r.ensureSession(publication.IncarnationID); err != nil { | |||
| 63 | + | return err | |||
| 64 | + | } | |||
| 65 | + | ||||
| 64 | // Retained refs from the authoritative generation were verified when their | 66 | // Retained refs from the authoritative generation were verified when their | ||
| 65 | // immutable files were installed. They are intentionally not re-read when a | 67 | // immutable files were installed. They are intentionally not re-read when a | ||
| 66 | // producer supplies them again. Every other referenced object is checked | 68 | // producer supplies them again. Every other referenced object is checked | ||
| @@ -194,11 +196,12 @@ func (r *Repository) currentIncarnationPublication(ctx context.Context, publicat | |||||
| 194 | if manifest.IncarnationID != publication.IncarnationID || manifest.Generation != generation { | 196 | if manifest.IncarnationID != publication.IncarnationID || manifest.Generation != generation { | ||
| 195 | return 0, nil, nil, fmt.Errorf("current manifest identity mismatch") | 197 | return 0, nil, nil, fmt.Errorf("current manifest identity mismatch") | ||
| 196 | } | 198 | } | ||
| 197 | - | // Parent checkpoints are optional in manifest v2. When a publisher supplies | 199 | + | // Every normal child checkpoint is bound to the exact authoritative HEAD. |
| 198 | - | // one, it must bind the next generation to the authoritative current HEAD. | 200 | + | // Nil-parent checkpoints beyond generation one are admitted only through the |
| 199 | - | if publication.ParentCheckpoint != nil { | 201 | + | // explicitly migration-only MigrateV1Checkpoint path. |
| 202 | + | if publication.Generation == generation+1 { | |||
| 200 | wantParent := &domain.CheckpointRef{Generation: generation, ManifestDigest: digest} | 203 | wantParent := &domain.CheckpointRef{Generation: generation, ManifestDigest: digest} | ||
| 201 | - | if publication.Generation == generation+1 && !checkpointRefEqual(publication.ParentCheckpoint, wantParent) { | 204 | + | if !checkpointRefEqual(publication.ParentCheckpoint, wantParent) { |
| 202 | return 0, nil, nil, fmt.Errorf("publication parent does not match current checkpoint") | 205 | return 0, nil, nil, fmt.Errorf("publication parent does not match current checkpoint") | ||
| 203 | } | 206 | } | ||
| 204 | } | 207 | } | ||
M internal/adapters/snapshot/publish_efficiency_test.go +8 -5
| @@ -8,6 +8,7 @@ import ( | |||
| 8 | 8 | "path/filepath" | |
| 9 | 9 | "testing" | |
| 10 | 10 | ||
| 11 | + | "github.com/bnema/vev/internal/domain" | |
| 11 | 12 | "github.com/bnema/vev/internal/ports" | |
| 12 | 13 | codec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 13 | 14 | ) | |
| @@ -117,7 +118,7 @@ func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) { | |||
| 117 | 118 | if err := repo.Publish(context.Background(), first); err != nil { | |
| 118 | 119 | t.Fatal(err) | |
| 119 | 120 | } | |
| 120 | - | second := repositoryPublication(t, "named", 2, []byte("second")) | |
| 121 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("second")) | |
| 121 | 122 | newTail := second.Objects[0] | |
| 122 | 123 | key := legacyIncarnationID(second.Name).String() | |
| 123 | 124 | if err := os.MkdirAll(filepath.Dir(repo.legacyObjectPath(key, newTail.Digest)), 0o700); err != nil { | |
| @@ -154,7 +155,7 @@ func largeIncrementalPublications(t testing.TB, name string, count int) (ports.S | |||
| 154 | 155 | sealed = append(sealed, codec.ObjectRef{Kind: codec.HistoryChunk, Digest: object.Digest, Size: uint32(len(object.Data))}) | |
| 155 | 156 | objects = append(objects, object) | |
| 156 | 157 | } | |
| 157 | - | makePublication := func(generation uint64, tailPayload, visiblePayload string) ports.SnapshotPublication { | |
| 158 | + | makePublication := func(generation uint64, parent *domain.CheckpointRef, tailPayload, visiblePayload string) ports.SnapshotPublication { | |
| 158 | 159 | t.Helper() | |
| 159 | 160 | tail, err := codec.MarshalObject(codec.HistoryTail, []byte(tailPayload)) | |
| 160 | 161 | if err != nil { | |
| @@ -164,13 +165,15 @@ func largeIncrementalPublications(t testing.TB, name string, count int) (ports.S | |||
| 164 | 165 | if err != nil { | |
| 165 | 166 | t.Fatal(err) | |
| 166 | 167 | } | |
| 167 | - | manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, IncarnationID: id, Name: name, Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Sealed: sealed, Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}) | |
| 168 | + | manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, IncarnationID: id, ParentCheckpoint: parent, Name: name, Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Sealed: sealed, Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}) | |
| 168 | 169 | if err != nil { | |
| 169 | 170 | t.Fatal(err) | |
| 170 | 171 | } | |
| 171 | - | return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)} | |
| 172 | + | return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, ParentCheckpoint: parent, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)} | |
| 172 | 173 | } | |
| 173 | - | return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2") | |
| 174 | + | first := makePublication(1, nil, "tail-1", "visible-1") | |
| 175 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |
| 176 | + | return first, makePublication(2, parent, "tail-2", "visible-2") | |
| 174 | 177 | } | |
| 175 | 178 | ||
| 176 | 179 | func seedCompletePublication(t testing.TB, repo *Repository, publication ports.SnapshotPublication) { | |
| @@ -8,6 +8,7 @@ import ( | |||||
| 8 | "path/filepath" | 8 | "path/filepath" | ||
| 9 | "testing" | 9 | "testing" | ||
| 10 | 10 | ||||
| 11 | + | "github.com/bnema/vev/internal/domain" | |||
| 11 | "github.com/bnema/vev/internal/ports" | 12 | "github.com/bnema/vev/internal/ports" | ||
| 12 | codec "github.com/bnema/vev/internal/usecase/snapshot" | 13 | codec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 13 | ) | 14 | ) | ||
| @@ -117,7 +118,7 @@ func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) { | |||||
| 117 | if err := repo.Publish(context.Background(), first); err != nil { | 118 | if err := repo.Publish(context.Background(), first); err != nil { | ||
| 118 | t.Fatal(err) | 119 | t.Fatal(err) | ||
| 119 | } | 120 | } | ||
| 120 | - | second := repositoryPublication(t, "named", 2, []byte("second")) | 121 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("second")) |
| 121 | newTail := second.Objects[0] | 122 | newTail := second.Objects[0] | ||
| 122 | key := legacyIncarnationID(second.Name).String() | 123 | key := legacyIncarnationID(second.Name).String() | ||
| 123 | if err := os.MkdirAll(filepath.Dir(repo.legacyObjectPath(key, newTail.Digest)), 0o700); err != nil { | 124 | if err := os.MkdirAll(filepath.Dir(repo.legacyObjectPath(key, newTail.Digest)), 0o700); err != nil { | ||
| @@ -154,7 +155,7 @@ func largeIncrementalPublications(t testing.TB, name string, count int) (ports.S | |||||
| 154 | sealed = append(sealed, codec.ObjectRef{Kind: codec.HistoryChunk, Digest: object.Digest, Size: uint32(len(object.Data))}) | 155 | sealed = append(sealed, codec.ObjectRef{Kind: codec.HistoryChunk, Digest: object.Digest, Size: uint32(len(object.Data))}) | ||
| 155 | objects = append(objects, object) | 156 | objects = append(objects, object) | ||
| 156 | } | 157 | } | ||
| 157 | - | makePublication := func(generation uint64, tailPayload, visiblePayload string) ports.SnapshotPublication { | 158 | + | makePublication := func(generation uint64, parent *domain.CheckpointRef, tailPayload, visiblePayload string) ports.SnapshotPublication { |
| 158 | t.Helper() | 159 | t.Helper() | ||
| 159 | tail, err := codec.MarshalObject(codec.HistoryTail, []byte(tailPayload)) | 160 | tail, err := codec.MarshalObject(codec.HistoryTail, []byte(tailPayload)) | ||
| 160 | if err != nil { | 161 | if err != nil { | ||
| @@ -164,13 +165,15 @@ func largeIncrementalPublications(t testing.TB, name string, count int) (ports.S | |||||
| 164 | if err != nil { | 165 | if err != nil { | ||
| 165 | t.Fatal(err) | 166 | t.Fatal(err) | ||
| 166 | } | 167 | } | ||
| 167 | - | manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, IncarnationID: id, Name: name, Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Sealed: sealed, Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}) | 168 | + | manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, IncarnationID: id, ParentCheckpoint: parent, Name: name, Tabs: []codec.ManifestTab{{Cols: 1, Rows: 1, Panes: []codec.ManifestPane{{ID: "p", Sealed: sealed, Tail: codec.ObjectRef{Kind: codec.HistoryTail, Digest: tail.Digest, Size: uint32(len(tail.Data))}, Visible: codec.ObjectRef{Kind: codec.Visible, Digest: visible.Digest, Size: uint32(len(visible.Data))}}}}}}) |
| 168 | if err != nil { | 169 | if err != nil { | ||
| 169 | t.Fatal(err) | 170 | t.Fatal(err) | ||
| 170 | } | 171 | } | ||
| 171 | - | return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)} | 172 | + | return ports.SnapshotPublication{IncarnationID: id, Name: name, Generation: generation, ParentCheckpoint: parent, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)} |
| 172 | } | 173 | } | ||
| 173 | - | return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2") | 174 | + | first := makePublication(1, nil, "tail-1", "visible-1") |
| 175 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |||
| 176 | + | return first, makePublication(2, parent, "tail-2", "visible-2") | |||
| 174 | } | 177 | } | ||
| 175 | 178 | ||||
| 176 | func seedCompletePublication(t testing.TB, repo *Repository, publication ports.SnapshotPublication) { | 179 | func seedCompletePublication(t testing.TB, repo *Repository, publication ports.SnapshotPublication) { | ||
A internal/adapters/snapshot/publish_parent_test.go +109 -0
| @@ -0,0 +1,109 @@ | |||
| 1 | + | package snapshot | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "crypto/sha256" | |
| 6 | + | "os" | |
| 7 | + | "path/filepath" | |
| 8 | + | "testing" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/vev/internal/domain" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | func TestRepositoryPublishAcceptsGenerationOneWithoutParent(t *testing.T) { | |
| 14 | + | repo := NewRepository(privateDir(t)) | |
| 15 | + | publication := repositoryPublication(t, "named", 1, []byte("one")) | |
| 16 | + | ||
| 17 | + | if err := repo.Publish(context.Background(), publication); err != nil { | |
| 18 | + | t.Fatalf("Publish generation 1 without parent: %v", err) | |
| 19 | + | } | |
| 20 | + | } | |
| 21 | + | ||
| 22 | + | func TestRepositoryPublishRejectsInvalidLaterParentWithoutMutation(t *testing.T) { | |
| 23 | + | tests := []struct { | |
| 24 | + | name string | |
| 25 | + | parent func(domain.CheckpointRef) *domain.CheckpointRef | |
| 26 | + | }{ | |
| 27 | + | {name: "nil", parent: func(domain.CheckpointRef) *domain.CheckpointRef { return nil }}, | |
| 28 | + | {name: "stale generation", parent: func(current domain.CheckpointRef) *domain.CheckpointRef { | |
| 29 | + | current.Generation-- | |
| 30 | + | return ¤t | |
| 31 | + | }}, | |
| 32 | + | {name: "mismatched digest", parent: func(current domain.CheckpointRef) *domain.CheckpointRef { | |
| 33 | + | current.ManifestDigest[0]++ | |
| 34 | + | return ¤t | |
| 35 | + | }}, | |
| 36 | + | } | |
| 37 | + | ||
| 38 | + | for _, test := range tests { | |
| 39 | + | t.Run(test.name, func(t *testing.T) { | |
| 40 | + | repo := NewRepository(privateDir(t)) | |
| 41 | + | first := repositoryPublication(t, "named", 1, []byte("one")) | |
| 42 | + | if err := repo.Publish(context.Background(), first); err != nil { | |
| 43 | + | t.Fatalf("Publish generation 1: %v", err) | |
| 44 | + | } | |
| 45 | + | firstRef := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |
| 46 | + | second := incarnationPublication(t, first.IncarnationID, first.Name, 2, &firstRef) | |
| 47 | + | if err := repo.Publish(context.Background(), second); err != nil { | |
| 48 | + | t.Fatalf("Publish generation 2: %v", err) | |
| 49 | + | } | |
| 50 | + | current := domain.CheckpointRef{Generation: 2, ManifestDigest: sha256.Sum256(second.Manifest)} | |
| 51 | + | third := incarnationPublication(t, first.IncarnationID, first.Name, 3, test.parent(current)) | |
| 52 | + | before := snapshotRepositoryFiles(t, repo.dir) | |
| 53 | + | ||
| 54 | + | if err := repo.Publish(context.Background(), third); err == nil { | |
| 55 | + | t.Fatal("Publish accepted invalid parent") | |
| 56 | + | } | |
| 57 | + | after := snapshotRepositoryFiles(t, repo.dir) | |
| 58 | + | if len(after) != len(before) { | |
| 59 | + | t.Fatalf("repository file count changed: before=%d after=%d", len(before), len(after)) | |
| 60 | + | } | |
| 61 | + | for path, data := range before { | |
| 62 | + | if string(after[path]) != string(data) { | |
| 63 | + | t.Fatalf("repository file %q mutated", path) | |
| 64 | + | } | |
| 65 | + | } | |
| 66 | + | }) | |
| 67 | + | } | |
| 68 | + | } | |
| 69 | + | ||
| 70 | + | func TestRepositoryPublishAcceptsExactCurrentParent(t *testing.T) { | |
| 71 | + | repo := NewRepository(privateDir(t)) | |
| 72 | + | first := repositoryPublication(t, "named", 1, []byte("one")) | |
| 73 | + | if err := repo.Publish(context.Background(), first); err != nil { | |
| 74 | + | t.Fatalf("Publish generation 1: %v", err) | |
| 75 | + | } | |
| 76 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |
| 77 | + | second := incarnationPublication(t, first.IncarnationID, first.Name, 2, parent) | |
| 78 | + | ||
| 79 | + | if err := repo.Publish(context.Background(), second); err != nil { | |
| 80 | + | t.Fatalf("Publish generation 2 with exact parent: %v", err) | |
| 81 | + | } | |
| 82 | + | } | |
| 83 | + | ||
| 84 | + | func snapshotRepositoryFiles(t *testing.T, root string) map[string][]byte { | |
| 85 | + | t.Helper() | |
| 86 | + | files := make(map[string][]byte) | |
| 87 | + | err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { | |
| 88 | + | if err != nil { | |
| 89 | + | return err | |
| 90 | + | } | |
| 91 | + | if entry.IsDir() { | |
| 92 | + | return nil | |
| 93 | + | } | |
| 94 | + | data, err := os.ReadFile(path) | |
| 95 | + | if err != nil { | |
| 96 | + | return err | |
| 97 | + | } | |
| 98 | + | relative, err := filepath.Rel(root, path) | |
| 99 | + | if err != nil { | |
| 100 | + | return err | |
| 101 | + | } | |
| 102 | + | files[relative] = data | |
| 103 | + | return nil | |
| 104 | + | }) | |
| 105 | + | if err != nil { | |
| 106 | + | t.Fatalf("snapshot repository files: %v", err) | |
| 107 | + | } | |
| 108 | + | return files | |
| 109 | + | } | |
| @@ -0,0 +1,109 @@ | |||||
| 1 | + | package snapshot | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "crypto/sha256" | |||
| 6 | + | "os" | |||
| 7 | + | "path/filepath" | |||
| 8 | + | "testing" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/vev/internal/domain" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | func TestRepositoryPublishAcceptsGenerationOneWithoutParent(t *testing.T) { | |||
| 14 | + | repo := NewRepository(privateDir(t)) | |||
| 15 | + | publication := repositoryPublication(t, "named", 1, []byte("one")) | |||
| 16 | + | ||||
| 17 | + | if err := repo.Publish(context.Background(), publication); err != nil { | |||
| 18 | + | t.Fatalf("Publish generation 1 without parent: %v", err) | |||
| 19 | + | } | |||
| 20 | + | } | |||
| 21 | + | ||||
| 22 | + | func TestRepositoryPublishRejectsInvalidLaterParentWithoutMutation(t *testing.T) { | |||
| 23 | + | tests := []struct { | |||
| 24 | + | name string | |||
| 25 | + | parent func(domain.CheckpointRef) *domain.CheckpointRef | |||
| 26 | + | }{ | |||
| 27 | + | {name: "nil", parent: func(domain.CheckpointRef) *domain.CheckpointRef { return nil }}, | |||
| 28 | + | {name: "stale generation", parent: func(current domain.CheckpointRef) *domain.CheckpointRef { | |||
| 29 | + | current.Generation-- | |||
| 30 | + | return ¤t | |||
| 31 | + | }}, | |||
| 32 | + | {name: "mismatched digest", parent: func(current domain.CheckpointRef) *domain.CheckpointRef { | |||
| 33 | + | current.ManifestDigest[0]++ | |||
| 34 | + | return ¤t | |||
| 35 | + | }}, | |||
| 36 | + | } | |||
| 37 | + | ||||
| 38 | + | for _, test := range tests { | |||
| 39 | + | t.Run(test.name, func(t *testing.T) { | |||
| 40 | + | repo := NewRepository(privateDir(t)) | |||
| 41 | + | first := repositoryPublication(t, "named", 1, []byte("one")) | |||
| 42 | + | if err := repo.Publish(context.Background(), first); err != nil { | |||
| 43 | + | t.Fatalf("Publish generation 1: %v", err) | |||
| 44 | + | } | |||
| 45 | + | firstRef := domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |||
| 46 | + | second := incarnationPublication(t, first.IncarnationID, first.Name, 2, &firstRef) | |||
| 47 | + | if err := repo.Publish(context.Background(), second); err != nil { | |||
| 48 | + | t.Fatalf("Publish generation 2: %v", err) | |||
| 49 | + | } | |||
| 50 | + | current := domain.CheckpointRef{Generation: 2, ManifestDigest: sha256.Sum256(second.Manifest)} | |||
| 51 | + | third := incarnationPublication(t, first.IncarnationID, first.Name, 3, test.parent(current)) | |||
| 52 | + | before := snapshotRepositoryFiles(t, repo.dir) | |||
| 53 | + | ||||
| 54 | + | if err := repo.Publish(context.Background(), third); err == nil { | |||
| 55 | + | t.Fatal("Publish accepted invalid parent") | |||
| 56 | + | } | |||
| 57 | + | after := snapshotRepositoryFiles(t, repo.dir) | |||
| 58 | + | if len(after) != len(before) { | |||
| 59 | + | t.Fatalf("repository file count changed: before=%d after=%d", len(before), len(after)) | |||
| 60 | + | } | |||
| 61 | + | for path, data := range before { | |||
| 62 | + | if string(after[path]) != string(data) { | |||
| 63 | + | t.Fatalf("repository file %q mutated", path) | |||
| 64 | + | } | |||
| 65 | + | } | |||
| 66 | + | }) | |||
| 67 | + | } | |||
| 68 | + | } | |||
| 69 | + | ||||
| 70 | + | func TestRepositoryPublishAcceptsExactCurrentParent(t *testing.T) { | |||
| 71 | + | repo := NewRepository(privateDir(t)) | |||
| 72 | + | first := repositoryPublication(t, "named", 1, []byte("one")) | |||
| 73 | + | if err := repo.Publish(context.Background(), first); err != nil { | |||
| 74 | + | t.Fatalf("Publish generation 1: %v", err) | |||
| 75 | + | } | |||
| 76 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: sha256.Sum256(first.Manifest)} | |||
| 77 | + | second := incarnationPublication(t, first.IncarnationID, first.Name, 2, parent) | |||
| 78 | + | ||||
| 79 | + | if err := repo.Publish(context.Background(), second); err != nil { | |||
| 80 | + | t.Fatalf("Publish generation 2 with exact parent: %v", err) | |||
| 81 | + | } | |||
| 82 | + | } | |||
| 83 | + | ||||
| 84 | + | func snapshotRepositoryFiles(t *testing.T, root string) map[string][]byte { | |||
| 85 | + | t.Helper() | |||
| 86 | + | files := make(map[string][]byte) | |||
| 87 | + | err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { | |||
| 88 | + | if err != nil { | |||
| 89 | + | return err | |||
| 90 | + | } | |||
| 91 | + | if entry.IsDir() { | |||
| 92 | + | return nil | |||
| 93 | + | } | |||
| 94 | + | data, err := os.ReadFile(path) | |||
| 95 | + | if err != nil { | |||
| 96 | + | return err | |||
| 97 | + | } | |||
| 98 | + | relative, err := filepath.Rel(root, path) | |||
| 99 | + | if err != nil { | |||
| 100 | + | return err | |||
| 101 | + | } | |||
| 102 | + | files[relative] = data | |||
| 103 | + | return nil | |||
| 104 | + | }) | |||
| 105 | + | if err != nil { | |||
| 106 | + | t.Fatalf("snapshot repository files: %v", err) | |||
| 107 | + | } | |||
| 108 | + | return files | |||
| 109 | + | } | |||
M internal/adapters/snapshot/repository_durability_test.go +1 -1
| @@ -78,7 +78,7 @@ func TestRepositoryFaultsAtEveryPublicationBoundary(t *testing.T) { | |||
| 78 | 78 | if err := repo.Publish(context.Background(), first); err != nil { | |
| 79 | 79 | t.Fatal(err) | |
| 80 | 80 | } | |
| 81 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | |
| 81 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) | |
| 82 | 82 | if err := repo.ensureSession(second.IncarnationID); err != nil { | |
| 83 | 83 | t.Fatal(err) | |
| 84 | 84 | } | |
| @@ -78,7 +78,7 @@ func TestRepositoryFaultsAtEveryPublicationBoundary(t *testing.T) { | |||||
| 78 | if err := repo.Publish(context.Background(), first); err != nil { | 78 | if err := repo.Publish(context.Background(), first); err != nil { | ||
| 79 | t.Fatal(err) | 79 | t.Fatal(err) | ||
| 80 | } | 80 | } | ||
| 81 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | 81 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) |
| 82 | if err := repo.ensureSession(second.IncarnationID); err != nil { | 82 | if err := repo.ensureSession(second.IncarnationID); err != nil { | ||
| 83 | t.Fatal(err) | 83 | t.Fatal(err) | ||
| 84 | } | 84 | } | ||
M internal/adapters/snapshot/repository_fault_test.go +3 -3
| @@ -18,7 +18,7 @@ func TestRepositoryHeadFailureKeepsOldCompleteGeneration(t *testing.T) { | |||
| 18 | 18 | if err := repo.Publish(context.Background(), first); err != nil { | |
| 19 | 19 | t.Fatal(err) | |
| 20 | 20 | } | |
| 21 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | |
| 21 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) | |
| 22 | 22 | repo.hooks.beforeHeadWrite = func(string) error { return errors.New("injected HEAD failure") } | |
| 23 | 23 | if err := repo.Publish(context.Background(), second); err == nil { | |
| 24 | 24 | t.Fatal("Publish succeeded with injected HEAD failure") | |
| @@ -55,7 +55,7 @@ func TestRepositoryCleanupFaultsSurfaceForImmutableAndMutablePublication(t *test | |||
| 55 | 55 | if err := repo.Publish(context.Background(), first); err != nil { | |
| 56 | 56 | t.Fatal(err) | |
| 57 | 57 | } | |
| 58 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | |
| 58 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) | |
| 59 | 59 | key := legacyIncarnationID(second.Name).String() | |
| 60 | 60 | if tc.location == "manifest" { | |
| 61 | 61 | prepareObjects(t, repo, second) | |
| @@ -187,7 +187,7 @@ func TestRepositoryLoadFallsBackFromIncompleteNewestGeneration(t *testing.T) { | |||
| 187 | 187 | if err := repo.Publish(context.Background(), first); err != nil { | |
| 188 | 188 | t.Fatal(err) | |
| 189 | 189 | } | |
| 190 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | |
| 190 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) | |
| 191 | 191 | if err := repo.Publish(context.Background(), second); err != nil { | |
| 192 | 192 | t.Fatal(err) | |
| 193 | 193 | } | |
| @@ -18,7 +18,7 @@ func TestRepositoryHeadFailureKeepsOldCompleteGeneration(t *testing.T) { | |||||
| 18 | if err := repo.Publish(context.Background(), first); err != nil { | 18 | if err := repo.Publish(context.Background(), first); err != nil { | ||
| 19 | t.Fatal(err) | 19 | t.Fatal(err) | ||
| 20 | } | 20 | } | ||
| 21 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | 21 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) |
| 22 | repo.hooks.beforeHeadWrite = func(string) error { return errors.New("injected HEAD failure") } | 22 | repo.hooks.beforeHeadWrite = func(string) error { return errors.New("injected HEAD failure") } | ||
| 23 | if err := repo.Publish(context.Background(), second); err == nil { | 23 | if err := repo.Publish(context.Background(), second); err == nil { | ||
| 24 | t.Fatal("Publish succeeded with injected HEAD failure") | 24 | t.Fatal("Publish succeeded with injected HEAD failure") | ||
| @@ -55,7 +55,7 @@ func TestRepositoryCleanupFaultsSurfaceForImmutableAndMutablePublication(t *test | |||||
| 55 | if err := repo.Publish(context.Background(), first); err != nil { | 55 | if err := repo.Publish(context.Background(), first); err != nil { | ||
| 56 | t.Fatal(err) | 56 | t.Fatal(err) | ||
| 57 | } | 57 | } | ||
| 58 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | 58 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) |
| 59 | key := legacyIncarnationID(second.Name).String() | 59 | key := legacyIncarnationID(second.Name).String() | ||
| 60 | if tc.location == "manifest" { | 60 | if tc.location == "manifest" { | ||
| 61 | prepareObjects(t, repo, second) | 61 | prepareObjects(t, repo, second) | ||
| @@ -187,7 +187,7 @@ func TestRepositoryLoadFallsBackFromIncompleteNewestGeneration(t *testing.T) { | |||||
| 187 | if err := repo.Publish(context.Background(), first); err != nil { | 187 | if err := repo.Publish(context.Background(), first); err != nil { | ||
| 188 | t.Fatal(err) | 188 | t.Fatal(err) | ||
| 189 | } | 189 | } | ||
| 190 | - | second := repositoryPublication(t, "named", 2, []byte("two")) | 190 | + | second := repositoryPublicationAfter(t, repo, "named", 2, []byte("two")) |
| 191 | if err := repo.Publish(context.Background(), second); err != nil { | 191 | if err := repo.Publish(context.Background(), second); err != nil { | ||
| 192 | t.Fatal(err) | 192 | t.Fatal(err) | ||
| 193 | } | 193 | } | ||
M internal/adapters/snapshot/repository_test.go +38 -0
| @@ -2,8 +2,11 @@ package snapshot | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "crypto/sha256" | |
| 5 | 6 | "testing" | |
| 6 | 7 | ||
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | ||
| 7 | 10 | "github.com/bnema/vev/internal/ports" | |
| 8 | 11 | codec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 9 | 12 | ) | |
| @@ -41,6 +44,9 @@ func TestRepositoryDoesNotRewriteVerifiedImmutableBlob(t *testing.T) { | |||
| 41 | 44 | t.Fatal(err) | |
| 42 | 45 | } | |
| 43 | 46 | manifest.Generation = 2 | |
| 47 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: codecDigestForTest(pub.Manifest)} | |
| 48 | + | manifest.ParentCheckpoint = parent | |
| 49 | + | pub.ParentCheckpoint = parent | |
| 44 | 50 | pub.Manifest, err = codec.MarshalManifest(manifest) | |
| 45 | 51 | if err != nil { | |
| 46 | 52 | t.Fatal(err) | |
| @@ -53,6 +59,38 @@ func TestRepositoryDoesNotRewriteVerifiedImmutableBlob(t *testing.T) { | |||
| 53 | 59 | } | |
| 54 | 60 | } | |
| 55 | 61 | ||
| 62 | + | func codecDigestForTest(data []byte) [32]byte { | |
| 63 | + | return sha256.Sum256(data) | |
| 64 | + | } | |
| 65 | + | ||
| 66 | + | func repositoryPublicationAfter(t *testing.T, repo *Repository, name string, generation uint64, payload []byte) ports.SnapshotPublication { | |
| 67 | + | t.Helper() | |
| 68 | + | return publicationWithCurrentParent(t, repo, repositoryPublication(t, name, generation, payload)) | |
| 69 | + | } | |
| 70 | + | ||
| 71 | + | func publicationWithCurrentParent(t *testing.T, repo *Repository, publication ports.SnapshotPublication) ports.SnapshotPublication { | |
| 72 | + | t.Helper() | |
| 73 | + | if publication.Generation == 1 { | |
| 74 | + | return publication | |
| 75 | + | } | |
| 76 | + | currentGeneration, digest, err := repo.readHead(publication.IncarnationID) | |
| 77 | + | if err != nil { | |
| 78 | + | t.Fatalf("read parent HEAD: %v", err) | |
| 79 | + | } | |
| 80 | + | parent := &domain.CheckpointRef{Generation: currentGeneration, ManifestDigest: digest} | |
| 81 | + | manifest, err := codec.UnmarshalManifest(publication.Manifest) | |
| 82 | + | if err != nil { | |
| 83 | + | t.Fatal(err) | |
| 84 | + | } | |
| 85 | + | manifest.ParentCheckpoint = parent | |
| 86 | + | publication.Manifest, err = codec.MarshalManifest(manifest) | |
| 87 | + | if err != nil { | |
| 88 | + | t.Fatal(err) | |
| 89 | + | } | |
| 90 | + | publication.ParentCheckpoint = parent | |
| 91 | + | return publication | |
| 92 | + | } | |
| 93 | + | ||
| 56 | 94 | func repositoryPublication(t *testing.T, name string, generation uint64, payload []byte) ports.SnapshotPublication { | |
| 57 | 95 | t.Helper() | |
| 58 | 96 | id := legacyIncarnationID(name) | |
| @@ -2,8 +2,11 @@ package snapshot | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "crypto/sha256" | |||
| 5 | "testing" | 6 | "testing" | ||
| 6 | 7 | ||||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | ||||
| 7 | "github.com/bnema/vev/internal/ports" | 10 | "github.com/bnema/vev/internal/ports" | ||
| 8 | codec "github.com/bnema/vev/internal/usecase/snapshot" | 11 | codec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 9 | ) | 12 | ) | ||
| @@ -41,6 +44,9 @@ func TestRepositoryDoesNotRewriteVerifiedImmutableBlob(t *testing.T) { | |||||
| 41 | t.Fatal(err) | 44 | t.Fatal(err) | ||
| 42 | } | 45 | } | ||
| 43 | manifest.Generation = 2 | 46 | manifest.Generation = 2 | ||
| 47 | + | parent := &domain.CheckpointRef{Generation: 1, ManifestDigest: codecDigestForTest(pub.Manifest)} | |||
| 48 | + | manifest.ParentCheckpoint = parent | |||
| 49 | + | pub.ParentCheckpoint = parent | |||
| 44 | pub.Manifest, err = codec.MarshalManifest(manifest) | 50 | pub.Manifest, err = codec.MarshalManifest(manifest) | ||
| 45 | if err != nil { | 51 | if err != nil { | ||
| 46 | t.Fatal(err) | 52 | t.Fatal(err) | ||
| @@ -53,6 +59,38 @@ func TestRepositoryDoesNotRewriteVerifiedImmutableBlob(t *testing.T) { | |||||
| 53 | } | 59 | } | ||
| 54 | } | 60 | } | ||
| 55 | 61 | ||||
| 62 | + | func codecDigestForTest(data []byte) [32]byte { | |||
| 63 | + | return sha256.Sum256(data) | |||
| 64 | + | } | |||
| 65 | + | ||||
| 66 | + | func repositoryPublicationAfter(t *testing.T, repo *Repository, name string, generation uint64, payload []byte) ports.SnapshotPublication { | |||
| 67 | + | t.Helper() | |||
| 68 | + | return publicationWithCurrentParent(t, repo, repositoryPublication(t, name, generation, payload)) | |||
| 69 | + | } | |||
| 70 | + | ||||
| 71 | + | func publicationWithCurrentParent(t *testing.T, repo *Repository, publication ports.SnapshotPublication) ports.SnapshotPublication { | |||
| 72 | + | t.Helper() | |||
| 73 | + | if publication.Generation == 1 { | |||
| 74 | + | return publication | |||
| 75 | + | } | |||
| 76 | + | currentGeneration, digest, err := repo.readHead(publication.IncarnationID) | |||
| 77 | + | if err != nil { | |||
| 78 | + | t.Fatalf("read parent HEAD: %v", err) | |||
| 79 | + | } | |||
| 80 | + | parent := &domain.CheckpointRef{Generation: currentGeneration, ManifestDigest: digest} | |||
| 81 | + | manifest, err := codec.UnmarshalManifest(publication.Manifest) | |||
| 82 | + | if err != nil { | |||
| 83 | + | t.Fatal(err) | |||
| 84 | + | } | |||
| 85 | + | manifest.ParentCheckpoint = parent | |||
| 86 | + | publication.Manifest, err = codec.MarshalManifest(manifest) | |||
| 87 | + | if err != nil { | |||
| 88 | + | t.Fatal(err) | |||
| 89 | + | } | |||
| 90 | + | publication.ParentCheckpoint = parent | |||
| 91 | + | return publication | |||
| 92 | + | } | |||
| 93 | + | ||||
| 56 | func repositoryPublication(t *testing.T, name string, generation uint64, payload []byte) ports.SnapshotPublication { | 94 | func repositoryPublication(t *testing.T, name string, generation uint64, payload []byte) ports.SnapshotPublication { | ||
| 57 | t.Helper() | 95 | t.Helper() | ||
| 58 | id := legacyIncarnationID(name) | 96 | id := legacyIncarnationID(name) | ||
M internal/usecase/daemon/session.go +1 -0
| @@ -77,6 +77,7 @@ type session struct { | |||
| 77 | 77 | snapshotMu sync.Mutex | |
| 78 | 78 | snapshotGeneration uint64 // newest mutation revision | |
| 79 | 79 | snapshotPublishedGeneration uint64 // newest repository generation | |
| 80 | + | snapshotPublishedCheckpoint *domain.CheckpointRef | |
| 80 | 81 | snapshotPublishedMutationRevision uint64 | |
| 81 | 82 | snapshotNextEligibleAt time.Time | |
| 82 | 83 | // The coordinator state below is guarded by snapshotMu. A capture can be | |
| @@ -77,6 +77,7 @@ type session struct { | |||||
| 77 | snapshotMu sync.Mutex | 77 | snapshotMu sync.Mutex | ||
| 78 | snapshotGeneration uint64 // newest mutation revision | 78 | snapshotGeneration uint64 // newest mutation revision | ||
| 79 | snapshotPublishedGeneration uint64 // newest repository generation | 79 | snapshotPublishedGeneration uint64 // newest repository generation | ||
| 80 | + | snapshotPublishedCheckpoint *domain.CheckpointRef | |||
| 80 | snapshotPublishedMutationRevision uint64 | 81 | snapshotPublishedMutationRevision uint64 | ||
| 81 | snapshotNextEligibleAt time.Time | 82 | snapshotNextEligibleAt time.Time | ||
| 82 | // The coordinator state below is guarded by snapshotMu. A capture can be | 83 | // The coordinator state below is guarded by snapshotMu. A capture can be | ||
M internal/usecase/daemon/snapshot.go +2 -0
| @@ -36,6 +36,8 @@ type snapshotCapture struct { | |||
| 36 | 36 | session *session | |
| 37 | 37 | attemptKind snapshotAttemptKind | |
| 38 | 38 | generation uint64 // repository publication generation | |
| 39 | + | parentCheckpoint *domain.CheckpointRef | |
| 40 | + | checkpoint domain.CheckpointRef // set by the encoder before publication | |
| 39 | 41 | mutationRevision uint64 | |
| 40 | 42 | name string | |
| 41 | 43 | incarnation domain.IncarnationID | |
| @@ -36,6 +36,8 @@ type snapshotCapture struct { | |||||
| 36 | session *session | 36 | session *session | ||
| 37 | attemptKind snapshotAttemptKind | 37 | attemptKind snapshotAttemptKind | ||
| 38 | generation uint64 // repository publication generation | 38 | generation uint64 // repository publication generation | ||
| 39 | + | parentCheckpoint *domain.CheckpointRef | |||
| 40 | + | checkpoint domain.CheckpointRef // set by the encoder before publication | |||
| 39 | mutationRevision uint64 | 41 | mutationRevision uint64 | ||
| 40 | name string | 42 | name string | ||
| 41 | incarnation domain.IncarnationID | 43 | incarnation domain.IncarnationID | ||
M internal/usecase/daemon/snapshot_acceptance_test.go +4 -3
| @@ -63,9 +63,10 @@ func snapshotCacheHistory(t *testing.T, chunks int, first rune) vt.HistorySnapsh | |||
| 63 | 63 | ||
| 64 | 64 | func snapshotCacheCapture(sess *session, history vt.HistorySnapshotView, generation uint64) *snapshotCapture { | |
| 65 | 65 | return &snapshotCapture{ | |
| 66 | - | session: sess, | |
| 67 | - | name: sess.name, | |
| 68 | - | generation: generation, | |
| 66 | + | session: sess, | |
| 67 | + | name: sess.name, | |
| 68 | + | incarnation: sess.incarnation, | |
| 69 | + | generation: generation, | |
| 69 | 70 | tabs: []snapshotCaptureTab{{ | |
| 70 | 71 | stableID: "tab", | |
| 71 | 72 | cols: 1, | |
| @@ -63,9 +63,10 @@ func snapshotCacheHistory(t *testing.T, chunks int, first rune) vt.HistorySnapsh | |||||
| 63 | 63 | ||||
| 64 | func snapshotCacheCapture(sess *session, history vt.HistorySnapshotView, generation uint64) *snapshotCapture { | 64 | func snapshotCacheCapture(sess *session, history vt.HistorySnapshotView, generation uint64) *snapshotCapture { | ||
| 65 | return &snapshotCapture{ | 65 | return &snapshotCapture{ | ||
| 66 | - | session: sess, | 66 | + | session: sess, |
| 67 | - | name: sess.name, | 67 | + | name: sess.name, |
| 68 | - | generation: generation, | 68 | + | incarnation: sess.incarnation, |
| 69 | + | generation: generation, | |||
| 69 | tabs: []snapshotCaptureTab{{ | 70 | tabs: []snapshotCaptureTab{{ | ||
| 70 | stableID: "tab", | 71 | stableID: "tab", | ||
| 71 | cols: 1, | 72 | cols: 1, | ||
M internal/usecase/daemon/snapshot_coordinator.go +6 -0
| @@ -90,6 +90,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||
| 90 | 90 | // dirty and wake the scheduler. | |
| 91 | 91 | sess.snapshotGeneration = 1 | |
| 92 | 92 | sess.snapshotPublishedGeneration = 0 | |
| 93 | + | sess.snapshotPublishedCheckpoint = nil | |
| 93 | 94 | sess.snapshotPublishedMutationRevision = 0 | |
| 94 | 95 | sess.snapshotForcedGeneration = 0 | |
| 95 | 96 | sess.snapshotNextEligibleAt = time.Time{} | |
| @@ -238,6 +239,11 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||
| 238 | 239 | mutationRevision = capture.generation | |
| 239 | 240 | } | |
| 240 | 241 | if succeeded { | |
| 242 | + | if capture.generation >= capture.session.snapshotPublishedGeneration && | |
| 243 | + | capture.checkpoint.Generation == capture.generation && capture.checkpoint.ManifestDigest != ([32]byte{}) { | |
| 244 | + | checkpoint := capture.checkpoint | |
| 245 | + | capture.session.snapshotPublishedCheckpoint = &checkpoint | |
| 246 | + | } | |
| 241 | 247 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | |
| 242 | 248 | capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision) | |
| 243 | 249 | } | |
| @@ -90,6 +90,7 @@ func resumeSnapshotCoordinatorForNewIdentity(sess *session, quarantine snapshotC | |||||
| 90 | // dirty and wake the scheduler. | 90 | // dirty and wake the scheduler. | ||
| 91 | sess.snapshotGeneration = 1 | 91 | sess.snapshotGeneration = 1 | ||
| 92 | sess.snapshotPublishedGeneration = 0 | 92 | sess.snapshotPublishedGeneration = 0 | ||
| 93 | + | sess.snapshotPublishedCheckpoint = nil | |||
| 93 | sess.snapshotPublishedMutationRevision = 0 | 94 | sess.snapshotPublishedMutationRevision = 0 | ||
| 94 | sess.snapshotForcedGeneration = 0 | 95 | sess.snapshotForcedGeneration = 0 | ||
| 95 | sess.snapshotNextEligibleAt = time.Time{} | 96 | sess.snapshotNextEligibleAt = time.Time{} | ||
| @@ -238,6 +239,11 @@ func (d *Daemon) finishSnapshotCapture(capture *snapshotCapture, succeeded bool) | |||||
| 238 | mutationRevision = capture.generation | 239 | mutationRevision = capture.generation | ||
| 239 | } | 240 | } | ||
| 240 | if succeeded { | 241 | if succeeded { | ||
| 242 | + | if capture.generation >= capture.session.snapshotPublishedGeneration && | |||
| 243 | + | capture.checkpoint.Generation == capture.generation && capture.checkpoint.ManifestDigest != ([32]byte{}) { | |||
| 244 | + | checkpoint := capture.checkpoint | |||
| 245 | + | capture.session.snapshotPublishedCheckpoint = &checkpoint | |||
| 246 | + | } | |||
| 241 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | 247 | capture.session.snapshotPublishedGeneration = max(capture.session.snapshotPublishedGeneration, capture.generation) | ||
| 242 | capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision) | 248 | capture.session.snapshotPublishedMutationRevision = max(capture.session.snapshotPublishedMutationRevision, mutationRevision) | ||
| 243 | } | 249 | } | ||
M internal/usecase/daemon/snapshot_incremental.go +5 -2
| @@ -1,8 +1,10 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "crypto/sha256" | |
| 4 | 5 | "fmt" | |
| 5 | 6 | ||
| 7 | + | "github.com/bnema/vev/internal/domain" | |
| 6 | 8 | "github.com/bnema/vev/internal/ports" | |
| 7 | 9 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 8 | 10 | "github.com/bnema/vev/pkg/vt" | |
| @@ -113,7 +115,7 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||
| 113 | 115 | if err := prepareSnapshotChunkCache(capture); err != nil { | |
| 114 | 116 | return ports.SnapshotPublication{}, err | |
| 115 | 117 | } | |
| 116 | - | manifest := snapcodec.Manifest{Generation: capture.generation, IncarnationID: capture.incarnation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | |
| 118 | + | manifest := snapcodec.Manifest{Generation: capture.generation, IncarnationID: capture.incarnation, ParentCheckpoint: capture.parentCheckpoint, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | |
| 117 | 119 | objects := make([]ports.SnapshotObject, 0) | |
| 118 | 120 | capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef) | |
| 119 | 121 | for _, tab := range capture.tabs { | |
| @@ -160,7 +162,8 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||
| 160 | 162 | if err != nil { | |
| 161 | 163 | return ports.SnapshotPublication{}, err | |
| 162 | 164 | } | |
| 163 | - | return ports.SnapshotPublication{IncarnationID: capture.incarnation, Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | |
| 165 | + | capture.checkpoint = domain.CheckpointRef{Generation: capture.generation, ManifestDigest: sha256.Sum256(encoded)} | |
| 166 | + | return ports.SnapshotPublication{IncarnationID: capture.incarnation, Name: capture.name, Generation: capture.generation, ParentCheckpoint: capture.parentCheckpoint, Manifest: encoded, Objects: objects}, nil | |
| 164 | 167 | } | |
| 165 | 168 | ||
| 166 | 169 | // marshalSnapshotTail selects the canonical empty-tail encoding before any | |
| @@ -1,8 +1,10 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "crypto/sha256" | |||
| 4 | "fmt" | 5 | "fmt" | ||
| 5 | 6 | ||||
| 7 | + | "github.com/bnema/vev/internal/domain" | |||
| 6 | "github.com/bnema/vev/internal/ports" | 8 | "github.com/bnema/vev/internal/ports" | ||
| 7 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 9 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 8 | "github.com/bnema/vev/pkg/vt" | 10 | "github.com/bnema/vev/pkg/vt" | ||
| @@ -113,7 +115,7 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||||
| 113 | if err := prepareSnapshotChunkCache(capture); err != nil { | 115 | if err := prepareSnapshotChunkCache(capture); err != nil { | ||
| 114 | return ports.SnapshotPublication{}, err | 116 | return ports.SnapshotPublication{}, err | ||
| 115 | } | 117 | } | ||
| 116 | - | manifest := snapcodec.Manifest{Generation: capture.generation, IncarnationID: capture.incarnation, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} | 118 | + | manifest := snapcodec.Manifest{Generation: capture.generation, IncarnationID: capture.incarnation, ParentCheckpoint: capture.parentCheckpoint, Name: capture.name, CreatedAt: capture.createdAt, Active: capture.active, Tabs: make([]snapcodec.ManifestTab, 0, len(capture.tabs))} |
| 117 | objects := make([]ports.SnapshotObject, 0) | 119 | objects := make([]ports.SnapshotObject, 0) | ||
| 118 | capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef) | 120 | capture.sealedRefs = make(map[*vt.HistoryChunk]snapcodec.ObjectRef) | ||
| 119 | for _, tab := range capture.tabs { | 121 | for _, tab := range capture.tabs { | ||
| @@ -160,7 +162,8 @@ func (d *Daemon) incrementalPublication(capture *snapshotCapture) (ports.Snapsho | |||||
| 160 | if err != nil { | 162 | if err != nil { | ||
| 161 | return ports.SnapshotPublication{}, err | 163 | return ports.SnapshotPublication{}, err | ||
| 162 | } | 164 | } | ||
| 163 | - | return ports.SnapshotPublication{IncarnationID: capture.incarnation, Name: capture.name, Generation: capture.generation, Manifest: encoded, Objects: objects}, nil | 165 | + | capture.checkpoint = domain.CheckpointRef{Generation: capture.generation, ManifestDigest: sha256.Sum256(encoded)} |
| 166 | + | return ports.SnapshotPublication{IncarnationID: capture.incarnation, Name: capture.name, Generation: capture.generation, ParentCheckpoint: capture.parentCheckpoint, Manifest: encoded, Objects: objects}, nil | |||
| 164 | } | 167 | } | ||
| 165 | 168 | ||||
| 166 | // marshalSnapshotTail selects the canonical empty-tail encoding before any | 169 | // marshalSnapshotTail selects the canonical empty-tail encoding before any | ||
M internal/usecase/daemon/snapshot_incremental_test.go +9 -1
| @@ -1,11 +1,13 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "crypto/sha256" | |
| 4 | 5 | "testing" | |
| 5 | 6 | "time" | |
| 6 | 7 | ||
| 7 | 8 | "github.com/stretchr/testify/require" | |
| 8 | 9 | ||
| 10 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | 11 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | |
| 10 | 12 | "github.com/bnema/vev/pkg/renderer" | |
| 11 | 13 | "github.com/bnema/vev/pkg/vt" | |
| @@ -124,12 +126,14 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||
| 124 | 126 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | |
| 125 | 127 | d := New(nil, nil, nil) | |
| 126 | 128 | sess := newSnapshotTestSession(t, "work", false, "/work") | |
| 127 | - | capture := &snapshotCapture{session: sess, name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", sealed: view, tail: view.Tail(), visible: visible}}}}} | |
| 129 | + | capture := &snapshotCapture{session: sess, name: "work", incarnation: domain.IncarnationID{1}, generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", sealed: view, tail: view.Tail(), visible: visible}}}}} | |
| 128 | 130 | first, err := d.incrementalPublication(capture) | |
| 129 | 131 | if err != nil { | |
| 130 | 132 | t.Fatal(err) | |
| 131 | 133 | } | |
| 134 | + | firstRef := domain.CheckpointRef{Generation: first.Generation, ManifestDigest: sha256.Sum256(first.Manifest)} | |
| 132 | 135 | capture.generation = 2 | |
| 136 | + | capture.parentCheckpoint = &firstRef | |
| 133 | 137 | second, err := d.incrementalPublication(capture) | |
| 134 | 138 | if err != nil { | |
| 135 | 139 | t.Fatal(err) | |
| @@ -137,6 +141,10 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||
| 137 | 141 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | |
| 138 | 142 | t.Fatal("publication omitted required objects") | |
| 139 | 143 | } | |
| 144 | + | require.Equal(t, &firstRef, second.ParentCheckpoint) | |
| 145 | + | manifest, err := snapcodec.UnmarshalManifest(second.Manifest) | |
| 146 | + | require.NoError(t, err) | |
| 147 | + | require.Equal(t, &firstRef, manifest.ParentCheckpoint) | |
| 140 | 148 | if got, want := sess.snapshotChunkCache.used, sess.snapshotChunkCache.limit; got > want { | |
| 141 | 149 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | |
| 142 | 150 | } | |
| @@ -1,11 +1,13 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "crypto/sha256" | |||
| 4 | "testing" | 5 | "testing" | ||
| 5 | "time" | 6 | "time" | ||
| 6 | 7 | ||||
| 7 | "github.com/stretchr/testify/require" | 8 | "github.com/stretchr/testify/require" | ||
| 8 | 9 | ||||
| 10 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | 11 | snapcodec "github.com/bnema/vev/internal/usecase/snapshot" | ||
| 10 | "github.com/bnema/vev/pkg/renderer" | 12 | "github.com/bnema/vev/pkg/renderer" | ||
| 11 | "github.com/bnema/vev/pkg/vt" | 13 | "github.com/bnema/vev/pkg/vt" | ||
| @@ -124,12 +126,14 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||||
| 124 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | 126 | visible := vt.NewScreen(1, 1).PrimaryVisibleSnapshot() | ||
| 125 | d := New(nil, nil, nil) | 127 | d := New(nil, nil, nil) | ||
| 126 | sess := newSnapshotTestSession(t, "work", false, "/work") | 128 | sess := newSnapshotTestSession(t, "work", false, "/work") | ||
| 127 | - | capture := &snapshotCapture{session: sess, name: "work", generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", sealed: view, tail: view.Tail(), visible: visible}}}}} | 129 | + | capture := &snapshotCapture{session: sess, name: "work", incarnation: domain.IncarnationID{1}, generation: 1, tabs: []snapshotCaptureTab{{stableID: "t", cols: 1, rows: 1, panes: []snapshotCapturePane{{id: "p", stableID: "p", sealed: view, tail: view.Tail(), visible: visible}}}}} |
| 128 | first, err := d.incrementalPublication(capture) | 130 | first, err := d.incrementalPublication(capture) | ||
| 129 | if err != nil { | 131 | if err != nil { | ||
| 130 | t.Fatal(err) | 132 | t.Fatal(err) | ||
| 131 | } | 133 | } | ||
| 134 | + | firstRef := domain.CheckpointRef{Generation: first.Generation, ManifestDigest: sha256.Sum256(first.Manifest)} | |||
| 132 | capture.generation = 2 | 135 | capture.generation = 2 | ||
| 136 | + | capture.parentCheckpoint = &firstRef | |||
| 133 | second, err := d.incrementalPublication(capture) | 137 | second, err := d.incrementalPublication(capture) | ||
| 134 | if err != nil { | 138 | if err != nil { | ||
| 135 | t.Fatal(err) | 139 | t.Fatal(err) | ||
| @@ -137,6 +141,10 @@ func TestIncrementalPublicationReusesSealedChunkObject(t *testing.T) { | |||||
| 137 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | 141 | if len(first.Objects) == 0 || len(second.Objects) == 0 { | ||
| 138 | t.Fatal("publication omitted required objects") | 142 | t.Fatal("publication omitted required objects") | ||
| 139 | } | 143 | } | ||
| 144 | + | require.Equal(t, &firstRef, second.ParentCheckpoint) | |||
| 145 | + | manifest, err := snapcodec.UnmarshalManifest(second.Manifest) | |||
| 146 | + | require.NoError(t, err) | |||
| 147 | + | require.Equal(t, &firstRef, manifest.ParentCheckpoint) | |||
| 140 | if got, want := sess.snapshotChunkCache.used, sess.snapshotChunkCache.limit; got > want { | 148 | if got, want := sess.snapshotChunkCache.used, sess.snapshotChunkCache.limit; got > want { | ||
| 141 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | 149 | t.Fatalf("cache bytes = %d, limit = %d", got, want) | ||
| 142 | } | 150 | } | ||
M internal/usecase/daemon/snapshot_queue.go +11 -1
| @@ -1,6 +1,10 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | - | import "context" | |
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | ||
| 6 | + | "github.com/bnema/vev/internal/domain" | |
| 7 | + | ) | |
| 4 | 8 | ||
| 5 | 9 | func (d *Daemon) scheduleSnapshot(sess *session) bool { | |
| 6 | 10 | return d.scheduleSnapshotWithFinalFallback(sess, false) | |
| @@ -54,6 +58,11 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||
| 54 | 58 | sess.snapshotGeneration = mutationRevision | |
| 55 | 59 | } | |
| 56 | 60 | publicationContext := sess.snapshotPublicationContext | |
| 61 | + | var parentCheckpoint *domain.CheckpointRef | |
| 62 | + | if sess.snapshotPublishedCheckpoint != nil { | |
| 63 | + | parent := *sess.snapshotPublishedCheckpoint | |
| 64 | + | parentCheckpoint = &parent | |
| 65 | + | } | |
| 57 | 66 | sess.snapshotPendingCaptures++ | |
| 58 | 67 | sess.snapshotPending = true | |
| 59 | 68 | sess.signalSnapshotChangedLocked() | |
| @@ -64,6 +73,7 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||
| 64 | 73 | capture.mutationRevision = mutationRevision | |
| 65 | 74 | capture.attemptKind = kind | |
| 66 | 75 | capture.publicationContext = publicationContext | |
| 76 | + | capture.parentCheckpoint = parentCheckpoint | |
| 67 | 77 | } | |
| 68 | 78 | if !ok { | |
| 69 | 79 | d.finishSnapshotCapture(capture, false) | |
| @@ -1,6 +1,10 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | - | import "context" | 3 | + | import ( |
| 4 | + | "context" | |||
| 5 | + | ||||
| 6 | + | "github.com/bnema/vev/internal/domain" | |||
| 7 | + | ) | |||
| 4 | 8 | ||||
| 5 | func (d *Daemon) scheduleSnapshot(sess *session) bool { | 9 | func (d *Daemon) scheduleSnapshot(sess *session) bool { | ||
| 6 | return d.scheduleSnapshotWithFinalFallback(sess, false) | 10 | return d.scheduleSnapshotWithFinalFallback(sess, false) | ||
| @@ -54,6 +58,11 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||||
| 54 | sess.snapshotGeneration = mutationRevision | 58 | sess.snapshotGeneration = mutationRevision | ||
| 55 | } | 59 | } | ||
| 56 | publicationContext := sess.snapshotPublicationContext | 60 | publicationContext := sess.snapshotPublicationContext | ||
| 61 | + | var parentCheckpoint *domain.CheckpointRef | |||
| 62 | + | if sess.snapshotPublishedCheckpoint != nil { | |||
| 63 | + | parent := *sess.snapshotPublishedCheckpoint | |||
| 64 | + | parentCheckpoint = &parent | |||
| 65 | + | } | |||
| 57 | sess.snapshotPendingCaptures++ | 66 | sess.snapshotPendingCaptures++ | ||
| 58 | sess.snapshotPending = true | 67 | sess.snapshotPending = true | ||
| 59 | sess.signalSnapshotChangedLocked() | 68 | sess.signalSnapshotChangedLocked() | ||
| @@ -64,6 +73,7 @@ func (d *Daemon) scheduleSnapshotWithFinalFallback(sess *session, final bool) bo | |||||
| 64 | capture.mutationRevision = mutationRevision | 73 | capture.mutationRevision = mutationRevision | ||
| 65 | capture.attemptKind = kind | 74 | capture.attemptKind = kind | ||
| 66 | capture.publicationContext = publicationContext | 75 | capture.publicationContext = publicationContext | ||
| 76 | + | capture.parentCheckpoint = parentCheckpoint | |||
| 67 | } | 77 | } | ||
| 68 | if !ok { | 78 | if !ok { | ||
| 69 | d.finishSnapshotCapture(capture, false) | 79 | d.finishSnapshotCapture(capture, false) | ||