Merge pull request 'Registry recovery: the grants part takes a first view of every binding (issue 1966)' (#203) from fix/1966-bindings-first-view into main

This commit was merged in pull request #203.
This commit is contained in:
2026-11-11 10:32:34 +00:00
10 changed files with 1060 additions and 21 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,65 @@ 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
// 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)
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": 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)
}
}
// 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 = 10*time.Second, 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 +301,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 {
@@ -305,18 +365,81 @@ 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)
}
}
// 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"})
until("mod-c's binding in its membership after created", func() bool {
// 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"
})
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.
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.
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)
}
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"
})
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 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
}
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) {
t.Errorf("the bucket %s: %v, want none", bucket, err)
@@ -424,3 +547,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")
}
+28 -1
View File
@@ -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)
@@ -182,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
}
}
+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)
}},
+282
View File
@@ -3051,6 +3051,288 @@ 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.
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
}
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)}})
}
// 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) }
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 || st.v.Status().Answered.IsZero() {
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)
}
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)
}
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("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)
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,24 @@ 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): 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) {
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 +111,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,543 @@
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 {
seq uint64
event string
body []byte
ack func()
}
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. 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
}
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
}
// 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, most := v.Bound, v.Wait, v.MaxWait
if bound == 0 {
bound = views.Bound
}
if wait == 0 {
wait = 10 * time.Second
}
if most == 0 {
most = 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()
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 > most {
wait = most
}
}
}
// 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, left unacknowledged to be delivered again", err)
continue
} else if c {
changed = true
}
if e.ack != nil {
e.ack()
}
}
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 && 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",