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"}}},