Say the assigner's conditions as findings alone, the keeper being the one source (issue 1126, B8)
ADR 0590 clause 3 ends the choice of a condition source by the keeper's take-over key, which mesh-sdk no longer defines. The assigner watched that key and said every condition straight to the router while it named the controller. It now writes each as a finding the keeper carries, and says one straight only where it has no findings bucket at all. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
@@ -26,6 +26,7 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose/judge"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
@@ -419,8 +420,8 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
bkey, _ := delivery.BuilderBuildKey("web", fNew)
|
||||
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
|
||||
for _, fp := range []string{fNew, fOld} {
|
||||
gkey, _ := delivery.ControllerGrantKey("web", fp)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
gkey, _ := delivery.PartGrantKey("web", fp)
|
||||
grant, _ := delivery.EncodeGrant(delivery.Grant{})
|
||||
b.putJSON(t, delivery.PartGrantsBucket, gkey, grant)
|
||||
}
|
||||
runs := map[string]string{"ace": fOld, "g14": "build-9", "shanks": fOld}
|
||||
@@ -446,12 +447,12 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
var neither asks.Condition
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && neither.Key == ""; time.Sleep(100 * time.Millisecond) {
|
||||
for k, c := range b.conditionsSaid(t) {
|
||||
if strings.Contains(k, ".neither.") && c.State == asks.StateRaised {
|
||||
if strings.HasSuffix(k, ".neither") && c.State == asks.StateRaised {
|
||||
neither = c
|
||||
}
|
||||
}
|
||||
}
|
||||
if neither.Subject.Machine != "g14" || neither.Subject.ID != "web" || neither.Severity != asks.SeverityUrgent || !strings.Contains(neither.Explanation, "canary") {
|
||||
if neither.Subject.Machine != "g14" || !strings.HasPrefix(neither.Subject.ID, "g14.web") || neither.Severity != asks.SeverityUrgent || !strings.Contains(neither.Explanation, "canary") {
|
||||
t.Errorf("the condition of web on neither list on g14 = %+v", neither)
|
||||
}
|
||||
if keys, err := b.buckets[delivery.AssignerDifferencesBucket].Keys(b.ctx); err == nil && len(keys) != 0 {
|
||||
@@ -532,36 +533,34 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// conditionsSaid reads what the assigner said on the operator channel's subject for conditions, by condition key.
|
||||
// conditionsSaid reads what the assigner said of its conditions, by finding key: each a finding in its own bucket, which
|
||||
// the condition keeper carries (novox/hq ADR 0590 §3, issue 1126), raised while it stands and cleared once deleted, each
|
||||
// with the words its last finding said and the subject the finding names (`<module>[.<fingerprint>]`).
|
||||
func (b *bed) conditionsSaid(t *testing.T) map[string]asks.Condition {
|
||||
t.Helper()
|
||||
stream, err := b.js.Stream(b.ctx, "CONDITIONS")
|
||||
w, err := b.buckets[assign.FindingsBucket].WatchAll(b.ctx, jetstream.IncludeHistory())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer w.Stop()
|
||||
out := map[string]asks.Condition{}
|
||||
cons, err := stream.CreateOrUpdateConsumer(b.ctx, jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for {
|
||||
batch, err := cons.FetchNoWait(100)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
for e := range w.Updates() {
|
||||
if e == nil {
|
||||
return out
|
||||
}
|
||||
n := 0
|
||||
for m := range batch.Messages() {
|
||||
n++
|
||||
c, err := asks.DecodeCondition(m.Data())
|
||||
c := out[e.Key()]
|
||||
c.Key, c.State = e.Key(), asks.StateCleared
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
f, err := health.DecodeFinding(e.Value())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
out[c.Key] = c
|
||||
}
|
||||
if n == 0 {
|
||||
return out
|
||||
c = asks.Condition{Key: e.Key(), State: asks.StateRaised, Severity: string(f.Severity), Explanation: f.Explanation,
|
||||
Resolved: f.Resolved, Subject: asks.Scope{Scope: string(f.Subject.Scope), ID: f.Subject.ID, Machine: f.Subject.Machine}}
|
||||
}
|
||||
out[e.Key()] = c
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// 1906/D3 (ADR 0539 §19), whole on the bus: a module holding data whose new fingerprint failed its health check on the
|
||||
@@ -594,7 +593,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) {
|
||||
key := ""
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && key == ""; time.Sleep(100 * time.Millisecond) {
|
||||
for k, c := range b.conditionsSaid(t) {
|
||||
if c.State == asks.StateRaised && c.Subject.ID == "mail" && strings.Contains(k, ".held.") {
|
||||
if c.State == asks.StateRaised && strings.HasPrefix(c.Subject.ID, "mail.") && strings.HasSuffix(k, ".held") {
|
||||
if c.Subject.Machine != "ace" || !strings.Contains(c.Explanation, "connection refused") {
|
||||
t.Fatalf("the condition = %+v, want it to name ace and the check's words", c)
|
||||
}
|
||||
@@ -629,8 +628,8 @@ func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
|
||||
}
|
||||
bkey, _ := delivery.BuilderBuildKey(name, fp)
|
||||
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
|
||||
gkey, _ := delivery.ControllerGrantKey(name, fp)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
gkey, _ := delivery.PartGrantKey(name, fp)
|
||||
grant, _ := delivery.EncodeGrant(delivery.Grant{})
|
||||
b.putJSON(t, delivery.PartGrantsBucket, gkey, grant)
|
||||
b.compose(t, "ace", 'c')
|
||||
b.place(t, "ace", name)
|
||||
@@ -672,8 +671,8 @@ func (b *bed) fixture(t *testing.T, name, fp, manifest string) {
|
||||
built, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
|
||||
bkey, _ := delivery.BuilderBuildKey(name, fp)
|
||||
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
|
||||
gkey, _ := delivery.ControllerGrantKey(name, fp)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
gkey, _ := delivery.PartGrantKey(name, fp)
|
||||
grant, _ := delivery.EncodeGrant(delivery.Grant{})
|
||||
b.putJSON(t, delivery.PartGrantsBucket, gkey, grant)
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
b.putJSON(t, delivery.PresenceBucket, "ace", on)
|
||||
@@ -762,7 +761,7 @@ func TestAReleaseAnswerWritesTheWantAgain(t *testing.T) {
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
|
||||
cleared := false
|
||||
for k, c := range b.conditionsSaid(t) {
|
||||
cleared = cleared || (c.Subject.ID == "web" && c.State == asks.StateCleared && strings.Contains(k, ".held."))
|
||||
cleared = cleared || (strings.HasPrefix(k, "module.web.") && c.State == asks.StateCleared && strings.HasSuffix(k, ".held"))
|
||||
}
|
||||
if cleared {
|
||||
return
|
||||
@@ -852,7 +851,7 @@ func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(t *testing.T) {
|
||||
continue
|
||||
}
|
||||
for k, c := range b.conditionsSaid(t) {
|
||||
if c.Subject.ID == "web" && c.State == asks.StateCleared && strings.Contains(k, ".held.") && strings.Contains(c.Resolved, "lifted its hold by itself") {
|
||||
if strings.HasPrefix(k, "module.web.") && c.State == asks.StateCleared && strings.HasSuffix(k, ".held") {
|
||||
if w := b.wantNow(t, "ace", "web", "the want at the lifted fingerprint", func(w delivery.Want) bool { return w.FP == fNew }); w.Rule != 0 {
|
||||
t.Errorf("the want = %+v", w)
|
||||
}
|
||||
|
||||
@@ -66,7 +66,6 @@ var readBuckets = []string{
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
takeoverBucket,
|
||||
provisions.ProvisionsBucket,
|
||||
keptBucket,
|
||||
}
|
||||
@@ -75,10 +74,6 @@ var readBuckets = []string{
|
||||
// newest done trunk build of a module names the commit a reader of the manifests runs at (novox/hq ADR 0571 clause 1).
|
||||
var readLogs = []string{delivery.BuildLogRead}
|
||||
|
||||
// takeoverBucket is the condition keeper's take-over key's bucket (novox/hq ADR 0564 §9): whether the assigner's said
|
||||
// conditions go straight to the router or only as findings (issue 1952 WM6).
|
||||
var takeoverBucket = health.Bucket(health.KeeperModule, health.TakeoverState)
|
||||
|
||||
// goBuildModule is the Go build module, which takes the assigner's compose asks on its seat's check stream and writes
|
||||
// their steps in its steps state (novox/hq ADR 0482 clause 4).
|
||||
const goBuildModule = "mesh-build-go"
|
||||
@@ -157,9 +152,9 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err)
|
||||
continue
|
||||
}
|
||||
if (name == assign.FindingsBucket || name == takeoverBucket) && err != nil {
|
||||
// New with health leaving the controller (issue 1952 WM6): without them the conditions are said straight to the
|
||||
// router and kept as no finding, as before.
|
||||
if name == assign.FindingsBucket && err != nil {
|
||||
// New with health leaving the controller (issue 1952 WM6): without it the conditions are said straight to the
|
||||
// router and kept as no finding.
|
||||
logf("the bucket %s cannot be opened (%v): the assigner's conditions are said straight to the router until it starts again", name, err)
|
||||
continue
|
||||
}
|
||||
@@ -180,7 +175,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
|
||||
// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source.
|
||||
func startsWithout(name string) bool {
|
||||
switch name {
|
||||
case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, takeoverBucket,
|
||||
case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket,
|
||||
delivery.AssignerPutsBucket:
|
||||
return true
|
||||
}
|
||||
@@ -222,10 +217,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, buildLog, logger)
|
||||
}
|
||||
|
||||
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, and the
|
||||
// condition keeper's take-over key, watched, so a condition is said straight to the router only until the take-over. A
|
||||
// bucket it cannot open is said: without its findings the conditions are said straight, as before; without the take-over
|
||||
// key they are said straight too, the controller being followed.
|
||||
// findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
|
||||
// condition keeper carries (novox/hq ADR 0590 §3). Without it the conditions are said straight to the router.
|
||||
func findingsOf(ctx context.Context, kvs buckets, logf func(string, ...any)) assign.Findings {
|
||||
var f assign.Findings
|
||||
if kv := kvs[assign.FindingsBucket]; kv != nil {
|
||||
@@ -239,46 +232,6 @@ func findingsOf(ctx context.Context, kvs buckets, logf func(string, ...any)) ass
|
||||
return keys, err
|
||||
}
|
||||
}
|
||||
var keeper atomic.Bool
|
||||
f.ToKeeper = keeper.Load
|
||||
takeover := takeoverBucket
|
||||
kv := kvs[takeover]
|
||||
if kv == nil {
|
||||
return f
|
||||
}
|
||||
w, err := kv.Watch(ctx, health.TakeoverKey)
|
||||
if err != nil {
|
||||
process.NotWatched(logf, takeover, err, "the controller is followed: the conditions are said straight to the router")
|
||||
return f
|
||||
}
|
||||
go func() {
|
||||
defer w.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case e, ok := <-w.Updates():
|
||||
if !ok {
|
||||
logf("the watch of %s ended: the last source it named is followed until the assigner starts again", takeover)
|
||||
return
|
||||
}
|
||||
if e == nil {
|
||||
continue
|
||||
}
|
||||
var raw []byte
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
raw = e.Value()
|
||||
}
|
||||
t, err := health.DecodeTakeover(raw, raw != nil)
|
||||
if err != nil {
|
||||
logf("the take-over key cannot be read (%v): the controller is followed", err)
|
||||
}
|
||||
if was := keeper.Swap(t.Follow == health.FollowKeeper); was != keeper.Load() {
|
||||
logf("the assigner's conditions are said by %s from now (the take-over key follows %s)", t.Seat(), t.Follow)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return f
|
||||
}
|
||||
|
||||
|
||||
@@ -78,7 +78,7 @@ func TestTheRestoreAskIsValid(t *testing.T) {
|
||||
func TestTheAssignersConditions(t *testing.T) {
|
||||
t.Run("the held condition names module, node and words", heldConditionNamesModuleNodeAndWords)
|
||||
t.Run("a lifted hold clears its condition saying why", liftedHoldClearsItsConditionSayingWhy)
|
||||
t.Run("1952/WM6: each said condition is a finding, its clear deletes it, said straight only until the take-over", saidConditionsAreFindings)
|
||||
t.Run("1952/WM6: each said condition is a finding, its clear deletes it, said straight only with no findings bucket", saidConditionsAreFindings)
|
||||
}
|
||||
|
||||
func heldConditionNamesModuleNodeAndWords(t *testing.T) {
|
||||
@@ -106,8 +106,9 @@ func liftedHoldClearsItsConditionSayingWhy(t *testing.T) {
|
||||
}
|
||||
|
||||
// 1952/WM6 (ADR 0564 §2, issue 1952 note 16): every condition the assigner says is a finding in its own bucket, under
|
||||
// `<scope>.<id>.<kind>`, written at the raise and deleted at the clear; it is said straight to the router while the take-over
|
||||
// key follows the controller, and only as a finding from the take-over.
|
||||
// `<scope>.<id>.<kind>`, written at the raise and deleted at the clear, and only as a finding: the keeper is the one
|
||||
// source since the controller's end (novox/hq ADR 0590 §3, issue 1126 B8). Said straight to the router only where the
|
||||
// findings bucket cannot be written to at all.
|
||||
func saidConditionsAreFindings(t *testing.T) {
|
||||
at := time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC)
|
||||
for _, c := range []struct {
|
||||
@@ -121,24 +122,25 @@ func saidConditionsAreFindings(t *testing.T) {
|
||||
{BackupCondition("ace", "mail", fNew, "disk full", true), BackupCondition("ace", "mail", fNew, "", false), "module.ace.mail." + strings.TrimPrefix(fNew, delivery.SourcePrefix2) + ".backup"},
|
||||
} {
|
||||
puts, deletes, said := map[string]health.Finding{}, []string{}, 0
|
||||
keeper := false
|
||||
f := Findings{Put: func(k string, v []byte) error {
|
||||
got, err := health.DecodeFinding(v)
|
||||
puts[k] = got
|
||||
return err
|
||||
}, Delete: func(k string) error { deletes = append(deletes, k); return nil }, ToKeeper: func() bool { return keeper }, Now: func() time.Time { return at }}
|
||||
}, Delete: func(k string) error { deletes = append(deletes, k); return nil }, Now: func() time.Time { return at }}
|
||||
say := f.Say(func(asks.Condition) error { said++; return nil })
|
||||
if err := say(c.raised); err != nil {
|
||||
t.Fatalf("%s: %v", c.key, err)
|
||||
}
|
||||
got, ok := puts[c.key]
|
||||
if !ok || got.Severity != health.Severity(c.raised.Severity) || got.Headline != c.raised.Headline || got.Observer != delivery.AssignerModule ||
|
||||
got.Subject.Machine != c.raised.Subject.Machine || said != 1 {
|
||||
got.Subject.Machine != c.raised.Subject.Machine || said != 0 {
|
||||
t.Fatalf("%s: finding %+v (written %v), said %d", c.key, got, puts, said)
|
||||
}
|
||||
keeper = true
|
||||
if err := say(c.cleared); err != nil || !slices.Equal(deletes, []string{c.key}) || said != 1 {
|
||||
t.Fatalf("%s: deleted %v, said %d after the take-over (%v)", c.key, deletes, said, err)
|
||||
if err := say(c.cleared); err != nil || !slices.Equal(deletes, []string{c.key}) || said != 0 {
|
||||
t.Fatalf("%s: deleted %v, said %d (%v)", c.key, deletes, said, err)
|
||||
}
|
||||
if err := (Findings{}).Say(func(asks.Condition) error { said++; return nil })(c.raised); err != nil || said != 1 {
|
||||
t.Fatalf("%s: with no findings bucket said %d (%v), want straight once", c.key, said, err)
|
||||
}
|
||||
}
|
||||
if _, _, _, err := FindingOf(asks.Condition{Key: "another.x", State: asks.StateRaised}, at); err == nil {
|
||||
|
||||
@@ -59,18 +59,17 @@ func FindingOf(c asks.Condition, at time.Time) (key string, value []byte, raised
|
||||
return key, value, true, err
|
||||
}
|
||||
|
||||
// Findings is where the assigner's findings go: the bucket's put and delete, and whether the keeper's conditions are the
|
||||
// ones said (the take-over key names the keeper).
|
||||
// Findings is where the assigner's findings go: the bucket's put and delete.
|
||||
type Findings struct {
|
||||
Put func(key string, value []byte) error
|
||||
Delete func(key string) error
|
||||
Keys func() ([]string, error)
|
||||
ToKeeper func() bool
|
||||
Now func() time.Time
|
||||
Put func(key string, value []byte) error
|
||||
Delete func(key string) error
|
||||
Keys func() ([]string, error)
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// Say is say with each condition also written as a finding (or its finding deleted), and said straight to the router only
|
||||
// while the take-over key follows the controller. A finding that cannot be written is the error, beside the say's.
|
||||
// Say is say with each condition written as a finding (or its finding deleted), which the condition keeper carries, the
|
||||
// one source since the controller's end (novox/hq ADR 0590 §3); said straight to the router only where there is no
|
||||
// findings bucket at all. A finding that cannot be written is the error.
|
||||
func (f Findings) Say(say func(asks.Condition) error) func(asks.Condition) error {
|
||||
return func(c asks.Condition) error {
|
||||
var errs []error
|
||||
@@ -93,7 +92,7 @@ func (f Findings) Say(say func(asks.Condition) error) func(asks.Condition) error
|
||||
}
|
||||
}
|
||||
}
|
||||
if f.ToKeeper == nil || !f.ToKeeper() {
|
||||
if f.Put == nil {
|
||||
if say != nil {
|
||||
if err := say(c); err != nil {
|
||||
errs = append(errs, err)
|
||||
|
||||
@@ -33,7 +33,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved",
|
||||
want := []string{"mesh-build-go.steps", "mesh-config.kept", "mesh-config.resolved",
|
||||
"mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-provisioner.provisions", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
|
||||
@@ -25,7 +25,6 @@
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-conditions.takeover",
|
||||
"mesh-config.kept",
|
||||
"mesh-config.resolved",
|
||||
"mesh-delivery-builder.build-log",
|
||||
|
||||
Reference in New Issue
Block a user