From fcfaa413489006fa7355f9102bb3e787bb993640 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 10:30:22 +0100 Subject: [PATCH 1/3] The grants part takes a first view of every binding from the seat mesh-broker's list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit At each start and on grants_first_view_again with state bindings, the part asks list, writes the bindings the answer carries that it holds none of or another of, keeps and raises binding-unlisted for those the answer omits, never writes over an event heard since the ask, holds the seat's events until the answer or the bound, and raises first-view-unanswered naming mesh-broker and list past two minutes while composing as before. grants_first_views shows the first view beside the others. Cases are rows and subtests of the existing tests. Builds: design 25 bindings-first-view §1, §3, §4, §5, §6, §7, §8 (hq d0760629, PR 543) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-grants/integration_test.go | 93 +++- modules/mesh-grants/cmd/mesh-grants/main.go | 13 + modules/mesh-grants/cmd/mesh-grants/tools.go | 8 +- modules/mesh-grants/grants_test.go | 250 +++++++++ .../mesh-grants/internal/grants/bindings.go | 20 + .../internal/grants/bindingsview.go | 525 ++++++++++++++++++ .../mesh-grants/internal/grants/firstviews.go | 31 +- modules/mesh-grants/internal/grants/inputs.go | 2 +- modules/mesh-grants/internal/grants/part.go | 22 +- modules/mesh-grants/module.json | 3 +- 10 files changed, 948 insertions(+), 19 deletions(-) create mode 100644 modules/mesh-grants/internal/grants/bindingsview.go diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index 1931e920..687c1610 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -24,6 +24,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" @@ -216,6 +217,50 @@ func TestTheGrantsPartWhole(t *testing.T) { t.Fatal(err) } } + // 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 + 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 + } + 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{}}, + "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"}, + }}) + }); err != nil { + t.Fatal(err) + } + } + // The conditions the part said, by key, as last said. + conditionsSaid := func() map[string]string { + out := map[string]string{} + st, err := js.Stream(ctx, "CONDITIONS") + if err != nil { + return out + } + info, _ := st.Info(ctx) + for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ { + raw, err := st.GetMsg(ctx, seq) + if err != nil { + continue + } + if c, err := occ.DecodeCondition(raw.Data); err == nil { + out[c.Key] = string(c.State) + } + } + return out + } + bindingsBoundAndWaits := func() { + bindingsTiming.bound, bindingsTiming.wait, bindingsTiming.max = 300*time.Millisecond, 50*time.Millisecond, 100*time.Millisecond + } + bindingsBoundAndWaits() + t.Cleanup(func() { bindingsTiming.bound, bindingsTiming.wait, bindingsTiming.max = 0, 0, 0 }) // A membership is never said on its subject any more (issue 1126). var said atomic.Int32 if _, err := nc.Subscribe("mesh.assignment.>", func(*nats.Msg) { said.Add(1) }); err != nil { @@ -241,7 +286,7 @@ func TestTheGrantsPartWhole(t *testing.T) { return } } - t.Fatalf("never saw: %s", what) + t.Fatalf("never saw: %s\n%s", what, logged.tail("binding")) } until("every first view asked and refused once", func() bool { return refusedAsks.Load() >= int32(len(grants.FirstViews)) }) until("the part saying it waits on the indexer and the assigner", func() bool { @@ -310,13 +355,42 @@ func TestTheGrantsPartWhole(t *testing.T) { t.Fatal(err) } } + // 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. sayBroker("created", membership.BusBinding{Local: "kept", Storage: "key-value", Template: "state", Bucket: "mod-c_kept"}) - until("mod-c's binding in its membership after created", func() bool { + 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" + }) + // 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. + listing.Store(true) + until("the binding of the entry created never said, kept beside the one the answer does not list", func() bool { + b := bindingOf() + return len(b) == 2 && b[0].Bucket == "mod-c_kept" && b[1].Bucket == "mod-c_late" + }) + until("binding-unlisted raised for the kept binding and first-view-unanswered cleared", func() bool { + said := conditionsSaid() + return said[grants.BindingUnlisted+".mod-c.kept"] == "raised" && said[grants.FirstViewUnanswered+".mesh-broker.bindings"] == "cleared" && + 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}) + if _, err := kvs[grants.Module+"."+grants.StateUserList].Put(ctx, grants.KeyAsked, askedAgain); err != nil { + t.Fatal(err) + } + until("the bindings asked once more by hand", func() bool { + return logged.has("the first view of the bindings (" + grants.CallBrokerList + ") once more by hand") + }) sayBroker("dropped", membership.BusBinding{Local: "kept"}) - until("mod-c's binding gone after dropped", func() bool { return len(bindingOf()) == 0 }) + until("mod-c's kept binding gone after dropped, its unlisted condition cleared", func() bool { + b := bindingOf() + return len(b) == 1 && b[0].Bucket == "mod-c_late" && conditionsSaid()[grants.BindingUnlisted+".mod-c.kept"] == "cleared" + }) 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) @@ -424,3 +498,16 @@ func (l *lockedLog) has(s string) bool { defer l.mu.Unlock() return strings.Contains(l.buf.String(), s) } + +// tail is the lines of the log that contain s, for a failure to name what the part said of it. +func (l *lockedLog) tail(s string) string { + l.mu.Lock() + defer l.mu.Unlock() + var out []string + for _, line := range strings.Split(l.buf.String(), "\n") { + if strings.Contains(line, s) { + out = append(out, line) + } + } + return strings.Join(out, "\n") +} diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 91a00b36..288b6b6f 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -20,6 +20,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -60,6 +61,10 @@ func main() { } } +// bindingsTiming is the bound and the waits of the first view of the bindings; zero is the design's (two minutes; from ten +// seconds, doubling, at most five minutes: design 25 bindings-first-view §5). The whole-module test sets small ones. +var bindingsTiming struct{ bound, wait, max time.Duration } + // keptStates is what the part keeps (ADR 0495 §2-§3): each readable, for the view and a session, but the wanted accounts, // by the vault alone (ADR 0561 §2). func keptStates() []announce.State { return grants.Announced() } @@ -131,6 +136,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Objects: sealedObjects{ctx: ctx, js: js}, Purging: ownPurging{ctx: ctx, js: js, module: account.Module}, Logf: logger.Printf, Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}} f.Watch = fv + // The first view of the bindings (design 25 bindings-first-view): the seat mesh-broker's list, asked at each start and by + // hand, with the seat's events held until its answer is applied or its bound passes. + f.Part.Bindings = &grants.BindingsView{Say: sayCond, Changed: f.Wake, Logf: logger.Printf, + Bound: bindingsTiming.bound, Wait: bindingsTiming.wait, MaxWait: bindingsTiming.max, + Ask: func(context.Context) (json.RawMessage, error) { + return grants.Unwrapped(conn.Call)(grants.CallBrokerList, map[string]any{}) + }} + f.Part.Bindings.Start(ctx, f.Part) // The seat list: read over the connection and the hub, given to catalogue at every change, and a pass composes again then. gate := &seatGate{} gate.follow(seatlist.ProcessBus(conn.Call, hub), f.Wake, logger.Printf) diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 53268412..954b8f47 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -236,13 +236,15 @@ func toolsOf(p *grants.Part) []stdio.Tool { {Name: grants.ToolFirstViews, Description: "The grants part's first views of the owners' states (novox/hq ADR 0615 §4 to §6, " + "ADR 0622): the assigner's assignments and wants, the indexer's manifests, modules and verdicts, each with its owner, tool " + "and the revision its picture holds, when it asked and when it was answered, how many asks and hand acts, and whether " + - "its condition is raised, as the part's process last kept them. Read-only.", + "its condition is raised, as the part's process last kept them; and the first view of the bindings, asked of the seat " + + "mesh-broker's list: how many entries it listed, how many bindings it wrote and rewrote, and the bindings it found " + + "unlisted (design 25 bindings-first-view). Read-only.", Run: func(map[string]any) (any, error) { return p.FirstViewsOf() }}, {Name: grants.ToolFirstViewAgain, Description: "Ask one of the grants part's first views once more now (assignments, wants, " + - "manifests, modules or verdicts; none for all), by a person's act with why, recorded as a hand act: the step after an " + + "manifests, modules, verdicts or bindings, the bus module's list of every entry's binding; none for all), by a person's act with why, recorded as a hand act: the step after an " + "owner's state was written again from its sources (an empty answer at revision 0, condition " + "mesh-grants.first-view-empty). Asks the part's process, which alone follows them; grants_first_views shows the answer.", - Input: schema(map[string]string{"state": "assignments, wants, manifests, modules or verdicts; none for all", "why": "why, kept in the hand act"}, "why"), + Input: schema(map[string]string{"state": "assignments, wants, manifests, modules, verdicts or bindings; none for all", "why": "why, kept in the hand act"}, "why"), Run: func(a map[string]any) (any, error) { return p.AskFirstViewAgain(stdio.CallerOf(a), grants.Answer(a, "state"), grants.Answer(a, "why"), recordHandAct) }}, diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 62a54c96..3a6a4947 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -3051,6 +3051,256 @@ func TestGrants(t *testing.T) { } }) + // Design 25 bindings-first-view §3 to §6, §8: the part's first view of the bindings, the seat mesh-broker's list. One row + // per case of §4 (a binding written, the same one writes nothing, a different one rewritten, an entry listed without a + // binding, a binding the answer omits kept with its condition and cleared when listed again or dropped, an event heard + // since the ask not written over, a bus module from before the field), then the start (events held until the answer, a + // created heard while the bus module does not answer applied past the bound, the condition naming the seat and the verb) + // and the tools. + t.Run("bindings first view: the answer of mesh-broker's list is reconciled with what the events brought", func(t *testing.T) { + entry := func(local, bucket string) map[string]any { + return map[string]any{"module": "mod-x", "local-name": local, "storage-type": "key-value", "template": "state", "bus-name": "mod-x_" + local, + "settings": map[string]any{"mod-x_" + local: map[string]any{}}, + "binding": map[string]any{"local": local, "storage": "key-value", "template": "state", "bucket": bucket}} + } + withoutBinding := map[string]any{"module": "mod-x", "local-name": "waiting", "storage-type": "key-value", "bus-name": "mod-x_waiting", "state": "waiting"} + oldBusModule := map[string]any{"module": "mod-x", "local-name": "late", "storage-type": "key-value", "bus-name": "mod-x_late", + "settings": map[string]any{"mod-x_late": map[string]any{}}} + // A bus module's setup: the part with the first view, the conditions it said, the bindings it keeps for mod-x. + type setup struct { + b *fakeBus + p *grants.Part + v *grants.BindingsView + said map[string]string + raised map[string]oc.Condition + logs []string + ask func() []any + mu sync.Mutex + } + newSetup := func() *setup { + st := &setup{b: fixture(), said: map[string]string{}, raised: map[string]oc.Condition{}} + st.p = partOn(st.b) + st.p.Compose() + st.v = &grants.BindingsView{Changed: func() {}, Pause: func(context.Context, time.Duration) bool { return true }, + Say: func(c oc.Condition) error { + st.mu.Lock() + st.said[c.Key] = string(c.State) + if string(c.State) == "raised" { + st.raised[c.Key] = c + } + st.mu.Unlock() + return nil + }, + Logf: func(f string, a ...any) { + st.mu.Lock() + st.logs = append(st.logs, fmt.Sprintf(f, a...)) + st.mu.Unlock() + }} + st.v.Ask = func(context.Context) (json.RawMessage, error) { + return json.Marshal(map[string]any{"entries": st.ask()}) + } + st.p.Bindings = st.v + return st + } + kept := func(st *setup) map[string]string { + raw, _, found, _ := st.b.ownGet(grants.StateMemberships, grants.BoundKey("mod-x")) + out := map[string]string{} + if found { + var bound grants.Bound + _ = json.Unmarshal(raw, &bound) + for local, b := range bound.Bindings { + out[local] = b.Bucket + } + } + return out + } + hear := func(st *setup, event, local, bucket string) { + body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: local, Storage: "key-value", Template: "state", + Binding: membership.BusBinding{Local: local, Storage: "key-value", Template: "state", Bucket: bucket}}) + if _, err := st.p.HearBroker("mesh.seat.mesh-broker.event."+event, body, st.p.Brokers()); err != nil { + t.Fatal(err) + } + } + for _, row := range []struct { + name string + held map[string]string // kept before, by local name + answer []any + during func(*setup) // heard while the ask runs + want map[string]string + status [3]int // written, rewritten, listed + said map[string]string + legacy bool + again []any // a second answer + wantAgn map[string]string + saidAgn map[string]string + }{ + {name: "a binding the events never brought is written", answer: []any{entry("late", "mod-x_late")}, want: map[string]string{"late": "mod-x_late"}, status: [3]int{1, 0, 1}}, + {name: "the same binding writes nothing", held: map[string]string{"late": "mod-x_late"}, answer: []any{entry("late", "mod-x_late")}, + want: map[string]string{"late": "mod-x_late"}, status: [3]int{0, 0, 1}}, + {name: "a different binding is rewritten", held: map[string]string{"late": "mod-x_old"}, answer: []any{entry("late", "mod-x_late")}, + want: map[string]string{"late": "mod-x_late"}, status: [3]int{0, 1, 1}}, + {name: "an entry listed without a binding changes nothing", held: map[string]string{"waiting": "mod-x_waiting"}, answer: []any{withoutBinding}, + want: map[string]string{"waiting": "mod-x_waiting"}, status: [3]int{0, 0, 1}}, + {name: "a binding the answer omits is kept, raises its condition, and a later answer that lists it clears it", + held: map[string]string{"gone": "mod-x_gone"}, answer: []any{}, want: map[string]string{"gone": "mod-x_gone"}, + said: map[string]string{grants.BindingUnlisted + ".mod-x.gone": "raised"}, + again: []any{entry("gone", "mod-x_gone")}, wantAgn: map[string]string{"gone": "mod-x_gone"}, + saidAgn: map[string]string{grants.BindingUnlisted + ".mod-x.gone": "cleared"}}, + {name: "an event heard since the ask is not written over", answer: []any{entry("late", "mod-x_answer")}, + during: func(st *setup) { hear(st, "created", "late", "mod-x_event") }, want: map[string]string{"late": "mod-x_event"}, status: [3]int{0, 0, 1}}, + {name: "a dropped heard since the ask is not written back, and raises nothing", held: map[string]string{"late": "mod-x_late"}, + answer: []any{entry("late", "mod-x_late")}, during: func(st *setup) { hear(st, "dropped", "late", "") }, + want: map[string]string{}, status: [3]int{0, 0, 1}, said: map[string]string{}}, + {name: "an answer of a bus module from before the field writes nothing and says so once", answer: []any{oldBusModule}, + want: map[string]string{}, status: [3]int{0, 0, 1}, legacy: true, again: []any{oldBusModule}, wantAgn: map[string]string{}}, + } { + t.Run(row.name, func(t *testing.T) { + st := newSetup() + for local, bucket := range row.held { + hear(st, "created", local, bucket) + } + answer := row.answer + st.ask = func() []any { + if row.during != nil { + row.during(st) + } + return answer + } + check := func(want map[string]string, said map[string]string) { + t.Helper() + if got := kept(st); !maps.Equal(got, want) { + t.Errorf("kept %v, want %v", got, want) + } + for k, v := range said { + if st.said[k] != v { + t.Errorf("condition %s is %q, want %q (said %v)", k, st.said[k], v, st.said) + } + } + } + if err := st.v.Run(context.Background(), st.p, "start", false); err != nil { + t.Fatal(err) + } + check(row.want, row.said) + if s := st.v.Status(); [3]int{s.Written, s.Rewritten, s.Listed} != row.status && !row.legacy && row.again == nil { + t.Errorf("status wrote %d, rewrote %d, listed %d; want %v", s.Written, s.Rewritten, s.Listed, row.status) + } + if row.again != nil { + answer, row.during = row.again, nil + if err := st.v.Run(context.Background(), st.p, "again", true); err != nil { + t.Fatal(err) + } + check(row.wantAgn, row.saidAgn) + if st.v.Status().ByHand != 1 { + t.Errorf("by hand %d, want 1", st.v.Status().ByHand) + } + } + if told := strings.Count(strings.Join(st.logs, "\n"), "no binding (a bus module from before the field)"); row.legacy && told != 1 || !row.legacy && told != 0 { + t.Errorf("a bus module from before the field was told %d times", told) + } + }) + } + + // A dropped for an entry the part holds unlisted clears its condition. + t.Run("a dropped clears the unlisted condition of its binding", func(t *testing.T) { + st := newSetup() + hear(st, "created", "gone", "mod-x_gone") + st.ask = func() []any { return []any{} } + _ = st.v.Run(context.Background(), st.p, "start", false) + if st.said[grants.BindingUnlisted+".mod-x.gone"] != "raised" { + t.Fatalf("not raised: %v", st.said) + } + hear(st, "dropped", "gone", "") + if st.said[grants.BindingUnlisted+".mod-x.gone"] != "cleared" || len(kept(st)) != 0 { + t.Errorf("after dropped: %v, kept %v", st.said, kept(st)) + } + }) + + // 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") + } + 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()) + } + } + 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) + } + }) + + // 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) { + st := newSetup() + st.ask = func() []any { return []any{entry("late", "mod-x_late")} } + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + st.p.FirstViewContext, st.p.Consumers = ctx, st.b // the process reads a tool's ask only with its consumers + 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") { + t.Fatalf("bindings again: %v %v", err, acts) + } + changed := make(chan struct{}, 64) + st.v.Changed = func() { changed <- struct{}{} } + st.p.Compose() + for st.v.Status().ByHand == 0 || !st.v.Status().Answered.After(time.Time{}) { + select { + case <-changed: + case <-time.After(10 * time.Second): + t.Fatalf("never asked by hand: %+v", st.v.Status()) + } + } + st.p.Compose() + got, err := st.p.FirstViewsOf() + if raw, _ := json.Marshal(got); err != nil || !strings.Contains(string(raw), `"seat":"mesh-broker"`) || !strings.Contains(string(raw), `"by-hand":1`) || !strings.Contains(string(raw), `"written":1`) { + t.Errorf("grants_first_views = %s, %v", raw, err) + } + }) + }) + // Issue 1966 (F2), the conformance row: an entry asked in `requires` is granted where its old section is. For a // key-value entry and an object store, the old section and the new form compose the same subjects for the module's // own account, its process naming it and the tool runner carrying it, and issue the same state; a stream is granted diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index 101900b8..357cf4c5 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -79,11 +79,27 @@ func brokerEvent(event string, brokers map[string]bool) (string, bool) { // HearBroker applies one event heard to the kept bindings: a `created` or `recreated` writes the entry's binding, a // `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. 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." + +// 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) { name, ok := brokerEvent(event, brokers) if !ok { return false, nil } + p.bindMu.Lock() + defer p.bindMu.Unlock() var e BrokerEvent if err := json.Unmarshal(body, &e); err != nil { return false, fmt.Errorf("the bus module's %s does not read: %w", name, err) @@ -98,6 +114,10 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b return false, err } was, had := b.Bindings[local] + if p.Bindings != nil { + // Named since the last ask of the first view was sent, so an answer never writes over it (§4). + p.Bindings.touch(e.Module, local, name == BrokerDropped) + } switch name { case BrokerDropped: if !had { diff --git a/modules/mesh-grants/internal/grants/bindingsview.go b/modules/mesh-grants/internal/grants/bindingsview.go new file mode 100644 index 00000000..65a7f7ef --- /dev/null +++ b/modules/mesh-grants/internal/grants/bindingsview.go @@ -0,0 +1,525 @@ +package grants + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-sdk/go/membership" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// **The first view of the bindings** (design 25 bindings-first-view). The bus module's events were the binding's only +// source, so an event the part never heard lost the binding for good. The part now asks the seat mesh-broker's `list` once +// at each start (§3) and again when a person asks (grants_first_view_again, state `bindings`), never on a timer, and takes +// each binding the answer carries into its kept bindings (§4). The live events stay the way a binding changes between starts. + +// The first view of the bindings: its name for grants_first_view_again, the seat and verb it asks, and its conditions. +const ( + FirstViewBindings = "bindings" + BrokerSeat = meshbroker.Seat + BrokerListVerb = meshbroker.VerbList + // CallBrokerList is the seat's verb as the manifest's invokes names it. + CallBrokerList = "seat:" + meshbroker.Seat + "." + meshbroker.VerbList + // BindingUnlisted is the condition of a binding the part holds that the bus module no longer lists (§4), one per binding. + BindingUnlisted = "mesh-grants.binding-unlisted" +) + +// BindingsStatus is what the part's tools say of the first view of the bindings (§6), kept beside the other first views. +type BindingsStatus struct { + Seat string `json:"seat"` + Verb string `json:"verb"` + Asked time.Time `json:"asked,omitzero"` + Answered time.Time `json:"answered,omitzero"` + // Listed is how many entries the last answer listed; Written and Rewritten how many bindings it wrote and rewrote. + Listed int `json:"listed"` + Written int `json:"written"` + Rewritten int `json:"rewritten"` + // Unlisted are the bindings the part holds that the last answer did not list, `/`. + Unlisted []string `json:"unlisted,omitempty"` + Asks int `json:"asks"` + ByHand int `json:"by-hand"` + // Raised is whether the condition first-view-unanswered is raised for it. + Raised bool `json:"raised"` + Why string `json:"why,omitempty"` + Error string `json:"error,omitempty"` +} + +// BindingsView is the first view of the bindings. +type BindingsView struct { + // Ask calls the seat's list and answers its reply as the tool's own text (the process's connection). + Ask func(ctx context.Context) (json.RawMessage, error) + // Say says a condition; nil says none. Changed wakes a pass; Logf logs; Now is the clock. + Say func(oc.Condition) error + Changed func() + Logf func(string, ...any) + Now func() time.Time + // Pause waits d between two asks and answers false once ctx ended; nil waits on the context's bound. + Pause func(ctx context.Context, d time.Duration) bool + // Bound is how long it asks before it raises its condition (two minutes: views.Bound when zero); Wait the first wait + // between two asks that failed (ten seconds when zero), doubling to MaxWait (five minutes when zero). + Bound time.Duration + Wait time.Duration + MaxWait time.Duration + + mu sync.Mutex + st BindingsStatus + asking bool + holding bool + held []heldEvent + touched map[string]bool + raised map[string]bool // the unlisted conditions raised, by `/` + legacy bool +} + +type heldEvent struct { + event string + body []byte +} + +func (v *BindingsView) now() time.Time { + if v.Now != nil { + return v.Now().UTC() + } + return time.Now().UTC() +} + +func (v *BindingsView) logf(format string, args ...any) { + if v.Logf != nil { + v.Logf(format, args...) + } +} + +func (v *BindingsView) wake() { + if v.Changed != nil { + v.Changed() + } +} + +func (v *BindingsView) say(c oc.Condition) { + if v.Say == nil { + return + } + if err := v.Say(c); err != nil { + v.logf("the condition %s could not be said: %v", c.Key, err) + } +} + +// Status is the first view's status now. +func (v *BindingsView) Status() BindingsStatus { + v.mu.Lock() + defer v.mu.Unlock() + return v.statusLocked() +} + +func (v *BindingsView) statusLocked() BindingsStatus { + s := v.st + s.Seat, s.Verb = BrokerSeat, BrokerListVerb + s.Unlisted = slices.Sorted(func(yield func(string) bool) { + for k := range v.raised { + if !yield(k) { + return + } + } + }) + return s +} + +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 { + v.mu.Lock() + defer v.mu.Unlock() + if !v.holding { + return false + } + v.held = append(v.held, heldEvent{event, slices.Clone(body)}) + return true +} + +// touch notes that an event named the entry since the last ask was sent, so an answer never writes over it (§4); a dropped +// one also clears the entry's unlisted condition. +func (v *BindingsView) touch(module, local string, dropped bool) { + v.mu.Lock() + if v.touched == nil { + v.touched = map[string]bool{} + } + v.touched[localKey(module, local)] = true + clear := dropped && v.raised[localKey(module, local)] + if clear { + delete(v.raised, localKey(module, local)) + } + v.mu.Unlock() + if clear { + v.say(unlistedCondition(module, local, false)) + } +} + +// Run asks the first view and answers when it is answered (or ctx ends); Start runs it on a goroutine of its own. +func (v *BindingsView) Run(ctx context.Context, p *Part, why string, byHand bool) error { + return v.askUntil(ctx, p, why, byHand) +} + +// Start asks the first view once at the part's start, on a goroutine of its own, holding the seat's events until the answer +// is applied or the bound passes (§3). The unlisted conditions raised before the start are read from the part's kept status. +func (v *BindingsView) Start(ctx context.Context, p *Part) { + v.mu.Lock() + v.holding = true + v.mu.Unlock() + v.seed(p) + go func() { _ = v.askUntil(ctx, p, "start", false) }() +} + +// Again asks the first view once more by hand (grants_first_view_again), on a goroutine of its own. +func (v *BindingsView) Again(ctx context.Context, p *Part, why string) { + go func() { _ = v.askUntil(ctx, p, "by hand: "+why, true) }() +} + +// seed reads the unlisted bindings the last run kept, so a later answer that lists them clears their conditions. +func (v *BindingsView) seed(p *Part) { + raw, _, found, err := p.Store.Get(StateUserList, KeyFirstViews) + if err != nil || !found { + return + } + var st FirstViewsState + if json.Unmarshal(raw, &st) != nil || st.Bindings == nil { + return + } + v.mu.Lock() + defer v.mu.Unlock() + v.raised = map[string]bool{} + for _, k := range st.Bindings.Unlisted { + v.raised[k] = true + } +} + +func (v *BindingsView) pause(ctx context.Context, d time.Duration) bool { + if v.Pause != nil { + return v.Pause(ctx, d) + } + t, cancel := context.WithTimeout(ctx, d) + <-t.Done() + cancel() + return ctx.Err() == nil +} + +// askUntil asks until the bus module answers or ctx ends: from ten seconds, doubling, at most every five minutes (§5). +// Past the bound it raises first-view-unanswered, releases the events it holds, and keeps asking; the answer clears it. +func (v *BindingsView) askUntil(ctx context.Context, p *Part, why string, byHand bool) error { + v.mu.Lock() + if v.asking { + v.mu.Unlock() + return nil // one ask at a time + } + v.asking = true + v.st.Why = why + if byHand { + v.st.ByHand++ + } + v.mu.Unlock() + defer func() { + v.mu.Lock() + v.asking = false + v.mu.Unlock() + }() + bound, wait, max := 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 + } + first := v.now() + for { + v.mu.Lock() + v.touched = map[string]bool{} // what an event names from here on is newer than this ask + v.st.Asked = v.now() + v.st.Asks++ + v.mu.Unlock() + raw, err := v.Ask(ctx) + if err == nil { + err = v.apply(p, raw) + } + if err == nil { + return nil + } + v.mu.Lock() + v.st.Error = err.Error() + raise := !v.st.Raised && v.now().Sub(first) >= bound + if raise { + v.st.Raised = true + } + v.mu.Unlock() + v.logf("the first view %s of the bindings is not answered (%v): asked again in %s", CallBrokerList, err, wait) + if raise { + v.say(bindingsUnansweredCondition(true, v.Status())) + v.release(p) + } + v.wake() + if !v.pause(ctx, wait) { + return ctx.Err() + } + if wait *= 2; wait > max { + wait = max + } + } +} + +// release takes the events held, in the order heard, after the answer or past the bound. +func (v *BindingsView) release(p *Part) { + v.mu.Lock() + held := v.held + v.held, v.holding = nil, false + v.mu.Unlock() + 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) + } else if c { + changed = true + } + } + if changed { + v.wake() + } +} + +// apply takes an answer of list into the kept bindings (§4) and then the events held (§3). +func (v *BindingsView) apply(p *Part, raw json.RawMessage) error { + var answer meshbroker.ListAnswer + if err := json.Unmarshal(raw, &answer); err != nil { + return fmt.Errorf("%s answered what is not a list: %w", CallBrokerList, err) + } + p.bindMu.Lock() + written, rewritten, holdsNoBinding, err := v.take(p, answer.Entries) + p.bindMu.Unlock() + if err != nil { + return err + } + if holdsNoBinding > 0 { + v.mu.Lock() + said := v.legacy + v.legacy = true + v.mu.Unlock() + if !said { + v.logf("the seat %s's verb %s lists %d entries whose objects the bus holds with no binding (a bus module from before the field): nothing is written from the answer, and the events are heard as before", BrokerSeat, BrokerListVerb, holdsNoBinding) + } + } + v.mu.Lock() + wasRaised := v.st.Raised + v.st.Raised, v.st.Error = false, "" + v.st.Answered = v.now() + v.st.Listed = len(answer.Entries) + v.st.Written, v.st.Rewritten = written, rewritten + status := v.statusLocked() + v.mu.Unlock() + if wasRaised { + v.say(bindingsUnansweredCondition(false, status)) + } + v.release(p) + v.wake() + return nil +} + +// take writes the bindings of the entries an event did not name since the ask was sent, keeps every binding the answer does +// not list and raises its unlisted condition, and answers how many bindings it wrote and rewrote, and how many entries it +// lists with their objects held and no binding. The caller holds the part's bindMu. +func (v *BindingsView) take(p *Part, entries []meshbroker.Held) (written, rewritten, noBinding int, err error) { + v.mu.Lock() + touched := copyOf(v.touched) + v.mu.Unlock() + listed := map[string]bool{} + byModule := map[string][]meshbroker.Held{} + for _, e := range entries { + listed[localKey(e.Module, e.LocalName)] = true + if e.Binding == nil { + if len(e.Settings) > 0 { + noBinding++ + } + continue + } + byModule[e.Module] = append(byModule[e.Module], e) + } + modules := make([]string, 0, len(byModule)) + for m := range byModule { + modules = append(modules, m) + } + sort.Strings(modules) + for _, module := range modules { + if strings.Contains(module, ".") || module == "" { + continue + } + b, berr := p.bound(module) + if berr != nil { + return written, rewritten, noBinding, berr + } + changed := false + for _, e := range byModule[module] { + if touched[localKey(module, e.LocalName)] || e.LocalName == "" { + continue + } + bind, ok := membershipBinding(e) + if !ok { + v.logf("the binding of %s %s in the answer of %s does not read: not written", module, e.LocalName, CallBrokerList) + continue + } + 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) + } + if had && sameBinding(was, bind) { + continue + } + b.Bindings[e.LocalName] = bind + changed = true + if had { + rewritten++ + v.logf("the binding of %s %s was rewritten, the cause the first view of the bindings (%s)", module, e.LocalName, CallBrokerList) + } else { + written++ + v.logf("the binding of %s %s was written, the cause the first view of the bindings (%s)", module, e.LocalName, CallBrokerList) + } + } + if changed { + raw, _ := json.Marshal(b) + if _, perr := p.Store.Put(StateMemberships, BoundKey(module), raw); perr != nil { + return written, rewritten, noBinding, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(module), perr) + } + } + } + return written, rewritten, noBinding, v.unlisted(p, listed, touched) +} + +// unlisted keeps every binding the answer does not list and raises its condition, one per binding; a binding it lists again +// clears it (§4). +func (v *BindingsView) unlisted(p *Part, listed, touched map[string]bool) error { + keys, err := p.Store.Keys(StateMemberships) + if err != nil { + return fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err) + } + holds := map[string]bool{} + for _, key := range keys { + module, ok := strings.CutPrefix(key, BoundPrefix) + if !ok { + continue + } + b, berr := p.bound(module) + if berr != nil { + return berr + } + for local := range b.Bindings { + holds[localKey(module, local)] = true + } + } + v.mu.Lock() + if v.raised == nil { + v.raised = map[string]bool{} + } + var raise, clearing []string + for k := range holds { + switch { + case listed[k] && v.raised[k]: + clearing = append(clearing, k) + case !listed[k] && !touched[k] && !v.raised[k]: + raise = append(raise, k) + } + } + for k := range v.raised { // a raised one the part no longer holds is gone with its binding + if !holds[k] { + clearing = append(clearing, k) + } + } + for _, k := range raise { + v.raised[k] = true + } + for _, k := range clearing { + delete(v.raised, k) + } + v.mu.Unlock() + sort.Strings(raise) + sort.Strings(clearing) + for _, k := range raise { + module, local, _ := strings.Cut(k, "/") + v.logf("the binding of %s %s is held and the bus module no longer lists it: kept, and %s raised", module, local, BindingUnlisted) + v.say(unlistedCondition(module, local, true)) + } + for _, k := range clearing { + module, local, _ := strings.Cut(k, "/") + v.say(unlistedCondition(module, local, false)) + } + return nil +} + +func copyOf(m map[string]bool) map[string]bool { + out := make(map[string]bool, len(m)) + for k, v := range m { + out[k] = v + } + return out +} + +// membershipBinding is an entry's binding in the form the membership holds: the seat's Binding is that form, so the JSON is +// the same. Its local name and storage are the entry's where it left them out. +func membershipBinding(e meshbroker.Held) (membership.BusBinding, bool) { + var b membership.BusBinding + raw, err := json.Marshal(e.Binding) + if err != nil || json.Unmarshal(raw, &b) != nil { + return b, false + } + b.Local = e.LocalName + if b.Storage == "" { + b.Storage = e.StorageType + } + if b.Template == "" { + b.Template = e.Template + } + return b, true +} + +// bindingsUnansweredCondition is FirstViewUnanswered for the first view of the bindings, naming the seat mesh-broker and its +// verb list. +func bindingsUnansweredCondition(raised bool, s BindingsStatus) oc.Condition { + c := oc.Condition{Key: FirstViewUnanswered + "." + BrokerSeat + "." + FirstViewBindings, Subject: oc.ConditionSubject{Scope: "module", ID: Module}} + if !raised { + c.State = occ.StateCleared + c.Resolved = CallBrokerList + " answered" + return c + } + c.State, c.Severity = occ.StateRaised, occ.SeverityWarning + c.Summary = fmt.Sprintf("the grants part has no first view of the bindings: the seat %s's verb %s has not answered since %s (%s)", + BrokerSeat, BrokerListVerb, s.Asked.Format(time.RFC3339), strings.TrimSpace(s.Error)) + c.Headline = "The grants part waits for the bus module" + needsYou(&c, "a look at the module holding the seat "+BrokerSeat+": is it running and connected?", "The grants part learns a binding it never "+ + "heard from the seat's "+BrokerListVerb+", and keeps asking it; it composes every account and keeps every binding it holds meanwhile, and "+ + "takes the bus module's events as they come. The answer clears this.") + return c +} + +// unlistedCondition is BindingUnlisted for one binding the part holds that the bus module no longer lists, raised or cleared. +func unlistedCondition(module, local string, raised bool) oc.Condition { + c := oc.Condition{Key: BindingUnlisted + "." + module + "." + local, Subject: oc.ConditionSubject{Scope: "module", ID: Module}} + if !raised { + c.State = occ.StateCleared + c.Resolved = "the bus module lists " + module + " " + local + " again, or dropped it" + return c + } + c.State, c.Severity = occ.StateRaised, occ.SeverityWarning + c.Summary = fmt.Sprintf("the grants part holds the binding of %s %s, which the seat %s's verb %s no longer lists", module, local, BrokerSeat, BrokerListVerb) + c.Headline = "A binding the bus module no longer lists" + needsYou(&c, "read the entry with the seat "+BrokerSeat+"'s show; an entry its manifest no longer asks for leaves by the bus module's drop", "The grants "+ + "part keeps the binding: it removes one only on the bus module's dropped, so a bus module whose own state was lost or rebuilt strips "+ + "no module of its data. A later answer that lists the entry, or its dropped, clears this.") + return c +} diff --git a/modules/mesh-grants/internal/grants/firstviews.go b/modules/mesh-grants/internal/grants/firstviews.go index 11aace93..77bd2818 100644 --- a/modules/mesh-grants/internal/grants/firstviews.go +++ b/modules/mesh-grants/internal/grants/firstviews.go @@ -257,7 +257,7 @@ func (v *Views) Again(ctx context.Context, state, why string) ([]string, error) asked = append(asked, fv.Tool) } if len(asked) == 0 { - return nil, fmt.Errorf("%q is no first view of the grants part: one of assignments, wants, manifests, modules, verdicts, or none for all", state) + return nil, fmt.Errorf("%q is no first view of the grants part: one of assignments, wants, manifests, modules, verdicts, bindings, or none for all", state) } return asked, nil } @@ -394,20 +394,35 @@ type firstViewsSaid struct { type FirstViewsState struct { V int `json:"v"` Views []views.Status `json:"views"` + // Bindings is the first view of the bindings, asked of the seat mesh-broker's list, beside the others (design 25 + // bindings-first-view §6); absent where the part's process holds none. + Bindings *BindingsStatus `json:"bindings,omitempty"` } // sayFirstViews keeps the first views' statuses in the part's own state (for grants_first_views) when they change, and // raises or clears FirstViewEmpty for each answered at revision 0. Each pass calls it; nil Views is none. func (p *Part) sayFirstViews(res *Result) { - if p.Views == nil { + if p.Views == nil && p.Bindings == nil { return } - statuses := p.Views.Statuses() + state := FirstViewsState{V: 1, Views: []views.Status{}} + if p.Views != nil { + state.Views = p.Views.Statuses() + } + statuses := state.Views for i := range statuses { statuses[i].Asked, statuses[i].Answered = statuses[i].Asked.Truncate(time.Second), statuses[i].Answered.Truncate(time.Second) } - data, _ := json.Marshal(FirstViewsState{V: 1, Views: statuses}) + if p.Bindings != nil { + b := p.Bindings.Status() + b.Asked, b.Answered = b.Asked.Truncate(time.Second), b.Answered.Truncate(time.Second) + state.Bindings = &b + } + data, _ := json.Marshal(state) p.changed(res, StateUserList, KeyFirstViews, data) + if p.Views == nil { + return + } p.fvSaid.mu.Lock() defer p.fvSaid.mu.Unlock() if p.fvSaid.empty == nil { @@ -443,7 +458,8 @@ func (p *Part) FirstViewsOf() (any, error) { if err := json.Unmarshal(raw, &st); err != nil { return nil, fmt.Errorf("its own %s %s does not read: %w", StateUserList, KeyFirstViews, err) } - return map[string]any{"views": st.Views, "revision": rev, "first-views": FirstViews}, nil + return map[string]any{"views": st.Views, "bindings": st.Bindings, "revision": rev, "first-views": FirstViews, + "bindings-first-view": map[string]string{"seat": BrokerSeat, "verb": BrokerListVerb}}, nil } // AskFirstViewAgain answers the tool grants_first_view_again: it asks the part's process to ask one first view (by its @@ -453,13 +469,16 @@ func (p *Part) AskFirstViewAgain(caller, state, why string, record func(what, wh which := AllFirstViews if state != "" { which = "" + if state == FirstViewBindings || state == CallBrokerList { + which = FirstViewBindings + } for _, fv := range FirstViews { if state == fv.State || state == fv.Tool || state == fv.Read { which = fv.State } } if which == "" { - return nil, fmt.Errorf("%q is no first view of the grants part: one of assignments, wants, manifests, modules, verdicts, or none for all", state) + return nil, fmt.Errorf("%q is no first view of the grants part: one of assignments, wants, manifests, modules, verdicts, bindings, or none for all", state) } } if strings.TrimSpace(why) == "" { diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index cc7dc116..d885b67f 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -75,7 +75,7 @@ const ( ) // Invokes is every tool the part calls. -var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions}, firstViewTools()...) +var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions, CallBrokerList}, firstViewTools()...) // CallSeatDefinitions reads the seat list (mesh-sdk seatlist): the verb of the seat mesh-catalog. const CallSeatDefinitions = "seat:mesh-catalog.seat-definitions" diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index bd118429..b6e126ac 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -216,6 +216,10 @@ type Part struct { Views *Views FirstViewContext context.Context fvSaid firstViewsSaid + // Bindings is the first view of the bindings, asked of the seat mesh-broker's list (bindingsview.go); nil in the tools. + Bindings *BindingsView + // bindMu orders the writes of the kept bindings: an event's and a first view's answer (bindings.go). + bindMu sync.Mutex } // keepBrokers keeps the modules whose manifest claims the seat mesh-broker: the only emitters whose created, recreated @@ -520,15 +524,23 @@ func (p *Part) Compose() Result { } p.consumers(&res, recs, ask.Consumer) p.sayFirstViews(&res) - if ask.FirstView != "" && p.Views != nil && p.FirstViewContext != nil { + if ask.FirstView != "" && p.FirstViewContext != nil { state := ask.FirstView if state == AllFirstViews { state = "" } - if asked, err := p.Views.Again(p.FirstViewContext, state, "by hand: "+ask.By); err != nil { - res.Errors = append(res.Errors, err.Error()) - } else if p.Logf != nil { - p.Logf("asked the first views %v once more by hand, for %s", asked, ask.By) + if p.Bindings != nil && (state == "" || state == FirstViewBindings) { + p.Bindings.Again(p.FirstViewContext, p, "by hand: "+ask.By) + if p.Logf != nil { + p.Logf("asked the first view of the bindings (%s) once more by hand, for %s", CallBrokerList, ask.By) + } + } + if p.Views != nil && state != FirstViewBindings { + if asked, err := p.Views.Again(p.FirstViewContext, state, "by hand: "+ask.By); err != nil { + res.Errors = append(res.Errors, err.Error()) + } else if p.Logf != nil { + p.Logf("asked the first views %v once more by hand, for %s", asked, ask.By) + } } } return res diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index c60bfcea..4ae5ec87 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -40,7 +40,8 @@ "mesh-delivery-indexer.modules", "mesh-delivery-indexer.verdicts", "seat:mesh-catalog.seat-definitions", - "seat:mesh-hand-acts.hand-act" + "seat:mesh-hand-acts.hand-act", + "seat:mesh-broker.list" ], "emits": [ "asserted", -- 2.54.0 From 20af25ea400e3f466b493dffcb964ce9b4538ba4 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:01:39 +0100 Subject: [PATCH 2/3] 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 -- 2.54.0 From 8501968154b4721bc3d075c58068c5ceb25a8e6e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:22:57 +0100 Subject: [PATCH 3/3] Raise and release the bindings' first view at the bound itself, so held events never reach dead letters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-grants/integration_test.go | 34 +++++++++++++------ modules/mesh-grants/grants_test.go | 14 +++++++- .../internal/grants/bindingsview.go | 22 +++++++----- 3 files changed, 49 insertions(+), 21 deletions(-) diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index f9725a7b..e438366b 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -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) { diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index ac194076..601792c7 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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) diff --git a/modules/mesh-grants/internal/grants/bindingsview.go b/modules/mesh-grants/internal/grants/bindingsview.go index dd138323..e596c596 100644 --- a/modules/mesh-grants/internal/grants/bindingsview.go +++ b/modules/mesh-grants/internal/grants/bindingsview.go @@ -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 } -- 2.54.0