diff --git a/modules/mesh-mail/cmd/mesh-mail/send.go b/modules/mesh-mail/cmd/mesh-mail/send.go index 8135ade8..46e402fe 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send.go +++ b/modules/mesh-mail/cmd/mesh-mail/send.go @@ -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 diff --git a/modules/mesh-mail/cmd/mesh-mail/send_test.go b/modules/mesh-mail/cmd/mesh-mail/send_test.go index 3e0b0fe0..53311284 100644 --- a/modules/mesh-mail/cmd/mesh-mail/send_test.go +++ b/modules/mesh-mail/cmd/mesh-mail/send_test.go @@ -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") + } +}