The grants part keeps and recovers every binding of state asked in requires, and the validator reads both forms (issue 1966) #190

Open
mesh-tools wants to merge 7 commits from fix/1966-requires-grants into main
19 changed files with 831 additions and 56 deletions
@@ -166,6 +166,15 @@ func BusStateProblems(m catalogue.Manifest) []string {
"%s: the name is also kept in the old section %s; one local name is one form, so a module that moves an "+
"entry removes it from the old section (novox/hq ADR 0595 §11)", at, section))
}
// A key-value entry kept one bucket key per machine is not issued in the new form yet (issue 1966, the lead's
// ruling): its binding names no per-machine key, so neither the grants part nor the tool runner could hold each
// machine to its own key (ADR 0260). Refused with the way that holds, never issued as a whole bucket.
if e.Storage == catalogue.StorageKeyValue && string(e.Options[catalogue.OptionPerMachine]) == "true" {
problems = append(problems, fmt.Sprintf(
"%s: per-machine is not issued in `requires` yet, since its binding names no machine's key and each "+
"machine would reach every key; keep it in the old section state with per-machine until the binding "+
"carries it (novox/hq ADR 0260, ADR 0595 §1)", at))
}
problems = append(problems, busEntryProblems(at, e, terminal)...)
}
return problems
@@ -28,6 +28,14 @@ func TestBusStateIsJudgedOnce(t *testing.T) {
{"within bounds", manifest(`{"name":"cache","storage":"key-value","template":"state","options":{"history":8,"key-ttl":true}},` +
`{"name":"audit","storage":"stream","template":"log","options":{"max-mib":64,"max-age-days":365}},` +
`{"name":"files","storage":"object-store","template":"objects","options":{"max-mib":2048}},` + ledger), nil},
// Issue 1966: a process's bus state names a key-value entry of `requires` by its local name, as an old state's
// (ADR 0595 §11); one per machine is refused with the way that holds, its binding naming no machine's key (Q3).
{"a process naming a requires key-value entry", `{"module":"a","version":"1","requires":[{"name":"cache","storage":"key-value","template":"state"}],` +
`"resources":[{"id":"u","type":"user","name":"a"},{"id":"p","type":"process","user":"a","run":["./a"],"bus":{"state":["cache"]}}]}`, nil},
{"a process naming a state it keeps in neither form", `{"module":"a","version":"1","requires":[{"name":"cache","storage":"key-value","template":"state"}],` +
`"resources":[{"id":"u","type":"user","name":"a"},{"id":"p","type":"process","user":"a","run":["./a"],"bus":{"state":["other"]}}]}`, []string{"bus reads and writes state \"other\""}},
{"per-machine in requires", manifest(`{"name":"x","storage":"key-value","template":"state","options":{"per-machine":true}}`),
[]string{"\"x\"", "per-machine is not issued in `requires` yet", "old section state"}},
// 1966/AC7: each refused case, naming the module, the entry and the field
{"unknown storage type", manifest(`{"name":"x","storage":"tape"}`), []string{"a's bus state \"x\"", "storage type \"tape\""}},
{"unknown template", manifest(`{"name":"x","storage":"stream","template":"pile"}`), []string{"\"x\"", "template \"pile\""}},
+7 -8
View File
@@ -3913,9 +3913,11 @@ func ProcessBusProblems(m catalogue.Manifest) []string {
for _, s := range m.Streams {
streams[s.Name] = true
}
// What the module keeps, in either form (catalogue.KeptStates, issue 1966): a process names a key-value entry of
// `requires` by its local name as it names an old state (ADR 0595 §11).
state := map[string]bool{}
for _, s := range m.State {
state[s.Name] = true
for _, s := range catalogue.KeptStates(m) {
state[s] = true
}
emits := map[string]bool{}
for _, e := range m.EmitsAll() {
@@ -4306,8 +4308,8 @@ func ownStateProblems(m catalogue.Manifest, i int, r catalogue.Receivable, c cat
return nil
}
keeps := map[string]bool{}
for _, s := range m.State {
keeps[s.Name] = true
for _, s := range catalogue.KeptStates(m) { // either form (issue 1966)
keeps[s] = true
}
var problems []string
for _, sr := range r.Shape.Reads(c.Data) {
@@ -5118,10 +5120,7 @@ func trafficProblems(m catalogue.Manifest, s catalogue.SeatDeclaration) []string
m.Module, s.Name))
}
for _, r := range s.Records {
kept := false
for _, st := range m.State {
kept = kept || st.Name == r
}
kept := slices.Contains(catalogue.KeptStates(m), r) // either form (issue 1966)
if !kept {
problems = append(problems, fmt.Sprintf(
"%s says %s's users read its records %q, which it keeps no state of", m.Module, s.Name, r))
@@ -1403,9 +1403,12 @@ func caseAModuleOffersItsOwnStateOnly(t *testing.T) {
t.Errorf("%s: accepted, or refused without %q: %v", raw, want, err)
}
}
ok := `{"module":"power","state":["draw"],"contributions":[` + fmt.Sprintf(block, "power.draw") + `]}`
if _, err := ParseManifest([]byte(ok)); err != nil {
t.Errorf("its own state was refused: %v", err)
// Issue 1966: its own state in either form, the old section's or a key-value entry of `requires` (ADR 0595 §11).
for _, ok := range []string{`{"module":"power","state":["draw"],"contributions":[` + fmt.Sprintf(block, "power.draw") + `]}`,
`{"module":"power","requires":[{"name":"draw","storage":"key-value","template":"state"}],"contributions":[` + fmt.Sprintf(block, "power.draw") + `]}`} {
if _, err := ParseManifest([]byte(ok)); err != nil {
t.Errorf("its own state was refused: %v", err)
}
}
}
@@ -4735,6 +4738,11 @@ func caseTheNewRulesAreHeldToWhatTheSeatSays(t *testing.T) {
t.Errorf("not refused: %q in %s", want, got)
}
}
// Issue 1966: records kept as a key-value entry of `requires` are kept state, as the old section's are.
m.BusState = []catalogue.BusStateEntry{{Name: "nothing", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState}}
if got := strings.Join(declaredSeatProblems(m), "; "); strings.Contains(got, "keeps no state of") {
t.Errorf("records kept in requires refused: %s", got)
}
}
// **A build machine has no store, so it may not judge a seat.** The set is data the control plane
@@ -216,6 +216,14 @@ func (s ownStore) Put(state, key string, value []byte) (uint64, error) {
func (s ownStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() }
// Update writes at a revision: a create where it is 0, refused as stdio.ErrChanged when another write came first.
func (s ownStore) Update(state, key string, value []byte, revision uint64) (uint64, error) {
if revision == 0 {
return s.c.State(state).Create(key, json.RawMessage(value))
}
return s.c.State(state).Update(key, json.RawMessage(value), revision)
}
func (s ownStore) Delete(state, key string) error { return s.c.State(state).Delete(key) }
// emitting is the process connection as the part's emitter: its events, with no options.
@@ -16,6 +16,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/membership"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seal"
"github.com/nats-io/nats-server/v2/server"
@@ -37,7 +38,10 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
"novox": {{Module: buscompose.GrantsModule, State: []buscompose.Bucket{{Module: buscompose.GrantsModule, Name: "user-list"}}}},
"ace": {{Module: "plex", Reads: []string{buscompose.ComposedRead}},
{Module: "mesh-delivery-view", Reads: []string{buscompose.ComposedRead}, ReadsObjects: []string{"mesh-knowledge_sections"}},
{Module: "mesh-knowledge", Objects: []buscompose.ObjectBucket{{Module: "mesh-knowledge", Name: "sections", MaxMiB: 1, Restricted: true}}}},
{Module: "mesh-knowledge", Objects: []buscompose.ObjectBucket{{Module: "mesh-knowledge", Name: "sections", MaxMiB: 1, Restricted: true}}},
// Issue 1966: a module whose state is asked in `requires`, bound as the bus module said it created it.
{Module: "mod-c", Bindings: []membership.BusBinding{{Local: "kept", Storage: catalogue.StorageKeyValue,
Template: catalogue.TemplateState, Bucket: "mod-c_kept"}}}},
},
}
// The bus module and the router as the part composes them from their own manifests (testdata: the catalogue's
@@ -50,6 +54,10 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
}
records.Assigned["novox"] = append(records.Assigned["novox"], fromManifests.Assigned["novox"]...)
records.BusState = fromManifests.BusState
if records.BusState == nil {
records.BusState = map[string][]catalogue.BusStateEntry{}
}
records.BusState["mod-c"] = []catalogue.BusStateEntry{{Name: "kept", Storage: catalogue.StorageKeyValue, Template: catalogue.TemplateState}}
// The node-engine's state as the part's records carry it: every node's compositions (ADR 0593 §6).
records.Engine = &buscompose.Declared{Module: buscompose.EngineModule, State: buscompose.EngineComposedBuckets(records.Nodes)}
users, err := buscompose.Users(records)
@@ -388,6 +396,34 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
if _, err := plex.GetLastMsg(sections.Stream(), "$O."+sections.Bucket()+".M.x", nats.DirectGet()); err == nil {
t.Error("a module the grant does not name read the bucket")
}
// Issue 1966 (ADR 0595 §1, design 25 § 12.7, the binding's usable half): the bus module's keeper creates a key-value
// entry asked in `requires` under its own account, and the module's own account, granted it from its binding, binds the
// bucket by its bus name and writes and reads a key; another module reads nothing of it.
keeperJS, err := jetstream.New(dial("novox.nats.keeper"))
if err != nil {
t.Fatal(err)
}
if _, err := keeperJS.CreateStream(ctx, jetstream.StreamConfig{Name: "KV_mod-c_kept", Subjects: []string{"$KV.mod-c_kept.>"},
MaxMsgsPerSubject: 1, AllowDirect: true, AllowRollup: true, DenyDelete: true, Discard: jetstream.DiscardNew}); err != nil {
t.Fatalf("the keeper creates mod-c's bound bucket: %v", err)
}
modC, err := jetstream.New(dial("ace.mod-c"))
if err != nil {
t.Fatal(err)
}
bound, err := modC.KeyValue(ctx, "mod-c_kept")
if err != nil {
t.Fatalf("mod-c's own account binds its bound bucket: %v", err)
}
if _, err := bound.Put(ctx, "k", []byte("v")); err != nil {
t.Fatalf("mod-c's own account writes its bound bucket: %v", err)
}
if e, err := bound.Get(ctx, "k"); err != nil || string(e.Value()) != "v" {
t.Errorf("mod-c reads back %v, %v", e, err)
}
if _, err := plex.GetLastMsg("KV_mod-c_kept", "$KV.mod-c_kept.k", nats.DirectGet()); err == nil {
t.Error("another module read mod-c's bound bucket")
}
}
// manifestIn is a module's manifest as testdata keeps it.
@@ -0,0 +1,75 @@
package main
import (
"context"
"encoding/json"
"fmt"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
"github.com/nats-io/nats.go"
"mesh-grants/internal/buscompose"
"mesh-grants/internal/grants"
)
// readAgainAtMost bounds one re-read of the seat mesh-broker's events (grants_bindings_again): a stream holding more of
// them than this is read up to here, and the answer says the rest was not read.
const readAgainAtMost = 20000
// brokerHistory reads the seat mesh-broker's events again from the shared events stream, by sequence and subject
// (STREAM.MSG.GET with next_by_subj, the grant buscompose.SeatEventsReadAgain): a get per event, no consumer, oldest first.
type brokerHistory struct {
ctx context.Context
nc *nats.Conn
}
// msgGetAnswer is the bus's answer to a STREAM.MSG.GET.
type msgGetAnswer struct {
Message *struct {
Subject string `json:"subject"`
Seq uint64 `json:"seq"`
Data []byte `json:"data"`
Time time.Time `json:"time"`
} `json:"message"`
Error *struct {
Code int `json:"code"`
ErrCode int `json:"err_code"`
Description string `json:"description"`
} `json:"error"`
}
// noMessage is the bus's error code for a get past the last message of the subject.
const noMessage = 10037
func (h brokerHistory) Said() ([]grants.SaidEvent, error) {
subject := "mesh.seat." + meshbroker.Seat + ".event.*"
var out []grants.SaidEvent
for seq := uint64(1); ; {
if len(out) >= readAgainAtMost {
return out, fmt.Errorf("the stream holds more than %d of the seat's events: read up to sequence %d, the rest not", readAgainAtMost, seq)
}
body, _ := json.Marshal(map[string]any{"seq": seq, "next_by_subj": subject})
ctx, cancel := context.WithTimeout(h.ctx, requestWithin)
reply, err := h.nc.RequestWithContext(ctx, buscompose.SeatEventsReadAgain, body)
cancel()
if err != nil {
return out, fmt.Errorf("%s at sequence %d: %w", buscompose.SeatEventsReadAgain, seq, err)
}
var a msgGetAnswer
if err := json.Unmarshal(reply.Data, &a); err != nil {
return out, fmt.Errorf("%s answered what does not read: %w", buscompose.SeatEventsReadAgain, err)
}
if a.Error != nil {
if a.Error.ErrCode == noMessage {
return out, nil
}
return out, fmt.Errorf("%s at sequence %d: %s", buscompose.SeatEventsReadAgain, seq, a.Error.Description)
}
if a.Message == nil {
return out, nil
}
out = append(out, grants.SaidEvent{Subject: a.Message.Subject, Data: a.Message.Data, At: a.Message.Time})
seq = a.Message.Seq + 1
}
}
@@ -316,6 +316,24 @@ func TestTheGrantsPartWhole(t *testing.T) {
})
sayBroker("dropped", membership.BusBinding{Local: "kept"})
until("mod-c's binding gone after dropped", func() bool { return len(bindingOf()) == 0 })
// Issue 1966, grants_bindings_again: asked as the tool asks, the process reads the seat's events again from EVENTS by
// sequence (brokerHistory) and keeps mod-c's created and dropped in their order, so the dropped entry stays dropped.
ask, _ := json.Marshal(grants.Ask{By: "jochen@shanks", At: time.Now(), Bindings: "mod-c"})
if _, err := kvs[grants.Module+"."+grants.StateUserList].Put(ctx, grants.KeyAsked, ask); err != nil {
t.Fatal(err)
}
until("mod-c's bindings read again from EVENTS", func() bool {
e, err := kvs[grants.Module+"."+grants.StateUserList].Get(ctx, grants.KeyBindingsAgain)
if err != nil {
return false
}
var again grants.BindingsAgain
return json.Unmarshal(e.Value(), &again) == nil && again.Module == "mod-c" && again.Read == 2 && again.Error == "" &&
slices.Equal(again.Dropped, []string{"kept"}) && len(again.Kept) == 0
})
if b := bindingOf(); len(b) != 0 {
t.Errorf("a dropped binding came back from the re-read: %+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)
+14 -2
View File
@@ -44,6 +44,10 @@ import (
"mesh-grants/internal/grants"
)
// EventRetryDelay is how long a seat event the part could not apply waits before the bus says it again: long enough for a
// refused write of its own state to clear, short against the pass that follows it.
const EventRetryDelay = 5 * time.Second
func main() {
if os.Getenv("MESH_BROKER_FILE") == "" {
if err := serveTools(); err != nil {
@@ -131,6 +135,7 @@ 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
f.Part.History = brokerHistory{ctx: ctx, nc: nc}
// 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)
@@ -154,7 +159,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// there yet is opened at the next change, and one opened is never opened twice.
opened := map[string]bool{}
registered := false
f.Hear = func(handle func(string, []byte)) error {
f.Hear = func(handle func(string, []byte) error) error {
if !registered {
registered = true
// An owner's views event goes to the follower of its state, which acknowledges it (firstviews.go).
@@ -182,7 +187,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
// The other seats' events the part follows its passes by (Consumes).
for _, subject := range grants.SeatEventSubjects() {
if err := hub.Handle(subject, func(m jetstream.Msg) { handle(m.Subject(), m.Data()); _ = m.Ack() }); err != nil {
// An event not applied is left unacknowledged, said again after a while (issue 1966), never acknowledged unread.
if err := hub.Handle(subject, func(m jetstream.Msg) {
if handle(m.Subject(), m.Data()) != nil {
_ = m.NakWithDelay(EventRetryDelay)
return
}
_ = m.Ack()
}); err != nil {
return err
}
}
+27 -4
View File
@@ -132,6 +132,11 @@ func (runtimeStore) Put(state, key string, value []byte) (uint64, error) {
func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
// Update is refused: the runtime writes no key at a revision, and a tool asks the part's process for what it writes so.
func (runtimeStore) Update(state, key string, _ []byte, _ uint64) (uint64, error) {
return 0, fmt.Errorf("%s %s is written at its revision by the part's process alone; a tool asks it", state, key)
}
func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }
func (runtimeStore) PutTTL(state, key string, value []byte, ttl time.Duration) (uint64, error) {
@@ -246,13 +251,31 @@ func toolsOf(p *grants.Part) []stdio.Tool {
Run: func(a map[string]any) (any, error) {
return p.AskFirstViewAgain(stdio.CallerOf(a), grants.Answer(a, "state"), grants.Answer(a, "why"), recordHandAct)
}},
{Name: grants.ToolBindings, Description: "The bindings the grants part keeps of a module's kept data on the bus (novox/hq " +
"ADR 0595 §1, §7): each entry the bus module said created or recreated, by local name with its names on the bus, beside " +
"every membership of the module the part wrote, and the kept bindings a membership lacks. One module, or every one. Read-only.",
Input: schema(map[string]string{"module": "one module, e.g. distribution; none for every one"}),
Run: func(a map[string]any) (any, error) { return p.BindingsOf(grants.Answer(a, "module")) }},
{Name: grants.ToolBindingsAgain, Description: "Read one module's bindings again from the seat mesh-broker's created, " +
"recreated and dropped that the shared events stream still holds, kept as if heard now (issue 1966), by a person's act " +
"with why, recorded as a hand act: the step when grants_bindings lacks an entry the bus module made. Asks the part's " +
"process; grants_bindings shows what it kept, what was already kept, and when the stream no longer holds the created.",
Input: schema(map[string]string{"module": "the module, e.g. distribution", "why": "why, kept in the hand act"}, "module", "why"),
Run: func(a map[string]any) (any, error) {
return p.AskBindingsAgain(stdio.CallerOf(a), grants.Answer(a, "module"), grants.Answer(a, "why"), recordHandActFor("bindings-again"))
}},
}
}
// recordHandAct records a person's act through the seat mesh-hand-acts.
func recordHandAct(what, why string) error {
_, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(meshhandactsclient.Request{What: what, Why: why, Cause: "first-view-again"})
return err
// recordHandAct records a person's act of grants_first_view_again through the seat mesh-hand-acts.
func recordHandAct(what, why string) error { return recordHandActFor("first-view-again")(what, why) }
// recordHandActFor records a person's act through the seat mesh-hand-acts, with its cause.
func recordHandActFor(cause string) func(what, why string) error {
return func(what, why string) error {
_, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
return err
}
}
// serveTools serves the tools to the runtime until it stops.
+301 -15
View File
@@ -8,6 +8,10 @@ import (
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
stdio "git.novox.be/novox/mesh-sdk/go"
"go/ast"
"go/parser"
"go/token"
"io/fs"
"maps"
"os"
@@ -51,17 +55,19 @@ import (
// tools, the part's own states and its events. It records every write, call and event, so a row can prove nothing was
// published or asserted.
type fakeBus struct {
mu sync.Mutex
states map[string]map[string][]byte
refused map[string]bool
loaded []grants.User
streams []map[string]any
own map[string]map[string][]byte
revs map[string]uint64
refuseTo map[string]bool
calls []string
events []string
writes []string
// beforeUpdate, when set, runs once before the next Update: a second writer heard at once.
beforeUpdate func()
mu sync.Mutex
states map[string]map[string][]byte
refused map[string]bool
loaded []grants.User
streams []map[string]any
own map[string]map[string][]byte
revs map[string]uint64
refuseTo map[string]bool
calls []string
events []string
writes []string
// Each membership's subject: the last membership said there, every notice, and whether it is refused.
stream map[string][]byte
publishes []string
@@ -332,6 +338,23 @@ func (b *fakeBus) ownGet(state, key string) ([]byte, uint64, bool, error) {
type ownStore struct{ b *fakeBus }
// Update writes at a revision as the bus does: refused, wrapping stdio.ErrChanged, when the key is at another; and the
// fake's beforeUpdate, when set, writes first (a second writer heard at once).
func (s ownStore) Update(state, key string, value []byte, revision uint64) (uint64, error) {
if f := s.b.beforeUpdate; f != nil {
s.b.beforeUpdate = nil
f()
}
s.b.mu.Lock()
rev := s.b.revs[state+"/"+key]
_, has := s.b.own[state][key]
s.b.mu.Unlock()
if (revision == 0 && has) || (revision != 0 && rev != revision) {
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
}
return s.Put(state, key, value)
}
func (s ownStore) Get(state, key string) ([]byte, uint64, bool, error) { return s.b.ownGet(state, key) }
func (s ownStore) Put(state, key string, value []byte) (uint64, error) {
b := s.b
@@ -617,6 +640,14 @@ func TestGrants(t *testing.T) {
}
src = []byte(strings.ReplaceAll(string(src), "CreateObjectStore", ""))
}
// The one key created: a kept binding's first write in the part's own state, at revision 0 (bus.go's
// ownStore.Update, issue 1966), a key and never a bucket or a stream.
if path == filepath.Join("cmd", "mesh-grants", "bus.go") {
if strings.Count(string(src), "s.c.State(state).Create(key, ") != 1 {
t.Errorf("bus.go creates other than one key of its own state")
}
src = []byte(strings.Replace(string(src), "s.c.State(state).Create(key, ", "", 1))
}
for _, f := range forbidden {
if strings.Contains(string(src), f) {
t.Errorf("%s holds %s: the part holds no timer and creates or asserts nothing", path, f)
@@ -1002,9 +1033,9 @@ func TestGrants(t *testing.T) {
}
// A root-not-free condition raised or cleared wakes a pass; another kind does not (never a timer).
wb := fixture()
var handle func(string, []byte)
var handle func(string, []byte) error
wf := &grants.Follower{Part: partOn(wb), Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {},
Hear: func(h func(string, []byte)) error { handle = h; return nil }}
Hear: func(h func(string, []byte) error) error { handle = h; return nil }}
wctx, wcancel := context.WithCancel(context.Background())
wdone := make(chan struct{})
go func() { wf.Run(wctx); close(wdone) }()
@@ -2842,6 +2873,68 @@ func TestGrants(t *testing.T) {
}
})
// Issue 1966 (ADR 0595 §1, §11): the state a manifest keeps and a membership issues is read through mesh-sdk's two rules,
// catalogue.KeptStates and Membership.StateIssued or StateNames, which answer the old `state` section and a key-value
// entry asked in `requires` alike. A reader of a manifest's or a membership's `State` beside them misses the new form,
// the kind of this issue's faults; only the rules that judge or compose the old section itself read it. Read by the
// syntax tree across the repository, since the field's name alone is shared by other types.
t.Run("the state a manifest keeps or a membership issues is read through mesh-sdk's two rules", func(t *testing.T) {
judgesTheOldSection := map[string]bool{
"modules/mesh-delivery-indexer/judge/busstate.go BusStateProblems": true,
"modules/mesh-delivery-indexer/judge/oldsections.go SectionsUsed": true,
"modules/mesh-delivery-indexer/judge/rules.go readOnlyStateProblems": true,
"modules/mesh-delivery-indexer/judge/rules.go parseManifest": true,
"modules/mesh-delivery-indexer/judge/rules.go StateProblems": true,
"modules/mesh-delivery-indexer/judge/rules.go ObjectsProblems": true,
"modules/mesh-grants/internal/grants/records.go bucketsOf": true,
}
planted := "package p\n\nimport \"git.novox.be/novox/mesh-sdk/go/catalogue\"\n\n" +
"func planted(m catalogue.Manifest) (n int) {\n\tfor range m.State {\n\t\tn++\n\t}\n\treturn n\n}\n"
if got, err := oldStateReaders("p.go", []byte(planted)); err != nil || !slices.Equal(got, []string{"p.go planted"}) {
t.Fatalf("a planted range m.State is read as %v (%v), want it found", got, err)
}
files, seen := 0, map[string]bool{}
err := filepath.WalkDir(filepath.Join("..", ".."), func(path string, d fs.DirEntry, err error) error {
if err != nil {
return err
}
if d.IsDir() && path != filepath.Join("..", "..") && (d.Name() == "testdata" || d.Name() == "node_modules" || (len(d.Name()) > 1 && d.Name()[0] == '.')) {
return filepath.SkipDir
}
if d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
return nil
}
src, err := os.ReadFile(path)
if err != nil {
return err
}
files++
rel, _ := filepath.Rel(filepath.Join("..", ".."), path)
readers, err := oldStateReaders(filepath.ToSlash(rel), src)
if err != nil {
return err
}
for _, r := range readers {
seen[r] = true
if !judgesTheOldSection[r] {
t.Errorf("%s reads a manifest's or a membership's State beside catalogue.KeptStates and Membership.StateIssued", r)
}
}
return nil
})
if err != nil {
t.Fatal(err)
}
if files < 100 {
t.Fatalf("read %d source files, want the repository's", files)
}
for r := range judgesTheOldSection {
if !seen[r] {
t.Errorf("%s no longer reads the old section: take it off the list", r)
}
}
})
// 774/S29/AC1 (ADR 0549 §5, ADR 0571 §10): the memberships composed from the assignments alone, with no read of the
// controller, are the ones the union of assignments and assignments gave on the same input: the fixture's (two nodes,
// the bus and mod-x on a, mod-y on b, as placed on the live mesh's own shape of nats beside its modules), whose
@@ -3024,6 +3117,53 @@ func TestGrants(t *testing.T) {
if !hear("mesh.seat.mesh-broker.event.created", first) || hear("mesh.seat.mesh-broker.event.created", first) {
t.Fatal("created: not kept once")
}
// Issue 1966: grants_bindings reads what the part keeps: a binding kept before the pass that writes it into the
// memberships is named lacking there, and no longer once the pass has written it; a module with none says so.
lacking := func() string {
got, err := p.BindingsOf("mod-x")
if err != nil {
t.Fatalf("grants_bindings: %v", err)
}
raw, _ := json.Marshal(got)
if !strings.Contains(string(raw), `"bucket":"mod-x_records"`) {
t.Errorf("grants_bindings names no kept binding: %s", raw)
}
return string(raw)
}
if got := lacking(); !strings.Contains(got, `"lacking":{"a.mod-x":["records"]}`) {
t.Errorf("a binding kept and not written is not named lacking: %s", got)
}
p.Compose()
if got := lacking(); strings.Contains(got, `"lacking"`) || !strings.Contains(got, `"a.mod-x":["records"]`) {
t.Errorf("a binding written is named lacking: %s", got)
}
if got, err := p.BindingsOf("mod-y"); err != nil || !strings.Contains(fmt.Sprint(got), "keeps no binding of mod-y") {
t.Errorf("grants_bindings of a module with none = %v, %v", got, err)
}
// Issue 1966 (F3): two created of one module heard at once are both kept: the second writer comes between the
// first's read and its write, and the first, refused at its revision, reads again and keeps both.
holds := membership.BusBinding{Local: "holds", Storage: "key-value", Template: "state", Bucket: "mod-x_holds"}
b.beforeUpdate = func() {
body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: "holds", Storage: "key-value", Template: "state", Binding: holds})
if _, err := p.HearBroker("mesh.seat.mesh-broker.event.created", body, p.Brokers()); err != nil {
t.Errorf("the second writer: %v", err)
}
}
lost := first
lost.Local, lost.Bucket = "lost", "mod-x_lost"
body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-x", Local: "lost", Storage: "key-value", Template: "state", Binding: lost})
if _, err := p.HearBroker("mesh.seat.mesh-broker.event.created", body, p.Brokers()); err != nil {
t.Fatalf("created beside another writer: %v", err)
}
p.Compose()
if got := bindingsOn("a.mod-x"); len(got) != 3 {
t.Errorf("a binding was lost to a write it did not see: %+v", got)
}
for _, local := range []string{"holds", "lost"} {
if _, err := p.HearBroker("mesh.seat.mesh-broker.event.dropped", []byte(`{"module":"mod-x","local-name":"`+local+`"}`), p.Brokers()); err != nil {
t.Fatal(err)
}
}
b.assign("b", "mod-x")
p.Compose()
for _, key := range []string{"a.mod-x", "b.mod-x"} {
@@ -3049,6 +3189,64 @@ func TestGrants(t *testing.T) {
if got := bindingsOn("a.mod-x"); len(got) != 0 {
t.Errorf("after dropped: %+v", got)
}
// Issue 1966, grants_bindings_again (the lead's ruling, option b): the process reads the seat's events the shared
// stream still holds again, oldest first, and keeps one module's created, recreated and dropped as if heard now: a
// binding never kept comes back, one dropped since stays dropped, another module's is left; the hand act is recorded
// before the ask, and an ask without why or a module is refused.
event := func(name, module, local, bucket string) grants.SaidEvent {
body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: module, LocalName: local, StorageType: "key-value", Template: "state",
Binding: membership.BusBinding{Local: local, Storage: "key-value", Template: "state", Bucket: bucket}})
return grants.SaidEvent{Subject: "mesh.seat.mesh-broker.event." + name, Data: body, At: time.Date(2026, 11, 3, 0, 0, 0, 0, time.UTC)}
}
p.Create = b
p.History = saidHistory{event("created", "mod-x", "windows", "mod-x_windows"), event("refused", "mod-x", "mirrors", ""),
event("created", "mod-y", "other", "mod-y_other"), event("created", "mod-x", "gone", "mod-x_gone"),
event("dropped", "mod-x", "gone", "")}
var acts []string
record := func(what, why string) error { acts = append(acts, what+": "+why); return nil }
for _, bad := range [][2]string{{"mod-x", ""}, {"", "x"}, {"a.mod-x", "x"}} {
if _, err := p.AskBindingsAgain("jochen@shanks", bad[0], bad[1], record); err == nil || len(acts) != 0 {
t.Errorf("grants_bindings_again %q %q was asked: %v %v", bad[0], bad[1], err, acts)
}
}
readAgain := func(module string) string {
if _, err := p.AskBindingsAgain("jochen@shanks", module, "distribution's lost bindings", record); err != nil {
t.Fatalf("grants_bindings_again %s: %v", module, err)
}
p.Compose()
got, err := p.BindingsOf(module)
if err != nil {
t.Fatal(err)
}
raw, _ := json.Marshal(got)
return string(raw)
}
if got := readAgain("mod-x"); !strings.Contains(got, `"kept":["windows"],"already":[],"dropped":["gone"]`) || !strings.Contains(got, `"read":3`) {
t.Errorf("the first re-read: %s", got)
}
if got := bindingsOn("a.mod-x"); len(got) != 1 || got[0].Bucket != "mod-x_windows" {
t.Errorf("after the re-read, mod-x's membership carries %+v", got)
}
if raw, _, found, _ := b.ownGet(grants.StateMemberships, grants.BoundKey("mod-y")); found {
t.Errorf("another module's binding was kept by mod-x's re-read: %s", raw)
}
if got := readAgain("mod-x"); !strings.Contains(got, `"kept":[],"already":["windows"]`) {
t.Errorf("the second re-read: %s", got)
}
if got := readAgain("mod-z"); !strings.Contains(got, "holds no created or recreated of mod-z") || !strings.Contains(got, "seat mesh-broker's list") {
t.Errorf("a module whose created the stream no longer holds: %s", got)
}
// A re-read that fails partway applies nothing it read: the created whose dropped lay past the failure is not kept.
p.History = failingHistory{saidHistory{event("created", "mod-x", "late", "mod-x_late")}}
if got := readAgain("mod-x"); !strings.Contains(got, "could not be read, so nothing was kept") || !strings.Contains(got, `"read":0`) {
t.Errorf("a failed re-read: %s", got)
}
if got := bindingsOn("a.mod-x"); len(got) != 1 || got[0].Bucket != "mod-x_windows" {
t.Errorf("a failed re-read kept what it read before the failure: %+v", got)
}
if len(acts) != 4 || !strings.HasPrefix(acts[0], "grants_bindings_again mod-x: ") {
t.Errorf("hand acts %v", acts)
}
})
// Issue 1966 (F2), the conformance row: an entry asked in `requires` is granted where its old section is. For a
@@ -3536,10 +3734,10 @@ func TestGrants(t *testing.T) {
})
t.Run("the vault's account-issued composes the list again", func(t *testing.T) {
b := fixture()
var handle func(string, []byte)
var handle func(string, []byte) error
p := partOn(b)
f := &grants.Follower{Part: p, Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {},
Hear: func(h func(string, []byte)) error { handle = h; return nil }}
Hear: func(h func(string, []byte) error) error { handle = h; return nil }}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { f.Run(ctx); close(done) }()
@@ -3561,6 +3759,21 @@ func TestGrants(t *testing.T) {
b.mu.Unlock()
handle(buscompose.AccountIssuedSubject, []byte(`{"user":"a.mod-x","kind":"module","node":"a","module":"mod-x","generation":1}`))
waitFor(t, held)
// Issue 1966 (F3): a bus module's created the part cannot keep is answered, so the hearer leaves it unacknowledged
// and the bus says it again; one kept is answered nil.
created := []byte(`{"module":"mod-x","local-name":"records","storage-type":"key-value","binding":{"bucket":"mod-x_records"}}`)
b.mu.Lock()
b.refuseTo[grants.StateMemberships] = true
b.mu.Unlock()
if err := handle("mesh.seat.mesh-broker.event.created", created); err == nil {
t.Error("a created not kept was answered nil, so it would be acknowledged and lost")
}
b.mu.Lock()
b.refuseTo[grants.StateMemberships] = false
b.mu.Unlock()
if err := handle("mesh.seat.mesh-broker.event.created", created); err != nil {
t.Errorf("a created kept was answered %v", err)
}
if !slices.Contains(grants.Consumes, grants.EventAccountIssued) || !grants.IsAccountIssued(grants.EventAccountIssued) ||
grants.IsAccountIssued("mesh.mod.mesh-vault.event.secret-changed") {
t.Errorf("the part hears %v", grants.Consumes)
@@ -4409,3 +4622,76 @@ func channelTakes(t *testing.T, say func(oc.Condition) error) func(oc.Condition)
return say(c)
}
}
// oldStateReaders names, as "<path> <function>", every function of the source that reads the field State of a
// catalogue.Manifest or a membership.Membership: a receiver, parameter or declared variable of either type, or a field
// named Manifest or Membership (issue 1966's source guard).
func oldStateReaders(path string, src []byte) ([]string, error) {
f, err := parser.ParseFile(token.NewFileSet(), path, src, 0)
if err != nil {
return nil, err
}
typed := func(e ast.Expr) bool {
if s, ok := e.(*ast.StarExpr); ok {
e = s.X
}
sel, ok := e.(*ast.SelectorExpr)
if !ok {
return false
}
x, ok := sel.X.(*ast.Ident)
return ok && ((x.Name == "catalogue" && sel.Sel.Name == "Manifest") || (x.Name == "membership" && sel.Sel.Name == "Membership"))
}
var out []string
for _, decl := range f.Decls {
fn, ok := decl.(*ast.FuncDecl)
if !ok || fn.Body == nil {
continue
}
names := map[string]bool{}
var fields []*ast.Field
if fn.Recv != nil {
fields = append(fields, fn.Recv.List...)
}
for _, fl := range append(fields, fn.Type.Params.List...) {
if typed(fl.Type) {
for _, n := range fl.Names {
names[n.Name] = true
}
}
}
reads := false
ast.Inspect(fn.Body, func(n ast.Node) bool {
if vs, ok := n.(*ast.ValueSpec); ok && vs.Type != nil && typed(vs.Type) {
for _, n := range vs.Names {
names[n.Name] = true
}
}
if s, ok := n.(*ast.SelectorExpr); ok && s.Sel.Name == "State" {
switch x := s.X.(type) {
case *ast.Ident:
reads = reads || names[x.Name]
case *ast.SelectorExpr:
reads = reads || x.Sel.Name == "Manifest" || x.Sel.Name == "Membership"
}
}
return true
})
if reads {
out = append(out, path+" "+fn.Name.Name)
}
}
return out, nil
}
// saidHistory is the seat mesh-broker's events a shared events stream holds, oldest first (grants.BrokerHistory).
type saidHistory []grants.SaidEvent
func (h saidHistory) Said() ([]grants.SaidEvent, error) { return h, nil }
// failingHistory is a re-read that fails partway: the events read so far, and the error.
type failingHistory struct{ read saidHistory }
func (h failingHistory) Said() ([]grants.SaidEvent, error) {
return h.read, errors.New("$JS.API.STREAM.MSG.GET.EVENTS at sequence 2: context deadline exceeded")
}
@@ -600,13 +600,18 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
t.Errorf("a one-filter consumer's grants = %v: its update is the no-filter create form", one)
}
for _, u := range users {
if u.Module != GrantsModule || u.Kind != KindModule {
continue
}
perms, err := PermissionsFor(u)
if err != nil {
t.Fatal(err)
}
// Issue 1966: the grants part alone reads the shared events stream by sequence, to read the seat mesh-broker's
// events again (grants_bindings_again).
if may := MayPublish(perms, "$JS.API.STREAM.MSG.GET.EVENTS"); may != (u.Module == GrantsModule && u.Kind == KindModule) {
t.Errorf("%s %s may read EVENTS by sequence: %v", u.Kind, u.Module, may)
}
if u.Module != GrantsModule || u.Kind != KindModule {
continue
}
if !MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_m") || MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_quiet") ||
MayPublish(perms, "$JS.API.CONSUMER.DELETE.EVENTS.a_m") || MayPublish(perms, "$JS.API.CONSUMER.CREATE.EVENTS.a_m.mesh.mod.x.event.happened") {
t.Errorf("the grants part creates consumers as %v", u.CreatesConsumers)
@@ -859,6 +859,10 @@ func PermissionsFor(p Principal) (Permissions, error) {
pub = append(pub, CreateConsumerGrants(p.CreatesConsumers)...)
// And its own object bucket, where it writes the user list sealed to the bus node (ADR 0547 clause 7).
pub = append(pub, GrantsUserListWrites()...)
// And the read of one message of the shared events stream by its sequence and subject, where the part reads the
// seat mesh-broker's created, recreated and dropped again for grants_bindings_again (issue 1966): a get, never a
// consumer, so no other account's consumer is touched.
pub = append(pub, SeatEventsReadAgain)
}
// The view alone reads the build seats' workers for the backlog of asks (issue 1887 A8a).
if p.Module == ViewModule {
@@ -1546,6 +1550,10 @@ func CreateConsumerGrants(consumers []ModuleConsumer) []string {
return out
}
// SeatEventsReadAgain is the grants part's read of the shared events stream by sequence (STREAM.MSG.GET with
// next_by_subj): a binding the bus module said that the part did not keep is read again from it (issue 1966).
const SeatEventsReadAgain = "$JS.API.STREAM.MSG.GET." + EventsStream
// GrantsModule is the one module whose principal publishes the modules' memberships (novox/hq ADR 0547 §5): the grants
// part, which composes them from the manifests and assignments. It defines no stream and is granted no other subject here.
const GrantsModule = "mesh-grants"
File diff suppressed because one or more lines are too long
+263 -6
View File
@@ -2,10 +2,15 @@ package grants
import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
stdio "git.novox.be/novox/mesh-sdk/go"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/membership"
"mesh-grants/internal/buscompose"
@@ -93,7 +98,18 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b
if e.Module == "" || local == "" || strings.Contains(e.Module, ".") {
return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local)
}
b, err := p.bound(e.Module)
// Written at the revision read, and read again once when another write came first (two `created` of one module heard
// at once): a binding is never lost to a write it did not see (issue 1966).
changed, err := p.keepBinding(e, name, local, storage)
if errors.Is(err, stdio.ErrChanged) {
changed, err = p.keepBinding(e, name, local, storage)
}
return changed, err
}
// keepBinding applies one event to the kept bindings of its module, written at the revision it read them at.
func (p *Part) keepBinding(e BrokerEvent, name, local, storage string) (bool, error) {
b, rev, err := p.boundAt(e.Module)
if err != nil {
return false, err
}
@@ -119,7 +135,7 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b
b.Bindings[local] = bind
}
raw, _ := json.Marshal(b)
if _, err := p.Store.Put(StateMemberships, BoundKey(e.Module), raw); err != nil {
if _, err := p.Store.Update(StateMemberships, BoundKey(e.Module), raw, rev); err != nil {
return false, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(e.Module), err)
}
return true, nil
@@ -127,20 +143,26 @@ func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (b
// bound reads the kept bindings of module; none kept is an empty value.
func (p *Part) bound(module string) (Bound, error) {
b, _, err := p.boundAt(module)
return b, err
}
// boundAt is bound with the revision it was read at, 0 when none is kept.
func (p *Part) boundAt(module string) (Bound, uint64, error) {
b := Bound{V: 1, Module: module, Bindings: map[string]membership.BusBinding{}}
raw, _, found, err := p.Store.Get(StateMemberships, BoundKey(module))
raw, rev, found, err := p.Store.Get(StateMemberships, BoundKey(module))
if err != nil {
return b, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err)
return b, 0, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err)
}
if found {
if err := json.Unmarshal(raw, &b); err != nil {
return b, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err)
return b, 0, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err)
}
if b.Bindings == nil {
b.Bindings = map[string]membership.BusBinding{}
}
}
return b, nil
return b, rev, nil
}
// bindingsOf is module's kept bindings in local-name order, for its memberships; nil when none.
@@ -161,6 +183,241 @@ func (p *Part) bindingsOf(module string) ([]membership.BusBinding, error) {
return out, nil
}
// BindingsOf answers the tool grants_bindings: every binding the part keeps of one module, or of every module when module
// is empty (its own state memberships, key BoundKey), each beside the memberships `<node>.<module>` it wrote, and by
// membership the kept local names its bindings lack, so a binding kept but not yet written into a membership is named.
func (p *Part) BindingsOf(module string) (any, error) {
keys, err := p.Store.Keys(StateMemberships)
if err != nil {
return nil, fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
}
type kept struct {
Module string `json:"module"`
Bindings []membership.BusBinding `json:"bindings"`
Memberships map[string][]string `json:"memberships"`
Lacking map[string][]string `json:"lacking,omitempty"`
}
out := []kept{}
for _, key := range keys {
mod, ok := strings.CutPrefix(key, BoundPrefix)
if !ok || (module != "" && mod != module) {
continue
}
binds, err := p.bindingsOf(mod)
if err != nil {
return nil, err
}
k := kept{Module: mod, Bindings: binds, Memberships: map[string][]string{}}
if k.Bindings == nil {
k.Bindings = []membership.BusBinding{}
}
for _, mk := range keys {
if _, of, _ := strings.Cut(mk, "."); strings.HasPrefix(mk, BoundPrefix) || of != mod {
continue
}
raw, _, found, err := p.Store.Get(StateMemberships, mk)
if err != nil {
return nil, fmt.Errorf("its own membership %s cannot be read: %w", mk, err)
}
if !found {
continue
}
bare, err := membership.BareGrantsMembership(raw)
if err != nil {
return nil, fmt.Errorf("its own membership %s does not read: %w", mk, err)
}
var m membership.Membership
if err := json.Unmarshal(bare, &m); err != nil {
return nil, fmt.Errorf("its own membership %s does not read: %w", mk, err)
}
written := []string{}
for _, b := range m.Bindings {
written = append(written, b.Local)
}
k.Memberships[mk] = written
for _, b := range binds {
if !slices.Contains(written, b.Local) {
if k.Lacking == nil {
k.Lacking = map[string][]string{}
}
k.Lacking[mk] = append(k.Lacking[mk], b.Local)
}
}
}
out = append(out, k)
}
again, err := p.lastBindingsAgain(module)
if err != nil {
return nil, err
}
answer := map[string]any{"modules": out,
"says": "each binding the part kept from the bus module's created and recreated, beside the memberships it wrote; lacking names a kept binding a membership does not carry yet"}
if module != "" && len(out) == 0 {
answer["says"] = "the part keeps no binding of " + module + ": the bus module said no created for it that the part heard"
}
if again != nil {
answer["read-again"] = again
}
return answer, nil
}
// **A binding the part did not keep is read again from the shared events stream** (issue 1966; the lead's ruling, option
// b): grants_bindings_again asks the part's process, by a person's act recorded as a hand act, to read the seat
// mesh-broker's events the stream still holds, oldest first, and to keep each of one module's created, recreated and
// dropped through HearBroker, as if heard now, so a binding dropped since is not brought back. The stream keeps its events
// for a bounded age: a created older than that is not there, and the answer says so, naming the path then (the seat's
// list answering each entry's binding, a record extending ADR 0595 §7).
// KeyBindingsAgain is, in StateUserList, the last re-read of one module's bindings: BindingsAgain.
const KeyBindingsAgain = "bindings-again"
// BrokerHistory reads the seat mesh-broker's events again from the shared events stream.
type BrokerHistory interface {
// Said is every event of the seat the stream still holds, oldest first.
Said() ([]SaidEvent, error)
}
// SaidEvent is one event of the seat mesh-broker as the shared events stream holds it.
type SaidEvent struct {
Subject string `json:"subject"`
Data []byte `json:"data"`
At time.Time `json:"at"`
}
// BindingsAgain is what the last re-read of one module's bindings found: the seat's events read, the local names whose
// binding it kept (Kept), those it found already kept as said (Already), those a dropped took away (Dropped), the oldest
// event the stream still holds, and in plain words what it means.
type BindingsAgain struct {
V int `json:"v"`
Module string `json:"module"`
By string `json:"by"`
At time.Time `json:"at"`
Read int `json:"read"`
Kept []string `json:"kept"`
Already []string `json:"already"`
Dropped []string `json:"dropped"`
Oldest time.Time `json:"oldest,omitzero"`
Says string `json:"says"`
Error string `json:"error,omitempty"`
}
// bindingsAgain reads module's bindings again from the seat's events the stream holds, keeps each, and writes what it
// found to its own state; an error is said in the pass's errors and in the kept result.
func (p *Part) bindingsAgain(res *Result, module, by string) {
out := BindingsAgain{V: 1, Module: module, By: by, At: p.now(), Kept: []string{}, Already: []string{}, Dropped: []string{}}
events, err := p.History.Said()
if err != nil {
// A read that failed partway applies nothing: a created whose dropped lies past the failure would be kept again.
out.Error = "the shared events stream could not be read, so nothing was kept: " + err.Error()
events = nil
}
heardCreate := false
for _, e := range events {
if out.Oldest.IsZero() || (!e.At.IsZero() && e.At.Before(out.Oldest)) {
out.Oldest = e.At
}
name, ok := brokerEvent(e.Subject, p.Brokers())
if !ok {
continue
}
var said BrokerEvent
if json.Unmarshal(e.Data, &said) != nil || said.Module != module {
continue
}
out.Read++
local := firstOf(said.LocalName, said.Local, said.Binding.Local)
changed, err := p.HearBroker(e.Subject, e.Data, p.Brokers())
if err != nil {
out.Error = fmt.Sprintf("the %s of %s could not be kept: %v", name, local, err)
break
}
switch {
case name == BrokerDropped && changed:
out.Dropped = appendOnce(out.Dropped, local)
out.Kept = slices.DeleteFunc(out.Kept, func(l string) bool { return l == local })
case name != BrokerDropped:
heardCreate = true
out.Dropped = slices.DeleteFunc(out.Dropped, func(l string) bool { return l == local })
if changed {
out.Kept = appendOnce(out.Kept, local)
out.Already = slices.DeleteFunc(out.Already, func(l string) bool { return l == local })
} else if !slices.Contains(out.Kept, local) {
out.Already = appendOnce(out.Already, local)
}
}
}
switch {
case out.Error != "":
out.Says = "the re-read of " + module + " stopped: " + out.Error
res.Errors = append(res.Errors, out.Says)
case !heardCreate:
out.Says = fmt.Sprintf("the shared events stream holds no created or recreated of %s (its oldest event: %s): a created older "+
"than the stream keeps is not there, and its binding comes back only by the seat mesh-broker's list answering each "+
"entry's binding (a record extending ADR 0595 §7), not by this tool", module, out.Oldest.Format(time.RFC3339))
default:
out.Says = fmt.Sprintf("read %d of the seat mesh-broker's events of %s again: kept %v, already kept %v, dropped %v; the pass "+
"writes them into its memberships", out.Read, module, out.Kept, out.Already, out.Dropped)
}
if p.Logf != nil {
p.Logf("the bindings of %s read again by hand, for %s: %s", module, by, out.Says)
}
raw, _ := json.Marshal(out)
p.changed(res, StateUserList, KeyBindingsAgain, raw)
}
func appendOnce(list []string, s string) []string {
if slices.Contains(list, s) {
return list
}
return append(list, s)
}
// AskBindingsAgain answers the tool grants_bindings_again: it asks the part's process to read one module's bindings
// again from the seat mesh-broker's events the shared events stream holds, and records the hand act through record (the
// seat mesh-hand-acts) first; a module that is no module's name, or an ask without why, is refused before anything is
// asked. grants_bindings shows what the process found.
func (p *Part) AskBindingsAgain(caller, module, why string, record func(what, why string) error) (any, error) {
if module == "" || strings.ContainsAny(module, ".*> ") {
return nil, fmt.Errorf("grants_bindings_again needs one module, by its name: %q is none", module)
}
if strings.TrimSpace(why) == "" {
return nil, errors.New("grants_bindings_again needs why: it is recorded as a hand act")
}
if record != nil {
if err := record(ToolBindingsAgain+" "+module, why); err != nil {
return nil, fmt.Errorf("the hand act could not be recorded, so nothing was asked: %w", err)
}
}
ask, _ := json.Marshal(Ask{By: caller, At: p.now(), Bindings: module})
rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
if err != nil {
return nil, fmt.Errorf("the ask could not be written: %w", err)
}
return map[string]any{"asked": rev, "module": module,
"says": "the part's process reads the seat mesh-broker's created, recreated and dropped of " + module + " again from the shared " +
"events stream at this ask, and keeps each as if heard now; grants_bindings shows what it kept, what was already kept, " +
"and whether the stream still holds its created"}, nil
}
// lastBindingsAgain is the last re-read kept, when it was of module (or any, for an empty module).
func (p *Part) lastBindingsAgain(module string) (*BindingsAgain, error) {
raw, _, found, err := p.Store.Get(StateUserList, KeyBindingsAgain)
if err != nil {
return nil, fmt.Errorf("its own %s %s cannot be read: %w", StateUserList, KeyBindingsAgain, err)
}
if !found {
return nil, nil
}
var b BindingsAgain
if err := json.Unmarshal(raw, &b); err != nil {
return nil, fmt.Errorf("its own %s %s does not read: %w", StateUserList, KeyBindingsAgain, err)
}
if module != "" && b.Module != module {
return nil, nil
}
return &b, nil
}
func sameBinding(a, b membership.BusBinding) bool {
ra, _ := json.Marshal(a)
rb, _ := json.Marshal(b)
@@ -35,7 +35,8 @@ type Follower struct {
// Hear hears the keeper's condition events and the vault's account-issued (Consumes) and hands each, by event name,
// with its body, to handle; nil hears none. A condition of the kind root-not-free raised or cleared wakes a pass:
// a machine's root-free word changed; and so does an account the vault minted (ADR 0561 §1), whose hash the list carries.
Hear func(handle func(event string, body []byte)) error
// handle answers an event it could not apply, which the hearer then leaves unacknowledged, so it is said again (issue 1966).
Hear func(handle func(event string, body []byte) error) error
wake chan struct{}
once sync.Once
@@ -127,25 +128,27 @@ func (f *Follower) open() {
}
}
if f.Hear != nil && !f.heard_ {
err := f.Hear(func(event string, body []byte) {
err := f.Hear(func(event string, body []byte) error {
// The bus module's created, recreated or dropped: the binding it carries is kept, and a pass composes the
// memberships with it (bindings.go).
// memberships with it (bindings.go). One not applied is answered, so the hearer leaves it unacknowledged and the
// bus says it again (issue 1966: acknowledged, it was lost while this line said it would come again).
if changed, err := f.Part.HearBroker(event, body, f.Part.Brokers()); err != nil {
f.logf("%v: not applied, said again at its redelivery", err)
return
return err
} else if changed {
f.Wake()
return
return nil
}
// An account the vault minted.
if IsAccountIssued(event) {
f.Wake()
return
return nil
}
// A condition event is heard from the keeper alone (ADR 0590 §3).
if FromKeeper(event) && IsRootNotFree(body) {
f.Wake()
}
return nil
})
if err != nil {
f.logf("the keeper's condition events cannot be heard yet (%v): tried again at the next change", err)
+18 -4
View File
@@ -57,6 +57,10 @@ type Store interface {
// Keys and Delete are for the wanted accounts (accounts.go) and the grants (grants.go): every other state is only put.
Keys(state string) ([]string, error)
Delete(state, key string) error
// Update writes a key only while it is still at revision (0: only while it holds nothing), and refuses one changed
// meanwhile with an error wrapping stdio.ErrChanged: the kept bindings are written so, never lost to a second writer
// (issue 1966).
Update(state, key string, value []byte, revision uint64) (uint64, error)
}
// Emitter says one of the part's events.
@@ -170,6 +174,9 @@ type Part struct {
// Say says a condition to the operator: the process's; nil in the tools.
Say func(oc.Condition) error
Now func() time.Time
// History reads the seat mesh-broker's events again from the shared events stream (bindings.go): the process's
// connection; nil in the tools, which ask the process.
History BrokerHistory
// SeatList is the seat list reader, which gates a module whose seats the list has not named (inputs.go, SeatGate); nil
// gates none.
SeatList SeatGate
@@ -254,6 +261,9 @@ type Ask struct {
Consumer string `json:"consumer,omitempty"`
// FirstView is, from grants_first_view_again, one first view to ask once more by hand (its state), or AllFirstViews.
FirstView string `json:"first-view,omitempty"`
// Bindings is, from grants_bindings_again, one module whose bindings the part reads again from the seat mesh-broker's
// events the shared events stream still holds (bindings.go, issue 1966).
Bindings string `json:"bindings,omitempty"`
}
// AllFirstViews asks every first view once more.
@@ -294,6 +304,14 @@ func (p *Part) Compose() Result {
p.Views.SayUnsaid()
in := p.inputs()
p.keepBrokers(in)
var ask Ask
if p.Create != nil || p.Consumers != nil {
ask = p.asked()
}
// A module's bindings read again by hand are kept before the pass reads the kept bindings, so it writes them now.
if ask.Bindings != "" && p.History != nil {
p.bindingsAgain(&res, ask.Bindings, ask.By)
}
if err := p.withBound(&in); err != nil {
res.Errors = append(res.Errors, err.Error())
}
@@ -489,10 +507,6 @@ func (p *Part) Compose() Result {
}
p.clearFull()
held, heldErr := HeldStreams(p.Call)
var ask Ask
if p.Create != nil || p.Consumers != nil {
ask = p.asked()
}
forced := ask.Assert
for _, module := range sortedKeys(in.Manifests) {
if !assigned(in, module) {
+5 -1
View File
@@ -31,11 +31,15 @@ const (
// ToolSeatList and ToolSeatListAgain read and read again the seat list the part composes from (mesh-sdk seatlist.Tools).
ToolSeatList = "grants_seat_list"
ToolSeatListAgain = "grants_seat_list_again"
// ToolBindings reads the bindings the part keeps from the bus module's events (bindings.go, issue 1966).
ToolBindings = "grants_bindings"
// ToolBindingsAgain reads one module's bindings again from the seat mesh-broker's events the shared events stream holds.
ToolBindingsAgain = "grants_bindings_again"
)
// Tools is every tool, in the manifest's order.
var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolAccounts, ToolConsumerMatch,
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolSeatList, ToolSeatListAgain}
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolSeatList, ToolSeatListAgain, ToolBindings, ToolBindingsAgain}
// WritesGrants is what grants says of the part's own grants: written once the list the bus loaded carries a build's
// accounts, and the one key the assigner reads (ADR 0571 clause 9, which changes ADR 0547 clause 5).
+3 -1
View File
@@ -61,7 +61,9 @@
"grants_first_views",
"grants_first_view_again",
"grants_seat_list",
"grants_seat_list_again"
"grants_seat_list_again",
"grants_bindings",
"grants_bindings_again"
],
"state": [
{