diff --git a/node-tools/cmd/node-tools/approvals.go b/node-tools/cmd/node-tools/approvals.go index 48dfe63..99153d6 100644 --- a/node-tools/cmd/node-tools/approvals.go +++ b/node-tools/cmd/node-tools/approvals.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "errors" "fmt" @@ -149,6 +150,15 @@ func warranted(conn *bus.Conn) func(asker, ask string, open func() (bool, error) } } +// hearOne waits on the runner's connection for the first event on the subjects whose body match takes (design 34 +// every-event §36: a marked wait's ask's decided). +func hearOne(conn *bus.Conn) func(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (bool, error) { + return func(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (bool, error) { + _, ok, _, err := conn.HearFirst(ctx, subjects, within, match) + return ok, err + } +} + // gateTools is the gate as mesh_approvals reads it, with its state's fill read from the bucket (§13, §18). type gateTools struct { *approval.Gate diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 8a81f04..302afde 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -80,7 +80,9 @@ func main() { gate := approval.New(approval.Deps{Node: conn.Node(), Runner: cred.Module, Catalog: source.ApprovalCatalog(definitionsOf(seatList)), Wait: source.WaitCatalog, Findings: source.Findings(), Logf: log.Printf, State: approvalsState{conn: conn, module: cred.Module}, Publish: publishAsk(conn), Record: routerRecord(conn), Warranted: warranted(conn), Bounds: boundState{conn: conn, module: cred.Module}, - Holder: approval.HoldersIn(func() (map[string][]byte, error) { return source.State(firstview.SeatHoldersRead) })}) + Holder: approval.HoldersIn(func() (map[string][]byte, error) { return source.State(firstview.SeatHoldersRead) }), + // The runner's own build, the one a marked wait's act binds (design 34 every-event §35). + Build: os.Getenv(runtime.FingerprintWord), Hear: hearOne(conn)}) // A failed read or write of the state approvals is made again at the next membership change the runner hears (A-157). conn.OnMembership(func(bus.Membership) { go gate.StateAgain() }) opts = append(opts, runtime.WithGate(gate)) diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index 230cd1e..f1cf18f 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -3,7 +3,6 @@ package approval import ( "encoding/json" "errors" - "sort" "strings" "sync" "testing" @@ -15,83 +14,14 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/seatdef" + + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) -// mem is the runner's state approvals as a fixture; fail makes every read and write fail, as a bus refusing them. -type mem struct { - mu sync.Mutex - vals map[string]json.RawMessage - revs map[string]uint64 - n uint64 - fail error -} +// mem is the runner's state approvals as a fixture (meshtest.State). +type mem = mt.State -func newMem() *mem { return &mem{vals: map[string]json.RawMessage{}, revs: map[string]uint64{}} } - -func (m *mem) Get(k string) (*stdio.StateEntry, error) { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return nil, m.fail - } - v, ok := m.vals[k] - if !ok { - return nil, nil - } - return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil -} -func (m *mem) Create(k string, v any) (uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return 0, m.fail - } - if _, ok := m.vals[k]; ok { - return 0, stdio.ErrChanged - } - return m.put(k, v) -} -func (m *mem) Update(k string, v any, rev uint64) (uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return 0, m.fail - } - if rev == 0 || m.revs[k] != rev { - return 0, stdio.ErrChanged - } - return m.put(k, v) -} -func (m *mem) Put(k string, v any) (uint64, error) { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return 0, m.fail - } - return m.put(k, v) -} -func (m *mem) Keys() ([]string, error) { - m.mu.Lock() - defer m.mu.Unlock() - if m.fail != nil { - return nil, m.fail - } - var out []string - for k := range m.vals { - out = append(out, k) - } - sort.Strings(out) - return out, nil -} -func (m *mem) put(k string, v any) (uint64, error) { - raw, ok := v.(json.RawMessage) - if !ok { - raw, _ = json.Marshal(v) - } - m.n++ - m.vals[k], m.revs[k] = raw, m.n - return m.n, nil -} +func newMem() *mem { return mt.NewState() } // world is a gate on a fixture mesh: the catalog half, the router's records as the router writes them (mesh-catalog's // messenger, asks.go), the asks published, the clock. @@ -262,11 +192,11 @@ func theActAndItsAsk(t *testing.T) { {strings.Repeat("x", 50), "n1", "Run " + strings.Repeat("x", 50) + "?"}, {strings.Repeat("y", 80), "n1", "Run " + strings.Repeat("y", 54) + "…?"}, } { - if got := headlineOf(tc.name, tc.node, true); got != tc.want || len([]rune(got)) > 60 { + if got := headlineOf("Run", tc.name, tc.node, true); got != tc.want || len([]rune(got)) > 60 { t.Errorf("headline %q, want %q", got, tc.want) } } - if got := headlineOf("wipe", "n1", false); got != "Run wipe on n1? Part shown" { + if got := headlineOf("Run", "wipe", "n1", false); got != "Run wipe on n1? Part shown" { t.Errorf("headline %q", got) } } @@ -314,9 +244,9 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Errorf("explanation %q", a.Explanation) } var k operatorchannelclient.Kept - _ = json.Unmarshal(w.state.vals["n1.ask."+a.ID], &k) - if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.vals["n1.act."+d]) != `"`+a.ID+`"` { - t.Errorf("kept %+v, act key %s", k, w.state.vals["n1.act."+d]) + _ = json.Unmarshal(w.state.Vals["n1.ask."+a.ID], &k) + if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.Vals["n1.act."+d]) != `"`+a.ID+`"` { + t.Errorf("kept %+v, act key %s", k, w.state.Vals["n1.act."+d]) } }}, {"§12: with no record within 5 seconds, Not run, asked; not yet taken by the router", func(t *testing.T, w *world) { @@ -385,10 +315,10 @@ func TestTheRefusalsAndAnswers(t *testing.T) { rec.Warrant.AskDigest = operatorchannelclient.Digest(a) w.records["shell."+a.ID] = rec var k operatorchannelclient.Kept - _ = json.Unmarshal(w.state.vals["n1.ask."+a.ID], &k) + _ = json.Unmarshal(w.state.Vals["n1.ask."+a.ID], &k) k.Ask = a raw, _ := json.Marshal(k) - w.state.vals["n1.ask."+a.ID] = raw + w.state.Vals["n1.ask."+a.ID] = raw if _, err := w.do(call("wipe", `{"dir":"scratch"}`)); err == nil || w.runs != 0 || !strings.Contains(err.Error(), "nothing is done") { t.Fatalf("%v, runs %d", err, w.runs) } @@ -495,7 +425,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("recorded %v, findings %v, bounds %v", w.recorded, w.findings.raised, w.bounds.keys) } var k operatorchannelclient.Kept - _ = json.Unmarshal(w.state.vals["n1.ask."+a.ID], &k) + _ = json.Unmarshal(w.state.Vals["n1.ask."+a.ID], &k) if !k.Recorded || !k.Acted { t.Fatalf("kept %+v", k) } @@ -517,7 +447,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("%v", err) } w.g.d.Publish = pub - w.state.vals = map[string]json.RawMessage{} + w.state.Vals = map[string]json.RawMessage{} _, _ = w.do(call("wipe", `{}`)) w.g.d.Record = func(string, string) (*stdio.StateEntry, error) { return nil, errors.New("timeout") } if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "the router's record of its ask") || w.runs != 0 { @@ -546,7 +476,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { {"§19: the state failing refuses, raises approvals-unreadable after five minutes, and clears when read again", func(t *testing.T, w *world) { stop, _ := w.g.WatchBounds() defer stop() - w.state.fail = errors.New("the bus is unreachable") + w.state.Fail = errors.New("the bus is unreachable") if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "cannot be read") { t.Fatalf("%v", err) } @@ -557,7 +487,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { if w.findings.raised[keys.EscapeName("n1.approvals")] != KindApprovalsUnreadable { t.Fatalf("findings %v", w.findings.raised) } - w.state.fail = nil + w.state.Fail = nil w.g.StateAgain() // a membership change heard, no further call if len(w.findings.raised) != 0 || len(w.bounds.keys) != 0 { t.Fatalf("findings %v, bounds %v", w.findings.raised, w.bounds.keys) diff --git a/node-tools/internal/approval/ask.go b/node-tools/internal/approval/ask.go index 43e1bc9..60a27da 100644 --- a/node-tools/internal/approval/ask.go +++ b/node-tools/internal/approval/ask.go @@ -70,7 +70,7 @@ func (g *Gate) askFor(c Call, digest, id string) seat.Ask { if c.Seat != "" { name = c.Seat + "." + c.Name } - headline := headlineOf(name, g.d.Node, whole) + headline := headlineOf("Run", name, g.d.Node, whole) first, _, _ := strings.Cut(strings.TrimSpace(c.Description), "\n") if first == "" { first = name @@ -92,22 +92,23 @@ func (g *Gate) askFor(c Call, digest, id string) seat.Ask { Level: operatorchannelclient.Approve, Binds: digest}}} } -// headlineOf is "Run on ?", "Run ?" when that is longer than the router allows, and the names cut with -// "…" when that too is longer; ending "Part shown" whenever an argument is not shown whole (§10, Q-HL). -func headlineOf(name, node string, whole bool) string { +// headlineOf is " on ?", " ?" when that is longer than the router allows, and the names cut +// with "…" when that too is longer; ending "Part shown" whenever an argument is not shown whole (§10, Q-HL). verb is "Run" +// for a call and "Wait on" for a wait (design 34 every-event §35). +func headlineOf(verb, name, node string, whole bool) string { suffix := "" if !whole { suffix = " Part shown" } max := operatorchannelclient.HeadlineLength fits := func(s string) bool { return len([]rune(s)) <= max } - if h := "Run " + name + " on " + node + "?" + suffix; fits(h) { + if h := verb + " " + name + " on " + node + "?" + suffix; fits(h) { return h } - if h := "Run " + name + "?" + suffix; fits(h) { + if h := verb + " " + name + "?" + suffix; fits(h) { return h } - room := max - len([]rune("Run …?"+suffix)) + room := max - len([]rune(verb+" …?"+suffix)) r := []rune(name) if room < 1 { room = 1 @@ -115,7 +116,7 @@ func headlineOf(name, node string, whole bool) string { if len(r) > room { r = r[:room] } - return "Run " + string(r) + "…?" + suffix + return verb + " " + string(r) + "…?" + suffix } // firstCall is a marked call whose act has no ask open: refused at once when its arguments may not be asked (§9), else @@ -140,7 +141,15 @@ func (g *Gate) firstCall(c Call) (any, error) { // the first answer (§12), always an error, never a result. func (g *Gate) raise(c Call, act operatorchannelclient.Act, digest, markedSince string) error { id := newID() - a := g.askFor(c, digest, id) + if err := g.keepAsk(c, act, digest, g.askFor(c, digest, id)); err != nil { + return err + } + return g.firstAnswer(c, id, digest, markedSince) +} + +// keepAsk raises an ask and keeps it in the runner's state with the act and the whole arguments, its act's key naming it +// (§10, §13); refused, as an error, when it cannot be. +func (g *Gate) keepAsk(c Call, act operatorchannelclient.Act, digest string, a seat.Ask) error { cl := g.client(c.Module) if size := keptSize(a, act, c.Args); size > MostKeptValue { return g.refused(c, digest, fmt.Sprintf("its ask would keep %d bytes, more than the %d a value of the state %s holds", size, MostKeptValue, StateName)) @@ -148,15 +157,15 @@ func (g *Gate) raise(c Call, act operatorchannelclient.Act, digest, markedSince if err := cl.Ask(a); err != nil { return g.refused(c, digest, "the ask could not be raised: "+err.Error()) } - if err := g.keepCall(id, act, c.Args); err != nil { + if err := g.keepCall(a.ID, act, c.Args); err != nil { g.stateFailed(err) return g.refused(c, digest, "the ask was raised and its call could not be kept: "+err.Error()) } - if _, err := g.d.State.Put(g.actKey(digest), id); err != nil { + if _, err := g.d.State.Put(g.actKey(digest), a.ID); err != nil { g.stateFailed(err) return g.refused(c, digest, "the ask was raised and its act could not be kept: "+err.Error()) } - return g.firstAnswer(c, id, digest, markedSince) + return nil } // keptSize is the bytes the kept ask would take. diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go index 787014a..315db22 100644 --- a/node-tools/internal/approval/gate.go +++ b/node-tools/internal/approval/gate.go @@ -2,6 +2,7 @@ package approval import ( "context" + "encoding/json" "fmt" "sort" "strings" @@ -55,6 +56,12 @@ type Deps struct { Bounds Bounds // Holder says whether a module holds a seat, from the assigner's view of the seats' holders the runner follows (§19). Holder func(seat string) (bool, error) + // Build is the tool runner's own fingerprint, read from its MESH_FINGERPRINT at its start: the build a marked wait's + // act binds; empty while unknown, when every marked wait is refused (design 34 every-event §35). + Build string + // Hear waits within the bound for the first event on the subjects whose body match takes, or the caller's going: a + // marked wait's ask's `decided` (every-event §36); false when none came. + Hear func(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (bool, error) // Go runs what goes on beside the call (a run's recording); nil is a goroutine. Go func(func()) } diff --git a/node-tools/internal/approval/marks.go b/node-tools/internal/approval/marks.go index a61d59a..824ef6e 100644 --- a/node-tools/internal/approval/marks.go +++ b/node-tools/internal/approval/marks.go @@ -29,6 +29,9 @@ type Catalog interface { Newest(module string) (catalogseat.CatalogEntry, bool) // Definitions are the seat definitions the catalog half carries, by seat. Definitions() map[string]seatdef.Definition + // Entries are the newest entry of every module in every followed branch's half, a marked wait's marks (lead answer + // A-168). + Entries() []catalogseat.CatalogEntry } // Marks is what one module's calls are gated by: its marked tools and seat verbs, and the subjects of its marked events and @@ -89,61 +92,67 @@ func Derive(c Catalog, module, fingerprint string) Marks { entries = append(entries, e) } for _, e := range entries { - source := "entry " + e.Module + " " + orNone(e.Fp) - var said marked - if raw, err := json.Marshal(e.Manifest); err == nil { - _ = json.Unmarshal(raw, &said) - } - serves := map[string]bool{} - for _, sv := range e.Serves { - serves[sv] = true - seat, verb, _ := strings.Cut(sv, ".") - for _, v := range defs[seat].Verbs { - if v.Name == verb && v.NeedsApproval { - m.Verbs[sv] = "seat " + seat - } + m.add(e, defs) + } + return m +} + +// add takes one entry's marks into m: its `needs-approval` (a tool, a verb it serves or an event it declares), the seat +// definitions' marks of every verb it serves, and every stream it declares with a `grant`, each with its source. +func (m *Marks) add(e catalogseat.CatalogEntry, defs map[string]seatdef.Definition) { + source := "entry " + e.Module + " " + orNone(e.Fp) + var said marked + if raw, err := json.Marshal(e.Manifest); err == nil { + _ = json.Unmarshal(raw, &said) + } + serves := map[string]bool{} + for _, sv := range e.Serves { + serves[sv] = true + seat, verb, _ := strings.Cut(sv, ".") + for _, v := range defs[seat].Verbs { + if v.Name == verb && v.NeedsApproval { + m.Verbs[sv] = "seat " + seat } } - events := map[string]bool{} - for _, ev := range said.Emits { + } + events := map[string]bool{} + for _, ev := range said.Emits { + events[ev] = true + } + for _, s := range said.Streams { + for _, ev := range s.Events { events[ev] = true } - for _, s := range said.Streams { - for _, ev := range s.Events { - events[ev] = true + } + for _, name := range said.NeedsApproval { + switch { + case serves[name]: + if _, bySeat := m.Verbs[name]; !bySeat { + m.Verbs[name] = source + } + case events[name]: + for _, s := range eventSubjects(e.Module, name) { + m.Subjects[s] = source } + default: + m.Tools[name] = source + } + } + for _, s := range said.Streams { + if len(s.Grant) == 0 || string(s.Grant) == "null" { + continue } - for _, name := range said.NeedsApproval { - switch { - case serves[name]: - if _, bySeat := m.Verbs[name]; !bySeat { - m.Verbs[name] = source - } - case events[name]: - for _, s := range eventSubjects(e.Module, name) { - m.Subjects[s] = source - } - default: - m.Tools[name] = source - } + wire := make([]string, 0, len(s.Events)) + for _, ev := range s.Events { + wire = append(wire, buscompose.WirePattern(ev)) } - for _, s := range said.Streams { - if len(s.Grant) == 0 || string(s.Grant) == "null" { - continue - } - wire := make([]string, 0, len(s.Events)) - for _, ev := range s.Events { - wire = append(wire, buscompose.WirePattern(ev)) - } - for _, sub := range (buscompose.ModuleStream{Module: e.Module, Events: wire}).Subjects() { - m.Subjects[sub] = source - if !strings.HasSuffix(sub, ">") { - m.Subjects[sub+".>"] = source - } + for _, sub := range (buscompose.ModuleStream{Module: e.Module, Events: wire}).Subjects() { + m.Subjects[sub] = source + if !strings.HasSuffix(sub, ">") { + m.Subjects[sub+".>"] = source } } } - return m } // eventSubjects are a marked event's subjects on the bus (design 34 every-event §33): said on the module's events and on its diff --git a/node-tools/internal/approval/marks_test.go b/node-tools/internal/approval/marks_test.go index 85053f6..39573cf 100644 --- a/node-tools/internal/approval/marks_test.go +++ b/node-tools/internal/approval/marks_test.go @@ -32,6 +32,13 @@ func (c *fakeCatalog) Newest(m string) (catalogseat.CatalogEntry, bool) { return e, ok } func (c *fakeCatalog) Definitions() map[string]seatdef.Definition { return c.defs } +func (c *fakeCatalog) Entries() []catalogseat.CatalogEntry { + var out []catalogseat.CatalogEntry + for _, e := range c.newest { + out = append(out, e) + } + return out +} func entry(fp string, serves []string, manifest map[string]any) catalogseat.CatalogEntry { return catalogseat.CatalogEntry{Module: "shell", Fp: fp, Serves: serves, Manifest: manifest} diff --git a/node-tools/internal/approval/wait.go b/node-tools/internal/approval/wait.go new file mode 100644 index 0000000..7fa963c --- /dev/null +++ b/node-tools/internal/approval/wait.go @@ -0,0 +1,296 @@ +package approval + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/outward" +) + +// Wait is one mesh_wait call as the gate judges it (design 34 every-event §34 to §36): the event names given, the subjects +// they became (after `from`), the filter and the timeout. +type Wait struct { + Names []string + From string + Subjects []string + Filter map[string]any + Timeout time.Duration +} + +// Waited is a wait through the gate (every-event §34 to §36). An unmarked wait is heard at once by hear, given how long it +// may wait. A marked one is heard only on the operator's warrant for that one wait, its timeout running from then; with +// none it subscribes to nothing, its ask raised once under the runner's own module's name, and is answered as an error at +// its timeout, when its caller goes, or as soon as the router's record says the ask was chosen, never in a hot loop. c is +// the call as the console makes it: its address, its tool's name, its caller and its arguments as given. +func (g *Gate) Waited(ctx context.Context, c Call, w Wait, hear func(within time.Duration) (any, error)) (any, error) { + deadline := g.d.Now().Add(w.Timeout) + // A wait from a seat is never marked, and a seat's event never is (§32, §34). + if w.From == "seat" || !outsideSeats(w.Subjects) { + return hear(w.Timeout) + } + if g.d.Catalog == nil { + return nil, g.holdRefused(ctx, deadline, "this wait is refused: the tool runner reads no catalog half, so the marks are not read") + } + if err := g.d.Catalog.Ready(); err != nil { + // Before the marks are read nothing is asked: the wait waits for the first view within its own timeout and is + // then judged, and is refused at its timeout when it does not come (§34). + if g.d.Wait != nil { + err = g.waitCatalog(ctx, w.Timeout) + } + if err != nil { + return nil, g.holdRefused(ctx, deadline, fmt.Sprintf("this wait is refused at its timeout: the marks are not yet read (%v); "+ + "mesh_first_views says what the tool runner waits for", err)) + } + } + if !overlapsAny(w.Subjects, g.waitMarks()) { + return hear(deadline.Sub(g.d.Now())) + } + c.Module, c.Seat, c.Fingerprint = g.d.Runner, "", g.d.Build + if c.Fingerprint == "" { + return nil, g.holdRefused(ctx, deadline, "this wait needs the operator's approval, and the tool runner's own fingerprint is "+ + "unknown (its MESH_FINGERPRINT was not set at its start), so no warrant can bind its build") + } + if b, ok := g.d.State.(Binding); g.d.State == nil || ok && !b.Bound() { + return nil, g.holdRefused(ctx, deadline, g.refused(c, "", "the tool runner's membership binds no state "+StateName+" to keep its asks in").Error()) + } + if why := refusedUnasked(c.Args); why != "" { + return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error()) + } + if why := g.held(); why != "" { + return nil, g.holdRefused(ctx, deadline, g.refused(c, "", why).Error()) + } + act, err := ActOf(c, g.d.Node) + if err != nil { + return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error()) + } + digest, err := operatorchannelclient.ActDigest(act) + if err != nil { + return nil, g.holdRefused(ctx, deadline, g.refused(c, "", err.Error()).Error()) + } + named, err := g.d.State.Get(g.actKey(digest)) + if err != nil { + g.stateFailed(err) + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()).Error()) + } + g.stateRead() + anew := func() (any, error) { + id := newID() + if err := g.keepAsk(c, act, digest, g.askForWait(c, w, digest, id)); err != nil { + return nil, g.holdRefused(ctx, deadline, err.Error()) + } + return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, true) + } + if named == nil { + return anew() + } + var id string + if err := json.Unmarshal(named.Value, &id); err != nil || id == "" { + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, fmt.Sprintf("its act's key in %s names no ask", StateName)).Error()) + } + rec, err := g.record(c.Module, id) + switch { + case err != nil: + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" cannot be read: "+err.Error()).Error()) + case rec == nil, rec.State == "open", rec.State == "held-back": + return nil, g.holdAsk(ctx, deadline, c.Module, id, digest, false) + case rec.State == "refused": + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router refused its ask "+id+": "+orNone(rec.Answer)).Error()) + case rec.State != "chosen": + return anew() // expired, cancelled, replaced: asked again (§15, §17) + } + wt := rec.Warrant + if wt == nil { + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "the router's record of its ask "+id+" is chosen and carries no warrant").Error()) + } + at, _ := operatorchannelclient.Time(wt.At) + if wt.Option == OptionDecline { + if g.d.Now().Before(at.Add(DeclinedFor)) { + return nil, g.holdRefused(ctx, deadline, declinedWords(id, at)) + } + return anew() + } + if g.d.Now().After(at.Add(RunWithin)) { + return anew() + } + var answer any + var ran error + h, err := g.client(c.Module).Take(*wt, func(o seat.Option, _ seat.Warrant) error { + if err := operatorchannelclient.Performs(o, act); err != nil { + ran = err + return err + } + // On the warrant it subscribes, and the wait's timeout runs from then (§35). + answer, ran = hear(w.Timeout) + return nil + }) + switch { + case h == operatorchannelclient.Done: + return anew() // already run: one warrant is one wait (§35) + case h == operatorchannelclient.Refused: + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its warrant does not hold: "+errText(err)).Error()) + case h == "" && err != nil: + g.stateFailed(err) + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, "its state "+StateName+" cannot be written: "+err.Error()).Error()) + case err != nil: + return nil, g.holdRefused(ctx, deadline, g.refused(c, digest, err.Error()).Error()) + } + return g.ran(c, id, rec, answer, ran) +} + +// outsideSeats says a subject is not a seat's: only such a subject can be marked (§32, §34). +func outsideSeats(subjects []string) bool { + for _, s := range subjects { + if !strings.HasPrefix(s, "mesh.seat.") { + return true + } + } + return false +} + +// waitMarks are the marked subjects of every module the catalog half lists (§33): the newest entry of every followed +// branch, and the served entry of each module this runner serves (lead answer A-168), so a mark is never lost. +func (g *Gate) waitMarks() map[string]string { + m := Marks{Tools: map[string]string{}, Verbs: map[string]string{}, Subjects: map[string]string{}} + defs := g.d.Catalog.Definitions() + for _, e := range g.d.Catalog.Entries() { + m.add(e, defs) + } + g.mu.Lock() + served := make(map[string]string, len(g.served)) + for module, fp := range g.served { + served[module] = fp + } + g.mu.Unlock() + for _, fp := range served { + if e, ok, _ := g.d.Catalog.Served(fp); ok { + m.add(e, defs) + } + } + return m.Subjects +} + +// overlapsAny says a subject a wait subscribes overlaps a marked subject (§34). +func overlapsAny(subjects []string, marked map[string]string) bool { + for _, s := range subjects { + for mk := range marked { + if buscompose.Overlaps(s, mk) { + return true + } + } + } + return false +} + +// waitCatalog waits for the catalog half's first view within the bound, or until the caller goes. +func (g *Gate) waitCatalog(ctx context.Context, within time.Duration) error { + done := make(chan error, 1) + go func() { done <- g.d.Wait(within) }() + select { + case err := <-done: + return err + case <-ctx.Done(): + return ctx.Err() + } +} + +// pause waits until the deadline or the caller's going. +func (g *Gate) pause(ctx context.Context, deadline time.Time) { + if left := deadline.Sub(g.d.Now()); left > 0 { + select { + case <-g.d.After(left): + case <-ctx.Done(): + } + } +} + +// holdRefused is a refused marked wait, said at its timeout or when its caller goes, never at once, so a watcher's loop +// is never a hot one (§36). +func (g *Gate) holdRefused(ctx context.Context, deadline time.Time, words string) error { + g.pause(ctx, deadline) + if !strings.HasPrefix(words, "Not run") { + words = "Not run: " + words + } + return fmt.Errorf("%s", words) +} + +// holdAsk holds a marked wait whose ask is open, subscribing to nothing but its ask's `decided` (§36): it answers, as an +// error, at its timeout, when its caller goes, or once the router's record says the ask was chosen, the event only waking it. +func (g *Gate) holdAsk(ctx context.Context, deadline time.Time, module, id, digest string, first bool) error { + for { + left := deadline.Sub(g.d.Now()) + heard := false + if left > 0 && ctx.Err() == nil { + var err error + if g.d.Hear != nil { + heard, err = g.d.Hear(ctx, []string{seat.SubjectDecided(module)}, left, func(body json.RawMessage) bool { + var w seat.Warrant + return json.Unmarshal(body, &w) == nil && w.Ask == id + }) + } + if g.d.Hear == nil || err != nil { + g.pause(ctx, deadline) + } + } + rec, _ := g.record(module, id) + if heard && (rec == nil || rec.State != "chosen") { + continue // the event only wakes it: the record says nothing chosen yet + } + return g.waitWords(rec, id, digest, first) + } +} + +// waitWords are §12's and §17's words for a marked wait not run, "this wait" for "the call" (§36). +func (g *Gate) waitWords(rec *seat.AsksByAskerIDValue, id, digest string, first bool) error { + switch { + case rec == nil: + return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make this wait again later", id, Fingerprint(digest)) + case rec.State == "chosen" && rec.Warrant != nil && rec.Warrant.Option == OptionDecline: + at, _ := operatorchannelclient.Time(rec.Warrant.At) + return fmt.Errorf("%s", declinedWords(id, at)) + case rec.State == "chosen", first: + return fmt.Errorf("Not run: this wait needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make this wait "+ + "again within 30 minutes of the approval; the ask expires at %s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened, + id, Fingerprint(digest), rec.Expires) + case rec.State == "open", rec.State == "held-back": + return fmt.Errorf("Not run: still waiting for the operator's answer to ask %s (fingerprint %s); it expires at %s", + id, Fingerprint(digest), rec.Expires) + } + return fmt.Errorf("Not run: the ask %s ended %s; make this wait again to ask anew", id, rec.State) +} + +func declinedWords(id string, at time.Time) string { + return fmt.Sprintf("Not run: the operator declined ask %s; this wait is not asked again until %s", id, + operatorchannelclient.FormatTime(at.Add(DeclinedFor))) +} + +// askForWait is a marked wait's ask (§35): "Wait on on ?", shortened as §10 says, and an explanation naming +// every event name one per line, the filter, and the bound as the timeout in minutes, then the act's fingerprint. +func (g *Gate) askForWait(c Call, w Wait, digest, id string) seat.Ask { + a := g.askFor(c, digest, id) + filter, whole := "none", true + if len(w.Filter) > 0 { + raw, _ := json.Marshal(w.Filter) + filter = string(raw) + if r, ok := outward.Check(filter); !ok { + filter, whole = shapeOf(r), false + } + } + a.Headline = headlineOf("Wait on", strings.Join(w.Names, ", "), g.d.Node, whole) + lines := append([]string{"Wait on " + g.d.Node + ", called through the mesh's tools, which cannot tell who called, for:"}, w.Names...) + lines = append(lines, "Filter: "+filter, fmt.Sprintf("Timeout: %s minutes", minutes(w.Timeout)), "Fingerprint "+Fingerprint(digest)) + a.Explanation = strings.Join(lines, "\n") + a.Options[0].Does = "the same wait, made again within 30 minutes, subscribes once, its timeout running from then" + a.Options[1].Does = "nothing is heard; the same wait is not asked again for 10 minutes" + return a +} + +// minutes is a timeout in minutes, as few digits as say it. +func minutes(d time.Duration) string { + return strings.TrimSuffix(strings.TrimRight(fmt.Sprintf("%.2f", d.Minutes()), "0"), ".") +} diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index a93a16d..541e620 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -20,6 +20,8 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" + + "github.com/novox/mesh-tools/node-tools/internal/approval" ) // WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last @@ -107,9 +109,42 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin } } + listen := func(within time.Duration) (any, error) { + return s.hear(ctx, names, subjects, undeclared, known.source, unchecked, filter, within) + } + // A wait whose subjects overlap a marked event's runs only on the operator's warrant for that one wait (design 34 + // every-event §34 to §36), judged by the runner's approval gate. + var answer any + if g, ok := s.Approvals.(waitGate); ok { + given := map[string]any{} + for _, k := range []string{"subjects", "from", "filter", "timeout"} { + if v, has := args[k]; has { + given[k] = v + } + } + raw, _ := json.Marshal(given) + answer, err = g.Waited(ctx, approval.Call{Address: verbWait, Name: verbWait, Caller: s.who, Args: raw}, + approval.Wait{Names: names, From: from, Subjects: subjects, Filter: filter, Timeout: timeout}, listen) + } else { + answer, err = listen(timeout) + } + if err != nil { + return failure(err.Error()) + } + return answerText(answer) +} + +// waitGate is the runner's approval gate as mesh_wait goes through it (design 34 every-event §34 to §36). +type waitGate interface { + Waited(ctx context.Context, c approval.Call, w approval.Wait, hear func(within time.Duration) (any, error)) (any, error) +} + +// hear subscribes the wait's subjects for within and answers the first matching event, or that it timed out. +func (s *Surface) hear(ctx context.Context, names, subjects, undeclared []string, source, unchecked string, filter map[string]any, + timeout time.Duration) (map[string]any, error) { heard, ok, refused, err := s.conn.HearFirst(ctx, subjects, timeout, func(body json.RawMessage) bool { return bodyMatches(body, filter) }) if err != nil { - return failure("mesh_wait could not wait: " + err.Error()) + return nil, fmt.Errorf("mesh_wait could not wait: %w", err) } var answer map[string]any if !ok { @@ -117,7 +152,7 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin "read": "nothing matching was heard; read the state once from the verb that owns it, then wait again"} if len(undeclared) > 0 { answer["undeclared"] = fmt.Sprintf("no declaration names %s: no seat definition and no module's entry in %s "+ - "declares that event", strings.Join(undeclared, ", "), known.source) + "declares that event", strings.Join(undeclared, ", "), source) } } else { name, from, _ := buscompose.EventName(heard.Subject) @@ -139,7 +174,7 @@ func (s *Surface) waitAnswer(ctx context.Context, args map[string]any) map[strin if len(refused) > 0 { answer["refused"] = refused } - return answerText(answer) + return answer, nil } // halfReader is a Source that also holds the catalog half of each followed branch (firstview.Views.Halves): mesh_wait diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 54e2aaf..1eb7263 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -10,11 +10,16 @@ import ( "reflect" "sort" "strings" + "sync" "testing" "time" + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + opseat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/seatdef" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -197,7 +202,7 @@ func TestMeshWaitsTimeoutIsBounded(t *testing.T) { // a wildcard, of one token, more than 16 names, none, or a from of another word is refused at once naming the rule; a // wait whose every subscription the bus refuses answers after its timeout, in words naming no wait grant. func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { - s, _ := waitingRuntime(t) + s, mesh := waitingRuntime(t) var seventeen []any for i := 0; i < 17; i++ { seventeen = append(seventeen, fmt.Sprintf("mesh-knowledge.reviewed%d", i)) @@ -274,8 +279,232 @@ func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { !strings.Contains(text, "novox/hq ADR 0670 §1") || strings.Contains(text, "wait grant") || strings.Contains(text, "0367 §2") { t.Errorf("a wait the bus refused whole: %s", text) } + // Design 34 every-event §37: a wait whose subjects overlap a marked event's, or a stream's with a `grant`, runs only on the + // operator's warrant for that one wait; with none it subscribes nothing, asks once and answers at its timeout or the choice. + markedWaits(t, s, mesh) } +// markedWaits are §37's rows, against a gate on the test bus, the router's records written there as the router writes them. +func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { + gw := newGatedWaits(t, s, mesh, "src2:runner") + g := gw.surface + took := func(args map[string]any) (string, bool, time.Duration) { + start := time.Now() + text, isErr := callWait(g, args) + return text, isErr, time.Since(start) + } + // A fixture module's marked event, no warrant: nothing of it subscribed (only its ask's decided), asked once, §12's + // words at its timeout; the same wait again asks nothing new. + before := g.conn.Subscriptions() + go func() { + listening(t, g, before, 1) + mesh.Emit(t, "mesh.mod.mail.event.message", map[string]any{"n": 1}) + }() + text, isErr, d := took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}) + if !isErr || !strings.Contains(text, "this wait needs the operator's approval. Asked on telegram") || gw.asks() != 1 || + d < time.Second || strings.Contains(text, "heard") { + t.Errorf("a marked event's wait, no warrant: %s after %s, %d asks", text, d, gw.asks()) + } + if text, isErr, _ = took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr || + !strings.Contains(text, "still waiting") || gw.asks() != 1 { + t.Errorf("the same marked wait again: %s, %d asks", text, gw.asks()) + } + // A stream with a `grant` is marked whole; a name with the node after a marked event's name is marked. + for i, name := range []string{"mail.inbox", "mail.message.desk"} { + if text, isErr, _ := took(map[string]any{"subjects": []any{name}, "timeout": 1}); !isErr || !strings.Contains(text, "Not run") || gw.asks() != 2+i { + t.Errorf("%s: %s, %d asks", name, text, gw.asks()) + } + } + // The same names from a seat ask nothing. + if text, isErr, _ := took(map[string]any{"subjects": []any{"mail.message"}, "from": "seat", "timeout": 1}); isErr || + !strings.Contains(text, `"timed-out": true`) || gw.asks() != 3 { + t.Errorf("from a seat: %s, %d asks", text, gw.asks()) + } + // The most timeout a wait takes, its ask chosen while it waits: answered at the choice; the same call made then + // subscribes and hears the event published after it, answered with its approval; a warrant for other names, another + // filter or another timeout is refused, asked anew. + long := map[string]any{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 2}, "timeout": 600} + done := make(chan string, 1) + go func() { text, _, d := took(long); done <- fmt.Sprintf("%s after %s", text, d) }() + mt.Until(t, func() error { + if gw.asks() < 4 { + return os.ErrNotExist + } + return nil + }) + gw.approve(t) + select { + case got := <-done: + if !strings.Contains(got, "make this wait again within 30 minutes") { + t.Errorf("a long wait chosen while it waits: %s", got) + } + case <-time.After(10 * time.Second): + t.Fatal("a long marked wait was not answered at the choice") + } + before = g.conn.Subscriptions() + go func() { + listening(t, g, before, 6) + mesh.Emit(t, "mesh.mod.mail.event.message", map[string]any{"n": 2}) + }() + if text, isErr, _ := took(long); isErr || !strings.Contains(text, `"approval": "run on the operator's approval`) || + !strings.Contains(text, `"subject": "mail.message"`) { + t.Errorf("the same wait on its warrant: %s", text) + } + for _, other := range []map[string]any{ + {"subjects": []any{"mail.message", "mail.inbox"}, "filter": map[string]any{"n": 2}, "timeout": 1}, + {"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 3}, "timeout": 1}, + {"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 2}, "timeout": 2}, + } { + asked := gw.asks() + if text, isErr, _ := took(other); !isErr || !strings.Contains(text, "Not run") || gw.asks() != asked+1 { + t.Errorf("another wait than the warrant's %v: %s", other, text) + } + } + // Before the catalog half is read nothing is asked: judged when its first view comes within the timeout, refused at + // the timeout when it does not. + gw.cat.mu.Lock() + gw.cat.ready = errors.New("no first view yet") + gw.cat.mu.Unlock() + asked := gw.asks() + if text, isErr, d := took(map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr || + !strings.Contains(text, "the marks are not yet read") || gw.asks() != asked || d < time.Second { + t.Errorf("before the catalog half is read: %s after %s", text, d) + } + gw.cat.came = make(chan chan struct{}) + go func() { close(<-gw.cat.came) }() + if text, isErr, _ := took(map[string]any{"subjects": []any{"mail.message"}, "filter": map[string]any{"n": 9}, "timeout": 1}); !isErr || + !strings.Contains(text, "this wait needs the operator's approval") || gw.asks() != asked+1 { + t.Errorf("the catalog half read within the timeout: %s", text) + } + // While the runner's own fingerprint is unknown a marked wait is refused, saying so. + unknown := newGatedWaits(t, s, mesh, "") + if text, isErr := callWait(unknown.surface, map[string]any{"subjects": []any{"mail.message"}, "timeout": 1}); !isErr || + !strings.Contains(text, "own fingerprint is unknown") || unknown.asks() != 0 { + t.Errorf("the runner's fingerprint unknown: %s", text) + } +} + +// gatedWaits is the surface with the runner's approval gate: the fixture module mail marks its event message and declares +// the stream inbox with a grant; the router takes each ask as it is published, its records on the test bus. +type gatedWaits struct { + surface *Surface + cat *waitCatalog + records nats.KeyValue + mu sync.Mutex + asked []opseat.Ask + say *bus.Conn +} + +func newGatedWaits(t *testing.T, s *Surface, mesh *mt.Mesh, build string) *gatedWaits { + gw := &gatedWaits{cat: &waitCatalog{entries: []catalogseat.CatalogEntry{{Module: "mail", Fp: "src2:mail", Manifest: map[string]any{ + "emits": []any{"message"}, "needs-approval": []any{"message"}, + "streams": []any{map[string]any{"name": "inbox", "events": []any{"inbox.*"}, "grant": map[string]any{"subscribe": []any{"x"}}}}}}}}, + records: mesh.Bucket(t, "operator-channel-asks"), say: connect(t, "messenger", "")} + gate := approval.New(approval.Deps{Node: "desk", Runner: "node-tools", Build: build, Catalog: gw.cat, Wait: gw.cat.wait, + State: mt.NewState(), Holder: func(string) (bool, error) { return true, nil }, + Hear: func(ctx context.Context, subjects []string, within time.Duration, match func(json.RawMessage) bool) (bool, error) { + _, ok, _, err := s.conn.HearFirst(ctx, subjects, within, match) + return ok, err + }, + Publish: func(module, subject string, body any, id string) error { + a, ok := body.(opseat.Ask) + if subject != opseat.AcceptSubjectAsk(module) || !ok { + return nil + } + gw.mu.Lock() + gw.asked = append(gw.asked, a) + gw.mu.Unlock() + gw.write(t, opseat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", Headline: a.Headline, + Expires: a.Expires, Opened: operatorchannelclient.FormatTime(time.Now()), OfferedOn: []string{"telegram"}}) + return nil + }, + Record: func(module, id string) (*stdio.StateEntry, error) { + e, err := gw.records.Get(module + "." + id) + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, nil + } else if err != nil { + return nil, err + } + return &stdio.StateEntry{Key: e.Key(), Value: e.Value(), Revision: e.Revision()}, nil + }}) + gw.surface = NewSurface(s.conn, busSource{s.conn}, "desk.node-tools") + gw.surface.Approvals = gateApprovals{gate} + return gw +} + +func (gw *gatedWaits) write(t *testing.T, rec opseat.AsksByAskerIDValue) { + raw, _ := json.Marshal(rec) + if _, err := gw.records.Put(rec.Key, raw); err != nil { + t.Error(err) + } +} + +func (gw *gatedWaits) asks() int { + gw.mu.Lock() + defer gw.mu.Unlock() + return len(gw.asked) +} + +// approve writes the router's record of the newest ask as a person's Approve, and says its decided. +func (gw *gatedWaits) approve(t *testing.T) { + gw.mu.Lock() + a := gw.asked[len(gw.asked)-1] + gw.mu.Unlock() + gw.write(t, opseat.AsksByAskerIDValue{Asker: "node-tools", Key: "node-tools." + a.ID, State: "chosen", Expires: a.Expires, + OfferedOn: []string{"telegram"}, Warrant: &opseat.Warrant{Ask: a.ID, Asker: "node-tools", Outcome: operatorchannelclient.OutcomeChosen, + Option: approval.OptionApprove, Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(time.Now()), + AskDigest: operatorchannelclient.Digest(a), Channel: "telegram", + By: &opseat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}}}) + if err := gw.say.Say(opseat.SubjectDecided("node-tools"), []byte(`{"ask":"`+a.ID+`"}`)); err != nil { + t.Fatal(err) + } +} + +// gateApprovals is the gate as the surface holds it: mesh_approvals' reads unused here, mesh_wait's gate. +type gateApprovals struct{ *approval.Gate } + +func (gateApprovals) Fill() (uint64, int64, error) { return 0, 0, nil } + +// waitCatalog is a catalog half as a fixture: its entries, read or not; with came set, its first view comes once a wait +// waits for it (waiting is told on it, then it closes). +type waitCatalog struct { + entries []catalogseat.CatalogEntry + mu sync.Mutex + ready error + came chan chan struct{} +} + +func (c *waitCatalog) Ready() error { + c.mu.Lock() + defer c.mu.Unlock() + return c.ready +} +func (c *waitCatalog) wait(within time.Duration) error { + if c.Ready() == nil { + return nil + } + if c.came != nil { + view := make(chan struct{}) + c.came <- view + <-view + c.mu.Lock() + c.ready = nil + c.mu.Unlock() + return nil + } + <-time.After(within) // the bound of the runner's own wait for the first view + return c.Ready() +} +func (c *waitCatalog) Revision() string { return "7" } +func (c *waitCatalog) Served(string) (catalogseat.CatalogEntry, bool, string) { + return catalogseat.CatalogEntry{}, false, "not-found" +} +func (c *waitCatalog) Newest(string) (catalogseat.CatalogEntry, bool) { + return catalogseat.CatalogEntry{}, false +} +func (c *waitCatalog) Definitions() map[string]seatdef.Definition { return nil } +func (c *waitCatalog) Entries() []catalogseat.CatalogEntry { return c.entries } + // The mesh MCP server announces seven tools (ADR 0367 v13): ADR 0195's five, mesh_runtimes and mesh_wait, its timeout bounded. func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) { var names []string diff --git a/node-tools/internal/firstview/approval.go b/node-tools/internal/firstview/approval.go index 45e1b6c..f230589 100644 --- a/node-tools/internal/firstview/approval.go +++ b/node-tools/internal/firstview/approval.go @@ -78,6 +78,20 @@ func (c approvalCatalog) Definitions() map[string]seatdef.Definition { return c.defs() } +func (c approvalCatalog) Entries() []catalogseat.CatalogEntry { + halves, err := c.v.Halves() + if err != nil { + return nil + } + var out []catalogseat.CatalogEntry + for _, half := range halves { + for _, e := range half.Entries { + out = append(out, e) + } + } + return out +} + type notFollowed struct{} func (notFollowed) Error() string { return "the catalog half is not followed by this runner" } diff --git a/node-tools/internal/meshtest/state.go b/node-tools/internal/meshtest/state.go new file mode 100644 index 0000000..87c9e4d --- /dev/null +++ b/node-tools/internal/meshtest/state.go @@ -0,0 +1,87 @@ +package meshtest + +import ( + "encoding/json" + "sort" + "sync" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// State is a module's state as a fixture, the shape a gate's Store takes: Fail makes every read and write fail, as a bus +// refusing them; Vals are its values by key. +type State struct { + mu sync.Mutex + Vals map[string]json.RawMessage + revs map[string]uint64 + n uint64 + Fail error +} + +// NewState is an empty State. +func NewState() *State { return &State{Vals: map[string]json.RawMessage{}, revs: map[string]uint64{}} } + +func (m *State) Get(k string) (*stdio.StateEntry, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.Fail != nil { + return nil, m.Fail + } + v, ok := m.Vals[k] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: k, Value: v, Revision: m.revs[k]}, nil +} +func (m *State) Create(k string, v any) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.Fail != nil { + return 0, m.Fail + } + if _, ok := m.Vals[k]; ok { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *State) Update(k string, v any, rev uint64) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.Fail != nil { + return 0, m.Fail + } + if rev == 0 || m.revs[k] != rev { + return 0, stdio.ErrChanged + } + return m.put(k, v) +} +func (m *State) Put(k string, v any) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.Fail != nil { + return 0, m.Fail + } + return m.put(k, v) +} +func (m *State) Keys() ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.Fail != nil { + return nil, m.Fail + } + var out []string + for k := range m.Vals { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} +func (m *State) put(k string, v any) (uint64, error) { + raw, ok := v.(json.RawMessage) + if !ok { + raw, _ = json.Marshal(v) + } + m.n++ + m.Vals[k], m.revs[k] = raw, m.n + return m.n, nil +}