diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go index 9310a73..51fe90a 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go @@ -91,6 +91,28 @@ func treesVia(ctx context.Context, call caller) bridge.Trees { } } +// listedVia asks the forge for the files directly under one directory of a commit through call: gitea_commit_trees with +// listed (issue 1126), answer {listed: {directory: {name: blob id}}}. +func listedVia(ctx context.Context, call caller) bridge.Listed { + return func(owner, repo, sha, dir string) (map[string]string, error) { + raw, err := call(ctx, commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha, "listed": []string{dir}}) + if err != nil { + return nil, err + } + var a struct { + Listed map[string]map[string]string `json:"listed"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("the answer of %s is not {listed}: %w", commitTreesTool, err) + } + if a.Listed == nil { + // A forge module from before listed answers none: placement cannot be read, never read as empty. + return nil, fmt.Errorf("the answer of %s lists no directory; the forge module does not list yet", commitTreesTool) + } + return a.Listed[dir], nil + } +} + // getFileTool is the forge's tool for one file at a ref (mesh-catalog modules/gitea tools/index.ts gitea_get_file): // arguments owner, repo, path and ref, answer {file: {path, ref, sha, size, content}}, the content decoded to text // (client.ts getFile). A path the forge does not have fails with `Gitea API : 404 ` (client.ts request). diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index e069a6c..dbdf411 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -117,8 +117,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, logger *log.Logger) { - msg, err := bridge.Convert(m.Subject(), m.Data(), trees, files) +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) switch { case errors.Is(err, bridge.ErrRetry): logger.Printf("%v; heard again in %s", err, retryDelay) @@ -237,7 +237,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), logger) + handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger) }) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) 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 9a16a64..85df582 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 @@ -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, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, 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, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, 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, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, 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), log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), 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), log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), 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) } diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 6f902ec..ed3098d 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index bdc8b83..010e0ee 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -2,6 +2,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYE git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a h1:JjQUrojy2rfxVlkTDC8tkP8yPh+rZ57mpuiPWqPKy3o= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index a6cbc7d..e65f8fd 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "path" "regexp" "slices" @@ -88,6 +89,17 @@ 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) +} + +// Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees +// with listed); a directory the commit lacks lists none. +type Listed func(owner, repo, sha, dir string) (map[string]string, error) + +// ConvertWith is Convert; with list, a merge of the catalogue repository (delivery.CatalogueRepository) also carries every +// placement file of its merge commit (novox/hq ADR 0549 §2, issue 1126), read whole or not at all: a forge that cannot +// say is ErrRetry, as for the manifests, so the indexer never takes a part of placement for all of it. +func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) { var action string switch subject { case UpdatedSubject: @@ -114,7 +126,7 @@ func Convert(subject string, data []byte, trees Trees, files Files) (Message, er return Message{}, unreadable(subject, "no branch") } if subject == MergedSubject && p.Base == Trunk { - return merged(subject, p, trees, files) + return merged(subject, p, trees, files, list) } f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Action: action} @@ -227,7 +239,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) (Message, error) { +func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) { if !sha40.MatchString(p.MergeSHA) { return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA) } @@ -252,8 +264,15 @@ func merged(subject string, p pull, trees Trees, files Files) (Message, error) { if err != nil { return Message{}, unreadable(subject, "%v", err) } - body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, - Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks}) + f := delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, + Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks} + if list != nil && delivery.SameRepository(repository, delivery.CatalogueRepository) { + if f.Placements, err = placementsOf(p.Owner, p.Repo, p.MergeSHA, list, files); err != nil { + return Message{}, fmt.Errorf("%w: %s: the placement files of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err) + } + f.PlacementsRead = true + } + body, err := delivery.EncodeForge(f) if err != nil { return Message{}, unreadable(subject, "%v", err) } @@ -350,3 +369,30 @@ func OpenMessage(answer []byte) (Message, error) { } return Message{Event: delivery.OpenEvent(), Data: body}, nil } + +// placementsOf is every placement file of the catalogue at a commit, by node, its bytes exact: each `.json` directly +// under delivery.PlacementDir. A file whose name is no node is left out and the rest read; the catalogue's check refuses +// such a file before it merges (ADR 0549 §1). Nil when there is none. +func placementsOf(owner, repo, sha string, list Listed, files Files) (map[string]string, error) { + names, err := list(owner, repo, sha, delivery.PlacementDir) + if err != nil { + return nil, err + } + var out map[string]string + for _, name := range slices.Sorted(maps.Keys(names)) { + file := delivery.PlacementDir + "/" + name + node, ok := delivery.PlacementFileNode(file) + if !ok { + continue + } + raw, err := files(owner, repo, file, sha) + if err != nil { + return nil, fmt.Errorf("%s: %w", file, err) + } + if out == nil { + out = map[string]string{} + } + out[node] = string(raw) + } + return out, nil +} 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 a5617ec..d4afe54 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -79,6 +79,47 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 if want := map[string]map[string]string{"modules/a": {"go.sum": hexSum(sum)}}; !reflect.DeepEqual(f.Locks, want) { t.Errorf("locks = %v, want %v", f.Locks, want) } + if f.Placements != nil || f.PlacementsRead { + t.Errorf("a merge of another repository than the catalogue carries placement: %v", f.Placements) + } + + // Issue 1126 (ADR 0549 §2): a merge of the catalogue carries every placement file of its merge commit, by node; a file + // whose name is no node is left out; a listing or a file the forge cannot give is a retry, never a part. + catalogue := mergedBody(map[string]any{"repo": "mesh-catalog"}) + forge.have["placement/ace.json"] = []byte(`{"v":1,"node":"ace","modules":[]}`) + forge.have["placement/g14.json"] = []byte(`{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`) + listing := map[string]string{"ace.json": "b1", "g14.json": "b2", "README.md": "b3", "a.b.json": "b4"} + var listedAt []string + list := func(owner, repo, sha, dir string) (map[string]string, error) { + listedAt = append(listedAt, owner+"/"+repo+"@"+sha+":"+dir) + return listing, nil + } + msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list) + if err != nil { + t.Fatal(err) + } + if f, err = delivery.DecodeForge(msg.Data); err != nil { + t.Fatal(err) + } + want := map[string]string{"ace": `{"v":1,"node":"ace","modules":[]}`, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`} + if !f.PlacementsRead || !reflect.DeepEqual(f.Placements, want) || !reflect.DeepEqual(listedAt, []string{"novox/mesh-catalog@" + merge + ":placement"}) { + t.Errorf("placements = %v (read %v), listed at %v", f.Placements, f.PlacementsRead, listedAt) + } + listing = map[string]string{} + if msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); err != nil { + t.Fatal(err) + } + if f, _ = delivery.DecodeForge(msg.Data); !f.PlacementsRead || f.Placements != nil { + t.Errorf("a catalogue with no placement file says it read none: %v, %v", f.Placements, f.PlacementsRead) + } + listing = map[string]string{"gone.json": "b5"} + if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); !errors.Is(err, ErrRetry) || !strings.Contains(err.Error(), "placement/gone.json") { + t.Errorf("a placement file the forge cannot give: %v", err) + } + failing := func(_, _, _, _ string) (map[string]string, error) { return nil, errors.New("forge down") } + if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, failing); !errors.Is(err, ErrRetry) { + t.Errorf("a listing the forge cannot give: %v", err) + } } func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2