Merge pull request 'Node agent: a switch never marks the new run's modules, a failed read or a bus pause keeps what runs, stale composer marks clear (issue 1126, group fix/1126-switch-safe)' (#155) from fix/1126-switch-safe into main

This commit was merged in pull request #155.
This commit is contained in:
2026-10-27 23:52:28 +00:00
9 changed files with 357 additions and 18 deletions
+22 -4
View File
@@ -72,6 +72,20 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool
Sealed: got.Sealed}, true
}
// ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr).
func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) {
if module == agent.NodeModule {
got, ok := h.Read(ctx, module)
return got, ok, nil
}
got, ok, err := h.session.ReadHaveErr(ctx, module)
if !ok || err != nil {
return agent.Have{}, false, err
}
return agent.Have{Module: module, FP: got.FP, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
Sealed: got.Sealed}, true, nil
}
func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
if x.Module == agent.NodeModule {
// The node's own key holds its facts and the digest of the composition it runs.
@@ -509,9 +523,12 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
run, stopRun := context.WithCancel(linked)
// What the node runs and how it opens a sealed value, for the composer to keep the boxes it runs; and the agent
// told when the node cannot compose, so each wanted module's have says it.
// The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in
// flight at the switch marked the next agent's modules failed).
own := &agentNow{}
hooks := composerHooks{
running: func() []delivery.Composed {
if now := current.get(); now != nil {
if now := own.get(); now != nil {
return now.Everything()
}
return nil
@@ -530,8 +547,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
return got.Sealed
},
refused: func(why string) {
if now := current.get(); now != nil {
now.NotComposed(linked, why)
if now := own.get(); now != nil && run.Err() == nil {
now.NotComposed(run, why)
}
},
}
@@ -543,7 +560,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
func(ctx context.Context, answers map[string]delivery.CandidateAnswer) {
session.NoteCandidates(ctx, answers)
})
go shadow.Keep(linked, link.Backoff)
go shadow.Keep(run, link.Backoff) // this run's alone: a switch begins a run, and its shadow, again
composer, why := selfComposer(linked, session, node, say, hooks)
if linked.Err() != nil {
// The link ended before the node knew whose compositions to apply: it chose neither, and waits for the next.
@@ -564,6 +581,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
go endWhenPlaced(run, session, node, stopRun, say)
}
ag = agent.New(env)
own.set(ag)
current.set(ag)
err = ag.Run(run)
stopRun()
+38 -2
View File
@@ -403,13 +403,13 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
hkey, place, fp := st.hkey, st.place, st.fp
within(t, 5*time.Second, "the node did not compose and apply its own module once placed", func() bool { return b.file("web") == "greeting=bonjour\n" })
h8rest(t, b, resolved, candidates, hkey, place, fp)
h8back(t, b, placement, candidates, hkey)
h8back(t, b, placement, resolved, candidates, hkey)
}
// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with
// the controller's compositions at its wants (here none: refused, naming web), and the placement key deleted returns the agent to them without a
// restart, said in the journal and the node's have, keeping what runs.
func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey string) {
func h8back(t *testing.T, b *itBed, placement, resolved, candidates nats.KeyValue, hkey string) {
t.Helper()
node := func() delivery.HaveNode {
e, err := b.haves.Get(hkey)
@@ -440,20 +440,56 @@ func h8back(t *testing.T, b *itBed, placement, candidates nats.KeyValue, hkey st
})
ran := b.have("web").Composed
pkey, _ := delivery.PlacementKey(itNode)
// A pass in flight at the switch (issue 1126, 2026-10-28: the last pass of the own run, failing on its run's end,
// marked the next agent's modules failed): a config value changed starts one, and the key goes at once.
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut"}`)); err != nil {
t.Fatal(err)
}
if err := placement.Delete(pkey); err != nil {
t.Fatal(err)
}
stopWatch := make(chan struct{})
marked := make(chan Have1126, 1)
go func() {
for {
select {
case <-stopWatch:
return
case <-time.After(20 * time.Millisecond):
}
if h := b.have("web"); h.Phase == delivery.PhaseFailed {
select {
case marked <- Have1126{h.Phase, h.Detail}:
default:
}
return
}
}
}()
defer close(stopWatch)
within(t, 15*time.Second, "the node did not return to the controller's compositions when its placement went", func() bool {
b.mu.Lock()
said := strings.Join(b.said, "\n")
b.mu.Unlock()
return strings.Contains(said, "the catalogue no longer places this node") && node().Composes == delivery.ComposesController
})
time.Sleep(3 * time.Second) // what a pass in flight would mark lands within it
select {
case h := <-marked:
t.Errorf("a pass in flight at the switch marked web failed: %+v", h)
default:
}
if h := b.have("web"); h.Composed != ran || h.Phase == delivery.PhaseFailed || b.file("web") != "greeting=salut\n" {
t.Errorf("returning to the controller's compositions moved what runs: %+v, %q", h, b.file("web"))
}
}
// Have1126 is what H8's way back watches of web's have.
type Have1126 struct {
Phase delivery.Phase
Detail string
}
// h8state is what h8 set up that the rest of H8 writes to.
type h8state struct {
placement, resolved, candidates nats.KeyValue
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
golang.org/x/crypto v0.57.0
+2
View File
@@ -24,6 +24,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466 h1:xzbkI9dO
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027222254-472d5f314466/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024 h1:MnmyspVoaCldYmm4Ut1O39RDR9WpiL4TfFgDEnY+QRg=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027223231-e32d77753024/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d h1:DhR7DRKaF0OtA5l/ikIk7gZGeIM1lRz65PUE1DGEkvU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027232145-7359d187db8d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+79 -3
View File
@@ -314,6 +314,49 @@ type notComposedMark struct {
prior Have
}
// NotComposedDetail opens the detail of a have NotComposed marks: the module keeps running what its build and digest
// name, and only the node's composition of it failed.
const NotComposedDetail = "the node cannot compose it, and keeps running what it ran: "
// composerMark says h is a mark NotComposed wrote: failed, its detail opening with NotComposedDetail.
func composerMark(h Have) bool {
return h.Phase == string(delivery.PhaseFailed) && strings.HasPrefix(h.Detail, NotComposedDetail)
}
// HavesReadErr is what a Haves may answer besides Read: a have absent (false, nil) told from one that could not be read
// (false, the error).
type HavesReadErr interface {
ReadErr(ctx context.Context, module string) (Have, bool, error)
}
// readHave reads a module's have: found, absent, or unread. A read that fails is tried again after a short wait while
// the bus does not answer (a reload pauses it for a few seconds, issue 1126), and is unread when it still fails. A
// Haves that cannot tell an absent key from a failed read says absent.
func (a *Agent) readHave(ctx context.Context, module string) (h Have, found, unread bool) {
re, tells := a.env.Haves.(HavesReadErr)
if !tells {
h, found = a.env.Haves.Read(ctx, module)
return h, found, false
}
for try := 0; ; try++ {
h, found, err := re.ReadErr(ctx, module)
if err == nil {
return h, found, false
}
if try == 2 || ctx.Err() != nil {
return Have{}, false, true
}
select {
case <-ctx.Done():
return Have{}, false, true
case <-time.After(time.Duration(1<<try) * readAgain):
}
}
}
// readAgain is the first wait before a have is read again.
var readAgain = time.Second
// NotComposed says the node composing its own could not compose its modules (ADR 0547 §2: a stand-in refused, an
// input that cannot be read): each module with a want gets its have failed, naming why, while it keeps running what it
// ran, so the failure is seen where every have is read and not only in the journal. The have keeps what runs: the
@@ -337,6 +380,9 @@ func (a *Agent) NotComposed(ctx context.Context, why string) {
a.mu.Unlock()
sort.Strings(modules)
for _, module := range modules {
if ctx.Err() != nil {
return // the run ended: what it would mark or clear is the next run's to say (issue 1126)
}
a.mu.Lock()
mark, marked := a.notComposed[module]
r, ran := a.running[module]
@@ -364,16 +410,39 @@ func (a *Agent) NotComposed(ctx context.Context, why string) {
if mark.prior.Phase == "" {
mark.prior.Phase = string(delivery.PhaseRunning)
}
} else if h, ok := a.env.Haves.Read(ctx, module); ok {
} else if h, found, unread := a.readHave(ctx, module); found {
mark.prior = h // what its last have said runs: this agent has run nothing of it since its start
} else if unread {
if ctx.Err() != nil {
return
}
// What its have names cannot be read (the bus paused, issue 1126): a mark written now would drop the build
// and digest it names. Nothing is written; the next refusal marks it once the have reads.
a.mu.Lock()
a.sayOnce("unmarked:"+module, fmt.Sprintf("%s's have cannot be read, so it is not marked: the node cannot compose it: %s", module, why))
a.mu.Unlock()
continue
}
}
// A have that already says another cause (an apply that failed, a backup held, an apply under way) keeps it: a
// composer's mark over it would hide that cause, and baseline would take the mark for what runs (review of the
// switch, round 1).
if p := mark.prior.Phase; !marked && p != "" && p != string(delivery.PhaseRunning) && !composerMark(mark.prior) {
a.mu.Lock()
a.sayOnce("unmarked-cause:"+module, fmt.Sprintf("%s is not marked: its have says %s (%s), and the node cannot compose it: %s",
module, p, mark.prior.Detail, why))
a.mu.Unlock()
continue
}
if ctx.Err() != nil {
return
}
mark.why = why
a.mu.Lock()
a.notComposed[module] = mark
a.mu.Unlock()
a.writeHave(ctx, Have{Module: module, FP: mark.prior.FP, Composed: mark.prior.Composed, Sealed: mark.prior.Sealed,
Phase: string(delivery.PhaseFailed), Detail: "the node cannot compose it, and keeps running what it ran: " + why})
Phase: string(delivery.PhaseFailed), Detail: NotComposedDetail + why})
}
}
@@ -900,8 +969,15 @@ func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp com
if h.Composed != "" {
// This node-engine wrote it: what it says is what runs, and the kept declaration says nothing more. A
// failed apply or a different composition is applied again.
// A have a composer marked failed (NotComposed) names what still runs: taken as running, so a mark this
// agent did not make clears once what it names is what the want names (issue 1126, the stale marks after
// the seed's revert).
marked := composerMark(h)
if marked {
h.Detail = ""
}
if h.FP == w.Want.FP && h.Composed == w.Want.Composed &&
(h.Phase == "" || h.Phase == string(delivery.PhaseRunning)) {
(h.Phase == "" || h.Phase == string(delivery.PhaseRunning) || marked) {
a.mu.Lock()
a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
a.mu.Unlock()
+86
View File
@@ -4,6 +4,7 @@ import (
"context"
"crypto/ed25519"
"encoding/json"
"fmt"
"slices"
"strings"
"sync"
@@ -72,6 +73,19 @@ type haves struct {
stored map[string]Have
log []Have
removed []string
// unreadable is each module whose have the bus cannot read now (ReadErr).
unreadable map[string]bool
}
func (h *haves) ReadErr(ctx context.Context, module string) (Have, bool, error) {
h.mu.Lock()
bad := h.unreadable[module]
h.mu.Unlock()
if bad {
return Have{}, false, fmt.Errorf("nats: no responders available for request")
}
x, ok := h.Read(ctx, module)
return x, ok, nil
}
func (h *haves) Read(_ context.Context, module string) (Have, bool) {
@@ -328,6 +342,42 @@ func TestStartingOnWhatTheKeptDeclarationHoldsAppliesNothing(t *testing.T) {
// What the node-engine itself last applied is what runs after a restart, not the kept declaration.
func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) {
t.Run("says the have again", testStartingOnItsOwnHaveSaysTheHaveAgainSoItsHealthIsThisEngines)
// An apply that failed is not what runs, though its have names the want's build and digest: the agent starting on
// it applies again (review of the switch, round 1).
t.Run("applies again after an apply failure", func(t *testing.T) {
b := newBed(t)
fp := fpOf("a")
ck, craw, dig := composition(t, "avahi", fp, false, file("avahi", "x", "1"))
wk, wraw := wantFor(t, "avahi", fp, dig, "")
b.haves.stored = map[string]Have{"avahi": {Module: "avahi", FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed),
Detail: "apply: unit failed to start"}}
b.comps.before(ck, craw)
b.wants.before(wk, wraw)
stop := b.run()
defer stop()
eventually(t, "a failed apply was not applied again", func() bool { return len(b.appliedList()) == 1 })
})
// Issue 1126, the stale marks after the seed's revert: a have a composer marked failed names what still runs; the
// agent starting on it applies nothing and says it running, so a mark it did not make clears.
t.Run("clears a composer's mark", func(t *testing.T) {
b := newBed(t)
fp := fpOf("a")
ck, craw, dig := composition(t, "avahi", fp, false, file("avahi", "x", "1"))
wk, wraw := wantFor(t, "avahi", fp, dig, "")
b.haves.stored = map[string]Have{"avahi": {Module: "avahi", FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed),
Detail: NotComposedDetail + "frombus: the credentials of ace: context canceled"}}
b.comps.before(ck, craw)
b.wants.before(wk, wraw)
stop := b.run()
defer stop()
eventually(t, "the mark was not cleared", func() bool {
h := b.haves.last("avahi")
return h.Phase == string(delivery.PhaseRunning) && h.FP == fp && h.Composed == dig && h.Detail == ""
})
if got := b.appliedList(); len(got) != 0 {
t.Fatalf("a module whose mark names what the want names was applied: %v", got)
}
})
b := newBed(t)
fp := fpOf("a")
ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1"))
@@ -1041,6 +1091,42 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(
t.Errorf("%s's have while the node cannot compose: %+v", row.module, h)
}
}
// A wanted module whose have cannot be read (the bus paused, issue 1126) is not marked: a mark would drop the build
// and digest its have names.
readAgain = time.Millisecond
b.haves.mu.Lock()
b.haves.unreadable = map[string]bool{"unread": true}
b.haves.mu.Unlock()
want("unread", fpOf("d"), "")
want("fresh", fpOf("e"), "") // never ran, its have absent: marked, naming nothing it runs
// A have that already says another cause keeps it (review of the switch, round 1): an apply that failed.
brokenFP, brokenDig := fpOf("f"), "sha256:"+strings.Repeat("8", 64)
b.haves.Write(context.Background(), Have{Module: "broken", FP: brokenFP, Composed: brokenDig, Phase: string(delivery.PhaseFailed),
Detail: "apply: unit failed to start"})
want("broken", brokenFP, "")
eventually(t, "unread's want was not taken", func() bool {
b.ag.mu.Lock()
defer b.ag.mu.Unlock()
return b.ag.wants["unread"].Present
})
b.ag.NotComposed(context.Background(), "a stand-in for web's password, again")
if h := b.haves.last("unread"); h.Module != "" {
t.Errorf("a module whose have could not be read was marked: %+v", h)
}
if h := b.haves.last("fresh"); h.Phase != string(delivery.PhaseFailed) || h.FP != "" {
t.Errorf("a module that never ran, its have absent, is not marked: %+v", h)
}
if h := b.haves.last("broken"); h.Detail != "apply: unit failed to start" || h.Composed != brokenDig {
t.Errorf("an apply failure was overwritten by a composer's mark: %+v", h)
}
// A run that ended marks nothing: what it would say is the next run's (review of the switch, round 1).
ended, end := context.WithCancel(context.Background())
end()
before := len(b.haves.log)
b.ag.NotComposed(ended, "a refusal from a run that ended")
if len(b.haves.log) != before {
t.Errorf("a run that ended wrote %d haves", len(b.haves.log)-before)
}
b.ag.NotComposed(context.Background(), "")
for _, row := range []struct{ module, composed string }{{"web", dig}, {"idle", idleDig}} {
if h := b.haves.last(row.module); h.Phase != string(delivery.PhaseRunning) || h.Composed != row.composed || len(h.Sealed) != 1 {
+19 -6
View File
@@ -811,33 +811,46 @@ func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) {
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs.
func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) {
h, ok, _ := q.ReadHaveErr(ctx, module)
return h, ok
}
// ErrNotLinkedHaves is a have read with no link to the bus open that can bind the have bucket.
var ErrNotLinkedHaves = errors.New("no link to the bus that binds the have bucket is open")
// ReadHaveErr is ReadHave telling a key absent (false, nil) from one that could not be read (false, the error): a bus
// that does not answer for a moment is not a module that never ran (issue 1126).
func (q *Session) ReadHaveErr(ctx context.Context, module string) (delivery.Have, bool, error) {
q.init()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
hb, ok := bus.(HaveBus)
if !ok || bus == nil {
return delivery.Have{}, false
return delivery.Have{}, false, ErrNotLinkedHaves
}
key, err := delivery.HaveKey(q.Membership.Node, module)
if err != nil {
return delivery.Have{}, false
return delivery.Have{}, false, err
}
read, cancel := context.WithTimeout(ctx, q.timeout())
defer cancel()
bucket, err := hb.HaveBind(read)
if err != nil {
return delivery.Have{}, false
return delivery.Have{}, false, err
}
value, _, err := bucket.Get(read, key)
if errors.Is(err, ErrHaveAbsent) {
return delivery.Have{}, false, nil
}
if err != nil {
return delivery.Have{}, false
return delivery.Have{}, false, err
}
h, err := delivery.DecodeHave(value)
if err != nil {
return delivery.Have{}, false
return delivery.Have{}, false, nil // a have no engine can read names nothing it runs
}
return h, true
return h, true, nil
}
// nothingJudged is the health of a module a look judged no resource of: ok when it was looked at and the module applied
+44 -2
View File
@@ -452,10 +452,29 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
if c.env.Running != nil {
running = c.env.Running()
}
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}, fps)
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
// 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++ {
c.env.Say(fmt.Sprintf("the node's composition met a bus that did not answer, and is tried again in %s: %v", composeAgain[try], err))
select {
case <-ctx.Done():
case <-time.After(composeAgain[try]):
}
if ctx.Err() == nil {
compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
}
}
if ctx.Err() != nil {
// The run ended (the agent begins again, on its own compositions or the controller's): what failed with it is
// no refusal, and nothing is marked (issue 1126: a pass in flight at the switch marked the next agent's modules).
return nil
}
if err != nil {
c.refused(fmt.Sprintf("%v", err))
// 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))
return nil
}
c.refused("")
@@ -904,3 +923,26 @@ 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}
// 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
// by value and by words, since the vault's and the reads' errors arrive wrapped in words (issue 1126).
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) {
return true
}
words := err.Error()
for _, w := range []string{"jetstream not enabled", "no responders"} {
if strings.Contains(words, w) {
return true
}
}
return false
}
+66
View File
@@ -240,6 +240,72 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey)
}
// Issue 1126, the switch made safe: a bus not answering for a moment (a reload of its accounts) is tried again and
// refuses nothing; a pass whose run ended refuses nothing either.
composeAgain = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond}
fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) {
var refusals []string
e := b.env()
e.Get = get
e.Refused = func(why string) { refusals = append(refusals, why) }
c2 := New(e)
c.mu.Lock()
for k, v := range c.wants {
c2.wants[k] = v
}
for k, v := range c.placements {
c2.placements[k] = v
}
for k, v := range c.records {
c2.records[k] = v
}
for k, v := range c.modules {
c2.modules[k] = v
}
c.mu.Unlock()
return c2, &refusals
}
for name, row := range map[string]struct {
failures int // reads failing before the bus answers
cancel bool
composes bool
}{
"the bus paused for one read": {failures: 1, composes: true},
"the bus paused past every try": {failures: 100},
"the run ended mid-pass": {failures: 100, cancel: true},
} {
failed := 0
pctx, pcancel := context.WithCancel(ctx)
c2, refusals := fresh(func(gctx context.Context, bucket, key string) ([]byte, error) {
if failed < row.failures {
failed++
if row.cancel {
pcancel()
return nil, fmt.Errorf("the vault's own could not be asked: %w", context.Canceled)
}
return nil, fmt.Errorf("the revisions of %s in %s: %w", key, bucket, nats.ErrJetStreamNotEnabled)
}
b.mu.Lock()
defer b.mu.Unlock()
return b.kv[bucket+"/"+key], nil
})
handed := c2.compose(pctx)
pcancel()
switch {
case row.composes && (len(handed) == 0 || len(*refusals) != 0):
t.Errorf("%s: handed %d, refusals %q; want composed and nothing refused", name, len(handed), *refusals)
case !row.composes && !row.cancel && (len(*refusals) != 1 || failed != 1+len(composeAgain)):
t.Errorf("%s: refusals %q after %d reads; want one refusal after %d tries", name, *refusals, failed, 1+len(composeAgain))
case row.cancel && len(*refusals) != 0:
t.Errorf("%s: a pass whose run ended refused %q", name, *refusals)
}
}
if !Transient(fmt.Errorf("the manifest of avahi: the revisions of x in mesh-delivery-indexer_manifests: nats: jetstream not enabled")) ||
!Transient(fmt.Errorf("x: %w", nats.ErrNoResponders)) || Transient(fmt.Errorf("nats says ${dir:mesh-state}, and nats declares no directory")) ||
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 config value of another module, and a value changing nothing composed, hand nothing.
b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`))
quiet(t, out)