novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Forge bridge: bridge_merged_message answers a merge's merged message again, built as the bridge builds the one it publishes (issue 1585)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-05 11:34:04 +01:00
parent 343321f11e
commit 0605962960
8 changed files with 269 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go
+82 -2
View File
@@ -8,7 +8,9 @@ import (
"context"
"encoding/json"
"fmt"
"maps"
"os"
"slices"
"strconv"
"strings"
"sync"
@@ -17,6 +19,9 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"mesh-delivery-forge-bridge/internal/bridge"
"mesh-delivery-forge-bridge/internal/groups"
)
@@ -24,15 +29,34 @@ import (
const (
GroupsTool = "bridge_groups"
EndAgainTool = "bridge_group_end_again"
// MergedTool answers the merged message of a merge on request (issue 1585).
MergedTool = "bridge_merged_message"
)
// asker writes an end-asks value and waits for the bridge's answer to it.
type asker func(key string, value []byte) (groups.EndAsk, error)
// Tools are the bridge's tools: groupsRead reads the groups the gate holds with what the bridge kept of them.
func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []string, error)) []stdio.Tool {
func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []string, error), merged func(repository, merge string) (json.RawMessage, error)) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
return []stdio.Tool{
{Name: MergedTool,
Description: "The merged message of one merge, asked again (issue 1585): the message the bridge wrote on its stream FORGE when the " +
"pull request was merged into the trunk, built the same way from the forge (the merge commit's trees and the manifests " +
"and lock files of its module directories) and from the gate's merges state (the pull request's number, head and " +
"group). For the indexer, which asks it when it takes in a held merge it does not hold in memory (a restart, a new " +
"build), and the ask-again a lost message lacked. About 0.5 MB for a repository of many modules. Answers {message} " +
"with the forge message whole, or why it cannot: the gate holds no merge with that commit, or the forge did not answer.",
Input: map[string]any{"type": "object", "required": []string{"repository", "merge"}, "properties": map[string]any{
"repository": map[string]any{"type": "string", "description": "the repository, owner/name, as the forge names it"},
"merge": map[string]any{"type": "string", "description": "the merge commit, 40 hex digits"}}},
Run: func(args map[string]any) (any, error) {
raw, err := merged(str(args, "repository"), str(args, "merge"))
if err != nil {
return nil, err
}
return map[string]any{"message": raw}, nil
}},
{Name: GroupsTool,
Description: "The delivery groups the gate holds (more than one member), each with its members' merge state and whether each " +
"merged message is written on the forge stream, and whether the group's end (forge.group.merged) is written: ended, " +
@@ -81,6 +105,43 @@ func Tools(ask asker, now func() time.Time, groupsRead func() ([]groups.Row, []s
}
}
// MergedOf is the merged message of the merge of repository at commit merge, as the bridge writes it: the pull request's
// number, head and group from the gate's merges (raw, by key), the rest from the forge through trees and files
// (bridge.MergedMessage), and, for a member of a delivery group, the group and its members as the process sets them.
func MergedOf(raws map[string][]byte, repository, merge string, trees bridge.Trees, files bridge.Files) (json.RawMessage, error) {
owner, repo, ok := strings.Cut(repository, "/")
if !ok || owner == "" || repo == "" {
return nil, fmt.Errorf("%q is not owner/name", repository)
}
for _, key := range slices.Sorted(maps.Keys(raws)) {
m, err := delivery.DecodeMerge(raws[key])
if err != nil {
continue
}
at := slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool {
return delivery.SameRepository(p.Repository, repository) && p.Commit == merge
})
if at < 0 {
continue
}
p := m.Pulls[at]
msg, err := bridge.MergedMessage(owner, repo, p.Number, m.Group, p.Head.Head, merge, trees, files)
if err != nil {
return nil, fmt.Errorf("the merged message of %s at %s cannot be built: %w", repository, merge, err)
}
f, err := forgemsg.DecodeForge(msg.Data)
if err != nil {
return nil, err
}
if _, approved := m.Approved[delivery.ApprovedKey(f.Repository, f.Number)]; approved && len(m.Approved) >= 2 {
f.Group, f.Members = m.Group, groups.MembersOf(m)
}
out, err := forgemsg.EncodeForge(f)
return out, err
}
return nil, fmt.Errorf("the gate's merges hold no merge of %s with the commit %s, so its pull request is not known", repository, merge)
}
// read is the groups the gate holds with what the bridge kept of them: their ends in performed, and the last ask of
// each in end-asks with the bridge's answer.
func read(merges, performed, asks stdio.KeptState) ([]groups.Row, []string, error) {
@@ -175,10 +236,29 @@ func askEnd(s stdio.KeptState, bound time.Duration) asker {
}
}
// askCaller calls a tool as this bundle's module, through the runtime (stdio.Ask), the module's own invokes granting it.
func askCaller(_ context.Context, key string, body any) (json.RawMessage, error) {
return stdio.Ask(key, body)
}
func main() {
merges, performed, asks := stdio.State(delivery.GateModule+".merges"), stdio.State(groups.PerformedState), stdio.State(groups.EndAsksState)
tools := Tools(askEnd(asks, 25*time.Second), time.Now,
func() ([]groups.Row, []string, error) { return read(merges, performed, asks) })
func() ([]groups.Row, []string, error) { return read(merges, performed, asks) },
func(repository, merge string) (json.RawMessage, error) {
keys, err := merges.Keys()
if err != nil {
return nil, fmt.Errorf("the gate's merges cannot be listed: %w", err)
}
raws := map[string][]byte{}
for _, k := range keys {
if e, err := merges.Get(k); err == nil && e != nil {
raws[k] = e.Value
}
}
ctx := context.Background()
return MergedOf(raws, repository, merge, bridge.TreesVia(ctx, askCaller), bridge.FilesVia(ctx, askCaller))
})
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-forge-bridge-tools] %v\n", err)
os.Exit(1)
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go
+75 -1
View File
@@ -1,13 +1,18 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-forge-bridge/internal/bridge"
"mesh-delivery-forge-bridge/internal/groups"
)
@@ -51,7 +56,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
func() ([]groups.Row, []string, error) {
rows, unread := groups.Rows(merges, performed, map[string]uint64{"ended": 7}, asks)
return rows, unread, nil
})
}, nil)
call := func(name string, args map[string]any) any {
for _, tl := range tools {
if tl.Name == name {
@@ -114,3 +119,72 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) {
}
}
}
// 1585: bridge_merged_message answers the merged message as the bridge publishes it: for a lone pull request the same bytes
// the event gives through Convert, for a member of a delivery group those with its group and members set as the process
// sets them; a merge the gate does not hold, or a forge that does not answer, is said.
func TestTheBridgeToolAnswersTheMergedMessageAsPublished(t *testing.T) {
h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40)
c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40)
trees := func(owner, repo, sha string) (string, map[string]string, error) {
return "root-" + sha[:4], map[string]string{"": "r", "modules/a": "ta-" + sha[:4]}, nil
}
files := func(owner, repo, path, ref string) ([]byte, error) {
if strings.HasSuffix(path, "module.json") {
return []byte(`{"module":"a"}`), nil
}
return nil, bridge.ErrNotFound
}
published := func(repo string, number int, head, branch, merge string) forgemsg.Forge {
body, _ := json.Marshal(map[string]any{"owner": "novox", "repo": repo, "number": number, "head": branch, "head_sha": head, "base": "main", "merge_commit_sha": merge})
msg, err := bridge.Convert(bridge.MergedSubject, body, trees, files)
if err != nil {
t.Fatal(err)
}
f, err := forgemsg.DecodeForge(msg.Data)
if err != nil {
t.Fatal(err)
}
return f
}
lone, _ := delivery.EncodeMerge(delivery.Merge{Group: "solo", Ask: "a", Approved: map[string]string{"novox/a@5": h1},
Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 5, Head: h1}, State: delivery.MergeMerged, Commit: c1}}})
group, _ := delivery.EncodeMerge(delivery.Merge{Group: "feat/g", Ask: "a", Approved: map[string]string{"novox/a@6": h1, "novox/b@7": h2},
Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 6, Head: h1}, State: delivery.MergeMerged, Commit: c1},
{Head: delivery.Head{Repository: "novox/b", Number: 7, Head: h2}, State: delivery.MergeMerging}}})
for _, c := range []struct {
name string
raws map[string][]byte
repository string
merge string
want forgemsg.Forge
fail string
}{
{name: "a lone pull request", raws: map[string][]byte{"solo": lone}, repository: "novox/a", merge: c1, want: published("a", 5, h1, "solo", c1)},
{name: "a member of a group", raws: map[string][]byte{"feat~g": group}, repository: "novox/a", merge: c1, want: func() forgemsg.Forge {
f := published("a", 6, h1, "feat/g", c1)
f.Group, f.Members = "feat/g", []delivery.Head{{Repository: "novox/a", Number: 6, Head: h1}, {Repository: "novox/b", Number: 7, Head: h2}}
return f
}()},
{name: "a merge the gate does not hold", raws: map[string][]byte{"solo": lone, "bad": []byte("{")}, repository: "novox/a", merge: c2, fail: "hold no merge"},
} {
t.Run(c.name, func(t *testing.T) {
raw, err := MergedOf(c.raws, c.repository, c.merge, trees, files)
if c.fail != "" {
if err == nil || !strings.Contains(err.Error(), c.fail) {
t.Fatalf("err %v, want %q", err, c.fail)
}
return
}
want, _ := forgemsg.EncodeForge(c.want)
if err != nil || !bytes.Equal(raw, want) {
t.Fatalf("err %v, answer\n%s\nwant the published message\n%s", err, raw, want)
}
})
}
if _, err := MergedOf(map[string][]byte{"solo": lone}, "novox/a", c1, func(string, string, string) (string, map[string]string, error) {
return "", nil, errors.New("the forge does not answer")
}, files); err == nil || !strings.Contains(err.Error(), "does not answer") {
t.Errorf("a forge that does not answer: %v", err)
}
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go
+2 -49
View File
@@ -5,7 +5,6 @@ import (
"encoding/json"
"errors"
"fmt"
"regexp"
"strings"
"time"
@@ -66,58 +65,12 @@ func callerOn(nc *nats.Conn, user string) caller {
}
}
// commitTreesTool is the forge's tool for a commit's trees (mesh-catalog modules/gitea, issue 1234): arguments owner,
// repo and sha, answer {root, trees}, the root directory spelled "" (1234/AC4).
const commitTreesTool = "gitea.gitea_commit_trees"
// treesVia asks the forge for a commit's trees through call.
func treesVia(ctx context.Context, call caller) bridge.Trees {
return func(owner, repo, sha string) (string, map[string]string, error) {
raw, err := call(ctx, commitTreesTool, map[string]string{"owner": owner, "repo": repo, "sha": sha})
if err != nil {
return "", nil, err
}
var a struct {
Root string `json:"root"`
Trees map[string]string `json:"trees"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return "", nil, fmt.Errorf("the answer of %s is not {root, trees}: %w", commitTreesTool, err)
}
if a.Root == "" {
return "", nil, fmt.Errorf("the answer of %s names no root tree", commitTreesTool)
}
return a.Root, a.Trees, nil
}
return bridge.TreesVia(ctx, bridge.Caller(call))
}
// 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).
const getFileTool = "gitea.gitea_get_file"
// notFound is how the forge's error says 404 (client.ts request: `Gitea API ${path}: ${status} ${text}`; client.ts:670
// recognises it as /: 404 /).
var notFound = regexp.MustCompile(`: 404 `)
// filesVia asks the forge for a file through call. A file the forge says is not there is bridge.ErrNotFound.
func filesVia(ctx context.Context, call caller) bridge.Files {
return func(owner, repo, path, ref string) ([]byte, error) {
raw, err := call(ctx, getFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref})
if err != nil {
if notFound.MatchString(err.Error()) {
return nil, fmt.Errorf("%w: %v", bridge.ErrNotFound, err)
}
return nil, err
}
var a struct {
File *struct {
Content *string `json:"content"`
} `json:"file"`
}
if err := json.Unmarshal(raw, &a); err != nil || a.File == nil || a.File.Content == nil {
return nil, fmt.Errorf("the answer of %s is not {file: {content}}: %v", getFileTool, err)
}
return []byte(*a.File.Content), nil
}
return bridge.FilesVia(ctx, bridge.Caller(call))
}
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go
+2 -27
View File
@@ -5,8 +5,6 @@ import (
"errors"
"log"
"slices"
"strconv"
"strings"
"github.com/nats-io/nats.go/jetstream"
@@ -86,29 +84,6 @@ func (g *grouper) groupOf(ctx context.Context, f forgemsg.Forge) (delivery.Merge
return m, rev, true
}
// membersOf are a merge's approved members with their heads, in the order of their keys.
func membersOf(m delivery.Merge) []delivery.Head {
var out []delivery.Head
for _, k := range slices.Sorted(func(yield func(string) bool) {
for k := range m.Approved {
if !yield(k) {
return
}
}
}) {
at := strings.LastIndex(k, "@")
if at < 0 {
continue
}
n, err := strconv.Atoi(k[at+1:])
if err != nil {
continue
}
out = append(out, delivery.Head{Repository: k[:at], Number: n, Head: m.Approved[k]})
}
return out
}
// member sets the group and its approved members on a merged message's body (ADR 0633 §4); any other body, and a
// merge of no group, is left as it is.
func (g *grouper) member(ctx context.Context, data []byte) ([]byte, forgemsg.Forge, bool) {
@@ -120,7 +95,7 @@ func (g *grouper) member(ctx context.Context, data []byte) ([]byte, forgemsg.For
if !ok {
return data, f, false
}
f.Group, f.Members = m.Group, membersOf(m)
f.Group, f.Members = m.Group, groupkeys.MembersOf(m)
out, err := forgemsg.EncodeForge(f)
if err != nil {
g.logger.Printf("%s#%d: its merged message cannot name the group %q, written without: %v", f.Repository, f.Number, m.Group, err)
@@ -165,7 +140,7 @@ func (g *grouper) end(ctx context.Context, key string, rev uint64, value []byte,
}
}
end := forgemsg.ForgeGroupEnd{Group: m.Group}
for _, member := range membersOf(m) {
for _, member := range groupkeys.MembersOf(m) {
at := slices.IndexFunc(m.Pulls, func(p delivery.MergePull) bool {
return delivery.SameRepository(p.Repository, member.Repository) && p.Number == member.Number
})
modules/mesh-delivery-forge-bridge/internal/bridge/call.go
+79
View File
@@ -0,0 +1,79 @@
package bridge
import (
"context"
"encoding/json"
"fmt"
"regexp"
)
// Caller calls a module's tool by its manifest key (`<module>.<tool>`) with a JSON body and returns the tool's result:
// the bridge's process calls through the bus as its own principal, its tools bundle through the runtime (stdio.Ask).
type Caller func(ctx context.Context, key string, body any) (json.RawMessage, error)
// CommitTreesTool is the forge's tool for a commit's trees (mesh-catalog modules/gitea, issue 1234): arguments owner,
// repo and sha, answer {root, trees}, the root directory spelled "" (1234/AC4).
const CommitTreesTool = "gitea.gitea_commit_trees"
// TreesVia asks the forge for a commit's trees through call.
func TreesVia(ctx context.Context, call Caller) Trees {
return func(owner, repo, sha string) (string, map[string]string, error) {
raw, err := call(ctx, CommitTreesTool, map[string]string{"owner": owner, "repo": repo, "sha": sha})
if err != nil {
return "", nil, err
}
var a struct {
Root string `json:"root"`
Trees map[string]string `json:"trees"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return "", nil, fmt.Errorf("the answer of %s is not {root, trees}: %w", CommitTreesTool, err)
}
if a.Root == "" {
return "", nil, fmt.Errorf("the answer of %s names no root tree", CommitTreesTool)
}
return a.Root, a.Trees, 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).
const GetFileTool = "gitea.gitea_get_file"
// notFound is how the forge's error says 404 (client.ts request: `Gitea API ${path}: ${status} ${text}`; client.ts:670
// recognises it as /: 404 /).
var notFound = regexp.MustCompile(`: 404 `)
// FilesVia asks the forge for a file through call. A file the forge says is not there is ErrNotFound.
func FilesVia(ctx context.Context, call Caller) Files {
return func(owner, repo, path, ref string) ([]byte, error) {
raw, err := call(ctx, GetFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref})
if err != nil {
if notFound.MatchString(err.Error()) {
return nil, fmt.Errorf("%w: %v", ErrNotFound, err)
}
return nil, err
}
var a struct {
File *struct {
Content *string `json:"content"`
} `json:"file"`
}
if err := json.Unmarshal(raw, &a); err != nil || a.File == nil || a.File.Content == nil {
return nil, fmt.Errorf("the answer of %s is not {file: {content}}: %v", GetFileTool, err)
}
return []byte(*a.File.Content), nil
}
}
// MergedMessage is the merged message of a merge into the trunk, asked again: built as the bridge builds the one it writes
// on FORGE (it is that event's body through ConvertKept), from the pull request's owner, repository, number, branch, head
// and merge commit, which the gate's merges state holds. It reads the forge for every module directory (no kept inputs).
func MergedMessage(owner, repo string, number int, branch, head, merge string, trees Trees, files Files) (Message, error) {
body, err := json.Marshal(pull{Owner: owner, Repo: repo, Number: number, Head: branch, HeadSHA: head, Base: Trunk, MergeSHA: merge})
if err != nil {
return Message{}, err
}
return ConvertKept(MergedSubject, body, trees, files, nil)
}
modules/mesh-delivery-forge-bridge/internal/groups/groups.go
+25
View File
@@ -10,7 +10,9 @@ import (
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strconv"
"strings"
"time"
@@ -171,3 +173,26 @@ func Rows(merges map[string][]byte, performed map[string]bool, endedAt map[strin
}
func sortRows(rows []Row) { sort.Slice(rows, func(i, j int) bool { return rows[i].Key < rows[j].Key }) }
// MembersOf are a merge's approved members with their heads, in the order of their keys.
func MembersOf(m delivery.Merge) []delivery.Head {
var out []delivery.Head
for _, k := range slices.Sorted(func(yield func(string) bool) {
for k := range m.Approved {
if !yield(k) {
return
}
}
}) {
at := strings.LastIndex(k, "@")
if at < 0 {
continue
}
n, err := strconv.Atoi(k[at+1:])
if err != nil {
continue
}
out = append(out, delivery.Head{Repository: k[:at], Number: n, Head: m.Approved[k]})
}
return out
}
modules/mesh-delivery-forge-bridge/manifest_test.go
+2 -2
View File
@@ -112,8 +112,8 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}, {Name: "inputs", History: 1}, {Name: "end-asks", History: 1}}) { // 1279/AC6, 1341, 1126, 1585
t.Errorf("state = %+v, want the buckets heads, performed, inputs and end-asks", m.State)
}
// 1585/T18: the tools read the groups and ask the bridge to say a group's end again, through its state end-asks.
if want := []string{"bridge_group_end_again", "bridge_groups"}; !reflect.DeepEqual(sorted(m.Tools), want) {
// 1585/T18: the tools read the groups and ask the bridge to say a group's end again, through its state end-asks. The tool bridge_merged_message answers a merge's merged message again (issue 1585).
if want := []string{"bridge_group_end_again", "bridge_groups", "bridge_merged_message"}; !reflect.DeepEqual(sorted(m.Tools), want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// No other state, no bus, no seat, no event (ADR 0437 §4; ADR 0521: no check.status).
modules/mesh-delivery-forge-bridge/module.json
+2 -1
View File
@@ -18,7 +18,8 @@
],
"tools": [
"bridge_groups",
"bridge_group_end_again"
"bridge_group_end_again",
"bridge_merged_message"
],
"reads": [
"mesh-delivery-gate.merges",