A read the bus did not answer is tried again after a bound, never kept as an answer: the shadow's answer, the node's composition and the own-secret step (issue 1126)

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.
2026-10-28 12:11:08 +01:00
parent 4182f8fbfb
commit cd1d2ff447
8 changed files with 306 additions and 66 deletions
@@ -20,6 +20,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"

 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -66,7 +67,8 @@ type Env struct {
 	// before the agent applied anything, and after the switch once the kept declaration is old. By slot
 	// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
 	// only within its own module.
-	Held func(module string) (slots delivery.Seals, legacy []string)
+	// An error is a have the bus did not answer for, never one absent: the pass is then tried again (BusFailed).
+	Held func(module string) (slots delivery.Seals, legacy []string, err error)
 	// Fetch reads an archive at a URL in the artifact store: a moved module's templates archive, whose texts its built
 	// manifest carries (issue 1126). Read once per digest. Nil reads none, and a moved module does not compose.
 	Fetch func(ctx context.Context, url string) ([]byte, error)
@@ -102,6 +104,25 @@ type Composer struct {
 	doing   string
 	caught  bool
 	rev     uint64
+	// missed counts the reads the bus did not answer (BusFailed), for a pass or an answer to tell whether it met one:
+	// a read that failed on the bus is tried again after a bound, never kept as an answer (issue 1126, 2026-10-28).
+	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
+}
+
+// note counts err when it is the bus not answering, and returns it.
+func (c *Composer) note(err error) error {
+	if BusFailed(err) {
+		c.missed.Add(1)
+	}
+	return err
+}
+
+// get is env.Get, its failures noted.
+func (c *Composer) get(ctx context.Context, bucket, key string) ([]byte, error) {
+	raw, err := c.env.Get(ctx, bucket, key)
+	return raw, c.note(err)
 }

 // New is a composer for one node.
@@ -204,11 +225,18 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
 		caught := make([]bool, len(ins))
 		ready := false
 		dirty := false
+		// 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
 		for {
 			// Drain whatever came, then compose once for all of it.
 			select {
 			case <-ctx.Done():
 				return
+			case <-again:
+				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))
@@ -247,6 +275,14 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
 					return
 				}
 			}
+			if c.owed.Load() {
+				wait := LaterWait(tries)
+				tries++
+				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
+			}
 			if !c.caughtSent() {
 				select {
 				case out <- agent.Update{Caught: true}:
@@ -459,7 +495,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
 		running = c.env.Running()
 	}
 	scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
+	missed := c.missed.Load()
 	compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
+	err = c.missedIn(missed, err)
 	// A bus that does not answer for a moment (a reload of its accounts pauses it a few seconds) is no refusal: the pass is
 	// tried again after a growing wait before anything is marked (issue 1126).
 	for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ {
@@ -469,7 +507,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
 		case <-time.After(composeAgain[try]):
 		}
 		if ctx.Err() == nil {
+			missed = c.missed.Load()
 			compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
+			err = c.missedIn(missed, err)
 		}
 	}
 	if ctx.Err() != nil {
@@ -481,14 +521,25 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
 		// 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))
+		// 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))
 		return nil
 	}
-	c.refused("")
-	c.sayOnce("")
 	// The boxes it runs, slot by slot: read before c.mu is taken, since the agent's lock is taken to read what runs.
+	missed = c.missed.Load()
 	for key, comp := range compositions {
 		compositions[key] = c.keepRunning(comp, running)
 	}
+	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)
+		return nil
+	}
+	c.owed.Store(false)
+	c.refused("")
+	c.sayOnce("")
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	var out []agent.Update
@@ -557,7 +608,8 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos
 	kept := delivery.Seals{}
 	var legacy []string
 	if c.env.Held != nil {
-		slots, old := c.env.Held(module)
+		slots, old, err := c.env.Held(module)
+		c.note(err)
 		for slot, box := range slots {
 			kept[slot] = box
 		}
@@ -687,7 +739,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
 		return nil, err
 	}
 	defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")()
-	value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key)
+	value, err := r.c.get(ctx, delivery.ManifestsBucket, key)
 	if err != nil {
 		return nil, err
 	}
@@ -710,7 +762,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder
 		return nil, err
 	}
 	defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")()
-	raw, err := r.c.env.Get(ctx, delivery.BuilderBuildsBucket, key)
+	raw, err := r.c.get(ctx, delivery.BuilderBuildsBucket, key)
 	if err != nil || raw == nil {
 		return nil, err
 	}
@@ -740,7 +792,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error)
 	fetching, cancel := context.WithTimeout(ctx, archiveWithin)
 	defer cancel()
 	body, err := r.c.env.Fetch(fetching, url)
-	if err != nil {
+	if r.c.note(err) != nil {
 		return nil, err
 	}
 	sum := sha256.Sum256(body)
@@ -759,7 +811,7 @@ func (r reader) Served(ctx context.Context, node, module string, keys []string)
 	defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
 	out := map[string]any{}
 	for _, k := range keys {
-		raw, err := r.c.env.Get(ctx, ResolvedBucket, node+"."+module+"."+k)
+		raw, err := r.c.get(ctx, ResolvedBucket, node+"."+module+"."+k)
 		if err != nil {
 			return nil, err
 		}
@@ -797,6 +849,7 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre
 	}
 	done := r.c.waiting("the vault's own secrets")
 	raw, err := r.c.env.Own(ctx, map[string]any{})
+	r.c.note(err)
 	done()
 	if err != nil {
 		return nil, fmt.Errorf("the vault's own: %w", err)
@@ -853,6 +906,7 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
 	}
 	done := r.c.waiting("the vault's credentials")
 	raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
+	r.c.note(err)
 	done()
 	if err != nil {
 		return nil, nil, fmt.Errorf("the vault's own: %w", err)
@@ -938,6 +992,46 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin
 // composeAgain are the waits before a pass that met a bus not answering is tried again.
 var composeAgain = []time.Duration{time.Second, 2 * time.Second, 4 * time.Second}

+// laterAgain are the waits, after a pass or an answer refused for a bus that did not answer past composeAgain, before
+// it is made again; the last repeats while the bus does not answer (issue 1126, 2026-10-28: a failed read was kept as
+// the answer until another input moved).
+var laterAgain = []time.Duration{5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute}
+
+// LaterWait is the wait before the try'th making again (from 0), of a pass, an answer or an ask the bus did not answer.
+func LaterWait(try int) time.Duration { return laterAgain[min(try, len(laterAgain)-1)] }
+
+// ErrBusMissed is a pass that met a read the bus did not answer, though the composition was made around it (a have of
+// sealed values that could not be read): never handed as made.
+var ErrBusMissed = errors.New("a read on the bus did not answer during the pass")
+
+// missedIn is err, or ErrBusMissed when no error came but a read since missed failed on the bus.
+func (c *Composer) missedIn(missed uint64, err error) error {
+	if err == nil && c.missed.Load() > missed {
+		return ErrBusMissed
+	}
+	return err
+}
+
+// BusFailed says err is a read the bus did not answer, for any reason but the run's own end: Transient, a read that
+// timed out, or no link to the bus. Such a read is tried again after a bound, never kept as an answer; a key absent, a
+// value refused or a verdict is an answer (issue 1126, 2026-10-28). Matched by value and by words, as Transient.
+func BusFailed(err error) bool {
+	if err == nil || errors.Is(err, context.Canceled) {
+		return false
+	}
+	if Transient(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) ||
+		errors.Is(err, nats.ErrConnectionClosed) || errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) {
+		return true
+	}
+	words := err.Error()
+	for _, w := range []string{"timeout", "deadline exceeded", "no link to the bus", "did not answer", "connection closed"} {
+		if strings.Contains(words, w) {
+			return true
+		}
+	}
+	return false
+}
+
 // Transient says err is the bus not answering for a moment, as a reload of its accounts makes it for a few seconds:
 // JetStream not enabled, or no responders; each fails at once, so a pass tried again costs only its wait. A read that
 // timed out is not one: it already waited longer than such a pause, and is most often a read the bus refuses. Matched
@@ -946,7 +1040,8 @@ func Transient(err error) bool {
 	if err == nil {
 		return false
 	}
-	if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) {
+	if errors.Is(err, nats.ErrJetStreamNotEnabled) || errors.Is(err, nats.ErrJetStreamNotEnabledForAccount) || errors.Is(err, nats.ErrNoResponders) ||
+		errors.Is(err, ErrBusMissed) {
 		return true
 	}
 	words := err.Error()
@@ -9,6 +9,7 @@ import (
 	"slices"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"

@@ -305,6 +306,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 		Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) {
 		t.Error("Transient misjudges a bus that does not answer, or a refusal")
 	}
+	// A read the bus did not answer, timed out or not, is tried again later; the run's own end and a refusal are not.
+	if !BusFailed(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) || !BusFailed(fmt.Errorf("x: %w", nats.ErrNoResponders)) ||
+		!BusFailed(fmt.Errorf("the have of web: %w", nats.ErrTimeout)) || BusFailed(fmt.Errorf("x: %w", context.Canceled)) ||
+		BusFailed(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) || BusFailed(nil) {
+		t.Error("BusFailed misjudges a read the bus did not answer, the run's end, or a refusal")
+	}

 	// Issue 1126, C2: a config value is laid as the controller lays it (catalogue.LaidValue), from the mesh's, the node's
 	// or the module's own and never from a default or an unset one, both of this node's and of what another node serves:
@@ -427,23 +434,28 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 		name, in, held string
 		slots          delivery.Seals
 		kept           bool
+		fails          bool // the have's read the bus did not answer: no composition, made again later (2026-10-28)
 	}{
-		{"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true},
-		{"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false},
-		{"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true},
-		{"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false},
-		{"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false},
+		{"the have holds the rotated value at its slot", "keeper", rotatedBox, delivery.Seals{slot: rotatedBox}, true, false},
+		{"the have holds the value before the rotation", "keeper", beforeBox, delivery.Seals{slot: beforeBox}, false, false},
+		{"a have written before slots holds the rotated value", "keeper", rotatedBox, nil, true, false},
+		{"another module holds the same value", "lemurs", rotatedBox, delivery.Seals{"lemurs.grant-x#sealed": rotatedBox}, false, false},
+		{"another resource of keeper holds the same value", "keeper", rotatedBox, delivery.Seals{"keeper.other#sealed": rotatedBox}, false, false},
+		{"the bus does not answer for the have", "keeper", rotatedBox, nil, false, true},
 	} {
 		restarted := env
 		restarted.Running = func() []delivery.Composed { return nil }
-		restarted.Held = func(module string) (delivery.Seals, []string) {
+		restarted.Held = func(module string) (delivery.Seals, []string, error) {
 			if module != row.in {
-				return nil, nil
+				return nil, nil, nil
+			}
+			if row.fails {
+				return nil, nil, fmt.Errorf("the have of %s: %w", module, nats.ErrTimeout)
 			}
 			if row.slots == nil {
-				return nil, []string{row.held}
+				return nil, []string{row.held}, nil
 			}
-			return row.slots, nil
+			return row.slots, nil, nil
 		}
 		restarted.Unseal = func(box string) ([]byte, error) {
 			b.mu.Lock()
@@ -474,6 +486,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 				composed = string(u.Value)
 			}
 		}
+		if row.fails {
+			if composed != "" || !again.owed.Load() {
+				t.Errorf("%s: composed %v and owes a pass again %v; want nothing handed and a pass owed", row.name, composed != "", again.owed.Load())
+			}
+			continue
+		}
 		if composed == "" {
 			t.Fatalf("%s: keeper was not composed", row.name)
 		}
@@ -496,11 +514,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 		}
 		return nil, fmt.Errorf("not sealed to n1: %s", box)
 	}
-	coreEnv.Held = func(module string) (delivery.Seals, []string) {
+	coreEnv.Held = func(module string) (delivery.Seals, []string, error) {
 		if module == agent.NodeModule {
-			return delivery.Seals{"core-box#sealed": rotatedBox}, nil
+			return delivery.Seals{"core-box#sealed": rotatedBox}, nil, nil
 		}
-		return nil, nil
+		return nil, nil, nil
 	}
 	if got := delivery.SealsIn(New(coreEnv).keepRunning(ownComp, nil))["core-box#sealed"]; got != rotatedBox {
 		t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox)
@@ -572,7 +590,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 	readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
 	senv := readOnly.env()
 	senv.Watch = sb.watch
+	var ownFails atomic.Bool // the vault's own on a bus that does not answer (2026-10-28, ace's answer to seed-7)
 	senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) {
+		if ownFails.Load() {
+			return nil, fmt.Errorf("the vault's own could not be asked: %w", nats.ErrNoResponders)
+		}
 		return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here
 	}
 	// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
@@ -780,6 +802,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
 	if got := answering(); got != settled+len(answers) {
 		t.Errorf("what runs moved, and %d answer(s) were composed again for %d group(s)", got-settled, len(answers))
 	}
+	// A read the bus did not answer is no answer to keep: the group is answered again after a bound, though nothing moved.
+	laterAgain = []time.Duration{20 * time.Millisecond}
+	ownFails.Store(true)
+	candidate("adds", 9, "lemurs", "keeper")
+	waitFor("the answer met the vault not answering", func(a map[string]delivery.CandidateAnswer) bool {
+		return a["adds"].Revision == 9 && strings.Contains(a["adds"].Refused, "no responders")
+	})
+	ownFails.Store(false)
+	waitFor("answered again once the bus answers, with nothing else moved", func(a map[string]delivery.CandidateAnswer) bool {
+		return a["adds"].Revision == 9 && a["adds"].Verdict == delivery.CandidateDiffers
+	})
+	drain()
 	k, _ := delivery.PlacementCandidateKey("same", "n1")
 	sb.del(delivery.PlacementCandidatesBucket, k)
 	waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 })
@@ -46,6 +46,10 @@ type Shadow struct {
 	// haves whose health words changed at every look).
 	inputs   uint64
 	answered map[string]answeredAt
+	// owed says an answer met a read the bus did not answer: it is no answer to keep, and its group is answered again
+	// after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not
+	// yet ... no responders" in its answer to seed-7).
+	owed bool
 }

 // answeredAt is what one group's answer was composed at: its keys' revisions, by node, and the count of input changes.
@@ -154,10 +158,15 @@ func (s *Shadow) Run(ctx context.Context) error {
 	// knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531).
 	waiting := time.NewTimer(notReadyAfter)
 	defer waiting.Stop()
+	var again <-chan time.Time
+	tries := 0
 	for {
 		select {
 		case <-ctx.Done():
 			return nil
+		case <-again:
+			again = nil
+			dirty = true
 		case <-waiting.C:
 			if !ready {
 				s.notReady(ctx, ins, caught)
@@ -205,6 +214,14 @@ func (s *Shadow) Run(ctx context.Context) error {
 		if dirty {
 			dirty = false
 			s.answerAll(ctx)
+			if s.owed {
+				wait := LaterWait(tries)
+				tries++
+				c.env.Say(fmt.Sprintf("this node answers again in %s the placement candidates whose answer met a bus that did not answer", wait))
+				again = time.After(wait)
+			} else {
+				tries, again = 0, nil
+			}
 		}
 	}
 }
@@ -390,6 +407,7 @@ func (s *Shadow) answerAll(ctx context.Context) {
 			next[l.group] = a
 		}
 	}
+	s.owed = false
 	s.mu.Unlock()
 	sort.Slice(groups, func(i, j int) bool {
 		if groups[i].rev != groups[j].rev {
@@ -449,8 +467,10 @@ func (s *Shadow) answerAll(ctx context.Context) {
 				}
 			}
 		}()
+		missed := c.missed.Load()
 		a, ok := s.answerGroup(ctx, group)
 		close(stop)
+		provisional := c.missed.Load() > missed && ctx.Err() == nil
 		took := time.Since(began)
 		if took > slowAfter {
 			c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second)))
@@ -468,7 +488,14 @@ func (s *Shadow) answerAll(ctx context.Context) {
 			c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group))
 		}
 		s.mu.Lock()
-		s.answered[group] = l.at
+		if provisional {
+			// What the bus did not answer is no answer: handed as it stands, and composed again after a bound.
+			s.owed = true
+			delete(s.answered, group)
+			c.env.Say(fmt.Sprintf("the answer to %s met a read on the bus that did not answer: it is answered again", group))
+		} else {
+			s.answered[group] = l.at
+		}
 		s.mu.Unlock()
 		hand()
 	}
@@ -522,7 +549,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
 			wants[m] = w
 		}
 		s.mu.Unlock()
-		return returning(ctx, c.env.Get, node, wants, running, head, rev), true
+		return returning(ctx, c.get, node, wants, running, head, rev), true
 	}

 	placement := given[node]