mesh-host/internal/link/have_test.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// What the node-engine writes beside its report: a module's have, and the node's own facts (novox/hq ADR 0429 §4,
// issue 1132).

// haveLink is a link with a bucket behind it that keeps each key's revision as the bus does.
type haveLink struct {
	*quietLink
	hmu                  sync.Mutex
	kv                   map[string]haveEntry
	sets                 []string // keys written, in order
	events               []string // "report" and "have <key>", in the order they happened
	failing              bool
	binds                int
	setTries             int
	bindFails, bindStuck bool
}

type haveEntry struct {
	value []byte
	rev   uint64
}

func newHaveLink() *haveLink { return &haveLink{quietLink: newQuietLink(), kv: map[string]haveEntry{}} }

func (l *haveLink) Report(ctx context.Context, node string, body []byte) error {
	err := l.quietLink.Report(ctx, node, body)
	if err == nil {
		l.hmu.Lock()
		l.events = append(l.events, "report")
		l.hmu.Unlock()
	}
	return err
}

func (l *haveLink) Health(context.Context, string, []byte) error { return nil }

func (l *haveLink) HaveBind(ctx context.Context) (HaveBucket, error) {
	l.hmu.Lock()
	l.binds++
	stuck, refuse := l.bindStuck, l.bindFails
	l.hmu.Unlock()
	if stuck {
		<-ctx.Done()
		return nil, ctx.Err()
	}
	if refuse {
		return nil, errors.New("no grant for the have bucket")
	}
	return l, nil
}

func (l *haveLink) Get(_ context.Context, key string) ([]byte, uint64, error) {
	l.hmu.Lock()
	defer l.hmu.Unlock()
	e, ok := l.kv[key]
	if !ok {
		return nil, 0, ErrHaveAbsent
	}
	return e.value, e.rev, nil
}

func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uint64) error {
	l.hmu.Lock()
	defer l.hmu.Unlock()
	l.setTries++
	if l.failing {
		return errors.New("the bucket refused")
	}
	e, ok := l.kv[key]
	if (revision == 0) == ok || (ok && e.rev != revision) {
		return errors.New("wrong last sequence")
	}
	l.kv[key] = haveEntry{value: value, rev: e.rev + 1}
	l.sets = append(l.sets, key)
	l.events = append(l.events, "have "+key)
	return nil
}

func (l *haveLink) have(t *testing.T, key string) delivery.Have {
	t.Helper()
	l.hmu.Lock()
	defer l.hmu.Unlock()
	h, err := delivery.DecodeHave(l.kv[key].value)
	if err != nil {
		t.Fatalf("%s holds no have: %v", key, err)
	}
	return h
}

func (l *haveLink) setCount() int {
	l.hmu.Lock()
	defer l.hmu.Unlock()
	return len(l.sets)
}

func (l *haveLink) put(t *testing.T, key string, h delivery.Have) {
	t.Helper()
	raw, err := delivery.EncodeHave(h)
	if err != nil {
		t.Fatal(err)
	}
	l.hmu.Lock()
	defer l.hmu.Unlock()
	l.kv[key] = haveEntry{value: raw, rev: l.kv[key].rev + 1}
}

func healthOf(states map[string]string) *Health {
	h := &Health{Contract: LivenessContract, At: time.Now().UTC()}
	for module, state := range states {
		h.Resources = append(h.Resources, ResourceHealth{Module: module, Resource: module + ".server", State: state})
	}
	return h
}

func appliedModules() Report {
	return Report{Applied: []string{"a.x", "b.x"},
		Modules: []ModuleFact{{Module: "a", Build: "b-1"}, {Module: "b", Build: "b-2", Failed: true, Detail: "b.x failed: no space"}},
		Health:  healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy})}
}

func aHaveQueue(l *haveLink) (*Session, *[]string) {
	var said []string
	var mu sync.Mutex
	q := &Session{Membership: Membership{Node: "n1"}, Timeout: time.Second, Say: func(s string) {
		mu.Lock()
		said = append(said, s)
		mu.Unlock()
	}}
	q.Attach(context.Background(), l)
	return q, &said
}

// 1132/AC3, AC5: a module that applied is running with its build and a module that failed is failed with the detail;
// the key does not exist yet, so it is created.
func TestAnAppliedModuleIsWrittenRunningAndAFailedOneFailed(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	a, b := l.have(t, "n1.a"), l.have(t, "n1.b")
	if a.Phase != delivery.PhaseRunning || a.Build != "b-1" || a.Health != delivery.HealthOK {
		t.Errorf("a: %+v", a)
	}
	if b.Phase != delivery.PhaseFailed || b.Build != "b-2" || !strings.Contains(b.Detail, "no space") ||
		b.Health != delivery.HealthFailed {
		t.Errorf("b: %+v", b)
	}
	if l.kv["n1.a"].rev != 1 {
		t.Errorf("a was created at revision %d", l.kv["n1.a"].rev)
	}
}

// 1132/AC5: a have equal to the bucket's value is not written; a different one is written over the last revision.
func TestAHaveIsWrittenOnlyWhenItDiffers(t *testing.T) {
	l := newHaveLink()
	l.put(t, "n1.a", delivery.Have{V: delivery.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	l.put(t, "n1.b", delivery.Have{V: delivery.Version, Build: "b-0", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	if l.kv["n1.a"].rev != 1 {
		t.Errorf("a equal to the bucket was written again, to revision %d", l.kv["n1.a"].rev)
	}
	if l.kv["n1.b"].rev != 2 || l.have(t, "n1.b").Phase != delivery.PhaseFailed {
		t.Errorf("b was not updated over its last revision: %+v at %d", l.have(t, "n1.b"), l.kv["n1.b"].rev)
	}
	before := l.setCount()
	q.SayFacts(context.Background(), appliedModules())
	if l.setCount() != before {
		t.Errorf("an apply that changed nothing wrote %d keys", l.setCount()-before)
	}
}

// 1132/AC4: a look's health reaches the module's have: ok, failed, waits or unknown, and only a change is written.
func TestAModulesHealthFollowsTheLook(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	rep := Report{Modules: []ModuleFact{{Module: "a", Build: "b"}, {Module: "b", Build: "b"},
		{Module: "c", Build: "b"}, {Module: "d", Build: "b"}}}
	q.SayFacts(context.Background(), rep)
	for _, m := range []string{"a", "b", "c", "d"} {
		if h := l.have(t, "n1."+m).Health; h != delivery.HealthUnknown {
			t.Errorf("%s with no look is %s, want unknown", m, h)
		}
	}
	h := healthOf(map[string]string{"a": StateHealthy, "b": StateUnhealthy, "c": StateStarting})
	h.Resources = append(h.Resources, ResourceHealth{Module: "a", Resource: "a.two", State: StateWaiting},
		ResourceHealth{Module: "b", Resource: "b.two", State: StateHealthy}, ResourceHealth{Module: "d", Resource: "d.x", State: StateUnknown})
	if !q.SayHealth(context.Background(), *h) {
		t.Fatal("not said")
	}
	for m, want := range map[string]delivery.Health{"a": delivery.HealthWaits, "b": delivery.HealthFailed,
		"c": delivery.HealthWaits, "d": delivery.HealthUnknown} {
		if got := l.have(t, "n1."+m).Health; got != want {
			t.Errorf("%s is %s, want %s", m, got, want)
		}
	}
	before := l.setCount()
	h2 := *h
	q.SayHealth(context.Background(), h2)
	if l.setCount() != before {
		t.Errorf("the same look wrote %d keys", l.setCount()-before)
	}
	h2.Resources = []ResourceHealth{{Module: "a", Resource: "a.server", State: StateHealthy}}
	q.SayHealth(context.Background(), h2)
	if got := l.have(t, "n1.a").Health; got != delivery.HealthOK {
		t.Errorf("a after a healthy look is %s", got)
	}
}

// 1132/AC6: the report is sent first and carries no new key; the node's key holds its windows, busy, strays, filters
// and profile.
func TestTheReportIsSentFirstAndTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	rep := appliedModules()
	rep.Windows = []Window{{Step: "backup", Holds: []string{"db"}}}
	rep.Busy = []Busy{{ID: "x", Container: "x", State: "waiting"}}
	rep.Strays = []Stray{{Kind: "container", Name: "old"}}
	rep.Filters = []Filter{{Where: "inet mesh", Owner: "mesh"}}
	rep.Profile = map[string]any{"cores": 8}
	q.SayFacts(context.Background(), rep)
	if len(l.events) < 2 || l.events[0] != "report" {
		t.Fatalf("the report was not first: %v", l.events)
	}
	raw, _ := json.Marshal(l.said()[0])
	if strings.Contains(string(raw), "Modules") || strings.Contains(string(raw), "\"modules\"") {
		t.Errorf("the report carries the module facts: %s", raw)
	}
	got, err := delivery.DecodeHaveNode(l.kv["n1.node"].value)
	if err != nil {
		t.Fatal(err)
	}
	enc := func(v any) json.RawMessage { b, _ := json.Marshal(v); return b }
	want := delivery.HaveNode{V: delivery.Version, Windows: enc(rep.Windows), Busy: enc(rep.Busy), Strays: enc(rep.Strays),
		Filters: enc(rep.Filters), Profile: enc(rep.Profile)}
	if !delivery.SameHaveNode(got, want) {
		t.Errorf("the node key holds %+v, want %+v", got, want)
	}
}

// 1132/AC7: a write the bucket refuses is said in the log; the report went out, and the next apply writes again.
func TestAFailedHaveWriteIsSaidAndNeverBlocksTheReport(t *testing.T) {
	l := newHaveLink()
	l.failing = true
	q, said := aHaveQueue(l)
	if !q.SayFacts(context.Background(), appliedModules()) {
		t.Fatal("the report did not go out")
	}
	if len(l.said()) != 1 {
		t.Fatalf("reports sent: %d", len(l.said()))
	}
	if !strings.Contains(strings.Join(*said, "\n"), "n1.a") || !strings.Contains(strings.Join(*said, "\n"), "refused") {
		t.Errorf("the failure is not in the log: %v", *said)
	}
	l.hmu.Lock()
	l.failing = false
	l.hmu.Unlock()
	q.SayFacts(context.Background(), appliedModules())
	if l.have(t, "n1.a").Phase != delivery.PhaseRunning {
		t.Error("the next apply did not write again")
	}
}

// A link without a bucket, and a report with no module facts (a refusal), write nothing and say nothing wrong.
func TestNoBucketAndNoFactsWriteNothing(t *testing.T) {
	plain := newQuietLink()
	q := &Session{Membership: Membership{Node: "n1"}, Timeout: time.Second}
	q.Attach(context.Background(), plain)
	if !q.SayFacts(context.Background(), appliedModules()) {
		t.Fatal("the report did not go out")
	}
	l := newHaveLink()
	q2, _ := aHaveQueue(l)
	q2.SayFacts(context.Background(), Report{Refused: "no"})
	if l.setCount() != 0 {
		t.Errorf("a refusal wrote %d keys", l.setCount())
	}
}

// 1132/AC7: a bind that fails, or never answers, writes nothing and returns within the queue's timeout; a write that
// fails ends the pass, and is said once.
func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
	for name, set := range map[string]func(*haveLink){
		"a bind refused":            func(l *haveLink) { l.bindFails = true },
		"a bind that never answers": func(l *haveLink) { l.bindStuck = true },
	} {
		l := newHaveLink()
		set(l)
		q, said := aHaveQueue(l)
		q.Timeout = 100 * time.Millisecond
		start := time.Now()
		if !q.SayFacts(context.Background(), appliedModules()) {
			t.Fatalf("%s: the report did not go out", name)
		}
		if took := time.Since(start); took > time.Second {
			t.Errorf("%s: the pass took %s", name, took)
		}
		if l.binds != 1 || l.setTries != 0 || len(*said) != 1 {
			t.Errorf("%s: binds %d, writes tried %d, said %v; want one bind, no write, one line", name, l.binds, l.setTries, *said)
		}
	}
	l := newHaveLink()
	l.failing = true
	q, said := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	if l.binds != 1 || l.setTries != 1 || len(*said) != 1 {
		t.Errorf("a failed write: binds %d, writes tried %d, said %v; want one of each and one line", l.binds, l.setTries, *said)
	}
}

func onlyA() Report {
	return Report{Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1"}},
		Health: healthOf(map[string]string{"a": StateHealthy})}
}

// 1369/AC1: after a declaration with a and b, one with a only writes n1.b with the phase removed, over the revision
// read, and leaves a as it was; the key is not deleted.
func TestAModuleLeavingTheDeclarationIsWrittenRemoved(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	revA, revB := l.kv["n1.a"].rev, l.kv["n1.b"].rev
	q.SayFacts(context.Background(), onlyA())
	b := l.have(t, "n1.b")
	if b.Phase != delivery.PhaseRemoved || b.Build != "b-2" {
		t.Errorf("b: %+v", b)
	}
	if l.kv["n1.b"].rev != revB+1 {
		t.Errorf("b was written at revision %d, want one over %d", l.kv["n1.b"].rev, revB)
	}
	if l.kv["n1.a"].rev != revA || l.have(t, "n1.a").Phase != delivery.PhaseRunning {
		t.Errorf("a changed: %+v at %d", l.have(t, "n1.a"), l.kv["n1.a"].rev)
	}
	before := l.setCount()
	q.SayFacts(context.Background(), onlyA())
	if l.setCount() != before {
		t.Errorf("a second apply without b wrote %d keys, removed is written once", l.setCount()-before)
	}
}

// 1369/AC2: a module removed and later added back is written with its new phase.
func TestARemovedModuleAddedBackIsWrittenWithItsNewPhase(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	q.SayFacts(context.Background(), onlyA())
	q.SayFacts(context.Background(), appliedModules())
	if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseFailed || b.Build != "b-2" {
		t.Errorf("b after it came back: %+v", b)
	}
	q.SayFacts(context.Background(), onlyA())
	if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
		t.Errorf("b removed again: %+v", b)
	}
}

// 1369/AC3: a module this node-engine never wrote a have key for is not written, and a key that is gone from the
// bucket is not created again.
func TestAModuleNeverWrittenIsNotWrittenRemoved(t *testing.T) {
	l := newHaveLink()
	l.put(t, "n1.c", delivery.Have{V: delivery.Version, Build: "b-9", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	q.SayFacts(context.Background(), onlyA())
	if c := l.have(t, "n1.c"); c.Phase != delivery.PhaseRunning || l.kv["n1.c"].rev != 1 {
		t.Errorf("c, which this node-engine never wrote, changed: %+v at %d", c, l.kv["n1.c"].rev)
	}
	// a key that vanished from the bucket is not created by the removal
	l2 := newHaveLink()
	q2, _ := aHaveQueue(l2)
	q2.SayFacts(context.Background(), appliedModules())
	l2.hmu.Lock()
	delete(l2.kv, "n1.b")
	l2.hmu.Unlock()
	q2.SayFacts(context.Background(), onlyA())
	if _, there := l2.kv["n1.b"]; there {
		t.Error("the removal created a key that was not there")
	}
}

// A removal the bucket refused is written by the next pass.
func TestARefusedRemovalIsWrittenByTheNextPass(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), appliedModules())
	l.hmu.Lock()
	l.failing = true
	l.hmu.Unlock()
	q.SayFacts(context.Background(), onlyA())
	l.hmu.Lock()
	l.failing = false
	l.hmu.Unlock()
	q.SayFacts(context.Background(), onlyA())
	if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
		t.Errorf("b: %+v", b)
	}
}

// 1369/F1: a node-engine that started since wrote nothing, and the report names the modules the kept declaration held
// before: one of them the new declaration lacks is written removed, one that is not in the bucket is not created.
func TestARestartedNodeEngineWritesTheRemovalFromTheReportsBefore(t *testing.T) {
	l := newHaveLink()
	l.put(t, "n1.a", delivery.Have{V: delivery.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	l.put(t, "n1.b", delivery.Have{V: delivery.Version, Build: "b-2", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
	q, _ := aHaveQueue(l) // fresh: nothing written by this process
	r := onlyA()
	r.Before = []string{"a", "b", "c"}
	q.SayFacts(context.Background(), r)
	if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
		t.Errorf("b: %+v", b)
	}
	if _, there := l.kv["n1.c"]; there {
		t.Error("c, which has no key, was created")
	}
	if l.kv["n1.a"].rev != 1 {
		t.Errorf("a was written, revision %d", l.kv["n1.a"].rev)
	}
}

// R6-A AC7: the first have this node-engine writes — or finds already saying what it would write — is told to the
// proof, which proves a node-engine on trial to its launcher (novox/hq ADR 0539 §9); a write the bucket refuses is not.
func TestTheFirstHaveWrittenIsToldToTheProof(t *testing.T) {
	l := newHaveLink()
	l.failing = true
	q, _ := aHaveQueue(l)
	var told int
	q.HaveWritten = func() { told++ }
	q.SayFacts(context.Background(), appliedModules())
	if told != 0 {
		t.Fatalf("a refused write proved the build %d time(s)", told)
	}
	l.hmu.Lock()
	l.failing = false
	l.hmu.Unlock()
	q.SayFacts(context.Background(), appliedModules())
	if told == 0 {
		t.Fatal("a have written under this build was not told")
	}

	// The bucket already holds what this node-engine would write: a restart with nothing changed still proves it.
	same := newHaveLink()
	q2, _ := aHaveQueue(same)
	q2.SayFacts(context.Background(), appliedModules())
	again, _ := aHaveQueue(same)
	told = 0
	again.HaveWritten = func() { told++ }
	again.SayFacts(context.Background(), appliedModules())
	if told == 0 {
		t.Fatal("a have already saying the same was not told")
	}
}

// R6-A AC6: an applied module's have names its fingerprint and the digest of the composition applied; a health check that
// fails after the module's start window gives health failed, and the have keeps both.
func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	fp := "src2:" + strings.Repeat("a", 64)
	dig := "sha256:" + strings.Repeat("b", 64)
	q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig})
	h := l.have(t, "n1.web")
	if h.FP != fp || h.Composed != dig || h.Phase != delivery.PhaseRunning || h.Health != delivery.HealthUnknown {
		t.Fatalf("the have after the apply: %+v", h)
	}
	// Starting: it waits. Past its start window the check fails: failed.
	q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateStarting}))
	if got := l.have(t, "n1.web"); got.Health != delivery.HealthWaits || got.FP != fp || got.Composed != dig {
		t.Fatalf("while it starts: %+v", got)
	}
	q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateUnhealthy}))
	if got := l.have(t, "n1.web"); got.Health != delivery.HealthFailed || got.FP != fp || got.Composed != dig {
		t.Fatalf("a failing check after the start window: %+v", got)
	}

	// A module held for a failed backup says so and what it keeps running; removed is written once, over its key.
	q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Phase: string(delivery.PhaseHeld), Detail: "the repository is locked"})
	if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseHeld || got.Detail == "" {
		t.Fatalf("held: %+v", got)
	}
	q.NoteRemoved(context.Background(), "web", "")
	if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved {
		t.Fatalf("removed: %+v", got)
	}
	q.NoteNode(context.Background(), dig)
	if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig {
		t.Fatalf("the node key: %+v %v", n, err)
	}
}

// A module applied and running that has nothing long-running to judge is ok once the engine has looked: there is nothing
// of it that could be down. Unknown kept it from ever being healthy, so the assigner's canary rule held every other node
// of mesh-host back for good (issue 1945, live on 2026-10-25: ace's have of mesh-host said unknown, and novox and shanks
// never got the new node-engine). A failed one stays failed, and before any look it is unknown.
func TestAModuleWithNothingToJudgeIsOkOnceLookedAt(t *testing.T) {
	l := newHaveLink()
	q, _ := aHaveQueue(l)
	q.SayFacts(context.Background(), Report{Modules: []ModuleFact{{Module: "mesh-host", Build: "b"}, {Module: "web", Build: "b"},
		{Module: "broken", Build: "b", Failed: true, Detail: "x"}}})
	if h := l.have(t, "n1.mesh-host").Health; h != delivery.HealthUnknown {
		t.Fatalf("before any look: %s", h)
	}
	if !q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) {
		t.Fatal("not said")
	}
	if h := l.have(t, "n1.mesh-host").Health; h != delivery.HealthOK {
		t.Fatalf("a running module with nothing to judge after a look: %s", h)
	}
	if h := l.have(t, "n1.broken").Health; h == delivery.HealthOK {
		t.Fatalf("a failed module was said ok")
	}
}

// A have an older engine left unknown is written over with this engine's health once the module is said again, as the
// node agent says a module it finds running on its own have: a packages-only module (xorg, nothing long-running) is ok
// after a look, so the assigner no longer holds back what requires its provision (2026-10-25, g14 and shanks).
func TestAModuleSaidAgainOverAnOlderEnginesUnknownIsWrittenWithThisEnginesHealth(t *testing.T) {
	l := newHaveLink()
	fp := "src2:" + strings.Repeat("a", 64)
	dig := "sha256:" + strings.Repeat("b", 64)
	l.put(t, "n1.xorg", delivery.Have{V: delivery.Version, FP: fp, Composed: dig, Phase: delivery.PhaseRunning,
		Health: delivery.HealthUnknown})
	q, _ := aHaveQueue(l)
	if !q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) {
		t.Fatal("not said")
	}
	if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthUnknown {
		t.Fatalf("a module the session does not hold was written: %+v", h)
	}
	q.NoteModule(context.Background(), ModuleFact{Module: "xorg", FP: fp, Composed: dig})
	if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthOK || h.FP != fp || h.Composed != dig || h.Phase != delivery.PhaseRunning {
		t.Fatalf("said again after a look: %+v", h)
	}
}