Merge pull request 'The node answers a placement candidate once, retries a read the bus did not answer, and keeps a file's secrets entry by entry (issue 1126; group fix/1126-seed-same-3)' (#160) from fix/1126-seed-same-3 into main
This commit was merged in pull request #160.
This commit is contained in:
@@ -75,8 +75,12 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool
|
|||||||
// ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr).
|
// 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) {
|
func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, bool, error) {
|
||||||
if module == agent.NodeModule {
|
if module == agent.NodeModule {
|
||||||
got, ok := h.Read(ctx, module)
|
// The node's own key, its read the bus did not answer told as such (issue 1126, 2026-10-28).
|
||||||
return got, ok, nil
|
got, ok, err := h.session.ReadNodeHaveErr(ctx)
|
||||||
|
if !ok || err != nil || got.Composed == "" {
|
||||||
|
return agent.Have{}, false, err
|
||||||
|
}
|
||||||
|
return agent.Have{Module: module, Composed: got.Composed, Sealed: got.SealedAt}, true, nil
|
||||||
}
|
}
|
||||||
got, ok, err := h.session.ReadHaveErr(ctx, module)
|
got, ok, err := h.session.ReadHaveErr(ctx, module)
|
||||||
if !ok || err != nil {
|
if !ok || err != nil {
|
||||||
@@ -535,16 +539,17 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
|
|||||||
},
|
},
|
||||||
unseal: sealOpener(opts.state),
|
unseal: sealOpener(opts.state),
|
||||||
// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
|
// And what each module's have says it runs sealed: after a restart the agent has applied nothing yet.
|
||||||
held: func(module string) (delivery.Seals, []string) {
|
// A have the bus did not answer for is said as such, never as one holding no seals (issue 1126, 2026-10-28).
|
||||||
|
held: func(module string) (delivery.Seals, []string, error) {
|
||||||
if module == agent.NodeModule {
|
if module == agent.NodeModule {
|
||||||
got, _ := session.ReadNodeHave(linked)
|
got, _, err := session.ReadNodeHaveErr(linked)
|
||||||
return got.SealedAt, got.Sealed
|
return got.SealedAt, got.Sealed, err
|
||||||
}
|
}
|
||||||
got, ok := session.ReadHave(linked, module)
|
got, ok, err := session.ReadHaveErr(linked, module)
|
||||||
if !ok {
|
if !ok {
|
||||||
return nil, nil
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
return got.SealedAt, got.Sealed
|
return got.SealedAt, got.Sealed, nil
|
||||||
},
|
},
|
||||||
refused: func(why string) {
|
refused: func(why string) {
|
||||||
if now := own.get(); now != nil && run.Err() == nil {
|
if now := own.get(); now != nil && run.Err() == nil {
|
||||||
@@ -879,7 +884,7 @@ type composerHooks struct {
|
|||||||
running func() []delivery.Composed
|
running func() []delivery.Composed
|
||||||
unseal func(string) ([]byte, error)
|
unseal func(string) ([]byte, error)
|
||||||
refused func(string)
|
refused func(string)
|
||||||
held func(module string) (delivery.Seals, []string)
|
held func(module string) (delivery.Seals, []string, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) {
|
func selfComposer(ctx context.Context, session *link.Session, node string, say func(string), hooks composerHooks) (*selfcompose.Composer, string) {
|
||||||
|
|||||||
@@ -403,9 +403,67 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) {
|
|||||||
hkey, place, fp := st.hkey, st.place, st.fp
|
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" })
|
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)
|
h8rest(t, b, resolved, candidates, hkey, place, fp)
|
||||||
|
t.Run("one answer per candidate revision", func(t *testing.T) { h8once(t, b, resolved, candidates, place) })
|
||||||
h8back(t, b, placement, resolved, candidates, hkey)
|
h8back(t, b, placement, resolved, candidates, hkey)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// h8once is issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, its bus flooded):
|
||||||
|
// a candidate revision is answered once. The node's own answer written, a module's have whose health words alone move
|
||||||
|
// (each look rewrites them) and a config value written again with the value it held answer nothing again; the
|
||||||
|
// candidate's next revision is answered once.
|
||||||
|
func h8once(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, place []byte) {
|
||||||
|
answering := func() int {
|
||||||
|
b.mu.Lock()
|
||||||
|
defer b.mu.Unlock()
|
||||||
|
n := 0
|
||||||
|
for _, l := range b.said {
|
||||||
|
if strings.HasPrefix(l, "answering placement candidate feat/x ") {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
time.Sleep(time.Second)
|
||||||
|
settled := answering()
|
||||||
|
if settled == 0 {
|
||||||
|
t.Fatal("feat/x was never answered")
|
||||||
|
}
|
||||||
|
hkey, _ := delivery.HaveKey(itNode, "web")
|
||||||
|
for _, detail := range []string{"its tool web_health: no link to the bus is open", "its tool web_health: no answer within 10s"} {
|
||||||
|
h := b.have("web")
|
||||||
|
h.Health, h.Detail = delivery.HealthFailed, detail
|
||||||
|
raw, err := delivery.EncodeHave(h)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := b.haves.Put(hkey, raw); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
time.Sleep(1500 * time.Millisecond)
|
||||||
|
if got := answering(); got != settled {
|
||||||
|
t.Errorf("feat/x answered %d more time(s) at the same revision, with nothing it is composed from changed", got-settled)
|
||||||
|
}
|
||||||
|
same, _ := delivery.DecodePlacement(place)
|
||||||
|
craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x",
|
||||||
|
Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1b"}, Placement: &same})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode)
|
||||||
|
if _, err := candidates.Put(ckey, craw); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
within(t, 10*time.Second, "the candidate's next revision was not answered", func() bool { return answering() > settled })
|
||||||
|
time.Sleep(1500 * time.Millisecond)
|
||||||
|
if got := answering(); got != settled+1 {
|
||||||
|
t.Errorf("the candidate's next revision was answered %d time(s), want 1", got-settled)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// h8back is the way back (issue 1126, the seed's revert): a candidate deleting the node's placement is compared with
|
// 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
|
// 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.
|
// restart, said in the journal and the node's have, keeping what runs.
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
@@ -30,6 +31,10 @@ type resyncStep struct {
|
|||||||
one sync.Mutex
|
one sync.Mutex
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
last map[string]delivery.CandidateAnswer
|
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 {
|
func ownResync(session *link.Session, node string, say func(string), env selfcompose.Env, composes *composesNow) *resyncStep {
|
||||||
@@ -48,14 +53,42 @@ func (s *resyncStep) take(ctx context.Context, answers map[string]delivery.Candi
|
|||||||
s.again(ctx)
|
s.again(ctx)
|
||||||
}
|
}
|
||||||
|
|
||||||
// again takes the last answers.
|
// 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) {
|
func (s *resyncStep) again(ctx context.Context) {
|
||||||
s.one.Lock()
|
s.one.Lock()
|
||||||
defer s.one.Unlock()
|
defer s.one.Unlock()
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
last := s.last
|
last := s.last
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
if last != nil {
|
if last == nil {
|
||||||
s.r.Take(ctx, last)
|
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)
|
||||||
|
}
|
||||||
|
}()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ module github.com/novox/mesh-host
|
|||||||
go 1.26.0
|
go 1.26.0
|
||||||
|
|
||||||
require (
|
require (
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960
|
||||||
github.com/nats-io/nats-server/v2 v2.11.17
|
github.com/nats-io/nats-server/v2 v2.11.17
|
||||||
github.com/nats-io/nats.go v1.54.0
|
github.com/nats-io/nats.go v1.54.0
|
||||||
golang.org/x/crypto v0.57.0
|
golang.org/x/crypto v0.57.0
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2 h1:eSwObLdxs3jxPrlcwdaMo9hzIDTUz2ogWw8cjX8To7I=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960 h1:+XJKxMzi1l0U9zAlv2wj8i3hTmHvc3XqBcIhCM2luV4=
|
||||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028090029-3028c7ec4ed2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028112237-8520a4c09960/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
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=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
+16
-6
@@ -779,33 +779,43 @@ func (q *Session) NoteComposes(ctx context.Context, composes, why string) {
|
|||||||
// ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition
|
// ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition
|
||||||
// it last ran is and carries sealed, read after a restart (ADR 0547 §3).
|
// it last ran is and carries sealed, read after a restart (ADR 0547 §3).
|
||||||
func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) {
|
func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) {
|
||||||
|
n, ok, _ := q.ReadNodeHaveErr(ctx)
|
||||||
|
return n, ok
|
||||||
|
}
|
||||||
|
|
||||||
|
// ReadNodeHaveErr is ReadNodeHave telling the key absent (false, nil) from one that could not be read (false, the
|
||||||
|
// error): a bus that did not answer is never kept as a node that ran nothing (issue 1126, 2026-10-28).
|
||||||
|
func (q *Session) ReadNodeHaveErr(ctx context.Context) (delivery.HaveNode, bool, error) {
|
||||||
q.init()
|
q.init()
|
||||||
q.mu.Lock()
|
q.mu.Lock()
|
||||||
bus := q.bus
|
bus := q.bus
|
||||||
q.mu.Unlock()
|
q.mu.Unlock()
|
||||||
hb, ok := bus.(HaveBus)
|
hb, ok := bus.(HaveBus)
|
||||||
if !ok || bus == nil {
|
if !ok || bus == nil {
|
||||||
return delivery.HaveNode{}, false
|
return delivery.HaveNode{}, false, ErrNotLinkedHaves
|
||||||
}
|
}
|
||||||
key, err := delivery.HaveNodeKey(q.Membership.Node)
|
key, err := delivery.HaveNodeKey(q.Membership.Node)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return delivery.HaveNode{}, false
|
return delivery.HaveNode{}, false, err
|
||||||
}
|
}
|
||||||
read, cancel := context.WithTimeout(ctx, q.timeout())
|
read, cancel := context.WithTimeout(ctx, q.timeout())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
bucket, err := hb.HaveBind(read)
|
bucket, err := hb.HaveBind(read)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return delivery.HaveNode{}, false
|
return delivery.HaveNode{}, false, err
|
||||||
}
|
}
|
||||||
value, _, err := bucket.Get(read, key)
|
value, _, err := bucket.Get(read, key)
|
||||||
|
if errors.Is(err, ErrHaveAbsent) {
|
||||||
|
return delivery.HaveNode{}, false, nil
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return delivery.HaveNode{}, false
|
return delivery.HaveNode{}, false, err
|
||||||
}
|
}
|
||||||
n, err := delivery.DecodeHaveNode(value)
|
n, err := delivery.DecodeHaveNode(value)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return delivery.HaveNode{}, false
|
return delivery.HaveNode{}, false, nil // a key no engine can read names nothing it runs
|
||||||
}
|
}
|
||||||
return n, true
|
return n, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
|
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
|
||||||
|
|||||||
@@ -28,6 +28,8 @@ import (
|
|||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
|
||||||
|
"github.com/novox/mesh-host/internal/selfcompose"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Verb is the vault's verb this step asks, with ReplaceRunning true in both steps.
|
// Verb is the vault's verb this step asks, with ReplaceRunning true in both steps.
|
||||||
@@ -60,6 +62,9 @@ type Outcome struct {
|
|||||||
Name string
|
Name string
|
||||||
Asked bool
|
Asked bool
|
||||||
Said string
|
Said string
|
||||||
|
// Retry says a read or an ask the bus did not answer stopped it: no answer, kept as none, and the step is to be
|
||||||
|
// taken again after a bound (issue 1126, 2026-10-28: "want.shanks.brother could not be read ... no responders").
|
||||||
|
Retry bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// Resync is the step of one node agent's run.
|
// Resync is the step of one node agent's run.
|
||||||
@@ -107,7 +112,7 @@ func (r *Resync) Take(ctx context.Context, answers map[string]delivery.Candidate
|
|||||||
var out []Outcome
|
var out []Outcome
|
||||||
for _, o := range drift {
|
for _, o := range drift {
|
||||||
res := r.one(ctx, composes, o)
|
res := r.one(ctx, composes, o)
|
||||||
if res.Said != "" {
|
if res.Said != "" || res.Retry {
|
||||||
out = append(out, res)
|
out = append(out, res)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -125,7 +130,13 @@ func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDi
|
|||||||
return r.said(key, "not:composes", res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
|
return r.said(key, "not:composes", res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
|
||||||
"nothing is handed: this node composes its own (%q), so the vault's value is the source", key, composes))
|
"nothing is handed: this node composes its own (%q), so the vault's value is the source", key, composes))
|
||||||
}
|
}
|
||||||
sealed, why := r.running(ctx, o)
|
sealed, why, failed := r.running(ctx, o)
|
||||||
|
if failed != nil {
|
||||||
|
// The bus did not answer: no reason to keep, and taken again (Retry); said once per reason.
|
||||||
|
res.Retry = true
|
||||||
|
return r.said(key, "retry:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
|
||||||
|
"nothing is handed yet: %s; asked again", key, why))
|
||||||
|
}
|
||||||
if why != "" {
|
if why != "" {
|
||||||
return r.said(key, "not:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
|
return r.said(key, "not:"+why, res, fmt.Sprintf("%s holds another value in the vault than runs here, and "+
|
||||||
"nothing is handed: %s", key, why))
|
"nothing is handed: %s", key, why))
|
||||||
@@ -149,7 +160,13 @@ func (r *Resync) one(ctx context.Context, composes string, o delivery.OwnValueDi
|
|||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
return res // the run ended: asked again in the next
|
return res // the run ended: asked again in the next
|
||||||
}
|
}
|
||||||
said = fmt.Sprintf("%s: the vault's %s with replace_running refused or did not answer: %v", key, Verb, err)
|
if selfcompose.BusFailed(err) {
|
||||||
|
// The vault did not answer on the bus: no word of the vault's, so not kept as asked (Retry).
|
||||||
|
res.Retry = true
|
||||||
|
return r.said(key, "retry:ask", res, fmt.Sprintf("%s: the vault's %s with replace_running did not answer, "+
|
||||||
|
"and is asked again: %v", key, Verb, err))
|
||||||
|
}
|
||||||
|
said = fmt.Sprintf("%s: the vault's %s with replace_running refused: %v", key, Verb, err)
|
||||||
}
|
}
|
||||||
// Once per running value, whatever the vault answered: a refusal is the vault's word, said once, and a new value
|
// Once per running value, whatever the vault answered: a refusal is the vault's word, said once, and a new value
|
||||||
// that runs asks again.
|
// that runs asks again.
|
||||||
@@ -176,8 +193,8 @@ func (r *Resync) said(key, reason string, res Outcome, why string) Outcome {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// running is the sealed value of o in the composition that runs, when that composition is the one the module's want
|
// running is the sealed value of o in the composition that runs, when that composition is the one the module's want
|
||||||
// names; or why it is not handed.
|
// names; or why it is not handed, with the error of a read the bus did not answer, which is no answer.
|
||||||
func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (string, string) {
|
func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (sealed, why string, failed error) {
|
||||||
var comp *delivery.Composed
|
var comp *delivery.Composed
|
||||||
if r.env.Running != nil {
|
if r.env.Running != nil {
|
||||||
for _, c := range r.env.Running() {
|
for _, c := range r.env.Running() {
|
||||||
@@ -189,39 +206,45 @@ func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (strin
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
if comp == nil {
|
if comp == nil {
|
||||||
return "", o.Module + " runs no composition here"
|
return "", o.Module + " runs no composition here", nil
|
||||||
}
|
}
|
||||||
wantKey, err := delivery.WantKey(r.env.Node, o.Module)
|
wantKey, err := delivery.WantKey(r.env.Node, o.Module)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err.Error()
|
return "", err.Error(), nil
|
||||||
}
|
}
|
||||||
raw, err := r.env.Get(ctx, delivery.AssignerWantsBucket, wantKey)
|
raw, err := r.env.Get(ctx, delivery.AssignerWantsBucket, wantKey)
|
||||||
if err != nil || raw == nil {
|
if err != nil {
|
||||||
return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err)
|
return "", fmt.Sprintf("its want %s could not be read (%v)", wantKey, err), err
|
||||||
|
}
|
||||||
|
if raw == nil {
|
||||||
|
return "", fmt.Sprintf("it has no want %s", wantKey), nil
|
||||||
}
|
}
|
||||||
want, err := delivery.DecodeWant(raw)
|
want, err := delivery.DecodeWant(raw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err)
|
return "", fmt.Sprintf("its want %s is not one: %v", wantKey, err), nil
|
||||||
}
|
}
|
||||||
if want.Self || want.Removed || want.Composed == "" {
|
if want.Self || want.Removed || want.Composed == "" {
|
||||||
return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey)
|
return "", fmt.Sprintf("its want %s names no composition of the controller's", wantKey), nil
|
||||||
}
|
}
|
||||||
// What runs is the composition the want names, as the module's have says: applied and running.
|
// What runs is the composition the want names, as the module's have says: applied and running.
|
||||||
haveKey, err := delivery.HaveKey(r.env.Node, o.Module)
|
haveKey, err := delivery.HaveKey(r.env.Node, o.Module)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", err.Error()
|
return "", err.Error(), nil
|
||||||
}
|
}
|
||||||
raw, err = r.env.Get(ctx, delivery.HaveBucket, haveKey)
|
raw, err = r.env.Get(ctx, delivery.HaveBucket, haveKey)
|
||||||
if err != nil || raw == nil {
|
if err != nil {
|
||||||
return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err)
|
return "", fmt.Sprintf("its have %s could not be read (%v)", haveKey, err), err
|
||||||
|
}
|
||||||
|
if raw == nil {
|
||||||
|
return "", fmt.Sprintf("it has no have %s", haveKey), nil
|
||||||
}
|
}
|
||||||
have, err := delivery.DecodeHave(raw)
|
have, err := delivery.DecodeHave(raw)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err)
|
return "", fmt.Sprintf("its have %s is not one: %v", haveKey, err), nil
|
||||||
}
|
}
|
||||||
if have.Composed != want.Composed || have.Phase != delivery.PhaseRunning {
|
if have.Composed != want.Composed || have.Phase != delivery.PhaseRunning {
|
||||||
return "", fmt.Sprintf("what runs (%s, %s) is not yet the composition its want names (%s)", have.Composed,
|
return "", fmt.Sprintf("what runs (%s, %s) is not yet the composition its want names (%s)", have.Composed,
|
||||||
have.Phase, want.Composed)
|
have.Phase, want.Composed), nil
|
||||||
}
|
}
|
||||||
// On a node the resource's id carries its module's name (`only-office.needs-jwt`); delivery.OwnNeedName reads it.
|
// On a node the resource's id carries its module's name (`only-office.needs-jwt`); delivery.OwnNeedName reads it.
|
||||||
id := o.Module + "." + catalogue.NeedID(o.Name)
|
id := o.Module + "." + catalogue.NeedID(o.Name)
|
||||||
@@ -237,11 +260,11 @@ func (r *Resync) running(ctx context.Context, o delivery.OwnValueDiffers) (strin
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if s, ok := body.Sealed.(string); ok && s != "" {
|
if s, ok := body.Sealed.(string); ok && s != "" {
|
||||||
return s, ""
|
return s, "", nil
|
||||||
}
|
}
|
||||||
return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id)
|
return "", fmt.Sprintf("its resource %s carries no value sealed to this machine", id), nil
|
||||||
}
|
}
|
||||||
return "", fmt.Sprintf("what runs holds no resource %s", id)
|
return "", fmt.Sprintf("what runs holds no resource %s", id), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// hand asks the vault's two steps with value, and checks what the vault holds after by its fingerprint.
|
// hand asks the vault's two steps with value, and checks what the vault holds after by its fingerprint.
|
||||||
|
|||||||
@@ -30,12 +30,18 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
|
|||||||
refuse string // the vault's refusal
|
refuse string // the vault's refusal
|
||||||
asks int // the vault's verb asked, both passes
|
asks int // the vault's verb asked, both passes
|
||||||
says string
|
says string
|
||||||
|
// Issue 1126, 2026-10-28: a read or an ask the bus did not answer is no answer: taken again (Retry), never kept.
|
||||||
|
wantRead string // the want's read: "" answers it, "fails" the bus does not answer, "absent" no key
|
||||||
|
retry bool
|
||||||
}{
|
}{
|
||||||
"applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2"},
|
"applies the controller's, runs its want": {delivery.ComposesController, digest, "", 2, "generation 1 replaced by 2", "", false},
|
||||||
"composes its own": {delivery.ComposesOwn, digest, "", 0, "composes its own"},
|
"composes its own": {delivery.ComposesOwn, digest, "", 0, "composes its own", "", false},
|
||||||
"not decided yet": {"", digest, "", 0, ""},
|
"not decided yet": {"", digest, "", 0, "", "", false},
|
||||||
"runs not yet what its want names": {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition"},
|
"runs not yet what its want names": {delivery.ComposesController, "sha256:" + strings.Repeat("c", 64), "", 0, "not yet the composition", "", false},
|
||||||
"the vault refuses": {delivery.ComposesController, digest, "switched off", 1, "switched off"},
|
"the vault refuses": {delivery.ComposesController, digest, "switched off", 1, "switched off", "", false},
|
||||||
|
"the vault does not answer": {delivery.ComposesController, digest, "nats: no responders available for request", 2, "asked again", "", true},
|
||||||
|
"the want's read does not answer": {delivery.ComposesController, digest, "", 0, "asked again", "fails", true},
|
||||||
|
"it has no want": {delivery.ComposesController, digest, "", 0, "it has no want", "absent", false},
|
||||||
} {
|
} {
|
||||||
var asked []map[string]any
|
var asked []map[string]any
|
||||||
var said []string
|
var said []string
|
||||||
@@ -50,6 +56,10 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
|
|||||||
},
|
},
|
||||||
Get: func(_ context.Context, bucket, key string) ([]byte, error) {
|
Get: func(_ context.Context, bucket, key string) ([]byte, error) {
|
||||||
switch {
|
switch {
|
||||||
|
case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office" && row.wantRead == "fails":
|
||||||
|
return nil, errors.New("the bucket mesh-delivery-assigner_wants: nats: no responders available for request")
|
||||||
|
case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office" && row.wantRead == "absent":
|
||||||
|
return nil, nil
|
||||||
case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office":
|
case bucket == delivery.AssignerWantsBucket && key == "want.novox.only-office":
|
||||||
return want, nil
|
return want, nil
|
||||||
case bucket == delivery.HaveBucket && key == "novox.only-office":
|
case bucket == delivery.HaveBucket && key == "novox.only-office":
|
||||||
@@ -90,7 +100,10 @@ func TestTheNodeHandsTheVaultWhatRunsOfADriftedOwnSecret(t *testing.T) {
|
|||||||
if len(asked) != row.asks {
|
if len(asked) != row.asks {
|
||||||
t.Errorf("%s: the vault asked %d times, want %d: %v", name, len(asked), row.asks, asked)
|
t.Errorf("%s: the vault asked %d times, want %d: %v", name, len(asked), row.asks, asked)
|
||||||
}
|
}
|
||||||
if len(again) != 0 {
|
if retry := len(first) == 1 && first[0].Retry; retry != row.retry {
|
||||||
|
t.Errorf("%s: taken again %v, want %v: %+v", name, retry, row.retry, first)
|
||||||
|
}
|
||||||
|
if row.retry && (len(again) != 1 || !again[0].Retry || again[0].Said != "") || !row.retry && len(again) != 0 {
|
||||||
t.Errorf("%s: the same answer did or said something again: %+v", name, again)
|
t.Errorf("%s: the same answer did or said something again: %+v", name, again)
|
||||||
}
|
}
|
||||||
all := strings.Join(said, "\n")
|
all := strings.Join(said, "\n")
|
||||||
|
|||||||
@@ -20,6 +20,7 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"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
|
// 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
|
// (delivery.Have.SealedAt); legacy is the list a have written before slots holds (delivery.Have.Sealed), reused
|
||||||
// only within its own module.
|
// 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
|
// 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.
|
// 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)
|
Fetch func(ctx context.Context, url string) ([]byte, error)
|
||||||
@@ -102,6 +104,25 @@ type Composer struct {
|
|||||||
doing string
|
doing string
|
||||||
caught bool
|
caught bool
|
||||||
rev uint64
|
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.
|
// 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))
|
caught := make([]bool, len(ins))
|
||||||
ready := false
|
ready := false
|
||||||
dirty := 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 {
|
for {
|
||||||
// Drain whatever came, then compose once for all of it.
|
// Drain whatever came, then compose once for all of it.
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
|
case <-again:
|
||||||
|
again = nil
|
||||||
|
dirty = true
|
||||||
case t := <-merged:
|
case t := <-merged:
|
||||||
if !t.ok {
|
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))
|
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
|
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() {
|
if !c.caughtSent() {
|
||||||
select {
|
select {
|
||||||
case out <- agent.Update{Caught: true}:
|
case out <- agent.Update{Caught: true}:
|
||||||
@@ -446,6 +482,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
|
// 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).
|
// this is no refusal of what it runs (issue 1126, the seed's revert).
|
||||||
c.sayOnce("the catalogue no longer places this node: it composes nothing of its own, and keeps running what it runs")
|
c.sayOnce("the catalogue no longer places this node: 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
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
|
// Composed from the build its want names (frombus.Scope.KeptAt, issue 1126).
|
||||||
@@ -459,7 +496,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
|||||||
running = c.env.Running()
|
running = c.env.Running()
|
||||||
}
|
}
|
||||||
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: 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)
|
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
|
// 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).
|
// 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++ {
|
for try := 0; err != nil && Transient(err) && ctx.Err() == nil && try < len(composeAgain); try++ {
|
||||||
@@ -469,7 +508,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
|||||||
case <-time.After(composeAgain[try]):
|
case <-time.After(composeAgain[try]):
|
||||||
}
|
}
|
||||||
if ctx.Err() == nil {
|
if ctx.Err() == nil {
|
||||||
|
missed = c.missed.Load()
|
||||||
compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
|
compositions, _, err = frombus.Compose(ctx, r, c.env.Mesh, scope, fps)
|
||||||
|
err = c.missedIn(missed, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
@@ -481,14 +522,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.
|
// 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.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err))
|
||||||
c.refused(fmt.Sprintf("%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
|
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.
|
// 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 {
|
for key, comp := range compositions {
|
||||||
compositions[key] = c.keepRunning(comp, running)
|
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()
|
c.mu.Lock()
|
||||||
defer c.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
var out []agent.Update
|
var out []agent.Update
|
||||||
@@ -557,7 +609,8 @@ func (c *Composer) keepRunning(comp delivery.Composed, running []delivery.Compos
|
|||||||
kept := delivery.Seals{}
|
kept := delivery.Seals{}
|
||||||
var legacy []string
|
var legacy []string
|
||||||
if c.env.Held != nil {
|
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 {
|
for slot, box := range slots {
|
||||||
kept[slot] = box
|
kept[slot] = box
|
||||||
}
|
}
|
||||||
@@ -687,7 +740,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
defer r.c.waiting("the manifest " + digest + " (" + delivery.ManifestsBucket + ")")()
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -710,7 +763,7 @@ func (r reader) Build(ctx context.Context, module, fp string) (*delivery.Builder
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
defer r.c.waiting("the build of " + module + " at " + fp + " (" + delivery.BuilderBuildsBucket + ")")()
|
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 {
|
if err != nil || raw == nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -740,7 +793,7 @@ func (r reader) Archive(ctx context.Context, url, digest string) ([]byte, error)
|
|||||||
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
|
fetching, cancel := context.WithTimeout(ctx, archiveWithin)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
body, err := r.c.env.Fetch(fetching, url)
|
body, err := r.c.env.Fetch(fetching, url)
|
||||||
if err != nil {
|
if r.c.note(err) != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
sum := sha256.Sum256(body)
|
sum := sha256.Sum256(body)
|
||||||
@@ -759,7 +812,7 @@ func (r reader) Served(ctx context.Context, node, module string, keys []string)
|
|||||||
defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
|
defer r.c.waiting("what " + module + " serves on " + node + " (" + ResolvedBucket + ")")()
|
||||||
out := map[string]any{}
|
out := map[string]any{}
|
||||||
for _, k := range keys {
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -797,6 +850,7 @@ func (r reader) OwnSecrets(ctx context.Context, node string) ([]compose.OwnSecre
|
|||||||
}
|
}
|
||||||
done := r.c.waiting("the vault's own secrets")
|
done := r.c.waiting("the vault's own secrets")
|
||||||
raw, err := r.c.env.Own(ctx, map[string]any{})
|
raw, err := r.c.env.Own(ctx, map[string]any{})
|
||||||
|
r.c.note(err)
|
||||||
done()
|
done()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("the vault's own: %w", err)
|
return nil, fmt.Errorf("the vault's own: %w", err)
|
||||||
@@ -853,6 +907,7 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv
|
|||||||
}
|
}
|
||||||
done := r.c.waiting("the vault's credentials")
|
done := r.c.waiting("the vault's credentials")
|
||||||
raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
|
raw, err := r.c.env.Own(ctx, map[string]any{"credentials": ask})
|
||||||
|
r.c.note(err)
|
||||||
done()
|
done()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, fmt.Errorf("the vault's own: %w", err)
|
return nil, nil, fmt.Errorf("the vault's own: %w", err)
|
||||||
@@ -938,6 +993,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.
|
// 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}
|
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:
|
// 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
|
// 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
|
// 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 +1041,8 @@ func Transient(err error) bool {
|
|||||||
if err == nil {
|
if err == nil {
|
||||||
return false
|
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
|
return true
|
||||||
}
|
}
|
||||||
words := err.Error()
|
words := err.Error()
|
||||||
|
|||||||
@@ -7,8 +7,10 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"slices"
|
"slices"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -242,7 +244,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
|
|
||||||
// Issue 1126, the switch made safe: a bus not answering for a moment (a reload of its accounts) is tried again and
|
// 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.
|
// refuses nothing; a pass whose run ended refuses nothing either.
|
||||||
composeAgain = []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond}
|
setFor(t, &composeAgain, []time.Duration{time.Millisecond, time.Millisecond, time.Millisecond})
|
||||||
fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) {
|
fresh := func(get func(context.Context, string, string) ([]byte, error)) (*Composer, *[]string) {
|
||||||
var refusals []string
|
var refusals []string
|
||||||
e := b.env()
|
e := b.env()
|
||||||
@@ -305,6 +307,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
Transient(fmt.Errorf("the revisions of it2.mailer.domain: %w", context.DeadlineExceeded)) {
|
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")
|
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
|
// 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:
|
// 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 +435,28 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
name, in, held string
|
name, in, held string
|
||||||
slots delivery.Seals
|
slots delivery.Seals
|
||||||
kept bool
|
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 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},
|
{"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},
|
{"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},
|
{"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},
|
{"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 := env
|
||||||
restarted.Running = func() []delivery.Composed { return nil }
|
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 {
|
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 {
|
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) {
|
restarted.Unseal = func(box string) ([]byte, error) {
|
||||||
b.mu.Lock()
|
b.mu.Lock()
|
||||||
@@ -474,6 +487,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
composed = string(u.Value)
|
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())
|
||||||
|
}
|
||||||
|
// Then the catalogue no longer places the node: it composes nothing of its own, and owes no pass either, so
|
||||||
|
// no timer is armed again (rev-seed-same-3, round 1).
|
||||||
|
again.mu.Lock()
|
||||||
|
clear(again.placements)
|
||||||
|
again.mu.Unlock()
|
||||||
|
if again.compose(ctx); again.owed.Load() {
|
||||||
|
t.Errorf("%s: an unplaced node still owes a pass", row.name)
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
if composed == "" {
|
if composed == "" {
|
||||||
t.Fatalf("%s: keeper was not composed", row.name)
|
t.Fatalf("%s: keeper was not composed", row.name)
|
||||||
}
|
}
|
||||||
@@ -496,15 +523,48 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
}
|
}
|
||||||
return nil, fmt.Errorf("not sealed to n1: %s", box)
|
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 {
|
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 {
|
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)
|
t.Errorf("the node's own core box after a restart is %s, want the one its key holds %s", got, rotatedBox)
|
||||||
}
|
}
|
||||||
|
// Issue 1126, C1, the seed-7 probe: each entry of a file's secrets (catalogue's intoFile, the shape captured from it
|
||||||
|
// with the module a composition names) is a slot of its own: the box that runs at that entry is kept where it opens
|
||||||
|
// to the same value; one at another entry is not, though it opens to the same.
|
||||||
|
b.mu.Lock()
|
||||||
|
sealedAt["box-again"] = sealedAt[rotatedBox]
|
||||||
|
sealedAt["box-other"] = sealedAt[rotatedBox] + 100
|
||||||
|
b.mu.Unlock()
|
||||||
|
ini := func(db, key string) delivery.Composed {
|
||||||
|
return delivery.Composed{Node: "n1", Module: "grafana", Resources: []json.RawMessage{json.RawMessage(
|
||||||
|
`{"content":"[database]\npassword = ${secret:postgres-database}\n[security]\nsecret_key = ${secret:secret-key}\n",` +
|
||||||
|
`"id":"grafana.ini","mode":"0600","module":"grafana","path":"/etc/grafana/grafana.ini","secrets":{"postgres-database":` +
|
||||||
|
strconv.Quote(db) + `,"secret-key":` + strconv.Quote(key) + `},"type":"file"}`)}}
|
||||||
|
}
|
||||||
|
for _, row := range []struct {
|
||||||
|
name string
|
||||||
|
held delivery.Seals
|
||||||
|
want string
|
||||||
|
}{
|
||||||
|
{"the box that runs at its entry", delivery.Seals{"grafana.ini#secrets.postgres-database": rotatedBox}, rotatedBox},
|
||||||
|
{"a box at another entry", delivery.Seals{"grafana.ini#secrets.secret-key": rotatedBox}, "box-again"},
|
||||||
|
} {
|
||||||
|
secretsEnv := coreEnv
|
||||||
|
secretsEnv.Held = func(module string) (delivery.Seals, []string, error) {
|
||||||
|
if module == "grafana" {
|
||||||
|
return row.held, nil, nil
|
||||||
|
}
|
||||||
|
return nil, nil, nil
|
||||||
|
}
|
||||||
|
got := delivery.SealsIn(New(secretsEnv).keepRunning(ini("box-again", "box-other"), nil))
|
||||||
|
if got["grafana.ini#secrets.postgres-database"] != row.want || got["grafana.ini#secrets.secret-key"] != "box-other" {
|
||||||
|
t.Errorf("%s: the secrets kept are %v", row.name, got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as
|
// The pair credentials asked (ADR 0550 §5): the keys go to the vault's own with their role, a consumer's comes back as
|
||||||
// a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept.
|
// a credential held, a provider's as the grant's copy, each sealed to n1's key, the value first had kept.
|
||||||
@@ -572,7 +632,11 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
|
readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}}
|
||||||
senv := readOnly.env()
|
senv := readOnly.env()
|
||||||
senv.Watch = sb.watch
|
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) {
|
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
|
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.
|
// What n1 runs: lemurs and its own composition, as its composer composes them from the trunk.
|
||||||
@@ -721,6 +785,77 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
t.Errorf("answers handed again for nothing changed: %+v", a)
|
t.Errorf("answers handed again for nothing changed: %+v", a)
|
||||||
case <-time.After(100 * time.Millisecond):
|
case <-time.After(100 * time.Millisecond):
|
||||||
}
|
}
|
||||||
|
// Issue 1126, 2026-10-28 (shanks answered probe/1126-seed-7 at revision 84 every 13-17 s): a candidate revision is
|
||||||
|
// answered once. A module's have whose health or words alone move (each look rewrites them), and a placement, a
|
||||||
|
// module or a config value written again with the value it held, answer nothing again; a new revision answers its
|
||||||
|
// group alone, and what runs moving answers every group once.
|
||||||
|
answering := func() int {
|
||||||
|
readOnly.mu.Lock()
|
||||||
|
defer readOnly.mu.Unlock()
|
||||||
|
n := 0
|
||||||
|
for _, l := range readOnly.said {
|
||||||
|
if strings.HasPrefix(l, "answering placement candidate ") {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
drain := func() {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case answers = <-answered:
|
||||||
|
case <-time.After(150 * time.Millisecond):
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
lemursHave := func(detail string, rev uint64) {
|
||||||
|
raw, err := delivery.EncodeHave(delivery.Have{FP: fp, Composed: lemursDigest, Phase: delivery.PhaseRunning,
|
||||||
|
Health: delivery.HealthFailed, Detail: detail})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
hk, _ := delivery.HaveKey("n1", "lemurs")
|
||||||
|
sb.put(delivery.HaveBucket, hk, raw, rev)
|
||||||
|
}
|
||||||
|
lemursHave("its tool lemurs_health: no link to the bus is open", 10)
|
||||||
|
drain()
|
||||||
|
settled := answering()
|
||||||
|
lemursHave("its tool lemurs_health: no answer within 10s", 11)
|
||||||
|
lemursHave("its tool lemurs_health: no answer within 20s", 12)
|
||||||
|
sb.put(delivery.PlacementBucket, "placement.n1", tp, 2)
|
||||||
|
lraw, _ := delivery.EncodeModule(mods["lemurs"])
|
||||||
|
sb.put(delivery.ModulesBucket, "lemurs", lraw, 2)
|
||||||
|
drain()
|
||||||
|
if got := answering(); got != settled {
|
||||||
|
t.Errorf("answered %d more time(s) with no candidate revision and no input of the node changed", got-settled)
|
||||||
|
}
|
||||||
|
candidate("same", 8, "lemurs")
|
||||||
|
waitFor("the new revision answered", func(a map[string]delivery.CandidateAnswer) bool { return a["same"].Revision == 8 })
|
||||||
|
drain()
|
||||||
|
if got := answering(); got != settled+1 {
|
||||||
|
t.Errorf("a new revision of one group answered %d group(s), want 1", got-settled)
|
||||||
|
}
|
||||||
|
settled = answering()
|
||||||
|
hk, _ := delivery.HaveKey("n1", "lemurs")
|
||||||
|
moved, _ := delivery.EncodeHave(delivery.Have{FP: fp, Composed: "sha256:" + strings.Repeat("e", 64), Phase: delivery.PhaseRunning})
|
||||||
|
sb.put(delivery.HaveBucket, hk, moved, 13)
|
||||||
|
drain()
|
||||||
|
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.
|
||||||
|
setFor(t, &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")
|
k, _ := delivery.PlacementCandidateKey("same", "n1")
|
||||||
sb.del(delivery.PlacementCandidatesBucket, k)
|
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 })
|
waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 4 })
|
||||||
@@ -776,7 +911,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
|
|
||||||
// Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in
|
// Issue 1126, the shadow shows its work: one input not read after the wait is said, naming it, in the journal and in
|
||||||
// a refused answer to each group known; once read, the group is answered for real.
|
// a refused answer to each group known; once read, the group is answered for real.
|
||||||
notReadyAfter = 30 * time.Millisecond
|
setFor(t, ¬ReadyAfter, 30*time.Millisecond)
|
||||||
nb := newPatternBus()
|
nb := newPatternBus()
|
||||||
nenv := readOnly.env()
|
nenv := readOnly.env()
|
||||||
nenv.Watch = nb.watch
|
nenv.Watch = nb.watch
|
||||||
@@ -821,8 +956,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) {
|
|||||||
t.Errorf("a shadow not ready said %q", nsaid)
|
t.Errorf("a shadow not ready said %q", nsaid)
|
||||||
}
|
}
|
||||||
nmu.Unlock()
|
nmu.Unlock()
|
||||||
slowAfter = time.Nanosecond // every answer is slow: said, and noted in words that do not move with the time
|
setFor(t, &slowAfter, time.Nanosecond) // every answer is slow: said, and noted in words that do not move with the time
|
||||||
defer func() { slowAfter = time.Minute }()
|
|
||||||
nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true})
|
nb.send(delivery.AssignerWantsBucket, agent.Update{Caught: true})
|
||||||
select {
|
select {
|
||||||
case a := <-nanswered:
|
case a := <-nanswered:
|
||||||
@@ -1183,3 +1317,12 @@ func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// setFor sets a package global (a wait the composer or the shadow reads) for one test, and puts back what it held when
|
||||||
|
// the test ends, so no later test runs on another test's waits.
|
||||||
|
func setFor[T any](t *testing.T, p *T, v T) {
|
||||||
|
t.Helper()
|
||||||
|
old := *p
|
||||||
|
*p = v
|
||||||
|
t.Cleanup(func() { *p = old })
|
||||||
|
}
|
||||||
|
|||||||
+140
-16
@@ -4,6 +4,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"reflect"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -39,6 +40,22 @@ type Shadow struct {
|
|||||||
// ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy
|
// ran is, by module, the digest the module's have says runs: what runs, where the agent's list holds only the copy
|
||||||
// from the declaration kept before the switch (issue 1126, probe #532 on shanks).
|
// from the declaration kept before the switch (issue 1126, probe #532 on shanks).
|
||||||
ran map[string]string
|
ran map[string]string
|
||||||
|
// inputs counts the changes of the node's own inputs (everything but the candidates), and answered is, by group, the
|
||||||
|
// candidate revisions and the count each answer was composed at: a group is answered again only when either moved
|
||||||
|
// (issue 1126, 2026-10-28: shanks answered probe/1126-seed-7 at revision 84 every 13-17 s, woken by its own modules'
|
||||||
|
// 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.
|
||||||
|
type answeredAt struct {
|
||||||
|
revs string
|
||||||
|
inputs uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
// candidateKey is one candidate key as last read: its value and revision.
|
// candidateKey is one candidate key as last read: its value and revision.
|
||||||
@@ -94,6 +111,7 @@ func (s *Shadow) Run(ctx context.Context) error {
|
|||||||
s.candidates = map[string]map[string]candidateKey{}
|
s.candidates = map[string]map[string]candidateKey{}
|
||||||
s.wants = map[string]delivery.Want{}
|
s.wants = map[string]delivery.Want{}
|
||||||
s.ran = map[string]string{}
|
s.ran = map[string]string{}
|
||||||
|
s.answered = map[string]answeredAt{}
|
||||||
s.c = New(s.c.env)
|
s.c = New(s.c.env)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
c := s.c
|
c := s.c
|
||||||
@@ -140,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).
|
// knows: a shadow that waits is never silent (issue 1126: ace, g14 and shanks answered nothing to seed #531).
|
||||||
waiting := time.NewTimer(notReadyAfter)
|
waiting := time.NewTimer(notReadyAfter)
|
||||||
defer waiting.Stop()
|
defer waiting.Stop()
|
||||||
|
var again <-chan time.Time
|
||||||
|
tries := 0
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return nil
|
return nil
|
||||||
|
case <-again:
|
||||||
|
again = nil
|
||||||
|
dirty = true
|
||||||
case <-waiting.C:
|
case <-waiting.C:
|
||||||
if !ready {
|
if !ready {
|
||||||
s.notReady(ctx, ins, caught)
|
s.notReady(ctx, ins, caught)
|
||||||
@@ -153,21 +176,29 @@ func (s *Shadow) Run(ctx context.Context) error {
|
|||||||
if !t.ok {
|
if !t.ok {
|
||||||
return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket)
|
return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket)
|
||||||
}
|
}
|
||||||
|
// Only a change of what an answer is composed from answers again: a key written again with the value it held,
|
||||||
|
// this node's own key (its answers), or a module's have whose health or words moved but not what it runs, is
|
||||||
|
// none (issue 1126, 2026-10-28: each look rewrote shanks' unhealthy modules' haves, and each woke a full answer).
|
||||||
|
changed := false
|
||||||
switch {
|
switch {
|
||||||
case t.u.Caught:
|
case t.u.Caught:
|
||||||
caught[t.in] = true
|
caught[t.in] = true
|
||||||
case t.in == 0:
|
case t.in == 0:
|
||||||
dirty = s.takeCandidate(t.u) || dirty
|
changed = s.takeCandidate(t.u)
|
||||||
case t.in == 2:
|
case t.in == 2:
|
||||||
dirty = s.takeRecord(t.u) || dirty
|
changed = s.takeRecord(t.u)
|
||||||
case t.in == 5:
|
case t.in == 5:
|
||||||
c.takeModuleHave(t.u)
|
folded := c.takeModuleHave(t.u)
|
||||||
s.takeRan(t.u)
|
changed = s.takeRan(t.u) || folded
|
||||||
dirty = !isNodeKey(t.u.Key) || dirty
|
|
||||||
case t.in == 6:
|
case t.in == 6:
|
||||||
dirty = s.takeWant(t.u) || dirty
|
changed = s.takeWant(t.u)
|
||||||
default:
|
default:
|
||||||
ins[t.in].take(c, t.u)
|
changed = s.takeChanged(ins[t.in], t.u)
|
||||||
|
}
|
||||||
|
if changed {
|
||||||
|
if t.in != 0 {
|
||||||
|
s.inputs++ // the candidates' own changes are their revisions
|
||||||
|
}
|
||||||
dirty = true
|
dirty = true
|
||||||
}
|
}
|
||||||
if len(merged) > 0 {
|
if len(merged) > 0 {
|
||||||
@@ -183,13 +214,77 @@ func (s *Shadow) Run(ctx context.Context) error {
|
|||||||
if dirty {
|
if dirty {
|
||||||
dirty = false
|
dirty = false
|
||||||
s.answerAll(ctx)
|
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
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func isNodeKey(key string) bool {
|
// takeChanged takes u through in.take and says whether the entry it names changed: a placement, a module or a config
|
||||||
_, _, isNode, err := delivery.ParseHaveKey(key)
|
// value written again with the value it held changes nothing composed.
|
||||||
return err == nil && isNode
|
func (s *Shadow) takeChanged(in input, u agent.Update) bool {
|
||||||
|
before, had, known := s.entry(in.bucket, u.Key)
|
||||||
|
in.take(s.c, u)
|
||||||
|
if !known {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
after, has, _ := s.entry(in.bucket, u.Key)
|
||||||
|
return had != has || !reflect.DeepEqual(before, after)
|
||||||
|
}
|
||||||
|
|
||||||
|
// entry is what the composer holds of one key of bucket, whether it holds it, and whether the bucket is one it knows.
|
||||||
|
func (s *Shadow) entry(bucket, key string) (v any, has, known bool) {
|
||||||
|
c := s.c
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
switch bucket {
|
||||||
|
case delivery.PlacementBucket:
|
||||||
|
node, err := delivery.ParsePlacementKey(key)
|
||||||
|
if err != nil {
|
||||||
|
return nil, false, true
|
||||||
|
}
|
||||||
|
p, ok := c.placements[node]
|
||||||
|
return p, ok, true
|
||||||
|
case delivery.ModulesBucket:
|
||||||
|
name, err := delivery.ParseModulesKey(key)
|
||||||
|
if err != nil {
|
||||||
|
return nil, false, true
|
||||||
|
}
|
||||||
|
m, ok := c.modules[name]
|
||||||
|
return m, ok, true
|
||||||
|
case ResolvedBucket:
|
||||||
|
rest, ok := strings.CutPrefix(key, c.env.Node+".")
|
||||||
|
if !ok {
|
||||||
|
return nil, false, true
|
||||||
|
}
|
||||||
|
module, k, ok := strings.Cut(rest, ".")
|
||||||
|
if !ok {
|
||||||
|
return nil, false, true
|
||||||
|
}
|
||||||
|
val, ok := c.values[module][k]
|
||||||
|
return val, ok, true
|
||||||
|
}
|
||||||
|
return nil, false, false
|
||||||
|
}
|
||||||
|
|
||||||
|
// revsOf is a group's keys' revisions, by node, as one word.
|
||||||
|
func revsOf(keys map[string]candidateKey) string {
|
||||||
|
nodes := make([]string, 0, len(keys))
|
||||||
|
for n := range keys {
|
||||||
|
nodes = append(nodes, n)
|
||||||
|
}
|
||||||
|
sort.Strings(nodes)
|
||||||
|
var b strings.Builder
|
||||||
|
for _, n := range nodes {
|
||||||
|
fmt.Fprintf(&b, "%s=%d;", n, keys[n].rev)
|
||||||
|
}
|
||||||
|
return b.String()
|
||||||
}
|
}
|
||||||
|
|
||||||
// takeCandidate takes one candidate key, and says whether it changed anything.
|
// takeCandidate takes one candidate key, and says whether it changed anything.
|
||||||
@@ -289,15 +384,22 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
|||||||
type listed struct {
|
type listed struct {
|
||||||
group string
|
group string
|
||||||
rev uint64
|
rev uint64
|
||||||
|
at answeredAt
|
||||||
}
|
}
|
||||||
var groups []listed
|
var groups []listed
|
||||||
for g, keys := range s.candidates {
|
for g, keys := range s.candidates {
|
||||||
l := listed{group: g}
|
l := listed{group: g, at: answeredAt{revs: revsOf(keys), inputs: s.inputs}}
|
||||||
for _, k := range keys {
|
for _, k := range keys {
|
||||||
l.rev = max(l.rev, k.rev)
|
l.rev = max(l.rev, k.rev)
|
||||||
}
|
}
|
||||||
groups = append(groups, l)
|
groups = append(groups, l)
|
||||||
}
|
}
|
||||||
|
// A group whose keys' revisions and the node's inputs are those its answer stands at is not answered again.
|
||||||
|
for g := range s.answered {
|
||||||
|
if _, ok := s.candidates[g]; !ok {
|
||||||
|
delete(s.answered, g)
|
||||||
|
}
|
||||||
|
}
|
||||||
// Each group's last answer stands until it is answered again; a group whose keys went is dropped.
|
// Each group's last answer stands until it is answered again; a group whose keys went is dropped.
|
||||||
next := map[string]delivery.CandidateAnswer{}
|
next := map[string]delivery.CandidateAnswer{}
|
||||||
for _, l := range groups {
|
for _, l := range groups {
|
||||||
@@ -305,6 +407,7 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
|||||||
next[l.group] = a
|
next[l.group] = a
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
s.owed = false
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
sort.Slice(groups, func(i, j int) bool {
|
sort.Slice(groups, func(i, j int) bool {
|
||||||
if groups[i].rev != groups[j].rev {
|
if groups[i].rev != groups[j].rev {
|
||||||
@@ -342,6 +445,12 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
group := l.group
|
group := l.group
|
||||||
|
s.mu.Lock()
|
||||||
|
at, done := s.answered[group]
|
||||||
|
s.mu.Unlock()
|
||||||
|
if done && at == l.at {
|
||||||
|
continue
|
||||||
|
}
|
||||||
began := time.Now()
|
began := time.Now()
|
||||||
c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev))
|
c.env.Say(fmt.Sprintf("answering placement candidate %s (revision %d)", group, l.rev))
|
||||||
stop := make(chan struct{})
|
stop := make(chan struct{})
|
||||||
@@ -358,8 +467,10 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
missed := c.missed.Load()
|
||||||
a, ok := s.answerGroup(ctx, group)
|
a, ok := s.answerGroup(ctx, group)
|
||||||
close(stop)
|
close(stop)
|
||||||
|
provisional := c.missed.Load() > missed && ctx.Err() == nil
|
||||||
took := time.Since(began)
|
took := time.Since(began)
|
||||||
if took > slowAfter {
|
if took > slowAfter {
|
||||||
c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second)))
|
c.env.Say(fmt.Sprintf("the answer to %s took %s", group, took.Round(time.Second)))
|
||||||
@@ -376,6 +487,16 @@ func (s *Shadow) answerAll(ctx context.Context) {
|
|||||||
delete(next, group)
|
delete(next, group)
|
||||||
c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group))
|
c.env.Say(fmt.Sprintf("placement candidate %s does not place this node: nothing to answer", group))
|
||||||
}
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
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()
|
hand()
|
||||||
}
|
}
|
||||||
hand() // a group whose keys went, when no group is left to answer
|
hand() // a group whose keys went, when no group is left to answer
|
||||||
@@ -428,7 +549,7 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
|||||||
wants[m] = w
|
wants[m] = w
|
||||||
}
|
}
|
||||||
s.mu.Unlock()
|
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]
|
placement := given[node]
|
||||||
@@ -470,20 +591,23 @@ func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.Candid
|
|||||||
return asHavesSay(a, ran), true
|
return asHavesSay(a, ran), true
|
||||||
}
|
}
|
||||||
|
|
||||||
// takeRan keeps the digest one of this node's module haves says runs.
|
// takeRan keeps the digest one of this node's module haves says runs, and says whether it changed: a have whose health
|
||||||
func (s *Shadow) takeRan(u agent.Update) {
|
// or words alone moved runs the same.
|
||||||
|
func (s *Shadow) takeRan(u agent.Update) bool {
|
||||||
node, module, isNode, err := delivery.ParseHaveKey(u.Key)
|
node, module, isNode, err := delivery.ParseHaveKey(u.Key)
|
||||||
if err != nil || isNode || node != s.c.env.Node {
|
if err != nil || isNode || node != s.c.env.Node {
|
||||||
return
|
return false
|
||||||
}
|
}
|
||||||
s.mu.Lock()
|
s.mu.Lock()
|
||||||
defer s.mu.Unlock()
|
defer s.mu.Unlock()
|
||||||
|
before, had := s.ran[module]
|
||||||
h, err := delivery.DecodeHave(u.Value)
|
h, err := delivery.DecodeHave(u.Value)
|
||||||
if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved {
|
if u.Deleted || err != nil || h.Composed == "" || h.Phase == delivery.PhaseRemoved {
|
||||||
delete(s.ran, module)
|
delete(s.ran, module)
|
||||||
return
|
return had
|
||||||
}
|
}
|
||||||
s.ran[module] = h.Composed
|
s.ran[module] = h.Composed
|
||||||
|
return !had || before != h.Composed
|
||||||
}
|
}
|
||||||
|
|
||||||
// asHavesSay is a with each module that differs only from the copy the agent keeps of the declaration before the switch
|
// asHavesSay is a with each module that differs only from the copy the agent keeps of the declaration before the switch
|
||||||
|
|||||||
Reference in New Issue
Block a user