Raise and release the bindings' first view at the bound itself, so held events never reach dead letters
mesh/delivery merged: source.main moved to novox/mesh-delivery@8f5cbce46845
mesh/delivery-group group fix/1966-bindings-first-view merged: every member is merged
mesh/check checking 8501968154b4 on 8f5cbce46845 | novox/mesh-delivery#203 in group fix/1966-bindings-first-view

No pause between two failed asks runs past the bound (two minutes): the ask at the bound raises
first-view-unanswered and releases the held events, well inside the events consumer's five deliveries
of 30 s. An answer's binding is taken as it lists it, with no fallback to an event's types (§4).
Checks: the unit start row requires the raise at exactly the bound; the whole test says the created
before the bound and requires the condition raised when it is applied (fails without the process's
hold), and waits on the kept first-views status for the held answer instead of a sleep.

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

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
2026-11-11 11:22:57 +01:00
parent 20af25ea40
commit 8501968154
3 changed files with 49 additions and 21 deletions
@@ -232,18 +232,22 @@ func TestTheGrantsPartWhole(t *testing.T) {
}
listAsked.Add(1)
defer listAnswered.Add(1)
var more []any
if g := gate.Load(); g != nil {
g.asked <- struct{}{}
<-g.release
defer answeredGated.Add(1)
// The held answer lists one entry more (waiting, no binding), so its taking shows in the kept status.
more = append(more, map[string]any{"module": "mod-c", "local-name": "also-waiting", "storage-type": "key-value",
"template": "state", "bus-name": "mod-c_also-waiting", "state": "waiting"})
}
answer(m, map[string]any{"entries": []any{
answer(m, map[string]any{"entries": append([]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{}},
"binding": map[string]any{"local": "late", "storage": "key-value", "template": "state", "bucket": "mod-c_late"}},
map[string]any{"module": "mod-c", "local-name": "waiting", "storage-type": "key-value", "template": "state", "bus-name": "mod-c_waiting",
"state": "waiting"},
}})
}, more...)})
}); err != nil {
t.Fatal(err)
}
@@ -268,7 +272,7 @@ func TestTheGrantsPartWhole(t *testing.T) {
return out
}
bindingsBoundAndWaits := func() {
bindingsTiming.bound, bindingsTiming.wait, bindingsTiming.max = 300*time.Millisecond, 50*time.Millisecond, 100*time.Millisecond
bindingsTiming.bound, bindingsTiming.wait, bindingsTiming.max = 10*time.Second, 50*time.Millisecond, 100*time.Millisecond
}
bindingsBoundAndWaits()
t.Cleanup(func() { bindingsTiming.bound, bindingsTiming.wait, bindingsTiming.max = 0, 0, 0 })
@@ -368,14 +372,18 @@ func TestTheGrantsPartWhole(t *testing.T) {
}
// With list unanswered the part has composed every account (mod-c's membership is above), says past its bound that the
// bindings' first view is unanswered, naming the seat and the verb, and applies the created heard meanwhile.
if said := conditionsSaid()[grants.FirstViewUnanswered+".mesh-broker.bindings"]; said != "" {
t.Fatalf("first-view-unanswered %q before the created is said: the bound (10s) passed first", said)
}
sayBroker("created", membership.BusBinding{Local: "kept", Storage: "key-value", Template: "state", Bucket: "mod-c_kept"})
// The created is held until the bound (§3): when it is applied, first-view-unanswered is already raised.
until("mod-c's binding in its membership after created, list unanswered", func() bool {
b := bindingOf()
return len(b) == 1 && b[0].Bucket == "mod-c_kept"
})
until("first-view-unanswered raised for mesh-broker's list", func() bool {
return conditionsSaid()[grants.FirstViewUnanswered+".mesh-broker.bindings"] == "raised"
})
if said := conditionsSaid()[grants.FirstViewUnanswered+".mesh-broker.bindings"]; said != "raised" {
t.Fatalf("the created held while list is unanswered was applied before the bound: first-view-unanswered %q, want raised", said)
}
// The bus module answers: the entry whose created was never said gets its binding; the one the answer does not list (kept)
// stays and raises binding-unlisted; an entry listed without a binding changes nothing; the answer clears the first
// view's condition.
@@ -421,12 +429,16 @@ func TestTheGrantsPartWhole(t *testing.T) {
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)
until("the held answer taken, as the kept status of the first views says", func() bool {
e, err := kvs[grants.Module+"."+grants.StateUserList].Get(ctx, grants.KeyFirstViews)
if err != nil {
return false
}
time.Sleep(50 * time.Millisecond)
var st grants.FirstViewsState
return json.Unmarshal(e.Value(), &st) == nil && st.Bindings != nil && st.Bindings.Listed == 3 && st.Bindings.Error == ""
})
if b := bindingOf(); len(b) != 0 {
t.Fatalf("after an answer that lists a binding dropped since its ask: %+v", b)
}
for _, bucket := range []string{"mesh-controller_lease", grants.Module + "_lease", grants.Module + "_takeover"} {
if _, err := js.KeyValue(ctx, bucket); !errors.Is(err, jetstream.ErrBucketNotFound) {
+13 -1
View File
@@ -3246,6 +3246,15 @@ func TestGrants(t *testing.T) {
}
return json.Marshal(map[string]any{"entries": []any{entry("late", row.answer)}})
}
// Raised at the bound itself, no pause running past it (§3, §5): the held events are released then.
start, raisedAt := now, time.Time{}
say := st.v.Say
st.v.Say = func(c oc.Condition) error {
if string(c.State) == "raised" && c.Key == grants.FirstViewUnanswered+".mesh-broker.bindings" {
raisedAt = st.v.Now()
}
return say(c)
}
hold := make(chan struct{})
ask := st.v.Ask
st.v.Ask = func(ctx context.Context) (json.RawMessage, error) { <-hold; return ask(ctx) }
@@ -3262,7 +3271,7 @@ func TestGrants(t *testing.T) {
}
close(hold)
deadline := time.After(10 * time.Second)
for acked.Load() == 0 {
for acked.Load() == 0 || st.v.Status().Answered.IsZero() {
select {
case <-changed:
case <-deadline:
@@ -3279,6 +3288,9 @@ func TestGrants(t *testing.T) {
if st.said[grants.FirstViewUnanswered+".mesh-broker.bindings"] != "cleared" {
t.Errorf("the unanswered condition after the answer: %v", st.said)
}
if got := raisedAt.Sub(start); got != 2*time.Minute {
t.Errorf("raised %s after the start, want at the bound (2m)", got)
}
c, ok := st.raised[grants.FirstViewUnanswered+".mesh-broker.bindings"]
if !ok {
t.Fatalf("never raised past its bound: %v", st.said)
@@ -241,15 +241,15 @@ func (v *BindingsView) askUntil(ctx context.Context, p *Part, why string, byHand
v.asking = false
v.mu.Unlock()
}()
bound, wait, max := v.Bound, v.Wait, v.MaxWait
bound, wait, most := v.Bound, v.Wait, v.MaxWait
if bound == 0 {
bound = views.Bound
}
if wait == 0 {
wait = 10 * time.Second
}
if max == 0 {
max = 5 * time.Minute
if most == 0 {
most = 5 * time.Minute
}
first := v.now()
for {
@@ -278,11 +278,18 @@ func (v *BindingsView) askUntil(ctx context.Context, p *Part, why string, byHand
v.release(p)
}
v.wake()
if !v.pause(ctx, wait) {
d := wait // no pause runs past the bound: the ask at the bound raises and releases (§3, §5)
v.mu.Lock()
raised := v.st.Raised
v.mu.Unlock()
if left := bound - v.now().Sub(first); !raised && left < d {
d = max(left, 0)
}
if !v.pause(ctx, d) {
return ctx.Err()
}
if wait *= 2; wait > max {
wait = max
if wait *= 2; wait > most {
wait = most
}
}
}
@@ -390,9 +397,6 @@ func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewrit
continue
}
was, had := b.Bindings[e.LocalName]
if had && len(bind.Types) == 0 {
bind.Types = was.Types // an answer that names none leaves the types an event brought
}
if had && sameBinding(was, bind) {
continue
}