diff --git a/node-tools/cmd/node-tools/approvals.go b/node-tools/cmd/node-tools/approvals.go index 99153d6..3f57f65 100644 --- a/node-tools/cmd/node-tools/approvals.go +++ b/node-tools/cmd/node-tools/approvals.go @@ -12,8 +12,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/busretry" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/approval" - "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/bus" ) diff --git a/node-tools/cmd/node-tools/approvals_integration_test.go b/node-tools/cmd/node-tools/approvals_integration_test.go new file mode 100644 index 0000000..2eb2ed1 --- /dev/null +++ b/node-tools/cmd/node-tools/approvals_integration_test.go @@ -0,0 +1,295 @@ +//go:build integration + +package main + +import ( + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "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" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/approval" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "github.com/nats-io/nats.go" + runnerapproval "github.com/novox/mesh-tools/node-tools/internal/approval" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// marked is the fixture marked module's catalog half: its build src2:1 marks its tool wipe (design 46 7-approval-mark §1). +type marked struct{} + +var wipeEntry = catalogseat.CatalogEntry{Module: "shell", Fp: "src2:1", Manifest: map[string]any{"needs-approval": []any{"wipe"}}} + +func (marked) Ready() error { return nil } +func (marked) Revision() string { return "7" } +func (marked) Served(fp string) (approval.Entry, bool, string) { + return runnerapproval.Entry(wipeEntry), fp == wipeEntry.Fp, "not-found" +} +func (marked) Newest(m string) (approval.Entry, bool) { + return runnerapproval.Entry(wipeEntry), m == "shell" +} +func (marked) Definitions() map[string]seatdef.Definition { return map[string]seatdef.Definition{} } +func (marked) Entries() []approval.Entry { return []approval.Entry{runnerapproval.Entry(wipeEntry)} } + +// bed is the runner's gate wired as main wires it, over the runner's own connection to a real bus, with stand-ins of the +// operator channel's router (it takes each ask into its record asks, as the router writes it) and of the hand-act seat. +type bed struct { + t *testing.T + conn *bus.Conn + asks nats.KeyValue + taken chan struct{} + done chan struct{} + + mu sync.Mutex + asked []seat.Ask + recorded []string + holders map[string][]byte +} + +func newBed(t *testing.T, bindState bool) *bed { + t.Helper() + mesh := mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := nc.JetStream() + if _, err := js.AddStream(&nats.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"}, + Retention: nats.WorkQueuePolicy}); err != nil { + t.Fatal(err) + } + b := &bed{t: t, asks: mesh.Bucket(t, "messenger_asks"), taken: make(chan struct{}, 1), done: make(chan struct{}), + holders: map[string][]byte{"operator-channel": []byte(`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"messenger"}]}`)}} + t.Cleanup(func() { close(b.done) }) + mesh.Bucket(t, "node-tools_approvals") + mesh.BucketKeyTTL(t, "node-tools_bound") + + // The router's stand-in: each ask accepted under the marked module's name is taken into its record, state open. + sub, err := js.PullSubscribe("mesh.seat.operator-channel.accept.>", "router", nats.BindStream("SEAT_OPERATOR_CHANNEL")) + if err != nil { + t.Fatal(err) + } + go func() { + for { + select { + case <-b.done: + return + default: + } + msgs, _ := sub.Fetch(1, nats.MaxWait(200*time.Millisecond)) + for _, m := range msgs { + var a seat.Ask + if err := json.Unmarshal(m.Data, &a); err == nil { + module := strings.TrimPrefix(m.Subject, "mesh.seat.operator-channel.accept.ask.") + b.mu.Lock() + b.asked = append(b.asked, a) + b.mu.Unlock() + b.write(seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", Headline: a.Headline, + Expires: a.Expires, Opened: operatorchannelclient.FormatTime(time.Now()), OfferedOn: []string{"telegram"}}) + select { + case b.taken <- struct{}{}: + default: + } + } + _ = m.Ack() + } + } + }() + // The hand-act seat's stand-in: warranted {asker, ask} answered, each one kept. + if _, err := nc.Subscribe("mesh.seat.mesh-hand-acts.>", func(m *nats.Msg) { + b.mu.Lock() + b.recorded = append(b.recorded, string(m.Data)) + b.mu.Unlock() + _ = m.Respond([]byte(`{"result":{"recorded":true}}`)) + }); err != nil { + t.Fatal(err) + } + + runner := mt.MembershipOf("node-tools", "n1", false, nil) + if bindState { + runner.State = []bus.StateIssued{{Name: approval.StateName, Bucket: "node-tools_approvals", Writes: true}, + {Name: BoundState, Bucket: "node-tools_bound", Writes: true}} + } + shell := mt.MembershipOf("shell", "n1", false, nil) + shell.SeatTraffic = &bus.SeatTraffic{Publish: []string{seat.AcceptSubjectAsk("shell")}, + Records: []string{"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.shell.>"}} + mesh.Issue(t, runner) + mesh.Issue(t, shell) + b.conn, err = bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "n1"}) + if err != nil { + t.Fatal(err) + } + b.conn.Logf = func(string, ...any) {} + t.Cleanup(b.conn.Close) + b.conn.Follow("node-tools") + b.conn.Follow("shell") + mt.Until(t, func() error { + if b.conn.Membership("node-tools") == nil || b.conn.Membership("shell") == nil { + return errors.New("not issued yet") + } + return nil + }) + return b +} + +// gate is a runner's gate as main makes it; record says how the run's recording goes beside the call (nil: at once). +func (b *bed) gate(record func(func())) *approval.Gate { + if record == nil { + record = func(f func()) { f() } + } + g := approval.New(approval.Deps{Node: "n1", Runner: "node-tools", Catalog: marked{}, Logf: b.t.Logf, + State: approvalsState{conn: b.conn, module: "node-tools"}, + Channel: &operatorchannelclient.GateChannel{State: approvalsState{conn: b.conn, module: "node-tools"}, KeyPrefix: "n1.", + Publish: publishAsk(b.conn), ReadRecord: routerRecord(b.conn)}, + Warranted: warranted(b.conn), Bounds: boundState{conn: b.conn, module: "node-tools"}, Go: record, + Holder: runnerapproval.HoldersIn(func() (map[string][]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.holders, nil + }), + // The pause while the router takes the ask ends when the stand-in has taken one: no sleeping. + After: func(time.Duration) <-chan time.Time { + c := make(chan time.Time, 1) + go func() { + select { + case <-b.taken: + case <-b.done: + } + c <- time.Now() + }() + return c + }}) + _ = g.Unserved("shell", "src2:1", []string{"wipe"}) + return g +} + +func (b *bed) write(rec seat.AsksByAskerIDValue) { + raw, _ := json.Marshal(rec) + if _, err := b.asks.Put(rec.Key, raw); err != nil { + b.t.Fatal(err) + } +} + +// decide writes the router's record of the newest ask as the router writes the operator's choice: chosen, with its warrant. +func (b *bed) decide(option string) seat.Ask { + b.mu.Lock() + a := b.asked[len(b.asked)-1] + b.mu.Unlock() + e, err := b.asks.Get("shell." + a.ID) + if err != nil { + b.t.Fatal(err) + } + var rec seat.AsksByAskerIDValue + _ = json.Unmarshal(e.Value(), &rec) + rec.State = "chosen" + rec.Warrant = &seat.Warrant{Ask: a.ID, Asker: "shell", Outcome: operatorchannelclient.OutcomeChosen, Option: option, + Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(time.Now()), AskDigest: operatorchannelclient.Digest(a), + Channel: "telegram", By: &seat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}} + b.write(rec) + return a +} + +func wipe(args string) approval.Call { + return approval.Call{Address: "shell.wipe", Module: "shell", Name: "wipe", Fingerprint: "src2:1", Caller: "agent", + Args: json.RawMessage(args), Description: "Wipes a scratch directory."} +} + +// Design 46 7-approval-mark §22: the runner's gate against the real bus with a fixture marked module: a call asked and taken +// by the operator channel's stand-in, the operator's Approve written as the router's record, the call again run once and +// recorded in the hand-act log, mesh_approvals' list, a restart between the run and warranted's answer recorded once; a +// Decline runs nothing; a gate missing a part of its check runs nothing. +func TestAMarkedCallOnTheBusIsAskedRunOnTheApprovalAndRecorded(t *testing.T) { + t.Run("approved: asked, run once, recorded, listed", func(t *testing.T) { + b := newBed(t, true) + g := b.gate(nil) + runs := 0 + run := func() (any, error) { runs++; return "wiped", nil } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.HasPrefix(err.Error(), "Not run: this needs the operator's approval. Asked on telegram at ") || runs != 0 { + t.Fatalf("first call: %v, runs %d", err, runs) + } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "still waiting for the operator's answer") { + t.Fatalf("before the answer: %v", err) + } + a := b.decide(approval.OptionApprove) + got, err := g.Run(wipe(`{"dir":"scratch"}`), run) + answered, ok := got.(approval.Answered) + if err != nil || !ok || answered.Answer != "wiped" || !strings.Contains(answered.Approval, "ask "+a.ID) || runs != 1 { + t.Fatalf("on the approval: %+v, %v, runs %d", got, err, runs) + } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || runs != 1 { + t.Fatalf("the warrant ran twice: %v, runs %d", err, runs) + } + b.mu.Lock() + recorded := strings.Join(b.recorded, "\n") + b.mu.Unlock() + if !strings.Contains(recorded, a.ID) || !strings.Contains(recorded, "shell") { + t.Fatalf("hand-act log heard %q", recorded) + } + list, err := g.List() + if err != nil || len(list) == 0 || list[len(list)-1].Ask != a.ID || list[len(list)-1].State != "run, recorded" { + t.Fatalf("mesh_approvals lists %+v, %v", list, err) + } + }) + t.Run("a restart between the run and warranted's answer records the run once", func(t *testing.T) { + b := newBed(t, true) + stopped := b.gate(func(func()) {}) // the runner stops before warranted answers + runs := 0 + run := func() (any, error) { runs++; return "wiped", nil } + _, _ = stopped.Run(wipe(`{"dir":"tmp"}`), run) + a := b.decide(approval.OptionApprove) + if _, err := stopped.Run(wipe(`{"dir":"tmp"}`), run); err != nil || runs != 1 { + t.Fatalf("run: %v, runs %d", err, runs) + } + if list, _ := stopped.List(); len(list) != 1 || list[0].State != "run, not yet recorded" { + t.Fatalf("before the restart: %+v", list) + } + started := b.gate(nil) + started.Recover() + started.Recover() + b.mu.Lock() + n := len(b.recorded) + b.mu.Unlock() + if list, _ := started.List(); n != 1 || len(list) != 1 || list[0].Ask != a.ID || list[0].State != "run, recorded" || runs != 1 { + t.Fatalf("after the restart: recorded %d, %+v, runs %d", n, list, runs) + } + }) + t.Run("declined: refused, nothing runs", func(t *testing.T) { + b := newBed(t, true) + g := b.gate(nil) + runs := 0 + run := func() (any, error) { runs++; return nil, nil } + _, _ = g.Run(wipe(`{"dir":"scratch"}`), run) + b.decide(approval.OptionDecline) + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "the operator declined ask") || runs != 0 { + t.Fatalf("declined: %v, runs %d", err, runs) + } + }) + t.Run("failing closed: no state bound, or nobody to ask, runs nothing and asks nothing", func(t *testing.T) { + unbound := newBed(t, false) + runs := 0 + run := func() (any, error) { runs++; return nil, nil } + if _, err := unbound.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.Contains(err.Error(), "binds no state approvals") || runs != 0 || len(unbound.asked) != 0 { + t.Fatalf("no state: %v, runs %d", err, runs) + } + b := newBed(t, true) + b.holders = map[string][]byte{} + if _, err := b.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.Contains(err.Error(), "no module holds operator-channel") || runs != 0 { + t.Fatalf("no holder: %v, runs %d", err, runs) + } + if got, err := b.gate(nil).Run(approval.Call{Address: "shell.tidy", Module: "shell", Name: "tidy", Fingerprint: "src2:1", + Args: json.RawMessage(`{}`)}, func() (any, error) { return "tidied", nil }); err != nil || got != "tidied" { + t.Fatalf("an unmarked tool: %v, %v", got, err) + } + }) +} diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 788dd82..07b66fe 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -26,9 +26,11 @@ import ( "time" handacts "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts" - "github.com/novox/mesh-tools/node-tools/internal/alive" + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + "git.novox.be/novox/mesh-sdk/go/approval" - "github.com/novox/mesh-tools/node-tools/internal/approval" + "github.com/novox/mesh-tools/node-tools/internal/alive" + runnerapproval "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/console" "github.com/novox/mesh-tools/node-tools/internal/firstview" @@ -77,10 +79,12 @@ func main() { // ADR 0615 §1; design 46 7-approval-mark §5). source, stopViews := firstview.Start(context.Background(), conn, log.Printf) defer stopViews() + approvals := approvalsState{conn: conn, module: cred.Module} 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) }), + Findings: runnerapproval.Findings(source.Findings(), conn.Node(), cred.Module, nil), Logf: log.Printf, State: approvals, + Channel: &operatorchannelclient.GateChannel{State: approvals, KeyPrefix: conn.Node() + ".", Publish: publishAsk(conn), ReadRecord: routerRecord(conn)}, + Warranted: warranted(conn), Bounds: boundState{conn: conn, module: cred.Module}, + Holder: runnerapproval.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). diff --git a/node-tools/cmd/node-tools/main_test.go b/node-tools/cmd/node-tools/main_test.go index a9d093a..0018932 100644 --- a/node-tools/cmd/node-tools/main_test.go +++ b/node-tools/cmd/node-tools/main_test.go @@ -13,10 +13,17 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/runtime" ) +// The runner's start waits for its credential and its bus within the start's bound (issue 1496): one test, two cases, so +// design 46 7-approval-mark §22's whole test of the gate adds no test to the module (design 47 G §419, §626). +func TestTheRunnersStartWaitsForItsCredentialAndItsBus(t *testing.T) { + t.Run("the credential", awaitsCredential) + t.Run("the bus", connectsWithin) +} + // A runtime started before its credential is on the machine waits for it, woken by the watch of its directory, and gives // up loudly at its bound; a file that is no credential ends the wait at once (issue 1496: on ace mesh-build-node's and // mesh-build-image's runtimes exited twice on "cannot read the broker credential ... no such file"). -func TestAwaitCredential(t *testing.T) { +func awaitsCredential(t *testing.T) { const good = `{"url":"tls://bus.example:4222","module":"m"}` rereadEvery = time.Hour // only the watch can wake the read in time settle := runtime.Settle @@ -82,7 +89,7 @@ func TestAwaitCredential(t *testing.T) { // A bus that refuses the account at a start is asked again until the start's bound, said once; a refusal past it, a pin // that does not match and an address that is no URL give up (issue 1496, issue 058). -func TestConnectWithin(t *testing.T) { +func connectsWithin(t *testing.T) { refused := errors.New("nats: Authorization Violation") unreachable := errors.New("nats: no servers available for connection") for _, c := range []struct { diff --git a/node-tools/internal/approval/act.go b/node-tools/internal/approval/act.go deleted file mode 100644 index 5ad9df8..0000000 --- a/node-tools/internal/approval/act.go +++ /dev/null @@ -1,293 +0,0 @@ -package approval - -import ( - "bytes" - "encoding/json" - "fmt" - "math" - "sort" - "strconv" - "strings" - "unicode/utf8" - - "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" - "git.novox.be/novox/mesh-sdk/go/outward" -) - -// Call is one call the runner serves, as the gate judges it: a module's tool, or a seat verb its module serves. -type Call struct { - // Address is the address as called: `.` or `.`. - Address string - // Module is the module whose bundle serves it; Seat the seat of a verb, empty for a tool; Name the tool or the verb. - Module, Seat, Name string - // Fingerprint is the build the bundle serves, from its composed environment (§5). - Fingerprint string - // Caller is the caller the bus proved. - Caller string - // Args are the call's arguments, a JSON object. - 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). -const MostArguments = 16 * 1024 - -// arguments are a call's arguments by name; a body that is not an object is refused. -func arguments(raw json.RawMessage) (map[string]json.RawMessage, error) { - args := map[string]json.RawMessage{} - if len(bytes.TrimSpace(raw)) == 0 || string(bytes.TrimSpace(raw)) == "null" { - return args, nil - } - if err := json.Unmarshal(raw, &args); err != nil { - return nil, fmt.Errorf("its arguments are not a JSON object: %v", err) - } - return args, nil -} - -// refusedUnasked says why a marked call is refused at once, asking nothing (§9): its arguments hold a secret's shape, or are -// more than MostArguments. -func refusedUnasked(raw json.RawMessage) string { - if len(raw) > MostArguments { - return fmt.Sprintf("its arguments are %d bytes, more than the %d a call that needs the operator's approval may carry", len(raw), MostArguments) - } - if r, ok := outward.CheckSecret(string(raw)); !ok { - return fmt.Sprintf("its arguments hold %s, which no ask may carry", r.What) - } - return "" -} - -// ActOf is the act a marked call binds (§8): `address`, `node`, `module` and `tool` or `seat` and `verb`, `build`, and -// `arg.` for each argument given, a string as itself and any other value as its JSON in RFC 8785's canonical form. -func ActOf(c Call, node string) (operatorchannelclient.Act, error) { - act := operatorchannelclient.Act{"address": c.Address, "node": node, "build": c.Fingerprint} - if c.Seat == "" { - act["module"], act["tool"] = c.Module, c.Name - } else { - act["seat"], act["verb"] = c.Seat, c.Name - } - args, err := arguments(c.Args) - if err != nil { - return nil, err - } - for name, raw := range args { - var s string - if json.Unmarshal(raw, &s) == nil { - act["arg."+name] = s - continue - } - canon, err := Canonical(raw) - if err != nil { - return nil, fmt.Errorf("the argument %s cannot be put in canonical form: %v", name, err) - } - act["arg."+name] = canon - } - return act, nil -} - -// Canonical is a JSON value in RFC 8785's canonical form: no white space, an object's members sorted by their names' UTF-16 -// code units, a number as ECMAScript writes a double, a string with only the escapes the RFC names. -func Canonical(raw json.RawMessage) (string, error) { - dec := json.NewDecoder(bytes.NewReader(raw)) - dec.UseNumber() - var v any - if err := dec.Decode(&v); err != nil { - return "", err - } - var b strings.Builder - if err := canonicalValue(&b, v); err != nil { - return "", err - } - return b.String(), nil -} - -func canonicalValue(b *strings.Builder, v any) error { - switch x := v.(type) { - case nil: - b.WriteString("null") - case bool: - b.WriteString(strconv.FormatBool(x)) - case json.Number: - f, err := strconv.ParseFloat(string(x), 64) - if err != nil || math.IsInf(f, 0) || math.IsNaN(f) { - return fmt.Errorf("the number %s is not a double", x) - } - b.WriteString(es6Number(f)) - case string: - canonicalString(b, x) - case []any: - b.WriteByte('[') - for i, e := range x { - if i > 0 { - b.WriteByte(',') - } - if err := canonicalValue(b, e); err != nil { - return err - } - } - b.WriteByte(']') - case map[string]any: - names := make([]string, 0, len(x)) - for k := range x { - names = append(names, k) - } - sort.Slice(names, func(i, j int) bool { return utf16Less(names[i], names[j]) }) - b.WriteByte('{') - for i, k := range names { - if i > 0 { - b.WriteByte(',') - } - canonicalString(b, k) - b.WriteByte(':') - if err := canonicalValue(b, x[k]); err != nil { - return err - } - } - b.WriteByte('}') - default: - return fmt.Errorf("a value of type %T", v) - } - return nil -} - -// es6Number is a double as ECMAScript's Number.prototype.toString writes it (RFC 8785 §3.2.2.3). -func es6Number(f float64) string { - if f == 0 { - return "0" - } - abs := math.Abs(f) - if abs >= 1e-6 && abs < 1e21 { - return strconv.FormatFloat(f, 'f', -1, 64) - } - s := strconv.FormatFloat(f, 'e', -1, 64) // 1e+21, 1.5e-07 - mant, exp, _ := strings.Cut(s, "e") - sign, digits := exp[:1], strings.TrimLeft(exp[1:], "0") - return mant + "e" + sign + digits -} - -func canonicalString(b *strings.Builder, s string) { - b.WriteByte('"') - for _, r := range s { - switch { - case r == '"': - b.WriteString(`\"`) - case r == '\\': - b.WriteString(`\\`) - case r == '\b': - b.WriteString(`\b`) - case r == '\f': - b.WriteString(`\f`) - case r == '\n': - b.WriteString(`\n`) - case r == '\r': - b.WriteString(`\r`) - case r == '\t': - b.WriteString(`\t`) - case r < 0x20: - fmt.Fprintf(b, `\u%04x`, r) - default: - b.WriteRune(r) - } - } - b.WriteByte('"') -} - -func utf16Less(a, b string) bool { - ua, ub := utf16Of(a), utf16Of(b) - for i := 0; i < len(ua) && i < len(ub); i++ { - if ua[i] != ub[i] { - return ua[i] < ub[i] - } - } - return len(ua) < len(ub) -} - -func utf16Of(s string) []uint16 { - out := make([]uint16, 0, utf8.RuneCountInString(s)) - for _, r := range s { - if r >= 0x10000 { - r -= 0x10000 - out = append(out, uint16(0xD800+(r>>10)), uint16(0xDC00+(r&0x3FF))) - } else { - out = append(out, uint16(r)) - } - } - return out -} - -// Fingerprint is how an act is shown: its digest's first 24 hexadecimal digits in groups of four (§10). -func Fingerprint(digest string) string { - hex := strings.TrimPrefix(digest, "sha256:") - if len(hex) > 24 { - hex = hex[:24] - } - var groups []string - for len(hex) > 4 { - groups, hex = append(groups, hex[:4]), hex[4:] - } - return strings.Join(append(groups, hex), " ") -} - -// The bounds of the arguments shown in an ask (§11). -const ( - ShownBytes, ShownLines = 2000, 40 - PartBytes, PartLines = 1200, 20 -) - -// shown is a call's arguments as an ask shows them (§11), one per line as "name: value", and whether all were shown whole: -// an argument whole where outward.Check passes it, else its shape, ‹path›, ‹address› or ‹withheld›; the whole when it is at -// most ShownBytes and ShownLines, else its first PartLines lines and PartBytes bytes, marked NOT SHOWN IN FULL. -func shown(raw json.RawMessage) (string, bool) { - args, _ := arguments(raw) - names := make([]string, 0, len(args)) - for n := range args { - names = append(names, n) - } - sort.Strings(names) - whole := true - var lines []string - for _, n := range names { - var value string - if json.Unmarshal(args[n], &value) != nil { - value = string(args[n]) - } - if r, ok := outward.Check(value); !ok { - value, whole = shapeOf(r), false - } - lines = append(lines, n+": "+value) - } - text := strings.Join(lines, "\n") - if all := strings.Split(text, "\n"); len(text) > ShownBytes || len(all) > ShownLines { - if len(all) > PartLines { - all = all[:PartLines] - } - text = strings.Join(all, "\n") - if len(text) > PartBytes { - text = cutBytes(text, PartBytes) - } - text += "\nNOT SHOWN IN FULL" - whole = false - } - return text, whole -} - -// shapeOf is what an argument outward.Check refuses is shown as. -func shapeOf(r outward.Refusal) string { - switch r.Class { - case "path": - return "‹path›" - case "address": - return "‹address›" - } - return "‹withheld›" -} - -// cutBytes is s at most n bytes, cut at a character's end. -func cutBytes(s string, n int) string { - for n > 0 && !utf8.RuneStart(s[n]) { - n-- - } - return s[:n] -} diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index ef5907f..8bc98ff 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -12,6 +12,7 @@ import ( seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" stdio "git.novox.be/novox/mesh-sdk/go" + gate "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/seatdef" "git.novox.be/novox/mesh-sdk/go/views" @@ -27,7 +28,7 @@ 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. type world struct { - g *Gate + g *gate.Gate state *mem cat *fakeCatalog records map[string]seat.AsksByAskerIDValue // by `.` @@ -43,6 +44,16 @@ type world struct { // warranted is what mesh-hand-acts.warranted answers, and recorded each ask it answered for. warranted error recorded []string + // deps is the gate's, made anew by rebuild; publish and record are the bus's publish and the router's record read. + deps gate.Deps + publish func(module, subject string, body any, id string) error + record func(module, id string) (*stdio.StateEntry, error) +} + +// rebuild makes the gate anew from w.deps, its module's bundle loaded. +func (w *world) rebuild() { + w.g = gate.New(w.deps) + _ = w.g.Unserved("shell", "src2:1", []string{"wipe", "tidy"}) } // bounds is the runner's key-ttl state bound as a fixture: expire hands a key's expiry to the watch. @@ -72,7 +83,11 @@ func newWorld() *world { newest: map[string]catalogseat.CatalogEntry{"shell": shell}, defs: map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", Verbs: []seatdef.Verb{{Name: "execute", NeedsApproval: true}}}}}} - w.g = New(Deps{Node: "n1", Runner: "node-tools", Catalog: w.cat, State: w.state, Findings: w.findings, Bounds: w.bounds, + w.deps = gate.Deps{Node: "n1", Runner: "node-tools", Catalog: w.cat, State: w.state, Bounds: w.bounds, + Findings: Findings(w.findings, "n1", "node-tools", func() time.Time { return w.now }), + Channel: &operatorchannelclient.GateChannel{State: w.state, KeyPrefix: "n1.", Now: func() time.Time { return w.now }, + Publish: func(module, subject string, body any, id string) error { return w.publish(module, subject, body, id) }, + ReadRecord: func(module, id string) (*stdio.StateEntry, error) { return w.record(module, id) }}, Go: func(f func()) { f() }, Holder: HoldersIn(func() (map[string][]byte, error) { return w.holders, w.holdersErr }), Warranted: func(asker, ask string, open func() (bool, error)) error { @@ -91,27 +106,28 @@ func newWorld() *world { c <- w.now return c }, - Publish: func(module, subject string, body any, id string) error { - if subject != seat.AcceptSubjectAsk(module) { - return nil - } - a := body.(seat.Ask) - w.asked = append(w.asked, a) - if w.takes { - w.records[module+"."+a.ID] = seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", - Headline: a.Headline, Expires: a.Expires, Opened: operatorchannelclient.FormatTime(w.now), OfferedOn: []string{"telegram"}} - } + } + w.publish = func(module, subject string, body any, id string) error { + if subject != seat.AcceptSubjectAsk(module) { return nil - }, - Record: func(module, id string) (*stdio.StateEntry, error) { - rec, ok := w.records[module+"."+id] - if !ok { - return nil, nil - } - raw, _ := json.Marshal(rec) - return &stdio.StateEntry{Key: module + "." + id, Value: raw}, nil - }}) - _ = w.g.Unserved("shell", "src2:1", []string{"wipe", "tidy"}) + } + a := body.(seat.Ask) + w.asked = append(w.asked, a) + if w.takes { + w.records[module+"."+a.ID] = seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", + Headline: a.Headline, Expires: a.Expires, Opened: operatorchannelclient.FormatTime(w.now), OfferedOn: []string{"telegram"}} + } + return nil + } + w.record = func(module, id string) (*stdio.StateEntry, error) { + rec, ok := w.records[module+"."+id] + if !ok { + return nil, nil + } + raw, _ := json.Marshal(rec) + return &stdio.StateEntry{Key: module + "." + id, Value: raw}, nil + } + w.rebuild() return w } @@ -129,102 +145,38 @@ func (w *world) decide(option string, at time.Time) seat.Ask { } // do makes a call through the gate, counting the runs. -func (w *world) do(c Call) (any, error) { +func (w *world) do(c gate.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), +func call(name, args string) gate.Call { + return gate.Call{Address: "shell." + name, Module: "shell", Name: name, Fingerprint: "src2:1", Caller: "agent", Args: json.RawMessage(args), Description: "Wipes a scratch directory.\nMore words."} } -// theActAndItsAsk is §8's and §11's rows: the act's fields, and how an ask shows a call's arguments. -func theActAndItsAsk(t *testing.T) { - base := call("wipe", `{"dir":"scratch","n":1}`) - digest := func(c Call) string { - act, err := ActOf(c, "n1") - if err != nil { - t.Fatal(err) - } - d, _ := operatorchannelclient.ActDigest(act) - return d - } - act, _ := ActOf(base, "n1") - want := operatorchannelclient.Act{"address": "shell.wipe", "node": "n1", "module": "shell", "tool": "wipe", "build": "src2:1", - "arg.dir": "scratch", "arg.n": "1"} - if len(act) != len(want) { - t.Fatalf("act %v", act) - } - for k, v := range want { - if act[k] != v { - t.Errorf("act[%s] = %q, want %q", k, act[k], v) - } - } - seen := map[string]string{digest(base): "the call"} - for name, c := range map[string]Call{ - "an omitted argument": call("wipe", `{"dir":"scratch"}`), - "another number": call("wipe", `{"dir":"scratch","n":2}`), - "an object": call("wipe", `{"dir":{"name":"scratch"},"n":1}`), - "a changed build": func() Call { c := base; c.Fingerprint = "src2:2"; return c }(), - "another node's seat": {Address: "node-login-shell.execute", Module: "shell", Seat: "node-login-shell", Name: "execute", Fingerprint: "src2:1"}, - } { - d := digest(c) - if other, dup := seen[d]; dup { - t.Errorf("%s gives the digest of %s", name, other) - } - seen[d] = name - } - if got, _ := Canonical(json.RawMessage(`{"b":[1.0,2.5e-7,1e21],"a":"é<"}`)); got != `{"a":"é<","b":[1,2.5e-7,1e+21]}` { - t.Errorf("canonical form %s", got) - } - // §11: a path, an address and a long argument are shown in their shape and cut. - text, whole := shown(json.RawMessage(`{"dir":"/home/someone/scratch","host":"10.0.0.7","ok":"yes"}`)) - if whole || text != "dir: ‹path›\nhost: ‹address›\nok: yes" { - t.Errorf("shown %q whole %v", text, whole) - } - long, _ := json.Marshal(map[string]string{"text": strings.Repeat("word ", 500)}) - if text, whole := shown(long); whole || !strings.HasSuffix(text, "NOT SHOWN IN FULL") || len(text) > PartBytes+len("\nNOT SHOWN IN FULL") { - t.Errorf("a long argument shown whole %v, %d bytes", whole, len(text)) - } - // §10: the headline, cut with … to 60 characters, ending Part shown. - for _, tc := range []struct{ name, node, want string }{ - {"wipe", "n1", "Run wipe on n1?"}, - {strings.Repeat("x", 50), "n1", "Run " + strings.Repeat("x", 50) + "?"}, - {strings.Repeat("y", 80), "n1", "Run " + strings.Repeat("y", 54) + "…?"}, - } { - 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("Run", "wipe", "n1", false); got != "Run wipe on n1? Part shown" { - t.Errorf("headline %q", got) - } -} - // Design 46 7-approval-mark §22: the refusals and answers of a marked call, on the fixture mesh, the router's record and // the warrant written as the router writes them. func TestTheRefusalsAndAnswers(t *testing.T) { - t.Run("§8, §10, §11: the act's fields give each its digest, and the ask shows the arguments", theActAndItsAsk) for _, tc := range []struct { name string run func(t *testing.T, w *world) }{ {"§9: a secret in the arguments is refused asking nothing", func(t *testing.T, w *world) { - _, err := w.g.firstCall(call("wipe", `{"token":"ghp_0123456789abcdefghijABCDEFGHIJ012345"}`)) + _, err := w.do(call("wipe", `{"token":"ghp_0123456789abcdefghijABCDEFGHIJ012345"}`)) if err == nil || !strings.Contains(err.Error(), "which no ask may carry") || len(w.asked) != 0 { t.Fatalf("%v, asked %d", err, len(w.asked)) } }}, {"§9: more than 16 KiB of arguments is refused asking nothing", func(t *testing.T, w *world) { big, _ := json.Marshal(map[string]string{"text": strings.Repeat("a b ", 5000)}) - _, err := w.g.firstCall(call("wipe", string(big))) + _, err := w.do(call("wipe", string(big))) if err == nil || !strings.Contains(err.Error(), "more than the 16384") || len(w.asked) != 0 { t.Fatalf("%v, asked %d", err, len(w.asked)) } }}, {"§10, §12: the published ask's fields, and Not run with the router's record", func(t *testing.T, w *world) { - _, err := w.g.firstCall(call("wipe", `{"dir":"scratch"}`)) + _, err := w.do(call("wipe", `{"dir":"scratch"}`)) if err == nil || !strings.HasPrefix(err.Error(), "Not run: this needs the operator's approval. Asked on telegram at ") || !strings.Contains(err.Error(), "make the same call again within 30 minutes of the approval; the ask expires at ") { t.Fatalf("answer %v", err) @@ -233,9 +185,9 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("asked %d", len(w.asked)) } a := w.asked[0] - act, _ := ActOf(call("wipe", `{"dir":"scratch"}`), "n1") + act, _ := gate.ActOf(call("wipe", `{"dir":"scratch"}`), "n1") d, _ := operatorchannelclient.ActDigest(act) - if a.About != d || a.Who != "operator" || a.OnExpiry != OnExpiry || a.Headline != "Run wipe on n1?" || + if a.About != d || a.Who != "operator" || a.OnExpiry != gate.OnExpiry || a.Headline != "Run wipe on n1?" || a.Expires != operatorchannelclient.FormatTime(time.Date(2026, 11, 12, 11, 0, 0, 0, time.UTC)) || len(a.Options) != 2 || a.Options[0].ID != "approve" || a.Options[1].ID != "decline" || a.Options[0].Binds != d || a.Options[1].Binds != d || a.Options[1].Level != "approve" { @@ -246,15 +198,15 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } 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]) + if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.Vals["n1.act."+keys.EscapeName(d)]) != `"`+a.ID+`"` { + t.Errorf("kept %+v, act key %s", k, w.state.Vals["n1.act."+keys.EscapeName(d)]) } }}, {"§12: with no record within 5 seconds, Not run, asked; not yet taken by the router", func(t *testing.T, w *world) { w.takes = false start := w.now - _, err := w.g.firstCall(call("wipe", `{}`)) - if err == nil || !strings.HasPrefix(err.Error(), "Not run, asked; not yet taken by the router (ask ") || w.now.Sub(start) > RecordWithin { + _, err := w.do(call("wipe", `{}`)) + if err == nil || !strings.HasPrefix(err.Error(), "Not run, asked; not yet taken by the router (ask ") || w.now.Sub(start) > gate.RecordWithin { t.Fatalf("answer %v after %s", err, w.now.Sub(start)) } }}, @@ -277,10 +229,10 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§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.decide(gate.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.(Answered).Answer != "done" || w.runs != 1 || - !strings.HasPrefix(got.(Answered).Approval, "run on the operator's approval via telegram at 2026-11-12T10:01:00Z, ask ") { + if got, err := w.do(call("wipe", `{"dir":"scratch"}`)); err != nil || got.(gate.Answered).Answer != "done" || w.runs != 1 || + !strings.HasPrefix(got.(gate.Answered).Approval, "run on the operator's approval via telegram at 2026-11-12T10:01:00Z, ask ") { t.Fatalf("%v %v, runs %d", got, err, w.runs) } _, err := w.do(call("wipe", `{"dir":"scratch"}`)) @@ -290,7 +242,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§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) + w.decide(gate.OptionApprove, w.now) var wg sync.WaitGroup var mu sync.Mutex runs := 0 @@ -308,9 +260,9 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§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) + a := w.decide(gate.OptionApprove, w.now) rec := w.records["shell."+a.ID] - other, _ := ActOf(call("wipe", `{"dir":"home"}`), "n1") + other, _ := gate.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) @@ -326,7 +278,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§14: approved 31 minutes ago asks anew", func(t *testing.T, w *world) { _, _ = w.do(call("wipe", `{}`)) - w.decide(OptionApprove, w.now) + w.decide(gate.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)) @@ -334,7 +286,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§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.decide(gate.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)) @@ -356,7 +308,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { }}, {"§14: another asker's warrant runs nothing", func(t *testing.T, w *world) { _, _ = w.do(call("wipe", `{}`)) - a := w.decide(OptionApprove, w.now) + a := w.decide(gate.OptionApprove, w.now) rec := w.records["shell."+a.ID] rec.Warrant.Asker = "mail" w.records["shell."+a.ID] = rec @@ -382,7 +334,8 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } }}, {"§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 + w.deps.State = nil + w.rebuild() if got, err := w.do(call("tidy", `{}`)); err != nil || got != "done" { t.Fatalf("unmarked: %v %v", got, err) } @@ -390,6 +343,18 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("marked: %v, runs %d", err, w.runs) } }}, + {"§13: the fixture state refuses a key the bus refuses, so every key the gate writes here is one the bus holds (A-209)", func(t *testing.T, w *world) { + if _, err := w.state.Put("n1.act.sha256:00", "x"); err == nil { + t.Fatal("the fixture state held a key with a colon, which the bus refuses") + } + if _, err := w.state.Create("n1..ask", "x"); err == nil { + t.Fatal("the fixture state held a key with an empty name") + } + _, _ = w.do(call("wipe", `{"dir":"scratch"}`)) + if len(w.asked) != 1 || len(w.state.Vals) == 0 { + t.Fatalf("the gate kept nothing through the fixture state: asked %d, keys %d", len(w.asked), len(w.state.Vals)) + } + }}, {"§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) @@ -398,37 +363,38 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } }}, {"§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"}`)} + c := gate.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)) } }}, {"§16: the run is recorded once; unanswered within five minutes it is a finding, cleared when warranted answers", func(t *testing.T, w *world) { _, _ = w.do(call("wipe", `{}`)) - a := w.decide(OptionApprove, w.now) + a := w.decide(gate.OptionApprove, w.now) stop, _ := w.g.WatchBounds() defer stop() w.warranted = errors.New("no responders") if _, err := w.do(call("wipe", `{}`)); err != nil || w.runs != 1 || len(w.recorded) != 0 { t.Fatalf("%v, runs %d, recorded %v", err, w.runs, w.recorded) } - if w.bounds.keys["n1.unanswered."+a.ID] != UnansweredAfter || w.bounds.keys["n1.unrecorded."+a.ID] != KeptFor-SaidBefore { + if w.bounds.keys["n1.unanswered."+a.ID] != gate.UnansweredAfter || w.bounds.keys["n1.unrecorded."+a.ID] != gate.KeptFor-gate.SaidBefore { t.Fatalf("bounds %v", w.bounds.keys) } w.bounds.expire("n1.unanswered." + a.ID) - if w.findings.raised[keys.EscapeName("n1."+a.ID)] != KindCallUnanswered { + if w.findings.raised[keys.EscapeName("n1."+a.ID)] != gate.KindCallUnanswered { t.Fatalf("findings %v", w.findings.raised) } w.warranted = nil // At a start each acted, unrecorded ask is pending in the runner's findings-stated until warranted is made again // (design 47 G §384). - pendingKey := views.FindingKey("machine", keys.EscapeName("n1."+a.ID), KindCallUnanswered) - warranted := w.g.d.Warranted + pendingKey := views.FindingKey("machine", keys.EscapeName("n1."+a.ID), gate.KindCallUnanswered) + warranted := w.deps.Warranted var pendingAtCall bool - w.g.d.Warranted = func(asker, ask string, open func() (bool, error)) error { + w.deps.Warranted = func(asker, ask string, open func() (bool, error)) error { _, pendingAtCall = w.findings.pends[pendingKey] return warranted(asker, ask, open) } + w.rebuild() // a start w.g.Recover() // a start, or the hand-act seat's holder heard again w.g.Recover() if !pendingAtCall || len(w.findings.pended) != 1 || len(w.findings.pends) != 0 || w.findings.pended[0] != pendingKey { @@ -454,15 +420,15 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } }}, {"§19: a publish that fails and a record that cannot be read refuse, running nothing", func(t *testing.T, w *world) { - pub := w.g.d.Publish - w.g.d.Publish = func(string, string, any, string) error { return errors.New("the bus refused it") } + pub := w.publish + w.publish = func(string, string, any, string) error { return errors.New("the bus refused it") } if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "the ask could not be raised") { t.Fatalf("%v", err) } - w.g.d.Publish = pub + w.publish = pub 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") } + w.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 { t.Fatalf("%v, runs %d", err, w.runs) } @@ -477,11 +443,13 @@ func TestTheRefusalsAndAnswers(t *testing.T) { {"§19: the marks unread, a call waits within 5 seconds and is refused naming mesh_first_views", func(t *testing.T, w *world) { w.cat.ready = errors.New("the catalog half has no first view of main yet") waited := time.Duration(0) - w.g.d.Wait = func(d time.Duration) error { waited = d; return w.cat.ready } - if _, err := w.do(call("tidy", `{}`)); err == nil || !strings.Contains(err.Error(), "mesh_first_views") || waited != RecordWithin || w.runs != 0 { + w.deps.Wait = func(d time.Duration) error { waited = d; return w.cat.ready } + w.rebuild() + if _, err := w.do(call("tidy", `{}`)); err == nil || !strings.Contains(err.Error(), "mesh_first_views") || waited != gate.RecordWithin || w.runs != 0 { t.Fatalf("%v, waited %s", err, waited) } - w.g.d.Wait = func(time.Duration) error { w.cat.ready = nil; return nil } + w.deps.Wait = func(time.Duration) error { w.cat.ready = nil; return nil } + w.rebuild() if _, err := w.do(call("tidy", `{}`)); err != nil || w.runs != 1 { t.Fatalf("read within the wait: %v, runs %d", err, w.runs) } @@ -493,11 +461,11 @@ func TestTheRefusalsAndAnswers(t *testing.T) { if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "cannot be read") { t.Fatalf("%v", err) } - if w.bounds.keys["n1.unreadable.approvals"] != UnreadableAfter || len(w.findings.raised) != 0 { + if w.bounds.keys["n1.unreadable.approvals"] != gate.UnreadableAfter || len(w.findings.raised) != 0 { t.Fatalf("bounds %v, findings %v", w.bounds.keys, w.findings.raised) } w.bounds.expire("n1.unreadable.approvals") - if w.findings.raised[keys.EscapeName("n1.approvals")] != KindApprovalsUnreadable { + if w.findings.raised[keys.EscapeName("n1.approvals")] != gate.KindApprovalsUnreadable { t.Fatalf("findings %v", w.findings.raised) } w.state.Fail = nil @@ -530,7 +498,7 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Error("an ask not run was asked its hand-act again") } cancelled := "" - w.g.d.Publish = func(module, subject string, body any, id string) error { cancelled = subject + " " + id; return nil } + w.publish = func(module, subject string, body any, id string) error { cancelled = subject + " " + id; return nil } if err := w.g.Cancel(list[0].Ask); err != nil || cancelled != "mesh.seat.operator-channel.accept.cancel.shell cancel."+list[0].Ask { t.Fatalf("cancel %v %q", err, cancelled) } diff --git a/node-tools/internal/approval/ask.go b/node-tools/internal/approval/ask.go deleted file mode 100644 index 60a27da..0000000 --- a/node-tools/internal/approval/ask.go +++ /dev/null @@ -1,241 +0,0 @@ -package approval - -import ( - "crypto/rand" - "encoding/hex" - "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" - stdio "git.novox.be/novox/mesh-sdk/go" -) - -// StateName is the runner's own state its asks and claims are kept in, per machine (§13). -const StateName = "approvals" - -// Store is the runner's state `approvals` (§13): a key's value, a create that refuses a key that exists and an update by -// compare-and-set, each failing with stdio.ErrChanged when another write came first; a put, and its keys. -type Store interface { - Get(key string) (*stdio.StateEntry, error) - Create(key string, value any) (uint64, error) - Update(key string, value any, revision uint64) (uint64, error) - Put(key string, value any) (uint64, error) - Keys() ([]string, error) -} - -// The bounds of an ask (§10, §12, §13, §14, §17). -const ( - AskLasts = 60 * time.Minute - RecordWithin = 5 * time.Second - RunWithin = 30 * time.Minute - DeclinedFor = 10 * time.Minute - MostKeptValue = 256 * 1024 - OnExpiry = "nothing runs; call again to ask anew" -) - -// The options every ask offers (§10), both at level approve, both binding the act's digest. -const ( - OptionApprove = "approve" - OptionDecline = "decline" -) - -// client is operator-channel's client asking under the marked module's name, its asks kept in the runner's state under -// this machine's keys, its publish and the router's record on the runner's own account (§7, §10). -func (g *Gate) client(module string) *operatorchannelclient.Client { - return &operatorchannelclient.Client{Module: module, State: g.d.State, KeyPrefix: g.d.Node + ".", Now: g.d.Now, - Publish: func(subject string, body any, id string) error { return g.d.Publish(module, subject, body, id) }, - Record: func(id string) (*stdio.StateEntry, error) { return g.d.Record(module, id) }} -} - -// actKey is the key naming the newest ask of an act (§13). -func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + digest } - -// askKey is the key the client keeps an ask under (§13). -func (g *Gate) askKey(id string) string { return g.d.Node + ".ask." + id } - -// newID is an ask's id: 128 random bits. -func newID() string { - var b [16]byte - _, _ = rand.Read(b[:]) - return hex.EncodeToString(b[:]) -} - -// askFor is the ask a marked call raises (§10, §11). -func (g *Gate) askFor(c Call, digest, id string) seat.Ask { - args, whole := shown(c.Args) - name := c.Name - if c.Seat != "" { - name = c.Seat + "." + c.Name - } - headline := headlineOf("Run", name, g.d.Node, whole) - first, _, _ := strings.Cut(strings.TrimSpace(c.Description), "\n") - if first == "" { - first = name - } - lines := []string{first, "On " + g.d.Node + ", called through the mesh's tools, which cannot tell who called."} - if args != "" { - lines = append(lines, args) - } - if !whole { - lines = append(lines, "The whole is read with mesh_approvals on "+g.d.Node+", whose fingerprint must be the one on the phone.") - } - lines = append(lines, "Fingerprint "+Fingerprint(digest)) - return seat.Ask{ID: id, About: digest, Who: operatorchannelclient.Operator, Headline: headline, - Explanation: strings.Join(lines, "\n"), Expires: operatorchannelclient.FormatTime(g.d.Now().Add(AskLasts)), OnExpiry: OnExpiry, - Options: []seat.Option{ - {ID: OptionApprove, Label: "Approve", Does: "the same call, made again within 30 minutes, runs once", - Level: operatorchannelclient.Approve, Binds: digest}, - {ID: OptionDecline, Label: "Decline", Does: "nothing runs; the same call is not asked again for 10 minutes", - Level: operatorchannelclient.Approve, Binds: digest}}} -} - -// 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 := verb + " " + name + " on " + node + "?" + suffix; fits(h) { - return h - } - if h := verb + " " + name + "?" + suffix; fits(h) { - return h - } - room := max - len([]rune(verb+" …?"+suffix)) - r := []rune(name) - if room < 1 { - room = 1 - } - if len(r) > room { - r = r[:room] - } - 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 -// asked (§10, §12). -func (g *Gate) firstCall(c Call) (any, error) { - 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()) - } - return nil, g.raise(c, act, digest, "") -} - -// raise raises the ask of a marked call whose act has none open (§10): kept in the runner's state with the act and the whole -// arguments, published under the module's name, its act's key naming it; then the 5 seconds for the router's record and -// 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() - 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)) - } - if err := cl.Ask(a); err != nil { - return g.refused(c, digest, "the ask could not be raised: "+err.Error()) - } - 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), 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 nil -} - -// keptSize is the bytes the kept ask would take. -func keptSize(a seat.Ask, act operatorchannelclient.Act, args json.RawMessage) int { - raw, _ := json.Marshal(operatorchannelclient.Kept{Ask: a, Act: act, Arguments: args}) - return len(raw) -} - -// keepCall writes the act and the whole arguments beside the ask the client kept (§13). -func (g *Gate) keepCall(id string, act operatorchannelclient.Act, args json.RawMessage) error { - e, err := g.d.State.Get(g.askKey(id)) - if err != nil || e == nil { - return fmt.Errorf("the kept ask %s cannot be read: %v", id, err) - } - var k operatorchannelclient.Kept - if err := json.Unmarshal(e.Value, &k); err != nil { - return err - } - k.Act, k.Arguments = act, args - _, err = g.d.State.Update(g.askKey(id), k, e.Revision) - return err -} - -// firstAnswer waits up to RecordWithin for the router's record of the ask and says the call was not run (§12). -func (g *Gate) firstAnswer(c Call, id, digest, markedSince string) error { - rec, _ := g.awaitRecord(c.Module, id) - since := "" - if markedSince != "" { - since = "; marked since revision " + markedSince - } - if rec == nil { - return fmt.Errorf("Not run, asked; not yet taken by the router (ask %s, fingerprint %s); make the same call again later%s", - id, Fingerprint(digest), since) - } - return fmt.Errorf("Not run: this needs the operator's approval. Asked on %s at %s (ask %s, fingerprint %s); make the same call "+ - "again within 30 minutes of the approval; the ask expires at %s%s", orNone(strings.Join(rec.OfferedOn, ", ")), rec.Opened, - id, Fingerprint(digest), rec.Expires, since) -} - -// record is the router's record of an ask under the module's name, nil when it holds none. -func (g *Gate) record(module, id string) (*seat.AsksByAskerIDValue, error) { - e, err := g.d.Record(module, id) - if err != nil || e == nil { - return nil, err - } - var rec seat.AsksByAskerIDValue - if err := json.Unmarshal(e.Value, &rec); err != nil { - return nil, fmt.Errorf("the router's record of %s cannot be read: %w", id, err) - } - return &rec, nil -} - -// awaitRecord reads the router's record of a new ask until it is there, within RecordWithin: the seat answers a key's -// value and says nothing when the router takes an ask, so it is read again after a growing pause (§12). -func (g *Gate) awaitRecord(module, id string) (*seat.AsksByAskerIDValue, error) { - until := g.d.Now().Add(RecordWithin) - pause := 50 * time.Millisecond - for { - rec, err := g.record(module, id) - if rec != nil || g.d.Now().Add(pause).After(until) { - return rec, err - } - <-g.d.After(pause) - pause *= 2 - } -} - -// refused is a marked call refused, run nothing, written in the runner's log with the ask's fingerprint and never its -// arguments (§19). -func (g *Gate) refused(c Call, digest, why string) error { - g.d.Logf("[mesh-tools] %s on %s is refused (fingerprint %s): %s", c.Address, g.d.Node, orNone(Fingerprint(digest)), why) - return fmt.Errorf("Not run: %s needs the operator's approval and %s", c.Address, why) -} diff --git a/node-tools/internal/approval/closed.go b/node-tools/internal/approval/closed.go deleted file mode 100644 index 96219a4..0000000 --- a/node-tools/internal/approval/closed.go +++ /dev/null @@ -1,119 +0,0 @@ -package approval - -import ( - "encoding/json" - "fmt" - "time" - - assignerseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" -) - -// AskedOn is the seat an ask is raised on, whose holder must be there for a marked call to be asked (§19). -const AskedOn = "operator-channel" - -// UnreadableAfter is how long the state approvals fails before the runner's finding says so (§19). -const UnreadableAfter = 5 * time.Minute - -// unreadableKey is the bound key whose expiry raises `approvals-unreadable` (lead answer A-157). -func (g *Gate) unreadableKey() string { return g.d.Node + ".unreadable." + StateName } - -// held says why no marked call can be asked now, or "" (§19): no module holds operator-channel, as the assigner's view -// of the seat's holders says, or that view is not read. -func (g *Gate) held() string { - if g.d.Holder == nil { - return "the tool runner reads no view of the seats' holders, so it cannot tell that a module holds " + AskedOn - } - ok, err := g.d.Holder(AskedOn) - switch { - case err != nil: - return fmt.Sprintf("whether a module holds %s cannot be read: %v", AskedOn, err) - case !ok: - return "no module holds " + AskedOn + ", so nobody can be asked" - } - return "" -} - -// stateFailed notes that the state approvals could not be read or written: no marked call runs on this machine until it -// can, the failed read or write is made again at the next marked call and at the next membership change (A-157), and -// five minutes on the runner's finding says it (§19). -func (g *Gate) stateFailed(err error) { - g.mu.Lock() - first := g.failingSince.IsZero() - if first { - g.failingSince = g.d.Now() - } - g.mu.Unlock() - if first { - g.d.Logf("[mesh-tools] the state %s on %s cannot be read or written (%v): no marked call runs here until it can", StateName, g.d.Node, err) - if g.d.Bounds != nil { - if e := g.d.Bounds.CreateTTL(g.unreadableKey(), map[string]string{"since": g.d.Now().UTC().Format(time.RFC3339)}, UnreadableAfter); e != nil { - g.d.Logf("[mesh-tools] the bound of the state %s's failure is not kept: %v", StateName, e) - } - } - } -} - -// stateRead notes that the state approvals was read or written: a failure noted before ends, its finding cleared. -func (g *Gate) stateRead() { - g.mu.Lock() - was := !g.failingSince.IsZero() - g.failingSince = time.Time{} - g.mu.Unlock() - if !was { - return - } - if g.d.Bounds != nil { - _ = g.d.Bounds.Delete(g.unreadableKey()) - } - g.finding(StateName, KindApprovalsUnreadable, false, "the state "+StateName+" is read again", "") -} - -// failing says the state approvals is failing now. -func (g *Gate) failing() bool { - g.mu.Lock() - defer g.mu.Unlock() - return !g.failingSince.IsZero() -} - -// StateAgain makes the failed read of the state approvals again: at the next membership change the runner hears (A-157). -func (g *Gate) StateAgain() { - if g.d.State == nil || !g.failing() { - return - } - if _, err := g.d.State.Keys(); err != nil { - return - } - g.stateRead() -} - -// unreadableExpired is the bound key of a failing state expiring: the finding raised while the state still fails. -func (g *Gate) unreadableExpired() { - g.mu.Lock() - since := g.failingSince - g.mu.Unlock() - if since.IsZero() { - return - } - g.finding(StateName, KindApprovalsUnreadable, true, fmt.Sprintf("the state %s on %s cannot be read or written since %s: no marked call "+ - "runs here until it can", StateName, g.d.Node, since.UTC().Format(time.RFC3339)), "approvals unreadable on "+g.d.Node) -} - -// HoldersIn is Holder over the assigner's view of the seats' holders as the runner holds it (§19): view answers the view's -// values by key, an error while its first view is not answered; a seat is held when its key names a holder. -func HoldersIn(view func() (map[string][]byte, error)) func(seat string) (bool, error) { - return func(seat string) (bool, error) { - values, err := view() - if err != nil { - return false, err - } - raw, ok := values[assignerseat.SeatHoldersBySeatKey(seat)] - if !ok { - return false, nil - } - var h assignerseat.SeatHolders - if err := json.Unmarshal(raw, &h); err != nil { - return false, fmt.Errorf("the holders of %s cannot be read: %w", seat, err) - } - return len(h.Holders) > 0, nil - } -} diff --git a/node-tools/internal/approval/contracts.go b/node-tools/internal/approval/contracts.go new file mode 100644 index 0000000..928821f --- /dev/null +++ b/node-tools/internal/approval/contracts.go @@ -0,0 +1,100 @@ +// Package approval is the tool runner's side of mesh-sdk's approval gate (go/approval; design 47 G lead answers A-224, +// A-228): it translates what the gate reads beyond itself from the generated contracts, the gate's findings into the +// contract's findings about the runner on its machine, the catalog's entries into the gate's, and the assigner's view of +// the seats' holders into its holder function. The gate itself is mesh-sdk's; none is written here. +package approval + +import ( + "context" + "encoding/json" + "fmt" + "time" + + assignerseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + gate "git.novox.be/novox/mesh-sdk/go/approval" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// HeldWriter is the runner's held finding writer (views.HeldWriter over the contract's Finding). +type HeldWriter interface { + Write(ctx context.Context, finding condseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + +// scopeMachine is the scope of the gate's findings on the tool runner: its machine (lead answer A-154). +const scopeMachine = "machine" + +// Findings is the gate's finding writer over the runner's held writer: each finding about the runner's module on node, at +// scope machine, and its pending named in the writer's `findings-stated` where the writer keeps one (design 47 G §384). +func Findings(w HeldWriter, node, runner string, now func() time.Time) gate.FindingWriter { + if w == nil { + return nil + } + if now == nil { + now = time.Now + } + return findingWriter{w: w, node: node, runner: runner, now: now} +} + +type findingWriter struct { + w HeldWriter + node, runner string + now func() time.Time +} + +func (f findingWriter) Raise(ctx context.Context, kind, id, summary, headline string) error { + return f.w.Write(ctx, condseat.Finding{Kind: kind, Scope: scopeMachine, ID: id, Severity: condseat.FindingSeverityWarning, + Source: "approval", Observer: f.runner, Subject: condseat.Subject{Scope: "module", ID: f.runner, Machine: f.node}, + Summary: summary, Headline: headline, At: f.now().UTC()}) +} + +func (f findingWriter) Clear(ctx context.Context, kind, id, why string) error { + return f.w.Clear(ctx, scopeMachine, id, kind, why) +} + +// pender is the held writer's pending (views.HeldWriter's Pend and Judged). +type pender interface { + Pend(views.PendingKey) + Judged(key string) +} + +func (f findingWriter) Pend(kind, id string, since time.Time, bound time.Duration) { + if p, ok := f.w.(pender); ok { + p.Pend(views.PendingKey{Key: views.FindingKey(scopeMachine, id, kind), Since: since, Bound: bound.String()}) + } +} + +func (f findingWriter) Judged(kind, id string) { + if p, ok := f.w.(pender); ok { + p.Judged(views.FindingKey(scopeMachine, id, kind)) + } +} + +// Entry is a catalog entry as the gate reads its marks. +func Entry(e catalogseat.CatalogEntry) gate.Entry { + manifest, _ := json.Marshal(e.Manifest) + return gate.Entry{Module: e.Module, Fingerprint: e.Fp, Serves: e.Serves, Manifest: manifest} +} + +// HoldersIn is the gate's holder function over the assigner's view of the seats' holders as the runner holds it (§19): +// view answers the view's values by key, an error while its first view is not answered; a seat is held when its key names +// a holder. +func HoldersIn(view func() (map[string][]byte, error)) func(seat string) (bool, error) { + return func(seat string) (bool, error) { + values, err := view() + if err != nil { + return false, err + } + raw, ok := values[assignerseat.SeatHoldersBySeatKey(seat)] + if !ok { + return false, nil + } + var h assignerseat.SeatHolders + if err := json.Unmarshal(raw, &h); err != nil { + return false, fmt.Errorf("the holders of %s cannot be read: %w", seat, err) + } + return len(h.Holders) > 0, nil + } +} diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go deleted file mode 100644 index 315db22..0000000 --- a/node-tools/internal/approval/gate.go +++ /dev/null @@ -1,179 +0,0 @@ -package approval - -import ( - "context" - "encoding/json" - "fmt" - "sort" - "strings" - "sync" - "time" - - condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" - stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/keys" -) - -// The runner's findings of the gate (design 46 7-approval-mark §4, §16, §19), written through its held writer. -const ( - KindMarkUnserved = "mark-unserved" - KindCallUnanswered = "call-unanswered" - KindApprovalsUnreadable = "approvals-unreadable" -) - -// Kinds are the finding kinds the gate writes through the runner's writer. -var Kinds = []string{KindMarkUnserved, KindCallUnanswered, KindApprovalsUnreadable} - -// FindingWriter is the runner's held finding writer (views.HeldWriter over the contract's Finding). -type FindingWriter interface { - Write(ctx context.Context, finding condseat.Finding) error - Clear(ctx context.Context, scope, id, kind, why string) error -} - -// Deps is what the gate reaches the mesh through: the runner's connection in its one place, fakes in a test. -type Deps struct { - // Node is the machine the runner serves; Runner its own module's name, the subject of its findings. - 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) - Now func() time.Time - // After is the clock's pause; nil is time.After. - After func(time.Duration) <-chan time.Time - // State is the runner's state approvals; nil when its membership binds none, so every marked call is refused (§13). - State Store - // Publish says an accept on a seat under the module's name, on the runner's own account (§7). - Publish func(module, subject string, body any, id string) error - // Record reads the router's record of an ask, `.` in the seat's state asks, on the runner's own account (§14). - Record func(module, id string) (*stdio.StateEntry, error) - // Warranted asks `mesh-hand-acts.warranted` {asker, ask}, each try a kept call open while open says so (§16, A-157). - Warranted func(asker, ask string, open func() (bool, error)) error - // Bounds is the runner's key-ttl state bound (A-157, A-158); nil keeps no bound. - 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()) -} - -// Gate is the runner's approval gate: every call of a module it serves goes through it before the bundle (§20). -type Gate struct { - d Deps - mu sync.Mutex - // registered is each module's tools as its bundle registered them at its last load. - 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 - // inFlight is each ask whose recording runs now. - inFlight map[string]bool - // failingSince is when the state approvals began to fail; zero while it is read. - failingSince time.Time - // served is each module this runner serves, with the build its bundle serves, as its last load said (§18). - served map[string]string -} - -// New is a gate over d. -func New(d Deps) *Gate { - if d.Logf == nil { - d.Logf = func(string, ...any) {} - } - if d.Now == nil { - d.Now = time.Now - } - if d.After == nil { - d.After = time.After - } - if d.Go == nil { - d.Go = func(f func()) { go f() } - } - return &Gate{d: d, registered: map[string]map[string]bool{}, unserved: map[string][]string{}} -} - -// Unserved is, of the module's marked tools, those its bundle does not register (§4): such a bundle is not loaded, its -// finding `mark-unserved` raised; a bundle that serves them, or marks that no longer name them, clear it. Before the marks -// are read nothing is known, and every marked call is refused until they are (§19). -func (g *Gate) Unserved(module, fingerprint string, registered []string) []string { - set := map[string]bool{} - for _, t := range registered { - set[t] = true - } - g.mu.Lock() - g.registered[module] = set - if g.served == nil { - g.served = map[string]string{} - } - g.served[module] = fingerprint - g.mu.Unlock() - if g.d.Catalog == nil || g.d.Catalog.Ready() != nil { - return nil - } - return g.checkServed(Derive(g.d.Catalog, module, fingerprint)) -} - -// checkServed raises or clears the module's `mark-unserved` as its marks and its bundle's registrations stand now, and is -// the tools it marks and does not register. -func (g *Gate) checkServed(m Marks) []string { - g.mu.Lock() - set := g.registered[m.Module] - var missing []string - for t := range m.Tools { - if !set[t] { - missing = append(missing, t) - } - } - sort.Strings(missing) - was := g.unserved[m.Module] - if len(missing) > 0 { - g.unserved[m.Module] = missing - } else { - delete(g.unserved, m.Module) - } - g.mu.Unlock() - switch { - case len(missing) > 0 && strings.Join(was, ",") != strings.Join(missing, ","): - g.finding(m.Module, KindMarkUnserved, true, fmt.Sprintf("%s's marks (%s, catalog revision %s) name %s, which its bundle on %s does not "+ - "register: no tool of %s is served here", m.Module, sourceOf(m, missing[0]), m.Revision, strings.Join(missing, ", "), g.d.Node, m.Module), - fmt.Sprintf("%s is not served on %s", m.Module, g.d.Node)) - case len(missing) == 0 && len(was) > 0: - g.finding(m.Module, KindMarkUnserved, false, "a bundle that serves every marked tool, or marks without the tool, is loaded", "") - } - return missing -} - -func sourceOf(m Marks, tool string) string { return m.Tools[tool] } - -// finding writes or clears one of the gate's findings: scope machine, its id `.` escaped as a key's name is, -// about the runner's module on this machine (lead answer A-154). -func (g *Gate) finding(what, kind string, raised bool, summary, headline string) { - id := keys.EscapeName(g.d.Node + "." + what) - if g.d.Findings == nil { - g.d.Logf("[mesh-tools] %s %s: %s", kind, id, summary) - return - } - var err error - if raised { - err = g.d.Findings.Write(context.Background(), condseat.Finding{Kind: kind, Scope: scopeMachine, ID: id, - Severity: condseat.FindingSeverityWarning, Source: "approval", Observer: g.d.Runner, - Subject: condseat.Subject{Scope: "module", ID: g.d.Runner, Machine: g.d.Node}, Summary: summary, Headline: headline, - At: g.d.Now().UTC()}) - } else { - err = g.d.Findings.Clear(context.Background(), scopeMachine, id, kind, summary) - } - if err != nil { - g.d.Logf("[mesh-tools] the finding %s of %s is not kept: %v", kind, id, err) - } -} - -const scopeMachine = "machine" diff --git a/node-tools/internal/approval/marks.go b/node-tools/internal/approval/marks.go deleted file mode 100644 index 824ef6e..0000000 --- a/node-tools/internal/approval/marks.go +++ /dev/null @@ -1,185 +0,0 @@ -// Package approval is the tool runner's gate on a call that needs the operator's approval (design 46 7-approval-mark): the -// marks it reads from the catalog half, the act it digests, the ask it raises on the seat operator-channel under the -// marked module's name, the judging of the same call again on the router's record, the claim, the recording in the -// hand-act log, and the refusals when a part of the check is missing (§5, §8 to §20). No module's name is written here. -package approval - -import ( - "encoding/json" - "fmt" - "sort" - "strings" - - catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-sdk/go/buscompose" - "git.novox.be/novox/mesh-sdk/go/seatdef" -) - -// Catalog is what the marks are read from: the catalog half the runner follows (design 46 7-approval-mark §5, design 47 G -// §196, §443) and the seat definitions it carries (§2). -type Catalog interface { - // Ready is nil once the catalog half's first view is read, else says what it waits for. - Ready() error - // Revision is the catalog revision the marks are read at, as the runner names it. - Revision() string - // Served is the entry of the build a bundle serves, by the fingerprint its composed environment carries; false when - // none is read (not yet, or not found), with why. - Served(fingerprint string) (catalogseat.CatalogEntry, bool, string) - // Newest is the module's entry at the newest revision; false when the catalog half lists none. - 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 -// restricted streams, each with its source (§18: its module's entry in the catalog half, or its seat's definition). -type Marks struct { - Module string - // Tools are its marked tools by name, Verbs its marked seat verbs as `.`, Subjects the bus subjects of its - // marked events and restricted streams (design 34 every-event §33); each to its source. - Tools, Verbs, Subjects map[string]string - // Revision is the catalog revision read; Served the served entry's fingerprint, empty when it could not be read. - Revision, Served string - // Unread says why the served entry could not be read: every marked call is then refused (§5, §19). - Unread string -} - -// Marked says whether a tool or a seat verb of the module needs the operator's approval, and its source. -func (m Marks) Marked(seat, name string) (bool, string) { - if seat == "" { - src, ok := m.Tools[name] - return ok, src - } - src, ok := m.Verbs[seat+"."+name] - return ok, src -} - -// Any says the module marks anything at all. -func (m Marks) Any() bool { return len(m.Tools)+len(m.Verbs)+len(m.Subjects) > 0 } - -// marked is what an entry's manifest says of its marks: `needs-approval`, its events and its streams. -type marked struct { - NeedsApproval []string `json:"needs-approval"` - Emits []string `json:"emits"` - // Streams are read for their events and whether each carries a `grant`, and no other word, so a word added to a - // stream's declaration later never hides its marks. - Streams []struct { - Events []string `json:"events"` - Grant json.RawMessage `json:"grant"` - } `json:"streams"` -} - -// Derive is a module's marks from its served entry and its newest one, unioned while they differ so a mark is never lost -// before or after the module moves (§5, ADR 0286 §3), and the seat definitions' verb marks for every verb an entry serves, -// which no holder unmarks (§2, §3). fingerprint is the build its bundle serves; empty, or an entry not read, leaves the -// marks of the newest entry and the seats, with Unread saying why. -func Derive(c Catalog, module, fingerprint string) Marks { - m := Marks{Module: module, Tools: map[string]string{}, Verbs: map[string]string{}, Subjects: map[string]string{}, - Revision: c.Revision()} - defs := c.Definitions() - var entries []catalogseat.CatalogEntry - if fingerprint == "" { - m.Unread = "its bundle's environment carries no fingerprint, so the build it serves cannot be read" - } else if e, ok, why := c.Served(fingerprint); ok { - entries, m.Served = append(entries, e), fingerprint - } else { - m.Unread = fmt.Sprintf("the catalog half's entry of %s cannot be read: %s", fingerprint, why) - } - if e, ok := c.Newest(module); ok && (len(entries) == 0 || e.Fp != entries[0].Fp) { - entries = append(entries, e) - } - for _, e := range entries { - 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[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 - } - 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 - } - } - } -} - -// eventSubjects are a marked event's subjects on the bus (design 34 every-event §33): said on the module's events and on its -// own stream, each also with every subject below it. -func eventSubjects(module, event string) []string { - ev := buscompose.WirePattern(event) - out := []string{"mesh.mod." + module + ".event." + ev, buscompose.ModuleStreamSubject(module, ev)} - if !strings.HasSuffix(ev, ">") { - out = append(out, out[0]+".>", out[1]+".>") - } - sort.Strings(out) - return out -} - -func orNone(s string) string { - if s == "" { - return "none" - } - return s -} - -// keysOf is a set's names, sorted; nil when empty. -func keysOf(set map[string]string) []string { - var out []string - for k := range set { - out = append(out, k) - } - sort.Strings(out) - return out -} diff --git a/node-tools/internal/approval/marks_test.go b/node-tools/internal/approval/marks_test.go index 000b379..a18bfd7 100644 --- a/node-tools/internal/approval/marks_test.go +++ b/node-tools/internal/approval/marks_test.go @@ -3,6 +3,7 @@ package approval import ( "context" "reflect" + "sort" "testing" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" @@ -10,6 +11,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/views" catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + gate "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" ) @@ -24,19 +26,19 @@ type fakeCatalog struct { func (c *fakeCatalog) Ready() error { return c.ready } func (c *fakeCatalog) Revision() string { return c.revision } -func (c *fakeCatalog) Served(fp string) (catalogseat.CatalogEntry, bool, string) { +func (c *fakeCatalog) Served(fp string) (gate.Entry, bool, string) { e, ok := c.byFp[fp] - return e, ok, "not-found" + return Entry(e), ok, "not-found" } -func (c *fakeCatalog) Newest(m string) (catalogseat.CatalogEntry, bool) { +func (c *fakeCatalog) Newest(m string) (gate.Entry, bool) { e, ok := c.newest[m] - return e, ok + return Entry(e), ok } func (c *fakeCatalog) Definitions() map[string]seatdef.Definition { return c.defs } -func (c *fakeCatalog) Entries() []catalogseat.CatalogEntry { - var out []catalogseat.CatalogEntry +func (c *fakeCatalog) Entries() []gate.Entry { + var out []gate.Entry for _, e := range c.newest { - out = append(out, e) + out = append(out, Entry(e)) } return out } @@ -90,7 +92,7 @@ func TestTheMarks(t *testing.T) { "mesh.stream.mail.event.mail.>", "mesh.stream.mail.event.mail.received", "mesh.stream.mail.event.mail.received.>"}}, } { t.Run(tc.name, func(t *testing.T) { - m := Derive(tc.cat, tc.module, tc.fp) + m := gate.Derive(tc.cat, tc.module, tc.fp) if got := keysOf(m.Tools); !reflect.DeepEqual(got, tc.tools) { t.Errorf("tools %v, want %v", got, tc.tools) } @@ -113,8 +115,8 @@ func TestTheMarks(t *testing.T) { t.Run("mark-unserved raised and cleared", func(t *testing.T) { cat := &fakeCatalog{revision: "7", byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1}, newest: map[string]catalogseat.CatalogEntry{"shell": v1}, defs: login} w := &findings{} - g := New(Deps{Node: "n1", Runner: "node-tools", Catalog: cat, Findings: w}) - if got := g.Unserved("shell", "src2:1", []string{"other"}); !reflect.DeepEqual(got, []string{"wipe"}) || w.raised[keys.EscapeName("n1.shell")] != KindMarkUnserved { + g := gate.New(gate.Deps{Node: "n1", Runner: "node-tools", Catalog: cat, Findings: Findings(w, "n1", "node-tools", nil)}) + if got := g.Unserved("shell", "src2:1", []string{"other"}); !reflect.DeepEqual(got, []string{"wipe"}) || w.raised[keys.EscapeName("n1.shell")] != gate.KindMarkUnserved { t.Fatalf("unserved %v, findings %v", got, w.raised) } if got := g.Unserved("shell", "src2:1", []string{"wipe"}); got != nil || w.raised[keys.EscapeName("n1.shell")] != "" { @@ -156,3 +158,12 @@ func (f *findings) Clear(_ context.Context, _, id, _, _ string) error { delete(f.raised, id) return nil } + +func keysOf(set map[string]string) []string { + var out []string + for k := range set { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/node-tools/internal/approval/record.go b/node-tools/internal/approval/record.go deleted file mode 100644 index 2826bed..0000000 --- a/node-tools/internal/approval/record.go +++ /dev/null @@ -1,256 +0,0 @@ -package approval - -import ( - "encoding/json" - "errors" - "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" - stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/keys" - "git.novox.be/novox/mesh-sdk/go/views" -) - -// The bounds of a run's recording (§13, §16; lead answers A-157, A-158). -const ( - UnansweredAfter = 5 * time.Minute - KeptFor = 48 * time.Hour - SaidBefore = time.Hour -) - -// Bounds is the runner's key-ttl state `bound` (lead answers A-157, A-158): a key created with its own time to live, which -// the bus removes when it passes, every watch hearing the removal as an expiry, so the gate's waits need no timer. -type Bounds interface { - CreateTTL(key string, value any, ttl time.Duration) error - Delete(key string) error - // Watch hands expired every key under the pattern whose time to live passed, until the stop is called. - Watch(pattern string, expired func(key string)) (stop func(), err error) -} - -// Answered is a call run on the operator's approval: the tool's own answer and the line naming the approval (§16), in the -// form a process holder's `gated-ended` says it (design 47 G §677; lead answer A-158). -type Answered struct { - Answer any `json:"answer"` - Approval string `json:"approval"` -} - -// The bound keys of a run (lead answer A-158; §13). -func (g *Gate) unansweredKey(id string) string { return g.d.Node + ".unanswered." + id } -func (g *Gate) unrecordedKey(id string) string { return g.d.Node + ".unrecorded." + id } - -// ran is a call run on the operator's approval: answered with the tool's answer and the approval's line, and its warrant -// recorded in the hand-act log (§16). -func (g *Gate) ran(c Call, id string, rec *seat.AsksByAskerIDValue, answer any, err error) (any, error) { - channel := "" - at := "" - if w := rec.Warrant; w != nil { - channel, at = w.Channel, w.At - } - if channel == "" { - channel = strings.Join(rec.OfferedOn, ", ") - } - line := fmt.Sprintf("run on the operator's approval via %s at %s, ask %s", orNone(channel), at, id) - g.d.Logf("[mesh-tools] %s on %s %s", c.Address, g.d.Node, line) - g.bindRecording(c.Module, id) - g.d.Go(func() { g.recordRun(c.Module, id) }) - if err != nil { - return nil, fmt.Errorf("%w (%s)", err, line) - } - return Answered{Answer: answer, Approval: line}, nil -} - -// bindRecording keeps the run's two bound keys: one that raises `call-unanswered` when `warranted` has not answered within -// five minutes, one that says an unrecorded run an hour before its key's 48 hours end (§13, §16). -func (g *Gate) bindRecording(module, id string) { - if g.d.Bounds == nil { - return - } - for key, ttl := range map[string]time.Duration{g.unansweredKey(id): UnansweredAfter, g.unrecordedKey(id): KeptFor - SaidBefore} { - if err := g.d.Bounds.CreateTTL(key, map[string]string{"asker": module, "ask": id}, ttl); err != nil { - g.d.Logf("[mesh-tools] the bound %s of the run of ask %s is not kept: %v", key, id, err) - } - } -} - -// recordRun calls `mesh-hand-acts.warranted` for one acted ask, each try a kept call open while the ask lacks `recorded` -// (§16, A-157), and on its answer writes `recorded` and clears what waited on it. One recording per ask at a time. -func (g *Gate) recordRun(module, id string) { - if !g.recording(id, true) { - return - } - defer g.recording(id, false) - if g.d.Warranted == nil { - g.d.Logf("[mesh-tools] the run of ask %s is not recorded: the tool runner reaches no hand-act log", id) - return - } - err := g.d.Warranted(module, id, func() (bool, error) { return g.unrecorded(id) }) - g.judged(id) // made again: its answer, or the bound of its key, judges it from now - if err != nil { - g.d.Logf("[mesh-tools] mesh-hand-acts.warranted has not answered for ask %s (%v): asked again when the hand-act seat's "+ - "holder is heard again and at the next start", id, err) - return - } - if err := g.markRecorded(id); err != nil { - g.d.Logf("[mesh-tools] ask %s is recorded in the hand-act log and its mark recorded is not kept: %v", id, err) - return - } - if g.d.Bounds != nil { - _ = g.d.Bounds.Delete(g.unansweredKey(id)) - _ = g.d.Bounds.Delete(g.unrecordedKey(id)) - } - g.finding(id, KindCallUnanswered, false, "mesh-hand-acts.warranted answered for ask "+id, "") -} - -// recording claims or releases an ask's recording: false when one runs already. -func (g *Gate) recording(id string, start bool) bool { - g.mu.Lock() - defer g.mu.Unlock() - if g.inFlight == nil { - g.inFlight = map[string]bool{} - } - if start { - if g.inFlight[id] { - return false - } - g.inFlight[id] = true - return true - } - delete(g.inFlight, id) - return true -} - -// kept is the ask the client keeps under its id, with its revision; nil when there is none. -func (g *Gate) kept(id string) (*operatorchannelclient.Kept, uint64, error) { - e, err := g.d.State.Get(g.askKey(id)) - if err != nil || e == nil { - return nil, 0, err - } - var k operatorchannelclient.Kept - if err := json.Unmarshal(e.Value, &k); err != nil { - return nil, 0, err - } - return &k, e.Revision, nil -} - -// unrecorded says the ask is acted on and its run not yet recorded: the key that keeps `warranted` open (§16). -func (g *Gate) unrecorded(id string) (bool, error) { - k, _, err := g.kept(id) - if err != nil || k == nil { - return false, err - } - return k.Acted && !k.Recorded, nil -} - -// markRecorded writes `recorded` on the kept ask by compare-and-set, again when another write came first. -func (g *Gate) markRecorded(id string) error { - for { - k, rev, err := g.kept(id) - if err != nil || k == nil { - return fmt.Errorf("the kept ask cannot be read: %v", err) - } - if k.Recorded { - return nil - } - k.Recorded = true - if _, err := g.d.State.Update(g.askKey(id), k, rev); err == nil { - return nil - } else if !isChanged(err) { - return err - } - } -} - -// Recover calls `warranted` again for every acted, unrecorded ask this machine's keys hold within their 48 hours (§16): at -// each start, and when the hand-act seat's holder is heard again (A-157). -func (g *Gate) Recover() { - if g.d.State == nil { - return - } - keys, err := g.d.State.Keys() - if err != nil { - g.d.Logf("[mesh-tools] the acted, unrecorded asks cannot be read (%v): read again at the next start or holder heard", err) - return - } - prefix := g.d.Node + ".ask." - for _, key := range keys { - if !strings.HasPrefix(key, prefix) { - continue - } - id := strings.TrimPrefix(key, prefix) - k, _, err := g.kept(id) - if err != nil || k == nil || !k.Acted || k.Recorded || g.d.Now().Sub(k.Asked) > KeptFor { - continue - } - module := moduleOf(k) - g.pend(id) - g.d.Go(func() { g.recordRun(module, id) }) - } -} - -// moduleOf is the module an ask was asked under: the warrant's asker. -func moduleOf(k *operatorchannelclient.Kept) string { - if k.Answer != nil { - return k.Answer.Asker - } - return "" -} - -// WatchBounds hears the run's bound keys expire (A-157): `call-unanswered` raised when `warranted` has not answered within -// five minutes; an unrecorded run said in the log an hour before its key's 48 hours end (§13). -func (g *Gate) WatchBounds() (func(), error) { - if g.d.Bounds == nil { - return func() {}, nil - } - return g.d.Bounds.Watch(g.d.Node+".**", func(key string) { - if key == g.unreadableKey() { - g.unreadableExpired() - return - } - rest := strings.TrimPrefix(key, g.d.Node+".") - kind, id, _ := strings.Cut(rest, ".") - if open, err := g.unrecorded(id); err != nil || !open { - return - } - switch kind { - case "unanswered": - g.finding(id, KindCallUnanswered, true, fmt.Sprintf("mesh-hand-acts.warranted has not answered for ask %s, run on the "+ - "operator's approval on %s, within %s: the run is not yet in the hand-act log", id, g.d.Node, UnansweredAfter), - "a run's hand-act is unrecorded on "+g.d.Node) - case "unrecorded": - g.d.Logf("[mesh-tools] ERROR the run of ask %s on %s is still unrecorded an hour before its key's 48 hours end", id, g.d.Node) - } - }) -} - -// Pender is the runner's held writer's pending (views.HeldWriter's Pend and Judged): the keys it cannot judge yet since its -// start, named in its `findings-stated` (design 47 G §384). -type Pender interface { - Pend(views.PendingKey) - Judged(key string) -} - -// pend names an acted, unrecorded ask's `call-unanswered` in the runner's `findings-stated` until its `warranted` is made -// again (§16; design 47 G §384), its bound the five minutes the finding waits. -func (g *Gate) pend(id string) { - if p, ok := g.d.Findings.(Pender); ok { - p.Pend(views.PendingKey{Key: g.unansweredFinding(id), Since: g.d.Now().UTC(), Bound: UnansweredAfter.String()}) - } -} - -// judged ends that pending. -func (g *Gate) judged(id string) { - if p, ok := g.d.Findings.(Pender); ok { - p.Judged(g.unansweredFinding(id)) - } -} - -// unansweredFinding is the key of an ask's `call-unanswered` in the runner's state findings (lead answer A-154). -func (g *Gate) unansweredFinding(id string) string { - return views.FindingKey(scopeMachine, keys.EscapeName(g.d.Node+"."+id), KindCallUnanswered) -} - -// isChanged says a compare-and-set lost to another write. -func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) } diff --git a/node-tools/internal/approval/run.go b/node-tools/internal/approval/run.go deleted file mode 100644 index b0f525c..0000000 --- a/node-tools/internal/approval/run.go +++ /dev/null @@ -1,184 +0,0 @@ -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) - } - if why := g.held(); 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 { - g.stateFailed(err) - return nil, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()) - } - g.stateRead() - 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: - g.stateFailed(err) - 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) -} - -func errText(err error) string { - if err == nil { - return "no reason given" - } - return err.Error() -} diff --git a/node-tools/internal/approval/tools.go b/node-tools/internal/approval/tools.go deleted file mode 100644 index 13b4cbf..0000000 --- a/node-tools/internal/approval/tools.go +++ /dev/null @@ -1,168 +0,0 @@ -package approval - -import ( - "encoding/json" - "fmt" - "sort" - "strings" - "time" -) - -// The mesh MCP server's mesh_approvals reads and steers what the gate asked on this machine (§18). - -// Listed is one ask of the last day as mesh_approvals lists it: its state and its act's fingerprint. -type Listed struct { - Ask string `json:"ask"` - Address string `json:"address"` - Module string `json:"module"` - Asked time.Time `json:"asked"` - State string `json:"state"` - Fingerprint string `json:"fingerprint"` -} - -// Whole is one ask read whole: its arguments and the act's digest. -type Whole struct { - Listed - Digest string `json:"digest"` - Act map[string]string `json:"act"` - Arguments json.RawMessage `json:"arguments"` - Recorded bool `json:"recorded"` -} - -// stateOf is a kept ask's state as the runner knows it. -func stateOf(acted, recorded bool, ended string) string { - switch { - case acted && recorded: - return "run, recorded" - case acted: - return "run, not yet recorded" - case ended != "": - return ended - } - return "asked" -} - -// List is the calls asked on this machine in the last day, newest first (§18). -func (g *Gate) List() ([]Listed, error) { - if g.d.State == nil { - return nil, fmt.Errorf("the tool runner's membership binds no state %s", StateName) - } - keys, err := g.d.State.Keys() - if err != nil { - return nil, err - } - prefix := g.d.Node + ".ask." - out := []Listed{} - for _, key := range keys { - if !strings.HasPrefix(key, prefix) { - continue - } - w, err := g.Read(strings.TrimPrefix(key, prefix)) - if err != nil || g.d.Now().Sub(w.Asked) > 24*time.Hour { - continue - } - out = append(out, w.Listed) - } - sort.Slice(out, func(i, j int) bool { return out[i].Asked.After(out[j].Asked) }) - return out, nil -} - -// Read is one ask whole: its arguments and the act's digest (§18). -func (g *Gate) Read(id string) (Whole, error) { - if g.d.State == nil { - return Whole{}, fmt.Errorf("the tool runner's membership binds no state %s", StateName) - } - k, _, err := g.kept(id) - if err != nil { - return Whole{}, err - } - if k == nil { - return Whole{}, fmt.Errorf("no ask %s is kept on %s", id, g.d.Node) - } - module := k.Act["module"] - if module == "" { - module = moduleOf(k) - } - return Whole{Listed: Listed{Ask: id, Address: k.Act["address"], Module: module, Asked: k.Asked, - State: stateOf(k.Acted, k.Recorded, k.Ended), Fingerprint: Fingerprint(k.Ask.About)}, - Digest: k.Ask.About, Act: k.Act, Arguments: k.Arguments, Recorded: k.Recorded}, nil -} - -// Applied are the marks the gate applies on this machine, each module's marked tools and verbs with their source and the -// catalog revision read (§18). -func (g *Gate) Applied() []Marks { - g.mu.Lock() - modules := make([]string, 0, len(g.served)) - for m := range g.served { - modules = append(modules, m) - } - g.mu.Unlock() - sort.Strings(modules) - var out []Marks - if g.d.Catalog == nil { - return out - } - for _, m := range modules { - if marks := Derive(g.d.Catalog, m, g.servedFingerprint(m)); marks.Any() { - out = append(out, marks) - } - } - return out -} - -func (g *Gate) servedFingerprint(module string) string { - g.mu.Lock() - defer g.mu.Unlock() - return g.served[module] -} - -// Describe says whether a tool or a seat verb of a module is marked, and its source, for mesh_describe (§18): the served -// build's marks where this machine serves the module, else the newest entry's and its seat's. -func (g *Gate) Describe(module, seat, name string) (bool, string) { - if g.d.Catalog == nil || g.d.Catalog.Ready() != nil { - return false, "" - } - if seat != "" && module == "" { - for s, d := range g.d.Catalog.Definitions() { - if s != seat { - continue - } - for _, v := range d.Verbs { - if v.Name == name && v.NeedsApproval { - return true, "seat " + seat - } - } - } - return false, "" - } - return Derive(g.d.Catalog, module, g.servedFingerprint(module)).Marked(seat, name) -} - -// Again asks `warranted` again for one acted, unrecorded ask, by hand, in the kept-call form of §16 (§18); the caller -// records the hand act first. -func (g *Gate) Again(id string) error { - w, err := g.Read(id) - if err != nil { - return err - } - if !strings.HasPrefix(w.State, "run, not yet") { - return fmt.Errorf("the ask %s is %s: only a run not yet recorded is asked again", id, w.State) - } - g.recordRun(w.Module, id) - if again, _ := g.Read(id); !again.Recorded { - return fmt.Errorf("mesh-hand-acts.warranted has not answered for ask %s yet", id) - } - return nil -} - -// Cancel takes an open ask back through operator-channel's client (§18); the caller records the hand act first. -func (g *Gate) Cancel(id string) error { - w, err := g.Read(id) - if err != nil { - return err - } - if w.State != "asked" { - return fmt.Errorf("the ask %s is %s, not open", id, w.State) - } - return g.client(w.Module).Cancel(id) -} diff --git a/node-tools/internal/approval/wait.go b/node-tools/internal/approval/wait.go deleted file mode 100644 index 7fa963c..0000000 --- a/node-tools/internal/approval/wait.go +++ /dev/null @@ -1,296 +0,0 @@ -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/bus/chunked.go b/node-tools/internal/bus/chunked.go new file mode 100644 index 0000000..d1da74c --- /dev/null +++ b/node-tools/internal/bus/chunked.go @@ -0,0 +1,95 @@ +package bus + +import ( + "bytes" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// FileField is the member of a chunked verb's answer that carries the file received: its whole bytes, base64 (design 47 G +// A-249). +const FileField = "file" + +// AskAsChunked is AskAs of a seat's verb whose definition says `chunks` (design 47 G §681, A-237, A-249): it chooses the call +// token (the body's `call`), hears the chunk subjects of that call before it asks, receives the chunks the verb's answer +// counts with buscalls.SeatChunks, each acknowledged before the next, and answers the verb's own answer with the whole +// file added as FileField. The asking bundle receives no chunk itself. +func (c *Conn) AskAsChunked(module, key string, body any) (Answered, error) { + seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + if !strings.HasPrefix(key, "seat:") || !ok || strings.Contains(verb, "@") { + return Answered{}, fmt.Errorf("%q is no seat:. of no one machine, which a chunked answer needs", key) + } + args := map[string]any{} + if raw, err := json.Marshal(body); err != nil { + return Answered{}, err + } else if string(raw) != "null" { + if err := json.Unmarshal(raw, &args); err != nil { + return Answered{}, fmt.Errorf("the arguments of %s are not an object: %w", key, err) + } + } + var bits [16]byte + if _, err := rand.Read(bits[:]); err != nil { + return Answered{}, err + } + send := buscalls.SeatChunks{Seat: seat, Verb: verb, Call: hex.EncodeToString(bits[:])} + args["call"] = send.Call + chunks := make(chan *nats.Msg, 4) + sub, err := c.nc.ChanSubscribe(send.Chunks(), chunks) + if err != nil { + return Answered{}, fmt.Errorf("the chunks of %s cannot be heard: %w", key, err) + } + defer func() { _ = sub.Unsubscribe() }() + answered, err := c.AskAs(module, key, args) + if err != nil { + return answered, err + } + fields := map[string]json.RawMessage{} + if err := json.Unmarshal(answered.Result, &fields); err != nil { + return answered, fmt.Errorf("%s answered what is no object: %w", key, err) + } + var count int64 + if raw, has := fields["chunks"]; has { + if err := json.Unmarshal(raw, &count); err != nil { + return answered, fmt.Errorf("%s answered chunks that is no count: %w", key, err) + } + } + var file bytes.Buffer + if count > 0 { + next := func() (string, []byte, error) { + select { + case m := <-chunks: + return m.Subject, m.Data, nil + case <-time.After(buscalls.SeatAckWithin): + return "", nil, buscalls.ErrSeatChunkMissed + } + } + ack := func(subject string) error { + _, err := c.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin) + return err + } + if _, err := send.Receive(count, next, ack, &file); err != nil { + return answered, fmt.Errorf("the chunks of %s: %w", key, err) + } + } + if _, taken := fields[FileField]; taken { + return answered, errors.New(key + " answers a member named " + FileField + " itself") + } + enc, _ := json.Marshal(base64.StdEncoding.EncodeToString(file.Bytes())) + fields[FileField] = enc + out, err := json.Marshal(fields) + if err != nil { + return answered, err + } + answered.Result = out + return answered, nil +} diff --git a/node-tools/internal/bus/chunked_test.go b/node-tools/internal/bus/chunked_test.go new file mode 100644 index 0000000..a14741e --- /dev/null +++ b/node-tools/internal/bus/chunked_test.go @@ -0,0 +1,75 @@ +package bus_test + +import ( + "encoding/base64" + "encoding/json" + "testing" + + "github.com/nats-io/nats.go" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// Design 47 G A-249, §681: the runner asks a chunked verb with a call token of its own, receives the chunks the answer counts +// (each acknowledged before the next) and answers the verb's own answer with the whole file, so the asker receives none. +func TestAChunkedSeatAnswerIsReceivedByTheRunnerAndHandedWhole(t *testing.T) { + url := mt.URL(t) + holder := dial(t, url, "mesh-delivery-indexer", "anchor", "") + asker := dial(t, url, "mesh-build-go", "anchor", "") + raw, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(raw.Close) + file := make([]byte, buscalls.SeatChunkBytes*2+5) + for i := range file { + file[i] = byte(i) + } + // the holder's process: answers the count, then sends each chunk and waits for its acknowledgement + stop, err := holder.Handle("seat:mesh-catalog.input", func(body json.RawMessage) (any, error) { + var in struct{ Call, Digest string } + if err := json.Unmarshal(body, &in); err != nil || in.Call == "" { + t.Errorf("the verb was asked with no call token: %s (%v)", body, err) + return nil, err + } + send := buscalls.SeatChunks{Seat: "mesh-catalog", Verb: "input", Call: in.Call} + go func() { + for n := int64(1); n <= 3; n++ { + end := min(int64(len(file)), n*buscalls.SeatChunkBytes) + data, _ := json.Marshal(buscalls.SeatChunk{N: n, Of: 3, Data: file[(n-1)*buscalls.SeatChunkBytes : end]}) + ackSub, _ := raw.SubscribeSync(send.Ack(n)) + _ = raw.Flush() + if err := raw.Publish(send.Chunk(n), data); err != nil { + return + } + m, err := ackSub.NextMsg(buscalls.SeatAckWithin) + if err != nil { + t.Errorf("chunk %d not acknowledged: %v", n, err) + return + } + _ = m.Respond([]byte("{}")) + } + }() + return map[string]any{"digest": in.Digest, "chunks": 3}, nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + got, err := asker.AskAsChunked("mesh-build-go", "seat:mesh-catalog.input", map[string]any{"digest": "d1"}) + if err != nil { + t.Fatal(err) + } + var ans struct { + Digest string `json:"digest"` + Chunks int `json:"chunks"` + File string `json:"file"` + } + if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Digest != "d1" || ans.Chunks != 3 { + t.Fatalf("the answer %s: %v", got.Result, err) + } + if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) { + t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err) + } +} diff --git a/node-tools/internal/console/approvals.go b/node-tools/internal/console/approvals.go index f189e8b..17bcaf3 100644 --- a/node-tools/internal/console/approvals.go +++ b/node-tools/internal/console/approvals.go @@ -4,7 +4,7 @@ import ( "log" "strings" - "github.com/novox/mesh-tools/node-tools/internal/approval" + "git.novox.be/novox/mesh-sdk/go/approval" ) // verbApprovals is the mesh MCP server's tool that reads and steers what the gate asked on this machine (design 46 diff --git a/node-tools/internal/console/wait.go b/node-tools/internal/console/wait.go index 541e620..d15df4d 100644 --- a/node-tools/internal/console/wait.go +++ b/node-tools/internal/console/wait.go @@ -21,7 +21,7 @@ import ( "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" + "git.novox.be/novox/mesh-sdk/go/approval" ) // WaitDefault is how long a wait lasts when no timeout is given, and WaitLongest the most it may last diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index e66fe91..c0af795 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -19,11 +19,12 @@ import ( "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/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" + runnerapproval "github.com/novox/mesh-tools/node-tools/internal/approval" - "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -400,13 +401,14 @@ func newGatedWaits(t *testing.T, s *Surface, mesh *mt.Mesh, build string) *gated "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", "")} + state := mt.NewState() 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 }, + State: state, 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 { + Channel: &operatorchannelclient.GateChannel{State: state, KeyPrefix: "desk.", Publish: func(module, subject string, body any, id string) error { a, ok := body.(opseat.Ask) if subject != opseat.AcceptSubjectAsk(module) || !ok { return nil @@ -418,15 +420,15 @@ func newGatedWaits(t *testing.T, s *Surface, mesh *mt.Mesh, build string) *gated 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 - }}) + ReadRecord: 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 @@ -496,14 +498,20 @@ func (c *waitCatalog) wait(within time.Duration) error { 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) Served(string) (approval.Entry, bool, string) { + return approval.Entry{}, false, "not-found" } -func (c *waitCatalog) Newest(string) (catalogseat.CatalogEntry, bool) { - return catalogseat.CatalogEntry{}, false +func (c *waitCatalog) Newest(string) (approval.Entry, bool) { + return approval.Entry{}, false } func (c *waitCatalog) Definitions() map[string]seatdef.Definition { return nil } -func (c *waitCatalog) Entries() []catalogseat.CatalogEntry { return c.entries } +func (c *waitCatalog) Entries() []approval.Entry { + var out []approval.Entry + for _, e := range c.entries { + out = append(out, runnerapproval.Entry(e)) + } + return out +} // 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) { diff --git a/node-tools/internal/firstview/approval.go b/node-tools/internal/firstview/approval.go index f230589..e6f911b 100644 --- a/node-tools/internal/firstview/approval.go +++ b/node-tools/internal/firstview/approval.go @@ -7,15 +7,15 @@ import ( "time" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" - catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" - "github.com/novox/mesh-tools/node-tools/internal/approval" + runnerapproval "github.com/novox/mesh-tools/node-tools/internal/approval" ) // Findings is the runner's held finding writer, the one the gate writes its findings through too (design 46 7-approval-mark // §4, §16, §19); nil when it was not made. -func (v *Views) Findings() approval.FindingWriter { +func (v *Views) Findings() runnerapproval.HeldWriter { if v.deps.Findings == nil { return nil } @@ -55,20 +55,20 @@ func (c approvalCatalog) Revision() string { return "" } -func (c approvalCatalog) Served(fingerprint string) (catalogseat.CatalogEntry, bool, string) { +func (c approvalCatalog) Served(fingerprint string) (approval.Entry, bool, string) { if c.v.branches == nil { - return catalogseat.CatalogEntry{}, false, errNotFollowed.Error() + return approval.Entry{}, false, errNotFollowed.Error() } e, state := c.v.branches.Resolve(context.Background(), fingerprint) - return e, state == meshcatalogclient.EntryAnswered, string(state) + return runnerapproval.Entry(e), state == meshcatalogclient.EntryAnswered, string(state) } -func (c approvalCatalog) Newest(module string) (catalogseat.CatalogEntry, bool) { +func (c approvalCatalog) Newest(module string) (approval.Entry, bool) { if c.v.branches == nil { - return catalogseat.CatalogEntry{}, false + return approval.Entry{}, false } e, ok, err := c.v.branches.Entry(module) - return e, ok && err == nil + return runnerapproval.Entry(e), ok && err == nil } func (c approvalCatalog) Definitions() map[string]seatdef.Definition { @@ -78,15 +78,15 @@ func (c approvalCatalog) Definitions() map[string]seatdef.Definition { return c.defs() } -func (c approvalCatalog) Entries() []catalogseat.CatalogEntry { +func (c approvalCatalog) Entries() []approval.Entry { halves, err := c.v.Halves() if err != nil { return nil } - var out []catalogseat.CatalogEntry + var out []approval.Entry for _, half := range halves { for _, e := range half.Entries { - out = append(out, e) + out = append(out, runnerapproval.Entry(e)) } } return out diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 77fc83b..604c93c 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -26,11 +26,11 @@ import ( catalogseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/bus" ) diff --git a/node-tools/internal/meshtest/state.go b/node-tools/internal/meshtest/state.go index 87c9e4d..4ec6417 100644 --- a/node-tools/internal/meshtest/state.go +++ b/node-tools/internal/meshtest/state.go @@ -2,14 +2,18 @@ package meshtest import ( "encoding/json" + "fmt" "sort" + "strings" "sync" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" ) // 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. +// refusing them; Vals are its values by key. A key the bus refuses (a name outside letters, digits and -/_=, in +// dot-separated names) is refused here too, so no test passes a key the live bus refuses (lead answer A-209). type State struct { mu sync.Mutex Vals map[string]json.RawMessage @@ -18,12 +22,25 @@ type State struct { Fail error } +// checkKey refuses a key the bus would. +func checkKey(k string) error { + for _, part := range strings.Split(k, ".") { + if keys.CheckName("key", part) != nil { + return fmt.Errorf("%q is not a key the bus can hold: letters, digits and -/_=, in dot-separated names", k) + } + } + return nil +} + // 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 err := checkKey(k); err != nil { + return nil, err + } if m.Fail != nil { return nil, m.Fail } @@ -36,6 +53,9 @@ func (m *State) Get(k string) (*stdio.StateEntry, error) { func (m *State) Create(k string, v any) (uint64, error) { m.mu.Lock() defer m.mu.Unlock() + if err := checkKey(k); err != nil { + return 0, err + } if m.Fail != nil { return 0, m.Fail } @@ -47,6 +67,9 @@ func (m *State) Create(k string, v any) (uint64, error) { func (m *State) Update(k string, v any, rev uint64) (uint64, error) { m.mu.Lock() defer m.mu.Unlock() + if err := checkKey(k); err != nil { + return 0, err + } if m.Fail != nil { return 0, m.Fail } @@ -58,6 +81,9 @@ func (m *State) Update(k string, v any, rev uint64) (uint64, error) { func (m *State) Put(k string, v any) (uint64, error) { m.mu.Lock() defer m.mu.Unlock() + if err := checkKey(k); err != nil { + return 0, err + } if m.Fail != nil { return 0, m.Fail } diff --git a/node-tools/internal/runtime/chunked_test.go b/node-tools/internal/runtime/chunked_test.go new file mode 100644 index 0000000..3b929c1 --- /dev/null +++ b/node-tools/internal/runtime/chunked_test.go @@ -0,0 +1,29 @@ +package runtime + +import ( + "testing" + + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// Design 47 G A-249: a bundle's seat call is received by the runner in chunks only for a verb whose definition says `chunks`, +// of no one machine. +func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) { + c := &consumers{definitions: func() map[string]seatdef.Definition { + return map[string]seatdef.Definition{"mesh-catalog": {Verbs: []seatdef.Verb{{Name: "input", Chunks: true}, {Name: "entry"}}}} + }} + for key, want := range map[string]bool{ + "seat:mesh-catalog.input": true, + "seat:mesh-catalog.entry": false, + "seat:mesh-catalog.input@ace": false, + "seat:mesh-unread.input": false, + "mesh-delivery-indexer.indexer": false, + } { + if got := c.chunked(key); got != want { + t.Errorf("chunked(%s) = %v, want %v", key, got, want) + } + } + if (&consumers{}).chunked("seat:mesh-catalog.input") { + t.Error("a runner that holds no seat definitions asks no chunks") + } +} diff --git a/node-tools/internal/runtime/gate.go b/node-tools/internal/runtime/gate.go index 7c8020e..960f299 100644 --- a/node-tools/internal/runtime/gate.go +++ b/node-tools/internal/runtime/gate.go @@ -2,10 +2,11 @@ package runtime import ( "fmt" + "strings" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" - "github.com/novox/mesh-tools/node-tools/internal/approval" "github.com/novox/mesh-tools/node-tools/internal/launch" ) @@ -86,3 +87,21 @@ func (r *runner) unservedMarks(module string, regs []launch.Registration) []stri } return r.events.gate.Unserved(module, r.fingerprintOf(module), registered) } + +// chunked says whether key names a seat's verb whose definition says `chunks` (design 47 G A-249): its answer is received in +// chunks by the runner, whose call hands the bundle the whole. +func (c *consumers) chunked(key string) bool { + if c.definitions == nil || !strings.HasPrefix(key, "seat:") { + return false + } + seat, verb, ok := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + if !ok || strings.Contains(verb, "@") { + return false + } + for _, v := range c.definitions()[seat].Verbs { + if v.Name == verb { + return v.Chunks + } + } + return false +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 0b3011c..10dfa64 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -16,10 +16,10 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" "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" ) @@ -937,7 +937,11 @@ func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { if len(asked.Body) == 0 { body = map[string]any{} } - answered, err := b.all.conn.AskAs(b.module, asked.Key, body) + ask := b.all.conn.AskAs + if b.all.chunked(asked.Key) { + ask = b.all.conn.AskAsChunked // the runner receives the chunks of the answer; the bundle receives none (design 47 G A-249) + } + answered, err := ask(b.module, asked.Key, body) if err != nil { return nil, err }