Retry only a failed apply that is not held, answer a repeated compose from the last pass, and say the try the refusal named (issue 774, review round 1)
mesh/delivery merged: source.main moved to novox/mesh-host@3d1465606bed
mesh/delivery-group group fix/compose-retries-transient merged: every member is merged
mesh/check being carried to b0d6684d3731 on 91fb323ce0cf | from mesh-sdk dec3bfe7a8cb | novox/mesh-host#185 in group fix/compose-retries-transient
mesh/delivery merged: source.main moved to novox/mesh-host@3d1465606bed
mesh/delivery-group group fix/compose-retries-transient merged: every member is merged
mesh/check being carried to b0d6684d3731 on 91fb323ce0cf | from mesh-sdk dec3bfe7a8cb | novox/mesh-host#185 in group fix/compose-retries-transient
The verb compose has no warrant (ADR 0563 §2): Agent.Retry released a module held for a failed backup and re-applied one already running its want. RetryFailed tries again only a failed apply not held, and refuses in words otherwise. A hand ask no longer restarts the growing bound, and a pass less than askedWithin old is the answer, so a caller in a loop drives no passes against the vault. A composer whose loop ended refuses at once. The retry line said try N+1. The test's first shadow now ends before slowAfter is set, a race the race detector found under the new timing.
This commit is contained in:
@@ -135,6 +135,8 @@ type Composer struct {
|
||||
asked chan chan Pass
|
||||
// last is how the last pass went, under mu.
|
||||
last Pass
|
||||
// watching says the composer's loop runs (Watch), so an ask of a pass can be answered.
|
||||
watching atomic.Bool
|
||||
}
|
||||
|
||||
// Pass is how one pass went, as the node-engine's verb compose answers it.
|
||||
@@ -150,8 +152,14 @@ type Pass struct {
|
||||
// Again is when the node composes again on its own, after a seat or a read that did not answer; zero when it does
|
||||
// not need to.
|
||||
Again time.Time `json:"again,omitempty"`
|
||||
// Earlier says the answer is the last pass, made less than askedWithin ago, and no new one was made for the ask.
|
||||
Earlier bool `json:"earlier,omitempty"`
|
||||
}
|
||||
|
||||
// askedWithin is how recent a pass answers an ask of the verb compose without a new one: the first wait after a seat
|
||||
// that did not answer (laterAgain's first).
|
||||
var askedWithin = 5 * time.Second
|
||||
|
||||
// passSaysEvery is how often a pass that runs on says, in the journal, how long it has run and what it waits on.
|
||||
var passSaysEvery = time.Minute
|
||||
|
||||
@@ -174,6 +182,12 @@ func (c *Composer) paid() {
|
||||
c.owed.Store(false)
|
||||
}
|
||||
|
||||
// againLine is what a retry pass says as it starts: the try its refusal named.
|
||||
func (c *Composer) againLine() string {
|
||||
_, try := c.waitNow()
|
||||
return fmt.Sprintf("the node composes again now (try %d), after a seat or a read that did not answer", try)
|
||||
}
|
||||
|
||||
// waitNow is the bound before the pass owed, and its try.
|
||||
func (c *Composer) waitNow() (time.Duration, int) {
|
||||
c.mu.Lock()
|
||||
@@ -209,7 +223,20 @@ func New(env Env) *Composer {
|
||||
// ComposeNow asks the composer for a pass now (the node-engine's verb compose, issue 774): the growing bound after a seat
|
||||
// that did not answer starts again, and the answer is how the pass went. Refused in words when no pass is made before
|
||||
// ctx ends: the composer not running, or its inputs not all read yet.
|
||||
//
|
||||
// A pass ended less than askedWithin ago is the answer, and no new one is made, so a caller asking in a loop never
|
||||
// drives passes back to back against the vault; an ask never restarts the growing bound either (review round 1).
|
||||
func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) {
|
||||
if !c.watching.Load() {
|
||||
return Pass{}, errors.New("the node's composer runs no loop now: it composes nothing until its agent starts again")
|
||||
}
|
||||
c.mu.Lock()
|
||||
last := c.last
|
||||
c.mu.Unlock()
|
||||
if !last.At.IsZero() && time.Since(last.At) < askedWithin {
|
||||
last.Earlier = true
|
||||
return last, nil
|
||||
}
|
||||
reply := make(chan Pass, 1)
|
||||
select {
|
||||
case c.asked <- reply:
|
||||
@@ -333,8 +360,10 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
|
||||
}()
|
||||
}
|
||||
out := make(chan agent.Update, 64)
|
||||
c.watching.Store(true)
|
||||
go func() {
|
||||
defer close(out)
|
||||
defer c.watching.Store(false)
|
||||
caught := make([]bool, len(ins))
|
||||
ready := false
|
||||
dirty := false
|
||||
@@ -350,14 +379,11 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
|
||||
case <-again:
|
||||
again = nil
|
||||
dirty = true
|
||||
_, try := c.waitNow()
|
||||
c.env.Say(fmt.Sprintf("the node composes again now (try %d), after a seat or a read that did not answer", try+1))
|
||||
c.env.Say(c.againLine())
|
||||
case reply := <-c.asked:
|
||||
// Asked by hand (the verb compose): composed now, the bound starting again.
|
||||
// Asked by hand (the verb compose): composed now; the growing bound is not restarted, so a pass that
|
||||
// fails again waits longer, not less.
|
||||
replies = append(replies, reply)
|
||||
c.mu.Lock()
|
||||
c.tries = 0
|
||||
c.mu.Unlock()
|
||||
again = nil
|
||||
dirty = true
|
||||
case t := <-merged:
|
||||
|
||||
Reference in New Issue
Block a user