The TypeScript writer holds no FindingBody of its own and works on the wire form (checkFinding, views reader findings laid out by viewFindingBody); both writers name findings-asked as the seat mesh-conditions' event. The view-waits test waits on the raise, not on a count of yields. Builds: design 47 G §686, §640, §687, A-3, A-13, A-17 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
1183 lines
49 KiB
Go
1183 lines
49 KiB
Go
package views
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"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/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 TestViews(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 = func() time.Time { return at }
|
|
defer func() { clock = 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
|
|
waitBound = func(context.Context, time.Duration) bool { return <-ended }
|
|
defer func() { waitBound = 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)
|
|
}
|
|
})
|
|
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
|
|
waitBound = func(context.Context, time.Duration) bool { return <-ended }
|
|
defer func() { waitBound = kept }()
|
|
fs := &signalFindings{put: make(chan Finding, 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)
|
|
}
|
|
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 Finding, 4)}
|
|
pr := &ProcessReader{Module: "reader", Findings: func() FindingWriter { 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)
|
|
}
|
|
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, 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")
|
|
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" || f.Observer != "o" || !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]Finding{key: left0}}
|
|
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)
|
|
}
|
|
})
|
|
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")
|
|
}
|
|
})
|
|
}
|
|
|
|
// findingsMap is a module's own findings state in memory, keyed `<kind>/<id>`.
|
|
type findingsMap struct {
|
|
m map[string]Finding
|
|
puts, deletes int
|
|
}
|
|
|
|
func (f *findingsMap) Put(v Finding) error {
|
|
if f.m == nil {
|
|
f.m = map[string]Finding{}
|
|
}
|
|
f.m[v.Kind+"/"+v.ID] = v
|
|
f.puts++
|
|
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
|
|
}
|
|
|
|
func (f *findingsMap) Delete(v Finding) error { delete(f.m, v.Kind+"/"+v.ID); f.deletes++; return nil }
|
|
|
|
// signalFindings is a findingsMap that hands each finding written to put, once written.
|
|
type signalFindings struct {
|
|
findingsMap
|
|
put chan Finding
|
|
}
|
|
|
|
func (f *signalFindings) Put(v Finding) error { _ = f.findingsMap.Put(v); f.put <- v; 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) Finding {
|
|
t.Helper()
|
|
select {
|
|
case v := <-f.put:
|
|
return v
|
|
case <-time.After(time.Second):
|
|
t.Fatal("no finding was written")
|
|
return Finding{}
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
func (k kvEntry) Bucket() string { return "b" }
|
|
func (k kvEntry) Key() string { return k.key }
|
|
func (k kvEntry) Value() []byte { 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
|
|
}
|