detached 0e3ca7c

perf(snapshot): avoid full-history publication reads

commit
0e3ca7c7f2297e4388a2897f1064a7147dbc4c7f
author
brice · · 2026-07-22 02:30 UTC
parent
058bbc8
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +280 -31
Split
UnifiedSplit
M internal/adapters/snapshot/publish.go +139 -29
@@ -21,7 +21,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
2121 return err
2222 }
2323
24- refs, supplied, err := validatePublication(publication)
24+ refs, err := validatePublicationManifest(publication)
2525 if err != nil {
2626 return err
2727 }
@@ -35,7 +35,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
3535 return err
3636 }
3737
38- current, currentManifest, err := r.currentGeneration(ctx, publication.Name, key)
38+ current, currentManifest, currentRefs, err := r.currentPublication(ctx, publication.Name, key)
3939 if err != nil {
4040 return err
4141 }
@@ -49,22 +49,37 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
4949 return nil
5050 }
5151
52+ // Retained refs from the authoritative generation were verified when their
53+ // immutable files were installed. They are intentionally not re-read when a
54+ // producer supplies them again. Every other referenced object is checked
55+ // from disk once, or validated and installed from the supplied bytes.
56+ supplied, err := suppliedNecessaryObjects(refs, currentRefs, publication.Objects)
57+ if err != nil {
58+ return err
59+ }
5260 for digest, ref := range refs {
61+ if retainedObject(currentRefs, digest, ref) {
62+ continue
63+ }
5364 if err := ctx.Err(); err != nil {
5465 return err
5566 }
5667 path := r.objectPath(key, digest)
57- exists, err := verifyObjectFile(path, digest, ref)
68+ exists, err := r.verifyObjectFile(path, digest, ref)
5869 if err != nil {
5970 return fmt.Errorf("verify object %x: %w", digest, err)
6071 }
6172 if exists {
6273 continue
6374 }
64- object, ok := supplied[digest]
65- if !ok {
75+ objects := supplied[digest]
76+ if len(objects) == 0 {
6677 return fmt.Errorf("missing referenced object %x", digest)
6778 }
79+ object, err := r.validateSuppliedObject(objects, digest, ref)
80+ if err != nil {
81+ return err
82+ }
6883 if r.hooks.beforeBlobWrite != nil {
6984 if err := r.hooks.beforeBlobWrite(path); err != nil {
7085 return err
@@ -74,7 +89,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
7489 return err
7590 }
7691 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {
77- if sha256.Sum256(existing) != digest || !validObject(existing, ref) {
92+ if r.objectDigest(existing) != digest || !validObject(existing, ref) {
7893 return fmt.Errorf("existing immutable object is invalid")
7994 }
8095 return nil
@@ -123,40 +138,124 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
123138 }
124139 return nil
125140 }
141+
142+// currentPublication takes the normal HEAD path without loading every object
143+// in the current generation. A damaged HEAD or manifest still uses the full
144+// fallback scan, because only Load's object validation can establish a safe
145+// replacement authoritative generation in that recovery case.
146+func (r *Repository) currentPublication(ctx context.Context, name, key string) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {
147+ generation, digest, err := r.readHead(key)
148+ if err == nil {
149+ data, readErr := readBounded(r.manifestPath(key, generation))
150+ if readErr == nil && sha256.Sum256(data) == digest {
151+ refs, validateErr := validateManifest(data, name, generation)
152+ if validateErr == nil {
153+ return generation, data, refs, nil
154+ }
155+ }
156+ }
157+
158+ generation, data, err := r.currentGeneration(ctx, name, key)
159+ if err != nil || generation == 0 {
160+ return generation, data, nil, err
161+ }
162+ refs, err := validateManifest(data, name, generation)
163+ if err != nil {
164+ return 0, nil, nil, err
165+ }
166+ return generation, data, refs, nil
167+}
168+
126169 func validatePublication(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, map[ports.SnapshotDigest]ports.SnapshotObject, error) {
127- manifest, err := codec.UnmarshalManifest(p.Manifest)
170+ refs, err := validatePublicationManifest(p)
171+ if err != nil {
172+ return nil, nil, err
173+ }
174+ supplied, err := suppliedNecessaryObjects(refs, nil, p.Objects)
128175 if err != nil {
129- return nil, nil, fmt.Errorf("invalid manifest: %w", err)
176+ return nil, nil, err
130177 }
131- if manifest.Name != p.Name || manifest.Generation != p.Generation || p.Generation == 0 {
132- return nil, nil, fmt.Errorf("manifest name or generation does not match publication")
178+ objects := make(map[ports.SnapshotDigest]ports.SnapshotObject, len(supplied))
179+ for digest, entries := range supplied {
180+ object, err := validateSuppliedObject(entries, digest, refs[digest], nil)
181+ if err != nil {
182+ return nil, nil, err
183+ }
184+ objects[digest] = object
185+ }
186+ return refs, objects, nil
187+}
188+
189+func validatePublicationManifest(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
190+ return validateManifest(p.Manifest, p.Name, p.Generation)
191+}
192+
193+func validateManifest(data []byte, name string, generation uint64) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
194+ manifest, err := codec.UnmarshalManifest(data)
195+ if err != nil {
196+ return nil, fmt.Errorf("invalid manifest: %w", err)
197+ }
198+ if manifest.Name != name || manifest.Generation != generation || generation == 0 {
199+ return nil, fmt.Errorf("manifest name or generation does not match publication")
133200 }
134201 refs := manifestRefs(manifest)
135202 if refs == nil {
136- return nil, nil, fmt.Errorf("conflicting manifest references")
203+ return nil, fmt.Errorf("conflicting manifest references")
137204 }
138- if !withinGenerationBudget(len(p.Manifest), refs) {
139- return nil, nil, fmt.Errorf("snapshot generation too large")
205+ if !withinGenerationBudget(len(data), refs) {
206+ return nil, fmt.Errorf("snapshot generation too large")
140207 }
141- objects := make(map[ports.SnapshotDigest]ports.SnapshotObject, len(p.Objects))
142- for _, object := range p.Objects {
143- if sha256.Sum256(object.Data) != object.Digest {
144- return nil, nil, fmt.Errorf("object digest mismatch")
208+ return refs, nil
209+}
210+
211+// suppliedNecessaryObjects first rejects unreferenced input without inspecting
212+// its payload. Objects retained from currentRefs cannot affect publication and
213+// are left out so callers do not pay to hash or copy a full history again.
214+func suppliedNecessaryObjects(refs, currentRefs map[ports.SnapshotDigest]codec.ObjectRef, supplied []ports.SnapshotObject) (map[ports.SnapshotDigest][]ports.SnapshotObject, error) {
215+ out := make(map[ports.SnapshotDigest][]ports.SnapshotObject)
216+ for _, object := range supplied {
217+ if _, used := refs[object.Digest]; !used {
218+ return nil, fmt.Errorf("unreferenced object")
219+ }
220+ if retainedObject(currentRefs, object.Digest, refs[object.Digest]) {
221+ continue
145222 }
146- ref, used := refs[object.Digest]
147- if !used {
148- return nil, nil, fmt.Errorf("unreferenced object")
223+ out[object.Digest] = append(out[object.Digest], object)
224+ }
225+ return out, nil
226+}
227+
228+func retainedObject(currentRefs map[ports.SnapshotDigest]codec.ObjectRef, digest ports.SnapshotDigest, ref codec.ObjectRef) bool {
229+ currentRef, ok := currentRefs[digest]
230+ return ok && currentRef == ref
231+}
232+
233+func (r *Repository) validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef) (ports.SnapshotObject, error) {
234+ return validateSuppliedObject(objects, digest, ref, r)
235+}
236+
237+func validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef, repository *Repository) (ports.SnapshotObject, error) {
238+ object := objects[0]
239+ for _, candidate := range objects {
240+ actual := sha256.Sum256(candidate.Data)
241+ if repository != nil {
242+ actual = repository.objectDigest(candidate.Data)
149243 }
150- kind, payload, err := codec.PreflightObject(object.Data)
151- if err != nil || kind != ref.Kind || len(object.Data) != int(ref.Size) || len(payload) == 0 {
152- return nil, nil, fmt.Errorf("invalid object envelope")
244+ if actual != candidate.Digest || candidate.Digest != digest {
245+ return ports.SnapshotObject{}, fmt.Errorf("object digest mismatch")
153246 }
154- if old, duplicate := objects[object.Digest]; duplicate && !equalBytes(old.Data, object.Data) {
155- return nil, nil, fmt.Errorf("conflicting object")
247+ if !equalBytes(candidate.Data, object.Data) {
248+ return ports.SnapshotObject{}, fmt.Errorf("conflicting object")
156249 }
157- objects[object.Digest] = ports.SnapshotObject{Digest: object.Digest, Data: append([]byte(nil), object.Data...)}
158250 }
159- return refs, objects, nil
251+ kind, payload, err := codec.PreflightObject(object.Data)
252+ if err != nil || kind != ref.Kind || len(object.Data) != int(ref.Size) || len(payload) == 0 {
253+ return ports.SnapshotObject{}, fmt.Errorf("invalid object envelope")
254+ }
255+ if repository != nil && repository.hooks.beforeObjectCopy != nil {
256+ repository.hooks.beforeObjectCopy(object.Data)
257+ }
258+ return ports.SnapshotObject{Digest: digest, Data: append([]byte(nil), object.Data...)}, nil
160259 }
161260
162261 func withinGenerationBudget(manifestSize int, refs map[ports.SnapshotDigest]codec.ObjectRef) bool {
@@ -196,17 +295,28 @@ func manifestRefs(manifest codec.Manifest) map[ports.SnapshotDigest]codec.Object
196295 }
197296 return refs
198297 }
298+
199299 func validObject(data []byte, ref codec.ObjectRef) bool {
200300 kind, payload, err := codec.PreflightObject(data)
201301 return err == nil && kind == ref.Kind && len(payload) > 0 && len(data) == int(ref.Size)
202302 }
203303
204-func verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.ObjectRef) (bool, error) {
304+func (r *Repository) objectDigest(data []byte) ports.SnapshotDigest {
305+ if r.hooks.beforeObjectHash != nil {
306+ r.hooks.beforeObjectHash(data)
307+ }
308+ return sha256.Sum256(data)
309+}
310+
311+func (r *Repository) verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.ObjectRef) (bool, error) {
312+ if r.hooks.beforeObjectRead != nil {
313+ r.hooks.beforeObjectRead(path)
314+ }
205315 data, exists, err := readOptionalBounded(path)
206316 if err != nil || !exists {
207317 return exists, err
208318 }
209- if sha256.Sum256(data) != digest || !validObject(data, ref) {
319+ if r.objectDigest(data) != digest || !validObject(data, ref) {
210320 return true, fmt.Errorf("existing immutable object is invalid")
211321 }
212322 return true, nil
@@ -21,7 +21,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
21 return err21 return err
22 }22 }
23 23
24- refs, supplied, err := validatePublication(publication)24+ refs, err := validatePublicationManifest(publication)
25 if err != nil {25 if err != nil {
26 return err26 return err
27 }27 }
@@ -35,7 +35,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
35 return err35 return err
36 }36 }
37 37
38- current, currentManifest, err := r.currentGeneration(ctx, publication.Name, key)38+ current, currentManifest, currentRefs, err := r.currentPublication(ctx, publication.Name, key)
39 if err != nil {39 if err != nil {
40 return err40 return err
41 }41 }
@@ -49,22 +49,37 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
49 return nil49 return nil
50 }50 }
51 51
52+ // Retained refs from the authoritative generation were verified when their
53+ // immutable files were installed. They are intentionally not re-read when a
54+ // producer supplies them again. Every other referenced object is checked
55+ // from disk once, or validated and installed from the supplied bytes.
56+ supplied, err := suppliedNecessaryObjects(refs, currentRefs, publication.Objects)
57+ if err != nil {
58+ return err
59+ }
52 for digest, ref := range refs {60 for digest, ref := range refs {
61+ if retainedObject(currentRefs, digest, ref) {
62+ continue
63+ }
53 if err := ctx.Err(); err != nil {64 if err := ctx.Err(); err != nil {
54 return err65 return err
55 }66 }
56 path := r.objectPath(key, digest)67 path := r.objectPath(key, digest)
57- exists, err := verifyObjectFile(path, digest, ref)68+ exists, err := r.verifyObjectFile(path, digest, ref)
58 if err != nil {69 if err != nil {
59 return fmt.Errorf("verify object %x: %w", digest, err)70 return fmt.Errorf("verify object %x: %w", digest, err)
60 }71 }
61 if exists {72 if exists {
62 continue73 continue
63 }74 }
64- object, ok := supplied[digest]75+ objects := supplied[digest]
65- if !ok {76+ if len(objects) == 0 {
66 return fmt.Errorf("missing referenced object %x", digest)77 return fmt.Errorf("missing referenced object %x", digest)
67 }78 }
79+ object, err := r.validateSuppliedObject(objects, digest, ref)
80+ if err != nil {
81+ return err
82+ }
68 if r.hooks.beforeBlobWrite != nil {83 if r.hooks.beforeBlobWrite != nil {
69 if err := r.hooks.beforeBlobWrite(path); err != nil {84 if err := r.hooks.beforeBlobWrite(path); err != nil {
70 return err85 return err
@@ -74,7 +89,7 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
74 return err89 return err
75 }90 }
76 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {91 if err := r.writeImmutable(path, object.Data, func(existing []byte) error {
77- if sha256.Sum256(existing) != digest || !validObject(existing, ref) {92+ if r.objectDigest(existing) != digest || !validObject(existing, ref) {
78 return fmt.Errorf("existing immutable object is invalid")93 return fmt.Errorf("existing immutable object is invalid")
79 }94 }
80 return nil95 return nil
@@ -123,40 +138,124 @@ func (r *Repository) Publish(ctx context.Context, publication ports.SnapshotPubl
123 }138 }
124 return nil139 return nil
125 }140 }
141+
142+// currentPublication takes the normal HEAD path without loading every object
143+// in the current generation. A damaged HEAD or manifest still uses the full
144+// fallback scan, because only Load's object validation can establish a safe
145+// replacement authoritative generation in that recovery case.
146+func (r *Repository) currentPublication(ctx context.Context, name, key string) (uint64, []byte, map[ports.SnapshotDigest]codec.ObjectRef, error) {
147+ generation, digest, err := r.readHead(key)
148+ if err == nil {
149+ data, readErr := readBounded(r.manifestPath(key, generation))
150+ if readErr == nil && sha256.Sum256(data) == digest {
151+ refs, validateErr := validateManifest(data, name, generation)
152+ if validateErr == nil {
153+ return generation, data, refs, nil
154+ }
155+ }
156+ }
157+
158+ generation, data, err := r.currentGeneration(ctx, name, key)
159+ if err != nil || generation == 0 {
160+ return generation, data, nil, err
161+ }
162+ refs, err := validateManifest(data, name, generation)
163+ if err != nil {
164+ return 0, nil, nil, err
165+ }
166+ return generation, data, refs, nil
167+}
168+
126 func validatePublication(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, map[ports.SnapshotDigest]ports.SnapshotObject, error) {169 func validatePublication(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, map[ports.SnapshotDigest]ports.SnapshotObject, error) {
127- manifest, err := codec.UnmarshalManifest(p.Manifest)170+ refs, err := validatePublicationManifest(p)
171+ if err != nil {
172+ return nil, nil, err
173+ }
174+ supplied, err := suppliedNecessaryObjects(refs, nil, p.Objects)
128 if err != nil {175 if err != nil {
129- return nil, nil, fmt.Errorf("invalid manifest: %w", err)176+ return nil, nil, err
130 }177 }
131- if manifest.Name != p.Name || manifest.Generation != p.Generation || p.Generation == 0 {178+ objects := make(map[ports.SnapshotDigest]ports.SnapshotObject, len(supplied))
132- return nil, nil, fmt.Errorf("manifest name or generation does not match publication")179+ for digest, entries := range supplied {
180+ object, err := validateSuppliedObject(entries, digest, refs[digest], nil)
181+ if err != nil {
182+ return nil, nil, err
183+ }
184+ objects[digest] = object
185+ }
186+ return refs, objects, nil
187+}
188+
189+func validatePublicationManifest(p ports.SnapshotPublication) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
190+ return validateManifest(p.Manifest, p.Name, p.Generation)
191+}
192+
193+func validateManifest(data []byte, name string, generation uint64) (map[ports.SnapshotDigest]codec.ObjectRef, error) {
194+ manifest, err := codec.UnmarshalManifest(data)
195+ if err != nil {
196+ return nil, fmt.Errorf("invalid manifest: %w", err)
197+ }
198+ if manifest.Name != name || manifest.Generation != generation || generation == 0 {
199+ return nil, fmt.Errorf("manifest name or generation does not match publication")
133 }200 }
134 refs := manifestRefs(manifest)201 refs := manifestRefs(manifest)
135 if refs == nil {202 if refs == nil {
136- return nil, nil, fmt.Errorf("conflicting manifest references")203+ return nil, fmt.Errorf("conflicting manifest references")
137 }204 }
138- if !withinGenerationBudget(len(p.Manifest), refs) {205+ if !withinGenerationBudget(len(data), refs) {
139- return nil, nil, fmt.Errorf("snapshot generation too large")206+ return nil, fmt.Errorf("snapshot generation too large")
140 }207 }
141- objects := make(map[ports.SnapshotDigest]ports.SnapshotObject, len(p.Objects))208+ return refs, nil
142- for _, object := range p.Objects {209+}
143- if sha256.Sum256(object.Data) != object.Digest {210+
144- return nil, nil, fmt.Errorf("object digest mismatch")211+// suppliedNecessaryObjects first rejects unreferenced input without inspecting
212+// its payload. Objects retained from currentRefs cannot affect publication and
213+// are left out so callers do not pay to hash or copy a full history again.
214+func suppliedNecessaryObjects(refs, currentRefs map[ports.SnapshotDigest]codec.ObjectRef, supplied []ports.SnapshotObject) (map[ports.SnapshotDigest][]ports.SnapshotObject, error) {
215+ out := make(map[ports.SnapshotDigest][]ports.SnapshotObject)
216+ for _, object := range supplied {
217+ if _, used := refs[object.Digest]; !used {
218+ return nil, fmt.Errorf("unreferenced object")
219+ }
220+ if retainedObject(currentRefs, object.Digest, refs[object.Digest]) {
221+ continue
145 }222 }
146- ref, used := refs[object.Digest]223+ out[object.Digest] = append(out[object.Digest], object)
147- if !used {224+ }
148- return nil, nil, fmt.Errorf("unreferenced object")225+ return out, nil
226+}
227+
228+func retainedObject(currentRefs map[ports.SnapshotDigest]codec.ObjectRef, digest ports.SnapshotDigest, ref codec.ObjectRef) bool {
229+ currentRef, ok := currentRefs[digest]
230+ return ok && currentRef == ref
231+}
232+
233+func (r *Repository) validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef) (ports.SnapshotObject, error) {
234+ return validateSuppliedObject(objects, digest, ref, r)
235+}
236+
237+func validateSuppliedObject(objects []ports.SnapshotObject, digest ports.SnapshotDigest, ref codec.ObjectRef, repository *Repository) (ports.SnapshotObject, error) {
238+ object := objects[0]
239+ for _, candidate := range objects {
240+ actual := sha256.Sum256(candidate.Data)
241+ if repository != nil {
242+ actual = repository.objectDigest(candidate.Data)
149 }243 }
150- kind, payload, err := codec.PreflightObject(object.Data)244+ if actual != candidate.Digest || candidate.Digest != digest {
151- if err != nil || kind != ref.Kind || len(object.Data) != int(ref.Size) || len(payload) == 0 {245+ return ports.SnapshotObject{}, fmt.Errorf("object digest mismatch")
152- return nil, nil, fmt.Errorf("invalid object envelope")
153 }246 }
154- if old, duplicate := objects[object.Digest]; duplicate && !equalBytes(old.Data, object.Data) {247+ if !equalBytes(candidate.Data, object.Data) {
155- return nil, nil, fmt.Errorf("conflicting object")248+ return ports.SnapshotObject{}, fmt.Errorf("conflicting object")
156 }249 }
157- objects[object.Digest] = ports.SnapshotObject{Digest: object.Digest, Data: append([]byte(nil), object.Data...)}
158 }250 }
159- return refs, objects, nil251+ kind, payload, err := codec.PreflightObject(object.Data)
252+ if err != nil || kind != ref.Kind || len(object.Data) != int(ref.Size) || len(payload) == 0 {
253+ return ports.SnapshotObject{}, fmt.Errorf("invalid object envelope")
254+ }
255+ if repository != nil && repository.hooks.beforeObjectCopy != nil {
256+ repository.hooks.beforeObjectCopy(object.Data)
257+ }
258+ return ports.SnapshotObject{Digest: digest, Data: append([]byte(nil), object.Data...)}, nil
160 }259 }
161 260
162 func withinGenerationBudget(manifestSize int, refs map[ports.SnapshotDigest]codec.ObjectRef) bool {261 func withinGenerationBudget(manifestSize int, refs map[ports.SnapshotDigest]codec.ObjectRef) bool {
@@ -196,17 +295,28 @@ func manifestRefs(manifest codec.Manifest) map[ports.SnapshotDigest]codec.Object
196 }295 }
197 return refs296 return refs
198 }297 }
298+
199 func validObject(data []byte, ref codec.ObjectRef) bool {299 func validObject(data []byte, ref codec.ObjectRef) bool {
200 kind, payload, err := codec.PreflightObject(data)300 kind, payload, err := codec.PreflightObject(data)
201 return err == nil && kind == ref.Kind && len(payload) > 0 && len(data) == int(ref.Size)301 return err == nil && kind == ref.Kind && len(payload) > 0 && len(data) == int(ref.Size)
202 }302 }
203 303
204-func verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.ObjectRef) (bool, error) {304+func (r *Repository) objectDigest(data []byte) ports.SnapshotDigest {
305+ if r.hooks.beforeObjectHash != nil {
306+ r.hooks.beforeObjectHash(data)
307+ }
308+ return sha256.Sum256(data)
309+}
310+
311+func (r *Repository) verifyObjectFile(path string, digest ports.SnapshotDigest, ref codec.ObjectRef) (bool, error) {
312+ if r.hooks.beforeObjectRead != nil {
313+ r.hooks.beforeObjectRead(path)
314+ }
205 data, exists, err := readOptionalBounded(path)315 data, exists, err := readOptionalBounded(path)
206 if err != nil || !exists {316 if err != nil || !exists {
207 return exists, err317 return exists, err
208 }318 }
209- if sha256.Sum256(data) != digest || !validObject(data, ref) {319+ if r.objectDigest(data) != digest || !validObject(data, ref) {
210 return true, fmt.Errorf("existing immutable object is invalid")320 return true, fmt.Errorf("existing immutable object is invalid")
211 }321 }
212 return true, nil322 return true, nil
A internal/adapters/snapshot/publish_efficiency_test.go +133 -0
@@ -0,0 +1,133 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "fmt"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+
11+ "github.com/bnema/vev/internal/ports"
12+ codec "github.com/bnema/vev/internal/usecase/snapshot"
13+)
14+
15+func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
16+ const historyObjects = 10_000
17+
18+ repo := NewRepository(privateDir(t))
19+ first, second := largeIncrementalPublications(t, "named", historyObjects)
20+ seedCompletePublication(t, repo, first)
21+
22+ var reads, hashes, copies int
23+ repo.hooks.beforeObjectRead = func(string) { reads++ }
24+ repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
25+ repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
26+
27+ if err := repo.Publish(context.Background(), second); err != nil {
28+ t.Fatal(err)
29+ }
30+ // Only the new mutable tail and visible objects need work. The 10k sealed
31+ // objects are retained history supplied again by the producer, not input
32+ // that Publish needs to read, hash, or copy.
33+ if reads != 2 || hashes != 2 || copies != 2 {
34+ t.Fatalf("unchanged 10k history reads/hashes/copies = %d/%d/%d, want 2/2/2", reads, hashes, copies)
35+ }
36+ got, err := repo.Load(context.Background(), second.Name)
37+ if err != nil {
38+ t.Fatal(err)
39+ }
40+ if got.Generation != second.Generation {
41+ t.Fatalf("published generation = %d, want %d", got.Generation, second.Generation)
42+ }
43+}
44+
45+func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
46+ repo := NewRepository(privateDir(t))
47+ first := repositoryPublication(t, "named", 1, []byte("first"))
48+ if err := repo.Publish(context.Background(), first); err != nil {
49+ t.Fatal(err)
50+ }
51+ second := repositoryPublication(t, "named", 2, []byte("second"))
52+ newTail := second.Objects[0]
53+ key := sessionKey(second.Name)
54+ if err := os.MkdirAll(filepath.Dir(repo.objectPath(key, newTail.Digest)), 0o700); err != nil {
55+ t.Fatal(err)
56+ }
57+ if err := os.WriteFile(repo.objectPath(key, newTail.Digest), newTail.Data, 0o600); err != nil {
58+ t.Fatal(err)
59+ }
60+
61+ var reads, hashes, copies int
62+ repo.hooks.beforeObjectRead = func(string) { reads++ }
63+ repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
64+ repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
65+ if err := repo.Publish(context.Background(), second); err != nil {
66+ t.Fatal(err)
67+ }
68+ // The existing new tail is checked from disk once. The unchanged visible
69+ // object is retained from generation one and needs no further work.
70+ if reads != 1 || hashes != 1 || copies != 0 {
71+ t.Fatalf("necessary object reads/hashes/copies = %d/%d/%d, want 1/1/0", reads, hashes, copies)
72+ }
73+}
74+
75+func largeIncrementalPublications(t *testing.T, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {
76+ t.Helper()
77+ sealed := make([]codec.ObjectRef, 0, count)
78+ objects := make([]ports.SnapshotObject, 0, count+2)
79+ for i := range count {
80+ object, err := codec.MarshalObject(codec.HistoryChunk, []byte(fmt.Sprintf("history-%d", i)))
81+ if err != nil {
82+ t.Fatal(err)
83+ }
84+ sealed = append(sealed, codec.ObjectRef{Kind: codec.HistoryChunk, Digest: object.Digest, Size: uint32(len(object.Data))})
85+ objects = append(objects, object)
86+ }
87+ makePublication := func(generation uint64, tailPayload, visiblePayload string) ports.SnapshotPublication {
88+ t.Helper()
89+ tail, err := codec.MarshalObject(codec.HistoryTail, []byte(tailPayload))
90+ if err != nil {
91+ t.Fatal(err)
92+ }
93+ visible, err := codec.MarshalObject(codec.Visible, []byte(visiblePayload))
94+ if err != nil {
95+ t.Fatal(err)
96+ }
97+ manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, 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))}}}}}})
98+ if err != nil {
99+ t.Fatal(err)
100+ }
101+ return ports.SnapshotPublication{Name: name, Generation: generation, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)}
102+ }
103+ return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2")
104+}
105+
106+func seedCompletePublication(t *testing.T, repo *Repository, publication ports.SnapshotPublication) {
107+ t.Helper()
108+ key := sessionKey(publication.Name)
109+ if err := repo.ensureSession(key); err != nil {
110+ t.Fatal(err)
111+ }
112+ manifest, err := codec.UnmarshalManifest(publication.Manifest)
113+ if err != nil {
114+ t.Fatal(err)
115+ }
116+ for _, object := range publication.Objects {
117+ if err := os.MkdirAll(filepath.Dir(repo.objectPath(key, object.Digest)), 0o700); err != nil {
118+ t.Fatal(err)
119+ }
120+ if err := os.WriteFile(repo.objectPath(key, object.Digest), object.Data, 0o600); err != nil {
121+ t.Fatal(err)
122+ }
123+ }
124+ if err := os.WriteFile(repo.manifestPath(key, publication.Generation), publication.Manifest, 0o600); err != nil {
125+ t.Fatal(err)
126+ }
127+ if err := os.WriteFile(repo.headPath(key), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest)), 0o600); err != nil {
128+ t.Fatal(err)
129+ }
130+ if refs := manifestRefs(manifest); refs == nil {
131+ t.Fatal("invalid seeded manifest")
132+ }
133+}
@@ -0,0 +1,133 @@
1+package snapshot
2+
3+import (
4+ "context"
5+ "crypto/sha256"
6+ "fmt"
7+ "os"
8+ "path/filepath"
9+ "testing"
10+
11+ "github.com/bnema/vev/internal/ports"
12+ codec "github.com/bnema/vev/internal/usecase/snapshot"
13+)
14+
15+func TestRepositoryPublishSkipsUnchangedTenThousandObjectHistory(t *testing.T) {
16+ const historyObjects = 10_000
17+
18+ repo := NewRepository(privateDir(t))
19+ first, second := largeIncrementalPublications(t, "named", historyObjects)
20+ seedCompletePublication(t, repo, first)
21+
22+ var reads, hashes, copies int
23+ repo.hooks.beforeObjectRead = func(string) { reads++ }
24+ repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
25+ repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
26+
27+ if err := repo.Publish(context.Background(), second); err != nil {
28+ t.Fatal(err)
29+ }
30+ // Only the new mutable tail and visible objects need work. The 10k sealed
31+ // objects are retained history supplied again by the producer, not input
32+ // that Publish needs to read, hash, or copy.
33+ if reads != 2 || hashes != 2 || copies != 2 {
34+ t.Fatalf("unchanged 10k history reads/hashes/copies = %d/%d/%d, want 2/2/2", reads, hashes, copies)
35+ }
36+ got, err := repo.Load(context.Background(), second.Name)
37+ if err != nil {
38+ t.Fatal(err)
39+ }
40+ if got.Generation != second.Generation {
41+ t.Fatalf("published generation = %d, want %d", got.Generation, second.Generation)
42+ }
43+}
44+
45+func TestRepositoryPublishVerifiesNecessaryExistingObjectOnce(t *testing.T) {
46+ repo := NewRepository(privateDir(t))
47+ first := repositoryPublication(t, "named", 1, []byte("first"))
48+ if err := repo.Publish(context.Background(), first); err != nil {
49+ t.Fatal(err)
50+ }
51+ second := repositoryPublication(t, "named", 2, []byte("second"))
52+ newTail := second.Objects[0]
53+ key := sessionKey(second.Name)
54+ if err := os.MkdirAll(filepath.Dir(repo.objectPath(key, newTail.Digest)), 0o700); err != nil {
55+ t.Fatal(err)
56+ }
57+ if err := os.WriteFile(repo.objectPath(key, newTail.Digest), newTail.Data, 0o600); err != nil {
58+ t.Fatal(err)
59+ }
60+
61+ var reads, hashes, copies int
62+ repo.hooks.beforeObjectRead = func(string) { reads++ }
63+ repo.hooks.beforeObjectHash = func([]byte) { hashes++ }
64+ repo.hooks.beforeObjectCopy = func([]byte) { copies++ }
65+ if err := repo.Publish(context.Background(), second); err != nil {
66+ t.Fatal(err)
67+ }
68+ // The existing new tail is checked from disk once. The unchanged visible
69+ // object is retained from generation one and needs no further work.
70+ if reads != 1 || hashes != 1 || copies != 0 {
71+ t.Fatalf("necessary object reads/hashes/copies = %d/%d/%d, want 1/1/0", reads, hashes, copies)
72+ }
73+}
74+
75+func largeIncrementalPublications(t *testing.T, name string, count int) (ports.SnapshotPublication, ports.SnapshotPublication) {
76+ t.Helper()
77+ sealed := make([]codec.ObjectRef, 0, count)
78+ objects := make([]ports.SnapshotObject, 0, count+2)
79+ for i := range count {
80+ object, err := codec.MarshalObject(codec.HistoryChunk, []byte(fmt.Sprintf("history-%d", i)))
81+ if err != nil {
82+ t.Fatal(err)
83+ }
84+ sealed = append(sealed, codec.ObjectRef{Kind: codec.HistoryChunk, Digest: object.Digest, Size: uint32(len(object.Data))})
85+ objects = append(objects, object)
86+ }
87+ makePublication := func(generation uint64, tailPayload, visiblePayload string) ports.SnapshotPublication {
88+ t.Helper()
89+ tail, err := codec.MarshalObject(codec.HistoryTail, []byte(tailPayload))
90+ if err != nil {
91+ t.Fatal(err)
92+ }
93+ visible, err := codec.MarshalObject(codec.Visible, []byte(visiblePayload))
94+ if err != nil {
95+ t.Fatal(err)
96+ }
97+ manifest, err := codec.MarshalManifest(codec.Manifest{Generation: generation, 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))}}}}}})
98+ if err != nil {
99+ t.Fatal(err)
100+ }
101+ return ports.SnapshotPublication{Name: name, Generation: generation, Manifest: manifest, Objects: append(append([]ports.SnapshotObject(nil), objects...), tail, visible)}
102+ }
103+ return makePublication(1, "tail-1", "visible-1"), makePublication(2, "tail-2", "visible-2")
104+}
105+
106+func seedCompletePublication(t *testing.T, repo *Repository, publication ports.SnapshotPublication) {
107+ t.Helper()
108+ key := sessionKey(publication.Name)
109+ if err := repo.ensureSession(key); err != nil {
110+ t.Fatal(err)
111+ }
112+ manifest, err := codec.UnmarshalManifest(publication.Manifest)
113+ if err != nil {
114+ t.Fatal(err)
115+ }
116+ for _, object := range publication.Objects {
117+ if err := os.MkdirAll(filepath.Dir(repo.objectPath(key, object.Digest)), 0o700); err != nil {
118+ t.Fatal(err)
119+ }
120+ if err := os.WriteFile(repo.objectPath(key, object.Digest), object.Data, 0o600); err != nil {
121+ t.Fatal(err)
122+ }
123+ }
124+ if err := os.WriteFile(repo.manifestPath(key, publication.Generation), publication.Manifest, 0o600); err != nil {
125+ t.Fatal(err)
126+ }
127+ if err := os.WriteFile(repo.headPath(key), marshalHead(publication.Generation, sha256.Sum256(publication.Manifest)), 0o600); err != nil {
128+ t.Fatal(err)
129+ }
130+ if refs := manifestRefs(manifest); refs == nil {
131+ t.Fatal("invalid seeded manifest")
132+ }
133+}
M internal/adapters/snapshot/repository.go +8 -2
@@ -48,8 +48,14 @@ type Repository struct {
4848 // immediately before their respective syscall and are intentionally package
4949 // private so production callers cannot weaken repository guarantees.
5050 type repositoryHooks struct {
51- beforeBlobWrite func(string) error
52- beforeManifestWrite func(string) error
51+ beforeBlobWrite func(string) error
52+ beforeManifestWrite func(string) error
53+ // Object hooks instrument the publication path in package tests. They keep
54+ // the steady-state cost of retained history observable without exposing a
55+ // production metrics surface.
56+ beforeObjectRead func(string)
57+ beforeObjectHash func([]byte)
58+ beforeObjectCopy func([]byte)
5359 beforeHeadWrite func(string) error
5460 createTemp func(string) error
5561 writeTemp func(string) error
@@ -48,8 +48,14 @@ type Repository struct {
48 // immediately before their respective syscall and are intentionally package48 // immediately before their respective syscall and are intentionally package
49 // private so production callers cannot weaken repository guarantees.49 // private so production callers cannot weaken repository guarantees.
50 type repositoryHooks struct {50 type repositoryHooks struct {
51- beforeBlobWrite func(string) error51+ beforeBlobWrite func(string) error
52- beforeManifestWrite func(string) error52+ beforeManifestWrite func(string) error
53+ // Object hooks instrument the publication path in package tests. They keep
54+ // the steady-state cost of retained history observable without exposing a
55+ // production metrics surface.
56+ beforeObjectRead func(string)
57+ beforeObjectHash func([]byte)
58+ beforeObjectCopy func([]byte)
53 beforeHeadWrite func(string) error59 beforeHeadWrite func(string) error
54 createTemp func(string) error60 createTemp func(string) error
55 writeTemp func(string) error61 writeTemp func(string) error

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close