novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The seat's events wait for the first view unacknowledged; the checks prove the order and the dropped case

Browse Source 
The process hands an event of the seat mesh-broker to BindingsView.Hold, which
applies and acknowledges it after the answer or past the bound, so a restart
finds it delivered again (one delivered again while held keeps its place by its
stream sequence). Unit rows: the answer's bucket differs from the held created's
and the held event's stands. Whole test: an answer that lists a binding dropped
since its ask does not write it back (fails without the check). Neutral caller
name in the by-hand asks.

Builds: design 25 bindings-first-view §3, §4, §8 (hq d0760629, PR 543)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 11:01:39 +01:00
parent fcfaa41348
commit 20af25ea40
5 changed files with 147 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/cmd/mesh-grants/integration_test.go
+39 -2
View File
@@ -220,12 +220,23 @@ func TestTheGrantsPartWhole(t *testing.T) {
// Design 25 bindings-first-view §8: the bus module answers the seat mesh-broker's list with an entry whose created was never
// said, once it is answering; until then the ask is refused as of a holder not yet connected.
var listing atomic.Bool
// A gate on the bus module's answer: while set, the answer waits (after saying it was asked) until the test releases it.
type listGate struct{ asked, release chan struct{} }
var gate atomic.Pointer[listGate]
var answeredGated, listAsked, listAnswered atomic.Int32
for _, subject := range []string{"mesh.seat.mesh-broker.call.list.>", "mesh.seat.mesh-broker.tool.list"} {
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
if !listing.Load() {
_ = m.Respond([]byte(`{"error":"not connected yet","node":"a"}`))
return
}
listAsked.Add(1)
defer listAnswered.Add(1)
if g := gate.Load(); g != nil {
g.asked <- struct{}{}
<-g.release
defer answeredGated.Add(1)
}
answer(m, map[string]any{"entries": []any{
map[string]any{"module": "mod-c", "local-name": "late", "storage-type": "key-value", "template": "state", "bus-name": "mod-c_late",
"settings": map[string]any{"mod-c_late": map[string]any{}},
@@ -350,7 +361,7 @@ func TestTheGrantsPartWhole(t *testing.T) {
return m.Bindings
}
sayBroker := func(event string, bind membership.BusBinding) {
body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-c", Local: "kept", Storage: "key-value", Template: "state", Binding: bind})
body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-c", Local: bind.Local, Storage: "key-value", Template: "state", Binding: bind})
if _, err := js.Publish(ctx, "mesh.seat.mesh-broker.event."+event, body); err != nil {
t.Fatal(err)
}
@@ -379,7 +390,8 @@ func TestTheGrantsPartWhole(t *testing.T) {
said[grants.BindingUnlisted+".mod-c.late"] == "" && said[grants.BindingUnlisted+".mod-c.waiting"] == ""
})
// A person asks again by hand (state bindings): asked once more, the same answer changes nothing it holds.
askedAgain, _ := json.Marshal(grants.Ask{By: "jochen@shanks", At: time.Now(), FirstView: grants.FirstViewBindings})
askedBefore := listAsked.Load()
askedAgain, _ := json.Marshal(grants.Ask{By: "someone@desk", At: time.Now(), FirstView: grants.FirstViewBindings})
if _, err := kvs[grants.Module+"."+grants.StateUserList].Put(ctx, grants.KeyAsked, askedAgain); err != nil {
t.Fatal(err)
}
@@ -391,6 +403,31 @@ func TestTheGrantsPartWhole(t *testing.T) {
b := bindingOf()
return len(b) == 1 && b[0].Bucket == "mod-c_late" && conditionsSaid()[grants.BindingUnlisted+".mod-c.kept"] == "cleared"
})
until("the ask by hand answered", func() bool { return listAsked.Load() > askedBefore && listAnswered.Load() == listAsked.Load() })
// An answer that comes after a dropped heard since the ask does not write the dropped binding back (§4): the bus module
// holds the answer, which still lists the entry, while the dropped is said and heard.
g := &listGate{asked: make(chan struct{}, 1), release: make(chan struct{})}
gate.Store(g)
askedAgain, _ = json.Marshal(grants.Ask{By: "someone@desk", At: time.Now().Add(time.Second), FirstView: grants.FirstViewBindings})
if _, err := kvs[grants.Module+"."+grants.StateUserList].Put(ctx, grants.KeyAsked, askedAgain); err != nil {
t.Fatal(err)
}
select {
case <-g.asked:
case <-time.After(20 * time.Second):
t.Fatalf("the bus module was not asked again by hand\n%s", logged.tail("by hand")+logged.tail("bindings"))
}
sayBroker("dropped", membership.BusBinding{Local: "late"})
until("mod-c's late binding gone after dropped, the answer held", func() bool { return len(bindingOf()) == 0 })
close(g.release)
gate.Store(nil)
until("the held answer given", func() bool { return answeredGated.Load() == 1 })
for range 20 { // the part takes the answer within milliseconds; it must not write the dropped binding back
if b := bindingOf(); len(b) != 0 {
t.Fatalf("after an answer that lists a binding dropped since its ask: %+v", b)
}
time.Sleep(50 * time.Millisecond)
}
for _, bucket := range []string{"mesh-controller_lease", grants.Module + "_lease", grants.Module + "_takeover"} {
if _, err := js.KeyValue(ctx, bucket); !errors.Is(err, jetstream.ErrBucketNotFound) {
t.Errorf("the bucket %s: %v, want none", bucket, err)
modules/mesh-grants/cmd/mesh-grants/main.go
+15 -1
View File
@@ -195,7 +195,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
// The other seats' events the part follows its passes by (Consumes).
for _, subject := range grants.SeatEventSubjects() {
if err := hub.Handle(subject, func(m jetstream.Msg) { handle(m.Subject(), m.Data()); _ = m.Ack() }); err != nil {
if err := hub.Handle(subject, func(m jetstream.Msg) {
// The seat mesh-broker's events wait for the first view's answer, unacknowledged (design 25
// bindings-first-view §3): a restart finds them delivered again.
if strings.HasPrefix(m.Subject(), grants.BrokerEventPrefix) && f.Part.Bindings != nil {
var seq uint64
if md, err := m.Metadata(); err == nil {
seq = md.Sequence.Stream
}
if f.Part.Bindings.Hold(seq, m.Subject(), m.Data(), func() { _ = m.Ack() }) {
return
}
}
handle(m.Subject(), m.Data())
_ = m.Ack()
}); err != nil {
return err
}
}
modules/mesh-grants/grants_test.go
+70 -50
View File
@@ -3218,57 +3218,77 @@ func TestGrants(t *testing.T) {
// The start: the events heard before the answer wait for it and come after it, whichever was said later; with the bus
// module not answering the part has composed every account, raises past its bound the condition naming the seat and
// the verb, takes the events then, and the answer clears it.
t.Run("at a start the answer comes first; unanswered, the part raises past its bound and takes the events", func(t *testing.T) {
st := newSetup()
now := time.Date(2026, 11, 11, 3, 0, 0, 0, time.UTC)
var failures atomic.Int32
failures.Store(5)
changed := make(chan struct{}, 64)
st.v.Changed = func() { changed <- struct{}{} }
st.v.Now = func() time.Time { st.mu.Lock(); defer st.mu.Unlock(); return now }
st.v.Pause = func(_ context.Context, d time.Duration) bool {
st.mu.Lock()
now = now.Add(d)
st.mu.Unlock()
return true
}
st.v.Ask = func(context.Context) (json.RawMessage, error) {
if failures.Add(-1) >= 0 {
return nil, errors.New("nats: no responders available for request")
for _, row := range []struct {
name string
failures int32
answer string // the answer's bucket for the entry; the held created names mod-x_event
}{
{"at a start the answer comes first, and the event held after it stands over it", 0, "mod-x_answer"},
{"unanswered, the part raises past its bound naming the seat and the verb, and takes the events", 5, "mod-x_event"},
} {
t.Run(row.name, func(t *testing.T) {
st := newSetup()
now := time.Date(2026, 11, 11, 3, 0, 0, 0, time.UTC)
var failures atomic.Int32
failures.Store(row.failures)
changed := make(chan struct{}, 64)
st.v.Changed = func() { changed <- struct{}{} }
st.v.Now = func() time.Time { st.mu.Lock(); defer st.mu.Unlock(); return now }
st.v.Pause = func(_ context.Context, d time.Duration) bool {
st.mu.Lock()
now = now.Add(d)
st.mu.Unlock()
return true
}
return json.Marshal(map[string]any{"entries": []any{entry("late", "mod-x_event")}})
}
hold := make(chan struct{})
ask := st.v.Ask
st.v.Ask = func(ctx context.Context) (json.RawMessage, error) { <-hold; return ask(ctx) }
st.v.Start(context.Background(), st.p)
hear(st, "created", "late", "mod-x_event") // heard before the answer: held
if got := kept(st); len(got) != 0 {
t.Fatalf("an event was taken before the answer: %v", got)
}
close(hold)
deadline := time.After(10 * time.Second)
for !st.v.Status().Answered.After(time.Time{}) {
select {
case <-changed:
case <-deadline:
t.Fatalf("never answered: %+v", st.v.Status())
st.v.Ask = func(context.Context) (json.RawMessage, error) {
if failures.Add(-1) >= 0 {
return nil, errors.New("nats: no responders available for request")
}
return json.Marshal(map[string]any{"entries": []any{entry("late", row.answer)}})
}
}
if got := kept(st); !maps.Equal(got, map[string]string{"late": "mod-x_event"}) {
t.Errorf("kept %v after the held created", got)
}
if st.said[grants.FirstViewUnanswered+".mesh-broker.bindings"] != "cleared" {
t.Errorf("the unanswered condition after the answer: %v", st.said)
}
c, ok := st.raised[grants.FirstViewUnanswered+".mesh-broker.bindings"]
if !ok {
t.Fatalf("never raised past its bound: %v", st.said)
}
if !strings.Contains(c.Summary, "mesh-broker") || !strings.Contains(c.Summary, "list") {
t.Errorf("the condition does not name the seat and the verb: %s", c.Summary)
}
})
hold := make(chan struct{})
ask := st.v.Ask
st.v.Ask = func(ctx context.Context) (json.RawMessage, error) { <-hold; return ask(ctx) }
st.v.Start(context.Background(), st.p)
// Heard before the answer: held, not applied and not acknowledged.
var acked atomic.Int32
ev, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: "late", Storage: "key-value", Template: "state",
Binding: membership.BusBinding{Local: "late", Storage: "key-value", Template: "state", Bucket: "mod-x_event"}})
if !st.v.Hold(7, "mesh.seat.mesh-broker.event.created", ev, func() { acked.Add(1) }) {
t.Fatal("the event was not held at a start")
}
if got := kept(st); len(got) != 0 || acked.Load() != 0 {
t.Fatalf("an event was taken before the answer: %v, acknowledged %d", got, acked.Load())
}
close(hold)
deadline := time.After(10 * time.Second)
for acked.Load() == 0 {
select {
case <-changed:
case <-deadline:
t.Fatalf("never answered: %+v", st.v.Status())
}
}
// The held event is taken after the answer, so its binding stands, not the answer's.
if got := kept(st); !maps.Equal(got, map[string]string{"late": "mod-x_event"}) {
t.Errorf("kept %v after the held created", got)
}
if row.failures == 0 {
return
}
if st.said[grants.FirstViewUnanswered+".mesh-broker.bindings"] != "cleared" {
t.Errorf("the unanswered condition after the answer: %v", st.said)
}
c, ok := st.raised[grants.FirstViewUnanswered+".mesh-broker.bindings"]
if !ok {
t.Fatalf("never raised past its bound: %v", st.said)
}
if !strings.Contains(c.Summary, "mesh-broker") || !strings.Contains(c.Summary, "list") {
t.Errorf("the condition does not name the seat and the verb: %s", c.Summary)
}
})
}
// The tools: the first view is shown beside the others and asked again by a person's act, recorded as a hand act.
t.Run("the tools show and ask the bindings", func(t *testing.T) {
@@ -3280,7 +3300,7 @@ func TestGrants(t *testing.T) {
st.p.Compose()
var acts []string
record := func(what, why string) error { acts = append(acts, what+": "+why); return nil }
if _, err := st.p.AskFirstViewAgain("jochen@shanks", "bindings", "a created was missed", record); err != nil || len(acts) != 1 || !strings.Contains(acts[0], "bindings") {
if _, err := st.p.AskFirstViewAgain("someone@desk", "bindings", "a created was missed", record); err != nil || len(acts) != 1 || !strings.Contains(acts[0], "bindings") {
t.Fatalf("bindings again: %v %v", err, acts)
}
changed := make(chan struct{}, 64)
modules/mesh-grants/internal/grants/bindings.go
+4 -7
View File
@@ -80,17 +80,14 @@ func brokerEvent(event string, brokers map[string]bool) (string, bool) {
// `dropped` removes it. It answers whether the kept bindings changed (so a pass composes again), or why the event was
// not applied; an event that is no bus module's is no change and no error. brokers are the modules holding mesh-broker.
//
// At a start the answer of the first view comes first (design 25 bindings-first-view §3): an event of the seat heard before
// the answer is applied, or before its bound passes, is held and taken after it, in the order heard.
// At a start the answer of the first view comes first (design 25 bindings-first-view §3): the process hands an event of the
// seat to BindingsView.Hold, which takes it after the answer is applied, or past the bound, and acknowledges it then.
func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (bool, error) {
if p.Bindings != nil && strings.HasPrefix(event, brokerEventPrefix) && p.Bindings.hold(event, body) {
return false, nil
}
return p.hearBroker(event, body, brokers)
}
// brokerEventPrefix starts the subject of every event of the seat mesh-broker.
const brokerEventPrefix = "mesh.seat." + meshbroker.Seat + ".event."
// BrokerEventPrefix starts the subject of every event of the seat mesh-broker.
const BrokerEventPrefix = "mesh.seat." + meshbroker.Seat + ".event."
// hearBroker applies one event, the first view's hold already passed.
func (p *Part) hearBroker(event string, body []byte, brokers map[string]bool) (bool, error) {
modules/mesh-grants/internal/grants/bindingsview.go
+19 -5
View File
@@ -81,8 +81,10 @@ type BindingsView struct {
}
type heldEvent struct {
seq uint64
event string
body []byte
ack func()
}
func (v *BindingsView) now() time.Time {
@@ -135,14 +137,22 @@ func (v *BindingsView) statusLocked() BindingsStatus {
func localKey(module, local string) string { return module + "/" + local }
// hold keeps an event of the seat heard before the first view is applied (§3), and says whether it did.
func (v *BindingsView) hold(event string, body []byte) bool {
// Hold keeps an event of the seat heard before the first view is applied (§3), and says whether it did. A held event is not
// acknowledged: ack is called only when the event is applied after the answer (or past the bound), so a restart finds it
// delivered again. seq is the event's sequence on its stream; one delivered again while held replaces its ack and keeps its place.
func (v *BindingsView) Hold(seq uint64, event string, body []byte, ack func()) bool {
v.mu.Lock()
defer v.mu.Unlock()
if !v.holding {
return false
}
v.held = append(v.held, heldEvent{event, slices.Clone(body)})
for i := range v.held {
if seq != 0 && v.held[i].seq == seq {
v.held[i].ack = ack
return true
}
}
v.held = append(v.held, heldEvent{seq, event, slices.Clone(body), ack})
return true
}
@@ -286,10 +296,14 @@ func (v *BindingsView) release(p *Part) {
changed := false
for _, e := range held {
if c, err := p.hearBroker(e.event, e.body, p.Brokers()); err != nil {
v.logf("%v: not applied", err)
v.logf("%v: not applied, left unacknowledged to be delivered again", err)
continue
} else if c {
changed = true
}
if e.ack != nil {
e.ack()
}
}
if changed {
v.wake()
@@ -377,7 +391,7 @@ func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewrit
}
was, had := b.Bindings[e.LocalName]
if had && len(bind.Types) == 0 {
bind.Types = was.Types // the bus module's events leave the types to the manifest's verdict (§2)
bind.Types = was.Types // an answer that names none leaves the types an event brought
}
if had && sameBinding(was, bind) {
continue