From 82c89bec6505aeb40a99bc56850b66e8781dada1 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 25 Oct 2026 01:42:28 +0200 Subject: [PATCH] Split compositions from the inline body so their large files' objects are put, and refuse a composition naming an object the bucket lacks (issue 1945) --- cmd/mesh-controller/composed_live.go | 12 +++- cmd/mesh-controller/composed_objects_test.go | 58 ++++++++++++++++++++ cmd/mesh-controller/composed_writer.go | 37 +++++++++++++ 3 files changed, 106 insertions(+), 1 deletion(-) create mode 100644 cmd/mesh-controller/composed_objects_test.go diff --git a/cmd/mesh-controller/composed_live.go b/cmd/mesh-controller/composed_live.go index a1ea3ead..436be543 100644 --- a/cmd/mesh-controller/composed_live.go +++ b/cmd/mesh-controller/composed_live.go @@ -83,7 +83,7 @@ func composeModules(ctx context.Context, open *stores, nodes []string) (map[stri errs = append(errs, fmt.Errorf("%s: %w", node, err)) continue } - body, err := declared.Body() + body, err := compositionBody(declared) if err != nil { errs = append(errs, fmt.Errorf("%s: %w", node, err)) continue @@ -360,3 +360,13 @@ func sliceOfModule(resources []json.RawMessage, module string) []json.RawMessage } return out } + +// compositionBody is the body a node's compositions are split from: every file's content inline, never by digest, so +// writeComposed puts each object of a file over 64 KiB in the node's bucket before it writes the composition that names +// it (ADR 0539 §4). A send's body names those files by digest for a node that reads references, and a send put their +// objects itself; with sends gone, a composition split from that body named objects nothing put (issue 1945, live on +// 2026-10-25: novox's nats.bus-users). +func compositionBody(declared sendable) ([]byte, error) { + declared.fileReferences = false + return declared.Body() +} diff --git a/cmd/mesh-controller/composed_objects_test.go b/cmd/mesh-controller/composed_objects_test.go new file mode 100644 index 00000000..5716ec24 --- /dev/null +++ b/cmd/mesh-controller/composed_objects_test.go @@ -0,0 +1,58 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "strings" + "testing" + + sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration" +) + +// A composition's file over 64 KiB reaches the node by its digest, and its object is in the node's bucket before the +// composition is written (ADR 0539 §4; issue 1945, live on 2026-10-25 at 01:22: novox's node-engine could not fetch +// nats.bus-users, because the body the compositions were split from already named the file by digest, so the writer +// found no content to put, and nothing else put it once sends were gone). +func TestALargeFileOfACompositionHasItsObjectInTheBucket(t *testing.T) { + big := strings.Repeat("user = x\n", sdkdecl.FileReferenceThreshold/8) + declared := sendable{fileReferences: true, Resources: []map[string]any{{"id": "nats.bus-users", "type": "file", + "path": "/var/lib/nats-module/conf/accounts.conf", "mode": "0644", "content": big}}} + body, err := compositionBody(declared) + if err != nil { + t.Fatal(err) + } + comps, _, err := compositionsOf("novox", body, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + at, _, _, _ := aPlace(t, "novox") + if _, err := writeNodeCompositions(context.Background(), at, comps); err != nil { + t.Fatal(err) + } + sum := sha256.Sum256([]byte(big)) + if _, ok := at.objects.(fakeObjects)[hex.EncodeToString(sum[:])]; !ok { + t.Fatalf("the object of nats.bus-users is not in the bucket: %d objects", len(at.objects.(fakeObjects))) + } +} + +// A composition that names a file by a digest whose object the bucket does not hold is refused, never written: a node +// that read it could not apply it. +func TestACompositionNamingAnObjectTheBucketLacksIsRefused(t *testing.T) { + big := strings.Repeat("user = x\n", sdkdecl.FileReferenceThreshold/8) + declared := sendable{fileReferences: true, Resources: []map[string]any{{"id": "nats.bus-users", "type": "file", + "path": "/var/lib/nats-module/conf/accounts.conf", "mode": "0644", "content": big}}} + body, err := declared.Body() // the body as a send named it: by digest, the content gone + if err != nil { + t.Fatal(err) + } + comps, _, err := compositionsOf("novox", body, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + at, _, _, _ := aPlace(t, "novox") + if _, err := writeNodeCompositions(context.Background(), at, comps); err == nil || + !strings.Contains(err.Error(), "does not hold") { + t.Fatalf("a composition naming a missing object was written: %v", err) + } +} diff --git a/cmd/mesh-controller/composed_writer.go b/cmd/mesh-controller/composed_writer.go index 2529e613..be268e93 100644 --- a/cmd/mesh-controller/composed_writer.go +++ b/cmd/mesh-controller/composed_writer.go @@ -178,6 +178,9 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv return false, err } } + if err := objectsHeld(ctx, at, key, c, objects); err != nil { + return false, err + } digest, err := sdkdelivery.ComposedDigestOf(c) if err != nil { return false, err @@ -213,6 +216,40 @@ func writeComposed(ctx context.Context, at composedPlace, key string, c sdkdeliv return wrote || written, err } +// objectsHeld refuses a composition that names a file by a digest whose object the node's bucket does not hold: a node +// that read it could not apply it (issue 1945). An object this write just put is held. +func objectsHeld(ctx context.Context, at composedPlace, key string, c sdkdelivery.Composed, put map[string][]byte) error { + for _, raw := range c.Resources { + r, err := sdkdecl.ParseResource(raw) + if err != nil { + return err + } + f, ok := r.(*sdkdecl.File) + if !ok || !f.IsReference() { + continue + } + if _, ok := put[f.Digest]; ok { + continue + } + name, err := sdkdecl.ObjectName(f.Digest) + if err != nil { + return fmt.Errorf("%s: file %s: %w", key, f.ID, err) + } + if at.objects == nil { + return fmt.Errorf("%s names file %s by digest and %s's object bucket is not open", key, f.ID, at.node) + } + have, _, err := at.objects.Has(ctx, name) + if err != nil { + return fmt.Errorf("%s: whether the object %s of file %s is in the bucket: %w", key, name[:12], f.ID, err) + } + if !have { + return fmt.Errorf("%s names file %s by the object %s, which %s's bucket does not hold: not written", key, f.ID, + name[:12], at.node) + } + } + return nil +} + // readComposed is the composition a bucket holds under key and its revision; revision 0 when there is none. func readComposed(ctx context.Context, kv composedBucketsKV, key string) (sdkdelivery.Composed, uint64, error) { e, err := kv.Get(ctx, key)