diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index 0d55bda..46a0100 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -23,7 +23,7 @@ type judgeDeps struct { kept check.KV previews check.KV repos *repoModules - announce func(ctx context.Context, key string) + announce check.Announcer logf func(format string, args ...any) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 537680e..8b03ec9 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -32,9 +32,11 @@ import ( "mesh-delivery-checker/internal/check" ) -// previewWrittenSubject is the subject of the checker's preview.written event, on its stream previews (the bus names the -// stream PREVIEWS); the indexer's watch of the previews bucket is woken by it. -const previewWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.preview.written" +// eventSubject is the subject of an event of the checker's stream checks (the bus names the stream CHECKS): the indexer's +// watch of the previews bucket and the forge bridge's watch of the owed bucket are woken by them. +func eventSubject(event string) string { + return "mesh.stream." + delivery.CheckerModule + ".event." + event +} const ( // defaultStepBound is how long the checker waits for one step of a build module (a fetch downloads the Go modules, a @@ -196,18 +198,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Link: link, Logf: logger.Printf, } + announce := func(ctx context.Context, event, key string) { + body, _ := json.Marshal(map[string]string{"key": key}) + if _, err := js.Publish(ctx, eventSubject(event), body); err != nil { + logger.Printf("%s of %s was written and the part that reads it could not be told: %v", event, key, err) + } + } repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf} ch.Judge = newJudge(judgeDeps{ runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, logf: logger.Printf, - announce: func(ctx context.Context, key string) { - body, _ := json.Marshal(map[string]string{"key": key}) - if _, err := js.Publish(ctx, previewWrittenSubject, body); err != nil { - logger.Printf("previews.%s was written and the indexer could not be told: %v", key, err) - } - }, + announce: announce, }) + ch.Announce = announce // The toolchains first, so a check that starts from the first candidate finds the nodes. watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger) diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 9084403..84753ed 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -18,6 +18,17 @@ const DefaultBound = 30 * time.Minute // is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8). const NodesWords = "build node qualifies" +// The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads +// the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no +// timer). The body is {"key": ...}. +const ( + EventPreviewWritten = "preview.written" + EventOwedWritten = "owed.written" +) + +// Announcer says an event of the checker's stream, with the key written. +type Announcer func(ctx context.Context, event, key string) + // Outcome is what judging a case gives: the verdict (state, failure, levels, times, composition; the checker adds the // group, the heads and the trunk) and, for the line, what the check knows of the builds of the modules it touches. type Outcome struct { @@ -56,8 +67,10 @@ type Checker struct { Hold func() bool // Link is the base of the address a status points at; the verdict key follows. Link string - Now func() time.Time - Logf func(format string, args ...any) + // Announce says a status was written (EventOwedWritten); nil says nothing. + Announce Announcer + Now func() time.Time + Logf func(format string, args ...any) once sync.Once sem chan struct{} @@ -294,7 +307,7 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { return } began := ch.now() - ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil) + ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil, nil) out := ch.judge(ctx, cs) if ctx.Err() != nil { return @@ -322,7 +335,7 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) { ch.nodesFail[cs.Key] = epoch ch.mu.Unlock() } - ch.writeOwed(ctx, cs, v, out.Builds) + ch.writeOwed(ctx, cs, v, out.Builds, out.Slowest) ch.logf("%s: %s", cs.Key, v.State) } @@ -360,7 +373,7 @@ func failure(machine, words string) delivery.Verdict { // writeOwed writes, for each pull request judged, the status the forge bridge performs: pending while the check runs, success // or failure at its end (ADR 0484 clause 2). A key already holding the same value is not written again, so the bridge // performs one act for each change; an older head's key is never touched. -func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, builds []BuildNote) { +func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, builds []BuildNote, slowest []string) { for _, p := range cs.Pulls { key, err := delivery.CheckerOwedKey(p.Repository, p.Number, p.Head) if err != nil { @@ -369,32 +382,35 @@ func (ch *Checker) writeOwed(ctx context.Context, cs *Case, v delivery.Verdict, } state := map[string]string{delivery.VerdictRunning: "pending", delivery.VerdictPass: "success", delivery.VerdictFail: "failure"}[v.State] o := delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, - Target: strings.TrimSuffix(ch.Link, "/") + "/" + cs.Key, Text: Line(v, cs, p, builds)} + Target: strings.TrimSuffix(ch.Link, "/") + "/" + cs.Key, Text: Line(v, cs, p, builds, slowest)} data, err := delivery.EncodeCheckerOwed(o) if err != nil { ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) continue } - if err := putOnce(ctx, ch.Owed, key, data, func(cur []byte) bool { + wrote, err := putOnce(ctx, ch.Owed, key, data, func(cur []byte) bool { held, err := delivery.DecodeCheckerOwed(cur) o.V = delivery.Version return err == nil && delivery.SameOwed(held, o) - }); err != nil { + }) + if err != nil { ch.logf("%s: the status of %s cannot be written: %v", cs.Key, key, err) + } else if wrote && ch.Announce != nil { + ch.Announce(ctx, EventOwedWritten, key) } } } // putOnce writes data under key by create, or by compare-and-set when the key exists, and writes nothing when same says the // key holds the value already. -func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur []byte) bool) error { +func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur []byte) bool) (bool, error) { for try := 0; try < 4; try++ { cur, rev, found, err := kv.Get(ctx, key) if err != nil { - return err + return false, err } if found && same(cur) { - return nil + return false, nil } if !found { _, err = kv.Create(ctx, key, data) @@ -404,9 +420,9 @@ func putOnce(ctx context.Context, kv KV, key string, data []byte, same func(cur if errors.Is(err, ErrConflict) { continue } - return err + return err == nil, err } - return fmt.Errorf("%s changed under every one of 4 tries", key) + return false, fmt.Errorf("%s changed under every one of 4 tries", key) } // Sweep closes, at start, every verdict left running for a candidate that no longer holds those heads: nothing will check diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index 4a791f4..f57ed71 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -365,3 +365,23 @@ func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T) t.Errorf("the live key = %+v, want it left for the check to take over", v) } } + +func TestTheForgeBridgeIsToldOnceForEachChangeOfAStatus(t *testing.T) { // 1683 (the checker's half of AC5) + ctx, j := context.Background(), passJudge() + r := newRig(t, j) + var told []string + r.ch.Announce = func(_ context.Context, event, key string) { told = append(told, event+" "+key) } + cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1)) + r.ch.Candidate(ctx, "feat%2Fx", cand) + r.ch.Wait() + key, _ := delivery.CheckerOwedKey(repoA, 7, head1) + want := []string{EventOwedWritten + " " + key, EventOwedWritten + " " + key} // pending, then success + if len(told) != 2 || told[0] != want[0] || told[1] != want[1] { + t.Fatalf("told %v, want the status said when pending and when success", told) + } + r.ch.Candidate(ctx, "feat%2Fx", cand) // nothing changed: nothing written, nothing said + r.ch.Wait() + if len(told) != 2 { + t.Errorf("told %v after a check that changed nothing", told) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/examiner_test.go b/modules/mesh-delivery-checker/internal/check/examiner_test.go index 270a504..ba1109d 100644 --- a/modules/mesh-delivery-checker/internal/check/examiner_test.go +++ b/modules/mesh-delivery-checker/internal/check/examiner_test.go @@ -508,11 +508,11 @@ func TestAPreviewIsWrittenOnceAndAnExistingKeyStands(t *testing.T) { func TestTheIndexerIsToldOfAPreviewOnceWhenItIsCreated(t *testing.T) { // 1445/AC4 (the checker's half) b := newBench(t) var told []string - b.ex.Announce = func(_ context.Context, key string) { told = append(told, key) } + b.ex.Announce = func(_ context.Context, event, key string) { told = append(told, event+" "+key) } b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) key, _ := delivery.PreviewKey(repoA, head1, trunkA) - if len(told) != 1 || told[0] != key { + if len(told) != 1 || told[0] != EventPreviewWritten+" "+key { t.Errorf("told %v, want once of %s", told, key) } } diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index c29508d..0a9ce4b 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -50,8 +50,8 @@ type Examiner struct { // Previews is the previews bucket; nil writes none. Previews KV Modules BuildModules - // Announce says a preview was written under the key (the checker's preview.written event); nil says nothing. - Announce func(ctx context.Context, key string) + // Announce says a preview was written under the key (EventPreviewWritten); nil says nothing. + Announce Announcer // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } Now func() time.Time diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go index d2aaced..db675c2 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -2,13 +2,26 @@ package check import ( "fmt" + "strings" + "unicode/utf8" "git.novox.be/novox/mesh-sdk/go/delivery" ) -// Line is the words of a status for one pull request of a case (ADR 0447 clause 3, ADR 0484 clause 7). -func Line(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote) string { - return fmt.Sprintf("%s: %s", v.State, short12(p.Head)) +// MaxLine is the most characters a status's line has: the forge shows a status's description in a short space, and the full +// report is linked (ADR 0447 clause 3, ADR 0484 clause 7). A line that would be longer drops its lesser parts in a fixed +// order, and is cut last. +const MaxLine = 140 + +// seconds is a time in milliseconds in the unit a person reads: 350ms, 4.1s, 72s. +func seconds(ms int64) string { + switch { + case ms < 1000: + return fmt.Sprintf("%dms", ms) + case ms < 10000: + return fmt.Sprintf("%.1fs", float64(ms)/1000) + } + return fmt.Sprintf("%ds", (ms+500)/1000) } func short12(s string) string { @@ -17,3 +30,155 @@ func short12(s string) string { } return s } + +// firstFailed is what failed first, in words: a conflict, a refused step or a test, named before anything else. +func firstFailed(v delivery.Verdict) string { + f := v.Failed + switch { + case f.Test != "": + if f.Words != "" { + return f.Test + ": " + f.Words + } + return f.Test + case f.Machine != "" && f.Words != "": + return f.Machine + ": " + f.Words + case f.Machine != "": + return f.Machine + } + return f.Words +} + +// pieces are the parts of a line, from the most to the least needed. +type pieces struct { + head, commits, counts, times, over, builds, composition, group string +} + +func (p pieces) join(drop int) string { + parts := []string{p.head, p.commits, p.counts} + if drop < 3 && p.times != "" { + parts = append(parts, p.times) + } + if p.over != "" { + parts = append(parts, p.over) + } + if drop < 1 && p.builds != "" { + parts = append(parts, p.builds) + } + if drop < 2 && p.group != "" { + parts = append(parts, p.group) + } + if p.composition != "" { + parts = append(parts, p.composition) + } + out := make([]string, 0, len(parts)) + for _, s := range parts { + if s != "" { + out = append(out, s) + } + } + return strings.Join(out, " | ") +} + +func levelWords(name string, r delivery.LevelResult) string { + s := fmt.Sprintf("%s %d run, %d kept", name, r.Run, r.Kept) + if r.Skipped > 0 { + s += fmt.Sprintf(", %d skipped", r.Skipped) + } + return s +} + +func compositionWords(c *delivery.Composition) string { + switch { + case c == nil, c.State == delivery.CompositionNotRun: + return "composition not yet run" + case c.State == delivery.CompositionPass: + return "composition passed with " + short12(c.SDK) + } + return "composition failed: " + c.Words +} + +func buildWords(builds []BuildNote) string { + if len(builds) == 0 { + return "" + } + parts := make([]string, len(builds)) + for i, b := range builds { + parts[i] = b.Module + " " + b.State + } + return "builds: " + strings.Join(parts, ", ") +} + +// parts makes the pieces of the line of a pull request of a case. +func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []string) pieces { + var pc pieces + commits := short12(p.Head) + " on " + short12(p.Trunk) + switch v.State { + case delivery.VerdictRunning: + pc.head = "checking " + commits + case delivery.VerdictPass: + pc.head = "pass " + commits + default: + pc.head = "failed: " + firstFailed(v) + pc.commits = commits + } + var counts []string + for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} { + if r, ok := v.Levels[level]; ok { + counts = append(counts, levelWords(level, r)) + } + } + pc.counts = strings.Join(counts, " | ") + if v.State != delivery.VerdictRunning { + t := v.Took + var times []string + for _, x := range []struct { + name string + ms int64 + }{{"queue", t.Queue}, {"clone", t.Clone}, {"unit", t.Unit}, {"int setup", t.IntegrationSetup}, {"int run", t.IntegrationRun}, {"composition", t.Composition}} { + if x.ms > 0 { + times = append(times, x.name+" "+seconds(x.ms)) + } + } + if len(times) > 0 { + pc.times = strings.Join(times, ", ") + } + } + if len(v.OverBound) > 0 { + pc.over = "over bound: " + strings.Join(v.OverBound, ", ") + if len(slowest) > 0 { + pc.over += " (slowest " + strings.Join(slowest, ", ") + ")" + } + } + pc.builds = buildWords(builds) + if v.State != delivery.VerdictRunning { + pc.composition = compositionWords(v.Composition) + } + if cs != nil && len(cs.Pulls) > 1 { + pc.group = fmt.Sprintf("group %s of %d pull requests", cs.Group, len(cs.Pulls)) + if v.State != delivery.VerdictPass { + pc.group = fmt.Sprintf("%s#%d in group %s", p.Repository, p.Number, cs.Group) + } + } + return pc +} + +// Report is the full line of a pull request's status, with no limit: what the read tools and the full report say. +func Report(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []string) string { + return parts(v, cs, p, builds, slowest).join(0) +} + +// Line is the words of a status for one pull request of a case (ADR 0447 clause 3, ADR 0484 clause 7): the verdict first (a +// failure names the conflict, the refused step or the test before anything), the head and the trunk commit as twelve +// characters each, the tests run, kept and skipped per level, the time of each part, a level over its bound with its slowest +// tests, the builds of the touched modules, and that the composition was not yet run. It is at most MaxLine characters: the +// builds, then the group, then the times drop in that order, and the line is cut at last. +func Line(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []string) string { + pc := parts(v, cs, p, builds, slowest) + for drop := 0; drop <= 3; drop++ { + if s := pc.join(drop); utf8.RuneCountInString(s) <= MaxLine { + return s + } + } + runes := []rune(pc.join(3)) + return string(runes[:MaxLine-1]) + "…" +} diff --git a/modules/mesh-delivery-checker/internal/check/line_test.go b/modules/mesh-delivery-checker/internal/check/line_test.go new file mode 100644 index 0000000..ccbce03 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/line_test.go @@ -0,0 +1,134 @@ +package check + +import ( + "strings" + "testing" + "unicode/utf8" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func lineCase(t testing.TB, pulls ...delivery.CandidatePull) (*Case, Pull) { + cs := caseOf(t, pulls...) + return cs, cs.Pulls[0] +} + +func passVerdict() delivery.Verdict { + return delivery.Verdict{State: delivery.VerdictPass, + Levels: map[string]delivery.LevelResult{delivery.LevelUnit: {Run: 12, Kept: 301}, delivery.LevelIntegration: {Run: 3, Kept: 40}}, + Took: delivery.Took{Queue: 800, Clone: 4100, Unit: 400, IntegrationSetup: 3200, IntegrationRun: 12000}, + Composition: &delivery.Composition{State: delivery.CompositionNotRun}} +} + +func TestAPassOpensWithTheVerdictAndSaysTheCommitsTheCountsTheTimesAndThatTheCompositionWasNotRun(t *testing.T) { // 1300/AC1 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + full := Report(passVerdict(), cs, p, nil, nil) + for _, want := range []string{"pass 111111111111 on aaaaaaaaaaaa", "unit 12 run, 301 kept", "integration 3 run, 40 kept", "queue 800ms", "clone 4.1s", "unit 400ms", "int setup 3.2s", "int run 12s", "composition not yet run"} { + if !strings.Contains(full, want) { + t.Errorf("the report %q lacks %q", full, want) + } + } + if !strings.HasPrefix(full, "pass ") || strings.Contains(full, "composition passed") { + t.Errorf("report = %q", full) + } + line := Line(passVerdict(), cs, p, nil, nil) + if n := utf8.RuneCountInString(line); n > MaxLine { + t.Errorf("the line is %d characters, want at most %d: %q", n, MaxLine, line) + } + if !strings.HasPrefix(line, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(line, "composition not yet run") || !strings.Contains(line, "unit 12 run, 301 kept") { + t.Errorf("line = %q", line) + } +} + +func TestAFailureNamesTheTestTheConflictOrTheRefusedStepFirst(t *testing.T) { // 1300/AC2 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + for name, c := range map[string]struct { + v delivery.Verdict + want string + }{ + "test": {delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Test: "unit: m/a.TestX", Words: "want 4"}}, "failed: unit: m/a.TestX: want 4"}, + "conflict": {delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "shanks", Words: "novox/mesh-delivery#7 does not merge into the trunk: files in conflict: go.mod"}}, "failed: shanks: novox/mesh-delivery#7 does not merge into the trunk: files in conflict: go.mod"}, + "refused": {delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "mesh-build-go", Words: "no go build node qualifies: builders disagree"}}, "failed: mesh-build-go: no go build node qualifies"}, + } { + line := Line(c.v, cs, p, nil, nil) + if !strings.HasPrefix(line, c.want) || !strings.Contains(line, "111111111111 on aaaaaaaaaaaa") { + t.Errorf("%s: line = %q, want it to open with %q and name the commits", name, line, c.want) + } + } +} + +func TestALevelOverItsBoundIsNamedWithItsSlowestTestsAndTheVerdictIsNotChanged(t *testing.T) { // 1300/AC3 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + v := passVerdict() + v.OverBound = []string{"unit"} + line := Line(v, cs, p, nil, []string{"unit pkg.TestSlow 7.0s"}) + if !strings.HasPrefix(line, "pass ") || !strings.Contains(Report(v, cs, p, nil, []string{"unit pkg.TestSlow 7.0s"}), "over bound: unit (slowest unit pkg.TestSlow 7.0s)") { + t.Errorf("line = %q", line) + } +} + +func TestSkippedTestsAreSaidAndNeverAFlatPass(t *testing.T) { // 1300/AC6 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + v := passVerdict() + v.Levels[delivery.LevelUnit] = delivery.LevelResult{Run: 12, Kept: 301, Skipped: 2} + if line := Line(v, cs, p, nil, nil); !strings.Contains(line, "unit 12 run, 301 kept, 2 skipped") { + t.Errorf("line = %q", line) + } +} + +func TestAGroupsMembersEachCarryTheSameVerdict(t *testing.T) { // 1300/AC4 + cs := caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2)) + a := Line(passVerdict(), cs, cs.Pulls[0], nil, nil) + b := Line(passVerdict(), cs, cs.Pulls[1], nil, nil) + if !strings.HasPrefix(a, "pass ") || !strings.HasPrefix(b, "pass ") || !strings.Contains(Report(passVerdict(), cs, cs.Pulls[0], nil, nil), "group feat/x of 2 pull requests") { + t.Errorf("lines %q and %q", a, b) + } + if !strings.Contains(a, "111111111111") || !strings.Contains(b, "222222222222") { + t.Errorf("each member's line names its own head: %q and %q", a, b) + } +} + +func TestTheBuildsOfTheTouchedModulesAreSaidAndDroppedFirstWhenTheLineIsTooLong(t *testing.T) { // ADR 0484 clause 7 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + builds := []BuildNote{{"avahi", "built"}, {"mesh-tools", "waiting"}} + if r := Report(passVerdict(), cs, p, builds, nil); !strings.Contains(r, "builds: avahi built, mesh-tools waiting") { + t.Errorf("report = %q", r) + } + line := Line(passVerdict(), cs, p, builds, nil) + if utf8.RuneCountInString(line) > MaxLine { + t.Errorf("line %q is too long", line) + } + short := passVerdict() + short.Took = delivery.Took{} + if l := Line(short, cs, p, builds[:1], nil); !strings.Contains(l, "builds: avahi built") { + t.Errorf("a short line keeps its builds: %q", l) + } +} + +func TestALineIsNeverLongerThanTheLimit(t *testing.T) { // 1300/AC1 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + v := delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Test: "integration: " + strings.Repeat("pkg/", 50) + ".TestSomething", Words: strings.Repeat("because ", 40)}} + if line := Line(v, cs, p, nil, nil); utf8.RuneCountInString(line) > MaxLine || !strings.HasPrefix(line, "failed: integration: ") { + t.Errorf("line = %q (%d)", line, utf8.RuneCountInString(line)) + } +} + +func TestARunningCheckIsPending(t *testing.T) { + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + if line := Line(delivery.Verdict{State: delivery.VerdictRunning}, cs, p, nil, nil); line != "checking 111111111111 on aaaaaaaaaaaa" { + t.Errorf("line = %q", line) + } +} + +func TestThePassedCompositionIsSaidOnlyWhenItWasRun(t *testing.T) { // ADR 0484 clause 6 + cs, p := lineCase(t, pullOf(repoA, 7, head1)) + v := passVerdict() + v.Composition = &delivery.Composition{State: delivery.CompositionPass, SDK: "afbe371003a570f088be4a394d83aa7a9458f7ab"} + if r := Report(v, cs, p, nil, nil); !strings.Contains(r, "composition passed with afbe371003a5") { + t.Errorf("report = %q", r) + } + v.Composition = nil + if r := Report(v, cs, p, nil, nil); !strings.Contains(r, "composition not yet run") || strings.Contains(r, "composition passed") { + t.Errorf("a verdict with no composition: %q", r) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/preview.go b/modules/mesh-delivery-checker/internal/check/preview.go index e44a9d5..a037246 100644 --- a/modules/mesh-delivery-checker/internal/check/preview.go +++ b/modules/mesh-delivery-checker/internal/check/preview.go @@ -63,7 +63,7 @@ func (j *Examiner) writePreview(ctx context.Context, p Pull, fetched []Fetched) return } if err == nil && j.Announce != nil { - j.Announce(ctx, key) + j.Announce(ctx, EventPreviewWritten, key) } } } diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 1517d59..08d68ba 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -111,8 +111,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 if m.SecretsOwner != "mesh-delivery-checker" { t.Errorf("secrets-owner = %q", m.SecretsOwner) } - // The checker says each time it wrote a preview, so the indexer's watch finds the bucket (ADR 0484 clause 3). - if want := []stream{{Name: "previews", Events: []string{"preview.**"}}}; !reflect.DeepEqual(m.Streams, want) { + // The checker says each time it wrote a preview or a status, so the indexer's and the forge bridge's watches find the buckets (ADR 0484 clauses 2 and 3). + if want := []stream{{Name: "checks", Events: []string{"preview.**", "owed.**"}}}; !reflect.DeepEqual(m.Streams, want) { t.Errorf("streams = %+v, want %+v", m.Streams, want) } for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats", "bus"} { diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index eef0cc1..bb08186 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -7,7 +7,7 @@ "mesh-build-node.toolchain.reported" ], "streams": [ - { "name": "previews", "events": ["preview.**"] } + { "name": "checks", "events": ["preview.**", "owed.**"] } ], "state": [ { "name": "kept", "history": 1 }, diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go index a98b7cd..410ed63 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate.go @@ -91,9 +91,34 @@ func (p *performer) refuse(ctx context.Context, doneKey string, rev uint64, a br return true } -// owed performs the act an owed key asks at revision rev, once. +// owed performs the act an owed key of the gate asks at revision rev, once. func (p *performer) owed(ctx context.Context, key string, rev uint64, value []byte) { - doneKey := "owed." + key + p.perform(ctx, "owed."+key, key, rev, value) +} + +// checkOwed performs the status the checker owes on a head (mesh-delivery-checker_owed; novox/hq ADR 0484 clause 2) at +// revision rev, once. Its keys are kept apart from the gate's, and a key that is not the checker's one act, the status +// mesh/check, is left. +func (p *performer) checkOwed(ctx context.Context, key string, rev uint64, value []byte) { + doneKey := "check." + key + if p.seen(ctx, doneKey, rev) { + return + } + if _, _, _, err := delivery.ParseCheckerOwedKey(key); err != nil { + p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err) + p.keep(ctx, doneKey, rev) + return + } + if _, err := delivery.DecodeCheckerOwed(value); err != nil { + p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err) + p.keep(ctx, doneKey, rev) + return + } + p.perform(ctx, doneKey, key, rev, value) +} + +// perform does the act of an owed key once for the revision, under the bookkeeping key doneKey. +func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) { if p.seen(ctx, doneKey, rev) { return } diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go index 608685e..2a7b64b 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/gate_test.go @@ -208,3 +208,57 @@ func TestAMergeWithTwoHandedStopsAtTheFirstRefusal(t *testing.T) { // ADR 0441 t.Errorf("calls %d messages %d, want the first refusal to stop the pass", len(f.calls), len(pub.data)) } } + +// checkerOwed is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed). +func checkerOwed(t *testing.T, number int, state, text string) (string, []byte) { + t.Helper() + key, err := delivery.CheckerOwedKey("novox/mesh-delivery", number, headA) + if err != nil { + t.Fatal(err) + } + v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"}) + if err != nil { + t.Fatal(err) + } + return key, v +} + +func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // 1683/AC5 (ADR 0484 clause 2) + f := &forgeFake{} + p, _, done, _ := newPerformer(f) + key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb") + p.checkOwed(context.Background(), key, 3, v) + p.checkOwed(context.Background(), key, 3, v) + if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_commit_status" || + f.calls[0].body != `{"context":"mesh/check","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://view.example/checks/g.0123456789ab"}` { + t.Fatalf("calls = %+v", f.calls) + } + p.checkOwed(context.Background(), key, 4, v) + if len(f.calls) != 2 { + t.Errorf("a new revision was not performed: %+v", f.calls) + } + if _, ok := done["check."+key]; !ok { + t.Errorf("revision kept under %v, want check.%s", done, key) + } + // The gate's own owed bookkeeping is another key: the same owed key seen in the gate's bucket is its own act. + if _, ok := done["owed."+key]; ok { + t.Errorf("the checker's status was kept under the gate's key: %v", done) + } +} + +func TestAnythingButTheCheckersStatusInItsBucketIsLeft(t *testing.T) { // ADR 0484 clause 2: the checker owes one act + f := &forgeFake{} + p, _, done, logs := newPerformer(f) + // A comment, and a status of another context: the SDK refuses both to the checker's encoder, so they are written raw. + commentKey, commentValue := owedKV(t, "novox/mesh-delivery", 12, "comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"}) + gateKey, gateValue := owedKV(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate", + delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"}) + p.checkOwed(context.Background(), commentKey, 1, commentValue) + p.checkOwed(context.Background(), gateKey, 1, gateValue) + if len(f.calls) != 0 { + t.Errorf("the forge was called: %+v", f.calls) + } + if !strings.Contains(logs.String(), commentKey) || !strings.Contains(logs.String(), gateKey) || len(done) != 2 { + t.Errorf("log %q, kept %v: both keys are named and kept so they are not tried again", logs.String(), done) + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go index 3281f5c..679a8ae 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/integration_test.go @@ -30,6 +30,10 @@ import ( // The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the // bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342). // +// The checker's owed bucket is read like the gate's (reads) and the checker's stream checks is heard like EVENTS is (consumes +// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller +// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file. +// // The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal // with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls, // gitea.gitea_commit_trees and gitea.gitea_get_file, state heads, and the stream forge; `**` is written `>` on the wire (modstreams.go @@ -104,6 +108,14 @@ accounts { "mesh.mod.gitea.tool.gitea_add_comment" "mesh.mod.gitea.tool.gitea_add_comment.*" "mesh.stream.mesh-delivery-forge-bridge.event.forge.>" + "$JS.ACK.CHECKS.n1_mesh-delivery-forge-bridge.>" + "$JS.API.CONSUMER.INFO.CHECKS.n1_mesh-delivery-forge-bridge" + "$JS.API.CONSUMER.MSG.NEXT.CHECKS.n1_mesh-delivery-forge-bridge" + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-checker_owed.>" + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-checker_owed.>" + "$JS.API.DIRECT.GET.KV_mesh-delivery-checker_owed.>" + "$JS.API.STREAM.INFO.KV_mesh-delivery-checker_owed" + "$JS.FC.KV_mesh-delivery-checker_owed.>" ] } subscribe { allow: [ "_INBOX.n1.mesh-delivery-forge-bridge.>" @@ -205,6 +217,15 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context, if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil { t.Fatal(err) } + // The checker's stream checks and the bridge's durable consumer on its owed.written event; the checker's owed bucket is made + // only by the test that needs it, as when the checker is assigned after the bridge started. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, "CHECKS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy, + FilterSubjects: []string{"mesh.stream.mesh-delivery-checker.event.owed.written"}}); err != nil { + t.Fatal(err) + } // The bridge's state bucket, made by the controller from the manifest's state entry before the process starts. if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil { t.Fatal(err) @@ -706,3 +727,65 @@ func forgeMerged(t *testing.T, b *bed) []string { } func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] } + +// TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started; +// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker +// itself writes with) and says so on its stream; the bridge sets the status on the forge within 10 s, with the context +// mesh/check, the state and the line; the same revision is not set again, a new revision is (ADR 0484 clause 2). +func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2 + b := startBed(t, `{"result":{"pulls":[]}}`, nil) + calls := make(chan string, 8) + sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_status", func(m *nats.Msg) { + calls <- string(m.Data) + _ = m.Respond([]byte(`{"result":{},"node":"n0"}`)) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + if err := b.ctl.Flush(); err != nil { + t.Fatal(err) + } + kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.CheckerOwedBucket, History: delivery.CheckerOwedHistory}) + if err != nil { + t.Fatal(err) + } + key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, head) + if err != nil { + t.Fatal(err) + } + put := func(state, text string) { + v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"}) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(b.ctx, key, v); err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil { + t.Fatal(err) + } + } + next := func() string { + select { + case got := <-calls: + return got + case <-time.After(10 * time.Second): + t.Fatal("the status was not set within 10 s") + } + return "" + } + put("pending", "checking 0123456789ab on aaaaaaaaaaaa") + if got := next(); !strings.Contains(got, `"context":"mesh/check"`) || !strings.Contains(got, `"state":"pending"`) || !strings.Contains(got, `"sha":"`+head+`"`) || !strings.Contains(got, `"repo":"mesh-delivery"`) { + t.Fatalf("first call = %s", got) + } + put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept") + if got := next(); !strings.Contains(got, `"state":"success"`) || !strings.Contains(got, "unit 2 run, 3 kept") { + t.Fatalf("second call = %s", got) + } + select { + case extra := <-calls: + t.Errorf("a third call was made: %s", extra) + case <-time.After(500 * time.Millisecond): + } +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go new file mode 100644 index 0000000..c25f87a --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/late.go @@ -0,0 +1,82 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// checksStream is the stream the checker carries its events on, as the bus names it (declared by the checker as the stream +// checks, heard by this module through consumes; novox/hq ADR 0484 clause 2). +const checksStream = "CHECKS" + +// owedWrittenSubject is the subject of the checker's owed.written event. +const owedWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.owed.written" + +// lateWatch watches a bucket that may be made after the bridge started: the checker's owed bucket. The bridge holds no timer +// (ADR 0437 §8), so it watches the bucket where it exists at start, and otherwise when the checker says it wrote a status +// (owed.written, handled by Written). +type lateWatch struct { + ctx context.Context + js interface { + KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) + } + bucket string + react func(ctx context.Context, key string, rev uint64, value []byte) + logger *log.Logger + errs chan error + mu sync.Mutex + watched bool +} + +// ensure starts the watch if the bucket exists and is not watched. +func (w *lateWatch) ensure() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.watched { + return nil + } + kv, err := w.js.KeyValue(w.ctx, w.bucket) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a status", w.bucket) + return nil + case err != nil: + return fmt.Errorf("the bucket %s: %w", w.bucket, err) + } + w.watched = true + go func() { + w.errs <- watch(w.ctx, kv, func(key string, rev uint64, value []byte) { + hctx, cancel := context.WithTimeout(w.ctx, syncDeadline) + defer cancel() + w.react(hctx, key, rev, value) + }) + }() + return nil +} + +// Written takes one owed.written event of the checker in: its bucket is watched if it is not. A subject that is not the +// checker's is terminated; a failure stays unacknowledged, and the server delivers it again. +func (w *lateWatch) Written(m interface { + Subject() string + Ack() error + Term() error +}) { + if !strings.HasPrefix(m.Subject(), owedWrittenSubject) { + w.logger.Printf("%s: not the checker's owed.written, dropped", m.Subject()) + _ = m.Term() + return + } + if err := w.ensure(); err != nil { + w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) + return + } + _ = m.Ack() +} diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index 0438f1c..fc1aaff 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -202,7 +202,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the bucket %s: %w", performedBucket, err) } perf := &performer{call: call, pub: rec, done: kvRevisions{performedKV}, logger: logger} - watching := make(chan error, 2) + watching := make(chan error, 3) for _, w := range []struct { bucket string react func(ctx context.Context, key string, rev uint64, value []byte) @@ -225,6 +225,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro }) }() } + // The checker's statuses (mesh/check, unrequired until release R4f): its bucket is made when the checker is assigned, which + // may be after the bridge started, and the bridge holds no timer; the checker says on its stream each time it wrote a + // status, and the bucket is then watched (late.go). + late := &lateWatch{ctx: ctx, js: js, bucket: delivery.CheckerOwedBucket, react: perf.checkOwed, logger: logger, errs: watching} + if err := late.ensure(); err != nil { + return err + } + switch ccons, err := js.Consumer(ctx, checksStream, cred.Node+"_"+cred.Module); { + case err != nil: + logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's statuses are not heard until the bridge starts again: %v", cred.Node, cred.Module, checksStream, err) + default: + cc, err := ccons.Consume(func(m jetstream.Msg) { late.Written(m) }, + jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) + if err != nil { + return err + } + defer cc.Stop() + } if err := said(ctx); err != nil { return err } diff --git a/modules/mesh-delivery-forge-bridge/manifest_test.go b/modules/mesh-delivery-forge-bridge/manifest_test.go index d6682cc..40fb8a8 100644 --- a/modules/mesh-delivery-forge-bridge/manifest_test.go +++ b/modules/mesh-delivery-forge-bridge/manifest_test.go @@ -86,10 +86,10 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1 if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) { t.Errorf("streams = %+v", m.Streams) } - if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated", "mesh-delivery-checker.owed.written"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } - if want := []string{"mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { + if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view", diff --git a/modules/mesh-delivery-forge-bridge/module.json b/modules/mesh-delivery-forge-bridge/module.json index 25b68be..d411e5f 100644 --- a/modules/mesh-delivery-forge-bridge/module.json +++ b/modules/mesh-delivery-forge-bridge/module.json @@ -5,7 +5,8 @@ "consumes": [ "gitea.pull.updated", "gitea.pull.merged", - "gitea.pull.closed" + "gitea.pull.closed", + "mesh-delivery-checker.owed.written" ], "streams": [ { "name": "forge", "events": ["forge.**"] } @@ -15,6 +16,7 @@ { "name": "performed", "history": 1 } ], "reads": [ + "mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed" ], diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go index fcf6a31..c42221c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/previews.go @@ -15,9 +15,9 @@ import ( "mesh-delivery-indexer/internal/index" ) -// PreviewsStream is the stream the checker carries its preview.written event on, as the bus names it (declared by the -// checker as the stream previews, heard by this module through consumes; novox/hq ADR 0484 clause 3). -const PreviewsStream = "PREVIEWS" +// PreviewsStream is the stream the checker carries its events on, as the bus names it (declared by the +// checker as the stream checks, heard by this module through consumes; novox/hq ADR 0484 clause 3). +const PreviewsStream = "CHECKS" // previewWrittenSubject is the subject of the checker's preview.written event. const previewWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.preview.written"