Merge branch 'fix/1945-credential-on-assignment' into fix/1945-take-in-consumer

cmd/mesh-controller/assignment_accounts.go

@@ -282,44 +282,65 @@ func untilSettled(ctx context.Context, what string, settle func(context.Context)
 // bucket not there yet, is watched again a minute later.
 func watchAssignmentAccounts(ctx context.Context, open *stores) {
 	steps := liveAssignmentAccountSteps(open)
-	turns := &assignmentTurns{}
 	goServing(ctx, func() {
-		first := true
+		runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
-		for {
+			inBackground(ctx, func() {
-			var atStart []chain.Assignment
+				untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
-			if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, first, func(a chain.Assignment, fresh bool) {
+					return steps.settle(ctx, a.Node, a.Module)
-				if !fresh {
-					atStart = append(atStart, a)
-					return
-				}
-				inBackground(ctx, func() {
-					untilSettled(ctx, a.Module+" on "+a.Node, func(ctx context.Context) (string, error) {
-						return steps.settle(ctx, a.Node, a.Module)
-					})
 				})
-			}, func() {
+			})
-				// The start's look, once in the process: one reading of the catalogue, one assignment after another,
+		}, func(look []chain.Assignment) {
-				// then one composition of the bus's node and the grants (issues 1945 and 1277).
+			// The start's look, once in the process: one reading of the catalogue, one assignment after another,
-				look := atStart
+			// then one composition of the bus's node and the grants (issues 1945 and 1277).
-				atStart = nil
+			inBackground(ctx, func() {
-				inBackground(ctx, func() {
+				untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
-					untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
+					return steps.look(ctx, look)
-						return steps.look(ctx, look)
-					})
 				})
-			}); err != nil && ctx.Err() == nil {
+			})
-				fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
+		}, func() bool {
-			}
-			first = false
 			select {
 			case <-ctx.Done():
-				return
+				return false
 			case <-time.After(time.Minute):
+				return true
 			}
-		}
+		})
 	})
 }
 
+// runAssignmentWatches watches the assignments again each time a watch ends, until ctx ends or pause says stop: each
+// assignment that comes to stand is handed to fresh, and the standing ones of the first watch that reads them all (its
+// Caught mark) to look, once. A watch that ends, or fails to open, before the start's look ran is no start's look: the
+// next one is, with what it hears from nothing (a fresh assignmentTurns), so no key it handed is dropped and no burst of
+// fresh settles replaces the look (the review of issue 1945).
+func runAssignmentWatches(ctx context.Context, watch func(context.Context) (<-chan chain.Entry, error),
+	fresh func(chain.Assignment), look func([]chain.Assignment), pause func() bool) {
+	turns := &assignmentTurns{}
+	first := true
+	for {
+		var atStart []chain.Assignment
+		if err := watchAssignmentsOnce(ctx, watch, turns, first, func(a chain.Assignment, isFresh bool) {
+			if !isFresh {
+				atStart = append(atStart, a)
+				return
+			}
+			fresh(a)
+		}, func() {
+			first = false
+			look(atStart)
+			atStart = nil
+		}); err != nil && ctx.Err() == nil {
+			fmt.Printf("assignments: the assignments are not heard, so a new assignment's bus accounts wait: %v\n", err)
+		}
+		if first {
+			turns = &assignmentTurns{} // the start's look has not run: the next watch hears every standing key anew
+		}
+		if ctx.Err() != nil || !pause() {
+			return
+		}
+	}
+}
+
 // watchAssignmentsOnce watches the assignments until the watch or ctx ends, handing each assignment that comes to stand
 // to settle. On the process's first watch, the current values are the start's look (not fresh) and read is called once
 // they are all handed (the chain's Caught mark); every other value, and every value of a watch opened again after one

cmd/mesh-controller/assignment_accounts_test.go

@@ -135,6 +135,43 @@ func TestAnAssignmentThatComesToStand(t *testing.T) {
 			}
 		})
 	}
+	// A first watch that fails to open, or ends before its Caught mark, is no start's look: the next watch is, and hands
+	// every standing assignment to it once, none as fresh (the review of issue 1945: a burst of fresh settles, or keys
+	// handed and dropped).
+	for _, firstWatch := range []string{"fails to open", "ends before its Caught mark"} {
+		t.Run("a first watch that "+firstWatch, func(t *testing.T) {
+			bucket := chain.NewMemory()
+			standing := chain.Assignment{Node: "g14", Module: "mesh-build-go", Assigned: at}
+			if _, err := bucket.Create(t.Context(), standing.Key(), anAssignmentValue(t, standing)); err != nil {
+				t.Fatal(err)
+			}
+			watches := 0
+			watch := func(ctx context.Context) (<-chan chain.Entry, error) {
+				watches++
+				if watches > 1 {
+					return bucket.Watch(ctx)
+				}
+				if firstWatch == "fails to open" {
+					return nil, errors.New("the chain is not open yet")
+				}
+				cut := make(chan chain.Entry, 1)
+				cut <- chain.Entry{Key: standing.Key(), Value: anAssignmentValue(t, standing)}
+				close(cut)
+				return cut, nil
+			}
+			var looks [][]chain.Assignment
+			var fresh []chain.Assignment
+			ctx, stop := context.WithCancel(t.Context())
+			defer stop()
+			runAssignmentWatches(ctx, watch, func(a chain.Assignment) { fresh = append(fresh, a) },
+				func(look []chain.Assignment) { looks = append(looks, look); stop() },
+				func() bool { return ctx.Err() == nil })
+			if len(looks) != 1 || len(looks[0]) != 1 || looks[0][0].Key() != standing.Key() || len(fresh) != 0 {
+				t.Fatalf("after %d watch(es): looks %v, fresh %v; want one look at %s and nothing fresh",
+					watches, looks, fresh, standing.Key())
+			}
+		})
+	}
 }
 
 // settle issues what is missing and then puts it on the bus: the node holding the bus composed (its user list) and the