diff --git a/modules/messenger/cmd/messenger/asks.go b/modules/messenger/cmd/messenger/asks.go index 008f264b..08a53a95 100644 --- a/modules/messenger/cmd/messenger/asks.go +++ b/modules/messenger/cmd/messenger/asks.go @@ -702,6 +702,39 @@ func (h *Holder) releaseHeld() { } } +// offerUnsent offers again every open ask that no channel has been sent and nothing is queued for: the outbox +// is in memory, so an ask queued when the process stopped was lost with it, and an ask is otherwise offered only +// when it is taken (issue 1507). It goes through the path a new ask takes, so the content rule, the channels' +// capabilities and the burst window apply as they do there; one already sent, queued, held or ended is left. +func (h *Holder) offerUnsent() { + h.mu.Lock() + var keys []string + for k, e := range h.asks { + if e.rec.State == StateOpen { + keys = append(keys, k) + } + } + h.mu.Unlock() + sort.Strings(keys) + now := h.Now() + for _, key := range keys { + e := h.askOf(key) + if e == nil || e.rec.State != StateOpen || !now.Before(e.rec.Ask.Expires) { + continue + } + rec := e.rec + h.mu.Lock() + r := h.open[rec.Record] + skip := r != nil && (!r.Cleared.IsZero() || len(r.Sent) > 0 || r.Pending != "" || (r.Ask != "" && r.Ask != key)) + h.mu.Unlock() + if skip || (r != nil && (h.queued(r.Key) || r.silenced(now))) || (r == nil && h.queued("ask."+key)) { + continue + } + h.Logf("[messenger] the ask %s was never sent on a channel; offering it", key) + h.showAsk(key, rec, now) + } +} + // saveHeld records a held ask's change by compare-and-set, and keeps it; false when another writer was first. func (h *Holder) saveHeld(key string, rec AskRecord, rev uint64) bool { if h.Asks != nil { diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index b203c24a..076fc270 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -770,6 +770,7 @@ func (h *Holder) Tick() { sort.Slice(recs, func(i, j int) bool { return recs[i].Raised.Before(recs[j].Raised) }) h.sweepAsks(now) h.releaseHeld() + h.offerUnsent() h.breakGlass() h.boundWalks(now) h.duePasses(now) diff --git a/modules/messenger/cmd/messenger/offer_unsent_test.go b/modules/messenger/cmd/messenger/offer_unsent_test.go new file mode 100644 index 00000000..690576da --- /dev/null +++ b/modules/messenger/cmd/messenger/offer_unsent_test.go @@ -0,0 +1,68 @@ +package main + +import ( + "testing" + "time" +) + +// Issue 1507: an ask was taken and queued for its burst window, the process was replaced before the flush, and +// the new process never offered it: the outbox is in memory, and an ask is offered only when it is taken. An +// open ask no channel has been sent is offered again, once, by the minute tick; one already sent is not. + +// restarted is a new holder over the same stores and channels, as a replaced process is. +func restarted(t *testing.T, old *Holder, c *clock, st *memStore) *Holder { + t.Helper() + h := &Holder{Away: old.Away, Desk: old.Desk, Store: st, Now: c.now, Logf: t.Logf, Emit: old.Emit, + Bench: old.Bench, Asks: old.Asks, Node: old.Node, RootFree: old.RootFree, Zone: old.Zone, Publish: old.Publish} + h.init() + if err := h.Load(); err != nil { + t.Fatal(err) + } + h.Start() + return h +} + +func TestAnAskQueuedWhenTheProcessStoppedIsOfferedByTheNextProcess(t *testing.T) { + h, tg, dt, c, store, _ := newRouter(t) + st := h.Store.(*memStore) + link(h, store, c, "42", c.now().Add(-2*time.Hour)) + submit(t, h, "mesh-config", heldAsk(c, "cfg-1")) + if len(tg.sends)+len(dt.sends) != 0 { + t.Fatalf("sent before the burst window") + } + // The process stops here, before the flush. + h2 := restarted(t, h, c, st) + for i := 0; i < 5; i++ { + c.pass(time.Minute) + stand(h2, c) + h2.Tick() + h2.Flush() + } + if n := len(tg.sends); n != 1 { + t.Fatalf("the ask was sent %d times on telegram, want once", n) + } + if n := len(dt.sends); n != 0 { + t.Fatalf("an approval ask was sent on the desk: %d", n) + } +} + +func TestAnAskAlreadySentIsNotSentAgainAfterARestart(t *testing.T) { + h, tg, _, c, store, _ := newRouter(t) + st := h.Store.(*memStore) + link(h, store, c, "42", c.now().Add(-2*time.Hour)) + submit(t, h, "mesh-config", heldAsk(c, "cfg-1")) + settle(h, c) + if len(tg.sends) != 1 { + t.Fatalf("not sent: %d", len(tg.sends)) + } + h2 := restarted(t, h, c, st) + for i := 0; i < 5; i++ { + c.pass(time.Minute) + stand(h2, c) + h2.Tick() + h2.Flush() + } + if len(tg.sends) != 1 { + t.Fatalf("sent again after a restart: %d", len(tg.sends)) + } +}