A3e item 1: the approval gate is mesh-sdk's go/approval, reading the channel, catalog, findings and holders through interfaces
Builds: design 47 G §677, §700, lead answers A-219, A-224, A-228 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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: `<module>.<tool>` or `<seat>.<verb>`.
|
||||
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.<name>` 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]
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 "<verb> <names> on <node>?", "<verb> <names>?" 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)
|
||||
}
|
||||
@@ -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 `<seat>.<verb>`, 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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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 `<node>.<what>` 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)
|
||||
}
|
||||
}
|
||||
@@ -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 `<seat>.<verb>`, 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
|
||||
}
|
||||
@@ -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, `<node>.<what>` 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) }
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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 <names> on <node>?", 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"), ".")
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user