From 1989dd7cf2f645f22385f6e030a1964a6ca48d6d Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 00:48:50 +0100 Subject: [PATCH] mesh-build-go: start the integration level's store and bus beside its compile, and compile the go.mod directories side by side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The setup started the store, then the bus, then compiled each directory one after another: 132 s of setup in the L3 group check. Now the three start together and the directories compile up to the bound at once; the compile runs no test binary (-exec true), so it needs neither server and is given nothing a test reads. Ported from feat/try-speed S5 (issue 1921; ADR 0541 §1). --- .../cmd/mesh-build-go/integration.go | 72 +++++++++--- .../mesh-build-go/integration_level_test.go | 15 ++- .../cmd/mesh-build-go/preview_test.go | 4 +- .../cmd/mesh-build-go/selected_dirs_test.go | 104 ++++++++++++++---- 4 files changed, 153 insertions(+), 42 deletions(-) diff --git a/modules/mesh-build-go/cmd/mesh-build-go/integration.go b/modules/mesh-build-go/cmd/mesh-build-go/integration.go index 62b4d0d1..1c89d4fb 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/integration.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/integration.go @@ -12,6 +12,7 @@ import ( "path/filepath" "strconv" "strings" + "sync" "time" "git.novox.be/novox/mesh-sdk/go/buildstep" @@ -62,30 +63,73 @@ func (m *Module) integrationHere(ctx context.Context, workspace, tree string, pa if err != nil { return nil, err } - postgres, stopStore, err := m.startStore(ctx, run) - if err != nil { - return nil, fmt.Errorf("the integration level's store: %w", err) - } - defer stopStore() - nats, stopBus, err := m.startBus(ctx, filepath.Join(run, "nats")) - if err != nil { - return nil, fmt.Errorf("the integration level's bus: %w", err) - } - defer stopBus() - env := append([]string{"MESH_TEST_POSTGRES=" + postgres, "MESH_TEST_NATS=" + nats}, m.checkEnv(workspace, nil)...) - // Compiling counts as setup: the tests are built once, here, with the addresses a TestMain may read, and the run - // below finds them built. Whether they compile is for the run to say; its failure is not read here. // Only the directories the selected packages lie in compile (ADR 0536 §3); no selection is every directory. byDir, err := packagesByDir(tree, dirs, packages) if err != nil { return nil, err } + // The store, the bus and the compile start together, and the directories compile side by side up to the bound + // (issue 1921, try speed S5): the setup is the longest of them, not their sum. The compile runs no test binary + // (-exec true), so it needs neither server and is given nothing a test reads; whether the tests compile is for the + // run to say, its failure is not read here, and the run below finds them built. + var ( + wg sync.WaitGroup + postgres, nats string + stopStore, stopBus func() + storeErr, busErr error + crash any + crashMu sync.Mutex + ) + guard := func() { + if p := recover(); p != nil { + crashMu.Lock() + crash = p + crashMu.Unlock() + } + } + wg.Add(2) + go func() { + defer wg.Done() + defer guard() + postgres, stopStore, storeErr = m.startStore(ctx, run) + }() + go func() { + defer wg.Done() + defer guard() + nats, stopBus, busErr = m.startBus(ctx, filepath.Join(run, "nats")) + }() + slots := make(chan struct{}, m.parallel()) for _, d := range dirs { if len(byDir[d]) == 0 { continue } - m.Run(ctx, append(m.goEnvFor(filepath.Join(tree, d)), env...), "go", append([]string{"-C", filepath.Join(tree, d), "test", "-tags", strings.Join(integrationTags, ","), "-run", "^$", "-count=1"}, byDir[d]...)...) + wg.Add(1) + slots <- struct{}{} + go func() { + defer wg.Done() + defer func() { <-slots }() + defer guard() + m.Run(ctx, m.goEnvFor(filepath.Join(tree, d)), "go", append([]string{"-C", filepath.Join(tree, d), "test", "-tags", strings.Join(integrationTags, ","), "-exec", "true", "-count=1"}, byDir[d]...)...) + }() } + wg.Wait() + // Whatever started is stopped however the level ends, a half-started store included. + if stopStore != nil { + defer stopStore() + } + if stopBus != nil { + defer stopBus() + } + if crash != nil { + panic(crash) + } + if storeErr != nil { + return nil, fmt.Errorf("the integration level's store: %w", storeErr) + } + if busErr != nil { + return nil, fmt.Errorf("the integration level's bus: %w", busErr) + } + env := append([]string{"MESH_TEST_POSTGRES=" + postgres, "MESH_TEST_NATS=" + nats}, m.checkEnv(workspace, nil)...) setup := m.clock().Sub(began) return m.runLevel(ctx, tree, levelOpts{ diff --git a/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go b/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go index fafb7bda..19a1873b 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go @@ -35,6 +35,9 @@ type integFake struct { // command runs, with its line. gomods string onTest func(line string) + // onCompile is called, unlocked, as each compile of the setup runs, and onStart as the store starts (issue 1921, S5). + onCompile func(line string, env []string) + onStart func() } func newIntegFake(t *testing.T, captured string) *integFake { @@ -71,7 +74,10 @@ func (f *integFake) run(ctx context.Context, env []string, name string, args ... return forgekey.Ran{Stdout: f.gomods} } return forgekey.Ran{Stdout: "go.mod\n"} - case name == "go" && strings.Contains(line, "-run ^$"): + case name == "go" && strings.Contains(line, "-exec true"): + if f.onCompile != nil { + f.onCompile(line, env) + } f.tick("compile") case name == "go": if f.onTest != nil { @@ -84,6 +90,9 @@ func (f *integFake) run(ctx context.Context, env []string, name string, args ... } return forgekey.Ran{Stdout: f.captured, Status: status} case name == "pg_ctl" && strings.HasSuffix(line, " start"): + if f.onStart != nil { + f.onStart() + } f.tick("start") for i, a := range args { if a == "-D" { @@ -314,8 +323,8 @@ func TestAStoreThatDoesNotStartEndsTheStepFailedAndLeavesNothing(t *testing.T) { if !strings.Contains(f.lines(), " stop") { t.Errorf("the half-started store was not stopped:\n%s", f.lines()) } - if f.busDir != "" { - t.Errorf("the bus was started after the store failed") + if f.busDir != "" && !f.busDown { + t.Errorf("the bus, started beside the store, was not stopped after the store failed") } if entries, _ := os.ReadDir(filepath.Join(m.Work, "ws1")); len(entries) != 1 || entries[0].Name() != "tree" { t.Errorf("the workspace holds %v, want only its tree", entries) diff --git a/modules/mesh-build-go/cmd/mesh-build-go/preview_test.go b/modules/mesh-build-go/cmd/mesh-build-go/preview_test.go index 6df51b81..62b88a2d 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/preview_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/preview_test.go @@ -290,8 +290,8 @@ func TestTheIntegrationLevelIsGivenTheBesideDirectoryOnlyWhenSiblingsWerePlaced( } tested := 0 for _, c := range f.calls { - if !strings.HasPrefix(c.line, "go ") || !strings.Contains(c.line, "-tags integration") { - continue + if !strings.HasPrefix(c.line, "go ") || !strings.Contains(c.line, "-tags integration") || strings.Contains(c.line, "-exec true") { + continue // the setup's compile runs no test binary, so it is given nothing a test reads (issue 1921, S5) } tested++ got := "" diff --git a/modules/mesh-build-go/cmd/mesh-build-go/selected_dirs_test.go b/modules/mesh-build-go/cmd/mesh-build-go/selected_dirs_test.go index fa86bb54..67873f03 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/selected_dirs_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/selected_dirs_test.go @@ -72,7 +72,7 @@ func TestOnlyTheSelectedDirectoryCompilesAndRuns(t *testing.T) { if !strings.HasPrefix(l, "go -C "+filepath.Join(tree, "b")+" test ") { t.Errorf("go test ran outside the selected directory: %s", l) } - if strings.Contains(l, "-run ^$") { + if strings.Contains(l, "-exec true") { compiled++ } else { ran++ @@ -91,7 +91,7 @@ func TestWithoutASelectionEveryDirectoryCompilesAndRuns(t *testing.T) { compiled, ran := 0, 0 for _, l := range strings.Split(f.lines(), "\n") { if strings.HasPrefix(l, "go ") && strings.Contains(l, " test ") { - if strings.Contains(l, "-run ^$") { + if strings.Contains(l, "-exec true") { compiled++ } else { ran++ @@ -146,27 +146,85 @@ func TestSelectedDirectoriesRunAtOnceAndAnswerInDirectoryOrder(t *testing.T) { } } -// The bound holds: with one directory at a time, no two go test commands run at once. +// The bound holds for the setup's compiles and for the tests: no more of them run at once than the bound, and with a +// bound of 3 the three directories compile at once, beside the store's start, given nothing a test reads, since the +// compile runs no test binary (issue 1921, try speed S5). func TestTheDirectoriesAtOnceNeverPassTheBound(t *testing.T) { - f := newIntegFake(t, "slow") - var mu sync.Mutex - running, most := 0, 0 - f.onTest = func(string) { - mu.Lock() - running++ - if running > most { - most = running - } - mu.Unlock() - time.Sleep(20 * time.Millisecond) - mu.Lock() - running-- - mu.Unlock() - } - m, states := threeModules(t, f) - m.Parallel = 1 - runIntegrationOf(t, m, states) - if most != 1 { - t.Errorf("%d go test commands ran at once with a bound of 1", most) + for _, tc := range []struct { + name string + compile bool // the setup's compiles are counted, not the tests + bound int + want int // how many must run at once (at most the bound always) + }{ + {"tests one at a time", false, 1, 1}, + {"compiles one at a time", true, 1, 1}, + {"compiles at once beside the store", true, 3, 3}, + } { + t.Run(tc.name, func(t *testing.T) { + f := newIntegFake(t, "slow") + var mu sync.Mutex + running, most := 0, 0 + reached := make(chan struct{}) + storeStarting, beside := false, false + count := func() { + mu.Lock() + running++ + most = max(most, running) + if running == tc.want && tc.want > 1 { + close(reached) + } + mu.Unlock() + if tc.want > 1 { + select { // held until all of them run, so "at once" does not hang on timing + case <-reached: + case <-time.After(2 * time.Second): + } + } else { + time.Sleep(20 * time.Millisecond) + } + mu.Lock() + running-- + mu.Unlock() + } + if tc.compile { + f.onCompile = func(_ string, env []string) { + for _, e := range env { + if strings.HasPrefix(e, "MESH_TEST_") { + t.Errorf("the compile was given %s: it runs no test binary", e) + } + } + mu.Lock() + beside = beside || storeStarting + mu.Unlock() + count() + } + if tc.want > 1 { + f.onStart = func() { // the store starts until the compiles have all run at once, or 2 s pass + mu.Lock() + storeStarting = true + mu.Unlock() + select { + case <-reached: + case <-time.After(2 * time.Second): + } + } + } + } else { + f.onTest = func(string) { count() } + } + m, states := threeModules(t, f) + m.Parallel = tc.bound + if _, st := runIntegrationOf(t, m, states); st.State != delivery.StepPassed { + t.Fatalf("the step ended %s: %s", st.State, st.Reason) + } + mu.Lock() + defer mu.Unlock() + if most > tc.bound || most < tc.want { + t.Errorf("%d ran at once with a bound of %d, want %d", most, tc.bound, tc.want) + } + if tc.compile && tc.want > 1 && !beside { + t.Errorf("no compile ran while the store was starting:\n%s", f.lines()) + } + }) } }