F-host 3: the agent's tests of a pull waiting on its holder wait for the loop to come to rest, not 50-100 ms, before they count applies (review of issue 1966)
waitIdle returns once every change sent to the loop is taken in and applied and no backoff wait runs, so a retry the clock or a wrong settled would start is counted, with no real sleep. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
+42
-1
@@ -21,6 +21,7 @@ import (
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -260,6 +261,10 @@ type Agent struct {
|
||||
again chan againAsk
|
||||
transient chan struct{}
|
||||
stopped chan struct{}
|
||||
// quiet is asked by waitIdle: the loop closes the channel it is given once nothing waits to be taken in or applied.
|
||||
quiet chan chan struct{}
|
||||
// armed counts the backoff waits running: the loop is not idle while one may still ask an apply again.
|
||||
armed atomic.Int32
|
||||
}
|
||||
|
||||
type againAsk struct {
|
||||
@@ -301,7 +306,7 @@ func New(env Env) *Agent {
|
||||
attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}, notComposed: map[string]notComposedMark{},
|
||||
pulls: map[string]*pullWait{}, settledAt: map[string]time.Time{}, settledSeq: map[string]int{},
|
||||
retry: make(chan []string, 8), reconcile: make(chan []string, 64), failedAgain: make(chan string, 8), tries: map[string]int{}, again: make(chan againAsk, 64),
|
||||
transient: make(chan struct{}, 1), stopped: make(chan struct{})}
|
||||
transient: make(chan struct{}, 1), stopped: make(chan struct{}), quiet: make(chan chan struct{})}
|
||||
}
|
||||
|
||||
// Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition
|
||||
@@ -355,6 +360,19 @@ func (a *Agent) Run(ctx context.Context) error {
|
||||
dirty[m] = true
|
||||
}
|
||||
a.mu.Unlock()
|
||||
case done := <-a.quiet:
|
||||
// Idle only when every change already sent is taken in and applied; asked again otherwise, after the rest.
|
||||
if len(dirty) == 0 && len(a.retry)+len(a.reconcile)+len(a.failedAgain)+len(a.again)+len(a.transient)+len(wants)+len(comps) == 0 &&
|
||||
a.armed.Load() == 0 {
|
||||
close(done)
|
||||
} else {
|
||||
go func() {
|
||||
select {
|
||||
case a.quiet <- done:
|
||||
case <-a.stopped:
|
||||
}
|
||||
}()
|
||||
}
|
||||
case ask := <-a.again:
|
||||
// The wait of a transiently failed apply ran out: tried again from the keys it has, if nothing tried it since.
|
||||
a.mu.Lock()
|
||||
@@ -413,6 +431,27 @@ func (a *Agent) Run(ctx context.Context) error {
|
||||
}
|
||||
}
|
||||
|
||||
// waitIdle returns once the loop has taken in and applied every change sent to it before the call (a want or composition
|
||||
// put, a settled heard, a retry asked), with no backoff wait still running, so a caller can count what was applied with no clock; false when ctx ends first.
|
||||
func (a *Agent) waitIdle(ctx context.Context) bool {
|
||||
done := make(chan struct{})
|
||||
select {
|
||||
case a.quiet <- done:
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-a.stopped:
|
||||
return false
|
||||
}
|
||||
select {
|
||||
case <-done:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-a.stopped:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Retry asks for the modules to be applied again from the keys they have, though nothing changed: what ADR 0360 §11
|
||||
// needs for a container that waits because it said busy. It never blocks; a retry asked while eight wait is dropped, and
|
||||
// asked again by its caller's next look.
|
||||
@@ -1571,7 +1610,9 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
|
||||
if a.env.Backoff != nil {
|
||||
wait := a.env.Backoff(try)
|
||||
when = "in " + wait.String()
|
||||
a.armed.Add(1)
|
||||
time.AfterFunc(wait, func() {
|
||||
defer a.armed.Add(-1)
|
||||
select {
|
||||
case a.again <- againAsk{module, gen}:
|
||||
case <-a.stopped:
|
||||
|
||||
@@ -193,6 +193,17 @@ func newBed(t *testing.T) *bed {
|
||||
return b
|
||||
}
|
||||
|
||||
// idle waits until the agent has taken in and applied everything sent to it so far: what is counted after it is all there
|
||||
// will be, with no clock.
|
||||
func (b *bed) idle() {
|
||||
b.t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if !b.ag.waitIdle(ctx) {
|
||||
b.t.Fatalf("the agent's loop did not come to rest")
|
||||
}
|
||||
}
|
||||
|
||||
func (b *bed) run() (stop func()) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
@@ -499,13 +510,13 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||
h := b.haves.last("web")
|
||||
return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "tried again once distribution has settled")
|
||||
})
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
b.idle()
|
||||
if n := len(b.appliedList()); n != 1 {
|
||||
t.Fatalf("a failed pull was tried again by the clock (%d applies)", n)
|
||||
}
|
||||
// A settled of another module starts nothing.
|
||||
b.ag.Settled("gitea")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
b.idle()
|
||||
if n := len(b.appliedList()); n != 1 {
|
||||
t.Fatalf("a settled of another module applied the module (%d applies)", n)
|
||||
}
|
||||
@@ -517,7 +528,7 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||
h := b.haves.last("web")
|
||||
return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "last tried again at") && strings.Contains(h.Detail, "distribution settled")
|
||||
})
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
b.idle()
|
||||
if n := len(b.appliedList()); n != 2 {
|
||||
t.Fatalf("a second failure after the same settled started a further apply (%d applies)", n)
|
||||
}
|
||||
@@ -539,7 +550,7 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||
t.Fatalf("a running module's have still waits on %+v", w)
|
||||
}
|
||||
b.ag.Settled("distribution")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
b.idle()
|
||||
if n := len(b.appliedList()); n != 3 {
|
||||
t.Fatalf("a running module was applied again by a settled (%d applies)", n)
|
||||
}
|
||||
@@ -577,7 +588,7 @@ func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) {
|
||||
b.comps.put(ck, craw)
|
||||
b.wants.put(wk, wraw)
|
||||
eventually(t, "the settled heard during the pull was missed", func() bool { return len(b.appliedList()) == 2 })
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
b.idle()
|
||||
if n := len(b.appliedList()); n != 2 {
|
||||
t.Fatalf("the race retry failing again started a further apply (%d applies)", n)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user