From 0def1a70d7c5f4416279cc86e581af8d41aada83 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:35:25 +0100 Subject: [PATCH 1/9] Watchman: list the machines failing toward x in also of machine.x.network and machine.x.silent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §706 (the watchman's change) Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../internal/watch/findings_test.go | 18 +++++++++ .../internal/watch/nodehealth.go | 37 ++++++++++++++++++- .../internal/watch/signals.go | 4 +- 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 6a99ed7b..fea28393 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -3,6 +3,7 @@ package watch import ( "encoding/json" "errors" + "slices" "strings" "sync/atomic" "testing" @@ -602,6 +603,23 @@ func TestTheWatchmansFindings(t *testing.T) { if _, ok := o.findings["machine.hub.unreachable"]; ok { t.Error("the silent hub is said unreachable as well") } + // design 47 G §706: the silence lists the machines failing toward it in `also`, as `unreachable` does. + if f := o.findings["machine.hub.silent"]; !slices.Equal(f.Also, []string{"lap"}) { + t.Errorf("the hub's silence lists %v in also", f.Also) + } + w.NodeHealth("lap", nil, Delete) + if f := o.findings["machine.hub.silent"]; len(f.Also) != 0 { + t.Errorf("the hub's silence still lists %v once nothing fails toward it", f.Also) + } + w.NodeHealth("lap", toward("lap", "tunnel", "hub"), Put) + // the hub's own network open, two machines failing toward it: its `network` finding lists both. + w.NodeHealth("hub", toward("hub", "bus"), Put) + w.NodeHealth("ace", toward("ace", "tunnel", "hub"), Put) + if f := o.findings["machine.hub.network"]; !slices.Equal(f.Also, []string{"ace", "lap"}) { + t.Errorf("the hub's network lists %v in also: %v", f.Also, o.findings) + } + w.NodeHealth("hub", nil, Delete) + w.NodeHealth("ace", nil, Delete) w.Connect("node.hub") // back: lap's tunnel failing toward a hub up is its own again if _, own := o.findings["machine.lap.network"]; !own { t.Errorf("hub back: findings %v", o.findings) diff --git a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go index da6adfe1..7295e1b5 100644 --- a/modules/mesh-delivery-watchman/internal/watch/nodehealth.go +++ b/modules/mesh-delivery-watchman/internal/watch/nodehealth.go @@ -274,6 +274,34 @@ func orModule(m string) string { return m } +// silentAlso sets the machines `also` lists on the open `silent` finding of x, when there is one. +func (w *Watchman) silentAlso(x string, also []string) { + value, ok := w.findings[FindingKey(ScopeMachine, x, KindSilent)] + if !ok { + return + } + f, err := DecodeFinding(value) + if err != nil || slices.Equal(f.Also, also) { + return + } + f.Also = also + w.find(f) +} + +// keepSilentAlso gives a `silent` finding about to be written the `also` the open one holds, so a write of its other +// words (its urgency after half an hour) never drops the machines listed there. +func (w *Watchman) keepSilentAlso(f Finding) Finding { + if f.Kind != KindSilent { + return f + } + if value, ok := w.findings[f.Key()]; ok { + if cur, err := DecodeFinding(value); err == nil { + f.Also = cur.Also + } + } + return f +} + // pointed is one machine's failing part that points at another machine. type pointed struct { from string @@ -355,6 +383,13 @@ func (w *Watchman) networks() { } return len(xs) > 0 } + // A machine silent on the record holds the failures toward it under its silence, so its finding lists in `also` the + // machines whose failing parts point at it, as `unreachable` does (design 47 G §706); a change of `also` rewrites it. + for key, value := range w.findings { + if f, err := DecodeFinding(value); err == nil && f.Kind == KindSilent && key == f.Key() { + w.silentAlso(f.ID, from(f.ID)) + } + } want := map[string]bool{} saysOwn := map[string]bool{} for _, m := range machines { @@ -406,7 +441,7 @@ func (w *Watchman) networks() { severity = Urgent summary += fmt.Sprintf("; %s cannot reach it", strings.Join(waiting, ", ")) } - f := Finding{Scope: ScopeMachine, ID: m, Token: "network", Kind: KindMachineNetwork, Machine: m, Severity: severity, + f := Finding{Scope: ScopeMachine, ID: m, Token: "network", Kind: KindMachineNetwork, Machine: m, Also: from(m), Severity: severity, Source: "health-report", Since: w.health[m].Network.Since, Summary: summary, Said: strings.Join(said, " | ")} want[f.Key()] = true saysOwn[m] = true diff --git a/modules/mesh-delivery-watchman/internal/watch/signals.go b/modules/mesh-delivery-watchman/internal/watch/signals.go index e72c31da..c82d83ad 100644 --- a/modules/mesh-delivery-watchman/internal/watch/signals.go +++ b/modules/mesh-delivery-watchman/internal/watch/signals.go @@ -179,7 +179,7 @@ func (w *Watchman) signalExpired(pk string) { if due.While != "" && !w.stillWrong(due.While, due.Finding.Kind) { return } - w.find(due.Finding) + w.find(w.keepSilentAlso(due.Finding)) } } @@ -288,7 +288,7 @@ func (w *Watchman) stuckFinding(sk string, s meshwatchdogsclient.Stuck) { w.signal("S1", node, ControlNodeUrgentAfter-age+time.Second, signalDue{Do: dueFind, While: "silent:" + node, Finding: u}) } } - w.find(f) + w.find(w.keepSilentAlso(f)) w.applyCheck(node) return } From 24d8e4e987b92d8ea142527f5838e051a9769c5a Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:39:00 +0100 Subject: [PATCH 2/9] Contract: the seat mesh-wants' wait gains the step outside MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §706 (step outside added to §498's list), §498 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- contracts/go/seats/meshwants/contract.go | 3 +- contracts/go/seats/meshwants/contract_test.go | 4 +++ .../seats/mesh-wants/fixtures/events.json | 31 +++++++++++++++++++ contracts/seats/mesh-wants/seat.json | 5 +-- contracts/ts/seats/meshwants/index.d.ts | 4 +-- 5 files changed, 42 insertions(+), 5 deletions(-) diff --git a/contracts/go/seats/meshwants/contract.go b/contracts/go/seats/meshwants/contract.go index e86953fe..c4faef83 100644 --- a/contracts/go/seats/meshwants/contract.go +++ b/contracts/go/seats/meshwants/contract.go @@ -675,7 +675,7 @@ type Wait struct { // the acting part waited for Part string `json:"part,omitempty"` Since time.Time `json:"since"` - // which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498) + // which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706) Step string `json:"step"` // its new version's fingerprint Version string `json:"version,omitempty"` @@ -697,6 +697,7 @@ const ( WaitStepDelivery = "delivery" WaitStepKeptBuild = "kept-build" WaitStepBackup = "backup" + WaitStepOutside = "outside" ) // UpgradeInput is what upgrade takes. diff --git a/contracts/go/seats/meshwants/contract_test.go b/contracts/go/seats/meshwants/contract_test.go index d18fcac6..29449e7f 100644 --- a/contracts/go/seats/meshwants/contract_test.go +++ b/contracts/go/seats/meshwants/contract_test.go @@ -55,6 +55,10 @@ func TestFixtures(t *testing.T) { decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, {name: "the builder's toolchain wait ended by a newer fingerprint", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"superseded\",\"module\":\"mesh-sdk\",\"on\":\"toolchain\",\"owner\":\"mesh-delivery-builder\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"build\"}", wire: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"superseded\",\"module\":\"mesh-sdk\",\"on\":\"toolchain\",\"owner\":\"mesh-delivery-builder\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"build\"}", decode: func(b []byte) (any, error) { var v WaitEndedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-ended$")}, + {name: "an outside wait held on a hub's silence, counted from its first start", subject: "mesh.seat.mesh-wants.event.wait-held", given: "{\"bound\":\"30m\",\"clock-from\":\"2026-11-11T10:00:00Z\",\"fingerprint\":\"src2:aaaa\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", wire: "{\"bound\":\"30m\",\"clock-from\":\"2026-11-11T10:00:00Z\",\"fingerprint\":\"src2:aaaa\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", + decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, + {name: "an outside wait ended by the operator's Go on", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"go-on\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", wire: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"go-on\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", + decode: func(b []byte) (any, error) { var v WaitEndedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-ended$")}, {name: "a want", subject: "", given: "{\"fp\":\"ab12\",\"rule\":3,\"v\":1}", wire: "{\"fp\":\"ab12\",\"rule\":3,\"v\":1}", decode: func(b []byte) (any, error) { var v WantsByWantNodeModuleValue diff --git a/contracts/seats/mesh-wants/fixtures/events.json b/contracts/seats/mesh-wants/fixtures/events.json index 85b2e909..908e63d6 100644 --- a/contracts/seats/mesh-wants/fixtures/events.json +++ b/contracts/seats/mesh-wants/fixtures/events.json @@ -253,5 +253,36 @@ "since": "2026-11-11T10:00:00Z", "how": "superseded" } + }, + { + "name": "an outside wait held on a hub's silence, counted from its first start", + "of": "event:wait-held", + "subject": "mesh.seat.mesh-wants.event.wait-held", + "given": { + "step": "outside", + "owner": "mesh-delivery-assigner", + "node": "ace", + "module": "nats", + "fingerprint": "src2:aaaa", + "on": "silent:g14", + "since": "2026-11-11T10:00:00Z", + "clock-from": "2026-11-11T10:00:00Z", + "bound": "30m" + } + }, + { + "name": "an outside wait ended by the operator's Go on", + "of": "event:wait-ended", + "subject": "mesh.seat.mesh-wants.event.wait-ended", + "given": { + "step": "outside", + "owner": "mesh-delivery-assigner", + "node": "ace", + "module": "nats", + "fingerprint": "src2:aaaa", + "on": "silent:g14", + "since": "2026-11-11T10:00:00Z", + "how": "go-on" + } } ] diff --git a/contracts/seats/mesh-wants/seat.json b/contracts/seats/mesh-wants/seat.json index 05966ed3..fb0d2654 100644 --- a/contracts/seats/mesh-wants/seat.json +++ b/contracts/seats/mesh-wants/seat.json @@ -1038,9 +1038,10 @@ "put", "delivery", "kept-build", - "backup" + "backup", + "outside" ], - "description": "which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498)" + "description": "which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706)" }, "part": { "type": "string", diff --git a/contracts/ts/seats/meshwants/index.d.ts b/contracts/ts/seats/meshwants/index.d.ts index 399bdeaa..974bc4d0 100644 --- a/contracts/ts/seats/meshwants/index.d.ts +++ b/contracts/ts/seats/meshwants/index.d.ts @@ -328,8 +328,8 @@ export interface Wait { /** the acting part waited for */ "part"?: string; "since": string; - /** which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498) */ - "step": "take-in" | "validate" | "check" | "build" | "compose" | "grant" | "place" | "pipeline" | "put" | "delivery" | "kept-build" | "backup"; + /** which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706) */ + "step": "take-in" | "validate" | "check" | "build" | "compose" | "grant" | "place" | "pipeline" | "put" | "delivery" | "kept-build" | "backup" | "outside"; /** its new version's fingerprint */ "version"?: string; /** the wait in plain words (design 47 G section 512, section 629) */ From 48dfe360847b18b8490433ccf63a73a258396641 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:46:58 +0100 Subject: [PATCH 3/9] Assigner: a moved module's failing check on an outside cause is a wait that ends by itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The wait (step outside, on, bound, clock-from, the settle, conditions-unread), its events, finding and ask, assigner_waits and assigner_release with how, the settings, and a machine's own open network holding its wants. Builds: design 47 G §706 (assigner), §498, §276, §633, §700 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/mesh-delivery-assigner/asks/asks.go | 29 +- .../cmd/mesh-delivery-assigner/main.go | 33 +- .../cmd/mesh-delivery-assigner/outside.go | 469 ++++++++++++++++++ .../mesh-delivery-assigner/outside_test.go | 273 ++++++++++ .../mesh-delivery-assigner/outside_tools.go | 120 +++++ .../cmd/mesh-delivery-assigner/pass.go | 26 + .../cmd/mesh-delivery-assigner/tools.go | 8 +- .../internal/assign/asks.go | 40 ++ .../internal/assign/ended.go | 1 + .../internal/assign/wants.go | 11 + .../internal/buckets/buckets.go | 3 + .../internal/outside/outside.go | 313 ++++++++++++ .../internal/outside/outside_test.go | 169 +++++++ .../mesh-delivery-assigner/manifest_test.go | 22 +- modules/mesh-delivery-assigner/module.json | 30 +- 15 files changed, 1536 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go create mode 100644 modules/mesh-delivery-assigner/internal/outside/outside.go create mode 100644 modules/mesh-delivery-assigner/internal/outside/outside_test.go diff --git a/modules/mesh-delivery-assigner/asks/asks.go b/modules/mesh-delivery-assigner/asks/asks.go index 41e01ef1..0ddf4dba 100644 --- a/modules/mesh-delivery-assigner/asks/asks.go +++ b/modules/mesh-delivery-assigner/asks/asks.go @@ -38,6 +38,9 @@ const ( // `assignments.`. The kind needs no proof, as the bus's put does not: its yes is an Approve, which only a // channel proving who answered takes and the view never does (ViewAnswerFits). AskKindAssignments = "assignments" + // AskKindOutside is the kind of the assigner's ask about a moved module whose failing health check has waited past its bound + // on an outside cause (design 47 G §706): go on, or count it as failed. `outside...`. + AskKindOutside = "outside" ) // AskKey is an ask's key read whole: its Kind, and the parts its kind has: a merge's Group and heads Hash, a hold's Module @@ -60,11 +63,11 @@ func AskNeedsProof(kind string) (bool, error) { switch kind { case meshdeliveryclient.AskKindMerge: return true, nil - case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments: + case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside: return false, nil } - return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold, - AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments) + return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold, + AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside) } // HoldAskKey is the key of the ask about a module's held fingerprint: `hold..`. @@ -89,6 +92,19 @@ func RestoreAskKey(node, module, fingerprint string) (string, error) { return AskKindRestore + "." + node + "." + key, nil } +// OutsideAskKey is the key of the ask about a module's failing check on a node that waited past its bound on an outside cause: +// `outside...`. +func OutsideAskKey(node, module, fingerprint string) (string, error) { + key, err := RestoreAskKey(node, module, fingerprint) + if err != nil { + return "", err + } + return AskKindOutside + strings.TrimPrefix(key, AskKindRestore), nil +} + +// OutsideAbout is what the outside ask is about: `..` (design 47 G §706). +func OutsideAbout(key string) string { return strings.TrimPrefix(key, AskKindOutside+".") } + // BackupAskKey is the key of the ask about a module's failed backup on a node before it applies a fingerprint: // `backup...`. func BackupAskKey(node, module, fingerprint string) (string, error) { @@ -157,6 +173,13 @@ func ParseAnyAskKey(key string) (AskKey, error) { return refused(err) } return AskKey{Kind: kind, Node: node, Module: module, FP: fp}, nil + case AskKindOutside: + k, err := ParseAnyAskKey(AskKindRestore + "." + rest) + if err != nil { + return refused(fmt.Errorf("an outside ask is outside...")) + } + k.Kind = kind + return k, nil case AskKindBackup: node, module, part, isNode, err := nodeengineclient.ParseComposedKey(rest) if err != nil || isNode { 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 ffd7ab90..8a45b708 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -79,7 +79,7 @@ var readFamilies = []string{buckets.ComposedRead} var ownBuckets = []string{buckets.Asks, buckets.Differences, buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket, - engineReadsBucket, retiredBucket, copiesBucket, writtenBucket} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits} // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. @@ -119,6 +119,7 @@ var keptStates = []announce.State{ {Name: retiredState}, // read only by the copy into assignments (migrate.go) {Name: copiesState, Readable: true}, {Name: writtenState, Readable: true}, + {Name: buckets.WaitsState, Readable: true}, } // kvSet is every bucket the assigner opened, by name. @@ -177,6 +178,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): what each node's engine reads is kept in memory only until it starts again", name, err) continue } + if name == buckets.Waits && err != nil { + // New with the wait on an outside cause (design 47 G §706): without it no wait is made, and a failing check is judged + // as before. + logf("the bucket %s cannot be opened (%v): no outside wait is kept until the assigner starts again", name, err) + continue + } if name == retiredBucket && err != nil { // The retired state is read only for the copy into `assignments` (migrate.go): a mesh made after the switch has // none, and nothing is copied. @@ -201,7 +208,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket, - buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket: + buckets.Puts, buckets.Source, buckets.Merges, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -330,6 +337,7 @@ var watched = []string{ buckets.Merges, engineReadsBucket, writtenBucket, + buckets.Waits, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -508,6 +516,27 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{}, reads: reads, seats: gate} + // The wait on an outside cause (outside.go, design 47 G §706): the keeper's three events say when to judge again, and the + // seat mesh-wants' wait-held and wait-ended are said; registered before the hub starts. + if bus != nil && kvs[buckets.Waits] != nil { + bound, settle := outsideSettings(os.Getenv) + p.outside = newOutsideActs(bus.CallKept, wake, logger.Printf, bound, settle) + for _, subject := range []string{mcseat.SubjectConditionRaised, mcseat.SubjectConditionChanged, mcseat.SubjectConditionCleared} { + subject := subject + if err := hub.Handle(subject, func(m jetstream.Msg) { + p.outside.Heard(m.Subject(), m.Data()) + if err := m.Ack(); err != nil { + logger.Printf("acknowledging %s: %v", m.Subject(), err) + } + }); err != nil { + logger.Printf("%s is not heard, so a wait on an outside cause is judged again only when its check changes: %v", subject, err) + } + } + p.outside.readStart() + } + if js != nil { + p.publishWants = wantsPublisher(ctx, js, thisNode) + } // The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer. // Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it // is not there yet, a node is asked again only when it comes back online. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside.go new file mode 100644 index 00000000..12844b81 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside.go @@ -0,0 +1,469 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogkey" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/views" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assignment" + "mesh-delivery-assigner/internal/buckets" + "mesh-delivery-assigner/internal/outside" +) + +// The wait on an outside cause (design 47 G §706, ADR 0677): the assigner judges a node's health after a want moves, and a moved +// module's failing check on a machine whose open conditions name an outside cause is held as a wait (internal/outside has the +// rules). The conditions come only through the seat mesh-conditions: its verb `conditions`, read at each judging as a kept +// call, and its events, which only say when to judge again. + +// conditionsVerb is the seat verb the keeper answers `conditions` by. +const conditionsVerb = "seat:" + mcseat.Seat + "." + mcseat.VerbConditions + +// wantsPublisher says one of the seat mesh-wants' events, `wait-held` and `wait-ended` among them, under a message id the bus +// keeps one of. +func wantsPublisher(ctx context.Context, js jetstream.JetStream, node string) publisher { + return func(event string, body any, id string) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + msg := nats.NewMsg(seatdef.EventSubject(meshwants.Seat, event)) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", assignment.Module) + msg.Header.Set("x-node", node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) + return err + } +} + +// outsideActs keeps, while the process runs, what the judging of outside waits needs beside the kept waits: the conditions read +// for each machine (one answer is one judging), the machines a heard event says to judge again, the machines whose own network +// is held open, and each check's last signature. +type outsideActs struct { + mu sync.Mutex + call func(key string, body any) (json.RawMessage, error) // a kept call + wake func() + logf func(string, ...any) + // bound and settle are the settings outside-wait-minutes and outside-settle-seconds. + bound, settle time.Duration + // reading is a read in flight by machine, answered the answers it brought, not yet judged by. + reading map[string]bool + answered map[string][]outside.Cond + // dirty is a machine whose waits are judged again, cleared a wait's key whose cause's clearing was heard. + dirty map[string]bool + cleared map[string]bool + // sigs is each check's last signature, by node.module. + sigs map[string]string + // own is the machines whose own network is open; ownRead says the read at the start was answered. + own map[string]bool + ownRead bool + // causes is, by condition key, the nodes whose waits stand on it, as the last pass kept them. + causes map[string][]string + // started says the first pass judged every wait again. + started bool +} + +func newOutsideActs(call func(string, any) (json.RawMessage, error), wake func(), logf func(string, ...any), bound, settle time.Duration) *outsideActs { + if bound <= 0 { + bound = outside.DefaultBound + } + if settle <= 0 { + settle = outside.DefaultSettle + } + return &outsideActs{call: call, wake: wake, logf: logf, bound: bound, settle: settle, reading: map[string]bool{}, answered: map[string][]outside.Cond{}, + dirty: map[string]bool{}, cleared: map[string]bool{}, sigs: map[string]string{}, own: map[string]bool{}, causes: map[string][]string{}} +} + +// outsideSettings are the bounds of the wait, the settings outside-wait-minutes and outside-settle-seconds (design 47 G §633, §706), +// from the environment the manifest's config fills; zero for one not set, which is its default. +func outsideSettings(getenv func(string) string) (bound, settle time.Duration) { + if n, err := strconv.Atoi(strings.TrimSpace(getenv("MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES"))); err == nil && n > 0 { + bound = time.Duration(n) * time.Minute + } + if n, err := strconv.Atoi(strings.TrimSpace(getenv("MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS"))); err == nil && n > 0 { + settle = time.Duration(n) * time.Second + } + return bound, settle +} + +// readStart reads the open conditions of every machine once at the start, for the machines whose own network is held. +func (o *outsideActs) readStart() { + go func() { + answer, err := o.call(conditionsVerb, map[string]any{"scope": "machine"}) + if err != nil { + o.logf("the open conditions are not read at the start (%v): no machine is held for its own network until its event is heard", err) + return + } + conds, err := outside.ParseConditions(answer) + if err != nil { + o.logf("%v", err) + return + } + o.mu.Lock() + for m := range outside.OwnNetwork(conds) { + o.own[m] = true + } + o.ownRead = true + o.mu.Unlock() + o.wake() + }() +} + +// Heard is one of the seat's events condition-raised, -changed or -cleared: it says which machines to judge again and keeps the +// machines whose own network is open; nothing is kept of the condition. +func (o *outsideActs) Heard(subject string, body []byte) { + var e mcseat.ConditionEvent + if err := json.Unmarshal(body, &e); err != nil || e.Key == "" { + o.logf("%s holds what is not a condition's event, and is dropped: %v", subject, err) + return + } + cleared := strings.HasSuffix(subject, "."+mcseat.EventConditionCleared) + o.mu.Lock() + if p := strings.Split(e.Key, "."); len(p) == 3 && p[0] == "machine" && p[2] == "network" { + o.own[p[1]] = !cleared + } + if cleared { + for _, node := range o.causes[e.Key] { + o.dirty[node] = true + o.cleared[node+"|"+e.Key] = true + } + } else if e.Subject != nil { + o.dirty[e.Subject.Machine] = true + for _, m := range e.Subject.Also { + o.dirty[m] = true + } + } + o.mu.Unlock() + o.wake() +} + +// Held says whether machine's own network is open: no want moves to it. +func (o *outsideActs) Held(machine string) bool { + if o == nil { + return false + } + o.mu.Lock() + defer o.mu.Unlock() + return o.own[machine] +} + +// take is the conditions read for machine, consumed (one answer is one judging), or false while the kept call is unanswered, +// which it starts when none is in flight. +func (o *outsideActs) take(machine string) ([]outside.Cond, bool) { + o.mu.Lock() + defer o.mu.Unlock() + if c, ok := o.answered[machine]; ok { + delete(o.answered, machine) + return c, true + } + if o.reading[machine] { + return nil, false + } + o.reading[machine] = true + go func() { + answer, err := o.call(conditionsVerb, map[string]any{"scope": "machine", "machine": machine}) + var conds []outside.Cond + if err == nil { + conds, err = outside.ParseConditions(answer) + } + o.mu.Lock() + o.reading[machine] = false + if err != nil { + o.logf("the conditions of %s are not read (%v): the move waits on %s", machine, err, outside.OnUnread) + } else { + o.answered[machine] = conds + o.dirty[machine] = true // the kept call answered: the move is judged by it + } + o.mu.Unlock() + if err == nil { + o.wake() + } + }() + return nil, false +} + +// outsideReview is the judging of module's check at fp on each of its nodes (§706): it returns the nodes as the rules read them, +// the have of a waiting move no failure (its health unknown) and a go-on's counted healthy, and keeps the waits, says their +// events, finding and ask. moved says the delivery moved owner to node. +func (p *passer) outsideReview(ctx context.Context, module, fp string, nodes []string, n assign.Nodes, keeperSelf bool, moved func(owner, node string) bool) assign.Nodes { + o := p.outside + if o == nil || fp == "" || p.kvs[buckets.Waits] == nil { + return n + } + hex, err := meshcatalogkey.KeyPart(fp) + if err != nil { + return n + } + now := p.clock() + kept := p.w.read(buckets.Waits) + override := map[string]string{} + // A wait of another fingerprint of this module is replaced by the newer want. + for key, raw := range kept { + if w, err := outside.Decode(raw); err == nil && w.Module == module && w.FP != fp { + p.outsideEnd(ctx, key, w, outside.HowReplaced, now) + } + } + for _, node := range nodes { + key := outside.Key(node, module, hex) + var prev *outside.Wait + if raw := kept[key]; raw != nil { + if w, err := outside.Decode(raw); err == nil { + prev = &w + } + } + h, ok := n.Haves[node][module] + sig := fmt.Sprintf("%s|%s|%s", h.Fp, h.Phase, h.Health) + o.mu.Lock() + first := !o.started + changed := o.sigs[node+"."+module] != sig + o.sigs[node+"."+module] = sig + dirty := o.dirty[node] + cleared := prev != nil && prev.Cause != "" && o.cleared[node+"|"+prev.Cause] + o.mu.Unlock() + failing, healthy := ok && h.Fp == fp && assign.Failed(h), ok && assign.Healthy(h, fp) + if !failing && !healthy && prev == nil { + continue + } + if prev != nil && p.outsideAnswered(prev, node, module, fp) { + dirty = true + } + due := prev != nil && (prev.Release != "" || (prev.Settling && !now.Before(prev.SettleEnds))) + var d outside.Decision + if prev != nil && !(first || changed || dirty || due) { + // nothing to judge by: the wait stands as kept; the bound's ask is made when it has run + d = outside.Decision{Wait: prev, Waiting: prev.Standing() || prev.Settling, Own: prev.Decided == outside.HowFailed || prev.Decided == outside.HowOwnFailure, + Counted: prev.Decided == outside.HowGoOn} + if prev.Standing() && !now.Before(prev.ClockFrom.Add(o.bound)) { + d.Ask = true + } + } else { + f := outside.Facts{Now: now, Failing: failing, Healthy: healthy, Node: node, KeeperSelf: keeperSelf, Bound: o.bound, Settle: o.settle, + Moved: func(owner string) bool { return moved(owner, node) }, CauseCleared: cleared} + if failing && !keeperSelf { + f.Conds, f.Read = o.take(node) + } + d = outside.Judge(prev, node, module, fp, f) + o.mu.Lock() + delete(o.dirty, node) + if prev != nil { + delete(o.cleared, node+"|"+prev.Cause) + } + o.mu.Unlock() + } + switch { + case d.Waiting: + override[node] = nodeengineclient.HealthUnknown + case d.Counted: + override[node] = nodeengineclient.HealthOK + } + p.outsideApply(ctx, key, hex, node, module, fp, prev, d, now) + } + o.mu.Lock() + o.started = true + o.mu.Unlock() + if len(override) == 0 { + return n + } + out := n + out.Haves = map[string]map[string]nodeengine.HaveByNodeModuleValue{} + for node, byModule := range n.Haves { + out.Haves[node] = map[string]nodeengine.HaveByNodeModuleValue{} + for m, h := range byModule { + out.Haves[node][m] = h + } + } + for node, health := range override { + h := out.Haves[node][module] + h.Health = health + if health == nodeengineclient.HealthOK { + h.Phase = nodeengineclient.PhaseRunning + } + out.Haves[node][module] = h + } + return out +} + +// outsideAnswered reads the operator's answer to the wait's ask, from the teller's record of it under the ask's key (answeredWith, +// §700's path as the assigner reads an ask's answer), and notes it on the wait as a release; it says whether it did. +func (p *passer) outsideAnswered(w *outside.Wait, node, module, fp string) bool { + if w.Ask == "" || w.Release != "" { + return false + } + key, err := asks.OutsideAskKey(node, module, fp) + if err != nil { + return false + } + for _, how := range []string{assign.OptionGoOn, assign.OptionFailed} { + if p.answeredWith(key, how) { + w.Release, w.By, w.Why = how, "the operator", "the answer to the ask "+w.Ask + return true + } + } + return false +} + +// outsideApply keeps what a judging came to: the wait's record, its events, its finding and its ask. +func (p *passer) outsideApply(ctx context.Context, key, hex, node, module, fp string, prev *outside.Wait, d outside.Decision, now time.Time) { + o := p.outside + switch { + case d.End != "": + p.outsideEnd(ctx, key, *prev, d.End, now) + if d.Wait != nil { + p.keepWait(ctx, key, *d.Wait) + } + return + case d.Wait == nil: + if prev != nil { + p.outsideEnd(ctx, key, *prev, "", now) + } + return + } + w := *d.Wait + if d.Held { + body := p.outsideBody(w) + if err := p.wantsSay(meshwants.EventWaitHeld, body, key+"."+w.On+"."+w.ClockFrom.UTC().Format(time.RFC3339Nano)); err != nil { + p.logf("the wait of %s on %s is not said, and is said at the next pass: %v", module, node, err) + return + } + p.outsideFinding(w, true) + } + if d.Ask && w.Ask == "" { + if id := p.outsideAsk(ctx, node, module, fp, w, now); id != "" { + w.Ask = id + } + } + if w.Standing() || w.Settling || w.Decided != "" { + o.mu.Lock() + if w.Cause != "" && w.Standing() { + if !containsString(o.causes[w.Cause], node) { + o.causes[w.Cause] = append(o.causes[w.Cause], node) + } + } + o.mu.Unlock() + if w.Settling { + p.wakeAt(w.SettleEnds, now) + } else if w.Standing() { + p.wakeAt(w.ClockFrom.Add(o.bound), now) + } + p.keepWait(ctx, key, w) + } +} + +func containsString(l []string, s string) bool { + for _, x := range l { + if x == s { + return true + } + } + return false +} + +// keepWait writes the wait's record. +func (p *passer) keepWait(ctx context.Context, key string, w outside.Wait) { + p.put(ctx, buckets.Waits, key, func() ([]byte, error) { return outside.Encode(w) }) +} + +func (p *passer) outsideBody(w outside.Wait) meshwants.Wait { + since, from := w.Since, w.ClockFrom + return meshwants.Wait{Step: meshwants.WaitStepOutside, Owner: assignment.Module, Node: w.Node, Module: w.Module, Fingerprint: w.FP, On: w.On, + Since: since, ClockFrom: &from, Bound: w.Bound, + Words: fmt.Sprintf("%s on %s waits on %s", w.Module, w.Node, w.On)} +} + +// outsideEnd ends a wait: its wait-ended with how (none for a settle, which was never said), its finding and ask taken away, and +// its record deleted unless the end decided it (the caller keeps that record). +func (p *passer) outsideEnd(ctx context.Context, key string, w outside.Wait, how string, now time.Time) { + if w.Standing() && how != "" { + body := p.outsideBody(w) + body.How = how + if err := p.wantsSay(meshwants.EventWaitEnded, body, key+"."+how+"."+now.UTC().Format(time.RFC3339Nano)); err != nil { + p.logf("the end of the wait of %s on %s is not said, and is said at the next pass: %v", w.Module, w.Node, err) + return + } + } + p.outsideFinding(w, false) + if ask, err := asks.OutsideAskKey(w.Node, w.Module, w.FP); err == nil { + p.remove(ctx, buckets.Asks, ask) + } + if p.outside != nil && w.Cause != "" { + p.outside.mu.Lock() + p.outside.causes[w.Cause] = removeString(p.outside.causes[w.Cause], w.Node) + p.outside.mu.Unlock() + } + p.remove(ctx, buckets.Waits, key) +} + +func removeString(l []string, s string) []string { + var out []string + for _, x := range l { + if x != s { + out = append(out, x) + } + } + return out +} + +// outsideFinding writes the finding delivery-waits-outside, or clears it. +func (p *passer) outsideFinding(w outside.Wait, raised bool) { + if p.findings.Put == nil { + return + } + o := assign.WaitsOutsideFinding(w.Node, w.Module, w.Cause, w.On, p.clock()) + if raised { + if err := p.findings.Put(o.Finding(p.clock())); err != nil { + p.logf("the finding %s is not written: %v", "delivery."+o.ID, err) + } + return + } + if p.findings.Delete != nil { + if err := p.findings.Delete(mcseatKey(o)); err != nil { + p.logf("the finding %s is not cleared: %v", "delivery."+o.ID, err) + } + } +} + +func mcseatKey(o assign.Observed) string { + return views.FindingKey(o.Scope, assign.KeyID(o.ID), o.Kind) +} + +// outsideAsk asks the operator at the wait's bound, once, and returns the asking's id. +func (p *passer) outsideAsk(ctx context.Context, node, module, fp string, w outside.Wait, now time.Time) string { + key, err := asks.OutsideAskKey(node, module, fp) + if err != nil { + return "" + } + asking := assign.AskingSeconds(now) + a := assign.OutsideAsk(key, asks.OutsideAbout(key), node, module, fp, w.On, w.ClockFrom, asking, now) + p.put(ctx, buckets.Asks, key, func() ([]byte, error) { return assign.EncodeAsk(key, a) }) + if ask, err := assign.AskOf(a); err == nil { + return ask.ID + } + return "" +} + +// wakeAt makes a pass when t comes: the settle's end and the bound are the waits' own times, the one wake (no polling). +func (p *passer) wakeAt(t, now time.Time) { + if p.outside == nil || p.outside.wake == nil { + return + } + time.AfterFunc(max(t.Sub(now), 0)+time.Millisecond, p.outside.wake) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go new file mode 100644 index 00000000..f61e7ba1 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_test.go @@ -0,0 +1,273 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/buckets" + "mesh-delivery-assigner/internal/outside" +) + +// The whole of the wait on an outside cause in the pass (design 47 G §706): a fixture keeper answering `conditions` and saying +// its events, a failing check of nats on ace, and the events, finding, ask and record the pass leaves. +type outsideRig struct { + t *testing.T + p *passer + now time.Time + conds []map[string]any + unread bool + events []string + bodies []meshwants.Wait + found map[string]bool + woke chan struct{} + health string +} + +func newOutsideRig(t *testing.T) *outsideRig { + r := &outsideRig{t: t, now: time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC), found: map[string]bool{}, woke: make(chan struct{}, 16), health: nodeengineclient.HealthFailed} + r.p = newPasser(kvSet{buckets.Waits: &memKV{values: map[string][]byte{}}, buckets.Asks: &memKV{values: map[string][]byte{}}}, r.now) + r.p.now = func() time.Time { return r.now } + r.p.w.put(buckets.Asks, "x", nil) + r.p.outside = newOutsideActs(func(key string, body any) (json.RawMessage, error) { + if r.unread { + return nil, context.DeadlineExceeded + } + return json.Marshal(map[string]any{"conditions": r.conds}) + }, func() { r.woke <- struct{}{} }, func(string, ...any) {}, 30*time.Minute, 90*time.Second) + r.p.publishWants = func(event string, body any, id string) error { + r.events = append(r.events, event) + r.bodies = append(r.bodies, body.(meshwants.Wait)) + return nil + } + r.p.findings.Put = func(f mcseat.Finding) error { r.found[f.Scope+"."+f.ID+"."+f.Kind] = true; return nil } + r.p.findings.Delete = func(key string) error { delete(r.found, key); return nil } + return r +} + +func (r *outsideRig) cond(key string, also ...string) { + r.conds = append(r.conds, map[string]any{"key": key, "subject": map[string]any{"machine": strings.Split(key, ".")[1], "also": also}}) +} + +// review is one pass's judging of nats on ace at the fingerprint, with the read answered at once when one is started. +func (r *outsideRig) review(keeperSelf bool, moved func(string, string) bool) assign.Nodes { + r.t.Helper() + n := assign.Nodes{Online: map[string]bool{"ace": true}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{ + "ace": {"nats": {Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: r.health}}}} + if moved == nil { + moved = func(string, string) bool { return false } + } + out := r.p.outsideReview(context.Background(), "nats", fpTest, []string{"ace"}, n, keeperSelf, moved) + select { + case <-r.woke: // the read was answered: the next pass judges by it + out = r.p.outsideReview(context.Background(), "nats", fpTest, []string{"ace"}, n, keeperSelf, moved) + case <-time.After(50 * time.Millisecond): + } + return out +} + +func (r *outsideRig) wait() outside.Wait { + r.t.Helper() + for _, v := range r.p.w.read(buckets.Waits) { + w, err := outside.Decode(v) + if err != nil { + r.t.Fatal(err) + } + return w + } + return outside.Wait{} +} + +func TestTheWaitOnAnOutsideCause(t *testing.T) { + healthOf := func(n assign.Nodes) string { return n.Haves["ace"]["nats"].Health } + t.Run("a cause listing the machine is a wait said at once, ended cleared when it clears healthy", func(t *testing.T) { + r := newOutsideRig(t) + r.cond("machine.hub.unreachable", "ace") + n := r.review(false, nil) + if healthOf(n) == nodeengineclient.HealthFailed || len(r.events) != 1 || r.events[0] != meshwants.EventWaitHeld { + t.Fatalf("health %s events %v", healthOf(n), r.events) + } + b := r.bodies[0] + if b.Step != meshwants.WaitStepOutside || b.On != "unreachable:hub" || b.Bound != "30m0s" || b.ClockFrom == nil || !b.ClockFrom.Equal(r.now) || b.Node != "ace" || b.Module != "nats" { + t.Errorf("wait-held %+v", b) + } + if !r.found["delivery.ace.nats.delivery-waits-outside"] { + t.Errorf("findings %v", r.found) + } + // the cause clears and the module runs healthy + r.conds, r.health, r.now = nil, nodeengineclient.HealthOK, r.now.Add(time.Minute) + r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-cleared", []byte(`{"key":"machine.hub.unreachable","event":"cleared","change":"cleared","at":"2026-11-11T10:01:00Z","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`)) + <-r.woke + r.review(false, nil) + if len(r.events) != 2 || r.events[1] != meshwants.EventWaitEnded || r.bodies[1].How != outside.HowCleared || len(r.found) != 0 || len(r.p.w.read(buckets.Waits)) != 0 { + t.Errorf("events %v bodies %+v findings %v", r.events, r.bodies, r.found) + } + }) + t.Run("a cause not listing the machine settles, then the failure is the delivery's", func(t *testing.T) { + r := newOutsideRig(t) + r.cond("machine.hub.unreachable", "g14") + n := r.review(false, nil) + if healthOf(n) == nodeengineclient.HealthFailed || len(r.events) != 0 || !r.wait().Settling { + t.Fatalf("settle: health %s events %v wait %+v", healthOf(n), r.events, r.wait()) + } + r.now = r.now.Add(91 * time.Second) + if n := r.review(false, nil); healthOf(n) != nodeengineclient.HealthFailed { + t.Errorf("past the settle the check still counts as failing: %s", healthOf(n)) + } + }) + t.Run("a cause turning into another keeps the wait and its clock; a rewritten naming a moved module fails at once", func(t *testing.T) { + r := newOutsideRig(t) + r.cond("machine.hub.unreachable", "ace") + r.review(false, nil) + r.conds = nil + r.cond("machine.hub.network", "ace") + r.now = r.now.Add(5 * time.Minute) + r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.hub.network","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`)) + <-r.woke + r.review(false, nil) + last := len(r.bodies) - 1 + if len(r.events) != 2 || r.events[last] != meshwants.EventWaitHeld || r.bodies[last].On != "network:hub" || !r.bodies[last].ClockFrom.Equal(*r.bodies[0].ClockFrom) { + t.Errorf("events %v bodies %+v", r.events, r.bodies) + } + r2 := newOutsideRig(t) + r2.cond("machine.ace.nats.rewritten") + n := r2.review(false, func(owner, node string) bool { return owner == "nats" }) + if healthOf(n) == nodeengineclient.HealthFailed || !r2.wait().Settling { + t.Errorf("a rewritten of the moved module is its failure: %s %+v", healthOf(n), r2.wait()) + } + }) + t.Run("the keeper silent waits on conditions-unread, and its own module is judged with no read", func(t *testing.T) { + r := newOutsideRig(t) + r.unread = true + r.review(false, nil) + r.now = r.now.Add(90 * time.Second) // the settle has run with no answer + r.review(false, nil) + if len(r.bodies) != 1 || r.bodies[0].On != outside.OnUnread { + t.Fatalf("bodies %+v", r.bodies) + } + r2 := newOutsideRig(t) + r2.cond("machine.hub.unreachable", "ace") + if n := r2.review(true, nil); healthOf(n) != nodeengineclient.HealthFailed || len(r2.events) != 0 { + t.Errorf("the keeper's own failing check waits: %s %v", healthOf(n), r2.events) + } + }) + t.Run("past the bound the operator is asked; Go on counts it healthy and Count as failed holds it, each said ended", func(t *testing.T) { + for how, want := range map[string]string{assign.OptionGoOn: nodeengineclient.HealthOK, assign.OptionFailed: nodeengineclient.HealthFailed} { + r := newOutsideRig(t) + r.cond("machine.hub.silent", "ace") + r.review(false, nil) + r.now = r.now.Add(31 * time.Minute) + r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-changed", []byte(`{"key":"machine.hub.silent","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`)) + <-r.woke + r.review(false, nil) + key, _ := asks.OutsideAskKey("ace", "nats", fpTest) + raw := r.p.w.value(buckets.Asks, key) + if raw == nil { + t.Fatalf("%s: no ask at %s", how, key) + } + _, ask, err := assign.DecodeAsk(key, raw) + if err != nil || ask.About != "ace.nats."+strings.TrimPrefix(key[strings.LastIndex(key, ".")+1:], "") || len(ask.Options) != 2 { + t.Fatalf("ask %+v %v", ask, err) + } + v, _ := mdc.EncodeAnswer(mdc.Answer{Option: how, Warrant: ask.ID + "@2026-11-11T10:32:00Z", Words: "the operator"}) + r.p.w.put(buckets.Answers, key, v) + n := r.review(false, nil) + if r.events[len(r.events)-1] != meshwants.EventWaitEnded || r.bodies[len(r.bodies)-1].How != how || healthOf(n) != want { + t.Errorf("%s: events %v last %+v health %s", how, r.events, r.bodies[len(r.bodies)-1], healthOf(n)) + } + if r.p.w.value(buckets.Asks, key) != nil || len(r.found) != 0 { + t.Errorf("%s: the ask or the finding stands after the end", how) + } + // a check that goes on failing is not waited on again + r.now = r.now.Add(time.Minute) + r.p.outside.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.hub.silent","subject":{"id":"hub","scope":"machine","machine":"hub","also":["ace"]}}`)) + <-r.woke + before := len(r.events) + r.review(false, nil) + if len(r.events) != before { + t.Errorf("%s: a decided check is waited on again", how) + } + } + }) + t.Run("a newer want ends the wait replaced, and a restart judges it again", func(t *testing.T) { + r := newOutsideRig(t) + r.cond("machine.hub.unreachable", "ace") + r.review(false, nil) + n := assign.Nodes{Online: map[string]bool{"ace": true}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}} + r.p.outsideReview(context.Background(), "nats", "src2:"+strings.Repeat("c", 64), []string{"ace"}, n, false, func(string, string) bool { return false }) + if r.events[len(r.events)-1] != meshwants.EventWaitEnded || r.bodies[len(r.bodies)-1].How != outside.HowReplaced { + t.Errorf("events %v", r.events) + } + }) +} + +func TestAMachineWhoseOwnNetworkIsOpenGetsNoWant(t *testing.T) { + m := assign.Module{Name: "nats", FP: fpTest, Built: true, Nodes: []string{"ace"}, MachineHeld: func(node string) bool { return node == "ace" }} + ok, rule, why := assign.Holds(m, "ace", assign.Nodes{Online: map[string]bool{"ace": true}}) + if ok || rule == 0 || !strings.Contains(why, "machine.ace.network") { + t.Errorf("holds = %v %d %q", ok, rule, why) + } + o := newOutsideActs(nil, func() {}, func(string, ...any) {}, 0, 0) + o.Heard("mesh.seat.mesh-conditions.event.condition-raised", []byte(`{"key":"machine.ace.network","subject":{"id":"ace","scope":"machine","machine":"ace"}}`)) + if !o.Held("ace") || o.Held("g14") { + t.Error("the own network condition's event does not hold the machine") + } + o.Heard("mesh.seat.mesh-conditions.event.condition-cleared", []byte(`{"key":"machine.ace.network","subject":{"id":"ace","scope":"machine","machine":"ace"}}`)) + if o.Held("ace") { + t.Error("the machine is held after its condition cleared") + } +} + +func TestTheToolsReadAndReleaseAnOutsideWait(t *testing.T) { + r := newOutsideRig(t) + r.cond("machine.hub.unreachable", "ace") + r.review(false, nil) + var acts []string + s := newPutTools(memStore{waitsState: &memKV{values: map[string][]byte{}}}, func(key string, body any) (json.RawMessage, error) { + acts = append(acts, key) + return json.RawMessage(`{"ok":true,"log":"logged"}`), nil + }) + for k, v := range r.p.w.read(buckets.Waits) { + s.st.put(waitsState, k, v) + } + got, err := s.doWaits(map[string]any{"module": "nats"}) + if err != nil || len(got.(map[string]any)["waits"].([]waitSaid)) != 1 { + t.Fatalf("assigner_waits = %+v %v", got, err) + } + if w := got.(map[string]any)["waits"].([]waitSaid)[0]; w.On != "unreachable:hub" || w.BoundEnds == "" { + t.Errorf("wait = %+v", w) + } + if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "sideways", "why": "x"}, "shanks.node-tools"); err == nil { + t.Error("an unknown how is taken") + } + if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "go-on"}, "shanks.node-tools"); err == nil || len(acts) != 0 { + t.Error("a release with no why is taken") + } + if _, err := s.doRelease(map[string]any{"node": "ace", "module": "nats", "how": "go-on", "why": "the hub is being rebuilt"}, "shanks.node-tools"); err != nil { + t.Fatal(err) + } + if len(acts) == 0 || !strings.Contains(acts[0], "hand-act") { + t.Errorf("the hand act is not recorded first: %v", acts) + } + keys, _ := s.st.keys(waitsState) + raw, _, _, _ := s.st.get(waitsState, keys[0]) + if w, _ := outside.Decode(raw); w.Release != "go-on" || !strings.Contains(w.By, "the hub is being rebuilt") { + t.Errorf("wait = %+v", w) + } + // the process takes the release at its next pass: ended go-on, counted healthy + r.p.w.put(buckets.Waits, keys[0], raw) + n := r.review(false, nil) + if r.bodies[len(r.bodies)-1].How != "go-on" || n.Haves["ace"]["nats"].Health != nodeengineclient.HealthOK { + t.Errorf("last %+v health %s", r.bodies[len(r.bodies)-1], n.Haves["ace"]["nats"].Health) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go new file mode 100644 index 00000000..9d22b933 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/outside_tools.go @@ -0,0 +1,120 @@ +package main + +import ( + "errors" + "fmt" + "slices" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-assigner/internal/buckets" + "mesh-delivery-assigner/internal/outside" +) + +// The tools of the waits on an outside cause (design 47 G §706, §276's row *delivery on an outside cause*): `assigner_waits` reads +// each, the settle included, and `assigner_release` with `how` takes one. The tool writes the release into the wait's record and the +// assigner's process ends the wait at its next pass, saying wait-ended with that how. + +// toolWaits is the tool that reads the outside waits. +const toolWaits = "assigner_waits" + +// waitsState is the assigner's state of waits, as the runtime names a module's own state. +const waitsState = buckets.WaitsState + +// waitSaid is one wait as assigner_waits answers it: its record and where it stands. +type waitSaid struct { + Key string `json:"key"` + outside.Wait + // Settling's end is in the record; Ends is when the bound runs from clock-from, empty for a settle. + BoundEnds string `json:"bound-ends,omitempty"` +} + +// doWaits is the tool assigner_waits. +func (s *putTools) doWaits(args map[string]any) (any, error) { + keys, err := s.st.keys(waitsState) + if err != nil { + return nil, fmt.Errorf("the waits cannot be listed: %w", err) + } + sort.Strings(keys) + out := []waitSaid{} + for _, key := range keys { + raw, _, found, err := s.st.get(waitsState, key) + if err != nil || !found { + continue + } + w, err := outside.Decode(raw) + if err != nil || (arg(args, "node") != "" && w.Node != arg(args, "node")) || (arg(args, "module") != "" && w.Module != arg(args, "module")) { + continue + } + said := waitSaid{Key: key, Wait: w} + if w.Standing() { + if d, err := time.ParseDuration(w.Bound); err == nil { + said.BoundEnds = w.ClockFrom.Add(d).UTC().Format("2006-01-02T15:04:05Z") + } + } + out = append(out, said) + } + return map[string]any{"waits": out}, nil +} + +// doReleaseWait is assigner_release with how: the hand act is recorded first, then the release is written on the wait of node and +// module, which the assigner's process takes at its next pass. +func (s *putTools) doReleaseWait(args map[string]any, by string) (any, error) { + node, module, how, why := arg(args, "node"), arg(args, "module"), arg(args, "how"), arg(args, "why") + if !slices.Contains([]string{outside.HowGoOn, outside.HowFailed, outside.HowRejudge}, how) { + return nil, fmt.Errorf("how is %s, %s or %s, not %q", outside.HowGoOn, outside.HowFailed, outside.HowRejudge, how) + } + if why == "" { + return nil, errors.New("release says why: it is kept in the hand-act log") + } + keys, err := s.st.keys(waitsState) + if err != nil { + return nil, fmt.Errorf("the waits cannot be listed: %w", err) + } + prefix := node + "." + module + "." + var key string + var w outside.Wait + for _, k := range keys { + if !strings.HasPrefix(k, prefix) { + continue + } + raw, _, found, err := s.st.get(waitsState, k) + if err != nil || !found { + continue + } + if c, err := outside.Decode(raw); err == nil && c.Decided == "" { + key, w = k, c + } + } + if key == "" { + return nil, fmt.Errorf("%s holds no wait of %s on an outside cause: assigner_waits lists them", node, module) + } + rec := meshhandactsclient.Recorder{Ask: s.ask} + taken, err := rec.Record(meshhandactsclient.Request{What: fmt.Sprintf("assigner_release %s %s %s", node, module, how), Why: why, + Cause: "outside-wait", Verb: "mesh-wants.assigner_release"}) + if err != nil { + return nil, fmt.Errorf("the hand act could not be recorded, so nothing was released: %w", err) + } + w.Release, w.By, w.Why = how, by+": "+why, why + data, err := outside.Encode(w) + if err != nil { + return nil, err + } + if err := s.st.put(waitsState, key, data); err != nil { + return nil, fmt.Errorf("the release of %s on %s is not written: %w", module, node, err) + } + return map[string]any{"released": key, "how": how, "by": by, "hand-act": taken.Log}, nil +} + +// waitsTool is assigner_waits as the runtime serves it. +func (s *putTools) waitsTool() stdio.Tool { + return stdio.Tool{Name: toolWaits, Description: "Every wait of the delivery on an outside cause and where it stands: a moved module's failing health check on a " + + "machine whose open conditions name a cause outside the delivery (design 47 G §706), with its module, machine, fingerprint, the " + + "condition's key, what it stands `on`, its clock-from, bound, its ask and its state; a settle shows as settling with its end. Narrowed by " + + "machine or module. Read-only.", + Input: schema(map[string]string{"node": "one machine", "module": "one module"}, nil), Run: s.doWaits} +} 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 1999a44b..bad14a3f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -29,6 +29,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "mesh-delivery-assigner/internal/assign" plc "mesh-delivery-assigner/internal/assignment" @@ -96,6 +97,18 @@ type passer struct { // each node's record, as the rules did before the verb. online is, by node, whether the last pass saw it online. reads *enginereads.Keeper online map[string]bool + // outside is the judging of waits on an outside cause (outside.go, design 47 G §706); publishWants says the seat mesh-wants' + // wait-held and wait-ended. Nil: no wait is made. + outside *outsideActs + publishWants publisher +} + +// wantsSay says one event of the seat mesh-wants; with no bus to say it on, nothing is said. +func (p *passer) wantsSay(event string, body any, id string) error { + if p.publishWants == nil { + return nil + } + return p.publishWants(event, body, id) } // engineReads sets what the rules read of each node's engine (ADR 0618 §3, §4): the statement the verb manifest-reads @@ -329,6 +342,10 @@ func (p *passer) wants(ctx context.Context) { for _, node := range m.Nodes { last[node] = p.w.lastHealthy(node, name, fp) } + if p.outside != nil { + m.MachineHeld = p.outside.Held + nodes = p.outsideReview(ctx, name, fp, m.Nodes, nodes, manifests[name].ClaimsSeat(mcseat.Seat), p.movedTo(wanted, nodes)) + } out := assign.Wants(m, nodes, wantedOf[name], last) for _, node := range m.Nodes { for _, prov := range m.MissingProvisions(node) { @@ -384,6 +401,15 @@ func (p *passer) wants(ctx context.Context) { } } +// movedTo says whether the delivery moved owner to node: owner's want there names a fingerprint its have there does not yet run +// healthy (design 47 G §706, *What counts as an outside cause*: a rewritten file of a module the delivery moved is that module's). +func (p *passer) movedTo(wanted map[string]map[string]mwc.Want, n assign.Nodes) func(owner, node string) bool { + return func(owner, node string) bool { + w, ok := wanted[owner][node] + return ok && !w.Removed && !assign.Healthy(n.Haves[node][owner], w.Fp) + } +} + // sayNoWant says in the journal why a node of module gets no want after this pass, from the rule the outcome names for it // (issue 1945: a new assignment whose trunk build was not done was held back with nothing said), once per reason, and says // when the node has a want again. A node with a live want, written before and kept or written in this pass, has one. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 7d9be903..37b9eba8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -214,6 +214,9 @@ func followPut(node, module string) string { // doRelease is the tool assigner_release. func (s *putTools) doRelease(args map[string]any, by string) (any, error) { + if arg(args, "how") != "" { + return s.doReleaseWait(args, by) // a wait on an outside cause (outside_tools.go, design 47 G §706) + } node, module, why := arg(args, "node"), arg(args, "module"), arg(args, "why") key, err := puts.Key(node, module) if err != nil { @@ -344,9 +347,12 @@ func (s *putTools) tools() []stdio.Tool { Run: func(a map[string]any) (any, error) { return s.doPut(a, stdio.CallerOf(a)) }}, {Name: toolRelease, Description: "End a put on one machine (novox/hq ADR 0546): its hold is taken away and the assigner's " + "rules write the machine's want again from their next pass. Nothing changes on the machine until they move it.", - Input: schema(map[string]string{"node": "the machine", "module": "the module", "why": "why, kept with the put"}, + Input: schema(map[string]string{"node": "the machine", "module": "the module", "why": "why, kept with the put", + "how": "with how, the wait of the delivery on an outside cause (assigner_waits) is released, not a put: go-on (the module counts healthy there), " + + "failed (the failing check is the delivery's) or rejudge (judge it again now); written in the hand-act log first"}, []string{"node", "module", "why"}), Run: func(a map[string]any) (any, error) { return s.doRelease(a, stdio.CallerOf(a)) }}, + s.waitsTool(), {Name: toolPuts, Description: "Every put and where it stands (novox/hq ADR 0546): its version, hold, why, each step with " + "its time and words. Narrowed by machine or module. Read-only.", Input: schema(map[string]string{"node": "one machine", "module": "one module"}, nil), diff --git a/modules/mesh-delivery-assigner/internal/assign/asks.go b/modules/mesh-delivery-assigner/internal/assign/asks.go index f73ae7ff..4ed5b876 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks.go @@ -66,6 +66,46 @@ func RestoreAsk(key, node, module, fp, words string, asking int, now time.Time) }) } +// The options of the outside ask (design 47 G §706). +const ( + OptionGoOn = "go-on" + OptionFailed = "failed" +) + +// OutsideAsk is the ask at a wait's bound (design 47 G §706): the move of module on node waits on cause since since, and the +// operator says to go on (the module counts healthy there by the operator's word) or to count it as failed (held as the canary +// rule has it). about is the ask's About, `..`. +func OutsideAsk(key, about, node, module, fp, cause string, since time.Time, asking int, now time.Time) asks.Ask { + return assignerAsk(asking, oc.Ask{ + ID: meshdeliveryclient.AskingID(key, asking), + Headline: fmt.Sprintf("%s on %s waits on %s", module, node, cause), + Explanation: fmt.Sprintf("The health check of %s at %s fails on %s, and the delivery has waited since %s on %s, which is outside it. "+ + "Go on counts %s healthy there by your word and the delivery goes on; count it as failed holds it as a failing check is held.", + module, short(fp), node, since.UTC().Format(time.RFC3339), cause, module), + Options: []oc.Option{ + {ID: OptionGoOn, Label: "Go on", Does: module + " counts healthy on " + node + " by your word and the delivery goes on", Level: occ.Acknowledge}, + {ID: OptionFailed, Label: "Count as failed", Does: "the failing check is the delivery's and is held as the canary rule has it", Level: occ.Acknowledge}, + }, + Who: occ.Operator, + Expires: occ.FormatTime(now.Add(AskLasts)), + OnExpiry: "the delivery keeps waiting", + About: about, + }) +} + +// WaitsOutsideFinding is the finding `delivery-waits-outside` of a wait on an outside cause (design 47 G §706): scope delivery, +// id `.`, severity warning, naming the module, the machine and the cause by its condition's key, and needing +// nothing, so it is sent to no one: the operator's one message is the ask at the bound. +func WaitsOutsideFinding(node, module, causeKey, on string, at time.Time) Observed { + return Observed{Scope: "delivery", ID: node + "." + module, Kind: "delivery-waits-outside", Machine: node, + Severity: "warning", + Summary: fmt.Sprintf("%s on %s waits on %s (the condition %s)", module, node, on, causeKey), + Headline: fmt.Sprintf("%s on %s waits on an outside cause", module, node), + Explanation: fmt.Sprintf("The health check of %s fails on %s while %s is open, which is outside the delivery. The delivery "+ + "waits, judges again when it clears and goes on by itself.", module, node, causeKey), + Needs: "Nothing for you to do.", Source: "assigner.delivery-waits-outside"} +} + // HeldCondition is the condition the assigner raises when it holds a fingerprint of a module (ADR 0429 §6, ADR 0539 §19), or // clears (raised false) when the hold is released. It names the module, the node whose health check failed and the check's // words, under the assigner's own name (ADR 0437 clause 9), keyed by the held fingerprint. needsYou is true for a module holding diff --git a/modules/mesh-delivery-assigner/internal/assign/ended.go b/modules/mesh-delivery-assigner/internal/assign/ended.go index a59e1bbf..52e03a77 100644 --- a/modules/mesh-delivery-assigner/internal/assign/ended.go +++ b/modules/mesh-delivery-assigner/internal/assign/ended.go @@ -29,6 +29,7 @@ var writtenAgainOnEnd = map[string]string{ asks.AskKindHold: "", asks.AskKindRestore: "", asks.AskKindBackup: "", + asks.AskKindOutside: "", // raised again at once while the wait stands (design 47 G §706); the wait's end takes the key away asks.AskKindRecord: "", asks.AskKindAssignments: "a proposal's authorising ask that lapsed ends the proposal (ADR 0567 §5); the verb again asks it again (§11)", } diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go index f11b29b7..7d4774fe 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -95,6 +95,9 @@ type Module struct { // source of a want's Self: every want the rules write or keep is derived from it (selfDerived), so a want written Self while // the node was placed does not outlive the assignments (issue 1126). Nil: no node does. Self func(node string) bool + // MachineHeld says whether the node's own network condition is open, which holds the machine as a whole: no want moves to it + // until that clears (design 47 G §706 *M's own network*). Nil: none is held. + MachineHeld func(node string) bool } func (m Module) selfOn(node string) bool { return m.Self != nil && m.Self(node) } @@ -126,6 +129,12 @@ func healthy(h nodeengine.HaveByNodeModuleValue, fp string) bool { return h.Fp == fp && h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK } +// Healthy says whether a have runs fp healthy. +func Healthy(h nodeengine.HaveByNodeModuleValue, fp string) bool { return healthy(h, fp) } + +// Failed says whether a have's health check failed once the node is done moving. +func Failed(h nodeengine.HaveByNodeModuleValue) bool { return failed(h) } + // Canary is the module's canary: the first of its nodes, by name, that is online (ADR 0435 §3; one ring, ordered by // name, ADR 0528 clause 3). Empty when none is online. func Canary(nodes []string, online map[string]bool) string { @@ -155,6 +164,8 @@ func holds(m Module, node string, n Nodes, ordering bool) (bool, int, string) { return false, mwc.RuleBuildDone, fmt.Sprintf("the build of %s at %s is not done", m.Name, short(m.FP)) case m.Held: return false, mwc.RuleNotHeld, fmt.Sprintf("%s is held at %s", m.Name, short(m.FP)) + case m.MachineHeld != nil && m.MachineHeld(node): + return false, mwc.RuleNotHeld, fmt.Sprintf("the condition machine.%s.network is open: no want moves to %s until its own network is healthy", node, node) } if ordering { canary := Canary(m.Nodes, n.Online) diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 5e56fd5e..ad8514ed 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -22,6 +22,9 @@ const ( Differences = plc.Module + "_" + wants.StateDifferences Asks = plc.Module + "_asks" Puts = plc.Module + "_puts" + // Waits holds the waits on an outside cause, one key `..` (design 47 G §706). + WaitsState = "waits" + Waits = plc.Module + "_" + WaitsState // PoliciesState is the state of the policies a person set for a module's upgrade; Policies its bucket. PoliciesState = "policies" Policies = plc.Module + "_" + PoliciesState diff --git a/modules/mesh-delivery-assigner/internal/outside/outside.go b/modules/mesh-delivery-assigner/internal/outside/outside.go new file mode 100644 index 00000000..f7b4209e --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/outside/outside.go @@ -0,0 +1,313 @@ +// Package outside is the assigner's wait on an outside cause (design 47 G §706, ADR 0677): a moved module's failing health +// check on a machine whose open conditions name a cause outside the delivery is held as a wait, judged again on every change +// that can end it, and ended by itself or by the operator. The package holds the rules and the kept record; the assigner's +// process reads the conditions, says the events and writes the asks. +package outside + +import ( + "encoding/json" + "fmt" + "slices" + "strings" + "time" +) + +// The ways a wait ends, as `wait-ended` says them (§706 *How it ends*). +const ( + HowHealthy = "healthy" + HowCleared = "cleared" + HowOwnFailure = "own-failure" + HowGoOn = "go-on" + HowFailed = "failed" + HowReplaced = "replaced" + // HowRejudge is a release by hand that asks to judge the wait again now; it ends nothing. + HowRejudge = "rejudge" +) + +// OnUnread is what a wait stands on while the keeper does not answer. +const OnUnread = "conditions-unread" + +// The defaults of the settings (§633): `outside-wait-minutes` and `outside-settle-seconds`. +const ( + DefaultBound = 30 * time.Minute + DefaultSettle = 90 * time.Second +) + +// Cond is an open condition as the keeper's verb `conditions` answers it: its key and the subject's machine and `also`. +type Cond struct { + Key string + Machine string + Also []string +} + +// ParseConditions reads the keeper's answer of `conditions`. +func ParseConditions(answer []byte) ([]Cond, error) { + var a struct { + Conditions []struct { + Key string `json:"key"` + Subject struct { + Machine string `json:"machine"` + Also []string `json:"also"` + } `json:"subject"` + } `json:"conditions"` + } + if err := json.Unmarshal(answer, &a); err != nil { + return nil, fmt.Errorf("outside: the keeper's conditions do not read: %w", err) + } + out := make([]Cond, 0, len(a.Conditions)) + for _, c := range a.Conditions { + out = append(out, Cond{Key: c.Key, Machine: c.Subject.Machine, Also: c.Subject.Also}) + } + return out, nil +} + +// Cause is an outside cause open for a machine: the condition's key and what a wait says it stands `on`. +type Cause struct{ Key, On string } + +// Causes are the outside causes open for machine m among conds, sorted by key (§706 *What counts as an outside cause*): +// `machine.M..rewritten` or `machine.M.rewritten` unless the delivery moved to M (moved), and +// `machine..unreachable`, `.network` or `.silent` of another machine whose `also` lists M. No other condition is one. +func Causes(conds []Cond, m string, moved func(owner string) bool) []Cause { + var out []Cause + for _, c := range conds { + p := strings.Split(c.Key, ".") + if len(p) < 3 || p[0] != "machine" { + continue + } + switch { + case len(p) == 3 && p[1] == m && p[2] == "rewritten": + out = append(out, Cause{c.Key, "rewritten"}) + case len(p) == 4 && p[1] == m && p[3] == "rewritten" && (moved == nil || !moved(p[2])): + out = append(out, Cause{c.Key, "rewritten:" + p[2]}) + case len(p) == 3 && p[1] != m && slices.Contains([]string{"unreachable", "network", "silent"}, p[2]) && slices.Contains(c.Also, m): + out = append(out, Cause{c.Key, p[2] + ":" + p[1]}) + } + } + slices.SortFunc(out, func(a, b Cause) int { return strings.Compare(a.Key, b.Key) }) + return out +} + +// OwnNetworkKey is the key of a machine's own open network condition, which holds the machine as a whole (§706 *M's own +// network*). +func OwnNetworkKey(m string) string { return "machine." + m + ".network" } + +// OwnNetwork are the machines whose own network condition is open among conds. +func OwnNetwork(conds []Cond) map[string]bool { + out := map[string]bool{} + for _, c := range conds { + if p := strings.Split(c.Key, "."); len(p) == 3 && p[0] == "machine" && p[2] == "network" && c.Key == OwnNetworkKey(p[1]) { + out[p[1]] = true + } + } + return out +} + +// Wait is one outside wait as the assigner keeps it: the move of Module at FP on Node, or its settle (Settling). Decided is +// what ended it for the rest of that fingerprint on that node: `go-on` (counted healthy by the operator's word) or `failed` +// (the delivery's failure), so a check that goes on failing is not waited on again. +type Wait struct { + V int `json:"v"` + Node string `json:"node"` + Module string `json:"module"` + FP string `json:"fp"` + On string `json:"on,omitempty"` + Cause string `json:"cause,omitempty"` + Since time.Time `json:"since"` + ClockFrom time.Time `json:"clock-from"` + Bound string `json:"bound,omitempty"` + // Settling: no outside cause was open at the failing check; the judgement is held until SettleEnds. + Settling bool `json:"settling,omitempty"` + SettleEnds time.Time `json:"settle-ends,omitempty"` + // Ask is the id of the asking made at the bound, and State where the wait stands in words (settling, waiting, asked). + Ask string `json:"ask,omitempty"` + State string `json:"state,omitempty"` + // Release is a release by hand not taken up yet: go-on, failed or rejudge, by By and why Why. + Release string `json:"release,omitempty"` + By string `json:"by,omitempty"` + Why string `json:"why,omitempty"` + // Decided is `go-on` or `failed` once the wait ended so. + Decided string `json:"decided,omitempty"` +} + +// Key is the key of a wait in the state: `..`. +func Key(node, module, hex string) string { return node + "." + module + "." + hex } + +// Encode is w as JSON. +func Encode(w Wait) ([]byte, error) { w.V = 1; return json.Marshal(w) } + +// Decode reads a kept wait. +func Decode(raw []byte) (Wait, error) { + var w Wait + if err := json.Unmarshal(raw, &w); err != nil { + return w, err + } + if w.V != 1 || w.Node == "" || w.Module == "" { + return w, fmt.Errorf("outside: %q is not a kept outside wait", raw) + } + return w, nil +} + +// Standing says a wait is said (`wait-held`) and not only settling, nor decided. +func (w Wait) Standing() bool { return !w.Settling && w.Decided == "" } + +// Facts are what a judging reads: the module's check on the node at the wait's fingerprint (Failing, Healthy), the keeper's +// conditions as read now (Read false: the kept call is unanswered), whether the module holds the seat mesh-conditions itself, +// and the settings. +type Facts struct { + Now time.Time + Failing bool + Healthy bool + Read bool + Conds []Cond + Node string + // Moved says the delivery moved the module to the node. + Moved func(owner string) bool + // CauseCleared: the condition the wait stands on was heard cleared (the keeper's event), so a healthy module ends `cleared`. + CauseCleared bool + // KeeperSelf: the module holds the seat mesh-conditions, which cannot answer for its own failing check. + KeeperSelf bool + Bound time.Duration + Settle time.Duration +} + +// Decision is what a judging comes to. +type Decision struct { + // Wait is the record to keep: nil when none stands (none is kept; a decided one is kept with Decided). + Wait *Wait + // Held says wait-held is to be said (a new wait, or a new `on`); End is the `how` of a wait-ended to say, "" for none. + Held bool + End string + // Own: the failure is the delivery's (the rules hold it as the canary rule has it). Counted: the module counts healthy + // there by the operator's word. Waiting: the move waits (the rules see no failure, and the canary holds the rest). + Own, Counted, Waiting bool + // Ask: the bound has run from clock-from, so the operator is asked. + Ask bool + // Due is when the judging is due again by itself (the settle's end); zero for none. + Due time.Time +} + +// Judge is one judging of the check at the node (§706). prev is the kept wait of that module at that fingerprint, nil for none. +func Judge(prev *Wait, node, module, fp string, f Facts) Decision { + if prev != nil && prev.Decided != "" { + // Decided by the operator: it holds for this fingerprint until the module runs healthy. + if f.Healthy { + return Decision{} + } + d := Decision{Wait: prev, Own: prev.Decided == HowFailed, Counted: prev.Decided == HowGoOn} + return d + } + if prev != nil && prev.Release != "" && prev.Release != HowRejudge { + how := prev.Release + w := *prev + w.Release, w.Decided, w.Settling, w.Ask = "", how, false, "" + d := Decision{Wait: &w, Own: how == HowFailed, Counted: how == HowGoOn} + if prev.Standing() { + d.End = how + } + return d + } + if f.Healthy { + if prev == nil { + return Decision{} + } + d := Decision{} + if prev.Standing() { + d.End = HowHealthy + if prev.Cause != "" && (f.CauseCleared || (f.Read && !openKey(f.Conds, prev.Cause))) { + d.End = HowCleared + } + } + return d + } + if !f.Failing { + // still moving: nothing is judged, what stands stands + if prev == nil { + return Decision{} + } + return Decision{Wait: prev, Waiting: prev.Standing()} + } + if f.KeeperSelf { + d := Decision{Own: true} + if prev != nil && prev.Standing() { + d.End = HowOwnFailure + } + return d + } + next := Wait{Node: node, Module: module, FP: fp, Bound: f.Bound.String(), Since: f.Now, ClockFrom: f.Now} + standing := prev != nil && prev.Standing() + if standing { + next = *prev + next.Release = "" + } + if !f.Read { + // The keeper does not answer. A judging that has only just asked holds, as the settle does; once the settle has run + // from the failing check with no answer, the move waits on conditions-unread (§706 *When the keeper does not answer*). + if standing { + return stand(prev, *prev, f) // a wait that stands keeps its cause until the keeper answers + } + { + if prev != nil && prev.Settling { + next = *prev + } else { + next.Settling, next.SettleEnds, next.State = true, f.Now.Add(f.Settle), "settling" + } + if f.Now.Before(next.SettleEnds) { + return Decision{Wait: &next, Waiting: true, Due: next.SettleEnds} + } + next.Settling, next.SettleEnds = false, time.Time{} + } + next.On, next.Cause, next.Settling, next.State = OnUnread, "", false, "waiting" + return stand(prev, next, f) + } + causes := Causes(f.Conds, node, f.Moved) + if len(causes) > 0 { + c := causes[0] + if standing { + for _, k := range causes { + if k.Key == prev.Cause { + c = k + } + } + } + next.On, next.Cause, next.Settling, next.SettleEnds, next.State = c.On, c.Key, false, time.Time{}, "waiting" + return stand(prev, next, f) + } + if standing { + // judged again, failing, no outside cause open: the delivery's failure (after the settle the check has long had) + w := *prev + w.Decided, w.Release, w.Ask = HowOwnFailure, "", "" + return Decision{Wait: &w, End: HowOwnFailure, Own: true} + } + // no outside cause open: settle from the failing check, judged again when it ends + if prev != nil && prev.Settling { + if !f.Now.Before(prev.SettleEnds) { + w := *prev + w.Settling, w.Decided = false, HowOwnFailure + return Decision{Wait: &w, Own: true} + } + return Decision{Wait: prev, Waiting: true, Due: prev.SettleEnds} + } + next.Settling, next.SettleEnds, next.State, next.On, next.Cause = true, f.Now.Add(f.Settle), "settling", "", "" + return Decision{Wait: &next, Waiting: true, Due: next.SettleEnds} +} + +// stand is the decision of a wait that stands on next.On: said when it is new or stands on another cause, its clock kept. +func stand(prev *Wait, next Wait, f Facts) Decision { + d := Decision{Wait: &next, Waiting: true} + if prev == nil || !prev.Standing() || prev.On != next.On { + d.Held = true + } + if prev != nil && prev.Standing() { + next.Ask = prev.Ask + } + if f.Bound > 0 && !f.Now.Before(next.ClockFrom.Add(f.Bound)) { + d.Ask = true + next.State = "asked" + } + d.Wait = &next + return d +} + +func openKey(conds []Cond, key string) bool { + return slices.ContainsFunc(conds, func(c Cond) bool { return c.Key == key }) +} diff --git a/modules/mesh-delivery-assigner/internal/outside/outside_test.go b/modules/mesh-delivery-assigner/internal/outside/outside_test.go new file mode 100644 index 00000000..eb39c173 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/outside/outside_test.go @@ -0,0 +1,169 @@ +package outside + +import ( + "slices" + "testing" + "time" +) + +var t0 = time.Date(2026, 11, 11, 10, 0, 0, 0, time.UTC) + +func TestCauses(t *testing.T) { + conds := []Cond{ + {Key: "machine.ace.rewritten"}, + {Key: "machine.ace.systemd-resolved.rewritten"}, + {Key: "machine.ace.nats.rewritten"}, + {Key: "machine.hub.unreachable", Also: []string{"ace", "g14"}}, + {Key: "machine.lap.unreachable", Also: []string{"g14"}}, + {Key: "machine.hub2.network", Also: []string{"ace"}}, + {Key: "machine.hub3.silent", Also: []string{"ace"}}, + {Key: "machine.ace.network", Also: []string{"ace"}}, + {Key: "module.nats.ace.unhealthy", Machine: "ace"}, + } + moved := func(o string) bool { return o == "nats" } + var got []string + for _, c := range Causes(conds, "ace", moved) { + got = append(got, c.On) + } + want := []string{"unreachable:hub", "network:hub2", "silent:hub3", "rewritten", "rewritten:systemd-resolved"} + slices.Sort(got) + slices.Sort(want) + if !slices.Equal(got, want) { + t.Errorf("causes = %v, want %v (a rewritten naming a moved module, an unreachable not listing ace and ace's own network are none)", got, want) + } + if own := OwnNetwork(conds); !own["ace"] || !own["hub2"] || own["hub"] { + t.Errorf("own network = %v", own) + } +} + +func facts(now time.Time, conds ...Cond) Facts { + return Facts{Now: now, Failing: true, Read: true, Conds: conds, Node: "ace", Bound: 30 * time.Minute, Settle: 90 * time.Second} +} + +func TestJudge(t *testing.T) { + hub := Cond{Key: "machine.hub.unreachable", Also: []string{"ace"}} + hubNet := Cond{Key: "machine.hub.network", Also: []string{"ace"}} + step := func(prev *Wait, f Facts) (Decision, *Wait) { + d := Judge(prev, "ace", "nats", "src2:aa", f) + return d, d.Wait + } + tests := []struct { + name string + run func(t *testing.T) + }{ + {"a failing check with a cause open is a wait said at once, ended cleared on its clearing", func(t *testing.T) { + d, w := step(nil, facts(t0, hub)) + if !d.Held || !d.Waiting || w.On != "unreachable:hub" || w.Bound != "30m0s" || !w.ClockFrom.Equal(t0) || d.Own { + t.Fatalf("decision %+v wait %+v", d, w) + } + f := facts(t0.Add(time.Minute)) + f.Failing, f.Healthy = false, true + if d, _ := step(w, f); d.End != HowCleared || d.Wait != nil { + t.Errorf("healthy once cleared: %+v", d) + } + f.Conds = []Cond{hub} + if d, _ := step(w, f); d.End != HowHealthy { + t.Errorf("healthy with the cause open: %+v", d) + } + }}, + {"a cause that does not list the machine leaves the failure to settle, then the delivery's", func(t *testing.T) { + d, w := step(nil, facts(t0, Cond{Key: "machine.hub.unreachable", Also: []string{"g14"}})) + if d.Held || !w.Settling || !w.SettleEnds.Equal(t0.Add(90*time.Second)) || !d.Waiting || d.Own { + t.Fatalf("settle: %+v %+v", d, w) + } + if d, _ := step(w, facts(t0.Add(89*time.Second))); !d.Waiting || d.Own { + t.Errorf("within the settle: %+v", d) + } + if d, _ := step(w, facts(t0.Add(90*time.Second))); !d.Own || d.Waiting { + t.Errorf("past the settle: %+v", d) + } + }}, + {"a condition raised within the settle turns it into a wait", func(t *testing.T) { + _, w := step(nil, facts(t0)) + d, w2 := step(w, facts(t0.Add(30*time.Second), hub)) + if !d.Held || w2.Settling || w2.On != "unreachable:hub" { + t.Fatalf("%+v %+v", d, w2) + } + }}, + {"a cause that turns into another keeps the wait and its clock", func(t *testing.T) { + _, w := step(nil, facts(t0, hub)) + d, w2 := step(w, facts(t0.Add(5*time.Minute), hubNet)) + if !d.Held || w2.On != "network:hub" || !w2.ClockFrom.Equal(t0) || d.End != "" { + t.Fatalf("%+v %+v", d, w2) + } + }}, + {"judged again failing with no cause open ends own-failure", func(t *testing.T) { + _, w := step(nil, facts(t0, hub)) + d, w2 := step(w, facts(t0.Add(time.Minute))) + if d.End != HowOwnFailure || !d.Own || w2.Decided != HowOwnFailure { + t.Fatalf("%+v", d) + } + }}, + {"the keeper silent holds as the settle does, then waits on conditions-unread, and is judged when it answers", func(t *testing.T) { + f := facts(t0) + f.Read = false + d, w := step(nil, f) + if d.Held || !w.Settling || !d.Waiting { + t.Fatalf("%+v %+v", d, w) + } + f.Now = t0.Add(90 * time.Second) + d, w = step(w, f) + if !d.Held || w.On != OnUnread || w.Settling || !w.ClockFrom.Equal(t0) { + t.Fatalf("%+v %+v", d, w) + } + if d, w2 := step(w, facts(t0.Add(100*time.Second), hub)); !d.Held || w2.On != "unreachable:hub" || !w2.ClockFrom.Equal(w.ClockFrom) { + t.Errorf("answered: %+v %+v", d, w2) + } + }}, + {"the keeper's own module failing is its failure with no read", func(t *testing.T) { + f := facts(t0, hub) + f.KeeperSelf = true + if d, w := step(nil, f); !d.Own || w != nil { + t.Fatalf("%+v", d) + } + }}, + {"past the bound the operator is asked; Go on counts it healthy, Count as failed is the delivery's", func(t *testing.T) { + _, w := step(nil, facts(t0, hub)) + d, w := step(w, facts(t0.Add(30*time.Minute), hub)) + if !d.Ask || d.Held || d.End != "" { + t.Fatalf("%+v", d) + } + for how, want := range map[string]Decision{HowGoOn: {Counted: true}, HowFailed: {Own: true}} { + w2 := *w + w2.Release = how + d, kept := step(&w2, facts(t0.Add(31*time.Minute), hub)) + if d.End != how || d.Counted != want.Counted || d.Own != want.Own || kept.Decided != how { + t.Errorf("%s: %+v", how, d) + } + // decided: a check that goes on failing is not waited on again, until it runs healthy + if d, _ := step(kept, facts(t0.Add(32*time.Minute), hub)); d.Held || d.Counted != want.Counted || d.Own != want.Own { + t.Errorf("%s again: %+v", how, d) + } + } + }}, + {"a release to judge again changes nothing by itself", func(t *testing.T) { + _, w := step(nil, facts(t0, hub)) + w.Release = HowRejudge + if d, w2 := step(w, facts(t0.Add(time.Minute))); d.End != HowOwnFailure || w2.Release != "" { + t.Fatalf("%+v", d) + } + }}, + } + for _, tt := range tests { + t.Run(tt.name, tt.run) + } +} + +func TestWaitRoundTrip(t *testing.T) { + raw, err := Encode(Wait{Node: "ace", Module: "nats", FP: "src2:aa", On: "silent:g14", Since: t0, ClockFrom: t0}) + if err != nil { + t.Fatal(err) + } + w, err := Decode(raw) + if err != nil || w.On != "silent:g14" || !w.ClockFrom.Equal(t0) { + t.Fatalf("%+v %v", w, err) + } + if _, err := Decode([]byte(`{"v":2}`)); err == nil { + t.Error("an unknown version is read") + } +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 0430440d..87417cc8 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -1,6 +1,7 @@ package manifest import ( + "encoding/json" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-teller/tellerevents" @@ -65,7 +66,7 @@ func TestManifest(t *testing.T) { } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1, - "engine-reads": 1, "written": 5}; !reflect.DeepEqual(states, want) { + "engine-reads": 1, "written": 5, "waits": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -98,7 +99,7 @@ func TestManifest(t *testing.T) { // ADR 0621 §1: the first-view tools assignments, seat-holders and wants and design 25 § 12.2's six read tools, of the // program's bundle, which runs in a runtime of its own on the assigner's account; ADR 0618 §3: and assigner_engine_reads, // which lists and asks again what each node's engine reads. - if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants", + if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assigner_waits", "assignments", "seat-holders", "wants", "assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies", "assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again", "assigner_ask_again", "assigner_written"}) { t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads, the seat list's two (ADR 0636 §7) and assigner_ask_again (ADR 0668 §5)", m.Tools) @@ -128,17 +129,32 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + "seat:mesh-conditions.conditions", // design 47 G §706: the conditions read at each judging of an outside wait "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", "seat:node-engine.manifest-reads", // ADR 0618 §3 "seat:operator-channel.asks"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0668 §4: the ended asks at bind t.Errorf("invokes = %v, want %v", m.Invokes, want) } + // design 47 G §633, §706: the bounds of the wait on an outside cause are `config` keys filling the process's environment. + var raw struct { + Config map[string]struct { + Default any `json:"default"` + } `json:"config"` + } + file, _ := os.ReadFile("module.json") + if err := json.Unmarshal(file, &raw); err != nil || raw.Config["outside-wait-minutes"].Default != float64(30) || raw.Config["outside-settle-seconds"].Default != float64(90) || len(raw.Config) != 2 { + t.Errorf("config = %+v (%v), want outside-wait-minutes of 30 and outside-settle-seconds of 90", raw.Config, err) + } + if !strings.Contains(string(file), `"MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES": "${config:outside-wait-minutes}"`) || + !strings.Contains(string(file), `"MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS": "${config:outside-settle-seconds}"`) { + t.Errorf("the process's env does not carry the two settings from config") + } if want := []string{"assignments-copied", "put", "views.asks", "views.copies", "views.differences", "views.findings", "views.assignments", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) { t.Errorf("emits = %v, want the copy, put and its six views", m.Emits) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index f9d921a5..eda7c859 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -43,6 +43,9 @@ ], "consumes": [ "mesh-conditions.findings-asked", + "mesh-conditions.condition-raised", + "mesh-conditions.condition-changed", + "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", @@ -59,6 +62,7 @@ "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", + "seat:mesh-conditions.conditions", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", @@ -90,6 +94,7 @@ "assigner_put", "assigner_release", "assigner_puts", + "assigner_waits", "assignments", "seat-holders", "wants", @@ -125,8 +130,27 @@ { "name": "view-marks", "history": 1 }, { "name": "first-views", "history": 1 }, { "name": "engine-reads", "history": 1 }, - { "name": "written", "history": 5 } + { "name": "written", "history": 5 }, + { "name": "waits", "history": 5 } ], + "config": { + "outside-wait-minutes": { + "type": "number", + "description": "minutes a moved module's failing check waits on an outside cause before the operator is asked to go on or count it as failed (design 47 G section 706); a bound a whole test sets in seconds.", + "minimum": 1, + "maximum": 10080, + "trusted": false, + "default": 30 + }, + "outside-settle-seconds": { + "type": "number", + "description": "seconds the assigner holds its judgement of a failing check when no outside cause is open yet, since the node-engine states the failing check and the network in one report while the condition rises only after the watchman's two looks and the keeper's take-in (design 47 G section 706).", + "minimum": 1, + "maximum": 3600, + "trusted": false, + "default": 90 + } + }, "objects": [ { "name": "assignment-revisions", "max-mib": 512, "form": 1 } ], @@ -159,7 +183,9 @@ "./mesh-delivery-assigner" ], "env": { - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_ASSIGNER_OUTSIDE_WAIT_MINUTES": "${config:outside-wait-minutes}", + "MESH_ASSIGNER_OUTSIDE_SETTLE_SECONDS": "${config:outside-settle-seconds}" }, "user": "mesh-delivery-assigner", "health": { From 00ec7af34e4250d6b9f933400448472a58a60a80 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:48:20 +0100 Subject: [PATCH 4/9] Assigner: serves the seat mesh-wants' verb input and reads the kept inputs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §354, §338 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/inputs.go | 117 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/inputs_test.go | 45 +++++++ .../cmd/mesh-delivery-assigner/main.go | 10 +- .../mesh-delivery-assigner/seat_tools_test.go | 2 +- .../cmd/mesh-delivery-assigner/tools.go | 1 + .../cmd/mesh-delivery-assigner/wants_tools.go | 3 +- .../internal/buckets/buckets.go | 3 + .../mesh-delivery-assigner/manifest_test.go | 6 +- modules/mesh-delivery-assigner/module.json | 7 +- 9 files changed, 185 insertions(+), 9 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go new file mode 100644 index 00000000..0bbb0bf7 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go @@ -0,0 +1,117 @@ +package main + +import ( + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// A step's input by its digest (design 47 G §354): the assigner keeps the input of the composition step it asked a build seat to +// run in its state `inputs`, one key `