From ee8958b0463e5a9dda4647542818a095d2debc7a Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:25:56 +0100 Subject: [PATCH] A3e item 1: the approval gate is mesh-sdk's go/approval, reading the channel, catalog, findings and holders through interfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §677, §700, lead answers A-219, A-224, A-228 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/approval/act.go | 293 +++++++++++++++++++++++++++++++++++ go/approval/approval_test.go | 80 ++++++++++ go/approval/ask.go | 222 ++++++++++++++++++++++++++ go/approval/channel.go | 115 ++++++++++++++ go/approval/closed.go | 97 ++++++++++++ go/approval/gate.go | 170 ++++++++++++++++++++ go/approval/marks.go | 172 ++++++++++++++++++++ go/approval/record.go | 255 ++++++++++++++++++++++++++++++ go/approval/run.go | 176 +++++++++++++++++++++ go/approval/tools.go | 172 ++++++++++++++++++++ go/approval/wait.go | 290 ++++++++++++++++++++++++++++++++++ go/packages_list_test.go | 2 +- 12 files changed, 2043 insertions(+), 1 deletion(-) create mode 100644 go/approval/act.go create mode 100644 go/approval/approval_test.go create mode 100644 go/approval/ask.go create mode 100644 go/approval/channel.go create mode 100644 go/approval/closed.go create mode 100644 go/approval/gate.go create mode 100644 go/approval/marks.go create mode 100644 go/approval/record.go create mode 100644 go/approval/run.go create mode 100644 go/approval/tools.go create mode 100644 go/approval/wait.go diff --git a/go/approval/act.go b/go/approval/act.go new file mode 100644 index 00000000..2d61d27e --- /dev/null +++ b/go/approval/act.go @@ -0,0 +1,293 @@ +package approval + +import ( + "bytes" + "encoding/json" + "fmt" + "math" + "sort" + "strconv" + "strings" + "unicode/utf8" + + "git.novox.be/novox/mesh-sdk/go/operatorask" + "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) (operatorask.Act, error) { + act := operatorask.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/go/approval/approval_test.go b/go/approval/approval_test.go new file mode 100644 index 00000000..29ca678e --- /dev/null +++ b/go/approval/approval_test.go @@ -0,0 +1,80 @@ +package approval + +import ( + "encoding/json" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// call is a fixture call of the marked tool wipe. +func call(name, args string) Call { + return Call{Address: "shell." + name, Module: "shell", Name: name, Fingerprint: "src2:1", Caller: "agent", Args: json.RawMessage(args), + Description: "Wipes a scratch directory.\nMore words."} +} + +// Design 46 7-approval-mark §8, §10, §11: the act's fields give each call its digest, and the ask shows a call's arguments +// in their shape and its headline cut (the gate's other cases run on the tool runner's fixture mesh, node-tools +// internal/approval). +func TestTheActAndItsAsk(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, _ := operatorask.ActDigest(act) + return d + } + act, _ := ActOf(base, "n1") + want := operatorask.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) + } +} diff --git a/go/approval/ask.go b/go/approval/ask.go new file mode 100644 index 00000000..1cbc3143 --- /dev/null +++ b/go/approval/ask.go @@ -0,0 +1,222 @@ +package approval + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// 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" +) + +// actKey is the key naming the newest ask of an act (§13), its digest escaped as a key's name is, since the bus holds no `:` in a key (lead answer A-209). +func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + keys.EscapeName(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) 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 Ask{ID: id, About: digest, Headline: headline, + Explanation: strings.Join(lines, "\n"), Expires: formatTime(g.d.Now().Add(AskLasts)), OnExpiry: OnExpiry, + Options: []Option{ + {ID: OptionApprove, Label: "Approve", Does: "the same call, made again within 30 minutes, runs once", Binds: digest}, + {ID: OptionDecline, Label: "Decline", Does: "nothing runs; the same call is not asked again for 10 minutes", 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 := 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 := operatorask.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 operatorask.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 operatorask.Act, digest string, a Ask) error { + 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 := g.d.Channel.Ask(c.Module, 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 Ask, act operatorask.Act, args json.RawMessage) int { + raw, _ := json.Marshal(Kept[Ask, Warrant]{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 operatorask.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 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) (*Record, error) { + return g.d.Channel.Record(module, id) +} + +// 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) (*Record, 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("[approval] %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/go/approval/channel.go b/go/approval/channel.go new file mode 100644 index 00000000..b86689b7 --- /dev/null +++ b/go/approval/channel.go @@ -0,0 +1,115 @@ +package approval + +import ( + "encoding/json" + "time" + + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// What the gate reads beyond itself crosses in the gate's own terms, its caller translating from the generated contracts +// (design 47 G lead answers A-219, A-224, A-228): the operator channel (Channel), the catalog's entries (Entry, through +// Catalog), the finding writer (FindingWriter) and the seat holders (Deps.Holder). No form of a generated contract is +// copied here: what the router keeps whole (an ask, a warrant) the gate holds as the bytes it was given. + +// Channel is the operator channel as the gate reaches it (A-219, A-228): operatorchannelclient implements it over the +// generated types, keeping each ask in the gate's own state under the marked module's name. +type Channel interface { + // Ask checks the ask, keeps it in the gate's state as a Kept and publishes it under module's name, at the level + // approve, to the operator (design 46 7-approval-mark §10). + Ask(module string, a Ask) error + // Record is the router's record of module's ask id, read under module's own name; nil while the router holds none + // (design 47 G §700). + Record(module, id string) (*Record, error) + // Claim takes the warrant once (design 46 7-approval-mark §14, §15): it holds it to the ask kept, checks that the + // chosen option binds act and marks the kept ask acted, then calls run; act's run happens only when it answers + // Claimed. A warrant already taken answers ClaimDone, one that does not hold ClaimRefused with why. + Claim(module string, w Warrant, act operatorask.Act, run func()) (Claim, error) + // Cancel takes module's ask back. + Cancel(module, id string) error + // Decided is the subject the router's `decided` for module is heard on. + Decided(module string) string + // Seat is the seat an ask is raised on, whose holder must be there for a marked call to be asked (§19). + Seat() string +} + +// Claim is what came of a warrant handed to Channel.Claim. +type Claim string + +const ( + Claimed Claim = "acted" + ClaimRefused Claim = "refused" + ClaimDone Claim = "already acted on" + // ClaimEnded is a warrant of an ask that ended without a choice. + ClaimEnded Claim = "ended without a choice" +) + +// Ask is an ask the gate raises: two options, each binding the act's digest (design 46 7-approval-mark §10, §11). +type Ask struct { + ID, About, Headline, Explanation, Expires, OnExpiry string + Options []Option +} + +// Option is one option of an ask. +type Option struct{ ID, Label, Does, Binds string } + +// Record is the router's record of an ask as the gate reads it. +type Record struct { + State, Answer, Opened, Expires string + OfferedOn []string + Warrant *Warrant +} + +// Warrant is a person's choice as the gate reads it; Whole is the warrant as the router recorded it, which Claim holds to +// the kept ask. +type Warrant struct { + Ask, Asker, Option, Channel, At string + Whole json.RawMessage +} + +// Kept is one ask as its asker keeps it (design 46 7-approval-mark §13): the ask, beside it the act the ask binds, the +// call's whole arguments, and whether the act's run was recorded in the hand-act log. A is the ask's form and W the +// warrant's: operatorchannelclient's Kept is Kept of the generated contract's, the gate's of the bytes kept (A-228). +type Kept[A, W any] struct { + Ask A `json:"ask"` + Asked time.Time `json:"asked"` + Acted bool `json:"acted,omitempty"` + Ended string `json:"ended,omitempty"` + Answer *W `json:"answer,omitempty"` + // Act is the act the ask's options bind, as its asker digested it. + Act operatorask.Act `json:"act,omitempty"` + // Arguments are the asked call's arguments whole. + Arguments json.RawMessage `json:"arguments,omitempty"` + // Recorded says the act's run is in the hand-act log (`mesh-hand-acts.warranted` answered). + Recorded bool `json:"recorded,omitempty"` +} + +// kept is the gate's own reading of a Kept: the ask and the warrant as the bytes kept, so a write back keeps them whole. +type kept = Kept[json.RawMessage, json.RawMessage] + +// HeadlineLength is the longest headline the router takes, in characters; operatorchannelclient re-exports it. +const HeadlineLength = 60 + +// Entry is a module's entry in the catalog as the marks are read from it (design 46 7-approval-mark §5): its module, the +// build's fingerprint, the seat verbs it serves as `.`, and its manifest. +type Entry struct { + Module, Fingerprint string + Serves []string + Manifest json.RawMessage +} + +// formatTime is a time as the router's records carry it: RFC 3339 to the nanosecond; the zero time is empty. +func formatTime(t time.Time) string { + if t.IsZero() { + return "" + } + return t.Format(time.RFC3339Nano) +} + +// parseTime reads such a time; empty is the zero time. +func parseTime(s string) (time.Time, error) { + if s == "" { + return time.Time{}, nil + } + return time.Parse(time.RFC3339Nano, s) +} diff --git a/go/approval/closed.go b/go/approval/closed.go new file mode 100644 index 00000000..6ee0d1c1 --- /dev/null +++ b/go/approval/closed.go @@ -0,0 +1,97 @@ +package approval + +import ( + "fmt" + "time" +) + +// 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.Channel == nil { + return "the tool runner reaches no operator channel to ask on" + } + if g.d.Holder == nil { + return "the tool runner reads no view of the seats' holders, so it cannot tell that a module holds " + g.d.Channel.Seat() + } + on := g.d.Channel.Seat() + ok, err := g.d.Holder(on) + switch { + case err != nil: + return fmt.Sprintf("whether a module holds %s cannot be read: %v", on, err) + case !ok: + return "no module holds " + on + ", 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("[approval] 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("[approval] 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) +} diff --git a/go/approval/gate.go b/go/approval/gate.go new file mode 100644 index 00000000..170bae16 --- /dev/null +++ b/go/approval/gate.go @@ -0,0 +1,170 @@ +package approval + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" +) + +// 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 raises and clears the gate's findings by kind and id (A-228): its caller writes each as the contract's +// finding, about its own module on its machine, through its held writer. +type FindingWriter interface { + Raise(ctx context.Context, kind, id, summary, headline string) error + Clear(ctx context.Context, kind, id, 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 + // Channel is the operator channel, its asks kept in State under this machine's keys (A-219, A-228). + Channel Channel + // 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 approval gate: every call of a module the tool runner serves goes through it before the bundle (§20), and a +// marked verb a module's process serves before its handler (design 47 G §677; lead answer A-224). +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, 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 := g.findingID(what) + if g.d.Findings == nil { + g.d.Logf("[approval] %s %s: %s", kind, id, summary) + return + } + var err error + if raised { + err = g.d.Findings.Raise(context.Background(), kind, id, summary, headline) + } else { + err = g.d.Findings.Clear(context.Background(), kind, id, summary) + } + if err != nil { + g.d.Logf("[approval] the finding %s of %s is not kept: %v", kind, id, err) + } +} diff --git a/go/approval/marks.go b/go/approval/marks.go new file mode 100644 index 00000000..e162edff --- /dev/null +++ b/go/approval/marks.go @@ -0,0 +1,172 @@ +// Package approval is the one gate on a call that needs the operator's approval (design 46 7-approval-mark; design 47 G +// §677, lead answer A-224): the tool runner's and a module's process's seat server's. It holds 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" + + "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) (Entry, bool, string) + // Newest is the module's entry at the newest revision; false when the catalog half lists none. + Newest(module string) (Entry, 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() []Entry +} + +// 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 []Entry + 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.Fingerprint != entries[0].Fingerprint) { + 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 Entry, defs map[string]seatdef.Definition) { + source := "entry " + e.Module + " " + orNone(e.Fingerprint) + var said marked + _ = json.Unmarshal(e.Manifest, &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 +} diff --git a/go/approval/record.go b/go/approval/record.go new file mode 100644 index 00000000..22c1af32 --- /dev/null +++ b/go/approval/record.go @@ -0,0 +1,255 @@ +package approval + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// 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 *Record, 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("[approval] %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("[approval] 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("[approval] 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("[approval] 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("[approval] 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) (*kept, uint64, error) { + e, err := g.d.State.Get(g.askKey(id)) + if err != nil || e == nil { + return nil, 0, err + } + var k 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("[approval] 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 *kept) string { + var w struct { + Asker string `json:"asker"` + } + if k.Answer != nil { + _ = json.Unmarshal(*k.Answer, &w) + } + return w.Asker +} + +// 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("[approval] 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 held writer's pending (views.HeldWriter's Pend and Judged), by a finding's kind and id: the findings it +// cannot judge yet since its start, named in its `findings-stated` (design 47 G §384). +type Pender interface { + Pend(kind, id string, since time.Time, bound time.Duration) + Judged(kind, id 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(KindCallUnanswered, g.findingID(id), g.d.Now().UTC(), UnansweredAfter) + } +} + +// judged ends that pending. +func (g *Gate) judged(id string) { + if p, ok := g.d.Findings.(Pender); ok { + p.Judged(KindCallUnanswered, g.findingID(id)) + } +} + +// findingID is the id of a finding about what on this machine, `.` escaped as a key's name is (lead answer +// A-154). +func (g *Gate) findingID(what string) string { return keys.EscapeName(g.d.Node + "." + what) } + +// isChanged says a compare-and-set lost to another write. +func isChanged(err error) bool { return errors.Is(err, stdio.ErrChanged) } diff --git a/go/approval/run.go b/go/approval/run.go new file mode 100644 index 00000000..5cefae6f --- /dev/null +++ b/go/approval/run.go @@ -0,0 +1,176 @@ +package approval + +import ( + "encoding/json" + "fmt" + "strings" + + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// 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 := operatorask.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 operatorask.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, _ := parseTime(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, 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.d.Channel.Claim(c.Module, *w, act, func() { answer, ran = run() }) + switch { + case h == ClaimDone: + return again("was already run") + case h == ClaimRefused: + 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/go/approval/tools.go b/go/approval/tools.go new file mode 100644 index 00000000..7b1d78db --- /dev/null +++ b/go/approval/tools.go @@ -0,0 +1,172 @@ +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) + } + var ask struct { + About string `json:"about"` + } + _ = json.Unmarshal(k.Ask, &ask) + 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(ask.About)}, + Digest: 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.d.Channel.Cancel(w.Module, id) +} diff --git a/go/approval/wait.go b/go/approval/wait.go new file mode 100644 index 00000000..aea3a59a --- /dev/null +++ b/go/approval/wait.go @@ -0,0 +1,290 @@ +package approval + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/operatorask" + "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 := operatorask.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, _ := parseTime(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 + // On the warrant it subscribes, and the wait's timeout runs from then (§35). + h, err := g.d.Channel.Claim(c.Module, *wt, act, func() { answer, ran = hear(w.Timeout) }) + switch { + case h == ClaimDone: + return anew() // already run: one warrant is one wait (§35) + case h == ClaimRefused: + 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{g.d.Channel.Decided(module)}, left, func(body json.RawMessage) bool { + var w struct { + Ask string `json:"ask"` + } + 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 *Record, 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, _ := parseTime(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, + 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) 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/go/packages_list_test.go b/go/packages_list_test.go index f246289f..f0be789b 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -26,7 +26,7 @@ var meshSDKPackages = []string{ ".", "announce", "announcetest", "busarchive", "buscalls", "buscompose", "busstate", "bussnap", "busretry", "calladdress", "catalogue", "declaration", "forgekey", "health", "internal/beside", "keys", "kvlimit", "membership", "motion", "objectlink", "outward", "pernode", "process", - "sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "operatorask", "stateobjects", "testbus", "views", "views/viewstest", + "sdkversion", "seal", "seatdef", "seatlist", "seatlist/seatlisttest", "operatorask", "approval", "stateobjects", "testbus", "views", "views/viewstest", } // clientSeats are the seats whose logic is their client's in mesh-contracts (novox/hq ADR 0627 §1, §4): mesh-sdk names