Files
mesh-sdk/go/views/views_test.go
T
jschoubben cb5e62dfac
mesh/delivery-group group feat/47-G-join11 checking: 17 of 17 member(s) ready
mesh/delivery closed: a newer head of the same pull request
mesh/check checking cb5e62dfacd9 on 8fbba810609e | novox/mesh-sdk#189 in group feat/47-G-join11
views.ProcessReader.HolderBucket names the holder's real bucket of a seat's view FollowSeat follows
A part that follows a seat's view names its picture in the seat form, but where another part names the holder's own
bucket (the watchman's stuck key of the assigner's wants) the name must be the holder's: HolderBucket answers
`<holder>_<state>` of the holder followed now, "" while none, read at each use so a moved holder is named as it moves.

Builds: design 47 G §717, A-303 (corrected)

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 20:32:16 +01:00

1477 lines
62 KiB
Go

package views
import (
"context"
"encoding/json"
"errors"
"fmt"
"maps"
"reflect"
"runtime"
"slices"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/process"
)
// mapSource is a state held in memory, as a Source.
type mapSource map[string]Entry
func (m mapSource) Keys() ([]string, error) {
var out []string
for k := range m {
out = append(out, k)
}
return out, nil
}
func (m mapSource) Get(key string) (*Entry, bool, error) {
e, ok := m[key]
if !ok {
return nil, false, nil
}
return &e, true, nil
}
// keptSource is a mapSource that keeps each key's history, as a Historian.
type keptSource struct {
mapSource
kept map[string][]Revision
}
func (k keptSource) History(key string) ([]Revision, error) { return k.kept[key], nil }
func markOf(state string, rev uint64, made time.Time) Entry {
raw, _ := json.Marshal(Mark{V: Version, State: state, Revision: rev, Created: made, Fill: Fill{Keys: 2, Bytes: 100, MaxBytes: 1000}})
return Entry{Value: raw, Revision: 1}
}
// The owners' and the readers' side of a first view (ADR 0615 §4-§6, ADR 0621 §2-§3, ADR 0622): one table of cases,
// since mesh-delivery is over its test budget.
func subViews(t *testing.T) {
made := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
now := func() time.Time { return made.Add(time.Hour) }
written := made.Add(-time.Minute)
values := mapSource{"a": {Value: json.RawMessage(`{"n":1}`), Revision: 7, Written: written}, "b": {Value: json.RawMessage(`{"n":2}`), Revision: 9}}
marks := mapSource{"modules": markOf("modules", 9, made)}
t.Run("a first view answers every value at the mark's revision, the mark read first, with each key's written time", func(t *testing.T) {
a, err := FirstView("o", "modules", values, marks, "", now())
if err != nil || a.Revision != 9 || !a.Created.Equal(made) || len(a.Values) != 2 || a.Next != "" {
t.Fatalf("answer = %+v, %v", a, err)
}
raw, _ := json.Marshal(a)
if back, err := DecodeAnswer(raw); err != nil || !back.Written["a"].Equal(written) || len(back.Written) != 1 {
t.Errorf("written times = %v (%v), want a's alone", back.Written, err)
}
if a, _ := FirstView("o", "modules", values, mapSource{}, "", now()); a.Revision != 0 {
t.Errorf("a state not marked answers revision %d, want 0", a.Revision)
}
})
t.Run("a first view larger than a page is paged by key", func(t *testing.T) {
big := mapSource{}
for i := range 4 {
big[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`"` + strings.Repeat("x", PageBytes/3) + `"`), Revision: uint64(i + 1)}
}
got := map[string]bool{}
after := ""
for pages := 0; ; pages++ {
a, err := FirstView("o", "s", big, nil, after, now())
if err != nil || pages > 4 {
t.Fatalf("page %d: %+v %v", pages, a.Next, err)
}
for k := range a.Values {
got[k] = true
}
if a.Next == "" {
break
}
after = a.Next
}
if len(got) != 4 {
t.Errorf("pages hold %v, want the four keys", got)
}
})
t.Run("a first view from a slow source stops at the page budget, with one value at least, and pages on to every key", func(t *testing.T) {
at := time.Unix(0, 0)
clock.set(func() time.Time { return at })
defer clock.set(time.Now)
slow := mapSource{}
for i := range 5 {
slow[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`{"n":1}`), Revision: uint64(i + 1)}
}
got, after := map[string]bool{}, ""
for pages := 0; ; pages++ {
a, err := FirstView("o", "s", &tickSource{slow, &at, PageBudget / 2}, nil, after, now())
if err != nil || pages > 5 || len(a.Values) == 0 || len(a.Values) > 3 {
t.Fatalf("page %d: %d values, %v", pages, len(a.Values), err)
}
for k := range a.Values {
got[k] = true
}
if a.Next == "" {
break
}
after = a.Next
}
if len(got) != 5 {
t.Errorf("pages hold %v, want the five keys", got)
}
})
t.Run("a first view from a source that pages reads it in one request, and from one that cannot, per key", func(t *testing.T) {
p := &pagedSource{mapSource: mapSource{"a": {Value: json.RawMessage(`1`), Revision: 1, Written: written}, "b": {Value: json.RawMessage(`2`), Revision: 2}, "c": {Value: json.RawMessage(`3`), Revision: 3}}}
a, err := FirstView("o", "s", p, nil, "", now())
if err != nil || len(a.Values) != 2 || a.Next != "b" || p.pages != 1 || p.gets != 0 || !a.Written["a"].Equal(written) {
t.Fatalf("answer = %+v, pages %d, gets %d, %v", a, p.pages, p.gets, err)
}
p.unsupported = true
if a, err = FirstView("o", "s", p, nil, "", now()); err != nil || len(a.Values) != 3 || a.Next != "" || p.gets != 3 {
t.Fatalf("fallback answer = %+v, gets %d, %v", a, p.gets, err)
}
})
t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) {
callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner")
for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true,
"novox.mesh-delivery-indexer": false, "novox.mesh-config": false, "person.jochen": false, stdio.Unproven: false} {
if err := Refusal("manifests", callers, caller); (err == nil) != allowed {
t.Errorf("%q: err = %v, allowed %v", caller, err, allowed)
}
}
// A caller argument names nobody: a call with no proof is refused.
tool := FirstViewTool("mesh-delivery-indexer", "manifests", "modules", "w.", callers, values, marks, now)
if _, err := tool.Run(map[string]any{"caller": "novox.mesh-grants"}); err == nil {
t.Errorf("an unproven call naming a caller is answered")
}
})
t.Run("a read tool says the keys and the fill, one key, one key's kept history, or a page of values", func(t *testing.T) {
r, err := Read("o", "modules", values, marks, "", false, "", now())
ra, ok := r.(ReadAnswer)
if err != nil || !ok || len(ra.Keys) != 2 || ra.Fill == nil || ra.Fill.MaxBytes != 1000 || ra.Revision != 9 {
t.Fatalf("read = %+v, %v", r, err)
}
r, _ = Read("o", "modules", values, marks, "b", false, "", now())
if k := r.(KeyAnswer); !k.Found || k.Revision != 9 {
t.Errorf("one key = %+v", k)
}
r, _ = Read("o", "modules", values, nil, "", false, "", now())
if ra := r.(ReadAnswer); ra.Fill != nil || ra.Words == "" {
t.Errorf("an unmarked state says its fill as %+v", ra)
}
kept := []Revision{{Revision: 3, At: written, Op: "put", Value: json.RawMessage(`{"n":0}`)},
{Revision: 5, At: written.Add(time.Second), Op: "delete", Value: json.RawMessage(`null`)}, {Revision: 7, At: written, Op: "put", Value: json.RawMessage(`{"n":1}`)}}
tool := ReadTool("o", "o_read_modules", "modules", "w", keptSource{values, map[string][]Revision{"a": kept}}, marks, now)
r, err = tool.Run(map[string]any{"key": "a", "history": true})
raw, _ := json.Marshal(r)
var h struct {
History []Revision `json:"history"`
}
if err != nil || json.Unmarshal(raw, &h) != nil || len(h.History) != 3 || h.History[1].Op != "delete" || h.History[2].Revision != 7 || !h.History[0].At.Equal(written) {
t.Errorf("a key's history = %s, %v", raw, err)
}
if r, err = tool.Run(map[string]any{"key": "none", "history": true}); err != nil || r.(HistoryAnswer).History == nil {
t.Errorf("a key with nothing kept answers %+v, %v, not an empty history", r, err)
}
if _, err := tool.Run(map[string]any{"history": true}); err == nil {
t.Error("a history naming no key is answered")
}
if _, err := ReadTool("o", "o_read_modules", "modules", "w", values, marks, now).Run(map[string]any{"key": "a", "history": true}); err == nil {
t.Error("a source keeping no history answers one")
}
})
// A follower whose owner answers from values and marks.
follower := func(ask func(context.Context, string) (Answer, error)) (*Follower, *[]string) {
var said []string
f := &Follower{Owner: "o", Tool: "o.modules", State: "modules", Ask: ask, Bound: time.Nanosecond, Wait: time.Millisecond,
Raise: func(s Status) { said = append(said, "raised "+s.Tool) }, Clear: func(s Status) { said = append(said, "cleared") }}
return f, &said
}
answer := func(context.Context, string) (Answer, error) {
return FirstView("o", "modules", values, marks, "", now())
}
ev := func(rev uint64, key, value string) Event {
e := Event{V: Version, Owner: "o", State: "modules", Key: key, Revision: rev, Created: made}
if value == "" {
e.Deleted = true
} else {
e.Value = json.RawMessage(value)
}
return e
}
t.Run("a reader applies a put and a delete from the events alone, and passes over what it holds", func(t *testing.T) {
asks := 0
f, _ := follower(func(ctx context.Context, after string) (Answer, error) { asks++; return answer(ctx, after) })
if err := f.Start(context.Background()); err != nil {
t.Fatal(err)
}
for _, e := range []Event{ev(8, "a", `{"n":0}`), ev(10, "c", `{"n":3}`), ev(11, "a", "")} {
if gap, why := f.Apply(e); gap {
t.Fatalf("revision %d is a gap: %s", e.Revision, why)
}
}
if _, ok := f.Get("a"); ok || f.Status().Revision != 11 || asks != 1 {
t.Errorf("picture = %v at %d after %d asks", f.Values(), f.Status().Revision, asks)
}
if v, _ := f.Get("c"); string(v) != `{"n":3}` {
t.Errorf("c = %s", v)
}
})
t.Run("a gap, a state made again and a value too large each ask once more", func(t *testing.T) {
for name, e := range map[string]Event{
"a revision missed": ev(12, "a", `{}`),
"made again": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 1, Created: made.Add(time.Minute), Value: json.RawMessage(`{}`)},
"too large": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 10, Created: made, Gap: true},
} {
f, _ := follower(answer)
_ = f.Start(context.Background())
if gap, _ := f.Apply(e); !gap {
t.Errorf("%s: not taken as a gap", name)
}
}
})
t.Run("a change said after revisions written over is no gap; one said after a revision the picture lacks, or a mark past the picture, asks once more and changes nothing", func(t *testing.T) {
f, _ := follower(answer)
_ = f.Start(context.Background())
over := ev(12, "a", `{"n":5}`)
over.Prev = 9 // 10 and 11 were written over before the owner's watch was handed them
if gap, why := f.Apply(over); gap || f.Status().Revision != 12 {
t.Fatalf("a change said after 9 on a picture at 9: gap %v (%s), at %d", gap, why, f.Status().Revision)
}
missed := ev(14, "b", `{"n":6}`)
missed.Prev = 13
mark := func(rev uint64) Event {
return Event{V: Version, Owner: "o", State: "modules", Revision: rev, Created: made, Mark: true}
}
for _, c := range []struct {
e Event
gap bool
}{{missed, true}, {mark(12), false}, {mark(11), false}, {mark(13), true}} {
if gap, why := f.Apply(c.e); gap != c.gap {
t.Errorf("revision %d (mark %v) on a picture at 12: gap %v, want %v (%s)", c.e.Revision, c.e.Mark, gap, c.gap, why)
}
}
if v, _ := f.Get("a"); string(v) != `{"n":5}` || f.Status().Revision != 12 || len(f.Values()) != 2 {
t.Errorf("a gap or a mark changed the picture: %v at %d", f.Values(), f.Status().Revision)
}
})
t.Run("a reader that never heard the owner's last change asks its first view on the owner's mark alone, and holds what it read", func(t *testing.T) {
later := mapSource{"a": values["a"], "b": values["b"], "c": {Value: json.RawMessage(`{"n":3}`), Revision: 10}}
asks := 0
f, _ := follower(func(ctx context.Context, after string) (Answer, error) {
asks++
if asks == 1 {
return answer(ctx, after)
}
return FirstView("o", "modules", later, mapSource{"modules": markOf("modules", 10, made)}, after, now())
})
_ = f.Start(context.Background())
got := make(chan string, 4)
f.Changed = func(key string, _ json.RawMessage, _ bool, _ time.Time) { got <- key }
// The change at 10 (c) never reached this reader; only the mark the owner said after it does.
data, _ := Event{V: Version, Owner: "o", State: "modules", Revision: 10, Created: made, Mark: true}.Encode()
if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, data, t.Logf); err != nil {
t.Fatal(err)
}
if k := <-got; k != "c" {
t.Fatalf("changed %s", k)
}
if v, _ := f.Get("c"); string(v) != `{"n":3}` || asks != 2 {
t.Errorf("c = %s after %d asks", v, asks)
}
})
t.Run("a reader whose owner is down keeps asking, raises past its bound, and the answer clears it", func(t *testing.T) {
down := 3
f, said := follower(func(ctx context.Context, after string) (Answer, error) {
if down > 0 {
down--
return Answer{}, errors.New("no responders")
}
return answer(ctx, after)
})
if err := f.Start(context.Background()); err != nil {
t.Fatal(err)
}
if strings.Join(*said, ",") != "raised o.modules,cleared" || f.Status().Asks != 4 || f.Status().Raised {
t.Errorf("said %v after %d asks", *said, f.Status().Asks)
}
})
// W1.4 (design 47 *A part with no view says it waits*): the status says the view held, the one waited for and since
// when; past the bound the part's view-waits is raised once, and holding the view clears it.
t.Run("a part that waits for a view says so, raises view-waits past its bound, and holding it clears that", func(t *testing.T) {
ended := make(chan bool)
kept := waitBound.get()
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
defer waitBound.set(kept)
var raised, cleared, said []Status
var mu sync.Mutex
raisedCh := make(chan struct{}, 4) // told after the part's raise, so the test reads it only once it is made
f := &Follower{Owner: "o", Tool: "o.revision", State: BranchState("catalog", "trunk"), Now: now,
RaiseWaits: func(s Status) { mu.Lock(); raised = append(raised, s); mu.Unlock(); raisedCh <- struct{}{} },
ClearWaits: func(s Status) { mu.Lock(); cleared = append(cleared, s); mu.Unlock() },
Said: func(s Status) { mu.Lock(); said = append(said, s); mu.Unlock() }}
f.Holds("trunk.11")
f.WaitFor(context.Background(), "trunk.12")
f.WaitFor(context.Background(), "trunk.12") // the same view again: nothing new, no second bound
if s := f.Status(); s.View != "trunk.11" || s.WaitsFor != "trunk.12" || !s.WaitingSince.Equal(now()) || s.WaitsRaised {
t.Fatalf("status = %+v", s)
}
ended <- true
select {
case <-raisedCh:
case <-time.After(10 * time.Second):
t.Fatal("the bound ended and view-waits was not raised")
}
mu.Lock()
if len(raised) != 1 || raised[0].WaitsFor != "trunk.12" {
t.Errorf("raised = %+v", raised)
}
mu.Unlock()
fd := viewWaitsFinding("mesh-delivery-builder", f.Status(), now())
if fd.kind != ViewWaitsKind || fd.id != "mesh-delivery-builder.o.catalog.trunk" || !strings.Contains(fd.summary, "waits for the view o.catalog.trunk at revision trunk.12") ||
!strings.Contains(fd.summary, "it holds trunk.11") {
t.Errorf("finding = %+v", fd)
}
none := &Follower{Owner: "o", Tool: "o.revision", State: BranchState("catalog", "new"), Now: now,
Ask: func(context.Context, string) (Answer, error) {
return Answer{V: 1, Owner: "o", State: "catalog.new", At: now()}, nil
}}
if err := none.Start(context.Background()); err != nil {
t.Fatal(err)
}
select {
case <-none.Ready():
default:
t.Errorf("a view of a branch with no revision is not ready")
}
if len(none.Values()) != 0 {
t.Errorf("not empty")
}
none.WaitFor(context.Background(), NoneYet) // a branch with no revision yet: ready, empty, nothing awaited (§495)
if s := none.Status(); s.WaitsFor != "" || !s.WaitingSince.IsZero() {
t.Errorf("a view of none yet is awaited: %+v", s)
}
f.Holds("trunk.12")
if s := f.Status(); s.View != "trunk.12" || s.WaitsFor != "" || s.WaitsRaised {
t.Errorf("after holding: %+v", s)
}
mu.Lock()
if len(cleared) != 1 || len(said) == 0 || said[len(said)-1].View != "trunk.12" {
t.Errorf("cleared %d, last said %+v", len(cleared), said)
}
mu.Unlock()
f.WaitFor(context.Background(), "trunk.13")
f.Holds("trunk.13")
ended <- true // the bound of a view now held raises nothing
for i := 0; i < 100; i++ {
runtime.Gosched()
}
mu.Lock()
if len(raised) != 1 {
t.Errorf("a view held before its bound raised: %+v", raised)
}
mu.Unlock()
})
t.Run("an event heard while an ask runs is taken again after its answer", func(t *testing.T) {
var f *Follower
f, _ = follower(func(ctx context.Context, after string) (Answer, error) {
a, err := answer(ctx, after) // the picture read at 9, before revision 10 was written
if f.Status().Asks == 2 {
f.Apply(ev(10, "d", `{"n":4}`))
}
return a, err
})
_ = f.Start(context.Background())
_ = f.Again(context.Background(), "test")
if v, ok := f.Get("d"); !ok || string(v) != `{"n":4}` || f.Status().Revision != 10 || f.Status().ByHand != 1 {
t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision)
}
})
t.Run("a reader that heard an event its answer does not hold asks again by itself on its Wait, with no further event", func(t *testing.T) {
// The owner's state: revision 9 at first, 12 once its writer is done. The picture a first view reads lags the events.
for name, drop := range map[string]func(f *Follower, owner *uint64){
// A gap heard while the ask runs: the running ask swallowed the gap's own ask and the event was not consecutive.
"a gap heard while an ask runs": func(f *Follower, owner *uint64) { *owner = 12; f.Apply(ev(12, "e", `{"n":5}`)) },
// An event heard before the first view, passed over, the owner answering from before its write.
"an event heard before the first view": func(f *Follower, owner *uint64) { f.Apply(ev(10, "e", `{"n":5}`)) },
} {
t.Run(name, func(t *testing.T) {
owner := uint64(9)
var f *Follower
asks := 0
f, _ = follower(func(context.Context, string) (Answer, error) {
asks++
a := Answer{V: Version, Owner: "o", State: "modules", Revision: owner, Created: made, Values: map[string]json.RawMessage{"e": json.RawMessage(`{"n":5}`)}}
if asks == 1 && name == "a gap heard while an ask runs" {
drop(f, &owner) // heard while this ask runs; the answer is the one read before it
}
if asks == 2 {
owner = 12 // the owner has written it by the second ask
a.Revision = owner
}
return a, nil
})
if name == "an event heard before the first view" {
drop(f, &owner)
}
caught := make(chan struct{})
var once sync.Once
f.Said = func(s Status) {
if s.Revision >= 10 {
once.Do(func() { close(caught) })
}
}
if err := f.Start(context.Background()); err != nil {
t.Fatal(err)
}
// No further event follows: the follower asks once more by itself and holds what was heard.
select {
case <-caught:
case <-time.After(10 * time.Second):
t.Fatal("the reader stayed behind the event it heard")
}
if s := f.Status(); s.Revision < 10 || s.Heard < 10 || s.Behind != "" || f.Behind() != "" || s.Asks != 2 || !strings.HasPrefix(s.Why, "behind") {
t.Errorf("revision %d after %d asks (%q), behind: %q", s.Revision, s.Asks, s.Why, f.Behind())
}
})
}
})
t.Run("a consumer made after the first view was asked asks it again by itself, and the status says why until then", func(t *testing.T) {
asked := make(chan struct{}, 8)
var f *Follower
f, _ = follower(func(ctx context.Context, after string) (Answer, error) {
asked <- struct{}{}
return answer(ctx, after)
})
var said []Status
var mu sync.Mutex
var clock atomic.Int64
clock.Store(int64(time.Hour))
f.Now = func() time.Time { return made.Add(time.Duration(clock.Load())) }
f.Said = func(s Status) { mu.Lock(); said = append(said, s); mu.Unlock() }
_ = f.Start(context.Background())
<-asked
// The bus made the consumer after the ask: an event said between is never delivered to it.
f.ConsumerMade(context.Background(), made.Add(2*time.Hour))
if b := f.Status().Behind; !strings.Contains(b, "consumer") {
t.Errorf("behind = %q", b)
}
clock.Store(int64(3 * time.Hour)) // a later ask is after the consumer's making
select {
case <-asked:
case <-time.After(10 * time.Second):
t.Fatal("a consumer made after the first view asked nothing")
}
})
t.Run("an event heard but not in the picture is in the status as heard, with why it is behind", func(t *testing.T) {
f, _ := follower(answer)
_ = f.Start(context.Background())
f.Apply(ev(12, "a", `{}`)) // a gap: heard, not taken
if s := f.Status(); s.Heard != 12 || !strings.Contains(s.Behind, "12") {
t.Errorf("status = %+v", s)
}
})
t.Run("an event taken into the picture is said in the status, so a read tool shows the revision held", func(t *testing.T) {
var told []uint64
f, _ := follower(answer)
f.Said = func(s Status) { told = append(told, s.Revision) }
_ = f.Start(context.Background())
told = nil
f.Apply(ev(10, "c", `{"n":3}`))
if len(told) != 1 || told[0] != 10 {
t.Errorf("said revisions %v, want [10]", told)
}
})
t.Run("an event heard by its data asks a gap once more as a gap, never by hand", func(t *testing.T) {
asked := make(chan struct{}, 4)
f, _ := follower(func(ctx context.Context, after string) (Answer, error) {
asked <- struct{}{}
return answer(ctx, after)
})
_ = f.Start(context.Background())
<-asked
raw, _ := ev(12, "a", `{}`).Encode()
if err := HearEvent(context.Background(), map[string]*Follower{"modules": f}, raw, t.Logf); err != nil {
t.Fatal(err)
}
select {
case <-asked:
case <-time.After(3 * time.Second):
t.Fatal("a gap heard by its data asked nothing")
}
if s := f.Status(); s.Asks != 2 || s.ByHand != 0 {
t.Errorf("a gap was counted by hand: %+v", s)
}
if err := HearEvent(context.Background(), nil, []byte("not an event"), t.Logf); err == nil {
t.Error("data that is no views event was taken")
}
})
t.Run("a module's reader answers Keys and Get from its pictures, refuses a state it does not follow or has no first view of, and writes and clears its finding", func(t *testing.T) {
acted := 0
fs := &findingsMap{}
r := &Reader{Module: "reader", Findings: fs, Bound: time.Nanosecond, Wait: time.Millisecond, Acts: actsFunc(func(HandAct) error {
acted++
return nil
})}
f := r.Follow("o", "modules", "o.indexer_read_modules")
down := 2
f.Ask = func(ctx context.Context, after string) (Answer, error) {
if down > 0 {
down--
return Answer{}, errors.New("no responders")
}
return answer(ctx, after)
}
if _, err := r.Keys("o.modules"); err == nil {
t.Errorf("a state with no first view is read")
}
if _, err := r.Keys("o.other"); err == nil {
t.Errorf("a state not followed is read")
}
if err := f.Start(context.Background()); err != nil {
t.Fatal(err)
}
keys, err := r.Keys("o.modules")
v, ok, _ := r.Get("o.modules", "b")
if err != nil || len(keys) != 2 || !ok || string(v) != `{"n":2}` || len(fs.m) != 0 || fs.puts != 1 || fs.deletes != 1 {
t.Errorf("keys %v, b %s, findings %+v", keys, v, fs)
}
handed := map[string]bool{}
if err := r.Watch("o.modules", func(k string, _ []byte, gone bool) { handed[k] = !gone }); err != nil || len(handed) != 2 || !r.Follows("o.modules") || r.Follows("o.x") {
t.Errorf("watch handed %v, %v", handed, err)
}
f.Apply(ev(10, "c", `{"n":3}`))
f.Apply(ev(11, "a", ""))
if !handed["c"] || handed["a"] {
t.Errorf("a change was not handed to the watch: %v", handed)
}
if got := r.Tools("reader"); len(got) != 2 || got[0].Name != "reader_first_views" {
t.Errorf("tools = %v", got)
}
again, err := r.Tools("reader")[1].Run(map[string]any{"state": "o.modules"})
if err != nil || len(again.(map[string]any)["views"].([]Status)) != 1 || f.Status().ByHand != 1 || acted != 1 {
t.Errorf("asked again by hand: %v, %v", again, err)
}
// a list added by Rows is said beside the statuses (design 47 G A-103), and the entry-not-found finding is written and cleared
r.Rows("entries", func() any { return []string{"fp1"} })
seen, _ := r.Tools("reader")[0].Run(nil)
if got := seen.(map[string]any)["entries"]; !reflect.DeepEqual(got, []string{"fp1"}) {
t.Errorf("first views entries = %v", got)
}
nf := &findingsMap{}
if err := WriteEntryNotFound(nf, "fp1", now(), now()); err != nil || nf.m[EntryNotFoundKind+"/fp1"].Summary == "" {
t.Errorf("entry-not-found not written: %+v, %v", nf.m, err)
}
if err := ClearEntryNotFound(nf, "fp1"); err != nil || len(nf.m) != 0 {
t.Errorf("entry-not-found not cleared: %+v, %v", nf.m, err)
}
})
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 {
return nil, fmt.Errorf("asked %s with %v", tool, body)
}
a, err := FirstView("o", "modules", values, marks, "", now())
raw, _ := json.Marshal(map[string]any{"result": a})
return raw, err
}}
p, err := r.Follow("o", "modules", "o.o_read_modules")
if err != nil {
t.Fatal(err)
}
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
}
}
if _, err := p.Get(context.Background(), "a"); err == nil {
t.Errorf("a picture with no first view is read")
}
if _, err := r.KeyValue(context.Background(), "o_other"); err == nil {
t.Errorf("a bucket not followed is opened")
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
r.Ask(ctx, nil)
<-r.Ready()
kv, err := r.KeyValue(ctx, "o_modules")
if err != nil {
t.Fatal(err)
}
e, err := kv.Get(ctx, "a")
if err != nil || string(e.Value()) != `{"n":1}` || !e.Created().Equal(written) {
t.Fatalf("a = %v, %v", e, err)
}
if _, err := kv.Put(ctx, "a", nil); !errors.Is(err, ErrReadOnly) {
t.Errorf("a write to a picture: %v", err)
}
w, _ := kv.Watch(ctx, "*")
var seen []string
for e := range w.Updates() {
if e == nil {
break
}
seen = append(seen, e.Key())
}
at := made.Add(2 * time.Minute)
ch := ev(10, "c", `{"n":3}`)
ch.At = at
p.Follower().Apply(ch)
if e := <-w.Updates(); strings.Join(seen, ",") != "a,b" || e.Key() != "c" || !e.Created().Equal(at) {
t.Errorf("watch handed %v then %v", seen, e)
}
_ = w.Stop()
})
// A seat's client makes the follower (its Ask a seat's verb) and both readers adopt it, so its callers write no glue:
// the reader's findings (unanswered, view-waits), Keys, Get, Watch, statuses and tools, and the seat's event handed to
// the client's handler.
// Design 47 G §479: a per-branch state's view is named <owner>.<state>.<branch>, several branches in one reader.
t.Run("a per-branch state is followed once per branch, each view its own row named with its escaped branch", func(t *testing.T) {
if got := BranchState("catalog-revisions", "feat/x y"); got != "catalog-revisions.feat/x=20y" {
t.Errorf("BranchState = %q", got)
}
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")
r.Follow("s", "plain", "s.read")
if a.State != "catalog-revisions.main" || b.State != "catalog-revisions.feat/x" {
t.Errorf("states %q %q", a.State, b.State)
}
for _, n := range []string{"s.catalog-revisions.main", "s.catalog-revisions.feat/x", "s.plain"} {
if !r.Follows(n) {
t.Errorf("%s is not followed", n)
}
}
if r.Follows("s.catalog-revisions") {
t.Errorf("the unnamed per-branch state is followed")
}
if st := r.Statuses(); len(st) != 3 || st[0].State != "catalog-revisions.main" || st[1].State != "catalog-revisions.feat/x" || st[2].State != "plain" {
t.Errorf("rows = %+v", st)
}
pr := &ProcessReader{Module: "reader"}
p1, err1 := pr.FollowBranch("s", "catalog-revisions", "main", "s.read")
p2, err2 := pr.FollowBranch("s", "catalog-revisions", "feat/x", "s.read")
if err1 != nil || err2 != nil || p1 == p2 || p1.Bucket() != "s_catalog-revisions.main" || pr.Picture("s_catalog-revisions.feat/x") != p2 {
t.Errorf("pictures %v %v %v", p1, p2, err1)
}
})
// 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.get()
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
defer waitBound.set(kept)
fs := &signalFindings{put: make(chan seenFinding, 4)}
r := &Reader{Module: "reader", Findings: fs, Now: now}
var own []string
f := &Follower{Owner: "s", Tool: "seat:s.revision", State: "revisions", Ask: answer,
Changed: func(k string, _ json.RawMessage, _ bool, _ time.Time) { own = append(own, k) }}
var heard []string
hear := func(_ context.Context, data []byte) error { heard = append(heard, string(data)); return nil }
if err := r.Adopt(f, "s", "revision-said", hear); err != nil {
t.Fatal(err)
}
if err := r.Adopt(f, "s", "revision-said", hear); err == nil {
t.Errorf("a state followed already is adopted twice")
}
if err := r.Adopt(&Follower{Owner: "s", State: "x"}, "s", "", hear); err == nil {
t.Errorf("a follower with no seat event is adopted")
}
if err := f.Start(context.Background()); err != nil {
t.Fatal(err)
}
keys, err := r.Keys("s.revisions")
if v, ok, _ := r.Get("s.revisions", "b"); err != nil || len(keys) != 2 || !ok || string(v) != `{"n":2}` || len(own) != 2 {
t.Errorf("keys %v (%v), b %s, own Changed %v", keys, err, v, own)
}
if st := r.Statuses(); len(st) != 1 || st[0].Tool != "seat:s.revision" {
t.Errorf("statuses = %+v", st)
}
_ = r.hearAdopted(context.Background(), r.adopted["s.revisions"])(stdio.Envelope{Key: "s.revision-said", Body: []byte(`{"revision":3}`)})
if r.adopted["s.revisions"].local() != "s.revision-said.>" || len(heard) != 1 {
t.Errorf("the seat event %v reached the client %v", r.adopted, heard)
}
// A later revision, said with a further part (`<seat>.<event>.<part>`), reaches the follower by the runtime's own matching.
if pat := r.adopted["s.revisions"].local(); !stdio.TopicMatches(pat, "s.revision-said.catalog-revision-9") || !stdio.TopicMatches(pat, "s.revision-said.a.b") {
t.Errorf("the pattern %q does not hear a later catalog revision", pat)
}
f.WaitFor(context.Background(), "3")
ended <- true
if fd := fs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {
t.Errorf("view-waits is not the reader's finding: %+v", fd)
}
f.Holds("3")
if len(fs.m) != 0 {
t.Errorf("holding the view left %+v", fs.m)
}
// The process form: the picture reads the adopted follower, and the seat's event on the durable goes to the client.
pfs := &signalFindings{put: make(chan seenFinding, 4)}
pr := &ProcessReader{Module: "reader", Findings: func() WireWriter { return pfs }, Now: now}
pf := &Follower{Owner: "s", Tool: "seat:s.revision", State: "revisions", Ask: answer}
p, err := pr.Adopt(pf, "s", "revision-said", hear)
if err != nil || p.Bucket() != "s_revisions" || len(pr.Followers()) != 1 {
t.Fatalf("adopted %v, %v", p, err)
}
if err := pf.Start(context.Background()); err != nil {
t.Fatal(err)
}
if e, err := p.Get(context.Background(), "a"); err != nil || string(e.Value()) != `{"n":1}` {
t.Errorf("a = %v, %v", e, err)
}
m := &seatMsg{subject: "mesh.seat.s.event.revision-said", data: []byte(`{"revision":4}`)}
pr.Hear(context.Background())(m)
if len(heard) != 2 || heard[1] != `{"revision":4}` || !m.acked {
t.Errorf("the seat event on the durable: heard %v, acked %v", heard, m.acked)
}
// G §521, §532: an event's further parts reach the handler too
pr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.runner.main.abc", data: []byte(`{"revision":5}`)})
if len(heard) != 3 {
t.Errorf("an event with further parts: heard %v", heard)
}
// G §532: one subscription per event, `<event>.>`, and every follower adopted on the event hears it, two
// branches' views on one reader each
two := &Reader{Module: "reader", Now: now}
var one, other []string
for branch, got := range map[string]*[]string{"main": &one, "next": &other} {
f := &Follower{Owner: "s", State: BranchState("revisions", branch), Ask: answer}
if err := two.Adopt(f, "s", "revision-said", func(_ context.Context, d []byte) error { *got = append(*got, string(d)); return nil }); err != nil {
t.Fatal(err)
}
}
var fol []*Follower
for _, n := range two.order {
fol = append(fol, two.followers[n])
}
if err := two.hearAdoptedBy(context.Background(), fol, two.adopted)(stdio.Envelope{Key: "s.revision-said.main", Body: []byte(`{"revision":6}`)}); err != nil || len(one) != 1 || len(other) != 1 {
t.Errorf("two adopted views on one reader heard %v and %v (%v)", one, other, err)
}
var a, b []string
prr := &ProcessReader{Module: "reader", Now: now}
for i, got := range []*[]string{&a, &b} {
f := &Follower{Owner: "s", State: BranchState("revisions", []string{"main", "next"}[i]), Ask: answer}
if _, err := prr.Adopt(f, "s", "revision-said", func(_ context.Context, d []byte) error { *got = append(*got, string(d)); return nil }); err != nil {
t.Fatal(err)
}
}
prr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.main", data: []byte(`{"revision":7}`)})
if len(a) != 1 || len(b) != 1 {
t.Errorf("two adopted views on one process reader heard %v and %v", a, b)
}
// A-185, A-199: an event adopted with no picture is heard once on the same subscription, beside the views; a reader
// following no view hears it too.
var hooked []string
hook := func(_ context.Context, d []byte) error { hooked = append(hooked, string(d)); return nil }
if err := prr.AdoptEvent("s", "revision-said", hook); err != nil || prr.AdoptEvent("s", "", hook) == nil {
t.Fatalf("AdoptEvent: %v, and one naming no event was taken", err)
}
prr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.main", data: []byte(`{"revision":8}`)})
if len(a) != 2 || len(b) != 2 || len(hooked) != 1 || len(prr.Followers()) != 2 {
t.Errorf("beside two views the adopted event heard %v (views %v, %v)", hooked, a, b)
}
alone := &ProcessReader{Module: "reader", Now: now}
if err := alone.AdoptEvent("s", "revision-said", hook); err != nil {
t.Fatal(err)
}
am := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":9}`)}
alone.Hear(context.Background())(am)
if len(hooked) != 2 || !am.acked {
t.Errorf("a reader following no view heard %v, acked %v", hooked, am.acked)
}
// A-222: an adopted event whose handler fails is left unacknowledged, so the bus asks again.
failing := &ProcessReader{Module: "reader", Now: now}
var tries int
if err := failing.AdoptEvent("s", "revision-said", func(context.Context, []byte) error {
tries++
if tries == 1 {
return errors.New("not placed yet")
}
return nil
}); err != nil {
t.Fatal(err)
}
fm := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":10}`)}
failing.Hear(context.Background())(fm)
if fm.acked {
t.Error("a failed adopted event was acknowledged")
}
fm2 := &seatMsg{subject: fm.subject, data: fm.data}
failing.Hear(context.Background())(fm2)
if !fm2.acked || tries != 2 {
t.Errorf("the event asked again: acked %v after %d tries", fm2.acked, tries)
}
pf.WaitFor(context.Background(), "4")
ended <- true
if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {
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.get(), subscribe.get()
waitBound.set(func(context.Context, time.Duration) bool { return <-ended })
var hmu sync.Mutex
handlers := map[string]func(stdio.Envelope) error{}
subscribe.set(func(p string, h func(stdio.Envelope) error) error {
hmu.Lock()
defer hmu.Unlock()
handlers[p] = h
return nil
})
defer func() { waitBound.set(keptBound); subscribe.set(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 seenFinding, 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 seenFinding, 4)}
pr := &ProcessReader{Module: "reader", Findings: func() WireWriter { 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)
}
})
// Design 47 G §717: a view found by its seat starts without a holder, waits naming the seat, follows the holder the
// seat-holders view names at its change, and a moved holder is followed as it moves; nothing is asked between changes.
t.Run("a seat's view starts with no holder, follows the one the seat-holders view names, and moves with it, asking nothing between changes", func(t *testing.T) {
ended := make(chan bool, 2)
keptBound := waitBound.get()
waitBound.set(func(ctx context.Context, _ time.Duration) bool {
select {
case v := <-ended:
return v
case <-ctx.Done():
return false
}
})
defer waitBound.set(keptBound)
var mu sync.Mutex
asked := map[string]int{}
states := map[string]map[string]json.RawMessage{
"assigner.seat-holders": {},
"h1.asks": {"hold.a": json.RawMessage(`{"n":1}`), "hold.b": json.RawMessage(`{"n":2}`)},
"h2.asks": {"hold.a": json.RawMessage(`{"n":3}`)},
}
fs := &signalFindings{put: make(chan seenFinding, 4)}
r := &ProcessReader{Module: "teller", Findings: func() WireWriter { return fs }, Now: now, Wait: time.Millisecond,
Call: func(tool string, _ any) (json.RawMessage, error) {
mu.Lock()
defer mu.Unlock()
asked[tool]++
vs, ok := states[tool]
if !ok {
return nil, fmt.Errorf("no tool %s", tool)
}
owner, state, _ := strings.Cut(tool, ".")
raw, _ := json.Marshal(Answer{V: Version, Owner: owner, State: state, Revision: 1, Created: made, Values: vs})
return raw, nil
}}
sh, err := r.Follow("assigner", "seat-holders", "assigner.seat-holders")
if err != nil {
t.Fatal(err)
}
holders := func(seat string) []string {
raw, ok := sh.Follower().Get(seat)
var held struct {
Holders []struct{ Module string } `json:"holders"`
}
if !ok || json.Unmarshal(raw, &held) != nil {
return nil
}
var out []string
for _, h := range held.Holders {
out = append(out, h.Module)
}
return out
}
p, err := r.FollowSeat("mesh-wants", "asks", holders)
if err != nil || p.Bucket() != "mesh-wants_asks" {
t.Fatalf("FollowSeat = %v, %v", p, err)
}
if again, _ := r.FollowSeat("mesh-wants", "asks", holders); again != p {
t.Errorf("a seat's view followed twice is two pictures")
}
if _, err := p.Get(context.Background(), "hold.a"); err == nil {
t.Errorf("a seat's view with no holder reads as held")
}
if b := r.HolderBucket("mesh-wants", "asks"); b != "" {
t.Errorf("with no holder the holder's bucket is %q", b)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
r.Ask(ctx, nil)
<-sh.Follower().Ready()
// no holder: past the bound the reader writes view-waits naming the seat and the state
ended <- true
if fd := fs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "teller.mesh-wants.asks" || !strings.Contains(fd.Summary, "seat mesh-wants") {
t.Fatalf("view-waits of a seat with no holder: %+v", fd)
}
if s := p.Follower().Status(); s.Owner != "mesh-wants" || s.WaitsFor != HolderWait {
t.Errorf("the waiting status names %q waiting for %q", s.Owner, s.WaitsFor)
}
ready := func(want string) {
t.Helper()
deadline := time.After(2 * time.Second)
for {
if e, err := p.Get(ctx, "hold.a"); err == nil && string(e.Value()) == want {
return
}
select {
case <-deadline:
t.Fatalf("the seat's picture never held hold.a %s", want)
case <-time.After(time.Millisecond):
}
}
}
w, _ := p.Watch(ctx, "hold.*")
// the seat-holders view names h1: its first view is asked of h1.asks, and its family heard
sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 2, Created: made,
Value: json.RawMessage(`{"holders":[{"module":"h1"}]}`)})
ready(`{"n":1}`)
if b := r.HolderBucket("mesh-wants", "asks"); b != "h1_asks" { // A-303: the holder's real bucket, at run time
t.Errorf("the holder's bucket is %q, want h1_asks", b)
}
if len(fs.m) != 0 {
t.Errorf("the holder's first view left the findings %v", fs.m)
}
if f := r.Picture("h1_asks"); f == nil {
t.Fatalf("h1's asks is not followed")
} else {
f.Follower().Apply(Event{V: Version, Owner: "h1", State: "asks", Key: "hold.c", Revision: 2, Created: made, Value: json.RawMessage(`{"n":4}`)})
}
if e, err := p.Get(ctx, "hold.c"); err != nil || string(e.Value()) != `{"n":4}` {
t.Errorf("the holder's event did not reach the seat's picture: %v, %v", e, err)
}
// another holder: h1's view is dropped, h2's taken whole (hold.b and hold.c leave)
sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 3, Created: made,
Value: json.RawMessage(`{"holders":[{"module":"h2"}]}`)})
ready(`{"n":3}`)
if b := r.HolderBucket("mesh-wants", "asks"); b != "h2_asks" {
t.Errorf("after the move the holder's bucket is %q, want h2_asks", b)
}
if keys, _ := p.Keys(ctx); strings.Join(keys, ",") != "hold.a" || r.Picture("h1_asks") != nil {
t.Errorf("after the move the picture holds %v, h1 followed %v", keys, r.Picture("h1_asks") != nil)
}
var seen []string
for !slices.Contains(seen, "hold.b:KeyValueDeleteOp") || !slices.Contains(seen, "hold.c:KeyValueDeleteOp") {
select {
case e := <-w.Updates():
if e != nil {
seen = append(seen, e.Key()+":"+e.Operation().String())
}
case <-time.After(2 * time.Second):
t.Fatalf("the watch saw %v", seen)
}
}
// no holder left: the picture is kept, h2 no longer followed
sh.Follower().Apply(Event{V: Version, Owner: "assigner", State: "seat-holders", Key: "mesh-wants", Revision: 4, Created: made, Deleted: true})
deadline := time.After(2 * time.Second)
for r.Picture("h2_asks") != nil {
select {
case <-deadline:
t.Fatalf("h2 is still followed with no holder")
case <-time.After(time.Millisecond):
}
}
if e, err := p.Get(ctx, "hold.a"); err != nil || string(e.Value()) != `{"n":3}` {
t.Errorf("with no holder the picture is not kept: %v, %v", e, err)
}
mu.Lock()
defer mu.Unlock()
if asked["h1.asks"] != 1 || asked["h2.asks"] != 1 || asked["assigner.seat-holders"] != 1 {
t.Errorf("asked %v: one first view per holder and none between changes", asked)
}
})
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")
acts := actsFunc(func(q HandAct) error {
recorded = append(recorded, q)
return take
})
again := StatusTools("reader", "reader_first_views", "reader_first_view_again", refusingStatus{}, acts, now)[1]
args := map[string]any{"owner": "o", "state": "modules", "why": "doubted"}
if _, err := again.Run(args); err == nil || !strings.Contains(err.Error(), "nothing was asked again") || len(recorded) != 1 {
t.Errorf("a hand act the seat did not take: %v, recorded %v", err, recorded)
}
take = nil // taken: the ask is written next (here refused)
if _, err := again.Run(args); err == nil || !strings.Contains(err.Error(), "hand ask is not written") || len(recorded) != 2 ||
recorded[1].Why != "doubted" || recorded[1].Verb != "reader_first_view_again" || !strings.Contains(recorded[1].What, "o's modules") {
t.Errorf("a hand act taken: %v, recorded %+v", err, recorded)
}
})
t.Run("an ask through a tool reads the answer bare or in the tool runner's envelope, and its error", func(t *testing.T) {
bare := `{"v":1,"owner":"o","state":"s","revision":3,"values":{"k":"1"}}`
for name, c := range map[string]struct {
answer string
rev uint64
err string
}{
"bare": {bare, 3, ""},
"in an envelope": {`{"result":` + bare + `}`, 3, ""},
"an error": {`{"error":"not yet"}`, 0, "not yet"},
} {
ask := AskVia(func(string, any) (json.RawMessage, error) { return json.RawMessage(c.answer), nil }, "o.first_view")
a, err := ask(context.Background(), "")
if (c.err == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.err)) || a.Revision != c.rev {
t.Errorf("%s: %+v, %v", name, a, err)
}
}
})
t.Run("a change not said is said again on a backoff before the next, and the mark never passes it", func(t *testing.T) {
var said []string
fails := 2
em := Emitter{Owner: "o", Subject: func(e string) string { return "s." + e },
Publish: func(_ context.Context, msg *nats.Msg, id string) error {
if fails > 0 && strings.HasSuffix(id, "-1") {
fails--
return errors.New("bus away")
}
said = append(said, id)
return nil
},
Wait: func(_ context.Context, attempt int) bool { return true }}
updates := make(chan jetstream.KeyValueEntry, 2)
updates <- kvEntry{key: "a", rev: 1, at: made}
updates <- kvEntry{key: "b", rev: 2, at: made}
close(updates)
errs := make(chan error, 1)
em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs)
if len(said) != 2 || !strings.HasSuffix(said[0], "-1") || !strings.HasSuffix(said[1], "-2") || fails != 0 {
t.Fatalf("said %v (fails left %d): the change at 1 must be said, again, before the one at 2", said, fails)
}
// A context that ends while a change waits to be said again stops there: nothing after it is said.
said, fails = nil, 1
ctx, cancel := context.WithCancel(context.Background())
em.Wait = func(context.Context, int) bool { cancel(); return false }
updates = make(chan jetstream.KeyValueEntry, 2)
updates <- kvEntry{key: "a", rev: 1, at: made}
updates <- kvEntry{key: "b", rev: 2, at: made}
em.follow(ctx, "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, errs)
if len(said) != 0 {
t.Fatalf("said %v after the context ended on a change not said", said)
}
// A change not said past its bound (Bound) is the owner's finding naming the state, cleared once it is said;
// one said within the bound raises none.
for _, c := range []struct {
fails, puts int
}{{1, 0}, {3, 1}} {
clock := made
fs := &findingsMap{}
said, fails = nil, c.fails
em.Now, em.Findings = func() time.Time { return clock }, fs
em.Wait = func(context.Context, int) bool { clock = clock.Add(time.Minute); return true }
updates = make(chan jetstream.KeyValueEntry, 1)
updates <- kvEntry{key: "a", rev: 1, at: made}
close(updates)
em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil)
if fs.puts != c.puts || fs.deletes != 1 || len(fs.m) != 0 || len(said) != 1 {
t.Errorf("%d failures: %d findings written, %d cleared, %d left, said %v", c.fails, fs.puts, fs.deletes, len(fs.m), said)
}
if c.puts > 0 {
f := unsaidFinding("o", "s", "a", 1, made, "bus away", clock)
if f.kind != UnsaidKind || f.id != "o.s" || !strings.Contains(f.summary, "o has not said the change of s.a at 1") {
t.Errorf("the finding is %+v", f)
}
}
}
// A finding a run before a restart raised is cleared by the first change said after it.
left0 := unsaidFinding("o", "s", "a", 1, made, "bus away", made)
key := left0.kind + "/" + left0.id
fs := &findingsMap{m: map[string]seenFinding{key: {Kind: left0.kind, ID: left0.id}}}
said, fails = nil, 0
em.Findings = fs
updates = make(chan jetstream.KeyValueEntry, 1)
updates <- kvEntry{key: "b", rev: 2, at: made}
close(updates)
em.follow(context.Background(), "s", made, 0, updates, func() (Fill, error) { return Fill{}, nil }, nil)
if _, left := fs.m[key]; left || len(said) != 1 {
t.Errorf("after a restart the leftover finding stands (%v), said %v", left, said)
}
// A value too large for an event is said as a gap, and the emitter writes the owner's view-too-large itself (design 47 G §687).
tl := &findingsMap{}
te := Emitter{Owner: "o", Findings: tl, Now: func() time.Time { return made }}
if ev := te.Event("s", made, kvEntry{key: "big", rev: 3, at: made, size: MaxValueBytes + 1}); !ev.Gap || tl.m[TooLargeKind+"/o.s.big"].Kind != TooLargeKind {
t.Errorf("a value too large: gap %v, findings %v", ev.Gap, tl.m)
}
})
t.Run("each change names the revision said before it, and the mark is said as an event once written", func(t *testing.T) {
var said []Event
marks := &marksKV{}
em := Emitter{Owner: "o", Subject: func(e string) string { return "s." + e }, Marks: marks,
Publish: func(_ context.Context, msg *nats.Msg, id string) error {
e, err := DecodeEvent(msg.Data)
if err != nil || e.ID() != id {
t.Errorf("said %s as %s: %v", msg.Data, id, err)
}
said = append(said, e)
return nil
}}
// 4 to 6 were written over before the watch was handed them: the watch hands 3, then 7.
updates := make(chan jetstream.KeyValueEntry, 2)
updates <- kvEntry{key: "a", rev: 3, at: made}
updates <- kvEntry{key: "a", rev: 7, at: made}
close(updates)
em.follow(context.Background(), "s", made, 2, updates, func() (Fill, error) { return Fill{}, nil }, nil)
if len(said) != 3 || said[0].Prev != 2 || said[1].Prev != 3 || !said[2].Mark || said[2].Revision != 7 || marks.puts != 1 {
t.Fatalf("said %+v with %d marks written: 3 after 2, 7 after 3, then the mark at 7", said, marks.puts)
}
})
t.Run("an event names its state on its subject, and decodes as it was encoded", func(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}` {
t.Errorf("round trip = %+v %v %v", got, err, err2)
}
if _, err := (Event{Owner: "o", State: "s"}).Encode(); err == nil {
t.Errorf("an event with no key or revision is encoded")
}
})
}
// seenFinding is a finding a fake writer was handed, as its wire form names it.
type seenFinding struct{ Kind, ID, Summary string }
// findingsMap is a module's own findings writer in memory, keyed `<kind>/<id>` (the id as the finding names it).
type findingsMap struct {
m map[string]seenFinding
puts, deletes int
}
func (f *findingsMap) ModuleNode() (string, string) { return "m", "n1" }
func (f *findingsMap) WriteWire(_ context.Context, w map[string]any) error {
id, _ := keys.UnescapeName(w["id"].(string))
v := seenFinding{Kind: w["kind"].(string), ID: id, Summary: w["summary"].(string)}
if f.m == nil {
f.m = map[string]seenFinding{}
}
f.m[v.Kind+"/"+v.ID] = v
f.puts++
return nil
}
func (f *findingsMap) Clear(_ context.Context, _, id, kind, _ string) error {
plain, _ := keys.UnescapeName(id)
delete(f.m, kind+"/"+plain)
f.deletes++
return nil
}
// marksKV is an owner's marks state that takes puts and nothing else.
type marksKV struct {
jetstream.KeyValue
puts int
}
func (m *marksKV) Put(context.Context, string, []byte) (uint64, error) {
m.puts++
return uint64(m.puts), nil
}
// signalFindings is a findingsMap that hands each finding written to put, once written.
type signalFindings struct {
findingsMap
put chan seenFinding
}
func (f *signalFindings) WriteWire(ctx context.Context, w map[string]any) error {
_ = f.findingsMap.WriteWire(ctx, w)
id, _ := keys.UnescapeName(w["id"].(string))
f.put <- seenFinding{Kind: w["kind"].(string), ID: id, Summary: w["summary"].(string)}
return nil
}
// next is the next finding written; none within a second fails the test (a bound, never waited out when it comes).
func (f *signalFindings) next(t *testing.T) seenFinding {
t.Helper()
select {
case v := <-f.put:
return v
case <-time.After(time.Second):
t.Fatal("no finding was written")
return seenFinding{}
}
}
// seatMsg is one message of a seat's event on a module's durable consumer.
type seatMsg struct {
jetstream.Msg
subject string
data []byte
acked bool
}
func (m *seatMsg) Subject() string { return m.subject }
func (m *seatMsg) Data() []byte { return m.data }
func (m *seatMsg) Ack() error { m.acked = true; return nil }
// refusingStatus is a state `first-views` whose write the runtime refuses at once (no runtime serves this test, and the
// real binding waits 30 s for one).
type refusingStatus struct{}
func (refusingStatus) Keys() ([]string, error) { return nil, nil }
func (refusingStatus) Get(string) (*stdio.StateEntry, error) { return nil, nil }
func (refusingStatus) Put(string, any) (uint64, error) {
return 0, errors.New("no runtime serves this test")
}
// actsFunc records a hand act through a test's function.
type actsFunc func(HandAct) error
func (f actsFunc) RecordHandAct(a HandAct) error { return f(a) }
// kvEntry is one change of a bucket as a watch hands it.
type kvEntry struct {
key string
rev uint64
at time.Time
size int // the value's length when over the one event's: zero is a small value
}
func (k kvEntry) Bucket() string { return "b" }
func (k kvEntry) Key() string { return k.key }
func (k kvEntry) Value() []byte {
if k.size > 0 {
return make([]byte, k.size)
}
return []byte(`{"n":1}`)
}
func (k kvEntry) Revision() uint64 { return k.rev }
func (k kvEntry) Created() time.Time { return k.at }
func (k kvEntry) Delta() uint64 { return 0 }
func (k kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
// tickSource is a source whose every Get takes a step of the fake clock.
type tickSource struct {
mapSource
at *time.Time
step time.Duration
}
func (s *tickSource) Get(k string) (*Entry, bool, error) {
*s.at = s.at.Add(s.step)
return s.mapSource.Get(k)
}
// pagedSource is a source with a bulk read, two values a page.
type pagedSource struct {
mapSource
pages, gets int
unsupported bool
}
func (s *pagedSource) Get(k string) (*Entry, bool, error) {
s.gets++
return s.mapSource.Get(k)
}
func (s *pagedSource) Page(after string, _ int, _ time.Duration) ([]KeyedEntry, string, error) {
if s.unsupported {
return nil, "", errors.ErrUnsupported
}
s.pages++
keys, _ := s.Keys()
slices.Sort(keys)
var out []KeyedEntry
for _, k := range keys {
if k > after && len(out) < 2 {
out = append(out, KeyedEntry{Key: k, Entry: s.mapSource[k]})
}
}
if len(out) == 2 && out[1].Key != keys[len(keys)-1] {
return out, out[1].Key, nil
}
return out, "", nil
}
// TestViews runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
func TestViews(t *testing.T) {
t.Run("Views", subViews)
t.Run("StdioWriterPublishesAndListensThroughTheModulesOwnPlumbing", subStdioWriterPublishesAndListensThroughTheModulesOwnPlumbing)
}
type putRecorder struct{ puts map[string][]byte }
func (p *putRecorder) Put(_ context.Context, key string, value []byte) (uint64, error) {
p.puts[key] = value
return 1, nil
}
// statusWith is a state `first-views` holding the given values.
type statusWith map[string][]byte
func (s statusWith) Keys() ([]string, error) { return slices.Collect(maps.Keys(s)), nil }
func (s statusWith) Get(k string) (*stdio.StateEntry, error) {
if v, ok := s[k]; ok {
return &stdio.StateEntry{Value: v}, nil
}
return nil, nil
}
func (statusWith) Put(string, any) (uint64, error) { return 0, errors.New("read only") }
// design 47 G §674: both answers of a part's first views say its own build beside their list.
func TestFirstViewsSayTheOwnBuild(t *testing.T) {
for name, c := range map[string]struct{ env, want string }{"a fingerprint": {"abc123", "abc123"}, "none": {"", "unknown"}} {
getenv := func(string) string { return c.env }
// the bundle reader's answer
r := &Reader{Module: "m", Getenv: getenv}
got, _ := r.Tools("m")[0].Run(nil)
if v := got.(map[string]any)[OwnBuildKey]; v != c.want {
t.Errorf("%s: the bundle reader's first views say %v, want %s", name, v, c.want)
}
// the process writes it at its start, and its status tool reads it
rec := &putRecorder{puts: map[string][]byte{}}
KeepOwnBuild(context.Background(), rec, process.ReadOwnBuild(getenv), nil)
st := statusWith{OwnBuildKey: rec.puts[OwnBuildKey]}
read, err := StatusTools("m", "m_first_views", "m_first_view_again", st, nil, time.Now)[0].Run(nil)
m := read.(map[string]any)
if err != nil || m[OwnBuildKey] != c.want || len(m["first-views"].([]Status)) != 0 {
t.Errorf("%s: the status tool says %v, %v, want %s", name, m, err, c.want)
}
}
// a process that kept nothing yet says unknown
read, _ := StatusTools("m", "r", "a", statusWith{}, nil, time.Now)[0].Run(nil)
if read.(map[string]any)[OwnBuildKey] != "unknown" {
t.Errorf("nothing kept: %v", read)
}
}