Ask a hold again after a restart, leave a merge ask room, take an answer only for its asking (issue 1906, review)

A hold written again after a lift and a restart reused its first ask id, which
the router drops; the hold's time is now its asking. Seven holds could take the
router's three places for the teller and keep a merge Approve from being asked;
the teller keeps at most two assigner asks open. A release or keep answered for
one hold carried over to a later hold of the same fingerprint; an answer now
counts only for the asking its warrant names.

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

@@ -348,7 +348,7 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) {
 	}
 	akey, _ := delivery.HoldAskKey("web", fNew)
 	a, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey))
-	if err != nil || a.Ask.ID != delivery.AskingID(akey, 0) || len(a.Ask.Options) != 2 {
+	if err != nil || a.Asking <= 0 || a.Ask.ID != delivery.AskingID(akey, a.Asking) || len(a.Ask.Options) != 2 {
 		t.Fatalf("the ask = %+v, %v", a, err)
 	}
 }
@@ -582,8 +582,15 @@ func (b *bed) wantNow(t *testing.T, node, module, what string, ok func(delivery.
 	return last
 }
 
-// answer says the teller wrote the operator's answer to the ask at key.
-func (b *bed) answer(t *testing.T, key, option string) { b.answerTo(t, key, option, "r1") }
+// answer says the teller wrote the operator's answer to the ask that stands at key (its warrant names that asking's id).
+func (b *bed) answer(t *testing.T, key, option string) {
+	t.Helper()
+	a, err := delivery.DecodeAssignerAsk(key, b.get(t, delivery.AssignerAsksBucket, key))
+	if err != nil {
+		t.Fatal(err)
+	}
+	b.answerTo(t, key, option, a.Ask.ID)
+}
 
 // answerTo is answer to the ask with this id (the router's warrant names it).
 func (b *bed) answerTo(t *testing.T, key, option, askID string) {
@@ -809,3 +816,47 @@ func TestAHoldIsLiftedOnceEveryNodeRunsItHealthy(t *testing.T) {
 	}
 	t.Fatalf("the hold was not lifted: held %v, ask %v", b.hasKey(delivery.AssignerWantsBucket, hkey), b.hasKey(delivery.AssignerAsksBucket, akey))
 }
+
+// 1906/H2, whole on the bus: a hold lifted by itself and written again after the assigner was started again is asked under a
+// new id, so the teller raises it; the answer to the first hold does not carry over (1906/H4).
+func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) {
+	b := startBed(t, "")
+	fNew := delivery.SourcePrefix2 + strings.Repeat("b", 64)
+	b.fixture(t, "web", fNew, `{"module":"web","version":"1"}`)
+	b.compose(t, "ace", "web", fNew, 'b', delivery.JudgedNeither)
+	failed, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"})
+	healthy, _ := delivery.EncodeHave(delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
+	b.putJSON(t, delivery.HaveBucket, "ace.web", failed)
+	b.putWant(t, "ace", "web", delivery.Want{FP: fNew, Composed: "sha256:" + strings.Repeat("b", 64)})
+	_, stop := b.startStoppable(t)
+	hkey, _ := delivery.HeldKey("web", fNew)
+	akey, _ := delivery.HoldAskKey("web", fNew)
+	first, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey))
+	if err != nil {
+		t.Fatal(err)
+	}
+	b.putJSON(t, delivery.HaveBucket, "ace.web", healthy)
+	for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("the hold was not lifted")
+		}
+	}
+	stop()
+	time.Sleep(1100 * time.Millisecond) // the next hold is written in another second
+	b.answerTo(t, akey, "keep", first.Ask.ID) // an answer to the first hold, which the later hold must not take
+	b.putJSON(t, delivery.HaveBucket, "ace.web", failed)
+	b.start(t)
+	again, err := delivery.DecodeAssignerAsk(akey, b.get(t, delivery.AssignerAsksBucket, akey))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if again.Ask.ID == first.Ask.ID {
+		t.Fatalf("the hold written again is asked under the first id %s", first.Ask.ID)
+	}
+	b.putJSON(t, delivery.HaveBucket, "ace.web", healthy)
+	for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.AssignerWantsBucket, hkey); time.Sleep(100 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("the stale \"keep\" of the first hold kept the second one held")
+		}
+	}
+}

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go

@@ -28,9 +28,15 @@ type passer struct {
 	// answeredAt is, by the id of a backup ask, how many values the have had when its answer was written into the want.
 	answeredAt map[string]uint64
 	asks       *composeAsks
-	// lifted is, by hold ask key, the holds this process lifted by itself: a hold of the same fingerprint written again is
-	// asked under another asking, since the router drops a second ask under an id it keeps (issue 1906).
-	lifted map[string]bool
+	// now is the clock; nil is time.Now.
+	now func() time.Time
+}
+
+func (p *passer) clock() time.Time {
+	if p.now != nil {
+		return p.now()
+	}
+	return time.Now()
 }
 
 func (p *passer) pass(ctx context.Context) {
@@ -222,10 +228,10 @@ func (p *passer) holdAsk(ctx context.Context, module, fp string, fresh *delivery
 	if err != nil || p.w.has(delivery.AssignerAsksBucket, key) {
 		return
 	}
-	asking, now := 0, time.Now()
-	if p.lifted[key] {
-		asking = int(now.Unix()) // another asking than any before it, also one of an earlier process
-	}
+	// The asking is the hold's time in seconds: a hold of the same fingerprint written again after a lift is another asking, under
+	// an id of its own, also after a restart, since the router drops a second ask under an id it keeps (issue 1906).
+	now := p.clock()
+	asking := int(now.Unix())
 	h := p.heldHold(module, fp, fresh)
 	a := assign.HoldAsk(key, module, fp, h.Node, h.Words, asking, now)
 	p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) })
@@ -256,10 +262,6 @@ func (p *passer) lift(ctx context.Context, module, fp string) {
 		p.said[c.Key+".lifted"] = true
 		delete(p.said, c.Key+".raised") // a hold written again is said again
 	}
-	if p.lifted == nil {
-		p.lifted = map[string]bool{}
-	}
-	p.lifted[ask] = true
 	p.remove(ctx, delivery.AssignerWantsBucket, held)
 	p.remove(ctx, delivery.AssignerAsksBucket, ask)
 }
@@ -317,11 +319,21 @@ func (p *passer) restoreAsk(ctx context.Context, module, node, fp string) {
 	p.put(ctx, delivery.AssignerAsksBucket, key, func() ([]byte, error) { return delivery.EncodeAssignerAsk(key, a) })
 }
 
-// answeredWith says whether the teller wrote an answer to the ask at key and the option chosen is option. The answer is the
-// operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19).
+// answeredWith says whether the teller wrote an answer to the ask at key as it stands now and the option chosen is option. The
+// answer is the operator's, given in the view or on the phone, and stays under its key (ADR 0539 §14, §16, §19); it counts only
+// for the asking it answered (its warrant names that asking's id), so an answer to an ask taken away, or to an earlier hold of
+// the same fingerprint, does not carry over to a later one (issue 1906).
 func (p *passer) answeredWith(key, option string) bool {
 	a, ok := p.w.answer(key)
-	return ok && a.Option == option
+	if !ok || a.Option != option {
+		return false
+	}
+	ask, err := delivery.DecodeAssignerAsk(key, p.w.value(delivery.AssignerAsksBucket, key))
+	if err != nil {
+		return false
+	}
+	id, _, _ := strings.Cut(a.Warrant, "@")
+	return id == ask.Ask.ID
 }
 
 // clearHeld clears, once, the condition of a hold the operator released.

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go

@@ -0,0 +1,100 @@
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-assigner/internal/assign"
+)
+
+// memKV is the part of a bucket a pass writes: puts and deletes.
+type memKV struct {
+	jetstream.KeyValue
+	values map[string][]byte
+}
+
+func (m *memKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
+	m.values[key] = value
+	return uint64(len(m.values)), nil
+}
+
+func (m *memKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
+	delete(m.values, key)
+	return nil
+}
+
+// newPasser is a pass on a fresh world, as after a start, writing to the buckets given, at the clock given.
+func newPasser(kvs buckets, at time.Time) *passer {
+	return &passer{kvs: kvs, w: newWorld(), logf: func(string, ...any) {}, said: map[string]bool{}, answeredAt: map[string]uint64{},
+		now: func() time.Time { return at }}
+}
+
+var fpTest = delivery.SourcePrefix2 + strings.Repeat("b", 64)
+
+// 1906/H2: given a hold lifted by itself, when the assigner is started again and the same fingerprint is held again, then the
+// hold ask is written under another asking, with an id of its own, so the teller raises it (the router drops a second ask under
+// an id it keeps).
+func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) {
+	asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
+	kvs := buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV}
+	key, _ := delivery.HoldAskKey("web", fpTest)
+	held := &delivery.Held{Node: "ace", Words: "health failed"}
+	first := newPasser(kvs, time.Date(2026, 10, 25, 1, 0, 0, 0, time.UTC))
+	first.holdAsk(context.Background(), "web", fpTest, held)
+	a1, err := delivery.DecodeAssignerAsk(key, asksKV.values[key])
+	if err != nil {
+		t.Fatal(err)
+	}
+	first.lift(context.Background(), "web", fpTest)
+	if _, there := asksKV.values[key]; there {
+		t.Fatal("the lift left the hold ask")
+	}
+	again := newPasser(kvs, time.Date(2026, 10, 25, 2, 0, 0, 0, time.UTC)) // started again: it remembers nothing
+	again.holdAsk(context.Background(), "web", fpTest, held)
+	a2, err := delivery.DecodeAssignerAsk(key, asksKV.values[key])
+	if err != nil {
+		t.Fatal(err)
+	}
+	if a2.Ask.ID == a1.Ask.ID {
+		t.Errorf("the ask written again after a lift and a restart has the id %s of the first", a1.Ask.ID)
+	}
+}
+
+// 1906/H4: an answer counts only for the asking it answered: an answer to an earlier hold of the same fingerprint, or to an ask
+// taken away, is not taken for the hold that stands now.
+func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) {
+	key, _ := delivery.HoldAskKey("web", fpTest)
+	p := newPasser(buckets{}, time.Now())
+	put := func(asking int) string {
+		a := assign.HoldAsk(key, "web", fpTest, "ace", "health failed", asking, time.Now())
+		v, err := delivery.EncodeAssignerAsk(key, a)
+		if err != nil {
+			t.Fatal(err)
+		}
+		p.w.put(delivery.AssignerAsksBucket, key, v)
+		return a.Ask.ID
+	}
+	first := put(100)
+	v, _ := delivery.EncodeAnswer(delivery.Answer{Option: "release", Warrant: first + "@2026-10-25T01:00:00Z", Words: "the operator"})
+	p.w.put(delivery.AnswersBucket, key, v)
+	if !p.answeredWith(key, "release") {
+		t.Fatal("the answer to the ask that stands is not taken")
+	}
+	if p.answeredWith(key, "keep") {
+		t.Error("an answer was taken for an option not chosen")
+	}
+	put(200) // the same fingerprint held again, after a lift
+	if p.answeredWith(key, "release") {
+		t.Error("the answer to an earlier hold carried over to a later one")
+	}
+	p.w.put(delivery.AssignerAsksBucket, key, nil) // the ask taken away
+	if p.answeredWith(key, "release") {
+		t.Error("an answer to an ask taken away was taken")
+	}
+}

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go

@@ -326,6 +326,13 @@ func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[stri
 	return n
 }
 
+// value is what a bucket holds at a key now, nil when nothing.
+func (w *world) value(bucket, key string) []byte {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.values[bucket][key]
+}
+
 // has says whether a bucket holds a key now.
 func (w *world) has(bucket, key string) bool {
 	w.mu.Lock()

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

@@ -1,7 +1,11 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
+	"slices"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
 
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -16,6 +20,71 @@ import (
 // under the ask's own key in the bucket answers, which the assigner reads (mesh-delivery-assigner pass.go answeredWith); none
 // of these asks needs proof of who answered (mesh-sdk go/delivery AskNeedsProof).
 
+// mostAssignerOpen is how many of the assigner's asks the teller keeps open at the router at once. The router keeps at most
+// asks.MostOpen open asks of one asker, and the teller asks both the gate's merge asks and the assigner's as one asker, so a burst
+// of holds (seven at once on 2026-10-25) would take all the room and a merge Approve could not be asked; one place is always
+// left for a merge ask. An assigner ask beyond these waits in the wait for room and is asked when one of them ends.
+const mostAssignerOpen = asks.MostOpen - 1
+
+// assignerOpenRecord is the key of the router ids of the assigner's asks the teller asked.
+const assignerOpenRecord = "assigner.open"
+
+// assignerOpen is the router ids of the assigner's asks open now: those asked whose ask is neither answered nor ended. An id
+// that ended is dropped from the record as it is read.
+func (t *Teller) assignerOpen() ([]string, *stdio.StateEntry) {
+	e, err := t.Store.Get(assignerOpenRecord)
+	if err != nil || e == nil {
+		return nil, e
+	}
+	var ids []string
+	if err := json.Unmarshal(e.Value, &ids); err != nil {
+		t.logf("the record %s cannot be read: %v", assignerOpenRecord, err)
+		return nil, e
+	}
+	return slices.DeleteFunc(ids, func(id string) bool {
+		k, _, err := t.keptOf(id)
+		return err == nil && !open(k)
+	}), e
+}
+
+// assignerRoom says whether another of the assigner's asks may be asked now.
+func (t *Teller) assignerRoom() bool {
+	ids, _ := t.assignerOpen()
+	return len(ids) < mostAssignerOpen
+}
+
+// assignerAsked keeps that the assigner's ask with this router id is asked.
+func (t *Teller) assignerAsked(id string) {
+	ids, e := t.assignerOpen()
+	if !slices.Contains(ids, id) {
+		ids = append(ids, id)
+	}
+	var err error
+	if e == nil {
+		_, err = t.Store.Create(assignerOpenRecord, ids)
+	} else {
+		_, err = t.Store.Update(assignerOpenRecord, ids, e.Revision)
+	}
+	if err != nil {
+		t.logf("the record %s could not be kept: %v", assignerOpenRecord, err)
+	}
+}
+
+// waitForAssignerRoom puts an assigner ask in the wait for room, once.
+func (t *Teller) waitForAssignerRoom(key, 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
+	}
+	if t.keepWaiting(append(ws, waitingAsk{Key: key, ID: id}), rec) == nil {
+		t.logf("%s: the assigner's ask %s waits: %d of the assigner's asks are open, the most the teller keeps open so a merge ask always has room", key, id, mostAssignerOpen)
+	}
+}
+
 // isAssignerKey says whether an ask key is one of the assigner's, not the gate's merge ask.
 func isAssignerKey(key string) bool {
 	k, err := delivery.ParseAnyAskKey(key)
@@ -83,6 +152,10 @@ func (t *Teller) raiseAssignerAsk(key string) {
 		t.logf("%s: the record of the ask %s could not be kept, so it is not raised: %v", key, id, err)
 		return
 	}
+	if !t.assignerRoom() {
+		t.waitForAssignerRoom(key, id)
+		return
+	}
 	if err := t.Asker.Ask(a.Ask); err != nil {
 		t.logf("%s: the ask %s was not asked: %v", key, id, err)
 		if kept, _, _ := t.keptOf(id); open(kept) {
@@ -90,6 +163,7 @@ func (t *Teller) raiseAssignerAsk(key string) {
 		}
 		return
 	}
+	t.assignerAsked(id)
 	t.logf("%s: the assigner's ask %s is asked of the operator (%s)", key, id, a.Ask.Headline)
 }
 

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

@@ -157,3 +157,53 @@ func ackFor(a asks.Ask, option string) asks.Warrant {
 	w.Level, w.Label = asks.Acknowledge, option
 	return w
 }
+
+// holdAskOf is the assigner's hold ask of module plex at the n-th fingerprint, shaped as holdAsk.
+func holdAskOf(t *testing.T, n int) (string, []byte, asks.Ask) {
+	t.Helper()
+	key, err := delivery.HoldAskKey("plex", "src2:"+strings.Repeat(string(rune('0'+n)), 64))
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, _, ask := holdAsk(t)
+	ask.ID = delivery.AskID(key)
+	value, err := delivery.EncodeAssignerAsk(key, delivery.AssignerAsk{Ask: ask})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return key, value, ask
+}
+
+// 1906/H5: given seven holds at once (as on 2026-10-25), when they are raised, then at most two are open at the router, a merge
+// ask is still asked, and each hold answered lets the next waiting one be asked.
+func TestABurstOfHoldsLeavesRoomForAMergeAsk(t *testing.T) {
+	b := newBed(t)
+	assigner, _ := withView(b)
+	var held []asks.Ask
+	for n := 0; n < 7; n++ {
+		key, value, ask := holdAskOf(t, n)
+		assigner[key] = value
+		b.t.OnAssignerAsk(key)
+		held = append(held, ask)
+	}
+	if got := b.askedIDs(); len(got) != mostAssignerOpen || got[0] != held[0].ID || got[1] != held[1].ID {
+		t.Fatalf("asked %v, want the first %d holds only; log:\n%s", got, mostAssignerOpen, b.logs.String())
+	}
+	_, merge := b.askOf(t, 1, 1)
+	if got := b.askedIDs(); len(got) != mostAssignerOpen+1 || got[mostAssignerOpen] != merge.Ask.ID {
+		t.Fatalf("asked %v, want the merge ask asked beside the holds", got)
+	}
+	if len(b.dropped) != 0 || len(b.conds) != 0 {
+		t.Errorf("router dropped %v, conditions %+v", b.dropped, b.conds)
+	}
+	b.deliver(t, ackFor(held[0], "release"))
+	got := b.askedIDs()
+	if len(got) != mostAssignerOpen+2 || got[len(got)-1] != held[2].ID {
+		t.Fatalf("asked %v after a hold was answered, want the third hold asked next; log:\n%s", got, b.logs.String())
+	}
+	// The merge ask answered frees its own place, and no hold takes it.
+	b.deliver(t, warrantFor(merge.Ask, "approve"))
+	if got := b.askedIDs(); len(got) != mostAssignerOpen+2 {
+		t.Errorf("asked %v after the merge ask was answered, want no further hold", got)
+	}
+}

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

@@ -708,6 +708,11 @@ func (t *Teller) askWaiting(n int) {
 			i++
 			continue
 		}
+		assigner := isAssignerKey(x.Key)
+		if assigner && !t.assignerRoom() {
+			i++ // the assigner's asks hold their most open: it waits, and a merge ask after it may be asked
+			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)
@@ -720,6 +725,9 @@ func (t *Teller) askWaiting(n int) {
 			return
 		}
 		ws[i].Asked = true
+		if assigner {
+			t.assignerAsked(x.ID)
+		}
 		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