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

@@ -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.

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
 				}
 			}

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) {

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)

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.

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)

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.