Take-in holds a merge by reference, so a mesh-catalog merge over the state value cap is taken in (issue 1585, ADR 0633) #154
@@ -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)
|
||||
|
||||
+75
-1
@@ -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 {
|
||||
@@ -113,4 +118,73 @@ 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.
|
||||
t.Run("bridge_merged_message answers the merged message as published", func(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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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
|
||||
})
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -18,7 +18,8 @@
|
||||
],
|
||||
"tools": [
|
||||
"bridge_groups",
|
||||
"bridge_group_end_again"
|
||||
"bridge_group_end_again",
|
||||
"bridge_merged_message"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-gate.merges",
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// mergedTool is the forge bridge's tool that answers a merge's merged message again (issue 1585): arguments repository
|
||||
// and merge, answer {message}, the message whole as the bridge writes it on FORGE. The indexer's manifest lists it in
|
||||
// invokes, which is the grant to call it.
|
||||
const mergedTool = delivery.ForgeBridgeModule + ".bridge_merged_message"
|
||||
|
||||
// messageVia asks the forge bridge for a merge's merged message, for a held merge the indexer does not hold in memory (a
|
||||
// restart, a new build). An error is the bridge's own words.
|
||||
func messageVia(call toolCaller) func(ctx context.Context, repository, merge string) ([]byte, error) {
|
||||
return func(_ context.Context, repository, merge string) ([]byte, error) {
|
||||
raw, err := call(mergedTool, map[string]string{"repository": repository, "merge": merge})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a struct {
|
||||
Message json.RawMessage `json:"message"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil || len(a.Message) == 0 {
|
||||
return nil, fmt.Errorf("the answer of %s is not {message}: %v", mergedTool, err)
|
||||
}
|
||||
return a.Message, nil
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -62,6 +63,9 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
var running atomic.Pointer[string]
|
||||
first := "fp-the-indexer-that-runs"
|
||||
running.Store(&first)
|
||||
// The forge bridge's tool bridge_merged_message (issue 1585): answers the merged message published for a merge, by
|
||||
// repository@merge, as the bridge does; a restarted indexer holds none in memory and asks it.
|
||||
var bridgeMessages sync.Map
|
||||
js, ctx, candidates, _, restart := startIndexerRun(t, func(ctx context.Context, builds buildLog) {
|
||||
// The builder's log entries of two builds' ends (ADR 0571 clause 1), as the builder appends them.
|
||||
builds.end(t, "novox/mesh-delivery", "a", "modules/a")
|
||||
@@ -97,6 +101,23 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"runs": map[string]string{"fp": *running.Load()}}, "node": "n1"})
|
||||
_ = m.Respond(raw)
|
||||
}
|
||||
bridge := func(m *nats.Msg) {
|
||||
var ask struct{ Repository, Merge string }
|
||||
_ = json.Unmarshal(m.Data, &ask)
|
||||
data, ok := bridgeMessages.Load(ask.Repository + "@" + ask.Merge)
|
||||
if !ok {
|
||||
raw, _ := json.Marshal(map[string]any{"error": "the gate holds no merge of " + ask.Repository + " with that commit"})
|
||||
_ = m.Respond(raw)
|
||||
return
|
||||
}
|
||||
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"message": json.RawMessage(data.([]byte))}, "node": "n1"})
|
||||
_ = m.Respond(raw)
|
||||
}
|
||||
for _, subject := range []string{"mesh.mod.mesh-delivery-forge-bridge.call.bridge_merged_message.>", "mesh.mod.mesh-delivery-forge-bridge.tool.bridge_merged_message"} {
|
||||
if _, err := js.Conn().Subscribe(subject, bridge); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, subject := range []string{"mesh.seat.node-engine.call.composition.n1.>", "mesh.seat.node-engine.tool.composition.n1"} {
|
||||
if _, err := js.Conn().Subscribe(subject, answer); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -157,6 +178,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15
|
||||
t.Fatal(err)
|
||||
}
|
||||
event, _ := forgemsg.ForgeEvent(f.Repository, forgemsg.ForgeMerged)
|
||||
bridgeMessages.Store(f.Repository+"@"+f.Merge, data)
|
||||
ack, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event."+event, data)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
@@ -348,7 +348,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf,
|
||||
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
|
||||
ReadOpen: openVia(calls.Call), Packages: awaitedStore{packages}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call),
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph},
|
||||
OwnBuild: ownBuildVia(calls.Call, cred.Node), Graph: awaitedStore{graph}, Message: messageVia(calls.Call),
|
||||
// The take-in step (ADR 0633): its state pending and its stream take-in, both this delivery's, bound when they appear;
|
||||
// until then every merge is taken in at once, as before.
|
||||
Pending: awaitedStore{process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)},
|
||||
|
||||
@@ -18,6 +18,7 @@ type takerIn interface {
|
||||
JudgeAgain(ctx context.Context, module, digest string) index.TakenIn
|
||||
TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn
|
||||
DropPending(ctx context.Context, key, by, why string) index.TakenIn
|
||||
TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn
|
||||
}
|
||||
|
||||
// takenIn answers one value of the takeins state (the tool indexer_take_in, issue 1945): a value not yet answered takes the
|
||||
@@ -36,7 +37,10 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
|
||||
return
|
||||
}
|
||||
asked, r := "take-in", index.TakenIn{}
|
||||
if a.Pending != "" && a.Drop {
|
||||
if a.Merge != "" {
|
||||
asked, r = "hand take-in of a merge", ix.TakeInMerge(ctx, a.MergeRepository, a.Merge, a.Sequence, a.By, a.Why)
|
||||
a.Module = r.Module
|
||||
} else if a.Pending != "" && a.Drop {
|
||||
asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why)
|
||||
a.Module = a.Pending
|
||||
} else if a.Pending != "" {
|
||||
|
||||
@@ -123,6 +123,11 @@ type Indexer struct {
|
||||
ownMu sync.Mutex // guards ownFP
|
||||
ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild)
|
||||
judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go)
|
||||
// Message asks the forge bridge for the merged message of a merge, repository and merge commit in, the message whole out (a
|
||||
// pending value keeps a reference, not a copy: issue 1585); an error is the bridge's words. Nil asks none.
|
||||
Message func(ctx context.Context, repository, merge string) ([]byte, error)
|
||||
heldMu sync.Mutex
|
||||
held map[uint64]forgemsg.Forge // the merged messages Arrive was handed, by stream sequence, until their merge is taken in or dropped
|
||||
// Pending is the state of merges held for the indexer their revision names (pending.go, novox/hq ADR 0633 §1); nil
|
||||
// takes every merge in at once, as before the take-in step.
|
||||
Pending Store
|
||||
@@ -215,15 +220,26 @@ func (ix *Indexer) ApplyMessage(ctx context.Context, f forgemsg.Forge) (Applied,
|
||||
ix.logf("%s#%d: its config declarations are not written: %v", f.Repository, f.Number, err)
|
||||
}
|
||||
}
|
||||
return applied, ix.writeCandidate(ctx, f, digests, held)
|
||||
}
|
||||
|
||||
// removeCandidate takes a merged pull request out of its group's candidate and writes nothing else of its message: for a
|
||||
// merge a newer merge of the same repository covers (it needs only the repository, number and branch).
|
||||
func (ix *Indexer) removeCandidate(ctx context.Context, f forgemsg.Forge) error {
|
||||
return ix.writeCandidate(ctx, f, nil, nil)
|
||||
}
|
||||
|
||||
// writeCandidate writes the group's candidate with the pull request f names (taken out for a merged or closed one).
|
||||
func (ix *Indexer) writeCandidate(ctx context.Context, f forgemsg.Forge, digests map[string]string, held map[string]delivery.Module) error {
|
||||
if trunkLike[f.Branch] {
|
||||
ix.logf("%s#%d: its branch %q is no delivery group, so no candidate", f.Repository, f.Number, f.Branch)
|
||||
return applied, nil
|
||||
return nil
|
||||
}
|
||||
key, err := delivery.CandidatesKey(f.Branch)
|
||||
if err != nil {
|
||||
return applied, fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
|
||||
return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
|
||||
}
|
||||
return applied, write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
return write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
|
||||
var cur *delivery.Candidate
|
||||
if found {
|
||||
c, err := delivery.DecodeCandidate(value)
|
||||
|
||||
@@ -2,6 +2,7 @@ package index
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -27,6 +28,15 @@ type memStore struct {
|
||||
writes int
|
||||
beforeWrite func()
|
||||
failWrite func(key string) error // a write's own error, asked before each Create and Update
|
||||
maxValue int // the bucket's value cap (buscompose.StateMaxValueBytes); a larger value is refused as the bus refuses it
|
||||
}
|
||||
|
||||
// tooBig is the bus's refusal of a value over a bucket's cap.
|
||||
func (m *memStore) tooBig(value []byte) error {
|
||||
if m.maxValue > 0 && len(value) > m.maxValue {
|
||||
return errors.New("nats: API error: code=400 err_code=10054 description=message size exceeds maximum allowed")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} }
|
||||
@@ -68,6 +78,9 @@ func (m *memStore) Create(_ context.Context, key string, value []byte) error {
|
||||
if _, ok := m.vals[key]; ok {
|
||||
return ErrConflict
|
||||
}
|
||||
if err := m.tooBig(value); err != nil {
|
||||
return err
|
||||
}
|
||||
m.writes++
|
||||
m.put(key, value)
|
||||
return nil
|
||||
@@ -82,6 +95,9 @@ func (m *memStore) Update(_ context.Context, key string, value []byte, rev uint6
|
||||
if _, ok := m.vals[key]; !ok || m.revs[key] != rev {
|
||||
return ErrConflict
|
||||
}
|
||||
if err := m.tooBig(value); err != nil {
|
||||
return err
|
||||
}
|
||||
m.writes++
|
||||
m.put(key, value)
|
||||
return nil
|
||||
|
||||
@@ -44,14 +44,32 @@ const NoBuild = "no build: "
|
||||
// the message is left unacknowledged, so the bus delivers it again, and nothing is written.
|
||||
var ErrOwnBuildUnknown = errors.New("the build of the indexer this program runs is not known")
|
||||
|
||||
// Pending is one value of the pending state: the merged message whole, as the forge bridge encoded it, the group it
|
||||
// PendingRef is what a pending value keeps of its merged message instead of a copy (a merge of a repository with many
|
||||
// modules is about 0.5 MB, over the state's 256 KiB value cap; issue 1585): the small fields the waits are worked out
|
||||
// from. The message itself is read from the forge bridge's stream by Pending.Seq when a take-in needs it.
|
||||
type PendingRef struct {
|
||||
Repository string `json:"repository"`
|
||||
Number int `json:"number,omitempty"`
|
||||
Merge string `json:"merge"`
|
||||
// Branch is the pull request's branch, which names its candidate: a merge covered by a newer one is taken out of it.
|
||||
Branch string `json:"branch,omitempty"`
|
||||
Members []delivery.Head `json:"members,omitempty"`
|
||||
// Holds are the modules the merge's manifests name, which indexer_pending says it holds back.
|
||||
Holds []string `json:"holds,omitempty"`
|
||||
}
|
||||
|
||||
// Pending is one value of the pending state: a reference to the merged message (Seq, its sequence on the forge bridge's
|
||||
// stream, and Ref), or, in a value an earlier indexer wrote, the message whole as the forge bridge encoded it; the group it
|
||||
// was merged in (Ended once the group's end was heard), since when it is held, the indexer build its revision waits
|
||||
// for (absent while it waits for nothing yet, or behind a group being gathered), the build of the indexer that last
|
||||
// wrote it, the wait it was last said under on the take-in stream (Said, SaidAt), and Taking, set by a take-in that
|
||||
// claimed it: a second indexer that finds it claimed takes nothing in (ADR 0633 §1: never two take-ins of one merge).
|
||||
type Pending struct {
|
||||
V int `json:"v"`
|
||||
Forge json.RawMessage `json:"forge"`
|
||||
Seq uint64 `json:"seq,omitempty"`
|
||||
Ref *PendingRef `json:"ref,omitempty"`
|
||||
Forge json.RawMessage `json:"forge,omitempty"`
|
||||
Lost string `json:"lost,omitempty"`
|
||||
Group string `json:"group,omitempty"`
|
||||
Ended bool `json:"ended,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
@@ -63,12 +81,19 @@ type Pending struct {
|
||||
Taking string `json:"taking,omitempty"`
|
||||
}
|
||||
|
||||
// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded.
|
||||
// PendingEntry is one key of the pending state as read: its key, revision, value and the merged message decoded. For a
|
||||
// value that keeps a reference, Forge holds only the fields of the reference (Light) until Indexer.load reads the message
|
||||
// from the stream; Holds are the modules the merge's manifests name.
|
||||
type PendingEntry struct {
|
||||
Key string
|
||||
Revision uint64
|
||||
Value Pending
|
||||
Forge forgemsg.Forge
|
||||
Light bool
|
||||
Holds []string
|
||||
// Covered says a newer merge of the same repository is in the same revision: its message is not read or applied, since
|
||||
// applying the newest writes every module whose manifest differs from what is held (modules.go).
|
||||
Covered bool
|
||||
}
|
||||
|
||||
// DecodePending reads one pending value and its merged message.
|
||||
@@ -80,6 +105,11 @@ func DecodePending(data []byte) (Pending, forgemsg.Forge, error) {
|
||||
if p.V != 1 {
|
||||
return p, forgemsg.Forge{}, fmt.Errorf("a pending value of version %d, not 1", p.V)
|
||||
}
|
||||
if len(p.Forge) == 0 && p.Ref != nil {
|
||||
r := p.Ref
|
||||
return p, forgemsg.Forge{V: delivery.Version, Kind: forgemsg.ForgeMerged, Repository: r.Repository, Number: r.Number,
|
||||
Merge: r.Merge, Branch: r.Branch, Group: p.Group, Members: r.Members}, nil
|
||||
}
|
||||
f, err := forgemsg.DecodeForge(p.Forge)
|
||||
if err != nil {
|
||||
return p, f, fmt.Errorf("its merged message: %w", err)
|
||||
@@ -119,11 +149,115 @@ func ReadPending(ctx context.Context, st Store, logf func(string, ...any)) ([]Pe
|
||||
}
|
||||
continue
|
||||
}
|
||||
out = append(out, PendingEntry{Key: key, Revision: rev, Value: p, Forge: f})
|
||||
e := PendingEntry{Key: key, Revision: rev, Value: p, Forge: f, Light: len(p.Forge) == 0}
|
||||
if p.Ref != nil {
|
||||
e.Holds = p.Ref.Holds
|
||||
} else {
|
||||
e.Holds = holdsOf(f)
|
||||
}
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// holdsOf are the modules a merged message's manifests name, sorted.
|
||||
func holdsOf(f forgemsg.Forge) []string {
|
||||
var out []string
|
||||
for _, raw := range f.Manifests {
|
||||
if name := manifestModule(raw); name != "" && !slices.Contains(out, name) {
|
||||
out = append(out, name)
|
||||
}
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// refOf is what a pending value keeps of the merged message f (PendingRef).
|
||||
func refOf(f forgemsg.Forge) *PendingRef {
|
||||
return &PendingRef{Repository: f.Repository, Number: f.Number, Merge: f.Merge, Branch: f.Branch, Members: f.Members, Holds: holdsOf(f)}
|
||||
}
|
||||
|
||||
// heldMessage is the message Arrive was handed for a sequence, kept in memory until its merge is taken in or dropped.
|
||||
func (ix *Indexer) heldMessage(seq uint64) (forgemsg.Forge, bool) {
|
||||
ix.heldMu.Lock()
|
||||
defer ix.heldMu.Unlock()
|
||||
f, ok := ix.held[seq]
|
||||
return f, ok
|
||||
}
|
||||
|
||||
func (ix *Indexer) hold(seq uint64, f forgemsg.Forge) {
|
||||
ix.heldMu.Lock()
|
||||
defer ix.heldMu.Unlock()
|
||||
if ix.held == nil {
|
||||
ix.held = map[uint64]forgemsg.Forge{}
|
||||
}
|
||||
ix.held[seq] = f
|
||||
}
|
||||
|
||||
func (ix *Indexer) forget(entries ...PendingEntry) {
|
||||
ix.heldMu.Lock()
|
||||
defer ix.heldMu.Unlock()
|
||||
for _, e := range entries {
|
||||
delete(ix.held, e.Value.Seq)
|
||||
}
|
||||
}
|
||||
|
||||
// load gives the entry the merged message whole: the copy an earlier value holds, the message Arrive was handed (kept in
|
||||
// memory), else the message read from the forge bridge's stream by the value's sequence. A stream that no longer holds it
|
||||
// answers ErrForgeGone, a bus that refuses the read its error; a message that is not the merge the value refers to (a
|
||||
// stream rebuilt under the same sequence) is refused.
|
||||
func (ix *Indexer) load(ctx context.Context, e PendingEntry) (PendingEntry, error) {
|
||||
if !e.Light {
|
||||
return e, nil
|
||||
}
|
||||
if f, ok := ix.heldMessage(e.Value.Seq); ok {
|
||||
e.Forge, e.Light = f, false
|
||||
return e, nil
|
||||
}
|
||||
if ix.Message == nil {
|
||||
return e, fmt.Errorf("the forge bridge is not asked by this indexer, so %s (%s#%d) cannot be read", e.Key, e.Forge.Repository, e.Forge.Number)
|
||||
}
|
||||
data, err := ix.Message(ctx, e.Forge.Repository, e.Forge.Merge)
|
||||
if err != nil {
|
||||
return e, fmt.Errorf("the forge bridge could not answer for the merge of %s#%d at %s: %w", e.Forge.Repository, e.Forge.Number, e.Forge.Merge, err)
|
||||
}
|
||||
f, err := forgemsg.DecodeForge(data)
|
||||
if err != nil {
|
||||
return e, fmt.Errorf("the forge bridge's answer for %s is not a forge message: %w", e.Key, err)
|
||||
}
|
||||
if f.Kind != forgemsg.ForgeMerged || f.Repository != e.Forge.Repository || f.Merge != e.Forge.Merge {
|
||||
return e, fmt.Errorf("the forge bridge's answer is not the merge %s#%d at %s that %s refers to", e.Forge.Repository, e.Forge.Number, e.Forge.Merge, e.Key)
|
||||
}
|
||||
e.Forge, e.Light = f, false
|
||||
return e, nil
|
||||
}
|
||||
|
||||
// loadNewest prepares the ready merges of one revision, in merge order. Of the merges of one repository only the newest is
|
||||
// loaded: applying it writes every module whose manifest differs from what is held, so it covers the older ones, which are
|
||||
// marked Covered and never read. A newest merge the forge bridge cannot give is answered in lost with its reason, by key, and its revision is held (holdLost).
|
||||
func (ix *Indexer) loadNewest(ctx context.Context, entries []PendingEntry) (out, lost []PendingEntry, reasons map[string]string) {
|
||||
out = slices.Clone(entries)
|
||||
reasons = map[string]string{}
|
||||
newest := map[string]bool{}
|
||||
for i := len(out) - 1; i >= 0; i-- {
|
||||
repo := strings.ToLower(out[i].Forge.Repository)
|
||||
if newest[repo] {
|
||||
out[i].Covered = true
|
||||
continue
|
||||
}
|
||||
newest[repo] = true
|
||||
full, err := ix.load(ctx, out[i])
|
||||
if err != nil {
|
||||
lost = append(lost, out[i])
|
||||
reasons[out[i].Key] = err.Error()
|
||||
continue
|
||||
}
|
||||
out[i] = full
|
||||
}
|
||||
slices.Reverse(lost)
|
||||
return out, lost, reasons
|
||||
}
|
||||
|
||||
// Waits is the pending state read as what waits (ADR 0633 §3, §4): Ready are the merges no group still being gathered
|
||||
// holds back, in merge order, which are one revision together; Groups are the groups still being gathered, each with
|
||||
// its members held so far; Behind are the merges in a gathered group's member repositories that wait behind it, by the
|
||||
@@ -268,12 +402,9 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f forgemsg.Forge) (Ap
|
||||
}
|
||||
ix.pendingMu.Lock()
|
||||
defer ix.pendingMu.Unlock()
|
||||
raw, err := forgemsg.EncodeForge(f)
|
||||
if err != nil {
|
||||
return Applied{}, "", fmt.Errorf("%w: %v", ErrUnreadable, err)
|
||||
}
|
||||
key := takein.PendingKey(seq)
|
||||
data, err := encodePending(Pending{Forge: raw, Group: f.Group, Since: ix.now(), LastWriter: own})
|
||||
ix.hold(seq, f)
|
||||
data, err := encodePending(Pending{Seq: seq, Ref: refOf(f), Group: f.Group, Since: ix.now(), LastWriter: own})
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
@@ -461,18 +592,65 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie
|
||||
}
|
||||
}
|
||||
}
|
||||
names, dry, err := ix.computeOver(ctx, w.Ready)
|
||||
// Of each repository's merges only the newest is read (from memory, else asked of the forge bridge: a pending value
|
||||
// keeps a reference); the older ones are covered by it. A newest merge that cannot be read is said and
|
||||
// holds its revision: nothing of it is taken in silently.
|
||||
ready, lost, reasons := ix.loadNewest(ctx, w.Ready)
|
||||
if len(lost) > 0 {
|
||||
return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, lost, reasons)
|
||||
}
|
||||
names, dry, err := ix.computeOver(ctx, ready)
|
||||
if err != nil {
|
||||
return Applied{}, "", err
|
||||
}
|
||||
if names == own {
|
||||
a, err := ix.takeInWhole(ctx, w.Ready, own, "")
|
||||
a, err := ix.takeInWhole(ctx, ready, own, "")
|
||||
if errors.Is(err, errClaimLost) {
|
||||
return a, "claimed by another take-in", nil
|
||||
}
|
||||
return a, ix.waitingLine(w), err
|
||||
}
|
||||
return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, w.Ready, names, own, dry)
|
||||
return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, ready, names, own, dry)
|
||||
}
|
||||
|
||||
// holdLost holds a revision whose newest merge of a repository cannot be read (the forge bridge's stream no longer holds
|
||||
// it, or the bus refuses the read): said once on take-in.pending with the lost keys (TakeIn.Lost), and on each lost entry
|
||||
// (Pending.Lost, the reason, which indexer_pending shows), until a newer merge of its repository covers it or
|
||||
// indexer_drop_pending drops it. Nothing of the revision is taken in meanwhile.
|
||||
func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, reasons map[string]string) error {
|
||||
t := revisionWait(takein.TakeInPending, ready, "")
|
||||
for _, g := range lost {
|
||||
t.Lost = append(t.Lost, g.Key)
|
||||
}
|
||||
said := true
|
||||
for _, e := range ready {
|
||||
isLost := slices.Contains(t.Lost, e.Key)
|
||||
if e.Value.Said != t.Key || e.Value.SaidKind != takein.TakeInPending || (isLost && e.Value.Lost == "") {
|
||||
said = false
|
||||
}
|
||||
}
|
||||
if said {
|
||||
return nil
|
||||
}
|
||||
for _, g := range lost {
|
||||
ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision, the forge bridge could not give its message: %s; a newer merge of %s covers it, indexer_drop_pending drops it",
|
||||
g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], g.Forge.Repository)
|
||||
}
|
||||
if !ix.sayTakeIn(ctx, t) {
|
||||
return nil
|
||||
}
|
||||
var kept []PendingEntry
|
||||
for _, e := range ready {
|
||||
if !slices.Contains(t.Lost, e.Key) {
|
||||
kept = append(kept, e)
|
||||
}
|
||||
}
|
||||
for _, g := range lost {
|
||||
reason := reasons[g.Key]
|
||||
ix.markSaid(ctx, []PendingEntry{g}, &t, func(p *Pending) { p.Lost = reason })
|
||||
}
|
||||
ix.markSaid(ctx, kept, &t, nil)
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitingLine says what waits after an evaluation, for the heard state.
|
||||
@@ -510,7 +688,7 @@ func (ix *Indexer) dry() *dryRun {
|
||||
d := &dryRun{}
|
||||
x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum,
|
||||
GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains,
|
||||
ReadOpen: nil, OwnBuild: nil, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil}
|
||||
ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil}
|
||||
x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) }
|
||||
x.Store, d.candidates = over(ix.Store)
|
||||
x.Modules, d.modules = over(ix.Modules)
|
||||
@@ -544,6 +722,9 @@ func (ix *Indexer) dry() *dryRun {
|
||||
func (ix *Indexer) computeOver(ctx context.Context, ready []PendingEntry) (string, *dryRun, error) {
|
||||
d := ix.dry()
|
||||
for _, e := range ready {
|
||||
if e.Covered {
|
||||
continue
|
||||
}
|
||||
if _, err := d.ix.ApplyMessage(ctx, e.Forge); err != nil {
|
||||
if errors.Is(err, ErrUnreadable) {
|
||||
continue
|
||||
@@ -646,7 +827,23 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow
|
||||
return Applied{}, errClaimLost
|
||||
}
|
||||
var applied Applied
|
||||
newest := map[string]PendingEntry{}
|
||||
for _, e := range claimed {
|
||||
if !e.Covered {
|
||||
newest[strings.ToLower(e.Forge.Repository)] = e
|
||||
}
|
||||
}
|
||||
for _, e := range claimed {
|
||||
if e.Covered {
|
||||
by := newest[strings.ToLower(e.Forge.Repository)]
|
||||
ix.logf("take-in: %s (%s#%d) is covered by %s, the newest merge of %s in this revision: its message is not read, and it is ended with it", e.Key, e.Forge.Repository, e.Forge.Number, by.Key, e.Forge.Repository)
|
||||
// Its pull request leaves its group's candidate as a merged message takes it out; nothing else of it is applied.
|
||||
if err := ix.removeCandidate(ctx, e.Forge); err != nil {
|
||||
ix.release(ctx, claimed)
|
||||
return applied, fmt.Errorf("taking %s (%s#%d) out of its candidate: %w", e.Key, e.Forge.Repository, e.Forge.Number, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
a, err := ix.ApplyMessage(ctx, e.Forge)
|
||||
applied.Written = append(applied.Written, a.Written...)
|
||||
if a.Late != "" {
|
||||
@@ -666,6 +863,7 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow
|
||||
ix.logf("take-in: %s.%s is taken in and not removed: %v", PendingState, e.Key, err)
|
||||
}
|
||||
}
|
||||
ix.forget(claimed...)
|
||||
ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key)
|
||||
for _, t := range saidWaits(claimed) {
|
||||
t.By, t.Why = who, why
|
||||
@@ -809,11 +1007,14 @@ func waitOf(entries []PendingEntry, what, behind string) takein.PendingWait {
|
||||
}
|
||||
for _, e := range entries {
|
||||
pw.Keys = append(pw.Keys, e.Key)
|
||||
for _, raw := range e.Forge.Manifests {
|
||||
if name := manifestModule(raw); name != "" && !slices.Contains(pw.Holds, name) {
|
||||
for _, name := range e.Holds {
|
||||
if !slices.Contains(pw.Holds, name) {
|
||||
pw.Holds = append(pw.Holds, name)
|
||||
}
|
||||
}
|
||||
if e.Value.Lost != "" {
|
||||
pw.Lost = append(pw.Lost, e.Key)
|
||||
}
|
||||
if e.Value.Said != "" && e.Value.SaidAt.After(pw.SaidAt) {
|
||||
pw.Said, pw.SaidAt = "take-in."+e.Value.SaidKind, e.Value.SaidAt
|
||||
}
|
||||
@@ -856,6 +1057,11 @@ func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string)
|
||||
r.Answer = fmt.Sprintf("nothing waits under %s: indexer_pending names the waits", key)
|
||||
return r
|
||||
}
|
||||
take, lost, reasons := ix.loadNewest(ctx, take)
|
||||
if len(lost) > 0 {
|
||||
r.Answer = fmt.Sprintf("nothing is taken in: the message of %s is not read (%s); a newer merge of its repository covers it, indexer_drop_pending drops it", lost[0].Key, reasons[lost[0].Key])
|
||||
return r
|
||||
}
|
||||
ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why)
|
||||
hand := fmt.Sprintf("%s (%s)", by, why)
|
||||
if _, err := ix.takeInWholeBy(ctx, take, own, hand, by, why); err != nil {
|
||||
@@ -897,6 +1103,7 @@ func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn
|
||||
r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err)
|
||||
return r
|
||||
}
|
||||
ix.forget(e)
|
||||
ix.logf("take-in: %s (%s#%d at %s) dropped by hand, asked by %s: %s; nothing of it is taken in", key, e.Forge.Repository, e.Forge.Number, e.Forge.Merge, by, why)
|
||||
if e.Value.Said != "" && e.Value.SaidKind == takein.TakeInPending {
|
||||
var same []PendingEntry
|
||||
@@ -919,3 +1126,42 @@ func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// TakeInMerge is the hand take-in of one merge (indexer_take_in_merge, issue 1585): its merged message is asked of the forge
|
||||
// bridge and handed to the take-in as a delivered merged message is (Arrive, under the stream sequence seq, which keeps the
|
||||
// merge order), so it is held for the indexer its revision names or taken in, and a merge already held under that
|
||||
// sequence is not written twice. For a merge the take-in never took in. The bridge's own words answer when it cannot give
|
||||
// the message.
|
||||
func (ix *Indexer) TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) TakenIn {
|
||||
r := TakenIn{Module: takein.PendingKey(seq)}
|
||||
if ix.Message == nil {
|
||||
r.Answer = "this indexer does not ask the forge bridge"
|
||||
return r
|
||||
}
|
||||
data, err := ix.Message(ctx, repository, merge)
|
||||
if err != nil {
|
||||
r.Answer = fmt.Sprintf("the forge bridge could not answer for the merge of %s at %s: %v", repository, merge, err)
|
||||
return r
|
||||
}
|
||||
f, err := forgemsg.DecodeForge(data)
|
||||
if err != nil {
|
||||
r.Answer = fmt.Sprintf("the forge bridge's answer for %s at %s is not a forge message: %v", repository, merge, err)
|
||||
return r
|
||||
}
|
||||
if f.Kind != forgemsg.ForgeMerged || f.Merge != merge || f.Repository != repository {
|
||||
r.Answer = fmt.Sprintf("the forge bridge's answer is not the merge of %s at %s", repository, merge)
|
||||
return r
|
||||
}
|
||||
ix.logf("take-in: a hand take-in of the merge %s#%d at %s (sequence %d) asked by %s: %s", f.Repository, f.Number, f.Merge, seq, by, why)
|
||||
a, words, err := ix.Arrive(ctx, seq, f)
|
||||
if err != nil {
|
||||
r.Answer = err.Error()
|
||||
return r
|
||||
}
|
||||
r.Stored, r.Repository, r.Commit = true, f.Repository, f.Merge
|
||||
r.Answer = fmt.Sprintf("%s#%d at %s handed to the take-in as a delivered merge (%d keys written)", f.Repository, f.Number, f.Merge, len(a.Written))
|
||||
if words != "" {
|
||||
r.Answer += "; " + words
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
)
|
||||
@@ -22,6 +23,7 @@ type takeInBed struct {
|
||||
mods *memStore
|
||||
pending *memStore
|
||||
said []takein.TakeIn
|
||||
bridge map[string][]byte // what the forge bridge answers (bridge_merged_message), by repository@merge
|
||||
base forgemsg.Forge
|
||||
logs *strings.Builder
|
||||
}
|
||||
@@ -35,9 +37,16 @@ func newTakeInBed(t *testing.T) *takeInBed {
|
||||
}
|
||||
reportToolchain(t, ix, "mesh-build-go", "novox", "go", "go1.27.1")
|
||||
reportToolchain(t, ix, "mesh-build-go", "ace", "go", "go1.27.1")
|
||||
b := &takeInBed{ix: ix, mods: mods, pending: newMem(), base: f, logs: &strings.Builder{}}
|
||||
b := &takeInBed{ix: ix, mods: mods, pending: newMem(), bridge: map[string][]byte{}, base: f, logs: &strings.Builder{}}
|
||||
ix.Verdicts = newMem()
|
||||
b.pending.maxValue = buscompose.StateMaxValueBytes
|
||||
ix.Pending = b.pending
|
||||
ix.Message = func(_ context.Context, repository, merge string) ([]byte, error) {
|
||||
if data, ok := b.bridge[repository+"@"+merge]; ok {
|
||||
return data, nil
|
||||
}
|
||||
return nil, errors.New("the gate holds no merge of " + repository + " with that commit")
|
||||
}
|
||||
ix.SayTakeIn = func(_ context.Context, t takein.TakeIn) error { b.said = append(b.said, t); return nil }
|
||||
logf := ix.Logf
|
||||
ix.Logf = func(format string, args ...any) {
|
||||
@@ -85,11 +94,22 @@ func (b *takeInBed) merge(number int, c, avahi, indexer string) forgemsg.Forge {
|
||||
|
||||
func (b *takeInBed) arrive(t *testing.T, seq uint64, f forgemsg.Forge) {
|
||||
t.Helper()
|
||||
b.publish(t, f)
|
||||
if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil {
|
||||
t.Fatalf("arrive %d: %v", seq, err)
|
||||
}
|
||||
}
|
||||
|
||||
// publish lets the bed's forge bridge answer the merged message f, as bridge_merged_message does.
|
||||
func (b *takeInBed) publish(t *testing.T, f forgemsg.Forge) {
|
||||
t.Helper()
|
||||
data, err := forgemsg.EncodeForge(f)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.bridge[f.Repository+"@"+f.Merge] = data
|
||||
}
|
||||
|
||||
func (b *takeInBed) keys() []string {
|
||||
k, _ := b.pending.Keys(context.Background())
|
||||
slices.Sort(k)
|
||||
@@ -179,7 +199,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
// The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits.
|
||||
next := &Indexer{}
|
||||
*next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum}
|
||||
next.toolchains = b.ix.toolchains
|
||||
asked := 0
|
||||
next.OwnBuild = func(context.Context) (string, error) {
|
||||
@@ -333,9 +353,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
t.Fatalf("%s: pending %v, claimed %v: want both pending and no claim left", c.name, b.keys(), got)
|
||||
}
|
||||
// The next take-in, not at a start, is not refused by a claim left behind: it takes both in.
|
||||
if _, _, err := b.ix.Arrive(ctx, 4, b.merge(23, sha('e'), "tree-avahi-5", "tree-indexer-2")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", "tree-indexer-2"))
|
||||
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('e') {
|
||||
t.Errorf("%s: after: pending %v, avahi at %s: want all taken in", c.name, b.keys(), b.commit(t, "avahi"))
|
||||
}
|
||||
@@ -354,7 +372,8 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
key, _ := delivery.ToolchainKey(node, "go")
|
||||
x.PutToolchain("mesh-build-go", key, toolchainValue(t, "go", "go1.27.2", "1"), false)
|
||||
}
|
||||
next, _, err := x.computeOver(ctx, WaitsOf(entries).Ready)
|
||||
ready, _, _ := b.ix.loadNewest(ctx, WaitsOf(entries).Ready)
|
||||
next, _, err := x.computeOver(ctx, ready)
|
||||
if err != nil || next == "" || next == b.said[0].WaitsFor {
|
||||
t.Fatalf("the build under the next toolchain: %q %v (the revision names %s)", next, err, b.said[0].WaitsFor)
|
||||
}
|
||||
@@ -403,12 +422,105 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said)
|
||||
}
|
||||
})
|
||||
t.Run("1585 a merge over the value cap is held by reference, a new indexer with an empty memory takes it in through the forge bridge, and a covered pull request leaves its candidate", func(t *testing.T) {
|
||||
big := func(f forgemsg.Forge) forgemsg.Forge {
|
||||
f.Words = strings.Repeat("x", 300*1024)
|
||||
f.Branch = "feat/g"
|
||||
return f
|
||||
}
|
||||
b := newTakeInBed(t)
|
||||
avahi0 := b.fp(t, "avahi")
|
||||
// The forge bridge is not asked at all while the merges are in memory: a 300 KiB merge, over the bucket's cap whole,
|
||||
// is held by reference and a lone one taken in.
|
||||
bridgeDown := func(context.Context, string, string) ([]byte, error) {
|
||||
return nil, errors.New("the bridge does not answer")
|
||||
}
|
||||
answers := b.ix.Message
|
||||
b.ix.Message = bridgeDown
|
||||
b.arrive(t, 1, big(b.merge(20, sha('b'), "tree-avahi-2", "")))
|
||||
if b.fp(t, "avahi") == avahi0 || b.commit(t, "avahi") != sha('b') || len(b.keys()) != 0 || len(b.ix.held) != 0 {
|
||||
t.Fatalf("a big merge alone: avahi at %s, pending %v, in memory %d: want it taken in and forgotten", b.commit(t, "avahi"), b.keys(), len(b.ix.held))
|
||||
}
|
||||
// Two pull requests of one group are open; both then merge (the indexer changed: held for the new build).
|
||||
for n, head := range map[int]string{21: sha('1'), 22: sha('2')} {
|
||||
if err := b.ix.Apply(ctx, pull(catalogueRepo, n, "feat/g", head, "modules/avahi")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
key, _ := delivery.CandidatesKey("feat/g")
|
||||
if _, _, found, _ := b.ix.Store.Get(ctx, key); !found {
|
||||
t.Fatal("the group has no candidate")
|
||||
}
|
||||
b.arrive(t, 2, big(b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")))
|
||||
b.arrive(t, 3, big(b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")))
|
||||
raw, _, _, _ := b.pending.Get(ctx, takein.PendingKey(2))
|
||||
if len(b.keys()) != 2 || len(raw) > 4096 || b.commit(t, "avahi") != sha('b') {
|
||||
t.Fatalf("a big merge held: pending %v, value %d bytes, avahi at %s: want two small values", b.keys(), len(raw), b.commit(t, "avahi"))
|
||||
}
|
||||
named := b.fp(t, delivery.IndexerModule)
|
||||
// The new indexer starts with nothing in memory and the bridge down: the newest merge is said lost with the bridge's
|
||||
// words (event, state), the older one is covered and not lost, and nothing is taken in.
|
||||
next := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts,
|
||||
Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: bridgeDown, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf,
|
||||
GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }}
|
||||
next.toolchains = b.ix.toolchains
|
||||
b.said = nil
|
||||
if _, _, err := next.TakeInAtStart(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waits := PendingWaits(mustReadPending(t, b))
|
||||
if len(b.said) != 1 || !slices.Equal(b.said[0].Lost, []string{takein.PendingKey(3)}) || len(waits) != 1 || !slices.Equal(waits[0].Lost, []string{takein.PendingKey(3)}) || b.commit(t, "avahi") != sha('b') {
|
||||
t.Fatalf("lost: said %+v, waits %+v, avahi at %s: want m.3 lost on the event and in indexer_pending, nothing taken in", b.said, waits, b.commit(t, "avahi"))
|
||||
}
|
||||
if v, _, _, _ := b.pending.Get(ctx, takein.PendingKey(3)); !strings.Contains(string(v), "the bridge does not answer") {
|
||||
t.Errorf("m.3 = %s: want the bridge's words", v)
|
||||
}
|
||||
if err := next.PendingAgain(ctx); err != nil || len(b.said) != 1 {
|
||||
t.Errorf("lost again: err %v, said %v: want it said once", err, b.events())
|
||||
}
|
||||
if r := next.TakeInPendingByHand(ctx, takein.PendingKey(2), "lead@shanks", "try"); r.Stored || b.commit(t, "avahi") != sha('b') {
|
||||
t.Errorf("hand take-in of a lost merge: %+v, avahi at %s: want nothing taken in", r, b.commit(t, "avahi"))
|
||||
}
|
||||
// The bridge answers: the newest merge is read through it, the older one is covered, and the revision is taken in.
|
||||
b.publish(t, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) // the answer is the message as published, small here
|
||||
next.Message = answers
|
||||
b.bridge[catalogueRepo+"@"+sha('d')], _ = forgemsg.EncodeForge(big(b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")))
|
||||
b.logs.Reset()
|
||||
if err := next.PendingAgain(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || !strings.Contains(b.logs.String(), "is covered by") {
|
||||
t.Fatalf("taken in: pending %v, avahi at %s, journal %q: want both merges taken in, the older covered", b.keys(), b.commit(t, "avahi"), b.logs.String())
|
||||
}
|
||||
if _, _, found, _ := b.ix.Store.Get(ctx, key); found {
|
||||
t.Errorf("the group's candidate stays after both its pull requests merged (the covered one is not taken out)")
|
||||
}
|
||||
// A hand take-in of a merge by its commit asks the bridge and runs the message through Arrive.
|
||||
b.ix.Message = answers
|
||||
b.publish(t, big(b.merge(30, sha('e'), "tree-avahi-6", b.base.Trees["modules/"+delivery.IndexerModule])))
|
||||
if r := b.ix.TakeInMerge(ctx, catalogueRepo, sha('e'), 9, "lead@shanks", "held by the cap"); !r.Stored || b.commit(t, "avahi") != sha('e') {
|
||||
t.Fatalf("merge take-in: %+v, avahi at %s: want it taken in", r, b.commit(t, "avahi"))
|
||||
}
|
||||
if r := b.ix.TakeInMerge(ctx, catalogueRepo, sha('9'), 10, "lead@shanks", "no such"); r.Stored || !strings.Contains(r.Answer, "gate holds no merge") {
|
||||
t.Errorf("merge take-in of a merge the bridge does not know: %+v", r)
|
||||
}
|
||||
// An old value that holds the message whole reads as before.
|
||||
old := b.merge(31, sha('f'), "tree-avahi-7", b.base.Trees["modules/"+delivery.IndexerModule])
|
||||
rawOld, _ := forgemsg.EncodeForge(old)
|
||||
data, _ := encodePending(Pending{Forge: rawOld, Since: b.ix.now(), LastWriter: b.fp(t, delivery.IndexerModule)})
|
||||
if err := b.pending.Create(ctx, takein.PendingKey(11), data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := b.ix.PendingAgain(ctx); err != nil || b.commit(t, "avahi") != sha('f') {
|
||||
t.Errorf("an old value with the message whole: err %v, avahi at %s: want it taken in", err, b.commit(t, "avahi"))
|
||||
}
|
||||
})
|
||||
t.Run("the dry run names every field of the indexer", func(t *testing.T) {
|
||||
// dry (pending.go) builds an indexer field by field: a field added to Indexer is named there, or this fails.
|
||||
named := map[string]bool{}
|
||||
for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections",
|
||||
"JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk",
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Pending", "SayTakeIn"} {
|
||||
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn"} {
|
||||
named[f] = true
|
||||
}
|
||||
ty := reflect.TypeOf(Indexer{})
|
||||
@@ -419,3 +531,12 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func mustReadPending(t *testing.T, b *takeInBed) []PendingEntry {
|
||||
t.Helper()
|
||||
entries, err := ReadPending(context.Background(), b.pending, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return entries
|
||||
}
|
||||
|
||||
@@ -27,6 +27,18 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
|
||||
return a.Result, nil
|
||||
}
|
||||
}
|
||||
merge := func(args map[string]any) (any, error) {
|
||||
seq, _ := args["sequence"].(float64)
|
||||
key, value, err := NewMergeAsk(str(args, "repository"), str(args, "merge"), uint64(seq), str(args, "why"), stdio.CallerOf(args), now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a, err := ask(key, value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return a.Result, nil
|
||||
}
|
||||
return []stdio.Tool{
|
||||
{Name: PendingTool,
|
||||
Description: "What waits for a take-in (ADR 0633): merges the indexer holds in its pending state instead of taking them in. " +
|
||||
@@ -34,7 +46,8 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
|
||||
"(waits-for): the running indexer wrote only what building and moving that indexer needs, and the new indexer takes " +
|
||||
"it in at its start. A group (what group, key g.<group>) is a delivery group whose members are gathered until the " +
|
||||
"forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " +
|
||||
"merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " +
|
||||
"merges, since when it waits, the modules it holds back (holds), the merges whose message the forge bridge's stream no " +
|
||||
"longer holds (lost: they are not taken in; indexer_drop_pending drops them) and the last take-in event said of it. Empty waits: " +
|
||||
"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
|
||||
"indexer_take_in_pending and indexer_drop_pending steer a wait.",
|
||||
Input: map[string]any{"type": "object", "properties": map[string]any{}},
|
||||
@@ -65,5 +78,17 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
|
||||
"key": map[string]any{"type": "string", "description": "the pending merge's key, m.<sequence>, as indexer_pending names its merges' wait"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
|
||||
Run: hand(true)},
|
||||
{Name: TakeInMergeTool,
|
||||
Description: "Take one merge in by hand (ADR 0633 §5, issue 1585): the indexer asks the forge bridge for the merge's merged message " +
|
||||
"(bridge_merged_message) and runs it through the take-in as a delivered merged message, so it is held for the " +
|
||||
"indexer its revision names or taken in. For a merge the take-in did not take in (indexer_heard names its sequence " +
|
||||
"and why). A hand act: who and why are said in the indexer's journal. The bridge's own words answer when it cannot " +
|
||||
"build the message (the gate holds no such merge, the forge does not answer).",
|
||||
Input: map[string]any{"type": "object", "required": []string{"repository", "merge", "sequence", "why"}, "properties": map[string]any{
|
||||
"repository": map[string]any{"type": "string", "description": "the repository, owner/name"},
|
||||
"merge": map[string]any{"type": "string", "description": "the merge commit, 40 hex digits"},
|
||||
"sequence": map[string]any{"type": "integer", "description": "the merge's sequence on the forge bridge's stream, as indexer_heard names it"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is taken in by hand, in plain words"}}},
|
||||
Run: merge},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -36,12 +37,17 @@ type TakeInAsk struct {
|
||||
Digest string `json:"digest,omitempty"`
|
||||
// Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the
|
||||
// one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5).
|
||||
Pending string `json:"pending,omitempty"`
|
||||
Drop bool `json:"drop,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
Result *TakenIn `json:"result,omitempty"`
|
||||
Answered time.Time `json:"answered,omitzero"`
|
||||
Pending string `json:"pending,omitempty"`
|
||||
// Merge asks the merge of MergeRepository at this commit to be taken in by hand (indexer_take_in_merge): its merged message
|
||||
// is asked of the forge bridge and run through the take-in as a delivered one, under its stream sequence Sequence.
|
||||
Merge string `json:"merge,omitempty"`
|
||||
MergeRepository string `json:"merge-repository,omitempty"`
|
||||
Sequence uint64 `json:"sequence,omitempty"`
|
||||
Drop bool `json:"drop,omitempty"`
|
||||
By string `json:"by,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
Result *TakenIn `json:"result,omitempty"`
|
||||
Answered time.Time `json:"answered,omitzero"`
|
||||
}
|
||||
|
||||
// Done says the ask is answered.
|
||||
@@ -56,8 +62,8 @@ func DecodeTakeIn(data []byte) (TakeInAsk, error) {
|
||||
if a.V != 1 {
|
||||
return a, fmt.Errorf("a takeins value of version %d, not 1", a.V)
|
||||
}
|
||||
if a.Module == "" && a.Repository == "" && a.Pending == "" {
|
||||
return a, errors.New("a takeins value naming no module, repository or pending wait")
|
||||
if a.Module == "" && a.Repository == "" && a.Pending == "" && a.Merge == "" {
|
||||
return a, errors.New("a takeins value naming no module, repository, pending wait or merge")
|
||||
}
|
||||
return a, nil
|
||||
}
|
||||
@@ -87,6 +93,30 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by
|
||||
return "repository." + key, data, err
|
||||
}
|
||||
|
||||
// TakeInMergeTool takes one merge in by hand, its merged message asked of the forge bridge (ADR 0633 §5, issue 1585).
|
||||
const TakeInMergeTool = "indexer_take_in_merge"
|
||||
|
||||
var sha40 = regexp.MustCompile(`^[0-9a-f]{40}$`)
|
||||
|
||||
// NewMergeAsk is the value indexer_take_in_merge writes: its key in takeins, `merge.<commit>`, and the value. sequence is
|
||||
// the merge's sequence on the forge bridge's stream, as indexer_heard names it: the pending merge is keyed by it, which
|
||||
// keeps the merge order.
|
||||
func NewMergeAsk(repository, merge string, sequence uint64, why, by string, at time.Time) (string, []byte, error) {
|
||||
repository, merge = strings.TrimSpace(repository), strings.TrimSpace(merge)
|
||||
switch {
|
||||
case !strings.Contains(repository, "/"):
|
||||
return "", nil, errors.New("name the repository as owner/name (repository)")
|
||||
case !sha40.MatchString(merge):
|
||||
return "", nil, errors.New("name the merge commit, 40 hex digits (merge)")
|
||||
case sequence == 0:
|
||||
return "", nil, errors.New("name the merge's sequence on the forge bridge's stream (sequence), as indexer_heard names it")
|
||||
case strings.TrimSpace(why) == "":
|
||||
return "", nil, errors.New("say why (why): a hand act is said in the journal")
|
||||
}
|
||||
data, err := json.Marshal(TakeInAsk{V: 1, Merge: merge, MergeRepository: repository, Sequence: sequence, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
|
||||
return "merge." + merge, data, err
|
||||
}
|
||||
|
||||
// NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key
|
||||
// in takeins, `pending.<take-in|drop>.<key>`, and the value. key is a wait's key as indexer_pending names it, for a
|
||||
// drop one pending merge's.
|
||||
|
||||
@@ -129,15 +129,15 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
// Three invokes: the file of a registered module's manifest and lock files (registered.go), the forge's trees (the named paths' objects, since the indexer is the one reader of a manifest's paths:
|
||||
// ADR 0499 clauses 2 and 3, ADR 0495 clause 1; and the module directories of the trunk, issue 1900) and whether a
|
||||
// commit is in another (issue 1900).
|
||||
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "seat:node-engine.composition"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
if want := []string{"gitea.gitea_commit_trees", "gitea.gitea_contains", "gitea.gitea_get_file", "gitea.gitea_open_pulls", "mesh-delivery-forge-bridge.bridge_merged_message", "seat:node-engine.composition"}; !reflect.DeepEqual(m.Invokes, want) {
|
||||
t.Errorf("invokes = %v, want %v", m.Invokes, want)
|
||||
}
|
||||
// 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle
|
||||
// reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state
|
||||
// (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections).
|
||||
// 1585/T12, T13 (ADR 0633 §5): indexer_pending reads the pending state; indexer_take_in_pending and indexer_drop_pending
|
||||
// ask through takeins.
|
||||
if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool}; !reflect.DeepEqual(m.Tools, want) {
|
||||
// ask through takeins; indexer_take_in_merge takes one merge in, its message asked of the forge bridge (issue 1585).
|
||||
if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TakeInMergeTool}; !reflect.DeepEqual(m.Tools, want) {
|
||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||
}
|
||||
// ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a
|
||||
|
||||
@@ -21,7 +21,8 @@
|
||||
"indexer_judge",
|
||||
"indexer_pending",
|
||||
"indexer_take_in_pending",
|
||||
"indexer_drop_pending"
|
||||
"indexer_drop_pending",
|
||||
"indexer_take_in_merge"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-build-go.toolchain.reported",
|
||||
@@ -63,6 +64,7 @@
|
||||
"gitea.gitea_contains",
|
||||
"gitea.gitea_get_file",
|
||||
"gitea.gitea_open_pulls",
|
||||
"mesh-delivery-forge-bridge.bridge_merged_message",
|
||||
"seat:node-engine.composition"
|
||||
],
|
||||
"own-secrets": {
|
||||
|
||||
@@ -133,6 +133,9 @@ type TakeIn struct {
|
||||
By string `json:"by,omitempty"`
|
||||
Why string `json:"why,omitempty"`
|
||||
Dropped bool `json:"dropped,omitempty"`
|
||||
// Lost are the keys of merges whose message the forge bridge's stream no longer holds: they are not taken in, and the
|
||||
// revision waits until indexer_drop_pending drops them (issue 1585).
|
||||
Lost []string `json:"lost,omitempty"`
|
||||
}
|
||||
|
||||
func (t TakeIn) check() error {
|
||||
@@ -230,6 +233,8 @@ type PendingWait struct {
|
||||
Holds []string `json:"holds"`
|
||||
Said string `json:"said,omitempty"`
|
||||
SaidAt time.Time `json:"said-at,omitzero"`
|
||||
// Lost are the keys of its merges whose message the forge bridge's stream no longer holds (TakeIn.Lost).
|
||||
Lost []string `json:"lost,omitempty"`
|
||||
}
|
||||
|
||||
// PendingAnswer is indexer_pending's answer: every wait, oldest first, and the pending keys that do not read.
|
||||
|
||||
Reference in New Issue
Block a user