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)

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
 		}

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)
+		reads = append(reads, &read{dir: dir, path: at(dir, "module.json"), manifest: true})
-		if err != nil {
-			return nil, nil, fmt.Errorf("%s: %w", at(dir, "module.json"), err)
-		}
-		manifests[dir] = string(raw)
 		for _, name := range LockFiles {
-			raw, err := files(owner, repo, at(dir, name), ref)
+			reads = append(reads, &read{dir: dir, path: at(dir, name)})
-			if errors.Is(err, ErrNotFound) {
+		}
-				continue
+	}
+	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 {
+		}(r)
-				return nil, nil, fmt.Errorf("%s: %w", at(dir, name), err)
+	}
+	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)
+			locks[r.dir][path.Base(r.path)] = hex.EncodeToString(sum[:])
-			if locks[dir] == nil {
-				locks[dir] = map[string]string{}
-			}
-			locks[dir][name] = hex.EncodeToString(sum[:])
 		}
 	}
 	if len(manifests) == 0 {

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)
+	}
+}