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
This commit is contained in:
2026-11-05 18:35:22 +01:00
parent 92c38981ea
commit c2b8ac3370
4 changed files with 57 additions and 12 deletions
@@ -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.
@@ -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)
}
})
}
@@ -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())
}
@@ -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"