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

Merge pull request 'The node-engine says its events once through a transport fault, and an unsaid one at the next link (ADR 0668; issue 1966; group fix/1966-call-made-once)' (#222) from fix/1966-call-made-once into main

Browse Source
This commit was merged in pull request #222.
This commit is contained in:
mesh-tools
2026-11-10 07:20:32 +00:00
parent f50442ac7a d581acdeda
commit 783a2d2207
4 changed files with 98 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/compose/resolution/toolchain_manifests_test.go
+3 -2
View File
@@ -10,10 +10,11 @@ import (
// builderNode is a machine that can run a build module: the package manager and user namespaces it asks of
// the machine, and the forge account (mesh-scope, provided by the forge) and the artifact store the build
// modules publish to (ADR 0513, provided by distribution) stood in for, and node-tools, the tool runner they pull their
// asks through (mesh-catalog issue 1911; mesh-tools' module, not in the catalogue).
// asks through (mesh-catalog issue 1911; mesh-tools' module, not in the catalogue), with the tool runner's
// seat-settle-delay the build modules ask of it since issue 1748 (ADR 0664 §5).
func builderNode() mcat.Node {
return mcat.Node{Name: "builder", Site: "house",
Capabilities: map[string]bool{"package-manager": true, "user-namespaces": true}}
Capabilities: map[string]bool{"package-manager": true, "user-namespaces": true, "seat-settle-delay": true}}
}
func toolchainShelf(t *testing.T) map[string]mcat.Manifest {
internal/link/engineseat.go
+22 -4
View File
@@ -2,6 +2,8 @@ package link
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -12,6 +14,7 @@ import (
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatlist"
"github.com/nats-io/nats.go"
@@ -198,15 +201,30 @@ func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string {
"tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)}
}
// SayEngine says one of the seat's events about this node, on the events stream.
// SayEngine says one of the seat's events about this node, on the events stream, said again through a transport fault
// (sayOnce).
func (l *natsLink) SayEngine(event string, body any) error {
raw, err := json.Marshal(body)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, err = l.js.Publish(EngineEventSubject(event, l.node), raw, nats.Context(ctx))
subject := EngineEventSubject(event, l.node)
return sayOnce(func(id string) error {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_, err := l.js.Publish(subject, raw, nats.Context(ctx), nats.MsgId(id))
return err
}, subject, raw)
}
// sayOnce publishes one event under a message id made of its subject and its body, and publishes it again through a
// transport fault (no responders, a deadline) within the bus's retry bound: the stream takes the id once, so a publish
// that was stored and whose acknowledgement was lost is not stored twice (novox/hq ADR 0668 §1; issue 1966 note 155, a
// manifest-reads-changed that one publish lost was never said, and the assigner kept a bootstrapped statement).
func sayOnce(publish func(id string) error, subject string, raw []byte) error {
sum := sha256.Sum256(raw)
id := subject + "." + hex.EncodeToString(sum[:12])
_, err := busretry.DoRepeatable("saying "+subject, func() (struct{}, error) { return struct{}{}, publish(id) }, nil)
return err
}
internal/link/have_test.go
+46
View File
@@ -15,6 +15,7 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"github.com/nats-io/nats.go"
)
@@ -40,9 +41,20 @@ type haveLink struct {
statedErr error
// readsSaid are the node seat's manifest-reads-changed events said on the link (ADR 0618 §4).
readsSaid []nodeengine.ManifestReadsChangedEvent
// sayFails fails that many of the node seat's says, as a publish past the bus's retry bound; said is every event said.
sayFails int
said []string
}
func (l *haveLink) SayEngine(event string, body any) error {
l.hmu.Lock()
if l.sayFails > 0 {
l.sayFails--
l.hmu.Unlock()
return errors.New("nats: timeout")
}
l.said = append(l.said, event)
l.hmu.Unlock()
if e, ok := body.(nodeengine.ManifestReadsChangedEvent); ok && event == nodeengine.EventManifestReadsChanged {
l.hmu.Lock()
l.readsSaid = append(l.readsSaid, e)
@@ -372,6 +384,40 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); !slices.Equal(got.Record.ManifestReads, catalogue.ManifestReads()) {
t.Errorf("the list back whole: the record states %v, want its contract's", got.Record.ManifestReads)
}
// Issue 1966 note 155, ADR 0668 §4: a seat list read while no link is open is said at the link's opening with what
// the engine states then; and an event whose publish failed past the bus's retry bound is kept and said there too,
// never only a line in the log.
q.Detach(l)
saidBefore = len(l.readsSaid)
catalogue.UseSeats(slices.DeleteFunc(slices.Clone(all), func(s catalogue.Seat) bool { return s.Name == "mesh-vault" }))
q.RestateReads(context.Background())
if len(l.readsSaid) != saidBefore {
t.Errorf("a statement made with no link open was said on none")
}
q.Attach(context.Background(), l)
l.sayFails = 1
q.SayEngine("running", map[string]string{"module": "m"})
q.SayReads()
if said := l.readsSaid[saidBefore:]; len(said) != 1 || slices.Contains(said[0].ManifestReads, "seat:mesh-vault") ||
!slices.Contains(said[0].ManifestReads, "seat:mesh-catalog") {
t.Errorf("at the link's opening said %+v, want the statement made while no link was open", said)
}
if !slices.Contains(l.said, "running") {
t.Errorf("an event whose publish failed was not said at the next link: said %v", l.said)
}
catalogue.UseSeats(all)
q.RestateReads(context.Background())
// And the publish itself (sayOnce): through no responders, as the old JetStream client says it, and a deadline it is published again under one message id,
// so the stream takes it once.
t.Cleanup(busretry.Waits(10*time.Second, time.Millisecond, 2*time.Millisecond))
var ids []string
faults := []error{nats.ErrNoStreamResponse, context.DeadlineExceeded, nil}
if err := sayOnce(func(id string) error { ids = append(ids, id); return faults[len(ids)-1] }, "s", []byte("{}")); err != nil ||
len(ids) != 3 || ids[0] != ids[2] || ids[0] == "" {
t.Errorf("a publish through two faults: %v, ids %q", err, ids)
}
}
// 1132/AC7: a write the bucket refuses is said in the log, and the next apply writes again.
internal/link/session.go
+27 -4
View File
@@ -6,6 +6,8 @@ import (
"errors"
"fmt"
sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
"maps"
"slices"
"strings"
"sync"
"time"
@@ -92,6 +94,9 @@ type Session struct {
reads struct {
sync.Mutex
linked bool
// unsaid is, per event, the last of the node seat's events that could not be said (no link open, or its
// publish failed past the bus's retry bound): said at the next link's SayReads (novox/hq ADR 0668; issue 1966).
unsaid map[string]any
}
// restating takes RestateReads one at a time.
restating sync.Mutex
@@ -446,22 +451,40 @@ func (q *Session) SayReads() {
why = ReadsReturned
}
q.reads.linked = true
unsaid := q.reads.unsaid
q.reads.unsaid = nil
q.reads.Unlock()
for _, event := range slices.Sorted(maps.Keys(unsaid)) {
if event != nodeengine.EventManifestReadsChanged { // said below with what the engine states now
q.SayEngine(event, unsaid[event])
}
}
q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: sdkkeys.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.
// SayEngine says one of the node seat's events over the link open now. With none open, or a publish that failed past the
// bus's retry bound, it is kept and said at the next link's opening (SayReads; novox/hq ADR 0668 §4, issue 1966): an
// event a reader waits on is never only a line in this engine's log.
func (q *Session) SayEngine(event string, body any) {
q.init()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
if sayer, ok := bus.(interface{ SayEngine(string, any) error }); ok {
if err := sayer.SayEngine(event, body); err != nil {
q.Say("could not say the node seat's " + event + ": " + err.Error())
sayer, ok := bus.(interface{ SayEngine(string, any) error })
var err error
if ok {
if err = sayer.SayEngine(event, body); err == nil {
return
}
q.Say("could not say the node seat's " + event + ", said again at the next link: " + err.Error())
}
q.reads.Lock()
if q.reads.unsaid == nil {
q.reads.unsaid = map[string]any{}
}
q.reads.unsaid[event] = body
q.reads.Unlock()
}
func (q *Session) operatorAsks() (OperatorAsks, error) {