From 20af25ea400e3f466b493dffcb964ce9b4538ba4 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:01:39 +0100 Subject: [PATCH] The seat's events wait for the first view unacknowledged; the checks prove the order and the dropped case MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-grants/integration_test.go | 41 +++++- modules/mesh-grants/cmd/mesh-grants/main.go | 16 ++- modules/mesh-grants/grants_test.go | 120 ++++++++++-------- .../mesh-grants/internal/grants/bindings.go | 11 +- .../internal/grants/bindingsview.go | 24 +++- 5 files changed, 147 insertions(+), 65 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 687c1610..f9725a7b 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -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) diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 288b6b6f..67114a96 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -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 } } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 3a6a4947..ac194076 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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) diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index 357cf4c5..e9256c4f 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -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) { diff --git a/modules/mesh-grants/internal/grants/bindingsview.go b/modules/mesh-grants/internal/grants/bindingsview.go index 65a7f7ef..dd138323 100644 --- a/modules/mesh-grants/internal/grants/bindingsview.go +++ b/modules/mesh-grants/internal/grants/bindingsview.go @@ -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