Area 3 (join 11): the held finding writer says its events under its finding-events binding

A part handed the binding finding-events publishes on mesh.bus.<module>.finding-events.<event>.<node>, a part with none on
the old stream; the stdio writer reads the event of either form.

Builds: design 47 G A-274, A-13, A-30, §529

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-12 11:02:11 +01:00
parent 5908076f33
commit 22ded4011f
5 changed files with 146 additions and 7 deletions
+7
View File
@@ -71,6 +71,13 @@ func (b *Binding) Names(local string) (membership.BusBinding, error) {
return e.names, nil
}
// Asks says whether local is an entry this Binding was made for.
func (b *Binding) Asks(local string) bool {
b.mu.Lock()
defer b.mu.Unlock()
return b.entries[local] != nil
}
// Waiting is why the process waits, naming each entry `show` has not answered for, or empty when every one is bound.
func (b *Binding) Waiting() string {
b.mu.Lock()
+6 -2
View File
@@ -65,7 +65,8 @@ func withStdioPlumbing[F, C, S, A any](cfg WriterConfig[F, C, S, A],
return cfg
}
// eventOfSubject is the event of a subject of FindingSubjects: `mesh.stream.<module>.event.<event>.<node>`.
// eventOfSubject is the event of a subject of the writer: the old stream's `mesh.stream.<module>.event.<event>.<node>` or the
// binding's `mesh.bus.<module>.finding-events.<event>.<node>` (design 47 G A-274).
func eventOfSubject(subject string) (string, error) {
p := strings.Split(subject, ".")
for i, s := range p {
@@ -73,7 +74,10 @@ func eventOfSubject(subject string) (string, error) {
return p[i+1], nil
}
}
return "", fmt.Errorf("the subject %q is not of the form mesh.stream.<module>.event.<event>.<node>", subject)
if len(p) == 6 && p[0] == "mesh" && p[1] == "bus" && p[3] == FindingEventsLocal {
return p[4], nil
}
return "", fmt.Errorf("the subject %q is not of the form mesh.stream.<module>.event.<event>.<node> or mesh.bus.<module>.finding-events.<event>.<node>", subject)
}
// seatEventOfSubject is `<seat>.<event>` of a seat event's subject `mesh.seat.<seat>.event.<event>`, the name a runtime
+15
View File
@@ -128,3 +128,18 @@ func (noState) Get(string) (*stdio.StateEntry, error) { return nil, nil }
func (noState) Put(string, any) (uint64, error) { return 0, nil }
func (noState) Delete(string) error { return nil }
func (noState) Keys() ([]string, error) { return nil, nil }
// Design 47 G A-274: the event of a writer's subject is read in both forms, the old stream's and the binding's.
func TestEventOfSubjectReadsBothForms(t *testing.T) {
for subject, want := range map[string]string{
"mesh.stream.m.event.finding-written.n1": "finding-written",
"mesh.bus.m.finding-events.findings-stated.n1": "findings-stated",
} {
if got, err := eventOfSubject(subject); err != nil || got != want {
t.Errorf("%s: %q, %v, want %q", subject, got, err, want)
}
}
if _, err := eventOfSubject("mesh.bus.m.other.finding-written.n1"); err == nil {
t.Error("another binding's subject was read as a finding event")
}
}
+46 -5
View File
@@ -17,6 +17,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/membership"
)
// The finding writer (design 47 G §461, §485, §503): the one writer every observer uses to keep its findings in its own
@@ -124,6 +125,39 @@ func FindingSubjects(module, node string) WriterSubjects {
return WriterSubjects{Written: of("finding-written"), Cleared: of("finding-cleared"), Stated: of("findings-stated")}
}
// FindingEventsLocal is the local name of the `events` template a part requires for its finding stream (design 47 G §529, A-30).
const FindingEventsLocal = "finding-events"
// FindingBindings is the part's bus bindings as the writer reads them (process.Binding is one): Asks says whether the part
// requires the entry of that local name, Names gives its names on the bus.
type FindingBindings interface {
Asks(local string) bool
Names(local string) (membership.BusBinding, error)
}
// subject is where the writer says event: under the binding `finding-events` when the process was handed one, whose subject
// is `mesh.bus.<module>.finding-events` and each message `<subject>.<event>.<node>`; with none, on the old stream
// (FindingSubjects). A binding not answered yet is an error, so the event is sent again until it is (design 47 G A-274, §529).
func (w *Writer[F, C, S, A]) subject(event string) (string, error) {
if w.cfg.Bindings != nil && w.cfg.Bindings.Asks(FindingEventsLocal) {
b, err := w.cfg.Bindings.Names(FindingEventsLocal)
if err != nil {
return "", err
}
if b.Subject == "" {
return "", fmt.Errorf("the bus entry %q of %s names no subject", FindingEventsLocal, w.cfg.Module)
}
return b.Subject + "." + event + "." + w.cfg.Node, nil
}
switch event {
case "finding-written":
return w.subj.Written, nil
case "finding-cleared":
return w.subj.Cleared, nil
}
return w.subj.Stated, nil
}
// WriterConfig makes a Writer.
type WriterConfig[F, C, S, A any] struct {
// Module is the observer's module name: it fills the observer of a finding and tells an ask for another module.
@@ -137,6 +171,10 @@ 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
// Bindings are the part's bus bindings: when it asks the entry `finding-events` (an `events` template its manifest
// requires) the writer says its events under that binding's subjects, else on the old stream (design 47 G A-274); a
// module passes no subjects. nil is the old stream.
Bindings FindingBindings
// 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
@@ -540,7 +578,7 @@ func (w *Writer[F, C, S, A]) Write(ctx context.Context, finding F) error {
return nil
}
raw, _ := json.Marshal(f)
w.say(ctx, w.subj.Written, "finding-written", keyOf(f), raw)
w.say(ctx, "finding-written", keyOf(f), raw)
return nil
}
@@ -561,7 +599,7 @@ func (w *Writer[F, C, S, A]) Clear(ctx context.Context, scope, id, kind, why str
return err
}
raw, _ := json.Marshal(&cleared)
w.say(ctx, w.subj.Cleared, "finding-cleared", key, raw)
w.say(ctx, "finding-cleared", key, raw)
return nil
}
@@ -629,18 +667,21 @@ func (w *Writer[F, C, S, A]) State(ctx context.Context, asked string) error {
return nil
}
}
w.say(ctx, w.subj.Stated, "findings-stated", w.cfg.Module, raw)
w.say(ctx, "findings-stated", w.cfg.Module, raw)
return nil
}
// say publishes one event awaiting the bus's acknowledgement, sent again for PublishRetryFor when unacknowledged, then
// written in the journal as an error naming what it said (design 47 G §383); the next statement covers it.
func (w *Writer[F, C, S, A]) say(ctx context.Context, subject, event, what string, data []byte) {
func (w *Writer[F, C, S, A]) say(ctx context.Context, event, what string, data []byte) {
w.seq++
id := fmt.Sprintf("%s-%s-%d-%d", w.cfg.Module, event, w.now().UnixNano(), w.seq)
start := w.now()
for delay := time.Second; ; delay = min(delay*2, 10*time.Second) {
err := w.cfg.Publish(ctx, subject, data, id)
subject, err := w.subject(event)
if err == nil {
err = w.cfg.Publish(ctx, subject, data, id)
}
if err == nil {
return
}
+72
View File
@@ -13,6 +13,7 @@ import (
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/membership"
"git.novox.be/novox/mesh-sdk/go/views"
)
@@ -91,6 +92,76 @@ func subFindingSubjectsAreTheEventsStreamForm(t *testing.T) {
}
}
// fakeBindings is a process's binding of the entries it asks.
type fakeBindings struct {
asks bool
names membership.BusBinding
err error
}
func (f fakeBindings) Asks(local string) bool { return f.asks && local == views.FindingEventsLocal }
func (f fakeBindings) Names(string) (membership.BusBinding, error) {
return f.names, f.err
}
// Design 47 G A-274: a part handed the binding `finding-events` says its three events under that binding's subject, a part
// with none on the old stream; a binding not answered yet is sent again until it is.
func subWriterSaysUnderItsFindingEventsBindingOrOnTheOldStream(t *testing.T) {
say := func(b views.FindingBindings) (*rig, []said) {
r := newRig(t)
w, err := views.NewWriter(testConfig{Module: "m", State: r.state, Node: "n1", Asked: asked, Bindings: b,
Observe: func(context.Context) (testObserved, error) { return testObserved{}, nil },
Wait: func(ctx context.Context, d time.Duration) bool {
r.mu.Lock()
r.now = r.now.Add(d)
r.mu.Unlock()
return ctx.Err() == nil
},
Now: func() time.Time { r.mu.Lock(); defer r.mu.Unlock(); return r.now },
Publish: func(_ context.Context, subject string, data []byte, _ string) error {
r.mu.Lock()
defer r.mu.Unlock()
r.sent = append(r.sent, said{subject, data})
return nil
},
Listen: func(context.Context, string, func([]byte)) (func(), error) { return func() {}, nil }})
if err != nil {
t.Fatal(err)
}
if err := w.Write(context.Background(), finding("a", "one")); err != nil {
t.Fatal(err)
}
if err := w.Clear(context.Background(), "module", "a", "k", "gone"); err != nil {
t.Fatal(err)
}
if err := w.State(context.Background(), ""); err != nil {
t.Fatal(err)
}
return r, r.events()
}
subjectsOf := func(ev []said) []string {
var out []string
for _, e := range ev {
out = append(out, e.subject)
}
return out
}
bound := membership.BusBinding{Local: views.FindingEventsLocal, Subject: "mesh.bus.m.finding-events"}
_, ev := say(fakeBindings{asks: true, names: bound})
if got := subjectsOf(ev); !slices.Equal(got, []string{"mesh.bus.m.finding-events.finding-written.n1",
"mesh.bus.m.finding-events.finding-cleared.n1", "mesh.bus.m.finding-events.findings-stated.n1"}) {
t.Errorf("under the binding: %v", got)
}
for _, b := range []views.FindingBindings{nil, fakeBindings{asks: false}} {
if _, ev := say(b); !slices.Equal(subjectsOf(ev), []string{subjects.Written, subjects.Cleared, subjects.Stated}) {
t.Errorf("with no binding: %v", subjectsOf(ev))
}
}
if _, ev := say(fakeBindings{asks: true, err: errors.New("not bound yet")}); len(ev) != 0 {
t.Errorf("an unbound entry said %v", subjectsOf(ev))
}
}
func newRig(t *testing.T) *rig { return newRigKinds(t) }
func newRigKinds(t *testing.T, kinds ...string) *rig {
@@ -574,6 +645,7 @@ func subHeldWriterSeedsAtStart(t *testing.T) {
// TestFindingSubjectsAreTheEventsStreamForm runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
func TestFindingSubjectsAreTheEventsStreamForm(t *testing.T) {
t.Run("FindingSubjectsAreTheEventsStreamForm", subFindingSubjectsAreTheEventsStreamForm)
t.Run("WriterSaysUnderItsFindingEventsBindingOrOnTheOldStream", subWriterSaysUnderItsFindingEventsBindingOrOnTheOldStream)
t.Run("WriterWritesAndClearsFindingsAsEvents", subWriterWritesAndClearsFindingsAsEvents)
t.Run("WriterSendsAgainForAMinuteThenLogs", subWriterSendsAgainForAMinuteThenLogs)
t.Run("WriterStatesAtStartAndOnAsksAndRecovers", subWriterStatesAtStartAndOnAsksAndRecovers)