novox/mesh-sdk
9
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

G A-17 (corrected): mesh-sdk names no seat; WriterConfig.Asked and the TypeScript writer's asked are the subject of findings-asked, given from the seat's generated contract

Browse Source 
views.FindingsAskedSubject, views.FindingsAskedEvent and FINDINGS_ASKED_EVENT are gone; the stdio writer subscribes the seat event
the given subject names.

Builds: design 47 G A-17, A-13, §461
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 21:58:38 +01:00
parent 948cc1b34e
commit 24f876a348
7 changed files with 59 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/views/stdio_writer.go
+16 -7
View File
@@ -10,13 +10,9 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
)
// FindingsAskedEvent is the event `findings-asked`, always the seat mesh-conditions' event, named with that seat (design 47 G
// A-17); a stdio module's writer subscribes to it by this name, and no module supplies it.
const FindingsAskedEvent = "mesh-conditions.findings-asked"
// NewStdioWriter makes the finding writer of a stdio module (design 47 G A-16, §461, §485, §503): it fills `Publish` from
// the module's own event publish (stdio.Emit, answered once the bus acknowledged it) and `Listen` from its subscription
// (stdio.SubscribeNew of FindingsAskedEvent, only the asks published after it is made), so no module writes that adapter. A `Publish` or `Listen`
// (stdio.SubscribeNew of the seat event the config's Asked subject names, only the asks published after it is made), so no module writes that adapter. A `Publish` or `Listen`
// the config already holds is kept. The runtime publishes and subscribes by the event's name on the subject the module's
// membership names for it, so each of the writer's four subjects (FindingSubjects) is told by its event.
func NewStdioWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*Writer[F, C, S, A], error) {
@@ -36,8 +32,11 @@ func newStdioWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A],
}
}
if cfg.Listen == nil {
cfg.Listen = func(_ context.Context, _ string, handle func([]byte)) (func(), error) {
event := FindingsAskedEvent
cfg.Listen = func(_ context.Context, subject string, handle func([]byte)) (func(), error) {
event, err := seatEventOfSubject(subject)
if err != nil {
return nil, err
}
var stopped atomic.Bool
// the runtime keeps a subscription until the process ends: a stopped listener hands on nothing
if err := subscribe(event, func(env stdio.Envelope) error {
@@ -64,3 +63,13 @@ func eventOfSubject(subject string) (string, error) {
}
return "", fmt.Errorf("the subject %q is not of the form mesh.stream.<module>.event.<event>.<node>", subject)
}
// seatEventOfSubject is `<seat>.<event>` of a seat event's subject `mesh.seat.<seat>.event.<event>`, the name a runtime
// subscribes a seat's event by.
func seatEventOfSubject(subject string) (string, error) {
p := strings.Split(subject, ".")
if len(p) == 5 && p[0] == "mesh" && p[1] == "seat" && p[3] == "event" {
return p[2] + "." + p[4], nil
}
return "", fmt.Errorf("the subject %q is not of the form mesh.seat.<seat>.event.<event>", subject)
}
go/views/stdio_writer_test.go
+4 -4
View File
@@ -48,7 +48,7 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T
var subscribed []string
var deliver func(stdio.Envelope) error
w, err := newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{
Module: "m", Node: "n1", State: nil,
Module: "m", Node: "n1", State: nil, Asked: "mesh.seat.s.event.findings-asked",
Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil },
}, func(event string, body any, o ...stdio.EmitOptions) error {
raw, _ := json.Marshal(body)
@@ -63,7 +63,7 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T
t.Fatalf("the writer still needs its state: %v", err)
}
w, err = newStdioWriter(WriterConfig[swFinding, swCleared, swStated, swAsked]{
Module: "m", Node: "n1", State: noState{},
Module: "m", Node: "n1", State: noState{}, Asked: "mesh.seat.s.event.findings-asked",
Observe: func(context.Context) (Observed[swFinding], error) { return Observed[swFinding]{}, nil },
}, func(event string, body any, o ...stdio.EmitOptions) error {
raw, _ := json.Marshal(body)
@@ -84,8 +84,8 @@ func TestStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing(t *testing.T
t.Fatalf("the publish: %+v", out)
}
var heard []string
stop, err := w.cfg.Listen(context.Background(), w.subj.Asked, func(b []byte) { heard = append(heard, string(b)) })
if err != nil || len(subscribed) != 1 || subscribed[0] != "mesh-conditions.findings-asked" {
stop, err := w.cfg.Listen(context.Background(), w.cfg.Asked, func(b []byte) { heard = append(heard, string(b)) })
if err != nil || len(subscribed) != 1 || subscribed[0] != "s.findings-asked" {
t.Fatalf("the listen: %v %v", subscribed, err)
}
_ = deliver(stdio.Envelope{Body: json.RawMessage(`{"ask":"x"}`)})
go/views/writer.go
+12 -11
View File
@@ -111,21 +111,17 @@ type FindingState interface {
Keys() ([]string, error)
}
// WriterSubjects are the subjects of the four events of an observer's module and node.
// WriterSubjects are the subjects of the three events an observer's module says on its own stream.
type WriterSubjects struct {
Written, Cleared, Stated, Asked string
Written, Cleared, Stated string
}
// FindingsAskedSubject is the subject the writer hears an ask for findings on: the seat mesh-conditions' event
// `findings-asked` (design 47 G §387, §390, A-17), never a stream of the part's own.
const FindingsAskedSubject = "mesh.seat.mesh-conditions.event.findings-asked"
// FindingSubjects are the subjects of the module's three own events on node, in §383's form
// `mesh.stream.<module>.event.<event>.<node>` (design 47 G A-13, §461, §439, §383), and the seat's subject of the ask it hears
// (FindingsAskedSubject). The writer forms them itself; a module passes none.
// `mesh.stream.<module>.event.<event>.<node>` (design 47 G A-13, §461, §439, §383). The writer forms them itself; a module
// passes none. The fourth, the subject of the ask it hears, is the seat's and is given (WriterConfig.Asked, A-17).
func FindingSubjects(module, node string) WriterSubjects {
of := func(event string) string { return "mesh.stream." + module + ".event." + event + "." + node }
return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated"), Asked: FindingsAskedSubject}
return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated")}
}
// WriterConfig makes a Writer.
@@ -141,6 +137,9 @@ type WriterConfig[F, C, S, A any] struct {
// JS is the bus connection's JetStream; Conn, when given, lets the writer hear the connection made again.
JS jetstream.JetStream
Conn *nats.Conn
// Asked is the subject the writer hears `findings-asked` on: the seat's event, which the module fills from the generated
// contract of the seat that defines it (design 47 G A-17); mesh-sdk names no seat.
Asked string
// Node is the machine the module runs on: it forms the subjects of the four events (FindingSubjects) and a views
// reader's findings name it in their subject (design 47 G §687, A-13); none given, MESH_NODE.
Node string
@@ -186,6 +185,8 @@ func NewWriter[F, C, S, A any](cfg WriterConfig[F, C, S, A]) (*Writer[F, C, S, A
return nil, errors.New("a finding writer needs the state `findings`")
case cfg.Observe == nil:
return nil, errors.New("a finding writer needs the observe function")
case cfg.Asked == "":
return nil, errors.New("a finding writer needs the subject of findings-asked, from the generated contract of its seat")
case cfg.JS == nil && (cfg.Publish == nil || cfg.Listen == nil):
return nil, errors.New("a finding writer needs the bus connection's JetStream")
}
@@ -309,7 +310,7 @@ func (w *Writer[F, C, S, A]) Close() {
}
func (w *Writer[F, C, S, A]) listen() error {
stop, err := w.cfg.Listen(w.run, w.subj.Asked, w.heard)
stop, err := w.cfg.Listen(w.run, w.cfg.Asked, w.heard)
if err != nil {
return err
}
@@ -329,7 +330,7 @@ func (w *Writer[F, C, S, A]) Reconnected() {
return
}
if err := w.listen(); err != nil {
w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.subj.Asked, err)
w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.cfg.Asked, err)
}
w.asks, w.want = append(w.asks, ""), true
w.signal()
go/views/writer_integration_test.go
+4 -4
View File
@@ -31,12 +31,12 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
subj := views.FindingSubjects("m", "n1")
for name, subjects := range map[string][]string{"EVENTS": {subj.Written, subj.Cleared, subj.Stated}, "ASKS": {subj.Asked}} {
for name, subjects := range map[string][]string{"EVENTS": {subj.Written, subj.Cleared, subj.Stated}, "ASKS": {asked}} {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: subjects}); err != nil {
t.Fatal(err)
}
}
if _, err := js.Publish(ctx, subj.Asked, []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
if _, err := js.Publish(ctx, asked, []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
t.Fatal(err)
}
listen, err := js.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy})
@@ -44,7 +44,7 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) {
t.Fatal(err)
}
state := &fakeState{m: map[string][]byte{}}
w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1",
w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1", Asked: asked,
Observe: func(context.Context) (testObserved, error) {
return testObserved{Findings: []testFinding{finding("a", "now")}}, nil
}, Logf: t.Logf})
@@ -82,7 +82,7 @@ func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) {
for _, ask := range []string{"new", "again"} {
w.Reconnected()
next(subj.Stated) // the statement after the connection was made again
if _, err := js.Publish(ctx, subj.Asked, []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
if _, err := js.Publish(ctx, asked, []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
t.Fatal(err)
}
if st := next(subj.Stated); st.Asked != ask {
go/views/writer_test.go
+15 -4
View File
@@ -78,10 +78,13 @@ type rig struct {
// subjects are the four subjects the writer forms itself in §383's form (design 47 G A-13); the literal is the form.
var subjects = views.FindingSubjects("m", "n1")
// asked is the subject the module gives the writer for findings-asked, as its seat's generated contract has it (design 47 G A-17).
const asked = "mesh.seat.s.event.findings-asked"
func TestFindingSubjectsAreTheEventsStreamForm(t *testing.T) {
want := views.WriterSubjects{
Written: "mesh.stream.m.event.finding-written.n1", Cleared: "mesh.stream.m.event.finding-cleared.n1",
Stated: "mesh.stream.m.event.findings-stated.n1", Asked: "mesh.seat.mesh-conditions.event.findings-asked",
Stated: "mesh.stream.m.event.findings-stated.n1",
}
if subjects != want {
t.Fatalf("the subjects: %+v, want %+v", subjects, want)
@@ -94,7 +97,7 @@ func newRigKinds(t *testing.T, kinds ...string) *rig {
r := &rig{t: t, state: &fakeState{m: map[string][]byte{}}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), ch: make(chan struct{}, 100)}
r.observe = func(context.Context) (testObserved, error) { return testObserved{}, nil }
w, err := views.NewWriter(testConfig{
Module: "m", State: r.state, Kinds: kinds, Node: "n1",
Module: "m", State: r.state, Kinds: kinds, Node: "n1", Asked: asked,
Observe: func(ctx context.Context) (testObserved, error) { return r.observe(ctx) },
Logf: func(f string, a ...any) {
r.mu.Lock()
@@ -123,7 +126,7 @@ func newRigKinds(t *testing.T, kinds ...string) *rig {
Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) {
r.mu.Lock()
defer r.mu.Unlock()
if subject != subjects.Asked {
if subject != asked {
t.Errorf("listening on %s", subject)
}
r.listens++
@@ -444,11 +447,19 @@ func TestWriterRefusesABodyWithoutAContractField(t *testing.T) {
At time.Time `json:"at"`
Subject testSubject
}
_, err := views.NewWriter(views.WriterConfig[noSince, testCleared, testStated, testAsked]{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1",
_, err := views.NewWriter(views.WriterConfig[noSince, testCleared, testStated, testAsked]{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1", Asked: asked,
Observe: func(context.Context) (views.Observed[noSince], error) { return views.Observed[noSince]{}, nil },
Publish: func(context.Context, string, []byte, string) error { return nil },
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }})
if err == nil || !strings.Contains(err.Error(), `"kind"`) {
t.Fatalf("a body of other wire names: %v", err)
}
// the subject of findings-asked is the seat's and the module's to give; none given is refused (design 47 G A-17)
_, err = views.NewWriter(testConfig{Module: "m", State: &fakeState{m: map[string][]byte{}}, Node: "n1",
Observe: func(context.Context) (testObserved, error) { return testObserved{}, nil },
Publish: func(context.Context, string, []byte, string) error { return nil },
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }})
if err == nil || !strings.Contains(err.Error(), "findings-asked") {
t.Fatalf("a writer with no subject for findings-asked: %v", err)
}
}
src/views/writer.ts
+5 -5
View File
@@ -3,13 +3,11 @@
// working on their wire form, with no layout of a finding of its own (design 47 G A-3, §640). It keeps an observer's findings in its own state
// `findings` and says them as events (§383, §384). It names the four events itself (design 47 G A-13, A-17): it says
// `finding-written`, `finding-cleared` and `findings-stated` by their names, which the runtime publishes on the subject the
// module's membership names for each, and hears `findings-asked` as the seat mesh-conditions' event; a module passes none.
// module's membership names for each; the event `findings-asked` is the seat's and is given as `asked` (A-17: mesh-sdk names no seat).
// Its acts, `write`, `clear` and a statement, run one at a time in the order they are asked, on one chain of promises.
import type { State } from "../state/index.js";
/** The event `findings-asked` is always the seat mesh-conditions' (design 47 G A-17), named with that seat. */
export const FINDINGS_ASKED_EVENT = "mesh-conditions.findings-asked";
/** The events the writer says, by their names (design 47 G §383, A-13). */
export const FINDING_WRITTEN_EVENT = "finding-written";
export const FINDING_CLEARED_EVENT = "finding-cleared";
@@ -94,6 +92,8 @@ export interface WriterOptions<F extends FindingFields, C extends object, S exte
module: string;
/** The observer's own state `findings`. */
state: State<F>;
/** The event the writer hears `findings-asked` as: the seat's, which the module fills from the seat's generated contract (design 47 G A-17). */
asked: string;
/** The finding kinds this writer owns (design 47 G §580): a write or clear of another kind is refused, naming it, and a statement carries them as `kinds` and deletes only keys of them. None: every kind. The views reader's kinds are always its own besides. */
kinds?: string[];
/** The findings the observer observes now and its pending keys. It is called inside the writer's one chain: it must not call `write` or `clear`. */
@@ -174,7 +174,7 @@ export class Writer<F extends FindingFields, C extends object, S extends object,
try {
await this.listen();
} catch (e) {
this.log(`error: the finding writer could not make its consumer of ${FINDINGS_ASKED_EVENT} again: ${e}`);
this.log(`error: the finding writer could not make its consumer of ${this.o.asked} again: ${e}`);
}
this.request("");
}
@@ -188,7 +188,7 @@ export class Writer<F extends FindingFields, C extends object, S extends object,
}
private async listen(): Promise<void> {
const stop = await this.o.listen(FINDINGS_ASKED_EVENT, (body) => this.heard(body));
const stop = await this.o.listen(this.o.asked, (body) => this.heard(body));
this.stop?.();
this.stop = stop;
}
test/sdk.test.ts
+3 -2
View File
@@ -860,6 +860,7 @@ test("the finding writer writes, clears and states findings as the Go writer doe
const logs: string[] = [];
const w = new Writer({
module: "m",
asked: "s.findings-asked",
state: st,
observe: async () => {
@@ -870,7 +871,7 @@ test("the finding writer writes, clears and states findings as the Go writer doe
if (refuse > 0) { refuse--; throw new Error("no ack"); }
sent.push({ subject, body });
},
listen: async (s, h) => { assert.equal(s, "mesh-conditions.findings-asked"); consumers++; ask = h; return () => {}; },
listen: async (s, h) => { assert.equal(s, "s.findings-asked"); consumers++; ask = h; return () => {}; },
log: (l) => logs.push(l),
now: () => new Date(clock),
wait: async (ms) => { waits.push(ms); clock += ms; return true; },
@@ -945,7 +946,7 @@ test("the finding writer owns its kinds, keeps a views reader's findings in its
};
const sent: { subject: string; body: any }[] = [];
const w = new Writer({
module: "m", state: st, kinds: ["k"],
module: "m", asked: "s.findings-asked", state: st, kinds: ["k"],
observe: async () => ({ findings: [], pending: [] }),
publish: async (subject, body) => void sent.push({ subject, body }),
listen: async () => () => {},