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

G V-5: the views readers adopt a view after their start, and drop one (issue 1966)

Browse Source 
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
<module>_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
This commit is contained in:
jschoubben
2026-11-11 16:04:43 +01:00
parent 6a864aa1b2
commit 7eb6f31c53
5 changed files with 534 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/views/process_reader.go
+207 -50
View File
@@ -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 (`<owner>.<tool>`, from the owner's manifest's tools). The owner's changes
// are ordinary events (`<owner>.views.<state>`), 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 <owner>.<tool>, not %q", r.Module, owner, state, tool)
return nil, false, fmt.Errorf("%s follows %s's %s by its read tool <owner>.<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 `<owner>.<state>.<branch>` (the
// branch escaped as a key's name is, BranchState) and its picture `<owner>_<state>.<branch>`, 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 (`<owner>_<state>`) 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.<owner>.event.views.<state>`) 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
// `<module>_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 (`<event>.<module>.<branch>.<commit or group>`, 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
}
go/views/process_reader_integration_test.go
+16 -1
View File
@@ -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 <module>_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
go/views/reader.go
+192 -54
View File
@@ -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 "<owner>.<state>.<branch>", 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 "<owner>.<state>".
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 (`<module>.<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 `<owner>.<state>.<branch>`
// (BranchState), so a reader follows several branches of one state, each its own view and its own row of
// `<module>_first_views` (design 47 G §479). Call it before Start.
// `<module>_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 `<owner>.<state>`.
// Its events are the seat's event (`<seat>.<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`: `<owner>.views.<state>`) 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 (`<owner>.<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 `<module>_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 {
go/views/status.go
+7
View File
@@ -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())
go/views/views_test.go
+112
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/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 `<module>_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")