Forge bridge: carry no placement files, placement is the seat mesh-placement's (issue 1126)

A catalogue merge or pull request no longer lists or reads placement/, so the Listed read and ConvertWith have no use and go.
This commit is contained in:
2026-10-28 17:58:58 +01:00
parent f4dd334b49
commit 730e44d762
5 changed files with 27 additions and 188 deletions
@@ -91,28 +91,6 @@ 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 <path>: 404 <body>` (client.ts request).
@@ -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, kept bridge.Kept, logger *log.Logger) {
msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, list, kept)
func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, kept bridge.Kept, logger *log.Logger) {
msg, err := bridge.ConvertKept(m.Subject(), m.Data(), trees, files, kept)
switch {
case errors.Is(err, bridge.ErrRetry):
logger.Printf("%v; heard again in %s", err, retryDelay)
@@ -175,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), inputs, logger)
handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), inputs, logger)
})
}, logger.Printf)
if err != nil {
@@ -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, nil, 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, nil, nil, 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, nil, nil, 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), nil, nil, 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), nil, nil, 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)
}
@@ -86,20 +86,6 @@ type pull struct {
PathsTruncated bool `json:"paths_truncated"`
}
// touchesPlacement says whether a pull request may change a placement file: a changed path under
// delivery.PlacementDir, or paths the forge did not say or cut (the safe direction: the files are read).
func (p pull) touchesPlacement() bool {
if p.Paths == nil || p.PathsTruncated {
return true
}
for _, path := range *p.Paths {
if strings.HasPrefix(path, delivery.PlacementDir+"/") {
return true
}
}
return false
}
func unreadable(subject, why string, args ...any) error {
return fmt.Errorf("%w: %s: %s", ErrUnreadable, subject, fmt.Sprintf(why, args...))
}
@@ -109,25 +95,12 @@ 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 ConvertKept(subject, data, trees, files, nil, nil)
return ConvertKept(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. So does an
// 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
// ConvertKept is Convert; 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) {
func ConvertKept(subject string, data []byte, trees Trees, files Files, kept Kept) (Message, error) {
var action string
switch subject {
case UpdatedSubject:
@@ -154,7 +127,7 @@ func ConvertKept(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, kept)
return merged(subject, p, trees, files, kept)
}
said := ""
if action == delivery.PullUpdated && p.Base != "" && p.Base != Trunk {
@@ -175,13 +148,6 @@ func ConvertKept(subject string, data []byte, trees Trees, files Files, list Lis
return Message{}, fmt.Errorf("%w: %s: the manifests of %s/%s at the head %s: %v", ErrRetry, subject, p.Owner, p.Repo, p.HeadSHA, err)
}
}
if action == delivery.PullUpdated && list != nil && delivery.SameRepository(f.Repository, delivery.CatalogueRepository) && p.touchesPlacement() {
var err error
if f.Placements, err = placementsOf(p.Owner, p.Repo, p.HeadSHA, list, files); err != nil {
return Message{}, fmt.Errorf("%w: %s: the placement files of %s at the head %s: %v", ErrRetry, subject, f.Repository, p.HeadSHA, err)
}
f.PlacementsRead = true
}
event, err := delivery.ForgeEvent(f.Repository, delivery.ForgePull)
if err != nil {
return Message{}, unreadable(subject, "%v", err)
@@ -311,7 +277,7 @@ func inputsRead(owner, repo, ref string, dirs []string, files Files) (manifests
}
// 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, kept Kept) (Message, error) {
func merged(subject string, p pull, trees Trees, files Files, kept Kept) (Message, error) {
if !sha40.MatchString(p.MergeSHA) {
return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA)
}
@@ -349,12 +315,6 @@ func merged(subject string, p pull, trees Trees, files Files, list Listed, kept
}
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)
@@ -452,30 +412,3 @@ 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 `<node>.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
}
@@ -79,46 +79,23 @@ 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)
// ADR 0567: placement is the seat mesh-placement's: a merge of the catalogue carries no placement file and asks the
// forge for none.
forge.calls = nil
msg, err = Convert(mergedSubject, mergedBody(map[string]any{"repo": "mesh-catalog"}), treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get)
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)
if f.Placements != nil || f.PlacementsRead {
t.Errorf("a merge of the catalogue carries placement: %v", f.Placements)
}
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)
for _, c := range forge.calls {
if strings.HasPrefix(c.path, "placement") {
t.Errorf("the forge was asked for %s", c.path)
}
}
// Issue 1126: a merge reads the forge only for the module directories whose tree changed.
t.Run("a merge reads only the directories whose tree changed", mergeReadsOnlyChangedDirs)
@@ -148,55 +125,6 @@ func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2
if f.Manifests != nil || f.Locks != nil || len(forge.calls) != 0 {
t.Errorf("manifests %v locks %v calls %v", f.Manifests, f.Locks, forge.calls)
}
// Issue 1126 (ADR 0549 §4): an updated pull request of the catalogue that touches placement/ carries every placement
// file at its head, read whole or not at all; one that touches none, or of another repository, carries none and lists
// nothing; paths the forge did not say or cut are read as touching; a closed one carries none.
forge = &forgeFiles{have: map[string][]byte{"modules/a/module.json": manifest,
"placement/ace.json": []byte(`{"v":1,"node":"ace","modules":[]}`), "placement/g14.json": []byte(`{"v":1,"node":"g14","modules":[]}`)}}
listing := map[string]string{"ace.json": "b1", "g14.json": "b2", "README.md": "b3"}
var listedAt []string
list := func(owner, repo, sha, dir string) (map[string]string, error) {
listedAt = append(listedAt, owner+"/"+repo+"@"+sha+":"+dir)
return listing, nil
}
all := map[string]string{"ace": `{"v":1,"node":"ace","modules":[]}`, "g14": `{"v":1,"node":"g14","modules":[]}`}
catalogue := map[string]any{"repo": "mesh-catalog", "paths": []string{"placement/g14.json"}}
for _, row := range []struct {
name string
subject string
body map[string]any
want map[string]string
read bool
}{
{"a catalogue pull touching placement", updatedSubject, catalogue, all, true},
{"a catalogue pull touching no placement", updatedSubject, map[string]any{"repo": "mesh-catalog"}, nil, false},
{"a catalogue pull whose paths are cut", updatedSubject, map[string]any{"repo": "mesh-catalog", "paths_truncated": true}, all, true},
{"a catalogue pull whose paths are not said", updatedSubject, map[string]any{"repo": "mesh-catalog", "paths": nil}, all, true},
{"another repository's pull touching placement/", updatedSubject, map[string]any{"paths": []string{"placement/g14.json"}}, nil, false},
{"a closed catalogue pull", closedSubject, catalogue, nil, false},
} {
listedAt = nil
body := updated(row.body)
msg, err := ConvertWith(row.subject, body, nil, forge.get, list)
if err != nil {
t.Fatalf("%s: %v", row.name, err)
}
f, err := delivery.DecodeForge(msg.Data)
if err != nil {
t.Fatalf("%s: %v", row.name, err)
}
if f.PlacementsRead != row.read || !reflect.DeepEqual(f.Placements, row.want) || (len(listedAt) > 0) != row.read {
t.Errorf("%s: placements %v (read %v), listed %v", row.name, f.Placements, f.PlacementsRead, listedAt)
}
if row.read && listedAt[0] != "novox/mesh-catalog@"+head+":placement" {
t.Errorf("%s: listed at %v, want the head", row.name, listedAt)
}
}
listing = map[string]string{"gone.json": "b5"}
if _, err := ConvertWith(updatedSubject, updated(catalogue), nil, forge.get, list); !errors.Is(err, ErrRetry) || !strings.Contains(err.Error(), "placement/gone.json") {
t.Errorf("a placement file the forge cannot give at the head: %v", err)
}
}
func TestAForgeErrorOrAMissingManifestIsRetryAndALockNotFoundIsLeftOut(t *testing.T) { // 1448/AC3
@@ -385,7 +313,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) {
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)
first, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r0", map[string]string{"modules/a": "t1", "modules/b": "t2"}, nil), forge.get, kept)
if err != nil {
t.Fatal(err)
}
@@ -396,7 +324,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) {
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)
next, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), forge.get, kept)
if err != nil {
t.Fatal(err)
}
@@ -422,7 +350,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) {
// rest: a merge too large for one try's bound converges over its tries.
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) {
if _, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2); !errors.Is(err, ErrRetry) {
t.Fatalf("err = %v, want ErrRetry", err)
}
if _, ok := kept2["t1"]; !ok || len(kept2) != 1 { // modules/a's files are asked first, before the failure is seen
@@ -430,7 +358,7 @@ func mergeReadsOnlyChangedDirs(t *testing.T) {
}
failing.calls = nil
failing.errs = nil
again, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, nil, kept2)
again, err := ConvertKept(mergedSubject, mergedBody(nil), treesOf(&calls, "r1", trees, nil), failing.get, kept2)
if err != nil {
t.Fatal(err)
}