diff --git a/go/approval/ask.go b/go/approval/ask.go index 1cbc3143..1033882e 100644 --- a/go/approval/ask.go +++ b/go/approval/ask.go @@ -43,10 +43,19 @@ const ( ) // 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) } +func (g *Gate) actKey(digest string) string { return g.prefix() + "act." + keys.EscapeName(digest) } + +// prefix is what the gate's keys begin with: the tool runner's machine, `.`, since its state is shared by every +// machine's runner (§13); none for a module's process, whose state approvals is its module's own (design 47 G §677). +func (g *Gate) prefix() string { + if g.d.Holding { + return "" + } + return g.d.Node + "." +} // askKey is the key the client keeps an ask under (§13). -func (g *Gate) askKey(id string) string { return g.d.Node + ".ask." + id } +func (g *Gate) askKey(id string) string { return g.prefix() + "ask." + id } // newID is an ask's id: 128 random bits. func newID() string { @@ -67,12 +76,20 @@ func (g *Gate) askFor(c Call, digest, id string) Ask { if first == "" { first = name } - lines := []string{first, "On " + g.d.Node + ", called through the mesh's tools, which cannot tell who called."} + how := "called through the mesh's tools, which cannot tell who called." + if g.d.Holding { + how = "called by " + c.Caller + "." + } + lines := []string{first, "On " + g.d.Node + ", " + how} 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.") + read := "mesh_approvals on " + g.d.Node + if g.d.Holding { + read = c.Seat + ".approvals" + } + lines = append(lines, "The whole is read with "+read+", whose fingerprint must be the one on the phone.") } lines = append(lines, "Fingerprint "+Fingerprint(digest)) return Ask{ID: id, About: digest, Headline: headline, diff --git a/go/approval/closed.go b/go/approval/closed.go index 6ee0d1c1..00310a6d 100644 --- a/go/approval/closed.go +++ b/go/approval/closed.go @@ -17,6 +17,9 @@ 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 && g.d.Holding { + return "" // a process holder reads no view of the seats' holders: an ask nobody takes is said by its gate-unanswered (design 47 G §677) + } 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() } diff --git a/go/approval/gate.go b/go/approval/gate.go index 170bae16..2f6d9cf2 100644 --- a/go/approval/gate.go +++ b/go/approval/gate.go @@ -59,6 +59,11 @@ type Deps struct { 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()) + // Holding says the gate is a module's process's seat server's (design 47 G §677): its keys are its module's own, its + // calls held as waiting calls and their ends said through Say; Runner is then the holder's module. + Holding bool + // Say publishes one gated-ended on its subject with the message id given (§677). + Say func(subject string, body any, id string) error } // Gate is the approval gate: every call of a module the tool runner serves goes through it before the bundle (§20), and a @@ -76,6 +81,8 @@ type Gate struct { inFlight map[string]bool // failingSince is when the state approvals began to fail; zero while it is read. failingSince time.Time + // handlers are a process holder's marked verbs' handlers, by `.` (design 47 G §677). + handlers map[string]func(string, json.RawMessage) (any, error) // served is each module this runner serves, with the build its bundle serves, as its last load said (§18). served map[string]string } diff --git a/go/approval/held.go b/go/approval/held.go new file mode 100644 index 00000000..3c0b0ff1 --- /dev/null +++ b/go/approval/held.go @@ -0,0 +1,653 @@ +package approval + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// A marked verb a module's process serves is one call with a gate before its running (design 47 G §677): the holder keeps +// the call, asks under its own module's name, and on the Approve runs that same call, saying its end on the seat's event +// `gated-ended`. These are the holder's half of the gate; the caller's is process.Conn's call. + +// WaitHeader is the bus header a call carries to say it waits for its end on `gated-ended` (lead answer A-244): a header, +// never an argument, so no verb's input and no act's digest changes. WaitHeld is its one value. +const ( + WaitHeader = "x-wait" + WaitHeld = "held" +) + +// Held is the holder's answer to a call that waits (§677): the ask it waits on and its own call's id, in the reply's field +// `held` beside result, error and node (A-244). +type Held struct { + Asked string `json:"asked"` + Call string `json:"call"` +} + +// GatedEndedEvent is the seat's event a call's end is said on, kept last per subject for GatedEndedKeptFor (§677). +const ( + GatedEndedEvent = "gated-ended" + GatedEndedKeptFor = 24 * time.Hour +) + +// The outcomes of a held call (§677). +const ( + OutcomeRan = "ran" + OutcomeDeclined = "declined" + OutcomeExpired = "expired" + OutcomeRefused = "refused" +) + +// KindGateUnanswered is the holder's finding of a waiting call past its ask's 60 minutes with no decided heard (§677), +// its id the call's. +const KindGateUnanswered = "gate-unanswered" + +// GatedEnded is a held call's end as the holder says it (§677): no argument is in it. +type GatedEnded struct { + Call string `json:"call"` + Ask string `json:"ask"` + Act string `json:"act"` + Verb string `json:"verb"` + Outcome string `json:"outcome"` + Answer any `json:"answer,omitempty"` + Approval string `json:"approval,omitempty"` + Error string `json:"error,omitempty"` +} + +// GatedEndedSubject is the subject one call's end is said on: `mesh.seat..event.gated-ended..`, the +// caller the account the bus proved, escaped as a key's name is (§677, §73). +func GatedEndedSubject(seat, caller, call string) string { + return "mesh.seat." + seat + ".event." + GatedEndedEvent + "." + keys.EscapeName(caller) + "." + call +} + +// WaitingCall is one call the holder keeps under `call.` (§677): the call whole, the caller the bus proved, its ask, +// and once it ended its outcome, and whether that end was said. +type WaitingCall struct { + ID string `json:"id"` + Caller string `json:"caller"` + Ask string `json:"ask"` + Digest string `json:"digest"` + Act operatorask.Act `json:"act"` + Address string `json:"address"` + Seat string `json:"seat"` + Verb string `json:"verb"` + Arguments json.RawMessage `json:"arguments,omitempty"` + // Made is when the call was made; Asked when its ask (the newest, after an ask anew) was raised or taken. + Made time.Time `json:"made"` + Asked time.Time `json:"asked"` + // End is its end once it ended; Said says End was said on gated-ended. + End *GatedEnded `json:"end,omitempty"` + Said bool `json:"said,omitempty"` +} + +func callKey(id string) string { return "call." + id } +func claimedKey(id string) string { return "claimed." + id } + +// Serves names the handler a waiting call of seat's verb runs on its Approve (§677): process.Conn's AnswerSeat gives each +// marked verb's. +func (g *Gate) Serves(seat, verb string, h func(caller string, args json.RawMessage) (any, error)) { + g.mu.Lock() + defer g.mu.Unlock() + if g.handlers == nil { + g.handlers = map[string]func(string, json.RawMessage) (any, error){} + } + g.handlers[seat+"."+verb] = h +} + +func (g *Gate) handler(seat, verb string) func(string, json.RawMessage) (any, error) { + g.mu.Lock() + defer g.mu.Unlock() + return g.handlers[seat+"."+verb] +} + +// Hold is a call of a marked verb the holder serves (§677). A call whose arguments may not be asked, or that comes while +// the state approvals cannot be read or written, is refused at once and nothing is kept. One that waits (the header +// WaitHeader) is kept as a waiting call and answered at once with Held; one that does not takes ADR 0286 §37's path: it is +// refused naming its act's ask, and the same call made again after the Approve runs. A call of an act declined within 10 +// minutes ends declined at once, unasked. +func (g *Gate) Hold(c Call, wait bool, run func() (any, error)) (any, *Held, error) { + if c.Fingerprint == "" { + c.Fingerprint = g.d.Build + } + if c.Fingerprint == "" { + return nil, nil, g.refused(c, "", "the holder's own fingerprint is unknown, so no warrant can bind its build") + } + if g.d.State == nil { + return nil, nil, g.refused(c, "", "its holder keeps no state "+StateName+" to keep its calls in") + } + if why := refusedUnasked(c.Args); why != "" { + return nil, nil, g.refused(c, "", why) + } + if why := g.held(); why != "" { + return nil, nil, g.refused(c, "", why) + } + act, err := ActOf(c, g.d.Node) + if err != nil { + return nil, nil, g.refused(c, "", err.Error()) + } + digest, err := operatorask.ActDigest(act) + if err != nil { + return nil, nil, g.refused(c, "", err.Error()) + } + if !wait { + answer, err := g.unheld(c, act, digest, run) + return answer, nil, err + } + id, err := g.askOf(c, act, digest) + if err != nil { + return nil, nil, err + } + now := g.d.Now().UTC() + w := WaitingCall{ID: newID(), Caller: c.Caller, Ask: id, Digest: digest, Act: act, Address: c.Address, Seat: c.Seat, + Verb: c.Name, Arguments: c.Args, Made: now, Asked: now} + if _, err := g.d.State.Create(callKey(w.ID), w); err != nil { + g.stateFailed(err) + return nil, nil, g.refused(c, digest, "its call cannot be kept in "+StateName+": "+err.Error()) + } + g.stateRead() + g.armUnanswered(w) + g.d.Go(func() { g.Decided(id) }) // a choice made before the call was kept is acted on now + return nil, &Held{Asked: id, Call: w.ID}, nil +} + +// askOf is the ask a call of the act takes (§677): the one `act.` names while it is open, else a new one, written +// with the bus's create (so two instances raise one ask) or, over an ended one, by compare-and-set on its revision. A call +// of an act declined within 10 minutes is refused with the time it may be asked again. +func (g *Gate) askOf(c Call, act operatorask.Act, digest string) (string, error) { + key := g.actKey(digest) + for { + e, err := g.d.State.Get(key) + if err != nil { + g.stateFailed(err) + return "", g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()) + } + g.stateRead() + id := newID() + if e == nil { + if _, err := g.d.State.Create(key, id); err != nil { + if isChanged(err) { + continue + } + g.stateFailed(err) + return "", g.refused(c, digest, "its act cannot be kept in "+StateName+": "+err.Error()) + } + return id, g.raiseHeld(c, act, digest, id) + } + var named string + if err := json.Unmarshal(e.Value, &named); err != nil || named == "" { + return "", g.refused(c, digest, fmt.Sprintf("its act's key in %s names no ask", StateName)) + } + open, until, err := g.openAsk(c.Module, named) + if err != nil { + return "", g.refused(c, digest, err.Error()) + } + if !until.IsZero() { + return "", fmt.Errorf("Not run: the operator declined ask %s; the same call is not asked again until %s", named, formatTime(until)) + } + if open { + return named, nil + } + if _, err := g.d.State.Update(key, id, e.Revision); err != nil { + if isChanged(err) { + continue + } + g.stateFailed(err) + return "", g.refused(c, digest, "its act cannot be kept in "+StateName+": "+err.Error()) + } + return id, g.raiseHeld(c, act, digest, id) + } +} + +// openAsk says whether an ask is open for another call of its act: raised and not yet ended by the router; and, when it was +// declined within 10 minutes, until when no call of its act is asked. +func (g *Gate) openAsk(module, id string) (bool, time.Time, error) { + rec, err := g.record(module, id) + if err != nil { + return false, time.Time{}, fmt.Errorf("the router's record of its ask %s cannot be read: %v", id, err) + } + if rec == nil { + k, _, err := g.kept(id) + return err == nil && k != nil && k.Ended == "", time.Time{}, nil // raised and not yet taken by the router + } + switch rec.State { + case "open", "held-back": + return true, time.Time{}, nil + case "chosen": + if w := rec.Warrant; w != nil && w.Option == OptionDecline { + at, _ := parseTime(w.At) + if until := at.Add(DeclinedFor); g.d.Now().Before(until) { + return false, until, nil + } + } + } + return false, time.Time{}, nil +} + +// raiseHeld raises the ask the act's key names, under the holder's module's name, keeping the act and the arguments beside +// it (§677; design 46 7-approval-mark §10). +func (g *Gate) raiseHeld(c Call, act operatorask.Act, digest, id string) error { + return g.keepAskHeld(c, act, digest, g.askFor(c, digest, id)) +} + +// keepAskHeld is keepAsk without its write of the act's key, which askOf made first. +func (g *Gate) keepAskHeld(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()) + } + return nil +} + +// unheld is a call that does not wait (ADR 0286 §37's path, §677): run when its act's ask is chosen Approve within 30 +// minutes and unclaimed, claimed once; otherwise refused naming the ask, raising one when none is open. +func (g *Gate) unheld(c Call, act operatorask.Act, digest string, run func() (any, error)) (any, 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 { + var id string + if json.Unmarshal(named.Value, &id) == nil && id != "" { + 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 && rec.State == "chosen" && rec.Warrant != nil && rec.Warrant.Option == OptionApprove { + if at, _ := parseTime(rec.Warrant.At); !g.d.Now().After(at.Add(RunWithin)) { + out := g.claimRun(c.Module, id, *rec.Warrant, act, run) + switch out.outcome { + case OutcomeRan: + return g.ran(c, id, rec, out.answer, out.err) + case OutcomeRefused: + return nil, g.refused(c, digest, out.why) + } + } + } + } + } + id, err := g.askOf(c, act, digest) + if err != nil { + return nil, err + } + return nil, g.firstAnswer(c, id, digest, "") +} + +// claimed is what came of a claim of a warrant for one call. +type claimed struct { + outcome string // OutcomeRan, OutcomeRefused, or "" when another took it + answer any + err error + why string +} + +// claimRun holds the warrant to the ask and the call's act, claims it once at `claimed.` with the bus's create, and +// runs the call (§677; ADR 0286 §27, §28, §30, §31): of two instances, or two deliveries, one runs. +func (g *Gate) claimRun(module, id string, w Warrant, act operatorask.Act, run func() (any, error)) claimed { + var out claimed + var claimErr error + h, err := g.d.Channel.Claim(module, w, act, func() { + if _, err := g.d.State.Create(claimedKey(id), map[string]string{"at": formatTime(g.d.Now().UTC())}); err != nil { + claimErr = err + return + } + out.outcome = OutcomeRan + out.answer, out.err = run() + }) + switch { + case h == Claimed && claimErr != nil && !isChanged(claimErr): + g.stateFailed(claimErr) + return claimed{outcome: OutcomeRefused, why: "its claim cannot be written in " + StateName + ": " + claimErr.Error()} + case h == Claimed: + return out + case h == ClaimDone: + return claimed{} + case h == "" && err != nil: + g.stateFailed(err) + return claimed{outcome: OutcomeRefused, why: "its state " + StateName + " cannot be written: " + err.Error()} + default: // refused, or the act about to run is not the one the option bound: nothing runs + return claimed{outcome: OutcomeRefused, why: "its warrant does not hold: " + errText(err)} + } +} + +// Decided acts on the router's own record of one of the holder's asks (§677): heard on the holder's `decided`, never from +// its body, and at the start for each ask a waiting call waits on. On the Approve the oldest waiting call of the ask runs, +// within 30 minutes of the choice, else it ends expired; every other is asked anew. A Decline ends every one declined, an +// ask that ended without a choice every one expired, and one the router refused every one refused with its words. +func (g *Gate) Decided(id string) { + module := g.d.Runner + rec, err := g.record(module, id) + if err != nil { + g.d.Logf("[approval] the router's record of ask %s cannot be read (%v): read again at the next start", id, err) + return + } + if rec == nil || rec.State == "open" || rec.State == "held-back" { + return + } + calls, err := g.waiting(func(w WaitingCall) bool { return w.Ask == id }) + if err != nil { + g.stateFailed(err) + g.d.Logf("[approval] the waiting calls of ask %s cannot be read (%v): read again at the next start", id, err) + return + } + if len(calls) == 0 { + return + } + end := func(w WaitingCall, outcome, why string) { + g.end(w, GatedEnded{Outcome: outcome, Error: why}) + } + switch { + case rec.State == "refused": + for _, w := range calls { + end(w, OutcomeRefused, "the router refused its ask "+id+": "+orNone(rec.Answer)) + } + return + case rec.State != "chosen": + for _, w := range calls { + end(w, OutcomeExpired, "its ask "+id+" ended "+rec.State+" with no choice; nothing ran") + } + return + case rec.Warrant == nil: + for _, w := range calls { + end(w, OutcomeRefused, "the router's record of its ask "+id+" is chosen and carries no warrant") + } + return + case rec.Warrant.Option == OptionDecline: + for _, w := range calls { + end(w, OutcomeDeclined, "the operator declined ask "+id+"; nothing ran") + } + return + } + first, rest := calls[0], calls[1:] + at, _ := parseTime(rec.Warrant.At) + if g.d.Now().After(at.Add(RunWithin)) { + end(first, OutcomeExpired, "the Approve of ask "+id+" was heard more than 30 minutes after it was given; nothing ran") + } else { + g.runWaiting(first, rec) + } + for _, w := range rest { + g.anew(w) + } +} + +// runWaiting claims and runs one waiting call on its ask's warrant, saying its end. +func (g *Gate) runWaiting(w WaitingCall, rec *Record) { + h := g.handler(w.Seat, w.Verb) + c := Call{Address: w.Address, Module: g.d.Runner, Seat: w.Seat, Name: w.Verb, Caller: w.Caller, Args: w.Arguments} + if h == nil { + g.end(w, GatedEnded{Outcome: OutcomeRefused, Error: "the holder serves no handler of " + w.Address + " now; nothing ran"}) + return + } + out := g.claimRun(g.d.Runner, w.Ask, *rec.Warrant, w.Act, func() (any, error) { return h(w.Caller, w.Arguments) }) + switch out.outcome { + case "": // another instance, or another call of the same ask, took the warrant + return + case OutcomeRefused: + g.end(w, GatedEnded{Outcome: OutcomeRefused, Error: out.why}) + return + } + answered, err := g.ran(c, w.Ask, rec, out.answer, out.err) + if err != nil { + g.end(w, GatedEnded{Outcome: OutcomeRan, Error: err.Error()}) + return + } + a := answered.(Answered) + g.end(w, GatedEnded{Outcome: OutcomeRan, Answer: a.Answer, Approval: a.Approval}) +} + +// anew asks a waiting call's act anew, as a new ask (ADR 0286 §33: a second run takes a second approval); its end comes +// when its own ask ends. +func (g *Gate) anew(w WaitingCall) { + c := Call{Address: w.Address, Module: g.d.Runner, Seat: w.Seat, Name: w.Verb, Caller: w.Caller, Args: w.Arguments, + Fingerprint: w.Act["build"]} + id, err := g.askOf(c, w.Act, w.Digest) + if err != nil { + g.end(w, GatedEnded{Outcome: OutcomeRefused, Error: err.Error()}) + return + } + if err := g.updateCall(w.ID, func(k *WaitingCall) { k.Ask, k.Asked = id, g.d.Now().UTC() }); err != nil { + g.d.Logf("[approval] the call %s, asked anew as ask %s, cannot be kept (%v)", w.ID, id, err) + return + } + w.Ask, w.Asked = id, g.d.Now().UTC() + g.armUnanswered(w) + g.d.Go(func() { g.Decided(id) }) +} + +// end keeps a waiting call's end, then says it on gated-ended (§677); an end not said is said again at the next start. +func (g *Gate) end(w WaitingCall, e GatedEnded) { + e.Call, e.Ask, e.Act, e.Verb = w.ID, w.Ask, w.Digest, w.Verb + ended := false + if err := g.updateCall(w.ID, func(k *WaitingCall) { + if k.End == nil { + k.End, ended = &e, true + } + }); err != nil { + g.d.Logf("[approval] the end %s of call %s cannot be kept (%v)", e.Outcome, w.ID, err) + return + } + if ended { + g.say(w.ID) + } +} + +// say says a kept end on gated-ended and marks it said; its finding gate-unanswered is cleared. +func (g *Gate) say(id string) { + e, err := g.d.State.Get(callKey(id)) + if err != nil || e == nil { + return + } + var w WaitingCall + if json.Unmarshal(e.Value, &w) != nil || w.End == nil || w.Said { + return + } + if g.d.Say == nil { + g.d.Logf("[approval] the end %s of call %s is not said: the holder says no gated-ended", w.End.Outcome, id) + return + } + if err := g.d.Say(GatedEndedSubject(w.Seat, w.Caller, w.ID), *w.End, GatedEndedEvent+"."+w.ID); err != nil { + g.d.Logf("[approval] the end %s of call %s is not said (%v): said again at the next start", w.End.Outcome, id, err) + return + } + g.d.Logf("[approval] %s by %s ended %s (call %s, ask %s)", w.Address, w.Caller, w.End.Outcome, id, w.Ask) + _ = g.updateCall(id, func(k *WaitingCall) { k.Said = true }) + if g.d.Now().Sub(w.Asked) >= AskLasts && g.d.Findings != nil { + _ = g.d.Findings.Clear(nilCtx(), KindGateUnanswered, id, "the call's end, "+w.End.Outcome+", is said") + } +} + +// updateCall changes a kept call by compare-and-set, again when another write came first. +func (g *Gate) updateCall(id string, change func(*WaitingCall)) error { + for { + e, err := g.d.State.Get(callKey(id)) + if err != nil || e == nil { + return fmt.Errorf("the call %s cannot be read: %v", id, err) + } + var w WaitingCall + if err := json.Unmarshal(e.Value, &w); err != nil { + return err + } + change(&w) + if _, err := g.d.State.Update(callKey(id), w, e.Revision); err == nil { + return nil + } else if !isChanged(err) { + return err + } + } +} + +// waiting is every kept call that has not ended and that keep takes, oldest first. +func (g *Gate) waiting(keep func(WaitingCall) bool) ([]WaitingCall, error) { + all, err := g.calls() + if err != nil { + return nil, err + } + var out []WaitingCall + for _, w := range all { + if w.End == nil && keep(w) { + out = append(out, w) + } + } + return out, nil +} + +// calls is every kept call, oldest first. +func (g *Gate) calls() ([]WaitingCall, error) { + names, err := g.d.State.Keys() + if err != nil { + return nil, err + } + var out []WaitingCall + for _, key := range names { + if !strings.HasPrefix(key, "call.") { + continue + } + e, err := g.d.State.Get(key) + if err != nil { + return nil, err + } + var w WaitingCall + if e == nil || json.Unmarshal(e.Value, &w) != nil { + continue + } + out = append(out, w) + } + sort.Slice(out, func(i, j int) bool { return out[i].Made.Before(out[j].Made) }) + return out, nil +} + +// armUnanswered raises the waiting call's gate-unanswered when its ask's 60 minutes pass with no end of it (§677). +func (g *Gate) armUnanswered(w WaitingCall) { + left := max(w.Asked.Add(AskLasts).Sub(g.d.Now()), 0) + at := g.d.After(left) + go func() { // its own goroutine whatever Go is: it waits out the ask's 60 minutes on the clock's pause + <-at + e, err := g.d.State.Get(callKey(w.ID)) + if err != nil || e == nil { + return + } + var now WaitingCall + if json.Unmarshal(e.Value, &now) != nil || now.End != nil || now.Ask != w.Ask || g.d.Findings == nil { + return + } + if err := g.d.Findings.Raise(nilCtx(), KindGateUnanswered, w.ID, fmt.Sprintf("the call %s of %s by %s waits on ask %s, "+ + "past its %s with no decided heard: nothing ran; the holder's %s.approvals reads it", w.ID, w.Address, w.Caller, w.Ask, + AskLasts, w.Seat), "a call waits on an unanswered ask"); err != nil { + g.d.Logf("[approval] the finding %s of call %s is not kept: %v", KindGateUnanswered, w.ID, err) + } + }() +} + +// StartHolding is the holder's read at its start (§677): every waiting call is read, each ask already decided acted on as +// when its decided is heard, each end not yet said said, each waiting call's 60 minutes armed, and every acted, unrecorded +// run recorded. A state that cannot be read leaves every marked call refused until it can. +func (g *Gate) StartHolding() error { + all, err := g.calls() + if err != nil { + g.stateFailed(err) + return fmt.Errorf("the waiting calls in %s cannot be read: %w", StateName, err) + } + g.stateRead() + asks := map[string]bool{} + for _, w := range all { + switch { + case w.End != nil && !w.Said: + g.say(w.ID) + case w.End == nil: + g.armUnanswered(w) + asks[w.Ask] = true + } + } + ids := make([]string, 0, len(asks)) + for id := range asks { + ids = append(ids, id) + } + sort.Strings(ids) + for _, id := range ids { + g.Decided(id) + } + g.Recover() + return nil +} + +// Approvals is the seat's verb `approvals` (§677): the waiting and ended calls with their caller, ask and state, the +// `act.` keys and the claims; a call's arguments only to that call's caller. +type Approvals struct { + Calls []ListedCall `json:"calls"` + Acts []ListedAct `json:"acts"` + Claims []string `json:"claims"` +} + +// ListedCall is one kept call as `approvals` lists it. +type ListedCall struct { + Call string `json:"call"` + Caller string `json:"caller"` + Ask string `json:"ask"` + Verb string `json:"verb"` + State string `json:"state"` + Made time.Time `json:"made"` + End *GatedEnded `json:"end,omitempty"` + Arguments json.RawMessage `json:"arguments,omitempty"` +} + +// ListedAct is one `act.` key: its digest, escaped as kept, and the ask it names. +type ListedAct struct { + Act string `json:"act"` + Ask string `json:"ask"` +} + +// Approvals answers the verb for caller, the account the bus proved. +func (g *Gate) Approvals(caller string) (Approvals, error) { + out := Approvals{Calls: []ListedCall{}, Acts: []ListedAct{}, Claims: []string{}} + if g.d.State == nil { + return out, fmt.Errorf("the holder keeps no state %s", StateName) + } + all, err := g.calls() + if err != nil { + return out, fmt.Errorf("the state %s cannot be read: %w", StateName, err) + } + for _, w := range all { + l := ListedCall{Call: w.ID, Caller: w.Caller, Ask: w.Ask, Verb: w.Verb, State: "waiting", Made: w.Made, End: w.End} + if w.End != nil { + l.State = w.End.Outcome + } + if caller != "" && caller == w.Caller { + l.Arguments = w.Arguments + } + out.Calls = append(out.Calls, l) + } + names, err := g.d.State.Keys() + if err != nil { + return out, fmt.Errorf("the state %s cannot be read: %w", StateName, err) + } + sort.Strings(names) + for _, key := range names { + switch { + case strings.HasPrefix(key, "act."): + e, err := g.d.State.Get(key) + var id string + if err == nil && e != nil { + _ = json.Unmarshal(e.Value, &id) + } + out.Acts = append(out.Acts, ListedAct{Act: strings.TrimPrefix(key, "act."), Ask: id}) + case strings.HasPrefix(key, "claimed."): + out.Claims = append(out.Claims, strings.TrimPrefix(key, "claimed.")) + } + } + return out, nil +} + +func nilCtx() context.Context { return context.Background() } diff --git a/go/approval/held_test.go b/go/approval/held_test.go new file mode 100644 index 00000000..81cd9c81 --- /dev/null +++ b/go/approval/held_test.go @@ -0,0 +1,333 @@ +package approval + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/operatorask" +) + +// memStore is the holder's state approvals in memory: create refuses a key that exists, update is compare-and-set. +type memStore struct { + mu sync.Mutex + vals map[string][]byte + revs map[string]uint64 + rev uint64 + fail error +} + +func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} } + +func (m *memStore) Get(key string) (*stdio.StateEntry, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return nil, m.fail + } + v, ok := m.vals[key] + if !ok { + return nil, nil + } + return &stdio.StateEntry{Key: key, Value: v, Revision: m.revs[key]}, nil +} + +func (m *memStore) write(key string, value any, check func(exists bool, rev uint64) bool) (uint64, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return 0, m.fail + } + _, exists := m.vals[key] + if !check(exists, m.revs[key]) { + return 0, stdio.ErrChanged + } + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + m.rev++ + m.vals[key], m.revs[key] = raw, m.rev + return m.rev, nil +} + +func (m *memStore) Create(key string, value any) (uint64, error) { + return m.write(key, value, func(exists bool, _ uint64) bool { return !exists }) +} +func (m *memStore) Update(key string, value any, revision uint64) (uint64, error) { + return m.write(key, value, func(exists bool, rev uint64) bool { return exists && rev == revision }) +} +func (m *memStore) Put(key string, value any) (uint64, error) { + return m.write(key, value, func(bool, uint64) bool { return true }) +} +func (m *memStore) Keys() ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + if m.fail != nil { + return nil, m.fail + } + var out []string + for k := range m.vals { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +// router is the operator channel as a fake: it keeps each ask the gate raises, and the test decides its record. +type router struct { + mu sync.Mutex + state Store + asks []Ask + records map[string]*Record +} + +func (r *router) Ask(module string, a Ask) error { + raw, _ := json.Marshal(a) + if _, err := r.state.Create("ask."+a.ID, kept{Ask: raw, Asked: time.Now()}); err != nil { + return err + } + r.mu.Lock() + defer r.mu.Unlock() + r.asks = append(r.asks, a) + return nil +} +func (r *router) Record(module, id string) (*Record, error) { + r.mu.Lock() + defer r.mu.Unlock() + return r.records[id], nil +} +func (r *router) Claim(module string, w Warrant, act operatorask.Act, run func()) (Claim, error) { + e, _ := r.state.Get("ask." + w.Ask) + if e == nil { + return ClaimRefused, fmt.Errorf("no ask %s", w.Ask) + } + var k kept + _ = json.Unmarshal(e.Value, &k) + if k.Acted { + return ClaimDone, nil + } + var a Ask + _ = json.Unmarshal(k.Ask, &a) + digest, _ := operatorask.ActDigest(act) + if a.About != digest { + return ClaimRefused, fmt.Errorf("the option binds another act") + } + k.Acted = true + if _, err := r.state.Update("ask."+w.Ask, k, e.Revision); err != nil { + return ClaimDone, nil + } + run() + return Claimed, nil +} +func (r *router) Cancel(module, id string) error { return nil } +func (r *router) Decided(module string) string { return "decided." + module } +func (r *router) Seat() string { return "operator-channel" } + +func (r *router) choose(id, option string, at time.Time) { + r.mu.Lock() + defer r.mu.Unlock() + r.records[id] = &Record{State: "chosen", OfferedOn: []string{"telegram"}, + Warrant: &Warrant{Ask: id, Asker: "holder", Option: option, Channel: "telegram", At: formatTime(at)}} +} + +type findings struct { + mu sync.Mutex + raised map[string]bool + heard chan string +} + +func (f *findings) Raise(_ context.Context, kind, id, _, _ string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.raised[kind+" "+id] = true + if f.heard != nil { + f.heard <- kind + " " + id + } + return nil +} +func (f *findings) Clear(_ context.Context, kind, id, _ string) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.raised, kind+" "+id) + return nil +} + +// Design 47 G §677 (lead answers A-158, A-209, A-244): a process holder's marked verb is one call with a gate before its +// running, its waiting calls, act keys and claims kept in its own state. +func TestAHolderHoldsAMarkedCall(t *testing.T) { + now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) + type world struct { + g *Gate + state *memStore + router *router + said map[string]GatedEnded + ran *int + recs *int + finds *findings + after chan time.Time + } + newGate := func(state *memStore, r *router, ran, recs *int, said map[string]GatedEnded, after chan time.Time, f *findings) *Gate { + g := New(Deps{Node: "n1", Runner: "holder", Holding: true, Build: "fp1", State: state, Channel: r, Findings: f, + Now: func() time.Time { return now }, Go: func(f func()) { f() }, + After: func(d time.Duration) <-chan time.Time { + if d < time.Minute { // the record's short pauses pass at once + fired := make(chan time.Time) + close(fired) + return fired + } + return after + }, + Warranted: func(asker, ask string, open func() (bool, error)) error { + *recs++ + return nil + }, + Say: func(subject string, body any, id string) error { + said[subject] = body.(GatedEnded) + return nil + }}) + g.Serves("wiper", "wipe", func(caller string, args json.RawMessage) (any, error) { *ran++; return "wiped", nil }) + return g + } + setup := func() world { + state := newMem() + r := &router{state: state, records: map[string]*Record{}} + ran, recs := 0, 0 + w := world{state: state, router: r, said: map[string]GatedEnded{}, ran: &ran, recs: &recs, + finds: &findings{raised: map[string]bool{}}, after: make(chan time.Time)} + w.g = newGate(state, r, w.ran, w.recs, w.said, w.after, w.finds) + return w + } + call := func(caller, args string) Call { + return Call{Address: "wiper.wipe", Module: "holder", Seat: "wiper", Name: "wipe", Caller: caller, Args: json.RawMessage(args)} + } + hold := func(t *testing.T, w world, c Call, wait bool) (any, *Held, error) { + return w.g.Hold(c, wait, func() (any, error) { *w.ran++; return "wiped", nil }) + } + ended := func(w world, caller, id string) GatedEnded { return w.said[GatedEndedSubject("wiper", caller, id)] } + + t.Run("asked, run once on the Approve, its end said to its caller", func(t *testing.T) { + w := setup() + _, h, err := hold(t, w, call("n1.caller", `{"dir":"scratch"}`), true) + if err != nil || h == nil || h.Asked == "" || h.Call == "" { + t.Fatalf("held %+v, %v", h, err) + } + if e, _ := w.state.Get("call." + h.Call); e == nil { + t.Fatal("no call kept") + } + w.router.choose(h.Asked, OptionApprove, now) + w.g.Decided(h.Asked) + w.g.Decided(h.Asked) // heard twice: run once + end := ended(w, "n1.caller", h.Call) + if *w.ran != 1 || end.Outcome != OutcomeRan || end.Answer != "wiped" || !strings.Contains(end.Approval, h.Asked) || *w.recs != 1 { + t.Fatalf("ran %d, recorded %d, end %+v", *w.ran, *w.recs, end) + } + if e, _ := w.state.Get("claimed." + h.Asked); e == nil { + t.Error("no claim kept") + } + a, _ := w.g.Approvals("someone.else") + if len(a.Calls) != 1 || a.Calls[0].Arguments != nil || a.Calls[0].State != OutcomeRan || len(a.Acts) != 1 || len(a.Claims) != 1 { + t.Errorf("approvals %+v", a) + } + if a, _ := w.g.Approvals("n1.caller"); string(a.Calls[0].Arguments) != `{"dir":"scratch"}` { + t.Errorf("its caller reads %s", a.Calls[0].Arguments) + } + }) + t.Run("a second call of the act takes the same ask, asked anew after the first runs", func(t *testing.T) { + w := setup() + _, h1, _ := hold(t, w, call("n1.a", `{}`), true) + now = now.Add(time.Second) + _, h2, _ := hold(t, w, call("n1.b", `{}`), true) + if h1.Asked != h2.Asked || h1.Call == h2.Call { + t.Fatalf("asks %s and %s", h1.Asked, h2.Asked) + } + w.router.choose(h1.Asked, OptionApprove, now) + w.g.Decided(h1.Asked) + if *w.ran != 1 || ended(w, "n1.a", h1.Call).Outcome != OutcomeRan || ended(w, "n1.b", h2.Call).Outcome != "" { + t.Fatalf("ran %d; ends %v", *w.ran, w.said) + } + e, _ := w.state.Get("call." + h2.Call) + var second WaitingCall + _ = json.Unmarshal(e.Value, &second) + if second.Ask == h1.Asked || len(w.router.asks) != 2 { + t.Errorf("the second call waits on %s (asks %d)", second.Ask, len(w.router.asks)) + } + }) + t.Run("a decline runs nothing; a call within 10 minutes ends declined unasked", func(t *testing.T) { + w := setup() + _, h, _ := hold(t, w, call("n1.a", `{}`), true) + w.router.choose(h.Asked, OptionDecline, now) + w.g.Decided(h.Asked) + if *w.ran != 0 || ended(w, "n1.a", h.Call).Outcome != OutcomeDeclined { + t.Fatalf("ran %d, end %+v", *w.ran, ended(w, "n1.a", h.Call)) + } + if _, h2, err := hold(t, w, call("n1.a", `{}`), true); err == nil || h2 != nil || !strings.Contains(err.Error(), "declined") || len(w.router.asks) != 1 { + t.Errorf("a call within 10 minutes: %v, %d asks", err, len(w.router.asks)) + } + }) + t.Run("heard 31 minutes after the choice, it ends expired", func(t *testing.T) { + w := setup() + _, h, _ := hold(t, w, call("n1.a", `{}`), true) + w.router.choose(h.Asked, OptionApprove, now.Add(-31*time.Minute)) + w.g.Decided(h.Asked) + if *w.ran != 0 || ended(w, "n1.a", h.Call).Outcome != OutcomeExpired { + t.Fatalf("ran %d, end %+v", *w.ran, ended(w, "n1.a", h.Call)) + } + }) + t.Run("a secret-shaped argument refused, nothing kept", func(t *testing.T) { + w := setup() + if _, h, err := hold(t, w, call("n1.a", `{"token":"ghp_0123456789abcdefghijklmnopqrstuvwxyzAB"}`), true); err == nil || h != nil { + t.Fatalf("held %+v, %v", h, err) + } + if keys, _ := w.state.Keys(); len(keys) != 0 || len(w.router.asks) != 0 { + t.Errorf("kept %v, asked %d", keys, len(w.router.asks)) + } + }) + t.Run("a call with no wait refused with the ask, and run when made again after the Approve", func(t *testing.T) { + w := setup() + _, h, err := hold(t, w, call("n1.a", `{}`), false) + if err == nil || h != nil || len(w.router.asks) != 1 || !strings.Contains(err.Error(), w.router.asks[0].ID) { + t.Fatalf("first: %v", err) + } + w.router.choose(w.router.asks[0].ID, OptionApprove, now) + answer, _, err := hold(t, w, call("n1.a", `{}`), false) + if a, ok := answer.(Answered); err != nil || !ok || a.Answer != "wiped" || *w.ran != 1 { + t.Fatalf("again: %+v, %v", answer, err) + } + }) + t.Run("a restart between the ask and the Approve runs the waiting call once", func(t *testing.T) { + w := setup() + _, h, _ := hold(t, w, call("n1.a", `{}`), true) + w.router.choose(h.Asked, OptionApprove, now) + again := newGate(w.state, w.router, w.ran, w.recs, w.said, w.after, w.finds) + if err := again.StartHolding(); err != nil || *w.ran != 1 || ended(w, "n1.a", h.Call).Outcome != OutcomeRan { + t.Fatalf("start: %v, ran %d", err, *w.ran) + } + _ = again.StartHolding() + if *w.ran != 1 { + t.Errorf("ran %d after a second start", *w.ran) + } + }) + t.Run("past its ask's 60 minutes with no decided, gate-unanswered", func(t *testing.T) { + w := setup() + w.finds.heard = make(chan string, 1) + _, h, _ := hold(t, w, call("n1.a", `{}`), true) + w.after <- now + if got := <-w.finds.heard; got != KindGateUnanswered+" "+h.Call { + t.Errorf("raised %s", got) + } + }) + t.Run("a state that cannot be read refuses every marked call", func(t *testing.T) { + w := setup() + w.state.fail = fmt.Errorf("the bus is away") + if _, h, err := hold(t, w, call("n1.a", `{}`), true); err == nil || h != nil || *w.ran != 0 { + t.Fatalf("held %+v, %v", h, err) + } + }) +} diff --git a/go/approval/record.go b/go/approval/record.go index 22c1af32..97dc8a44 100644 --- a/go/approval/record.go +++ b/go/approval/record.go @@ -171,7 +171,7 @@ func (g *Gate) Recover() { 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." + prefix := g.prefix() + "ask." for _, key := range keys { if !strings.HasPrefix(key, prefix) { continue diff --git a/go/process/call.go b/go/process/call.go index 791af6e1..b1ad9e9e 100644 --- a/go/process/call.go +++ b/go/process/call.go @@ -10,6 +10,7 @@ import ( "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/calladdress" ) @@ -188,7 +189,13 @@ func callSubjectsOn(key, node, token string) (called, fallback string, err error func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { ctx, cancel := context.WithTimeout(context.Background(), CallTimeout) defer cancel() - msg, err := c.nc.RequestWithContext(ctx, subject, data) + req := nats.NewMsg(subject) + req.Data = data + if strings.HasPrefix(subject, "mesh.seat.") { + // Every seat call says it waits for a held call's end; a verb that is not marked ignores it (lead answer A-244). + req.Header.Set(approval.WaitHeader, approval.WaitHeld) + } + msg, err := c.nc.RequestMsgWithContext(ctx, req) if err != nil { return nil, err } @@ -196,6 +203,7 @@ func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { Result json.RawMessage `json:"result"` Error string `json:"error"` Node string `json:"node"` + Held *approval.Held `json:"held"` } if err := json.Unmarshal(msg.Data, &r); err != nil { return nil, fmt.Errorf("%s answered something that is not a seat's answer: %w", subject, err) @@ -203,5 +211,8 @@ func (c *Conn) request(subject string, data []byte) (json.RawMessage, error) { if r.Error != "" { return nil, answered(r.Error) } + if r.Held != nil { + return c.awaitGated(subject, *r.Held) + } return r.Result, nil } diff --git a/go/process/call_test.go b/go/process/call_test.go index 3ea2413a..188df54d 100644 --- a/go/process/call_test.go +++ b/go/process/call_test.go @@ -175,10 +175,13 @@ func TestUserOfTokenAndAnswerSeatRefusals(t *testing.T) { {"a verb open to none", "novox.mesh-recovery", func() (seatdef.Definition, bool) { return seatdef.Definition{Seat: "seat", Verbs: []seatdef.Verb{{Name: "v", Callers: &seatdef.Callers{List: []string{}}}}}, true }, "open to no caller"}, + {"a marked verb with no gate (§677)", "novox.mesh-recovery", func() (seatdef.Definition, bool) { + return seatdef.Definition{Seat: "seat", Verbs: []seatdef.Verb{{Name: "v", NeedsApproval: true}}}, true + }, "through no gate"}, {"a proven caller", "novox.mesh-recovery", defined, ""}, } { before := ran - _, err := admitted("seat", "v", row.def, run)(row.caller, nil) + _, _, err := c.admitted("seat", "v", row.def, run, nil)(row.caller, nil, false) if row.says == "" && (err != nil || ran != before+1) || row.says != "" && (err == nil || !strings.Contains(err.Error(), row.says) || ran != before) { t.Errorf("%s: %v, ran %d times", row.name, err, ran-before) } diff --git a/go/process/gate.go b/go/process/gate.go new file mode 100644 index 00000000..f4c372fb --- /dev/null +++ b/go/process/gate.go @@ -0,0 +1,129 @@ +package process + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/approval" + "git.novox.be/novox/mesh-sdk/go/busretry" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// Gating is what a holder's gate reaches beyond this connection (design 47 G §677; lead answers A-219, A-224, A-228): the +// operator channel (mesh-contracts' operatorchannelclient.GateChannel over this connection, reading the router's record +// under the module's own name, §700) and the holder's held finding writer. +type Gating struct { + Channel approval.Channel + Findings approval.FindingWriter + // Build is the holder's own fingerprint; empty reads it from the process's environment (OwnBuild). + Build string + Logf func(string, ...any) +} + +// Gate is the holder's gate over this connection (§677): its state `approvals`, its calls of the seat `mesh-hand-acts`'s +// `warranted` to record a run, and its seat's `gated-ended`, each on the process's own account. AnswerSeat holds the +// seat's marked verbs through it (Gated); the process hears its decided with HearDecided and reads it at its start with +// the gate's StartHolding. +func (c *Conn) Gate(g Gating) *approval.Gate { + build := g.Build + if build == "" { + build = OwnBuild().Fingerprint + } + return approval.New(approval.Deps{Node: c.account.Node, Runner: c.account.Module, Holding: true, Build: build, + State: c.State(approval.StateName), Channel: g.Channel, Findings: g.Findings, Logf: g.Logf, + Warranted: func(asker, ask string, open func() (bool, error)) error { + _, err := c.CallKeptWhileOpen(busretry.Open{Key: ask, Open: func(string) (bool, error) { return open() }, Logf: g.Logf}, + "seat:mesh-hand-acts.warranted", map[string]string{"asker": asker, "ask": ask}) + return err + }, + Say: func(subject string, body any, id string) error { + return c.emitOn(subject, body, []stdio.EmitOptions{{MessageID: id}}) + }}) +} + +// HearDecided hands the gate each `decided` of the holder's own asks that hub hears (§677): only the ask's id is read +// from it, the gate reading the router's own record. +func HearDecided(g *approval.Gate, ch approval.Channel, module string, hub *EventHub) error { + return hub.Handle(ch.Decided(module), func(m jetstream.Msg) { + var w struct { + Ask string `json:"ask"` + } + if err := json.Unmarshal(m.Data(), &w); err != nil || w.Ask == "" { + _ = m.Term() + return + } + g.Decided(w.Ask) + _ = m.Ack() + }) +} + +// HeldWithin bounds a caller's wait on a held call's end: the ask's 60 minutes and the 30 a run may come after its choice. +var HeldWithin = approval.AskLasts + approval.RunWithin + +// awaitGated waits on the held call's own end on its seat's gated-ended, from the last said on its subject (so an end said +// before the wait began is read), and is its answer: `{answer, approval}` when it ran (lead answer A-158), its words as +// the call's error otherwise. Every failure is an answer, never asked again: the call reached its holder (ADR 0668 §2). +func (c *Conn) awaitGated(subject string, held approval.Held) (json.RawMessage, error) { + seat, ok := seatOfSubject(subject) + if !ok { + return nil, answered(fmt.Sprintf("%s answered held, and is no seat's subject", subject)) + } + js, err := jetstream.New(c.nc) + if err != nil { + return nil, answered(err.Error()) + } + on := approval.GatedEndedSubject(seat, c.account.User, held.Call) + stream := seatdef.KeptStream(seat, seatdef.Message{Name: approval.GatedEndedEvent}) + ctx, cancel := context.WithTimeout(context.Background(), HeldWithin) + defer cancel() + cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{on}, + DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}) + if err != nil { + return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and its end on %s cannot be heard: %v; %s.approvals reads it", + held.Call, held.Asked, stream, err, seat)) + } + for { + m, err := cons.Next(jetstream.FetchMaxWait(time.Minute)) + if err == nil { + _ = m.Ack() + return endOf(m.Data()) + } + if ctx.Err() != nil || !errors.Is(err, jetstream.ErrNoMessages) && !errors.Is(err, context.DeadlineExceeded) { + return nil, answered(fmt.Sprintf("the call %s waits on ask %s, and no end of it was heard (%v); %s.approvals reads it", + held.Call, held.Asked, err, seat)) + } + } +} + +// endOf is a gated-ended's body as the call's answer. +func endOf(data []byte) (json.RawMessage, error) { + var e approval.GatedEnded + if err := json.Unmarshal(data, &e); err != nil { + return nil, answered("the end of the call is not read: " + err.Error()) + } + if e.Outcome != approval.OutcomeRan || e.Error != "" { + words := e.Error + if words == "" { + words = "the call ended " + e.Outcome + } + return nil, answered(fmt.Sprintf("Not run: %s (ask %s, call %s)", words, e.Ask, e.Call)) + } + return json.Marshal(approval.Answered{Answer: e.Answer, Approval: e.Approval}) +} + +// seatOfSubject is the seat a call subject `mesh.seat..…` names. +func seatOfSubject(subject string) (string, bool) { + rest, ok := strings.CutPrefix(subject, "mesh.seat.") + if !ok { + return "", false + } + seat, _, ok := strings.Cut(rest, ".") + return seat, ok && seat != "" +} diff --git a/go/process/serve.go b/go/process/serve.go index 6fb217db..03be2886 100644 --- a/go/process/serve.go +++ b/go/process/serve.go @@ -9,6 +9,7 @@ import ( "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/calladdress" "git.novox.be/novox/mesh-sdk/go/seatdef" ) @@ -42,13 +43,31 @@ func userOfToken(token string) string { // while it is not read. An unproven caller is refused, and so is every caller while the definition is unread, names no // such verb or opens it to none; a proven caller is among them, since the grants part composes a verb's call subject // only for its effective callers. -func (c *Conn) AnswerSeat(seat string, definition func() (seatdef.Definition, bool), verbs map[string]SeatHandler) (stop func(), err error) { +// +// A verb the definition marks `needs-approval` is held through the gate Gated gives (design 47 G §677; lead answers A-224, +// A-244): a call that carries the header approval.WaitHeader is kept and answered at once with the reply's field `held`, +// its end said on the seat's `gated-ended`; one that does not takes ADR 0286 §37's path. With no gate a marked verb is +// refused to every caller. With a gate the seat's verb `approvals` is served too, unless verbs answers it. +func (c *Conn) AnswerSeat(seat string, definition func() (seatdef.Definition, bool), verbs map[string]SeatHandler, opts ...SeatOption) (stop func(), err error) { + var o seatOptions + for _, f := range opts { + f(&o) + } if definition == nil { return nil, fmt.Errorf("AnswerSeat(%q) is given no definition of the seat to read its verbs' callers from", seat) } if len(verbs) == 0 { return nil, fmt.Errorf("AnswerSeat(%q) answers no verb", seat) } + if o.gate != nil && verbs["approvals"] == nil { + all := make(map[string]SeatHandler, len(verbs)+1) + for v, h := range verbs { + all[v] = h + } + gate := o.gate + all["approvals"] = func(caller string, _ json.RawMessage) (any, error) { return gate.Approvals(caller) } + verbs = all + } names := make([]string, 0, len(verbs)) for verb, h := range verbs { if h == nil { @@ -68,7 +87,11 @@ func (c *Conn) AnswerSeat(seat string, definition func() (seatdef.Definition, bo } } for _, verb := range names { - h := admitted(seat, verb, definition, verbs[verb]) + if o.gate != nil { + h := verbs[verb] + o.gate.Serves(seat, verb, func(caller string, args json.RawMessage) (any, error) { return h(caller, args) }) + } + h := c.admitted(seat, verb, definition, verbs[verb], o.gate) base := "mesh.seat." + seat for _, subject := range []string{base + ".call." + verb + ".*", base + ".tool." + verb} { subject := subject @@ -87,16 +110,28 @@ func (c *Conn) AnswerSeat(seat string, definition func() (seatdef.Definition, bo return undo, nil } -// admitted is h behind the check of its caller against the verb's effective callers (A-179, A-186). -func admitted(seat, verb string, definition func() (seatdef.Definition, bool), h SeatHandler) SeatHandler { - return func(caller string, body json.RawMessage) (any, error) { +// SeatOption is an option of AnswerSeat. +type SeatOption func(*seatOptions) + +type seatOptions struct{ gate *approval.Gate } + +// Gated holds the seat's marked verbs through g, the holder's gate (design 47 G §677): Conn.Gate makes it. +func Gated(g *approval.Gate) SeatOption { return func(o *seatOptions) { o.gate = g } } + +// heldHandler is a verb's handler behind its checks: its answer, the held answer of a kept call, or its error. +type heldHandler func(caller string, body json.RawMessage, wait bool) (any, *approval.Held, error) + +// admitted is h behind the check of its caller against the verb's effective callers (A-179, A-186), and, for a verb the +// definition marks, behind the gate (§677). +func (c *Conn) admitted(seat, verb string, definition func() (seatdef.Definition, bool), h SeatHandler, gate *approval.Gate) heldHandler { + return func(caller string, body json.RawMessage, wait bool) (any, *approval.Held, error) { if caller == "" { - return nil, fmt.Errorf("%s.%s takes no unproven caller: call it on its subject that names its caller, "+ + return nil, nil, fmt.Errorf("%s.%s takes no unproven caller: call it on its subject that names its caller, "+ "mesh.seat.%s.call.%s.", seat, verb, seat, verb) } def, read := definition() if !read { - return nil, fmt.Errorf("the definition of the seat %s is not yet read by this process's seat list, so no caller of %s.%s "+ + return nil, nil, fmt.Errorf("the definition of the seat %s is not yet read by this process's seat list, so no caller of %s.%s "+ "is admitted", seat, seat, verb) } for _, v := range def.Verbs { @@ -104,16 +139,24 @@ func admitted(seat, verb string, definition func() (seatdef.Definition, bool), h continue } if len(v.EffectiveCallers()) == 0 { - return nil, fmt.Errorf("%s.%s is open to no caller", seat, verb) + return nil, nil, fmt.Errorf("%s.%s is open to no caller", seat, verb) } - return h(caller, body) + if !v.NeedsApproval { + answer, err := h(caller, body) + return answer, nil, err + } + if gate == nil { + return nil, nil, fmt.Errorf("Not run: %s.%s needs the operator's approval, and its holder serves it through no gate", seat, verb) + } + return gate.Hold(approval.Call{Address: seat + "." + verb, Module: c.account.Module, Seat: seat, Name: verb, + Caller: caller, Args: body, Description: v.Description}, wait, func() (any, error) { return h(caller, body) }) } - return nil, fmt.Errorf("the seat %s's definition names no verb %s, so no caller is admitted", seat, verb) + return nil, nil, fmt.Errorf("the seat %s's definition names no verb %s, so no caller is admitted", seat, verb) } } // answerCall runs h for one call and replies {result, error, node}; a panic in h is the call's error, never silence. -func (c *Conn) answerCall(msg *nats.Msg, pattern string, h SeatHandler) { +func (c *Conn) answerCall(msg *nats.Msg, pattern string, h heldHandler) { caller := "" if strings.HasSuffix(pattern, ".*") { caller = userOfToken(msg.Subject[strings.LastIndex(msg.Subject, ".")+1:]) @@ -122,17 +165,22 @@ func (c *Conn) answerCall(msg *nats.Msg, pattern string, h SeatHandler) { Result any `json:"result,omitempty"` Error string `json:"error,omitempty"` Node string `json:"node,omitempty"` + // Held is a kept call's answer (design 47 G §677; lead answer A-244). + Held *approval.Held `json:"held,omitempty"` } + wait := msg.Header != nil && msg.Header.Get(approval.WaitHeader) == approval.WaitHeld func() { defer func() { if r := recover(); r != nil { reply.Error = fmt.Sprintf("%s failed: %v", msg.Subject, r) } }() - result, err := h(caller, json.RawMessage(msg.Data)) + result, held, err := h(caller, json.RawMessage(msg.Data), wait) switch { case err != nil: reply.Error = err.Error() + case held != nil: + reply.Held = held case result == nil: reply.Result = json.RawMessage("null") default: