Let only a watch that read every standing assignment be the start's look (issue 1945, ADR 0543, review)

The flag that made the first watch the start's look was cleared after every watch, even one that failed to open or ended before its Caught mark: the next watch then handed every standing assignment as fresh, about 300 settles at once with a catalogue read each, or dropped the keys the cut watch had handed. The flag is now cleared by the start's look itself, and a watch cut before it opens the next with nothing heard. Two rows of the existing table test cover it. Research 072: the grants piece (memberships and credentials heard from placement).
2026-10-25 18:07:21 +01:00
parent 9ff47f6aaa
commit 3e39ccf958
2 changed files with 86 additions and 28 deletions
@@ -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
-		for {
-			var atStart []chain.Assignment
-			if err := watchAssignmentsOnce(ctx, open.inventory.WatchAssignments, turns, first, func(a chain.Assignment, fresh bool) {
-				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)
-					})
+		runAssignmentWatches(ctx, open.inventory.WatchAssignments, func(a chain.Assignment) {
+			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,
-				// then one composition of the bus's node and the grants (issues 1945 and 1277).
-				look := atStart
-				atStart = nil
-				inBackground(ctx, func() {
-					untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
-						return steps.look(ctx, look)
-					})
+			})
+		}, func(look []chain.Assignment) {
+			// The start's look, once in the process: one reading of the catalogue, one assignment after another,
+			// then one composition of the bus's node and the grants (issues 1945 and 1277).
+			inBackground(ctx, func() {
+				untilSettled(ctx, "the start's look", func(ctx context.Context) (string, error) {
+					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)
-			}
-			first = false
+			})
+		}, func() bool {
 			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
@@ -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