diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go index 7e5dd9d4..e5f7ab28 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks.go @@ -191,10 +191,11 @@ func (t *Teller) raiseAssignerAsk(key string) { return } if err := t.Asker.Ask(a.Ask); err != nil { - t.logf("%s: the ask %s was not asked: %v", key, id, err) - if kept, _, _ := t.keptOf(id); open(kept) { - _ = t.Store.Delete(keptKey(id)) // kept but not published: forgotten, so the next writing of the key asks it - } + // Kept but not published (a transport fault, the router restarting): nothing writes the key again, so the ask is not + // forgotten but waits, kept on the bus, to be asked again at the next end and at the teller's start (issue 1966, + // ADR 0668 §1). Before, the record was deleted for "the next writing of the key", which never came. + t.logf("%s: the ask %s was not asked: %v; it waits to be asked again at the next end and at the start", key, id, err) + t.unheard(id) return } t.assignerAsked(id) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go index 8420dcbc..1604f9df 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/assignerasks_test.go @@ -1,6 +1,7 @@ package main import ( + "errors" "slices" "encoding/json" @@ -34,6 +35,28 @@ func TestAnAssignerAskIsRaisedOnTheOperatorChannelOnce(t *testing.T) { if got.ID != ask.ID || occ.Digest(got) != occ.Digest(ask) { t.Errorf("the router's ask %+v differs from the assigner's", got) } + // Issue 1966, ADR 0668 §1: a publish that fails (the router restarting) is not forgotten: the ask waits, kept, and is + // asked at the teller's next start (or the next end), though nothing writes the assigner's key again. + b = newBed(t) + assigner, _ = withView(b) + assigner[key] = value + publish, failed := b.client.Publish, false + b.client.Publish = func(subject string, body any, id string) error { + if subject == oc.AcceptSubjectAsk(tellerModule) && !failed { + failed = true + return errors.New("nats: no responders available for request") + } + return publish(subject, body, id) + } + b.t.Publish = b.client.Publish + b.t.OnAssignerAsk(key) + if asked := b.askedOf(oc.AcceptSubjectAsk(tellerModule)); !failed || len(asked) != 0 { + t.Fatalf("published %d asks through a failing publish", len(asked)) + } + b.t.OnCurrent() + if asked := b.askedOf(oc.AcceptSubjectAsk(tellerModule)); len(asked) != 1 { + t.Errorf("after a failed publish and a start, published %d asks, want one; log:\n%s", len(asked), b.logs.String()) + } } // 1906/H1: without the view's bucket the assigner's asks are still raised.