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

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
+	}
+}

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

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 != "" {

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

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

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
+}

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
+}

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.<group>) is a delivery group whose members are gathered until the " +
 				"forge bridge says its end; a merge in one of its members' repositories waits behind it (behind). Each names its " +
-				"merges, since when it waits, the modules it holds back (holds) and the last take-in event said of it. Empty waits: " +
+				"merges, since when it waits, the modules it holds back (holds), the merges whose message the forge bridge's stream no " +
+				"longer holds (lost: they are not taken in; indexer_drop_pending drops them) and the last take-in event said of it. Empty waits: " +
 				"every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " +
 				"indexer_take_in_pending and indexer_drop_pending steer a wait.",
 			Input: map[string]any{"type": "object", "properties": map[string]any{}},
@@ -65,5 +78,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.<sequence>, as indexer_pending names its merges' wait"},
 				"why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}},
 			Run: hand(true)},
+		{Name: TakeInForgeTool,
+			Description: "Take one merged message of the forge bridge's stream in by hand, by its stream sequence (ADR 0633 §5, issue 1585): " +
+				"the indexer reads it from the stream and hands it to the take-in as a delivered merged message is, so it is held for " +
+				"the indexer its revision names or taken in. For a merge the take-in did not take in (indexer_heard names its " +
+				"sequence and why) or whose pending merge was dropped. A hand act: who and why are said in the indexer's journal. " +
+				"A sequence the stream no longer holds answers so.",
+			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},
 	}
 }

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.<sequence>`, and the value.
+func NewForgeAsk(seq uint64, why, by string, at time.Time) (string, []byte, error) {
+	if seq == 0 {
+		return "", nil, errors.New("name the forge message to take in by its stream sequence (sequence), as indexer_heard names it")
+	}
+	if strings.TrimSpace(why) == "" {
+		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.<take-in|drop>.<key>`, and the value. key is a wait's key as indexer_pending names it, for a
 // drop one pending merge's.

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

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

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.