From bf59b5a27b79bd67d3a78ba3217d35f1cc13a89a Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 01:35:26 +0000 Subject: [PATCH] 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) --- .../cmd/mesh-delivery-teller/kv.go | 26 ++++--- .../cmd/mesh-delivery-teller/teller.go | 43 +++++++++--- .../cmd/mesh-delivery-teller/teller_test.go | 67 ++++++++++++++++--- .../mesh-delivery-view/integration_test.go | 2 +- .../internal/view/asks_test.go | 34 +++++++++- .../mesh-delivery-view/internal/view/view.go | 15 +++-- 6 files changed, 150 insertions(+), 37 deletions(-) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go index 8301912..288d07d 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/kv.go @@ -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 diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go index a4ed730..a04c742 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -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()) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go index 20cdcf5..394f022 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 09b17f5..52d601e 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -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) } diff --git a/modules/mesh-delivery-view/internal/view/asks_test.go b/modules/mesh-delivery-view/internal/view/asks_test.go index 1d32b1b..d5e9c08 100644 --- a/modules/mesh-delivery-view/internal/view/asks_test.go +++ b/modules/mesh-delivery-view/internal/view/asks_test.go @@ -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 `@