1package snapshot
2
3import (
4 "bytes"
5 "context"
6 "crypto/sha256"
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
16func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPublication) error {
17 if err := ctx.Err(); err != nil {
18 return err
19 }
20 key, err := incarnationKey(publication.IncarnationID)
21 if err != nil {
22 return err
23 }
24 lock := r.lockSession(key)
25 defer r.unlockSession(lock)
26 if err := ctx.Err(); err != nil {
27 return err
28 }
29
30 refs, err := validatePublicationManifest(publication)
31 if err != nil {
32 return err
33 }
34 if err := ctx.Err(); err != nil {
35 return err
36 }
37 // A replay may find a forward orphan left after publication but before its
38 // catalogue commit. Accept it only after the canonical checkpoint loader has
39 // revalidated the pointer digest, manifest, and every referenced object.
40 if unchanged, err := r.unchangedPublication(ctx, publication); err != nil {
41 return err
42 } else if unchanged {
43 return nil
44 }
45
46 current, currentManifest, currentRefs, err := r.currentIncarnationPublication(ctx, publication)
47 if err != nil {
48 return err
49 }
50 if publication.Generation < current || publication.Generation > current+1 {
51 return fmt.Errorf("snapshot generation %d, current %d: immutable conflict", publication.Generation, current)
52 }
53 if publication.Generation == current {
54 if !bytes.Equal(currentManifest, publication.Manifest) {
55 return fmt.Errorf("snapshot generation %d: immutable conflict", publication.Generation)
56 }
57 return nil
58 }
59
60 // Parent validation and all other publication checks above happen before
61 // creating repository state. A rejected child must leave the authoritative
62 // checkpoint and its storage byte-for-byte unchanged.
63 if err := r.ensureSession(publication.IncarnationID); err != nil {
64 return err
65 }
66
67 // Retained refs from the authoritative generation were verified when their
68 // immutable files were installed. They are intentionally not re-read when a
69 // producer supplies them again. Every other referenced object is checked
70 // from disk once, or validated and installed from the supplied bytes.
71 supplied, err := suppliedNecessaryObjects(refs, currentRefs, publication.Objects)
72 if err != nil {
73 return err
74 }
75 for digest, ref := range refs {
76 if retainedObject(currentRefs, digest, ref) {
77 continue
78 }
79 if err := ctx.Err(); err != nil {
80 return err
81 }
82 path := r.objectPath(publication.IncarnationID, digest)
83 exists, err := r.verifyObjectFile(path, digest, ref)
84 if err != nil {
85 return fmt.Errorf("verify object %x: %w", digest, err)
86 }
87 if exists {
88 continue
89 }
90 objects := supplied[digest]
91 if len(objects) == 0 {
92 return fmt.Errorf("missing referenced object %x", digest)
93 }
94 object, err := r.validateSuppliedObject(objects, digest, ref)
95 if err != nil {
96 return err
97 }
98 if err := ctx.Err(); err != nil {
99 return err
100 }
101 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {
102 if r.objectDigest(existing) != digest || !validObject(existing, ref) {
103 return fmt.Errorf("existing immutable object is invalid")
104 }
105 return nil
106 }); err != nil {
107 return fmt.Errorf("write object: %w", err)
108 }
109 }
110
111 manifestPath := r.manifestPath(publication.IncarnationID, publication.Generation)
112 existing, exists, err := r.readOptionalBounded(manifestPath)
113 if err != nil {
114 return err
115 }
116 if exists {
117 if !bytes.Equal(existing, publication.Manifest) {
118 return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
119 }
120 } else {
121 if err := ctx.Err(); err != nil {
122 return err
123 }
124 if err := r.writeImmutable(manifestPath, publication.Manifest, func(existing []byte) error {
125 if !bytes.Equal(existing, publication.Manifest) {
126 return fmt.Errorf("manifest generation %d: immutable conflict", publication.Generation)
127 }
128 return nil
129 }); err != nil {
130 return fmt.Errorf("write manifest: %w", err)
131 }
132 }
133 if err := ctx.Err(); err != nil {
134 return err
135 }
136 if err := r.writeMutable(r.headPath(publication.IncarnationID), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest))); err != nil {
137 return fmt.Errorf("write HEAD: %w", err)
138 }
139 return nil
140}
141
142// unchangedPublication validates an existing forward orphan through the same
143// loader used for recovery. A corrupt replay must not be accepted merely
144// because its manifest bytes happen to match the requested publication.
145func (r *Repository) unchangedPublication(ctx context.Context, publication ports.SnapshotPublication) (bool, error) {
146 generation, digest, err := r.readHead(publication.IncarnationID)
147 _ = err // A missing or unreadable HEAD is not treated as a replay.
148 if err != nil || generation != publication.Generation {
149 return false, nil
150 }
151 current, _, err := r.loadCheckpointLocked(ctx, publication.IncarnationID, publication.Name, ports.CheckpointRef{
152 Generation: generation,
153 ManifestDigest: digest,
154 })
155 if err != nil {
156 return false, fmt.Errorf("validate replayed checkpoint: %w", err)
157 }
158 return bytes.Equal(current.Manifest, publication.Manifest), nil
159}
160
161func (r *Repository) currentIncarnationPublication(ctx context.Context, publication ports.SnapshotPublication) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {
162 generation, digest, err := r.readHead(publication.IncarnationID)
163 if errors.Is(err, os.ErrNotExist) {
164 if publication.Generation != 1 || publication.ParentCheckpoint != nil {
165 return 0, nil, nil, fmt.Errorf("first snapshot generation must be 1 with no parent")
166 }
167 return 0, nil, nil, nil
168 }
169 if err != nil {
170 return 0, nil, nil, fmt.Errorf("read snapshot HEAD: %w", err)
171 }
172 data, err := r.readBounded(r.manifestPath(publication.IncarnationID, generation))
173 if err != nil {
174 return 0, nil, nil, fmt.Errorf("validate current checkpoint: %w", err)
175 }
176 if sha256.Sum256(data) != digest {
177 return 0, nil, nil, fmt.Errorf("validate current checkpoint: manifest digest mismatch")
178 }
179 manifest, err := codec.UnmarshalManifest(data)
180 if err != nil {
181 return 0, nil, nil, fmt.Errorf("validate current manifest: %w", err)
182 }
183 if manifest.IncarnationID != publication.IncarnationID || manifest.Generation != generation {
184 return 0, nil, nil, fmt.Errorf("current manifest identity mismatch")
185 }
186 // Every child checkpoint is bound to the exact authoritative HEAD.
187 if publication.Generation == generation+1 {
188 wantParent := &domain.CheckpointRef{Generation: generation, ManifestDigest: digest}
189 if !checkpointRefEqual(publication.ParentCheckpoint, wantParent) {
190 return 0, nil, nil, fmt.Errorf("publication parent does not match current checkpoint")
191 }
192 }
193 refs := manifestRefs(manifest)
194 if refs == nil || !withinGenerationBudget(len(data), refs) {
195 return 0, nil, nil, fmt.Errorf("invalid current generation")
196 }
197 return generation, data, refs, ctx.Err()
198}
199
200func checkpointRefEqual(left, right *domain.CheckpointRef) bool {
201 if left == nil || right == nil {
202 return left == nil && right == nil
203 }
204 return *left == *right
205}
206
207func validatePublication(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, map[ports.SnapshotDigest]ports.SnapshotObject, error) {
208 refs, err := validatePublicationManifest(p)
209 if err != nil {
210 return nil, nil, err
211 }
212 supplied, err := suppliedNecessaryObjects(refs, nil, p.Objects)
213 if err != nil {
214 return nil, nil, err
215 }
216 objects := make(map[ports.SnapshotDigest]ports.SnapshotObject, len(supplied))
217 for digest, entries := range supplied {
218 object, err := validateSuppliedObject(entries, digest, refs[digest])
219 if err != nil {
220 return nil, nil, err
221 }
222 objects[digest] = object
223 }
224 return refs, objects, nil
225}
226
227func validatePublicationManifest(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
228 return validateManifest(p.Manifest, p.IncarnationID, p.Name, p.Generation, p.ParentCheckpoint)
229}
230
231func validateManifest(data []byte, incarnationID domain.IncarnationID, name string, generation uint64, parent *domain.CheckpointRef) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
232 manifest, err := codec.UnmarshalManifest(data)
233 if err != nil {
234 return nil, fmt.Errorf("invalid manifest: %w", err)
235 }
236 if manifest.IncarnationID != incarnationID || manifest.Name != name || manifest.Generation != generation || generation == 0 || !checkpointRefEqual(manifest.ParentCheckpoint, parent) {
237 return nil, fmt.Errorf("manifest identity, name, generation, or parent does not match publication")
238 }
239 refs := manifestRefs(manifest)
240 if refs == nil {
241 return nil, fmt.Errorf("conflicting manifest references")
242 }
243 if !withinGenerationBudget(len(data), refs) {
244 return nil, fmt.Errorf("snapshot generation too large")
245 }
246 return refs, nil
247}
248
249// suppliedNecessaryObjects first rejects unreferenced input without inspecting
250// its payload. Objects retained from currentRefs cannot affect publication and
251// are left out so callers do not pay to hash or copy a full history again.
252func suppliedNecessaryObjects(refs, currentRefs map[ports.SnapshotDigest]codec.ObjectRef, supplied []ports.SnapshotObject) (map[ports.SnapshotDigest][]ports.SnapshotObject, error) {
253 out := make(map[ports.SnapshotDigest][]ports.SnapshotObject)
254 for _, object := range supplied {
255 if _, used := refs[object.Digest]; !used {
256 return nil, fmt.Errorf("unreferenced object")
257 }
258 if retainedObject(currentRefs, object.Digest, refs[object.Digest]) {
259 continue
260 }
261 out[object.Digest] = append(out[object.Digest], object)
262 }
263 return out, nil
264}
265
266func retainedObject(currentRefs map[ports.SnapshotDigest]codec.ObjectRef, digest ports.SnapshotDigest, ref codec.ObjectRef) bool {
267 currentRef, ok := currentRefs[digest]
268 return ok && currentRef == ref
269}
270
271func (r *Repository) validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef) (ports.SnapshotObject, error) {
272 return validateSuppliedObject(objects, digest, ref)
273}
274
275func validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef) (ports.SnapshotObject, error) {
276 object := objects[0]
277 for _, candidate := range objects {
278 actual := sha256.Sum256(candidate.Data)
279 if actual != candidate.Digest || candidate.Digest != digest {
280 return ports.SnapshotObject{}, fmt.Errorf("object digest mismatch")
281 }
282 if !bytes.Equal(candidate.Data, object.Data) {
283 return ports.SnapshotObject{}, fmt.Errorf("conflicting object")
284 }
285 }
286 kind, payload, err := codec.PreflightObject(object.Data)
287 if err != nil || kind != ref.Kind || len(object.Data) != int(ref.Size) || len(payload) == 0 {
288 return ports.SnapshotObject{}, fmt.Errorf("invalid object envelope")
289 }
290 return ports.SnapshotObject{Digest: digest, Data: append([]byte(nil), object.Data...)}, nil
291}
292
293func withinGenerationBudget(manifestSize int, refs map[ports.SnapshotDigest]codec.ObjectRef) bool {
294 if manifestSize < 0 || manifestSize > maxRepositoryRead {
295 return false
296 }
297 total := uint64(manifestSize)
298 for _, ref := range refs {
299 if uint64(ref.Size) > uint64(maxRepositoryRead)-total {
300 return false
301 }
302 total += uint64(ref.Size)
303 }
304 return true
305}
306
307func manifestRefs(manifest codec.Manifest) map[ports.SnapshotDigest]codec.ObjectRef {
308 refs := make(map[ports.SnapshotDigest]codec.ObjectRef)
309 add := func(ref codec.ObjectRef) bool {
310 if old, ok := refs[ref.Digest]; ok && old != ref {
311 return false
312 }
313 refs[ref.Digest] = ref
314 return true
315 }
316 for _, tab := range manifest.Tabs {
317 for _, pane := range tab.Panes {
318 for _, ref := range pane.Sealed {
319 if !add(ref) {
320 return nil
321 }
322 }
323 if !add(pane.Tail) || !add(pane.Transcript) {
324 return nil
325 }
326 }
327 }
328 return refs
329}
330
331func validObject(data []byte, ref codec.ObjectRef) bool {
332 kind, payload, err := codec.PreflightObject(data)
333 return err == nil && kind == ref.Kind && len(payload) > 0 && len(data) == int(ref.Size)
334}
335
336func (r *Repository) objectDigest(data []byte) ports.SnapshotDigest {
337 return sha256.Sum256(data)
338}
339
340func (r *Repository) verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.ObjectRef) (bool, error) {
341 data, exists, err := r.readOptionalBounded(path)
342 if err != nil || !exists {
343 return exists, err
344 }
345 if r.objectDigest(data) != digest || !validObject(data, ref) {
346 return true, fmt.Errorf("existing immutable object is invalid")
347 }
348 return true, nil
349}