The node-engine speaks its seat's generated contract: manifest-reads and manifest-reads-changed in mesh-contracts' types (ADR 0618 §3, §4, ADR 0613 §3)
The verb's answer and the event's body are mesh-contracts' nodeengine types, not mesh-sdk's, and the seat's verb and event names are the contract's. The two reasons of the event, started and returned, are the engine's own words beside it. Requires mesh-contracts and the registry milestone's mesh-sdk (a go.work, uncommitted, until the group pins them). Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
+18
-16
@@ -13,6 +13,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/asker"
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
@@ -33,33 +34,33 @@ import (
|
||||
// time and at most one started in an hour. Every act, with who asked as the runtime proved it, is kept in the engine's own
|
||||
// declared state acts (ActsBucket, created by the grants part), under its own node's keys alone, read by the verb acts.
|
||||
|
||||
// The acts of the seat, as its verbs name them.
|
||||
// The acts of the seat, as its verbs name them: the seat's contract's names (mesh-contracts, seats/node-engine).
|
||||
const (
|
||||
ActHandOver = "hand-over"
|
||||
ActAccess = "access-act"
|
||||
ActJudgeDirectories = "judge-directories"
|
||||
ActSetuidSearch = "setuid-search"
|
||||
ActHandOver = nodeengine.VerbHandOver
|
||||
ActAccess = nodeengine.VerbAccessAct
|
||||
ActJudgeDirectories = nodeengine.VerbJudgeDirectories
|
||||
ActSetuidSearch = nodeengine.VerbSetuidSearch
|
||||
// ActComposition reads a module's composition beside what runs of it; read-only, no warrant (issue 774).
|
||||
ActComposition = "composition"
|
||||
ActComposition = nodeengine.VerbComposition
|
||||
// ActCompose composes the node now, the recovery of a failed composition (issue 774).
|
||||
ActCompose = "compose"
|
||||
ActCompose = nodeengine.VerbCompose
|
||||
// ActPackages reads the packages this node's compositions declare, and with write writes its key again (novox/hq
|
||||
// ADR 0565 §6, §15).
|
||||
ActPackages = "packages"
|
||||
ActPackages = nodeengine.VerbPackages
|
||||
// ActComposeCandidate composes a check's candidate on this node in shadow, read-only (novox/hq ADR 0593 §7).
|
||||
ActComposeCandidate = "compose-candidate"
|
||||
ActComposeCandidate = nodeengine.VerbComposeCandidate
|
||||
// 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"
|
||||
ActPullWaits = nodeengine.VerbPullWaits
|
||||
// ActManifestReads reads the reader capabilities this engine states (novox/hq ADR 0618 §3).
|
||||
ActManifestReads = "manifest-reads"
|
||||
ActManifestReads = nodeengine.VerbManifestReads
|
||||
)
|
||||
|
||||
// The states of an act.
|
||||
// The states of an act: asked, done and refused are also the seat's events.
|
||||
const (
|
||||
HandAsked = "asked"
|
||||
HandDone = "done"
|
||||
HandRefused = "refused"
|
||||
HandAsked = nodeengine.EventAsked
|
||||
HandDone = nodeengine.EventDone
|
||||
HandRefused = nodeengine.EventRefused
|
||||
HandAnswered = "answered"
|
||||
)
|
||||
|
||||
@@ -251,7 +252,8 @@ func (h *HandActs) Call(c EngineCall) (any, error) {
|
||||
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
|
||||
return nodeengine.ManifestReadsAnswer{V: delivery.Version, Node: h.Node, ManifestReads: reads,
|
||||
StatedAt: at.UTC().Format(time.RFC3339Nano)}, nil
|
||||
case "acts":
|
||||
if id := arg("id"); id != "" {
|
||||
return h.Act(id)
|
||||
|
||||
@@ -11,8 +11,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
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"
|
||||
)
|
||||
@@ -195,15 +195,18 @@ type handingLink struct {
|
||||
// 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
|
||||
said chan nodeengine.ManifestReadsChangedEvent
|
||||
}
|
||||
|
||||
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 {
|
||||
// As the bus carries it: the body written and read back as the seat's contract.
|
||||
raw, err := json.Marshal(body)
|
||||
var e nodeengine.ManifestReadsChangedEvent
|
||||
if err == nil && event == nodeengine.EventManifestReadsChanged && json.Unmarshal(raw, &e) == nil {
|
||||
l.said <- e
|
||||
}
|
||||
return nil
|
||||
return err
|
||||
}
|
||||
|
||||
func (l handingLink) HearHandOvers(h HandOverFunc) { l.given <- h }
|
||||
@@ -230,8 +233,8 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
|
||||
<-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} {
|
||||
said := make(chan nodeengine.ManifestReadsChangedEvent, 4)
|
||||
for _, why := range []string{ReadsStarted, ReadsReturned} {
|
||||
el := readsLink{quietLink: newQuietLink(), said: said}
|
||||
eq := q
|
||||
eq.Acts = &HandActs{Node: "n1", State: &MemState{}}
|
||||
@@ -240,7 +243,7 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
|
||||
go func() { edone <- serve(ectx, el, m, eq, nil, time.Second) }()
|
||||
select {
|
||||
case got := <-said:
|
||||
if got.Why != why || got.Node != "n1" {
|
||||
if at, err := time.Parse(time.RFC3339Nano, got.At); got.Why != why || got.Node != "n1" || got.V != 1 || err != nil || at.IsZero() {
|
||||
t.Errorf("the link said %+v, want %s", got, why)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
@@ -298,7 +301,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"`},
|
||||
{"manifest-reads", EngineToolSubject("manifest-reads", "n1"), `"manifest-reads":["form:a","seat:b"],"node":"n1","stated-at":"2026-11-04T10:00:00Z","v":1`},
|
||||
{"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, " ")
|
||||
|
||||
+13
-11
@@ -5,12 +5,13 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
@@ -394,25 +395,26 @@ func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) {
|
||||
// nothing of when an unchanged statement was made.
|
||||
func (q *Session) StatedReads() ([]string, time.Time) { return manifestReads(), readsStatedAt }
|
||||
|
||||
// The reasons the engine says its reads, in the event's why (novox/hq ADR 0618 §4): it started, or its node returned
|
||||
// after being away (a link after the first).
|
||||
const (
|
||||
ReadsStarted = "started"
|
||||
ReadsReturned = "returned"
|
||||
)
|
||||
|
||||
// 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
|
||||
why := ReadsStarted
|
||||
if q.reads.linked {
|
||||
why = delivery.ReadsReturned
|
||||
why = 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))
|
||||
q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: delivery.Version,
|
||||
Node: q.Membership.Node, ManifestReads: manifestReads(), At: time.Now().UTC().Format(time.RFC3339Nano), Why: why})
|
||||
}
|
||||
|
||||
// SayEngine says one of the node seat's events over the link open now; with none open it is not said.
|
||||
|
||||
Reference in New Issue
Block a user