novox/mesh-controller
6
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

Merge pull request 'Ask again after a refusal, with a growing wait (issues 369 and 373)' (#198) from fix/373-one-message-per-condition into main

Browse Source
This commit was merged in pull request #198.
This commit is contained in:
mesh-admin
2026-10-10 12:45:07 +00:00
parent b9ceeace41 2e0a7d1cbd
commit d1cc9b4e20
2 changed files with 188 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/asker.go
+69 -7
View File
@@ -9,7 +9,8 @@ package main
// actions as options at their levels (Silence acknowledges; Release, Stop, Start and Restart approve),
// answered by the operator, expiring after a day (a week when every option only acknowledges). A
// condition that clears, is silenced, or changes its answers has its ask cancelled; an ask that expired
// unanswered is asked again while the condition lasts. Each ask is kept in the controller's bucket
// unanswered is asked again while the condition lasts; one the router refused is asked again after a wait
// that grows with each refusal in a row, at most half an hour, so a refusal is never believed longer. Each ask is kept in the controller's bucket
// `asked`, so a restart neither asks twice nor forgets.
// - **On a warrant**, heard on the seat's event under the controller's own name (which only the router may
// say), the controller acts once per ask: only for an ask it holds, only for the option it offered at
@@ -58,8 +59,29 @@ const (
// 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
// askRefusedRetryMost is the longest a refusal is believed without asking again (novox/hq issue 369): the
// router refused while its channels had not yet said they could send, the channels could send twenty
// minutes later, and the controller repeated that refusal for eleven hours. A refused ask is asked again
// after askEvery, then twice as long after each refusal in a row, never longer than this — and at once when
// the channels change.
askRefusedRetryMost = 30 * time.Minute
// askRefusedCounted is how far back refusals in a row are counted for that wait.
askRefusedCounted = 6 * time.Hour
// askVerdictWait is how long an ask made again after a refusal waits for the router's word before it is
// taken as taken: the router refuses an ask as it reads it, so a refusal comes within seconds. Meanwhile
// the condition saying it was not delivered stands as it was, neither cleared nor raised again.
askVerdictWait = 2 * time.Minute
)
// refusedRetryAfter is how long a part refused n times in a row waits before it is asked again.
func refusedRetryAfter(n int) time.Duration {
wait := askEvery
for i := 1; i < n && wait < askRefusedRetryMost; i++ {
wait *= 2
}
return min(wait, askRefusedRetryMost)
}
// What became of an ask, as the controller keeps it.
const (
askOpen = "open"
@@ -70,8 +92,9 @@ const (
type asked struct {
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 is what the channels were when it was asked (asker.channels): an ask the router refused is asked
// again at once when the condition's answers or the channels change, and otherwise after a wait that grows
// with each refusal in a row (refusedRetryAfter, novox/hq issue 369).
Channels string `json:"channels,omitempty"`
Ask asks.Ask `json:"ask"`
Actions []conditions.Action `json:"actions"`
@@ -346,6 +369,32 @@ func (a *asker) reconcile(ctx context.Context) error {
}
}
}
// Refusals in a row, by part: those since the last ask that was not refused, within askRefusedCounted.
// superseded: a part asked since its last refusal, by an ask the router did not refuse — open, answered,
// expired or cancelled. Its refusal is history then, never said again (the review of PR 198: an answered
// retry brought the refusal back).
inRow, superseded := map[string]int{}, map[string]asked{}
for k, last := range refused {
for _, r := range all {
if partKey(r.Condition, r.Part) == k && r.State != string(asks.OutcomeRefused) && r.State != askUnsent &&
r.Opened.After(last.Opened) && r.Opened.After(superseded[k].Opened) {
superseded[k] = r
}
}
var since time.Time
for _, r := range all {
if partKey(r.Condition, r.Part) == k && r.State != string(asks.OutcomeRefused) && r.State != askUnsent &&
!r.Opened.After(last.Opened) && r.Opened.After(since) {
since = r.Opened
}
}
for _, r := range all {
if partKey(r.Condition, r.Part) == k && r.State == string(asks.OutcomeRefused) && r.Opened.After(since) &&
now.Sub(r.Opened) < askRefusedCounted {
inRow[k]++
}
}
}
wanted := map[string]bool{}
var unasked []conditions.Condition // refused by the router, and nothing it was refused for changed
var refusedWords []string
@@ -379,15 +428,28 @@ func (a *asker) reconcile(ctx context.Context) error {
for _, p := range partsOf(c) {
key := partKey(c.Key, p.name)
wanted[key] = true
if r, was := refused[key]; was && sameAsked(r.Actions, p.actions) && r.Channels == channels {
if _, held := byCondition[key]; !held {
if r, was := refused[key]; was && sameAsked(r.Actions, p.actions) {
cur, held := byCondition[key]
ended := r.Ended
if ended.IsZero() {
ended = r.Opened
}
later, asked := superseded[key]
waiting := !held && !asked && r.Channels == channels && now.Sub(ended) < refusedRetryAfter(inRow[key])
// Asked again now (the wait over), or lately and the router's word not in yet: said as it was until
// that word, so the condition neither clears nor is raised again at each try.
retrying := !held && !asked && !waiting
verdictDue := held && asked && later.ID == cur.ID && now.Sub(cur.Opened) < askVerdictWait
if waiting || retrying || verdictDue {
if !saidUnasked {
unasked, saidUnasked = append(unasked, c), true
}
if r.Warrant != nil && r.Warrant.Words != "" {
refusedWords = append(refusedWords, r.Warrant.Words)
}
continue // refused, and nothing it was refused for has changed
}
if waiting {
continue // refused lately, and nothing it was refused for has changed: asked again after the wait
}
}
if r, done := answered[key]; done && sameAsked(r.Actions, p.actions) {
@@ -489,7 +551,7 @@ func (a *asker) sayUnasked(ctx context.Context, unasked []conditions.Condition,
Summary: fmt.Sprintf("%d condition(s) that need the operator could not be asked on any channel: %s; %s",
len(keys), strings.Join(keys, ", "), why),
Headline: "Questions for you not delivered",
Explanation: "Needs you: answer them from the mesh MCP server. The mesh could not send you its questions on any channel.",
Explanation: "The mesh could not send you its questions on any channel.",
Needs: "answer them from the mesh MCP server, and check why no channel carries them.",
Resolved: "The mesh can ask you again"})
}
cmd/mesh-controller/asker_test.go
+119 -14
View File
@@ -378,33 +378,50 @@ func TestAWarrantMissedWhileAwayIsReadFromTheRoutersRecord(t *testing.T) {
}
}
// After review (2026-10-08): a refused ask is not asked again until its answers or the channels change.
func TestAnAskTheRouterRefusedWaitsUntilSomethingChanges(t *testing.T) {
// After review (2026-10-08): a refused ask is not asked again at once; issue 369: nor is the refusal believed
// for ever — it is asked again after a wait that doubles with each refusal in a row, and at once when the
// channels change.
func TestAnAskTheRouterRefusedIsAskedAgainAfterAGrowingWait(t *testing.T) {
r := newAskerRig(t)
channels := "channel/telegram=telegram@anchor[choice]own:true"
r.a.channels = func(context.Context) string { return channels }
r.open = []conditions.Condition{heldCondition()}
refuse := func() {
t.Helper()
sent := r.asksSent(t)
refusal, _ := json.Marshal(asks.Warrant{Ask: sent[len(sent)-1].ID, Asker: "mesh-controller",
Outcome: asks.OutcomeRefused, Words: "no channel can carry any of its answers now", At: r.now})
if err := r.a.Decided(context.Background(), refusal); err != nil {
t.Fatal(err)
}
}
_ = r.a.reconcile(context.Background())
first := r.asksSent(t)[0]
refusal, _ := json.Marshal(asks.Warrant{Ask: first.ID, Asker: "mesh-controller", Outcome: asks.OutcomeRefused,
Words: "no channel can carry any of its answers now", At: r.now})
if err := r.a.Decided(context.Background(), refusal); err != nil {
t.Fatal(err)
}
refuse()
if got := r.store[first.ID]; got.State != string(asks.OutcomeRefused) || !strings.Contains(got.Acted, "nothing") {
t.Fatalf("the refusal was kept as %+v", got)
}
for i := 0; i < 3; i++ {
r.now = r.now.Add(askEvery)
// Each wait: nothing asked before it ends, asked once when it has.
for i, wait := range []time.Duration{time.Minute, 2 * time.Minute, 4 * time.Minute, 8 * time.Minute,
16 * time.Minute, 30 * time.Minute, 30 * time.Minute} {
before := len(r.asksSent(t))
r.now = r.now.Add(wait - 10*time.Second)
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != before {
t.Fatalf("refusal %d: asked again before %s", i+1, wait)
}
r.now = r.now.Add(10 * time.Second)
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != before+1 {
t.Fatalf("refusal %d: not asked again after %s (%d asks)", i+1, wait, n)
}
refuse()
}
if n := len(r.asksSent(t)); n != 1 {
t.Fatalf("asked again %d time(s) though nothing changed", n-1)
}
before := len(r.asksSent(t))
channels = "channel/telegram=telegram@anchor[choice,verified-sender]own:true"
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != 2 {
t.Errorf("not asked again once the channels changed: %d", n)
if n := len(r.asksSent(t)); n != before+1 {
t.Errorf("not asked again once the channels changed: %d", n-before)
}
}
@@ -703,3 +720,91 @@ func TestNothingIsAskedWhileTheBusLacksTheControllersGrant(t *testing.T) {
t.Errorf("the undelivered condition was not cleared: %+v", last)
}
}
// novox/hq issue 369: the router refused while its channels had not yet said they could send; they could twenty
// minutes later, and "Questions for you not delivered" repeated that refusal for eleven hours, escalating on it.
// The ask is made again; while the router's word on it is awaited the condition stands unchanged (neither
// cleared nor raised again); once the router took it, the condition clears; a new refusal is said in its own
// words.
func TestARefusalIsAskedAgainAndTheUndeliveredConditionClearsOnceTaken(t *testing.T) {
r := newAskerRig(t)
var raised [][]conditions.Observation
r.a.raise = func(_ context.Context, obs []conditions.Observation) error {
raised = append(raised, obs)
return nil
}
last := func() []conditions.Observation { return raised[len(raised)-1] }
r.a.channels = func(context.Context) string { return "channel/telegram=telegram@anchor[choice]own:true" }
r.open = []conditions.Condition{unitsCondition()}
refuse := func(words string) {
t.Helper()
sent := r.asksSent(t)
refusal, _ := json.Marshal(asks.Warrant{Ask: sent[len(sent)-1].ID, Asker: "mesh-controller",
Outcome: asks.OutcomeRefused, Words: words, At: r.now})
if err := r.a.Decided(context.Background(), refusal); err != nil {
t.Fatal(err)
}
}
_ = r.a.reconcile(context.Background())
refuse("no channel can carry any of its answers now: telegram: telegram has not said whether it can send")
_ = r.a.reconcile(context.Background())
if got := last(); len(got) != 1 || !strings.Contains(got[0].Summary, "telegram has not said") {
t.Fatalf("the refusal, said as %+v", got)
}
// The wait ends: asked again; until the router's word on it is in, the condition stands as it was.
r.now = r.now.Add(askEvery)
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != 2 {
t.Fatalf("not asked again: %d asks", n)
}
if got := last(); len(got) != 1 {
t.Fatalf("cleared while the router's word on the new ask was awaited: the condition would flap")
}
r.now = r.now.Add(askVerdictWait / 2)
_ = r.a.reconcile(context.Background())
if got := last(); len(got) != 1 {
t.Fatalf("cleared inside the verdict wait: the condition would flap")
}
// Refused again, now for a reason of today: said in those words.
refuse("no channel can carry any of its answers now: telegram: no account is linked")
_ = r.a.reconcile(context.Background())
if got := last(); len(got) != 1 || !strings.Contains(got[0].Summary, "no account is linked") ||
strings.Contains(got[0].Summary, "has not said") {
t.Fatalf("an old refusal's words repeated: %+v", got)
}
// Asked again after the doubled wait, and taken: no refusal comes, and the condition clears.
r.now = r.now.Add(2 * askEvery)
_ = r.a.reconcile(context.Background())
if n := len(r.asksSent(t)); n != 3 {
t.Fatalf("not asked again after the doubled wait: %d asks", n)
}
r.now = r.now.Add(askVerdictWait)
_ = r.a.reconcile(context.Background())
if got := last(); len(got) != 0 {
t.Fatalf("taken, and still said undelivered: %+v", got)
}
if n := len(r.asksSent(t)); n != 3 {
t.Fatalf("an ask taken was asked again: %d asks", n)
}
// The operator answers it; the condition stays open (its fix takes a while): the old refusal is not said again.
taken := r.asksSent(t)[2]
r.store.Change(context.Background(), taken.ID, func(x *asked) bool {
x.State, x.Ended = string(asks.OutcomeChosen), r.now
return true
})
for i := 0; i < 5; i++ {
r.now = r.now.Add(askEvery)
_ = r.a.reconcile(context.Background())
if got := last(); len(got) != 0 {
t.Fatalf("an answered ask brought its old refusal back: %+v", got)
}
}
// Its explanation opens with one verdict, not two.
r.open = append(r.open, heldCondition())
_ = r.a.reconcile(context.Background())
refuse("no channel can carry any of its answers now")
_ = r.a.reconcile(context.Background())
if e := conditions.Verdict(last()[0].Needs, last()[0].Explanation); strings.Count(e, "Needs you") != 1 {
t.Errorf("the explanation says its verdict twice: %q", e)
}
}