From 77af82cfa2fc537a0c60bf95e75823e14432ade3 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:11:07 +0200 Subject: [PATCH 1/3] Keep the asks state and write the ask about a held fingerprint, so the view's and the teller's read of mesh-delivery-assigner.asks is granted and has something to answer (issue 1297, review round 1 item 1; ADR 0528 clause 6) --- .../integration_test.go | 32 +++++++++++++++++-- .../cmd/mesh-delivery-assigner/main.go | 7 ++-- .../cmd/mesh-delivery-assigner/pass.go | 15 +++++++++ .../internal/assign/asks.go | 29 +++++++++++++++++ .../internal/assign/asks_test.go | 28 ++++++++++++++++ .../mesh-delivery-assigner/manifest_test.go | 2 +- modules/mesh-delivery-assigner/module.json | 1 + 7 files changed, 108 insertions(+), 6 deletions(-) create mode 100644 modules/mesh-delivery-assigner/internal/assign/asks.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/asks_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 2682b37..6a79f7c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -99,7 +99,7 @@ func (b *bed) start(t *testing.T) <-chan error { return done } -// 1297/B2b: started whole on its credential, the assigner says on the bus the three states it keeps, each readable, +// 1297/B2b: started whole on its credential, the assigner says on the bus the four states it keeps, each readable, // having opened every bucket it names. func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) { b := startBed(t, "") @@ -124,9 +124,9 @@ func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) { } break } - if said.Module != "mesh-delivery-assigner" || said.Node != "n1" || len(said.States) != 3 || + if said.Module != "mesh-delivery-assigner" || said.Node != "n1" || len(said.States) != 4 || !said.Readable("asks") || !said.Readable("wants") || !said.Readable("differences") || !said.Readable("compositions") { - t.Fatalf("the assigner said %+v, want its three states, each readable", said) + t.Fatalf("the assigner said %+v, want its four states, each readable", said) } } @@ -274,3 +274,29 @@ func TestWantsAndTheirDifferencesOnTheBus(t *testing.T) { t.Errorf("g14, held back by the canary with no want before, was given one: %s", e.Value()) } } + +// 1297/B9 (the assigner's half, ADR 0528 clause 6): started whole, the assigner holds a fingerprint whose health check +// failed on a node and writes the ask about it in its own asks, which the view and the teller read; its asks bucket is +// the one its manifest declares (ownBuckets, held equal to the manifest by TestOpensWhatItsManifestNames). +func TestAHeldFingerprintIsAskedAbout(t *testing.T) { + b := startBed(t, "") + fNew := delivery.SourcePrefix2 + strings.Repeat("c", 64) + digest := b.manifest(t, "web", `{"module":"web","version":"1"}`) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + b.putJSON(t, delivery.ModulesBucket, "web", mod) + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + b.putJSON(t, delivery.PresenceBucket, "ace", on) + failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}) + b.putJSON(t, delivery.HaveBucket, "ace.web", failed) + b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-1"}`)) + b.start(t) + hkey, _ := delivery.HeldKey("web", fNew) + if h, err := delivery.DecodeHeld(b.get(t, delivery.AssignerWantsBucket, hkey)); err != nil || h.Node != "ace" { + t.Fatalf("held = %+v, %v", h, err) + } + akey, _ := delivery.HoldAskKey("web", fNew) + a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey)) + if err != nil || a.Ask.ID != delivery.AskingID(akey, 0) || len(a.Ask.Options) != 2 { + t.Fatalf("the ask = %+v, %v", a, err) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index dc34b09..8900bd8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -50,11 +50,13 @@ var readBuckets = []string{ } // ownBuckets are the assigner's own states, as its manifest declares them. -var ownBuckets = []string{delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket} +var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket} // keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view -// (and the node agent from release 6), its differences by the view, its compositions by the checker. +// (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view +// and the teller. var keptStates = []announce.State{ + {Name: "asks", Readable: true}, {Name: "compositions", Readable: true}, {Name: "differences", Readable: true}, {Name: "wants", Readable: true}, @@ -135,6 +137,7 @@ var watched = []string{ delivery.ModulesBucket, delivery.PresenceBucket, delivery.HaveBucket, + delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket, diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 754a812..a626f95 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -137,6 +137,9 @@ func (p *passer) wants(ctx context.Context) map[string]map[string]assign.Holding p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) }) } } + if out.Held != nil || m.Held { + p.holdAsk(ctx, name, fp) + } for node, want := range out.Wants { key, err := delivery.WantKey(node, name) if err != nil { @@ -205,3 +208,15 @@ func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]b } p.w.put(bucket, key, value) } + +// holdAsk writes, once, the ask about a held fingerprint (ADR 0528 clause 6, ADR 0437 clause 10): the first ask that needs +// no proof of who answered, which the view may answer and the teller fills. In release 5 an answer changes nothing on a +// node (wants are only compared, ADR 0429 §7), and the options say so. +func (p *passer) holdAsk(ctx context.Context, module, fp string) { + key, err := delivery.HoldAskKey(module, fp) + if err != nil || p.w.has(delivery.AssignerAsksBucket, key) { + return + } + a := assign.HoldAsk(key, module, fp, time.Now()) + p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) }) +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go new file mode 100644 index 0000000..52d7cb7 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -0,0 +1,29 @@ +package assign + +import ( + "fmt" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// HoldAsk is the ask about a module's fingerprint the assigner holds, at its key (delivery.HoldAskKey), first asking: keep +// it held or release it. Both options acknowledge and bind nothing: in release 5 the wants are only compared, so no answer +// changes what a node runs (ADR 0429 §7); the node agent acts on the hold from release 6. +func HoldAsk(key, module, fp string, now time.Time) delivery.AssignerAsk { + return delivery.AssignerAsk{Ask: asks.Ask{ + ID: delivery.AskingID(key, 0), + Headline: fmt.Sprintf("%s is held at %s", module, short(fp)), + Explanation: fmt.Sprintf("A node's health check of %s failed at %s, so the assigner holds that version and wants "+ + "no node to move to it. Until release 6 this changes nothing a node runs: the controller still sends.", module, short(fp)), + Options: []asks.Option{ + {ID: "keep", Label: "Keep it held", Does: "the version stays held", Level: asks.Acknowledge}, + {ID: "release", Label: "Release it", Does: "the version is released when wants are applied, from release 6", Level: asks.Acknowledge}, + }, + Who: asks.Operator, + Expires: now.Add(7 * 24 * time.Hour), + OnExpiry: "the version stays held", + About: key, + }} +} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go new file mode 100644 index 0000000..7de1348 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -0,0 +1,28 @@ +package assign + +import ( + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1297/B9 (the assigner's half): the ask about a held fingerprint is a valid ask of the router and of mesh-sdk's layout at +// its key, of a kind the view may answer. +func TestTheHoldAskIsValid(t *testing.T) { + key, err := delivery.HoldAskKey("mail", fNew) + if err != nil { + t.Fatal(err) + } + now := time.Now() + a := HoldAsk(key, "mail", fNew, now) + if err := a.Ask.Check(now); err != nil { + t.Errorf("the router refuses the ask: %v", err) + } + if _, err := delivery.EncodeAssignerAsk(key, a); err != nil { + t.Errorf("the ask cannot be written: %v", err) + } + if _, err := delivery.ViewAnswerKey(key); err != nil { + t.Errorf("the view may not answer it: %v", err) + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 417ac68..03b879a 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -42,7 +42,7 @@ func TestManifest(t *testing.T) { for _, s := range m.State { states[s.Name] = s.History } - if want := map[string]int{"compositions": 5, "differences": 64, "wants": 20}; !reflect.DeepEqual(states, want) { + if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "wants": 20}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 599c3f6..418e8fb 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -14,6 +14,7 @@ "node-engine.have" ], "state": [ + { "name": "asks", "history": 5 }, { "name": "compositions", "history": 5 }, { "name": "differences", "history": 64 }, { "name": "wants", "history": 20 } From d4977d9c9d673dc6c655f5f6af64aff40df4809c Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:11:31 +0200 Subject: [PATCH 2/3] Start the checker whole with the assigner's verdict on the bus and read its line, the level 1297/AC3 asks (issue 1297, review round 1 item 2) --- .../composition_line_integration_test.go | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go new file mode 100644 index 0000000..e41c96c --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go @@ -0,0 +1,63 @@ +//go:build integration + +package main + +import ( + "strings" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// assignerWrote raises the assigner's compositions bucket, the assigner online in the watchman's presence, and its verdict +// under the check's key, as the assigner leaves them on the bus. +func (b *bed) assignerWrote(t *testing.T, key string, v delivery.CompositionVerdict) { + t.Helper() + kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory}) + if err != nil { + t.Fatal(err) + } + pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox") + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + if _, err := b.buckets[delivery.PresenceBucket].Put(b.ctx, pkey, on); err != nil { + t.Fatal(err) + } + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(b.ctx, key, data); err != nil { + t.Fatal(err) + } +} + +// 1297/AC3: started whole with the assigner's verdict on the bus, the checker's line says the composition's verdict, not +// "not yet run": a pass names its machine, a fail fails the check with the assigner's words. +func TestTheCheckersLineSaysTheAssignersComposition(t *testing.T) { + for _, c := range []struct { + state, owed, words string + }{ + {delivery.CompositionPass, "success", "composition passed"}, + {delivery.CompositionFail, "failure", "g14 runs web, which the change removes from its source"}, + } { + b := startBed(t) + key := b.candidate(t, "feat/x", pull(repo, 7, head1)) + group, heads, _ := delivery.ParseCheckerVerdictKey(key) + v := delivery.CompositionVerdict{Group: group, Heads: heads, State: c.state, SDK: "v0.1.14-x", Machines: []string{"g14"}} + if c.state == delivery.CompositionFail { + v.Machine, v.Words = "g14", c.words + } + b.assignerWrote(t, key, v) + b.start(t, nil) + got := b.verdict(t, key, ended) + if got.Composition == nil || got.Composition.State != c.state { + t.Fatalf("%s: the verdict's composition is %+v", c.state, got.Composition) + } + o := b.owed(t, 7, head1, c.owed) + if !strings.Contains(o.Text, c.words) || strings.Contains(o.Text, "not yet run") { + t.Errorf("%s: the line says %q", c.state, o.Text) + } + } +} From d0354574c69a034a69a8f0ce6b2b5d00853968c1 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:12:18 +0200 Subject: [PATCH 3/3] Look the assigner's compositions bucket up at each check until it is there, so an assigner delivered after the checker started is read without a restart (issue 1297, review round 1 item 3) --- .../cmd/mesh-delivery-checker/compositions.go | 32 +++++++++++++ .../compositions_test.go | 46 +++++++++++++++++++ .../cmd/mesh-delivery-checker/main.go | 25 +++++----- 3 files changed, 90 insertions(+), 13 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_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 5770315..c5a3fe3 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -2,6 +2,8 @@ package main import ( "context" + "errors" + "sync" "github.com/nats-io/nats.go/jetstream" @@ -97,3 +99,33 @@ func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Ch }() return nil } + +// lazyCompositions opens the assigner's compositions when a check first asks for one and the bucket is there: until it +// is, no assigner ever ran, so the check says at once that its composition is not judged (check.ErrAssignerNotRunning), +// and the next check looks again. No timer: each look is a check asking. +type lazyCompositions struct { + open func(ctx context.Context) (check.CompositionReader, error) + + mu sync.Mutex + reader check.CompositionReader +} + +// Await is the opened reader's Await, opening it first when it is not yet. +func (l *lazyCompositions) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { + l.mu.Lock() + r := l.reader + if r == nil { + opened, err := l.open(ctx) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + l.mu.Unlock() + return delivery.CompositionVerdict{}, check.ErrAssignerNotRunning + case err != nil: + l.mu.Unlock() + return delivery.CompositionVerdict{}, err + } + l.reader, r = opened, opened + } + l.mu.Unlock() + return r.Await(ctx, key) +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go new file mode 100644 index 0000000..64afd0b --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "context" + "errors" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-checker/internal/check" +) + +type answering struct{ v delivery.CompositionVerdict } + +func (a answering) Await(context.Context, string) (delivery.CompositionVerdict, error) { + return a.v, nil +} + +// 1297 review item 3: the compositions bucket missing when the checker starts is looked up again at the next check: while +// it is missing a check says at once the assigner is not running; once it is there the check reads its composition, and +// it is opened once. +func TestTheCompositionsBucketIsLookedUpAtEachCheck(t *testing.T) { + opens, there := 0, false + l := &lazyCompositions{open: func(context.Context) (check.CompositionReader, error) { + opens++ + if !there { + return nil, jetstream.ErrBucketNotFound + } + return answering{delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "x"}}, nil + }} + if _, err := l.Await(context.Background(), "k"); !errors.Is(err, check.ErrAssignerNotRunning) { + t.Fatalf("without the bucket: %v", err) + } + there = true + for range 2 { + v, err := l.Await(context.Background(), "k") + if err != nil || v.State != delivery.CompositionPass { + t.Fatalf("with the bucket: %+v, %v", v, err) + } + } + if opens != 2 { + t.Errorf("opened %d times, want twice (missing, then once)", opens) + } +} 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 08b3c0b..e3f6669 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -222,24 +222,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro default: builds = buildReader{kvStore{kv}} } - // The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the - // composition was not run, until the checker starts again. - var compositions check.CompositionReader - switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); { - case errors.Is(err, jetstream.ErrBucketNotFound): - logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket) - case err != nil: - return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) - default: + // The assigner's compositions (ADR 0528 clause 5, issue 1297): their bucket is looked up at each check until it is + // there, so an assigner delivered after the checker started is read at the next check, and followed from then on. + compositions := &lazyCompositions{open: func(at context.Context) (check.CompositionReader, error) { + kv, err := js.KeyValue(at, delivery.AssignerCompositionsBucket) + if err != nil { + return nil, err + } cw := compositionWatch{kv: kv} - if pkv, err := js.KeyValue(ctx, delivery.PresenceBucket); err == nil { + if pkv, err := js.KeyValue(at, 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) + return nil, fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) } - } + logger.Printf("the assigner's compositions are read from %s", delivery.AssignerCompositionsBucket) + return cw, nil + }} // 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} holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger)