jschoubben
cd1d2ff447
Met twice on 2026-10-28: shanks kept 'want.shanks.brother could not be read ... no responders' as the own-secret step's last word, and ace kept 'the vault's own: not yet ... no responders' as its answer to seed-7, each until another input moved. Every bus read of the composer (manifests, builds, archives, served values, the vault's own and credentials, the have of what runs) is noted (BusFailed); an answer or a pass that met one is made again after a growing bound (5 s to 5 min), and a pass is not handed when the have of its sealed values did not answer. The own-secret step tells a read or an ask that did not answer (Retry, taken again) from a key absent or the vault's word.
95 lines
2.6 KiB
Go
95 lines
2.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"github.com/novox/mesh-host/internal/identity"
|
|
"github.com/novox/mesh-host/internal/link"
|
|
"github.com/novox/mesh-host/internal/ownresync"
|
|
"github.com/novox/mesh-host/internal/selfcompose"
|
|
)
|
|
|
|
// composesNow is whose compositions this run of the agent applies, once decided ("" before).
|
|
type composesNow struct {
|
|
mu sync.Mutex
|
|
now string
|
|
}
|
|
|
|
func (c *composesNow) set(s string) { c.mu.Lock(); c.now = s; c.mu.Unlock() }
|
|
func (c *composesNow) get() string { c.mu.Lock(); defer c.mu.Unlock(); return c.now }
|
|
|
|
// resyncStep is the step that hands the vault what runs of an own secret a shadow answer says another value of (issue
|
|
// 1126, cause C4; internal/ownresync): one pass at a time, and the last answers taken again once the agent has decided
|
|
// whose compositions it applies.
|
|
type resyncStep struct {
|
|
r *ownresync.Resync
|
|
one sync.Mutex
|
|
mu sync.Mutex
|
|
last map[string]delivery.CandidateAnswer
|
|
// tries and pending are the step taken again after a bound when a read or an ask the bus did not answer stopped it
|
|
// (Outcome.Retry, issue 1126, 2026-10-28), one wait at a time.
|
|
tries int
|
|
pending bool
|
|
}
|
|
|
|
func ownResync(session *link.Session, node string, say func(string), env selfcompose.Env, composes *composesNow) *resyncStep {
|
|
return &resyncStep{r: ownresync.New(ownresync.Env{Node: node, Composes: composes.get, Running: env.Running,
|
|
Get: env.Get, Unseal: env.Unseal, Seal: identity.Seal, Say: say,
|
|
Ask: func(ctx context.Context, verb string, args map[string]any) (json.RawMessage, error) {
|
|
return session.AskVault(ctx, verb, args)
|
|
}})}
|
|
}
|
|
|
|
// take keeps answers as the last and takes them.
|
|
func (s *resyncStep) take(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
|
|
s.mu.Lock()
|
|
s.last = answers
|
|
s.mu.Unlock()
|
|
s.again(ctx)
|
|
}
|
|
|
|
// again takes the last answers, and once more after a bound (selfcompose.LaterWait) while the bus did not answer for one.
|
|
func (s *resyncStep) again(ctx context.Context) {
|
|
s.one.Lock()
|
|
defer s.one.Unlock()
|
|
s.mu.Lock()
|
|
last := s.last
|
|
s.mu.Unlock()
|
|
if last == nil {
|
|
return
|
|
}
|
|
retry := false
|
|
for _, o := range s.r.Take(ctx, last) {
|
|
retry = retry || o.Retry
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if !retry {
|
|
s.tries = 0
|
|
return
|
|
}
|
|
if s.pending || ctx.Err() != nil {
|
|
return
|
|
}
|
|
wait := selfcompose.LaterWait(s.tries)
|
|
s.tries++
|
|
s.pending = true
|
|
go func() {
|
|
select {
|
|
case <-ctx.Done():
|
|
case <-time.After(wait):
|
|
}
|
|
s.mu.Lock()
|
|
s.pending = false
|
|
s.mu.Unlock()
|
|
if ctx.Err() == nil {
|
|
s.again(ctx)
|
|
}
|
|
}()
|
|
}
|