G §521: a per-branch state's event names the state, and the key's first token routes it to the branch's view
Builds: design 47 G §521, §479 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
+9
-1
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -553,7 +554,7 @@ func Hear(ctx context.Context, followers map[string]*Follower, logf func(string,
|
||||
|
||||
// HearEvent is Hear for one event given as its data, for a reader that takes the owner's events by another path than
|
||||
// a consumer of its own (the tool runner's): routed to the follower of its state, an event of a state it does not
|
||||
// follow passed over, and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is
|
||||
// follow passed over (a per-branch state's event goes by the first token of its key, §521), and a gap asked once more on its own goroutine as a gap, never as a hand ask (Gap). Data that is
|
||||
// not a views event is answered as an error, and nothing is asked.
|
||||
func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte, logf func(string, ...any)) error {
|
||||
ev, err := DecodeEvent(data)
|
||||
@@ -561,6 +562,13 @@ func HearEvent(ctx context.Context, followers map[string]*Follower, data []byte,
|
||||
return err
|
||||
}
|
||||
f := followers[ev.State]
|
||||
if f == nil && ev.Key != "" {
|
||||
// A per-branch state's event names the state as its owner declares it and the key `<branch>.<revision>`: it goes to
|
||||
// the view `<state>.<branch>` of the key's first token, the branch escaped as a key's name is (design 47 G §521).
|
||||
if branch, _, ok := strings.Cut(ev.Key, "."); ok && branch != "" {
|
||||
f = followers[ev.State+"."+branch]
|
||||
}
|
||||
}
|
||||
if f == nil {
|
||||
logf("views: %s's %s is a state this reader does not follow: event at revision %d passed over", ev.Owner, ev.State, ev.Revision)
|
||||
return nil
|
||||
|
||||
+43
-9
@@ -45,6 +45,8 @@ type Reader struct {
|
||||
// adopted are the followers a seat's client made (Adopt), by "<owner>.<state>": their events are a seat's, heard by
|
||||
// the client's own handler, not an owner's views family.
|
||||
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
|
||||
}
|
||||
|
||||
// adoption is how an adopted follower's events are heard: the seat's event, by its local name, and the client's handler.
|
||||
@@ -95,7 +97,14 @@ func BranchState(state, branch string) string { return state + "." + keys.Escape
|
||||
// (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.
|
||||
func (r *Reader) FollowBranch(owner, state, branch, tool string) *Follower {
|
||||
return r.Follow(owner, BranchState(state, branch), tool)
|
||||
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()
|
||||
return f
|
||||
}
|
||||
|
||||
// Adopt takes a follower a seat's client made (its Ask a seat's verb, such as the catalog half's view) as one of the
|
||||
@@ -218,13 +227,30 @@ func (r *Reader) clear(s Status) {
|
||||
func (r *Reader) Start(ctx context.Context) {
|
||||
r.mu.Lock()
|
||||
names := append([]string(nil), r.order...)
|
||||
adopted := maps.Clone(r.adopted)
|
||||
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, handle := f.Owner+"."+EventName(f.State), r.hear(ctx, f)
|
||||
pattern := f.Owner + "." + EventName(f.State)
|
||||
if a, ok := adopted[name]; ok {
|
||||
pattern, handle = a.local(), r.hearAdopted(ctx, a)
|
||||
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)
|
||||
}
|
||||
for _, pattern := range patterns {
|
||||
fs := subs[pattern]
|
||||
handle := r.hear(ctx, fs)
|
||||
if a, ok := adopted[fs[0].Owner+"."+fs[0].State]; ok {
|
||||
handle = r.hearAdopted(ctx, a)
|
||||
}
|
||||
go func() {
|
||||
for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
|
||||
@@ -232,15 +258,19 @@ func (r *Reader) Start(ctx context.Context) {
|
||||
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)
|
||||
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):
|
||||
}
|
||||
}
|
||||
if err := f.Start(ctx); err != nil {
|
||||
r.logf("the first view %s was not answered: %v", f.Tool, err)
|
||||
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)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -248,9 +278,13 @@ func (r *Reader) Start(ctx context.Context) {
|
||||
|
||||
// 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, f *Follower) func(stdio.Envelope) error {
|
||||
func (r *Reader) hear(ctx context.Context, fs []*Follower) func(stdio.Envelope) error {
|
||||
byState := map[string]*Follower{}
|
||||
for _, f := range fs {
|
||||
byState[f.State] = f
|
||||
}
|
||||
return func(e stdio.Envelope) error {
|
||||
if err := HearEvent(ctx, map[string]*Follower{f.State: f}, e.Body, r.logf); err != nil {
|
||||
if err := HearEvent(ctx, byState, e.Body, r.logf); err != nil {
|
||||
r.logf("views: %s is not a views event, dropped: %v", e.Key, err)
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -642,6 +642,41 @@ func TestViews(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
// Design 47 G §521: the owner's event names the state, the key names the branch; one event reaches the branch's own view.
|
||||
t.Run("an event of a per-branch state goes to the view of its key's first token, one subscription for the branches", func(t *testing.T) {
|
||||
made := now()
|
||||
ask := func(context.Context, string) (Answer, error) {
|
||||
return FirstView("s", "catalog-revisions", mapSource{}, mapSource{"catalog-revisions": markOf("catalog-revisions", 1, made)}, "", now())
|
||||
}
|
||||
r := &Reader{Module: "reader", Now: now}
|
||||
a, b := r.FollowBranch("s", "catalog-revisions", "main", "s.read"), r.FollowBranch("s", "catalog-revisions", "feat/x", "s.read")
|
||||
for _, f := range []*Follower{a, b} {
|
||||
f.Ask = ask
|
||||
if err := f.Start(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
followers := map[string]*Follower{a.State: a, b.State: b}
|
||||
say := func(key string, rev uint64) {
|
||||
data, _ := Event{Owner: "s", State: "catalog-revisions", Key: key, Revision: rev, Created: made, Value: json.RawMessage(`{}`)}.Encode()
|
||||
if err := HearEvent(context.Background(), followers, data, t.Logf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
say("main.r1", 2)
|
||||
say("feat/x.r7", 2)
|
||||
ka, kb := a.Values(), b.Values()
|
||||
if _, ok := ka["main.r1"]; !ok || len(ka) != 1 {
|
||||
t.Errorf("main's view holds %v", ka)
|
||||
}
|
||||
if _, ok := kb["feat/x.r7"]; !ok || len(kb) != 1 {
|
||||
t.Errorf("feat/x's view holds %v", kb)
|
||||
}
|
||||
if r.events["s."+a.State] != "catalog-revisions" || r.events["s."+b.State] != "catalog-revisions" {
|
||||
t.Errorf("events = %v", r.events)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a follower a seat's client made is adopted: the reader's findings, reads and tools hold it, and its seat event goes to the client", func(t *testing.T) {
|
||||
ended := make(chan bool, 1)
|
||||
kept := waitBound
|
||||
|
||||
Reference in New Issue
Block a user