1package snapshot
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/binary"
7 "errors"
8 "fmt"
9 "os"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 codec "github.com/bnema/vev/internal/usecase/snapshot"
14)
15
16var ErrInvalidHEAD = errors.New("invalid snapshot HEAD")
17
18func (r *Repository) LoadCheckpoint(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, error) {
19 if err := ctx.Err(); err != nil {
20 return ports.SnapshotGeneration{}, err
21 }
22 key, err := incarnationKey(id)
23 if err != nil {
24 return ports.SnapshotGeneration{}, err
25 }
26 lock := r.lockSession(key)
27 defer r.unlockSession(lock)
28 generation, _, err := r.loadCheckpointLocked(ctx, id, name, ref)
29 return generation, err
30}
31
32func (r *Repository) loadCheckpointLocked(ctx context.Context, id domain.IncarnationID, name string, ref ports.CheckpointRef) (ports.SnapshotGeneration, codec.Manifest, error) {
33 if ref.Generation == 0 || ref.ManifestDigest == ([32]byte{}) {
34 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("invalid checkpoint reference")
35 }
36 data, err := r.readBounded(r.manifestPath(id, ref.Generation))
37 if err != nil {
38 return ports.SnapshotGeneration{}, codec.Manifest{}, err
39 }
40 if sha256.Sum256(data) != ref.ManifestDigest {
41 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("manifest digest mismatch")
42 }
43 manifest, err := codec.UnmarshalManifest(data)
44 if err != nil {
45 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("invalid manifest: %w", err)
46 }
47 if manifest.IncarnationID != id || manifest.Generation != ref.Generation {
48 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("invalid manifest")
49 }
50 refs := manifestRefs(manifest)
51 if refs == nil || !withinGenerationBudget(len(data), refs) {
52 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("snapshot generation too large")
53 }
54 objects := make(map[ports.SnapshotDigest][]byte, len(refs))
55 for digest, objectRef := range refs {
56 if err := ctx.Err(); err != nil {
57 return ports.SnapshotGeneration{}, codec.Manifest{}, err
58 }
59 object, err := r.readBounded(r.objectPath(id, digest))
60 if err != nil {
61 return ports.SnapshotGeneration{}, codec.Manifest{}, err
62 }
63 if sha256.Sum256(object) != digest || !validObject(object, objectRef) {
64 return ports.SnapshotGeneration{}, codec.Manifest{}, fmt.Errorf("invalid object")
65 }
66 objects[digest] = object
67 }
68 return ports.SnapshotGeneration{IncarnationID: id, Name: name, Generation: ref.Generation, ParentCheckpoint: manifest.ParentCheckpoint, Manifest: data, Objects: objects}, manifest, nil
69}
70
71func (r *Repository) ReconcileCheckpoint(ctx context.Context, id domain.IncarnationID, ref ports.CheckpointRef) error {
72 if err := ctx.Err(); err != nil {
73 return err
74 }
75 key, err := incarnationKey(id)
76 if err != nil {
77 return err
78 }
79 lock := r.lockSession(key)
80 defer r.unlockSession(lock)
81 // HEAD normally already names the committed checkpoint. Skip the full
82 // reload and the durable rewrite then; only a divergent or unreadable HEAD
83 // needs repair, and repair still validates the checkpoint first.
84 if generation, digest, err := r.readHead(id); err == nil && generation == ref.Generation && digest == ref.ManifestDigest {
85 return nil
86 }
87 _, manifest, err := r.loadCheckpointLocked(ctx, id, "", ref)
88 if err != nil {
89 return fmt.Errorf("validate checkpoint before HEAD repair: %w", err)
90 }
91 if manifest.Name == "" {
92 return fmt.Errorf("validate checkpoint before HEAD repair: invalid manifest name")
93 }
94 return r.writeMutable(r.headPath(id), marshalHead(ref.Generation, ref.ManifestDigest))
95}
96
97func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {
98 out := make([]byte, 4+8+sha256.Size)
99 copy(out, "VEVH")
100 binary.BigEndian.PutUint64(out[4:12], generation)
101 copy(out[12:], digest[:])
102 return out
103}
104func (r *Repository) readHead(id domain.IncarnationID) (uint64, ports.SnapshotDigest, error) {
105 return r.readHeadWithRoot(nil, id)
106}
107
108func (r *Repository) readHeadWithRoot(root *os.Root, id domain.IncarnationID) (uint64, ports.SnapshotDigest, error) {
109 return r.readHeadAt(root, r.headPath(id))
110}
111
112func (r *Repository) readHeadAt(root *os.Root, path string) (uint64, ports.SnapshotDigest, error) {
113 var data []byte
114 var err error
115 if root == nil {
116 data, err = r.readBounded(path)
117 } else {
118 data, err = r.readBoundedRoot(root, path)
119 }
120 if err != nil {
121 return 0, ports.SnapshotDigest{}, err
122 }
123 if len(data) != 4+8+sha256.Size || string(data[:4]) != "VEVH" {
124 return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
125 }
126 generation := binary.BigEndian.Uint64(data[4:12])
127 var digest ports.SnapshotDigest
128 copy(digest[:], data[12:])
129 if generation == 0 {
130 return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
131 }
132 return generation, digest, nil
133}