From c9f742d5f20ea8d2e7489d1a5917802434c8e02e Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 19:06:03 +0000 Subject: [PATCH] Read a merge's manifests and lock files at most eight at once, so a catalogue merge stays inside the bridge's bound (issues 1448, 1458, design 47 R4c-2) --- .../mesh-delivery-forge-bridge/main_test.go | 4 + .../internal/bridge/bridge.go | 82 ++++++++--- .../internal/bridge/files_test.go | 139 ++++++++++++++++++ 3 files changed, 208 insertions(+), 17 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index b74d8da..c4ab010 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -8,6 +8,7 @@ import ( "log" "slices" "strings" + "sync" "testing" "time" @@ -139,8 +140,11 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/ m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.merged", data: mergedEvent()}, &fakePub{} var keys []string var args []any + var mu sync.Mutex // the files are asked for concurrently (issue 1458) call := func(_ context.Context, key string, body any) (json.RawMessage, error) { + mu.Lock() keys, args = append(keys, key), append(args, body) + mu.Unlock() if key == "gitea.gitea_get_file" { // the manifests, 1448; the trees are what this test is about return json.RawMessage(`{"file":{"path":"p","sha":"0123","size":2,"content":"{}"}}`), nil } diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index b1c74f8..a6cbc7d 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -7,10 +7,12 @@ import ( "encoding/json" "errors" "fmt" + "path" "regexp" "slices" "strconv" "strings" + "sync" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -135,38 +137,84 @@ func Convert(subject string, data []byte, trees Trees, files Files) (Message, er return Message{Event: event, Data: body}, nil } +// MaxReadsInFlight is how many files of the forge inputsOf asks for at once. A merge into a repository of many +// modules (131 in the catalogue) reads three files for each, one after another inside the bound of a handled event +// (30 s) that leaves about 76 ms a call; eight at once keeps the forge's load small and the read well inside the bound +// (issue 1458). +const MaxReadsInFlight = 8 + +// read is one file asked for: the path in the directory dir, and whether it is the module's manifest. +type read struct { + dir, path string + manifest bool + raw []byte + err error +} + // inputsOf reads, at ref, the module.json of each directory (the root spelled "") and the lock files beside it // (LockFiles): the manifests as the forge answered them and the lock files' hex sha256 by name. A manifest the forge // does not have is an error, since the directory was said to hold one; a lock file it does not have is left out; any other // failure is an error (issue 1448). +// +// The files are asked for with at most MaxReadsInFlight calls at once (issue 1458). The result does not depend on the +// order the calls finish in, and the error is that of the first failing file in the order of dirs and LockFiles. After +// a failure no further file is asked for, and inputsOf returns only when every call it started has returned. func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests map[string]string, locks map[string]map[string]string, err error) { - manifests = map[string]string{} - locks = map[string]map[string]string{} at := func(dir, name string) string { if dir == "" { return name } return dir + "/" + name } + var reads []*read for _, dir := range dirs { - raw, err := files(owner, repo, at(dir, "module.json"), ref) - if err != nil { - return nil, nil, fmt.Errorf("%s: %w", at(dir, "module.json"), err) - } - manifests[dir] = string(raw) + reads = append(reads, &read{dir: dir, path: at(dir, "module.json"), manifest: true}) for _, name := range LockFiles { - raw, err := files(owner, repo, at(dir, name), ref) - if errors.Is(err, ErrNotFound) { - continue + reads = append(reads, &read{dir: dir, path: at(dir, name)}) + } + } + var ( + wg sync.WaitGroup + mu sync.Mutex + failed bool + slots = make(chan struct{}, MaxReadsInFlight) + ) + for _, r := range reads { + slots <- struct{}{} + mu.Lock() + stop := failed + mu.Unlock() + if stop { + <-slots + break + } + wg.Add(1) + go func(r *read) { + defer func() { <-slots; wg.Done() }() + r.raw, r.err = files(owner, repo, r.path, ref) + if r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)) { + mu.Lock() + failed = true + mu.Unlock() } - if err != nil { - return nil, nil, fmt.Errorf("%s: %w", at(dir, name), err) + }(r) + } + wg.Wait() + manifests = map[string]string{} + locks = map[string]map[string]string{} + for _, r := range reads { + switch { + case r.err != nil && (r.manifest || !errors.Is(r.err, ErrNotFound)): + return nil, nil, fmt.Errorf("%s: %w", r.path, r.err) + case r.err != nil: // a lock file the forge does not have + case r.manifest: + manifests[r.dir] = string(r.raw) + default: + sum := sha256.Sum256(r.raw) + if locks[r.dir] == nil { + locks[r.dir] = map[string]string{} } - sum := sha256.Sum256(raw) - if locks[dir] == nil { - locks[dir] = map[string]string{} - } - locks[dir][name] = hex.EncodeToString(sum[:]) + locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:]) } } if len(manifests) == 0 { diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index 67e33b1..a5617ec 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -4,10 +4,14 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "fmt" "os" "reflect" + "sort" "strings" + "sync" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -32,11 +36,14 @@ type fileCall struct{ owner, repo, path, ref string } type forgeFiles struct { have map[string][]byte errs map[string]error + mu sync.Mutex calls []fileCall } func (f *forgeFiles) get(owner, repo, path, ref string) ([]byte, error) { + f.mu.Lock() f.calls = append(f.calls, fileCall{owner, repo, path, ref}) + f.mu.Unlock() if err, ok := f.errs[path]; ok { return nil, err } @@ -136,7 +143,139 @@ func TestTheFilesAreAskedAtTheMergeCommitByOwnerRepoPathAndRef(t *testing.T) { / } want := []fileCall{{"novox", "mesh-delivery", "modules/a/module.json", merge}, {"novox", "mesh-delivery", "modules/a/go.sum", merge}, {"novox", "mesh-delivery", "modules/a/package-lock.json", merge}} + sort.Slice(forge.calls, func(i, j int) bool { return forge.calls[i].path < forge.calls[j].path }) // they are asked for concurrently + sort.Slice(want, func(i, j int) bool { return want[i].path < want[j].path }) if !reflect.DeepEqual(forge.calls, want) { t.Errorf("calls = %v, want %v", forge.calls, want) } } + +// slowForge answers like a forge that takes d a call: a module.json of the directory's name everywhere, a go.sum in +// the directories whose number is even and in the root, and counts the calls in flight. +type slowForge struct { + d time.Duration + mu sync.Mutex + inFlight, most, started, n int + failOn string + finished int +} + +func (f *slowForge) get(_, _, path, _ string) ([]byte, error) { + f.mu.Lock() + f.inFlight++ + f.started++ + if f.inFlight > f.most { + f.most = f.inFlight + } + f.mu.Unlock() + time.Sleep(f.d) + defer func() { f.mu.Lock(); f.inFlight--; f.finished++; f.mu.Unlock() }() + switch { + case path == f.failOn: + return nil, errors.New("Gitea API " + path + ": 502 bad gateway") + case strings.HasSuffix(path, "module.json"): + return []byte(`{"module":"` + path + `"}`), nil + case strings.HasSuffix(path, "go.sum") && (path == "go.sum" || path[len(path)-8]%2 == 0): + return []byte("sum of " + path), nil + } + return nil, ErrNotFound +} + +func manyDirs(n int) []string { + dirs := []string{""} + for i := 1; i < n; i++ { + dirs = append(dirs, fmt.Sprintf("modules/m%02d", i)) + } + return dirs +} + +func TestFortyDirectoriesAreReadEightAtATimeWithTheResultOfOneAtATime(t *testing.T) { // 1458/AC1 + dirs := manyDirs(40) + slow := &slowForge{d: 10 * time.Millisecond} + manifests, locks, err := inputsOf("novox", "mesh-catalog", merge, dirs, slow.get) + if err != nil { + t.Fatal(err) + } + if slow.most > MaxReadsInFlight || slow.most < 2 { + t.Errorf("at most %d calls were in flight, want more than one and at most %d", slow.most, MaxReadsInFlight) + } + if slow.started != 40*3 { + t.Errorf("%d calls, want 120", slow.started) + } + // The same read one file at a time. + wantM, wantL := map[string]string{}, map[string]map[string]string{} + one := &slowForge{} + for _, dir := range dirs { + m, l, err := inputsOfOne(dir, one.get) + if err != nil { + t.Fatal(err) + } + wantM[dir] = m + if l != nil { + wantL[dir] = l + } + } + if !reflect.DeepEqual(manifests, wantM) || !reflect.DeepEqual(locks, wantL) { + t.Errorf("the concurrent read differs from a one-at-a-time read: %d manifests %d locks, want %d and %d", len(manifests), len(locks), len(wantM), len(wantL)) + } + if len(locks) == 0 || len(locks) == 40 { + t.Errorf("%d directories with locks, the fixture should have some and not all", len(locks)) + } +} + +// inputsOfOne reads one directory with a single call at a time, as the bridge did before issue 1458. +func inputsOfOne(dir string, files Files) (string, map[string]string, error) { + at := func(name string) string { + if dir == "" { + return name + } + return dir + "/" + name + } + raw, err := files("o", "r", at("module.json"), "x") + if err != nil { + return "", nil, err + } + var locks map[string]string + for _, name := range LockFiles { + b, err := files("o", "r", at(name), "x") + if errors.Is(err, ErrNotFound) { + continue + } + if err != nil { + return "", nil, err + } + if locks == nil { + locks = map[string]string{} + } + s := sha256.Sum256(b) + locks[name] = hex.EncodeToString(s[:]) + } + return string(raw), locks, nil +} + +func TestOneFailingFileIsRetryAndEveryCallHasReturnedBeforeTheReadDoes(t *testing.T) { // 1458/AC2 + dirs := manyDirs(40) + slow := &slowForge{d: 10 * time.Millisecond, failOn: "modules/m05/go.sum"} + _, _, err := inputsOf("novox", "mesh-catalog", merge, dirs, slow.get) + if err == nil || !strings.Contains(err.Error(), "modules/m05/go.sum") || !strings.Contains(err.Error(), "502") { + t.Fatalf("err = %v, want the failing file named", err) + } + slow.mu.Lock() + defer slow.mu.Unlock() + if slow.inFlight != 0 || slow.finished != slow.started { + t.Errorf("%d calls in flight, %d of %d finished, when the read returned", slow.inFlight, slow.finished, slow.started) + } + if slow.started >= 40*3 { + t.Errorf("%d calls were made, want the read to stop asking after the failure", slow.started) + } + // Through Convert it is ErrRetry. + var calls []call + dirsMap := map[string]string{} + for _, d := range dirs[1:] { + dirsMap[d] = "t" + } + slow2 := &slowForge{failOn: "modules/m05/module.json"} + if _, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", dirsMap, nil), slow2.get); !errors.Is(err, ErrRetry) { + t.Errorf("err = %v, want ErrRetry", err) + } +}