package main import ( "context" "slices" "sort" "sync" "testing" "time" ) // A burst of done keys composes each node once per pass and writes the grants once, never two passes at a time: the // take-ins of a burst each composed every node of their module at once, collided on the compositions' revisions and // retried every 30 s, while the one that mattered (mesh-host's) waited (issue 1945, live on 2026-10-25 at 11:18 and // 12:32). func TestABurstComposesEachNodeOncePerPass(t *testing.T) { var mu sync.Mutex var passes [][]string grants := 0 running, maxRunning := 0, 0 release := make(chan struct{}) first := true c := newComposer(func(_ context.Context, nodes []string) map[string]error { mu.Lock() running++ maxRunning = max(maxRunning, running) passes = append(passes, slices.Clone(nodes)) wait := first first = false mu.Unlock() if wait { <-release // the first pass runs while the rest of the burst arrives } mu.Lock() running-- mu.Unlock() return nil }, func(context.Context) error { mu.Lock() grants++ mu.Unlock() return nil }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go c.run(ctx) var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done(); _ = c.compose(ctx, []string{"ace"}, true) }() time.Sleep(20 * time.Millisecond) for _, nodes := range [][]string{{"ace", "g14"}, {"g14"}, {"shanks", "ace"}, {"novox"}, {"ace"}} { wg.Add(1) go func() { defer wg.Done(); _ = c.compose(ctx, nodes, true) }() } time.Sleep(20 * time.Millisecond) close(release) wg.Wait() mu.Lock() defer mu.Unlock() if maxRunning != 1 { t.Fatalf("%d passes ran at once", maxRunning) } if len(passes) != 2 { t.Fatalf("passes %v: want the first and one more for the rest of the burst", passes) } second := slices.Clone(passes[1]) sort.Strings(second) if !slices.Equal(second, []string{"ace", "g14", "novox", "shanks"}) { t.Fatalf("the second pass composed %v: each node once", passes[1]) } if grants != 2 { t.Fatalf("the grants were written %d times for two passes", grants) } } // A caller hears its own pass's failure, and only a pass that began after it asked counts as its own. func TestACallerHearsThePassThatComposedItsNodes(t *testing.T) { calls := 0 c := newComposer(func(_ context.Context, nodes []string) map[string]error { calls++ if calls == 1 { return map[string]error{"ace": context.DeadlineExceeded} } return nil }, func(context.Context) error { return nil }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go c.run(ctx) if err := c.compose(ctx, []string{"ace"}, false); err == nil { t.Fatal("a failed pass was not heard") } if err := c.compose(ctx, []string{"ace"}, false); err != nil { t.Fatalf("a good pass said %v", err) } } // One node that fails its composition fails only the callers that asked for it, and the grants are written whenever a // caller asked for them (the review of issue 1945: one joined error failed every take-in of the pass and stopped their // grants). func TestANodeThatFailsFailsOnlyItsCallers(t *testing.T) { grants := 0 release := make(chan struct{}) first := true c := newComposer(func(_ context.Context, nodes []string) map[string]error { if first { first = false <-release } out := map[string]error{} for _, n := range nodes { if n == "broken" { out[n] = context.DeadlineExceeded } } return out }, func(context.Context) error { grants++; return nil }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() go c.run(ctx) go func() { _ = c.compose(ctx, []string{"warm"}, false) }() time.Sleep(20 * time.Millisecond) errs := make(chan error, 2) go func() { errs <- c.compose(ctx, []string{"ace"}, true) }() go func() { errs <- c.compose(ctx, []string{"broken"}, false) }() time.Sleep(20 * time.Millisecond) close(release) var failed, fine int for range 2 { if err := <-errs; err != nil { failed++ } else { fine++ } } if failed != 1 || fine != 1 { t.Fatalf("failed %d, fine %d: only the caller of the broken node fails", failed, fine) } if grants != 1 { t.Fatalf("the grants were written %d times; a caller asked for them", grants) } }