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 _, manifest, err := r.loadCheckpointLocked(ctx, id, "", ref)
82 if err != nil {
83 return fmt.Errorf("validate checkpoint before HEAD repair: %w", err)
84 }
85 if manifest.Name == "" {
86 return fmt.Errorf("validate checkpoint before HEAD repair: invalid manifest name")
87 }
88 return r.writeMutable(r.headPath(id), marshalHead(ref.Generation, ref.ManifestDigest))
89}
90
91func marshalHead(generation uint64, digest ports.SnapshotDigest) []byte {
92 out := make([]byte, 4+8+sha256.Size)
93 copy(out, "VEVH")
94 binary.BigEndian.PutUint64(out[4:12], generation)
95 copy(out[12:], digest[:])
96 return out
97}
98func (r *Repository) readHead(id domain.IncarnationID) (uint64, ports.SnapshotDigest, error) {
99 return r.readHeadWithRoot(nil, id)
100}
101
102func (r *Repository) readHeadWithRoot(root *os.Root, id domain.IncarnationID) (uint64, ports.SnapshotDigest, error) {
103 return r.readHeadAt(root, r.headPath(id))
104}
105
106func (r *Repository) readHeadAt(root *os.Root, path string) (uint64, ports.SnapshotDigest, error) {
107 var data []byte
108 var err error
109 if root == nil {
110 data, err = r.readBounded(path)
111 } else {
112 data, err = r.readBoundedRoot(root, path)
113 }
114 if err != nil {
115 return 0, ports.SnapshotDigest{}, err
116 }
117 if len(data) != 4+8+sha256.Size || string(data[:4]) != "VEVH" {
118 return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
119 }
120 generation := binary.BigEndian.Uint64(data[4:12])
121 var digest ports.SnapshotDigest
122 copy(digest[:], data[12:])
123 if generation == 0 {
124 return 0, ports.SnapshotDigest{}, ErrInvalidHEAD
125 }
126 return generation, digest, nil
127}