novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

Say when the node composes again after a seat that did not answer, say what a long pass waits on, and serve the verb compose (issue 774)

Browse Source 
On 2026-10-31 ace's route-proxy said only that the vault could not be asked, and ace said nothing for five minutes
while a pass waited after the vault restarted; the retry that already existed was invisible. A refusal for a seat or
a read that did not answer now says when the node composes again ("the node composes again at HH:MM:SS (try N)"), a
retry pass says it starts, a pass that runs on says every minute what it waits on, and the node-engine's verb
compose makes a pass now and says how it went, so a person recovers without a restart.
This commit is contained in:
jschoubben
2026-10-31 00:56:47 +01:00
parent 91fb323ce0
commit 9c15610f70
7 changed files with 255 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/selfcompose/selfcompose.go
+142 -13
View File
@@ -127,6 +127,58 @@ type Composer struct {
missed atomic.Uint64
// owed is set when a pass refused for a bus that did not answer: the composer composes again after a bound.
owed atomic.Bool
// tries counts the passes owed in a row, and wait is the bound before the next (LaterWait): set by the pass that owes
// one, so the refusal it marks says when the node composes again (issue 774, 2026-10-31). Under mu.
tries int
wait time.Duration
// asked are the compose verb's asks (ComposeNow), each answered once the pass it asks is made.
asked chan chan Pass
// last is how the last pass went, under mu.
last Pass
}
// Pass is how one pass went, as the node-engine's verb compose answers it.
type Pass struct {
// At is when the pass ended.
At time.Time `json:"at"`
// Refused is why the node composed none of its modules; empty when it composed.
Refused string `json:"refused,omitempty"`
// Held is each module refused alone, with why.
Held map[string]string `json:"held,omitempty"`
// Handed is how many compositions the pass handed the agent that differ from what it handed before.
Handed int `json:"handed"`
// 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"`
}
// 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
// owe marks a pass owed after a seat or a read that did not answer: the next wait of the growing bound (LaterWait), and
// the words that say when the node composes again, for the refusal it marks.
func (c *Composer) owe() string {
c.mu.Lock()
defer c.mu.Unlock()
c.wait = LaterWait(c.tries)
c.tries++
c.owed.Store(true)
return fmt.Sprintf("the node composes again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.tries)
}
// paid clears what owe marked: the pass composed, or owes nothing.
func (c *Composer) paid() {
c.mu.Lock()
defer c.mu.Unlock()
c.tries, c.wait = 0, 0
c.owed.Store(false)
}
// waitNow is the bound before the pass owed, and its try.
func (c *Composer) waitNow() (time.Duration, int) {
c.mu.Lock()
defer c.mu.Unlock()
return c.wait, c.tries
}
// note counts err when it is the bus not answering, and returns it.
@@ -150,7 +202,26 @@ func New(env Env) *Composer {
}
return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{},
records: map[string]delivery.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{},
manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{}}
manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, sealedOwn: map[string]ownSealed{},
asked: make(chan chan Pass, 8)}
}
// 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.
func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) {
reply := make(chan Pass, 1)
select {
case c.asked <- reply:
case <-ctx.Done():
return Pass{}, errors.New("the node's composer took no ask: it composes nothing now")
}
select {
case p := <-reply:
return p, nil
case <-ctx.Done():
return Pass{}, fmt.Errorf("the node made no pass in time; it waits on %s", c.waitingOn())
}
}
// ownSealed is one own secret's sealed value at a generation and the key it is sealed to.
@@ -270,7 +341,7 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
// A pass refused for a bus that did not answer is made again after a bound (laterAgain), not kept until another
// input moves (issue 1126, 2026-10-28).
var again <-chan time.Time
tries := 0
var replies []chan Pass
for {
// Drain whatever came, then compose once for all of it.
select {
@@ -279,6 +350,16 @@ 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))
case reply := <-c.asked:
// Asked by hand (the verb compose): composed now, the bound starting again.
replies = append(replies, reply)
c.mu.Lock()
c.tries = 0
c.mu.Unlock()
again = nil
dirty = true
case t := <-merged:
if !t.ok {
c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket))
@@ -318,13 +399,19 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
}
}
if c.owed.Load() {
wait := LaterWait(tries)
tries++
wait, _ := c.waitNow()
c.env.Say(fmt.Sprintf("the node composes again in %s: a read on the bus did not answer", wait))
again = time.After(wait)
} else {
tries, again = 0, nil
again = nil
}
c.mu.Lock()
last := c.last
c.mu.Unlock()
for _, r := range replies {
r <- last
}
replies = nil
if !c.caughtSent() {
select {
case out <- agent.Update{Caught: true}:
@@ -518,6 +605,35 @@ func (c *Composer) takeValue(u agent.Update) {
// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one
// handed, and a deletion for each handed one no want names any more.
func (c *Composer) compose(ctx context.Context) []agent.Update {
pass := Pass{}
defer func() {
pass.At = time.Now()
if c.owed.Load() {
wait, _ := c.waitNow()
pass.Again = pass.At.Add(wait)
}
c.mu.Lock()
c.last = pass
c.mu.Unlock()
}()
// A pass that runs on says so, and what it waits on, every passSaysEvery (issue 774, 2026-10-31: ace said nothing for
// five minutes while a pass waited after the vault restarted).
began := time.Now()
long, stopLong := context.WithCancel(ctx)
defer stopLong()
go func() {
t := time.NewTicker(passSaysEvery)
defer t.Stop()
for {
select {
case <-long.Done():
return
case <-t.C:
c.env.Say(fmt.Sprintf("the node's composition has run for %s and waits on %s",
time.Since(began).Round(time.Second), c.waitingOn()))
}
}
}()
c.mu.Lock()
fps := map[string]string{}
kept := map[string]string{}
@@ -542,7 +658,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
// The catalogue no longer places it: the agent returns to the controller's compositions (endWhenUnplaced), and
// this is no refusal of what it runs (issue 1126, the seed's revert).
c.sayOnce("this node's key in the seat's placement state is gone: it composes nothing of its own, and keeps running what it runs")
c.owed.Store(false) // nothing of its own to compose, so no pass is owed and no timer armed again
c.paid() // nothing of its own to compose, so no pass is owed and no timer armed again
return nil
}
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
@@ -581,11 +697,20 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
return nil
}
if err != nil {
why := fmt.Sprintf("%v", err)
// A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28),
// and the refusal says when (issue 774, 2026-10-31). A refusal is final until an input moves.
if BusFailed(err) {
why += "; " + c.owe()
} else {
c.paid()
}
pass.Refused = why
// Said before the agent marks the modules: the marks read each have first, which a paused bus delays.
c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
c.refused(fmt.Sprintf("%v", err), nil)
// A bus that did not answer past every try is no answer: composed again after a bound (issue 1126, 2026-10-28).
c.owed.Store(BusFailed(err))
c.sayOnce("the node cannot compose its modules, and keeps running what it runs: " + why)
done := c.waiting("the agent marking each wanted module's have")
c.refused(why, nil)
done()
return nil
}
// The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs.
@@ -596,14 +721,17 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
if c.missed.Load() > missed {
// A have of what runs the bus did not answer for: its boxes would be the vault's, sealed again, and the
// composition another; nothing is handed, and the pass is made again after a bound (issue 1126, 2026-10-28).
c.sayOnce("the node composes nothing new for now, and keeps running what it runs: the have of what it runs did not answer on the bus")
c.owed.Store(true)
pass.Refused = "the have of what it runs did not answer on the bus; " + c.owe()
c.sayOnce("the node composes nothing new for now, and keeps running what it runs: " + pass.Refused)
return nil
}
c.owed.Store(false)
c.paid()
pass.Held = held
// A module refused alone is held: it keeps what it runs, its have says why, and every other module is applied (issue
// 1126: one module's refusal froze the whole node).
done := c.waiting("the agent marking each held module's have")
c.refused("", held)
done()
if len(held) == 0 {
c.sayOnce("")
} else {
@@ -658,6 +786,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev})
}
}
pass.Handed = len(out)
return out
}