From c2b8ac3370278ba040d5313881a2f302e3f08b3d Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 18:35:22 +0100 Subject: [PATCH] The indexer hears the checker's events after it lists the open tries; rows for the start order and the config skip (issue 1921) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- .../cmd/mesh-delivery-indexer/main.go | 21 ++++++++------ .../internal/index/control_records_test.go | 17 +++++++++++ .../internal/index/previews_test.go | 28 +++++++++++++++++++ .../internal/index/trycandidates.go | 3 -- 4 files changed, 57 insertions(+), 12 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 b9677157..86ba8676 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -494,15 +494,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer tc.Stop() } - switch pcons, err := cred.EventConsumer(ctx, js, PreviewsStream); { - case err != nil: - logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, err) - default: - pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.heardChecks(m) }, logger.Printf) - if err != nil { - return err - } - defer pc.Stop() + // The consumer on the checker's stream exists before the open tries are listed and is heard after (ADR 0638 §3): an event + // said between the call and the hearing waits in it, and a try that started or ended meanwhile is found either way. + pcons, perr := cred.EventConsumer(ctx, js, PreviewsStream) + if perr != nil { + logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's previews and tries are not heard until the indexer starts again: %v", cred.Node, cred.Module, PreviewsStream, perr) } // 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). When the build this program runs is not known yet, nothing waits for it: the consumer @@ -518,6 +514,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := ix.RecoverTries(ctx); err != nil { logger.Printf("%v", err) } + if perr == nil { + pc, err := process.HearEvents(pcons, func(m jetstream.Msg) { previews.heardChecks(m) }, logger.Printf) + if err != nil { + return err + } + defer pc.Stop() + } 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/internal/index/control_records_test.go b/modules/mesh-delivery-indexer/internal/index/control_records_test.go index 358d802b..0b2ddaba 100644 --- a/modules/mesh-delivery-indexer/internal/index/control_records_test.go +++ b/modules/mesh-delivery-indexer/internal/index/control_records_test.go @@ -85,5 +85,22 @@ func TestTheControlRecords(t *testing.T) { if _, ok := cands.vals[key]; ok { t.Error("a closed pull request's declarations stayed") } + // A group try's candidate (number 0) holds no pull request's declarations: a merge that touches its module writes none + // for it (fails when configCandidatesAfterMerge no longer skips a try's group). + tryPull := delivery.CandidatePull{Head: delivery.Head{Repository: catalogueRepo, Head: h1}, Modules: []string{"avahi"}, + Fps: map[string]delivery.CandidateModule{"avahi": {Manifest: delivery.ManifestDigest([]byte(trunk)), None2: NoneNoPreview}}} + tryCand := delivery.Candidate{V: delivery.Version, Group: "try-5", Pulls: []delivery.CandidatePull{tryPull}} + tryCand.HeadsHash, _ = delivery.HeadsHash([]delivery.Head{tryPull.Head}) + tryCand.FPSet = delivery.FingerprintSet(tryCand.Pulls) + data, err := delivery.EncodeCandidate(tryCand) + if err != nil { + t.Fatal(err) + } + if err := ix.Store.Create(ctx, "try-5", data); err != nil { + t.Fatal(err) + } + if err := ix.configCandidatesAfterMerge(ctx, mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "t2"}), []string{"avahi"}); err != nil || len(cands.vals) != 0 { + t.Errorf("a merge touching a try's module wrote %v for it (%v), want none", cands.vals, err) + } }) } diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 5a81a83d..701fdc8f 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -308,6 +308,34 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW if err := ix.RecoverTries(ctx); err != nil || len(held()) != 0 { t.Errorf("recovered with no open try: %v, candidates %v, want none", err, held()) } + // ADR 0638 §3, §4: the consumer exists before the call, so what happens between the call and the hearing is heard after + // it. A try the call listed and whose start is then heard gets one candidate; a try started after the list gets one; a + // try that ended during the call, listed as open, leaves none once its end is heard. + ix.OpenTries = func(context.Context) ([]tryevent.Open, error) { + return []tryevent.Open{{ID: "try-7", Whole: true, Members: []tryevent.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}, + {ID: "try-9", Whole: true, Members: []tryevent.Member{{Repository: catalogueRepo, Head: h1, Trunk: trunkNow}}}}, nil + } + if err := ix.RecoverTries(ctx); err != nil { + t.Fatal(err) + } + for _, step := range []struct { + what string + do func() error + want int + }{ + {"the start heard after the call, of a try the call listed", func() error { return ix.StartTry(ctx, "try-7", members) }, 2}, + {"a start after the list", func() error { return ix.StartTry(ctx, "try-10", members) }, 3}, + {"the end heard of a try ended during the call", func() error { return ix.EndTry(ctx, "try-9") }, 2}, + } { + if err := step.do(); err != nil || len(held()) != step.want { + t.Fatalf("%s: %v, candidates %v, want %d", step.what, err, held(), step.want) + } + } + for _, id := range []string{"try-10"} { + if err := ix.EndTry(ctx, id); err != nil { + t.Fatal(err) + } + } if err := ix.StartTry(ctx, "try-7", members); err != nil || len(held()) != 1 { t.Fatalf("started again: %v, candidates %v", err, held()) } diff --git a/modules/mesh-delivery-indexer/internal/index/trycandidates.go b/modules/mesh-delivery-indexer/internal/index/trycandidates.go index 6044ebe3..ae2b0fb7 100644 --- a/modules/mesh-delivery-indexer/internal/index/trycandidates.go +++ b/modules/mesh-delivery-indexer/internal/index/trycandidates.go @@ -21,9 +21,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" ) -// TriesBucket is the checker's state of tries, which the indexer reads. -const TriesBucket = delivery.CheckerModule + "_tries" - // TryCandidateTools: the tools that read and drop a try's candidate. const ( TryCandidatesTool = "indexer_try_candidates"