novox/mesh-controller
13
0
Fork 0
Code Issues Pull Requests 3 Releases Wiki Activity

Ask at most three at a time, wait out a refusal, need a router, and act only on a claimed open ask, as the review asked (hq ADR 0259)

Browse Source
This commit is contained in:
jochen
2026-10-09 16:22:34 +02:00
committed by jschoubben
parent 8de4dc7951
commit 744b0b9162
10 changed files with 384 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/asker.go
+108 -17
View File
@@ -43,7 +43,9 @@ const askerName = broker.ControllerSeat
// How long an ask lasts: a day when an answer approves, a week when every answer only acknowledges.
const (
askApproveFor = 24 * time.Hour
// askApproveFor is a day less a margin, so an ask is never refused at the router for lasting a day and
// a moment (the SDK's bound is a day).
askApproveFor = 24*time.Hour - 10*time.Minute
askAcknowledgeFor = 7 * 24 * time.Hour
// askEvery is how often what is open is asked about again, beside every change.
askEvery = time.Minute
@@ -53,6 +55,9 @@ const (
// askAgainAfterAnswer is how long a condition the operator answered is not asked about again with the
// same answers: what was chosen takes a while to clear it, and asking again at once would ask twice.
askAgainAfterAnswer = time.Hour
// askMostOpen is how many asks the controller holds open at once (the router refuses a fourth): the
// most urgent conditions first, then the oldest.
askMostOpen = asks.MostOpen
)
// What became of an ask, as the controller keeps it.
@@ -63,10 +68,13 @@ const (
// asked is one ask the controller made, as it keeps it.
type asked struct {
ID string `json:"id"`
Condition string `json:"condition"`
Ask asks.Ask `json:"ask"`
Actions []conditions.Action `json:"actions"`
ID string `json:"id"`
Condition string `json:"condition"`
// Channels is what the channels were when it was asked (asker.channels): an ask the router refused is not
// asked again until the condition's answers or the channels change.
Channels string `json:"channels,omitempty"`
Ask asks.Ask `json:"ask"`
Actions []conditions.Action `json:"actions"`
// Options are the actions by option id.
Options map[string]int `json:"options"`
State string `json:"state"`
@@ -83,6 +91,9 @@ type askedStore interface {
Get(ctx context.Context, id string) (*asked, error)
Put(ctx context.Context, a asked) error
All(ctx context.Context) ([]asked, error)
// Claim marks an open ask acting, by compare-and-set, and says whether this write stood: of two
// deliveries of one warrant, or two controllers, only the one whose write stands acts.
Claim(ctx context.Context, id string, w asks.Warrant) (bool, error)
}
// asker is the controller asking the operator and acting on the answer.
@@ -98,8 +109,14 @@ type asker struct {
record func(ctx context.Context, act link.HandAct) error
// routerRecord reads the router's record of an ask for a warrant missed; nil reads nothing.
routerRecord func(ctx context.Context, id string) (*asks.Warrant, error)
now func() time.Time
logf func(string, ...any)
// routerHere says whether a router holds the seat and takes asks under the asker's name; nil is yes.
routerHere func(ctx context.Context) (bool, error)
// channels is what the channels are now, as a fingerprint: who holds which kind, promising what.
channels func(ctx context.Context) string
now func() time.Time
logf func(string, ...any)
saidNoRouter bool
mu sync.Mutex
nudged chan struct{}
@@ -156,6 +173,25 @@ func sameAsked(a []conditions.Action, b []conditions.Action) bool {
// reconcile brings what is asked in line with what is open.
func (a *asker) reconcile(ctx context.Context) error {
now := a.now()
if a.routerHere != nil {
here, err := a.routerHere(ctx)
if err != nil {
return err
}
if !here {
if !a.saidNoRouter {
a.logf("no router takes asks under the controller's name (a module declaring %s with its ask "+
"named by its caller, assigned): the operator is asked nothing until one is", broker.AsksSeat)
a.saidNoRouter = true
}
return nil
}
a.saidNoRouter = false
}
channels := ""
if a.channels != nil {
channels = a.channels(ctx)
}
open, err := a.open(ctx)
if err != nil {
return err
@@ -195,20 +231,47 @@ func (a *asker) reconcile(ctx context.Context) error {
}
}
}
// What the operator answered lately, by condition: not asked again at once.
answered := map[string]asked{}
// What the operator answered lately, by condition: not asked again at once; and what the router refused,
// newest first: not asked again until the answers or the channels change.
answered, refused := map[string]asked{}, map[string]asked{}
for _, r := range all {
if r.State == string(asks.OutcomeChosen) && now.Sub(r.Ended) < askAgainAfterAnswer {
answered[r.Condition] = r
}
if r.State == string(asks.OutcomeRefused) {
if prior, has := refused[r.Condition]; !has || r.Opened.After(prior.Opened) {
refused[r.Condition] = r
}
}
}
wanted := map[string]bool{}
sort.Slice(open, func(i, j int) bool { return open[i].Key < open[j].Key })
// The most urgent first, then the oldest: those are asked when no more than askMostOpen may be.
sort.SliceStable(open, func(i, j int) bool {
ui, uj := open[i].Severity == conditions.Urgent, open[j].Severity == conditions.Urgent
if ui != uj {
return ui
}
if !open[i].Raised.Equal(open[j].Raised) {
return open[i].Raised.Before(open[j].Raised)
}
return open[i].Key < open[j].Key
})
openNow := 0
for _, c := range open {
if r, held := byCondition[c.Key]; held && wants(c, now) && sameAsked(r.Actions, c.Actions) && now.Before(r.Ask.Expires) {
openNow++
}
}
for _, c := range open {
if !wants(c, now) {
continue
}
wanted[c.Key] = true
if r, was := refused[c.Key]; was && sameAsked(r.Actions, c.Actions) && r.Channels == channels {
if _, held := byCondition[c.Key]; !held {
continue // refused, and nothing it was refused for has changed
}
}
if r, done := answered[c.Key]; done && sameAsked(r.Actions, c.Actions) {
if _, held := byCondition[c.Key]; !held {
continue
@@ -226,13 +289,19 @@ func (a *asker) reconcile(ctx context.Context) error {
if err := a.store.Put(ctx, r); err != nil {
return err
}
openNow--
default:
continue
}
}
if err := a.ask(ctx, c); err != nil {
a.logf("the operator could not be asked about %s: %v", c.Key, err)
if openNow >= askMostOpen {
continue // asked when one of the open ones ends, most urgent first
}
if err := a.ask(ctx, c, channels); err != nil {
a.logf("the operator could not be asked about %s: %v", c.Key, err)
continue
}
openNow++
}
for key, r := range byCondition {
if !wanted[key] {
@@ -280,9 +349,18 @@ func doesWords(act conditions.Action) string {
return strings.ToLower(act.Label)
}
// askText is a condition's words as an ask says them: without where an answer is given when no channel can
// give it (FromMeshMCPServer), since the ask is answered on a channel and the router says where else.
func askText(s string) string {
for _, with := range []string{", " + FromMeshMCPServer, " " + FromMeshMCPServer} {
s = strings.ReplaceAll(s, with, ".")
}
return strings.ReplaceAll(s, "..", ".")
}
// askOf is the ask a condition is asked with.
func askOf(id string, c conditions.Condition, now time.Time) (asks.Ask, map[string]int) {
q := asks.Ask{ID: id, Headline: c.Headline, Explanation: c.Explanation, Who: asks.Operator,
q := asks.Ask{ID: id, Headline: c.Headline, Explanation: askText(c.Explanation), Who: asks.Operator,
OnExpiry: "nothing is done, and you are asked again while it lasts", About: c.Key,
Urgent: c.Severity == conditions.Urgent}
options := map[string]int{}
@@ -311,7 +389,7 @@ func newAskID() string {
}
// ask publishes one ask about a condition, and keeps it.
func (a *asker) ask(ctx context.Context, c conditions.Condition) error {
func (a *asker) ask(ctx context.Context, c conditions.Condition, channels string) error {
now := a.now()
id := newAskID()
q, options := askOf(id, c, now)
@@ -327,7 +405,7 @@ func (a *asker) ask(ctx context.Context, c conditions.Condition) error {
}
a.logf("asked the operator about %s (%s): %d answer(s)", c.Key, id, len(q.Options))
return a.store.Put(ctx, asked{ID: id, Condition: c.Key, Ask: q, Actions: c.Actions, Options: options,
State: askOpen, Opened: now})
State: askOpen, Opened: now, Channels: channels})
}
// cancel takes an ask back.
@@ -375,6 +453,11 @@ func (a *asker) Decided(ctx context.Context, body []byte) error {
a.logf("the ask %s about %s ended %s; nothing is done", r.ID, r.Condition, w.Outcome)
return a.store.Put(ctx, *r)
}
if r.State != askOpen {
// Cancelled, replaced or expired in the controller's own record: no answer to it is acted on.
a.logf("REFUSED a warrant for the ask %s, which is %s in the controller's own record", r.ID, r.State)
return nil
}
option, err := w.For(askerName, r.Ask)
if err != nil {
a.logf("REFUSED a warrant for the ask %s: %v", r.ID, err)
@@ -401,10 +484,18 @@ func (a *asker) Decided(ctx context.Context, body []byte) error {
a.logf("%s, for %s, which ended meanwhile: nothing is done", w.Says(), r.Condition)
return a.store.Put(ctx, *r)
}
r.Acted = "acting"
if err := a.store.Put(ctx, *r); err != nil {
// Claimed before acting, by compare-and-set: only the delivery whose write stands acts (security review
// of 2026-10-08, finding 9).
claimed, err := a.store.Claim(ctx, r.ID, w)
if err != nil {
return err
}
if !claimed {
a.logf("the warrant for the ask %s was already taken by another delivery; nothing more is done", r.ID)
return nil
}
r.Acted = "acting"
why := fmt.Sprintf("%s (ask %s)", w.Says(), r.ID)
args := map[string]string{}
for k, v := range act.Arguments {