Indexer: a held merge's message is asked of the forge bridge, not read from its stream; covered merges leave their candidate; indexer_take_in_merge (issue 1585)

Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
2026-11-05 11:37:13 +01:00
parent 0605962960
commit 3c832d1a2d
12 changed files with 236 additions and 155 deletions
@@ -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
}
}
@@ -1,31 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-indexer/internal/index"
)
// forgeVia reads one message of the forge bridge's stream by its sequence (issue 1585): what a pending merge keeps instead
// of a copy of the message. A sequence the stream no longer holds (its age or size cap removed it) answers
// index.ErrForgeGone.
func forgeVia(js jetstream.JetStream) func(ctx context.Context, seq uint64) ([]byte, error) {
return func(ctx context.Context, seq uint64) ([]byte, error) {
s, err := js.Stream(ctx, forgeStream)
if err != nil {
return nil, fmt.Errorf("the stream %s: %w", forgeStream, err)
}
m, err := s.GetMsg(ctx, seq)
if errors.Is(err, jetstream.ErrMsgNotFound) {
return nil, index.ErrForgeGone
}
if err != nil {
return nil, err
}
return m.Data, 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}, Forge: forgeVia(js),
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,7 +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
TakeInForge(ctx context.Context, seq uint64, 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
@@ -37,8 +37,8 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream
return
}
asked, r := "take-in", index.TakenIn{}
if a.Forge != 0 {
asked, r = "hand take-in of a forge message", ix.TakeInForge(ctx, a.Forge, a.By, a.Why)
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)
@@ -123,11 +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)
// Forge reads the message of a sequence of the forge bridge's stream (a pending value keeps the sequence, not a copy of the
// message: issue 1585); ErrForgeGone says the stream no longer holds it. Nil reads none.
Forge func(ctx context.Context, seq uint64) ([]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
// 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
@@ -220,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)
@@ -44,18 +44,16 @@ 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")
// ErrForgeGone says the forge bridge's stream no longer holds the message a pending value refers to (its age or size
// cap removed it). The merge is then said held, never taken in silently; indexer_drop_pending drops it.
var ErrForgeGone = errors.New("the forge bridge's stream no longer holds the message")
// 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"`
Members []delivery.Head `json:"members,omitempty"`
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"`
}
@@ -110,7 +108,7 @@ func DecodePending(data []byte) (Pending, forgemsg.Forge, error) {
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, Group: p.Group, Members: r.Members}, nil
Merge: r.Merge, Branch: r.Branch, Group: p.Group, Members: r.Members}, nil
}
f, err := forgemsg.DecodeForge(p.Forge)
if err != nil {
@@ -176,7 +174,7 @@ func holdsOf(f forgemsg.Forge) []string {
// 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, Members: f.Members, Holds: holdsOf(f)}
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.
@@ -216,20 +214,19 @@ func (ix *Indexer) load(ctx context.Context, e PendingEntry) (PendingEntry, erro
e.Forge, e.Light = f, false
return e, nil
}
if ix.Forge == nil {
return e, fmt.Errorf("the forge bridge's stream is not read by this indexer, so %s (sequence %d) cannot be read", e.Key, e.Value.Seq)
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.Forge(ctx, e.Value.Seq)
data, err := ix.Message(ctx, e.Forge.Repository, e.Forge.Merge)
if err != nil {
return e, fmt.Errorf("sequence %d: %w", e.Value.Seq, err)
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("sequence %d of the forge bridge's stream is not a forge message: %w", e.Value.Seq, err)
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("sequence %d of the forge bridge's stream is not the merge %s#%d at %s that %s refers to", e.Value.Seq,
e.Forge.Repository, e.Forge.Number, e.Forge.Merge, e.Key)
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
@@ -237,8 +234,7 @@ func (ix *Indexer) load(ctx context.Context, e PendingEntry) (PendingEntry, erro
// 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 that cannot be read (the stream no longer holds it, or the bus refuses the
// read) is answered in lost with its reason, by key, and its revision is held (holdLost).
// 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{}
@@ -596,8 +592,8 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie
}
}
}
// Of each repository's merges only the newest is read (from memory, else the forge bridge's stream by its sequence: a
// pending value keeps a reference); the older ones are covered by it. A newest merge that cannot be read is said and
// 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 {
@@ -637,7 +633,7 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea
return nil
}
for _, g := range lost {
ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision: %s; a newer merge of %s covers it, indexer_drop_pending drops it",
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) {
@@ -692,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, Forge: ix.Forge, 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)
@@ -841,6 +837,11 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow
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)
@@ -1126,32 +1127,32 @@ func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn
return r
}
// TakeInForge is the hand take-in of one forge message (indexer_take_in_forge, issue 1585): the merged message of
// sequence seq is read from the forge bridge's stream and handed to the take-in as a delivered merged message is (Arrive),
// 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 (a pending value that could not be written) or whose pending value
// was dropped. A stream that no longer holds the sequence answers so.
func (ix *Indexer) TakeInForge(ctx context.Context, seq uint64, by, why string) TakenIn {
// 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.Forge == nil {
r.Answer = "this indexer does not read the forge bridge's stream"
if ix.Message == nil {
r.Answer = "this indexer does not ask the forge bridge"
return r
}
data, err := ix.Forge(ctx, seq)
data, err := ix.Message(ctx, repository, merge)
if err != nil {
r.Answer = fmt.Sprintf("sequence %d is not read (reading the forge bridge's stream by sequence needs the indexer's read grant on it): %v", seq, err)
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("sequence %d is not a forge message: %v", seq, err)
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 {
r.Answer = fmt.Sprintf("sequence %d is a %s message, not a merged one: only a merge is taken in", seq, f.Kind)
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 forge message %d (%s#%d at %s) asked by %s: %s", seq, f.Repository, f.Number, f.Merge, by, why)
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()
@@ -23,7 +23,7 @@ type takeInBed struct {
mods *memStore
pending *memStore
said []takein.TakeIn
stream map[uint64][]byte // the forge bridge's stream: a pending merge is read from it by its sequence
bridge map[string][]byte // what the forge bridge answers (bridge_merged_message), by repository@merge
base forgemsg.Forge
logs *strings.Builder
}
@@ -37,15 +37,15 @@ 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(), stream: map[uint64][]byte{}, 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.Forge = func(_ context.Context, seq uint64) ([]byte, error) {
if data, ok := b.stream[seq]; ok {
ix.Message = func(_ context.Context, repository, merge string) ([]byte, error) {
if data, ok := b.bridge[repository+"@"+merge]; ok {
return data, nil
}
return nil, ErrForgeGone
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
@@ -94,20 +94,20 @@ 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, seq, f)
b.publish(t, f)
if _, _, err := b.ix.Arrive(context.Background(), seq, f); err != nil {
t.Fatalf("arrive %d: %v", seq, err)
}
}
// publish puts the merged message on the bed's stream under seq, as the forge bridge does.
func (b *takeInBed) publish(t *testing.T, seq uint64, f forgemsg.Forge) {
// 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.stream[seq] = data
b.bridge[f.Repository+"@"+f.Merge] = data
}
func (b *takeInBed) keys() []string {
@@ -199,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, Forge: b.ix.Forge, 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) {
@@ -422,61 +422,93 @@ 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 with no stream read, covered after a restart without the read grant, taken in by hand by its sequence", func(t *testing.T) {
big := func(f forgemsg.Forge) forgemsg.Forge { f.Words = strings.Repeat("x", 300*1024); return f }
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)
own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi")
// No grant on the stream: every read is refused. A merge of 300 KiB, over the bucket's cap whole, is held by
// reference and taken in from the message Arrive was handed, with no stream read at all.
readable := b.ix.Forge
b.ix.Forge = func(context.Context, uint64) ([]byte, error) { return nil, errors.New("nats: permissions violation") }
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"))
}
// A restart: the messages in memory are gone and the read is refused. The newest merge of the repository is said lost
// (event, state, reason); the older one is covered by it and not lost; nothing is taken in, even by hand.
b.ix.held = nil
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 := b.ix.PendingAgain(ctx); err != 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 err := b.ix.PendingAgain(ctx); err != nil || len(b.said) != 1 {
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 := b.ix.TakeInPendingByHand(ctx, takein.PendingKey(2), "lead@shanks", "try"); r.Stored || b.commit(t, "avahi") != sha('b') {
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"))
}
// A newer merge of the repository arrives, back on the running indexer: it covers both held merges and ends them.
// 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()
b.arrive(t, 4, big(b.merge(23, sha('g'), "tree-avahi-5", b.base.Trees["modules/"+delivery.IndexerModule])))
if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('g') || b.fp(t, delivery.IndexerModule) != own || !strings.Contains(b.logs.String(), "is covered by") {
t.Fatalf("covered: pending %v, avahi at %s, journal %q: want both held merges taken in by the newer one", b.keys(), b.commit(t, "avahi"), b.logs.String())
if err := next.PendingAgain(ctx); err != nil {
t.Fatal(err)
}
// With the read grant, a merge the take-in never held is taken in by its stream sequence, as a delivered one; a
// sequence the stream lost answers so; an old value that holds the message whole reads as before.
b.ix.Forge = readable
b.publish(t, 9, big(b.merge(30, sha('e'), "tree-avahi-6", b.base.Trees["modules/"+delivery.IndexerModule])))
if r := b.ix.TakeInForge(ctx, 9, "lead@shanks", "held by the cap"); !r.Stored || b.commit(t, "avahi") != sha('e') {
t.Fatalf("forge take-in: %+v, avahi at %s: want it taken in", r, b.commit(t, "avahi"))
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 r := b.ix.TakeInForge(ctx, 77, "lead@shanks", "no such"); r.Stored || !strings.Contains(r.Answer, "no longer holds") {
t.Errorf("forge take-in of a sequence the stream lost: %+v", r)
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: own})
if err := b.pending.Create(ctx, takein.PendingKey(10), data); err != nil {
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') {
@@ -488,7 +520,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
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", "Forge", "Pending", "SayTakeIn"} {
"Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn"} {
named[f] = true
}
ty := reflect.TypeOf(Indexer{})
@@ -27,9 +27,9 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk,
return a.Result, nil
}
}
forge := func(args map[string]any) (any, error) {
merge := func(args map[string]any) (any, error) {
seq, _ := args["sequence"].(float64)
key, value, err := NewForgeAsk(uint64(seq), str(args, "why"), stdio.CallerOf(args), now())
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
}
@@ -78,16 +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: TakeInForgeTool,
Description: "Take one merged message of the forge bridge's stream in by hand, by its stream sequence (ADR 0633 §5, issue 1585): " +
"the indexer reads it from the stream and hands it to the take-in as a delivered merged message is, 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) or whose pending merge was dropped. A hand act: who and why are said in the indexer's journal. " +
"A sequence the stream no longer holds answers so. It reads the stream by sequence, which needs the indexer's read grant " +
"on the forge bridge's stream; until that grant is given it answers the bus's refusal.",
Input: map[string]any{"type": "object", "required": []string{"sequence", "why"}, "properties": map[string]any{
"sequence": map[string]any{"type": "integer", "description": "the message'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: forge},
{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"
@@ -37,14 +38,16 @@ type TakeInAsk struct {
// 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"`
// Forge asks the merged message of this sequence on the forge bridge's stream to be taken in by hand
// (indexer_take_in_forge): read from the stream and handed to the take-in as a delivered merged message is.
Forge uint64 `json:"forge,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"`
// 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.
@@ -59,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 == "" && a.Forge == 0 {
return a, errors.New("a takeins value naming no module, repository, pending wait or forge message")
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
}
@@ -90,19 +93,28 @@ func NewTakeInRepository(repository, why, by string, at time.Time) (string, []by
return "repository." + key, data, err
}
// TakeInForgeTool takes one merged message of the forge bridge's stream in by its sequence (ADR 0633 §5, issue 1585).
const TakeInForgeTool = "indexer_take_in_forge"
// 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"
// NewForgeAsk is the value indexer_take_in_forge writes: its key in takeins, `forge.<sequence>`, and the value.
func NewForgeAsk(seq uint64, why, by string, at time.Time) (string, []byte, error) {
if seq == 0 {
return "", nil, errors.New("name the forge message to take in by its stream sequence (sequence), as indexer_heard names it")
}
if strings.TrimSpace(why) == "" {
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, Forge: seq, Why: strings.TrimSpace(why), By: by, At: at.UTC()})
return fmt.Sprintf("forge.%020d", seq), data, err
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
@@ -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; indexer_take_in_forge takes one forge message in by its sequence (issue 1585).
if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge", index.PendingTool, index.TakeInPendingTool, index.DropPendingTool, index.TakeInForgeTool}; !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
+2 -1
View File
@@ -22,7 +22,7 @@
"indexer_pending",
"indexer_take_in_pending",
"indexer_drop_pending",
"indexer_take_in_forge"
"indexer_take_in_merge"
],
"consumes": [
"mesh-build-go.toolchain.reported",
@@ -64,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": {