diff --git a/go/views/emit.go b/go/views/emit.go index 3a68d69..8119ba3 100644 --- a/go/views/emit.go +++ b/go/views/emit.go @@ -18,9 +18,9 @@ import ( type Emitter struct { // Owner is the owner module, said in every event. Owner string - // Subject is where an event of the given local name lands on the owner's own stream (process.Account.StreamSubject). + // Subject is where an event of the given local name lands: the owner's ordinary event (process.Account.EventSubject). Subject func(event string) string - // Publish sends one message to the owner's stream, its id the bus's dedup id. + // Publish sends one message to the shared events stream, its id the bus's dedup id (JetStreamPublish). Publish func(ctx context.Context, msg *nats.Msg, id string) error // Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end. Marks jetstream.KeyValue diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 2c63d11..1ab7611 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -21,7 +21,6 @@ import ( "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -43,10 +42,14 @@ type ProcessReader struct { Bound time.Duration Wait time.Duration - mu sync.Mutex - pics map[string]*Picture // by bucket name _ - order []string - streams map[string]string // each owner's views stream, by owner + // Other is told of each message of the module's durable consumer on the shared events stream that is not an owner's + // views event, so a module that hears other events through the same consumer (a module has one) handles them there. + // Nil says them in the log, acknowledged: the module hears views events alone. + Other func(jetstream.Msg) + + mu sync.Mutex + pics map[string]*Picture // by bucket name _ + order []string } func (r *ProcessReader) logf(format string, args ...any) { @@ -55,23 +58,19 @@ func (r *ProcessReader) logf(format string, args ...any) { } } -// Follow adds the owner's state to follow, as the owner declares them (design 25 § 12.2): tool is its ordinary read tool -// of state as the reader's `invokes` names it (`.`, from the owner's manifest's tools), stream its views -// family as the owner's manifest's `streams` names it (its bus name is catalogue.StreamBusName's). views names no owner: -// the caller gives both. An owner has one views stream; a second name for it is refused. Call it before Start. -func (r *ProcessReader) Follow(owner, state, tool, stream string) (*Picture, error) { +// Follow adds the owner's state to follow, as the owner declares it (design 25 § 12.2): tool is its ordinary read tool +// of state as the reader's `invokes` names it (`.`, from the owner's manifest's tools). The owner's changes +// are ordinary events (`.views.`), which the reader's manifest `consumes` one entry per followed state. +// views names no owner: the caller gives it. Call it before Start. +func (r *ProcessReader) Follow(owner, state, tool string) (*Picture, error) { r.mu.Lock() defer r.mu.Unlock() - if owner == "" || state == "" || !strings.HasPrefix(tool, owner+".") || stream == "" { - return nil, fmt.Errorf("%s follows %s's %s by its read tool . and its views stream, not %q and %q", r.Module, owner, state, tool, stream) - } - if have, ok := r.streams[owner]; ok && have != stream { - return nil, fmt.Errorf("%s's views stream is %s, not %s", owner, have, stream) + if owner == "" || state == "" || !strings.HasPrefix(tool, owner+".") { + return nil, fmt.Errorf("%s follows %s's %s by its read tool ., not %q", r.Module, owner, state, tool) } if r.pics == nil { - r.pics, r.streams = map[string]*Picture{}, map[string]string{} + r.pics = map[string]*Picture{} } - r.streams[owner] = stream bucket := owner + "_" + state if p := r.pics[bucket]; p != nil { return p, nil @@ -153,16 +152,35 @@ func (r *ProcessReader) finding(s Status, raised bool) { } } -// Hear is the handler of the owner's views family on the reader's durable consumer (cred.EventConsumer on the owner's -// views stream): each event to the follower of its state. -func (r *ProcessReader) Hear(ctx context.Context, owner string) func(jetstream.Msg) { - fs := map[string]*Follower{} +// Hear is the handler of the module's durable consumer on the shared events stream (cred.EventConsumer on +// process.EventsStream): an owner's views event (`mesh.mod..event.views.`) goes to the follower of that +// owner's state, any other message to Other. +func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) { + fs := map[string]map[string]*Follower{} for _, f := range r.Followers() { - if f.Owner == owner { - fs[f.State] = f + if fs[f.Owner] == nil { + fs[f.Owner] = map[string]*Follower{} } + fs[f.Owner][f.State] = f + } + views := map[string]func(jetstream.Msg){} + for owner, m := range fs { + views[owner] = Hear(ctx, m, r.logf) + } + return func(m jetstream.Msg) { + if owner, ok := OwnerOfEvent(m.Subject()); ok { + if h := views[owner]; h != nil { + h(m) + return + } + } + if r.Other != nil { + r.Other(m) + return + } + r.logf("views: %s is no views event of an owner followed, acknowledged", m.Subject()) + _ = m.Ack() } - return Hear(ctx, fs, r.logf) } // Owners are the owners followed, sorted. @@ -510,23 +528,18 @@ func (e entry) Created() time.Time { return e.at } func (e entry) Delta() uint64 { return 0 } func (e entry) Operation() jetstream.KeyValueOp { return e.op } -// Start hears each owner's views family on the process's durable consumer of it (its manifest's `consumes` -// `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the tools -// bundle's `_first_view_again` writes there (StatusTools), then asks every first view (Ask). A consumer -// the bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand. +// Start hears the owners' views events on the process's durable consumer of the shared events stream (its manifest's +// `consumes` `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the +// tools bundle's `_first_view_again` writes there (StatusTools), then asks every first view (Ask). A consumer the +// bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand. func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) { - for _, owner := range r.Owners() { - r.mu.Lock() - stream := catalogue.StreamBusName(r.streams[owner]) - r.mu.Unlock() - cons, err := cred.EventConsumer(ctx, js, stream) - if err == nil { - _, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf) - } - if err != nil { - r.logf("the consumer %s on %s is not heard (%v): %s's states are the first view's alone, asked again by hand, until the %s starts again", - cred.EventConsumerName(), stream, err, owner, r.Module) - } + cons, err := cred.EventConsumer(ctx, js, process.EventsStream) + if err == nil { + _, err = process.HearEvents(cons, r.Hear(ctx), r.logf) + } + if err != nil { + r.logf("the consumer %s on %s is not heard (%v): the states followed are the first views' alone, asked again by hand, until the %s starts again", + cred.EventConsumerName(), process.EventsStream, err, r.Module) } var said func(Status) if status != nil { diff --git a/go/views/process_reader_integration_test.go b/go/views/process_reader_integration_test.go new file mode 100644 index 0000000..ab740da --- /dev/null +++ b/go/views/process_reader_integration_test.go @@ -0,0 +1,80 @@ +//go:build integration + +package views_test + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" + "git.novox.be/novox/mesh-sdk/go/views" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" +) + +// An owner's change is an ordinary event on the shared events stream (novox/hq ADR 0635 §5): a reader that follows the +// owner's state by its read tool hears it on the module's durable consumer of that stream, with no stream of the owner's, +// and a message that is no views event goes to Other. +func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cred := process.Account{Node: "n1", Module: "reader"} + kv := viewstest.Serve(t, ctx, js, nc, "o", "modules", "o.o_read_modules", cred.EventConsumerName()) + if _, err := kv.Put(ctx, "a", []byte(`{"n":1}`)); err != nil { + t.Fatal(err) + } + other := make(chan string, 1) + r := &views.ProcessReader{Module: "reader", Other: func(m jetstream.Msg) { other <- m.Subject(); _ = m.Ack() }, Logf: t.Logf, + Call: func(tool string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + m, err := nc.Request("mesh.mod.o.call.o_read_modules.n0", raw, 5*time.Second) + if err != nil { + return nil, err + } + return m.Data, nil + }} + if _, err := r.Follow("o", "modules", "o.o_read_modules"); err != nil { + t.Fatal(err) + } + r.Start(ctx, js, cred, nil) + <-r.Ready() + pic, err := r.KeyValue(ctx, "o_modules") + if err != nil { + t.Fatal(err) + } + w, err := pic.Watch(ctx, "b") + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, "b", []byte(`{"n":2}`)); err != nil { + t.Fatal(err) + } + for e := range w.Updates() { + if e != nil && e.Key() == "b" { + break + } + } + if s := r.Followers()[0].Status(); s.Asks != 1 || s.Keys != 2 { + t.Fatalf("the change was not heard as an event alone: %+v", s) + } + if _, err := js.Publish(ctx, "mesh.mod.o.event.something-else", []byte("{}")); err != nil { + t.Fatal(err) + } + if got := <-other; got != "mesh.mod.o.event.something-else" { + t.Fatalf("other got %s", got) + } +} diff --git a/go/views/views.go b/go/views/views.go index 4e1a515..7414677 100644 --- a/go/views/views.go +++ b/go/views/views.go @@ -17,12 +17,11 @@ import ( // Version is the version of every value this package writes and the only one it reads. const Version = 1 -// The owners' views families (design 25 § 12.2; ADR 0615 §5, ADR 0619 §2): one stream each on its owner's old `streams` -// section, kept by limits 7 days and at most 256 MiB, dropping its oldest when full, so a reader that missed a dropped -// event sees the gap in revisions and asks once more. A stream's name is its owner's to declare; views names none. +// The owners' views families (design 25 § 12.2; ADR 0615 §5, ADR 0619 §2, novox/hq ADR 0635 §5): ordinary events of the +// owner, `mesh.mod..event.views.` on the shared events stream, kept as every event is; a reader that missed a +// dropped one sees the gap in revisions and asks its first view once more. An owner declares them among its `emits` +// (`views.`), a reader `consumes` `.views.`; views names no owner. const ( - StreamDays = 7 - StreamMiB = 256 // EventPrefix starts each event's local name: `views.`, one event per changed key of that state. EventPrefix = "views." // EventPattern is every event of a views family, as a stream's events and a reader's consumes name it. @@ -48,6 +47,19 @@ const Bound = 2 * time.Minute // EventName is the local name of the event saying a change of state: `views.`. func EventName(state string) string { return EventPrefix + state } +// OwnerOfEvent is the owner whose views event a subject is, `mesh.mod..event.views.`. +func OwnerOfEvent(subject string) (string, bool) { + rest, ok := strings.CutPrefix(subject, "mesh.mod.") + if !ok { + return "", false + } + owner, ev, ok := strings.Cut(rest, ".event.") + if !ok || owner == "" || strings.Contains(owner, ".") || StateOfEvent(ev) == "" { + return "", false + } + return owner, true +} + // StateOfEvent is the state an event's local name (or the last tokens of its subject) says, or "". func StateOfEvent(name string) string { if i := strings.LastIndex(name, ".event."+EventPrefix); i >= 0 { diff --git a/go/views/views_test.go b/go/views/views_test.go index 2aa94b5..272afb7 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -295,7 +295,7 @@ func TestViews(t *testing.T) { t.Errorf("asked again by hand: %v, %v", again, err) } }) - t.Run("a process's reader follows the read tool and views stream its caller names, reads the state as a read-only bucket, each key's written time handed on, and opens no other", func(t *testing.T) { + t.Run("a process's reader follows the read tool its caller names, reads the state as a read-only bucket, each key's written time handed on, and opens no other", func(t *testing.T) { r := &ProcessReader{Module: "reader", Bound: time.Nanosecond, Wait: time.Millisecond, Now: now, Call: func(tool string, body any) (json.RawMessage, error) { if m, _ := body.(map[string]any); tool != "o.o_read_modules" || m["values"] != true { @@ -305,13 +305,12 @@ func TestViews(t *testing.T) { raw, _ := json.Marshal(map[string]any{"result": a}) return raw, err }} - p, err := r.Follow("o", "modules", "o.o_read_modules", "o-views") + p, err := r.Follow("o", "modules", "o.o_read_modules") if err != nil { t.Fatal(err) } - for name, c := range map[string][2]string{"another owner's tool": {"x.read_more", "o-views"}, "no stream": {"o.read_more", ""}, - "a second stream for the owner": {"o.read_more", "o-other-views"}} { - if _, err := r.Follow("o", "more", c[0], c[1]); err == nil { + for name, tool := range map[string]string{"another owner's tool": "x.read_more", "no owner in the tool": "read_more"} { + if _, err := r.Follow("o", "more", tool); err == nil { t.Fatalf("%s is followed", name) // a picture no owner answers would hold Ready } } @@ -464,6 +463,12 @@ func TestViews(t *testing.T) { if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { t.Errorf("state of subject = %q", s) } + for subject, want := range map[string]string{"mesh.mod.o.event.views.modules": "o", "mesh.mod.o.event.views.a.b": "", "mesh.mod.o.event.other": "", + "mesh.stream.o.event.views.modules": "", "mesh.mod..event.views.modules": ""} { + if got, _ := OwnerOfEvent(subject); got != want { + t.Errorf("owner of %s = %q, want %q", subject, got, want) + } + } raw, err := ev(3, "k", `{"x":1}`).Encode() got, err2 := DecodeEvent(raw) if err != nil || err2 != nil || got.Revision != 3 || string(got.Value) != `{"x":1}` { diff --git a/go/views/viewstest/owner.go b/go/views/viewstest/owner.go index 4b62535..30321a3 100644 --- a/go/views/viewstest/owner.go +++ b/go/views/viewstest/owner.go @@ -14,18 +14,17 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" ) // Serve is owner's state on the bus with the owner's side of it: the bucket `_` (made when absent, history -// 1), the owner's marks, its views stream with a durable consumer per reader (`_`), the emitter saying -// each change, and the read tool answering on nc: tool is the owner's read tool as a reader's invokes names it -// (`.`), stream its views stream as its manifest's `streams` names it, both given as the reader's -// ProcessReader.Follow takes them. The tool answers a key's history too (key with history), from what the bucket keeps: +// 1), the owner's marks, the shared events stream (made when absent) with a durable consumer per reader +// (`_`), the emitter saying each change as the owner's ordinary event, and the read tool answering on nc: +// tool is the owner's read tool as a reader's invokes names it (`.`), as the reader's ProcessReader.Follow +// takes it. The tool answers a key's history too (key with history), from what the bucket keeps: // a test that reads history makes the bucket first with the history it needs. It returns the bucket the test writes. -func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state, tool, stream string, readers ...string) jetstream.KeyValue { +func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state, tool string, readers ...string) jetstream.KeyValue { t.Helper() bucket := owner + "_" + state kv, err := js.KeyValue(ctx, bucket) @@ -38,16 +37,17 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C if err != nil { t.Fatal(err) } - stream = catalogue.StreamBusName(stream) - if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil { - t.Fatal(err) - } - for _, r := range readers { - if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + if _, err := js.Stream(ctx, process.EventsStream); err != nil { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { t.Fatal(err) } } - em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.ModuleStreamSubject(owner, e) }, + for _, r := range readers { + if _, err := js.CreateOrUpdateConsumer(ctx, process.EventsStream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } + em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.Account{Module: owner}.EventSubject(e) }, Publish: views.JetStreamPublish(js), Marks: marks} if err := em.Start(ctx, state, kv, nil); err != nil { t.Fatal(err) @@ -82,10 +82,10 @@ func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.C } // Grants are the subjects a reader's account publishes to follow owner's state (its `invokes` and `consumes`): the -// read tool's call subjects, its durable consumer's info, next and acknowledgements on the owner's views stream. -func Grants(owner, tool, stream, durable string) []string { +// read tool's call subjects, its durable consumer's info, next and acknowledgements on the shared events stream. +func Grants(owner, tool, durable string) []string { tool = strings.TrimPrefix(tool, owner+".") - stream = catalogue.StreamBusName(stream) + stream := process.EventsStream return []string{"mesh.mod." + owner + ".call." + tool + ".>", "mesh.mod." + owner + ".call." + tool, "$JS.API.CONSUMER.INFO." + stream + "." + durable, "$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable, "$JS.ACK." + stream + "." + durable + ".>"}