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)