diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 572df1b..ba1e5ae 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -896,7 +896,7 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil, err.Error() } env := composerEnv(session, node, say, hooks) - placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say) + placed, err := selfcompose.Placed(ctx, env.Get, node, link.Backoff, say, composedItsOwn(ctx, env.Get, node)) if err != nil { return nil, "its placement could not be read before the link ended: " + err.Error() } @@ -910,6 +910,24 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return selfcompose.New(env), why } +// composedItsOwn says whether the node's own have, kept on the bus, says it composed from its placement key at its last +// start. A have that cannot be read says yes: the safe side, a node waits and does not fall back. +func composedItsOwn(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string) bool { + key, err := delivery.HaveNodeKey(node) + if err != nil { + return true + } + raw, err := get(ctx, delivery.HaveBucket, key) + if err != nil { + return true + } + if len(raw) == 0 { + return false + } + n, err := delivery.DecodeHaveNode(raw) + return err != nil || n.Composes == delivery.ComposesOwn +} + // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to placement candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 5631a22..6781bee 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -993,8 +993,14 @@ func (r reader) Credentials(ctx context.Context, node string, asConsumer, asProv // marker. A read that fails chooses neither (review of the shadow, round 1: a timeout or a bus not ready at a restart // flipped a self-composing node to the controller's compositions for a whole run): it is tried again after wait(try), // each distinct failure said once, until ctx ends, which is then its error. +// +// switched says the node composed from its own placement key before (its have says it composes its own, +// delivery.ComposesOwn, and that is kept on the bus, so it survives a restart). Only a node that never switched takes a +// read that fails (a timeout, or a state its account may not read yet, as the seat's before its grant arrives) as "not +// placed": it stays on the controller's compositions, says why once, and switches on a positive read of its key (the +// watch endWhenPlaced keeps). A node that switched never goes back on a failed read, however long it lasts. func Placed(ctx context.Context, get func(ctx context.Context, bucket, key string) ([]byte, error), node string, - wait func(try int) time.Duration, say func(string)) (bool, error) { + wait func(try int) time.Duration, say func(string), switched bool) (bool, error) { key, err := delivery.PlacementKey(node) if err != nil { return false, err @@ -1011,6 +1017,10 @@ func Placed(ctx context.Context, get func(ctx context.Context, bucket, key strin if ctx.Err() != nil { return false, ctx.Err() } + if !switched { + say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): it has never composed its own, so it applies the controller's compositions and switches when its key is read: %v", key, err)) + return false, nil + } if why := err.Error(); why != said { say(fmt.Sprintf("whether the catalogue places this node cannot be read (%s): the node agent waits, choosing neither its own compositions nor the controller's: %v", key, err)) said = why diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 75e87d8..98c1ec2 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -957,12 +957,15 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { placed bool tries int quiet bool // read and known: nothing to say + // switched: the node's have says it composed its own before (it never falls back on a failed read). + switched bool }{ - "placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3}, - "no key": {placed: false, tries: 1}, - "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, - "failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), - fmt.Errorf("jetstream not ready")}, end: true, tries: 3}, + "switched: placed after two failures": {answers: []error{fmt.Errorf("nats: timeout"), fmt.Errorf("nats: timeout")}, value: []byte(`{}`), placed: true, tries: 3, switched: true}, + "never switched: a timeout is not placed, said once": {answers: []error{fmt.Errorf("nats: timeout")}, placed: false, tries: 1}, + "no key": {placed: false, tries: 1}, + "no placement state yet": {answers: []error{fmt.Errorf("the bucket %s: %w", delivery.PlacementBucket, nats.ErrBucketNotFound)}, placed: false, tries: 1, quiet: true}, + "switched: failing until the link ends": {answers: []error{fmt.Errorf("jetstream not ready"), fmt.Errorf("jetstream not ready"), + fmt.Errorf("jetstream not ready")}, end: true, tries: 3, switched: true}, } { lctx, lcancel := context.WithCancel(ctx) tries := 0 @@ -977,7 +980,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } return row.value, nil } - placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }) + placed, err := Placed(lctx, get, "n1", func(int) time.Duration { return time.Millisecond }, func(s string) { said = append(said, s) }, row.switched) lcancel() if placed != row.placed || tries != row.tries || (err != nil) != row.end || (len(row.answers) > 0 && !row.quiet) != (len(said) == 1) { t.Errorf("%s: placed %v, err %v after %d reads, said %q", name, placed, err, tries, said)