novox/mesh-catalog
12
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

mesh-mail: a send whose ask was never heard settles as ended, and the sweep reads only undecided sends (issue 811)

Browse Source 
A send whose ask never reached the router (the publish failed, or it was otherwise never recorded) stayed kept and
undecided forever, so mail_send_status kept saying "waiting" and the minute sweep kept reading it. Settle it as
ended once its asked time plus SendAskLasts has passed, with the words that say why. Keep an index of undecided
send ids beside the sends (written when kept, removed once decided) so the sweep's work grows with undecided
sends only, not with every send ever kept.
This commit is contained in:
jochen
2026-10-15 15:27:56 +02:00
parent 2f2daea415
commit 8b2e2300d8
2 changed files with 259 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-mail/cmd/mesh-mail/send.go
+147 -26
View File
@@ -52,6 +52,103 @@ const StaleSendAfter = 10 * time.Minute
// still acted on (novox/hq issue 790, review round 1).
const SweepEvery = time.Minute
// undecidedIndexKey is a reserved key in the `sends` state (never a send id, which always starts "send-") that
// keeps the ids of the sends still undecided, so the sweep can list only those instead of every send ever kept
// (novox/hq issue 811). It stays in the same declared state rather than a state of its own, which a manifest
// change would need.
const undecidedIndexKey = "undecided-index"
// undecidedIndex is the undecidedIndexKey document: the ids of the sends still undecided.
type undecidedIndex struct {
IDs []string `json:"ids"`
}
// addUndecided adds id to the undecided index, retrying its compare-and-set write on a race. Best-effort: the send
// itself is already kept by the time this is called, so a failure here costs the sweep sight of it, never the ask
// (mail_send_status and a direct Heard still decide it).
func addUndecided(store SendStore, id string) error {
for {
raw, rev, ok, err := store.Get(undecidedIndexKey)
if err != nil {
return err
}
var idx undecidedIndex
if ok {
if json.Unmarshal(raw, &idx) != nil {
idx = undecidedIndex{}
}
for _, existing := range idx.IDs {
if existing == id {
return nil
}
}
}
idx.IDs = append(idx.IDs, id)
if !ok {
if _, err := store.Create(undecidedIndexKey, idx); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
}
// removeUndecided drops id from the undecided index, retrying its compare-and-set write on a race. A send never
// read by the index (already absent) is left as it is.
func removeUndecided(store SendStore, id string) error {
for {
raw, rev, ok, err := store.Get(undecidedIndexKey)
if err != nil || !ok {
return err
}
var idx undecidedIndex
if json.Unmarshal(raw, &idx) != nil {
return nil
}
at := -1
for i, existing := range idx.IDs {
if existing == id {
at = i
break
}
}
if at == -1 {
return nil
}
idx.IDs = append(idx.IDs[:at], idx.IDs[at+1:]...)
if _, err := store.Update(undecidedIndexKey, idx, rev); err != nil {
if errors.Is(err, stdio.ErrChanged) {
continue
}
return err
}
return nil
}
}
// undecidedIDs is every id the undecided index currently names, for the sweep to read.
func undecidedIDs(store SendStore) ([]string, error) {
raw, _, ok, err := store.Get(undecidedIndexKey)
if err != nil || !ok {
return nil, err
}
var idx undecidedIndex
if json.Unmarshal(raw, &idx) != nil {
return nil, nil
}
return idx.IDs, nil
}
// The bounds of what mail_send takes.
const (
MostRecipients = 20
@@ -277,6 +374,7 @@ func AskMailSend(in SendInput, store SendStore, bus AskBus, now time.Time) (map[
if _, err := store.Create(req.ID, req); err != nil {
return nil, fmt.Errorf("the send could not be kept, so the operator was not asked: %w. Nothing was sent", err)
}
_ = addUndecided(store, req.ID) // best-effort (see addUndecided); the send is already kept either way
if err := bus.Publish(asks.AskSubject(Module), ask, ask.ID); err != nil {
return nil, fmt.Errorf("the operator could not be asked (%v). Nothing was sent; mail_send_status %s says so", err, req.ID)
}
@@ -308,6 +406,25 @@ type AccountLookup func(name string) (Account, bool)
// called at most once per request (Decide claims the request first).
type SendActor func(a Account, to []string, subject, body string, now time.Time) error
// settleSend writes a decision onto a kept send (claiming it, or ending it outright) and, once the write lands,
// drops it from the undecided index: an index entry is never left behind once the record it points to carries a
// decision (novox/hq issue 811). On a race lost to another writer, it reads back whatever that writer decided.
func settleSend(store SendStore, req SendRequest, rev uint64, d SendDecision) (SendDecision, uint64, error) {
next := req
next.Decided = &d
newRev, err := store.Update(req.ID, next, rev)
if err != nil {
if errors.Is(err, stdio.ErrChanged) {
if again, _, ok, err2 := getSend(store, req.ID); err2 == nil && ok && again.Decided != nil {
return *again.Decided, 0, nil
}
}
return d, 0, err
}
_ = removeUndecided(store, req.ID) // best-effort (see removeUndecided)
return d, newRev, nil
}
// Decide holds one closed ask's warrant to the request it answers (asks.Warrant.For, asks.Option.Performs), and —
// on Approve, exactly once — performs the send. A request already decided is never decided again: decide answers
// its kept decision at once, and two warrants for the same request (a redelivery, or two readers racing) have only
@@ -328,17 +445,8 @@ func Decide(store SendStore, lookup AccountLookup, send SendActor, w asks.Warran
return SendDecision{}, fmt.Errorf("the warrant is for %s's ask %s, not %s's", w.Asker, w.Ask, Module)
}
settle := func(d SendDecision) (SendDecision, error) {
next := req
next.Decided = &d
if _, err := store.Update(req.ID, next, rev); err != nil {
if errors.Is(err, stdio.ErrChanged) {
if again, _, ok2, err2 := getSend(store, req.ID); err2 == nil && ok2 && again.Decided != nil {
return *again.Decided, nil
}
}
return d, err
}
return d, nil
final, _, err := settleSend(store, req, rev, d)
return final, err
}
if w.Outcome != asks.OutcomeChosen {
outcome := "ended: " + string(w.Outcome)
@@ -371,16 +479,15 @@ func Decide(store SendStore, lookup AccountLookup, send SendActor, w asks.Warran
func decideSend(store SendStore, lookup AccountLookup, send SendActor, req SendRequest, rev uint64, w asks.Warrant, now time.Time) (SendDecision, error) {
claimed := req
claim := SendDecision{At: now, Outcome: "sending", Warrant: w.Says()}
claimed.Decided = &claim
newRev, err := store.Update(req.ID, claimed, rev)
claimedDecision, newRev, err := settleSend(store, req, rev, claim)
if err != nil {
if errors.Is(err, stdio.ErrChanged) {
if again, _, ok, err2 := getSend(store, req.ID); err2 == nil && ok && again.Decided != nil {
return *again.Decided, nil
}
}
return SendDecision{}, err
}
if newRev == 0 {
// Lost the claim race: another writer already decided this send.
return claimedDecision, nil
}
claimed.Decided = &claim
acc, found := lookup(req.Account)
var final SendDecision
switch {
@@ -409,7 +516,7 @@ func decideSend(store SendStore, lookup AccountLookup, send SendActor, req SendR
// decided (and so sent), exactly once (Decide).
func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send SendActor, id string, now time.Time) (map[string]any, error) {
trimmed := strings.TrimSpace(id)
req, _, ok, err := getSend(store, trimmed)
req, rev, ok, err := getSend(store, trimmed)
if err != nil {
return nil, fmt.Errorf("the send %q cannot be read now (%v); ask again", trimmed, err)
}
@@ -427,6 +534,17 @@ func ReadSendStatus(store SendStore, bus AskBus, lookup AccountLookup, send Send
return nil, fmt.Errorf("the router's record of the ask cannot be read now (%v); ask again", err)
}
if entry == nil {
// The router never recorded this ask at all (its Publish failed, or it was never heard): once its asked
// time plus SendAskLasts has passed, it never will be, so it is settled as ended rather than left
// "waiting" forever (novox/hq issue 811).
if now.Sub(req.At) > SendAskLasts {
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "ended: never heard"})
if err != nil {
return nil, fmt.Errorf("the send %q could not be settled as ended (%v); ask again", trimmed, err)
}
reportDecision(out, d, now)
return out, nil
}
out["status"], out["words"] = "waiting", "not yet recorded by the router; nothing is sent until the operator answers"
return out, nil
}
@@ -465,6 +583,8 @@ func reportDecision(out map[string]any, d SendDecision, now time.Time) {
"on at " + d.At.Format(time.RFC3339) + ", with no final word since"
case d.Outcome == "sending":
out["status"], out["words"] = "waiting", "the operator's Approve is being acted on; ask again shortly"
case d.Outcome == "ended: never heard":
out["status"], out["words"] = "ended", "its ask to the operator was never heard, so it was not sent"
case strings.HasPrefix(d.Outcome, "failed: "):
out["status"], out["words"] = "failed", "nothing reached the server: "+strings.TrimPrefix(d.Outcome, "failed: ")
case strings.HasPrefix(d.Outcome, "ended: "):
@@ -489,20 +609,21 @@ func (b *Bundle) Heard(w asks.Warrant) {
}
}
// Sweep reads every send this bundle still keeps undecided and asks the router's own record for its answer, the
// same path mail_send_status takes (ReadSendStatus): an Approve recorded while this module did not hear it
// directly (down, or not yet subscribed) is still acted on, once (novox/hq ADR 0361 §2, review round 1 of issue
// 790).
// Sweep reads only the sends the undecided index still names, and asks the router's own record for each one's
// answer, the same path mail_send_status takes (ReadSendStatus): an Approve recorded while this module did not
// hear it directly (down, or not yet subscribed) is still acted on, once. Unlike reading every send ever kept, its
// work grows only with the sends still undecided, never with every one a module has ever sent (novox/hq ADR 0361
// §2, review round 1 of issue 790; the index, issue 811).
func (b *Bundle) Sweep(now time.Time) {
store := b.sends()
keys, err := store.Keys()
ids, err := undecidedIDs(store)
if err != nil {
if b.Logf != nil {
b.Logf("[mesh-mail] the sends kept cannot be listed: %v", err)
b.Logf("[mesh-mail] the undecided index cannot be read: %v", err)
}
return
}
for _, id := range keys {
for _, id := range ids {
req, _, ok, err := getSend(store, id)
if err != nil || !ok || req.Decided != nil {
continue
modules/mesh-mail/cmd/mesh-mail/send_test.go
+112 -5
View File
@@ -7,6 +7,7 @@ package main
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"net"
"net/smtp"
@@ -80,14 +81,15 @@ func (m *memSends) Keys() ([]string, error) {
// sendRouter is the operator channel's holder as this module reaches it, kept by a test: what was asked, and the
// router's own record of each ask.
type sendRouter struct {
mu sync.Mutex
published map[string]asks.Ask
records map[string]json.RawMessage
publishErr error
mu sync.Mutex
published map[string]asks.Ask
records map[string]json.RawMessage
publishErr error
recordCalls map[string]int
}
func newSendRouter() *sendRouter {
return &sendRouter{published: map[string]asks.Ask{}, records: map[string]json.RawMessage{}}
return &sendRouter{published: map[string]asks.Ask{}, records: map[string]json.RawMessage{}, recordCalls: map[string]int{}}
}
func (r *sendRouter) Publish(subject string, body any, id string) error {
@@ -103,9 +105,25 @@ func (r *sendRouter) Publish(subject string, body any, id string) error {
r.records[a.ID], _ = json.Marshal(map[string]any{"state": "open"})
return nil
}
// recordCallsFor is how many times Record was asked about id since the last resetRecordCalls, for a test to check
// the sweep read only the sends still undecided (novox/hq issue 811).
func (r *sendRouter) recordCallsFor(id string) int {
r.mu.Lock()
defer r.mu.Unlock()
return r.recordCalls[id]
}
func (r *sendRouter) resetRecordCalls() {
r.mu.Lock()
defer r.mu.Unlock()
r.recordCalls = map[string]int{}
}
func (r *sendRouter) Record(id string) (*stdio.StateEntry, error) {
r.mu.Lock()
defer r.mu.Unlock()
r.recordCalls[id]++
raw, ok := r.records[id]
if !ok {
return nil, nil
@@ -585,3 +603,92 @@ func TestSendUsesTheParsedAddressForRcptAndTheFormattedAddressForTheToHeader(t *
t.Fatalf("the To header is not the formatted address:\n%s", got.data)
}
}
// (10) A send whose ask was never published (so the router never recorded it) is settled as ended, with its own
// words, once its asked time plus SendAskLasts has passed — never left "waiting" forever (novox/hq issue 811).
func TestASendWhoseAskNeverPublishedEndsAfterTheAskWouldHaveLasted(t *testing.T) {
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
router.publishErr = errors.New("the bus is down")
rt := startBundle(t, b.Tools())
_, isErr := rt.call(SendTool, "novox.triage", sendArgs)
if !isErr {
t.Fatalf("mail_send did not report the publish failure")
}
// The send was kept despite the publish failure (so a later sweep can still find it).
keys, err := sends.Keys()
if err != nil {
t.Fatalf("%v", err)
}
var id string
for _, k := range keys {
if strings.HasPrefix(k, "send-") {
id = k
}
}
if id == "" {
t.Fatalf("no send was kept despite the publish failure: %v", keys)
}
// Before the ask would have lasted, it still reads waiting: the router may yet come up and record it.
text, isErr := rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
if isErr || !strings.Contains(text, `"status":"waiting"`) {
t.Fatalf("before expiry: %v %s", isErr, text)
}
// Once a sweep runs after the ask would have lasted, it settles as ended, and nothing was sent.
b.Sweep(judgedAt.Add(SendAskLasts + time.Minute))
text, isErr = rt.call(SendStatusTool, "novox.triage", map[string]any{"id": id})
if isErr {
t.Fatalf("%s", text)
}
if !strings.Contains(text, `"status":"ended"`) {
t.Fatalf("does not read ended once the ask would have lasted: %s", text)
}
if !strings.Contains(text, "its ask to the operator was never heard, so it was not sent") {
t.Fatalf("does not say why: %s", text)
}
}
// (11) The sweep reads only the sends still undecided, by the undecided index, not every send ever kept: a send
// already decided is never asked about again (novox/hq issue 811).
func TestSweepReadsOnlyUndecidedSends(t *testing.T) {
fake := newFakeSMTPServer(t)
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
b.Dial = fake.dialer()
rt := startBundle(t, b.Tools())
// One send, declined right away: decided, so it leaves the undecided index.
text, isErr := rt.call(SendTool, "novox.triage", sendArgs)
if isErr {
t.Fatalf("%s", text)
}
var out map[string]any
_ = json.Unmarshal([]byte(text), &out)
declinedID := out["ask"].(string)
w := router.chose(declinedID, OptionDecline, judgedAt.Add(time.Minute))
router.close(w)
if _, err := Decide(sends, b.lookupAccount, b.sendActor(), w, judgedAt.Add(2*time.Minute)); err != nil {
t.Fatalf("deciding: %v", err)
}
// A second send, left undecided: still waiting on the operator.
text, isErr = rt.call(SendTool, "novox.triage", sendArgs)
if isErr {
t.Fatalf("%s", text)
}
_ = json.Unmarshal([]byte(text), &out)
waitingID := out["ask"].(string)
router.resetRecordCalls()
b.Sweep(judgedAt.Add(3 * time.Minute))
if n := router.recordCallsFor(declinedID); n != 0 {
t.Fatalf("the sweep read the decided send's router record %d time(s): it already left the undecided index", n)
}
if n := router.recordCallsFor(waitingID); n == 0 {
t.Fatalf("the sweep never read the undecided send's router record")
}
}