diff --git a/node-tools/cmd/node-tools/approvals.go b/node-tools/cmd/node-tools/approvals.go new file mode 100644 index 0000000..3b3b4ed --- /dev/null +++ b/node-tools/cmd/node-tools/approvals.go @@ -0,0 +1,101 @@ +package main + +import ( + "encoding/json" + "errors" + + stdio "git.novox.be/novox/mesh-sdk/go" + + "github.com/novox/mesh-tools/node-tools/internal/approval" + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// The gate's reach through the runner's own connection, in this one place (design 46 7-approval-mark §7, §13, §14): its +// state approvals, its publish of an ask under the marked module's name, and its read of the router's record. + +// approvalsState is the runner's state approvals over its connection: a lost compare-and-set is stdio.ErrChanged, as +// operator-channel's client reads it. +type approvalsState struct { + conn *bus.Conn + module string +} + +func (s approvalsState) Get(key string) (*stdio.StateEntry, error) { + e, err := s.conn.StateGet(s.module, approval.StateName, key) + if err != nil || e == nil { + return nil, err + } + return &stdio.StateEntry{Key: e.Key, Value: e.Value, Revision: e.Revision, At: e.At}, nil +} + +func (s approvalsState) Create(key string, value any) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + return changedAs(s.conn.StateCreate(s.module, approval.StateName, key, raw)) +} + +func (s approvalsState) Update(key string, value any, revision uint64) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + return changedAs(s.conn.StateUpdate(s.module, approval.StateName, key, raw, revision)) +} + +func (s approvalsState) Put(key string, value any) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + return s.conn.StatePut(s.module, approval.StateName, key, raw) +} + +func (s approvalsState) Keys() ([]string, error) { + return s.conn.StateKeys(s.module, approval.StateName) +} + +func changedAs(rev uint64, err error) (uint64, error) { + if errors.Is(err, bus.ErrStateChanged) { + return 0, stdio.ErrChanged + } + return rev, err +} + +// Bound says the runner's membership binds its state approvals now (§13): a membership issued later binds it with no restart. +func (s approvalsState) Bound() bool { + m := s.conn.Membership(s.module) + if m == nil { + return false + } + for _, st := range m.State { + if st.Name == approval.StateName && st.Writes { + return true + } + } + return false +} + +// publishAsk says an accept on a seat under the module's name, on the runner's own account (§7). +func publishAsk(conn *bus.Conn) func(module, subject string, body any, id string) error { + return func(module, subject string, body any, id string) error { + raw, err := json.Marshal(body) + if err != nil { + return err + } + _, err = conn.SeatPublish(module, subject, raw, id) + return err + } +} + +// routerRecord reads the router's record `.` of the seat's state asks, on the runner's own account (§14). +func routerRecord(conn *bus.Conn) func(module, id string) (*stdio.StateEntry, error) { + return func(module, id string) (*stdio.StateEntry, error) { + e, err := conn.SeatRecord(module, "", id) + if err != nil || e == nil { + return nil, err + } + return &stdio.StateEntry{Key: e.Key, Value: e.Value, Revision: e.Revision, At: e.At}, nil + } +} diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index bd861b1..78e15cd 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -75,8 +75,9 @@ func main() { // ADR 0615 §1; design 46 7-approval-mark §5). source, stopViews := firstview.Start(context.Background(), conn, log.Printf) defer stopViews() - gate := approval.New(approval.Deps{Node: conn.Node(), Runner: cred.Module, Catalog: source.ApprovalCatalog(definitionsOf(seatList)), - Findings: source.Findings(), Logf: log.Printf}) + 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)}) opts = append(opts, runtime.WithGate(gate)) stop, err := runtime.Run(conn, served, envs, log.Printf, opts...) if err != nil { diff --git a/node-tools/internal/approval/act.go b/node-tools/internal/approval/act.go index 7f9445d..5ad9df8 100644 --- a/node-tools/internal/approval/act.go +++ b/node-tools/internal/approval/act.go @@ -28,6 +28,8 @@ type Call struct { Args json.RawMessage // Description is the tool's or verb's own description, whose first line the ask explains it by. Description string + // Own says it is one of the runner's own tools served under the module's name (its ledger repairs), not its bundle's. + Own bool } // MostArguments is the most a marked call's arguments may hold: more is refused, asking nothing (§9). diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index 42b25ff..95ed780 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -145,6 +145,24 @@ func newWorld() *world { return w } +// decide writes the router's record of the newest ask as the router writes a person's choice: its state and warrant. +func (w *world) decide(option string, at time.Time) seat.Ask { + a := w.asked[len(w.asked)-1] + key := "shell." + a.ID + rec := w.records[key] + rec.State = "chosen" + rec.Warrant = &seat.Warrant{Ask: a.ID, Asker: "shell", Outcome: operatorchannelclient.OutcomeChosen, Option: option, + Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(at), AskDigest: operatorchannelclient.Digest(a), + By: &seat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}} + w.records[key] = rec + return a +} + +// do makes a call through the gate, counting the runs. +func (w *world) do(c Call) (any, error) { + return w.g.Run(c, func() (any, error) { w.runs++; return "done", nil }) +} + // call is a fixture call of the marked tool wipe, or of another tool of the module. func call(name, args string) Call { return Call{Address: "shell." + name, Module: "shell", Name: name, Fingerprint: "src2:1", Caller: "agent", Args: json.RawMessage(args), @@ -270,6 +288,150 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("answer %v after %s", err, w.now.Sub(start)) } }}, + {"§10, §17: a second identical call publishes nothing and is told it still waits", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{"dir":"scratch"}`)) + _, err := w.do(call("wipe", `{"dir":"scratch"}`)) + if len(w.asked) != 1 || err == nil || !strings.HasPrefix(err.Error(), "Not run: still waiting for the operator's answer") || w.runs != 0 { + t.Fatalf("asked %d, %v, runs %d", len(w.asked), err, w.runs) + } + }}, + {"§17: an ask the router holds back as the fourth open one still waits", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + a := w.asked[0] + rec := w.records["shell."+a.ID] + rec.State, rec.HeldSince = "held-back", operatorchannelclient.FormatTime(w.now) + w.records["shell."+a.ID] = rec + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "still waiting") || len(w.asked) != 1 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§14, §15: approved, the same call runs once; made again it is already run and asks anew", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{"dir":"scratch"}`)) + w.decide(OptionApprove, w.now.Add(time.Minute)) + w.now = w.now.Add(2 * time.Minute) + if got, err := w.do(call("wipe", `{"dir":"scratch"}`)); err != nil || got != "done" || w.runs != 1 { + t.Fatalf("%v %v, runs %d", got, err, w.runs) + } + _, err := w.do(call("wipe", `{"dir":"scratch"}`)) + if err == nil || !strings.Contains(err.Error(), "was already run; asked again") || w.runs != 1 || len(w.asked) != 2 { + t.Fatalf("%v, runs %d, asked %d", err, w.runs, len(w.asked)) + } + }}, + {"§15: two calls at the same moment run it once", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + w.decide(OptionApprove, w.now) + var wg sync.WaitGroup + var mu sync.Mutex + runs := 0 + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, _ = w.g.Run(call("wipe", `{}`), func() (any, error) { mu.Lock(); runs++; mu.Unlock(); return "done", nil }) + }() + } + wg.Wait() + if runs != 1 { + t.Fatalf("ran %d times", runs) + } + }}, + {"§14: approved for other arguments runs nothing", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{"dir":"scratch"}`)) + a := w.decide(OptionApprove, w.now) + rec := w.records["shell."+a.ID] + other, _ := ActOf(call("wipe", `{"dir":"home"}`), "n1") + d, _ := operatorchannelclient.ActDigest(other) + a.Options[0].Binds = d // the ask the person was shown bound another act + 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) + k.Ask = a + raw, _ := json.Marshal(k) + 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) + } + }}, + {"§14: approved 31 minutes ago asks anew", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + w.decide(OptionApprove, w.now) + w.now = w.now.Add(31 * time.Minute) + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "approved too long ago") || w.runs != 0 || len(w.asked) != 2 { + t.Fatalf("%v, runs %d, asked %d", err, w.runs, len(w.asked)) + } + }}, + {"§14, §17: declined, the same call is not asked again for 10 minutes", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + w.decide(OptionDecline, w.now) + w.now = w.now.Add(9 * time.Minute) + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "declined") || len(w.asked) != 1 || w.runs != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + w.now = w.now.Add(2 * time.Minute) + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "asked again") || len(w.asked) != 2 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§14, §17: expired asks anew", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + a := w.asked[0] + rec := w.records["shell."+a.ID] + rec.State = "expired" + w.records["shell."+a.ID] = rec + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "ended expired; asked again") || len(w.asked) != 2 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§14: another asker's warrant runs nothing", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + a := w.decide(OptionApprove, w.now) + rec := w.records["shell."+a.ID] + rec.Warrant.Asker = "mail" + w.records["shell."+a.ID] = rec + if _, err := w.do(call("wipe", `{}`)); err == nil || w.runs != 0 || !strings.Contains(err.Error(), "its warrant does not hold") { + t.Fatalf("%v, runs %d", err, w.runs) + } + }}, + {"§17: the router's refusal is said with its words", func(t *testing.T, w *world) { + _, _ = w.do(call("wipe", `{}`)) + a := w.asked[0] + rec := w.records["shell."+a.ID] + rec.State, rec.Answer = "refused", "the asker has too many open asks" + w.records["shell."+a.ID] = rec + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "the asker has too many open asks") { + t.Fatalf("%v", err) + } + }}, + {"§20: a runner's own tool marked is refused unasked, never run", func(t *testing.T, w *world) { + c := call("wipe", `{}`) + c.Own = true + if _, err := w.do(c); err == nil || w.runs != 0 || len(w.asked) != 0 { + t.Fatalf("%v, runs %d, asked %d", err, w.runs, len(w.asked)) + } + }}, + {"§13: with no binding, an unmarked call runs and a marked one is refused", func(t *testing.T, w *world) { + w.g.d.State = nil + if got, err := w.do(call("tidy", `{}`)); err != nil || got != "done" { + t.Fatalf("unmarked: %v %v", got, err) + } + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "binds no state approvals") || w.runs != 1 { + t.Fatalf("marked: %v, runs %d", err, w.runs) + } + }}, + {"§13: an ask over 256 KiB is refused naming the cause", func(t *testing.T, w *world) { + c := call("wipe", `{}`) + c.Description = strings.Repeat("w", 300*1024) + if _, err := w.do(c); err == nil || !strings.Contains(err.Error(), "more than the 262144 a value of the state approvals holds") || len(w.asked) != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§3, §2: a seat verb its seat marks is asked for its holder", func(t *testing.T, w *world) { + c := Call{Address: "node-login-shell.execute", Module: "shell", Seat: "node-login-shell", Name: "execute", Fingerprint: "src2:1", Args: json.RawMessage(`{"command":"ls"}`)} + if _, err := w.do(c); err == nil || !strings.HasPrefix(err.Error(), "Not run") || len(w.asked) != 1 || w.asked[0].Headline != "Run node-login-shell.execute on n1?" { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, } { t.Run(tc.name, func(t *testing.T) { tc.run(t, newWorld()) }) } diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go index 0aa5e32..a8c8498 100644 --- a/node-tools/internal/approval/gate.go +++ b/node-tools/internal/approval/gate.go @@ -35,6 +35,8 @@ type Deps struct { Node, Runner string // Catalog is the catalog half the marks are read from. Catalog Catalog + // Wait waits, within the bound, for the catalog half's first view, and is nil once it is read or why not. + Wait func(within time.Duration) error // Findings is the runner's held writer; nil says a finding in the log only. Findings FindingWriter Logf func(string, ...any) @@ -57,6 +59,8 @@ type Gate struct { registered map[string]map[string]bool // unserved is each module whose marks name a tool its bundle does not register, with those tools. unserved map[string][]string + // unmarkedAt is each caller's address it last called unmarked, so a mark added since is named (§12). + unmarkedAt map[string]bool } // New is a gate over d. diff --git a/node-tools/internal/approval/run.go b/node-tools/internal/approval/run.go new file mode 100644 index 0000000..b1d3bef --- /dev/null +++ b/node-tools/internal/approval/run.go @@ -0,0 +1,183 @@ +package approval + +import ( + "encoding/json" + "fmt" + "strings" + + "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" +) + +// Run is a call through the gate (§20): an unmarked call runs; a marked one runs only on the operator's approval of that +// one call, judged on the router's record and claimed once (§14, §15), and is otherwise answered as an error saying why +// (§12, §17, §19). run is called at most once. +func (g *Gate) Run(c Call, run func() (any, error)) (any, error) { + m, err := g.marksOf(c) + if err != nil { + return nil, err + } + if missing := g.unservedOf(c.Module); len(missing) > 0 { + return nil, fmt.Errorf("Not run: %s's marks name %s, which its bundle on %s does not register, so none of its tools is "+ + "served here (design 46 7-approval-mark §4)", c.Module, strings.Join(missing, ", "), g.d.Node) + } + marked, _ := m.Marked(c.Seat, c.Name) + since := g.markedSince(c, marked, m.Revision) + if !marked { + return run() + } + if c.Own { + return nil, g.refused(c, "", "it is one of the tool runner's own tools, which no ask covers") + } + if m.Unread != "" { + return nil, g.refused(c, "", m.Unread) + } + if b, ok := g.d.State.(Binding); g.d.State == nil || ok && !b.Bound() { + return nil, g.refused(c, "", "the tool runner's membership binds no state "+StateName+" to keep its asks in") + } + if why := refusedUnasked(c.Args); why != "" { + return nil, g.refused(c, "", why) + } + act, err := ActOf(c, g.d.Node) + if err != nil { + return nil, g.refused(c, "", err.Error()) + } + digest, err := operatorchannelclient.ActDigest(act) + if err != nil { + return nil, g.refused(c, "", err.Error()) + } + named, err := g.d.State.Get(g.actKey(digest)) + if err != nil { + return nil, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()) + } + if named == nil { + return nil, g.raise(c, act, digest, since) + } + var id string + if err := json.Unmarshal(named.Value, &id); err != nil || id == "" { + return nil, g.refused(c, digest, fmt.Sprintf("its act's key in %s names no ask", StateName)) + } + return g.judge(c, act, digest, id, since, run) +} + +// Binding is a Store that says whether the runner's membership binds the state now (§13). +type Binding interface{ Bound() bool } + +// marksOf is the module's marks as the catalog half has them now; before its first view, the call waits for it within +// RecordWithin and is then judged, refused naming mesh_first_views when it is still unread (§19). +func (g *Gate) marksOf(c Call) (Marks, error) { + if g.d.Catalog == nil { + return Marks{}, g.refused(c, "", "the tool runner reads no catalog half, so its marks are not read") + } + if err := g.d.Catalog.Ready(); err != nil { + if g.d.Wait != nil { + err = g.d.Wait(RecordWithin) + } + if err != nil { + return Marks{}, fmt.Errorf("Not run: the marks of %s are not read yet (%v); mesh_first_views says what the tool runner "+ + "waits for", c.Module, err) + } + } + m := Derive(g.d.Catalog, c.Module, c.Fingerprint) + g.checkServed(m) + return m, nil +} + +func (g *Gate) unservedOf(module string) []string { + g.mu.Lock() + defer g.mu.Unlock() + return g.unserved[module] +} + +// markedSince is the revision a call was marked at when the same caller last called it unmarked (§12), empty otherwise. +func (g *Gate) markedSince(c Call, marked bool, revision string) string { + key := c.Caller + " " + c.Address + g.mu.Lock() + defer g.mu.Unlock() + if g.unmarkedAt == nil { + g.unmarkedAt = map[string]bool{} + } + was := g.unmarkedAt[key] + if marked { + delete(g.unmarkedAt, key) + if was { + return revision + } + return "" + } + g.unmarkedAt[key] = true + return "" +} + +// judge is the same call made again, judged on the router's own record of its act's newest ask (§14 to §17). +func (g *Gate) judge(c Call, act operatorchannelclient.Act, digest, id, since string, run func() (any, error)) (any, error) { + rec, err := g.record(c.Module, id) + if err != nil { + return nil, g.refused(c, digest, "the router's record of its ask "+id+" cannot be read: "+err.Error()) + } + if rec == nil { + return nil, g.firstAnswer(c, id, digest, since) // §17: not yet taken by the router + } + again := func(why string) (any, error) { + err := g.raise(c, act, digest, since) + return nil, fmt.Errorf("The ask %s %s; asked again. %w", id, why, err) + } + switch rec.State { + case "open", "held-back": + return nil, 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) + case "refused": + return nil, g.refused(c, digest, "the router refused its ask "+id+": "+orNone(rec.Answer)) + case "chosen": + default: // expired, cancelled, replaced: each asks again (§15, §17) + return again("ended " + rec.State) + } + w := rec.Warrant + if w == nil { + return nil, g.refused(c, digest, "the router's record of its ask "+id+" is chosen and carries no warrant") + } + at, _ := operatorchannelclient.Time(w.At) + if w.Option == OptionDecline { + if until := at.Add(DeclinedFor); g.d.Now().Before(until) { + return nil, fmt.Errorf("Not run: the operator declined ask %s; the same call is not asked again until %s", + id, operatorchannelclient.FormatTime(until)) + } + return again("was declined more than 10 minutes ago") + } + if g.d.Now().After(at.Add(RunWithin)) { + return again("was approved too long ago (more than 30 minutes)") + } + var answer any + var ran error + h, err := g.client(c.Module).Take(*w, func(o seat.Option, _ seat.Warrant) error { + if err := operatorchannelclient.Performs(o, act); err != nil { + ran = err + return err + } + answer, ran = run() + return nil + }) + switch { + case h == operatorchannelclient.Done: + return again("was already run") + case h == operatorchannelclient.Refused: + return nil, g.refused(c, digest, "its warrant does not hold: "+errText(err)) + case h == "" && err != nil: + return nil, g.refused(c, digest, "its state "+StateName+" cannot be written: "+err.Error()) + case err != nil: // the act about to run is not the one the option bound: nothing runs, the warrant stays used + return nil, g.refused(c, digest, err.Error()) + } + return g.ran(c, id, rec, answer, ran) +} + +// ran is a call run on the operator's approval: the tool's own answer. +func (g *Gate) ran(_ Call, _ string, _ *seat.AsksByAskerIDValue, answer any, err error) (any, error) { + return answer, err +} + +func errText(err error) string { + if err == nil { + return "no reason given" + } + return err.Error() +} diff --git a/node-tools/internal/firstview/approval.go b/node-tools/internal/firstview/approval.go index bf317d6..587612e 100644 --- a/node-tools/internal/firstview/approval.go +++ b/node-tools/internal/firstview/approval.go @@ -3,6 +3,7 @@ package firstview import ( "context" "strconv" + "time" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" @@ -81,3 +82,76 @@ type notFollowed struct{} func (notFollowed) Error() string { return "the catalog half is not followed by this runner" } var errNotFollowed error = notFollowed{} + +// heardOne wakes every wait on the picture. +func (v *Views) heardOne() { + v.mu.Lock() + defer v.mu.Unlock() + if v.heard != nil { + close(v.heard) + } + v.heard = make(chan struct{}) +} + +// heardNext is closed at the next event heard. +func (v *Views) heardNext() <-chan struct{} { + v.mu.Lock() + defer v.mu.Unlock() + if v.heard == nil { + v.heard = make(chan struct{}) + } + return v.heard +} + +// WaitCatalog waits within the bound for the catalog half's first view of every watched branch (design 46 7-approval-mark +// §19), waking at each event the runner hears and at each first view answered; nil once it is read, else why not. +func (v *Views) WaitCatalog(within time.Duration) error { + c := approvalCatalog{v: v} + timer := time.NewTimer(within) + defer timer.Stop() + for { + next := v.heardNext() + err := c.Ready() + if err == nil { + return nil + } + var ready []<-chan struct{} + if v.branches != nil { + for _, branch := range v.branches.Followed() { + if cat, ok := v.branches.Catalog(branch); ok { + select { + case <-cat.Ready(): // read already: waits on nothing + default: + ready = append(ready, cat.Ready()) + } + } + } + } + if !waitAny(next, timer.C, ready) { + return c.Ready() + } + } +} + +// waitAny waits for next or one of ready, true, or for the bound, false. +func waitAny(next <-chan struct{}, bound <-chan time.Time, ready []<-chan struct{}) bool { + woke := make(chan struct{}, 1) + stop := make(chan struct{}) + defer close(stop) + for _, r := range ready { + go func(r <-chan struct{}) { + select { + case <-r: + woke <- struct{}{} + case <-stop: + } + }(r) + } + select { + case <-next: + case <-woke: + case <-bound: + return false + } + return true +} diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index c187bcb..1251cac 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -67,6 +67,8 @@ type Views struct { cancel context.CancelFunc wg sync.WaitGroup deps Deps + // heard is closed and made again at every event heard, so a wait for the picture wakes on events, never on a timer. + heard chan struct{} } // FindingsState is the runner's own state the findings are written in. @@ -205,8 +207,10 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View return v, func() { stop(); cancel(); v.wg.Wait(); stopFindings() } } -// hear applies one views event; a gap asks the first view once more. +// hear applies one views event; a gap asks the first view once more. Every event heard wakes what waits on the picture +// (heardOne). func (v *Views) hear(env bus.Envelope) error { + defer v.heardOne() v.mu.Lock() hs := append([]*adoption(nil), v.hearers[env.Key]...) v.mu.Unlock() diff --git a/node-tools/internal/runtime/gate.go b/node-tools/internal/runtime/gate.go index ed7cd5f..649bec5 100644 --- a/node-tools/internal/runtime/gate.go +++ b/node-tools/internal/runtime/gate.go @@ -12,6 +12,20 @@ const FingerprintWord = "MESH_FINGERPRINT" type Gate interface { // Unserved is, of a module's marked tools, those its bundle does not register, saying it as the module's finding. Unserved(module, fingerprint string, registered []string) []string + // Run is a call through the gate: run is called only when the call may run now, once. + Run(call approval.Call, run func() (any, error)) (any, error) +} + +// gatedRun is how a handler calls a tool: through the gate where there is one. +type gatedRun func(call approval.Call, t launch.Tool, caller string) (any, error) + +// gated calls the tool through the gate, or straight where the runner has none. +func (c *consumers) gated(call approval.Call, t launch.Tool, caller string) (any, error) { + run := func() (any, error) { return t.Run(caller, call.Args) } + if c.gate == nil { + return run() + } + return c.gate.Run(call, run) } // WithGate serves every call through the approval gate. diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 7dae3af..5bbea33 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -17,6 +17,7 @@ import ( "time" "github.com/novox/mesh-tools/node-tools/internal/announce" + "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/bus" "github.com/novox/mesh-tools/node-tools/internal/launch" ) @@ -303,7 +304,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log } logf("%s", line) conn.OnMembership(func(bus.Membership) { go func() { r.refreshRepairs(); conn.Flush() }() }) - stopSeats, again := serveSeats(conn, r.modules, r.registrations, logf) + stopSeats, again := serveSeats(conn, r.modules, r.registrations, r.events.gated, r.fingerprintOf, logf) r.seatsAgain = again r.stops = append(r.stops, stopSeats) @@ -464,8 +465,11 @@ func (r *runner) serve(l *loadedModule) { for _, t := range reg.tools { t := t // Told who called, from the subject the call arrived on, never from its body (novox/hq issue 365). + // Through the approval gate, before the bundle or the runner's own tool (design 46 7-approval-mark §20). + module, own := reg.module, reg.repairs stop, err := r.conn.HandleCalled(reg.module+"."+t.Name, func(caller string, body json.RawMessage) (any, error) { - return t.Run(caller, argsOf(body)) + return r.events.gated(approval.Call{Address: module + "." + t.Name, Module: module, Name: t.Name, Own: own, + Fingerprint: r.fingerprintOf(module), Caller: caller, Args: argsOf(body), Description: t.Description}, t, caller) }) if err != nil { r.logf("[mesh-tools] cannot serve %s.%s: %v", reg.module, t.Name, err) @@ -703,7 +707,8 @@ func subjectsOf(conn *bus.Conn, module, tool string) []string { // the tool of the same name registered under the seat's name (ADR 0159, 0160) — and serves again // whenever a membership changes, and when again is called because a bundle was loaded again. Whether // this machine holds the seat is the bus's to decide. It answers its stop and again. -func serveSeats(conn *bus.Conn, modules []string, current func() []registration, logf func(string, ...any)) (func(), func()) { +func serveSeats(conn *bus.Conn, modules []string, current func() []registration, gated gatedRun, fingerprintOf func(string) string, + logf func(string, ...any)) (func(), func()) { var mu sync.Mutex var stops []func() stopped := false @@ -742,8 +747,11 @@ func serveSeats(conn *bus.Conn, modules []string, current func() []registration, logf("[mesh-tools] %s claims %s and implements no %s, which that seat promises; not served", module, v.Seat, v.Verb) continue } + // Through the approval gate, before the bundle (design 46 7-approval-mark §20). + module, v := module, v stop, err := conn.HandleSubjectCalled(v.Subject, func(caller string, body json.RawMessage) (any, error) { - return t.Run(caller, argsOf(body)) + return gated(approval.Call{Address: v.Seat + "." + v.Verb, Module: module, Seat: v.Seat, Name: v.Verb, + Fingerprint: fingerprintOf(module), Caller: caller, Args: argsOf(body), Description: t.Description}, t, caller) }) if err != nil { logf("[mesh-tools] cannot serve %s's %s on %s: %v", v.Seat, v.Verb, v.Subject, err)