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)

This commit is contained in:
2026-10-28 10:02:27 +01:00
parent 3fb7951524
commit 6cea12a023
3 changed files with 73 additions and 5 deletions
+6 -4
View File
@@ -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.",
@@ -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))
}
}
+30 -1
View File
@@ -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.