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.