Node agent: a node that never composed its own takes a failed placement read as not placed; one that did keeps waiting (issue 1126)

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 {

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

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)