diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index e794ebc8..b2afc02c 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -81,6 +81,9 @@ type Try struct { Tests string `json:"tests,omitempty"` Level string `json:"level,omitempty"` State string `json:"state"` + // Told is true from the moment a whole try's start is said on the checker's stream until its end is said there: kept in the + // try's own state, so an end that is written and not said (the checker stopped between) is said when the checker starts. + Told bool `json:"told,omitempty"` // Said is the try's report in words: what failed first, the tests run, kept and skipped per level, the time of each part. Said string `json:"said,omitempty"` // Took is the time from its start to its end, as Go writes a duration. @@ -519,7 +522,7 @@ type Tryer struct { Say func(ctx context.Context, event string, body tryevent.Event) mu sync.Mutex - told map[string]bool // whole tries whose start was said and whose end was not + ending map[string]bool // tries whose end is being said now wg sync.WaitGroup inflight map[string]context.CancelFunc seen map[string]bool // looked at since this process started: a try left running is taken over the first time @@ -558,12 +561,8 @@ func (tr *Tryer) Put(ctx context.Context, key string, value []byte) { if stop := tr.inflight[key]; stop != nil { stop() } - said := tr.told[key] - delete(tr.told, key) tr.mu.Unlock() - if said { // ended by hand while it ran: the indexer drops its candidate - tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State}) - } + tr.sendEnd(ctx, key) // ended by hand while it ran, or its end was written and not said: the indexer drops its candidate return } if tr.inflight[key] != nil || (t.State == TryRunning && tr.seen[key]) { @@ -619,6 +618,7 @@ func (tr *Tryer) run(ctx context.Context, key string) { return } t.State, t.Started = TryRunning, tr.now().UTC() + t.Told = (&TryOptions{Level: t.Level, Run: t.Tests}).Whole() // its start is said below; its end is owed from now on data, err := encodeTry(t) if err != nil { return @@ -630,17 +630,11 @@ func (tr *Tryer) run(ctx context.Context, key string) { return } progress := &tryWriter{tr: tr, ctx: ctx, key: key, t: t, rev: rev} - if (&TryOptions{Level: t.Level, Run: t.Tests}).Whole() { + if t.Told { started := tryevent.Event{ID: key} for _, m := range t.MembersOf() { started.Members = append(started.Members, tryevent.Member{Repository: m.Repository, Head: m.Head, Trunk: m.Trunk}) } - tr.mu.Lock() - if tr.told == nil { - tr.told = map[string]bool{} - } - tr.told[key] = true - tr.mu.Unlock() tr.say(ctx, tryevent.Started, started) } cs := TryCase(key, t, progress.heard) @@ -653,13 +647,6 @@ func (tr *Tryer) run(ctx context.Context, key string) { return // left running: the checker takes it over when it starts; or ended by hand, which wrote its end } t, rev = progress.end() - tr.mu.Lock() - said := tr.told[key] - delete(tr.told, key) - tr.mu.Unlock() - if said { - defer tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key}) - } v := out.Verdict t.Ended = tr.now().UTC() t.Took = t.Ended.Sub(t.Started).Round(time.Second).String() @@ -707,6 +694,41 @@ func (tr *Tryer) run(ctx context.Context, key string) { return } tr.logf("%s: %s", key, t.State) + tr.sendEnd(ctx, key) +} + +// sendEnd says the end of an ended try whose start was said (Told) with the state it ended in, then clears Told in the try's +// own state; a checker that stopped between the two says it again at its start. Every end is said so: pass, fail, error and +// by hand. +func (tr *Tryer) sendEnd(ctx context.Context, key string) { + tr.mu.Lock() + if tr.ending[key] { + tr.mu.Unlock() + return + } + if tr.ending == nil { + tr.ending = map[string]bool{} + } + tr.ending[key] = true + tr.mu.Unlock() + defer func() { tr.mu.Lock(); delete(tr.ending, key); tr.mu.Unlock() }() + raw, rev, found, err := tr.Tries.Get(ctx, key) + if err != nil || !found { + return + } + t, err := DecodeTry(raw) + if err != nil || !t.Over() || !t.Told { + return + } + tr.say(ctx, tryevent.Ended, tryevent.Event{ID: key, State: t.State}) + t.Told = false + data, err := encodeTry(t) + if err == nil { + _, err = tr.Tries.Update(ctx, key, data, rev) + } + if err != nil && !errors.Is(err, ErrConflict) { + tr.logf("%s: its end was said and Told could not be cleared: %v", key, err) + } } // say publishes a try event; a failure is said in the log (the indexer finds the open tries again at its start). diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index 46133c57..9aabfd73 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "testing" "time" @@ -181,16 +182,19 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR written := &writtenTries{memKV: kv} var said []string tr := &Tryer{Tries: written, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { - said = append(said, fmt.Sprintf("%s %s %d", event, e.ID, len(e.Members))) + said = append(said, fmt.Sprintf("%s %s %d %s", event, e.ID, len(e.Members), e.State)) }} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) tr.Wait() // The indexer hears the start with the members at their commits, and the end (it writes and deletes the group's candidate). - if len(said) != 2 || said[0] != "try.started "+id+" 2" || said[1] != "try.ended "+id+" 0" { + if len(said) != 2 || said[0] != "try.started "+id+" 2 " || said[1] != "try.ended "+id+" 0 "+TryPass { t.Errorf("events = %v, want the start with 2 members and the end", said) } got := readTry(t, kv, id) + if got.Told { // the end was said: nothing is owed + t.Errorf("the ended try still owes its end: %+v", got) + } if open, err := OpenTries(memTries{kv}); err != nil || len(open) != 0 { t.Errorf("open tries after the end = %v, %v", open, err) } @@ -282,26 +286,24 @@ func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) { if got := readTry(t, kv, id); got.State != TryPass || !strings.Contains(got.Said, "only the tests TestBus|TestStore") { t.Errorf("try = %+v", got) } -} - -func TestATryOfOneLevelKeepsOnlyThatLevel(t *testing.T) { - ctx := context.Background() - kv := newMemKV() - id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now()) - b := newBench(t) - tr := &Tryer{Tries: kv, Judge: b.ex} - raw, _, _, _ := kv.Get(ctx, id) - tr.Put(ctx, id, raw) - tr.Wait() - keys, _ := b.kept.Keys(ctx) - if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 { - t.Fatalf("try = %+v, kept %v", got, keys) - } - for _, k := range keys { - if !strings.HasPrefix(k, delivery.LevelUnit+".") { - t.Errorf("a try of the unit level kept %s", k) + t.Run("a try of one level keeps only that level", func(t *testing.T) { + kv := newMemKV() + id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelUnit}, time.Now()) + b := newBench(t) + tr := &Tryer{Tries: kv, Judge: b.ex} + raw, _, _, _ := kv.Get(ctx, id) + tr.Put(ctx, id, raw) + tr.Wait() + keys, _ := b.kept.Keys(ctx) + if got := readTry(t, kv, id); got.State != TryPass || len(keys) == 0 { + t.Fatalf("try = %+v, kept %v", got, keys) } - } + for _, k := range keys { + if !strings.HasPrefix(k, delivery.LevelUnit+".") { + t.Errorf("a try of the unit level kept %s", k) + } + } + }) } func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) { @@ -340,10 +342,18 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) comps.v.State = delivery.CompositionPass } id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) - tr := &Tryer{Tries: kv, Judge: b.ex} + var ends []string + tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { + if event == tryevent.Ended { + ends = append(ends, e.State) + } + }} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) tr.Wait() + if len(ends) != 1 || ends[0] != row.state { // every end says its state: error as well as fail + t.Errorf("%s: ends said = %v, want once %s", row.name, ends, row.state) + } if got := readTry(t, kv, id); got.State != row.state || !strings.Contains(strings.Join(got.Failed, " | ")+got.Said, row.said) { t.Errorf("%s: try = %s, failed %v, said %q, want %s naming %q", row.name, got.State, got.Failed, got.Said, row.state, row.said) } @@ -353,22 +363,56 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) func TestATryLeftRunningIsTakenOver(t *testing.T) { ctx := context.Background() - kv := newMemKV() - id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) - // Left running by a checker that stopped. - raw, rev, _, _ := kv.Get(ctx, id) - cur, _ := DecodeTry(raw) - cur.State = TryRunning - data, _ := encodeTry(cur) - if _, err := kv.Update(ctx, id, data, rev); err != nil { - t.Fatal(err) - } - b := newBench(t) - wholeBench(b, delivery.Candidate{}) - tr := &Tryer{Tries: kv, Judge: b.ex} - tr.Reconsider(ctx) - tr.Wait() - if got := readTry(t, kv, id); got.State != TryPass { - t.Errorf("the try left running = %+v, want taken over and passed", got) + // Whether the start was said is in the try's own state (Told), so a checker that starts later says the end of a try it did + // not start, whatever way the try ended. + for _, row := range []struct { + name string + told bool + after func(t *testing.T, kv *memKV, id string) + ends []string + }{ + {name: "left running after its start was said", told: true, ends: []string{TryPass}}, + {name: "left running with no start said", told: false, ends: []string{TryPass}}, + {name: "ended by hand after the start was said", told: true, ends: []string{TryError}, after: func(t *testing.T, kv *memKV, id string) { + if _, err := EndTry(endable{kv}, id, "stuck", time.Now()); err != nil { + t.Fatal(err) + } + }}, + } { + t.Run(row.name, func(t *testing.T) { + kv := newMemKV() + id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now()) + // Left running by a checker that stopped. + raw, rev, _, _ := kv.Get(ctx, id) + cur, _ := DecodeTry(raw) + cur.State, cur.Told = TryRunning, row.told + data, _ := encodeTry(cur) + if _, err := kv.Update(ctx, id, data, rev); err != nil { + t.Fatal(err) + } + if row.after != nil { + row.after(t, kv, id) + } + b := newBench(t) + wholeBench(b, delivery.Candidate{}) + var ends []string + tr := &Tryer{Tries: kv, Judge: b.ex, Say: func(_ context.Context, event string, e tryevent.Event) { + if event == tryevent.Ended { + ends = append(ends, e.State) + } + }} + tr.Reconsider(ctx) + tr.Wait() + tr.Reconsider(ctx) // said once: the second look finds nothing owed + tr.Wait() + if row.after == nil { + if got := readTry(t, kv, id); got.State != TryPass { + t.Errorf("the try left running = %+v, want taken over and passed", got) + } + } + if !slices.Equal(ends, row.ends) || readTry(t, kv, id).Told { + t.Errorf("ends said = %v, want %v, and nothing owed after", ends, row.ends) + } + }) } } diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 276a1a89..1f71c257 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -134,6 +134,32 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 if want := []stream{{Name: "checks", Events: []string{"preview.**", "owed.**", "try.**"}}}; !reflect.DeepEqual(m.Streams, want) { t.Errorf("streams = %+v, want %+v", m.Streams, want) } + // Only the checker's account publishes try.* on its stream checks: the stream is declared by its owner alone, and no other + // module of the repository declares a stream carrying a try event (the indexer and the others only consume them). + siblings, err := filepath.Glob("../*/module.json") + if err != nil || len(siblings) < 2 { + t.Fatalf("the sibling manifests = %v, %v", siblings, err) + } + for _, f := range siblings { + raw, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + var other struct { + Module string `json:"module"` + Streams []stream `json:"streams"` + } + if err := json.Unmarshal(raw, &other); err != nil { + t.Fatalf("%s: %v", f, err) + } + for _, st := range other.Streams { + for _, e := range st.Events { + if strings.HasPrefix(e, "try.") && other.Module != m.Module { + t.Errorf("%s declares the stream %s with %s: only the checker's account publishes try events", other.Module, st.Name, e) + } + } + } + } // The read tools (issue 1686) are served by the tools bundle, not by the checker's process. // The try's tools (ADR 0501 ยง5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check // is asked again by hand (checker_ask_again).