Merge pull request 'Ask again a merge ask the router refused for want of room, so the operator is asked (issue 1610)' (#15) from fix/1610-ask-reaches-channel into main

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go

@@ -88,8 +88,9 @@ func sayVia(ctx context.Context, js jetstream.JetStream, module string) func(c a
 }
 
 // watchAsks hands each key of the gate's bucket asks, with its revision, to put, and each key taken away to gone, until
-// ctx ends or the watch fails.
-func watchAsks(ctx context.Context, kv jetstream.KeyValue, put func(key string, rev uint64, value []byte), gone func(key string)) error {
+// ctx ends or the watch fails. current is called once, when the watch has handed over every key the bucket held at its
+// start (the watch's nil mark).
+func watchAsks(ctx context.Context, kv jetstream.KeyValue, put func(key string, rev uint64, value []byte), gone func(key string), current func()) error {
 	w, err := kv.Watch(ctx, ">")
 	if err != nil {
 		return err
@@ -105,6 +106,10 @@ func watchAsks(ctx context.Context, kv jetstream.KeyValue, put func(key string,
 			}
 			switch {
 			case e == nil:
+				if current != nil {
+					current()
+					current = nil
+				}
 			case e.Operation() == jetstream.KeyValuePut:
 				put(e.Key(), e.Revision(), e.Value())
 			default:

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go

@@ -396,3 +396,75 @@ func TestTellerCarriesAnAskAndWritesTheAnswer(t *testing.T) { // 1343/AC1, AC2
 		t.Errorf("the withdrawn ask was not cancelled: %v %v\n%s", raw, err, logs.String())
 	}
 }
+
+// TestTellerAsksAgainAnAskRefusedForWantOfRoom starts the program against the bus: the router (the test) refuses the gate's
+// ask for want of room, as the live router did on 2026-10-21 (issue 1610), and then says another of the teller's asks
+// lapsed; the refused ask is on the seat's work queue again, which the bus's duplicate window would have dropped under its
+// first message id (1610/AC1 on the real bus).
+func TestTellerAsksAgainAnAskRefusedForWantOfRoom(t *testing.T) { // 1610/AC1
+	now = time.Now().UTC()
+	b := startFull(t)
+	tools := b.as(t, "n1.node-tools")
+	broker := filepath.Join(t.TempDir(), "broker")
+	cred, _ := json.Marshal(credential{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-teller", User: "n1.mesh-delivery-teller", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker}
+	logs := &syncBuffer{}
+	rn := startRunner(b.ctx, func(c context.Context) error { return run(c, func(k string) string { return env[k] }, logs) })
+	t.Cleanup(func() { rn.finish(t, 10*time.Second) })
+	kv, err := tools.KeyValue(b.ctx, gateAsksKV)
+	if err != nil {
+		t.Fatal(err)
+	}
+	seat, err := b.js.Stream(b.ctx, "SEAT_OPERATOR_CHANNEL")
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(500 * time.Millisecond)
+	// lastAsk waits for the seat's queue to hold an ask newer than seq under id, and gives its sequence.
+	lastAsk := func(id string, after uint64) uint64 {
+		t.Helper()
+		for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+			raw, err := seat.GetLastMsgForSubject(b.ctx, seatAsk)
+			if err != nil || raw.Sequence <= after {
+				continue
+			}
+			var a asks.Ask
+			if json.Unmarshal(raw.Data, &a) == nil && a.ID == id {
+				return raw.Sequence
+			}
+		}
+		t.Fatalf("the ask %s is not on the seat's queue after sequence %d\n%s", id, after, logs.String())
+		return 0
+	}
+	say := func(w asks.Warrant) {
+		t.Helper()
+		raw, _ := json.Marshal(w)
+		if err := b.ctl.Publish(seatDecided, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	key, value, held := gateAsk(t, twoHeads(), twoHeads())
+	if _, err := kv.Put(b.ctx, key, value); err != nil {
+		t.Fatal(err)
+	}
+	first := lastAsk(held.Ask.ID, 0)
+	say(asks.Warrant{Ask: held.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeRefused, Words: roomRefusal, At: 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}}
+	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(asks.Warrant{Ask: ga2.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeExpired, At: now, AskDigest: ga2.Ask.Digest()})
+	lastAsk(held.Ask.ID, second)
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go

@@ -128,15 +128,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 	}
 	state := kvStore{ctx: ctx, kv: kvs[1]}
-	client := &asker.Client{Module: delivery.TellerModule, State: state, Publish: publishVia(ctx, js),
+	publish := publishVia(ctx, js)
+	client := &asker.Client{Module: delivery.TellerModule, State: state, Publish: publish,
 		Record: recordVia(ctx, conn, delivery.TellerModule), Subscribe: decidedVia(conn, delivery.TellerModule, 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}
+		Say: sayVia(ctx, js, delivery.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)
 	}
-	if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone); err != nil {
+	if err := watchAsks(ctx, gate, t.OnAsk, t.OnAskGone, t.OnCurrent); err != nil {
 		return fmt.Errorf("watching the gate's asks: %w", err)
 	}
 	return nil

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go

@@ -0,0 +1,252 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// roomRefusal is the router's words when an asker already holds asks.MostOpen open asks, as the live router said them to
+// the teller on 2026-10-21 at 11:21:08 local (novox journal of mesh-delivery-teller.service: "the router refused the ask
+// merge-design_61_print_and_scan-8cc11c5b8c32-d916c290f50f: its asker already has 3 open asks"; mesh-catalog
+// modules/messenger cmd/messenger/asks.go:529-531).
+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)}}
+}
+
+// askOf puts the gate's key for n and hands it to the teller, as the watch would.
+func (b *bed) askOf(t *testing.T, n int, rev uint64) (string, delivery.Ask) {
+	t.Helper()
+	key, value, ga := gateAsk(t, headsN(n), headsN(n))
+	b.gate[key] = value
+	b.t.OnAsk(key, rev, value)
+	return key, ga
+}
+
+// refuse is the router's refusal of an ask: it keeps no record of a refused ask (asks.go refuseAsk only tells its asker),
+// so the id may be asked again.
+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})
+}
+
+// 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 delivery.Ask) {
+	t.Helper()
+	b.deliver(t, asks.Warrant{Ask: ga.Ask.ID, Asker: delivery.TellerModule, Outcome: asks.OutcomeExpired, At: now, AskDigest: ga.Ask.Digest()})
+}
+
+// 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(asks.AskSubject(delivery.TellerModule)) {
+		var a struct {
+			ID string `json:"id"`
+		}
+		_ = json.Unmarshal(p.body, &a)
+		out = append(out, a.ID)
+	}
+	return out
+}
+
+// The live fault of issue 1610: three asks open, a fourth refused for want of room, and the operator then answers one of
+// the three. The fourth is asked again, under its own router id and a new bus message id, and nothing is said as a fault.
+func TestAnAskRefusedForWantOfRoomIsAskedAgainWhenAnotherEnds(t *testing.T) { // 1610/AC1
+	b := newBed(t)
+	_, first := b.askOf(t, 1, 1)
+	b.askOf(t, 2, 2)
+	b.askOf(t, 3, 3)
+	_, fourth := b.askOf(t, 4, 4)
+	b.refuse(t, fourth.Ask.ID, roomRefusal)
+	if len(b.conds) != 0 {
+		t.Errorf("a refusal for want of room is said as a fault before it was asked again: %+v", b.conds)
+	}
+	if got := b.askedIDs(); len(got) != 4 {
+		t.Fatalf("asked %v before room came", got)
+	}
+	b.deliver(t, warrantFor(first.Ask, "approve"))
+	got := b.askedIDs()
+	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(asks.AskSubject(delivery.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)
+	}
+	if len(b.busDropped) != 0 || len(b.dropped) != 0 || len(b.conds) != 0 {
+		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))
+	b.deliver(t, warrantFor(fourth.Ask, "approve"))
+	if _, ok := b.answers.written[delivery.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.
+	_, second := b.askOf(t, 5, 5)
+	b.expire(t, second)
+	if got := b.askedIDs(); len(got) != 6 || got[5] == fourth.Ask.ID {
+		t.Errorf("asks after the answer = %v, want only the fifth asked", got)
+	}
+}
+
+func TestAnAskRefusedForWantOfRoomIsNotAskedAgainOnceTheGateMovedOn(t *testing.T) { // 1610/AC2
+	b := newBed(t)
+	_, first := b.askOf(t, 1, 1)
+	key, gone := b.askOf(t, 2, 2)
+	b.refuse(t, gone.Ask.ID, roomRefusal)
+	// Withdrawn by the gate (a head moved on) before room came.
+	delete(b.gate, key)
+	b.t.OnAskGone(key)
+	key3, g3 := b.askOf(t, 3, 3)
+	b.refuse(t, g3.Ask.ID, roomRefusal)
+	// Rewritten by the gate under its next asking before room came: the later asking is asked under its own id, and the
+	// refused earlier asking is not asked again.
+	_, v3b, g3b := gateAskN(t, headsN(3), headsN(3), 2)
+	b.gate[key3] = v3b
+	b.t.OnAsk(key3, 4, v3b)
+	before := len(b.askedIDs())
+	if ids := b.askedIDs(); ids[before-1] != g3b.Ask.ID {
+		t.Fatalf("asks = %v, want the next asking %s asked", ids, g3b.Ask.ID)
+	}
+	b.deliver(t, warrantFor(first.Ask, "approve"))
+	b.expire(t, g3b)
+	if after := b.askedIDs()[before:]; len(after) != 0 {
+		t.Errorf("asked again %v, which the gate no longer asks", after)
+	}
+}
+
+func TestAnAskRefusedForWantOfRoomTooOftenIsSaidAsACondition(t *testing.T) { // 1610/AC3
+	b := newBed(t)
+	_, held := b.askOf(t, 1, 1)
+	for i := 0; i < mostRoomRefusals; i++ {
+		if len(b.conds) != 0 {
+			t.Fatalf("a condition after %d refusals: %+v", i, b.conds)
+		}
+		b.refuse(t, held.Ask.ID, roomRefusal)
+		_, other := b.askOf(t, 2+i, uint64(2+i))
+		b.expire(t, other)
+	}
+	var again int
+	for _, id := range b.askedIDs() {
+		if id == held.Ask.ID {
+			again++
+		}
+	}
+	if again != mostRoomRefusals {
+		t.Errorf("asked %d times, want %d (once, and again after each refusal but the last)", again, mostRoomRefusals)
+	}
+	asked := len(b.askedIDs())
+	if len(b.conds) != 1 || b.conds[0].State != asks.StateRaised || !strings.Contains(b.conds[0].Summary, "open asks") {
+		t.Fatalf("conditions = %+v", b.conds)
+	}
+	_, last := b.askOf(t, 20, 20)
+	b.expire(t, last)
+	if len(b.askedIDs()) != asked+1 {
+		t.Errorf("asked again after the bound: %v", b.askedIDs())
+	}
+}
+
+func TestAnAskWaitingForRoomIsAskedAgainAtStart(t *testing.T) { // 1610/AC4
+	b := newBed(t)
+	key, held := b.askOf(t, 1, 1)
+	b.refuse(t, held.Ask.ID, roomRefusal)
+	// The teller starts again: a new teller on the same buckets, and the watch hands over the key it holds.
+	b2 := newBed(t)
+	b2.store, b2.answers, b2.gate = b.store, b.answers, b.gate
+	b2.client.State = b.store
+	b2.t.Store, b2.t.Answers, b2.t.Gate = b.store, b.answers, b.gate
+	b2.t.OnAsk(key, 1, b.gate[key])
+	if len(b2.askedIDs()) != 0 {
+		t.Fatalf("asked before the watch handed over its keys: %v", b2.askedIDs())
+	}
+	b2.t.OnCurrent()
+	if got := b2.askedIDs(); len(got) != 1 || got[0] != held.Ask.ID {
+		t.Errorf("asks at start = %v, want %s asked again", got, held.Ask.ID)
+	}
+}
+
+func TestAnAskRefusedForItsWordsIsSaidAtOnceAndNotAskedAgain(t *testing.T) { // 1610/AC5
+	b := newBed(t)
+	_, first := b.askOf(t, 1, 1)
+	_, bad := b.askOf(t, 2, 2)
+	b.refuse(t, bad.Ask.ID, "its words carry an address, which may not leave the mesh")
+	if len(b.conds) != 1 {
+		t.Fatalf("conditions = %+v", b.conds)
+	}
+	b.deliver(t, warrantFor(first.Ask, "approve"))
+	if got := b.askedIDs(); len(got) != 2 {
+		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

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -60,6 +61,11 @@ type Teller struct {
 	// Say publishes a condition to the operator (mesh.seat.operator-channel.accept.condition.<module>).
 	Say  func(c asks.Condition) error
 	Logf func(format string, args ...any)
+	// Publish is the bus the asker publishes on, used to ask again an ask the router refused for want of room under a bus
+	// message id of its own (the bus drops a second message under an id it holds).
+	Publish func(subject string, body any, id string) error
+	// Now is the clock; nil is time.Now.
+	Now func() time.Time
 
 	mu sync.Mutex
 }
@@ -263,11 +269,17 @@ 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)
 	h, err := t.Asker.Take(*w, t.Answer)
-	t.Said(h, *w, err)
+	t.said(h, *w, err)
 }
 
 // mergeAct is what the approve option of the gate's ask binds: this ask, the answer, and the heads it covers. It is
@@ -376,7 +388,31 @@ func (t *Teller) Answer(o asks.Option, w asks.Warrant) error {
 
 // Said is what the teller does with what became of a warrant: a good answer clears the condition of its ask; a warrant
 // that does not hold, or that the router refused, is said as a condition (ADR 0437 §9). Nothing is written for either.
+// An ask the router refused for want of room is not said: it waits, and is asked again when one of the teller's asks
+// ends (issue 1610).
 func (t *Teller) Said(h asker.Handled, w asks.Warrant, err error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.said(h, w, err)
+}
+
+// said is Said under the lock (catchUp holds it).
+func (t *Teller) said(h asker.Handled, w asks.Warrant, err error) {
+	if w.Outcome == asks.OutcomeRefused && forWantOfRoom(w.Words) {
+		if h == asker.Recorded {
+			t.roomRefused(w)
+		}
+		return // a second word of the same refusal is no end of an ask
+	}
+	t.saidOnce(h, w, err)
+	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(1)
+	}
+}
+
+func (t *Teller) saidOnce(h asker.Handled, w asks.Warrant, err error) {
 	switch {
 	case h == asker.Acted && err == nil:
 		t.logf("the warrant of %s was %s", w.Ask, h)
@@ -444,3 +480,261 @@ func (t *Teller) clear(askID string) {
 	}
 	_ = t.Store.Delete(condKey(id))
 }
+
+// Room at the router (issue 1610). The router holds at most asks.MostOpen open asks of one asker and refuses another at once
+// in words, keeping no record of it (mesh-catalog modules/messenger cmd/messenger/asks.go:524-531, refuseAsk), so the
+// same router id may be asked again. Merges come in bursts: on 2026-10-21 at 11:21 a fourth merge ask was refused so, and
+// nothing asked it again, so the operator was never asked. Now an ask refused for want of room waits in the record
+// `room`, and each time one of the teller's asks ends at the router (a warrant of any outcome: the room it held is free)
+// the first that waits is asked again, under its own router id and a bus message id of its own (the bus drops a second
+// message under an id it holds). It is asked again at most mostRoomRefusals times; refused that often it is said as a
+// condition, as any refusal is. An ask the gate took away or asked again under a newer asking meanwhile, or one already
+// answered, is the gate's no more, and is dropped from the wait. The teller holds no timer: an end, and its start, are the
+// events that ask again.
+
+// mostRoomRefusals is how many times an ask is refused for want of room before it is said as a condition and asked no
+// more: each refusal follows an end of another ask, so it is that many asks answered while this one could not be asked.
+const mostRoomRefusals = 6
+
+// roomRecord is the key of the asks that wait for room, in the order they were refused.
+const roomRecord = "room"
+
+// waitingAsk is one ask that waits for room: the gate key it was asked for, its router id, how often the router refused
+// it for want of room, and whether it is asked again and its word not yet heard.
+type waitingAsk struct {
+	Key      string `json:"key"`
+	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
+// ("its asker already has 3 open asks", asks.go:530).
+func forWantOfRoom(words string) bool {
+	return strings.Contains(words, "already has") && strings.Contains(words, "open asks")
+}
+
+func (t *Teller) now() time.Time {
+	if t.Now != nil {
+		return t.Now()
+	}
+	return time.Now()
+}
+
+func (t *Teller) waiting() ([]waitingAsk, *stdio.StateEntry, error) {
+	e, err := t.Store.Get(roomRecord)
+	if err != nil || e == nil {
+		return nil, nil, err
+	}
+	var ws []waitingAsk
+	if err := json.Unmarshal(e.Value, &ws); err != nil {
+		return nil, nil, fmt.Errorf("the record %s cannot be read: %w", roomRecord, err)
+	}
+	return ws, e, nil
+}
+
+func (t *Teller) keepWaiting(ws []waitingAsk, prior *stdio.StateEntry) error {
+	if ws == nil {
+		ws = []waitingAsk{}
+	}
+	var err error
+	if prior == nil {
+		_, err = t.Store.Create(roomRecord, ws)
+	} else {
+		_, err = t.Store.Update(roomRecord, ws, prior.Revision)
+	}
+	if err != nil {
+		t.logf("the record of the asks waiting for room could not be kept: %v", err)
+	}
+	return err
+}
+
+// roomRefused keeps an ask the router refused for want of room waiting, or, refused that way mostRoomRefusals times,
+// says it as a condition and drops it.
+func (t *Teller) roomRefused(w asks.Warrant) {
+	ws, rec, err := t.waiting()
+	if err != nil {
+		t.logf("%v", err)
+		return
+	}
+	i := slices.IndexFunc(ws, func(x waitingAsk) bool { return x.ID == w.Ask })
+	if i < 0 {
+		s, _, _ := t.seenOf(w.Ask)
+		ws = append(ws, waitingAsk{Key: s.Key, ID: w.Ask})
+		i = len(ws) - 1
+	}
+	ws[i].Refusals++
+	ws[i].Asked = false
+	n := ws[i].Refusals
+	if n >= mostRoomRefusals {
+		ws = slices.Delete(ws, i, i+1)
+		if t.keepWaiting(ws, rec) != nil {
+			return
+		}
+		t.logf("the router refused the ask %s for want of room %d times (%s): it is asked no more, and said", w.Ask, n, w.Words)
+		t.raise(w.Ask, "A merge ask could not be asked", fmt.Sprintf("the router refused it %d times for want of room: %s", n, w.Words))
+		return
+	}
+	if t.keepWaiting(ws, rec) != nil {
+		return
+	}
+	t.logf("the router refused the ask %s for want of room (%s): it waits, and is asked again when one of the teller's asks ends (refused %d of at most %d times)",
+		w.Ask, w.Words, n, mostRoomRefusals)
+}
+
+// endWait drops an ask from the wait once the router said it ended otherwise than for want of room.
+func (t *Teller) endWait(id string) {
+	ws, rec, err := t.waiting()
+	if err != nil || rec == nil {
+		return
+	}
+	i := slices.IndexFunc(ws, func(x waitingAsk) bool { return x.ID == id })
+	if i < 0 {
+		return
+	}
+	_ = t.keepWaiting(slices.Delete(ws, i, i+1), rec)
+}
+
+// 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()
+	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)
+}
+
+// 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)
+		return
+	}
+	if rec == nil {
+		return
+	}
+	changed := false
+	defer func() {
+		if changed {
+			_ = t.keepWaiting(ws, rec)
+		}
+	}()
+	for i := 0; i < len(ws); {
+		x := ws[i]
+		if x.Asked {
+			i++
+			continue
+		}
+		a, why := t.stillAsked(x)
+		if why != "" {
+			t.logf("%s: the ask %s, waiting for room, is not asked again: %s", x.Key, x.ID, why)
+			ws, changed = slices.Delete(ws, i, i+1), true
+			continue
+		}
+		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 = 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++
+	}
+}
+
+// stillAsked is the gate's ask a waiting ask is for, when the gate still asks it under the same id and it is unanswered;
+// otherwise why it is not.
+func (t *Teller) stillAsked(x waitingAsk) (asks.Ask, string) {
+	if x.Key == "" {
+		return asks.Ask{}, "the teller holds no gate key for it"
+	}
+	raw, found, err := t.Gate.Get(x.Key)
+	if err != nil {
+		return asks.Ask{}, "the gate's ask could not be read: " + err.Error()
+	}
+	if !found {
+		return asks.Ask{}, "the gate has taken it away"
+	}
+	ga, err := delivery.DecodeAsk(raw)
+	if err != nil {
+		return asks.Ask{}, "the gate's key holds no ask: " + err.Error()
+	}
+	if ga.Ask.ID != x.ID {
+		return asks.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 {
+		return asks.Ask{}, "it is answered"
+	}
+	if err := ga.Ask.Check(t.now()); err != nil {
+		return asks.Ask{}, "it no longer holds: " + err.Error()
+	}
+	return ga.Ask, ""
+}
+
+// askAgain keeps the ask afresh as the asker would (its record of the refusal is replaced, so its warrant is taken) and
+// publishes it under the bus message id given.
+func (t *Teller) askAgain(a asks.Ask, msgID string) error {
+	if t.Publish == nil {
+		return errors.New("the teller has no bus to ask on")
+	}
+	if err := t.Store.Delete(keptKey(a.ID)); err != nil {
+		return err
+	}
+	if _, err := t.Store.Create(keptKey(a.ID), asker.Kept{Ask: a, Asked: t.now()}); err != nil {
+		return err
+	}
+	if err := t.Publish(asks.AskSubject(delivery.TellerModule), a, msgID); err != nil {
+		_ = t.Store.Delete(keptKey(a.ID))
+		return err
+	}
+	return nil
+}

modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller_test.go

@@ -113,22 +113,37 @@ type bed struct {
 	// cmd/messenger/asks.go:487-489, keepAsk), so a reused id reaches no phone.
 	seenIDs map[string]bool
 	dropped []string
-	hear    func(stdio.Envelope) error
-	logs    strings.Builder
+	// msgIDs are the bus message ids published, and busDropped the messages the bus dropped as a second under one.
+	msgIDs     map[string]bool
+	busDropped []string
+	hear       func(stdio.Envelope) error
+	logs       strings.Builder
 }
 
 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{}}
+		gate: memGate{}, records: map[string]string{}, seenIDs: map[string]bool{}, msgIDs: 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 {
 			raw, err := json.Marshal(body)
+			if b.msgIDs[id] {
+				// The bus drops a second message under an id it holds (JetStream's duplicate window): it reaches nobody.
+				b.busDropped = append(b.busDropped, id)
+				return err
+			}
+			b.msgIDs[id] = true
 			if subject == asks.AskSubject(delivery.TellerModule) {
-				if b.seenIDs[id] {
-					b.dropped = append(b.dropped, id)
+				var a struct {
+					ID string `json:"id"`
 				}
-				b.seenIDs[id] = true
+				_ = 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
 			}
 			b.pubs = append(b.pubs, published{subject, raw, id})
 			return err
@@ -141,7 +156,7 @@ func newBed(t *testing.T) *bed {
 		},
 		Subscribe: func(_ string, handle func(stdio.Envelope) error) error { b.hear = handle; return nil },
 	}
-	b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Gate: b.gate,
+	b.t = &Teller{Asker: b.client, Store: b.store, Answers: b.answers, Gate: b.gate, Publish: b.client.Publish, Now: func() time.Time { return now },
 		Say: func(c asks.Condition) error {
 			if _, err := asks.EncodeCondition(c); err != nil {
 				t.Errorf("a condition the router would refuse: %v", err)