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

Teller: the gate's asks and answers are the delivery client's and the seat's types, the assigner's asks its asks package's, the view's answers the view's; the holders and buckets it names are its own constants checked against its manifest (D-11, D-17; ADR 0634 §1, §2, §5; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 16:06:00 +01:00
parent bd6855e1b7
commit e000ed0f66
13 changed files with 273 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go
+9 -8
View File
@@ -10,7 +10,8 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
)
// The assigner's asks reach the operator (issue 1906). The assigner writes its asks (the hold of a fingerprint, the restore
@@ -89,8 +90,8 @@ 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.
func isAssignerKey(key string) bool {
k, err := delivery.ParseAnyAskKey(key)
return err == nil && k.Kind != delivery.AskKindMerge
k, err := asks.ParseAnyAskKey(key)
return err == nil && k.Kind != meshdeliveryclient.AskKindMerge
}
// assignerAsk reads the assigner's ask at key: found false when the key holds none.
@@ -115,14 +116,14 @@ type assignerAsked struct {
Ask oc.Ask
}
// decodeAssignerAsk reads the assigner's ask at key as delivery.DecodeAssignerAsk did before the ask was JSON: it refuses an
// decodeAssignerAsk reads the assigner's ask at key as asks.DecodeAsk did before the ask was JSON: it refuses an
// id other than the one the key and the asking give, and an ask with no option (occ.Carried).
func decodeAssignerAsk(key string, data []byte) (assignerAsked, error) {
a, err := delivery.DecodeAssignerAsk(key, data)
a, err := asks.DecodeAsk(key, data)
if err != nil {
return assignerAsked{}, err
}
ask, err := occ.Carried(a.Ask, delivery.AskingID(key, a.Asking))
ask, err := occ.Carried(a.Ask, meshdeliveryclient.AskingID(key, a.Asking))
if err != nil {
return assignerAsked{}, err
}
@@ -142,7 +143,7 @@ func (t *Teller) raiseAssignerAsk(key string) {
return
}
id := a.Ask.ID
if raw, _, has, err := t.Answers.Get(delivery.AnswerKey(key)); err != nil {
if raw, _, has, err := t.Answers.Get(answerKey(key)); err != nil {
t.logf("%s: the answers could not be read, so the ask is not raised: %v", key, err)
return
} else if has && answeredAsking(raw) == id {
@@ -227,7 +228,7 @@ func (t *Teller) answerAssigner(key string, o oc.Option, w oc.Warrant) error {
if occ.Digest(a.Ask) != w.AskDigest {
return fmt.Errorf("the assigner's ask %s is no longer the ask the operator answered (its digest is %s, the warrant's %s)", key, occ.Digest(a.Ask), w.AskDigest)
}
body, err := delivery.EncodeAnswer(delivery.Answer{Option: o.ID, Warrant: warrantID(w), Words: occ.Says(w)})
body, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: o.ID, Warrant: warrantID(w), Words: occ.Says(w)})
if err != nil {
return err
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go
+11 -10
View File
@@ -7,7 +7,8 @@ import (
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
)
// 1906/H1: given the assigner's hold ask in its bucket and no answer, when the teller hears the key, then it raises the ask on
@@ -19,7 +20,7 @@ func TestAnAssignerAskIsRaisedOnTheOperatorChannelOnce(t *testing.T) {
assigner[key] = value
b.t.OnAssignerAsk(key)
b.t.OnAssignerAsk(key)
asked := b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))
asked := b.askedOf(oc.AcceptSubjectAsk(tellerModule))
if len(asked) != 1 {
t.Fatalf("published %d asks, want one; log:\n%s", len(asked), b.logs.String())
}
@@ -40,7 +41,7 @@ func TestAnAssignerAskIsRaisedWithoutTheView(t *testing.T) {
key, value, _ := holdAsk(t)
assigner[key] = value
b.t.OnAssignerAsk(key)
if n := len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))); n != 1 {
if n := len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))); n != 1 {
t.Fatalf("published %d asks, want one; log:\n%s", n, b.logs.String())
}
}
@@ -55,11 +56,11 @@ func TestTheOperatorsAnswerToAnAssignerAskIsWrittenUnderItsKey(t *testing.T) {
b.t.OnAssignerAsk(key)
w := ackFor(ask, "release")
b.deliver(t, w)
raw, ok := b.answers.written[delivery.AnswerKey(key)]
raw, ok := b.answers.written[answerKey(key)]
if !ok {
t.Fatalf("no answer under %s; log:\n%s", key, b.logs.String())
}
if a, err := delivery.DecodeAnswer(raw); err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID {
if a, err := meshdeliveryclient.DecodeAnswer(raw); err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID {
t.Errorf("answer = %+v, %v", a, err)
}
b.deliver(t, w)
@@ -86,7 +87,7 @@ func TestAnAnswerToAnAssignerAskTakenAwayWritesNothing(t *testing.T) {
// itself) is taken back too; an ask answered before it is heard is never raised.
func TestAnAssignerAskAnsweredInTheViewOrTakenAwayIsTakenBack(t *testing.T) {
cancelled := func(b *bed, id string) bool {
for _, p := range b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule)) {
for _, p := range b.askedOf(oc.AcceptSubjectCancel(tellerModule)) {
if string(p.body) == `{"id":"`+id+`"}` {
return true
}
@@ -119,7 +120,7 @@ func TestAnAssignerAskAnsweredInTheViewOrTakenAwayIsTakenBack(t *testing.T) {
assigner[key] = value
view[key+".view"] = viewAnswer(t, "keep", "")
b.t.OnAssignerAsk(key)
if n := len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))); n != 0 {
if n := len(b.askedOf(oc.AcceptSubjectAsk(tellerModule))); n != 0 {
t.Errorf("an ask answered in the view was raised %d times", n)
}
}
@@ -162,13 +163,13 @@ func ackFor(a oc.Ask, option string) oc.Warrant {
// 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 := delivery.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64))
key, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64))
if err != nil {
t.Fatal(err)
}
_, _, ask := holdAsk(t)
ask.ID = delivery.AskID(key)
value, err := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: rawAsk(t, ask)})
ask.ID = meshdeliveryclient.AskID(key)
value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask)})
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go
+24 -23
View File
@@ -20,7 +20,8 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-teller/internal/ends"
@@ -189,7 +190,7 @@ func fullConf(dir string) string {
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(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...)
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)...)
assigner := keepsOf(assignerAsks)
tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...)
@@ -256,7 +257,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: delivery.AsksHistory, tellerAnswers: delivery.AnswersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: delivery.AssignerAsksHistory, viewAnswers: delivery.ViewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} {
for name, history := range map[string]uint8{gateAsksKV: gateAsksHistory, tellerAnswers: answersHistory, tellerAsked: 1, ends.Bucket: 1, messengerAsks: 1, assignerAsks: assignerAsksHistory, viewAnswers: viewAnswersHistory, firstViewsBucket: 1, findingsBucket: 1} {
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: history}); err != nil {
t.Fatal(err)
}
@@ -273,7 +274,7 @@ func startFull(t *testing.T) *fullBed {
t.Fatal(err)
}
// The assigner as the owner of its asks: its read tool and its views family (ADR 0615).
viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", tellerConsumer)
viewstest.Serve(t, ctx, js, ctl, assignerModule, "asks", assignerModule+".assigner_read_asks", tellerConsumer)
return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx}
}
@@ -423,18 +424,18 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
}
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, delivery.AnswerKey(key)); err == nil {
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 := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err != nil || a.Option != "approve" {
t.Errorf("answer %+v err %v", a, err)
}
// A second ask, withdrawn by the gate, is cancelled at the router.
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
other := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
key2, value2, ga2 := gateAsk(t, other, other)
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
t.Fatal(err)
@@ -512,20 +513,20 @@ func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
t.Fatal(err)
}
first := lastAsk(held.Ask.ID, 0)
say(oc.Warrant{Ask: held.Ask.ID, Asker: delivery.TellerModule, Outcome: occ.OutcomeRefused, Words: roomRefusal, At: occ.FormatTime(now)})
say(oc.Warrant{Ask: held.Ask.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 := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
other := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 13, Head: headB}}
key2, value2, ga2 := gateAsk(t, other, other)
if _, err := kv.Put(b.ctx, key2, value2); err != nil {
t.Fatal(err)
}
second := lastAsk(ga2.Ask.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: delivery.TellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga2.Ask)})
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)
}
@@ -569,14 +570,14 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
}
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, delivery.AnswerKey(key)); err == nil {
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 := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "release" || answeredAsking(e.Value()) != ask.ID {
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)
}
// The view's answer written again with the same option changes nothing under the ask (another option, later, is the
@@ -585,7 +586,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
t.Fatal(err)
}
time.Sleep(time.Second)
if again, err := answers.Get(b.ctx, delivery.AnswerKey(key)); err != nil || again.Revision() != e.Revision() {
if again, err := answers.Get(b.ctx, answerKey(key)); err != nil || again.Revision() != e.Revision() {
t.Errorf("the answer was written a second time: %v", err)
}
// An answer under the merge's key fills nothing.
@@ -597,7 +598,7 @@ func TestTellerFillsAnAskFromTheViewsAnswerOnce(t *testing.T) { // 1297/B10 AC
t.Fatal(err)
}
time.Sleep(time.Second)
if _, err := answers.Get(b.ctx, delivery.AnswerKey(mkey)); err == nil {
if _, err := answers.Get(b.ctx, answerKey(mkey)); err == nil {
t.Error("the merge's ask was answered from a view key")
}
}
@@ -656,24 +657,24 @@ func TestTellerRaisesAnAssignerAskAndWritesTheOperatorsAnswer(t *testing.T) {
}
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, delivery.AnswerKey(key)); err == nil {
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 := delivery.DecodeAnswer(e.Value()); err != nil || a.Option != "release" || answeredAsking(e.Value()) != ask.ID {
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 := delivery.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64))
key2, err := asks.HoldAskKey("plex", "src2:"+strings.Repeat("3", 64))
if err != nil {
t.Fatal(err)
}
ask2 := ask
ask2.ID = delivery.AskID(key2)
value2, err := delivery.EncodeAssignerAsk(key2, delivery.AssignerAsk{Ask: rawAsk(t, ask2)})
ask2.ID = meshdeliveryclient.AskID(key2)
value2, err := asks.EncodeAsk(key2, asks.Ask{Ask: rawAsk(t, ask2)})
if err != nil {
t.Fatal(err)
}
@@ -714,8 +715,8 @@ func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
if err != nil {
t.Fatal(err)
}
kept, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: ask.ID + "@" + time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator chose keep"})
if _, err := answers.Put(b.ctx, delivery.AnswerKey(key), kept); err != nil {
kept, _ := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: "keep", Warrant: ask.ID + "@" + time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator chose keep"})
if _, err := answers.Put(b.ctx, answerKey(key), kept); err != nil {
t.Fatal(err)
}
askKV, err := assigner.KeyValue(b.ctx, assignerAsks)
@@ -738,8 +739,8 @@ func TestTellerTakesAChangedHoldAnswerFromTheView(t *testing.T) {
t.Fatal(err)
}
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
if e, err := answers.Get(b.ctx, delivery.AnswerKey(key)); err == nil {
if a, err := delivery.DecodeAnswer(e.Value()); err == nil && a.Option == "release" && answeredAsking(e.Value()) == ask.ID {
if e, err := answers.Get(b.ctx, answerKey(key)); err == nil {
if a, err := meshdeliveryclient.DecodeAnswer(e.Value()); err == nil && a.Option == "release" && answeredAsking(e.Value()) == ask.ID {
return
}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+13 -15
View File
@@ -25,8 +25,6 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-teller/internal/ends"
)
@@ -51,7 +49,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
defer conn.Close()
gate, err := js.KeyValue(ctx, delivery.AsksBucket)
gate, err := js.KeyValue(ctx, gateAsksBucket)
if errors.Is(err, jetstream.ErrBucketNotFound) {
logger.Print(waiting)
<-ctx.Done()
@@ -59,11 +57,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
if err != nil {
// Essential: the gate's asks are what the teller tells (issue 1945: the error names the grant).
return process.EssentialErr("teller", delivery.AsksBucket, "tell any of the gate's asks", err)
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 [2]jetstream.KeyValue
for i, name := range []string{delivery.AnswersBucket, askedBucket} {
for i, name := range []string{answersBucket, askedBucket} {
if kvs[i], err = process.EssentialBucket(ctx, js, "teller", name, "write an answer or remember what it asked"); err != nil {
return err
}
@@ -78,10 +76,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// started once both are registered.
hub := process.NewEventHub(logger.Printf)
defer hub.Stop()
client := &occ.Client{Module: delivery.TellerModule, State: state, Publish: publish,
Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(hub.Handle, logger.Printf)}
client := &occ.Client{Module: tellerModule, State: state, Publish: publish,
Record: recordVia(ctx, conn, tellerModule), 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},
Say: sayVia(ctx, js, delivery.TellerModule), Logf: logger.Printf, Publish: publish}
Say: sayVia(ctx, js, tellerModule), Logf: logger.Printf, Publish: publish}
// Hear the warrants first, so none that comes while the asks are read is missed.
if err := client.Listen(t.Answer, t.Said); err != nil {
return fmt.Errorf("hearing the router's warrants: %w", err)
@@ -96,10 +94,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return fmt.Errorf("the teller's connection for its tool calls: %w", err)
}
defer calls.Close()
reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub}
reader := &views.ProcessReader{Module: tellerModule, Call: calls.Call, Logf: logger.Printf, Hub: hub}
findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf)
reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) }
asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks")
asks, err := reader.Follow(assignerModule, assignerAsksState, assignerModule+".assigner_read_asks")
if err != nil {
return err
}
@@ -109,7 +107,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
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, delivery.ViewAnswersBucket, teller, logger.Printf)
viewKV := process.OptionalBucket(ctx, js, viewAnswersBucket, teller, logger.Printf)
// 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}
@@ -153,16 +151,16 @@ const teller = "the asks and answers it holds are not carried until the teller i
// 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 = delivery.TellerModule + "_asked"
const askedBucket = tellerModule + "_asked"
// The teller's own states of its first view of the assigner's asks (views.StatusState) and of its findings, where the
// condition keeper raises a first view unanswered past its bound (ADR 0615 §6).
var (
firstViewsBucket = delivery.TellerModule + "_" + views.StatusState
findingsBucket = delivery.TellerModule + "_findings"
firstViewsBucket = tellerModule + "_" + views.StatusState
findingsBucket = tellerModule + "_findings"
)
// assignerAsksState is the assigner's state of its asks (delivery.AssignerAsksBucket).
// assignerAsksState is the assigner's state of its asks (assignerAsksBucket).
const assignerAsksState = "asks"
// waiting is said once at start.
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names.go
+29
View File
@@ -0,0 +1,29 @@
package main
// 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).
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"
// 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
)
// answerKey is the key, in the teller's answers, of its answer to the ask at askKey: the same string.
func answerKey(askKey string) string { return askKey }
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/names_test.go
+42
View File
@@ -0,0 +1,42 @@
package main
import (
"encoding/json"
"os"
"slices"
"testing"
)
// 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).
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 {
Name string `json:"name"`
} `json:"state"`
}
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
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)
}
}
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)
}
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go
+9 -9
View File
@@ -7,7 +7,7 @@ import (
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/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
@@ -17,8 +17,8 @@ import (
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) []delivery.Head {
return []delivery.Head{{Repository: "novox/mesh-delivery", Number: 10 + n, Head: strings.Repeat(string(rune('a'+n)), 40)}}
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.
@@ -36,19 +36,19 @@ func (b *bed) refuse(t *testing.T, id, words string) {
t.Helper()
delete(b.seenIDs, id)
delete(b.records, id)
b.deliver(t, oc.Warrant{Ask: id, Asker: delivery.TellerModule, Outcome: occ.OutcomeRefused, Words: words, At: occ.FormatTime(now)})
b.deliver(t, oc.Warrant{Ask: id, Asker: tellerModule, Outcome: occ.OutcomeRefused, Words: words, At: occ.FormatTime(now)})
}
// 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) {
t.Helper()
b.deliver(t, oc.Warrant{Ask: ga.Ask.ID, Asker: delivery.TellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga.Ask)})
b.deliver(t, oc.Warrant{Ask: ga.Ask.ID, Asker: tellerModule, Outcome: occ.OutcomeExpired, At: occ.FormatTime(now), AskDigest: occ.Digest(ga.Ask)})
}
// askedIDs is the router ids of every ask the bus carried, in order.
func (b *bed) askedIDs() []string {
var out []string
for _, p := range b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule)) {
for _, p := range b.askedOf(oc.AcceptSubjectAsk(tellerModule)) {
var a struct {
ID string `json:"id"`
}
@@ -78,7 +78,7 @@ func TestAnAskRefusedForWantOfRoomIsAskedAgainWhenAnotherEnds(t *testing.T) { //
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)
}
pubs := b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))
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)
}
@@ -86,9 +86,9 @@ func TestAnAskRefusedForWantOfRoomIsAskedAgainWhenAnotherEnds(t *testing.T) { //
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, _ := delivery.MergeAskKey("g", headsN(4))
key4, _ := meshdeliveryclient.MergeAskKey("g", headsN(4))
b.deliver(t, warrantFor(fourth.Ask, "approve"))
if _, ok := b.answers.written[delivery.AnswerKey(key4)]; !ok {
if _, ok := b.answers.written[answerKey(key4)]; !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.
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go
+25 -20
View File
@@ -13,8 +13,8 @@ import (
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"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The teller carries the gate's merge asks to the operator and writes the operator's answer back (novox/hq ADR 0437
@@ -24,30 +24,35 @@ import (
// 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.
// gateAsked is the gate's ask with the router's ask decoded from its JSON (delivery.Ask.Ask is operator-channel's ask as
// 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 []delivery.Head
Heads []meshdeliveryclient.Head
Asking int
Ask oc.Ask
}
// decodeGateAsk reads the gate's ask value and the router's ask in it, as delivery.DecodeAsk did before the ask was JSON.
// 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 := delivery.DecodeAsk(value)
ga, err := meshdeliveryclient.DecodeAsk(value)
if err != nil {
return gateAsked{}, err
}
key, err := delivery.MergeAskKey(ga.Group, ga.Heads)
heads := meshdeliveryclient.HeadsOf(ga.Heads)
key, err := meshdeliveryclient.MergeAskKey(ga.Group, heads)
if err != nil {
return gateAsked{}, err
}
a, err := occ.Carried(ga.Ask, delivery.AskingID(key, ga.Asking))
raw, err := json.Marshal(ga.Ask)
if err != nil {
return gateAsked{}, err
}
return gateAsked{Group: ga.Group, Heads: ga.Heads, Asking: ga.Asking, Ask: a}, nil
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
@@ -189,7 +194,7 @@ func (t *Teller) OnAsk(key string, rev uint64, value []byte) {
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 := delivery.MergeAskKey(ga.Group, ga.Heads); err != nil || want != key {
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
}
@@ -277,7 +282,7 @@ func (t *Teller) cancelOpen(key, id string) {
func (t *Teller) OnAskGone(key string) {
t.mu.Lock()
defer t.mu.Unlock()
if _, _, _, err := delivery.ParseAskKey(key); err != nil {
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
}
@@ -313,7 +318,7 @@ 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 (delivery.AskVerb: mesh-delivery.merge, or
// mergeAct is what the approve option of the gate's ask binds: the ask's verb (meshdeliveryclient.AskVerb: mesh-delivery.merge, or
// mesh-delivery.settle for a person's settle ask, novox/hq ADR 0556 clause 5), 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).
@@ -329,7 +334,7 @@ func mergeAct(verb, ask, option string, heads map[string]string) occ.Act {
// 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 []delivery.Head) map[string]string {
func headsOf(heads []meshdeliveryclient.Head) map[string]string {
out := map[string]string{}
for _, h := range heads {
short := h.Head
@@ -344,7 +349,7 @@ func headsOf(heads []delivery.Head) map[string]string {
// 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 {
a, err := delivery.DecodeAnswer(raw)
a, err := meshdeliveryclient.DecodeAnswer(raw)
if err != nil {
return ""
}
@@ -395,10 +400,10 @@ func (t *Teller) Answer(o oc.Option, w oc.Warrant) error {
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(delivery.AskVerb(ga.Group), w.Ask, o.ID, headsOf(ga.Heads))); err != nil {
if err := occ.Performs(o, mergeAct(meshdeliveryclient.AskVerb(ga.Group), w.Ask, o.ID, headsOf(ga.Heads))); err != nil {
return err
}
body, err := delivery.EncodeAnswer(delivery.Answer{Option: o.ID, Warrant: warrantID(w), Words: occ.Says(w)})
body, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: o.ID, Warrant: warrantID(w), Words: occ.Says(w)})
if err != nil {
return err
}
@@ -409,7 +414,7 @@ func (t *Teller) Answer(o oc.Option, w oc.Warrant) error {
// updated at the revision read when it holds an earlier asking's answer (the reader ignores that one); left when it holds this
// asking's.
func (t *Teller) writeAnswer(key, askID string, body []byte, words string) error {
akey := delivery.AnswerKey(key)
akey := answerKey(key)
cur, crev, found, err := t.Answers.Get(akey)
if err != nil {
return fmt.Errorf("the answer under %s could not be read: %w", key, err)
@@ -484,7 +489,7 @@ func oneLine(err error) string {
}
// conditionKey is the key of the condition about an ask.
func conditionKey(id string) string { return delivery.TellerModule + ".merge-ask." + id }
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.
type askWords struct {
@@ -791,7 +796,7 @@ func (t *Teller) stillAsked(x waitingAsk) (oc.Ask, string) {
if ga.Ask.ID != x.ID {
return oc.Ask{}, "the gate asks again under " + ga.Ask.ID
}
if cur, _, has, err := t.Answers.Get(delivery.AnswerKey(x.Key)); err == nil && has && answeredAsking(cur) == x.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 {
@@ -811,7 +816,7 @@ func (t *Teller) stillAskedByAssigner(x waitingAsk) (oc.Ask, string) {
case a.Ask.ID != x.ID:
return oc.Ask{}, "the assigner asks again under " + a.Ask.ID
}
if cur, _, has, err := t.Answers.Get(delivery.AnswerKey(x.Key)); err == nil && has && answeredAsking(cur) == x.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(a.Ask, t.now()); err != nil {
@@ -832,7 +837,7 @@ func (t *Teller) askAgain(a oc.Ask, msgID string) error {
if _, err := t.Store.Create(keptKey(a.ID), occ.Kept{Ask: a, Asked: t.now()}); err != nil {
return err
}
if err := t.Publish(oc.AcceptSubjectAsk(delivery.TellerModule), a, msgID); err != nil {
if err := t.Publish(oc.AcceptSubjectAsk(tellerModule), a, msgID); err != nil {
_ = t.Store.Delete(keptKey(a.ID))
return err
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
+70 -56
View File
@@ -14,8 +14,9 @@ import (
"git.novox.be/novox/mesh-sdk/go/process"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
meshdelivery "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
const (
@@ -137,7 +138,7 @@ func newBed(t *testing.T) *bed {
t.Helper()
b := &bed{store: &memStore{entries: map[string]*stdio.StateEntry{}}, answers: &memAnswers{written: map[string][]byte{}, revs: map[string]uint64{}},
gate: memGate{}, records: map[string]string{}, seenIDs: map[string]bool{}, msgIDs: map[string]bool{}}
b.client = &occ.Client{Module: delivery.TellerModule, State: b.store, Now: func() time.Time { return now },
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)
if b.msgIDs[id] {
@@ -146,7 +147,7 @@ func newBed(t *testing.T) *bed {
return err
}
b.msgIDs[id] = true
if subject == oc.AcceptSubjectAsk(delivery.TellerModule) {
if subject == oc.AcceptSubjectAsk(tellerModule) {
var a struct {
ID string `json:"id"`
}
@@ -188,26 +189,26 @@ func newBed(t *testing.T) *bed {
// `mesh-delivery.merge` for this ask and the heads (mesh-catalog modules/mesh-delivery approval.go:123-191 composeMergeAsk
// and :63-65 mergeAct), under the id its key gives (mesh-sdk go/delivery AskID). boundTo are the heads the options bind,
// which are the ask's own unless a test says another.
func gateAsk(t *testing.T, heads, boundTo []delivery.Head) (key string, value []byte, ask gateAsked) {
func gateAsk(t *testing.T, heads, boundTo []meshdeliveryclient.Head) (key string, value []byte, ask gateAsked) {
t.Helper()
return gateAskN(t, heads, boundTo, 1)
}
// gateAskN is the asking-th writing of the group's ask at these heads: its router id is AskingID(key, asking) (mesh-sdk
// go/delivery Ask.Asking), the first asking's the key's own.
func gateAskN(t *testing.T, heads, boundTo []delivery.Head, asking int) (key string, value []byte, ask gateAsked) {
func gateAskN(t *testing.T, heads, boundTo []meshdeliveryclient.Head, asking int) (key string, value []byte, ask gateAsked) {
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 []delivery.Head, asking int) (key string, value []byte, ask gateAsked) {
func gateAskOf(t *testing.T, group, verb string, heads, boundTo []meshdeliveryclient.Head, asking int) (key string, value []byte, ask gateAsked) {
t.Helper()
key, err := delivery.MergeAskKey(group, heads)
key, err := meshdeliveryclient.MergeAskKey(group, heads)
if err != nil {
t.Fatal(err)
}
id := delivery.AskingID(key, asking)
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"}} {
@@ -221,7 +222,7 @@ func gateAskOf(t *testing.T, group, verb string, heads, boundTo []delivery.Head,
return key, encodeGate(t, ask), ask
}
// rawAsk is the router's ask as the JSON a gate or assigner value carries (delivery.Ask.Ask).
// rawAsk is the router's ask as the JSON a gate or assigner value carries (the seat's ask value).
func rawAsk(t *testing.T, a oc.Ask) json.RawMessage {
t.Helper()
b, err := json.Marshal(a)
@@ -235,27 +236,27 @@ func rawAsk(t *testing.T, a oc.Ask) json.RawMessage {
// (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 := delivery.MergeAskKey(ga.Group, ga.Heads)
key, err := meshdeliveryclient.MergeAskKey(ga.Group, ga.Heads)
if err != nil {
t.Fatal(err)
}
if want := delivery.AskingID(key, ga.Asking); ga.Ask.ID != want || len(ga.Ask.Options) == 0 {
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 := delivery.EncodeAsk(delivery.Ask{Group: ga.Group, Heads: ga.Heads, Asking: ga.Asking, Ask: rawAsk(t, ga.Ask)})
value, err := gateValue(ga.Group, ga.Heads, ga.Asking, rawAsk(t, ga.Ask))
if err != nil {
t.Fatal(err)
}
return value
}
func twoHeads() []delivery.Head {
return []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headA}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
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: delivery.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: "Approve", Level: occ.Approve,
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)}
}
@@ -263,7 +264,7 @@ func warrantFor(a oc.Ask, option string) oc.Warrant {
func (b *bed) deliver(t *testing.T, w oc.Warrant) {
t.Helper()
raw, _ := json.Marshal(w)
if err := b.hear(stdio.Envelope{Key: oc.Seat + ".decided." + delivery.TellerModule, Body: raw}); err != nil {
if err := b.hear(stdio.Envelope{Key: oc.Seat + ".decided." + tellerModule, Body: raw}); err != nil {
t.Fatal(err)
}
}
@@ -293,30 +294,30 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1
key, value, ga := gateAsk(t, twoHeads(), twoHeads())
b.gate[key] = value
b.t.OnAsk(key, 3, value)
asked := b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))
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, delivery.TellerModule, ga.Ask.ID)
t.Fatalf("published %+v, want one ask under %s with id %s", b.pubs, tellerModule, ga.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 != delivery.AskID(key) {
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
if n := len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))); n != 1 {
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)
cancels := b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule))
cancels := b.askedOf(oc.AcceptSubjectCancel(tellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+ga.Ask.ID+`"}` {
t.Fatalf("cancels = %+v", cancels)
}
b.t.OnAskGone(key)
if n := len(b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule))); n != 1 {
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,
@@ -337,15 +338,15 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1
_, 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")
cancels := b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule))
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+`"}` {
t.Fatalf("end = %+v, cancels %+v", r, cancels)
}
b.deliver(t, oc.Warrant{Ask: first.Ask.ID, Asker: delivery.TellerModule, Outcome: occ.OutcomeCancelled, At: occ.FormatTime(now), AskDigest: occ.Digest(first.Ask)})
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)
}
if again := b.t.EndAsk(k1, "again", "lead@shanks"); again.Ended || len(b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule))) != 1 {
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)
}
})
@@ -357,7 +358,7 @@ func TestAnAsksKeyBecomesOneRouterAskAndDeletingItCancelsIt(t *testing.T) { // 1
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(delivery.TellerModule))) != 0 {
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 {
t.Fatalf("end of a waiting ask = %+v", r)
}
b.deliver(t, warrantFor(first.Ask, "approve"))
@@ -377,11 +378,11 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { //
b.t.OnAsk(key, 1, value)
w := warrantFor(ga.Ask, "approve")
b.deliver(t, w)
raw, ok := b.answers.written[delivery.AnswerKey(key)]
raw, ok := b.answers.written[answerKey(key)]
if !ok || len(b.answers.written) != 1 {
t.Fatalf("answers = %v", b.answers.written)
}
a, err := delivery.DecodeAnswer(raw)
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)
}
@@ -398,19 +399,19 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { //
// 1280 (ADR 0556 clause 5): a settle ask holds when its options bind mesh-delivery.settle, and not when they bind a merge.
for verb, holds := range map[string]bool{"mesh-delivery.settle": true, "mesh-delivery.merge": false} {
b := newBed(t)
group := delivery.SettleGroup([]string{"novox/mesh-delivery@" + headA[:12]})
group := meshdeliveryclient.SettleGroup([]string{"novox/mesh-delivery@" + headA[:12]})
key, value, ga := gateAskOf(t, group, 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[delivery.AnswerKey(key)]; written != holds {
if _, written := b.answers.written[answerKey(key)]; written != holds {
t.Errorf("a settle ask binding %s: answer written %v, want %v", verb, written, holds)
}
}
}
func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T) { // 1343/AC3
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
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)
@@ -455,7 +456,7 @@ func TestAWarrantThatDoesNotHoldWritesNoAnswerAndRaisesTheCondition(t *testing.T
func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/AC3 (cleared when a good answer is written)
b := newBed(t)
other := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: headB}, {Repository: "novox/mesh-sdk", Number: 4, Head: headB}}
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)
@@ -467,11 +468,11 @@ func TestAGoodAnswerClearsTheConditionOfAnAskAskedAgain(t *testing.T) { // 1343/
_, good, gaGood := gateAskN(t, twoHeads(), twoHeads(), 2)
b.gate[key] = good
b.t.OnAsk(key, 2, good)
if n := len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))); n != 2 {
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)
}
b.deliver(t, warrantFor(gaGood.Ask, "approve"))
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
if _, ok := b.answers.written[answerKey(key)]; !ok {
t.Fatalf("no answer: %v\n%s", b.answers.written, b.logs.String())
}
if len(b.conds) != 2 || b.conds[1].State != occ.StateCleared || b.conds[1].Key != b.conds[0].Key {
@@ -487,7 +488,7 @@ func TestAWarrantComingWhileTheTellerWasAwayIsReadFromTheRouter(t *testing.T) {
w, _ := json.Marshal(warrantFor(ga.Ask, "approve"))
b.records[ga.Ask.ID] = `{"state":"chosen","warrant":` + string(w) + `}`
b.t.OnAsk(key, 1, value) // the teller starts again and reads the key at the same revision
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
if _, ok := b.answers.written[answerKey(key)]; !ok {
t.Errorf("no answer: %s", b.logs.String())
}
}
@@ -503,11 +504,11 @@ func TestARewriteWithAskingTwoCancelsAskingOneAndAsksUnderTheNewID(t *testing.T)
}
b.gate[key] = two
b.t.OnAsk(key, 2, two)
cancels := b.askedOf(oc.AcceptSubjectCancel(delivery.TellerModule))
cancels := b.askedOf(oc.AcceptSubjectCancel(tellerModule))
if len(cancels) != 1 || string(cancels[0].body) != `{"id":"`+g1.Ask.ID+`"}` {
t.Fatalf("cancels = %+v, want asking 1's router ask", cancels)
}
asked := b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))
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)
}
@@ -517,14 +518,14 @@ func TestARewriteWithAskingTwoCancelsAskingOneAndAsksUnderTheNewID(t *testing.T)
// The same revision again, and a warrant for the first asking (the gate has moved on), change nothing.
b.t.OnAsk(key, 2, two)
b.deliver(t, warrantFor(g1.Ask, "approve"))
if len(b.answers.written) != 0 || len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))) != 2 {
t.Errorf("answers %v asks %d", b.answers.written, len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))))
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")
b.deliver(t, w)
raw, ok := b.answers.written[delivery.AnswerKey(key)]
a, err := delivery.DecodeAnswer(raw)
raw, ok := b.answers.written[answerKey(key)]
a, err := meshdeliveryclient.DecodeAnswer(raw)
if !ok || err != nil || a.Warrant != warrantID(w) {
t.Errorf("answer %+v (%v) err %v", a, ok, err)
}
@@ -536,27 +537,27 @@ func TestAnAskWhoseWordsChangedUnderTheSameIDIsCancelledAndNotAskedAgain(t *test
b.t.OnAsk(key, 1, value)
changed := ga
changed.Ask.Explanation = "Merges two pull requests, now with a third point.\nReview: approved."
v2, err := delivery.EncodeAsk(delivery.Ask{Group: changed.Group, Heads: changed.Heads, Asking: changed.Asking, Ask: rawAsk(t, changed.Ask)})
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(delivery.TellerModule))); n != 1 {
if n := len(b.askedOf(oc.AcceptSubjectCancel(tellerModule))); n != 1 {
t.Errorf("cancels = %d", n)
}
if n := len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))); n != 1 || len(b.dropped) != 0 {
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 := []delivery.Head{{Repository: "novox/Mesh-Delivery", Number: 12, Head: headA}}
key, err := delivery.MergeAskKey("g", heads)
heads := []meshdeliveryclient.Head{{Repository: "novox/Mesh-Delivery", Number: 12, Head: headA}}
key, err := meshdeliveryclient.MergeAskKey("g", heads)
if err != nil {
t.Fatal(err)
}
id := delivery.AskID(key)
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 {
@@ -568,7 +569,7 @@ func TestACapitalisedRepositoryIsSpelledInLowerCaseInTheAct(t *testing.T) { // T
b.gate[key] = value
b.t.OnAsk(key, 1, value)
b.deliver(t, warrantFor(a, "approve"))
if _, ok := b.answers.written[delivery.AnswerKey(key)]; !ok {
if _, ok := b.answers.written[answerKey(key)]; !ok {
t.Errorf("no answer for the gate's spelling: %s", b.logs.String())
}
}
@@ -587,7 +588,7 @@ 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: delivery.TellerModule, Outcome: occ.OutcomeRefused, Words: "no channel can send", At: occ.FormatTime(now)}
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)
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)
@@ -600,22 +601,22 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R
b.gate[key] = one
b.t.OnAsk(key, 1, one)
b.deliver(t, warrantFor(g1.Ask, "approve"))
first, ok := b.answers.written[delivery.AnswerKey(key)]
if a, err := delivery.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.Ask.ID+"@") {
first, ok := b.answers.written[answerKey(key)]
if a, err := meshdeliveryclient.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.Ask.ID+"@") {
t.Fatalf("asking 1's answer: %+v %v %v", a, ok, err)
}
// The gate rewrites the key at asking 2 (a review record changed). Asking 1's answer stays in the bucket.
_, two, g2 := gateAskN(t, twoHeads(), twoHeads(), 2)
b.gate[key] = two
b.t.OnAsk(key, 2, two)
asked := b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))
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)
}
w := warrantFor(g2.Ask, "approve")
b.deliver(t, w)
raw := b.answers.written[delivery.AnswerKey(key)]
a, err := delivery.DecodeAnswer(raw)
raw := b.answers.written[answerKey(key)]
a, err := meshdeliveryclient.DecodeAnswer(raw)
if err != nil || a.Warrant != warrantID(w) {
t.Fatalf("the key holds %+v (%v), want asking 2's answer", a, err)
}
@@ -625,8 +626,8 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R
b.deliver(t, w2)
b.t.OnAsk(key, 2, two)
b.t.OnAsk(key, 3, two)
if b.answers.writes != 2 || len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))) != 2 || len(b.dropped) != 0 {
t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(oc.AcceptSubjectAsk(delivery.TellerModule))), b.dropped)
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)
}
}
@@ -634,3 +635,16 @@ func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R
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
+14 -12
View File
@@ -5,7 +5,9 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
)
// The view's answers (novox/hq ADR 0528 clause 6, ADR 0437 §10, issue 1297 B10). The operator may answer, in the view, the
@@ -19,7 +21,7 @@ import (
func (t *Teller) OnViewAnswer(key string) {
t.mu.Lock()
defer t.mu.Unlock()
askKey, err := delivery.ParseViewAnswerKey(key)
askKey, err := viewstate.ParseViewAnswerKey(key)
if err != nil {
t.logf("the view's key %s is left: %v", key, err)
return
@@ -32,7 +34,7 @@ func (t *Teller) OnViewAnswer(key string) {
func (t *Teller) OnAssignerAsk(key string) {
t.mu.Lock()
defer t.mu.Unlock()
if _, err := delivery.ViewAnswerKey(key); err != nil {
if _, err := viewstate.ViewAnswerKey(key); err != nil {
t.logf("the assigner's key %s is left and never answered from the view: %v", key, err)
return
}
@@ -49,7 +51,7 @@ func (t *Teller) fillFromView(askKey string) {
if t.Assigner == nil || t.View == nil {
return
}
viewKey, err := delivery.ViewAnswerKey(askKey) // refuses an ask that needs proof, naming the phone
viewKey, err := viewstate.ViewAnswerKey(askKey) // refuses an ask that needs proof, naming the phone
if err != nil {
t.logf("%s is not answered from the view: %v", askKey, err)
return
@@ -70,7 +72,7 @@ func (t *Teller) fillFromView(askKey string) {
if !found {
return // the answer waits for its ask: OnAssignerAsk fills it when it comes
}
va, err := delivery.DecodeViewAnswer(rawAnswer)
va, err := viewstate.DecodeViewAnswer(rawAnswer)
if err != nil {
t.logf("%s: the view's answer is not readable and fills nothing: %v", viewKey, err)
return
@@ -80,7 +82,7 @@ func (t *Teller) fillFromView(askKey string) {
t.logf("%s: the assigner's ask is not readable and takes no answer: %v", askKey, err)
return
}
if err := delivery.ViewAnswerFits(va, askKey, delivery.AssignerAsk{Asking: ask.Asking}); err != nil {
if err := viewstate.ViewAnswerFits(va, askKey, asks.Ask{Asking: ask.Asking}); err != nil {
t.logf("%s: nothing is written: %v", askKey, err)
return
}
@@ -95,7 +97,7 @@ func (t *Teller) fillFromView(askKey string) {
t.logf("%s: nothing is written: the view's answer chose %q, an option of level %s, which only a channel that proves who answered takes", askKey, va.Option, o.Level)
return
}
akey := delivery.AnswerKey(askKey)
akey := answerKey(askKey)
cur, crev, has, err := t.Answers.Get(akey)
if err != nil {
t.logf("%s: the answer could not be read: %v", askKey, err)
@@ -108,7 +110,7 @@ func (t *Teller) fillFromView(askKey string) {
if w := strings.TrimSpace(va.Words); w != "" {
words += ": " + w
}
body, err := delivery.EncodeAnswer(delivery.Answer{Option: va.Option, Warrant: ask.Ask.ID + "@" + at.UTC().Format(time.RFC3339Nano), Words: words})
body, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: va.Option, Warrant: ask.Ask.ID + "@" + at.UTC().Format(time.RFC3339Nano), Words: words})
if err != nil {
t.logf("%s: the answer could not be made: %v", askKey, err)
return
@@ -148,12 +150,12 @@ func (t *Teller) viewAnswer(key string) ([]byte, time.Time, bool, error) {
// changesMind says whether the view's answer va, written at at, replaces the answer cur of the same asking of the hold ask at
// askKey: it chose another option and was written after cur's warrant.
func (t *Teller) changesMind(askKey string, cur []byte, va delivery.ViewAnswer, at time.Time) bool {
k, err := delivery.ParseAnyAskKey(askKey)
if err != nil || k.Kind != delivery.AskKindHold {
func (t *Teller) changesMind(askKey string, cur []byte, va viewstate.ViewAnswer, at time.Time) bool {
k, err := asks.ParseAnyAskKey(askKey)
if err != nil || k.Kind != asks.AskKindHold {
return false
}
a, err := delivery.DecodeAnswer(cur)
a, err := meshdeliveryclient.DecodeAnswer(cur)
if err != nil || a.Option == va.Option {
return false
}
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/viewanswers_test.go
+16 -14
View File
@@ -7,7 +7,9 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
)
const fpHeld = "src2:2222222222222222222222222222222222222222222222222222222222222222"
@@ -17,14 +19,14 @@ const fpHeld = "src2:22222222222222222222222222222222222222222222222222222222222
// takes it; the value is `{"v":1,"ask":<the router's ask>}`).
func holdAsk(t *testing.T) (key string, value []byte, ask oc.Ask) {
t.Helper()
key, err := delivery.HoldAskKey("plex", fpHeld)
key, err := asks.HoldAskKey("plex", fpHeld)
if err != nil {
t.Fatal(err)
}
ask = oc.Ask{ID: delivery.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: occ.Operator,
ask = oc.Ask{ID: meshdeliveryclient.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check.", Who: occ.Operator,
Expires: occ.FormatTime(now.Add(occ.ApproveLasts)), OnExpiry: "the fingerprint stays held",
Options: []oc.Option{{ID: "keep", Label: "Keep it held", Does: "keeps it held", Level: occ.Acknowledge}, {ID: "release", Label: "Release it", Does: "lets the move go on", Level: occ.Acknowledge}}}
value, err = delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: rawAsk(t, ask)})
value, err = asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask)})
if err != nil {
t.Fatal(err)
}
@@ -32,13 +34,13 @@ func holdAsk(t *testing.T) (key string, value []byte, ask oc.Ask) {
}
func holdKey() string {
k, _ := delivery.HoldAskKey("plex", fpHeld)
k, _ := asks.HoldAskKey("plex", fpHeld)
return k
}
func viewAnswer(t *testing.T, option, words string) []byte {
t.Helper()
v, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: delivery.AskID(holdKey()), Option: option, Words: words})
v, err := viewstate.EncodeViewAnswer(viewstate.ViewAnswer{Asking: meshdeliveryclient.AskID(holdKey()), Option: option, Words: words})
if err != nil {
t.Fatal(err)
}
@@ -63,7 +65,7 @@ func TestAViewAnswerFillsTheAskOnce(t *testing.T) { // 1297/B10 AC
if !ok {
t.Fatalf("no answer under %s; log:\n%s", key, b.logs.String())
}
a, err := delivery.DecodeAnswer(raw)
a, err := meshdeliveryclient.DecodeAnswer(raw)
if err != nil || a.Option != "release" || answeredAsking(raw) != ask.ID || !strings.Contains(a.Words, "view") || !strings.Contains(a.Words, "the check was a flake") {
t.Errorf("answer = %+v, %v", a, err)
}
@@ -86,7 +88,7 @@ func TestAViewAnswerHeardBeforeItsAskFillsItWhenTheAskComes(t *testing.T) { // 1
}
assigner[key] = value
b.t.OnAssignerAsk(key)
if a, err := delivery.DecodeAnswer(b.answers.written[key]); err != nil || a.Option != "keep" {
if a, err := meshdeliveryclient.DecodeAnswer(b.answers.written[key]); err != nil || a.Option != "keep" {
t.Errorf("answer = %+v, %v", a, err)
}
}
@@ -129,7 +131,7 @@ func TestAViewAnswerNeverReplacesTheAnswerOfTheSameAsking(t *testing.T) { // 129
key, value, ask := holdAsk(t)
assigner[key] = value
view[key+".view"] = viewAnswer(t, "keep", "")
early, err := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: ask.ID + "@2026-10-20T09:00:00Z", Words: "answered elsewhere"})
early, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: "release", Warrant: ask.ID + "@2026-10-20T09:00:00Z", Words: "answered elsewhere"})
if err != nil {
t.Fatal(err)
}
@@ -139,7 +141,7 @@ func TestAViewAnswerNeverReplacesTheAnswerOfTheSameAsking(t *testing.T) { // 129
t.Errorf("the answer of the same asking was replaced")
}
// An earlier asking's answer (another router id) does not stop this one.
older, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: "an-older-id@2026-10-19T09:00:00Z", Words: "old"})
older, _ := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: "release", Warrant: "an-older-id@2026-10-19T09:00:00Z", Words: "old"})
b.answers.written[key], b.answers.revs[key] = older, 8
b.t.OnViewAnswer(key + ".view")
if b.answers.writes != 1 || answeredAsking(b.answers.written[key]) != ask.ID {
@@ -169,7 +171,7 @@ func TestAStaleViewAnswerOfAnotherAskingFillsNothing(t *testing.T) { // 1297/B10
assigner, view := withView(b)
key, value, _ := holdAsk(t)
assigner[key] = value
stale, err := delivery.EncodeViewAnswer(delivery.ViewAnswer{Asking: "an-earlier-asking", Option: "release"})
stale, err := viewstate.EncodeViewAnswer(viewstate.ViewAnswer{Asking: "an-earlier-asking", Option: "release"})
if err != nil {
t.Fatal(err)
}
@@ -203,7 +205,7 @@ func TestALaterViewAnswerChangesAHoldsAnswer(t *testing.T) {
b.t.Assigner, b.t.View = assigner, view
key, value, ask := holdAsk(t)
assigner[key] = value
kept, _ := delivery.EncodeAnswer(delivery.Answer{Option: "keep", Warrant: ask.ID + "@" + now.Add(time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator, via telegram, chose keep"})
kept, _ := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: "keep", Warrant: ask.ID + "@" + now.Add(time.Hour).UTC().Format(time.RFC3339Nano), Words: "the operator, via telegram, chose keep"})
b.answers.written[key], b.answers.revs[key] = kept, 3
view.memGate[key+".view"] = viewAnswer(t, option, "it was a flake")
view.at[key+".view"] = viewAt
@@ -212,7 +214,7 @@ func TestALaterViewAnswerChangesAHoldsAnswer(t *testing.T) {
}
later := now.Add(2 * time.Hour)
b, key, ask := setup(later, "release")
a, err := delivery.DecodeAnswer(b.answers.written[key])
a, err := meshdeliveryclient.DecodeAnswer(b.answers.written[key])
if err != nil || b.answers.writes != 1 || a.Option != "release" || a.Warrant != ask.ID+"@"+later.UTC().Format(time.RFC3339Nano) {
t.Fatalf("writes %d, answer %+v, %v; log:\n%s", b.answers.writes, a, err, b.logs.String())
}
@@ -226,7 +228,7 @@ func TestALaterViewAnswerChangesAHoldsAnswer(t *testing.T) {
}
// A restore ask answered stays answered: only a hold's answer may be changed.
bb := newBed(t)
if bb.t.changesMind("restore.ace.plex.2222222222222222222222222222222222222222222222222222222222222222", b.answers.written[key], delivery.ViewAnswer{Option: "keep"}, later.Add(time.Hour)) {
if bb.t.changesMind("restore.ace.plex.2222222222222222222222222222222222222222222222222222222222222222", b.answers.written[key], viewstate.ViewAnswer{Option: "keep"}, later.Add(time.Hour)) {
t.Error("a restore ask's answer was changed")
}
}
modules/mesh-delivery-teller/go.mod
+4
View File
@@ -3,6 +3,10 @@ module mesh-delivery-teller
go 1.26.0
require (
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
modules/mesh-delivery-teller/internal/ends/ends.go
+7 -5
View File
@@ -14,14 +14,16 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
)
const (
Tool = "teller_end_ask"
// module is the teller's own name (its manifest's module), the holder of the bucket.
module = "mesh-delivery-teller"
Tool = "teller_end_ask"
// State is the teller's state of ends asked by hand: the bucket mesh-delivery-teller_ends.
State = "ends"
Bucket = delivery.TellerModule + "_" + State
Bucket = module + "_" + State
)
// Result is the teller's answer to an end: which ask it found (its router id and the gate's or the assigner's key it was
@@ -64,7 +66,7 @@ func Decode(data []byte) (End, error) {
}
// New is the value teller_end_ask writes: its key and the value. The key is the router's spelling of the name given
// (delivery.AskID), which a bucket key takes whatever the name holds, and is the same for the same name, so a second end
// (meshdeliveryclient.AskID), which a bucket key takes whatever the name holds, and is the same for the same name, so a second end
// of one ask overwrites the first.
func New(ask, why, by string, at time.Time) (string, []byte, error) {
ask, why = strings.TrimSpace(ask), strings.TrimSpace(why)
@@ -75,5 +77,5 @@ func New(ask, why, by string, at time.Time) (string, []byte, error) {
return "", nil, errors.New("say why the ask is ended (why), in plain words: it is logged with who ended it")
}
data, err := json.Marshal(End{V: 1, Ask: ask, Why: why, By: by, At: at.UTC()})
return delivery.AskID(ask), data, err
return meshdeliveryclient.AskID(ask), data, err
}