Node-engine: reads the seat list through the seat mesh-catalog, holds what is new until it was read, writes its finding after five minutes; verbs seat-list and seat-list-again; whole-module case against a stand-in of the seat (H-7, ADR 0636 §1, §7, §8; issue 1966)

cmd/mesh-host/main.go

@@ -1017,6 +1017,7 @@ func runLink(ctx context.Context, opts options) error {
 	queue.Acts.Composition = compositionFrom(current)
 	queue.Acts.Compose = composeFrom(current, &composerNow)
 	queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow)
+	queue.Acts.SeatList, queue.Acts.SeatListAgain = seatListRead, seatListAgain
 	queue.Acts.ComposeCandidate = composeCandidateFrom(mine.Node, &shadowNow)
 	setRunsHere(runsFromAgent(current))
 	factsOfNow := func(ctx context.Context, ag *agent.Agent) {

cmd/mesh-host/nodeagent.go

@@ -586,7 +586,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		// The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in
 		// flight at the switch marked the next agent's modules failed).
 		own := &agentNow{}
+		// And the seat list it reads through the seat mesh-catalog, from its bootstrapped definitions until it has: the
+		// composer hands nothing new meanwhile, and the node says it has waited five minutes (novox/hq ADR 0636 §1).
+		seats, err := startSeatList(run, session, node, say)
+		if err != nil {
+			say(err.Error())
+		}
 		hooks := composerHooks{
+			seatList: seats,
 			running: func() []delivery.Composed {
 				if now := own.get(); now != nil {
 					return runsAsWritten(now.Everything(), opts.state)
@@ -628,7 +635,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		// And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into
 		// its own key, `<node>.node`, under candidates.
 		shadowHooks := hooks
-		shadowHooks.refused, shadowHooks.kept, shadowHooks.certify = nil, nil, nil
+		shadowHooks.refused, shadowHooks.kept, shadowHooks.certify, shadowHooks.seatList = nil, nil, nil, nil
 		shadowEnv := composerEnv(session, node, say, shadowHooks)
 		shadow := selfcompose.NewShadow(shadowEnv,
 			func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
@@ -982,6 +989,8 @@ type composerHooks struct {
 	// vaultAuthority and certify are selfcompose.Env's VaultAuthority and Certify.
 	vaultAuthority func() string
 	certify        func(ctx context.Context) error
+	// seatList is this run's reader of the seat list: the composer hands nothing new until it was read (ADR 0636 §1).
+	seatList *seatListState
 }
 
 // selfComposer is the node's own composer: it composes what the seat's placement places on the node, and nothing while
@@ -1003,7 +1012,7 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f
 // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer
 // to placement candidates (selfcompose.Shadow) alike.
 func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env {
-	return selfcompose.Env{Node: node,
+	env := selfcompose.Env{Node: node,
 		Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) {
 			if bucket == delivery.PlacementBucket || bucket == delivery.PlacementCandidatesBucket {
 				// The seat's states are made when their holder starts: a node-engine that starts first waits for the
@@ -1033,4 +1042,8 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com
 		Fetch:   apply.Fetch,
 		Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held, Kept: hooks.kept,
 		VaultAuthority: hooks.vaultAuthority, Certify: hooks.certify}
+	if hooks.seatList != nil {
+		env.SeatList, env.SeatListChanged = hooks.seatList.reader, hooks.seatList.changed
+	}
+	return env
 }

cmd/mesh-host/nodeagent_integration_test.go

@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -18,12 +19,14 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/declaration"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/seatlist"
 	"github.com/nats-io/nats-server/v2/server"
 	"github.com/nats-io/nats.go"
 	"github.com/novox/mesh-host/internal/agent"
 	"github.com/novox/mesh-host/internal/apply"
 	"github.com/novox/mesh-host/internal/link"
 	"github.com/novox/mesh-host/internal/nodegrant"
+	"github.com/novox/mesh-host/internal/seatwait"
 	"github.com/novox/mesh-host/internal/selfcompose"
 	"github.com/novox/mesh-host/internal/store"
 	"github.com/novox/mesh-host/internal/system"
@@ -55,6 +58,9 @@ type itBed struct {
 	// refusals are the bus's asynchronous errors on the node's connection: each permissions violation it met.
 	refusals []string
 	mu       sync.Mutex
+	// seatSilent makes the stand-in of the seat mesh-catalog refuse the seat list (it is served from the start otherwise).
+	seatSilent atomic.Bool
+	findings   nats.KeyValue
 }
 
 // aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what
@@ -133,6 +139,10 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed {
 		}
 		*c.into = kv
 	}
+	if b.findings, err = b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: link.FindingsBucket, History: 1}); err != nil {
+		t.Fatal(err)
+	}
+	b.standInSeat()
 	b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second}
 	if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil {
 		t.Fatal(err)
@@ -875,3 +885,137 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test
 			runtime.containers[server] == made, runtime.moved(server))
 	}
 }
+
+// standInSeat is the seat mesh-catalog as the node reads the seat list from it (novox/hq ADR 0636 §4): with no seat, every
+// bootstrapped seat with a digest; with one, its definition. While seatSilent it refuses.
+func (b *itBed) standInSeat() {
+	defs := map[string]json.RawMessage{}
+	var entries []seatlist.Entry
+	for _, d := range seatlist.Bootstrapped() {
+		raw, err := json.Marshal(d)
+		if err != nil {
+			b.t.Fatal(err)
+		}
+		defs[d.Seat] = raw
+		entries = append(entries, seatlist.Entry{Seat: d.Seat, Source: seatlist.SourceTrunk, Digest: "d-" + d.Seat})
+	}
+	sub, err := b.conn.Subscribe("mesh.seat."+seatlist.Seat+".call."+seatlist.Verb+".*", func(m *nats.Msg) {
+		var q struct {
+			Seat string `json:"seat"`
+		}
+		_ = json.Unmarshal(m.Data, &q)
+		var result any
+		switch {
+		case b.seatSilent.Load():
+			_ = m.Respond([]byte(`{"error":"the indexer has not started"}`))
+			return
+		case q.Seat == "":
+			result = seatlist.Listing{Revision: 1, Seats: entries}
+		default:
+			result = seatlist.Answer{Seat: q.Seat, Found: true, Revision: 1,
+				Entry: &seatlist.Entry{Seat: q.Seat, Source: seatlist.SourceTrunk, Digest: "d-" + q.Seat, Definition: defs[q.Seat]}}
+		}
+		out, _ := json.Marshal(map[string]any{"result": result})
+		_ = m.Respond(out)
+	})
+	if err != nil {
+		b.t.Fatal(err)
+	}
+	b.t.Cleanup(func() { _ = sub.Unsubscribe() })
+}
+
+// fiveMinutes is a clock of the test: the bound of the node's wait fires when the test says so, nothing sleeps.
+type fiveMinutes struct {
+	mu     sync.Mutex
+	timers []*time.Timer
+	fire   []func()
+}
+
+func (c *fiveMinutes) Now() time.Time { return time.Now() }
+func (c *fiveMinutes) AfterFunc(_ time.Duration, f func()) seatwait.Timer {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	t := &stoppable{}
+	c.fire = append(c.fire, func() {
+		if !t.stopped.Load() {
+			f()
+		}
+	})
+	return t
+}
+
+type stoppable struct{ stopped atomic.Bool }
+
+func (s *stoppable) Stop() bool { return !s.stopped.Swap(true) }
+
+func (c *fiveMinutes) pass() {
+	c.mu.Lock()
+	fire := c.fire
+	c.fire = nil
+	c.mu.Unlock()
+	for _, f := range fire {
+		f()
+	}
+}
+
+// ADR 0636 §1, §8, whole against a real bus and the stand-in of the seat: while the seat refuses, the node composes and
+// applies nothing new, and keeps running what it ran; after the bound it writes one finding in its state findings; when
+// the seat names its list the module is applied and the finding is cleared; a node started again with the seat gone
+// applies the composition it applied last.
+func TestNatsANodeThatCannotReadTheSeatListActsOnNothingNew(t *testing.T) {
+	clock := &fiveMinutes{}
+	was := seatWaitClock
+	seatWaitClock = clock
+	t.Cleanup(func() { seatWaitClock = was })
+
+	b := newItBedAs(t, nodegrant.For(itNode, "it2.mailer.domain"))
+	b.seatSilent.Store(true)
+	stop, st := h8(t, b, false)
+	defer stop()
+	_ = st
+	finding := "machine." + itNode + ".seat-list-unread"
+	within(t, 5*time.Second, "the node did not start its bound while the seat refused", func() bool {
+		clock.mu.Lock()
+		defer clock.mu.Unlock()
+		return len(clock.fire) > 0
+	})
+	if b.file("web") != "" {
+		t.Fatalf("the node applied web before it read the seat list: %q", b.file("web"))
+	}
+	if _, err := b.findings.Get(finding); err == nil {
+		t.Fatal("the node wrote its finding before the bound")
+	}
+	clock.pass()
+	within(t, 5*time.Second, "the node did not write its finding after the bound", func() bool {
+		_, err := b.findings.Get(finding)
+		return err == nil
+	})
+	if b.file("web") != "" {
+		t.Fatalf("the node applied web with its finding written: %q", b.file("web"))
+	}
+
+	b.seatSilent.Store(false)
+	// The hand act of the operator and the healers (the verb seat-list-again): a read now, as the seat's event would make.
+	if out, err := seatListAgain(); err != nil || out.(map[string]any)["error"] != nil {
+		t.Fatalf("seat-list-again: %v, %v", out, err)
+	}
+	within(t, 50*time.Second, "the node did not apply web once the seat named its list", func() bool { return b.file("web") == "greeting=bonjour\n" })
+	within(t, 5*time.Second, "the node did not clear its finding at the first full read", func() bool {
+		_, err := b.findings.Get(finding)
+		return errors.Is(err, nats.ErrKeyNotFound)
+	})
+
+	// Started again with the seat refusing: nothing new is composed, and what ran is neither removed nor changed.
+	stop()
+	b.seatSilent.Store(true)
+	stopAgain := b.start()
+	defer stopAgain()
+	within(t, 30*time.Second, "the node did not hold web as it was after a start with the seat gone", func() bool {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		return strings.Contains(strings.Join(b.said, "\n"), "it1 has read no complete seat list") || b.file("web") != ""
+	})
+	if b.file("web") != "greeting=bonjour\n" {
+		t.Errorf("web does not stay as it ran after a start with the seat gone: %q", b.file("web"))
+	}
+}

cmd/mesh-host/seatlist.go

@@ -0,0 +1,166 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	conditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
+	condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+	"git.novox.be/novox/mesh-sdk/go/seatdef"
+	"git.novox.be/novox/mesh-sdk/go/seatlist"
+	"git.novox.be/novox/mesh-sdk/go/views"
+	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/seatwait"
+)
+
+// The node-engine reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636): through the seat mesh-catalog,
+// from the bootstrapped definitions until it has been read, acting on nothing new meanwhile (the composer's gate), and
+// following the seat's events. What it holds is given to catalogue at every change, which compiles no seat list.
+
+// seatListNow is this run's reader, what the node-engine's verbs seat-list and seat-list-again answer from.
+var seatListNow atomic.Pointer[seatListState]
+
+// seatListRead answers the verb seat-list: what the engine holds of the list and what it waits for (ADR 0636 §7).
+func seatListRead() (any, error) {
+	st := seatListNow.Load()
+	if st == nil {
+		return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet")
+	}
+	status := st.reader.Status()
+	return map[string]any{"status": status, "health": status.Health()}, nil
+}
+
+// seatListAgain answers the verb seat-list-again: the list read now from the seat, and what is held afterwards.
+func seatListAgain() (any, error) {
+	st := seatListNow.Load()
+	if st == nil {
+		return nil, errors.New("this engine's node agent has not started, so it holds no seat list yet")
+	}
+	err := st.reader.Reload()
+	st.waiter.Changed() // a read that changed nothing held still ends the wait
+	out, _ := seatListRead()
+	if err != nil {
+		out.(map[string]any)["error"] = err.Error()
+	}
+	return out, nil
+}
+
+// seatListState is one run's reader, the finding that it waits, and the channel on which the composer hears a change.
+type seatListState struct {
+	reader  *seatlist.Reader
+	waiter  *seatwait.Waiter
+	changed chan struct{}
+}
+
+// sessionBus is the reader's bus over the link: a verb of the seat is asked in the engine's own name, an event heard on
+// the seat's event subject for as long as ctx lives.
+type sessionBus struct {
+	ctx     context.Context
+	session *link.Session
+}
+
+func (b sessionBus) Ask(key string, body any) (json.RawMessage, error) {
+	seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
+	return b.session.AskMeshSeat(b.ctx, seat, verb, body)
+}
+
+func (b sessionBus) Subscribe(pattern string, handle func(string, json.RawMessage) error) error {
+	seat, event, _ := strings.Cut(pattern, ".")
+	heard, stop, err := b.session.Hear(seatdef.EventSubject(seat, event))
+	if err != nil {
+		return err
+	}
+	go func() {
+		defer stop()
+		for {
+			select {
+			case <-b.ctx.Done():
+				return
+			case body, ok := <-heard:
+				if !ok {
+					return
+				}
+				_ = handle(pattern, body)
+			}
+		}
+	}()
+	return nil
+}
+
+// seatWaitClock is the clock of the node's bound; the tests give their own.
+var seatWaitClock seatwait.Clock = realClock{}
+
+type realClock struct{}
+
+func (realClock) Now() time.Time { return time.Now() }
+func (realClock) AfterFunc(d time.Duration, f func()) seatwait.Timer {
+	return time.AfterFunc(d, f)
+}
+
+// findingsOfTheNode writes the node's findings in its state `findings`, in the layout of the seat mesh-conditions.
+type findingsOfTheNode struct{ session *link.Session }
+
+func (w findingsOfTheNode) encode(f views.Finding) (string, []byte, error) {
+	key := conditions.Key(conditions.ScopeMachine, f.ID, f.Kind)
+	v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning,
+		Subject: condseat.Subject{Scope: conditions.ScopeMachine, ID: f.ID, Machine: f.ID}, Summary: f.Summary,
+		Headline: f.Headline, Explanation: f.Explanation, Source: "seat-list-unread", At: f.At.UTC().Format(time.RFC3339)}
+	if err := conditions.CheckFinding(v); err != nil {
+		return "", nil, err
+	}
+	raw, err := json.Marshal(v)
+	return key, raw, err
+}
+
+func (w findingsOfTheNode) Put(f views.Finding) error {
+	key, raw, err := w.encode(f)
+	if err != nil {
+		return err
+	}
+	return w.session.PutFinding(key, raw)
+}
+
+func (w findingsOfTheNode) Delete(f views.Finding) error {
+	key, _, err := w.encode(f)
+	if err != nil {
+		return err
+	}
+	return w.session.DeleteFinding(key)
+}
+
+// startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what
+// it holds again, saying so to the composer and to the waiter that writes the five-minute finding.
+func startSeatList(ctx context.Context, session *link.Session, node string, say func(string)) (*seatListState, error) {
+	st := &seatListState{changed: make(chan struct{}, 1)}
+	var announce func()
+	reader, err := seatlist.Start(seatlist.Config{Bus: sessionBus{ctx, session}, Bootstrap: seatlist.Bootstrapped(),
+		OnChange: func(seats []catalogue.Seat) {
+			catalogue.UseSeats(seats)
+			if announce != nil { // the first read happens inside Start, before the waiter exists
+				announce()
+			}
+		}})
+	if err != nil {
+		return nil, fmt.Errorf("the node-engine cannot follow the seat list: %w", err)
+	}
+	st.reader = reader
+	st.waiter = &seatwait.Waiter{Node: node, Observer: "node-engine", Reader: reader, Findings: findingsOfTheNode{session},
+		Clock: seatWaitClock, Say: say}
+	announce = func() {
+		st.waiter.Changed()
+		select {
+		case st.changed <- struct{}{}:
+		default:
+		}
+	}
+	catalogue.UseSeats(reader.Seats())
+	announce()
+	seatListNow.Store(st)
+	return st, nil
+}