diff --git a/cmd/mesh-controller/assignment_accounts.go b/cmd/mesh-controller/assignment_accounts.go index 52f80515..8c6ba982 100644 --- a/cmd/mesh-controller/assignment_accounts.go +++ b/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 - 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 diff --git a/cmd/mesh-controller/assignment_accounts_test.go b/cmd/mesh-controller/assignment_accounts_test.go index b0886dff..277e8c8d 100644 --- a/cmd/mesh-controller/assignment_accounts_test.go +++ b/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