From 7cbb1c4207456a9adf2f36ce540f7fbbb1c8d9cf Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:44:00 +0200 Subject: [PATCH] Wait for the assigner's composition only while the assigner is online, say at once that it is not judged when it is not, and take it in when the assigner writes it later, so no check waits its bound for a composition nobody makes (issue 1297, B3d) --- .../cmd/mesh-delivery-checker/compositions.go | 100 +++++++++--- .../compositions_integration_test.go | 26 ++- .../cmd/mesh-delivery-checker/main.go | 9 +- modules/mesh-delivery-checker/go.mod | 2 +- .../internal/check/awaiting.go | 141 +++++++++++++++++ .../internal/check/awaiting_test.go | 149 ++++++++++++++++++ .../internal/check/judge.go | 19 +-- .../internal/check/line.go | 3 + 8 files changed, 412 insertions(+), 37 deletions(-) create mode 100644 modules/mesh-delivery-checker/internal/check/awaiting.go create mode 100644 modules/mesh-delivery-checker/internal/check/awaiting_test.go 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" }