From 271de0108da1feeb3c76ecb82ff798e4a1ca2dcd Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 12:00:21 +0100 Subject: [PATCH] =?UTF-8?q?The=20node-engine=20serves=20manifest-reads=20a?= =?UTF-8?q?nd=20says=20manifest-reads-changed=20once=20per=20link=20(ADR?= =?UTF-8?q?=200618=20=C2=A73,=20=C2=A74,=20issue=20774)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- cmd/mesh-host/engineseat.go | 2 +- internal/link/handacts.go | 12 ++++++++++ internal/link/handover_test.go | 44 ++++++++++++++++++++++++++++++++++ internal/link/record.go | 4 ++++ internal/link/run.go | 2 ++ internal/link/session.go | 32 +++++++++++++++++++++++++ 6 files changed, 95 insertions(+), 1 deletion(-) diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index c6c79ff..307290d 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -255,7 +255,7 @@ func engineActs(opts options, node string, queue *link.Session, search *accounts }) return judgeRootFree(node, lastAccounts.Load(), time.Now(), served, why) }, - Say: queue.SayEngine} + ManifestReads: queue.StatedReads, Say: queue.SayEngine} asker := asks.EngineAsker(node) h.Asker = &meshasker.Client{Module: asker, State: state, Publish: queue.PublishAsk, Record: func(id string) (*stdio.StateEntry, error) { return queue.RouterRecord(asker + "." + id) }} diff --git a/internal/link/handacts.go b/internal/link/handacts.go index 90a1cda..2e70cd2 100644 --- a/internal/link/handacts.go +++ b/internal/link/handacts.go @@ -16,6 +16,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asker" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // The node seat node-engine's acts (novox/hq ADR 0563). @@ -50,6 +51,8 @@ const ( // ActPullWaits reads the failed applies that wait on a store's holder, and the last retry with its trigger (design 51, // issue 1966 ENG-9). ActPullWaits = "pull-waits" + // ActManifestReads reads the reader capabilities this engine states (novox/hq ADR 0618 §3). + ActManifestReads = "manifest-reads" ) // The states of an act. @@ -154,6 +157,9 @@ type HandActs struct { ComposeCandidate func(candidate string, input bool) (any, error) // PullWaits answers which failed applies wait on which store's holder, and the last retry of each with what triggered it. PullWaits func() (any, error) + // ManifestReads is what this engine states it reads of a manifest and when its record last carried them (the verb + // manifest-reads, novox/hq ADR 0618 §3). + ManifestReads func() ([]string, time.Time) // Say says one of the seat's events, asked, done or refused; nil says nothing. Say func(event string, body any) Now func() time.Time @@ -240,6 +246,12 @@ func (h *HandActs) Call(c EngineCall) (any, error) { return nil, errors.New("this engine waits on no store yet: its node agent has not started") } return h.PullWaits() + case ActManifestReads: + if h.ManifestReads == nil { + return nil, errors.New("this engine states no manifest reads yet: its node agent has not started") + } + reads, at := h.ManifestReads() + return delivery.ManifestReadsAnswer{V: delivery.Version, Node: h.Node, ManifestReads: reads, StatedAt: at}, nil case "acts": if id := arg("id"); id != "" { return h.Act(id) diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index 15e9442..c884172 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -12,6 +12,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/micro" ) @@ -191,6 +192,20 @@ type handingLink struct { given chan HandOverFunc } +// readsLink is a quiet link that serves the node seat and keeps the manifest-reads-changed events said on it. +type readsLink struct { + *quietLink + said chan delivery.ManifestReadsChanged +} + +func (l readsLink) ServeEngineSeat(EngineFunc) {} +func (l readsLink) SayEngine(event string, body any) error { + if e, err := delivery.DecodeManifestReadsChanged(body.(json.RawMessage)); event == "manifest-reads-changed" && err == nil { + l.said <- e + } + return nil +} + func (l handingLink) HearHandOvers(h HandOverFunc) { l.given <- h } // **The link is given the queue's hand-over** when it is served: what the controller asks reaches the engine's @@ -213,6 +228,30 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { } stop() <-served + // ADR 0618 §4: an engine that holds the seat says manifest-reads-changed once for each link it serves, "started" on + // the first and "returned" on every one after it, and never once per pass. + said := make(chan delivery.ManifestReadsChanged, 4) + for _, why := range []string{delivery.ReadsStarted, delivery.ReadsReturned} { + el := readsLink{quietLink: newQuietLink(), said: said} + eq := q + eq.Acts = &HandActs{Node: "n1", State: &MemState{}} + ectx, estop := context.WithCancel(context.Background()) + edone := make(chan error, 1) + go func() { edone <- serve(ectx, el, m, eq, nil, time.Second) }() + select { + case got := <-said: + if got.Why != why || got.Node != "n1" { + t.Errorf("the link said %+v, want %s", got, why) + } + case <-time.After(5 * time.Second): + t.Fatalf("the link %s said nothing", why) + } + estop() + <-edone + } + if len(said) != 0 { + t.Errorf("more than once per link: %d more said", len(said)) + } // 774 (ADR 0563 §1): a call of the node seat reaches its verb with its caller as the subject proves it, unproven on // the subject that names none; a verb the seat does not serve is refused, and the answer is a tool's envelope. t.Run("a call of the node seat reaches its verb with its proven caller", func(t *testing.T) { @@ -227,6 +266,10 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { PullWaits: func() (any, error) { return map[string]any{"waits": []map[string]any{{"module": "web", "holder": "distribution", "last-retry": "distribution settled"}}}, nil }, + // ADR 0618 §3: what this engine states it reads, and when it last stated it. + ManifestReads: func() ([]string, time.Time) { + return []string{"form:a", "seat:b"}, time.Date(2026, 11, 4, 10, 0, 0, 0, time.UTC) + }, // ADR 0565 §6, §15: the node's key of the engine's state packages. Packages: func(write bool) (any, error) { key := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false}]}`) @@ -255,6 +298,7 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) { {"packages", EngineToolSubject("packages", "n1"), `"source":"mesh-host.packages","key":{"node":"n1"`}, {"packages", EngineToolSubject("packages", "n1") + " " + `{"write":true}`, `"key":null,"wrote":true`}, {"packages", EngineToolSubject("packages", "n1") + " " + `{"write":"true"}`, `"note":"the engine has written no key for this node yet`}, + {"manifest-reads", EngineToolSubject("manifest-reads", "n1"), `"v":1,"node":"n1","manifest-reads":["form:a","seat:b"],"stated-at":"2026-11-04T10:00:00Z"`}, {"unserved", "mesh.seat.node-engine.call.unserved.n1.x", `"error":"the node seat node-engine serves no verb \"unserved\""`}, } { subject, body, ok := strings.Cut(row.subject, " ") diff --git a/internal/link/record.go b/internal/link/record.go index 740184f..fc19e2a 100644 --- a/internal/link/record.go +++ b/internal/link/record.go @@ -5,6 +5,7 @@ import ( "slices" "sort" "sync" + "time" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -37,6 +38,9 @@ type RecordKeys struct { // once: it is the build's, and never changes while it runs. var manifestReads = sync.OnceValue(catalogue.ManifestReads) +// readsStatedAt is when this engine computed what it reads: its start (the verb manifest-reads, ADR 0618 §3). +var readsStatedAt = time.Now().UTC() + // recordOf is the node's record from one facts report and its keys; nil for a report that carries no profile (one that // says nothing of the machine, such as a rekey), so the record last written stands. func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord { diff --git a/internal/link/run.go b/internal/link/run.go index 2e8b08d..545fc7b 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -218,6 +218,8 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann if queue.Acts != nil { if s, ok := link.(EngineSeats); ok { s.ServeEngineSeat(queue.Acts.Call) + // And what it reads of a manifest, said once for this link (novox/hq ADR 0618 §4). + queue.SayReads() } if a, ok := link.(OperatorAsks); ok { if err := a.HearWarrants(meshasks.EngineAsker(m.Node), queue.Acts.Heard); err != nil { diff --git a/internal/link/session.go b/internal/link/session.go index 7dd16af..4bb37f6 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -87,6 +87,12 @@ type Session struct { // saying is held while a facts event is noted, so the haves are written in the order the facts are made. saying sync.Mutex + // reads says whether the engine has said its manifest reads on a link before (novox/hq ADR 0618 §4). + reads struct { + sync.Mutex + linked bool + } + // haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132). haves haveState // healthWritten is this node's health report as last written or read (WriteNodeHealth): the health loop alone @@ -383,6 +389,32 @@ func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) { return asking.RouterRecord(key) } +// StatedReads is what this engine states it reads of a manifest and when it computed them, which is its start (the +// node seat's verb manifest-reads, novox/hq ADR 0618 §3): the record is written only when it differs, so its put says +// nothing of when an unchanged statement was made. +func (q *Session) StatedReads() ([]string, time.Time) { return manifestReads(), readsStatedAt } + +// SayReads says the node seat's event manifest-reads-changed once for the link just opened (novox/hq ADR 0618 §4): +// why "started" on the engine's first link and "returned" on every link after it, so a reader asks the verb again once +// per connection and never once per pass. +func (q *Session) SayReads() { + q.reads.Lock() + why := delivery.ReadsStarted + if q.reads.linked { + why = delivery.ReadsReturned + } + q.reads.linked = true + q.reads.Unlock() + raw, err := delivery.EncodeManifestReadsChanged(delivery.ManifestReadsChanged{Node: q.Membership.Node, + ManifestReads: manifestReads(), At: time.Now().UTC(), Why: why}) + if err != nil { + q.init() + q.Say("cannot say the node seat's manifest-reads-changed: " + err.Error()) + return + } + q.SayEngine("manifest-reads-changed", json.RawMessage(raw)) +} + // SayEngine says one of the node seat's events over the link open now; with none open it is not said. func (q *Session) SayEngine(event string, body any) { q.init()