From 38116fe00afec6343d269c0e30b8187e171ff77f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 15:54:08 +0200 Subject: [PATCH] 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)