The grants part takes a first view of every binding from the seat mesh-broker's list

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
This commit is contained in:
2026-11-11 10:30:22 +01:00
parent 532e4d740f
commit fcfaa41348
10 changed files with 948 additions and 19 deletions
@@ -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")
}
@@ -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)
+5 -3
View File
@@ -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)
}},
+250
View File
@@ -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
@@ -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 {
@@ -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, `<module>/<local name>`.
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 `<module>/<local name>`
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
}
@@ -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) == "" {
@@ -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"
+17 -5
View File
@@ -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
+2 -1
View File
@@ -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",