From c2e7249897cb05fbfb5f0e07b32f36f50c7aaea3 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 12:51:37 +0100 Subject: [PATCH] Keep every binding the bus module says created, and never acknowledge one not kept distribution's membership carried 4 of its 8 bindings and mesh-knowledge's 3 of 6 (issue 1966). The grants part wrote a module's kept bindings with a blind put, so two created heard at once lost one, and acknowledged every seat event whatever its handler answered, while the log said it would come again. The bindings are now written at the revision read and read again once on a conflict, and an event not applied is left unacknowledged (a delayed nak). Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- modules/mesh-grants/cmd/mesh-grants/bus.go | 8 ++ modules/mesh-grants/cmd/mesh-grants/main.go | 15 ++- modules/mesh-grants/cmd/mesh-grants/tools.go | 5 + modules/mesh-grants/grants_test.go | 97 ++++++++++++++++--- .../mesh-grants/internal/grants/bindings.go | 32 ++++-- modules/mesh-grants/internal/grants/follow.go | 15 +-- modules/mesh-grants/internal/grants/part.go | 4 + 7 files changed, 147 insertions(+), 29 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/bus.go b/modules/mesh-grants/cmd/mesh-grants/bus.go index 7f839eaa..09836eb9 100644 --- a/modules/mesh-grants/cmd/mesh-grants/bus.go +++ b/modules/mesh-grants/cmd/mesh-grants/bus.go @@ -216,6 +216,14 @@ func (s ownStore) Put(state, key string, value []byte) (uint64, error) { func (s ownStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() } +// Update writes at a revision: a create where it is 0, refused as stdio.ErrChanged when another write came first. +func (s ownStore) Update(state, key string, value []byte, revision uint64) (uint64, error) { + if revision == 0 { + return s.c.State(state).Create(key, json.RawMessage(value)) + } + return s.c.State(state).Update(key, json.RawMessage(value), revision) +} + func (s ownStore) Delete(state, key string) error { return s.c.State(state).Delete(key) } // emitting is the process connection as the part's emitter: its events, with no options. diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 91a00b36..e0db8af0 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -44,6 +44,10 @@ import ( "mesh-grants/internal/grants" ) +// EventRetryDelay is how long a seat event the part could not apply waits before the bus says it again: long enough for a +// refused write of its own state to clear, short against the pass that follows it. +const EventRetryDelay = 5 * time.Second + func main() { if os.Getenv("MESH_BROKER_FILE") == "" { if err := serveTools(); err != nil { @@ -154,7 +158,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // there yet is opened at the next change, and one opened is never opened twice. opened := map[string]bool{} registered := false - f.Hear = func(handle func(string, []byte)) error { + f.Hear = func(handle func(string, []byte) error) error { if !registered { registered = true // An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go). @@ -182,7 +186,14 @@ 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 { + // An event not applied is left unacknowledged, said again after a while (issue 1966), never acknowledged unread. + if err := hub.Handle(subject, func(m jetstream.Msg) { + if handle(m.Subject(), m.Data()) != nil { + _ = m.NakWithDelay(EventRetryDelay) + return + } + _ = m.Ack() + }); err != nil { return err } } diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 53268412..fee5debd 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -132,6 +132,11 @@ func (runtimeStore) Put(state, key string, value []byte) (uint64, error) { func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +// Update is refused: the runtime writes no key at a revision, and a tool asks the part's process for what it writes so. +func (runtimeStore) Update(state, key string, _ []byte, _ uint64) (uint64, error) { + return 0, fmt.Errorf("%s %s is written at its revision by the part's process alone; a tool asks it", state, key) +} + func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } func (runtimeStore) PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error) { diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 62a54c96..9338a153 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + stdio "git.novox.be/novox/mesh-sdk/go" "io/fs" "maps" "os" @@ -51,17 +52,19 @@ import ( // tools, the part's own states and its events. It records every write, call and event, so a row can prove nothing was // published or asserted. type fakeBus struct { - mu sync.Mutex - states map[string]map[string][]byte - refused map[string]bool - loaded []grants.User - streams []map[string]any - own map[string]map[string][]byte - revs map[string]uint64 - refuseTo map[string]bool - calls []string - events []string - writes []string + // beforeUpdate, when set, runs once before the next Update: a second writer heard at once. + beforeUpdate func() + mu sync.Mutex + states map[string]map[string][]byte + refused map[string]bool + loaded []grants.User + streams []map[string]any + own map[string]map[string][]byte + revs map[string]uint64 + refuseTo map[string]bool + calls []string + events []string + writes []string // Each membership's subject: the last membership said there, every notice, and whether it is refused. stream map[string][]byte publishes []string @@ -332,6 +335,23 @@ func (b *fakeBus) ownGet(state, key string) ([]byte, uint64, bool, error) { type ownStore struct{ b *fakeBus } +// Update writes at a revision as the bus does: refused, wrapping stdio.ErrChanged, when the key is at another; and the +// fake's beforeUpdate, when set, writes first (a second writer heard at once). +func (s ownStore) Update(state, key string, value []byte, revision uint64) (uint64, error) { + if f := s.b.beforeUpdate; f != nil { + s.b.beforeUpdate = nil + f() + } + s.b.mu.Lock() + rev := s.b.revs[state+"/"+key] + _, has := s.b.own[state][key] + s.b.mu.Unlock() + if (revision == 0 && has) || (revision != 0 && rev != revision) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + return s.Put(state, key, value) +} + func (s ownStore) Get(state, key string) ([]byte, uint64, bool, error) { return s.b.ownGet(state, key) } func (s ownStore) Put(state, key string, value []byte) (uint64, error) { b := s.b @@ -617,6 +637,14 @@ func TestGrants(t *testing.T) { } src = []byte(strings.ReplaceAll(string(src), "CreateObjectStore", "")) } + // The one key created: a kept binding's first write in the part's own state, at revision 0 (bus.go's + // ownStore.Update, issue 1966), a key and never a bucket or a stream. + if path == filepath.Join("cmd", "mesh-grants", "bus.go") { + if strings.Count(string(src), "s.c.State(state).Create(key, ") != 1 { + t.Errorf("bus.go creates other than one key of its own state") + } + src = []byte(strings.Replace(string(src), "s.c.State(state).Create(key, ", "", 1)) + } for _, f := range forbidden { if strings.Contains(string(src), f) { t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f) @@ -1002,9 +1030,9 @@ func TestGrants(t *testing.T) { } // A root-not-free condition raised or cleared wakes a pass; another kind does not (never a timer). wb := fixture() - var handle func(string, []byte) + var handle func(string, []byte) error wf := &grants.Follower{Part: partOn(wb), Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {}, - Hear: func(h func(string, []byte)) error { handle = h; return nil }} + Hear: func(h func(string, []byte) error) error { handle = h; return nil }} wctx, wcancel := context.WithCancel(context.Background()) wdone := make(chan struct{}) go func() { wf.Run(wctx); close(wdone) }() @@ -3024,6 +3052,30 @@ func TestGrants(t *testing.T) { if !hear("mesh.seat.mesh-broker.event.created", first) || hear("mesh.seat.mesh-broker.event.created", first) { t.Fatal("created: not kept once") } + // Issue 1966 (F3): two created of one module heard at once are both kept: the second writer comes between the + // first's read and its write, and the first, refused at its revision, reads again and keeps both. + holds := membership.BusBinding{Local: "holds", Storage: "key-value", Template: "state", Bucket: "mod-x_holds"} + b.beforeUpdate = func() { + body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: "holds", Storage: "key-value", Template: "state", Binding: holds}) + if _, err := p.HearBroker("mesh.seat.mesh-broker.event.created", body, p.Brokers()); err != nil { + t.Errorf("the second writer: %v", err) + } + } + lost := first + lost.Local, lost.Bucket = "lost", "mod-x_lost" + body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: "lost", Storage: "key-value", Template: "state", Binding: lost}) + if _, err := p.HearBroker("mesh.seat.mesh-broker.event.created", body, p.Brokers()); err != nil { + t.Fatalf("created beside another writer: %v", err) + } + p.Compose() + if got := bindingsOn("a.mod-x"); len(got) != 3 { + t.Errorf("a binding was lost to a write it did not see: %+v", got) + } + for _, local := range []string{"holds", "lost"} { + if _, err := p.HearBroker("mesh.seat.mesh-broker.event.dropped", []byte(`{"module":"mod-x","local-name":"`+local+`"}`), p.Brokers()); err != nil { + t.Fatal(err) + } + } b.assign("b", "mod-x") p.Compose() for _, key := range []string{"a.mod-x", "b.mod-x"} { @@ -3536,10 +3588,10 @@ func TestGrants(t *testing.T) { }) t.Run("the vault's account-issued composes the list again", func(t *testing.T) { b := fixture() - var handle func(string, []byte) + var handle func(string, []byte) error p := partOn(b) f := &grants.Follower{Part: p, Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {}, - Hear: func(h func(string, []byte)) error { handle = h; return nil }} + Hear: func(h func(string, []byte) error) error { handle = h; return nil }} ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { f.Run(ctx); close(done) }() @@ -3561,6 +3613,21 @@ func TestGrants(t *testing.T) { b.mu.Unlock() handle(buscompose.AccountIssuedSubject, []byte(`{"user":"a.mod-x","kind":"module","node":"a","module":"mod-x","generation":1}`)) waitFor(t, held) + // Issue 1966 (F3): a bus module's created the part cannot keep is answered, so the hearer leaves it unacknowledged + // and the bus says it again; one kept is answered nil. + created := []byte(`{"module":"mod-x","local-name":"records","storage-type":"key-value","binding":{"bucket":"mod-x_records"}}`) + b.mu.Lock() + b.refuseTo[grants.StateMemberships] = true + b.mu.Unlock() + if err := handle("mesh.seat.mesh-broker.event.created", created); err == nil { + t.Error("a created not kept was answered nil, so it would be acknowledged and lost") + } + b.mu.Lock() + b.refuseTo[grants.StateMemberships] = false + b.mu.Unlock() + if err := handle("mesh.seat.mesh-broker.event.created", created); err != nil { + t.Errorf("a created kept was answered %v", err) + } if !slices.Contains(grants.Consumes, grants.EventAccountIssued) || !grants.IsAccountIssued(grants.EventAccountIssued) || grants.IsAccountIssued("mesh.mod.mesh-vault.event.secret-changed") { t.Errorf("the part hears %v", grants.Consumes) diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index 101900b8..f3cac9df 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -2,8 +2,11 @@ package grants import ( "encoding/json" + "errors" "fmt" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + stdio "git.novox.be/novox/mesh-sdk/go" "sort" "strings" @@ -93,7 +96,18 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b if e.Module == "" || local == "" || strings.Contains(e.Module, ".") { return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local) } - b, err := p.bound(e.Module) + // Written at the revision read, and read again once when another write came first (two `created` of one module heard + // at once): a binding is never lost to a write it did not see (issue 1966). + changed, err := p.keepBinding(e, name, local, storage) + if errors.Is(err, stdio.ErrChanged) { + changed, err = p.keepBinding(e, name, local, storage) + } + return changed, err +} + +// keepBinding applies one event to the kept bindings of its module, written at the revision it read them at. +func (p *Part) keepBinding(e BrokerEvent, name, local, storage string) (bool, error) { + b, rev, err := p.boundAt(e.Module) if err != nil { return false, err } @@ -119,7 +133,7 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b b.Bindings[local] = bind } raw, _ := json.Marshal(b) - if _, err := p.Store.Put(StateMemberships, BoundKey(e.Module), raw); err != nil { + if _, err := p.Store.Update(StateMemberships, BoundKey(e.Module), raw, rev); err != nil { return false, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(e.Module), err) } return true, nil @@ -127,20 +141,26 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b // bound reads the kept bindings of module; none kept is an empty value. func (p *Part) bound(module string) (Bound, error) { + b, _, err := p.boundAt(module) + return b, err +} + +// boundAt is bound with the revision it was read at, 0 when none is kept. +func (p *Part) boundAt(module string) (Bound, uint64, error) { b := Bound{V: 1, Module: module, Bindings: map[string]membership.BusBinding{}} - raw, _, found, err := p.Store.Get(StateMemberships, BoundKey(module)) + raw, rev, found, err := p.Store.Get(StateMemberships, BoundKey(module)) if err != nil { - return b, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err) + return b, 0, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err) } if found { if err := json.Unmarshal(raw, &b); err != nil { - return b, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err) + return b, 0, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err) } if b.Bindings == nil { b.Bindings = map[string]membership.BusBinding{} } } - return b, nil + return b, rev, nil } // bindingsOf is module's kept bindings in local-name order, for its memberships; nil when none. diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index 1344e273..a30d1c82 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -35,7 +35,8 @@ type Follower struct { // Hear hears the keeper's condition events and the vault's account-issued (Consumes) and hands each, by event name, // with its body, to handle; nil hears none. A condition of the kind root-not-free raised or cleared wakes a pass: // a machine's root-free word changed; and so does an account the vault minted (ADR 0561 §1), whose hash the list carries. - Hear func(handle func(event string, body []byte)) error + // handle answers an event it could not apply, which the hearer then leaves unacknowledged, so it is said again (issue 1966). + Hear func(handle func(event string, body []byte) error) error wake chan struct{} once sync.Once @@ -127,25 +128,27 @@ func (f *Follower) open() { } } if f.Hear != nil && !f.heard_ { - err := f.Hear(func(event string, body []byte) { + err := f.Hear(func(event string, body []byte) error { // The bus module's created, recreated or dropped: the binding it carries is kept, and a pass composes the - // memberships with it (bindings.go). + // memberships with it (bindings.go). One not applied is answered, so the hearer leaves it unacknowledged and the + // bus says it again (issue 1966: acknowledged, it was lost while this line said it would come again). if changed, err := f.Part.HearBroker(event, body, f.Part.Brokers()); err != nil { f.logf("%v: not applied, said again at its redelivery", err) - return + return err } else if changed { f.Wake() - return + return nil } // An account the vault minted. if IsAccountIssued(event) { f.Wake() - return + return nil } // A condition event is heard from the keeper alone (ADR 0590 §3). if FromKeeper(event) && IsRootNotFree(body) { f.Wake() } + return nil }) if err != nil { f.logf("the keeper's condition events cannot be heard yet (%v): tried again at the next change", err) diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index bd118429..b769e0a7 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -57,6 +57,10 @@ type Store interface { // Keys and Delete are for the wanted accounts (accounts.go) and the grants (grants.go): every other state is only put. Keys(state string) ([]string, error) Delete(state, key string) error + // Update writes a key only while it is still at revision (0: only while it holds nothing), and refuses one changed + // meanwhile with an error wrapping stdio.ErrChanged: the kept bindings are written so, never lost to a second writer + // (issue 1966). + Update(state, key string, value []byte, revision uint64) (uint64, error) } // Emitter says one of the part's events.