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

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)

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-05 11:24:18 +01:00
parent 4b7a31b2c7
commit 343321f11e
4 changed files with 129 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/internal/index/candidates.go
+3 -1
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/pending.go
+102 -41
View File
@@ -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)
modules/mesh-delivery-indexer/internal/index/pending_test.go
+22 -20
View File
@@ -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") {
modules/mesh-delivery-indexer/internal/index/pending_tools.go
+2 -1
View File
@@ -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"}}},