Assigner: write each condition it says as a finding too, and say it straight only until the take-over, so its holds reach the operator through the condition keeper and never twice (issue 1952 WM6)
This commit is contained in:
@@ -15,6 +15,7 @@ import (
|
||||
"os"
|
||||
"os/signal"
|
||||
"slices"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -26,8 +27,11 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -66,8 +70,13 @@ var readBuckets = []string{
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.StepsBucket(goBuildModule),
|
||||
takeoverBucket,
|
||||
}
|
||||
|
||||
// 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"
|
||||
@@ -76,7 +85,7 @@ const goBuildModule = "mesh-build-go"
|
||||
// (ADR 0567 §1), named after the assigner as every module state is.
|
||||
var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket,
|
||||
delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket,
|
||||
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket}
|
||||
delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket}
|
||||
|
||||
// seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
|
||||
// them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
|
||||
@@ -98,6 +107,7 @@ var keptStates = []announce.State{
|
||||
{Name: "differences", Readable: true},
|
||||
{Name: "puts", Readable: true},
|
||||
{Name: "wants", Readable: true},
|
||||
{Name: health.FindingsState, Readable: true},
|
||||
}
|
||||
|
||||
// buckets is every bucket the assigner opened, by name.
|
||||
@@ -146,6 +156,12 @@ 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.
|
||||
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
|
||||
}
|
||||
if sourceOfGrants(name) && err != nil {
|
||||
// The grants part's grants and the controller's lease choose where grants are read from (issue 774): without
|
||||
// them the controller's grants are read, as before the lease is handed away.
|
||||
@@ -194,7 +210,60 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
thisNode = account.Node
|
||||
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
||||
return follow(ctx, js, conn, kvs, sayVia(ctx, js), emitting{said}, logger)
|
||||
return follow(ctx, js, conn, kvs, findingsOf(ctx, kvs, logger.Printf).Say(sayVia(ctx, js)), emitting{said}, 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.
|
||||
func findingsOf(ctx context.Context, kvs buckets, logf func(string, ...any)) assign.Findings {
|
||||
var f assign.Findings
|
||||
if kv := kvs[assign.FindingsBucket]; kv != nil {
|
||||
f.Put = func(key string, value []byte) error { _, err := kv.Put(ctx, key, value); return err }
|
||||
f.Delete = func(key string) error { return kv.Delete(ctx, key) }
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// thisNode is the node this process runs on, as its credential names it: said with each of the seat's events.
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a h1:nhRFmtdcI4VkghGUdJhpt1BnPuTQcekioOIRzW8fcG0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// 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
|
||||
@@ -73,7 +75,13 @@ func TestTheRestoreAskIsValid(t *testing.T) {
|
||||
|
||||
// 1906/D3: the condition for a held fingerprint names the module, the node and the check's words, under the assigner's own
|
||||
// name (ADR 0437 clause 9), and is cleared under the same key.
|
||||
func TestTheHeldConditionNamesModuleNodeAndWords(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)
|
||||
}
|
||||
|
||||
func heldConditionNamesModuleNodeAndWords(t *testing.T) {
|
||||
c := HeldCondition("mail", fNew, "ace", "connection refused", true, true)
|
||||
if _, err := asks.EncodeCondition(c); err != nil {
|
||||
t.Fatalf("the condition is refused: %v", err)
|
||||
@@ -87,7 +95,7 @@ func TestTheHeldConditionNamesModuleNodeAndWords(t *testing.T) {
|
||||
}
|
||||
|
||||
// 1906/H2: a hold lifted by itself clears the held condition under its key, saying why.
|
||||
func TestALiftedHoldClearsItsConditionSayingWhy(t *testing.T) {
|
||||
func liftedHoldClearsItsConditionSayingWhy(t *testing.T) {
|
||||
c := LiftedCondition("mail", fNew)
|
||||
if _, err := asks.EncodeCondition(c); err != nil {
|
||||
t.Fatalf("the condition is refused: %v", err)
|
||||
@@ -97,6 +105,48 @@ func TestALiftedHoldClearsItsConditionSayingWhy(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.
|
||||
func saidConditionsAreFindings(t *testing.T) {
|
||||
at := time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC)
|
||||
for _, c := range []struct {
|
||||
raised, cleared asks.Condition
|
||||
key string
|
||||
}{
|
||||
{HeldCondition("mail", fNew, "ace", "refused", true, true), LiftedCondition("mail", fNew), "module.mail." + strings.TrimPrefix(fNew, delivery.SourcePrefix2) + ".held"},
|
||||
{NodeFailedCondition("ace", "mail", fNew, NodeFailure{Words: "x", Others: []string{"g14"}}, true), NodeFailedCondition("ace", "mail", fNew, NodeFailure{}, false), "module.ace.mail.failed-alone"},
|
||||
{NeitherCondition("ace", "mail", Holding{Rule: 3, Words: "w"}, true), NeitherCondition("ace", "mail", Holding{}, false), "module.ace.mail.neither"},
|
||||
{PlacementCondition(Pair{Node: "ace", Module: "mail", Only: OnlyAssignments}, true), PlacementCondition(Pair{Node: "ace", Module: "mail"}, false), "module.ace.mail.placement"},
|
||||
{SeatWaitsCondition([]string{"placement"}), SeatWaitsCondition(nil), "module.mesh-delivery-assigner.seat-states"},
|
||||
{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 }}
|
||||
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 {
|
||||
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 := FindingOf(asks.Condition{Key: "another.x", State: asks.StateRaised}, at); err == nil {
|
||||
t.Error("a condition not the assigner's is a finding")
|
||||
}
|
||||
}
|
||||
|
||||
// 1906/D4: the record ask and the backup ask are valid asks of the router and of mesh-sdk's layout at their keys, answerable in
|
||||
// the view, and name the move and the node.
|
||||
func TestTheRecordAndBackupAsksAreValid(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// The assigner's said conditions become findings (novox/hq ADR 0564 §2, issue 1952 WM6, note 16): every condition it says
|
||||
// under its own name (ADR 0437 §9) — a hold and its lift, a node failing alone, a module on neither list, a backup, a pair
|
||||
// placed by one source, the seat waiting for its states — is also a finding in its own bucket `findings`, written at its
|
||||
// raise and deleted at its clear, which the condition keeper raises and clears. Until the operator's take-over the router
|
||||
// hears it straight from the assigner, as before; from the take-over the keeper's condition is the one said, so the
|
||||
// assigner no longer says it straight (never both).
|
||||
|
||||
// FindingsBucket is the assigner's findings.
|
||||
var FindingsBucket = health.FindingsBucket(delivery.AssignerModule)
|
||||
|
||||
// FindingOf is a said condition as a finding: its key `<scope>.<id>.<kind>`, from the condition's own key
|
||||
// `mesh-delivery-assigner.<kind>.<id…>` (the subject's id when the key names none), its words unchanged. raised is false for
|
||||
// a clear, whose key is to be deleted; value is nil then.
|
||||
func FindingOf(c asks.Condition, at time.Time) (key string, value []byte, raised bool, err error) {
|
||||
rest, ok := strings.CutPrefix(c.Key, delivery.AssignerModule+".")
|
||||
if !ok || rest == "" {
|
||||
return "", nil, false, fmt.Errorf("the condition %q is not the assigner's", c.Key)
|
||||
}
|
||||
kind, id, _ := strings.Cut(rest, ".")
|
||||
if id == "" {
|
||||
id = c.Subject.ID
|
||||
}
|
||||
scope := c.Subject.Scope
|
||||
if scope == "" {
|
||||
scope = health.ScopeModule
|
||||
}
|
||||
key = health.FindingKey(scope, id, kind)
|
||||
if c.State == asks.StateCleared {
|
||||
return key, nil, false, nil
|
||||
}
|
||||
if c.State != asks.StateRaised {
|
||||
return "", nil, false, fmt.Errorf("the condition %s is neither raised nor cleared", c.Key)
|
||||
}
|
||||
summary := c.Summary
|
||||
if summary == "" {
|
||||
summary = c.Headline
|
||||
}
|
||||
var actions []health.Action
|
||||
for _, a := range c.Actions {
|
||||
actions = append(actions, health.Action{Label: a.Label, Verb: a.Verb, Machine: a.Machine, Arguments: a.Arguments, Level: string(a.Level)})
|
||||
}
|
||||
value, err = health.EncodeFinding(health.Finding{Key: key, Kind: kind, Subject: health.Subject{Scope: scope, ID: id, Machine: c.Subject.Machine},
|
||||
Severity: health.Severity(c.Severity), Summary: summary, Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved,
|
||||
Needs: c.Needs, Actions: actions, Evidence: []health.Evidence{{At: at.UTC(), Said: summary}}, Source: "assigner." + kind,
|
||||
Observer: delivery.AssignerModule, At: at.UTC()})
|
||||
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).
|
||||
type Findings struct {
|
||||
Put func(key string, value []byte) error
|
||||
Delete func(key string) error
|
||||
ToKeeper func() bool
|
||||
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.
|
||||
func (f Findings) Say(say func(asks.Condition) error) func(asks.Condition) error {
|
||||
return func(c asks.Condition) error {
|
||||
var errs []error
|
||||
if f.Put != nil {
|
||||
now := time.Now
|
||||
if f.Now != nil {
|
||||
now = f.Now
|
||||
}
|
||||
key, value, raised, err := FindingOf(c, now())
|
||||
switch {
|
||||
case err != nil:
|
||||
errs = append(errs, err)
|
||||
case raised:
|
||||
if err := f.Put(key, value); err != nil {
|
||||
errs = append(errs, fmt.Errorf("the finding %s cannot be written: %w", key, err))
|
||||
}
|
||||
default:
|
||||
if err := f.Delete(key); err != nil {
|
||||
errs = append(errs, fmt.Errorf("the finding %s cannot be deleted: %w", key, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
if f.ToKeeper == nil || !f.ToKeeper() {
|
||||
if say != nil {
|
||||
if err := say(c); err != nil {
|
||||
errs = append(errs, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
}
|
||||
@@ -33,7 +33,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-build-go.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease",
|
||||
want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease",
|
||||
"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", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
@@ -44,7 +44,7 @@ func TestManifest(t *testing.T) {
|
||||
states[s.Name] = s.History
|
||||
}
|
||||
if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20,
|
||||
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0}; !reflect.DeepEqual(states, want) {
|
||||
"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
|
||||
@@ -19,6 +19,7 @@
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
"mesh-conditions.takeover",
|
||||
"mesh-config.resolved",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
@@ -66,7 +67,8 @@
|
||||
{ "name": "compositions", "history": 5 },
|
||||
{ "name": "differences", "history": 64 },
|
||||
{ "name": "puts", "history": 20 },
|
||||
{ "name": "wants", "history": 20 }
|
||||
{ "name": "wants", "history": 20 },
|
||||
{ "name": "findings", "history": 5 }
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user