mesh-mail: tell the operator once for a failed approved send and once more if its sender has not taken it, make an overdue send unhealthy, and give mail_send_status its full vocabulary (ADR 0436 clauses 11, 12; issue 1199)
This commit is contained in:
@@ -105,6 +105,24 @@ func Judge(accounts []Account, accountsErr, readyErr error, given func(Account)
|
||||
return health.Healthy(fmt.Sprintf("%d account(s) watched: %s", len(accounts), strings.Join(fine, "; ")))
|
||||
}
|
||||
|
||||
// WithOverdue folds any overdue send (Overdue, send.go) into base: unhealthy, naming each one, whatever
|
||||
// base itself said — an overdue send is never excused by an otherwise-healthy account (novox/hq ADR 0436
|
||||
// clause 11, issue 1199).
|
||||
func WithOverdue(base health.Answer, overdue []SendRequest) health.Answer {
|
||||
if len(overdue) == 0 {
|
||||
return base
|
||||
}
|
||||
names := make([]string, 0, len(overdue))
|
||||
for _, req := range overdue {
|
||||
names = append(names, fmt.Sprintf("%s (from %s)", req.ID, req.Account))
|
||||
}
|
||||
why := fmt.Sprintf("the send(s) %s have waited 10 minutes or more for the sender", strings.Join(names, ", "))
|
||||
if base.Why != "" {
|
||||
why = base.Why + "; and " + why
|
||||
}
|
||||
return health.Unhealthy(why, base.Waits...)
|
||||
}
|
||||
|
||||
// Status is the watch's verdict as it leaves it: the answer, and when it looked.
|
||||
type Status struct {
|
||||
health.Answer
|
||||
|
||||
@@ -191,12 +191,45 @@ func serve(env Env) {
|
||||
go p.Run(context.Background())
|
||||
go hear(b)
|
||||
go sweepSends(b)
|
||||
go waitsTellTicker(b)
|
||||
if err := stdio.Serve("", b.Tools()); err != nil {
|
||||
logf("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// sdkTellOperator is the tools' own TellOperator (send.go), wrapping stdio.Ask to the operator channel's
|
||||
// tell verb, exactly as mesh-files' own pattern (modules/mesh-files/cmd/mesh-files/main.go tellOperator):
|
||||
// no lines, no subject naming anything the content rule would refuse.
|
||||
func sdkTellOperator(headline string) error {
|
||||
_, err := stdio.Ask("seat:operator-channel.tell", map[string]any{"headline": headline, "lines": []string{}, "subject": "mesh-mail"})
|
||||
return err
|
||||
}
|
||||
|
||||
// WaitsTellEvery is how often the tools look for an overdue send still owing their own "waits" tell
|
||||
// (novox/hq ADR 0436 clause 11, issue 1199): a second ticker, apart from sweepSends, since the two tells —
|
||||
// the sender's own failure, the tools' own overdue wait — are never the same look.
|
||||
const WaitsTellEvery = time.Minute
|
||||
|
||||
// waitsTellTicker runs the tools' own "waits" tell for every overdue send, at least every WaitsTellEvery,
|
||||
// until the process ends (novox/hq ADR 0436 clause 11, issue 1199).
|
||||
func waitsTellTicker(b *Bundle) {
|
||||
tick := time.NewTicker(WaitsTellEvery)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
overdue, err := Overdue(b.sends(), time.Now())
|
||||
if err != nil {
|
||||
logf("[mesh-mail] the overdue check cannot be read: %v", err)
|
||||
}
|
||||
for _, req := range overdue {
|
||||
if err := tellWaits(b.sends(), sdkTellOperator, req.ID, time.Now()); err != nil {
|
||||
logf("[mesh-mail] telling the operator that %s waits: %v", req.ID, err)
|
||||
}
|
||||
}
|
||||
<-tick.C
|
||||
}
|
||||
}
|
||||
|
||||
// hear subscribes to the router's word on this module's own asks, asked again until the runtime takes it, exactly
|
||||
// as mesh-knowledge does. Each warrant heard goes to Heard, the same Decide that mail_send_status uses: an
|
||||
// approved send goes out as soon as the operator answers it, never only when something later calls
|
||||
|
||||
@@ -192,6 +192,14 @@ type SendDecision struct {
|
||||
// NextAttempt is when a "retrying" send is next claimed, the bus's own redelivery time (novox/hq ADR
|
||||
// 0436 clause 10, issue 1198): read by mail_health to say a send overdue (issue 1199).
|
||||
NextAttempt time.Time `json:"next-attempt,omitempty"`
|
||||
// Tell is the sender's own tell to the operator for a failed, refused or not-proven send, kept to once:
|
||||
// "" (not owed, or not yet attempted), "telling" (claimed, the channel not yet confirmed it took the
|
||||
// message) or "told" (novox/hq ADR 0436 clause 11, issue 1199).
|
||||
Tell string `json:"tell,omitempty"`
|
||||
// WaitsTell is the tools' own tell that an overdue send's sender has not taken it, kept to once the
|
||||
// same way as Tell, on its own field: the two are never the same tell (novox/hq ADR 0436 clause 11,
|
||||
// issue 1199).
|
||||
WaitsTell string `json:"waits-tell,omitempty"`
|
||||
}
|
||||
|
||||
// bodyDigest is the digest an act binds in place of the body itself: never the body, which the act's digest is
|
||||
@@ -696,6 +704,10 @@ func publishQueued(store SendStore, req SendRequest, rev uint64, approved SendDe
|
||||
}
|
||||
if newRev != 0 {
|
||||
_ = removeUndecided(store, req.ID) // from here the sender owns it; the tools' sweep is done with it
|
||||
// mail_health's own overdue check (issue 1199) needs to find every send the sender now owns, the
|
||||
// same way the sweep needed the undecided index: a reserved key, since neither side's own state
|
||||
// over mesh-sdk go/process lists its keys whole.
|
||||
_ = addSenderPending(store, req.ID)
|
||||
}
|
||||
return final, nil
|
||||
}
|
||||
@@ -743,17 +755,26 @@ func claimQueuedSend(store SendStore, record RouterRecord, id string, now time.T
|
||||
if !ok {
|
||||
return SendRequest{}, SendDecision{}, 0, claimLostRace, fmt.Errorf("no send %q is kept", id)
|
||||
}
|
||||
if req.Decided == nil {
|
||||
// Nothing was ever approved: no tell owed (clause 11 ties the tell to a *failed approved* send).
|
||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: the send is not yet approved"})
|
||||
return req, d, 0, claimRefused, err
|
||||
}
|
||||
// Every other refusal below follows a real Approve (the send reached at least queued): clause 9's own
|
||||
// refusal owes the operator a tell, same as clause 10's (novox/hq ADR 0436 clause 11, issue 1199).
|
||||
refuse := func(reason string) (SendRequest, SendDecision, uint64, claimOutcome, error) {
|
||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason})
|
||||
_ = removeSenderPending(store, req.ID)
|
||||
_ = addTellPending(store, req.ID)
|
||||
return req, d, 0, claimRefused, err
|
||||
}
|
||||
switch {
|
||||
case req.Decided == nil:
|
||||
return refuse("the send is not yet approved")
|
||||
case req.Decided.Outcome == "sending":
|
||||
// A previous delivery may have handed it to the server already; never sent twice, and never
|
||||
// retried automatically either way (ADR 0436 clause 10).
|
||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt})
|
||||
_ = removeSenderPending(store, req.ID)
|
||||
_ = addTellPending(store, req.ID)
|
||||
return req, d, 0, claimNotProven, err
|
||||
case req.Decided.Outcome != "queued" && req.Decided.Outcome != "retrying":
|
||||
return refuse(fmt.Sprintf("the send is %q, neither queued nor retrying", req.Decided.Outcome))
|
||||
@@ -849,20 +870,28 @@ const (
|
||||
// with the message next. Attempt (req's own, from the claim) decides retrying from failed on a not-taken
|
||||
// failure: the fifth attempt always fails rather than retries again (novox/hq ADR 0436 clauses 10, 11).
|
||||
func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDecision, rev uint64, sendErr error, now time.Time) (SendDecision, msgAction, time.Duration) {
|
||||
write := func(d SendDecision) SendDecision {
|
||||
final, _, err := settleSend(store, req, rev, d)
|
||||
write := func(d SendDecision, final bool) SendDecision {
|
||||
result, _, err := settleSend(store, req, rev, d)
|
||||
if final {
|
||||
// From here the sender is done with it, one way or another (novox/hq issue 1199): mail_health's
|
||||
// overdue check no longer needs to find it.
|
||||
_ = removeSenderPending(store, req.ID)
|
||||
}
|
||||
if err != nil {
|
||||
// The attempt (and its SMTP outcome) stand either way; the caller logs this. A redelivery (or the
|
||||
// sweep, for a send stuck mid-write) finds whatever the real kept state is and reconciles.
|
||||
return d
|
||||
}
|
||||
return final
|
||||
return result
|
||||
}
|
||||
say := func(outcome string) {
|
||||
tell := func(outcome string) {
|
||||
// Owed a tell (failed, refused or not proven, novox/hq ADR 0436 clause 11): the index the sender's
|
||||
// own look reads, kept to once by tell: telling then told (issue 1199).
|
||||
_ = addTellPending(store, req.ID)
|
||||
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: outcome})
|
||||
}
|
||||
if sendErr == nil {
|
||||
d := write(SendDecision{At: now, Outcome: "sent", Warrant: claimed.Warrant, Attempt: claimed.Attempt})
|
||||
d := write(SendDecision{At: now, Outcome: "sent", Warrant: claimed.Warrant, Attempt: claimed.Attempt}, true)
|
||||
_ = emit(MailSendSent, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
|
||||
return d, actionAck, 0
|
||||
}
|
||||
@@ -876,28 +905,302 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
|
||||
switch se.Outcome {
|
||||
case Refused:
|
||||
d := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
|
||||
Attempt: claimed.Attempt, Error: se.Err.Error()})
|
||||
say(d.Outcome)
|
||||
Attempt: claimed.Attempt, Error: se.Err.Error()}, true)
|
||||
tell(d.Outcome)
|
||||
return d, actionTerm, 0
|
||||
case NotProven:
|
||||
d := write(SendDecision{At: now, Outcome: "not proven", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
|
||||
Error: se.Err.Error()})
|
||||
say(d.Outcome)
|
||||
Error: se.Err.Error()}, true)
|
||||
tell(d.Outcome)
|
||||
return d, actionTerm, 0
|
||||
default: // NotTaken
|
||||
if claimed.Attempt >= MostAttempts {
|
||||
d := write(SendDecision{At: now, Outcome: "failed: " + se.Err.Error(), Warrant: claimed.Warrant,
|
||||
Attempt: claimed.Attempt, Error: se.Err.Error()})
|
||||
say(d.Outcome)
|
||||
Attempt: claimed.Attempt, Error: se.Err.Error()}, true)
|
||||
tell(d.Outcome)
|
||||
return d, actionTerm, 0
|
||||
}
|
||||
delay := RetryDelays[claimed.Attempt-1]
|
||||
d := write(SendDecision{At: now, Outcome: "retrying", Warrant: claimed.Warrant, Attempt: claimed.Attempt,
|
||||
Error: se.Err.Error(), NextAttempt: now.Add(delay)})
|
||||
Error: se.Err.Error(), NextAttempt: now.Add(delay)}, false)
|
||||
return d, actionRetry, delay // not a final outcome: no emit (clause 11 names only sent and failed)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- two more indices (novox/hq issue 1199), the same shape as the undecided index (issue 811) ------------------
|
||||
//
|
||||
// Neither the tools' nor the sender's own `sends` state lists every key it holds (mesh-sdk go/process's
|
||||
// State offers no such call, and listing it all on every health check or every look would grow with every
|
||||
// send ever sent, not with the few still live): a reserved key stands in for a list, exactly as the
|
||||
// undecided index already does.
|
||||
|
||||
// idIndex is the shape a send-id index keeps: generic across all three (undecided, sender-pending,
|
||||
// tells-pending).
|
||||
type idIndex struct {
|
||||
IDs []string `json:"ids"`
|
||||
}
|
||||
|
||||
// addToIndex adds id to the index at key, retrying its compare-and-set write on a race. A stored index
|
||||
// that cannot be parsed is never overwritten (which would silently drop every id it already named): it is
|
||||
// reported as an error naming the key instead (novox/hq issue 811, review round 3).
|
||||
func addToIndex(store SendStore, key, id string) error {
|
||||
for {
|
||||
raw, rev, ok, err := store.Get(key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var idx idIndex
|
||||
if ok {
|
||||
if err := json.Unmarshal(raw, &idx); err != nil {
|
||||
return fmt.Errorf("the index %q cannot be parsed: %w", key, err)
|
||||
}
|
||||
for _, existing := range idx.IDs {
|
||||
if existing == id {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
idx.IDs = append(idx.IDs, id)
|
||||
if !ok {
|
||||
if _, err := store.Create(key, idx); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if _, err := store.Update(key, idx, rev); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// removeFromIndex drops id from the index at key, retrying its compare-and-set write on a race. An id
|
||||
// never in the index (already absent) is left as it is.
|
||||
func removeFromIndex(store SendStore, key, id string) error {
|
||||
for {
|
||||
raw, rev, ok, err := store.Get(key)
|
||||
if err != nil || !ok {
|
||||
return err
|
||||
}
|
||||
var idx idIndex
|
||||
if err := json.Unmarshal(raw, &idx); err != nil {
|
||||
return fmt.Errorf("the index %q cannot be parsed: %w", key, err)
|
||||
}
|
||||
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(key, idx, rev); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
// indexIDs is every id the index at key currently names.
|
||||
func indexIDs(store SendStore, key string) ([]string, error) {
|
||||
raw, _, ok, err := store.Get(key)
|
||||
if err != nil || !ok {
|
||||
return nil, err
|
||||
}
|
||||
var idx idIndex
|
||||
if err := json.Unmarshal(raw, &idx); err != nil {
|
||||
return nil, fmt.Errorf("the index %q cannot be parsed: %w", key, err)
|
||||
}
|
||||
return idx.IDs, nil
|
||||
}
|
||||
|
||||
// senderPendingKey indexes every send the sender currently owns — queued, retrying or sending — so
|
||||
// mail_health's overdue check (Overdue, below) finds them without listing `sends` whole.
|
||||
const senderPendingKey = "sender-pending"
|
||||
|
||||
func addSenderPending(store SendStore, id string) error {
|
||||
return addToIndex(store, senderPendingKey, id)
|
||||
}
|
||||
func removeSenderPending(store SendStore, id string) error {
|
||||
return removeFromIndex(store, senderPendingKey, id)
|
||||
}
|
||||
func senderPendingIDs(store SendStore) ([]string, error) { return indexIDs(store, senderPendingKey) }
|
||||
|
||||
// tellsPendingKey indexes every send still owing the sender's own tell (needsTell, below), so its own
|
||||
// look (sender.go) finds them without listing `sends` whole.
|
||||
const tellsPendingKey = "tells-pending"
|
||||
|
||||
func addTellPending(store SendStore, id string) error { return addToIndex(store, tellsPendingKey, id) }
|
||||
func removeTellPending(store SendStore, id string) error {
|
||||
return removeFromIndex(store, tellsPendingKey, id)
|
||||
}
|
||||
func tellPendingIDs(store SendStore) ([]string, error) { return indexIDs(store, tellsPendingKey) }
|
||||
|
||||
// ---- overdue (novox/hq ADR 0436 clause 11, issue 1199) ------------------------------------------------------
|
||||
|
||||
// OverdueAfter is how long a send may sit queued, or wait past its own next attempt, before mail_health
|
||||
// calls it unhealthy (novox/hq ADR 0436 clause 11).
|
||||
const OverdueAfter = 10 * time.Minute
|
||||
|
||||
// Overdue is every send the sender currently owns (senderPendingIDs) that is overdue: queued more than
|
||||
// OverdueAfter ago, or retrying more than OverdueAfter past its own next attempt. A send waiting for its
|
||||
// next attempt within that window, or one actually being sent, is never overdue (novox/hq ADR 0436 clause
|
||||
// 11).
|
||||
func Overdue(store SendStore, now time.Time) ([]SendRequest, error) {
|
||||
ids, err := senderPendingIDs(store)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []SendRequest
|
||||
for _, id := range ids {
|
||||
req, _, ok, err := getSend(store, id)
|
||||
if err != nil || !ok || req.Decided == nil {
|
||||
continue
|
||||
}
|
||||
switch {
|
||||
case req.Decided.Outcome == "queued" && now.Sub(req.Decided.At) > OverdueAfter:
|
||||
out = append(out, req)
|
||||
case req.Decided.Outcome == "retrying" && now.Sub(req.Decided.NextAttempt) > OverdueAfter:
|
||||
out = append(out, req)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ---- telling the operator (novox/hq ADR 0436 clause 11, issue 1199) ----------------------------------------
|
||||
//
|
||||
// Two separate tells, each kept to once on its own field: the sender's own, for a failed, refused or
|
||||
// not-proven send (Tell); the tools' own, for an overdue one — still queued or retrying, its sender not
|
||||
// having taken it yet (WaitsTell). Neither ever names an address, held to the operator channel's content
|
||||
// rule by using only the account's own name.
|
||||
|
||||
// TellOperator calls seat:operator-channel.tell once with the one line to say (novox/hq ADR 0380 §8). The
|
||||
// real ones wrap a process.Conn's own Call (sender.go) or stdio.Ask (main.go); a test gives its own.
|
||||
type TellOperator func(headline string) error
|
||||
|
||||
// needsTell says whether a decided send's outcome owes the operator the sender's own tell (novox/hq ADR
|
||||
// 0436 clause 11): every outcome after the Approve but sent — failed, refused or not proven.
|
||||
func needsTell(outcome string) bool {
|
||||
return outcome == "not proven" || strings.HasPrefix(outcome, "failed: ") || strings.HasPrefix(outcome, "refused: ")
|
||||
}
|
||||
|
||||
// tellReason is the sender's own one line for a failed, refused or not-proven send (novox/hq ADR 0436
|
||||
// clause 11).
|
||||
func tellReason(account, outcome string) string {
|
||||
if outcome == "not proven" {
|
||||
return fmt.Sprintf("The mail you approved from %s may not have been sent; its sent folder says", account)
|
||||
}
|
||||
reason := outcome
|
||||
for _, prefix := range []string{"failed: ", "refused: "} {
|
||||
if r, ok := strings.CutPrefix(outcome, prefix); ok {
|
||||
reason = r
|
||||
break
|
||||
}
|
||||
}
|
||||
return fmt.Sprintf("The mail you approved from %s was not sent: %s", account, reason)
|
||||
}
|
||||
|
||||
// tellSend attempts the sender's own tell for one failed, refused or not-proven send, kept to once by
|
||||
// Tell: telling then told, by compare-and-set, on the record itself (novox/hq ADR 0436 clause 11). A tell
|
||||
// the channel refuses, or that nothing answers, leaves telling, so the next look (the sender's own ticker,
|
||||
// at least every 10 minutes) tries it again; a record already told, or whose outcome no longer needs one,
|
||||
// is dropped from the index without being told again.
|
||||
func tellSend(store SendStore, tell TellOperator, id string, now time.Time) error {
|
||||
req, rev, ok, err := getSend(store, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok || req.Decided == nil || !needsTell(req.Decided.Outcome) || req.Decided.Tell == "told" {
|
||||
_ = removeTellPending(store, id)
|
||||
return nil
|
||||
}
|
||||
d := *req.Decided
|
||||
if d.Tell != "telling" {
|
||||
d.Tell = "telling"
|
||||
next := req
|
||||
next.Decided = &d
|
||||
newRev, err := store.Update(id, next, rev)
|
||||
if err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return nil // raced with another writer; the next look sorts it out
|
||||
}
|
||||
return err
|
||||
}
|
||||
rev = newRev
|
||||
}
|
||||
if err := tell(tellReason(req.Account, d.Outcome)); err != nil {
|
||||
return nil // left telling; tried again on the next look
|
||||
}
|
||||
d.Tell = "told"
|
||||
next := req
|
||||
next.Decided = &d
|
||||
if _, err := store.Update(id, next, rev); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
_ = removeTellPending(store, id)
|
||||
return nil
|
||||
}
|
||||
|
||||
// waitsReason is the tools' own one line for an overdue send (novox/hq ADR 0436 clause 11).
|
||||
func waitsReason(account string) string {
|
||||
return fmt.Sprintf("The mail you approved from %s waits: its sender has not taken it", account)
|
||||
}
|
||||
|
||||
// tellWaits attempts the tools' own "waits" tell for one overdue send, kept to once by WaitsTell — the
|
||||
// same telling/told sequence as tellSend, on the record's own separate field, since the two tells are
|
||||
// never the same one (novox/hq ADR 0436 clause 11).
|
||||
func tellWaits(store SendStore, tell TellOperator, id string, now time.Time) error {
|
||||
req, rev, ok, err := getSend(store, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok || req.Decided == nil || (req.Decided.Outcome != "queued" && req.Decided.Outcome != "retrying") || req.Decided.WaitsTell == "told" {
|
||||
return nil
|
||||
}
|
||||
d := *req.Decided
|
||||
if d.WaitsTell != "telling" {
|
||||
d.WaitsTell = "telling"
|
||||
next := req
|
||||
next.Decided = &d
|
||||
newRev, err := store.Update(id, next, rev)
|
||||
if err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
rev = newRev
|
||||
}
|
||||
if err := tell(waitsReason(req.Account)); err != nil {
|
||||
return nil
|
||||
}
|
||||
d.WaitsTell = "told"
|
||||
next := req
|
||||
next.Decided = &d
|
||||
if _, err := store.Update(id, next, rev); err != nil {
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- reading the status -----------------------------------------------------------------------------------------
|
||||
|
||||
// ReadSendStatus is where a send asked with mail_send stands, by its ask id: waiting for the operator, sent,
|
||||
@@ -960,15 +1263,16 @@ func ReadSendStatus(store SendStore, bus AskBus, publish QueuePublish, id string
|
||||
// and the words that explain it. now is when it is asked, not when the send was decided: a claim still "sending"
|
||||
// longer than StaleSendAfter with no final write is "not proven" (the send may or may not have gone out), never
|
||||
// "waiting" forever — and it is never retried automatically either way (novox/hq issue 790, review round 1).
|
||||
// reportDecision's status word is one of: waiting (for the operator, or for the sender — approved, queued,
|
||||
// being sent or retrying, each said apart in words), sent, failed, refused, not proven, declined or expired
|
||||
// (novox/hq ADR 0436 clause 12, issue 1199) — the whole vocabulary mail_send_status answers.
|
||||
func reportDecision(out map[string]any, d SendDecision, now time.Time) {
|
||||
out["at"] = d.At.Format(time.RFC3339)
|
||||
switch {
|
||||
case d.Outcome == "approved":
|
||||
// Claimed, not yet queued (the publish to mail-sends is still to do, or failed and the sweep will try it
|
||||
// again): novox/hq ADR 0436 clause 8. The fuller status enumeration is issue 1199's to settle.
|
||||
case d.Outcome == "approved", d.Outcome == "queued":
|
||||
// Claimed, and either not yet published to mail-sends (the sweep will try it again) or already
|
||||
// queued for the sender: both read the same to the operator (novox/hq ADR 0436 clause 8).
|
||||
out["status"], out["words"] = "waiting", "approved and waiting for the sender; ask again shortly"
|
||||
case d.Outcome == "queued":
|
||||
out["status"], out["words"] = "waiting", "queued for the sender; ask again shortly"
|
||||
case d.Outcome == "sent":
|
||||
out["status"], out["words"] = "sent", "sent on the operator's Approve ("+d.Warrant+")"
|
||||
case d.Outcome == "declined":
|
||||
@@ -978,17 +1282,23 @@ func reportDecision(out map[string]any, d SendDecision, now time.Time) {
|
||||
out["words"] = "not proven: the send may or may not have gone out; the operator's Approve was being acted " +
|
||||
"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"
|
||||
out["status"], out["words"] = "waiting", "being sent; ask again shortly"
|
||||
case d.Outcome == "retrying":
|
||||
out["status"] = "waiting"
|
||||
out["words"] = "waiting for its next attempt, at " + d.NextAttempt.Format(time.RFC3339) + ": " + d.Error
|
||||
case d.Outcome == "not proven":
|
||||
out["status"] = "not proven"
|
||||
out["words"] = "not proven: the send may or may not have gone out: " + d.Error
|
||||
case d.Outcome == "ended: never heard":
|
||||
out["status"], out["words"] = "ended", "its ask to the operator was never heard, so it was not sent"
|
||||
out["status"], out["words"] = "expired", "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: "):
|
||||
out["status"], out["words"] = "ended", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "ended: ")
|
||||
out["status"], out["words"] = "expired", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "ended: ")
|
||||
case strings.HasPrefix(d.Outcome, "refused: "):
|
||||
out["status"], out["words"] = "not proven", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "refused: ")
|
||||
out["status"], out["words"] = "refused", "nothing was sent: "+strings.TrimPrefix(d.Outcome, "refused: ")
|
||||
default:
|
||||
out["status"], out["words"] = "ended", d.Outcome
|
||||
out["status"], out["words"] = "expired", d.Outcome
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -737,8 +737,8 @@ func TestASendWhoseAskNeverPublishedEndsAfterTheAskWouldHaveLasted(t *testing.T)
|
||||
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, `"status":"expired"`) {
|
||||
t.Fatalf("does not read expired 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)
|
||||
|
||||
@@ -98,8 +98,10 @@ func send(env Env) int {
|
||||
Accounts: func() ([]Account, error) { return ReadAccounts(env.Accounts) },
|
||||
SecretsDir: env.Secrets,
|
||||
Emit: emitOf(conn),
|
||||
Tell: tellOperatorOf(conn),
|
||||
Logf: logf,
|
||||
}
|
||||
go h.tellTicker(ctx)
|
||||
logf("[mesh-mail] send is consuming %s as %s", MailSendsStream, account.User)
|
||||
if err := conn.Consume(ctx, MailSendsStream, h.handle); err != nil && ctx.Err() == nil {
|
||||
logf("[mesh-mail] send stopped: %v", err)
|
||||
@@ -108,6 +110,49 @@ func send(env Env) int {
|
||||
return 0
|
||||
}
|
||||
|
||||
// tellOperatorOf is the real TellOperator (send.go), wrapping conn's own Call to the operator channel's
|
||||
// tell verb (novox/hq ADR 0380 §8, ADR 0436 clause 11): no lines, no subject naming anything the content
|
||||
// rule would refuse — the headline alone carries the one line.
|
||||
func tellOperatorOf(conn *process.Conn) TellOperator {
|
||||
return func(headline string) error {
|
||||
_, err := conn.Call("seat:operator-channel.tell", map[string]any{"headline": headline, "lines": []string{}, "subject": "mesh-mail"})
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// TellEvery is how often the sender looks for a send still owing its own tell (novox/hq ADR 0436 clause
|
||||
// 11: "at least every 10 minutes").
|
||||
const TellEvery = 10 * time.Minute
|
||||
|
||||
// tellTicker runs every send still owing the sender's own tell at least every TellEvery, until ctx ends
|
||||
// (novox/hq ADR 0436 clause 11, issue 1199): once right away, so a tell left over a restart is not made to
|
||||
// wait a full interval.
|
||||
func (h *senderHandler) tellTicker(ctx context.Context) {
|
||||
t := time.NewTicker(TellEvery)
|
||||
defer t.Stop()
|
||||
for {
|
||||
h.tellPending()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *senderHandler) tellPending() {
|
||||
ids, err := tellPendingIDs(h.Store)
|
||||
if err != nil {
|
||||
h.logf("[mesh-mail] send: the tells-pending index cannot be read: %v", err)
|
||||
return
|
||||
}
|
||||
for _, id := range ids {
|
||||
if err := tellSend(h.Store, h.Tell, id, h.now()); err != nil {
|
||||
h.logf("[mesh-mail] send: telling the operator about %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runSend connects as the sender's own account and consumes mail-sends until ctx ends, handing each message
|
||||
// to handle. Apart from send so a test can drive it over a bus of its own, with a handler of its own
|
||||
// (novox/hq issue 1195, AC2): on ctx ending (a signal), Consume stops pulling and returns without
|
||||
@@ -140,6 +185,7 @@ type senderHandler struct {
|
||||
SecretsDir string
|
||||
Dial SMTPDialer
|
||||
Emit Emit
|
||||
Tell TellOperator
|
||||
Now func() time.Time
|
||||
Logf func(format string, a ...any)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
package main
|
||||
|
||||
// The tells to the operator, overdue health and mail_send_status's words (novox/hq ADR 0436 clauses 11,
|
||||
// 12, issue 1199).
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// memTell collects every headline a TellOperator was called with, optionally refusing the first n calls
|
||||
// (novox/hq issue 1199).
|
||||
type memTell struct {
|
||||
headlines []string
|
||||
refuseN int
|
||||
}
|
||||
|
||||
func (m *memTell) tell(headline string) error {
|
||||
if m.refuseN > 0 {
|
||||
m.refuseN--
|
||||
return errors.New("the channel is down")
|
||||
}
|
||||
m.headlines = append(m.headlines, headline)
|
||||
return nil
|
||||
}
|
||||
|
||||
func tellTestSend(t *testing.T, sends *memSends, outcome string) SendRequest {
|
||||
t.Helper()
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, At: judgedAt}
|
||||
req.Decided = &SendDecision{At: judgedAt, Outcome: outcome}
|
||||
if _, err := sends.Create(req.ID, req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := addTellPending(sends, req.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return req
|
||||
}
|
||||
|
||||
// 1199/AC1: given a send recorded failed, refused or not proven after the operator's Approve, when the
|
||||
// sender looks, then it sets tell to telling by compare-and-set, calls seat:operator-channel.tell once
|
||||
// (the headline naming the account and the reason, or for not proven that it may not have been sent; no
|
||||
// address in it), and on the channel taking it sets told. A refused or unanswered tell leaves telling, and
|
||||
// the next look tells only records left telling. A sent send is never told.
|
||||
func TestTellSendTellsOnceAndSettlesToldOnlyOnceTheChannelTakesIt(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := tellTestSend(t, sends, "failed: 550 no such mailbox")
|
||||
tell := &memTell{}
|
||||
|
||||
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(time.Minute)); err != nil {
|
||||
t.Fatalf("telling: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 1 {
|
||||
t.Fatalf("told %d time(s): %v", len(tell.headlines), tell.headlines)
|
||||
}
|
||||
h := tell.headlines[0]
|
||||
if !strings.Contains(h, "mesh") || !strings.Contains(h, "550 no such mailbox") {
|
||||
t.Fatalf("headline: %q", h)
|
||||
}
|
||||
if strings.Contains(h, "@") {
|
||||
t.Fatalf("the headline carries an address: %q", h)
|
||||
}
|
||||
kept, _, ok, err := getSend(sends, req.ID)
|
||||
if err != nil || !ok || kept.Decided == nil || kept.Decided.Tell != "told" {
|
||||
t.Fatalf("tell not settled told: %v %v %+v", err, ok, kept.Decided)
|
||||
}
|
||||
ids, err := tellPendingIDs(sends)
|
||||
if err != nil || len(ids) != 0 {
|
||||
t.Fatalf("still indexed after being told: %v %v", err, ids)
|
||||
}
|
||||
|
||||
// A second look at an already-told send tells nothing more.
|
||||
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(2*time.Minute)); err != nil {
|
||||
t.Fatalf("telling again: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 1 {
|
||||
t.Fatalf("told again: %v", tell.headlines)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTellSendLeavesTellingOnARefusedChannelAndTriesAgain(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := tellTestSend(t, sends, "not proven")
|
||||
tell := &memTell{refuseN: 1}
|
||||
|
||||
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(time.Minute)); err != nil {
|
||||
t.Fatalf("telling: %v", err)
|
||||
}
|
||||
kept, _, ok, err := getSend(sends, req.ID)
|
||||
if err != nil || !ok || kept.Decided == nil || kept.Decided.Tell != "telling" {
|
||||
t.Fatalf("not left telling: %v %v %+v", err, ok, kept.Decided)
|
||||
}
|
||||
ids, err := tellPendingIDs(sends)
|
||||
if err != nil || len(ids) != 1 {
|
||||
t.Fatalf("no longer indexed despite the refusal: %v %v", err, ids)
|
||||
}
|
||||
|
||||
// The next look succeeds.
|
||||
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(2*time.Minute)); err != nil {
|
||||
t.Fatalf("telling again: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 1 {
|
||||
t.Fatalf("told: %v", tell.headlines)
|
||||
}
|
||||
if !strings.Contains(tell.headlines[0], "may not have been sent") {
|
||||
t.Fatalf("not-proven headline: %q", tell.headlines[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestTellSendNeverTellsForASentSend(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", At: judgedAt}
|
||||
req.Decided = &SendDecision{At: judgedAt, Outcome: "sent"}
|
||||
if _, err := sends.Create(req.ID, req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tell := &memTell{}
|
||||
if err := tellSend(sends, tell.tell, req.ID, judgedAt.Add(time.Minute)); err != nil {
|
||||
t.Fatalf("telling: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 0 {
|
||||
t.Fatalf("a sent send was told: %v", tell.headlines)
|
||||
}
|
||||
}
|
||||
|
||||
// 1199/AC2: given a send queued more than 10 minutes ago, or retrying 10 minutes past its next attempt,
|
||||
// when mail_health runs (Overdue), then it names the send; a send waiting for its next attempt is never
|
||||
// overdue.
|
||||
func TestOverdueFindsAQueuedOrRetryingSendPastTenMinutes(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
queueTestSend := func(outcome string, at, next time.Time) string {
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", At: judgedAt}
|
||||
req.Decided = &SendDecision{At: at, Outcome: outcome, NextAttempt: next}
|
||||
if _, err := sends.Create(req.ID, req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := addSenderPending(sends, req.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return req.ID
|
||||
}
|
||||
now := judgedAt
|
||||
overdueQueued := queueTestSend("queued", now.Add(-11*time.Minute), time.Time{})
|
||||
freshQueued := queueTestSend("queued", now.Add(-5*time.Minute), time.Time{})
|
||||
overdueRetrying := queueTestSend("retrying", now.Add(-20*time.Minute), now.Add(-11*time.Minute))
|
||||
notYetDueRetrying := queueTestSend("retrying", now.Add(-5*time.Minute), now.Add(3*time.Minute))
|
||||
sendingNotOverdue := queueTestSend("sending", now.Add(-20*time.Minute), time.Time{})
|
||||
|
||||
overdue, err := Overdue(sends, now)
|
||||
if err != nil {
|
||||
t.Fatalf("Overdue: %v", err)
|
||||
}
|
||||
var ids []string
|
||||
for _, r := range overdue {
|
||||
ids = append(ids, r.ID)
|
||||
}
|
||||
want := map[string]bool{overdueQueued: true, overdueRetrying: true}
|
||||
if len(ids) != len(want) {
|
||||
t.Fatalf("overdue: %v, want exactly %v", ids, want)
|
||||
}
|
||||
for _, id := range ids {
|
||||
if !want[id] {
|
||||
t.Fatalf("unexpectedly overdue: %s (fresh %s, not-yet-due %s, sending %s)", id, freshQueued, notYetDueRetrying, sendingNotOverdue)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1199/AC3: given an overdue send, when the tools' second ticker looks, then they tell once through
|
||||
// waits-tell (telling then told), separate from the sender's own tell.
|
||||
func TestTellWaitsTellsOnceThroughItsOwnField(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", At: judgedAt}
|
||||
req.Decided = &SendDecision{At: judgedAt.Add(-20 * time.Minute), Outcome: "queued"}
|
||||
if _, err := sends.Create(req.ID, req); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tell := &memTell{}
|
||||
if err := tellWaits(sends, tell.tell, req.ID, judgedAt); err != nil {
|
||||
t.Fatalf("telling waits: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 1 || !strings.Contains(tell.headlines[0], "waits") {
|
||||
t.Fatalf("waits-tell: %v", tell.headlines)
|
||||
}
|
||||
kept, _, ok, err := getSend(sends, req.ID)
|
||||
if err != nil || !ok || kept.Decided == nil || kept.Decided.WaitsTell != "told" || kept.Decided.Tell != "" {
|
||||
t.Fatalf("waits-tell not settled apart from the sender's own tell: %v %v %+v", err, ok, kept.Decided)
|
||||
}
|
||||
// Told once: a second look tells nothing more.
|
||||
if err := tellWaits(sends, tell.tell, req.ID, judgedAt.Add(time.Minute)); err != nil {
|
||||
t.Fatalf("telling waits again: %v", err)
|
||||
}
|
||||
if len(tell.headlines) != 1 {
|
||||
t.Fatalf("told again: %v", tell.headlines)
|
||||
}
|
||||
}
|
||||
|
||||
// 1199/AC4: given each state, when mail_send_status reads it (reportDecision), then it says: waiting for
|
||||
// the operator, approved and waiting for the sender, being sent, waiting for its next attempt, sent,
|
||||
// failed with its reason, refused, not proven, declined, or expired.
|
||||
func TestReportDecisionCoversTheWholeStatusVocabulary(t *testing.T) {
|
||||
for name, c := range map[string]struct {
|
||||
d SendDecision
|
||||
want string
|
||||
}{
|
||||
"approved": {SendDecision{At: judgedAt, Outcome: "approved"}, "waiting"},
|
||||
"queued": {SendDecision{At: judgedAt, Outcome: "queued"}, "waiting"},
|
||||
"being sent": {SendDecision{At: judgedAt, Outcome: "sending"}, "waiting"},
|
||||
"retrying": {SendDecision{At: judgedAt, Outcome: "retrying", NextAttempt: judgedAt.Add(time.Minute)}, "waiting"},
|
||||
"sent": {SendDecision{At: judgedAt, Outcome: "sent"}, "sent"},
|
||||
"failed with its reason": {SendDecision{At: judgedAt, Outcome: "failed: 550 no such mailbox"}, "failed"},
|
||||
"refused": {SendDecision{At: judgedAt, Outcome: "refused: the warrant does not bind the request"}, "refused"},
|
||||
"not proven": {SendDecision{At: judgedAt, Outcome: "not proven"}, "not proven"},
|
||||
"declined": {SendDecision{At: judgedAt, Outcome: "declined"}, "declined"},
|
||||
"expired (never heard)": {SendDecision{At: judgedAt, Outcome: "ended: never heard"}, "expired"},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
out := map[string]any{}
|
||||
reportDecision(out, c.d, judgedAt.Add(time.Minute))
|
||||
if out["status"] != c.want {
|
||||
t.Fatalf("status: %v, want %q (words: %v)", out["status"], c.want, out["words"])
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -139,10 +139,11 @@ func (b *Bundle) Tools() []stdio.Tool {
|
||||
Run: b.readTool},
|
||||
{Name: HealthTool,
|
||||
Description: "How mesh-mail is, as its watch last saw it: healthy when every account's server answered within five " +
|
||||
"minutes; unhealthy, naming the account and why, when a connection is down or its IDLE on the inbox went silent; " +
|
||||
"waits, naming the account's password, while the operator has not given it at the desk.",
|
||||
"minutes; unhealthy, naming the account and why, when a connection is down or its IDLE on the inbox went silent, " +
|
||||
"or an approved send has waited 10 minutes or more for its sender; waits, naming the account's password, while " +
|
||||
"the operator has not given it at the desk.",
|
||||
Input: schema(nil, map[string]any{}),
|
||||
Run: func(map[string]any) (any, error) { return ReadStatus(b.Env.Status, b.now()), nil }},
|
||||
Run: b.healthTool},
|
||||
{Name: SendTool,
|
||||
Description: "Send mail from an account you are granted (the same check as mail_read, novox/hq ADR 0361 §2). It " +
|
||||
"sends nothing itself: it raises an ask on the operator channel — the account, the recipients, the subject and " +
|
||||
@@ -206,6 +207,22 @@ func (b *Bundle) caller(args map[string]any) string {
|
||||
return stdio.CallerOf(args)
|
||||
}
|
||||
|
||||
// healthTool folds an overdue send (novox/hq ADR 0436 clause 11, issue 1199) into the watch's own verdict:
|
||||
// unhealthy, naming the send, no matter what the watch itself said. Overdue reads `sends` directly — the
|
||||
// watch process has no access to it — so a send stuck past its own 10 minutes is caught even while every
|
||||
// account's own connection is fine.
|
||||
func (b *Bundle) healthTool(map[string]any) (any, error) {
|
||||
base := ReadStatus(b.Env.Status, b.now())
|
||||
overdue, err := Overdue(b.sends(), b.now())
|
||||
if err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] mail_health: the overdue check failed: %v", err)
|
||||
}
|
||||
return base, nil
|
||||
}
|
||||
return WithOverdue(base, overdue), nil
|
||||
}
|
||||
|
||||
func (b *Bundle) accountsTool(args map[string]any) (any, error) {
|
||||
if err := only(args); err != nil {
|
||||
return nil, err
|
||||
|
||||
Reference in New Issue
Block a user