mesh-mail: decide kept by settleSend's own returned revision, not its error, so a changed revision is actually retried instead of silently accepted as another writer's state; emit mail.send.failed from claimQueuedSend only once its own write lands (review round 2; issue 1198)
This commit is contained in:
@@ -344,6 +344,40 @@ func TestDecideSendRecordsQueuedDirectlyAndPublishes(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 1196/AC2 (review round 2): a store that changes the revision between the claim and the refusal's own
|
||||||
|
// write — another writer's decision landing in between — never emits mail.send.failed for a write that did
|
||||||
|
// not actually land (refuse() does not retry: that is recordOutcome's own job, send.go, review round 2).
|
||||||
|
func TestClaimQueuedSendRefuseEmitsNothingWhenTheWriteDidNotLand(t *testing.T) {
|
||||||
|
sends, router := newMemSends(), newSendRouter()
|
||||||
|
req := rawQueuedSend(t, sends)
|
||||||
|
w := warrantFor(req, judgedAt.Add(time.Minute))
|
||||||
|
w.AskDigest = "sha256:" + strings.Repeat("0", 64) // mismatched: claimQueuedSend refuses it
|
||||||
|
router.close(w)
|
||||||
|
racy := &raceOnFirstUpdate{SendStore: sends, race: func() {
|
||||||
|
r, rv, ok, err := getSend(sends, req.ID)
|
||||||
|
if err != nil || !ok {
|
||||||
|
t.Fatalf("racing write could not read the send: %v %v", err, ok)
|
||||||
|
}
|
||||||
|
if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "not proven", Attempt: 1}); err != nil {
|
||||||
|
t.Fatalf("racing write: %v", err)
|
||||||
|
}
|
||||||
|
}}
|
||||||
|
emit := &memEmit{}
|
||||||
|
|
||||||
|
_, d, _, outcome, _ := claimQueuedSend(racy, router.Record, emit.emit, req.ID, judgedAt.Add(3*time.Minute))
|
||||||
|
if outcome != claimRefused {
|
||||||
|
t.Fatalf("outcome: %v, want claimRefused", outcome)
|
||||||
|
}
|
||||||
|
if len(emit.calls) != 0 {
|
||||||
|
t.Fatalf("emitted mail.send.failed despite the refusal's own write not landing: %+v", emit.calls)
|
||||||
|
}
|
||||||
|
// The kept record is the racing write's own decision, not this call's: refuse() never retries.
|
||||||
|
kept, _, ok, kerr := getSend(sends, req.ID)
|
||||||
|
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "not proven" {
|
||||||
|
t.Fatalf("kept: %v %v %+v (returned %+v)", kerr, ok, kept.Decided, d)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// raceOnFirstUpdate performs one competing write the first time Update is called — simulating another
|
// raceOnFirstUpdate performs one competing write the first time Update is called — simulating another
|
||||||
// delivery's claim landing first — so the caller's own compare-and-set loses with stdio.ErrChanged
|
// delivery's claim landing first — so the caller's own compare-and-set loses with stdio.ErrChanged
|
||||||
// (novox/hq issue 1196, AC4).
|
// (novox/hq issue 1196, AC4).
|
||||||
|
|||||||
@@ -231,3 +231,78 @@ func TestRecordOutcomeOnRefusedDoesNotTerminateOrEmitWhenTheWriteCannotBeKept(t
|
|||||||
t.Fatalf("emitted mail.send.failed despite the write never being kept: %+v", emit.calls)
|
t.Fatalf("emitted mail.send.failed despite the write never being kept: %+v", emit.calls)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 1198/AC1 (review round 2): a store that changes the revision between the claim and the outcome — another
|
||||||
|
// writer's decision landing in between — is retried once with the fresh revision; recordOutcome's own
|
||||||
|
// decision, once it does land, is what is kept, acked and emitted, never the other writer's state read
|
||||||
|
// back with a nil error mistaken for success.
|
||||||
|
func TestRecordOutcomeRetriesOnceWhenTheRevisionChangedBetweenClaimAndOutcome(t *testing.T) {
|
||||||
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
||||||
|
racy := &raceOnFirstUpdate{SendStore: sends, race: func() {
|
||||||
|
// A different delivery's own claim lands first, bumping the revision without ever finishing this
|
||||||
|
// call's own write — settleSend (via updateDecision) answers that with revision 0 and a nil error.
|
||||||
|
r, rv, ok, err := getSend(sends, req.ID)
|
||||||
|
if err != nil || !ok {
|
||||||
|
t.Fatalf("racing write could not read the send: %v %v", err, ok)
|
||||||
|
}
|
||||||
|
if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "sending", Attempt: 2}); err != nil {
|
||||||
|
t.Fatalf("racing write: %v", err)
|
||||||
|
}
|
||||||
|
}}
|
||||||
|
emit := &memEmit{}
|
||||||
|
|
||||||
|
d, action, _, err := recordOutcome(racy, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("recordOutcome: %v", err)
|
||||||
|
}
|
||||||
|
if d.Outcome != "sent" || action != actionAck {
|
||||||
|
t.Fatalf("outcome %+v, action %v, want sent/actionAck once retried with the fresh revision", d, action)
|
||||||
|
}
|
||||||
|
if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendSent || emit.calls[0].body.Outcome != "sent" {
|
||||||
|
t.Fatalf("emitted: %+v", emit.calls)
|
||||||
|
}
|
||||||
|
kept, _, ok, kerr := getSend(sends, req.ID)
|
||||||
|
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" {
|
||||||
|
t.Fatalf("the kept record is not this call's own decision: %v %v %+v", kerr, ok, kept.Decided)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Same finding: when the race repeats on the retry too (a second writer keeps landing first), recordOutcome
|
||||||
|
// gives up rather than looping or mistaking the other writer's state for its own — actionLeave, nothing
|
||||||
|
// emitted, the error reported.
|
||||||
|
func TestRecordOutcomeGivesUpWhenTheRevisionKeepsChanging(t *testing.T) {
|
||||||
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
||||||
|
always := &raceEveryUpdate{SendStore: sends, race: func() {
|
||||||
|
r, rv, ok, err := getSend(sends, req.ID)
|
||||||
|
if err != nil || !ok {
|
||||||
|
t.Fatalf("racing write could not read the send: %v %v", err, ok)
|
||||||
|
}
|
||||||
|
if _, _, err := settleSend(sends, r, rv, SendDecision{At: judgedAt, Outcome: "sending", Attempt: r.Decided.Attempt + 1}); err != nil {
|
||||||
|
t.Fatalf("racing write: %v", err)
|
||||||
|
}
|
||||||
|
}}
|
||||||
|
emit := &memEmit{}
|
||||||
|
|
||||||
|
_, action, _, err := recordOutcome(always, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
|
||||||
|
if err == nil {
|
||||||
|
t.Fatalf("a persistently changing revision was not reported")
|
||||||
|
}
|
||||||
|
if action != actionLeave {
|
||||||
|
t.Fatalf("action: %v, want actionLeave", action)
|
||||||
|
}
|
||||||
|
if len(emit.calls) != 0 {
|
||||||
|
t.Fatalf("emitted despite never keeping its own write: %+v", emit.calls)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// raceEveryUpdate performs the competing write before every Update call, not only the first — simulating a
|
||||||
|
// revision that keeps changing out from under every retry (novox/hq review round 2, issue 1198).
|
||||||
|
type raceEveryUpdate struct {
|
||||||
|
SendStore
|
||||||
|
race func()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (r *raceEveryUpdate) Update(key string, value any, revision uint64) (uint64, error) {
|
||||||
|
r.race()
|
||||||
|
return r.SendStore.Update(key, value, revision)
|
||||||
|
}
|
||||||
|
|||||||
@@ -759,11 +759,13 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string,
|
|||||||
// crash between this and the write below must still leave it findable once refused does land;
|
// crash between this and the write below must still leave it findable once refused does land;
|
||||||
// tellSend itself tolerates finding it still sending, queued or retrying in the meantime.
|
// tellSend itself tolerates finding it still sending, queued or retrying in the meantime.
|
||||||
indexErr := addTellPending(store, req.ID)
|
indexErr := addTellPending(store, req.ID)
|
||||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason})
|
d, newRev, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "refused: " + reason})
|
||||||
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
|
if newRev == 0 {
|
||||||
if err != nil {
|
// The write did not land (another writer's decision already stands, or it failed outright):
|
||||||
|
// nothing of this call's own making to emit (novox/hq review round 2, issue 1198).
|
||||||
return req, d, 0, claimRefused, errOr(indexErr, err)
|
return req, d, 0, claimRefused, errOr(indexErr, err)
|
||||||
}
|
}
|
||||||
|
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
|
||||||
// Only once refused is durably recorded is the sender-pending index's entry dropped.
|
// Only once refused is durably recorded is the sender-pending index's entry dropped.
|
||||||
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
|
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
|
||||||
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
|
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
|
||||||
@@ -775,11 +777,13 @@ func claimQueuedSend(store SendStore, record RouterRecord, emit Emit, id string,
|
|||||||
// A previous delivery may have handed it to the server already; never sent twice, and never
|
// A previous delivery may have handed it to the server already; never sent twice, and never
|
||||||
// retried automatically either way (ADR 0436 clause 10). Same ordering as refuse() above.
|
// retried automatically either way (ADR 0436 clause 10). Same ordering as refuse() above.
|
||||||
indexErr := addTellPending(store, req.ID)
|
indexErr := addTellPending(store, req.ID)
|
||||||
d, _, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt})
|
d, newRev, err := settleSend(store, req, rev, SendDecision{At: now, Outcome: "not proven", Attempt: req.Decided.Attempt})
|
||||||
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
|
if newRev == 0 {
|
||||||
if err != nil {
|
// The write did not land: nothing of this call's own making to emit (novox/hq review round 2,
|
||||||
|
// issue 1198).
|
||||||
return req, d, 0, claimNotProven, errOr(indexErr, err)
|
return req, d, 0, claimNotProven, errOr(indexErr, err)
|
||||||
}
|
}
|
||||||
|
_ = emit(MailSendFailed, req.ID, OutcomeEvent{Send: req.ID, Account: req.Account, Outcome: d.Outcome})
|
||||||
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
|
if rmErr := removeSenderPending(store, req.ID); rmErr != nil {
|
||||||
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
|
indexErr = errOr(indexErr, fmt.Errorf("dropping %s from the sender-pending index: %w", req.ID, rmErr))
|
||||||
}
|
}
|
||||||
@@ -887,7 +891,12 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
|
|||||||
// fresh one) before giving up: kept is false only when the record's own write could not be kept even
|
// fresh one) before giving up: kept is false only when the record's own write could not be kept even
|
||||||
// after that retry, in which case the caller emits nothing and leaves the message to the bus entirely —
|
// after that retry, in which case the caller emits nothing and leaves the message to the bus entirely —
|
||||||
// never announcing an outcome (sent, failed, retrying) that was not actually recorded (novox/hq review
|
// never announcing an outcome (sent, failed, retrying) that was not actually recorded (novox/hq review
|
||||||
// round 1 finding 5).
|
// round 1 finding 5). "Kept" is decided from settleSend's own returned revision, never from its error
|
||||||
|
// alone: on a changed revision whose record already carries another writer's decision, settleSend
|
||||||
|
// (through updateDecision) returns that other decision with revision 0 and a nil error — a nil error
|
||||||
|
// is not success here, and without checking the revision the retry never ran at all, so the caller
|
||||||
|
// could emit an outcome (and ack) while the record actually held someone else's state (novox/hq review
|
||||||
|
// round 2, issue 1198).
|
||||||
write := func(d SendDecision, final, tellWorthy bool) (result SendDecision, kept bool, outErr error) {
|
write := func(d SendDecision, final, tellWorthy bool) (result SendDecision, kept bool, outErr error) {
|
||||||
var indexErr error
|
var indexErr error
|
||||||
if tellWorthy {
|
if tellWorthy {
|
||||||
@@ -897,11 +906,18 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
|
|||||||
indexErr = addTellPending(store, req.ID)
|
indexErr = addTellPending(store, req.ID)
|
||||||
}
|
}
|
||||||
curReq, curRev := req, rev
|
curReq, curRev := req, rev
|
||||||
|
var newRev uint64
|
||||||
var writeErr error
|
var writeErr error
|
||||||
for attempt := 0; attempt < 2; attempt++ {
|
for attempt := 0; attempt < 2; attempt++ {
|
||||||
result, _, writeErr = settleSend(store, curReq, curRev, d)
|
result, newRev, writeErr = settleSend(store, curReq, curRev, d)
|
||||||
if writeErr == nil || !errors.Is(writeErr, stdio.ErrChanged) {
|
if newRev != 0 {
|
||||||
break
|
break // this write landed
|
||||||
|
}
|
||||||
|
if attempt == 1 {
|
||||||
|
break // retried once already; give up
|
||||||
|
}
|
||||||
|
if writeErr != nil && !errors.Is(writeErr, stdio.ErrChanged) {
|
||||||
|
break // a hard failure, not a revision race: retrying blindly would not help
|
||||||
}
|
}
|
||||||
fresh, freshRev, ok, rerr := getSend(store, req.ID)
|
fresh, freshRev, ok, rerr := getSend(store, req.ID)
|
||||||
if rerr != nil || !ok {
|
if rerr != nil || !ok {
|
||||||
@@ -909,8 +925,12 @@ func recordOutcome(store SendStore, emit Emit, req SendRequest, claimed SendDeci
|
|||||||
}
|
}
|
||||||
curReq, curRev = fresh, freshRev
|
curReq, curRev = fresh, freshRev
|
||||||
}
|
}
|
||||||
if writeErr != nil {
|
if newRev == 0 {
|
||||||
return d, false, errOr(indexErr, fmt.Errorf("the send %s's outcome could not be kept, even retried on a changed revision (%w)", req.ID, writeErr))
|
reportErr := writeErr
|
||||||
|
if reportErr == nil {
|
||||||
|
reportErr = fmt.Errorf("the send %s's outcome could not be kept: another writer's decision already stands, even retried on its fresh revision", req.ID)
|
||||||
|
}
|
||||||
|
return d, false, errOr(indexErr, reportErr)
|
||||||
}
|
}
|
||||||
if final {
|
if final {
|
||||||
// Only once the outcome is durably recorded is the sender-pending index's entry dropped: from
|
// Only once the outcome is durably recorded is the sender-pending index's entry dropped: from
|
||||||
|
|||||||
Reference in New Issue
Block a user