Merge pull request 'Offer an open ask no channel has sent again on the minute tick: a stopped messenger loses its outbox (issue 1507)' (#388) from fix/1507-offer-unsent-asks into main

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 {

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)

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))
+	}
+}