Compare commits

...
7 Commits
Author SHA1 Message Date
jschoubben ecad0c688b A re-read of the seat events that fails partway keeps nothing it read (issue 1966)
mesh/delivery ready: it is merged on the Approve
mesh/check pass ecad0c688b48 on fc21f2d1bd56 | carried from mesh-catalog 7630e3cc9d1e | unit 500 run, 1003 kept | integration 499 run, 1070 kept | com…
A created whose dropped lay past the failure was kept again. When the
stream cannot be read whole, only the error is written to bindings-again.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:22:18 +01:00
jschoubben aa89032f02 A tool reads one module's bindings again from the seat events the shared stream still holds (issue 1966)
A binding the bus module said created that the part did not keep had no
recovery: distribution's holds, windows, mirrors and events are missing on
the bus node's memberships. grants_bindings_again, a hand act, asks the
process to read the seat mesh-broker's created, recreated and dropped of
one module again from EVENTS by sequence and keep each through HearBroker
in order, so a dropped entry stays dropped. grants_bindings shows what it
kept, and says when the stream no longer holds a created: then the seat's
list answering bindings is the path (the record extending ADR 0595 §7).
The part alone is granted the stream's message get, no consumer.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:11:26 +01:00
jschoubben 44422f21db Prove on a server under the composed grants that a module writes its bound bucket (issue 1966)
The integration case after created proved only that the binding is in the
membership, never that the account can use it (design 25 § 12.7). Under the
list the composer writes, the keeper creates mod-c_kept, mod-c's own account
binds and writes it, and another module reads nothing of it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:01:41 +01:00
jschoubben 67e1c7594f A tool reads the bindings the grants part keeps, beside the memberships it wrote (issue 1966)
bound.<module> had no reader: a binding lost or kept but not yet written into
a membership could only be guessed from grants_membership. grants_bindings
names each kept binding and, by membership, the ones it lacks.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:01:41 +01:00
jschoubben 4d75646bf4 Guard that kept and issued state is read only through mesh-sdk's two rules (issue 1966)
A reader of a manifest's or a membership's State beside catalogue.KeptStates
and Membership.StateIssued misses a key-value entry asked in requires, the
kind of this issue's faults. The row reads the repository's syntax trees and
allows only the rules that judge or compose the old section itself.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:01:41 +01:00
jschoubben c2e7249897 Keep every binding the bus module says created, and never acknowledge one not kept
distribution's membership carried 4 of its 8 bindings and mesh-knowledge's 3
of 6 (issue 1966). The grants part wrote a module's kept bindings with a blind
put, so two created heard at once lost one, and acknowledged every seat event
whatever its handler answered, while the log said it would come again. The
bindings are now written at the revision read and read again once on a
conflict, and an event not applied is left unacknowledged (a delayed nak).

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:01:41 +01:00
jschoubben 35c3f5dd2f The manifest validator reads kept state in both forms; per-machine in requires is refused
A process's bus.state, a seat contribution and a seat's records naming a
key-value entry asked in requires were refused, where the old state section
is accepted (issue 1966, ADR 0595 §11). They now read catalogue.KeptStates.
A per-machine key-value entry in requires is refused with its reason: its
binding names no machine's key, so no machine could be held to its own.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-09 14:01:41 +01:00
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": [
{