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

An answer to an earlier asking no longer blocks the next asking: the teller and the view tell the asking by the answer's warrant id (review round 2, R2-1)

Browse Source
This commit is contained in:
mesh-tools
2026-10-20 01:35:26 +00:00
parent 14e9d56a22
commit bf59b5a27b
6 changed files with 150 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go
+17 -9
View File
@@ -61,23 +61,31 @@ func (s kvStore) Delete(key string) error {
return err
}
// kvAnswers is the bucket answers: the teller's, written by compare-and-set only when the key has no answer.
// kvAnswers is the bucket answers: the teller's, written by compare-and-set: created when the key has no answer, updated at the revision read when it holds an earlier asking's.
type kvAnswers struct {
ctx context.Context
kv jetstream.KeyValue
}
func (a kvAnswers) Create(key string, value []byte) (bool, error) {
_, err := a.kv.Create(a.ctx, key, value)
if errors.Is(err, jetstream.ErrKeyExists) {
return false, nil
func (a kvAnswers) Get(key string) ([]byte, uint64, bool, error) {
e, err := a.kv.Get(a.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, 0, false, nil
}
return err == nil, err
if err != nil {
return nil, 0, false, err
}
return e.Value(), e.Revision(), true, nil
}
func (a kvAnswers) Has(key string) (bool, error) {
_, err := a.kv.Get(a.ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
func (a kvAnswers) Write(key string, value []byte, rev uint64, found bool) (bool, error) {
var err error
if found {
_, err = a.kv.Update(a.ctx, key, value, rev)
} else {
_, err = a.kv.Create(a.ctx, key, value)
}
if errors.Is(err, jetstream.ErrKeyExists) {
return false, nil
}
return err == nil, err
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go
+33 -10
View File
@@ -39,10 +39,11 @@ type routerAsks interface {
// answers is the teller's bucket answers.
type answers interface {
// Create writes the answer under the ask's key only when there is none; created is false when there is one.
Create(key string, value []byte) (created bool, err error)
// Has says whether there is an answer under the key.
Has(key string) (bool, error)
// Get is the answer under the ask's key and its revision; found is false when there is none.
Get(key string) (value []byte, rev uint64, found bool, err error)
// Write creates the answer when found is false (the key had none when it was read) and otherwise updates it at the
// revision read; written is false when another writer was first.
Write(key string, value []byte, rev uint64, found bool) (written bool, err error)
}
// gateAsks reads the gate's bucket asks.
@@ -156,11 +157,11 @@ func (t *Teller) OnAsk(key string, rev uint64, value []byte) {
return
}
id := ga.Ask.ID
if has, err := t.Answers.Has(key); err != nil {
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 {
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 {
@@ -296,6 +297,17 @@ func headsOf(heads []delivery.Head) map[string]string {
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 {
a, err := delivery.DecodeAnswer(raw)
if err != nil {
return ""
}
id, _, _ := strings.Cut(a.Warrant, "@")
return id
}
// warrantID names a warrant, which carries no id of its own: the ask and the time the router gave it.
func warrantID(w asks.Warrant) string {
return w.Ask + "@" + w.At.UTC().Format(time.RFC3339Nano)
@@ -339,12 +351,23 @@ func (t *Teller) Answer(o asks.Option, w asks.Warrant) error {
if err != nil {
return err
}
created, err := t.Answers.Create(delivery.AnswerKey(s.Key), body)
// Compare-and-set: created when the key has no answer; updated at the revision read when it holds an earlier asking's
// answer (the gate ignores that one); left when it holds this asking's.
akey := delivery.AnswerKey(s.Key)
cur, crev, found, err := t.Answers.Get(akey)
if err != nil {
return fmt.Errorf("the answer under %s could not be read: %w", s.Key, err)
}
if found && answeredAsking(cur) == w.Ask {
t.logf("%s: an answer is there already, and is not written again", s.Key)
return nil
}
written, err := t.Answers.Write(akey, body, crev, found)
if err != nil {
return fmt.Errorf("the answer under %s could not be written: %w", s.Key, err)
}
if !created {
t.logf("%s: an answer is there already, and is not written again", s.Key)
if !written {
t.logf("%s: another writer answered first, and the answer is not written again", s.Key)
return nil
}
t.logf("%s: %s", s.Key, w.Says())
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go
+57 -10
View File
@@ -62,20 +62,31 @@ func (m *memStore) Delete(key string) error { delete(m.entries, key); return nil
type memAnswers struct {
written map[string][]byte
creates int
revs map[string]uint64
seq uint64
writes int
}
func (a *memAnswers) Create(key string, value []byte) (bool, error) {
a.creates++
if _, ok := a.written[key]; ok {
func (a *memAnswers) Get(key string) ([]byte, uint64, bool, error) {
v, ok := a.written[key]
return v, a.revs[key], ok, nil
}
// Write is the bucket's compare-and-set: a create of a key that has an answer, or an update at another revision, loses.
func (a *memAnswers) Write(key string, value []byte, rev uint64, found bool) (bool, error) {
_, has := a.written[key]
if found != has || (found && rev != a.revs[key]) {
return false, nil
}
a.written[key] = value
a.seq++
a.writes++
if a.revs == nil {
a.revs = map[string]uint64{}
}
a.written[key], a.revs[key] = value, a.seq
return true, nil
}
func (a *memAnswers) Has(key string) (bool, error) { _, ok := a.written[key]; return ok, nil }
type memGate map[string][]byte
func (g memGate) Get(key string) ([]byte, bool, error) { v, ok := g[key]; return v, ok, nil }
@@ -108,7 +119,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{}},
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{}}
b.client = &asker.Client{Module: delivery.TellerModule, State: b.store, Now: func() time.Time { return now },
Publish: func(subject string, body any, id string) error {
@@ -261,8 +272,8 @@ func TestAWarrantThatHoldsWritesOneAnswerAndASecondWritesNone(t *testing.T) { //
w2 := w
w2.At = w.At.Add(time.Minute) // a second warrant for the same ask
b.deliver(t, w2)
if b.answers.creates != 1 {
t.Errorf("the answer was written %d times, want once", b.answers.creates)
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)
@@ -456,3 +467,39 @@ func TestARefusedAskIsSaidAsACondition(t *testing.T) { // ADR 0437 §9
t.Errorf("conditions = %+v", b.conds)
}
}
func TestAnAnswerToAnEarlierAskingDoesNotBlockTheNextAsking(t *testing.T) { // R2-1 of review round 2
b := newBed(t)
key, one, g1 := gateAskN(t, twoHeads(), twoHeads(), 1)
b.gate[key] = one
b.t.OnAsk(key, 1, one)
b.deliver(t, warrantFor(g1.Ask, "approve"))
first, ok := b.answers.written[delivery.AnswerKey(key)]
if a, err := delivery.DecodeAnswer(first); !ok || err != nil || !strings.HasPrefix(a.Warrant, g1.Ask.ID+"@") {
t.Fatalf("asking 1's answer: %+v %v %v", a, ok, err)
}
// The gate rewrites the key at asking 2 (a review record changed). Asking 1's answer stays in the bucket.
_, two, g2 := gateAskN(t, twoHeads(), twoHeads(), 2)
b.gate[key] = two
b.t.OnAsk(key, 2, two)
asked := b.askedOf(asks.AskSubject(delivery.TellerModule))
if len(asked) != 2 || asked[1].id != g2.Ask.ID {
t.Fatalf("asks = %+v, want asking 2 asked under %s", asked, g2.Ask.ID)
}
w := warrantFor(g2.Ask, "approve")
b.deliver(t, w)
raw := b.answers.written[delivery.AnswerKey(key)]
a, err := delivery.DecodeAnswer(raw)
if err != nil || a.Warrant != g2.Ask.ID+"@"+w.At.UTC().Format(time.RFC3339Nano) {
t.Fatalf("the key holds %+v (%v), want asking 2's answer", a, err)
}
// A second warrant for the same asking writes nothing, and a replay of asking 2 asks nothing.
w2 := w
w2.At = w.At.Add(time.Minute)
b.deliver(t, w2)
b.t.OnAsk(key, 2, two)
b.t.OnAsk(key, 3, two)
if b.answers.writes != 2 || len(b.askedOf(asks.AskSubject(delivery.TellerModule))) != 2 || len(b.dropped) != 0 {
t.Errorf("writes %d asks %d dropped %v", b.answers.writes, len(b.askedOf(asks.AskSubject(delivery.TellerModule))), b.dropped)
}
}
modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go
+1 -1
View File
@@ -381,7 +381,7 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { //
wait(func(s view.Snapshot) bool {
return len(s.Asks) == 1 && s.Asks[0].Key == key && s.Asks[0].How == "answer on the phone"
}, "the open ask never showed")
answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "approve", Warrant: "w1", Words: "the operator, via telegram, chose Approve"})
answer, err := delivery.EncodeAnswer(delivery.Answer{Option: "approve", Warrant: ask.ID + "@2026-10-18T12:00:00Z", Words: "the operator, via telegram, chose Approve"})
if err != nil {
t.Fatal(err)
}
modules/mesh-delivery-view/internal/view/asks_test.go
+32 -2
View File
@@ -33,9 +33,13 @@ func askEntry(t *testing.T, group, headline, words string) Entry {
return Entry{Bucket: delivery.AsksBucket, Key: key, Value: v, At: t0}
}
func answerEntry(t *testing.T, key string) Entry {
// answerEntry is the teller's answer to the first asking of the key; answerToAsking to the asking with this router id. The
// answer's warrant is `<router id>@<time>` (teller.go warrantID).
func answerEntry(t *testing.T, key string) Entry { return answerToAsking(t, key, delivery.AskID(key)) }
func answerToAsking(t *testing.T, key, id string) Entry {
t.Helper()
v, err := delivery.EncodeAnswer(delivery.Answer{Option: "approve", Warrant: "w1", Words: "the operator, via telegram, chose Approve"})
v, err := delivery.EncodeAnswer(delivery.Answer{Option: "approve", Warrant: id + "@2026-10-18T12:00:00Z", Words: "the operator, via telegram, chose Approve"})
if err != nil {
t.Fatal(err)
}
@@ -154,3 +158,29 @@ func TestAHaveOfARemovedModuleIsShownRemovedNotLeftBehind(t *testing.T) { // 137
t.Errorf("table line %q", line)
}
}
func TestAnAnswerToAnEarlierAskingDoesNotHideTheNextAsking(t *testing.T) { // R2-1 of review round 2
m := New()
e := askEntry(t, "feat/x", "Merge into main: mesh-delivery#12", "Merges one pull request.")
fill(m, e, answerEntry(t, e.Key))
if got := m.Snapshot().Asks; len(got) != 0 {
t.Fatalf("asking 1 is answered and listed: %+v", got)
}
// The gate rewrites the key at asking 2; only asking 1's answer exists.
heads := []delivery.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: askHeadA}}
a := asks.Ask{ID: delivery.AskingID(e.Key, 2), Headline: "Merge into main: mesh-delivery#12", Explanation: "Merges one pull request.", Who: asks.Operator,
Expires: t0.Add(23 * time.Hour), OnExpiry: "the merge stays held",
Options: []asks.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: asks.Approve, Binds: "sha256:00"}}}
v, err := delivery.EncodeAsk(delivery.Ask{Group: "feat/x", Heads: heads, Asking: 2, Ask: a})
if err != nil {
t.Fatal(err)
}
fill(m, Entry{Bucket: delivery.AsksBucket, Key: e.Key, Value: v, At: t0})
if got := m.Snapshot().Asks; len(got) != 1 {
t.Fatalf("asking 2 is not listed while only asking 1's answer exists: %+v", got)
}
fill(m, answerToAsking(t, e.Key, a.ID))
if got := m.Snapshot().Asks; len(got) != 0 {
t.Errorf("asking 2 is answered and listed: %+v", got)
}
}
modules/mesh-delivery-view/internal/view/view.go
+10 -5
View File
@@ -148,7 +148,7 @@ type Model struct {
groups map[string]groupEntry // the indexer's candidates, by group
modules map[string]moduleEntry // the indexer's modules, by module
asks map[string]askRecord // the gate's asks, by key
answers map[string]bool // the keys the teller has an answer under
answers map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
newest map[string]string // module -> id
haves map[string]haveEntry
nodes map[string]nodeEntry
@@ -233,7 +233,7 @@ type nodeEntry struct {
// New is an empty model.
func New() *Model {
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, asks: map[string]askRecord{}, answers: map[string]bool{}, newest: map[string]string{},
return &Model{assigns: map[string]assignment{}, builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, asks: map[string]askRecord{}, answers: map[string]string{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
}
@@ -310,12 +310,17 @@ func (m *Model) Apply(e Entry) {
a, err := delivery.DecodeAsk(e.Value)
m.asks[e.Key] = askRecord{value: a, err: errText(err)}
case delivery.AnswersBucket:
// An answer under the ask's key, whatever it says, ends the ask here: the gate acts on it, and the view only shows.
if e.Deleted {
delete(m.answers, e.Key)
return
}
m.answers[e.Key] = true
// The answer's warrant is `<router id>@<time>`: it ends that asking, and only that one (a later asking of the key is
// still open). An answer that cannot be read ends none.
id := ""
if a, err := delivery.DecodeAnswer(e.Value); err == nil {
id, _, _ = strings.Cut(a.Warrant, "@")
}
m.answers[e.Key] = id
case BuildsBucket:
module, rest, ok := strings.Cut(e.Key, ".")
if !ok {
@@ -579,7 +584,7 @@ const (
func (m *Model) askRows() []AskRow {
out := []AskRow{}
for key, a := range m.asks {
if a.err != "" || m.answers[key] {
if a.err != "" || m.answers[key] == a.value.Ask.ID {
continue
}
row := AskRow{Key: key, Group: a.value.Group, Headline: a.value.Ask.Headline, Words: a.value.Ask.Explanation,