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)

2026-10-25 01:42:28 +02:00
parent b1f758ed71
commit 82c89bec65
3 changed files with 106 additions and 1 deletions
@@ -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()
+}
@@ -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)
+	}
+}
@@ -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)