The forge bridge reads a merge's module directories only where their tree changed (issue 1126)

mesh-catalog's 140 module directories are 420 forge reads per merge, past the 30 s bound
of a handled event while the forge is busy: the merged message of #535 timed out on every
try, and #3fb7951 needed seven (issue 1458 met the same kind at 131). A tree hash names its
bytes, so what was read at a tree is kept in the new state inputs and in memory, and a
merge reads only the directories it changed.

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go

@@ -72,11 +72,17 @@ accounts {
             "mesh.mod.gitea.tool.gitea_open_pulls"
             "mesh.mod.gitea.tool.gitea_open_pulls.*"
             "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.API.DIRECT.GET.KV_mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_performed"
+            "$JS.API.STREAM.INFO.KV_mesh-delivery-forge-bridge_inputs"
             "$JS.FC.KV_mesh-delivery-forge-bridge_performed.>"
+            "$JS.FC.KV_mesh-delivery-forge-bridge_inputs.>"
             "$KV.mesh-delivery-forge-bridge_performed.>"
+            "$KV.mesh-delivery-forge-bridge_inputs.>"
             "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-gate_merges.>"
             "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-gate_merges.>"
             "$JS.API.DIRECT.GET.KV_mesh-delivery-gate_merges.>"
@@ -219,7 +225,7 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
 	}
 	// Its other state bucket, and the gate's buckets the bridge reads, made by the controller from the gate's manifest
 	// (mesh-sdk go/delivery MergesBucket, OwedBucket with their histories).
-	for name, history := range map[string]uint8{performedBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
+	for name, history := range map[string]uint8{performedBucket: 1, inputsBucket: 1, delivery.MergesBucket: delivery.MergesHistory, delivery.OwedBucket: delivery.OwedHistory} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
 			t.Fatal(err)
 		}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/kept.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"regexp"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-forge-bridge/internal/bridge"
+)
+
+// inputsBucket is the module's state bucket inputs (manifest state entry inputs): what the bridge read of a module
+// directory, its manifest and lock-file hashes, by the directory's git tree hash, so a merge reads the forge only for the
+// directories it changed (issue 1126; internal/bridge kept.go). A tree hash names its bytes, so a value is never stale;
+// a merge adds a value only for each directory it changed (a few kilobytes each).
+const inputsBucket = delivery.ForgeBridgeModule + "_inputs"
+
+// keptTimeout bounds one read or write of the bucket: the bucket is a saving, never a reason to fail a merge.
+const keptTimeout = 2 * time.Second
+
+var treeHash = regexp.MustCompile(`^[0-9a-f]{40}$|^[0-9a-f]{64}$`)
+
+// kept is the bridge's Kept: in memory always, and in the bucket inputs when it is on the bus, so the first merge after a
+// start reads only what changed too. A value the bucket cannot give or take is read from the forge, and said.
+type kept struct {
+	kv     jetstream.KeyValue // nil when the bucket is not on the bus
+	logger *log.Logger
+	mu     sync.Mutex
+	mem    map[string]bridge.Inputs
+}
+
+func newKept(kv jetstream.KeyValue, logger *log.Logger) *kept {
+	return &kept{kv: kv, logger: logger, mem: map[string]bridge.Inputs{}}
+}
+
+func (k *kept) Get(tree string) (bridge.Inputs, bool) {
+	k.mu.Lock()
+	in, ok := k.mem[tree]
+	k.mu.Unlock()
+	if ok || k.kv == nil || !treeHash.MatchString(tree) {
+		return in, ok
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
+	defer cancel()
+	e, err := k.kv.Get(ctx, tree)
+	if err != nil {
+		return bridge.Inputs{}, false // not kept (or not readable): read from the forge
+	}
+	if err := json.Unmarshal(e.Value(), &in); err != nil || in.Manifest == "" {
+		k.logger.Printf("%s.%s cannot be read, so its directory is read from the forge: %v", inputsBucket, tree, err)
+		return bridge.Inputs{}, false
+	}
+	k.mu.Lock()
+	k.mem[tree] = in
+	k.mu.Unlock()
+	return in, true
+}
+
+func (k *kept) Put(tree string, in bridge.Inputs) {
+	k.mu.Lock()
+	k.mem[tree] = in
+	k.mu.Unlock()
+	if k.kv == nil || !treeHash.MatchString(tree) {
+		return
+	}
+	data, err := json.Marshal(in)
+	if err != nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), keptTimeout)
+	defer cancel()
+	if _, err := k.kv.Put(ctx, tree, data); err != nil {
+		k.logger.Printf("%s.%s is not kept, so the next merge reads its directory from the forge again: %v", inputsBucket, tree, err)
+	}
+}

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go

@@ -67,8 +67,8 @@ type publisher interface {
 // handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never
 // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server
 // delivers it again (the server's ack wait is the only clock).
-func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) {
-	msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list)
+func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, kept bridge.Kept, logger *log.Logger) {
+	msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, list, kept)
 	switch {
 	case errors.Is(err, bridge.ErrRetry):
 		logger.Printf("%v; heard again in %s", err, retryDelay)
@@ -163,6 +163,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	go onReconnect(ctx, again, said, logger)
+	// What a merge read of each module directory, by its tree (issue 1126): a merge reads only the directories it changed.
+	inputs := newKept(process.OptionalBucket(ctx, js, inputsBucket, "a merge's module directories are kept in memory only, so the first merge after a start reads every one", logger.Printf), logger)
 	cons, err := cred.EventConsumer(ctx, js, eventsStream)
 	if err != nil {
 		// Essential: the forge's events are what the bridge carries.
@@ -173,7 +175,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			// Bounded: the turn is held, and no sync or other event goes on without it.
 			hctx, cancel := context.WithTimeout(ctx, syncDeadline)
 			defer cancel()
-			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger)
+			handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), inputs, logger)
 		})
 	}, logger.Printf)
 	if err != nil {

modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go

@@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
 	})
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" {
 		t.Fatalf("published on %v", pub.subjects)
 	}
@@ -97,7 +97,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
 func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 0 {
 		t.Errorf("a message was written: %v", pub.subjects)
 	}
@@ -112,7 +112,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
 func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here)
 	var logs bytes.Buffer
 	m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")}
-	handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, nil, someFiles, nil, nil, log.New(&logs, "", 0))
 	if m.acked != 0 || m.termd != 0 {
 		t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd)
 	}
@@ -134,7 +134,7 @@ func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/
 	call := func(context.Context, string, any) (json.RawMessage, error) {
 		return nil, errors.New("the tree listing of novox/mesh-delivery was cut")
 	}
-	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0))
 	if len(pub.subjects) != 0 {
 		t.Errorf("a message was written: %v", pub.subjects)
 	}
@@ -163,7 +163,7 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/
 		}
 		return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil
 	}
-	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
+	handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, nil, log.New(&logs, "", 0))
 	if len(keys) == 0 || keys[0] != "gitea.gitea_commit_trees" || slices.Index(keys[1:], "gitea.gitea_commit_trees") >= 0 {
 		t.Fatalf("calls %v, want the trees once and first", keys)
 	}

modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go

@@ -109,7 +109,7 @@ func unreadable(subject, why string, args ...any) error {
 // the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the
 // pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules.
 func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) {
-	return ConvertWith(subject, data, trees, files, nil)
+	return ConvertKept(subject, data, trees, files, nil, nil)
 }
 
 // Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees
@@ -122,6 +122,12 @@ type Listed func(owner, repo, sha, dir string) (map[string]string, error)
 // updated pull request of the catalogue that touches placement/ (touchesPlacement), with every placement file at its
 // head, which the indexer writes as placement candidates (ADR 0549 §4: the nodes it places answer it before it merges).
 func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) {
+	return ConvertKept(subject, data, trees, files, list, nil)
+}
+
+// ConvertKept is ConvertWith; with kept, a merge reads the forge only for the module directories whose tree hash kept
+// does not hold, and keeps what it read (kept.go).
+func ConvertKept(subject string, data []byte, trees Trees, files Files, list Listed, kept Kept) (Message, error) {
 	var action string
 	switch subject {
 	case UpdatedSubject:
@@ -148,7 +154,7 @@ func ConvertWith(subject string, data []byte, trees Trees, files Files, list Lis
 		return Message{}, unreadable(subject, "no branch")
 	}
 	if subject == MergedSubject && p.Base == Trunk {
-		return merged(subject, p, trees, files, list)
+		return merged(subject, p, trees, files, list, kept)
 	}
 	said := ""
 	if action == delivery.PullUpdated && p.Base != "" && p.Base != Trunk {
@@ -277,7 +283,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma
 }
 
 // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once.
-func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) {
+func merged(subject string, p pull, trees Trees, files Files, list Listed, kept Kept) (Message, error) {
 	if !sha40.MatchString(p.MergeSHA) {
 		return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA)
 	}
@@ -294,9 +300,19 @@ func merged(subject string, p pull, trees Trees, files Files, list Listed) (Mess
 		names = append(names, dir)
 	}
 	slices.Sort(names)
-	manifests, locks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, names, files)
+	manifests, locks, toRead := keptInputs(names, dirs, kept)
+	read, readLocks, err := inputsOf(p.Owner, p.Repo, p.MergeSHA, toRead, files)
 	if err != nil {
-		return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err)
+		return Message{}, fmt.Errorf("%w: %s: the manifests of %s at the merge commit %s (%d of %d module directories read from the forge, the others kept by their tree): %v", ErrRetry, subject, repository, p.MergeSHA, len(toRead), len(names), err)
+	}
+	keep(toRead, dirs, read, readLocks, kept)
+	maps.Copy(manifests, read)
+	maps.Copy(locks, readLocks)
+	if len(manifests) == 0 {
+		manifests = nil
+	}
+	if len(locks) == 0 {
+		locks = nil
 	}
 	event, err := delivery.ForgeEvent(repository, delivery.ForgeMerged)
 	if err != nil {

modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go

@@ -369,3 +369,60 @@ func TestOneFailingFileIsRetryAndEveryCallHasReturnedBeforeTheReadDoes(t *testin
 		t.Errorf("err = %v, want ErrRetry", err)
 	}
 }
+
+// memKept is a Kept in memory.
+type memKept map[string]Inputs
+
+func (k memKept) Get(tree string) (Inputs, bool) { in, ok := k[tree]; return in, ok }
+func (k memKept) Put(tree string, in Inputs)     { k[tree] = in }
+
+// Issue 1126 (mesh-catalog#535; issue 1458 met the kind first): a merge reads the forge only for the module directories
+// whose tree changed. Its message is the same as one read whole; a directory whose tree is kept costs no read.
+func TestAMergeReadsOnlyTheDirectoriesWhoseTreeChanged(t *testing.T) {
+	manifest, sum := real(t, "module.json"), real(t, "go.sum")
+	forge := &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest, "modules/a/go.sum": sum, "modules/b/module.json": []byte(`{"module":"b"}`)}}
+	kept := memKept{}
+	var calls []call
+	first, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, nil, kept)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(forge.calls) != 6 {
+		t.Fatalf("the first merge read %d files, want 6 (a manifest and two lock files for each directory)", len(forge.calls))
+	}
+	// The next merge changes modules/b alone: only its three files are read, and the message equals one read whole.
+	forge.calls = nil
+	forge.have["modules/b/module.json"] = []byte(`{"module":"b","v":2}`)
+	trees := map[string]string{"modules/a": "t1", "modules/b": "t3"}
+	next, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, nil, kept)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, c := range forge.calls {
+		if !strings.HasPrefix(c.path, "modules/b/") {
+			t.Errorf("%s was read although its directory's tree is kept", c.path)
+		}
+	}
+	if len(forge.calls) != 3 {
+		t.Errorf("the second merge read %d files, want 3", len(forge.calls))
+	}
+	whole, err := Convert(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(next.Data) != string(whole.Data) {
+		t.Errorf("the message from kept inputs differs from one read whole:\n%s\n%s", next.Data, whole.Data)
+	}
+	if string(first.Data) == string(next.Data) {
+		t.Error("the second merge's message did not change")
+	}
+	// A failed read keeps nothing of the directories it read, so the next try reads them again.
+	kept2 := memKept{}
+	failing := &forgeFiles{have: forge.have, errs: map[string]error{"modules/b/module.json": errors.New("context deadline exceeded")}}
+	if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2); !errors.Is(err, ErrRetry) {
+		t.Fatalf("err = %v, want ErrRetry", err)
+	}
+	if len(kept2) != 0 {
+		t.Errorf("a failed read kept %v", kept2)
+	}
+}

modules/mesh-delivery-forge-bridge/internal/bridge/kept.go

@@ -0,0 +1,57 @@
+package bridge
+
+// A merge of a repository of many modules read three files of the forge for every module, whatever the merge changed:
+// the catalogue's 140 modules are 420 reads, which no longer fit the 30 s bound of a handled event while the forge is
+// busy after a merge, so the merged message of mesh-catalog#535 was never written and the indexer never saw it (issue
+// 1126; issue 1458 met the same kind at 131 modules and raised the reads in flight). A directory's git tree hash names
+// every byte under it, so the manifest and the lock-file hashes read once at a tree hash are those of every later commit
+// whose directory has the same tree: a merge reads the forge only for the directories it changed.
+
+// Inputs is what the bridge read of one module directory at one tree hash: its manifest and its lock files' hex sha256
+// by name (none when it has none).
+type Inputs struct {
+	Manifest string            `json:"manifest"`
+	Locks    map[string]string `json:"locks,omitempty"`
+}
+
+// Kept holds Inputs by a directory's git tree hash (the bridge's state bucket inputs). Get answers false for a tree it
+// does not hold or cannot read; Put is best effort: a value it could not keep is read from the forge next time.
+type Kept interface {
+	Get(tree string) (Inputs, bool)
+	Put(tree string, in Inputs)
+}
+
+// keptInputs splits dirs into the inputs kept for their tree hash and the directories still to read from the forge.
+func keptInputs(dirs []string, trees map[string]string, kept Kept) (manifests map[string]string, locks map[string]map[string]string, read []string) {
+	manifests, locks = map[string]string{}, map[string]map[string]string{}
+	for _, dir := range dirs {
+		tree := trees[dir]
+		if kept == nil || tree == "" {
+			read = append(read, dir)
+			continue
+		}
+		in, ok := kept.Get(tree)
+		if !ok {
+			read = append(read, dir)
+			continue
+		}
+		manifests[dir] = in.Manifest
+		if len(in.Locks) > 0 {
+			locks[dir] = in.Locks
+		}
+	}
+	return manifests, locks, read
+}
+
+// keep puts what was read of each directory under its tree hash.
+func keep(dirs []string, trees map[string]string, manifests map[string]string, locks map[string]map[string]string, kept Kept) {
+	if kept == nil {
+		return
+	}
+	for _, dir := range dirs {
+		raw, ok := manifests[dir]
+		if tree := trees[dir]; ok && tree != "" {
+			kept.Put(tree, Inputs{Manifest: raw, Locks: locks[dir]})
+		}
+	}
+}

modules/mesh-delivery-forge-bridge/manifest_test.go

@@ -103,11 +103,12 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
 	if m.SecretsOwner != "mesh-delivery-forge-bridge" {
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
-	// Two buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443
-	// §5), and the revision of each key of the gate's buckets the bridge performed, so a key is performed once per
-	// revision (ADR 0437 §8; issue 1341).
-	if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341
-		t.Errorf("state = %+v, want the buckets heads and performed", m.State)
+	// Three buckets: the head last written on FORGE for each pull request, so a sync writes only what differs (ADR 0443
+	// §5), the revision of each key of the gate's buckets the bridge performed, so a key is performed once per
+	// revision (ADR 0437 §8; issue 1341), and what a merge read of each module directory by its tree, so a merge reads
+	// the forge only for the directories it changed (issue 1126).
+	if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}}) { // 1279/AC6, 1341, 1126
+		t.Errorf("state = %+v, want the buckets heads, performed and inputs", m.State)
 	}
 	// No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status).
 	for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {

modules/mesh-delivery-forge-bridge/module.json

@@ -12,7 +12,8 @@
   ],
   "state": [
     { "name": "heads", "history": 1 },
-    { "name": "performed", "history": 1 }
+    { "name": "performed", "history": 1 },
+    { "name": "inputs", "history": 1 }
   ],
   "reads": [
     "mesh-delivery-gate.merges",