2026-10-28 09:18:00 +00:00
5 changed files with 112 additions and 8 deletions
@@ -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)
 	}
 }

@@ -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) {
@@ -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))
+	}
+}
@@ -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.