The node-engine serves manifest-reads and says manifest-reads-changed once per link (ADR 0618 §3, §4, issue 774)
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
@@ -255,7 +255,7 @@ func engineActs(opts options, node string, queue *link.Session, search *accounts
|
|||||||
})
|
})
|
||||||
return judgeRootFree(node, lastAccounts.Load(), time.Now(), served, why)
|
return judgeRootFree(node, lastAccounts.Load(), time.Now(), served, why)
|
||||||
},
|
},
|
||||||
Say: queue.SayEngine}
|
ManifestReads: queue.StatedReads, Say: queue.SayEngine}
|
||||||
asker := asks.EngineAsker(node)
|
asker := asks.EngineAsker(node)
|
||||||
h.Asker = &meshasker.Client{Module: asker, State: state, Publish: queue.PublishAsk,
|
h.Asker = &meshasker.Client{Module: asker, State: state, Publish: queue.PublishAsk,
|
||||||
Record: func(id string) (*stdio.StateEntry, error) { return queue.RouterRecord(asker + "." + id) }}
|
Record: func(id string) (*stdio.StateEntry, error) { return queue.RouterRecord(asker + "." + id) }}
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
"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).
|
// 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,
|
// 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).
|
// issue 1966 ENG-9).
|
||||||
ActPullWaits = "pull-waits"
|
ActPullWaits = "pull-waits"
|
||||||
|
// ActManifestReads reads the reader capabilities this engine states (novox/hq ADR 0618 §3).
|
||||||
|
ActManifestReads = "manifest-reads"
|
||||||
)
|
)
|
||||||
|
|
||||||
// The states of an act.
|
// The states of an act.
|
||||||
@@ -154,6 +157,9 @@ type HandActs struct {
|
|||||||
ComposeCandidate func(candidate string, input bool) (any, error)
|
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 answers which failed applies wait on which store's holder, and the last retry of each with what triggered it.
|
||||||
PullWaits func() (any, error)
|
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 says one of the seat's events, asked, done or refused; nil says nothing.
|
||||||
Say func(event string, body any)
|
Say func(event string, body any)
|
||||||
Now func() time.Time
|
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 nil, errors.New("this engine waits on no store yet: its node agent has not started")
|
||||||
}
|
}
|
||||||
return h.PullWaits()
|
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":
|
case "acts":
|
||||||
if id := arg("id"); id != "" {
|
if id := arg("id"); id != "" {
|
||||||
return h.Act(id)
|
return h.Act(id)
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
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"
|
||||||
"github.com/nats-io/nats.go/micro"
|
"github.com/nats-io/nats.go/micro"
|
||||||
)
|
)
|
||||||
@@ -191,6 +192,20 @@ type handingLink struct {
|
|||||||
given chan HandOverFunc
|
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 }
|
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
|
// **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()
|
stop()
|
||||||
<-served
|
<-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
|
// 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.
|
// 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) {
|
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) {
|
PullWaits: func() (any, error) {
|
||||||
return map[string]any{"waits": []map[string]any{{"module": "web", "holder": "distribution", "last-retry": "distribution settled"}}}, nil
|
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.
|
// ADR 0565 §6, §15: the node's key of the engine's state packages.
|
||||||
Packages: func(write bool) (any, error) {
|
Packages: func(write bool) (any, error) {
|
||||||
key := []byte(`{"node":"n1","packages":[{"package":"curl","modules":["backup","web"],"absent":false}]}`)
|
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"), `"source":"mesh-host.packages","key":{"node":"n1"`},
|
||||||
{"packages", EngineToolSubject("packages", "n1") + " " + `{"write":true}`, `"key":null,"wrote":true`},
|
{"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`},
|
{"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\""`},
|
{"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, " ")
|
subject, body, ok := strings.Cut(row.subject, " ")
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"slices"
|
"slices"
|
||||||
"sort"
|
"sort"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"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.
|
// once: it is the build's, and never changes while it runs.
|
||||||
var manifestReads = sync.OnceValue(catalogue.ManifestReads)
|
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
|
// 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.
|
// 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 {
|
func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
|
||||||
|
|||||||
@@ -218,6 +218,8 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
|
|||||||
if queue.Acts != nil {
|
if queue.Acts != nil {
|
||||||
if s, ok := link.(EngineSeats); ok {
|
if s, ok := link.(EngineSeats); ok {
|
||||||
s.ServeEngineSeat(queue.Acts.Call)
|
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 a, ok := link.(OperatorAsks); ok {
|
||||||
if err := a.HearWarrants(meshasks.EngineAsker(m.Node), queue.Acts.Heard); err != nil {
|
if err := a.HearWarrants(meshasks.EngineAsker(m.Node), queue.Acts.Heard); err != nil {
|
||||||
|
|||||||
@@ -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 is held while a facts event is noted, so the haves are written in the order the facts are made.
|
||||||
saying sync.Mutex
|
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 is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132).
|
||||||
haves haveState
|
haves haveState
|
||||||
// healthWritten is this node's health report as last written or read (WriteNodeHealth): the health loop alone
|
// 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)
|
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.
|
// 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) {
|
func (q *Session) SayEngine(event string, body any) {
|
||||||
q.init()
|
q.init()
|
||||||
|
|||||||
Reference in New Issue
Block a user