Merge branch 'feat/r4g-assigner' into feat/r4g

2026-10-23 16:19:45 +02:00
parent 271d1ca759 d0354574c6
commit 0463f21026
11 changed files with 261 additions and 19 deletions
@@ -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)
+	}
+}
@@ -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,
@@ -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) })
+}
@@ -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,
+	}}
+}
@@ -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)
+	}
+}
@@ -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 {
@@ -14,6 +14,7 @@
    "node-engine.have"
  ],
  "state": [
+    { "name": "asks", "history": 5 },
    { "name": "compositions", "history": 5 },
    { "name": "differences", "history": 64 },
    { "name": "wants", "history": 20 }
@@ -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)
+		}
+	}
+}
@@ -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)
+}
@@ -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)
+	}
+}
@@ -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)