indexer: hear a group's end, keep merges pending, take in at start (ADR 0633 §2, §4, issue 1585)

The consumer tells a group's end apart by its name before the forge decode,
a merged message goes through the pending state, and the pending revisions
naming this build are taken in before the consumer starts; take-in events go
on the indexer's own stream. Until the pending state is on the bus every
merge is taken in at once, as before.

Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go

@@ -16,6 +16,7 @@ import (
 	"log"
 	"os"
 	"os/signal"
+	"strings"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -153,6 +154,30 @@ type delivered interface {
 	Metadata() (*jetstream.MsgMetadata, error)
 }
 
+// handleGroupEnd takes a group's end in (index.EndGroup): a body that does not read is terminated and said; an end
+// whose members are not all held yet, or while the build this program runs is not known, is left for redelivery.
+func handleGroupEnd(ctx context.Context, m delivered, ix *index.Indexer, h *index.Heard, logger *log.Logger) {
+	end, err := delivery.DecodeForgeGroupEnd(m.Data())
+	if err != nil {
+		logger.Printf("%s: not a group's end, dropped: %v", m.Subject(), err)
+		h.Outcome, h.Words = index.HeardDropped, err.Error()
+		_ = m.Term()
+		return
+	}
+	h.Kind, h.Branch = delivery.ForgeGroupEndEvent, end.Group
+	applied, words, err := ix.EndGroup(ctx, end)
+	h.Written, h.Late = applied.Written, applied.Late
+	if err != nil {
+		logger.Printf("%s: the end of the group %q left for redelivery: %v", m.Subject(), end.Group, err)
+		h.Outcome, h.Words = index.HeardLeft, err.Error()
+		return
+	}
+	h.Outcome, h.Words = index.HeardApplied, words
+	if err := m.DoubleAck(ctx); err != nil {
+		logger.Printf("%s: the end of the group %q applied, its acknowledgement was not confirmed: %v", m.Subject(), end.Group, err)
+	}
+}
+
 // handle takes one message of FORGE into the candidates and acknowledges it. A message it can never read is terminated,
 // never redelivered, and the log names its subject and why. A write that failed stays unacknowledged: the server
 // delivers it again (the server's ack wait is the only clock). Each message handled is written to the heard state
@@ -171,6 +196,11 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto
 			logger.Printf("%s: sequence %d is not written to the heard state: %v", h.Subject, h.Sequence, err)
 		}
 	}()
+	// A delivery group's end (novox/hq ADR 0633 §4) has a body of its own: told apart by its name, before the decode.
+	if strings.HasSuffix(m.Subject(), "."+delivery.ForgeGroupEndEvent) {
+		handleGroupEnd(ctx, m, ix, &h, logger)
+		return
+	}
 	f, err := delivery.DecodeForge(m.Data())
 	if err != nil {
 		logger.Printf("%s: not a forge message, dropped: %v", m.Subject(), err)
@@ -182,7 +212,13 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, heard index.Sto
 	if f.Merge != "" {
 		h.Commit = f.Merge
 	}
-	applied, err := ix.ApplyMessage(ctx, f)
+	var applied index.Applied
+	if f.Kind == delivery.ForgeMerged && h.Sequence > 0 {
+		// A merge is taken in only by the indexer its revision names (ADR 0633 §1): kept in pending until then.
+		applied, h.Words, err = ix.Arrive(ctx, h.Sequence, f)
+	} else {
+		applied, err = ix.ApplyMessage(ctx, f)
+	}
 	h.Written, h.Late = applied.Written, applied.Late
 	switch {
 	case errors.Is(err, index.ErrUnreadable):
@@ -308,7 +344,11 @@ 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},
+		// 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)},
+		SayTakeIn: sayTakeIn(js, cred)}
 	// The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each
 	// new one indexes the module it names when that module is registered and the bucket holds none (registered.go).
 	var started atomic.Bool
@@ -458,6 +498,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		defer pc.Stop()
 	}
+	// The pending revisions that name this build are taken in now, oldest first, before the consumer on the forge bridge's
+	// stream starts (ADR 0633 §2); the build this program runs is asked until it is known, said every minute. A hand
+	// take-in (indexer_take_in_pending) is answered meanwhile: the takeins watch runs already.
+	if _, words, err := ix.TakeInAtStart(ctx, func(ctx context.Context) bool {
+		// A context's bound between two asks of the node-engine, no timer (ADR 0437 §8), as askTakeIn bounds its wait.
+		again, cancel := context.WithTimeout(ctx, time.Minute)
+		defer cancel()
+		<-again.Done()
+		return ctx.Err() == nil
+	}); err != nil {
+		return fmt.Errorf("the take-in at start: %w", err)
+	} else if words != "" {
+		logger.Printf("take-in at start: still waiting: %s", words)
+	}
 	cons, err := cred.EventConsumer(ctx, js, forgeStream)
 	if err != nil {
 		// Essential: the forge's merges are what the indexer indexes.

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go

@@ -102,6 +102,24 @@ func TestAMessageThatIsNoForgeMessageIsTerminatedAndLogged(t *testing.T) { // 12
 	if h := heardOf(t, heard); h.Outcome != index.HeardDropped || h.Sequence != 41 || h.Words == "" {
 		t.Errorf("heard = %+v", h)
 	}
+	// 1585/T9 (ADR 0633 §4): a group's end is told apart by its name before the forge decode: one that reads is acked
+	// (an indexer with no pending state takes the members in one by one), one that does not is dropped as a group's end.
+	for _, c := range []struct {
+		data   string
+		termd  int
+		logged string
+	}{{`{"v":1,"group":"feat/x","members":[{"repository":"novox/mesh-sdk","number":3,"merge":"` + head + `"}]}`, 0, ""},
+		{`{"v":1,"group":"feat/x"}`, 1, "not a group's end"}} {
+		logs.Reset()
+		m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event." + delivery.ForgeGroupEndEvent, data: []byte(c.data)}
+		handle(context.Background(), m, &index.Indexer{Store: st, Builds: index.NewBuilds()}, heard, log.New(&logs, "", 0))
+		if m.termd != c.termd || m.acked+m.termd != 1 || !strings.Contains(logs.String(), c.logged) {
+			t.Errorf("group end %s: terminated %d acked %d log %q", c.data, m.termd, m.acked, logs.String())
+		}
+		if h := heardOf(t, heard); h.Kind != delivery.ForgeGroupEndEvent && c.termd == 0 {
+			t.Errorf("heard of a group's end = %+v", h)
+		}
+	}
 }
 
 func TestTheSourceMainKeyIsTakenAtItsRevision(t *testing.T) { // 1344/AC1 (the watch's side: key, revision, value)

modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go

@@ -62,3 +62,28 @@ func sayAccepted(js jetstream.JetStream, cred process.Account, logf func(string,
 		}
 	}
 }
+
+// sayTakeIn publishes a take-in event on the indexer's own stream take-in (novox/hq ADR 0633 §5):
+// `mesh.stream.<module>.event.take-in.<kind>.<key>`. A publish that fails is the caller's to say.
+func sayTakeIn(js jetstream.JetStream, cred process.Account) func(context.Context, delivery.TakeIn) error {
+	return func(ctx context.Context, t delivery.TakeIn) error {
+		event, err := delivery.TakeInEvent(t.Kind, t.Key)
+		if err != nil {
+			return err
+		}
+		data, err := delivery.EncodeTakeIn(t)
+		if err != nil {
+			return err
+		}
+		msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, event))
+		msg.Header.Set("x-source", cred.Module)
+		msg.Header.Set("x-node", cred.Node)
+		msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+		msg.Header.Set("content-type", "application/json")
+		msg.Data = data
+		pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
+		defer cancel()
+		_, err = js.PublishMsg(pub, msg)
+		return err
+	}
+}

modules/mesh-delivery-indexer/internal/index/pending.go

@@ -265,7 +265,13 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f delivery.Forge) (Ap
 	if err != nil {
 		return Applied{}, "", err
 	}
-	if err := ix.Pending.Create(ctx, key, data); err != nil && !errors.Is(err, ErrConflict) {
+	switch err := ix.Pending.Create(ctx, key, data); {
+	case errors.Is(err, process.ErrNotYet):
+		// The state this delivery brings is not on the bus yet (issue 1790): the merge is taken in at once, as before.
+		ix.logf("take-in: the state %s is not on the bus yet, so %s#%d is taken in at once, as before ADR 0633", PendingBucket, f.Repository, f.Number)
+		a, err := ix.ApplyMessage(ctx, f)
+		return a, "", err
+	case err != nil && !errors.Is(err, ErrConflict):
 		return Applied{}, "", fmt.Errorf("%s.%s: %w", PendingState, key, err)
 	}
 	if f.Group != "" {
@@ -299,6 +305,9 @@ func (ix *Indexer) EndGroup(ctx context.Context, end delivery.ForgeGroupEnd) (Ap
 	ix.pendingMu.Lock()
 	defer ix.pendingMu.Unlock()
 	entries, err := ReadPending(ctx, ix.Pending, ix.logf)
+	if errors.Is(err, process.ErrNotYet) {
+		return Applied{}, "the pending state is not on the bus yet: the group's merges were taken in one by one", nil
+	}
 	if err != nil {
 		return Applied{}, "", err
 	}
@@ -392,7 +401,11 @@ func (ix *Indexer) TakeInAtStart(ctx context.Context, wait func(context.Context)
 	}
 	ix.pendingMu.Lock()
 	defer ix.pendingMu.Unlock()
-	return ix.evaluate(ctx, own, true)
+	a, words, err := ix.evaluate(ctx, own, true)
+	if errors.Is(err, process.ErrNotYet) {
+		return a, "the pending state is not on the bus yet: nothing waits", nil
+	}
+	return a, words, err
 }
 
 // evaluate works out what pending's ready merges name, together (ADR 0633 §3), and takes them in or holds them.