mesh-mail: on Approve the tools record the send queued directly, with no separate approved state, then publish; a stuck queued record is republished by the sweep past a short grace, never NakWithDelay'd by the sender (ADR 0436 clause 8, decided to replace the review round 1 item 2 work-around; issues 1194, 1196, 1199)
This commit is contained in:
jochen
@@ -224,46 +224,6 @@ func TestClaimQueuedSendOnALostRaceSendsNothingAndLeavesTheMessageToTheBus(t *te
|
||||
}
|
||||
}
|
||||
|
||||
// 1196/AC5: given a send still approved (publishQueued published to mail-sends before it recorded queued,
|
||||
// and a message was delivered inside that window), when taken, then it is delivered again later and
|
||||
// nothing is recorded or told (novox/hq review round 1 finding 2).
|
||||
func TestClaimQueuedSendOnASendStillApprovedAsksRedeliveryAndRecordsNothing(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := rawQueuedSend(t, sends)
|
||||
current, rev, ok, err := getSend(sends, req.ID)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("reading it back: %v %v", err, ok)
|
||||
}
|
||||
current.Decided.Outcome = "approved" // not yet advanced to "queued" by the tools' own second write
|
||||
if _, err := sends.Update(req.ID, current, rev); err != nil {
|
||||
t.Fatalf("leaving it approved: %v", err)
|
||||
}
|
||||
req = current
|
||||
|
||||
emit := &memEmit{}
|
||||
_, d, _, outcome, err := claimQueuedSend(sends, newSendRouter().Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute))
|
||||
if err != nil {
|
||||
t.Fatalf("claiming: %v", err)
|
||||
}
|
||||
if outcome != claimNotYetQueued {
|
||||
t.Fatalf("outcome: %v, want claimNotYetQueued", outcome)
|
||||
}
|
||||
if d.Outcome != "approved" {
|
||||
t.Fatalf("decision: %+v, want outcome still approved (unchanged)", d)
|
||||
}
|
||||
kept, _, ok, err := getSend(sends, req.ID)
|
||||
if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "approved" || kept.Decided.Tell != "" {
|
||||
t.Fatalf("something was recorded or told: %v %v %+v", err, ok, kept.Decided)
|
||||
}
|
||||
ids, err := tellPendingIDs(sends)
|
||||
if err != nil || len(ids) != 0 {
|
||||
t.Fatalf("a tell was queued despite nothing being told: %v %v", err, ids)
|
||||
}
|
||||
if len(emit.calls) != 0 {
|
||||
t.Fatalf("a send still approved emitted something: %+v", emit.calls)
|
||||
}
|
||||
}
|
||||
|
||||
// review round 1 finding 3: a failed write of the tells-pending index does not stop the refusal itself
|
||||
// from being recorded — the record write and the message's own fate are never undone by an index that
|
||||
// could not be kept — but the index failure is still reported, never silently dropped.
|
||||
@@ -316,21 +276,21 @@ func TestClaimQueuedSendStillRecordsNotProvenWhenTheTellsPendingIndexCannotBeWri
|
||||
assertEmittedFailed(t, emit, req.ID, d.Outcome)
|
||||
}
|
||||
|
||||
// Same finding, for publishQueued: a failed sender-pending index write does not stop "queued" from being
|
||||
// recorded, since the bus already took the message once.
|
||||
func TestPublishQueuedStillRecordsQueuedWhenTheSenderPendingIndexCannotBeWritten(t *testing.T) {
|
||||
// Same finding, for decideSend: a failed sender-pending index write does not stop "queued" from being
|
||||
// recorded, nor the publish from being attempted — there is no separate "approved" state any more (novox/hq
|
||||
// ADR 0436 clause 8, as decided to replace the two-state design, review round 1).
|
||||
func TestDecideSendStillRecordsQueuedAndPublishesWhenTheSenderPendingIndexCannotBeWritten(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Body: "body", At: judgedAt}
|
||||
claim := SendDecision{At: judgedAt, Outcome: "approved"}
|
||||
req.Decided = &claim
|
||||
rev, err := sends.Create(req.ID, req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sends.failKey, sends.failErr = senderPendingKey, errors.New("the store is down")
|
||||
queue := newMemQueue()
|
||||
w := warrantFor(req, judgedAt.Add(time.Minute))
|
||||
|
||||
d, err := publishQueued(sends, req, rev, claim, queue.publish, judgedAt.Add(time.Minute))
|
||||
d, err := decideSend(sends, queue.publish, req, rev, w, judgedAt.Add(time.Minute))
|
||||
if d.Outcome != "queued" {
|
||||
t.Fatalf("decision: %+v, want queued despite the index failing", d)
|
||||
}
|
||||
@@ -340,6 +300,48 @@ func TestPublishQueuedStillRecordsQueuedWhenTheSenderPendingIndexCannotBeWritten
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("published %d time(s)", n)
|
||||
}
|
||||
kept, _, ok, kerr := getSend(sends, req.ID)
|
||||
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "queued" {
|
||||
t.Fatalf("not kept as queued despite the index failing: %v %v %+v", kerr, ok, kept.Decided)
|
||||
}
|
||||
}
|
||||
|
||||
// 1194/AC1 (as decided to replace the two-state design, review round 1): on a verified warrant the record
|
||||
// moves straight from undecided to queued by one compare-and-set, with its time, and mail.send.approved is
|
||||
// published with the send id as its message id — no "approved" state in between.
|
||||
func TestDecideSendRecordsQueuedDirectlyAndPublishes(t *testing.T) {
|
||||
sends := newMemSends()
|
||||
req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Body: "body", At: judgedAt}
|
||||
rev, err := sends.Create(req.ID, req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
queue := newMemQueue()
|
||||
w := warrantFor(req, judgedAt.Add(time.Minute))
|
||||
now := judgedAt.Add(time.Minute)
|
||||
|
||||
d, err := decideSend(sends, queue.publish, req, rev, w, now)
|
||||
if err != nil {
|
||||
t.Fatalf("deciding: %v", err)
|
||||
}
|
||||
if d.Outcome != "queued" || !d.At.Equal(now) {
|
||||
t.Fatalf("decision: %+v, want queued at %v", d, now)
|
||||
}
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("published %d time(s), want exactly 1", n)
|
||||
}
|
||||
got := queue.last()
|
||||
if got.id != req.ID || got.body.Send != req.ID || got.body.Account != "mesh" || got.body.Ask != req.ID {
|
||||
t.Fatalf("published %+v, want the send id as the message id and body", got)
|
||||
}
|
||||
kept, _, ok, kerr := getSend(sends, req.ID)
|
||||
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "queued" {
|
||||
t.Fatalf("not kept as queued: %v %v %+v", kerr, ok, kept.Decided)
|
||||
}
|
||||
ids, err := senderPendingIDs(sends)
|
||||
if err != nil || len(ids) != 1 || ids[0] != req.ID {
|
||||
t.Fatalf("not indexed as sender-pending: %v %v", err, ids)
|
||||
}
|
||||
}
|
||||
|
||||
// raceOnFirstUpdate performs one competing write the first time Update is called — simulating another
|
||||
|
||||
@@ -591,8 +591,7 @@ func sdkQueuePublish(id string, body any) error {
|
||||
|
||||
// updateDecision writes a decision onto a kept send by compare-and-set, without touching the undecided index:
|
||||
// the index is settleSend's to drop, once the decision is one the sweep no longer needs to look at again
|
||||
// (novox/hq issue 811; ADR 0436 clause 8, where "approved" still needs the sweep). On a race lost to another
|
||||
// writer, it reads back whatever that writer decided.
|
||||
// (novox/hq issue 811). On a race lost to another writer, it reads back whatever that writer decided.
|
||||
func updateDecision(store SendStore, req SendRequest, rev uint64, d SendDecision) (SendDecision, uint64, error) {
|
||||
next := req
|
||||
next.Decided = &d
|
||||
@@ -671,52 +670,35 @@ func Decide(store SendStore, publish QueuePublish, w asks.Warrant, now time.Time
|
||||
}
|
||||
}
|
||||
|
||||
// decideSend claims the request as approved by a compare-and-set write before it is queued, so a redelivered
|
||||
// decideSend records the request queued by a compare-and-set write before it publishes, so a redelivered
|
||||
// warrant finds it already claimed (or decided) and queues nothing twice: a person's Approve is queued once
|
||||
// (novox/hq ADR 0436 clause 8). It never sends: the sender (clause 7) takes the queued message, verifies the
|
||||
// warrant again itself (clause 9) and performs the one handover to the server (clause 10).
|
||||
// (novox/hq ADR 0436 clause 8, as decided to replace the two-state approved/queued design — review round 1,
|
||||
// replacing finding 2's work-around with a decision changing the clause itself). There is no "approved"
|
||||
// state in a send's life at all: the record moves straight from undecided to queued; the publish that
|
||||
// follows is best-effort, and a publish that fails (or the tools stopping before it ever runs) leaves the
|
||||
// record "queued" with its own time, which the sweep finds and republishes once a short grace has passed
|
||||
// (Sweep, below) — the send id as the message id, and the stream's duplicate window, keep it one message
|
||||
// either way, and the sender's own claim (clause 9) keeps it one send. It never sends: the sender (clause 7)
|
||||
// takes the queued message, verifies the warrant again itself (clause 9) and performs the one handover to
|
||||
// the server (clause 10).
|
||||
func decideSend(store SendStore, publish QueuePublish, req SendRequest, rev uint64, w asks.Warrant, now time.Time) (SendDecision, error) {
|
||||
claim := SendDecision{At: now, Outcome: "approved", Warrant: w.Says()}
|
||||
claimed, newRev, err := updateDecision(store, req, rev, claim)
|
||||
// Indexed before the record carries "queued" (novox/hq review round 1 finding 3): a crash between this
|
||||
// and the write below must still leave it findable by mail_health's overdue check and the sweep's own
|
||||
// stale-queued retry once "queued" does land.
|
||||
indexErr := addSenderPending(store, req.ID)
|
||||
queued := SendDecision{At: now, Outcome: "queued", Warrant: w.Says()}
|
||||
claimed, newRev, err := settleSend(store, req, rev, queued)
|
||||
if err != nil {
|
||||
return SendDecision{}, err
|
||||
return SendDecision{}, errOr(indexErr, err)
|
||||
}
|
||||
if newRev == 0 {
|
||||
// Lost the claim race: another writer already decided (or is queueing) this send.
|
||||
return claimed, nil
|
||||
// Lost the claim race: another writer already decided this send.
|
||||
return claimed, indexErr
|
||||
}
|
||||
return publishQueued(store, req, newRev, claimed, publish, now)
|
||||
}
|
||||
|
||||
// publishQueued publishes the approved send on mail-sends, the send id its own message id, and records queued
|
||||
// once the bus took it. A publish that fails leaves the record approved — never an error returned to the
|
||||
// caller — so the sweep finds it and publishes it again (novox/hq ADR 0436 clause 8).
|
||||
func publishQueued(store SendStore, req SendRequest, rev uint64, approved SendDecision, publish QueuePublish, now time.Time) (SendDecision, error) {
|
||||
if err := publish(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID}); err != nil {
|
||||
return approved, nil
|
||||
}
|
||||
// mail_health's own overdue check (Overdue, issue 1199) needs to find every send the sender now owns,
|
||||
// the same way the sweep needs the undecided index: a reserved key, since neither side's own state over
|
||||
// mesh-sdk go/process lists its keys whole. Indexed BEFORE the record carries "queued" (novox/hq review
|
||||
// round 1 finding 3): a crash between this and the write below must still leave it findable once
|
||||
// "queued" does land — Overdue itself only ever matches "queued" or "retrying", so an entry that still
|
||||
// reads "approved" in the meantime is simply not yet counted overdue.
|
||||
indexErr := addSenderPending(store, req.ID)
|
||||
queued := SendDecision{At: now, Outcome: "queued", Warrant: approved.Warrant}
|
||||
final, newRev, err := updateDecision(store, req, rev, queued)
|
||||
if err != nil {
|
||||
// The publish stands either way: the bus already took the message once. The caller logs this; it is
|
||||
// not retried (a retry here would publish a second message under a new id).
|
||||
return approved, errOr(indexErr, fmt.Errorf("the send %s was queued, and that could not be kept (%w)", req.ID, err))
|
||||
}
|
||||
if newRev != 0 {
|
||||
// Only once "queued" is durably recorded is the undecided index's entry dropped: removing it first,
|
||||
// then crashing before the write, would have left the send unreachable by the sweep too.
|
||||
if rmErr := removeUndecided(store, req.ID); rmErr != nil {
|
||||
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the undecided index: %w", req.ID, rmErr))
|
||||
}
|
||||
}
|
||||
return final, indexErr
|
||||
// Best-effort: its own failure is never returned to the caller as this call's own error — the record is
|
||||
// already durably "queued" either way, and the sweep retries a publish that did not land.
|
||||
_ = publish(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID})
|
||||
return claimed, indexErr
|
||||
}
|
||||
|
||||
// ---- the sender's own claim (novox/hq ADR 0436 clauses 9, 10, issue 1196) --------------------------------------
|
||||
@@ -744,17 +726,8 @@ const (
|
||||
// message, racing this one); nothing is recorded here and nothing sent; the caller leaves the message to
|
||||
// the bus entirely — no Ack, no Nak, no Term.
|
||||
claimLostRace
|
||||
// claimNotYetQueued: publishQueued (above) publishes to mail-sends before it records queued, so a
|
||||
// message delivered in that narrow window finds the record still "approved" — the tools really did
|
||||
// approve it and are about to record queued, so this is not a refusal: nothing is recorded and nobody
|
||||
// is told; the caller asks the bus to redeliver shortly (novox/hq review round 1 finding 2).
|
||||
claimNotYetQueued
|
||||
)
|
||||
|
||||
// NotYetQueuedDelay is how long the sender asks the bus to wait before redelivering a message whose record
|
||||
// is still "approved" (claimNotYetQueued): long enough for the tools' own second write (queued) to land.
|
||||
const NotYetQueuedDelay = 5 * time.Second
|
||||
|
||||
// claimQueuedSend verifies the operator's warrant for id (mesh-mail's own ask) and claims the send by
|
||||
// compare-and-set from queued or retrying to sending, with the attempt and time, before any connection
|
||||
// (novox/hq ADR 0436 clauses 9, 10). It is never trusted from the queue's message alone: the ask is read
|
||||
@@ -772,8 +745,9 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string,
|
||||
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),
|
||||
// and nothing emitted either — clause 11's two events are for a send the operator did approve.
|
||||
// Nothing was ever decided at all (never queued): no tell owed (clause 11 ties the tell to a
|
||||
// *failed approved* send), and nothing emitted either — clause 11's two events are for a send the
|
||||
// operator did approve.
|
||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: the send is not yet approved"})
|
||||
return req, d, 0, claimRefused, err
|
||||
}
|
||||
@@ -810,10 +784,6 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string,
|
||||
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
|
||||
}
|
||||
return req, d, 0, claimNotProven, indexErr
|
||||
case req.Decided.Outcome == "approved":
|
||||
// Not yet visible as queued (publishQueued publishes before it records queued): the tools did
|
||||
// approve it, so this is not a refusal (novox/hq review round 1 finding 2).
|
||||
return req, *req.Decided, 0, claimNotYetQueued, nil
|
||||
case req.Decided.Outcome != "queued" && req.Decided.Outcome != "retrying":
|
||||
return refuse(fmt.Sprintf("the send is %q, neither queued nor retrying", req.Decided.Outcome))
|
||||
}
|
||||
@@ -1366,9 +1336,10 @@ func ReadSendStatus(store SendStore, bus AskBus, publish QueuePublish, id string
|
||||
func reportDecision(out map[string]any, d SendDecision, now time.Time) {
|
||||
out["at"] = d.At.Format(time.RFC3339)
|
||||
switch {
|
||||
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).
|
||||
case d.Outcome == "queued":
|
||||
// Recorded queued by the one compare-and-set the Approve makes (novox/hq ADR 0436 clause 8): there
|
||||
// is no separate "approved" state in a send's life. Published to mail-sends already, or the sweep
|
||||
// will publish it again once a short grace has passed.
|
||||
out["status"], out["words"] = "waiting", "approved and waiting for the sender; ask again shortly"
|
||||
case d.Outcome == "sent":
|
||||
out["status"], out["words"] = "sent", "sent on the operator's Approve ("+d.Warrant+")"
|
||||
@@ -1412,39 +1383,55 @@ func (b *Bundle) Heard(w asks.Warrant) {
|
||||
}
|
||||
}
|
||||
|
||||
// Sweep reads only the sends the undecided index still names. A send not yet decided is asked about the same
|
||||
// way mail_send_status does (ReadSendStatus): an Approve recorded while this module did not hear it directly
|
||||
// (down, or not yet subscribed) is still acted on, once. A send left approved (its publish to mail-sends
|
||||
// failed, or the tools stopped between the two compare-and-set writes) is published again, the send id still
|
||||
// its message id (novox/hq ADR 0436 clause 8). 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).
|
||||
// QueuedGrace is how long a newly queued send is given to be published before the sweep considers it
|
||||
// possibly stuck and publishes it again: long enough that the tools' own publish, right after the same
|
||||
// compare-and-set write, is never raced by the sweep under ordinary load (novox/hq ADR 0436 clause 8, as
|
||||
// decided to replace the two-state approved/queued design, review round 1).
|
||||
const QueuedGrace = time.Minute
|
||||
|
||||
// Sweep does two things, neither reading every send ever kept (its work grows only with the few still
|
||||
// live, never with history): a send the undecided index still names and that is not yet decided at all is
|
||||
// asked about the same way mail_send_status does (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; the index, issue 811); and a send the sender-pending index names that is
|
||||
// still "queued" past QueuedGrace — its publish to mail-sends may have failed, or the tools stopped before
|
||||
// it ever ran — is published again, the send id still its message id, so the duplicate window and the
|
||||
// sender's own claim keep it one send either way (novox/hq ADR 0436 clause 8, review round 1).
|
||||
func (b *Bundle) Sweep(now time.Time) {
|
||||
store := b.sends()
|
||||
ids, err := undecidedIDs(store)
|
||||
undecided, err := undecidedIDs(store)
|
||||
if err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] the undecided index cannot be read: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
for _, id := range ids {
|
||||
req, rev, ok, err := getSend(store, id)
|
||||
if err != nil || !ok {
|
||||
continue
|
||||
}
|
||||
switch {
|
||||
case req.Decided == nil:
|
||||
} else {
|
||||
for _, id := range undecided {
|
||||
req, _, ok, err := getSend(store, id)
|
||||
if err != nil || !ok || req.Decided != nil {
|
||||
continue
|
||||
}
|
||||
if _, err := ReadSendStatus(store, b.askBus(), b.queuePublish(), id, now); err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] sweeping the send %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
case req.Decided.Outcome == "approved":
|
||||
if _, err := publishQueued(store, req, rev, *req.Decided, b.queuePublish(), now); err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] sweeping the approved send %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
pending, err := senderPendingIDs(store)
|
||||
if err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] the sender-pending index cannot be read: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
for _, id := range pending {
|
||||
req, _, ok, err := getSend(store, id)
|
||||
if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "queued" || now.Sub(req.Decided.At) < QueuedGrace {
|
||||
continue
|
||||
}
|
||||
if err := b.queuePublish()(req.ID, QueuedSend{Send: req.ID, Account: req.Account, Ask: req.ID}); err != nil {
|
||||
if b.Logf != nil {
|
||||
b.Logf("[mesh-mail] republishing the queued send %s: %v", id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -566,16 +566,19 @@ func TestSweepActsOnAnApprovedSendItWasNotToldAboutDirectly(t *testing.T) {
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("the sweep did not act on the approved send: %d message(s)", n)
|
||||
}
|
||||
b.Sweep(judgedAt.Add(3 * time.Minute))
|
||||
// A second sweep, still within the just-queued record's own grace, does not publish it again (past the
|
||||
// grace, it would: TestSweepPublishesAgainAQueuedRecordPastItsGrace, novox/hq review round 1).
|
||||
b.Sweep(judgedAt.Add(2*time.Minute + QueuedGrace/2))
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("a second sweep queued again: %d message(s)", n)
|
||||
t.Fatalf("a second sweep within the grace queued again: %d message(s)", n)
|
||||
}
|
||||
}
|
||||
|
||||
// 1194/AC2: A send left "approved" (the publish to mail-sends failed, or the tools stopped between the two
|
||||
// compare-and-set writes) is still found by the undecided index, and the sweep publishes it again, still under
|
||||
// the send's own id (novox/hq ADR 0436 clause 8).
|
||||
func TestSweepPublishesAgainASendLeftApproved(t *testing.T) {
|
||||
// 1194/AC2 (as decided to replace the two-state approved/queued design, review round 1): the sweep
|
||||
// republishes a queued, unclaimed record whose queued time is older than QueuedGrace, the send id still
|
||||
// its message id — the stream's duplicate window and the sender's own claim keep it one send either way
|
||||
// (proved end to end over a real bus by the module-whole acceptance test, issue 1198 AC4).
|
||||
func TestSweepPublishesAgainAQueuedRecordPastItsGrace(t *testing.T) {
|
||||
b, router, sends := sendTestBundle(t, Server{Host: "127.0.0.1", Port: 25, Security: "tls"})
|
||||
queue := newMemQueue()
|
||||
queue.failErr = errors.New("the bus is down")
|
||||
@@ -592,31 +595,34 @@ func TestSweepPublishesAgainASendLeftApproved(t *testing.T) {
|
||||
|
||||
w := router.chose(id, OptionSend, judgedAt.Add(time.Minute))
|
||||
router.close(w)
|
||||
if _, err := Decide(sends, b.queuePublish(), w, judgedAt.Add(2*time.Minute)); err != nil {
|
||||
decidedAt := judgedAt.Add(2 * time.Minute)
|
||||
if _, err := Decide(sends, b.queuePublish(), w, decidedAt); err != nil {
|
||||
t.Fatalf("deciding: %v", err)
|
||||
}
|
||||
if n := queue.count(); n != 0 {
|
||||
t.Fatalf("the queue received %d message(s) despite failing", n)
|
||||
}
|
||||
// There is no "approved" state: the record is already "queued", its own publish just failed.
|
||||
req, _, ok, err := getSend(sends, id)
|
||||
if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "approved" {
|
||||
t.Fatalf("left approved: %v %v %+v", err, ok, req.Decided)
|
||||
if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "queued" {
|
||||
t.Fatalf("left queued despite the publish failing: %v %v %+v", err, ok, req.Decided)
|
||||
}
|
||||
|
||||
// Once the bus comes back, the sweep publishes the approved send again, and it reaches queued.
|
||||
queue.failErr = nil
|
||||
b.Sweep(judgedAt.Add(3 * time.Minute))
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("the sweep did not publish the approved send: %d message(s)", n)
|
||||
// Within its grace, the sweep leaves a freshly queued record alone.
|
||||
b.Sweep(decidedAt.Add(QueuedGrace / 2))
|
||||
if n := queue.count(); n != 0 {
|
||||
t.Fatalf("the sweep republished within the grace: %d message(s)", n)
|
||||
}
|
||||
req, _, ok, err = getSend(sends, id)
|
||||
if err != nil || !ok || req.Decided == nil || req.Decided.Outcome != "queued" {
|
||||
t.Fatalf("not queued after the sweep: %v %v %+v", err, ok, req.Decided)
|
||||
}
|
||||
// A further sweep does not publish a send that already reached queued (it left the undecided index).
|
||||
b.Sweep(judgedAt.Add(4 * time.Minute))
|
||||
|
||||
// Past its grace, the sweep publishes the still-queued send again.
|
||||
b.Sweep(decidedAt.Add(QueuedGrace + time.Second))
|
||||
if n := queue.count(); n != 1 {
|
||||
t.Fatalf("a further sweep published again: %d message(s)", n)
|
||||
t.Fatalf("the sweep did not republish the queued send past its grace: %d message(s)", n)
|
||||
}
|
||||
got := queue.last()
|
||||
if got.id != id || got.body.Send != id {
|
||||
t.Fatalf("republished %+v, want the send id as the message id", got)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -228,13 +228,6 @@ func (h *senderHandler) handle(msg *process.Message) {
|
||||
switch outcome {
|
||||
case claimLostRace:
|
||||
return // another delivery of the same message already owns it.
|
||||
case claimNotYetQueued:
|
||||
// The tools' own second write (queued) has not landed yet: not a refusal (novox/hq review round 1
|
||||
// finding 2). Nothing recorded, nobody told; ask the bus to redeliver shortly.
|
||||
if err := msg.NakWithDelay(NotYetQueuedDelay); err != nil {
|
||||
h.logf("[mesh-mail] send: asking %s redelivered: %v", q.Send, err)
|
||||
}
|
||||
return
|
||||
case claimRefused, claimNotProven:
|
||||
if err := msg.Term(); err != nil {
|
||||
h.logf("[mesh-mail] send: terminating %s: %v", q.Send, err)
|
||||
|
||||
@@ -225,29 +225,39 @@ func TestTellWaitsTellsOnceThroughItsOwnField(t *testing.T) {
|
||||
}
|
||||
|
||||
// 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.
|
||||
// the operator (covered by send_test.go's own asking tests, which read ReadSendStatus before any decision
|
||||
// exists — reportDecision itself only ever sees an already-decided SendDecision), approved and waiting for
|
||||
// the sender, being sent, waiting for its next attempt, sent, failed with its reason, refused, not proven,
|
||||
// declined, or expired — checking both the status word and the words that actually say which of these it
|
||||
// is (novox/hq review round 1 finding 6: checking status alone proves too little).
|
||||
func TestReportDecisionCoversTheWholeStatusVocabulary(t *testing.T) {
|
||||
for name, c := range map[string]struct {
|
||||
d SendDecision
|
||||
want string
|
||||
d SendDecision
|
||||
wantStatus string
|
||||
wantInWords 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"},
|
||||
// There is no "approved" state in a send's life: Decide's one compare-and-set write moves it
|
||||
// straight from undecided to "queued" (novox/hq ADR 0436 clause 8, as decided to replace the
|
||||
// two-state design, review round 1).
|
||||
"queued": {SendDecision{At: judgedAt, Outcome: "queued"}, "waiting", "approved and waiting for the sender"},
|
||||
"being sent": {SendDecision{At: judgedAt, Outcome: "sending"}, "waiting", "being sent"},
|
||||
"retrying": {SendDecision{At: judgedAt, Outcome: "retrying", NextAttempt: judgedAt.Add(time.Minute)}, "waiting", "waiting for its next attempt"},
|
||||
"sent": {SendDecision{At: judgedAt, Outcome: "sent"}, "sent", "sent on the operator's Approve"},
|
||||
"failed with its reason": {SendDecision{At: judgedAt, Outcome: "failed: 550 no such mailbox"}, "failed", "550 no such mailbox"},
|
||||
"refused": {SendDecision{At: judgedAt, Outcome: "refused: the warrant does not bind the request"}, "refused", "the warrant does not bind the request"},
|
||||
"not proven": {SendDecision{At: judgedAt, Outcome: "not proven"}, "not proven", "not proven"},
|
||||
"declined": {SendDecision{At: judgedAt, Outcome: "declined"}, "declined", "the operator declined"},
|
||||
"expired (never heard)": {SendDecision{At: judgedAt, Outcome: "ended: never heard"}, "expired", "never heard"},
|
||||
} {
|
||||
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"])
|
||||
if out["status"] != c.wantStatus {
|
||||
t.Fatalf("status: %v, want %q (words: %v)", out["status"], c.wantStatus, out["words"])
|
||||
}
|
||||
words, _ := out["words"].(string)
|
||||
if !strings.Contains(words, c.wantInWords) {
|
||||
t.Fatalf("words: %q, want it to contain %q", words, c.wantInWords)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user