A3f item 1: a process holder's marked seat verb is held through go/approval
conn.AnswerSeat holds a verb its seat definition marks needs-approval through the
holder's gate (process.Conn.Gate, go/approval held.go): waiting calls call.<id>,
act.<escaped digest> by create or compare-and-set, claimed.<ask id> by create; a
call carrying the x-wait: held header answered {asked, call} in the reply's field
held; on the router's record read under the module's name, the oldest waiting call
run once within 30 minutes of the Approve and recorded through
mesh-hand-acts.warranted, the others asked anew; declined unasked for 10 minutes;
expired and refused ends; every end said on gated-ended.<caller>.<call>; the read
at start; gate-unanswered past the ask's 60 minutes; the verb approvals. A call
without the header takes ADR 0286 §37's path. process.Conn's seat call sends the
header on every seat call and, answered held, waits on its own gated-ended.
Builds: design 47 G §677, §700, lead answers A-158, A-186, A-209, A-219, A-224, A-228, A-244
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
+21
-4
@@ -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, `<node>.`, 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,
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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 `<seat>.<verb>` (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
|
||||
}
|
||||
|
||||
@@ -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.<seat>.event.gated-ended.<caller>.<call>`, 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.<id>` (§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.<digest>` 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.<ask id>` 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() }
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+12
-1
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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.<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 != ""
|
||||
}
|
||||
+60
-12
@@ -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.<caller>", 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:
|
||||
|
||||
Reference in New Issue
Block a user