From b5fdcb88c8de72e6c73ab51f437f0afa1ed047e3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 22:12:08 +0100 Subject: [PATCH] Say checked, healer-acted and act-recorded on their seats' subjects, where their consumers hear them, and pin mesh-sdk fbd00c8 (issues 1952, 1953) stdio.Emit put them on the module's subject, which no consumer of a seat's event hears. The manifests no longer list a seat's events as the module's own, which the SDK now refuses. mesh-hand-acts takes its act layout and names from mesh-sdk's handacts, and counts a repeated cause by its Repair rule. --- checks/health/go.mod | 2 +- checks/health/go.sum | 4 +- .../mesh-hand-acts/cmd/mesh-hand-acts/acts.go | 106 +++++------------- .../cmd/mesh-hand-acts/acts_test.go | 4 +- .../mesh-hand-acts/cmd/mesh-hand-acts/main.go | 2 +- .../cmd/mesh-hand-acts/manifest_test.go | 3 +- .../mesh-hand-acts/whole_integration_test.go | 9 +- modules/mesh-hand-acts/go.mod | 2 +- modules/mesh-hand-acts/go.sum | 4 +- modules/mesh-hand-acts/module.json | 3 - .../mesh-healers/cmd/mesh-healers/healers.go | 13 ++- .../cmd/mesh-healers/healers_test.go | 4 +- modules/mesh-healers/cmd/mesh-healers/main.go | 6 +- .../cmd/mesh-healers/manifest_test.go | 2 +- .../mesh-healers/whole_integration_test.go | 9 +- modules/mesh-healers/go.mod | 2 +- modules/mesh-healers/go.sum | 4 +- modules/mesh-healers/module.json | 3 - .../cmd/mesh-self-check/main.go | 4 +- .../cmd/mesh-self-check/manifest_test.go | 3 +- .../cmd/mesh-self-check/selfcheck.go | 14 ++- .../cmd/mesh-self-check/selfcheck_test.go | 11 +- .../mesh-self-check/whole_integration_test.go | 9 +- modules/mesh-self-check/go.mod | 2 +- modules/mesh-self-check/go.sum | 4 +- modules/mesh-self-check/module.json | 3 - 26 files changed, 102 insertions(+), 130 deletions(-) diff --git a/checks/health/go.mod b/checks/health/go.mod index 30f01e1d..68e8f5fc 100644 --- a/checks/health/go.mod +++ b/checks/health/go.mod @@ -2,4 +2,4 @@ module checks/health go 1.26.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c diff --git a/checks/health/go.sum b/checks/health/go.sum index aacdf9f6..d598087e 100644 --- a/checks/health/go.sum +++ b/checks/health/go.sum @@ -1,2 +1,2 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go index 01e09440..6e8579e8 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go @@ -25,62 +25,35 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/handacts" ) -// Module and Seat are this module and the seat it holds once for the mesh. +// The seat, the log, the states and the event, as mesh-sdk's handacts names them. const ( - Module = "mesh-hand-acts" - Seat = "mesh-hand-acts" -) - -// The states it keeps and reads, and the event it says. -const ( - ActsLog = "acts" - WarrantsState = "warrants" - // ControllerActs is the controller's log, read until its lease ends. - ControllerActs = "mesh-controller.hand-acts" + Module = handacts.Module + Seat = handacts.Seat + ActsLog = handacts.ActsLog + WarrantsState = handacts.WarrantsState + ControllerActs = handacts.ControllerActs // RouterRecords is the router's record of every ask, `.` (ADR 0259): the warrant is read there. RouterRecords = "messenger.asks" - RecordedEvent = "act-recorded" -) + RecordedEvent = handacts.EventRecorded + // RecordedSubject is where the seat says it: a seat's events are on mesh.seat..event., which its + // consumers hear (catalogue.SeatSays). + RecordedSubject = "mesh.seat." + Seat + ".event." + RecordedEvent -// The verbs an act is recorded under, as the controller's log names them: what S15 reads to tell a repair from a -// person's decision by design. -const ( - VerbRecord = "hand-act record" - VerbDrill = "hand-act drill" - VerbWarrant = "warrant" - // CauseDrill is a drill's cause, and the one cause a repair may not give. - CauseDrill = "drill" - // CauseOperatorAnswer is an act the operator chose on a warrant. - CauseOperatorAnswer = "operator-answer" + VerbRecord = handacts.VerbRecord + VerbDrill = handacts.VerbDrill + VerbWarrant = handacts.VerbWarrant + CauseDrill = handacts.CauseDrill + CauseOperatorAnswer = handacts.CauseOperatorAnswer ) // RepeatedWithin is how far back a cause counts as repeated (S15). const RepeatedWithin = 14 * 24 * time.Hour -// Act is one entry of the log, in the layout of the controller's (mesh-controller internal/link HandAct), so one -// reader reads both logs alike. **A copy of the layout mesh-sdk holds for an act (issue 1953, L5-S2)**, replaced by it -// when it lands. -type Act struct { - ID string `json:"id"` - At time.Time `json:"at"` - By string `json:"by"` - Verb string `json:"verb"` - Args []string `json:"arguments,omitempty"` - Why string `json:"why"` - Cause string `json:"cause"` - Condition string `json:"condition,omitempty"` - Kind string `json:"kind,omitempty"` - Carried []string `json:"carried,omitempty"` - Via string `json:"via,omitempty"` - Ask string `json:"ask,omitempty"` - Proofs []string `json:"proofs,omitempty"` - RequestedBy string `json:"requested-by,omitempty"` - Outcome string `json:"outcome,omitempty"` - // Log is which log holds it, set when it is read: this seat's, or the controller's. Never written. - Log string `json:"log,omitempty"` -} +// Act is one entry of the log: mesh-sdk's handacts.Act, the controller's HandAct unchanged. +type Act = handacts.Act // Store is the bus's buckets and this module's log as it reaches them, through its runtime. Create writes a key only // when it has none, and says ErrExists otherwise. @@ -109,9 +82,10 @@ var ErrExists = errors.New("the key was made by another writer first") // Log is mesh-hand-acts: the log and what records into it. type Log struct { Store Store - Emit func(event string, body any) error - Now func() time.Time - Logf func(format string, args ...any) + // Publish says a seat event on its subject, de-duplicated by id (stdio.SeatPublish). + Publish func(subject string, body any, id string) error + Now func() time.Time + Logf func(format string, args ...any) } var seq atomic.Uint64 @@ -170,8 +144,8 @@ func (l *Log) record(act Act) error { if err := l.Store.Append(ActsLog, act.ID, act); err != nil { return fmt.Errorf("the act could not be recorded: %w", err) } - if l.Emit != nil { - if err := l.Emit(RecordedEvent, act); err != nil { + if l.Publish != nil { + if err := l.Publish(RecordedSubject, act, act.ID); err != nil { l.Logf("[%s] %s is recorded and could NOT be said as %s: %v", Module, act.ID, RecordedEvent, err) } } @@ -326,7 +300,7 @@ func (l *Log) HandActs(days int, cause, who string) Listing { all = append(all, acts...) } for _, a := range all { - if now.Sub(a.At) <= RepeatedWithin && !decision(a) { + if now.Sub(a.At) <= RepeatedWithin && a.Repair() { out.Repeated[a.Cause]++ } if a.At.Before(since) || (cause != "" && a.Cause != cause) || (who != "" && !strings.Contains(a.By, who)) { @@ -364,7 +338,7 @@ func (l *Log) read(log string, since time.Time) ([]Act, int, error) { return nil, 0, err } for _, raw := range entries { - keep(raw, Seat) + keep(raw, handacts.LogSeat) } return out, len(entries), nil } @@ -381,7 +355,7 @@ func (l *Log) read(log string, since time.Time) ([]Act, int, error) { return nil, len(keys), err } if found { - keep(raw, log) + keep(raw, handacts.LogController) } } return out, len(keys), nil @@ -399,29 +373,3 @@ func keyTime(key string) (time.Time, bool) { } return time.Unix(0, ns), true } - -// decided are the verbs whose acts are a person's decision by design, with the causes they hold for (every cause when -// none): the controller's handActVerbs as S15 reads them. A verb not listed is a repair. -var decided = map[string][]string{ - VerbWarrant: nil, "seat": {"seat-passed"}, "conditions silence": {CauseOperatorAnswer}, VerbRecord: {CauseDrill}, - "secret issued": nil, VerbDrill: nil, "retire approve": nil, "retire reject": nil, "cleanup delete": nil, - "dead-letters deliver": nil, "dead-letters drop": nil, "collect": nil, "mirrors": nil, "move-placement": nil, - "move-config": nil, "bus upgrade": nil, "upgrade release-backlog": nil, "secret accept": nil, - "secret rotate": {"leaked-in-logs"}, "rotate": {"leaked-in-logs"}, -} - -func decision(a Act) bool { - causes, ok := decided[a.Verb] - if !ok { - return false - } - if len(causes) == 0 { - return true - } - for _, c := range causes { - if c == a.Cause { - return true - } - } - return false -} diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go index b2762666..73d42cef 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go @@ -127,8 +127,8 @@ func newLog() (*Log, *memStore, *[]Act, *time.Time) { var said []Act now := time.Date(2026, 10, 29, 22, 0, 0, 0, time.UTC) l := &Log{Store: store, Logf: func(string, ...any) {}, Now: func() time.Time { return now }, - Emit: func(event string, body any) error { - if a, ok := body.(Act); ok && event == RecordedEvent { + Publish: func(subject string, body any, id string) error { + if a, ok := body.(Act); ok && subject == "mesh.seat.mesh-hand-acts.event.act-recorded" && id == a.ID { said = append(said, a) } return nil diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go index 0023c53c..b103fa23 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go @@ -76,7 +76,7 @@ func (runtimeStore) Fill(log string) (Fill, error) { func main() { l := &Log{Store: runtimeStore{}, Now: time.Now, Logf: logf, - Emit: func(event string, body any) error { return stdio.Emit(event, body) }} + Publish: stdio.SeatPublish} if err := stdio.Serve("", Tools(l)); err != nil { logf("%v", err) os.Exit(1) diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go index 90669933..1f6589ae 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/manifest_test.go @@ -39,7 +39,8 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if len(m.Claims) != 1 || m.Claims[0].Name != Seat || m.Claims[0].Scope != "mesh" || !slices.Equal(m.Claims[0].Serves, Verbs) { t.Errorf("claims %+v", m.Claims) } - if !slices.Equal(m.Emits, []string{RecordedEvent}) || !slices.Equal(m.Reads, []string{ControllerActs, RouterRecords}) { + // act-recorded is the seat's event, said on the seat's subject: a module lists no event of a seat it claims. + if len(m.Emits) != 0 || !slices.Equal(m.Reads, []string{ControllerActs, RouterRecords}) { t.Errorf("emits %v, reads %v", m.Emits, m.Reads) } readBy := map[string]string{ActsLog: HandActsTool, WarrantsState: HandActsTool, ControllerActs: HandActsTool, diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go index 3de340f2..64ec46d2 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/whole_integration_test.go @@ -204,6 +204,13 @@ func (p *player) handle(m wire) { err = kv.Delete(p.ctx, a.Key) } p.answer(m.ID, map[string]any{}, err) + case "mesh/seat.publish": + var e struct { + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish(e.Subject, e.Body)) case "mesh/publish": var e struct { Key string `json:"key"` @@ -343,7 +350,7 @@ func TestTheHandActLogWholeRecordsOnTheBus(t *testing.T) { if err != nil { t.Fatal(err) } - recorded, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + RecordedEvent) + recorded, err := nc.SubscribeSync("mesh.seat." + Seat + ".event." + RecordedEvent) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-hand-acts/go.mod b/modules/mesh-hand-acts/go.mod index f79a46d2..3b27171d 100644 --- a/modules/mesh-hand-acts/go.mod +++ b/modules/mesh-hand-acts/go.mod @@ -3,7 +3,7 @@ module mesh-hand-acts go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-hand-acts/go.sum b/modules/mesh-hand-acts/go.sum index ae67790f..0bfd755e 100644 --- a/modules/mesh-hand-acts/go.sum +++ b/modules/mesh-hand-acts/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-hand-acts/module.json b/modules/mesh-hand-acts/module.json index 768f3e23..6882e201 100644 --- a/modules/mesh-hand-acts/module.json +++ b/modules/mesh-hand-acts/module.json @@ -13,9 +13,6 @@ ] } ], - "emits": [ - "act-recorded" - ], "reads": [ "mesh-controller.hand-acts", "messenger.asks" diff --git a/modules/mesh-healers/cmd/mesh-healers/healers.go b/modules/mesh-healers/cmd/mesh-healers/healers.go index e18d8cae..3c6dde6c 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers.go @@ -45,6 +45,8 @@ const ( ConditionsRead = "mesh-conditions.conditions" HandActsRead = "mesh-controller.hand-acts" ActedEvent = health.EventHealerActed + // ActedSubject is where the seat says it, which its consumers hear (catalogue.SeatSays). + ActedSubject = "mesh.seat." + Seat + ".event." + ActedEvent // The events it hears: the keeper's, on the seat mesh-conditions. RaisedEvent = "mesh-conditions.condition-raised" ChangedEvent = "mesh-conditions.condition-changed" @@ -152,9 +154,10 @@ func Registry() []Healer { // Healers is mesh-healers: the registry, what its acts reach, and what it holds. type Healers struct { - Store Store - Ask func(key string, body any) (json.RawMessage, error) - Emit func(event string, body any) error + Store Store + Ask func(key string, body any) (json.RawMessage, error) + // Publish says a seat event on its subject, de-duplicated by id (stdio.SeatPublish). + Publish func(subject string, body any, id string) error Now func() time.Time Logf func(format string, args ...any) Registry []Healer @@ -354,10 +357,10 @@ func (h *Healers) escalate(row Healer, c Condition, budget string, spent []Attem func (h *Healers) tell(e Acted) { e.Event, e.At, e.By = ActedEvent, h.Now().UTC(), "healer "+e.Healer e.Show = "mesh-conditions.conditions key=" + e.Condition - if h.Emit == nil { + if h.Publish == nil { return } - if err := h.Emit(ActedEvent, e); err != nil { + if err := h.Publish(ActedSubject, e, e.Healer+"."+e.Condition+"."+e.Outcome+"."+strconv.FormatInt(e.At.UnixNano(), 10)); err != nil { h.Logf("[%s] healer %s %s %s, and that could NOT be said on the bus: %v", Module, e.Healer, e.Outcome, e.Condition, err) } } diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go index 00612cb8..50a69f42 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -112,8 +112,8 @@ func (w *world) fresh() { w.asked = append(w.asked, key+" "+string(raw)) return json.RawMessage(`{"says":"2 the bus lacks: created at this ask"}`), nil }, - Emit: func(event string, body any) error { - if a, ok := body.(Acted); ok && event == ActedEvent { + Publish: func(subject string, body any, _ string) error { + if a, ok := body.(Acted); ok && subject == "mesh.seat.mesh-healers.event.healer-acted" { w.said = append(w.said, a) } return nil diff --git a/modules/mesh-healers/cmd/mesh-healers/main.go b/modules/mesh-healers/cmd/mesh-healers/main.go index 2207e3c4..ddb4c7b1 100644 --- a/modules/mesh-healers/cmd/mesh-healers/main.go +++ b/modules/mesh-healers/cmd/mesh-healers/main.go @@ -61,9 +61,9 @@ func watched(state, key string, handle func(stdio.StateChange) error) { func main() { h := &Healers{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, - Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, - Emit: func(event string, body any) error { return stdio.Emit(event, body) }, - After: func(d time.Duration, f func()) { time.AfterFunc(d, f) }} + Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, + Publish: stdio.SeatPublish, + After: func(d time.Duration, f func()) { time.AfterFunc(d, f) }} // The lease first: until it is read the healers write would-acts only. Once it is first read, every settle is // re-armed from the attempts kept, so a restart loses no wake. var started sync.Once diff --git a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go index 86588a98..ab3528f7 100644 --- a/modules/mesh-healers/cmd/mesh-healers/manifest_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/manifest_test.go @@ -47,7 +47,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { name string got, want []string }{ - {"emits", m.Emits, []string{ActedEvent}}, + {"emits (healer-acted is the seat's, said on its subject)", m.Emits, nil}, {"consumes", m.Consumes, []string{RaisedEvent, ChangedEvent}}, {"invokes", m.Invokes, []string{AssertTool}}, {"reads", m.Reads, []string{LeaseRead, ConditionsRead, HandActsRead}}, diff --git a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go index 5c7ec972..a8e2733a 100644 --- a/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/whole_integration_test.go @@ -216,6 +216,13 @@ func (p *player) handle(m wire) { err = kv.Delete(p.ctx, a.Key) } p.answer(m.ID, map[string]any{}, err) + case "mesh/seat.publish": + var e struct { + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish(e.Subject, e.Body)) case "mesh/publish": var e struct { Key string `json:"key"` @@ -332,7 +339,7 @@ func TestTheHealersWholeActOnTheBusAndBrake(t *testing.T) { if _, err := lease.Put(ctx, LeaseKey, []byte(`{"instance":"novox/1/2026-10-27","epoch":57}`)); err != nil { t.Fatal(err) } - acted, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + ActedEvent) + acted, err := nc.SubscribeSync("mesh.seat." + Seat + ".event." + ActedEvent) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-healers/go.mod b/modules/mesh-healers/go.mod index 587a34e4..229731fa 100644 --- a/modules/mesh-healers/go.mod +++ b/modules/mesh-healers/go.mod @@ -3,7 +3,7 @@ module mesh-healers go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-healers/go.sum b/modules/mesh-healers/go.sum index ae67790f..0bfd755e 100644 --- a/modules/mesh-healers/go.sum +++ b/modules/mesh-healers/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-healers/module.json b/modules/mesh-healers/module.json index 186965ec..cc2438e1 100644 --- a/modules/mesh-healers/module.json +++ b/modules/mesh-healers/module.json @@ -13,9 +13,6 @@ ] } ], - "emits": [ - "healer-acted" - ], "consumes": [ "mesh-conditions.condition-raised", "mesh-conditions.condition-changed" diff --git a/modules/mesh-self-check/cmd/mesh-self-check/main.go b/modules/mesh-self-check/cmd/mesh-self-check/main.go index c423e855..4225eead 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/main.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/main.go @@ -40,8 +40,8 @@ func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(st // NewChecker is the self-check over the runtime. func NewChecker() *Checker { return &Checker{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"), - Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, - Emit: func(event string, body any) error { return stdio.Emit(event, body) }} + Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, + Publish: stdio.SeatPublish} } func main() { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go index 10f33dc9..fb245a78 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go @@ -47,7 +47,8 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if len(m.Claims) != 1 || m.Claims[0].Name != Seat || m.Claims[0].Scope != "mesh" || !slices.Equal(m.Claims[0].Serves, Verbs) { t.Errorf("claims %+v, want %s at mesh scope serving %v", m.Claims, Seat, Verbs) } - if !slices.Equal(m.Emits, []string{CheckedEvent}) { + // checked is the seat's event, said on the seat's subject: a module lists no event of a seat it claims. + if len(m.Emits) != 0 { t.Errorf("emits %v", m.Emits) } if !slices.Equal(m.Invokes, []string{StreamsTool, ClosedTool, SignalsVerb}) { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go index 8d07725d..53efb782 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck.go @@ -32,6 +32,9 @@ const ( VerdictsState = "verdicts" LastKey = "last" CheckedEvent = health.EventChecked + // CheckedSubject is where the seat says it: a seat's events are on mesh.seat..event., which its + // consumers hear (catalogue.SeatSays). + CheckedSubject = "mesh.seat." + Seat + ".event." + CheckedEvent ) // The clocks: the self-check's run is the one timer among the four holders (ADR 0564 ยง1). @@ -83,9 +86,10 @@ type ( // Checker is the self-check: the registry and what its probes reach. type Checker struct { - Store Store - Ask func(key string, body any) (json.RawMessage, error) - Emit func(event string, body any) error + Store Store + Ask func(key string, body any) (json.RawMessage, error) + // Publish says a seat event on its subject, de-duplicated by id (stdio.SeatPublish). + Publish func(subject string, body any, id string) error Now func() time.Time Logf func(format string, args ...any) Host string @@ -328,10 +332,10 @@ func (c *Checker) keep(run Run) { if err := c.Store.Put(VerdictsState, run.Run, Overview(run)); err != nil { c.Logf("[%s] the run %s is not kept among the week's verdicts: %v", Module, run.Run, err) } - if c.Emit == nil { + if c.Publish == nil { return } - if err := c.Emit(CheckedEvent, run); err != nil { + if err := c.Publish(CheckedSubject, run, run.Run); err != nil { c.mu.Lock() c.unsaid++ c.mu.Unlock() diff --git a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go index 311bf924..e79bb78f 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/selfcheck_test.go @@ -81,9 +81,9 @@ func checker(store *memStore, registry ...Probe) (*Checker, *[]string) { now := time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC) c := &Checker{Store: store, Registry: registry, Host: "novox", Logf: func(string, ...any) {}, Now: func() time.Time { now = now.Add(time.Second); return now }, - Emit: func(event string, body any) error { + Publish: func(subject string, body any, id string) error { raw, _ := json.Marshal(body) - said = append(said, event+" "+string(raw)) + said = append(said, subject+" "+id+" "+string(raw)) return nil }} return c, &said @@ -147,11 +147,14 @@ func TestARunWritesFindingsAsTheyChange(t *testing.T) { if keys, _ := store.Keys(VerdictsState); len(keys) != len(steps) { t.Errorf("verdicts kept %d runs, want %d", len(keys), len(steps)) } - if len(*said) != len(steps) || !strings.HasPrefix((*said)[0], CheckedEvent+" ") { + // Said on the seat's subject, where its consumers hear it, de-duplicated by the run's id. + if len(*said) != len(steps) || !strings.HasPrefix((*said)[0], "mesh.seat.mesh-self-check.event.checked doctor-") { t.Fatalf("said %v", *said) } + _, rest, _ := strings.Cut((*said)[0], " ") + _, rest, _ = strings.Cut(rest, " ") var body map[string]any - _ = json.Unmarshal([]byte(strings.TrimPrefix((*said)[0], CheckedEvent+" ")), &body) + _ = json.Unmarshal([]byte(rest), &body) for _, field := range []string{"run", "at", "started", "took", "interval-seconds", "counts", "probes", "controller", "why"} { if _, ok := body[field]; !ok { t.Errorf("checked's body lacks %s, which doctor-heartbeat carries: %v", field, body) diff --git a/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go b/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go index cc078779..54ae1679 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go @@ -146,6 +146,13 @@ func (p *player) handle(m wire) { err = kv.Delete(p.ctx, a.Key) } p.answer(m.ID, map[string]any{}, err) + case "mesh/seat.publish": + var e struct { + Subject string `json:"subject"` + Body json.RawMessage `json:"body"` + } + _ = json.Unmarshal(m.Params, &e) + p.answer(m.ID, map[string]any{}, p.nc.Publish(e.Subject, e.Body)) case "mesh/publish": var e struct { Key string `json:"key"` @@ -230,7 +237,7 @@ func TestTheSelfCheckWholeRunsOnTheBus(t *testing.T) { t.Fatal(err) } } - checked, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + CheckedEvent) + checked, err := nc.SubscribeSync("mesh.seat." + Seat + ".event." + CheckedEvent) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-self-check/go.mod b/modules/mesh-self-check/go.mod index a4d5f850..f96f4183 100644 --- a/modules/mesh-self-check/go.mod +++ b/modules/mesh-self-check/go.mod @@ -3,7 +3,7 @@ module mesh-self-check go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-self-check/go.sum b/modules/mesh-self-check/go.sum index ae67790f..0bfd755e 100644 --- a/modules/mesh-self-check/go.sum +++ b/modules/mesh-self-check/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba h1:uXwkcQWgbu/UthizGHFaKPiA4NBVJgqL9q2jJnnGR1o= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029204453-69fcb6c1c1ba/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c h1:Yd6AslGN7pCniQ9Q7ch05xAWHP57grXRoH3AsDQrSeY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029211044-fbd00c89436c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-self-check/module.json b/modules/mesh-self-check/module.json index b0e732fe..829af53e 100644 --- a/modules/mesh-self-check/module.json +++ b/modules/mesh-self-check/module.json @@ -12,9 +12,6 @@ ] } ], - "emits": [ - "checked" - ], "invokes": [ "nats.nats_streams", "nats.nats_closed_connections",