From 6cea12a023a93f1f19cf8ba359361f0908091374 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 10:02:27 +0100 Subject: [PATCH] messenger: say what waits in the outbox in the history, so an ask held for its burst window does not read as lost (issue 1507) --- modules/messenger/cmd/messenger/main.go | 10 +++-- .../cmd/messenger/offer_unsent_test.go | 37 +++++++++++++++++++ modules/messenger/cmd/messenger/status.go | 31 +++++++++++++++- 3 files changed, 73 insertions(+), 5 deletions(-) diff --git a/modules/messenger/cmd/messenger/main.go b/modules/messenger/cmd/messenger/main.go index 458dd236..e6c963f8 100644 --- a/modules/messenger/cmd/messenger/main.go +++ b/modules/messenger/cmd/messenger/main.go @@ -677,7 +677,8 @@ func tools(h *Holder, l *listening) []stdio.Tool { Run: func(map[string]any) (any, error) { return h.Open(), nil }}, {Name: "operator-channel.history", Description: "What was said to the operator lately, newest first — each send, digest, edit and failure with its " + - "channel, key and outcome — and every message refused for carrying an address, a path or a secret.", + "channel, key and outcome — every message refused for carrying an address, a path or a secret, and what " + + "waits to be sent on each channel, since when and when its burst window ends.", Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many of each (default 50)"}}, Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 50)), nil }}, {Name: "operator-channel.notify", @@ -750,9 +751,10 @@ func tools(h *Holder, l *listening) []stdio.Tool { return h.Unlink(kind) }}, {Name: "messenger_recent", - Description: "The recent sends — each message, digest, edit and failure on each channel, newest first — and the refusals.", - Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many (default 20)"}}, - Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 20)), nil }}, + Description: "The recent sends — each message, digest, edit and failure on each channel, newest first — the refusals, " + + "and what waits to be sent: each message held on a channel (its burst window is 30 s), since when and when its window ends.", + Input: map[string]any{"limit": map[string]any{"type": "integer", "description": "at most this many (default 20)"}}, + Run: func(a map[string]any) (any, error) { return h.History(limitArg(a, 20)), nil }}, {Name: "messenger_test", Description: "Send a test message now, to one channel kind or every one (default), through the content rule " + "and the cap: proves a channel reaches the operator. Answers per channel: put on its queue, or why not.", diff --git a/modules/messenger/cmd/messenger/offer_unsent_test.go b/modules/messenger/cmd/messenger/offer_unsent_test.go index 690576da..2e646716 100644 --- a/modules/messenger/cmd/messenger/offer_unsent_test.go +++ b/modules/messenger/cmd/messenger/offer_unsent_test.go @@ -66,3 +66,40 @@ func TestAnAskAlreadySentIsNotSentAgainAfterARestart(t *testing.T) { t.Fatalf("sent again after a restart: %d", len(tg.sends)) } } + +// Issue 1507, read again 2026-10-28: three asks taken at 08:50:03-08:50:14 went out together at 08:50:33, at the +// end of the first one's burst window; a reading of the history in between showed no send and no refusal, and the +// asks read as lost. What waits in an outbox is in the history, with when its window ends, until it is sent. +func TestTheHistorySaysAnAskWaitingForItsBurstWindow(t *testing.T) { + h, tg, _, c, store, _ := newRouter(t) + link(h, store, c, "42", c.now().Add(-2*time.Hour)) + submit(t, h, "mesh-config", heldAsk(c, "cfg-1")) + waiting := func() []Waiting { return h.History(20)["waiting"].([]Waiting) } + for _, step := range []struct { + name string + pass time.Duration + want int + }{ + {"taken, inside its window", 0, 1}, + {"its window passed and it was sent", BurstWindow + time.Second, 0}, + } { + c.pass(step.pass) + stand(h, c) + h.Flush() + w := waiting() + if len(w) != step.want { + t.Fatalf("%s: %d waiting, want %d: %+v", step.name, len(w), step.want, w) + } + if step.want == 1 { + if w[0].Key != "ask.mesh-config.cfg-1" || w[0].Channel != tg.Name() || w[0].Due == "" { + t.Fatalf("%s: waiting %+v, want the ask on %s with when its window ends", step.name, w[0], tg.Name()) + } + if len(tg.sends) != 0 { + t.Fatalf("%s: sent before its window ended", step.name) + } + } + } + if len(tg.sends) != 1 { + t.Fatalf("sent %d times, want once", len(tg.sends)) + } +} diff --git a/modules/messenger/cmd/messenger/status.go b/modules/messenger/cmd/messenger/status.go index 8680ac07..58831d3f 100644 --- a/modules/messenger/cmd/messenger/status.go +++ b/modules/messenger/cmd/messenger/status.go @@ -214,7 +214,36 @@ func (h *Holder) History(limit int) map[string]any { for i := len(h.refusals) - 1; i >= 0 && len(refusals) < limit; i-- { refusals = append(refusals, h.refusals[i]) } - return map[string]any{"sends": sends, "refusals": refusals} + return map[string]any{"sends": sends, "refusals": refusals, "waiting": h.waitingNow()} +} + +// Waiting is one message held in a channel's outbox, not sent yet: history says it, so a message held for its +// burst window reads as waiting and not as never sent (issue 1507, its reading of 2026-10-28). +type Waiting struct { + Channel string `json:"channel"` + Key string `json:"key"` + What string `json:"what"` + Since string `json:"since"` + // Due is when its burst window ends: it goes out at the first flush from then, unless a rule holds it longer. + Due string `json:"window_ends"` +} + +// waitingNow is every message held in an outbox, oldest first. Called with mu held. +func (h *Holder) waitingNow() []Waiting { + out := []Waiting{} + for ch, list := range h.outbox { + for _, it := range list { + out = append(out, Waiting{Channel: ch, Key: it.Key, What: it.What, + Since: it.At.Format(time.RFC3339Nano), Due: it.At.Add(BurstWindow).Format(time.RFC3339Nano)}) + } + } + sort.SliceStable(out, func(i, j int) bool { + if out[i].Since != out[j].Since { + return out[i].Since < out[j].Since + } + return out[i].Channel < out[j].Channel + }) + return out } // Test sends a test message on one channel or both, through the content rule and the cap.