H2: an engine event not said is kept and said at the next link's opening (issue 1966, ADR 0668 §4)
A failed say was only a line in the engine's log; now it is said again when a link opens, beside the statement made while no link was open (proved by a step of the node key test, with H1's publish). Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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 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.ErrNoResponders, 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.
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user