novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

The node-engine speaks operator-channel, mesh-vault and node-engine through their contracts and clients; it serves no pull-waits

Browse Source 
ADR 0627 §1, §4: mesh-sdk keeps no asks, asker, vault answers or compose-candidate checks, so the node-engine takes them
from mesh-contracts: operator-channel's contract and client (asks, warrants, the asker's client), mesh-vault's sign verb
and answer, and node-engine's own contract and client for compose-candidate's ask, answer, verdicts and bound. The
lead's ruling on pull-waits (no record gives the seat that verb): the verb and its handler go; the row that read the
waits now holds the seat refusing the verb, and failed before the change. The go.mod requires of mesh-contracts' modules
come with the group's pin.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 14:02:44 +01:00
parent d8207358b0
commit 2b14f166de
14 changed files with 129 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/authority_test.go
+3 -2
View File
@@ -13,6 +13,7 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/meshvault"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
@@ -108,13 +109,13 @@ func TestTheAppliedAuthorityIsKeptForTheRecord(t *testing.T) {
answer := signedBy(vaultCA, vault)
ask := func(_ context.Context, verb string, args any) (json.RawMessage, error) {
asked = append(asked, verb, args)
return json.Marshal(delivery.SignAnswer{Kind: delivery.KindNodeCertificate, Node: "n1", Certificate: answer})
return json.Marshal(meshvault.SignAnswer{Kind: delivery.KindNodeCertificate, Node: "n1", Certificate: answer})
}
written := 0
if err := certifyThisNode(context.Background(), state, "n1", trust, ask, func() { written++ }); err != nil {
t.Fatal(err)
}
if got := recordKeys(state, nil)(); got.Certificate != answer || written != 1 || len(asked) != 2 || asked[0] != delivery.SignVerb {
if got := recordKeys(state, nil)(); got.Certificate != answer || written != 1 || len(asked) != 2 || asked[0] != meshvault.VerbSign {
t.Errorf("the vault's certificate: kept %v, written %d, asked %v", got.Certificate == answer, written, asked)
}
answer = signedBy(testAuthority(t, controllerPub, controller), controller)
cmd/mesh-host/engineseat.go
+13 -24
View File
@@ -13,10 +13,10 @@ import (
"syscall"
"time"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
stdio "git.novox.be/novox/mesh-sdk/go"
meshasker "git.novox.be/novox/mesh-sdk/go/asker"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/accounts"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/link"
@@ -256,8 +256,8 @@ func engineActs(opts options, node string, queue *link.Session, search *accounts
return judgeRootFree(node, lastAccounts.Load(), time.Now(), served, why)
},
ManifestReads: queue.StatedReads, Say: queue.SayEngine}
asker := asks.EngineAsker(node)
h.Asker = &meshasker.Client{Module: asker, State: state, Publish: queue.PublishAsk,
asker := occ.EngineAsker(node)
h.Asker = &occ.Client{Module: asker, State: state, Publish: queue.PublishAsk,
Record: func(id string) (*stdio.StateEntry, error) { return queue.RouterRecord(asker + "." + id) }}
return h
}
@@ -284,17 +284,6 @@ const compositionReadFor = 20 * time.Second
// compositionFrom answers the verb composition from the node agent that runs now (issue 774); refused in words while
// none does.
// pullWaitsFrom answers the verb pull-waits from the agent running now (ENG-9).
func pullWaitsFrom(current *agentNow) func() (any, error) {
return func() (any, error) {
ag := current.get()
if ag == nil {
return nil, errors.New("this engine's node agent has not started, so it waits on no store yet")
}
return map[string]any{"waits": ag.PullWaits()}, nil
}
}
func compositionFrom(current *agentNow) func(module, digest string, diff bool) (any, error) {
return func(module, digest string, diff bool) (any, error) {
ag := current.get()
@@ -375,26 +364,26 @@ func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[s
// composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this
// node in shadow by the run's shadow, applying and writing nothing, within the shadow's own bound (shorter than
// delivery.ComposeAnswerBound, selfcompose candidateWithin). The answer is
// the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet
// nec.AnswerBound, selfcompose candidateWithin). The answer is
// the seat's shape (nodeengine.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet
// answers unready, which the checker takes as no answer.
func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) {
return func(candidate string, input bool) (any, error) {
ask := delivery.ComposeCandidateAsk{Candidate: candidate, Input: input}
if _, err := delivery.EncodeComposeCandidateAsk(ask); err != nil {
ask := nodeengine.ComposeCandidateInput{Candidate: candidate, Input: input}
if _, err := nec.EncodeAsk(ask); err != nil {
return nil, err
}
var a delivery.ComposeCandidateAnswer
var a nodeengine.ComposeCandidateAnswer
if s := shadow.Load(); s == nil {
// Unready, never an error the checker reads words of: its agent is starting (ADR 0593 §9).
a = delivery.ComposeCandidateAnswer{Node: node, Candidate: candidate, Verdict: delivery.ComposeCandidateUnready,
a = nodeengine.ComposeCandidateAnswer{Node: node, Candidate: candidate, Verdict: nec.Unready,
Refused: "this node's shadow does not run now (its node agent is starting), so it composes no candidate"}
} else {
ctx, cancel := context.WithTimeout(context.Background(), delivery.ComposeAnswerBound)
ctx, cancel := context.WithTimeout(context.Background(), nec.AnswerBound)
defer cancel()
a = s.ComposeCandidate(ctx, ask)
}
raw, err := delivery.EncodeComposeCandidateAnswer(a)
raw, err := nec.EncodeAnswer(a)
if err != nil {
return nil, fmt.Errorf("this node's answer to %s does not fit the seat's shape: %w", candidate, err)
}
cmd/mesh-host/libraries_test.go
+6 -6
View File
@@ -8,8 +8,8 @@ import (
"testing"
"time"
meshasker "git.novox.be/novox/mesh-sdk/go/asker"
"git.novox.be/novox/mesh-sdk/go/asks"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
"git.novox.be/novox/mesh-sdk/go/declaration"
"github.com/novox/mesh-host/internal/apply"
"github.com/novox/mesh-host/internal/link"
@@ -58,16 +58,16 @@ func TestALibrarySaysItsWaitWithoutWordsToParse(t *testing.T) {
// 774 (ADR 0563 §4): the library's home engine asks its take of the operator itself, once: no second ask while one is
// open, nor after it ended; and a library waiting for its disk is not asked.
t.Run("the home engine asks the take once", func(t *testing.T) {
var asked []asks.Ask
var asked []oc.Ask
state := &link.MemState{}
h := &link.HandActs{Node: "n1", State: state,
JudgeAccess: func(act, access string) (link.Judged, error) {
return link.Judged{Subject: access, Path: root, Sets: "the group mesh-media", Disk: "0:0 755"}, nil
},
AccessAct: func(link.AccessAct) (string, error) { return "taken", nil }}
h.Asker = &meshasker.Client{Module: asks.EngineAsker("n1"), State: state,
h.Asker = &occ.Client{Module: occ.EngineAsker("n1"), State: state,
Publish: func(subject string, body any, id string) error {
if a, ok := body.(asks.Ask); ok {
if a, ok := body.(oc.Ask); ok {
asked = append(asked, a)
}
return nil
@@ -78,7 +78,7 @@ func TestALibrarySaysItsWaitWithoutWordsToParse(t *testing.T) {
if len(waiting) != 1 || waiting[0] != "library:media" || len(asked) != 1 {
t.Fatalf("waiting %v asked %d times", waiting, len(asked))
}
h.Heard(asks.Warrant{Ask: asked[0].ID, Asker: asks.EngineAsker("n1"), Outcome: asks.OutcomeExpired, At: time.Now()})
h.Heard(oc.Warrant{Ask: asked[0].ID, Asker: occ.EngineAsker("n1"), Outcome: occ.OutcomeExpired, At: occ.FormatTime(time.Now())})
askTakes(h, waiting, nil)
if len(asked) != 1 {
t.Fatalf("the take was asked again after its ask ended: %d", len(asked))
cmd/mesh-host/main.go
-1
View File
@@ -1015,7 +1015,6 @@ func runLink(ctx context.Context, opts options) error {
go reconcileActs(ctx, queue.Acts)
current := &agentNow{}
queue.Acts.Composition = compositionFrom(current)
queue.Acts.PullWaits = pullWaitsFrom(current)
queue.Acts.Compose = composeFrom(current, &composerNow)
queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow)
queue.Acts.ComposeCandidate = composeCandidateFrom(mine.Node, &shadowNow)
cmd/mesh-host/nodeagent_integration_test.go
+4 -3
View File
@@ -14,6 +14,7 @@ import (
"testing"
"time"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/declaration"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -598,12 +599,12 @@ func h8candidate(t *testing.T, b *itBed, fp, running string) {
t.Fatal(err)
}
raw, _ := got.(json.RawMessage)
a, err := delivery.DecodeComposeCandidateAnswer(raw)
a, err := nec.DecodeAnswer(raw)
if err != nil {
t.Fatalf("the answer %s: %v", raw, err)
}
if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "web" ||
a.Modules[0].Fp != fpIt("c") || !a.Modules[0].Changed || a.Took > delivery.ComposeAnswerBound.Milliseconds() {
if a.Verdict != nec.Composed || len(a.Modules) != 1 || a.Modules[0].Module != "web" ||
a.Modules[0].Fp != fpIt("c") || !a.Modules[0].Changed || a.Took > nec.AnswerBound.Milliseconds() {
t.Errorf("the node's answer to a candidate changing web: %+v", a)
}
after, _ := b.composed.Status()
cmd/mesh-host/trust.go
+3 -2
View File
@@ -13,6 +13,7 @@ import (
"sync"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/meshvault"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
@@ -259,11 +260,11 @@ func certifyThisNode(ctx context.Context, state, node string, trust *meshTrust,
if err != nil || len(key) != ed25519.PublicKeySize {
return fmt.Errorf("this node's serving key is not an ed25519 key")
}
raw, err := ask(ctx, delivery.SignVerb, map[string]any{"kind": delivery.KindNodeCertificate, "node": node})
raw, err := ask(ctx, meshvault.VerbSign, map[string]any{"kind": delivery.KindNodeCertificate, "node": node})
if err != nil {
return err
}
var a delivery.SignAnswer
var a meshvault.SignAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return fmt.Errorf("the vault's answer is not a signing: %w", err)
}
internal/composer/composer.go
+1 -1
View File
@@ -306,7 +306,7 @@ const Package = "github.com/novox/mesh-host/cmd/mesh-compose"
// Decode reads a request (sdk.Decode); it refuses one that is not JSON or of another version.
func Decode(data []byte) (Request, error) { return sdk.Decode(data) }
// SentInput is the composition input a node-engine answers for the sandbox (delivery.ComposeCandidateAsk.Input), from the
// SentInput is the composition input a node-engine answers for the sandbox (nodeengine.ComposeCandidateInput.Input), from the
// input it composed node from: the one shape both read (issue 1126, checks of mesh-host #200). Every other node's modules
// are left to the cases (the indexer's manifests, as the bus reader reads them too); node's own placed modules stay as the
// bus reader took them in from their builds (their artifacts, the texts of their templates archive, what is handed to the
internal/link/access_act_test.go
+17 -17
View File
@@ -7,9 +7,9 @@ import (
"testing"
"time"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asker"
"git.novox.be/novox/mesh-sdk/go/asks"
)
// The take act's ask (novox/hq ADR 0325 §8): a hand-over's fields and the act's own in one signed object, under a
@@ -78,7 +78,7 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) {
t.Run("the node seat's hand-over acts only on the operator's warrant", func(t *testing.T) {
for _, row := range []struct {
name string
warrant func(asked asks.Ask, w asks.Warrant) asks.Warrant
warrant func(asked oc.Ask, w oc.Warrant) oc.Warrant
diskAfter string
twice bool
done int
@@ -87,19 +87,19 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) {
{name: "no warrant", done: 0, state: HandAsked},
{name: "the operator's yes", warrant: yes, done: 1, state: HandDone},
{name: "the yes heard twice", warrant: yes, twice: true, done: 1, state: HandDone},
{name: "a warrant for another digest", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
{name: "a warrant for another digest", warrant: func(a oc.Ask, w oc.Warrant) oc.Warrant {
w = yes(a, w)
w.AskDigest = "sha256:another"
return w
}, done: 0, state: HandAsked},
{name: "a no", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
{name: "a no", warrant: func(a oc.Ask, w oc.Warrant) oc.Warrant {
w = yes(a, w)
w.Option, w.Label, w.Level = "no", "No, leave it", asks.Acknowledge
w.Option, w.Label, w.Level = "no", "No, leave it", occ.Acknowledge
return w
}, done: 0, state: HandRefused},
{name: "the disk changed", warrant: yes, diskAfter: "0:0 700 9:2", done: 0, state: HandRefused},
{name: "expired", warrant: func(a asks.Ask, w asks.Warrant) asks.Warrant {
return asks.Warrant{Ask: a.ID, Asker: w.Asker, Outcome: asks.OutcomeExpired, At: w.At}
{name: "expired", warrant: func(a oc.Ask, w oc.Warrant) oc.Warrant {
return oc.Warrant{Ask: a.ID, Asker: w.Asker, Outcome: occ.OutcomeExpired, At: w.At}
}, done: 0, state: HandRefused},
} {
t.Run(row.name, func(t *testing.T) {
@@ -130,7 +130,7 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) {
disk = row.diskAfter
}
if row.warrant != nil {
w := row.warrant((*asked)[0], asks.Warrant{Asker: asks.EngineAsker("n1"), At: time.Now()})
w := row.warrant((*asked)[0], oc.Warrant{Asker: occ.EngineAsker("n1"), At: occ.FormatTime(time.Now())})
h.Heard(w)
if row.twice {
h.Heard(w)
@@ -153,25 +153,25 @@ func TestTheTakeActRunsOnlyOnAWarrant(t *testing.T) {
}
// yes is the operator's yes to an ask, as the router writes it.
func yes(a asks.Ask, w asks.Warrant) asks.Warrant {
w.Ask, w.Outcome, w.Option, w.Label, w.Level = a.ID, asks.OutcomeChosen, "yes", "Yes", asks.Approve
w.By = &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}
w.AskDigest, w.Proofs = a.Digest(), []string{"P1"}
func yes(a oc.Ask, w oc.Warrant) oc.Warrant {
w.Ask, w.Outcome, w.Option, w.Label, w.Level = a.ID, occ.OutcomeChosen, "yes", "Yes", occ.Approve
w.By = &oc.Person{Who: occ.Operator, Kind: "telegram", Verified: "user id verified"}
w.AskDigest, w.Proofs = occ.Digest(a), []string{"P1"}
return w
}
// anEngine is a node seat with its own state in a temporary folder, a router that keeps what it was asked, and a
// hand-over that counts what it performed.
func anEngine(t *testing.T) (*HandActs, *[]asks.Ask, *int) {
func anEngine(t *testing.T) (*HandActs, *[]oc.Ask, *int) {
t.Helper()
var asked []asks.Ask
var asked []oc.Ask
done := 0
state := &MemState{}
h := &HandActs{Node: "n1", State: state,
HandOver: func(path, by string) (string, error) { done++; return path + " handed over by " + by, nil }}
h.Asker = &asker.Client{Module: asks.EngineAsker("n1"), State: state,
h.Asker = &occ.Client{Module: occ.EngineAsker("n1"), State: state,
Publish: func(subject string, body any, id string) error {
if a, ok := body.(asks.Ask); ok && subject == asks.AskSubject(asks.EngineAsker("n1")) {
if a, ok := body.(oc.Ask); ok && subject == oc.AcceptSubjectAsk(occ.EngineAsker("n1")) {
asked = append(asked, a)
}
return nil
internal/link/engineseat.go
+6 -6
View File
@@ -9,8 +9,8 @@ import (
"strings"
"time"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/micro"
@@ -192,7 +192,7 @@ func (l *natsLink) SayEngine(event string, body any) error {
return err
}
// The engine asks the operator under its own asker name (asks.EngineAsker) on the router's seat (ADR 0259 §3, as ADR
// The engine asks the operator under its own asker name (occ.EngineAsker) on the router's seat (ADR 0259 §3, as ADR
// 0563 §3 changes it): its ask and cancel on the seat's work stream, its warrants heard on its own decided subject, and
// its asks' state read from the router's records under its own name.
@@ -204,7 +204,7 @@ const RouterRecords = "messenger_asks"
type OperatorAsks interface {
PublishAsk(subject string, body any, id string) error
RouterRecord(key string) (*stdio.StateEntry, error)
HearWarrants(asker string, heard func(asks.Warrant)) error
HearWarrants(asker string, heard func(oc.Warrant)) error
}
// PublishAsk publishes an ask or a cancel on the router's seat, once per id within the stream's duplicate window.
@@ -236,9 +236,9 @@ func (l *natsLink) RouterRecord(key string) (*stdio.StateEntry, error) {
}
// HearWarrants hears the warrants for asker on its decided subject, each handed to heard as it arrives.
func (l *natsLink) HearWarrants(asker string, heard func(asks.Warrant)) error {
sub, err := l.conn.Subscribe(asks.DecidedSubject(asker), func(msg *nats.Msg) {
var w asks.Warrant
func (l *natsLink) HearWarrants(asker string, heard func(oc.Warrant)) error {
sub, err := l.conn.Subscribe(oc.SubjectDecided(asker), func(msg *nats.Msg) {
var w oc.Warrant
if json.Unmarshal(msg.Data, &w) == nil {
heard(w)
}
internal/link/handacts.go
+24 -34
View File
@@ -13,10 +13,10 @@ import (
"sync"
"time"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
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"
)
@@ -24,9 +24,9 @@ import (
//
// A hand-over and an access act are the engine's own asks, performed only on the warrant (§3): a call performs nothing.
// The engine judges every value as it does for the controller's signed ask, then refuses in words or asks the operator
// under its own asker name (asks.EngineAsker) at the level approve, binding the digest of the verb, the node, every
// under its own asker name (occ.EngineAsker) at the level approve, binding the digest of the verb, the node, every
// argument and the owner, group and mode it will set. A warrant is claimed once by compare-and-set in the engine's own
// state (asker.Client.Take), held to the ask it answers, and to the act bound and the disk as judged, judged again now;
// state (occ.Client.Take), held to the ask it answers, and to the act bound and the disk as judged, judged again now;
// only then is the act performed and recorded with the warrant. A decline, an expiry or a changed disk says refused and
// changes nothing. One ask is open per path or access.
//
@@ -49,9 +49,6 @@ const (
ActPackages = nodeengine.VerbPackages
// ActComposeCandidate composes a check's candidate on this node in shadow, read-only (novox/hq ADR 0593 §7).
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 = nodeengine.VerbPullWaits
// ActManifestReads reads the reader capabilities this engine states (novox/hq ADR 0618 §3).
ActManifestReads = nodeengine.VerbManifestReads
)
@@ -129,8 +126,8 @@ type HandActs struct {
Node string
// State is where the acts and the asks are kept: this node's keys of ActsBucket (BusState), each key one act.
State ActStore
// Asker asks the operator as asks.EngineAsker(Node), with State as its store.
Asker *asker.Client
// Asker asks the operator as occ.EngineAsker(Node), with State as its store.
Asker *occ.Client
// JudgeHandOver judges a hand-over's path, refusing in words; HandOver records it on the warrant.
JudgeHandOver func(path string) (Judged, error)
HandOver HandOverFunc
@@ -156,8 +153,6 @@ type HandActs struct {
// module, applying and writing nothing; with input, also the composition input it composed from (the verb
// compose-candidate, ADR 0593 §7).
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)
@@ -242,11 +237,6 @@ func (h *HandActs) Call(c EngineCall) (any, error) {
input = b
}
return h.ComposeCandidate(arg("candidate"), input)
case ActPullWaits:
if h.PullWaits == nil {
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")
@@ -539,8 +529,8 @@ func (h *HandActs) id(act string) string {
}
// bound is the act an ask's yes binds: the verb, the node, every argument and what it will set (ADR 0563 §3).
func (h *HandActs) bound(a HandAct) asks.Act {
act := asks.Act{"verb": a.Act, "node": h.Node, "arg.subject": a.Subject, "arg.why": a.Why, "sets": a.Sets}
func (h *HandActs) bound(a HandAct) occ.Act {
act := occ.Act{"verb": a.Act, "node": h.Node, "arg.subject": a.Subject, "arg.why": a.Why, "sets": a.Sets}
if a.Path != "" {
act["path"] = a.Path
}
@@ -601,7 +591,7 @@ func (h *HandActs) ask(act, caller, why string, auto bool, j Judged) (any, error
defer h.mu.Unlock()
now := h.now().UTC()
a := HandAct{ID: h.id(act), Act: act, Subject: j.Subject, Path: j.Path, Caller: caller, Why: why, At: now,
State: HandAsked, Sets: j.Sets, Disk: j.Disk, Expires: now.Add(asks.ApproveLasts - time.Minute), Auto: auto}
State: HandAsked, Sets: j.Sets, Disk: j.Disk, Expires: now.Add(occ.ApproveLasts - time.Minute), Auto: auto}
a.Ask = a.ID
// One ask open per path or access: its open key made by a create, which a second ask cannot make.
if _, err := h.State.Create(openKey(act, j.Subject), map[string]string{"ask": a.Ask}); errors.Is(err, stdio.ErrChanged) {
@@ -622,7 +612,7 @@ func (h *HandActs) ask(act, caller, why string, auto bool, j Judged) (any, error
_ = h.State.Delete(openKey(act, j.Subject))
return nil, err
}
bind, err := asks.ActDigest(h.bound(a))
bind, err := occ.ActDigest(h.bound(a))
if err != nil {
return nil, err
}
@@ -633,16 +623,16 @@ func (h *HandActs) ask(act, caller, why string, auto bool, j Judged) (any, error
"The library's group, group write and setgid are set on everything under its root.",
"The library waits and nothing under its root changes."
}
if len([]rune(headline)) > asks.HeadlineLength {
headline = string([]rune(headline)[:asks.HeadlineLength])
if len([]rune(headline)) > occ.HeadlineLength {
headline = string([]rune(headline)[:occ.HeadlineLength])
}
asked := asks.Ask{ID: a.Ask, Headline: headline, Who: asks.Operator,
asked := oc.Ask{ID: a.Ask, Headline: headline, Who: occ.Operator,
Explanation: j.Words + " Your yes sets " + j.Sets + "; nothing changes until you answer.",
Whole: j.Words + " (" + j.Path + ") Your yes sets " + j.Sets + "; nothing changes until you answer.",
Details: "Asked by " + caller + " on " + h.Node + orWhy(why),
Options: []asks.Option{{ID: "yes", Label: "Yes", Does: does, Level: asks.Approve, Binds: bind},
{ID: "no", Label: "No, leave it", Does: stays, Level: asks.Acknowledge}},
Expires: a.Expires, OnExpiry: stays, About: EngineSeat + "-" + h.Node + "-" + act}
Options: []oc.Option{{ID: "yes", Label: "Yes", Does: does, Level: occ.Approve, Binds: bind},
{ID: "no", Label: "No, leave it", Does: stays, Level: occ.Acknowledge}},
Expires: occ.FormatTime(a.Expires), OnExpiry: stays, About: EngineSeat + "-" + h.Node + "-" + act}
if err := h.Asker.Ask(asked); err != nil {
_, _, _ = h.end(a.Ask, func(x *HandAct) { x.State, x.Said = HandRefused, "the operator could not be asked: "+err.Error() })
return nil, fmt.Errorf("the operator could not be asked, and nothing changed: %w", err)
@@ -663,10 +653,10 @@ func orWhy(why string) string {
return ": " + why
}
// Warranted is the act on the operator's answer, handed once by asker.Client.Take after it held the warrant to the
// Warranted is the act on the operator's answer, handed once by occ.Client.Take after it held the warrant to the
// ask: a no is recorded refused; a yes is held to the act the option bound and the disk as judged, judged again now,
// and only then performed and recorded done with the warrant.
func (h *HandActs) Warranted(o asks.Option, w asks.Warrant) error {
func (h *HandActs) Warranted(o oc.Option, w oc.Warrant) error {
a, _, err := h.readOne(actKey(w.Ask))
if err != nil {
return err
@@ -698,7 +688,7 @@ func (h *HandActs) Warranted(o asks.Option, w asks.Warrant) error {
}
now := *a
now.Sets, now.Path = j.Sets, j.Path
if err := o.Performs(h.bound(now)); err != nil {
if err := occ.Performs(o, h.bound(now)); err != nil {
return refuse("what the yes bound is not what would be done now: " + err.Error())
}
if j.Disk != a.Disk {
@@ -723,7 +713,7 @@ func (h *HandActs) Warranted(o asks.Option, w asks.Warrant) error {
return err
}
func warrantOn(a *HandAct, w asks.Warrant) {
func warrantOn(a *HandAct, w oc.Warrant) {
a.Proofs = w.Proofs
if w.By != nil {
a.By, a.Via = w.By.Who, w.By.Kind
@@ -733,22 +723,22 @@ func warrantOn(a *HandAct, w asks.Warrant) {
}
}
func operatorOf(w asks.Warrant) string {
func operatorOf(w oc.Warrant) string {
if w.By == nil {
return "the operator"
}
return "the " + w.By.Who + " via " + w.By.Kind
}
// Heard is a warrant heard on the engine's decided subject: taken once (asker.Client.Take), and an ask that ended with
// Heard is a warrant heard on the engine's decided subject: taken once (occ.Client.Take), and an ask that ended with
// no choice recorded refused.
func (h *HandActs) Heard(w asks.Warrant) {
func (h *HandActs) Heard(w oc.Warrant) {
if h.Asker == nil {
return
}
// A warrant the ask kept here does not hold to (another digest, another option, too late) ends nothing: it is no
// answer to the ask, which stays open for the operator's.
if handled, _ := h.Asker.Take(w, h.Warranted); handled == asker.Recorded {
if handled, _ := h.Asker.Take(w, h.Warranted); handled == occ.Recorded {
why := "the ask ended " + string(w.Outcome)
if w.Words != "" {
why += ": " + w.Words
internal/link/handover_test.go
+2 -5
View File
@@ -265,10 +265,6 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
return map[string]any{"module": module, "digest": digest, "diff": diff}, nil
},
Compose: func(module string) (any, error) { return map[string]any{"composed": true, "retried": module}, nil },
// ENG-9: the failed applies that wait on a store's holder.
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)
@@ -295,7 +291,8 @@ func TestALinkThatHearsHandOversIsGivenTheQueues(t *testing.T) {
// The recovery verb compose (issue 774) makes a pass now, with a module's apply tried again when named.
{"compose", EngineToolSubject("compose", "n1"), `"composed":true,"retried":""`},
{"compose", EngineToolSubject("compose", "n1") + " " + `{"module":"route-proxy"}`, `"retried":"route-proxy"`},
{"pull-waits", EngineToolSubject("pull-waits", "n1"), `"holder":"distribution","last-retry":"distribution settled","module":"web"`},
// ENG-9's pull waits are carried in the failed have, read through node_haves: the seat serves no verb of them.
{"pull-waits", EngineToolSubject("pull-waits", "n1"), `"error":"the node seat node-engine serves no verb \"pull-waits\""`},
// The read verb packages (ADR 0565 §6) names its source; with write, what it wrote, and a key the engine has not
// written is said and not an error.
{"packages", EngineToolSubject("packages", "n1"), `"source":"mesh-host.packages","key":{"node":"n1"`},
internal/link/run.go
+2 -2
View File
@@ -7,7 +7,7 @@ import (
"fmt"
"time"
meshasks "git.novox.be/novox/mesh-sdk/go/asks"
occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient"
)
// Membership is what a node needs to reach its mesh again, held by the caller.
@@ -222,7 +222,7 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
queue.SayReads()
}
if a, ok := link.(OperatorAsks); ok {
if err := a.HearWarrants(meshasks.EngineAsker(m.Node), queue.Acts.Heard); err != nil {
if err := a.HearWarrants(occ.EngineAsker(m.Node), queue.Acts.Heard); err != nil {
say("cannot hear the operator's answers to this engine's asks: " + err.Error())
}
go queue.Acts.Reconcile()
internal/selfcompose/candidate.go
+28 -26
View File
@@ -12,6 +12,8 @@ import (
"sync"
"time"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
@@ -33,11 +35,11 @@ import (
var errNotReady = errors.New("this node's shadow has not read every input yet")
// candidateWithin is the node's own bound on an answer to compose-candidate: shorter than the checker's
// (delivery.ComposeAnswerBound), because the bus lets the engine answer an ask only within a minute of it (the grants
// (nec.AnswerBound), because the bus lets the engine answer an ask only within a minute of it (the grants
// part's allow_responses ttl, mesh-grants ResponseTTL): an answer at the checker's 60 s bound was refused by the bus
// (issue 1126, 2026-11-03 01:17:04 on shanks, "Permissions Violation for Publish to _INBOX..."), so the checker heard
// nothing at all. A node not done by then answers unready, naming what it waits on.
var candidateWithin = delivery.ComposeAnswerBound - 10*time.Second
var candidateWithin = nec.AnswerBound - 10*time.Second
// maxSentInput is the most composition input an answer carries: the bus carries 1 MiB in one message (its max_payload),
// and the answer's own fields and envelope come beside it.
@@ -92,7 +94,7 @@ func (r *candidateRun) said(now time.Time) (waits string, on string) {
// ComposeCandidate is the node's answer to ask, within candidateWithin (or ctx's deadline, if sooner). Each answer is
// said in one line, with what it waited on and for how long (issue 1126: ace, g14 and shanks answered nothing for a
// minute while nothing said why). An answer not composed within the bound is unready, naming what the node waits on.
func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandidateAsk) delivery.ComposeCandidateAnswer {
func (s *Shadow) ComposeCandidate(ctx context.Context, ask nodeengine.ComposeCandidateInput) nodeengine.ComposeCandidateAnswer {
s.mu.Lock()
c := s.c
s.mu.Unlock()
@@ -100,14 +102,14 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
ctx, cancel := context.WithTimeout(ctx, candidateWithin)
defer cancel()
run := &candidateRun{}
done := make(chan delivery.ComposeCandidateAnswer, 1)
done := make(chan nodeengine.ComposeCandidateAnswer, 1)
go func() { done <- s.composeCandidate(ctx, c, ask, began, run) }()
var a delivery.ComposeCandidateAnswer
var a nodeengine.ComposeCandidateAnswer
select {
case a = <-done:
case <-ctx.Done():
_, on := run.said(c.now())
a = delivery.ComposeCandidateAnswer{Node: c.env.Node, Candidate: ask.Candidate, Verdict: delivery.ComposeCandidateUnready,
a = nodeengine.ComposeCandidateAnswer{Node: c.env.Node, Candidate: ask.Candidate, Verdict: nec.Unready,
Refused: fmt.Sprintf("the node did not compose the candidate within %s: it waits on %s", candidateWithin, on),
Took: c.now().Sub(began).Milliseconds()}
}
@@ -122,19 +124,19 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi
}
// composeCandidate composes ask on scratch copies of c, each added to run as it starts.
func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery.ComposeCandidateAsk, began time.Time, run *candidateRun) delivery.ComposeCandidateAnswer {
func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask nodeengine.ComposeCandidateInput, began time.Time, run *candidateRun) nodeengine.ComposeCandidateAnswer {
node := c.env.Node
a := delivery.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate}
refuse := func(why string) delivery.ComposeCandidateAnswer {
a.Verdict, a.Refused, a.Modules, a.Own = delivery.ComposeCandidateRefused, why, nil, delivery.ModuleComposed{}
a := nodeengine.ComposeCandidateAnswer{Node: node, Candidate: ask.Candidate}
refuse := func(why string) nodeengine.ComposeCandidateAnswer {
a.Verdict, a.Refused, a.Modules, a.Own = nec.Refused, why, nil, nodeengine.Module{}
a.Took = c.now().Sub(began).Milliseconds()
return a
}
// A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers
// unready: the checker takes it as no answer (ADR 0593 §9) by the verdict, the words only saying why.
unready := func(why string) delivery.ComposeCandidateAnswer {
unready := func(why string) nodeengine.ComposeCandidateAnswer {
a = refuse(why)
a.Verdict = delivery.ComposeCandidateUnready
a.Verdict = nec.Unready
return a
}
group, heads, err := delivery.ParseCompositionKey(ask.Candidate)
@@ -223,8 +225,8 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery
return unready("a read on the bus did not answer: " + laidErr.Error())
case laidErr != nil && baseErr != nil:
// The node composes none of its own without the candidate either: the candidate changes nothing of that.
a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{}
a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState, Words: laidErr.Error()}
a.Verdict, a.Modules = nec.Composed, []nodeengine.Module{}
a.Own = nodeengine.Module{Module: delivery.HaveNodeWord, State: nec.ModuleHeld, Words: laidErr.Error()}
a.Took = c.now().Sub(began).Milliseconds()
return a
case laidErr != nil:
@@ -238,15 +240,15 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery
return out
}
was, now := byModule(baseComps), byModule(laidComps)
one := func(module string) (delivery.ModuleComposed, bool) {
m := delivery.ModuleComposed{Module: module}
one := func(module string) (nodeengine.Module, bool) {
m := nodeengine.Module{Module: module}
if module == "" {
m.Module = delivery.HaveNodeWord
}
if why, held := laidHeld[module]; held {
m.State, m.Words = delivery.ModuleRefusedState, why
m.State, m.Words = nec.ModuleRefused, why
if _, before := baseHeld[module]; before || baseErr != nil {
m.State = delivery.ModuleHeldState
m.State = nec.ModuleHeld
}
return m, true
}
@@ -256,10 +258,10 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery
}
dig, err := delivery.ComposedDigestOf(comp)
if err != nil {
m.State, m.Words = delivery.ModuleRefusedState, err.Error()
m.State, m.Words = nec.ModuleRefused, err.Error()
return m, true
}
m.State, m.Fp, m.Digest = delivery.ModuleComposedState, comp.FP, dig
m.State, m.Fp, m.Digest = nec.ModuleComposed, comp.FP, dig
if before, ok := was[module]; !ok {
m.Changed = true
} else if d, err := delivery.ComposedDigestOf(before); err != nil || d != dig {
@@ -281,24 +283,24 @@ func (s *Shadow) composeCandidate(ctx context.Context, c *Composer, ask delivery
}
}
sort.Strings(names)
a.Modules = []delivery.ModuleComposed{}
a.Modules = []nodeengine.Module{}
refused := false
for _, m := range names {
if mc, ok := one(m); ok {
a.Modules = append(a.Modules, mc)
refused = refused || mc.State == delivery.ModuleRefusedState
refused = refused || mc.State == nec.ModuleRefused
}
}
if own, ok := one(""); ok {
a.Own = own
} else {
a.Own = delivery.ModuleComposed{Module: delivery.HaveNodeWord, State: delivery.ModuleHeldState,
a.Own = nodeengine.Module{Module: delivery.HaveNodeWord, State: nec.ModuleHeld,
Words: "the node composes no composition of its own"}
}
refused = refused || a.Own.State == delivery.ModuleRefusedState
a.Verdict = delivery.ComposeCandidateComposed
refused = refused || a.Own.State == nec.ModuleRefused
a.Verdict = nec.Composed
if refused {
a.Verdict = delivery.ComposeCandidateRefused
a.Verdict = nec.Refused
}
a.Took = c.now().Sub(began).Milliseconds()
return a
internal/selfcompose/candidate_test.go
+20 -18
View File
@@ -8,6 +8,8 @@ import (
"testing"
"time"
nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
@@ -76,7 +78,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go func() { _ = s.Run(ctx) }()
if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); a.Verdict != delivery.ComposeCandidateUnready || a.Refused == "" {
if a := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: accepted}); a.Verdict != nec.Unready || a.Refused == "" {
t.Errorf("a shadow not ready answered %+v", a)
}
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"},
@@ -98,46 +100,46 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
time.Sleep(5 * time.Millisecond)
}
a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted})
if _, err := delivery.EncodeComposeCandidateAnswer(a); err != nil {
a := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: accepted})
if _, err := nec.EncodeAnswer(a); err != nil {
t.Fatalf("the answer does not fit the seat's shape: %v (%+v)", err, a)
}
if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" ||
a.Modules[0].State != delivery.ModuleComposedState || a.Modules[0].Fp != fpNew || !a.Modules[0].Changed ||
a.Own.Module != delivery.HaveNodeWord || a.Took > delivery.ComposeAnswerBound.Milliseconds() {
if a.Verdict != nec.Composed || len(a.Modules) != 1 || a.Modules[0].Module != "lemurs" ||
a.Modules[0].State != nec.ModuleComposed || a.Modules[0].Fp != fpNew || !a.Modules[0].Changed ||
a.Own.Module != delivery.HaveNodeWord || a.Took > nec.AnswerBound.Milliseconds() {
t.Errorf("an accepted candidate answered %+v", a)
}
// Issue 774: the input it sends the sandbox says how it composed the node (the builds), so the sandbox composes it
// through the same path.
sent := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted, Input: true})
sent := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: accepted, Input: true})
var in compose.Input
raw, _ := json.Marshal(sent.Input)
if err := json.Unmarshal(raw, &in); err != nil || in.ComposedHow == nil || in.ComposedHow.Node != "n1" ||
in.ComposedHow.Builds["lemurs"] != fpNew {
t.Errorf("the input sent for the sandbox does not say how the node composes: %+v, %v", in.ComposedHow, err)
}
r := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: refusedKey})
if r.Verdict != delivery.ComposeCandidateRefused || len(r.Modules) != 1 || r.Modules[0].State != delivery.ModuleRefusedState ||
r := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: refusedKey})
if r.Verdict != nec.Refused || len(r.Modules) != 1 || r.Modules[0].State != nec.ModuleRefused ||
!strings.Contains(r.Modules[0].Words, "a field the judge refuses") {
t.Errorf("a candidate whose manifest the indexer refused answered %+v", r)
}
if _, err := delivery.EncodeComposeCandidateAnswer(r); err != nil {
if _, err := nec.EncodeAnswer(r); err != nil {
t.Errorf("the refused answer does not fit the seat's shape: %v", err)
}
missing, _ := delivery.CompositionKey("novox/x#8", "0123456789ab")
if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: missing}); !strings.Contains(m.Refused, "holds no candidate") {
if m := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: missing}); !strings.Contains(m.Refused, "holds no candidate") {
t.Errorf("a candidate the indexer does not hold answered %+v", m)
}
// The group's candidate moved to other heads since the ask: the node composes not the one asked, and answers
// unready (no answer), naming the heads it holds.
moved, _ := delivery.CompositionKey("fix/940-no-retire-on-partial-read", "0123456789ab")
if m := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: moved}); m.Verdict != delivery.ComposeCandidateUnready ||
if m := s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: moved}); m.Verdict != nec.Unready ||
!strings.Contains(m.Refused, "heads") {
t.Errorf("a candidate at other heads answered %+v", m)
}
ended, end := context.WithCancel(ctx)
end()
if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateUnready {
if e := s.ComposeCandidate(ended, nodeengine.ComposeCandidateInput{Candidate: accepted}); e.Verdict != nec.Unready {
t.Errorf("an ended bound answered %+v", e)
}
if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {
@@ -215,11 +217,11 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
words string // in the answer's words
said string // in the line said of the last answer
}{
{name: "one ask reads each manifest once", asks: 1, verdict: delivery.ComposeCandidateComposed, reads: 1,
{name: "one ask reads each manifest once", asks: 1, verdict: nec.Composed, reads: 1,
said: "compose-candidate fix/1126."},
{name: "a later ask reads none again", asks: 2, verdict: delivery.ComposeCandidateComposed, reads: 1,
{name: "a later ask reads none again", asks: 2, verdict: nec.Composed, reads: 1,
said: "read(s) waited on"},
{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: delivery.ComposeCandidateUnready,
{name: "a read that does not answer is named at the bound", stuck: true, asks: 1, verdict: nec.Unready,
words: "waits on", said: "answered unready"},
} {
t.Run(tc.name, func(t *testing.T) {
@@ -241,9 +243,9 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) {
return get(rctx, bucket, k)
}
}
var a delivery.ComposeCandidateAnswer
var a nodeengine.ComposeCandidateAnswer
for range tc.asks {
a = s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: key})
a = s.ComposeCandidate(ctx, nodeengine.ComposeCandidateInput{Candidate: key})
}
if a.Verdict != tc.verdict || !strings.Contains(a.Refused, tc.words) {
t.Errorf("answered %s %q; want %s with %q", a.Verdict, a.Refused, tc.verdict, tc.words)