diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go index 951b0a6..5770315 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -2,44 +2,98 @@ package main import ( "context" - "errors" - "fmt" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" ) -// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key: no poll and no -// timer, the check's own bound ends the wait. -type compositionWatch struct{ kv jetstream.KeyValue } +// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key and the +// assigner's presence: no poll and no timer. With no assigner online it answers check.ErrAssignerNotRunning at once. +type compositionWatch struct { + kv jetstream.KeyValue + // presence is the watchman's presence bucket; nil (not there) counts as no assigner online. + presence jetstream.KeyValue +} -// Await answers the composition under key once the assigner wrote one that does not wait; a value it cannot read is an -// error, and ctx ending is ctx's error. +// Await answers the composition under key once the assigner wrote one that does not wait (check.AwaitComposition). func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { - w, err := c.kv.Watch(ctx, key) + ctx, cancel := context.WithCancel(ctx) + defer cancel() + comps := make(chan check.CompositionEvent) + presence := make(chan check.PresenceEvent) + cw, err := c.kv.Watch(ctx, key) if err != nil { return delivery.CompositionVerdict{}, err } - defer w.Stop() - for { - select { - case <-ctx.Done(): - return delivery.CompositionVerdict{}, ctx.Err() - case e, ok := <-w.Updates(): - if !ok { - return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + defer cw.Stop() + go func() { + for e := range cw.Updates() { + ev := check.CompositionEvent{Settled: e == nil} + if e != nil && e.Operation() == jetstream.KeyValuePut { + ev.Value = e.Value() } + select { + case comps <- ev: + case <-ctx.Done(): + return + } + } + }() + if c.presence == nil { + go func() { + select { + case presence <- check.PresenceEvent{Settled: true}: + case <-ctx.Done(): + } + }() + } else { + pw, err := c.presence.Watch(ctx, "part."+delivery.AssignerModule+".*") + if err != nil { + return delivery.CompositionVerdict{}, err + } + defer pw.Stop() + go func() { + for e := range pw.Updates() { + ev := check.PresenceEvent{Settled: e == nil} + if e != nil { + ev.Key = e.Key() + if e.Operation() == jetstream.KeyValuePut { + if p, err := delivery.DecodePresence(e.Value()); err == nil { + ev.Online = p.Online + } + } + } + select { + case presence <- ev: + case <-ctx.Done(): + return + } + } + }() + } + return check.AwaitComposition(ctx, presence, comps) +} + +// followCompositions hands the checker each composition the assigner writes from now on, so a verdict that said its +// composition was not judged takes it in when the assigner starts (event-driven). +func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Checker) error { + w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + return err + } + go func() { + defer w.Stop() + for e := range w.Updates() { if e == nil || e.Operation() != jetstream.KeyValuePut { continue } - v, err := delivery.DecodeCompositionVerdict(e.Value()) - if err != nil { - return delivery.CompositionVerdict{}, fmt.Errorf("the composition %s cannot be read: %w", key, err) - } - if v.State != delivery.CompositionWaits { - return v, nil + if comp, err := delivery.DecodeCompositionVerdict(e.Value()); err == nil { + ch.Composed(ctx, comp) } } - } + }() + return nil } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go index 070abae..0325a35 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -4,6 +4,7 @@ package main import ( "context" + "errors" "os" "path/filepath" "testing" @@ -14,6 +15,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" ) // 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is @@ -45,10 +48,31 @@ func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { if err != nil { t.Fatal(err) } + pkv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PresenceBucket, History: 5}) + if err != nil { + t.Fatal(err) + } + pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox") + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + if _, err := pkv.Put(ctx, pkey, on); err != nil { + t.Fatal(err) + } + // With no assigner online the wait answers at once (a second key, before the assigner of this test is online). + off, _ := delivery.EncodePresence(delivery.Presence{Online: false}) + if _, err := pkv.Put(ctx, pkey, off); err != nil { + t.Fatal(err) + } + other, _ := delivery.CompositionKey("feat/y", "0123456789ab") + if _, err := (compositionWatch{kv: kv, presence: pkv}).Await(ctx, other); !errors.Is(err, check.ErrAssignerNotRunning) { + t.Fatalf("with no assigner online the wait answered %v", err) + } + if _, err := pkv.Put(ctx, pkey, on); err != nil { + t.Fatal(err) + } key, _ := delivery.CompositionKey("feat/x", "0123456789ab") got := make(chan delivery.CompositionVerdict, 1) go func() { - v, err := compositionWatch{kv}.Await(ctx, key) + v, err := compositionWatch{kv: kv, presence: pkv}.Await(ctx, key) if err != nil { t.Error(err) } 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 676d380..08b3c0b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -231,7 +231,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro case err != nil: return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) default: - compositions = compositionWatch{kv} + cw := compositionWatch{kv: kv} + if pkv, err := js.KeyValue(ctx, delivery.PresenceBucket); err == nil { + cw.presence = pkv + } + compositions = cw + if err := followCompositions(ctx, kv, ch); err != nil { + return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) + } } // The tries (ADR 0501 ยง5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 49477b2..53b499f 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/internal/check/awaiting.go b/modules/mesh-delivery-checker/internal/check/awaiting.go new file mode 100644 index 0000000..1fbae2c --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting.go @@ -0,0 +1,141 @@ +package check + +import ( + "context" + "errors" + "fmt" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ErrAssignerNotRunning is what a wait for a composition answers when no instance of the assigner is online: nobody is +// making it, so the check does not wait for it (issue 1297). The verdict says the composition is not judged, and when the +// assigner starts and writes it, the checker takes it in then (Checker.Composed). +var ErrAssignerNotRunning = errors.New(delivery.AssignerNotRunning) + +// PresenceEvent is a change of the watchman's presence bucket, or Settled once what stands was handed. +type PresenceEvent struct { + Key string + Online bool + Settled bool +} + +// CompositionEvent is a change of the composition's key (its value, nil for a delete), or Settled once what stands was +// handed. +type CompositionEvent struct { + Value []byte + Settled bool +} + +// AwaitComposition answers the assigner's composition once one is written that does not wait, while an instance of the +// assigner is online by the watchman's presence (`part.mesh-delivery-assigner.`). Once both watches handed +// what stands, no instance online is ErrAssignerNotRunning at once, and so is the last instance going offline during the +// wait. No timer: every answer follows an event; ctx ending is ctx's error. +func AwaitComposition(ctx context.Context, presence <-chan PresenceEvent, comps <-chan CompositionEvent) (delivery.CompositionVerdict, error) { + online := map[string]bool{} + presenceSettled, compSettled := false, false + for { + select { + case <-ctx.Done(): + return delivery.CompositionVerdict{}, ctx.Err() + case e, ok := <-presence: + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the assigner's presence ended") + } + if e.Settled { + presenceSettled = true + break + } + if _, part, instance, err := delivery.ParsePresenceKey(e.Key); err == nil && part == delivery.AssignerModule { + if e.Online { + online[instance] = true + } else { + delete(online, instance) + } + } + case e, ok := <-comps: + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + } + if e.Settled { + compSettled = true + break + } + if e.Value == nil { + break + } + v, err := delivery.DecodeCompositionVerdict(e.Value) + if err != nil { + return delivery.CompositionVerdict{}, fmt.Errorf("the composition cannot be read: %w", err) + } + if v.State != delivery.CompositionWaits { + return v, nil + } + } + if presenceSettled && compSettled && len(online) == 0 { + return delivery.CompositionVerdict{}, ErrAssignerNotRunning + } + } +} + +// withComposition is v with the assigner's composition taken in: a failed one fails the check, naming its machine. +func withComposition(v *delivery.Verdict, comp delivery.CompositionVerdict) { + v.Composition = comp.Composition() + v.Took.Composition = comp.Took + if comp.State == delivery.CompositionFail { + machine := comp.Machine + if machine == "" { + machine = "the composition" + } + v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words} + } +} + +// notJudged says whether a verdict's composition was not judged because the assigner was not running. +func notJudged(v delivery.Verdict) bool { + return v.Composition != nil && v.Composition.State == delivery.CompositionNotRun && v.Composition.Words == delivery.AssignerNotRunning +} + +// Composed takes in a composition the assigner wrote after the check's verdict said it was not judged because the +// assigner was not running (the assigner started since): the verdict and its status are written again with it. A +// composition of heads no candidate holds now, or of a verdict that judged its composition, is left. +func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdict) { + ch.init() + if comp.State == delivery.CompositionWaits { + return + } + ckey, err := delivery.CandidatesKey(comp.Group) + if err != nil { + return + } + ch.mu.Lock() + c, ok := ch.known[ckey] + ch.mu.Unlock() + if !ok { + return + } + cs := CaseOf(c, ch.heads()) + if cs == nil || cs.HeadsHash != comp.Heads { + return + } + cur, rev, found, err := ch.Verdicts.Get(ctx, cs.Key) + if err != nil || !found { + return + } + v, err := delivery.DecodeVerdict(cur) + if err != nil || v.State != delivery.VerdictPass || !notJudged(v) { + return + } + withComposition(&v, comp) + data, err := delivery.EncodeVerdict(v) + if err != nil { + ch.logf("%s: the verdict with its composition cannot be written: %v", cs.Key, err) + return + } + if _, err := ch.Verdicts.Update(ctx, cs.Key, data, rev); err != nil { + ch.logf("%s: the verdict with its composition could not be written: %v", cs.Key, err) + return + } + ch.writeOwed(ctx, cs, v, Outcome{Verdict: v}) + ch.logf("%s: the composition came after the verdict: %s", cs.Key, comp.State) +} diff --git a/modules/mesh-delivery-checker/internal/check/awaiting_test.go b/modules/mesh-delivery-checker/internal/check/awaiting_test.go new file mode 100644 index 0000000..c775c49 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting_test.go @@ -0,0 +1,149 @@ +package check + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func assignerOn(instance string, online bool) PresenceEvent { + key, _ := delivery.PresencePartKey(delivery.AssignerModule, instance) + return PresenceEvent{Key: key, Online: online} +} + +func compEvent(t *testing.T, v delivery.CompositionVerdict) CompositionEvent { + t.Helper() + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + return CompositionEvent{Value: data} +} + +var passed = delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}} + +// 1297/B3d: with no assigner online, the wait answers ErrAssignerNotRunning as soon as both watches handed what stands. +func TestNoAssignerAnswersAtOnce(t *testing.T) { + presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence <- assignerOn("novox", false) + presence <- PresenceEvent{Settled: true} + comps <- CompositionEvent{Settled: true} + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := AwaitComposition(ctx, presence, comps); !errors.Is(err, ErrAssignerNotRunning) { + t.Fatalf("answered %v, want the assigner not running", err) + } +} + +// 1297/B3d: with the assigner online it waits for the key, past a composition that waits, and answers the verdict. +func TestAnOnlineAssignerIsWaitedFor(t *testing.T) { + presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence <- assignerOn("novox", true) + presence <- PresenceEvent{Settled: true} + comps <- CompositionEvent{Settled: true} + done := make(chan error, 1) + var got delivery.CompositionVerdict + go func() { + var err error + got, err = AwaitComposition(context.Background(), presence, comps) + done <- err + }() + comps <- compEvent(t, delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "w"}) + select { + case err := <-done: + t.Fatalf("answered before the composition was written: %v", err) + case <-time.After(50 * time.Millisecond): + } + comps <- compEvent(t, passed) + if err := <-done; err != nil || got.State != delivery.CompositionPass { + t.Fatalf("answered %+v, %v", got, err) + } + // And the last instance going offline during a wait ends it as not running. + presence2, comps2 := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) + presence2 <- assignerOn("novox", true) + presence2 <- PresenceEvent{Settled: true} + comps2 <- CompositionEvent{Settled: true} + presence2 <- assignerOn("novox", false) + if _, err := AwaitComposition(context.Background(), presence2, comps2); !errors.Is(err, ErrAssignerNotRunning) { + t.Errorf("the assigner going offline answered %v", err) + } +} + +// 1297/B3d: a passing check with no assigner says at once, loud, that its composition is not judged; never a pass of it. +func TestAPassingCheckWithNoAssignerSaysTheCompositionIsNotJudged(t *testing.T) { + b := newBench(t) + b.ex.Compositions = &fakeCompositions{err: ErrAssignerNotRunning} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || v.Composition.Words != delivery.AssignerNotRunning { + t.Fatalf("composition = %+v", v.Composition) + } + if words := compositionWords(v.Composition, 0); words != "composition not judged: the assigner is not running" { + t.Errorf("the line says %q", words) + } +} + +// 1297/B3d: the assigner starting writes the composition later, and the checker takes it into the verdict that said it +// was not judged, writing the status again; a failed one fails the check. +func TestTheAssignerStartingBringsTheComposition(t *testing.T) { + for _, comp := range []delivery.CompositionVerdict{passed, + {Group: "feat/x", Heads: "", State: delivery.CompositionFail, SDK: "x", Machine: "g14", Words: "g14 runs web, which the change removes"}} { + ch, cs := checkerWithNotJudged(t) + comp.Heads = cs.HeadsHash + ch.Composed(context.Background(), comp) + raw, _, _, _ := ch.Verdicts.Get(context.Background(), cs.Key) + v, err := delivery.DecodeVerdict(raw) + if err != nil || v.Composition == nil || v.Composition.State != comp.State { + t.Fatalf("after the composition came, the verdict is %+v, %v", v, err) + } + if comp.State == delivery.CompositionFail && (v.State != delivery.VerdictFail || v.Failed.Machine != "g14") { + t.Errorf("a failed composition left the verdict %+v", v) + } + owed := ch.Owed.(*memKV) + keys, _ := owed.Keys(context.Background()) + if len(keys) == 0 { + t.Fatal("no status was written again") + } + data, _, _, _ := owed.Get(context.Background(), keys[0]) + o, _ := delivery.DecodeCheckerOwed(data) + if !strings.Contains(o.Text, "composition passed") && !strings.Contains(o.Text, "g14 runs web") { + t.Errorf("the status's line says %q", o.Text) + } + } +} + +// checkerWithNotJudged is a checker holding a candidate whose passing verdict said its composition was not judged. +func checkerWithNotJudged(t *testing.T) (*Checker, *Case) { + t.Helper() + cs := caseOf(t, pullOf(repoA, 7, head1)) + raw := candidateOf(t, cs.Group, pullOf(repoA, 7, head1)) + ch := &Checker{Verdicts: newMemKV(), Owed: newMemKV(), Trunks: fixedTrunks{cs.Trunk}} + ckey, _ := delivery.CandidatesKey(cs.Group) + ch.init() + c, err := delivery.DecodeCandidate(raw) + if err != nil { + t.Fatal(err) + } + ch.known[ckey] = c + if got := CaseOf(c, ch.heads()); got == nil || got.Key != cs.Key { + t.Fatalf("the case is %+v, want %s", got, cs.Key) + } + v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, State: delivery.VerdictPass, + Levels: map[string]delivery.LevelResult{delivery.LevelUnit: {Run: 1}, delivery.LevelIntegration: {Run: 1}}, + Composition: &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning}} + data, err := delivery.EncodeVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := ch.Verdicts.Create(context.Background(), cs.Key, data); err != nil { + t.Fatal(err) + } + return ch, cs +} + +type fixedTrunks struct{ heads map[string]string } + +func (f fixedTrunks) Heads() map[string]string { return f.heads } diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 099caca..d7e420c 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -3,6 +3,7 @@ package check import ( "context" "encoding/json" + "errors" "fmt" "path" "slices" @@ -253,25 +254,21 @@ func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { // composed reads the assigner's composition of the check into its verdict (ADR 0528 clause 5): a failed composition // fails the check, naming the machine it failed at. A try is not composed (it has no candidate); a composition that -// cannot be had leaves it not run, said. +// cannot be had leaves it not run, said; with no assigner running it is not judged, said at once. func (st *state) composed(ctx context.Context, v *delivery.Verdict) { if st.j.Compositions == nil || st.c.Try != nil { return } comp, err := st.j.Compositions.Await(ctx, st.c.Key) - if err != nil { + switch { + case errors.Is(err, ErrAssignerNotRunning): + v.Composition = &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning} + return + case err != nil: st.j.logf("%s: the composition could not be read, so it is said as not run: %v", st.c.Key, err) return } - v.Composition = comp.Composition() - v.Took.Composition = comp.Took - if comp.State == delivery.CompositionFail { - machine := comp.Machine - if machine == "" { - machine = "the composition" - } - v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words} - } + withComposition(v, comp) } // outcome fills the verdict's levels, times, machines and composition from what ran. diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go index 85e6830..1030f52 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -99,6 +99,9 @@ func compositionWords(c *delivery.Composition, took int64) string { return compo // composition is compositionWords; short counts the machines and leaves the SDK commit to the report. func composition(c *delivery.Composition, took int64, short bool) string { + if c != nil && c.State == delivery.CompositionNotRun && c.Words != "" { + return "composition not judged: " + c.Words + } if c == nil || c.State == delivery.CompositionNotRun { return "composition not yet run" }