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

Merged
mesh-tools merged 5 commits from fix/1585-pending-by-reference into main 2026-11-05 10:48:58 +00:00
21 changed files with 828 additions and 122 deletions
@@ -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)
@@ -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
+3 -1
View File
@@ -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.