diff --git a/internal/agent/agent.go b/internal/agent/agent.go index f42b747..1367c9f 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -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: diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 646856b..92137b6 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -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) }