mesh-controller/cmd/mesh-controller/declaration_files_test.go

package main

import (
	"context"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"

	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
)

// contentOf is a deterministic content of n bytes that does not compress to nothing and holds no character JSON escapes.
func contentOf(seed string, n int) string {
	var b strings.Builder
	for i := 0; b.Len() < n; i++ {
		fmt.Fprintf(&b, "%s-%07d-%x;", seed, i, i*2654435761%1000003)
	}
	return b.String()[:n]
}

func aFileResource(id, content string) map[string]any {
	return map[string]any{"id": id, "type": "file", "path": "/etc/" + id, "mode": "0644", "content": content}
}

// controlNodeShape is a declaration shaped like the control node's on 2026-10-21 (issue 1689), the sizes measured then: a bus users
// file of 252,678 bytes, the vault's export of 138,475 bytes (compact), the tool runtime 21 KB, the packet filter 19 KB, the launcher
// 14 KB, and about 600 small resources for the rest (about 286 KB). `more` scales the count of small ones: 1 is the shape, 1.4 forty
// percent more of them.
func controlNodeShape(more float64) []map[string]any {
	res := []map[string]any{
		aFileResource("nats.bus-users", contentOf("users", 252678)),
		aFileResource("mesh-vault.kept", contentOf("kept", 138475)),
		aFileResource("node-tools.runtime", contentOf("runtime", 21<<10)),
		aFileResource("nftables.filtering", contentOf("filter", 19<<10)),
		aFileResource("mesh-host.launcher", contentOf("launcher", 14<<10)),
	}
	for i := 0; i < int(600*more); i++ {
		res = append(res, aFileResource(fmt.Sprintf("module-%03d.config", i), contentOf(fmt.Sprintf("small%d", i), 380)))
	}
	return res
}

func wireSizeOf(t *testing.T, s sendable) int {
	t.Helper()
	body, err := s.Body()
	if err != nil {
		t.Fatal(err)
	}
	// The signature the controller's key makes is 64 bytes.
	msg, err := link.Envelope(body, make([]byte, 64))
	if err != nil {
		t.Fatal(err)
	}
	return len(msg)
}

// 1696/AC1: for a node that reports the capability, the two big files of a control-node-shaped declaration carry their digest and size
// and no content, and the declaration is far under the bus's 1 MiB.
func TestAFileOver64KiBTravelsByDigestToANodeThatReadsOne(t *testing.T) { // 1696/AC1
	users, kept := contentOf("users", 252678), contentOf("kept", 138475)
	s := sendable{fileReferences: true, Resources: []map[string]any{
		aFileResource("nats.bus-users", users), aFileResource("mesh-vault.kept", kept),
		aFileResource("small", "tiny\n"),
	}}
	body, err := s.Body()
	if err != nil {
		t.Fatal(err)
	}
	var got struct {
		Resources []map[string]any `json:"resources"`
	}
	if err := json.Unmarshal(body, &got); err != nil {
		t.Fatal(err)
	}
	for i, want := range []string{users, kept} {
		r := got.Resources[i]
		digest := sdkdecl.DigestOf([]byte(want))
		if _, has := r["content"]; has || r["digest"] != digest || r["size"] != float64(len(want)) {
			t.Errorf("%v goes with content %v, digest %v, size %v; want no content, %s and %d", r["id"], r["content"] != nil,
				r["digest"], r["size"], digest, len(want))
		}
		if r["path"] == nil || r["mode"] != "0644" {
			t.Errorf("%v lost its path or mode: %v", r["id"], r)
		}
	}
	if got.Resources[2]["content"] != "tiny\n" || got.Resources[2]["digest"] != nil {
		t.Errorf("a small file is not inline: %v", got.Resources[2])
	}
	if strings.Contains(string(body), users[:200]) || strings.Contains(string(body), kept[:200]) {
		t.Error("the content of a file that went by digest is in the body")
	}
	if n := wireSizeOf(t, s); n > 100<<10 {
		t.Errorf("the declaration is %d bytes on the wire, want under 100 KB", n)
	}
	// The objects to put are the contents, each under the hex of its digest, and a reference verifies against its object.
	objects, err := s.objects()
	if err != nil || len(objects) != 2 {
		t.Fatalf("objects: %v %v", objects, err)
	}
	for _, want := range []string{users, kept} {
		name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf([]byte(want)))
		if string(objects[name]) != want {
			t.Errorf("the object %s does not hold the content", name[:12])
		}
	}
	named, err := namedObjects(body)
	if err != nil || len(named) != 2 {
		t.Fatalf("the objects the body names: %v %v", named, err)
	}
	for name := range objects {
		if _, ok := named[name]; !ok {
			t.Errorf("the body does not name the object %s it was sent with", name[:12])
		}
		if _, err := hex.DecodeString(name); err != nil || len(name) != 64 {
			t.Errorf("the object name %q is not 64 hex characters", name)
		}
	}
}

// 1696/AC2: a node that does not report the capability is sent every content inline, as before, and has no object put.
func TestANodeWithoutTheCapabilityIsSentEveryContentInline(t *testing.T) { // 1696/AC2
	users := contentOf("users", 252678)
	s := sendable{Resources: []map[string]any{aFileResource("nats.bus-users", users)}}
	body, err := s.Body()
	if err != nil {
		t.Fatal(err)
	}
	var got struct {
		Resources []map[string]any `json:"resources"`
	}
	if err := json.Unmarshal(body, &got); err != nil {
		t.Fatal(err)
	}
	if got.Resources[0]["content"] != users || got.Resources[0]["digest"] != nil || got.Resources[0]["size"] != nil {
		t.Fatalf("a node that reads no references is not sent the content inline: %v", got.Resources[0]["digest"])
	}
	if objects, err := s.objects(); err != nil || len(objects) != 0 {
		t.Fatalf("objects for a node that reads none: %v %v", len(objects), err)
	}
}

// 1696/AC3: only a file that may travel by digest does: one of exactly 64 KiB, a sealed one, one filled with secrets, one written
// into another file, and a directory stay inline, whatever the node reads.
func TestOnlyAFileThatMayTravelByDigestDoes(t *testing.T) { // 1696/AC3
	exact := contentOf("exact", 64<<10)
	big := contentOf("big", 64<<10+1)
	sealed := aFileResource("sealed", big)
	sealed["sealed"], sealed["content"] = "box", ""
	sealed["content"] = big // a sealed file's content is none; the body must pass it as it is, whatever it holds
	secret := aFileResource("secret", big)
	secret["secrets"] = map[string]any{"token": "x"}
	into := aFileResource("into", big)
	into["into"] = "/etc/other"
	s := sendable{fileReferences: true, Resources: []map[string]any{
		aFileResource("exact", exact), sealed, secret, into,
		{"id": "dir", "type": "directory", "path": "/etc/d", "content": big},
		aFileResource("big", big),
	}}
	body, err := s.Body()
	if err != nil {
		t.Fatal(err)
	}
	var got struct {
		Resources []map[string]any `json:"resources"`
	}
	if err := json.Unmarshal(body, &got); err != nil {
		t.Fatal(err)
	}
	for i, id := range []string{"exact", "sealed", "secret", "into", "dir"} {
		if got.Resources[i]["id"] != id || got.Resources[i]["digest"] != nil || got.Resources[i]["content"] == nil {
			t.Errorf("%s went by digest or lost its content: %v", id, got.Resources[i]["digest"])
		}
	}
	if big := got.Resources[5]; big["id"] != "big" || big["digest"] == nil || big["content"] != nil {
		t.Errorf("a plain file of 64 KiB and a byte did not go by digest: %v", big["digest"])
	}
}

// A file given as bytes goes by digest on its decoded length, with the decoded bytes in the object.
func TestAFileOfBytesTravelsByDigestOnItsDecodedLength(t *testing.T) {
	raw := []byte(contentOf("png", 70<<10))
	r := map[string]any{"id": "icon", "type": "file", "path": "/usr/share/icon.png", "bytes": stdBase64(raw)}
	s := sendable{fileReferences: true, Resources: []map[string]any{r}}
	objects, err := s.objects()
	if err != nil || len(objects) != 1 {
		t.Fatalf("%v %v", len(objects), err)
	}
	name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf(raw))
	if string(objects[name]) != string(raw) {
		t.Fatal("the object is not the decoded bytes")
	}
	if r["bytes"] == nil {
		t.Fatal("the composed resource was changed in place")
	}
}

type memObjects struct {
	have   map[string][]byte
	marked map[string]bool
	puts   []string
	fail   map[string]error
}

func (m *memObjects) Has(_ context.Context, name string) (bool, bool, error) {
	_, ok := m.have[name]
	return ok, m.marked[name], nil
}
func (m *memObjects) Put(_ context.Context, name string, c []byte) error {
	if err := m.fail[name]; err != nil {
		return err
	}
	m.puts = append(m.puts, name)
	m.have[name] = c
	delete(m.marked, name)
	return nil
}

// 1696/AC4: an object already in the bucket is not put again, one absent is, and a put that fails stops the send, naming the object.
func TestObjectsAreInTheBucketBeforeTheDeclarationIsSent(t *testing.T) { // 1696/AC4
	objects := map[string][]byte{strings.Repeat("a", 64): []byte("one"), strings.Repeat("b", 64): []byte("two")}
	b := &memObjects{have: map[string][]byte{strings.Repeat("a", 64): []byte("one")}}
	if err := putObjects(t.Context(), b, objects); err != nil || len(b.puts) != 1 || b.puts[0] != strings.Repeat("b", 64) {
		t.Fatalf("put %v (%v), want the one absent", b.puts, err)
	}
	// An object the sweep marked unnamed is put again, which erases the mark: a send relies on it.
	b = &memObjects{have: map[string][]byte{strings.Repeat("a", 64): []byte("one"), strings.Repeat("b", 64): []byte("two")},
		marked: map[string]bool{strings.Repeat("a", 64): true}}
	if err := putObjects(t.Context(), b, objects); err != nil || len(b.puts) != 1 || b.puts[0] != strings.Repeat("a", 64) || b.marked[strings.Repeat("a", 64)] {
		t.Fatalf("a marked object: put %v (%v) marked %v, want it put again and the mark gone", b.puts, err, b.marked)
	}
	b = &memObjects{have: map[string][]byte{}, fail: map[string]error{strings.Repeat("a", 64): errors.New("permissions violation")}}
	err := putObjects(t.Context(), b, objects)
	if err == nil || !strings.Contains(err.Error(), "aaaaaaaaaaaa") || !strings.Contains(err.Error(), "permissions violation") {
		t.Fatalf("a failed put: %v", err)
	}
}

// fallbackDelivery refuses the first declaration it is sent with the bucket's failure, as a bus that has not yet been given the
// controller's grant does, and takes the second.
type fallbackDelivery struct {
	bodies [][]byte
	first  error
}

func (f *fallbackDelivery) grant(context.Context, []readyNode) error { return nil }
func (f *fallbackDelivery) declare(_ context.Context, _ readyNode, body []byte) (string, error) {
	f.bodies = append(f.bodies, body)
	if len(f.bodies) == 1 && f.first != nil {
		return "", fmt.Errorf("x: %w: %w", errFilesNotPut, f.first)
	}
	return "digest", nil
}

// 1696/AC5: when the bucket cannot be written yet (the grant travels in the declaration of the machine holding the bus), that send goes
// with every content inline and is made; a send that fails for another reason is not retried inline.
func TestASendWhoseFilesCannotBePutGoesInlineOnce(t *testing.T) { // 1696/AC5
	users := contentOf("users", 100<<10)
	s := readyNode{node: "novox", declared: sendable{fileReferences: true, Resources: []map[string]any{aFileResource("nats.bus-users", users)}}}
	f := &fallbackDelivery{first: errors.New("timeout")}
	digests, err := deliver(t.Context(), f, "novox", []readyNode{s})
	if err != nil || digests["novox"] != "digest" || len(f.bodies) != 2 {
		t.Fatalf("%v %v %d", digests, err, len(f.bodies))
	}
	if strings.Contains(string(f.bodies[0]), users[:100]) || !strings.Contains(string(f.bodies[1]), users[:100]) {
		t.Error("the first body should name the file by digest and the second carry it inline")
	}
	other := &fallbackDelivery{}
	other.first = nil
	failing := &failingDelivery{err: errors.New("the bus is down")}
	if _, err := deliver(t.Context(), failing, "novox", []readyNode{s}); err == nil || failing.calls != 1 {
		t.Fatalf("a failure of another kind: %v after %d send(s)", err, failing.calls)
	}
}

type failingDelivery struct {
	err   error
	calls int
}

func (f *failingDelivery) grant(context.Context, []readyNode) error { return nil }
func (f *failingDelivery) declare(context.Context, readyNode, []byte) (string, error) {
	f.calls++
	return "", f.err
}

// A bucket for the sweep.
type sweepBucketFake struct {
	objects map[string]time.Time // zero: named at the last look
	deleted []string
}

func (b *sweepBucketFake) Objects(context.Context) ([]bucketObject, error) {
	var out []bucketObject
	for n, at := range b.objects {
		out = append(out, bucketObject{Name: n, UnnamedSince: at})
	}
	return out, nil
}
func (b *sweepBucketFake) MarkUnnamed(_ context.Context, n string, at time.Time) error {
	b.objects[n] = at
	return nil
}
func (b *sweepBucketFake) ClearUnnamed(_ context.Context, n string) error {
	b.objects[n] = time.Time{}
	return nil
}
func (b *sweepBucketFake) Delete(_ context.Context, n string, markedAt time.Time) (bool, error) {
	if at, ok := b.objects[n]; !ok || !at.Equal(markedAt) {
		return false, nil // put again since it was judged: its mark is gone
	}
	delete(b.objects, n)
	b.deleted = append(b.deleted, n)
	return true, nil
}

// 1696/AC6: an object a latest declaration still names is not removed, however long; one unnamed is marked at the first look and removed
// only when the mark is 24 hours old; one named again loses its mark.
func TestAnObjectIsRemovedADayAfterNoLatestDeclarationNamesIt(t *testing.T) { // 1696/AC6
	named, unnamed, back := strings.Repeat("1", 64), strings.Repeat("2", 64), strings.Repeat("3", 64)
	t0 := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
	b := &sweepBucketFake{objects: map[string]time.Time{named: t0.Add(-100 * time.Hour), unnamed: {}, back: t0.Add(-5 * time.Hour)}}
	named3 := map[string]bool{named: true, back: true}
	r, err := sweepDeclarationFiles(t.Context(), b, named3, t0)
	if err != nil || r.Marked != 1 || r.Cleared != 2 || r.Removed != 0 || len(b.deleted) != 0 {
		t.Fatalf("first look: %+v %v %v", r, err, b.deleted)
	}
	if !b.objects[named].IsZero() || !b.objects[back].IsZero() || !b.objects[unnamed].Equal(t0) {
		t.Fatalf("marks after the first look: %v", b.objects)
	}
	// Still unnamed 23 hours 59 minutes later: kept. A day after the mark: removed. The named ones stay for ever.
	if r, err = sweepDeclarationFiles(t.Context(), b, map[string]bool{named: true, back: true}, t0.Add(24*time.Hour-time.Minute)); err != nil ||
		r.Removed != 0 || len(b.deleted) != 0 {
		t.Fatalf("a minute short of a day: %+v %v %v", r, err, b.deleted)
	}
	if r, err = sweepDeclarationFiles(t.Context(), b, map[string]bool{named: true, back: true}, t0.Add(24*time.Hour)); err != nil ||
		r.Removed != 1 || len(b.deleted) != 1 || b.deleted[0] != unnamed {
		t.Fatalf("a day after: %+v %v %v", r, err, b.deleted)
	}
	if _, ok := b.objects[named]; !ok {
		t.Fatal("an object a latest declaration names was removed")
	}
	// An object named again before the day is up loses its mark and is not removed.
	b = &sweepBucketFake{objects: map[string]time.Time{unnamed: t0}}
	if _, err := sweepDeclarationFiles(t.Context(), b, map[string]bool{unnamed: true}, t0.Add(23*time.Hour)); err != nil || !b.objects[unnamed].IsZero() {
		t.Fatalf("named again: %v %v", b.objects, err)
	}
}

// What a node's latest declaration names is read from the signed envelope as stored on the bus; an unreadable one is an error and
// never "names none", which would mark an object a node needs.
func TestTheObjectsALatestDeclarationNamesAreReadFromTheBus(t *testing.T) {
	s := sendable{fileReferences: true, Resources: []map[string]any{aFileResource("a", contentOf("a", 70<<10))}}
	body, _ := s.Body()
	msg, _ := link.Envelope(body, []byte("sig"))
	named, err := namedByLatest("novox", msg)
	name, _ := sdkdecl.ObjectName(sdkdecl.DigestOf([]byte(contentOf("a", 70<<10))))
	if err != nil || len(named) != 1 || !named[name] {
		t.Fatalf("%v %v", named, err)
	}
	if none, err := namedByLatest("ace", []byte(`{"declaration":"e30=","signature":""}`)); err != nil || len(none) != 0 {
		t.Fatalf("a declaration naming none: %v %v", none, err)
	}
	if _, err := namedByLatest("novox", []byte("not json")); err == nil {
		t.Fatal("an unreadable latest declaration was read as naming none")
	}
}

// 1698/AC2 (ADR 0486 §9): a declaration shaped like the control node's, for a node that reads references, stays under half of the
// bus's largest message on the wire; the inline form of the same does not fit at all; and the same shape with 40 percent more resources
// fails the test's own limit, so a change that eats the room fails in a pull request and not on the live mesh.
func TestTheControlNodesDeclarationInTheReferenceFormStaysUnderHalfTheBusLimit(t *testing.T) { // 1698/AC2
	const limit = 1048576
	shape := sendable{fileReferences: true, Resources: controlNodeShape(1)}
	n := wireSizeOf(t, shape)
	t.Logf("the control node's shape in the reference form: %d bytes on the wire, %.1f%% of %d", n, 100*float64(n)/limit, limit)
	if n > limit/2 {
		t.Errorf("the reference form is %d bytes on the wire, over half (%d) of the bus's largest message", n, limit/2)
	}
	if inline := wireSizeOf(t, sendable{Resources: controlNodeShape(1)}); inline < limit*3/4 {
		t.Errorf("the inline form is %d bytes, which is not the shape that was over the limit", inline)
	}
	grown := wireSizeOf(t, sendable{fileReferences: true, Resources: controlNodeShape(1.4)})
	if grown <= limit/2 {
		t.Errorf("the shape with 40%% more resources is %d bytes, which the test would let pass", grown)
	}
}

func stdBase64(b []byte) string { return base64.StdEncoding.EncodeToString(b) }

// A sweep that read an object's mark before a send put it again must not remove it: the object, now without the mark, is left.
func TestASweepDoesNotRemoveAnObjectASendPutAgainAfterItWasJudged(t *testing.T) { // 1696/AC8
	n := strings.Repeat("4", 64)
	t0 := time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)
	b := &sweepBucketFake{objects: map[string]time.Time{n: t0}}
	judged := bucketObject{Name: n, UnnamedSince: t0}
	b.objects[n] = time.Time{} // the send put it again: no mark
	if gone, err := b.Delete(t.Context(), judged.Name, judged.UnnamedSince); err != nil || gone || len(b.deleted) != 0 {
		t.Fatalf("an object put again was removed: %v %v %v", gone, err, b.deleted)
	}
}

// One node's bucket is held by one act at a time in this process: the sweep waits for a send that has put its objects and not yet
// published the declaration that names them (1696/AC9).
func TestTheSweepOfANodesBucketWaitsForASendThatHasPutItsObjects(t *testing.T) { // 1696/AC9
	unlock := lockFilesOf("lock-test-node")
	got := make(chan struct{})
	go func() {
		defer lockFilesOf("lock-test-node")()
		close(got)
	}()
	select {
	case <-got:
		t.Fatal("the sweep took the node's bucket while a send held it")
	case <-time.After(50 * time.Millisecond):
	}
	other := lockFilesOf("lock-test-other-node") // another node's bucket is not held
	other()
	unlock()
	select {
	case <-got:
	case <-time.After(2 * time.Second):
		t.Fatal("the sweep did not take the node's bucket once the send was done")
	}
}

// A put that fails send after send raises a condition; the first failure does not (the grant it waits for travels in that very
// declaration), and a put that works clears it (1696/AC10).
func TestAPutThatFailsSendAfterSendRaisesAConditionAndClearsWhenItWorks(t *testing.T) { // 1696/AC10
	store := conditions.NewInMemory()
	k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store,
		Teller: &conditions.Told{}, Now: func() time.Time { return time.Date(2026, 10, 21, 14, 0, 0, 0, time.UTC) }})
	t.Cleanup(func() { k.Close(t.Context()) })
	was := conditionsFrom
	conditionsFrom = k
	t.Cleanup(func() { conditionsFrom = was })
	key := filesNotPutObservation("cond-node", "").Key()
	open := func() bool {
		_, found, err := conditions.ReadOne(t.Context(), store, key)
		if err != nil {
			t.Fatal(err)
		}
		return found
	}
	boom := errors.New("permissions violation")
	notePut(t.Context(), "cond-node", boom)
	if open() {
		t.Fatal("the first failure raised the condition")
	}
	notePut(t.Context(), "cond-node", boom)
	if !open() {
		t.Fatal("the second failure in a row did not raise the condition")
	}
	notePut(t.Context(), "cond-node", nil)
	if open() {
		t.Fatal("a put that worked did not clear the condition")
	}
	notePut(t.Context(), "cond-node", boom)
	if open() {
		t.Fatal("the count did not start again after a put that worked")
	}
}

// aBusThatSays is a bus that keeps the declarations published to it and, at each, asks whether the node's bucket is held.
type aBusThatSays struct {
	link.Bus
	published     chan string
	heldAtPublish func() bool
}

func (b *aBusThatSays) PublishDeclaration(_ context.Context, node string, _ []byte) error {
	if b.heldAtPublish != nil && !b.heldAtPublish() {
		return errors.New("published while the node's bucket was not held")
	}
	b.published <- node
	return errors.New("stop here: nothing is recorded")
}

type aSignerOfZeros struct{}

func (aSignerOfZeros) Sign(context.Context, []byte) ([]byte, error) { return make([]byte, 64), nil }

// whileHeld says whether the node's lock is held by someone right now, without keeping it.
func whileHeld(node string) bool {
	got := make(chan struct{})
	go func() {
		defer lockFilesOf(node)()
		close(got)
	}()
	select {
	case <-got:
		return false
	case <-time.After(100 * time.Millisecond):
		// held: the goroutine stays waiting and takes and releases the lock once it is let go.
		return true
	}
}

// 1696/AC9: a send holds the node's lock from the put to the publication, so the sweep never removes an object between them. The
// test goes through overTheBus.declare: while another holds the lock the declaration is not published, and at the publication the
// lock is held.
func TestASendHoldsTheNodesLockFromThePutToThePublication(t *testing.T) { // 1696/AC9
	const node = "lock-send-node"
	bus := &aBusThatSays{published: make(chan string, 1)}
	bus.heldAtPublish = func() bool { return whileHeld(node) }
	send := overTheBus{server: link.SendingOn(bus), signer: aSignerOfZeros{}}
	release := lockFilesOf(node) // a sweep is judging this node's bucket
	done := make(chan error, 1)
	go func() {
		_, err := send.declare(t.Context(), readyNode{node: node}, []byte(`{}`))
		done <- err
	}()
	select {
	case n := <-bus.published:
		t.Fatalf("%s was sent its declaration while the sweep held the node's bucket", n)
	case <-time.After(150 * time.Millisecond):
	}
	release()
	select {
	case <-bus.published:
	case <-time.After(5 * time.Second):
		t.Fatal("the send did not go on once the sweep let go")
	}
	if err := <-done; err == nil || !strings.Contains(err.Error(), "nothing is recorded") {
		t.Fatalf("the send did not reach the publication (a lock held at it): %v", err)
	}
}