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

TL5: the teller carries the assigner's asks only, follows the view's answers and says its own on views.answers

Browse Source 
The merge-ask path goes whole with the Approve per merge: the reads of
mesh-delivery-gate.asks and mesh-delivery-view.answers, OnAsk, OnAskGone, the
gate branches of Answer and the room list, and the merge words. The teller
follows the view's answers through view_read_answers and views.answers, says
its state answers as views.answers from its state teller-marks, and serves the
read tool teller_read_answers. The room list stays for the assigner's asks
(the lead's answer). viewstate gains the names the view's own commit gives it
(the same hunk, so the two merge cleanly).

Builds: design 47 G §653, §431, §512; lead answers A-172, A-175, A-178

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:52:42 +01:00
parent 459c03933c
commit 0f4a33ce48
17 changed files with 518 additions and 1085 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go
+14
View File
@@ -104,6 +104,19 @@ func AnswerTools(s answers) []stdio.Tool {
}}}
}
// ReadAnswersTool is the teller's ordinary read tool of its state answers (design 47 G A-175, A-178; mesh-sdk views.ReadTool):
// what the view follows, first by it and then by the teller's views family views.answers. Its marks are the state teller-marks.
const ReadAnswersTool = "teller_read_answers"
// marksState is the teller's state of what its views family said each state up to (A-178).
const marksState = "teller-marks"
// ReadAnswersTools is the tool teller_read_answers over the state answers and its marks.
func ReadAnswersTools(values, marks views.Source, now func() time.Time) []stdio.Tool {
return []stdio.Tool{views.ReadTool("mesh-delivery-teller", ReadAnswersTool, "answers",
"the operator's answers to the assigner's asks, by the ask's key.", values, marks, now)}
}
// WidensSaidTool reads the teller's state widens-said (design 47 G §335).
const WidensSaidTool = "teller_widens_said"
@@ -205,6 +218,7 @@ func main() {
tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now),
views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...)
tools = append(tools, AnswerTools(stdio.State("answers"))...)
tools = append(tools, ReadAnswersTools(views.KeptSource{State: stdio.State("answers")}, views.KeptSource{State: stdio.State(marksState)}, time.Now)...)
tools = append(tools, WidensSaidTools(stdio.State(widensSaidState))...)
if err := stdio.Serve("", tools); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err)
modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main_test.go
+35
View File
@@ -4,8 +4,10 @@ import (
"encoding/json"
"fmt"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/views"
)
type fakeAnswers map[string]string
@@ -64,3 +66,36 @@ func TestTellerWidensSaidReadsOneKeyOrEveryKey(t *testing.T) {
t.Errorf("every key: %s, %v", raw, err)
}
}
// teller_read_answers is the teller's ordinary read tool of its state answers (A-175, A-178): the key's value and revision, the
// keys, over the answers state and the marks state teller-marks.
type entries map[string]views.Entry
func (e entries) Keys() ([]string, error) {
var out []string
for k := range e {
out = append(out, k)
}
return out, nil
}
func (e entries) Get(key string) (*views.Entry, bool, error) {
v, ok := e[key]
return &v, ok, nil
}
func TestTellerReadAnswersIsTheViewsReadToolOfTheAnswersState(t *testing.T) {
values := entries{"hold.k1": {Value: json.RawMessage(`{"option":"keep"}`), Revision: 3}}
tool := ReadAnswersTools(values, entries{}, time.Now)[0]
if tool.Name != "teller_read_answers" {
t.Fatalf("the tool is %s", tool.Name)
}
out, err := tool.Run(map[string]any{"key": "hold.k1"})
raw, _ := json.Marshal(out)
if err != nil || string(raw) != `{"owner":"mesh-delivery-teller","state":"answers","key":"hold.k1","found":true,"value":{"option":"keep"},"revision":3}` {
t.Errorf("one key: %s, %v", raw, err)
}
if marksState != "teller-marks" {
t.Errorf("the marks state is %s", marksState)
}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go
+4 -4
View File
@@ -25,9 +25,9 @@ import (
// of these asks needs proof of who answered (mesh-sdk go/delivery AskNeedsProof).
// mostAssignerOpen is how many of the assigner's asks the teller keeps open at the router at once. The router keeps at most
// occ.MostOpen open asks of one asker, and the teller asks both the gate's merge asks and the assigner's as one asker, so a burst
// of holds (seven at once on 2026-10-25) would take all the room and a merge Approve could not be asked; one place is always
// left for a merge ask. An assigner ask beyond these waits in the wait for room and is asked when one of them ends.
// occ.MostOpen open asks of one asker, and a burst of holds (seven at once on 2026-10-25) would take all the room; one place
// stays free (it was left for the merge ask, which went with the Approve per merge, design 47 G §653). An assigner ask beyond
// these waits in the wait for room and is asked when one of them ends.
const mostAssignerOpen = occ.MostOpen - 1
// assignerOpenRecord is the key of the router ids of the assigner's asks the teller asked.
@@ -89,7 +89,7 @@ func (t *Teller) waitForAssignerRoom(key, id string) {
}
}
// isAssignerKey says whether an ask key is one of the assigner's, not the gate's merge ask.
// isAssignerKey says whether an ask key is one of the assigner's, not the merge ask's key of before the Approve per merge went.
func isAssignerKey(key string) bool {
k, err := asks.ParseAnyAskKey(key)
return err == nil && k.Kind != meshdeliveryclient.AskKindMerge
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go
+11 -45
View File
@@ -151,22 +151,20 @@ func TestAnAssignerAskAnsweredInTheViewOrTakenAwayIsTakenBack(t *testing.T) {
}
}
// 1906/H1: an assigner ask refused for want of room waits and is asked again when another of the teller's asks ends, as a merge
// ask does; a refusal for its words is said as a condition about a delivery ask, not a merge.
// 1906/H1: an assigner ask refused for want of room waits and is asked again when another of the teller's asks ends; a refusal for
// its words is said as a condition about a delivery ask.
func TestAnAssignerAskWaitsForRoomAndARefusalIsSaidAsADeliveryAsk(t *testing.T) {
anAssignerAskWrittenAgainAfterBeingTakenAwayIsAskedAgainUnderANewAsking(t)
b := newBed(t)
assigner, _ := withView(b)
_, first := b.askOf(t, 1, 1)
b.askOf(t, 2, 2)
b.askOf(t, 3, 3)
_, first := b.holdOf(t, 1, 1)
key, value, ask := holdAsk(t)
assigner[key] = value
b.t.OnAssignerAsk(key)
b.refuse(t, ask.ID, roomRefusal)
b.deliver(t, warrantFor(first.Ask, "approve"))
b.deliver(t, warrantFor(first, "keep"))
got := b.askedIDs()
if len(got) != 5 || got[4] != ask.ID {
if len(got) != 3 || got[2] != ask.ID {
t.Fatalf("asks = %v, want the assigner's %s asked again", got, ask.ID)
}
@@ -181,60 +179,28 @@ func TestAnAssignerAskWaitsForRoomAndARefusalIsSaidAsADeliveryAsk(t *testing.T)
}
// ackFor is the router's warrant for an option of level acknowledge, as the assigner's asks offer.
func ackFor(a oc.Ask, option string) oc.Warrant {
w := warrantFor(a, option)
w.Level, w.Label = occ.Acknowledge, option
return w
}
func ackFor(a oc.Ask, option string) oc.Warrant { return warrantFor(a, option) }
// holdAskOf is the assigner's hold ask of module plex at the n-th fingerprint, shaped as holdAsk.
func holdAskOf(t *testing.T, n int) (string, []byte, oc.Ask) {
t.Helper()
key, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64))
if err != nil {
t.Fatal(err)
}
_, _, ask := holdAsk(t)
ask.ID = meshdeliveryclient.AskID(key)
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask)})
if err != nil {
t.Fatal(err)
}
return key, value, ask
}
// 1906/H5: given seven holds at once (as on 2026-10-25), when they are raised, then at most two are open at the router, a merge
// ask is still asked, and each hold answered lets the next waiting one be asked.
func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) {
// 1906/H5: given seven holds at once (as on 2026-10-25), when they are raised, then at most mostAssignerOpen are open at the
// router, and each hold answered lets the next waiting one be asked.
func TestABurstOfHoldsAsksNoMoreThanTheMostOpenAndTheNextWhenOneEnds(t *testing.T) {
b := newBed(t)
assigner, _ := withView(b)
var held []oc.Ask
for n := 0; n < 7; n++ {
key, value, ask := holdAskOf(t, n)
assigner[key] = value
b.t.OnAssignerAsk(key)
_, ask := b.holdOf(t, n, 1)
held = append(held, ask)
}
if got := b.askedIDs(); len(got) != mostAssignerOpen || got[0] != held[0].ID || got[1] != held[1].ID {
t.Fatalf("asked %v, want the first %d holds only; log:\n%s", got, mostAssignerOpen, b.logs.String())
}
_, merge := b.askOf(t, 1, 1)
if got := b.askedIDs(); len(got) != mostAssignerOpen+1 || got[mostAssignerOpen] != merge.Ask.ID {
t.Fatalf("asked %v, want the merge ask asked beside the holds", got)
}
if len(b.dropped) != 0 || len(b.conds) != 0 {
t.Errorf("router dropped %v, conditions %+v", b.dropped, b.conds)
}
b.deliver(t, ackFor(held[0], "release"))
got := b.askedIDs()
if len(got) != mostAssignerOpen+2 || got[len(got)-1] != held[2].ID {
if len(got) != mostAssignerOpen+1 || got[len(got)-1] != held[2].ID {
t.Fatalf("asked %v after a hold was answered, want the third hold asked next; log:\n%s", got, b.logs.String())
}
// The merge ask answered frees its own place, and no hold takes it.
b.deliver(t, warrantFor(merge.Ask, "approve"))
if got := b.askedIDs(); len(got) != mostAssignerOpen+2 {
t.Errorf("asked %v after the merge ask was answered, want no further hold", got)
}
}
// Issue 1966 (nats on novox): an assigner ask taken back when its key went is asked again when the assigner writes the key
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go
+80 -255
View File
@@ -25,25 +25,11 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"mesh-delivery-teller/internal/ends"
)
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
// controller's user and the teller's own principal <node>.<module>.
const busConf = `
listen: 127.0.0.1:-1
jetstream { store_dir: %q }
accounts {
MESH {
jetstream: enabled
users = [
{ user: "ctl", password: "pw" }
{ user: "n1.mesh-delivery-teller", password: "pw" }
]
}
}
`
type syncBuffer struct {
mu sync.Mutex
b strings.Builder
@@ -56,82 +42,6 @@ func (s *syncBuffer) Write(p []byte) (int, error) {
}
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
// TestTellerWaitsForTheGate starts the program the way the node does (its entry, run, with MESH_BROKER_FILE) against an
// embedded bus where mesh-delivery-gate_asks does not exist: it keeps running until stopped, says once that it waits for
// the gate's asks, and writes and publishes nothing (the revised 1240/AC2: the watch is release 3d's).
func TestTellerWaitsForTheGate(t *testing.T) { // 1240/AC2
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { os.RemoveAll(dir) })
conf := filepath.Join(dir, "bus.conf")
if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
t.Fatal(err)
}
opts, err := server.ProcessConfigFile(conf)
if err != nil {
t.Fatal(err)
}
opts.NoLog, opts.NoSigs = true, true
srv, err := server.NewServer(opts)
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("the bus did not come up")
}
t.Cleanup(srv.Shutdown)
ctl, err := nats.Connect(srv.ClientURL(), nats.UserInfo("ctl", "pw"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(ctl.Close)
js, err := jetstream.New(ctl)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
broker := filepath.Join(t.TempDir(), "broker")
cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
env := map[string]string{"MESH_BROKER_FILE": broker}
logs := &syncBuffer{}
rn := startRunner(ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
t.Cleanup(func() { rn.finish(t, 10*time.Second) })
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline) && !strings.Contains(logs.String(), "waiting for the gate's asks"); time.Sleep(20 * time.Millisecond) {
}
if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
t.Fatalf("the teller said it waits %d times, want once: %q", n, logs.String())
}
select {
case err := <-rn.done:
t.Fatalf("the teller stopped: %v", err)
case <-time.After(500 * time.Millisecond):
}
if n := strings.Count(logs.String(), "waiting for the gate's asks"); n != 1 {
t.Errorf("the teller said it waits %d times after a while, want once", n)
}
// It writes and publishes nothing: no stream, no bucket exists.
names := js.StreamNames(ctx)
for name := range names.Name() {
t.Errorf("a stream %s exists: the teller wrote something", name)
}
if _, err := js.KeyValue(ctx, "mesh-delivery-gate_asks"); err == nil {
t.Error("the gate's asks bucket exists")
}
if err := rn.finish(t, 10*time.Second); err != nil {
t.Errorf("the teller stopped with %v, want a clean stop", err)
}
}
// The principals of the composed accounts file for the teller's tests, with the permissions mesh-controller composes for
// them (internal/broker): a module's own state is the read grants of its bucket and `$KV.<bucket>.>` to write it
// (state.go stateGrants, readGrants), a read of another module's bucket is the read grants alone, and a module that uses
@@ -171,9 +81,8 @@ func user(name string, pub, sub []string) string {
const (
tellerAnswers = "mesh-delivery-teller_answers"
tellerAsked = "mesh-delivery-teller_asked"
gateAsksKV = "mesh-delivery-gate_asks"
assignerAsks = "mesh-delivery-assigner_asks"
viewAnswers = "mesh-delivery-view_answers"
viewAnswers = viewstate.ViewAnswersBucket
tellerConsumer = "n1_mesh-delivery-teller"
seatAsk = "mesh.seat.operator-channel.accept.ask.mesh-delivery-teller"
seatCancel = "mesh.seat.operator-channel.accept.cancel.mesh-delivery-teller"
@@ -187,7 +96,7 @@ const (
func fullConf(dir string) string {
// ends is the teller's own state (manifest state entry ends), which the controller makes before it starts: a bed
// without it held the start ten seconds on a read the bus never answered, and the tests raced that stall.
teller := append(append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...), readOf(gateAsksKV)...),
teller := append(append(append(keepsOf(tellerAnswers), keepsOf(tellerAsked)...), keepsOf(ends.Bucket)...),
seatAsk, seatCancel, seatCondition, "$JS.API.DIRECT.GET.KV_"+routerAsksBucket+".$KV."+routerAsksBucket+".mesh-delivery-teller.>")
teller = append(teller, keepsOf(widensSaidBucket)...) // design 47 G §335: the state widens-said
// The durable consumer on EVENTS the controller makes for the seat it uses, pulled and acknowledged (issue 1126).
@@ -195,14 +104,15 @@ func fullConf(dir string) string {
// The finding writer finds the stream of the keeper's findings-asked (design 47 G §384): the bed has none, so it is
// said and the findings stay in the state.
teller = append(teller, "$JS.API.STREAM.NAMES")
teller = append(teller, "mesh.mod.mesh-delivery-teller.event.answered") // the manifest's emits (ADR 0596 §2)
teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads
// The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and
// findings; the bucket itself is no longer the teller's to read.
teller = append(append(append(teller, viewstest.Grants(assignerModule, assignerModule+".assigner_read_asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...)
view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
// The manifest's emits (ADR 0596 §2; design 47 G A-175: its answers as its views family) and its marks state (A-178).
teller = append(teller, "mesh.mod.mesh-delivery-teller.event.answered", "mesh.mod.mesh-delivery-teller.event.ask-ended", "mesh.mod.mesh-delivery-teller.event.views.answers")
teller = append(teller, keepsOf(marksBucket)...)
// The assigner's asks and the view's answers, each by its read tool and its views family (ADR 0615, DLV-M3; A-175), and the
// teller's own first-views and findings; the buckets themselves are no longer the teller's to read.
teller = append(append(append(teller, viewstest.Grants(assignerModule, assignerModule+".assigner_read_asks", tellerConsumer)...), viewstest.Grants(viewModule, viewModule+"."+viewstate.ViewAnswersReadTool, tellerConsumer)...), keepsOf(firstViewsBucket)...)
teller = append(teller, keepsOf(findingsBucket)...)
view := append(append(readOf(tellerAnswers), readOf(assignerAsks)...), keepsOf(viewAnswers)...)
assigner := keepsOf(assignerAsks)
tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
return fmt.Sprintf(`
listen: 127.0.0.1:-1
jetstream { store_dir: %q }
@@ -211,14 +121,13 @@ accounts {
jetstream: enabled
users = [
{ user: "ctl", password: "pw" }
%s%s%s%s ]
%s%s%s ]
}
}
`, filepath.Join(dir, "js"),
user("n1.mesh-delivery-teller", teller, nil),
user("n1.mesh-delivery-view", view, nil),
user("n1.mesh-delivery-assigner", assigner, nil),
user("n1.node-tools", tools, nil))
user("n1.mesh-delivery-assigner", assigner, nil))
}
type fullBed struct {
@@ -229,7 +138,7 @@ type fullBed struct {
}
// startFull starts the bus with the principals above and what the controller makes before any module starts: the buckets
// of the gate's asks, the teller's answers and asked, the router's work queue SEAT_OPERATOR_CHANNEL.
// of the assigner's asks, the view's answers, the teller's answers, asked and marks, the router's work queue SEAT_OPERATOR_CHANNEL.
func startFull(t *testing.T) *fullBed {
t.Helper()
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "t")
@@ -266,7 +175,7 @@ func startFull(t *testing.T) *fullBed {
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
t.Cleanup(cancel)
for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, routerAsksBucket: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1, widensSaidBucket: 1} {
for name, history := range map[string]uint8{tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, routerAsksBucket: 1, assignerAsks: 5, viewAnswers: 5, marksBucket: 1, firstViewsBucket: 1, findingsBucket: 1, widensSaidBucket: 1} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
t.Fatal(err)
}
@@ -284,6 +193,8 @@ func startFull(t *testing.T) *fullBed {
}
// The assigner as the owner of its asks: its read tool and its views family (ADR 0615).
viewstest.Serve(t, ctx, js, ctl, assignerModule, "asks", assignerModule+".assigner_read_asks", tellerConsumer)
// The view as the owner of its answers, as the teller follows them (A-175, A-178).
viewstest.Serve(t, ctx, js, ctl, viewModule, viewstate.ViewAnswersState, viewModule+"."+viewstate.ViewAnswersReadTool, tellerConsumer)
return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
}
@@ -332,29 +243,29 @@ func (b *fullBed) put(t *testing.T, js jetstream.JetStream, bucket, key string)
return err
}
// TestCredentialsWriteOnlyTheirOwnBuckets: the teller's credential is refused writing the gate's asks, and the gate's
// (the tool runner's) and the view's are refused writing the teller's answers; each writes where it may, so a refusal is the
// permissions' and not a bus that does not take writes (1343/AC4, ADR 0437 §6's check).
// TestCredentialsWriteOnlyTheirOwnBuckets: the view's credential, which reads the teller's answers, is refused writing them; each
// credential writes its own buckets, so a refusal is the permissions' and not a bus that does not take writes (1343/AC4, ADR
// 0437 §6's check). The teller reads neither the assigner's asks nor the view's answers: it follows them.
func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
b := startFull(t)
teller, view, tools := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.node-tools")
if err := b.put(t, teller, tellerAnswers, "merge.g.0123456789ab"); err != nil {
t.Errorf("the teller cannot write its own answers: %v", err)
teller, view, assigner := b.as(t, "n1.mesh-delivery-teller"), b.as(t, "n1.mesh-delivery-view"), b.as(t, "n1.mesh-delivery-assigner")
for _, c := range []struct {
who jetstream.JetStream
bucket string
key string
}{{teller, tellerAnswers, "hold.a"}, {teller, marksBucket, "answers"}, {assigner, assignerAsks, "hold.a"}, {view, viewAnswers, "hold.a.view"}} {
if err := b.put(t, c.who, c.bucket, c.key); err != nil {
t.Errorf("a credential cannot write its own %s: %v", c.bucket, err)
}
}
if err := b.put(t, tools, gateAsksKV, "merge.g.0123456789ab"); err != nil {
t.Errorf("the gate's account cannot write the gate's asks: %v", err)
}
if err := b.put(t, teller, gateAsksKV, "merge.g.ffffffffffff"); err == nil {
t.Error("the teller's credential wrote the gate's asks")
}
if err := b.put(t, tools, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
t.Error("the gate's account wrote the teller's answers")
}
if err := b.put(t, view, tellerAnswers, "merge.g.ffffffffffff"); err == nil {
t.Error("the view's credential wrote the teller's answers")
}
// What was refused is not there.
for _, c := range []struct{ bucket, key string }{{gateAsksKV, "merge.g.ffffffffffff"}, {tellerAnswers, "merge.g.ffffffffffff"}} {
for _, c := range []struct {
who jetstream.JetStream
bucket string
key string
}{{view, tellerAnswers, "hold.f"}} {
if err := b.put(t, c.who, c.bucket, c.key); err == nil {
t.Errorf("a credential wrote %s, which is not its own", c.bucket)
}
kv, _ := b.js.KeyValue(b.ctx, c.bucket)
if _, err := kv.Get(b.ctx, c.key); err == nil {
t.Errorf("%s holds %s though its write was refused", c.bucket, c.key)
@@ -362,14 +273,14 @@ func TestCredentialsWriteOnlyTheirOwnBuckets(t *testing.T) { // 1343/AC4
}
}
// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the gate's account
// writes an ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the
// warrant on the teller's decided subject, and the answer is under the ask's key in the teller's bucket; withdrawing a
// second ask cancels it (1343/AC1, AC2).
// TestTellerCarriesAnAskAndWritesTheAnswer starts the program on the composed principal against the bus: the assigner writes
// a hold ask, the teller publishes it to the seat's work queue under its own name, the router (the test) says the warrant on
// the teller's decided subject, and the answer is under the ask's key in the teller's bucket and said as the event answered
// and on the views family views.answers (A-175); taking a second ask away cancels it (1343/AC1, AC2, 1906/H1, H2).
func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
t.Cleanup(nowIs(time.Now().UTC()))
b := startFull(t)
tools := b.as(t, "n1.node-tools")
assigner := b.as(t, "n1.mesh-delivery-assigner")
broker := filepath.Join(t.TempDir(), "broker")
cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
@@ -408,8 +319,8 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
t.Fatalf("the router's record read as the teller: %v, %v; want it", rec, err)
}
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
kv, err := tools.KeyValue(b.ctx, gateAsksKV)
key, value, ask := holdAsk(t)
kv, err := assigner.KeyValue(b.ctx, assignerAsks)
if err != nil {
t.Fatal(err)
}
@@ -432,11 +343,11 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
t.Fatalf("no ask on the seat's queue: %v\n%s", err, logs.String())
}
var asked oc.Ask
if err := json.Unmarshal(raw.Data, &asked); err != nil || occ.Digest(asked) != occ.Digest(ga.Ask) {
t.Fatalf("the ask on the queue is %+v (%v), want the gate's", asked, err)
if err := json.Unmarshal(raw.Data, &asked); err != nil || occ.Digest(asked) != occ.Digest(ask) {
t.Fatalf("the ask on the queue is %+v (%v), want the assigner's", asked, err)
}
// The router says the warrant.
w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
w, _ := json.Marshal(ackFor(ask, "release"))
if err := b.ctl.Publish(seatDecided, w); err != nil {
t.Fatal(err)
}
@@ -453,10 +364,10 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
if e == nil {
t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String())
}
if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err != nil || a.Option != "release" {
t.Errorf("answer %+v err %v", a, err)
}
// The answer is said as the event answered, which the gate hears (ADR 0596 §2).
// The answer is said as the event answered (ADR 0596 §2) and on the views family views.answers (A-175).
events, err := b.js.Stream(b.ctx, process.EventsStream)
if err != nil {
t.Fatal(err)
@@ -470,17 +381,28 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
if said == nil {
t.Fatalf("the answer was not said as the event answered: %v\n%s", err, logs.String())
}
if got := string(said.Data); !strings.Contains(got, `"key":"`+key+`"`) || !strings.Contains(got, `"option":"approve"`) {
if got := string(said.Data); !strings.Contains(got, `"key":"`+key+`"`) || !strings.Contains(got, `"option":"release"`) {
t.Errorf("the event answered is %s", got)
}
// A second ask, withdrawn by the gate, is cancelled at the router.
other := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
key2, value2, ga2 := gateAsk(t, other, other)
if vs, err := events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-teller.event.views.answers"); err != nil || !strings.Contains(string(vs.Data), key) {
t.Errorf("the answer was not said on the views family views.answers: %v\n%s", err, logs.String())
}
// A second ask, taken away by the assigner, is cancelled at the router.
key2, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64))
if err != nil {
t.Fatal(err)
}
ask2 := ask
ask2.ID = meshdeliveryclient.AskID(key2)
value2, err := asks.EncodeAsk(key2, asks.Ask{Ask: rawAsk(t, ask2)})
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if _, err := seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil && strings.Contains(logs.String(), ga2.Ask.ID) {
if strings.Contains(logs.String(), ask2.ID+" is asked") {
break
}
}
@@ -492,19 +414,19 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
break
}
}
if err != nil || string(raw.Data) != `{"id":"`+ga2.Ask.ID+`"}` {
t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String())
if err != nil || string(raw.Data) != `{"id":"`+ask2.ID+`"}` {
t.Errorf("the ask taken away was not cancelled: %v %v\n%s", raw, err, logs.String())
}
}
// TestTellerAsksAgainAnAskRefusedForWantOfRoom starts the program against the bus: the router (the test) refuses the gate's
// TestTellerAsksAgainAnAskRefusedForWantOfRoom starts the program against the bus: the router (the test) refuses the assigner's
// ask for want of room, as the live router did on 2026-10-21 (issue 1610), and then says another of the teller's asks
// lapsed; the refused ask is on the seat's work queue again, which the bus's duplicate window would have dropped under its
// first message id (1610/AC1 on the real bus).
func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
t.Cleanup(nowIs(time.Now().UTC()))
b := startFull(t)
tools := b.as(t, "n1.node-tools")
assigner := b.as(t, "n1.mesh-delivery-assigner")
broker := filepath.Join(t.TempDir(), "broker")
cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
@@ -514,7 +436,7 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
logs := &syncBuffer{}
rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
t.Cleanup(func() { rn.finish(t, 10*time.Second) })
kv, err := tools.KeyValue(b.ctx, gateAsksKV)
kv, err := assigner.KeyValue(b.ctx, assignerAsks)
if err != nil {
t.Fatal(err)
}
@@ -547,31 +469,30 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
}
}
key, value, held := gateAsk(t, twoHeads(), twoHeads())
key, value, held := holdAskAt(t, 1, 1)
if _, err := kv.Put(b.ctx, key, value); err != nil {
t.Fatal(err)
}
first := lastAsk(held.Ask.ID, 0)
say(oc.Warrant{Ask: held.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeRefused, Words: roomRefusal, At: occ.FormatTime(now)})
first := lastAsk(held.ID, 0)
say(oc.Warrant{Ask: held.ID, Asker: tellerModule, Outcome: occ.OutcomeRefused, Words: roomRefusal, At: occ.FormatTime(now)})
for deadline := time.Now().Add(10 * time.Second); !strings.Contains(logs.String(), "for want of room"); time.Sleep(50 * time.Millisecond) {
if time.Now().After(deadline) {
t.Fatalf("the refusal was not heard\n%s", logs.String())
}
}
other := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
key2, value2, ga2 := gateAsk(t, other, other)
key2, value2, ga2 := holdAskAt(t, 2, 1)
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
t.Fatal(err)
}
second := lastAsk(ga2.Ask.ID, first)
second := lastAsk(ga2.ID, first)
// The other ask lapses at the router: there is room, and the refused ask is asked again.
say(oc.Warrant{Ask: ga2.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga2.Ask)})
lastAsk(held.Ask.ID, second)
say(oc.Warrant{Ask: ga2.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga2)})
lastAsk(held.ID, second)
}
// TestTellerFillsAnAskFromTheViewsAnswerOnce starts the program against the bus with the permissions above: the assigner's
// fixture hold ask is written, then the view's answer under `<ask key>.view`; the answer is under the ask's own key, once
// (rewriting the view's answer does not write a second), and a view answer under the merge's key fills nothing
// (rewriting the view's answer does not write a second), and a view answer under a key that is no assigner's (a merge's) fills nothing
// (1297/B10 AC: a view answer fills the ask once; an ask needing proof is not answered from a view key).
func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
t.Cleanup(nowIs(time.Now().UTC()))
@@ -629,7 +550,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
t.Errorf("the answer was written a second time: %v", err)
}
// An answer under the merge's key fills nothing.
mkey, mvalue, _ := gateAsk(t, twoHeads(), twoHeads())
mkey, mvalue := "merge.g.0123456789ab", value
if _, err := askKV.Put(b.ctx, mkey, mvalue); err != nil {
t.Fatal(err)
}
@@ -642,102 +563,6 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
}
}
// TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer starts the program against the bus: the assigner's hold ask, put in
// its bucket, is on the operator channel's queue as the teller's ask; the router's warrant for "release" is written under the
// ask's key, where the assigner reads it; a second ask the assigner takes away (a hold it lifted) is cancelled at the router
// (1906/H1, H2 on the real bus).
func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
t.Cleanup(nowIs(time.Now().UTC()))
b := startFull(t)
assigner := b.as(t, "n1.mesh-delivery-assigner")
broker := filepath.Join(t.TempDir(), "broker")
cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
env := map[string]string{"MESH_BROKER_FILE": broker}
logs := &syncBuffer{}
rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
t.Cleanup(func() { rn.finish(t, 10*time.Second) })
key, value, ask := holdAsk(t)
askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
if err != nil {
t.Fatal(err)
}
b.listening(t) // the program hears warrants and watches
if _, err := askKV.Put(b.ctx, key, value); err != nil {
t.Fatal(err)
}
seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
if err != nil {
t.Fatal(err)
}
var raw *jetstream.RawStreamMsg
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatAsk); err == nil {
break
}
}
if raw == nil {
t.Fatalf("the assigner's ask is not on the seat's queue: %v\n%s", err, logs.String())
}
var asked oc.Ask
if err := json.Unmarshal(raw.Data, &asked); err != nil || asked.ID != ask.ID || occ.Digest(asked) != occ.Digest(ask) {
t.Fatalf("the ask on the queue is %+v (%v), want the assigner's", asked, err)
}
w, _ := json.Marshal(ackFor(ask, "release"))
if err := b.ctl.Publish(seatDecided, w); err != nil {
t.Fatal(err)
}
answers, err := b.js.KeyValue(b.ctx, tellerAnswers)
if err != nil {
t.Fatal(err)
}
var e jetstream.KeyValueEntry
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err = answers.Get(b.ctx, answerKey(key)); err == nil {
break
}
}
if e == nil {
t.Fatalf("no answer under %s: %v\n%s", key, err, logs.String())
}
if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err != nil || a.Option != "release" || answeredAsking(e.Value()) != ask.ID {
t.Fatalf("answer %+v err %v", a, err)
}
key2, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64))
if err != nil {
t.Fatal(err)
}
ask2 := ask
ask2.ID = meshdeliveryclient.AskID(key2)
value2, err := asks.EncodeAsk(key2, asks.Ask{Ask: rawAsk(t, ask2)})
if err != nil {
t.Fatal(err)
}
if _, err := askKV.Put(b.ctx, key2, value2); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if strings.Contains(logs.String(), ask2.ID+" is asked") {
break
}
}
if err := askKV.Delete(b.ctx, key2); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if raw, err = seat.GetLastMsgForSubject(b.ctx, seatCancel); err == nil {
break
}
}
if err != nil || string(raw.Data) != `{"id":"`+ask2.ID+`"}` {
t.Errorf("the ask taken away was not cancelled: %v %v\n%s", raw, err, logs.String())
}
}
// TestTellerTakesAChangedHoldAnswerFromTheView starts the program against the bus: a hold ask answered "keep" earlier, and the
// view's answer "release" written now, replace the answer of that asking with "release" (1906/N23 on the real bus).
func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go
+5 -4
View File
@@ -101,13 +101,14 @@ func (a kvAnswers) Write(key string, value []byte, rev uint64, found bool) (bool
return err == nil, err
}
// kvGate reads the gate's bucket asks, which the teller may watch and read and never write.
type kvGate struct {
// kvReader reads the picture of another module's state the teller follows (the assigner's asks, the view's answers), which it
// may watch and read and never write.
type kvReader struct {
ctx context.Context
kv jetstream.KeyValue
}
func (g kvGate) Get(key string) ([]byte, bool, error) {
func (g kvReader) Get(key string) ([]byte, bool, error) {
e, err := g.kv.Get(g.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, false, nil
@@ -119,7 +120,7 @@ func (g kvGate) Get(key string) ([]byte, bool, error) {
}
// GetAt is Get with the time the value was written.
func (g kvGate) GetAt(key string) ([]byte, time.Time, bool, error) {
func (g kvReader) GetAt(key string) ([]byte, time.Time, bool, error) {
e, err := g.kv.Get(g.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, time.Time{}, false, nil
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+31 -42
View File
@@ -1,15 +1,14 @@
// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, §6, §9; issues 1240, 1343). It carries the
// gate's merge asks to the operator and writes the answers back: it watches the gate's asks, asks each of the router as
// itself, and on the router's warrant, once it holds, writes the answer under the ask's key. Where the gate's asks
// bucket does not exist yet it says once that it waits for the gate's asks and does nothing: waiting for an absent bucket
// needs a timer (ADR 0437 §8 lets the parts hold none), so it is started again with the gate. It serves no request
// subject: its one tool, teller_end_ask (the program mesh-delivery-teller-tools, issue 774), asks through its state ends,
// which it watches.
// Command mesh-delivery-teller is delivery's teller (novox/hq ADR 0437 §4, §6, §9; issues 1240, 1343; design 47 G §653,
// A-172). It carries the assigner's asks to the operator and writes the answers back: it follows the assigner's asks (the
// first view of its read tool, then its views family), asks each of the router as itself, and on the router's warrant, once
// it holds, writes the answer under the ask's key, and says its state answers as its views family (A-175). It follows the
// view's answers the same way and fills an ask from the operator's answer in the view. It serves no request subject: its
// one tool beside the readers, teller_end_ask (the program mesh-delivery-teller-tools, issue 774), asks through its state
// ends, which it watches.
package main
import (
"context"
"errors"
"fmt"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
"io"
@@ -25,6 +24,8 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"mesh-delivery-teller/internal/ends"
)
@@ -49,16 +50,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
defer conn.Close()
gate, err := js.KeyValue(ctx, gateAsksBucket)
if errors.Is(err, jetstream.ErrBucketNotFound) {
logger.Print(waiting)
<-ctx.Done()
return nil
}
if err != nil {
// Essential: the gate's asks are what the teller tells (issue 1945: the error names the grant).
return process.EssentialErr("teller", gateAsksBucket, "tell any of the gate's asks", err)
}
// The answers it writes and what it asked: its own work, so each stops the start, naming its grant.
var kvs [3]jetstream.KeyValue
for i, name := range []string{answersBucket, askedBucket, widensSaidBucket} {
@@ -85,7 +76,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
defer calls.Close()
client := &occ.Client{Module: tellerModule, State: state, Publish: publish,
Record: recordVia(calls.Call, calls.Ask), Subscribe: decidedVia(hub.Handle, logger.Printf)}
t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]}, Gate: kvGate{ctx: ctx, kv: gate},
t := &Teller{Asker: client, Store: state, Answers: kvAnswers{ctx: ctx, kv: kvs[0]},
Say: sayVia(ctx, js, tellerModule), Logf: logger.Printf, Publish: publish,
// The teller's own events (its manifest's emits), on EVENTS under its module's subject.
Emit: func(event string, body []byte, id string) error {
@@ -111,8 +102,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := client.Listen(t.Answer, t.Said); err != nil {
return fmt.Errorf("hearing the router's warrants: %w", err)
}
// The view's answers to the assigner's asks (issue 1297 B10): both buckets are the controller's to create, and a bucket not
// there is said and leaves the teller without that path rather than without the gate's asks (a wait for it needs a timer).
failed := make(chan error, 4)
t.Answered = func(key, askID string, answer []byte) error {
return calls.Emit(answeredEvent, Answered{Key: key, Ask: askID, Answer: answer}, stdio.EmitOptions{MessageID: answeredID(key, askID)})
@@ -129,25 +118,33 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
assignerKV := jetstream.KeyValue(asks)
// The view's answers to the assigner's asks (issue 1297 B10, A-175): followed as the assigner's asks are, never its bucket.
answered, err := reader.Follow(viewModule, viewstate.ViewAnswersState, viewModule+"."+viewstate.ViewAnswersReadTool)
if err != nil {
return err
}
viewKV := jetstream.KeyValue(answered)
reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "teller_first_views reads nothing and teller_first_view_again is not heard", logger.Printf))
// Every part has registered: the hub pulls the module's durable.
if err := hub.Start(ctx, js, cred); err != nil {
return fmt.Errorf("the consumer %s on %s cannot be read (%v): the teller hears no answer without it, so it stops", cred.EventConsumerName(), process.EventsStream, err)
}
viewKV := process.OptionalBucket(ctx, js, viewAnswersBucket, teller, logger.Printf)
// The teller says its own answers as its views family (A-175, A-178): once its marks state is bound.
process.AwaitBucket(ctx, js, conn, marksBucket, "no answer is said on the views family, so the view waits on its first view", logger.Printf).Ready(func(mk jetstream.KeyValue) {
em := views.Emitter{Owner: tellerModule, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, Findings: writer, Logf: logger.Printf}
if err := em.Start(ctx, answersState, kvs[0], failed); err != nil {
logger.Printf("%v: the view asks the first view of the answers at each change until the teller starts again", err)
}
})
// The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906).
if viewKV != nil {
t.View = kvGate{ctx: ctx, kv: viewKV}
}
t.Assigner = kvGate{ctx: ctx, kv: assignerKV}
t.View = kvReader{ctx: ctx, kv: viewKV}
t.Assigner = kvReader{ctx: ctx, kv: assignerKV}
go func() {
failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, t.OnAssignerAskGone, nil)
failed <- watchAsks(ctx, assignerKV, func(key string, _ uint64, _ []byte) { t.OnAssignerAsk(key) }, t.OnAssignerAskGone, t.OnCurrent)
}()
go func() {
failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil)
}()
if viewKV != nil {
go func() {
failed <- watchAsks(ctx, viewKV, func(key string, _ uint64, _ []byte) { t.OnViewAnswer(key) }, func(string) {}, nil)
}()
}
// A person ends one ask by hand with teller_end_ask, which writes the state ends (issue 774): watched as the gate's asks
// are, once it is on the bus.
process.AwaitBucket(ctx, js, conn, ends.Bucket, "teller_end_ask is not answered; the asks still end when they are answered, lapse or their key is taken away", logger.Printf).Ready(func(kv jetstream.KeyValue) {
@@ -157,11 +154,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}()
})
go func() {
if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil {
failed <- fmt.Errorf("watching the gate's asks: %w", err)
}
}()
select {
case <-ctx.Done():
return nil
@@ -173,9 +165,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}
// teller is what an optional bucket of the teller leaves out.
const teller = "the asks and answers it holds are not carried until the teller is started again with it"
// askedBucket is the teller's state bucket asked, `<module>_<name>`: what it asked of the router and which condition it
// raised (manifest state entry asked, novox/hq ADR 0437 §4).
const askedBucket = tellerModule + "_asked"
@@ -190,5 +179,5 @@ var (
// assignerAsksState is the assigner's state of its asks (assignerAsksBucket).
const assignerAsksState = "asks"
// waiting is said once at start.
const waiting = "waiting for the gate's asks, whose bucket does not exist yet; the teller is started again with the gate, and nothing is read, written or published until then"
// marksBucket is the teller's state of what its views family said its answers up to (A-178).
const marksBucket = tellerModule + "_" + marksState
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go
+11 -18
View File
@@ -2,31 +2,24 @@ package main
import "encoding/json"
// The holders and buckets the teller names, taken from its own manifest (module.json: its name, its `reads`) and the
// states it keeps itself; a unit row checks them against it (TestTheNamesAreTheManifests). No contract names a holder
// (novox/hq ADR 0625 §4, ADR 0634 §1).
// The holders and buckets the teller names, taken from its own manifest (module.json: its name, what it invokes and
// consumes) and the states it keeps itself; a unit row checks them against it (TestTheNamesAreTheManifests). No contract
// names a holder (novox/hq ADR 0625 §4, ADR 0634 §1).
const (
tellerModule = "mesh-delivery-teller"
assignerModule = "mesh-delivery-assigner"
gateModule = "mesh-delivery-gate"
viewModule = "mesh-delivery-view"
// gateAsksBucket is the gate's state asks, read by the teller (reads: mesh-delivery-gate.asks).
gateAsksBucket = gateModule + "_asks"
// viewAnswersBucket is the view's state answers, read by the teller (reads: mesh-delivery-view.answers).
viewAnswersBucket = viewModule + "_answers"
// answersBucket is the teller's own state answers, which the gate reads.
answersBucket = tellerModule + "_answers"
// answersState is the teller's own state answers, said as its views family views.answers, and marksState the state of
// what that family said it up to.
answersState = "answers"
marksState = "teller-marks"
// answersBucket is the teller's own state answers.
answersBucket = tellerModule + "_" + answersState
// widensSaidBucket is the teller's own state widens-said, the last commit it said a widening of per module and branch.
widensSaidBucket = tellerModule + "_" + widensSaidState
// assignerAsksBucket is the assigner's state asks (the first view of assigner_read_asks stands in for it).
assignerAsksBucket = assignerModule + "_asks"
// The revisions each keeps per key.
gateAsksHistory = 5
answersHistory = 5
assignerAsksHistory = 5
viewAnswersHistory = 5
// The revisions the teller's answers keep per key.
answersHistory = 5
)
// answerKey is the key, in the teller's answers, of its answer to the ask at askKey: the same string.
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names_test.go
+26 -11
View File
@@ -5,19 +5,25 @@ import (
"os"
"slices"
"testing"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
)
// TestTheNamesAreTheManifests: the holders and buckets the teller names are its manifest's (its module, its `reads`, its
// own state answers), since no contract names a holder (novox/hq ADR 0634 §1).
// TestTheNamesAreTheManifests: the holders, tools and buckets the teller names are its manifest's (its module, what it invokes
// and consumes, its own states answers and teller-marks, no `reads`), since no contract names a holder (novox/hq ADR 0634 §1;
// design 47 G §653, A-175, A-178).
func TestTheNamesAreTheManifests(t *testing.T) {
raw, err := os.ReadFile("../../module.json")
if err != nil {
t.Fatal(err)
}
var m struct {
Module string `json:"module"`
Reads []string `json:"reads"`
State []struct {
Module string `json:"module"`
Reads []string `json:"reads"`
Invokes []string `json:"invokes"`
Consumes []string `json:"consumes"`
Emits []string `json:"emits"`
State []struct {
Name string `json:"name"`
} `json:"state"`
}
@@ -27,16 +33,25 @@ func TestTheNamesAreTheManifests(t *testing.T) {
if m.Module != tellerModule {
t.Errorf("the manifest's module is %q, the teller names itself %q", m.Module, tellerModule)
}
for _, read := range []string{gateModule + ".asks", viewModule + ".answers"} {
if !slices.Contains(m.Reads, read) {
t.Errorf("the teller reads %s, its manifest's reads are %v", read, m.Reads)
}
if len(m.Reads) != 0 {
t.Errorf("the teller reads no other module's state, its manifest's reads are %v", m.Reads)
}
if want := viewModule + "." + viewstate.ViewAnswersReadTool; !slices.Contains(m.Invokes, want) {
t.Errorf("the teller follows the view's answers by %s, its manifest's invokes are %v", want, m.Invokes)
}
if want := viewModule + ".views." + viewstate.ViewAnswersState; !slices.Contains(m.Consumes, want) {
t.Errorf("the teller follows %s, its manifest's consumes are %v", want, m.Consumes)
}
if want := "views." + answersState; !slices.Contains(m.Emits, want) {
t.Errorf("the teller says %s, its manifest's emits are %v", want, m.Emits)
}
var own []string
for _, s := range m.State {
own = append(own, tellerModule+"_"+s.Name)
}
if !slices.Contains(own, answersBucket) {
t.Errorf("the teller writes %s, its manifest's states are %v", answersBucket, own)
for _, want := range []string{answersBucket, marksBucket} {
if !slices.Contains(own, want) {
t.Errorf("the teller keeps %s, its manifest's states are %v", want, own)
}
}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go
+73 -93
View File
@@ -6,8 +6,6 @@ import (
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
)
// roomRefusal is the router's words when an asker already holds occ.MostOpen open asks, as the live router said them to
@@ -16,20 +14,6 @@ import (
// modules/messenger cmd/messenger/asks.go:529-531).
const roomRefusal = "its asker already has 3 open asks"
// headsN is one pull request's head, distinct per n, so each n is an ask of its own key.
func headsN(n int) []meshdeliveryclient.Head {
return []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 10 + n, Head: strings.Repeat(string(rune('a'+n)), 40)}}
}
// askOf puts the gate's key for n and hands it to the teller, as the watch would.
func (b *bed) askOf(t *testing.T, n int, rev uint64) (string, gateAsked) {
t.Helper()
key, value, ga := gateAsk(t, headsN(n), headsN(n))
b.gate[key] = value
b.t.OnAsk(key, rev, value)
return key, ga
}
// refuse is the router's refusal of an ask: it keeps no record of a refused ask (asks.go refuseAsk only tells its asker),
// so the id may be asked again.
func (b *bed) refuse(t *testing.T, id, words string) {
@@ -40,9 +24,9 @@ func (b *bed) refuse(t *testing.T, id, words string) {
}
// expire is the router's word that an ask of the teller's lapsed with nobody's answer: one of its asks ended.
func (b *bed) expire(t *testing.T, ga gateAsked) {
func (b *bed) expire(t *testing.T, ask oc.Ask) {
t.Helper()
b.deliver(t, oc.Warrant{Ask: ga.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga.Ask)})
b.deliver(t, oc.Warrant{Ask: ask.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ask)})
}
// askedIDs is the router ids of every ask the bus carried, in order.
@@ -58,87 +42,84 @@ func (b *bed) askedIDs() []string {
return out
}
// The live fault of issue 1610: three asks open, a fourth refused for want of room, and the operator then answers one of
// the three. The fourth is asked again, under its own router id and a new bus message id, and nothing is said as a fault.
// The live fault of issue 1610: asks open, one refused by the router for want of room, and the operator then answers one of the
// open. The refused is asked again, under its own router id and a new bus message id, and nothing is said as a fault.
func TestAnAskRefusedForWantOfRoomIsAskedAgainWhenAnotherEnds(t *testing.T) { // 1610/AC1
b := newBed(t)
_, first := b.askOf(t, 1, 1)
b.askOf(t, 2, 2)
b.askOf(t, 3, 3)
_, fourth := b.askOf(t, 4, 4)
b.refuse(t, fourth.Ask.ID, roomRefusal)
_, first := b.holdOf(t, 1, 1)
key2, second := b.holdOf(t, 2, 1)
b.refuse(t, second.ID, roomRefusal)
if len(b.conds) != 0 {
t.Errorf("a refusal for want of room is said as a fault before it was asked again: %+v", b.conds)
}
if got := b.askedIDs(); len(got) != 4 {
if got := b.askedIDs(); len(got) != 2 {
t.Fatalf("asked %v before room came", got)
}
b.deliver(t, warrantFor(first.Ask, "approve"))
b.deliver(t, warrantFor(first, "keep"))
got := b.askedIDs()
if len(got) != 5 || got[4] != fourth.Ask.ID {
t.Fatalf("asks = %v, want the refused %s asked again once the first was answered", got, fourth.Ask.ID)
if len(got) != 3 || got[2] != second.ID {
t.Fatalf("asks = %v, want the refused %s asked again once the first was answered", got, second.ID)
}
pubs := b.askedOf(oc.AcceptSubjectAsk(tellerModule))
if pubs[4].id == pubs[3].id {
t.Errorf("asked again under the bus message id %s of its first asking: the bus drops it as a duplicate", pubs[4].id)
if pubs[2].id == pubs[1].id {
t.Errorf("asked again under the bus message id %s of its first asking: the bus drops it as a duplicate", pubs[2].id)
}
if len(b.busDropped) != 0 || len(b.dropped) != 0 || len(b.conds) != 0 {
t.Errorf("bus dropped %v, router dropped %v, conditions %+v", b.busDropped, b.dropped, b.conds)
}
// Taken this time: an answer to it writes the answer as for any ask.
key4, _ := meshdeliveryclient.MergeAskKey("g", headsN(4))
b.deliver(t, warrantFor(fourth.Ask, "approve"))
if _, ok := b.answers.written[answerKey(key4)]; !ok {
b.deliver(t, warrantFor(second, "keep"))
if _, ok := b.answers.written[answerKey(key2)]; !ok {
t.Errorf("the answer to the ask asked again was not written; logs:\n%s", b.logs.String())
}
// And it is asked no more: its answer ends its wait.
_, second := b.askOf(t, 5, 5)
b.expire(t, second)
if got := b.askedIDs(); len(got) != 6 || got[5] == fourth.Ask.ID {
t.Errorf("asks after the answer = %v, want only the fifth asked", got)
_, third := b.holdOf(t, 3, 1)
b.expire(t, third)
if got := b.askedIDs(); len(got) != 4 || got[3] == second.ID {
t.Errorf("asks after the answer = %v, want only the third asked", got)
}
}
func TestAnAskRefusedForWantOfRoomIsNotAskedAgainOnceTheGateMovedOn(t *testing.T) { // 1610/AC2
func TestAnAskRefusedForWantOfRoomIsNotAskedAgainOnceTheAssignerMovedOn(t *testing.T) { // 1610/AC2
b := newBed(t)
_, first := b.askOf(t, 1, 1)
key, gone := b.askOf(t, 2, 2)
b.refuse(t, gone.Ask.ID, roomRefusal)
// Withdrawn by the gate (a head moved on) before room came.
delete(b.gate, key)
b.t.OnAskGone(key)
key3, g3 := b.askOf(t, 3, 3)
b.refuse(t, g3.Ask.ID, roomRefusal)
// Rewritten by the gate under its next asking before room came: the later asking is asked under its own id, and the
_, first := b.holdOf(t, 1, 1)
key, gone := b.holdOf(t, 2, 1)
b.refuse(t, gone.ID, roomRefusal)
// Taken away by the assigner (the hold lifted) before room came.
delete(b.assigner, key)
b.t.OnAssignerAskGone(key)
key3, g3 := b.holdOf(t, 3, 1)
b.refuse(t, g3.ID, roomRefusal)
// Rewritten by the assigner under its next asking before room came: the later asking is asked under its own id, and the
// refused earlier asking is not asked again.
_, v3b, g3b := gateAskN(t, headsN(3), headsN(3), 2)
b.gate[key3] = v3b
b.t.OnAsk(key3, 4, v3b)
_, v3b, g3b := holdAskAt(t, 3, 2)
b.assigner[key3] = v3b
b.t.OnAssignerAsk(key3)
before := len(b.askedIDs())
if ids := b.askedIDs(); ids[before-1] != g3b.Ask.ID {
t.Fatalf("asks = %v, want the next asking %s asked", ids, g3b.Ask.ID)
if ids := b.askedIDs(); ids[before-1] != g3b.ID {
t.Fatalf("asks = %v, want the next asking %s asked", ids, g3b.ID)
}
b.deliver(t, warrantFor(first.Ask, "approve"))
b.deliver(t, warrantFor(first, "keep"))
b.expire(t, g3b)
if after := b.askedIDs()[before:]; len(after) != 0 {
t.Errorf("asked again %v, which the gate no longer asks", after)
t.Errorf("asked again %v, which the assigner no longer asks", after)
}
}
func TestAnAskRefusedForWantOfRoomTooOftenIsSaidAsACondition(t *testing.T) { // 1610/AC3
b := newBed(t)
_, held := b.askOf(t, 1, 1)
_, held := b.holdOf(t, 1, 1)
for i := 0; i < mostRoomRefusals; i++ {
if len(b.conds) != 0 {
t.Fatalf("a condition after %d refusals: %+v", i, b.conds)
}
b.refuse(t, held.Ask.ID, roomRefusal)
_, other := b.askOf(t, 2+i, uint64(2+i))
b.refuse(t, held.ID, roomRefusal)
_, other := b.holdOf(t, 2+i, 1)
b.expire(t, other)
}
var again int
for _, id := range b.askedIDs() {
if id == held.Ask.ID {
if id == held.ID {
again++
}
}
@@ -149,7 +130,7 @@ func TestAnAskRefusedForWantOfRoomTooOftenIsSaidAsACondition(t *testing.T) { //
if len(b.conds) != 1 || b.conds[0].State != occ.StateRaised || !strings.Contains(b.conds[0].Summary, "open asks") {
t.Fatalf("conditions = %+v", b.conds)
}
_, last := b.askOf(t, 20, 20)
_, last := b.holdOf(t, 8, 1)
b.expire(t, last)
if len(b.askedIDs()) != asked+1 {
t.Errorf("asked again after the bound: %v", b.askedIDs())
@@ -158,47 +139,47 @@ func TestAnAskRefusedForWantOfRoomTooOftenIsSaidAsACondition(t *testing.T) { //
func TestAnAskWaitingForRoomIsAskedAgainAtStart(t *testing.T) { // 1610/AC4
b := newBed(t)
key, held := b.askOf(t, 1, 1)
b.refuse(t, held.Ask.ID, roomRefusal)
key, held := b.holdOf(t, 1, 1)
b.refuse(t, held.ID, roomRefusal)
// The teller starts again: a new teller on the same buckets, and the watch hands over the key it holds.
b2 := newBed(t)
b2.store, b2.answers, b2.gate = b.store, b.answers, b.gate
b2.store, b2.answers, b2.assigner = b.store, b.answers, b.assigner
b2.client.State = b.store
b2.t.Store, b2.t.Answers, b2.t.Gate = b.store, b.answers, b.gate
b2.t.OnAsk(key, 1, b.gate[key])
b2.t.Store, b2.t.Answers, b2.t.Assigner = b.store, b.answers, b.assigner
b2.t.OnAssignerAsk(key)
if len(b2.askedIDs()) != 0 {
t.Fatalf("asked before the watch handed over its keys: %v", b2.askedIDs())
}
b2.t.OnCurrent()
if got := b2.askedIDs(); len(got) != 1 || got[0] != held.Ask.ID {
t.Errorf("asks at start = %v, want %s asked again", got, held.Ask.ID)
if got := b2.askedIDs(); len(got) != 1 || got[0] != held.ID {
t.Errorf("asks at start = %v, want %s asked again", got, held.ID)
}
}
func TestAnAskRefusedForItsWordsIsSaidAtOnceAndNotAskedAgain(t *testing.T) { // 1610/AC5
b := newBed(t)
_, first := b.askOf(t, 1, 1)
_, bad := b.askOf(t, 2, 2)
b.refuse(t, bad.Ask.ID, "its words carry an address, which may not leave the mesh")
_, first := b.holdOf(t, 1, 1)
_, bad := b.holdOf(t, 2, 1)
b.refuse(t, bad.ID, "its words carry an address, which may not leave the mesh")
if len(b.conds) != 1 {
t.Fatalf("conditions = %+v", b.conds)
}
b.deliver(t, warrantFor(first.Ask, "approve"))
b.deliver(t, warrantFor(first, "keep"))
if got := b.askedIDs(); len(got) != 2 {
t.Errorf("asks = %v: an ask refused for its words was asked again", got)
}
}
// restart is a new teller on the same buckets and the same bus (its message ids still in the bus's duplicate window), and
// the router's records as they stand; the watch hands over the gate's keys, then says it has handed over all.
// the router's records as they stand; the watch hands over the assigner's keys, then says it has handed over all.
func (b *bed) restart(t *testing.T) *bed {
t.Helper()
n := newBed(t)
n.store, n.answers, n.gate, n.records, n.msgIDs, n.seenIDs = b.store, b.answers, b.gate, b.records, b.msgIDs, b.seenIDs
n.store, n.answers, n.assigner, n.records, n.msgIDs, n.seenIDs = b.store, b.answers, b.assigner, b.records, b.msgIDs, b.seenIDs
n.client.State = b.store
n.t.Store, n.t.Answers, n.t.Gate = b.store, b.answers, b.gate
for key, value := range b.gate {
n.t.OnAsk(key, 1, value)
n.t.Store, n.t.Answers, n.t.Assigner = b.store, b.answers, b.assigner
for key := range b.assigner {
n.t.OnAssignerAsk(key)
}
n.t.OnCurrent()
return n
@@ -209,22 +190,21 @@ func (b *bed) restart(t *testing.T) *bed {
// so no record there means it was not taken.
func TestAnAskAskedAgainWhoseRefusalWasMissedIsAskedAgainAtStart(t *testing.T) { // 1610/AC4
b := newBed(t)
b.askOf(t, 1, 1)
_, held := b.askOf(t, 2, 2)
b.refuse(t, held.Ask.ID, roomRefusal)
_, other := b.askOf(t, 3, 3)
b.expire(t, other)
b.records[other.Ask.ID] = `{"state":"expired"}`
if got := b.askedIDs(); got[len(got)-1] != held.Ask.ID {
t.Fatalf("asks = %v, want %s asked again", got, held.Ask.ID)
_, held := b.holdOf(t, 2, 1)
b.refuse(t, held.ID, roomRefusal)
_, other := b.holdOf(t, 3, 1)
b.deliver(t, warrantFor(other, "keep"))
b.records[other.ID] = `{"state":"chosen"}`
if got := b.askedIDs(); got[len(got)-1] != held.ID {
t.Fatalf("asks = %v, want %s asked again", got, held.ID)
}
// Refused again, and the teller does not hear it: the router forgets it.
delete(b.seenIDs, held.Ask.ID)
delete(b.records, held.Ask.ID)
delete(b.seenIDs, held.ID)
delete(b.records, held.ID)
n := b.restart(t)
got := n.askedIDs()
if len(got) != 1 || got[0] != held.Ask.ID {
t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.Ask.ID, n.logs.String())
if len(got) != 1 || got[0] != held.ID {
t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.ID, n.logs.String())
}
if len(n.busDropped) != 0 {
t.Errorf("the bus dropped %v as a duplicate", n.busDropped)
@@ -234,13 +214,13 @@ func TestAnAskAskedAgainWhoseRefusalWasMissedIsAskedAgainAtStart(t *testing.T) {
// The same for a first asking: kept open by the teller, its refusal never heard, and no record at the router.
func TestAFirstAskingWhoseRefusalWasMissedIsAskedAgainAtStart(t *testing.T) { // 1610/AC4
b := newBed(t)
_, held := b.askOf(t, 1, 1)
delete(b.seenIDs, held.Ask.ID) // refused for room, unheard: the router keeps no record of it
delete(b.records, held.Ask.ID)
_, held := b.holdOf(t, 1, 1)
delete(b.seenIDs, held.ID) // refused for room, unheard: the router keeps no record of it
delete(b.records, held.ID)
n := b.restart(t)
got := n.askedIDs()
if len(got) != 1 || got[0] != held.Ask.ID {
t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.Ask.ID, n.logs.String())
if len(got) != 1 || got[0] != held.ID {
t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.ID, n.logs.String())
}
if len(n.busDropped) != 0 {
t.Errorf("the bus dropped %v as a duplicate", n.busDropped)
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go
+45 -265
View File
@@ -5,7 +5,6 @@ import (
"errors"
"fmt"
"slices"
"sort"
"strings"
"sync"
"time"
@@ -17,45 +16,14 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
)
// The teller carries the gate's merge asks to the operator and writes the operator's answer back (novox/hq ADR 0437
// §6, §9). It reads the gate's asks, publishes each to the router unchanged as the asker mesh-delivery-teller
// (mesh-sdk go/asker), withdraws it when the gate takes its key away, and on the warrant the router says to it
// holds the warrant to the ask (ADR 0259 §6) and writes the answer under the ask's own key by compare-and-set. A
// warrant that does not hold writes no answer and is said as a condition. It holds no timer and serves no request:
// every reaction is a revision of the gate's key or a warrant.
// The teller carries the assigner's asks to the operator and writes the operator's answer back (novox/hq ADR 0437
// §6, §9; design 47 G §653, A-172: the merge ask went with the Approve per merge). It publishes each of the assigner's
// asks to the router unchanged as the asker mesh-delivery-teller (mesh-sdk go/asker), withdraws it when the assigner takes
// its key away, and on the warrant the router says to it holds the warrant to the ask (ADR 0259 §6) and writes the
// answer under the ask's own key by compare-and-set. A warrant that does not hold writes no answer and is said as a
// condition. It holds no timer and serves no request: every reaction is a revision of a key or a warrant.
// gateAsked is the gate's ask with the router's ask decoded from its JSON (the seat's ask value carries operator-channel's ask as
// JSON, novox/hq ADR 0627 §4).
type gateAsked struct {
Group string
Heads []meshdeliveryclient.Head
Asking int
Ask oc.Ask
}
// decodeGateAsk reads the gate's ask value and the router's ask in it, as meshdeliveryclient.DecodeAsk did before the ask was JSON.
func decodeGateAsk(value []byte) (gateAsked, error) {
ga, err := meshdeliveryclient.DecodeAsk(value)
if err != nil {
return gateAsked{}, err
}
heads := meshdeliveryclient.HeadsOf(ga.Heads)
key, err := meshdeliveryclient.MergeAskKey(ga.Group, heads)
if err != nil {
return gateAsked{}, err
}
raw, err := json.Marshal(ga.Ask)
if err != nil {
return gateAsked{}, err
}
a, err := occ.Carried(raw, meshdeliveryclient.AskingID(key, int(ga.Asking)))
if err != nil {
return gateAsked{}, err
}
return gateAsked{Group: ga.Group, Heads: heads, Asking: int(ga.Asking), Ask: a}, nil
}
// store is where the teller keeps what it asked: the asker's own records (`ask.<id>`) and, beside them, which gate key
// store is where the teller keeps what it asked: the asker's own records (`ask.<id>`) and, beside them, which ask key
// each ask was made for and at which revision of it (`key.<id>`), and which conditions it has raised (`cond.<id>`).
type store interface {
occ.Store
@@ -79,8 +47,9 @@ type answers interface {
Write(key string, value []byte, rev uint64, found bool) (written bool, err error)
}
// gateAsks reads the gate's bucket asks.
type gateAsks interface {
// asksReader reads the picture of a state another module keeps (the assigner's asks, the view's answers), as the
// reader of mesh-sdk's views gives it.
type asksReader interface {
Get(key string) (value []byte, found bool, err error)
}
@@ -89,11 +58,10 @@ type Teller struct {
Asker routerAsks
Store store
Answers answers
Gate gateAsks
// Assigner is the assigner's bucket asks and View the view's bucket answers (issue 1297 B10); with either nil the teller
// answers nothing from the view.
Assigner gateAsks
View gateAsks
// Assigner is the picture of the assigner's asks and View the picture of the view's answers (issue 1297 B10); with
// either nil the teller answers nothing from the view.
Assigner asksReader
View asksReader
// Say publishes a condition to the operator (mesh.seat.operator-channel.accept.condition.<module>).
Say func(c oc.Condition) error
Logf func(format string, args ...any)
@@ -118,8 +86,8 @@ func (t *Teller) logf(format string, args ...any) {
}
}
// seen is what the teller keeps beside an ask it made. Under `key.<router id>` it names the gate key the asking was made for;
// under `gate.<gate key>` it names the router id of the latest asking of that key and the revision of the key it was made
// seen is what the teller keeps beside an ask it made. Under `key.<router id>` it names the ask key (the assigner's) the asking was made for;
// under `gate.<ask key>` it names the router id of the latest asking of that key and the revision of the key it was made
// at, so a replay of the revision asks nothing again and an overwrite with a new id is known for a new asking.
type seen struct {
Key string `json:"key,omitempty"`
@@ -135,7 +103,7 @@ func recordKey(id string) string { return "key." + id }
func keptKey(id string) string { return "ask." + id }
func condKey(id string) string { return "cond." + id }
// condID is what a condition of an ask is kept and named by: the gate key it was asked for, so the asking that follows a bad
// condID is what a condition of an ask is kept and named by: the ask key it was asked for, so the asking that follows a bad
// answer clears the condition of the one before it. An id the teller holds no key for is its own.
func (t *Teller) condID(id string) string {
if s, e, _ := t.seenOf(id); e != nil && s.Key != "" {
@@ -156,7 +124,7 @@ func (t *Teller) seenAt(rk string) (seen, *stdio.StateEntry, error) {
return s, e, nil
}
// seenOf is the record of a router id: the gate key it was asked for.
// seenOf is the record of a router id: the ask key it was asked for.
func (t *Teller) seenOf(id string) (seen, *stdio.StateEntry, error) { return t.seenAt(recordKey(id)) }
func (t *Teller) keptOf(id string) (*occ.Kept, uint64, error) {
@@ -174,7 +142,7 @@ func (t *Teller) keptOf(id string) (*occ.Kept, uint64, error) {
// open says whether an ask is made and neither answered nor ended.
func open(k *occ.Kept) bool { return k != nil && !k.Acted && k.Ended == "" }
// note keeps that the asking with this router id is the latest of the gate key, made at revision rev of it.
// note keeps that the asking with this router id is the latest of the ask key, made at revision rev of it.
func (t *Teller) note(id, key string, rev uint64, prior *stdio.StateEntry) error {
if _, err := t.Store.Create(recordKey(id), seen{Key: key, Rev: rev}); err != nil && !errors.Is(err, stdio.ErrChanged) {
return err
@@ -188,82 +156,6 @@ func (t *Teller) note(id, key string, rev uint64, prior *stdio.StateEntry) error
return err
}
// OnAsk handles the gate's asks key put at revision rev. Each asking has its own router id, which is the value's ask.id and
// never one recomputed from the key (mesh-sdk go/delivery Ask.Asking, AskingID): the router keeps an ended ask and drops a
// second ask under its id. So the ask is published once per id; a key overwritten at a higher revision under a new id is a
// new asking, and the earlier asking's router ask is cancelled first. A key whose value is not the ask its name says, one
// already answered, and a revision the teller has handled are left. An ask still open and unchanged is only noted (and its
// warrant read from the router, in case it came while the teller was away). A value that changed its words under the id
// it was asked under is cancelled and not asked again: the router would drop it, so the gate must give it a new asking.
func (t *Teller) OnAsk(key string, rev uint64, value []byte) {
t.mu.Lock()
defer t.mu.Unlock()
ga, err := decodeGateAsk(value)
if err != nil {
t.logf("the gate's key %s at revision %d is not an ask, and is left: %v", key, rev, err)
return
}
if want, err := meshdeliveryclient.MergeAskKey(ga.Group, ga.Heads); err != nil || want != key {
t.logf("the gate's key %s at revision %d holds the ask of %q, not its own, and is left (%v)", key, rev, want, err)
return
}
id := ga.Ask.ID
if raw, _, has, err := t.Answers.Get(key); err != nil {
t.logf("%s: the answers could not be read, so the ask is left: %v", key, err)
return
} else if has && answeredAsking(raw) == id {
return // this asking is answered; an earlier asking's answer does not stop a later asking
}
prior, grec, err := t.seenAt(gateRecordKey(key))
if err != nil {
t.logf("%s: %v", key, err)
return
}
if grec != nil && rev <= prior.Rev {
if k, _, err := t.keptOf(prior.ID); err == nil && open(k) {
t.catchUp(prior.ID)
}
return
}
if grec != nil && prior.ID != id {
// A new asking of the group at these heads: the earlier one is taken back.
t.cancelOpen(key, prior.ID)
}
k, _, err := t.keptOf(id)
if err != nil {
t.logf("%s: %v", key, err)
return
}
if k != nil {
switch {
case open(k) && occ.Digest(k.Ask) == occ.Digest(ga.Ask):
_ = t.note(id, key, rev, grec)
t.catchUp(id)
case open(k):
t.cancelOpen(key, id)
_ = t.note(id, key, rev, grec)
t.logf("%s: the ask %s changed its words at revision %d under the same id: cancelled, and not asked again, for the router drops a second ask under an id (the gate gives a new asking a new id)", key, id, rev)
default:
_ = t.note(id, key, rev, grec)
t.logf("%s: the ask %s ended, and the gate wrote it again under the same id at revision %d: not asked again, the router would drop it", key, id, rev)
}
return
}
if err := t.note(id, key, rev, grec); err != nil {
t.logf("%s: the record of the ask %s could not be kept, so it is left: %v", key, id, err)
return
}
if err := t.Asker.Ask(ga.Ask); err != nil {
t.logf("%s at revision %d: the ask %s was not asked: %v", key, rev, id, err)
if kept, _, _ := t.keptOf(id); open(kept) {
// Kept but not published: forgotten, so the next revision of the key asks it.
_ = t.Store.Delete(keptKey(id))
}
return
}
t.logf("%s at revision %d: the ask %s is asked of the operator (asking %d)", key, rev, id, ga.Asking)
}
// cancelOpen takes an open ask back at the router and marks it ended here, since the router's word that it ended may come
// later. An ask that is not open is left.
func (t *Teller) cancelOpen(key, id string) {
@@ -286,25 +178,6 @@ func (t *Teller) cancelOpen(key, id string) {
t.logf("%s: the ask %s is cancelled", key, id)
}
// OnAskGone handles the gate's asks key taken away (a withdrawn ask): the latest asking's router ask is cancelled, and an
// answer to it merges nothing.
func (t *Teller) OnAskGone(key string) {
t.mu.Lock()
defer t.mu.Unlock()
if _, _, _, err := meshdeliveryclient.ParseAskKey(key); err != nil {
t.logf("the gate's key %s was taken away and is no ask key, and is left: %v", key, err)
return
}
prior, grec, err := t.seenAt(gateRecordKey(key))
if err != nil || grec == nil {
if err != nil {
t.logf("%s: %v", key, err)
}
return
}
t.cancelOpen(key, prior.ID)
}
// catchUp reads the router's record of an ask the teller made and hands its warrant on, as the router's event would:
// a warrant that came while the teller was away.
func (t *Teller) catchUp(id string) {
@@ -327,34 +200,6 @@ func (t *Teller) catchUp(id string) {
t.said(h, *w, err)
}
// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge; the settle ask
// went with the verb settle, design 47 G §431), this ask, the answer, and the heads it covers.
// It is spelled as the gate composes it (mesh-catalog modules/mesh-delivery-gate approval.go mergeAct, from
// modules/mesh-delivery approval.go:63-65) and as mesh-controller verifies it (cmd/mesh-controller/moved.go:98-106).
func mergeAct(verb, ask, option string, heads map[string]string) occ.Act {
var out []string
for id, sha := range heads {
out = append(out, id+"="+sha)
}
sort.Strings(out)
return occ.Act{"verb": verb, "ask": ask, "option": option, "heads": strings.Join(out, ",")}
}
// headsOf is the heads an ask covers as an act names them: a delivery's id, `owner/repository@<twelve characters of its
// commit>` with the repository in lower case, as the gate's IDOf spells it (mesh-catalog modules/mesh-delivery-gate model.go),
// and its whole commit.
func headsOf(heads []meshdeliveryclient.Head) map[string]string {
out := map[string]string{}
for _, h := range heads {
short := h.Head
if len(short) > 12 {
short = short[:12]
}
out[strings.ToLower(h.Repository)+"@"+short] = h.Head
}
return out
}
// answeredAsking is the router id of the asking an answer is for: its warrant's id before the "@". An answer that cannot be
// read answers no asking.
func answeredAsking(raw []byte) string {
@@ -376,47 +221,20 @@ func warrantID(w oc.Warrant) string {
}
// Answer is the act of a warrant that holds the ask (occ.Client.Take has checked its asker, the ask's digest, the
// option and its level, and the time): the option's bound act must be the act of this ask and the heads it covers,
// and then the answer is written under the ask's key, once. An error is a warrant that does not hold here.
// option and its level, and the time): the assigner's ask must still stand (answerAssigner), and then the answer is
// written under the ask's key, once. An error is a warrant that does not hold here.
func (t *Teller) Answer(o oc.Option, w oc.Warrant) error {
s, srec, err := t.seenOf(w.Ask)
if err != nil {
return err
}
if srec == nil {
return fmt.Errorf("the teller holds no gate key for the ask %s", w.Ask)
return fmt.Errorf("the teller holds no key for the ask %s", w.Ask)
}
if isAssignerKey(s.Key) {
return t.answerAssigner(s.Key, o, w)
if !isAssignerKey(s.Key) {
return fmt.Errorf("the ask %s was asked for %s, which is no ask of the assigner: the teller carries no merge ask", w.Ask, s.Key)
}
raw, found, err := t.Gate.Get(s.Key)
if err != nil {
return fmt.Errorf("the gate's ask %s could not be read: %w", s.Key, err)
}
if !found {
t.logf("%s: the ask %s was answered and the gate has taken it away: an answer to a withdrawn ask merges nothing", s.Key, w.Ask)
return nil
}
ga, err := decodeGateAsk(raw)
if err != nil {
return fmt.Errorf("the gate's ask %s is not an ask: %w", s.Key, err)
}
if ga.Ask.ID != w.Ask {
t.logf("%s: the ask %s was answered, and the gate has asked again under %s: an answer to an earlier asking merges nothing", s.Key, w.Ask, ga.Ask.ID)
return nil
}
if occ.Digest(ga.Ask) != w.AskDigest {
return fmt.Errorf("the gate's ask %s is no longer the ask the operator answered (its digest is %s, the warrant's %s)",
s.Key, occ.Digest(ga.Ask), w.AskDigest)
}
if err := occ.Performs(o, mergeAct(meshdeliveryclient.AskVerb(ga.Group), w.Ask, o.ID, headsOf(ga.Heads))); err != nil {
return err
}
body, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: o.ID, Warrant: warrantID(w), Words: occ.Says(w)})
if err != nil {
return err
}
return t.writeAnswer(s.Key, w.Ask, body, occ.Says(w))
return t.answerAssigner(s.Key, o, w)
}
// writeAnswer writes the answer to the asking askID under the ask's key by compare-and-set: created when the key has no answer;
@@ -500,10 +318,10 @@ func (t *Teller) saidOnce(h occ.Handled, w oc.Warrant, err error) {
t.clear(w.Ask)
case h == occ.Refused || (h == occ.Acted && err != nil):
t.logf("the warrant of %s does not hold, and no answer is written: %v", w.Ask, err)
t.raise(w.Ask, t.kindOf(w.Ask).answerHeadline, "its answer did not hold here: "+oneLine(err))
t.raise(w.Ask, assignerWords.answerHeadline, "its answer did not hold here: "+oneLine(err))
case h == occ.Recorded && w.Outcome == occ.OutcomeRefused:
t.logf("the router refused the ask %s: %s", w.Ask, w.Words)
t.raise(w.Ask, t.kindOf(w.Ask).refusedHeadline, "the router refused it: "+w.Words)
t.raise(w.Ask, assignerWords.refusedHeadline, "the router refused it: "+w.Words)
case err != nil:
t.logf("the warrant of %s: %v", w.Ask, err)
}
@@ -523,30 +341,19 @@ func oneLine(err error) string {
// conditionKey is the key of the condition about an ask.
func conditionKey(id string) string { return tellerModule + ".merge-ask." + id }
// askWords are the words the teller's conditions use for one kind of ask: the gate's merge ask, or one of the assigner's.
// askWords are the words the teller's conditions use for an ask: one of the assigner's.
type askWords struct {
scope, noun, needs, nothing, resolved string
answerHeadline, refusedHeadline, unaskedHeadline string
}
var (
mergeWords = askWords{scope: "merge", noun: "The merge ask ", needs: "to ask for the Approve again", nothing: "Nothing was merged.",
resolved: "A good answer to the merge ask was written.", answerHeadline: "A merge Approve did not hold",
refusedHeadline: "A merge ask was refused", unaskedHeadline: "A merge ask could not be asked"}
assignerWords = askWords{scope: "delivery", noun: "The delivery ask ", needs: "to answer it in the delivery view",
nothing: "Nothing was changed on a machine.", resolved: "A good answer to the delivery ask was written.",
answerHeadline: "An answer to a delivery ask did not hold", refusedHeadline: "A delivery ask was refused",
unaskedHeadline: "A delivery ask could not be asked"}
)
// kindOf is the words for the ask with this router id, by the key it was asked for.
func (t *Teller) kindOf(askID string) askWords {
if isAssignerKey(t.condID(askID)) {
return assignerWords
}
return mergeWords
}
func (t *Teller) raise(askID, headline, why string) {
if t.Say == nil {
return
@@ -558,7 +365,7 @@ func (t *Teller) raise(askID, headline, why string) {
}
t.logf("the condition of %s could not be kept: %v", id, err)
}
k := t.kindOf(askID)
k := assignerWords
needs := k.needs
c := oc.Condition{Key: conditionKey(id), Severity: occ.SeverityWarning, Subject: oc.ConditionSubject{Scope: k.scope, ID: id}, State: occ.StateRaised,
Summary: k.noun + id + ": " + why, Headline: headline, Needs: needs,
@@ -579,7 +386,7 @@ func (t *Teller) clear(askID string) {
if err != nil || e == nil {
return
}
c := oc.Condition{Key: conditionKey(id), Subject: oc.ConditionSubject{Scope: t.kindOf(askID).scope, ID: id}, State: occ.StateCleared}
c := oc.Condition{Key: conditionKey(id), Subject: oc.ConditionSubject{Scope: assignerWords.scope, ID: id}, State: occ.StateCleared}
if err := t.Say(c); err != nil {
t.logf("the condition of %s could not be cleared: %v", id, err)
return
@@ -594,8 +401,8 @@ func (t *Teller) clear(askID string) {
// `room`, and each time one of the teller's asks ends at the router (a warrant of any outcome: the room it held is free)
// the first that waits is asked again, under its own router id and a bus message id of its own (the bus drops a second
// message under an id it holds). It is asked again at most mostRoomRefusals times; refused that often it is said as a
// condition, as any refusal is. An ask the gate took away or asked again under a newer asking meanwhile, or one already
// answered, is the gate's no more, and is dropped from the wait. The teller holds no timer: an end, and its start, are the
// condition, as any refusal is. An ask the assigner took away or asked again under a newer asking meanwhile, or one already
// answered, is the assigner's no more, and is dropped from the wait. The teller holds no timer: an end, and its start, are the
// events that ask again.
// mostRoomRefusals is how many times an ask is refused for want of room before it is said as a condition and asked no
@@ -605,7 +412,7 @@ const mostRoomRefusals = 6
// roomRecord is the key of the asks that wait for room, in the order they were refused.
const roomRecord = "room"
// waitingAsk is one ask that waits for room: the gate key it was asked for, its router id, how often the router refused
// waitingAsk is one ask that waits for room: the ask key it was asked for, its router id, how often the router refused
// it for want of room, and whether it is asked again and its word not yet heard.
type waitingAsk struct {
Key string `json:"key"`
@@ -680,7 +487,7 @@ func (t *Teller) roomRefused(w oc.Warrant) {
return
}
t.logf("the router refused the ask %s for want of room %d times (%s): it is asked no more, and said", w.Ask, n, w.Words)
t.raise(w.Ask, t.kindOf(w.Ask).unaskedHeadline, fmt.Sprintf("the router refused it %d times for want of room: %s", n, w.Words))
t.raise(w.Ask, assignerWords.unaskedHeadline, fmt.Sprintf("the router refused it %d times for want of room: %s", n, w.Words))
return
}
if t.keepWaiting(ws, rec) != nil {
@@ -703,7 +510,7 @@ func (t *Teller) endWait(id string) {
_ = t.keepWaiting(slices.Delete(ws, i, i+1), rec)
}
// OnCurrent is the watch of the gate's asks having handed over every key it holds now, at the teller's start: an ask that
// OnCurrent is the watch of the assigner's asks having handed over every key it holds now, at the teller's start: an ask that
// waited for room while the teller was away is asked again (the asks that held the room may have ended meanwhile).
//
// An ask asked again whose word the teller did not hear (the router's warrants come on a plain subscription) waits once
@@ -754,7 +561,7 @@ func (t *Teller) unheard(id string) {
}
}
// askWaiting asks again the first n asks that wait for room and are not asked yet, dropping on the way every one the gate
// askWaiting asks again the first n asks that wait for room and are not asked yet, dropping on the way every one the assigner
// no longer asks. One at an end: one ask ended, so there is room for one.
func (t *Teller) askWaiting(n int) {
ws, rec, err := t.waiting()
@@ -777,9 +584,8 @@ func (t *Teller) askWaiting(n int) {
i++
continue
}
assigner := isAssignerKey(x.Key)
if assigner && !t.assignerRoom() {
i++ // the assigner's asks hold their most open: it waits, and a merge ask after it may be asked
if isAssignerKey(x.Key) && !t.assignerRoom() {
i++ // the assigner's asks hold their most open: it waits
continue
}
a, why := t.stillAsked(x)
@@ -794,9 +600,7 @@ func (t *Teller) askWaiting(n int) {
return
}
ws[i].Asked = true
if assigner {
t.assignerAsked(x.ID)
}
t.assignerAsked(x.ID)
t.logf("%s: the ask %s is asked of the operator again (refused %d times for want of room)", x.Key, x.ID, x.Refusals)
if n--; n <= 0 {
return
@@ -805,40 +609,16 @@ func (t *Teller) askWaiting(n int) {
}
}
// stillAsked is the gate's ask a waiting ask is for, when the gate still asks it under the same id and it is unanswered;
// otherwise why it is not.
// stillAsked is the assigner's ask a waiting ask is for, when the assigner still asks it under the same id and it is
// unanswered; otherwise why it is not. A waiting ask of a key that is no ask of the assigner (a merge ask kept before the
// Approve per merge went) is dropped.
func (t *Teller) stillAsked(x waitingAsk) (oc.Ask, string) {
if x.Key == "" {
return oc.Ask{}, "the teller holds no gate key for it"
return oc.Ask{}, "the teller holds no key for it"
}
if isAssignerKey(x.Key) {
return t.stillAskedByAssigner(x)
if !isAssignerKey(x.Key) {
return oc.Ask{}, "it is no ask of the assigner, and the teller carries no merge ask"
}
raw, found, err := t.Gate.Get(x.Key)
if err != nil {
return oc.Ask{}, "the gate's ask could not be read: " + err.Error()
}
if !found {
return oc.Ask{}, "the gate has taken it away"
}
ga, err := decodeGateAsk(raw)
if err != nil {
return oc.Ask{}, "the gate's key holds no ask: " + err.Error()
}
if ga.Ask.ID != x.ID {
return oc.Ask{}, "the gate asks again under " + ga.Ask.ID
}
if cur, _, has, err := t.Answers.Get(answerKey(x.Key)); err == nil && has && answeredAsking(cur) == x.ID {
return oc.Ask{}, "it is answered"
}
if err := occ.Check(ga.Ask, t.now()); err != nil {
return oc.Ask{}, "it no longer holds: " + err.Error()
}
return ga.Ask, ""
}
// stillAskedByAssigner is stillAsked for one of the assigner's asks.
func (t *Teller) stillAskedByAssigner(x waitingAsk) (oc.Ask, string) {
a, found, err := t.assignerAsk(x.Key)
switch {
case err != nil:
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
+147 -314
View File
@@ -1,9 +1,7 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
@@ -12,19 +10,11 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
meshdelivery "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
stdio "git.novox.be/novox/mesh-sdk/go"
)
const (
headA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
headB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
)
var now = time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)
// nowIs sets the tests' clock now to at and returns what puts it back; a test sets it only as t.Cleanup(nowIs(at)), so no
@@ -119,10 +109,11 @@ type bed struct {
client *occ.Client
store *memStore
answers *memAnswers
gate memGate
pubs []published
conds []oc.Condition
records map[string]string
// assigner and view are the pictures of the assigner's asks and the view's answers the teller follows.
assigner, view memGate
pubs []published
conds []oc.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.
@@ -140,7 +131,7 @@ type bed struct {
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{}}
assigner: memGate{}, view: memGate{}, records: map[string]string{}, seenIDs: map[string]bool{}, msgIDs: map[string]bool{}}
b.client = &occ.Client{Module: tellerModule, State: b.store, Now: func() time.Time { return now },
Publish: func(subject string, body any, id string) error {
raw, err := json.Marshal(body)
@@ -173,7 +164,7 @@ func newBed(t *testing.T) *bed {
},
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 },
b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Assigner: b.assigner, View: b.view, Publish: b.client.Publish, Now: func() time.Time { return now },
Say: func(c oc.Condition) error {
if _, err := occ.EncodeCondition(c); err != nil {
t.Errorf("a condition the router would refuse: %v", err)
@@ -201,41 +192,31 @@ func newBed(t *testing.T) *bed {
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 []meshdeliveryclient.Head) (key string, value []byte, ask gateAsked) {
// holdOf is the assigner's hold ask of module plex at the n-th fingerprint, written at its asking-th writing (its router id is
// AskingID(key, asking)), put in the bed's picture of the assigner's asks and handed to the teller as the watch would.
func (b *bed) holdOf(t *testing.T, n, asking int) (string, oc.Ask) {
t.Helper()
return gateAskN(t, heads, boundTo, 1)
key, value, ask := holdAskAt(t, n, asking)
b.assigner[key] = value
b.t.OnAssignerAsk(key)
return key, ask
}
// 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 []meshdeliveryclient.Head, asking int) (key string, value []byte, ask gateAsked) {
// holdAskAt is the assigner's hold ask at the n-th fingerprint and the asking-th writing, shaped as the assigner composes it
// (two options of level acknowledge that bind no act, so the router takes it).
func holdAskAt(t *testing.T, n, asking int) (string, []byte, oc.Ask) {
t.Helper()
return gateAskOf(t, "g", "mesh-delivery.merge", heads, boundTo, asking)
}
// gateAskOf is gateAskN for a group whose options bind the verb given.
func gateAskOf(t *testing.T, group, verb string, heads, boundTo []meshdeliveryclient.Head, asking int) (key string, value []byte, ask gateAsked) {
t.Helper()
key, err := meshdeliveryclient.MergeAskKey(group, heads)
key, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64))
if err != nil {
t.Fatal(err)
}
id := meshdeliveryclient.AskingID(key, asking)
a := oc.Ask{ID: id, Headline: "Merge into main: novox/mesh-delivery#12", Explanation: "Merges two pull requests.\nReview: approved.",
Who: occ.Operator, Expires: occ.FormatTime(now.Add(occ.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 := occ.ActDigest(mergeAct(verb, id, o.id, headsOf(boundTo)))
if err != nil {
t.Fatal(err)
}
a.Options = append(a.Options, oc.Option{ID: o.id, Label: o.label, Does: o.does, Level: occ.Approve, Binds: binds})
_, _, ask := holdAsk(t)
ask.ID = meshdeliveryclient.AskingID(key, asking)
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask), Asking: asking})
if err != nil {
t.Fatal(err)
}
ask = gateAsked{Group: group, Heads: heads, Asking: asking, Ask: a}
return key, encodeGate(t, ask), ask
return key, value, ask
}
// rawAsk is the router's ask as the JSON a gate or assigner value carries (the seat's ask value).
@@ -248,31 +229,9 @@ func rawAsk(t *testing.T, a oc.Ask) json.RawMessage {
return b
}
// encodeGate is the gate's value for ga, written as the gate writes it: the router's ask checked against its key and asking
// (an id other than the one they give, no option), then carried as JSON.
func encodeGate(t *testing.T, ga gateAsked) []byte {
t.Helper()
key, err := meshdeliveryclient.MergeAskKey(ga.Group, ga.Heads)
if err != nil {
t.Fatal(err)
}
if want := meshdeliveryclient.AskingID(key, ga.Asking); ga.Ask.ID != want || len(ga.Ask.Options) == 0 {
t.Fatalf("the ask's id %q, want %q, with %d options", ga.Ask.ID, want, len(ga.Ask.Options))
}
value, err := gateValue(ga.Group, ga.Heads, ga.Asking, rawAsk(t, ga.Ask))
if err != nil {
t.Fatal(err)
}
return value
}
func twoHeads() []meshdeliveryclient.Head {
return []meshdeliveryclient.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 oc.Ask, option string) oc.Warrant {
return oc.Warrant{Ask: a.ID, Asker: tellerModule, Outcome: occ.OutcomeChosen, Option: option, Label: "Approve", Level: occ.Approve,
return oc.Warrant{Ask: a.ID, Asker: tellerModule, Outcome: occ.OutcomeChosen, Option: option, Label: option, Level: occ.Acknowledge,
By: &oc.Person{Who: occ.Operator, Kind: "telegram", Identity: "42", Verified: "user id"}, Channel: "telegram",
At: occ.FormatTime(now.Add(time.Hour)), AskDigest: occ.Digest(a)}
}
@@ -295,72 +254,51 @@ func (b *bed) askedOf(subject string) []published {
return out
}
func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1343/AC1
t.Run("1945: a refused bucket is said, or stops the start naming its grant", func(t *testing.T) {
refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_x")
var said []string
if kv := process.OptionalBucket(context.Background(), refusingBus{refusal}, "x", "x is not heard", func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }); kv != nil || len(said) != 1 || !strings.Contains(said[0], "permissions violation") {
t.Errorf("an optional bucket refused: %v, said %v", kv, said)
}
if _, err := process.EssentialBucket(context.Background(), refusingBus{refusal}, "teller", "x", "work"); err == nil || !strings.Contains(err.Error(), "grant of x") {
t.Errorf("an essential bucket refused: %v, want its grant named", err)
}
})
func TestAnAssignersKeyBecomesOneRouterAskAndDeletingItCancelsIt(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)
key, value, ask := holdAskAt(t, 1, 1)
b.assigner[key] = value
b.t.OnAssignerAsk(key)
asked := b.askedOf(oc.AcceptSubjectAsk(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, tellerModule, ga.Ask.ID)
if len(asked) != 1 || asked[0].id != ask.ID {
t.Fatalf("published %+v, want one ask under %s with id %s", b.pubs, tellerModule, ask.ID)
}
var got oc.Ask
if err := json.Unmarshal(asked[0].body, &got); err != nil {
t.Fatal(err)
}
if occ.Digest(got) != occ.Digest(ga.Ask) || got.Headline != ga.Ask.Headline || got.Explanation != ga.Ask.Explanation || got.ID != meshdeliveryclient.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
b.t.OnAssignerAsk(key) // the key written again with the same ask, or heard again at a restart
if n := len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))); n != 1 {
t.Errorf("the ask was published %d times", n)
}
delete(b.gate, key)
b.t.OnAskGone(key)
delete(b.assigner, key)
b.t.OnAssignerAskGone(key)
cancels := b.askedOf(oc.AcceptSubjectCancel(tellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ga.Ask.ID+`"}` {
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ask.ID+`"}` {
t.Fatalf("cancels = %+v", cancels)
}
b.t.OnAskGone(key)
b.t.OnAssignerAskGone(key)
if n := len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))); n != 1 {
t.Errorf("the ask was cancelled %d times", n)
}
// Issue 774 (2026-10-31): two merge asks of pull requests merged by another path held two of the router's three places,
// their keys never taken away, and a fourth ask was refused for want of room. teller_end_ask ends one by hand, named by its
// gate key or its router id; the router's word that it ended asks the one waiting for room again.
// Issue 774: teller_end_ask ends one by hand, named by its key or its router id; the router's word that it ended asks the
// one waiting for room again (the assigner's asks hold at most mostAssignerOpen places, the third waits).
for _, c := range []struct {
name string
by func(key string, ga gateAsked) string
by func(key string, ask oc.Ask) string
}{
{"774: ended by hand, named by its gate key", func(key string, _ gateAsked) string { return key }},
{"774: ended by hand, named by its router id", func(_ string, ga gateAsked) string { return ga.Ask.ID }},
{"774: ended by hand, named by its key", func(key string, _ oc.Ask) string { return key }},
{"774: ended by hand, named by its router id", func(_ string, ask oc.Ask) string { return ask.ID }},
} {
t.Run(c.name, func(t *testing.T) {
b := newBed(t)
k1, first := b.askOf(t, 1, 1)
b.askOf(t, 2, 2)
b.askOf(t, 3, 3)
_, fourth := b.askOf(t, 4, 4)
b.refuse(t, fourth.Ask.ID, roomRefusal)
r := b.t.EndAsk(c.by(k1, first), "its pull request merged without the Approve", "lead@shanks")
k1, first := b.holdOf(t, 1, 1)
b.holdOf(t, 2, 1)
_, third := b.holdOf(t, 3, 1)
r := b.t.EndAsk(c.by(k1, first), "the hold no longer holds", "lead@shanks")
cancels := b.askedOf(oc.AcceptSubjectCancel(tellerModule))
if !r.Ended || r.ID != first.Ask.ID || r.Key != k1 || len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+first.Ask.ID+`"}` {
if !r.Ended || r.ID != first.ID || r.Key != k1 || len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+first.ID+`"}` {
t.Fatalf("end = %+v, cancels %+v", r, cancels)
}
b.deliver(t, oc.Warrant{Ask: first.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeCancelled, At: occ.FormatTime(now), AskDigest: occ.Digest(first.Ask)})
if got := b.askedIDs(); len(got) != 5 || got[4] != fourth.Ask.ID {
t.Errorf("asks = %v, want the refused %s asked again once the ask ended by hand left", got, fourth.Ask.ID)
b.deliver(t, oc.Warrant{Ask: first.ID, Asker: tellerModule, Outcome: occ.OutcomeCancelled, At: occ.FormatTime(now), AskDigest: occ.Digest(first)})
if got := b.askedIDs(); len(got) != 3 || got[2] != third.ID {
t.Errorf("asks = %v, want the waiting %s asked once the ask ended by hand left", got, third.ID)
}
if again := b.t.EndAsk(k1, "again", "lead@shanks"); again.Ended || len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))) != 1 {
t.Errorf("a second end = %+v: an ended ask is left", again)
@@ -369,62 +307,22 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1
}
t.Run("774: an ask waiting for room is dropped from the wait; a name the teller holds no ask under is said", func(t *testing.T) {
b := newBed(t)
_, first := b.askOf(t, 1, 1)
b.askOf(t, 2, 2)
b.askOf(t, 3, 3)
_, fourth := b.askOf(t, 4, 4)
b.refuse(t, fourth.Ask.ID, roomRefusal)
if r := b.t.EndAsk(fourth.Ask.ID, "asked about what no longer holds", "lead@shanks"); !r.Ended || len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))) != 0 {
_, first := b.holdOf(t, 1, 1)
b.holdOf(t, 2, 1)
_, third := b.holdOf(t, 3, 1)
if r := b.t.EndAsk(third.ID, "asked about what no longer holds", "lead@shanks"); !r.Ended || len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))) != 0 {
t.Fatalf("end of a waiting ask = %+v", r)
}
b.deliver(t, warrantFor(first.Ask, "approve"))
if got := b.askedIDs(); len(got) != 4 {
b.deliver(t, warrantFor(first, "keep"))
if got := b.askedIDs(); len(got) != 2 {
t.Errorf("asks = %v: the ask ended while it waited was asked again", got)
}
if r := b.t.EndAsk("merge.nothing.here", "a typo", "lead@shanks"); r.Ended || !strings.Contains(r.Says, "holds no ask") {
if r := b.t.EndAsk("hold.nothing.here", "a typo", "lead@shanks"); r.Ended || !strings.Contains(r.Says, "holds no ask") {
t.Errorf("end of an unknown ask = %+v", r)
}
})
}
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[answerKey(key)]
if !ok || len(b.answers.written) != 1 {
t.Fatalf("answers = %v", b.answers.written)
}
a, err := meshdeliveryclient.DecodeAnswer(raw)
if err != nil || a.Option != "approve" || a.Warrant != warrantID(w) || !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 = occ.FormatTime(now.Add(time.Hour + 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)
}
// design 47 G §431: the settle ask went with the verb settle, so an ask whose options bind it holds no more.
for verb, holds := range map[string]bool{"mesh-delivery.settle": false, "mesh-delivery.merge": true} {
b := newBed(t)
key, value, ga := gateAskOf(t, "g", verb, twoHeads(), twoHeads(), 1)
b.gate[key] = value
b.t.OnAsk(key, 1, value)
b.deliver(t, warrantFor(ga.Ask, "approve"))
if _, written := b.answers.written[answerKey(key)]; written != holds {
t.Errorf("an ask binding %s: answer written %v, want %v", verb, written, holds)
}
}
}
// ADR 0596 §2: the teller says answered after each answer it writes, under one message id per ask, and says it again for an
// answer written whose event was not said.
func TestEveryAnswerWrittenIsSaidAsTheEventAnswered(t *testing.T) {
@@ -438,89 +336,95 @@ func TestEveryAnswerWrittenIsSaidAsTheEventAnswered(t *testing.T) {
said = append(said, answeredID(key, askID)+" "+string(answer))
return nil
}
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)
key, ask := b.holdOf(t, 1, 1)
b.deliver(t, warrantFor(ask, "keep"))
if len(said) != 0 || b.answers.writes != 1 {
t.Fatalf("said %v with the bus down, written %d times", said, b.answers.writes)
}
fail = false
if err := b.t.writeAnswer(key, ga.Ask.ID, b.answers.written[answerKey(key)], "again"); err != nil { // the answer is there; its event was not said
if err := b.t.writeAnswer(key, ask.ID, b.answers.written[answerKey(key)], "again"); err != nil { // the answer is there; its event was not said
t.Fatal(err)
}
if len(said) != 1 || b.answers.writes != 1 {
t.Fatalf("said %v, written %d times; want the event said once and the answer written once", said, b.answers.writes)
}
raw := b.answers.written[answerKey(key)]
if want := answeredID(key, ga.Ask.ID) + " " + string(raw); said[0] != want {
if want := answeredID(key, ask.ID) + " " + string(raw); said[0] != want {
t.Errorf("said %q, want %q", said[0], want)
}
}
func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3
other := []meshdeliveryclient.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, oc.Warrant){
"heads that differ from the ones the option bound": func(t *testing.T) (*bed, string, oc.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, oc.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, oc.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
},
cases := map[string]func(w *oc.Warrant){
"an asker that differs": func(w *oc.Warrant) { w.Asker = "mesh-delivery" },
"a digest that differs": func(w *oc.Warrant) { w.AskDigest = "sha256:0000" },
"an option the ask lacks": func(w *oc.Warrant) { w.Option = "delete" },
}
for name, mk := range cases {
for name, change := range cases {
t.Run(name, func(t *testing.T) {
b, key, w := mk(t)
b := newBed(t)
key, ask := b.holdOf(t, 1, 1)
w := warrantFor(ask, "keep")
change(&w)
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 != occ.StateRaised || b.conds[0].Subject.ID != key ||
!strings.HasPrefix(b.conds[0].Explanation, occ.NeedsYou) {
if len(b.conds) != 1 || b.conds[0].State != occ.StateRaised || b.conds[0].Subject.ID != key || b.conds[0].Subject.Scope != "delivery" ||
!strings.HasPrefix(b.conds[0].Explanation, occ.NeedsYou) || strings.Contains(b.conds[0].Explanation, "merge") {
t.Fatalf("conditions = %+v", b.conds)
}
})
}
}
// design 47 G §653, A-172: the teller carries no merge ask. A router ask it holds under a key that is no assigner key (a merge
// ask kept before the Approve per merge went) answers nothing, and a merge ask waiting for room is dropped from the wait.
func TestAMergeAskOfBeforeTheApprovePerMergeWentAnswersNothingAndWaitsNoMore(t *testing.T) {
b := newBed(t)
_, _, ask := holdAskAt(t, 1, 1)
mergeKey := "merge.g.0123456789ab"
ask.ID = meshdeliveryclient.AskID(mergeKey)
if err := b.t.note(ask.ID, mergeKey, 3, nil); err != nil {
t.Fatal(err)
}
if err := b.client.Ask(ask); err != nil {
t.Fatal(err)
}
if ws, rec, _ := b.t.waiting(); rec != nil || len(ws) != 0 {
t.Fatalf("the wait holds %v", ws)
}
b.refuse(t, ask.ID, roomRefusal) // it waits for room, as a merge ask did
b.deliver(t, warrantFor(ask, "keep"))
if len(b.answers.written) != 0 {
t.Errorf("a merge ask was answered: %v", b.answers.written)
}
b.t.OnCurrent() // the start: the wait is read, and the merge ask is dropped, not asked again
if got := b.askedIDs(); len(got) != 1 {
t.Errorf("asks = %v: a merge ask was asked again", got)
}
if ws, _, _ := b.t.waiting(); len(ws) != 0 {
t.Errorf("the wait still holds %+v", ws)
}
}
func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/AC3 (cleared when a good answer is written)
b := newBed(t)
other := []meshdeliveryclient.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"))
key, bad := b.holdOf(t, 1, 1)
w := warrantFor(bad, "keep")
w.AskDigest = "sha256:0000"
b.deliver(t, w)
if len(b.conds) != 1 || b.conds[0].State != occ.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)
// The assigner writes the ask again at a new asking, under a new id.
_, value, good := holdAskAt(t, 1, 2)
b.assigner[key] = value
b.t.OnAssignerAsk(key)
if n := len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))); n != 2 {
t.Fatalf("the ask was published %d times, want again at the new revision", n)
t.Fatalf("the ask was published %d times, want again at the new asking", n)
}
b.deliver(t, warrantFor(gaGood.Ask, "approve"))
b.deliver(t, warrantFor(good, "keep"))
if _, ok := b.answers.written[answerKey(key)]; !ok {
t.Fatalf("no answer: %v\n%s", b.answers.written, b.logs.String())
}
@@ -531,12 +435,10 @@ func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/
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
key, ask := b.holdOf(t, 1, 1)
w, _ := json.Marshal(warrantFor(ask, "keep"))
b.records[ask.ID] = `{"state":"chosen","warrant":` + string(w) + `}`
b.t.OnAssignerAsk(key) // the teller starts again and reads the key
if _, ok := b.answers.written[answerKey(key)]; !ok {
t.Errorf("no answer: %s", b.logs.String())
}
@@ -544,34 +446,32 @@ func TestAWarrantComingWhileTheTellerWasAwayIsReadFromTheRouter(t *testing.T) {
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 {
key, g1 := b.holdOf(t, 1, 1)
_, two, g2 := holdAskAt(t, 1, 2)
if g1.ID == g2.ID {
t.Fatal("the two askings share an id")
}
b.gate[key] = two
b.t.OnAsk(key, 2, two)
b.assigner[key] = two
b.t.OnAssignerAsk(key)
cancels := b.askedOf(oc.AcceptSubjectCancel(tellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+g1.Ask.ID+`"}` {
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+g1.ID+`"}` {
t.Fatalf("cancels = %+v, want asking 1's router ask", cancels)
}
asked := b.askedOf(oc.AcceptSubjectAsk(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(asked) != 2 || asked[1].id != g2.ID {
t.Fatalf("asks = %+v, want the second under %s", asked, g2.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"))
// The same key again, and a warrant for the first asking (the assigner has moved on), change nothing.
b.t.OnAssignerAsk(key)
b.deliver(t, warrantFor(g1, "keep"))
if len(b.answers.written) != 0 || len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))) != 2 {
t.Errorf("answers %v asks %d", b.answers.written, len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))))
}
// The warrant of the second names it: its answer's warrant is the router id and the time.
w := warrantFor(g2.Ask, "approve")
w := warrantFor(g2, "keep")
b.deliver(t, w)
raw, ok := b.answers.written[answerKey(key)]
a, err := meshdeliveryclient.DecodeAnswer(raw)
@@ -580,65 +480,20 @@ func TestARewriteWithAskingTwoCancelsAskingOneAndAsksUnderTheNewID(t *testing.T)
}
}
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 := gateValue(changed.Group, changed.Heads, changed.Asking, rawAsk(t, changed.Ask))
if err != nil {
t.Fatal(err)
}
b.t.OnAsk(key, 2, v2)
if n := len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))); n != 1 {
t.Errorf("cancels = %d", n)
}
if n := len(b.askedOf(oc.AcceptSubjectAsk(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 := []meshdeliveryclient.Head{{Repository: "novox/Mesh-Delivery", Number: 12, Head: headA}}
key, err := meshdeliveryclient.MergeAskKey("g", heads)
if err != nil {
t.Fatal(err)
}
id := meshdeliveryclient.AskID(key)
// The act as the gate binds it: its IDOf lowercases the repository (mesh-catalog modules/mesh-delivery-gate model.go).
binds, err := occ.ActDigest(occ.Act{"verb": "mesh-delivery.merge", "ask": id, "option": "approve", "heads": "novox/mesh-delivery@" + headA[:12] + "=" + headA})
if err != nil {
t.Fatal(err)
}
a := oc.Ask{ID: id, Headline: "Merge into main: Mesh-Delivery#12", Explanation: "Merges.", Who: occ.Operator, Expires: occ.FormatTime(now.Add(occ.ApproveLasts - 10*time.Minute)),
OnExpiry: "the merge stays held", Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: binds}}}
value := encodeGate(t, gateAsked{Group: "g", Heads: heads, Asking: 0, Ask: a})
b.gate[key] = value
b.t.OnAsk(key, 1, value)
b.deliver(t, warrantFor(a, "approve"))
if _, ok := b.answers.written[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)
key, _, _ := holdAskAt(t, 1, 1)
b.assigner[key] = []byte(`{"v":1`)
b.t.OnAssignerAsk(key)
if len(b.pubs) != 0 || !strings.Contains(b.logs.String(), "the ask is not raised") {
t.Errorf("published %+v; log:\n%s", b.pubs, b.logs.String())
}
}
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 := oc.Warrant{Ask: ga.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeRefused, Words: "no channel can send", At: occ.FormatTime(now)}
b.deliver(t, w)
_, ask := b.holdOf(t, 1, 1)
b.deliver(t, oc.Warrant{Ask: ask.ID, Asker: tellerModule, Outcome: occ.OutcomeRefused, Words: "no channel can send", At: occ.FormatTime(now)})
if len(b.conds) != 1 || b.conds[0].State != occ.StateRaised || !strings.Contains(b.conds[0].Summary, "no channel can send") {
t.Errorf("conditions = %+v", b.conds)
}
@@ -646,26 +501,23 @@ func TestARefusedAskIsSaidAsACondition(t *testing.T) { // ADR 0437 §9
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"))
key, g1 := b.holdOf(t, 1, 1)
b.deliver(t, warrantFor(g1, "keep"))
first, ok := b.answers.written[answerKey(key)]
if a, err := meshdeliveryclient.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.Ask.ID+"@") {
if a, err := meshdeliveryclient.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.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)
// The assigner writes the key again at asking 2. Asking 1's answer stays in the bucket.
_, two, g2 := holdAskAt(t, 1, 2)
b.assigner[key] = two
b.t.OnAssignerAsk(key)
asked := b.askedOf(oc.AcceptSubjectAsk(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)
if len(asked) != 2 || asked[1].id != g2.ID {
t.Fatalf("asks = %+v, want asking 2 asked under %s", asked, g2.ID)
}
w := warrantFor(g2.Ask, "approve")
w := warrantFor(g2, "release")
b.deliver(t, w)
raw := b.answers.written[answerKey(key)]
a, err := meshdeliveryclient.DecodeAnswer(raw)
a, err := meshdeliveryclient.DecodeAnswer(b.answers.written[answerKey(key)])
if err != nil || a.Warrant != warrantID(w) {
t.Fatalf("the key holds %+v (%v), want asking 2's answer", a, err)
}
@@ -673,27 +525,8 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R
w2 := w
w2.At = occ.FormatTime(now.Add(time.Hour + time.Minute))
b.deliver(t, w2)
b.t.OnAsk(key, 2, two)
b.t.OnAsk(key, 3, two)
b.t.OnAssignerAsk(key)
if b.answers.writes != 2 || len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))) != 2 || len(b.dropped) != 0 {
t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))), b.dropped)
}
}
// refusingBus refuses every bucket with err (issue 1945).
type refusingBus struct{ err error }
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
// gateValue is the gate's ask value for the group, heads and asking, carrying the router's ask as its JSON.
func gateValue(group string, heads []meshdeliveryclient.Head, asking int, ask json.RawMessage) ([]byte, error) {
var asked any
if err := json.Unmarshal(ask, &asked); err != nil {
return nil, err
}
var items []meshdelivery.AsksByMergeGroupHeadsValueHeadsItem
for _, h := range heads {
items = append(items, meshdelivery.AsksByMergeGroupHeadsValueHeadsItem{Repository: h.Repository, Number: int64(h.Number), Head: h.Head})
}
return meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: group, Heads: items, Asking: int64(asking), Ask: asked})
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers.go
+5 -4
View File
@@ -12,11 +12,12 @@ import (
// The view's answers (novox/hq ADR 0528 clause 6, ADR 0437 §10, issue 1297 B10). The operator may answer, in the view, the
// asks of the assigner that need no proof of who answered (the hold of a fingerprint and the restore of a data module set back,
// mesh-sdk go/delivery AskNeedsProof); the view writes its answer under `<ask key>.view` in its bucket answers, and the teller,
// which alone fills an ask's answer, takes it from there as it takes a warrant from the phone: under the ask's own key, once,
// by compare-and-set. The merge's ask needs proof (ADR 0421 §1, issue 953) and is never answered from a view key.
// mesh-sdk go/delivery AskNeedsProof); the view writes its answer under `<ask key>.view` in its state answers, and the teller,
// which alone fills an ask's answer, takes it from the picture it follows of that state (the view's views family, first view
// by view_read_answers; design 47 G A-175) as it takes a warrant from the phone: under the ask's own key, once, by
// compare-and-set.
// OnViewAnswer handles a key put in the view's bucket answers. A key that is no view answer key, or is the answer to an ask that
// OnViewAnswer handles a key put in the followed picture of the view's answers. A key that is no view answer key, or is the answer to an ask that
// needs proof of who answered, is left and said.
func (t *Teller) OnViewAnswer(key string) {
t.mu.Lock()
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go
+6 -9
View File
@@ -47,12 +47,8 @@ func viewAnswer(t *testing.T, option, words string) []byte {
return v
}
// withView gives the bed's teller the assigner's asks and the view's answers, both empty.
func withView(b *bed) (assigner, view memGate) {
assigner, view = memGate{}, memGate{}
b.t.Assigner, b.t.View = assigner, view
return assigner, view
}
// withView is the bed's pictures of the assigner's asks and the view's answers, both empty.
func withView(b *bed) (assigner, view memGate) { return b.assigner, b.view }
func TestAViewAnswerFillsTheAskOnce(t *testing.T) { // 1297/B10 AC
b := newBed(t)
@@ -96,14 +92,15 @@ func TestAViewAnswerHeardBeforeItsAskFillsItWhenTheAskComes(t *testing.T) { // 1
func TestAnAskNeedingProofIsNotAnsweredFromAViewKey(t *testing.T) { // 1297/B10 AC
b := newBed(t)
assigner, view := withView(b)
key, value, _ := gateAsk(t, twoHeads(), twoHeads())
// Even with the merge's ask under the assigner's bucket and a view answer under its key, nothing is written.
// A key of a kind of ask that needs proof of who answered (the merge's, issue 953) with a view answer under it: nothing is written.
key := "merge.g.0123456789ab"
_, value, _ := holdAsk(t)
assigner[key] = value
view[key+".view"] = viewAnswer(t, "approve", "")
b.t.OnViewAnswer(key + ".view")
b.t.OnAssignerAsk(key)
if b.answers.writes != 0 || len(b.answers.written) != 0 {
t.Fatalf("the merge's ask was answered from the view: %v", b.answers.written)
t.Fatalf("an ask needing proof was answered from the view: %v", b.answers.written)
}
if !strings.Contains(b.logs.String(), "issue 953") {
t.Errorf("the refusal does not say why:\n%s", b.logs.String())
modules/mesh-delivery-teller/manifest_test.go
+11 -15
View File
@@ -92,28 +92,24 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
t.Errorf("module = %q", m.Module)
}
// answers is the teller's answer to each ask; asked is what it asked of the router, which condition it raised and which
// gate key and revision each ask was made at (issue 1343); ends is the ends asked by hand with teller_end_ask (issue 774).
// ask key and revision each ask was made at (issue 1343); ends is the ends asked by hand with teller_end_ask (issue 774).
// first-views is where its first view of the assigner's asks stands and findings its condition past the bound (ADR 0615).
if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}, {Name: "first-views", History: 1}, {Name: "widens-said", History: 1}}; !reflect.DeepEqual(m.State, want) {
if want := []state{{Name: "answers", History: 5}, {Name: "asked", History: 1}, {Name: "ends", History: 1}, {Name: "first-views", History: 1}, {Name: "widens-said", History: 1}, {Name: "teller-marks", History: 1}}; !reflect.DeepEqual(m.State, want) {
t.Errorf("state = %+v, want %+v", m.State, want)
}
// The view's answers to the assigner's asks are read so the view's answer fills an ask that needs no proof (1297/B10); the
// assigner's asks are its state, asked by its read tool and followed by its views family, never read (ADR 0615 §1, DLV-M3).
if want := []string{"mesh-delivery-gate.asks", "mesh-delivery-view.answers"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
}
for _, r := range m.Reads {
if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") {
t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r)
}
// design 47 G §653, A-172: the teller reads no other module's state: the assigner's asks and the view's answers are the
// owners' states, asked by their read tools and followed by their views families (ADR 0615 §1, DLV-M3; A-175).
if len(m.Reads) != 0 {
t.Errorf("reads = %v, want none", m.Reads)
}
// Its own states answers and teller-marks are said as its views family views.answers, read by teller_read_answers (A-175, A-178).
// The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools).
if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", "seat:operator-channel.notify"}; !reflect.DeepEqual(m.Invokes, want) {
if want := []string{"mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-view.view_read_answers", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.registrations", "seat:operator-channel.notify"}; !reflect.DeepEqual(m.Invokes, want) {
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// design 47 G §335 (TL1): the widening of a pipeline end is told through the seat operator-channel and kept in widens-said.
// design 47 G §387: the keeper's ask for the findings is heard, and the finding stream is declared (§383).
if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended", "mesh-delivery.recovery-merged", "mesh-delivery.record-opened"}; !reflect.DeepEqual(m.Consumes, want) {
if want := []string{"mesh-conditions.findings-asked", "mesh-delivery-assigner.views.asks", "mesh-delivery-view.views.answers", "mesh-catalog.registration-changed", "mesh-pipeline-runner.pipeline-ended", "mesh-delivery.recovery-merged", "mesh-delivery.record-opened"}; !reflect.DeepEqual(m.Consumes, want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
if want := []string{"operator-channel"}; !reflect.DeepEqual(sorted(m.Uses), want) {
@@ -126,12 +122,12 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1
t.Errorf("secrets-owner = %q", m.SecretsOwner)
}
// 774: teller_end_ask, served by its tools bundle through the state ends; the process serves none.
if want := []string{"teller_end_ask", "teller_answers", "teller_first_views", "teller_first_view_again", "teller_widens_said"}; !reflect.DeepEqual(m.Tools, want) {
if want := []string{"teller_end_ask", "teller_answers", "teller_read_answers", "teller_first_views", "teller_first_view_again", "teller_widens_said"}; !reflect.DeepEqual(m.Tools, want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Its two events: each answer it writes (ADR 0596 §2), and an assigner ask ended at the channel without an answer, heard by
// the assigner (issue 1966, tellerevents).
if want := []string{"answered", tellerevents.AskEnded}; !reflect.DeepEqual(m.Emits, want) {
if want := []string{"answered", tellerevents.AskEnded, "views.answers"}; !reflect.DeepEqual(m.Emits, want) {
t.Errorf("emits = %v, want %v", m.Emits, want)
}
// design 47 G §383, §387: the stream <module>-findings, 7 days and 16 MiB; §368: version is MAJOR.MINOR.
modules/mesh-delivery-teller/module.json
+9 -5
View File
@@ -5,12 +5,14 @@
"tools": [
"teller_end_ask",
"teller_answers",
"teller_read_answers",
"teller_first_views",
"teller_first_view_again",
"teller_widens_said"
],
"invokes": [
"mesh-delivery-assigner.assigner_read_asks",
"mesh-delivery-view.view_read_answers",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-catalog.registrations",
"seat:operator-channel.notify"
@@ -18,6 +20,7 @@
"consumes": [
"mesh-conditions.findings-asked",
"mesh-delivery-assigner.views.asks",
"mesh-delivery-view.views.answers",
"mesh-catalog.registration-changed",
"mesh-pipeline-runner.pipeline-ended",
"mesh-delivery.recovery-merged",
@@ -25,7 +28,8 @@
],
"emits": [
"answered",
"ask-ended"
"ask-ended",
"views.answers"
],
"state": [
{
@@ -47,12 +51,12 @@
{
"name": "widens-said",
"history": 1
},
{
"name": "teller-marks",
"history": 1
}
],
"reads": [
"mesh-delivery-gate.asks",
"mesh-delivery-view.answers"
],
"uses": [
"operator-channel"
],
modules/mesh-delivery-view/viewstate/reach.go
+5 -1
View File
@@ -136,8 +136,12 @@ func SameReach(a, b Reach) bool {
// ViewAnswersBucket is the delivery view's state `answers`: one key per ask it answered, and ViewAnswersHistory the revisions it
// keeps per key.
const (
ViewAnswersBucket = ViewModule + "_answers"
ViewAnswersState = "answers"
ViewAnswersBucket = ViewModule + "_" + ViewAnswersState
ViewAnswersHistory = 5
// ViewAnswersReadTool is the view's read tool of its state answers (mesh-sdk views.ReadTool), whose changes it says on its
// views family as views.answers; its marks are its state view-marks (views.MarksState) (design 47 G A-178).
ViewAnswersReadTool = "view_read_answers"
)
// ViewAnswer is the value of the view's answer to an ask of the assigner: the asking it answers, the option chosen and the words