From 4b7a31b2c7a0cad1525ef5f87d87065bde574376 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 11:21:42 +0100 Subject: [PATCH 1/5] Indexer: a pending merge keeps a reference to its FORGE message, not a copy, so a merge over the state value cap is held (issue 1585) The pending value holds the stream sequence and the small fields the waits read; the take-in reads the message from the forge bridge's stream by sequence. A message the stream no longer holds is said (take-in.pending lost, indexer_pending) and never taken in silently. indexer_take_in_forge takes one forge message in by its sequence. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../cmd/mesh-delivery-indexer/forgeread.go | 31 +++ .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/takein.go | 6 +- .../internal/index/candidates.go | 3 + .../internal/index/index_test.go | 16 ++ .../internal/index/pending.go | 214 ++++++++++++++++-- .../internal/index/pending_test.go | 101 ++++++++- .../internal/index/pending_tools.go | 25 +- .../internal/index/takein.go | 24 +- .../mesh-delivery-indexer/manifest_test.go | 4 +- modules/mesh-delivery-indexer/module.json | 3 +- .../mesh-delivery-indexer/takein/takein.go | 5 + 12 files changed, 403 insertions(+), 31 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go new file mode 100644 index 00000000..6cd3b277 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go @@ -0,0 +1,31 @@ +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 + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 3c24a297..8db24813 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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}, Forge: forgeVia(js), // 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)}, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 2ef5b43e..34bfa0f7 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -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 + TakeInForge(ctx context.Context, 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.Forge != 0 { + asked, r = "hand take-in of a forge message", ix.TakeInForge(ctx, a.Forge, 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 != "" { diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 31d23828..4663605f 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -123,6 +123,9 @@ 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) // 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 diff --git a/modules/mesh-delivery-indexer/internal/index/index_test.go b/modules/mesh-delivery-indexer/internal/index/index_test.go index 8c0cff64..57e1fcda 100644 --- a/modules/mesh-delivery-indexer/internal/index/index_test.go +++ b/modules/mesh-delivery-indexer/internal/index/index_test.go @@ -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 diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index b3a15633..15525c4c 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -44,14 +44,34 @@ 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 +// 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"` + // 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 +83,16 @@ 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 } // DecodePending reads one pending value and its merged message. @@ -80,6 +104,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, 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 +148,76 @@ 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, Members: f.Members, Holds: holdsOf(f)} +} + +// load gives the entry the merged message whole: the copy an earlier value holds, 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 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 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) + } + data, err := ix.Forge(ctx, e.Value.Seq) + if err != nil { + return e, fmt.Errorf("sequence %d: %w", e.Value.Seq, 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) + } + 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) + } + e.Forge, e.Light = f, false + return e, nil +} + +// loadAll gives each entry its message (load); the entries the stream no longer holds are answered apart, by key, and left out. +func (ix *Indexer) loadAll(ctx context.Context, entries []PendingEntry) (loaded, gone []PendingEntry, err error) { + for _, e := range entries { + full, err := ix.load(ctx, e) + switch { + case errors.Is(err, ErrForgeGone): + gone = append(gone, e) + case err != nil: + return nil, nil, err + default: + loaded = append(loaded, full) + } + } + return loaded, gone, nil +} + // 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 +362,8 @@ 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}) + 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 +551,61 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie } } } - names, dry, err := ix.computeOver(ctx, w.Ready) + // The merged messages are read from the forge bridge's stream by their sequence (a pending value keeps a reference). A + // merge the stream no longer holds is said and holds its revision: nothing of it is taken in silently. + ready, gone, err := ix.loadAll(ctx, w.Ready) + if err != nil { + return Applied{}, "", err + } + if len(gone) > 0 { + return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, gone) + } + 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 one of whose merges the forge bridge's stream no longer holds (its age or size cap removed +// it): said once on take-in.pending with the lost keys (TakeIn.Lost), and on each lost entry (Pending.Lost, which +// indexer_pending shows), until indexer_drop_pending drops the merge. Nothing of the revision is taken in meanwhile. +func (ix *Indexer) holdLost(ctx context.Context, ready, gone []PendingEntry) error { + t := revisionWait(takein.TakeInPending, ready, "") + for _, g := range gone { + t.Lost = append(t.Lost, g.Key) + } + said := true + for _, e := range ready { + lost := slices.Contains(t.Lost, e.Key) + if e.Value.Said != t.Key || e.Value.SaidKind != takein.TakeInPending || (lost && e.Value.Lost == "") { + said = false + } + } + if said { + return nil + } + ix.logf("take-in: %s are not taken in: the forge bridge's stream no longer holds their messages; indexer_drop_pending drops them", strings.Join(t.Lost, ", ")) + 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) + } + } + reason := "the forge bridge's stream no longer holds its message" + ix.markSaid(ctx, gone, &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 +643,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, Forge: ix.Forge, 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) @@ -809,11 +942,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 +992,15 @@ 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, gone, err := ix.loadAll(ctx, take) + if err != nil { + r.Answer = err.Error() + return r + } + if len(gone) > 0 { + r.Answer = fmt.Sprintf("nothing is taken in: the forge bridge's stream no longer holds the message of %s; indexer_drop_pending drops it", gone[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 { @@ -919,3 +1064,42 @@ 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 { + r := TakenIn{Module: takein.PendingKey(seq)} + if ix.Forge == nil { + r.Answer = "this indexer does not read the forge bridge's stream" + return r + } + data, err := ix.Forge(ctx, seq) + if err != nil { + r.Answer = fmt.Sprintf("sequence %d is not read: %v", seq, 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) + 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) + 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) + 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 +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 5a5396ce..bba40ba9 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -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 + stream map[uint64][]byte // the forge bridge's stream: a pending merge is read from it by its sequence 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(), stream: map[uint64][]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 { + return data, nil + } + return nil, ErrForgeGone + } 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, seq, 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) { + t.Helper() + data, err := forgemsg.EncodeForge(f) + if err != nil { + t.Fatal(err) + } + b.stream[seq] = 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, Forge: b.ix.Forge, 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,11 @@ 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, _, err := b.ix.loadAll(ctx, WaitsOf(entries).Ready) + if err != nil { + t.Fatal(err) + } + 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 +425,68 @@ 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 merged message over the value cap is held by reference, taken in by hand by its sequence, and said lost when the stream drops it", func(t *testing.T) { + big := func(f forgemsg.Forge) forgemsg.Forge { f.Words = strings.Repeat("x", 300*1024); return f } + b := newTakeInBed(t) + own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi") + // A merge of about 300 KiB is over the bucket's cap whole: held by reference it is written (small), and an ordinary + // one is taken in from the stream. + 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 { + t.Fatalf("a big merge alone: avahi at %s, pending %v: want it taken in from the stream", b.commit(t, "avahi"), b.keys()) + } + // A big merge changing the indexer is held; its pending value keeps the reference, far under the cap. + 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")) + } + // The stream no longer holds m.2: said lost (event and state), nothing taken in, even by hand. + delete(b.stream, 2) + b.said = nil + if err := b.ix.PendingAgain(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(2)}) || len(waits) != 1 || !slices.Equal(waits[0].Lost, []string{takein.PendingKey(2)}) || b.commit(t, "avahi") != sha('b') { + t.Fatalf("lost: said %+v, waits %+v, avahi at %s: want m.2 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 { + 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') { + t.Errorf("hand take-in of a lost merge: %+v, avahi at %s: want nothing taken in", r, b.commit(t, "avahi")) + } + // The hand drop ends it; m.3 is evaluated again and, held for the new indexer, is not lost. + if r := b.ix.DropPending(ctx, takein.PendingKey(2), "lead@shanks", "the stream lost it"); !r.Stored || len(b.keys()) != 1 { + t.Fatalf("drop: %+v, pending %v", r, b.keys()) + } + // A merge the take-in never held is taken in by its stream sequence, as a delivered one; an old value that holds the + // message whole reads as before. + 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') || b.fp(t, delivery.IndexerModule) != own { + t.Fatalf("forge take-in: %+v, avahi at %s: want it taken in", r, b.commit(t, "avahi")) + } + 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) + } + 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 { + 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", "Forge", "Pending", "SayTakeIn"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) @@ -419,3 +497,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 +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index bf3ab9b7..ef43dbe8 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -27,6 +27,18 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, return a.Result, nil } } + forge := func(args map[string]any) (any, error) { + seq, _ := args["sequence"].(float64) + key, value, err := NewForgeAsk(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.) 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,15 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, "key": map[string]any{"type": "string", "description": "the pending merge's key, m., 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.", + 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}, } } diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index f1982e05..899666c4 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -36,7 +36,10 @@ 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"` + 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"` @@ -56,8 +59,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.Forge == 0 { + return a, errors.New("a takeins value naming no module, repository, pending wait or forge message") } return a, nil } @@ -87,6 +90,21 @@ 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" + +// NewForgeAsk is the value indexer_take_in_forge writes: its key in takeins, `forge.`, 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) == "" { + 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 +} + // NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key // in takeins, `pending..`, and the value. key is a wait's key as indexer_pending names it, for a // drop one pending merge's. diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 5ce577c3..eb14f64d 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -136,8 +136,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 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_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) { 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 diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 10bb8227..48ee9cdd 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -21,7 +21,8 @@ "indexer_judge", "indexer_pending", "indexer_take_in_pending", - "indexer_drop_pending" + "indexer_drop_pending", + "indexer_take_in_forge" ], "consumes": [ "mesh-build-go.toolchain.reported", diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index 31fbfb8e..d5ce9866 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -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. From 343321f11eddd30716aea483d0c24b200c3b6827 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 11:24:18 +0100 Subject: [PATCH 2/5] Indexer: Arrive keeps its message in memory, only the newest merge of a repository is read, an unreadable one is said and covered by a newer one (issue 1585) The take-in unblocks merges without the read grant on the forge bridge's stream: the message Arrive was handed is used, the stream is read only for a pending merge not held in memory, and older merges of a repository in a revision are covered by its newest and never read. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../internal/index/candidates.go | 4 +- .../internal/index/pending.go | 143 +++++++++++++----- .../internal/index/pending_test.go | 42 ++--- .../internal/index/pending_tools.go | 3 +- 4 files changed, 129 insertions(+), 63 deletions(-) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 4663605f..4398f7e0 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -125,7 +125,9 @@ type Indexer struct { 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) + 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 // 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 diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 15525c4c..60f94e44 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -93,6 +93,9 @@ type PendingEntry struct { 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. @@ -176,13 +179,43 @@ func refOf(f forgemsg.Forge) *PendingRef { return &PendingRef{Repository: f.Repository, Number: f.Number, Merge: f.Merge, Members: f.Members, Holds: holdsOf(f)} } -// load gives the entry the merged message whole: the copy an earlier value holds, 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 message that is not -// the merge the value refers to (a stream rebuilt under the same sequence) is refused. +// 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.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) } @@ -202,20 +235,31 @@ func (ix *Indexer) load(ctx context.Context, e PendingEntry) (PendingEntry, erro return e, nil } -// loadAll gives each entry its message (load); the entries the stream no longer holds are answered apart, by key, and left out. -func (ix *Indexer) loadAll(ctx context.Context, entries []PendingEntry) (loaded, gone []PendingEntry, err error) { - for _, e := range entries { - full, err := ix.load(ctx, e) - switch { - case errors.Is(err, ErrForgeGone): - gone = append(gone, e) - case err != nil: - return nil, nil, err - default: - loaded = append(loaded, full) +// 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). +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 } - return loaded, gone, nil + 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 @@ -363,6 +407,7 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f forgemsg.Forge) (Ap ix.pendingMu.Lock() defer ix.pendingMu.Unlock() key := takein.PendingKey(seq) + 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 @@ -551,14 +596,12 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie } } } - // The merged messages are read from the forge bridge's stream by their sequence (a pending value keeps a reference). A - // merge the stream no longer holds is said and holds its revision: nothing of it is taken in silently. - ready, gone, err := ix.loadAll(ctx, w.Ready) - if err != nil { - return Applied{}, "", err - } - if len(gone) > 0 { - return Applied{}, ix.waitingLine(w), ix.holdLost(ctx, w.Ready, gone) + // 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 + // 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 { @@ -574,25 +617,29 @@ func (ix *Indexer) evaluate(ctx context.Context, own string, start bool) (Applie return Applied{}, ix.waitingLine(w), ix.holdFor(ctx, ready, names, own, dry) } -// holdLost holds a revision one of whose merges the forge bridge's stream no longer holds (its age or size cap removed -// it): said once on take-in.pending with the lost keys (TakeIn.Lost), and on each lost entry (Pending.Lost, which -// indexer_pending shows), until indexer_drop_pending drops the merge. Nothing of the revision is taken in meanwhile. -func (ix *Indexer) holdLost(ctx context.Context, ready, gone []PendingEntry) error { +// 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 gone { + for _, g := range lost { t.Lost = append(t.Lost, g.Key) } said := true for _, e := range ready { - lost := slices.Contains(t.Lost, e.Key) - if e.Value.Said != t.Key || e.Value.SaidKind != takein.TakeInPending || (lost && e.Value.Lost == "") { + 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 } - ix.logf("take-in: %s are not taken in: the forge bridge's stream no longer holds their messages; indexer_drop_pending drops them", strings.Join(t.Lost, ", ")) + 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", + g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], g.Forge.Repository) + } if !ix.sayTakeIn(ctx, t) { return nil } @@ -602,8 +649,10 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, gone []PendingEntry) err kept = append(kept, e) } } - reason := "the forge bridge's stream no longer holds its message" - ix.markSaid(ctx, gone, &t, func(p *Pending) { p.Lost = reason }) + 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 } @@ -677,6 +726,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 @@ -779,7 +831,18 @@ 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) + continue + } a, err := ix.ApplyMessage(ctx, e.Forge) applied.Written = append(applied.Written, a.Written...) if a.Late != "" { @@ -799,6 +862,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 @@ -992,13 +1056,9 @@ 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, gone, err := ix.loadAll(ctx, take) - if err != nil { - r.Answer = err.Error() - return r - } - if len(gone) > 0 { - r.Answer = fmt.Sprintf("nothing is taken in: the forge bridge's stream no longer holds the message of %s; indexer_drop_pending drops it", gone[0].Key) + 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) @@ -1042,6 +1102,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 @@ -1078,7 +1139,7 @@ func (ix *Indexer) TakeInForge(ctx context.Context, seq uint64, by, why string) } data, err := ix.Forge(ctx, seq) if err != nil { - r.Answer = fmt.Sprintf("sequence %d is not read: %v", seq, err) + 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) return r } f, err := forgemsg.DecodeForge(data) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index bba40ba9..0de75c2d 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -372,10 +372,7 @@ 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) } - ready, _, err := b.ix.loadAll(ctx, WaitsOf(entries).Ready) - if err != nil { - t.Fatal(err) - } + 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) @@ -425,32 +422,34 @@ 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 merged message over the value cap is held by reference, taken in by hand by its sequence, and said lost when the stream drops it", func(t *testing.T) { + 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 } b := newTakeInBed(t) own, avahi0 := b.fp(t, delivery.IndexerModule), b.fp(t, "avahi") - // A merge of about 300 KiB is over the bucket's cap whole: held by reference it is written (small), and an ordinary - // one is taken in from the stream. + // 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") } 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 { - t.Fatalf("a big merge alone: avahi at %s, pending %v: want it taken in from the stream", b.commit(t, "avahi"), b.keys()) + 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)) } - // A big merge changing the indexer is held; its pending value keeps the reference, far under the cap. 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")) } - // The stream no longer holds m.2: said lost (event and state), nothing taken in, even by hand. - delete(b.stream, 2) + // 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 b.said = nil if err := b.ix.PendingAgain(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(2)}) || len(waits) != 1 || !slices.Equal(waits[0].Lost, []string{takein.PendingKey(2)}) || b.commit(t, "avahi") != sha('b') { - t.Fatalf("lost: said %+v, waits %+v, avahi at %s: want m.2 lost on the event and in indexer_pending, nothing taken in", b.said, waits, b.commit(t, "avahi")) + 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 { t.Errorf("lost again: err %v, said %v: want it said once", err, b.events()) @@ -458,14 +457,17 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if r := b.ix.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 hand drop ends it; m.3 is evaluated again and, held for the new indexer, is not lost. - if r := b.ix.DropPending(ctx, takein.PendingKey(2), "lead@shanks", "the stream lost it"); !r.Stored || len(b.keys()) != 1 { - t.Fatalf("drop: %+v, pending %v", r, b.keys()) + // A newer merge of the repository arrives, back on the running indexer: it covers both held merges and ends them. + 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()) } - // A merge the take-in never held is taken in by its stream sequence, as a delivered one; an old value that holds the - // message whole reads as before. + // 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') || b.fp(t, delivery.IndexerModule) != own { + 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 r := b.ix.TakeInForge(ctx, 77, "lead@shanks", "no such"); r.Stored || !strings.Contains(r.Answer, "no longer holds") { diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index ef43dbe8..16f3570b 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -83,7 +83,8 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, "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.", + "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"}}}, From 060596296006b1b27f5d90fc5b70ba8da3ef749b Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 11:34:04 +0100 Subject: [PATCH 3/5] Forge bridge: bridge_merged_message answers a merge's merged message again, built as the bridge builds the one it publishes (issue 1585) Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../mesh-delivery-forge-bridge-tools/main.go | 84 ++++++++++++++++++- .../main_test.go | 76 ++++++++++++++++- .../cmd/mesh-delivery-forge-bridge/call.go | 51 +---------- .../cmd/mesh-delivery-forge-bridge/groups.go | 29 +------ .../internal/bridge/call.go | 79 +++++++++++++++++ .../internal/groups/groups.go | 25 ++++++ .../manifest_test.go | 4 +- .../mesh-delivery-forge-bridge/module.json | 3 +- 8 files changed, 269 insertions(+), 82 deletions(-) create mode 100644 modules/mesh-delivery-forge-bridge/internal/bridge/call.go diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go index 86faa05a..74a3f561 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main.go @@ -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) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go index 2d1b9e6f..b36dc76c 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go @@ -1,13 +1,18 @@ package main import ( + "bytes" "encoding/json" + "errors" "strings" "testing" "time" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg" "git.novox.be/novox/mesh-sdk/go/delivery" + "mesh-delivery-forge-bridge/internal/bridge" + "mesh-delivery-forge-bridge/internal/groups" ) @@ -51,7 +56,7 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { func() ([]groups.Row, []string, error) { rows, unread := groups.Rows(merges, performed, map[string]uint64{"ended": 7}, asks) return rows, unread, nil - }) + }, nil) call := func(name string, args map[string]any) any { for _, tl := range tools { if tl.Name == name { @@ -114,3 +119,72 @@ func TestTheBridgeToolsReadTheGroupsAndAskTheEndAgain(t *testing.T) { } } } + +// 1585: bridge_merged_message answers the merged message as the bridge publishes it: for a lone pull request the same bytes +// the event gives through Convert, for a member of a delivery group those with its group and members set as the process +// sets them; a merge the gate does not hold, or a forge that does not answer, is said. +func TestTheBridgeToolAnswersTheMergedMessageAsPublished(t *testing.T) { + h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) + c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) + trees := func(owner, repo, sha string) (string, map[string]string, error) { + return "root-" + sha[:4], map[string]string{"": "r", "modules/a": "ta-" + sha[:4]}, nil + } + files := func(owner, repo, path, ref string) ([]byte, error) { + if strings.HasSuffix(path, "module.json") { + return []byte(`{"module":"a"}`), nil + } + return nil, bridge.ErrNotFound + } + published := func(repo string, number int, head, branch, merge string) forgemsg.Forge { + body, _ := json.Marshal(map[string]any{"owner": "novox", "repo": repo, "number": number, "head": branch, "head_sha": head, "base": "main", "merge_commit_sha": merge}) + msg, err := bridge.Convert(bridge.MergedSubject, body, trees, files) + if err != nil { + t.Fatal(err) + } + f, err := forgemsg.DecodeForge(msg.Data) + if err != nil { + t.Fatal(err) + } + return f + } + lone, _ := delivery.EncodeMerge(delivery.Merge{Group: "solo", Ask: "a", Approved: map[string]string{"novox/a@5": h1}, + Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 5, Head: h1}, State: delivery.MergeMerged, Commit: c1}}}) + group, _ := delivery.EncodeMerge(delivery.Merge{Group: "feat/g", Ask: "a", Approved: map[string]string{"novox/a@6": h1, "novox/b@7": h2}, + Pulls: []delivery.MergePull{{Head: delivery.Head{Repository: "novox/a", Number: 6, Head: h1}, State: delivery.MergeMerged, Commit: c1}, + {Head: delivery.Head{Repository: "novox/b", Number: 7, Head: h2}, State: delivery.MergeMerging}}}) + for _, c := range []struct { + name string + raws map[string][]byte + repository string + merge string + want forgemsg.Forge + fail string + }{ + {name: "a lone pull request", raws: map[string][]byte{"solo": lone}, repository: "novox/a", merge: c1, want: published("a", 5, h1, "solo", c1)}, + {name: "a member of a group", raws: map[string][]byte{"feat~g": group}, repository: "novox/a", merge: c1, want: func() forgemsg.Forge { + f := published("a", 6, h1, "feat/g", c1) + f.Group, f.Members = "feat/g", []delivery.Head{{Repository: "novox/a", Number: 6, Head: h1}, {Repository: "novox/b", Number: 7, Head: h2}} + return f + }()}, + {name: "a merge the gate does not hold", raws: map[string][]byte{"solo": lone, "bad": []byte("{")}, repository: "novox/a", merge: c2, fail: "hold no merge"}, + } { + t.Run(c.name, func(t *testing.T) { + raw, err := MergedOf(c.raws, c.repository, c.merge, trees, files) + if c.fail != "" { + if err == nil || !strings.Contains(err.Error(), c.fail) { + t.Fatalf("err %v, want %q", err, c.fail) + } + return + } + want, _ := forgemsg.EncodeForge(c.want) + if err != nil || !bytes.Equal(raw, want) { + t.Fatalf("err %v, answer\n%s\nwant the published message\n%s", err, raw, want) + } + }) + } + if _, err := MergedOf(map[string][]byte{"solo": lone}, "novox/a", c1, func(string, string, string) (string, map[string]string, error) { + return "", nil, errors.New("the forge does not answer") + }, files); err == nil || !strings.Contains(err.Error(), "does not answer") { + t.Errorf("a forge that does not answer: %v", err) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go index 9310a73d..abf7f86b 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go @@ -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 : 404 ` (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)) } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go index 82b1eb12..c8026a15 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/groups.go @@ -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 }) diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/call.go b/modules/mesh-delivery-forge-bridge/internal/bridge/call.go new file mode 100644 index 00000000..d1160c79 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/call.go @@ -0,0 +1,79 @@ +package bridge + +import ( + "context" + "encoding/json" + "fmt" + "regexp" +) + +// Caller calls a module's tool by its manifest key (`.`) 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 : 404 ` (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) +} diff --git a/modules/mesh-delivery-forge-bridge/internal/groups/groups.go b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go index bd09dbb7..8b731a47 100644 --- a/modules/mesh-delivery-forge-bridge/internal/groups/groups.go +++ b/modules/mesh-delivery-forge-bridge/internal/groups/groups.go @@ -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 +} diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index f702fa2f..59b8ed3c 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -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). diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index 338d9cdc..4b677cce 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -18,7 +18,8 @@ ], "tools": [ "bridge_groups", - "bridge_group_end_again" + "bridge_group_end_again", + "bridge_merged_message" ], "reads": [ "mesh-delivery-gate.merges", From 3c832d1a2da0783a9f232cb75399dd041179248c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 11:37:13 +0100 Subject: [PATCH 4/5] 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 --- .../mesh-delivery-indexer/bridgemessage.go | 32 ++++++ .../cmd/mesh-delivery-indexer/forgeread.go | 31 ----- .../mesh-delivery-indexer/integration_test.go | 22 ++++ .../cmd/mesh-delivery-indexer/main.go | 2 +- .../cmd/mesh-delivery-indexer/takein.go | 6 +- .../internal/index/candidates.go | 27 +++-- .../internal/index/pending.go | 75 ++++++------ .../internal/index/pending_test.go | 108 ++++++++++++------ .../internal/index/pending_tools.go | 27 ++--- .../internal/index/takein.go | 52 +++++---- .../mesh-delivery-indexer/manifest_test.go | 6 +- modules/mesh-delivery-indexer/module.json | 3 +- 12 files changed, 236 insertions(+), 155 deletions(-) create mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go delete mode 100644 modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go new file mode 100644 index 00000000..98e9a850 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/bridgemessage.go @@ -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 + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go deleted file mode 100644 index 6cd3b277..00000000 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/forgeread.go +++ /dev/null @@ -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 - } -} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 345cb0f2..023478a4 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -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) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 8db24813..7f21ad2a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -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)}, diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 34bfa0f7..cfa942d2 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -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) diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 4398f7e0..381ce794 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -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) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 60f94e44..be323c74 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -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() diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 0de75c2d..7e6456fc 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -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{}) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index 16f3570b..db6cd499 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -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., 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}, } } diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index 899666c4..e6cb17c6 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -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.`, 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.`, 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 diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index eb14f64d..f04f6d04 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -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 diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 48ee9cdd..3ae6276e 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -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": { From 655791b8a351e178ef5b73983fa8abcc0e6f2f54 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 11:42:10 +0100 Subject: [PATCH 5/5] Bridge tool test: the merged-message rows are a subtest of the existing tools test, within the test budget (issue 1585) Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../main_test.go | 128 +++++++++--------- 1 file changed, 64 insertions(+), 64 deletions(-) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go index b36dc76c..2454e5ec 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge-tools/main_test.go @@ -118,73 +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. -func TestTheBridgeToolAnswersTheMergedMessageAsPublished(t *testing.T) { - h1, h2 := strings.Repeat("1", 40), strings.Repeat("2", 40) - c1, c2 := strings.Repeat("c", 40), strings.Repeat("d", 40) - trees := func(owner, repo, sha string) (string, map[string]string, error) { - return "root-" + sha[:4], map[string]string{"": "r", "modules/a": "ta-" + sha[:4]}, nil - } - files := func(owner, repo, path, ref string) ([]byte, error) { - if strings.HasSuffix(path, "module.json") { - return []byte(`{"module":"a"}`), nil + // 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 } - 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) + files := func(owner, repo, path, ref string) ([]byte, error) { + if strings.HasSuffix(path, "module.json") { + return []byte(`{"module":"a"}`), nil + } + return nil, bridge.ErrNotFound } - 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}} + 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 - }()}, - {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) + } + 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 } - 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) - } + 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) + } + }) }