From 7eb6f31c536e22789df1b7c5a98e7884e734ac97 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 16:04:43 +0100 Subject: [PATCH] G V-5: the views readers adopt a view after their start, and drop one (issue 1966) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A part adopts a watched branch registered while it runs and drops one removed: both go/views readers begin a view followed or adopted after Start at once (its event's one subscription taken first, or reused), read their followers at each event, and Drop ends a view, its row of _first_views (the process's status in first-views) and every finding it raised, cleared through the reader's findings and said in the log. Builds: design 47 G §532, §614 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- go/views/process_reader.go | 257 ++++++++++++++++---- go/views/process_reader_integration_test.go | 17 +- go/views/reader.go | 246 +++++++++++++++---- go/views/status.go | 7 + go/views/views_test.go | 112 +++++++++ 5 files changed, 534 insertions(+), 105 deletions(-) diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 36a3abe6..6ba49939 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -58,7 +58,25 @@ type ProcessReader struct { order []string // adopted are the handlers of the adopted followers' seat events (Adopt), by the event's subject. // Several followers adopt one event (two branches' views of the catalog half), each hearing it (design 47 G §532). - adopted map[string][]func(context.Context, []byte) error + adopted map[string][]adoptedHear + + // After Start and Ask (design 47 G §532): the hub the reader's events come through when it is the module's (Hub) and + // its handler, the time the bus made the consumer, the asks' life and Said, and of each first view begun, the cancel + // of its asks, by bucket; status is the state `first-views` each status is kept in. + hub *process.EventHub + handle func(jetstream.Msg) + created time.Time + ctx context.Context + said func(Status) + asked bool + begun map[string]context.CancelFunc + status *process.AwaitedBucket +} + +// adoptedHear is one adopted follower's handler of its seat event, by its picture's bucket. +type adoptedHear struct { + bucket string + hear func(context.Context, []byte) error } func (r *ProcessReader) logf(format string, args ...any) { @@ -70,19 +88,28 @@ func (r *ProcessReader) logf(format string, args ...any) { // 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. +// views names no owner: the caller gives it. Called after Start, the view takes its first view at once and then follows +// (design 47 G §532). func (r *ProcessReader) Follow(owner, state, tool string) (*Picture, error) { + p, fresh, err := r.follow(owner, state, tool) + if err == nil && fresh { + r.begin(p.bucket) + } + return p, err +} + +func (r *ProcessReader) follow(owner, state, tool string) (*Picture, bool, error) { r.mu.Lock() defer r.mu.Unlock() 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) + return nil, false, fmt.Errorf("%s follows %s's %s by its read tool ., not %q", r.Module, owner, state, tool) } if r.pics == nil { r.pics = map[string]*Picture{} } bucket := owner + "_" + state if p := r.pics[bucket]; p != nil { - return p, nil + return p, false, nil } p := &Picture{bucket: bucket, ctx: context.Background()} call := r.Call @@ -98,16 +125,27 @@ func (r *ProcessReader) Follow(owner, state, tool string) (*Picture, error) { return call(key, m) }, tool) p.f = &Follower{Owner: owner, Tool: tool, State: state, Ask: ask, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait, - Raise: func(s Status) { r.finding(s, true) }, Clear: func(s Status) { r.finding(s, false) }, + Raise: func(s Status) { + if r.holds(p) { + r.finding(s, true) + } + }, Clear: func(s Status) { r.finding(s, false) }, Changed: p.changed} r.pics[bucket] = p r.order = append(r.order, bucket) - return p, nil + return p, true, nil +} + +// holds says whether p is still a picture of the reader: a dropped view writes no finding after its drop cleared them. +func (r *ProcessReader) holds(p *Picture) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.pics[p.bucket] == p } // FollowBranch is Follow for a state kept per branch: the view of one branch is named `..` (the // branch escaped as a key's name is, BranchState) and its picture `_.`, so a process follows -// several branches of one state, each its own view (design 47 G §479). Call it before Start. +// several branches of one state, each its own view (design 47 G §479). Called after Start, it begins at once (§532). func (r *ProcessReader) FollowBranch(owner, state, branch, tool string) (*Picture, error) { return r.Follow(owner, BranchState(state, branch), tool) } @@ -117,12 +155,32 @@ func (r *ProcessReader) FollowBranch(owner, state, branch, tool string) (*Pictur // view and view-waits, in the reader's findings), its log and clock, and its picture (`_`) hands each // change on before the follower's own Changed, set before. Start keeps its status, hears its hand asks, and asks it with // the others; its events are the seat's event (seatdef's EventSubject of seat and event, on the module's durable -// consumer, as its `consumes` names it), handed to hear, the client's handler, and acknowledged. Call it before Start; -// a state followed already is refused. +// consumer, as its `consumes` names it), handed to hear, the client's handler, and acknowledged; a state followed already +// is refused. Adopted after Start, as a part adopts a watched branch added while it runs, the view takes its first view +// at once and then follows (design 47 G §532). func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, error) { if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil { return nil, errors.New("views: an adopted follower names its owner and state, and its seat's event and handler") } + p, subject, err := r.adopt(f, seat, event, hear) + if err != nil { + return nil, err + } + r.mu.Lock() + hub, handle := r.hub, r.handle + r.mu.Unlock() + if subject != "" && hub != nil { + // one subscription per event, with every further part (design 47 G §532) + if err := hub.Handle(subject+".>", handle); err != nil { + r.logf("views: %s's event %s is not heard (%v): the view %s is its first view's alone, asked again by hand", r.Module, subject, err, p.bucket) + } + } + r.begin(p.bucket) + return p, nil +} + +// adopt is Adopt's part under the lock; subject is the event's subject when no follower adopted it before. +func (r *ProcessReader) adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, string, error) { r.mu.Lock() defer r.mu.Unlock() if r.pics == nil { @@ -130,7 +188,7 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context } bucket := f.Owner + "_" + f.State if r.pics[bucket] != nil { - return nil, fmt.Errorf("views: %s's %s is followed already", f.Owner, f.State) + return nil, "", fmt.Errorf("views: %s's %s is followed already", f.Owner, f.State) } p := &Picture{bucket: bucket, ctx: context.Background(), f: f} if f.Logf == nil { @@ -145,8 +203,16 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context if f.Wait == 0 { f.Wait = r.Wait } - f.Raise, f.Clear = func(s Status) { r.finding(s, true) }, func(s Status) { r.finding(s, false) } - f.RaiseWaits, f.ClearWaits = func(s Status) { r.waits(s, true) }, func(s Status) { r.waits(s, false) } + f.Raise, f.Clear = func(s Status) { + if r.holds(p) { + r.finding(s, true) + } + }, func(s Status) { r.finding(s, false) } + f.RaiseWaits, f.ClearWaits = func(s Status) { + if r.holds(p) { + r.waits(s, true) + } + }, func(s Status) { r.waits(s, false) } own := f.Changed f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { p.changed(key, value, deleted, at) @@ -157,11 +223,15 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context r.pics[bucket] = p r.order = append(r.order, bucket) if r.adopted == nil { - r.adopted = map[string][]func(context.Context, []byte) error{} + r.adopted = map[string][]adoptedHear{} } subject := "mesh.seat." + seat + ".event." + event - r.adopted[subject] = append(r.adopted[subject], hear) - return p, nil + fresh := len(r.adopted[subject]) == 0 + r.adopted[subject] = append(r.adopted[subject], adoptedHear{bucket: bucket, hear: hear}) + if !fresh { + subject = "" + } + return p, subject, nil } // waits writes (raised) or clears the reader's view-waits finding of s (design 47 *A part with no view says it waits*). @@ -250,37 +320,36 @@ func (r *ProcessReader) finding(s Status, raised bool) { // 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. +// owner's state, any other message to Other. The followers are read at each message, so a view followed or adopted +// after Start hears its events and a dropped one no longer does (design 47 G §532). func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) { - fs := map[string]map[string]*Follower{} - for _, f := range r.Followers() { - 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) - } - r.mu.Lock() - adopted := maps.Clone(r.adopted) - r.mu.Unlock() return func(m jetstream.Msg) { - if hears := adoptedHears(adopted, m.Subject()); hears != nil { - for _, hear := range hears { - if err := hear(ctx, m.Data()); err != nil { + r.mu.Lock() + hears := adoptedHears(r.adopted, m.Subject()) + var owned map[string]*Follower + if owner, ok := OwnerOfEvent(m.Subject()); ok && hears == nil { + for _, b := range r.order { + if f := r.pics[b].f; f.Owner == owner { + if owned == nil { + owned = map[string]*Follower{} + } + owned[f.State] = f + } + } + } + r.mu.Unlock() + if hears != nil { + for _, h := range hears { + if err := h.hear(ctx, m.Data()); err != nil { r.logf("views: %s is not taken: %v", m.Subject(), err) } } _ = m.Ack() return } - if owner, ok := OwnerOfEvent(m.Subject()); ok { - if h := views[owner]; h != nil { - h(m) - return - } + if owned != nil { + Hear(ctx, owned, r.logf)(m) + return } if r.Other != nil { r.Other(m) @@ -304,20 +373,98 @@ func (r *ProcessReader) Owners() []string { } // Ask asks every first view until answered, each on a goroutine of its own, so the process starts at once and waits for -// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (KeepStatus). +// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (KeepStatus). A view followed or +// adopted after it is asked at once (design 47 G §532). func (r *ProcessReader) Ask(ctx context.Context, said func(Status)) { - for _, f := range r.Followers() { - if said != nil { - f.Said = said - } - go func() { - if err := f.Start(ctx); err != nil { - r.logf("the first view %s was not answered: %v", f.Tool, err) - } - }() + r.mu.Lock() + r.ctx, r.said, r.asked = ctx, said, true + buckets := slices.Clone(r.order) + r.mu.Unlock() + for _, b := range buckets { + r.begin(b) } } +// begin asks bucket's first view once Ask ran, on a goroutine of its own, until answered or the view is dropped. +func (r *ProcessReader) begin(bucket string) { + r.mu.Lock() + p := r.pics[bucket] + if !r.asked || p == nil || r.begun[bucket] != nil { + r.mu.Unlock() + return + } + if r.begun == nil { + r.begun = map[string]context.CancelFunc{} + } + ctx, cancel := context.WithCancel(r.ctx) + r.begun[bucket] = cancel + f, said, created := p.f, r.said, r.created + r.mu.Unlock() + if said != nil { + f.Said = func(s Status) { + if r.holds(p) { + said(s) + } + } + } + if !created.IsZero() { + f.ConsumerMade(ctx, created) + } + go func() { + if err := f.Start(ctx); err != nil && ctx.Err() == nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() +} + +// Drop ends the view of owner's state, as a part drops a watched branch removed while it runs (design 47 G §532, §614): +// it asks it no more and hears none of its events, its status leaves the state `first-views` and so its row leaves +// `_first_views`, and every finding it raised (its unanswered first view, its view-waits) is cleared through +// the reader's findings, the drop and each cleared finding said in the log. A state not followed is an error. +func (r *ProcessReader) Drop(owner, state string) error { + bucket := owner + "_" + state + r.mu.Lock() + p := r.pics[bucket] + if p == nil { + r.mu.Unlock() + return fmt.Errorf("%s does not follow %s's %s", r.Module, owner, state) + } + cancel := r.begun[bucket] + delete(r.pics, bucket) + delete(r.begun, bucket) + r.order = slices.DeleteFunc(r.order, func(b string) bool { return b == bucket }) + for subject, hs := range r.adopted { + if hs = slices.DeleteFunc(hs, func(h adoptedHear) bool { return h.bucket == bucket }); len(hs) == 0 { + delete(r.adopted, subject) + } else { + r.adopted[subject] = hs + } + } + status := r.status + r.mu.Unlock() + if cancel != nil { + cancel() + } + r.logf("views: %s drops its view %s.%s", r.Module, owner, state) + if status != nil { + if kv := status.KV(); kv != nil { + if err := kv.Delete(context.Background(), StatusKey(owner, state)); err != nil { + r.logf("views: the status of the dropped view %s.%s is not removed from %s: %v", owner, state, StatusState, err) + } + } + } + s := p.f.Status() + if s.Raised { + r.finding(s, false) + r.logf("views: %s clears its finding %s of the dropped view %s.%s", r.Module, UnansweredKind, owner, state) + } + if s.WaitsRaised { + r.waits(s, false) + r.logf("views: %s clears its finding %s of the dropped view %s.%s", r.Module, ViewWaitsKind, owner, state) + } + return nil +} + // Ready is closed once every followed state had its first view. func (r *ProcessReader) Ready() <-chan struct{} { fs := r.Followers() @@ -649,12 +796,22 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred hub, pattern, own = process.NewEventHub(r.logf), ">", true } hub.OnBound(func(created time.Time) { + r.mu.Lock() + r.created = created + r.mu.Unlock() for _, f := range r.Followers() { f.ConsumerMade(ctx, created) } }) hear := r.Hear(ctx) err := hub.Handle(pattern, hear) + r.mu.Lock() + r.status = status + if !own { + // an event adopted after the start is handled on the module's hub (Adopt); the reader's own hub takes every one + r.hub, r.handle = hub, hear + } + r.mu.Unlock() if !own { r.mu.Lock() subjects := slices.Sorted(maps.Keys(r.adopted)) @@ -676,7 +833,7 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred var said func(Status) if status != nil { status.Ready(func(kv jetstream.KeyValue) { - if err := HearAgain(ctx, kv, r.Followers(), r.logf); err != nil { + if err := hearAgain(ctx, kv, r.Followers, r.logf); err != nil { r.logf("views: the hand asks of %s's first views are not heard (%v) until it starts again", r.Module, err) } }) @@ -691,7 +848,7 @@ func (r *ProcessReader) Start(ctx context.Context, js jetstream.JetStream, cred // adoptedHears are the handlers of every follower adopted on the event whose subject is subject or has further parts // after it (`...`, design 47 G §521, §532), or nil. -func adoptedHears(adopted map[string][]func(context.Context, []byte) error, subject string) []func(context.Context, []byte) error { +func adoptedHears(adopted map[string][]adoptedHear, subject string) []adoptedHear { if h := adopted[subject]; h != nil { return h } diff --git a/go/views/process_reader_integration_test.go b/go/views/process_reader_integration_test.go index 9f73c431..0c4d75f2 100644 --- a/go/views/process_reader_integration_test.go +++ b/go/views/process_reader_integration_test.go @@ -5,6 +5,7 @@ package views_test import ( "context" "encoding/json" + "errors" "fmt" "testing" "time" @@ -51,7 +52,11 @@ func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) { if _, err := r.Follow("o", "modules", "o.o_read_modules"); err != nil { t.Fatal(err) } - r.Start(ctx, js, cred, nil) + statusKV, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "reader_first-views", Storage: jetstream.MemoryStorage}) + if err != nil { + t.Fatal(err) + } + r.Start(ctx, js, cred, process.AwaitBucket(ctx, js, nil, "reader_first-views", "", t.Logf)) <-r.Ready() pic, err := r.KeyValue(ctx, "o_modules") if err != nil { @@ -115,6 +120,16 @@ func TestAReaderHearsAnOwnersChangesAsOrdinaryEvents(t *testing.T) { if s := f.Status(); s.Asks != 1 { t.Errorf("the reader asked its first view %d times across the bursts (%s): a revision written over was taken as a gap", s.Asks, s.Why) } + // G §532, §614: a dropped view's status leaves the state first-views, and so its row of _first_views. + if _, err := statusKV.Get(ctx, views.StatusKey("o", "modules")); err != nil { + t.Fatalf("the status was not kept: %v", err) + } + if err := r.Drop("o", "modules"); err != nil { + t.Fatal(err) + } + if _, err := statusKV.Get(ctx, views.StatusKey("o", "modules")); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Errorf("the dropped view's status is still kept: %v", err) + } } // A module has one durable consumer on the shared events stream, so the loops that hear events in one module must share diff --git a/go/views/reader.go b/go/views/reader.go index 9aa24775..fcc76d2c 100644 --- a/go/views/reader.go +++ b/go/views/reader.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "maps" + "slices" "sort" "strings" "sync" @@ -47,8 +48,19 @@ type Reader struct { adopted map[string]adoption // events is, of each per-branch view (FollowBranch), by "..", the state as its owner's events name it (§521). events map[string]string + + // After Start (design 47 G §532): the reader's life, the event patterns whose subscription is taken (heard) or being + // asked (asking), and of each first view begun, the cancel of its asks, by ".". + ctx context.Context + started bool + heard map[string]bool + asking map[string]bool + begun map[string]context.CancelFunc } +// subscribe is how the reader asks the runtime for an event pattern; a variable so a test hears without a runtime. +var subscribe = stdio.Subscribe + // adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler. type adoption struct { seat, event string @@ -66,8 +78,16 @@ func (r *Reader) logf(format string, args ...any) { } // Follow adds a state of an owner to follow, asked through the owner's read tool (`.`, as the reader's -// `invokes` names it), with `values` so the answer is the first view. Call it before Start. +// `invokes` names it), with `values` so the answer is the first view. Called after Start, the view takes its first view +// at once and then follows (design 47 G §532). func (r *Reader) Follow(owner, state, tool string) *Follower { + f := r.follow(owner, state, tool, "") + r.begin(owner + "." + f.State) + return f +} + +// follow is Follow without beginning it: event, when given, is the state as the owner's events name it (FollowBranch). +func (r *Reader) follow(owner, state, tool, event string) *Follower { r.mu.Lock() defer r.mu.Unlock() if r.followers == nil { @@ -87,6 +107,12 @@ func (r *Reader) Follow(owner, state, tool string) *Follower { return DecodeAnswer(raw) } r.wireLocked(name, f, nil) + if event != "" { + if r.events == nil { + r.events = map[string]string{} + } + r.events[name] = event + } return f } @@ -96,15 +122,11 @@ func BranchState(state, branch string) string { return state + "." + keys.Escape // FollowBranch is Follow for a state kept per branch: the view of one branch is named `..` // (BranchState), so a reader follows several branches of one state, each its own view and its own row of -// `_first_views` (design 47 G §479). Call it before Start. +// `_first_views` (design 47 G §479). Called after Start, the view takes its first view at once and then follows +// (§532). func (r *Reader) FollowBranch(owner, state, branch, tool string) *Follower { - f := r.Follow(owner, BranchState(state, branch), tool) - r.mu.Lock() - if r.events == nil { - r.events = map[string]string{} - } - r.events[owner+"."+f.State] = state - r.mu.Unlock() + f := r.follow(owner, BranchState(state, branch), tool, state) + r.begin(owner + "." + f.State) return f } @@ -113,11 +135,20 @@ func (r *Reader) FollowBranch(owner, state, branch, tool string) *Follower { // and view-waits, in the reader's findings), its log and clock, and hands each change to Watch, Changed and the // follower's own Changed, set before; Keys, Get, Values, Watch, Statuses and the two tools name it `.`. // Its events are the seat's event (`.`, as the reader's `consumes` names it), which Start subscribes and -// hands to hear, the client's handler. Call it before Start; a state followed already is refused. +// hands to hear, the client's handler; a state followed already is refused. Adopted after Start, as a part adopts a +// watched branch added while it runs, the view takes its first view at once and then follows (design 47 G §532). func (r *Reader) Adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) error { if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil { return errors.New("views: an adopted follower names its owner and state, and its seat's event and handler") } + if err := r.adopt(f, seat, event, hear); err != nil { + return err + } + r.begin(f.Owner + "." + f.State) + return nil +} + +func (r *Reader) adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) error { r.mu.Lock() defer r.mu.Unlock() if r.followers == nil { @@ -150,9 +181,19 @@ func (r *Reader) Adopt(f *Follower, seat, event string, hear func(context.Contex // wireLocked makes f one of the reader's first views under name: its conditions are the reader's, and each change goes // to the watchers, Changed and own (the follower's own Changed, or nil). The caller holds mu. func (r *Reader) wireLocked(name string, f *Follower, own func(string, json.RawMessage, bool, time.Time)) { - f.Raise = func(s Status) { r.raise(s) } + // A view dropped writes no finding after its drop cleared them (Drop). + held := func() bool { r.mu.Lock(); defer r.mu.Unlock(); return r.followers[name] == f } + f.Raise = func(s Status) { + if held() { + r.raise(s) + } + } f.Clear = func(s Status) { r.clear(s) } - f.RaiseWaits = func(s Status) { r.waits(s, true) } + f.RaiseWaits = func(s Status) { + if held() { + r.waits(s, true) + } + } f.ClearWaits = func(s Status) { r.waits(s, false) } f.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) { r.mu.Lock() @@ -224,59 +265,156 @@ func (r *Reader) clear(s Status) { // Start subscribes each owner's views family (the module's `consumes`: `.views.`) and asks every first // view until it is answered, each on a goroutine of its own, so a module starts at once and a reader whose owner is down -// waits, saying so (ADR 0615 §6). It returns when the subscriptions are taken. +// waits, saying so (ADR 0615 §6). It returns when the subscriptions are asked. A view followed or adopted after it is +// begun the same way, and Drop ends one (design 47 G §532). func (r *Reader) Start(ctx context.Context) { r.mu.Lock() + r.ctx, r.started = ctx, true names := append([]string(nil), r.order...) - adopted, events := maps.Clone(r.adopted), maps.Clone(r.events) r.mu.Unlock() - // One subscription per event pattern: the followers of a per-branch state share the owner's one event, which - // HearEvent routes to the branch's view by the key's first token (§521). - subs := map[string][]*Follower{} - var patterns []string for _, name := range names { - f := r.followers[name] - pattern := f.Owner + "." + EventName(f.State) - if a, ok := adopted[name]; ok { - pattern = a.local() - } else if state, ok := events[name]; ok { - pattern = f.Owner + "." + EventName(state) - } - if subs[pattern] == nil { - patterns = append(patterns, pattern) - } - subs[pattern] = append(subs[pattern], f) + r.begin(name) } - for _, pattern := range patterns { - fs := subs[pattern] - handle := r.hear(ctx, fs) +} + +// patternLocked is the event pattern a followed state's events come on: one subscription per pattern, the followers of +// a per-branch state sharing the owner's one event, which HearEvent routes to the branch's view by the key's first token +// (§521). The caller holds mu. +func (r *Reader) patternLocked(name string) string { + if a, ok := r.adopted[name]; ok { + return a.local() + } + f := r.followers[name] + if state, ok := r.events[name]; ok { + return f.Owner + "." + EventName(state) + } + return f.Owner + "." + EventName(f.State) +} + +// begin starts the first view of name once the reader started: at once when its event pattern is heard, else once its +// subscription is taken, which it asks when none is asked yet, so no event said after the first view is missed. +func (r *Reader) begin(name string) { + r.mu.Lock() + defer r.mu.Unlock() + if !r.started || r.followers[name] == nil { + return + } + pattern := r.patternLocked(name) + if r.heard[pattern] { + r.askLocked(name) + return + } + if r.asking[pattern] { + return + } + if r.asking == nil { + r.asking, r.heard = map[string]bool{}, map[string]bool{} + } + r.asking[pattern] = true + ctx, f := r.ctx, r.followers[name] + handle := r.handler(ctx, pattern) + go func() { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := subscribe(pattern, handle) + if err == nil { + break + } + r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, f.Owner, f.State, err, wait) + select { + case <-ctx.Done(): + return + case <-time.After(wait): + } + } + r.mu.Lock() + defer r.mu.Unlock() + r.heard[pattern] = true + for _, n := range r.order { + if r.patternLocked(n) == pattern { + r.askLocked(n) + } + } + }() +} + +// askLocked asks name's first view on a goroutine of its own, once, until answered or the view is dropped. The caller +// holds mu. +func (r *Reader) askLocked(name string) { + if r.begun[name] != nil { + return + } + if r.begun == nil { + r.begun = map[string]context.CancelFunc{} + } + ctx, cancel := context.WithCancel(r.ctx) + r.begun[name] = cancel + f := r.followers[name] + go func() { + if err := f.Start(ctx); err != nil && ctx.Err() == nil { + r.logf("the first view %s was not answered: %v", f.Tool, err) + } + }() +} + +// handler is the handler of one event pattern: the followers on it are read at each event, so a view adopted after the +// subscription hears it and a dropped one no longer does. An event the adopted followers on it hear goes to each +// client's handler; otherwise HearEvent routes it to its state's view. +func (r *Reader) handler(ctx context.Context, pattern string) func(stdio.Envelope) error { + return func(e stdio.Envelope) error { + r.mu.Lock() + var fs []*Follower + for _, n := range r.order { + if r.patternLocked(n) == pattern { + fs = append(fs, r.followers[n]) + } + } + adopted := maps.Clone(r.adopted) + r.mu.Unlock() if h := r.hearAdoptedBy(ctx, fs, adopted); h != nil { - handle = h + return h(e) } - go func() { - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := stdio.Subscribe(pattern, handle) - if err == nil { - break - } - r.logf("not hearing %s yet for %s's %s (%v); asking again in %s", pattern, fs[0].Owner, fs[0].State, err, wait) - select { - case <-ctx.Done(): - return - case <-time.After(wait): - } - } - for _, f := range fs { - go func() { - if err := f.Start(ctx); err != nil { - r.logf("the first view %s was not answered: %v", f.Tool, err) - } - }() - } - }() + return r.hear(ctx, fs)(e) } } +// Drop ends the view of state (`.`), as a part drops a watched branch removed while it runs (design 47 G +// §532, §614): it asks it no more and hears none of its events, its row leaves `_first_views`, and every finding +// it raised (its unanswered first view, its view-waits) is cleared through the reader's findings, the drop and each +// cleared finding said in the log. A state not followed is an error. +func (r *Reader) Drop(state string) error { + r.mu.Lock() + f := r.followers[state] + if f == nil { + r.mu.Unlock() + return fmt.Errorf("%s is not a state this module follows", state) + } + cancel := r.begun[state] + delete(r.followers, state) + delete(r.begun, state) + delete(r.adopted, state) + delete(r.events, state) + delete(r.watchers, state) + r.order = slices.DeleteFunc(r.order, func(n string) bool { return n == state }) + if byState := r.byOwner[f.Owner]; byState[f.State] == f { + delete(byState, f.State) + } + r.mu.Unlock() + if cancel != nil { + cancel() + } + r.logf("views: %s drops its view %s", r.Module, state) + s := f.Status() + if s.Raised { + r.clear(s) + r.logf("views: %s clears its finding %s of the dropped view %s", r.Module, UnansweredKind, state) + } + if s.WaitsRaised { + r.waits(s, false) + r.logf("views: %s clears its finding %s of the dropped view %s", r.Module, ViewWaitsKind, state) + } + return nil +} + // hear is the handler of one followed state's events: HearEvent routes it, and a gap asks that first view once more as a // gap, said in the log. func (r *Reader) hear(ctx context.Context, fs []*Follower) func(stdio.Envelope) error { diff --git a/go/views/status.go b/go/views/status.go index e8e6986a..ecfd4c75 100644 --- a/go/views/status.go +++ b/go/views/status.go @@ -51,6 +51,12 @@ func KeepStatus(ctx context.Context, kv jetstream.KeyValue, logf func(string, .. // HearAgain takes each hand ask written at an AgainKey of kv, those standing at the start included, asks that // follower's first view again with the ask's words and who asked, and deletes the ask. It returns once the watch is set. func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower, logf func(string, ...any)) error { + return hearAgain(ctx, kv, func() []*Follower { return followers }, logf) +} + +// hearAgain is HearAgain over the followers as they are at each ask, so a view adopted after the start is asked again by +// hand too (design 47 G §532). +func hearAgain(ctx context.Context, kv jetstream.KeyValue, current func() []*Follower, logf func(string, ...any)) error { w, err := kv.Watch(ctx, "again.>") if err != nil { return err @@ -74,6 +80,7 @@ func HearAgain(ctx context.Context, kv jetstream.KeyValue, followers []*Follower _ = kv.Delete(ctx, e.Key()) continue } + followers := current() i := slices.IndexFunc(followers, func(f *Follower) bool { return f.Owner == a.Owner && f.State == a.State }) if i < 0 { logf("views: the hand ask %s names no first view this part holds", e.Key()) diff --git a/go/views/views_test.go b/go/views/views_test.go index 86e3ce65..9247b67d 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.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/process" ) // mapSource is a state held in memory, as a Source. @@ -783,6 +784,117 @@ func TestViews(t *testing.T) { t.Errorf("view-waits is not the process reader's finding: %+v", fd) } }) + // G §532, §614: a view adopted after the start takes its first view and follows on the event's one subscription; a + // dropped one hears nothing more, leaves `_first_views`, and the findings it raised are cleared. + t.Run("a view adopted after the start takes its first view and follows, and a dropped one leaves with its findings cleared", func(t *testing.T) { + ended := make(chan bool, 2) + keptBound, keptSubscribe := waitBound, subscribe + waitBound = func(context.Context, time.Duration) bool { return <-ended } + var hmu sync.Mutex + handlers := map[string]func(stdio.Envelope) error{} + subscribe = func(p string, h func(stdio.Envelope) error) error { + hmu.Lock() + defer hmu.Unlock() + handlers[p] = h + return nil + } + defer func() { waitBound, subscribe = keptBound, keptSubscribe }() + ready := func(f *Follower) { + t.Helper() + select { + case <-f.Ready(): + case <-time.After(time.Second): + t.Fatalf("%s has no first view", f.State) + } + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + heard := map[string]*[]string{} + hearer := func(branch string) func(context.Context, []byte) error { + got := &[]string{} + heard[branch] = got + return func(_ context.Context, d []byte) error { *got = append(*got, string(d)); return nil } + } + fs := &signalFindings{put: make(chan Finding, 4)} + r := &Reader{Module: "reader", Findings: fs, Now: now} + main := &Follower{Owner: "s", Tool: "seat:s.revision", State: BranchState("revisions", "main"), Ask: answer} + if err := r.Adopt(main, "s", "revision-said", hearer("main")); err != nil { + t.Fatal(err) + } + r.Start(ctx) + ready(main) + next := &Follower{Owner: "s", Tool: "seat:s.revision", State: BranchState("revisions", "next"), Ask: answer} + if err := r.Adopt(next, "s", "revision-said", hearer("next")); err != nil { + t.Fatal(err) + } + ready(next) + hmu.Lock() + h, n := handlers["s.revision-said.>"], len(handlers) + hmu.Unlock() + if h == nil || n != 1 { + t.Fatalf("subscriptions %v: want the event's one", handlers) + } + event := stdio.Envelope{Key: "s.revision-said.next", Body: []byte(`{"revision":8}`)} + if _ = h(event); len(*heard["main"]) != 1 || len(*heard["next"]) != 1 { + t.Errorf("the event reached %v and %v", *heard["main"], *heard["next"]) + } + next.WaitFor(ctx, "9") + ended <- true + if fd := fs.next(t); fd.Kind != ViewWaitsKind { + t.Fatalf("finding %+v", fd) + } + if err := r.Drop("s." + next.State); err != nil { + t.Fatal(err) + } + if len(fs.m) != 0 { + t.Errorf("the dropped view left its findings %v", fs.m) + } + if st := r.Statuses(); len(st) != 1 || st[0].State != main.State || r.Follows("s."+next.State) { + t.Errorf("first views after the drop: %+v", st) + } + if _ = h(event); len(*heard["next"]) != 1 || len(*heard["main"]) != 2 { + t.Errorf("after the drop the event reached %v and %v", *heard["main"], *heard["next"]) + } + if err := r.Drop("s." + next.State); err == nil { + t.Errorf("a view dropped twice") + } + + // The process form: a view adopted after the start on another event is handled on the module's hub. + pfs := &signalFindings{put: make(chan Finding, 4)} + pr := &ProcessReader{Module: "reader", Findings: func() FindingWriter { return pfs }, Now: now, Hub: process.NewEventHub(t.Logf)} + pm := &Follower{Owner: "s", Tool: "seat:s.revision", State: BranchState("revisions", "main"), Ask: answer} + if _, err := pr.Adopt(pm, "s", "revision-said", hearer("pmain")); err != nil { + t.Fatal(err) + } + pr.Start(ctx, nil, process.Account{}, nil) + ready(pm) + pn := &Follower{Owner: "s", Tool: "seat:s.revision", State: BranchState("revisions", "next"), Ask: answer} + if _, err := pr.Adopt(pn, "s", "revision-other", hearer("pnext")); err != nil { + t.Fatal(err) + } + ready(pn) + if !slices.Contains(pr.Hub.Patterns(), "mesh.seat.s.event.revision-other.>") { + t.Errorf("the hub handles %v", pr.Hub.Patterns()) + } + msg := func() *seatMsg { + return &seatMsg{subject: "mesh.seat.s.event.revision-other.next", data: []byte(`{"revision":8}`)} + } + if pr.Hear(ctx)(msg()); len(*heard["pnext"]) != 1 { + t.Errorf("the process's adopted view heard %v", *heard["pnext"]) + } + pn.WaitFor(ctx, "9") + ended <- true + if fd := pfs.next(t); fd.Kind != ViewWaitsKind { + t.Fatalf("finding %+v", fd) + } + if err := pr.Drop("s", pn.State); err != nil { + t.Fatal(err) + } + m := msg() + if pr.Hear(ctx)(m); len(pfs.m) != 0 || len(pr.Followers()) != 1 || len(*heard["pnext"]) != 1 || !m.acked { + t.Errorf("after the drop: findings %v, followers %d, heard %v, acked %v", pfs.m, len(pr.Followers()), *heard["pnext"], m.acked) + } + }) t.Run("a reader's again tool records its hand act before it writes the ask, and asks nothing when the act is not taken", func(t *testing.T) { var recorded []HandAct take := errors.New("no responders")