From 037bf1ef361b2a9a41979f99f1ddf4615ac1e7a9 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 16:45:31 +0200 Subject: [PATCH] Ask again at start a merge ask whose refusal the teller did not hear, since the router keeps no record of a refused ask (issue 1610, review round 1) --- .../cmd/mesh-delivery-teller/room_test.go | 64 ++++++++++++++++ .../cmd/mesh-delivery-teller/teller.go | 73 ++++++++++++++++--- .../cmd/mesh-delivery-teller/teller_test.go | 2 + 3 files changed, 130 insertions(+), 9 deletions(-) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go index 7259c44..abc12e8 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go @@ -34,6 +34,7 @@ func (b *bed) askOf(t *testing.T, n int, rev uint64) (string, delivery.Ask) { func (b *bed) refuse(t *testing.T, id, words string) { t.Helper() delete(b.seenIDs, id) + delete(b.records, id) b.deliver(t, asks.Warrant{Ask: id, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: words, At: now}) } @@ -186,3 +187,66 @@ func TestAnAskRefusedForItsWordsIsSaidAtOnceAndNotAskedAgain(t *testing.T) { // t.Errorf("asks = %v: an ask refused for its words was asked again", got) } } + +// restart is a new teller on the same buckets and the same bus (its message ids still in the bus's duplicate window), and +// the router's records as they stand; the watch hands over the gate's keys, then says it has handed over all. +func (b *bed) restart(t *testing.T) *bed { + t.Helper() + n := newBed(t) + n.store, n.answers, n.gate, n.records, n.msgIDs, n.seenIDs = b.store, b.answers, b.gate, b.records, b.msgIDs, b.seenIDs + n.client.State = b.store + n.t.Store, n.t.Answers, n.t.Gate = b.store, b.answers, b.gate + for key, value := range b.gate { + n.t.OnAsk(key, 1, value) + } + n.t.OnCurrent() + return n +} + +// Review of 38116fe0 (reviewer-1610): an ask asked again whose refusal the teller never heard (it was away; the router's +// warrants come on a plain subscription) is asked again at the next start: the router keeps no record of a refused ask, +// so no record there means it was not taken. +func TestAnAskAskedAgainWhoseRefusalWasMissedIsAskedAgainAtStart(t *testing.T) { // 1610/AC4 + b := newBed(t) + b.askOf(t, 1, 1) + _, held := b.askOf(t, 2, 2) + b.refuse(t, held.Ask.ID, roomRefusal) + _, other := b.askOf(t, 3, 3) + b.expire(t, other) + b.records[other.Ask.ID] = `{"state":"expired"}` + if got := b.askedIDs(); got[len(got)-1] != held.Ask.ID { + t.Fatalf("asks = %v, want %s asked again", got, held.Ask.ID) + } + // Refused again, and the teller does not hear it: the router forgets it. + delete(b.seenIDs, held.Ask.ID) + delete(b.records, held.Ask.ID) + n := b.restart(t) + got := n.askedIDs() + if len(got) != 1 || got[0] != held.Ask.ID { + t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.Ask.ID, n.logs.String()) + } + if len(n.busDropped) != 0 { + t.Errorf("the bus dropped %v as a duplicate", n.busDropped) + } +} + +// The same for a first asking: kept open by the teller, its refusal never heard, and no record at the router. +func TestAFirstAskingWhoseRefusalWasMissedIsAskedAgainAtStart(t *testing.T) { // 1610/AC4 + b := newBed(t) + _, held := b.askOf(t, 1, 1) + delete(b.seenIDs, held.Ask.ID) // refused for room, unheard: the router keeps no record of it + delete(b.records, held.Ask.ID) + n := b.restart(t) + got := n.askedIDs() + if len(got) != 1 || got[0] != held.Ask.ID { + t.Fatalf("asks after the start = %v, want %s asked again; logs:\n%s", got, held.Ask.ID, n.logs.String()) + } + if len(n.busDropped) != 0 { + t.Errorf("the bus dropped %v as a duplicate", n.busDropped) + } + // An ask the router holds (it took the one asked again) is not asked again. + m := n.restart(t) + if got := m.askedIDs(); len(got) != 0 { + t.Errorf("an ask the router holds was asked again: %v", got) + } +} 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 a80d8b9..5d64273 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go @@ -269,6 +269,12 @@ func (t *Teller) catchUp(id string) { return } if w == nil { + if state == "" { + // The router holds no record of an ask the teller holds open: it refused it (a refused ask is not kept) while + // the teller was away and did not hear it, or has not read it yet. It waits for room, and is asked again at the + // start or at the next end; the router drops a second ask under an id it holds (review of 38116fe0). + t.unheard(id) + } return } t.logf("the ask %s is %s at the router: its warrant is handled now", id, state) @@ -402,7 +408,7 @@ func (t *Teller) said(h asker.Handled, w asks.Warrant, err error) { if h == asker.Acted || h == asker.Recorded || h == asker.Done { // One of the teller's asks ended at the router, so it has room for one more: the first ask waiting is asked again. t.endWait(w.Ask) - t.askWaiting() + t.askWaiting(1) } } @@ -500,6 +506,8 @@ type waitingAsk struct { ID string `json:"id"` Refusals int `json:"refusals"` Asked bool `json:"asked,omitempty"` + // Tries is how many times it was asked again: each time under a bus message id of its own, also across a start. + Tries int `json:"tries,omitempty"` } // forWantOfRoom says whether a refusal's words are the router's for an asker that already holds its most open asks @@ -591,15 +599,58 @@ func (t *Teller) endWait(id string) { // OnCurrent is the watch of the gate's asks having handed over every key it holds now, at the teller's start: an ask that // waited for room while the teller was away is asked again (the asks that held the room may have ended meanwhile). +// +// An ask asked again whose word the teller did not hear (the router's warrants come on a plain subscription) waits once +// more when the router holds no record of it: the router keeps none of a refused ask (review of 38116fe0). Up to +// asks.MostOpen are asked, the most the router can have room for. func (t *Teller) OnCurrent() { t.mu.Lock() defer t.mu.Unlock() - t.askWaiting() + ws, rec, err := t.waiting() + if err != nil { + t.logf("%v", err) + return + } + changed := false + for i, x := range ws { + if !x.Asked { + continue + } + state, w, err := t.Asker.Where(x.ID) + if err != nil { + t.logf("the router's record of the ask %s could not be read: %v", x.ID, err) + continue + } + if state == "" && w == nil { + t.logf("%s: the ask %s, asked again, is not on the router's record: its refusal was not heard, and it waits again", x.Key, x.ID) + ws[i].Asked, changed = false, true + } + } + if changed && t.keepWaiting(ws, rec) != nil { + return + } + t.askWaiting(asks.MostOpen) } -// askWaiting asks again the first ask that waits for room and is not asked yet, dropping on the way every one the gate no -// longer asks. One at a time: one ask ended, so there is room for one. -func (t *Teller) askWaiting() { +// unheard puts an ask the teller holds open and the router holds no record of in the wait for room, once. +func (t *Teller) unheard(id string) { + ws, rec, err := t.waiting() + if err != nil { + t.logf("%v", err) + return + } + if slices.ContainsFunc(ws, func(x waitingAsk) bool { return x.ID == id }) { + return + } + s, _, _ := t.seenOf(id) + if t.keepWaiting(append(ws, waitingAsk{Key: s.Key, ID: id}), rec) == nil { + t.logf("%s: the ask %s is open here and not on the router's record: it waits to be asked again", s.Key, id) + } +} + +// askWaiting asks again the first n asks that wait for room and are not asked yet, dropping on the way every one the gate +// no longer asks. One at an end: one ask ended, so there is room for one. +func (t *Teller) askWaiting(n int) { ws, rec, err := t.waiting() if err != nil { t.logf("%v", err) @@ -626,13 +677,17 @@ func (t *Teller) askWaiting() { ws, changed = slices.Delete(ws, i, i+1), true continue } - if err := t.askAgain(a, fmt.Sprintf("%s.again-%d", a.ID, x.Refusals)); err != nil { + ws[i].Tries, changed = x.Tries+1, true + if err := t.askAgain(a, fmt.Sprintf("%s.again-%d", a.ID, ws[i].Tries)); err != nil { t.logf("%s: the ask %s, waiting for room, could not be asked again, and waits for the next end: %v", x.Key, x.ID, err) return } - ws[i].Asked, changed = true, true - t.logf("%s: the ask %s is asked of the operator again, now that one of the teller's asks ended (refused %d times for want of room)", x.Key, x.ID, x.Refusals) - return + ws[i].Asked = true + t.logf("%s: the ask %s is asked of the operator again (refused %d times for want of room)", x.Key, x.ID, x.Refusals) + if n--; n <= 0 { + return + } + i++ } } 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 9adcf1a..8986842 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 @@ -140,6 +140,8 @@ func newBed(t *testing.T) *bed { _ = json.Unmarshal(raw, &a) if b.seenIDs[a.ID] { b.dropped = append(b.dropped, a.ID) + } else if _, kept := b.records[a.ID]; !kept { + b.records[a.ID] = `{"state":"open"}` // the router keeps the ask it took (asks.go CreateAsk) } b.seenIDs[a.ID] = true }