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)

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)
+	}
+}

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++
 	}
 }
 

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
 			}