diff --git a/go/process/binding.go b/go/process/binding.go index 56fb669e..f572dec3 100644 --- a/go/process/binding.go +++ b/go/process/binding.go @@ -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() diff --git a/go/views/stdio_writer.go b/go/views/stdio_writer.go index 161957ed..4e5a802b 100644 --- a/go/views/stdio_writer.go +++ b/go/views/stdio_writer.go @@ -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..event..`. +// eventOfSubject is the event of a subject of the writer: the old stream's `mesh.stream..event..` or the +// binding's `mesh.bus..finding-events..` (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..event..", 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..event.. or mesh.bus..finding-events..", subject) } // seatEventOfSubject is `.` of a seat event's subject `mesh.seat..event.`, the name a runtime diff --git a/go/views/stdio_writer_test.go b/go/views/stdio_writer_test.go index c820d16f..088f094d 100644 --- a/go/views/stdio_writer_test.go +++ b/go/views/stdio_writer_test.go @@ -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") + } +} diff --git a/go/views/writer.go b/go/views/writer.go index befc8d92..1ec9a5f1 100644 --- a/go/views/writer.go +++ b/go/views/writer.go @@ -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..finding-events` and each message `..`; 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 } diff --git a/go/views/writer_test.go b/go/views/writer_test.go index ed5d4929..4e23a4fb 100644 --- a/go/views/writer_test.go +++ b/go/views/writer_test.go @@ -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)