diff --git a/modules/messenger/cmd/messenger/holder.go b/modules/messenger/cmd/messenger/holder.go index a2dcfaf3..120daa75 100644 --- a/modules/messenger/cmd/messenger/holder.go +++ b/modules/messenger/cmd/messenger/holder.go @@ -1130,6 +1130,8 @@ func (h *Holder) record(s Sent) { if h.Store != nil { if err := h.Store.PutRecent(recent); err != nil { h.noteStore(err) + } else { + h.storeWritten() } } } @@ -1143,6 +1145,20 @@ func (h *Holder) persist(r *Record) { h.mu.Unlock() if err := h.Store.Put(c); err != nil { h.noteStore(err) + } else { + h.storeWritten() + } +} + +// storeWritten clears a state problem once a write succeeds again: the status says what holds now, never a +// failure that passed (issue 1507's reading of 2026-10-28: a verdict kept a single failure of the day before). +func (h *Holder) storeWritten() { + h.mu.Lock() + was := h.storeErr + h.storeErr = "" + h.mu.Unlock() + if was != "" { + h.Logf("[messenger] writes its state again (it could not: %s)", was) } } diff --git a/modules/messenger/cmd/messenger/holder_test.go b/modules/messenger/cmd/messenger/holder_test.go index 980721c3..eff5a92e 100644 --- a/modules/messenger/cmd/messenger/holder_test.go +++ b/modules/messenger/cmd/messenger/holder_test.go @@ -71,9 +71,14 @@ type memStore struct { passes map[string]Pass // puts counts the writes, so a write of bytes identical to what is stored is still seen (issue 422's review). puts int + // fail, when set, refuses every write of a record or of the sends, as the bus does with no responders. + fail error } func (m *memStore) Put(r Record) error { + if m.fail != nil { + return m.fail + } if m.recs == nil { m.recs = map[string]Record{} } @@ -89,7 +94,13 @@ func (m *memStore) All() ([]Record, error) { } return out, nil } -func (m *memStore) PutRecent(s []Sent) error { m.recent = s; return nil } +func (m *memStore) PutRecent(s []Sent) error { + if m.fail != nil { + return m.fail + } + m.recent = s + return nil +} func (m *memStore) PutWalk(w Walk) error { if m.walks == nil { m.walks = map[string]Walk{} @@ -431,15 +442,21 @@ func TestEscalationIsSaidOnce(t *testing.T) { } func TestAChannelThatCannotSendSaysSoAndIsTriedAgain(t *testing.T) { - h, tg, _, c, _ := newHolder(t) + h, tg, _, c, store := newHolder(t) h.Desk = nil tg.fail = errors.New("telegram sendMessage: cannot connect (dial)") + // The state could not be written either, once: said while it lasts, and no longer once a write succeeds + // (issue 1507's reading of 2026-10-28: a verdict kept a day-old "no responders"). + store.fail = errors.New("nats: no responders available for request") h.Condition(EventRaised, cond("machine.ace.silent", Urgent, "silent")) st := h.Status("listening") if !strings.HasPrefix(st.Verdict, "CANNOT SEND") || len(st.Unsent) != 1 || st.Channels[0].LastError == "" { t.Fatalf("status: %+v", st) } - tg.fail = nil + if st.StateProblem == "" { + t.Fatalf("a failed state write is not said: %+v", st) + } + tg.fail, store.fail = nil, nil h.Flush() if len(tg.sends) != 0 { t.Fatalf("tried again before a minute") @@ -449,6 +466,9 @@ func TestAChannelThatCannotSendSaysSoAndIsTriedAgain(t *testing.T) { if len(tg.sends) != 1 || h.Status("listening").Verdict != "ok" { t.Fatalf("not retried: %d, %s", len(tg.sends), h.Status("listening").Verdict) } + if p := h.Status("listening").StateProblem; p != "" { + t.Fatalf("a state write that succeeded again still says the old failure: %s", p) + } } func TestNothingConfiguredIsSaidInTheStatus(t *testing.T) { 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.