From 38116fe00afec6343d269c0e30b8187e171ff77f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 15:54:08 +0200 Subject: [PATCH 1/2] Ask again a merge ask the router refused for want of room, once one of the teller's asks ends, so the operator is asked (issue 1610) The router holds at most three open asks of one asker and refuses a fourth without keeping it; the teller only logged the refusal, so a merge ask in a burst never reached the operator while the gate read it as asked (issues 1610, 1620). Bounded at six refusals, then said as a condition. --- .../cmd/mesh-delivery-teller/decided.go | 9 +- .../mesh-delivery-teller/integration_test.go | 72 ++++++ .../cmd/mesh-delivery-teller/main.go | 7 +- .../cmd/mesh-delivery-teller/room_test.go | 188 ++++++++++++++ .../cmd/mesh-delivery-teller/teller.go | 241 +++++++++++++++++- .../cmd/mesh-delivery-teller/teller_test.go | 27 +- 6 files changed, 531 insertions(+), 13 deletions(-) create mode 100644 modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go index 2d20503..f6f4e9b 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/decided.go +++ b/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: diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index 50ce770..bcadf13 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/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) +} diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index f9f3479..525e003 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/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 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 new file mode 100644 index 0000000..7259c44 --- /dev/null +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/room_test.go @@ -0,0 +1,188 @@ +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) + 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) + } +} 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 a04c742..a80d8b9 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/teller.go +++ b/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.). 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 } @@ -267,7 +273,7 @@ func (t *Teller) catchUp(id string) { } 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 +382,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() + } +} + +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 +474,212 @@ 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"` +} + +// 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). +func (t *Teller) OnCurrent() { + t.mu.Lock() + defer t.mu.Unlock() + t.askWaiting() +} + +// 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() { + 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 + } + if err := t.askAgain(a, fmt.Sprintf("%s.again-%d", a.ID, x.Refusals)); 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 + } +} + +// 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 +} 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 394f022..9adcf1a 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 @@ -113,22 +113,35 @@ 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) + } + b.seenIDs[a.ID] = true } b.pubs = append(b.pubs, published{subject, raw, id}) return err @@ -141,7 +154,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) From 037bf1ef361b2a9a41979f99f1ddf4615ac1e7a9 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 16:45:31 +0200 Subject: [PATCH 2/2] 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 }