From 531d05ab4b9f2c35d039338f5a57a4483da44358 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:08:37 +0100 Subject: [PATCH] =?UTF-8?q?indexer:=20hear=20a=20group's=20end,=20keep=20m?= =?UTF-8?q?erges=20pending,=20take=20in=20at=20start=20(ADR=200633=20?= =?UTF-8?q?=C2=A72,=20=C2=A74,=20issue=201585)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-indexer/main.go | 58 ++++++++++++++++++- .../cmd/mesh-delivery-indexer/main_test.go | 18 ++++++ .../cmd/mesh-delivery-indexer/verdicts.go | 25 ++++++++ .../internal/index/pending.go | 17 +++++- 4 files changed, 114 insertions(+), 4 deletions(-) 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 0d896889..ccdbe7ac 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/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. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index eb70b62f..f42d7972 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/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) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go index 79e1b316..495cb684 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/verdicts.go +++ b/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..event.take-in..`. 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 + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 1fdb7977..e88b20c0 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/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.