Files
mesh-delivery/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
T
jschoubben 037bf1ef36
mesh/merge-gate head 037bf1ef361b merged into trunk b04ee764739a — pass: builds mesh-delivery-teller → novox; no bus step; every machine composes with the …
mesh/repo-check head 037bf1ef361b merged into trunk b04ee764739a — pass: its merge-check.sh passed
mesh/delivery delivered
Ask again at start a merge ask whose refusal the teller did not hear, since the router keeps no record of a refused ask (issue 1610, review round 1)
2026-10-21 16:45:31 +02:00

521 lines
20 KiB
Go

package main
import (
"encoding/json"
"fmt"
"strings"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/asker"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
const (
headA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
)
var now = time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
// memStore is a state bucket in memory with the revisions and the compare-and-set of the runtime's state.
type memStore struct {
entries map[string]*stdio.StateEntry
seq uint64
}
func (m *memStore) Get(key string) (*stdio.StateEntry, error) {
if e, ok := m.entries[key]; ok {
c := *e
return &c, nil
}
return nil, nil
}
func (m *memStore) put(key string, value any) (uint64, error) {
b, err := json.Marshal(value)
if err != nil {
return 0, err
}
m.seq++
m.entries[key] = &stdio.StateEntry{Key: key, Value: b, Revision: m.seq}
return m.seq, nil
}
func (m *memStore) Create(key string, value any) (uint64, error) {
if _, ok := m.entries[key]; ok {
return 0, stdio.ErrChanged
}
return m.put(key, value)
}
func (m *memStore) Update(key string, value any, revision uint64) (uint64, error) {
if e, ok := m.entries[key]; !ok || e.Revision != revision {
return 0, stdio.ErrChanged
}
return m.put(key, value)
}
func (m *memStore) Delete(key string) error { delete(m.entries, key); return nil }
type memAnswers struct {
written map[string][]byte
revs map[string]uint64
seq uint64
writes int
}
func (a *memAnswers) Get(key string) ([]byte, uint64, bool, error) {
v, ok := a.written[key]
return v, a.revs[key], ok, nil
}
// Write is the bucket's compare-and-set: a create of a key that has an answer, or an update at another revision, loses.
func (a *memAnswers) Write(key string, value []byte, rev uint64, found bool) (bool, error) {
_, has := a.written[key]
if found != has || (found && rev != a.revs[key]) {
return false, nil
}
a.seq++
a.writes++
if a.revs == nil {
a.revs = map[string]uint64{}
}
a.written[key], a.revs[key] = value, a.seq
return true, nil
}
type memGate map[string][]byte
func (g memGate) Get(key string) ([]byte, bool, error) { v, ok := g[key]; return v, ok, nil }
type published struct {
subject string
body json.RawMessage
id string
}
// bed is a teller with the asker.Client of the SDK on in-memory ends: what is published, the router's records, the
// warrants it would hear, and the conditions it says.
type bed struct {
t *Teller
client *asker.Client
store *memStore
answers *memAnswers
gate memGate
pubs []published
conds []asks.Condition
records map[string]string
// seenIDs are the ids the fake router has taken an ask under, and dropped the asks it took a second time under: the
// messenger keeps an ask after it ends and returns for an ask whose id it holds (mesh-catalog modules/messenger
// cmd/messenger/asks.go:487-489, keepAsk), so a reused id reaches no phone.
seenIDs map[string]bool
dropped []string
// msgIDs are the bus message ids published, and busDropped the messages the bus dropped as a second under one.
msgIDs map[string]bool
busDropped []string
hear func(stdio.Envelope) error
logs strings.Builder
}
func newBed(t *testing.T) *bed {
t.Helper()
b := &bed{store: &memStore{entries: map[string]*stdio.StateEntry{}}, answers: &memAnswers{written: map[string][]byte{}, revs: map[string]uint64{}},
gate: memGate{}, records: map[string]string{}, seenIDs: map[string]bool{}, msgIDs: map[string]bool{}}
b.client = &asker.Client{Module: delivery.TellerModule, State: b.store, Now: func() time.Time { return now },
Publish: func(subject string, body any, id string) error {
raw, err := json.Marshal(body)
if b.msgIDs[id] {
// The bus drops a second message under an id it holds (JetStream's duplicate window): it reaches nobody.
b.busDropped = append(b.busDropped, id)
return err
}
b.msgIDs[id] = true
if subject == asks.AskSubject(delivery.TellerModule) {
var a struct {
ID string `json:"id"`
}
_ = json.Unmarshal(raw, &a)
if b.seenIDs[a.ID] {
b.dropped = append(b.dropped, a.ID)
} else if _, kept := b.records[a.ID]; !kept {
b.records[a.ID] = `{"state":"open"}` // the router keeps the ask it took (asks.go CreateAsk)
}
b.seenIDs[a.ID] = true
}
b.pubs = append(b.pubs, published{subject, raw, id})
return err
},
Record: func(id string) (*stdio.StateEntry, error) {
if v, ok := b.records[id]; ok {
return &stdio.StateEntry{Key: id, Value: json.RawMessage(v)}, nil
}
return nil, nil
},
Subscribe: func(_ string, handle func(stdio.Envelope) error) error { b.hear = handle; return nil },
}
b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Gate: b.gate, Publish: b.client.Publish, Now: func() time.Time { return now },
Say: func(c asks.Condition) error {
if _, err := asks.EncodeCondition(c); err != nil {
t.Errorf("a condition the router would refuse: %v", err)
}
b.conds = append(b.conds, c)
return nil
},
Logf: func(f string, a ...any) { fmt.Fprintf(&b.logs, f+"\n", a...) }}
if err := b.client.Listen(b.t.Answer, b.t.Said); err != nil {
t.Fatal(err)
}
return b
}
// gateAsk is an ask as the gate composes it: the router's ask with the approve and decline options binding the act
// `mesh-delivery.merge` for this ask and the heads (mesh-catalog modules/mesh-delivery approval.go:123-191 composeMergeAsk
// and :63-65 mergeAct), under the id its key gives (mesh-sdk go/delivery AskID). boundTo are the heads the options bind,
// which are the ask's own unless a test says another.
func gateAsk(t *testing.T, heads, boundTo []delivery.Head) (key string, value []byte, ask delivery.Ask) {
t.Helper()
return gateAskN(t, heads, boundTo, 1)
}
// gateAskN is the asking-th writing of the group's ask at these heads: its router id is AskingID(key, asking) (mesh-sdk
// go/delivery Ask.Asking), the first asking's the key's own.
func gateAskN(t *testing.T, heads, boundTo []delivery.Head, asking int) (key string, value []byte, ask delivery.Ask) {
t.Helper()
key, err := delivery.MergeAskKey("g", heads)
if err != nil {
t.Fatal(err)
}
id := delivery.AskingID(key, asking)
a := asks.Ask{ID: id, Headline: "Merge into main: novox/mesh-delivery#12", Explanation: "Merges two pull requests.\nReview: approved.",
Who: asks.Operator, Expires: now.Add(asks.ApproveLasts - 10*time.Minute), OnExpiry: "the merge stays held", About: "delivery.mesh-delivery.12"}
for _, o := range []struct{ id, label, does string }{{"approve", "Approve", "merges"}, {"decline", "Decline", "merges nothing"}} {
binds, err := asks.ActDigest(mergeAct(id, o.id, headsOf(boundTo)))
if err != nil {
t.Fatal(err)
}
a.Options = append(a.Options, asks.Option{ID: o.id, Label: o.label, Does: o.does, Level: asks.Approve, Binds: binds})
}
ask = delivery.Ask{Group: "g", Heads: heads, Asking: asking, Ask: a}
value, err = delivery.EncodeAsk(ask)
if err != nil {
t.Fatal(err)
}
return key, value, ask
}
func twoHeads() []delivery.Head {
return []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headA}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
}
// warrantFor is the router's warrant for a chosen option of an ask, as the router verified the operator.
func warrantFor(a asks.Ask, option string) asks.Warrant {
return asks.Warrant{Ask: a.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeChosen, Option: option, Label: "Approve", Level: asks.Approve,
By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "42", Verified: "user id"}, Channel: "telegram",
At: now.Add(time.Hour), AskDigest: a.Digest()}
}
func (b *bed) deliver(t *testing.T, w asks.Warrant) {
t.Helper()
raw, _ := json.Marshal(w)
if err := b.hear(stdio.Envelope{Key: asks.Seat + ".decided." + delivery.TellerModule, Body: raw}); err != nil {
t.Fatal(err)
}
}
func (b *bed) askedOf(subject string) []published {
var out []published
for _, p := range b.pubs {
if p.subject == subject {
out = append(out, p)
}
}
return out
}
func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 3, value)
asked := b.askedOf(asks.AskSubject(delivery.TellerModule))
if len(asked) != 1 || asked[0].id != ga.Ask.ID {
t.Fatalf("published %+v, want one ask under %s with id %s", b.pubs, delivery.TellerModule, ga.Ask.ID)
}
var got asks.Ask
if err := json.Unmarshal(asked[0].body, &got); err != nil {
t.Fatal(err)
}
if got.Digest() != ga.Ask.Digest() || got.Headline != ga.Ask.Headline || got.Explanation != ga.Ask.Explanation || got.ID != delivery.AskID(key) {
t.Errorf("the router's ask differs from the gate's: %+v", got)
}
b.t.OnAsk(key, 3, value) // the revision seen again, as at a restart
b.t.OnAsk(key, 4, value) // the key written again with the same ask
if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 1 {
t.Errorf("the ask was published %d times", n)
}
delete(b.gate, key)
b.t.OnAskGone(key)
cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ga.Ask.ID+`"}` {
t.Fatalf("cancels = %+v", cancels)
}
b.t.OnAskGone(key)
if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 {
t.Errorf("the ask was cancelled %d times", n)
}
}
func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { // 1343/AC2
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 1, value)
w := warrantFor(ga.Ask, "approve")
b.deliver(t, w)
raw, ok := b.answers.written[delivery.AnswerKey(key)]
if !ok || len(b.answers.written) != 1 {
t.Fatalf("answers = %v", b.answers.written)
}
a, err := delivery.DecodeAnswer(raw)
if err != nil || a.Option != "approve" || a.Warrant != w.Ask+"@"+w.At.UTC().Format(time.RFC3339Nano) || !strings.Contains(a.Words, "chose Approve") {
t.Errorf("answer %+v err %v", a, err)
}
b.deliver(t, w) // the same warrant again
w2 := w
w2.At = w.At.Add(time.Minute) // a second warrant for the same ask
b.deliver(t, w2)
if b.answers.writes != 1 {
t.Errorf("the answer was written %d times, want once", b.answers.writes)
}
if len(b.conds) != 0 {
t.Errorf("conditions said: %+v", b.conds)
}
}
func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
cases := map[string]func(t *testing.T) (*bed, string, asks.Warrant){
"heads that differ from the ones the option bound": func(t *testing.T) (*bed, string, asks.Warrant) {
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), other) // the options bind other heads than the ask covers
b.gate[key] = value
b.t.OnAsk(key, 1, value)
return b, key, warrantFor(ga.Ask, "approve")
},
"an asker that differs": func(t *testing.T) (*bed, string, asks.Warrant) {
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 1, value)
w := warrantFor(ga.Ask, "approve")
w.Asker = "mesh-delivery"
return b, key, w
},
"a digest that differs": func(t *testing.T) (*bed, string, asks.Warrant) {
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 1, value)
w := warrantFor(ga.Ask, "approve")
w.AskDigest = "sha256:0000"
return b, key, w
},
}
for name, mk := range cases {
t.Run(name, func(t *testing.T) {
b, key, w := mk(t)
b.deliver(t, w)
if len(b.answers.written) != 0 {
t.Errorf("an answer was written: %v", b.answers.written)
}
if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || b.conds[0].Subject.ID != key ||
!strings.HasPrefix(b.conds[0].Explanation, asks.NeedsYou) {
t.Fatalf("conditions = %+v", b.conds)
}
})
}
}
func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/AC3 (cleared when a good answer is written)
b := newBed(t)
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
key, bad, gaBad := gateAsk(t, twoHeads(), other)
b.gate[key] = bad
b.t.OnAsk(key, 1, bad)
b.deliver(t, warrantFor(gaBad.Ask, "approve"))
if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised {
t.Fatalf("conditions = %+v", b.conds)
}
// The gate writes the ask again, right this time, at a new revision of its key: the second asking, under a new id.
_, good, gaGood := gateAskN(t, twoHeads(), twoHeads(), 2)
b.gate[key] = good
b.t.OnAsk(key, 2, good)
if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 2 {
t.Fatalf("the ask was published %d times, want again at the new revision", n)
}
b.deliver(t, warrantFor(gaGood.Ask, "approve"))
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
t.Fatalf("no answer: %v\n%s", b.answers.written, b.logs.String())
}
if len(b.conds) != 2 || b.conds[1].State != asks.StateCleared || b.conds[1].Key != b.conds[0].Key {
t.Errorf("conditions = %+v", b.conds)
}
}
func TestAWarrantComingWhileTheTellerWasAwayIsReadFromTheRouter(t *testing.T) {
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 1, value)
w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
b.records[ga.Ask.ID] = `{"state":"chosen","warrant":` + string(w) + `}`
b.t.OnAsk(key, 1, value) // the teller starts again and reads the key at the same revision
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
t.Errorf("no answer: %s", b.logs.String())
}
}
func TestARewriteWithAskingTwoCancelsAskingOneAndAsksUnderTheNewID(t *testing.T) { // G2 of review round 1
b := newBed(t)
key, one, g1 := gateAskN(t, twoHeads(), twoHeads(), 1)
b.gate[key] = one
b.t.OnAsk(key, 1, one)
_, two, g2 := gateAskN(t, twoHeads(), twoHeads(), 2)
if g1.Ask.ID == g2.Ask.ID {
t.Fatal("the two askings share an id")
}
b.gate[key] = two
b.t.OnAsk(key, 2, two)
cancels := b.askedOf(asks.CancelSubject(delivery.TellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+g1.Ask.ID+`"}` {
t.Fatalf("cancels = %+v, want asking 1's router ask", cancels)
}
asked := b.askedOf(asks.AskSubject(delivery.TellerModule))
if len(asked) != 2 || asked[1].id != g2.Ask.ID {
t.Fatalf("asks = %+v, want the second under %s", asked, g2.Ask.ID)
}
if len(b.dropped) != 0 {
t.Errorf("the router dropped asks under a reused id: %v", b.dropped)
}
// The same revision again, and a warrant for the first asking (the gate has moved on), change nothing.
b.t.OnAsk(key, 2, two)
b.deliver(t, warrantFor(g1.Ask, "approve"))
if len(b.answers.written) != 0 || len(b.askedOf(asks.AskSubject(delivery.TellerModule))) != 2 {
t.Errorf("answers %v asks %d", b.answers.written, len(b.askedOf(asks.AskSubject(delivery.TellerModule))))
}
// The warrant of the second names it: its answer's warrant is the router id and the time.
w := warrantFor(g2.Ask, "approve")
b.deliver(t, w)
raw, ok := b.answers.written[delivery.AnswerKey(key)]
a, err := delivery.DecodeAnswer(raw)
if !ok || err != nil || a.Warrant != g2.Ask.ID+"@"+w.At.UTC().Format(time.RFC3339Nano) {
t.Errorf("answer %+v (%v) err %v", a, ok, err)
}
}
func TestAnAskWhoseWordsChangedUnderTheSameIDIsCancelledAndNotAskedAgain(t *testing.T) {
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.t.OnAsk(key, 1, value)
changed := ga
changed.Ask.Explanation = "Merges two pull requests, now with a third point.\nReview: approved."
v2, err := delivery.EncodeAsk(changed)
if err != nil {
t.Fatal(err)
}
b.t.OnAsk(key, 2, v2)
if n := len(b.askedOf(asks.CancelSubject(delivery.TellerModule))); n != 1 {
t.Errorf("cancels = %d", n)
}
if n := len(b.askedOf(asks.AskSubject(delivery.TellerModule))); n != 1 || len(b.dropped) != 0 {
t.Errorf("asks = %d dropped %v: the router would drop a second ask under the id", n, b.dropped)
}
}
func TestACapitalisedRepositoryIsSpelledInLowerCaseInTheAct(t *testing.T) { // T1 of review round 1
b := newBed(t)
heads := []delivery.Head{{Repository: "novox/Mesh-Delivery", Number: 12, Head: headA}}
key, err := delivery.MergeAskKey("g", heads)
if err != nil {
t.Fatal(err)
}
id := delivery.AskID(key)
// The act as the gate binds it: its IDOf lowercases the repository (mesh-catalog modules/mesh-delivery-gate model.go).
binds, err := asks.ActDigest(asks.Act{"verb": "mesh-delivery.merge", "ask": id, "option": "approve", "heads": "novox/mesh-delivery@" + headA[:12] + "=" + headA})
if err != nil {
t.Fatal(err)
}
a := asks.Ask{ID: id, Headline: "Merge into main: Mesh-Delivery#12", Explanation: "Merges.", Who: asks.Operator, Expires: now.Add(asks.ApproveLasts - 10*time.Minute),
OnExpiry: "the merge stays held", Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: binds}}}
value, err := delivery.EncodeAsk(delivery.Ask{Group: "g", Heads: heads, Ask: a})
if err != nil {
t.Fatal(err)
}
b.gate[key] = value
b.t.OnAsk(key, 1, value)
b.deliver(t, warrantFor(a, "approve"))
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
t.Errorf("no answer for the gate's spelling: %s", b.logs.String())
}
}
func TestAKeyWhoseValueIsNotItsOwnAskIsLeft(t *testing.T) {
b := newBed(t)
_, value, _ := gateAsk(t, twoHeads(), twoHeads())
b.t.OnAsk("merge.g.000000000000", 1, value)
b.t.OnAsk("merge.g.000000000000", 2, []byte(`{"v":1`))
if len(b.pubs) != 0 {
t.Errorf("published %+v", b.pubs)
}
}
func TestARefusedAskIsSaidAsACondition(t *testing.T) { // ADR 0437 §9
b := newBed(t)
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.t.OnAsk(key, 1, value)
w := asks.Warrant{Ask: ga.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: "no channel can send", At: now}
b.deliver(t, w)
if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || !strings.Contains(b.conds[0].Summary, "no channel can send") {
t.Errorf("conditions = %+v", b.conds)
}
}
func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R2-1 of review round 2
b := newBed(t)
key, one, g1 := gateAskN(t, twoHeads(), twoHeads(), 1)
b.gate[key] = one
b.t.OnAsk(key, 1, one)
b.deliver(t, warrantFor(g1.Ask, "approve"))
first, ok := b.answers.written[delivery.AnswerKey(key)]
if a, err := delivery.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.Ask.ID+"@") {
t.Fatalf("asking 1's answer: %+v %v %v", a, ok, err)
}
// The gate rewrites the key at asking 2 (a review record changed). Asking 1's answer stays in the bucket.
_, two, g2 := gateAskN(t, twoHeads(), twoHeads(), 2)
b.gate[key] = two
b.t.OnAsk(key, 2, two)
asked := b.askedOf(asks.AskSubject(delivery.TellerModule))
if len(asked) != 2 || asked[1].id != g2.Ask.ID {
t.Fatalf("asks = %+v, want asking 2 asked under %s", asked, g2.Ask.ID)
}
w := warrantFor(g2.Ask, "approve")
b.deliver(t, w)
raw := b.answers.written[delivery.AnswerKey(key)]
a, err := delivery.DecodeAnswer(raw)
if err != nil || a.Warrant != g2.Ask.ID+"@"+w.At.UTC().Format(time.RFC3339Nano) {
t.Fatalf("the key holds %+v (%v), want asking 2's answer", a, err)
}
// A second warrant for the same asking writes nothing, and a replay of asking 2 asks nothing.
w2 := w
w2.At = w.At.Add(time.Minute)
b.deliver(t, w2)
b.t.OnAsk(key, 2, two)
b.t.OnAsk(key, 3, two)
if b.answers.writes != 2 || len(b.askedOf(asks.AskSubject(delivery.TellerModule))) != 2 || len(b.dropped) != 0 {
t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(asks.AskSubject(delivery.TellerModule))), b.dropped)
}
}