309 lines
13 KiB
Go
309 lines
13 KiB
Go
package main
|
|
|
|
// recordOutcome writes the final decision for one claimed send and says what the caller does with the
|
|
// queue message next (novox/hq ADR 0436 clauses 10, 11, issue 1198).
|
|
|
|
import (
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// emitted is one call recordOutcome's Emit made.
|
|
type emitted struct {
|
|
eventType, id string
|
|
body OutcomeEvent
|
|
}
|
|
|
|
// memEmit collects every emit call, for a test to check the event and its body.
|
|
type memEmit struct{ calls []emitted }
|
|
|
|
func (m *memEmit) emit(eventType, id string, body any) error {
|
|
b, _ := body.(OutcomeEvent)
|
|
m.calls = append(m.calls, emitted{eventType: eventType, id: id, body: b})
|
|
return nil
|
|
}
|
|
|
|
// outcomeTestClaim keeps one send claimed "sending" at the given attempt, returning its request, the
|
|
// claimed decision and the revision recordOutcome needs.
|
|
func outcomeTestClaim(t *testing.T, attempt int) (SendRequest, SendDecision, uint64, *memSends) {
|
|
t.Helper()
|
|
sends := newMemSends()
|
|
req := SendRequest{ID: newSendID(), Account: "mesh", To: []string{"ops@example.org"}, Subject: "s", Body: "body", At: judgedAt}
|
|
claim := SendDecision{At: judgedAt.Add(time.Minute), Outcome: "sending", Warrant: "the operator, via telegram, chose Send", Attempt: attempt}
|
|
req.Decided = &claim
|
|
rev, err := sends.Create(req.ID, req)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return req, claim, rev, sends
|
|
}
|
|
|
|
// 1198/AC1: given a claimed send the server accepts, when sent, then sent is recorded, mail.send.sent is
|
|
// emitted (send id, account, outcome in words; no recipient, subject or body), and only then — by
|
|
// construction, since the caller acknowledges only once this call returns — the message is acknowledged.
|
|
func TestRecordOutcomeOnSuccessRecordsSentAndEmitsIt(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
emit := &memEmit{}
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
|
|
if err != nil {
|
|
t.Fatalf("recordOutcome: %v", err)
|
|
}
|
|
if d.Outcome != "sent" {
|
|
t.Fatalf("outcome: %+v", d)
|
|
}
|
|
if action != actionAck {
|
|
t.Fatalf("action: %v, want actionAck", action)
|
|
}
|
|
kept, _, ok, err := getSend(sends, req.ID)
|
|
if err != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sent" {
|
|
t.Fatalf("not kept as sent: %v %v %+v", err, ok, kept.Decided)
|
|
}
|
|
if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendSent || emit.calls[0].id != req.ID {
|
|
t.Fatalf("emitted: %+v", emit.calls)
|
|
}
|
|
body := emit.calls[0].body
|
|
if body.Send != req.ID || body.Account != "mesh" || body.Outcome == "" {
|
|
t.Fatalf("the event's body: %+v", body)
|
|
}
|
|
whole := body.Send + body.Account + body.Outcome
|
|
for _, forbidden := range []string{"ops@example.org", "s", "body"} {
|
|
if forbidden == "s" {
|
|
continue // too short to search for meaningfully without false positives in the id/account
|
|
}
|
|
if strings.Contains(whole, forbidden) {
|
|
t.Fatalf("the emitted body carries %q: %+v", forbidden, body)
|
|
}
|
|
}
|
|
}
|
|
|
|
// 1198/AC2: given a not-taken failure on attempt n (n < 5), then retrying is recorded with attempt, reason
|
|
// and next attempt, and the caller is told to ask the bus to deliver it again after 1, 5, 15, 30 minutes
|
|
// for n = 1..4; on the fifth, failed is recorded and mail.send.failed emitted.
|
|
func TestRecordOutcomeOnNotTakenRetriesUpToTheFourthAttempt(t *testing.T) {
|
|
for n, want := range map[int]time.Duration{1: time.Minute, 2: 5 * time.Minute, 3: 15 * time.Minute, 4: 30 * time.Minute} {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, n)
|
|
emit := &memEmit{}
|
|
now := judgedAt.Add(2 * time.Minute)
|
|
d, action, delay, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")}, now)
|
|
if err != nil {
|
|
t.Fatalf("attempt %d: recordOutcome: %v", n, err)
|
|
}
|
|
if d.Outcome != "retrying" || d.Attempt != n {
|
|
t.Fatalf("attempt %d: outcome %+v", n, d)
|
|
}
|
|
if !d.NextAttempt.Equal(now.Add(want)) {
|
|
t.Fatalf("attempt %d: next attempt %v, want %v", n, d.NextAttempt, now.Add(want))
|
|
}
|
|
if action != actionRetry || delay != want {
|
|
t.Fatalf("attempt %d: action %v, delay %v, want actionRetry, %v", n, action, delay, want)
|
|
}
|
|
if len(emit.calls) != 0 {
|
|
t.Fatalf("attempt %d: retrying emitted something: %+v", n, emit.calls)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRecordOutcomeOnNotTakenFailsOnTheFifthAttempt(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, MostAttempts)
|
|
emit := &memEmit{}
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: NotTaken, Err: errors.New("no connection")},
|
|
judgedAt.Add(2*time.Minute))
|
|
if err != nil {
|
|
t.Fatalf("recordOutcome: %v", err)
|
|
}
|
|
if !strings.HasPrefix(d.Outcome, "failed: ") || d.Attempt != MostAttempts {
|
|
t.Fatalf("outcome: %+v", d)
|
|
}
|
|
if action != actionTerm {
|
|
t.Fatalf("action: %v, want actionTerm", action)
|
|
}
|
|
if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed {
|
|
t.Fatalf("emitted: %+v", emit.calls)
|
|
}
|
|
}
|
|
|
|
// 1198/AC3: given a refusal, then failed is recorded with the reason, mail.send.failed emitted, and the
|
|
// message terminated; given not proven, the same with not proven.
|
|
func TestRecordOutcomeOnRefusedRecordsFailedAndTerminates(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
emit := &memEmit{}
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
|
|
judgedAt.Add(2*time.Minute))
|
|
if err != nil {
|
|
t.Fatalf("recordOutcome: %v", err)
|
|
}
|
|
if !strings.HasPrefix(d.Outcome, "failed: ") || !strings.Contains(d.Outcome, "550 no such mailbox") {
|
|
t.Fatalf("outcome: %+v", d)
|
|
}
|
|
if action != actionTerm {
|
|
t.Fatalf("action: %v, want actionTerm", action)
|
|
}
|
|
if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed || emit.calls[0].body.Outcome != d.Outcome {
|
|
t.Fatalf("emitted: %+v", emit.calls)
|
|
}
|
|
}
|
|
|
|
func TestRecordOutcomeOnNotProvenRecordsNotProvenAndTerminates(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
emit := &memEmit{}
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev,
|
|
&SendError{Outcome: NotProven, Err: errors.New("connection lost after the data was written")}, judgedAt.Add(2*time.Minute))
|
|
if err != nil {
|
|
t.Fatalf("recordOutcome: %v", err)
|
|
}
|
|
if d.Outcome != "not proven" {
|
|
t.Fatalf("outcome: %+v", d)
|
|
}
|
|
if action != actionTerm {
|
|
t.Fatalf("action: %v, want actionTerm", action)
|
|
}
|
|
if len(emit.calls) != 1 || emit.calls[0].eventType != MailSendFailed || emit.calls[0].body.Outcome != "not proven" {
|
|
t.Fatalf("emitted: %+v", emit.calls)
|
|
}
|
|
}
|
|
|
|
// review round 1 finding 3: a failed write of the tells-pending index does not stop the failed decision
|
|
// itself from being recorded and emitted — only the index write is reported as failing.
|
|
func TestRecordOutcomeStillRecordsFailedWhenTheTellsPendingIndexCannotBeWritten(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
sends.failKey, sends.failErr = tellsPendingKey, errors.New("the store is down")
|
|
emit := &memEmit{}
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
|
|
judgedAt.Add(2*time.Minute))
|
|
if err == nil {
|
|
t.Fatalf("the index failure was not reported")
|
|
}
|
|
if !strings.HasPrefix(d.Outcome, "failed: ") || action != actionTerm {
|
|
t.Fatalf("outcome %+v, action %v", d, action)
|
|
}
|
|
kept, _, ok, kerr := getSend(sends, req.ID)
|
|
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != d.Outcome {
|
|
t.Fatalf("the decision itself was not kept despite the index failing: %v %v %+v", kerr, ok, kept.Decided)
|
|
}
|
|
if len(emit.calls) != 1 {
|
|
t.Fatalf("emitted %d time(s) despite the index failing", len(emit.calls))
|
|
}
|
|
}
|
|
|
|
// review round 1 finding 5: if writing "sent" cannot be kept (even retried on a changed revision), the
|
|
// message is never acknowledged and mail.send.sent is never emitted — a mail actually sent is never
|
|
// silently announced successful when the record never actually carried it.
|
|
func TestRecordOutcomeOnSentDoesNotAckOrEmitWhenTheWriteCannotBeKept(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
sends.failKey, sends.failErr = req.ID, errors.New("the store is down")
|
|
emit := &memEmit{}
|
|
|
|
d, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, nil, judgedAt.Add(2*time.Minute))
|
|
if err == nil {
|
|
t.Fatalf("the write failure was not reported")
|
|
}
|
|
if action != actionLeave {
|
|
t.Fatalf("action: %v, want actionLeave (left to the bus, never acked)", action)
|
|
}
|
|
if len(emit.calls) != 0 {
|
|
t.Fatalf("emitted mail.send.sent despite the write never being kept: %+v", emit.calls)
|
|
}
|
|
// The kept record is unaffected: still "sending", as claimQueuedSend originally left it.
|
|
kept, _, ok, kerr := getSend(sends, req.ID)
|
|
if kerr != nil || !ok || kept.Decided == nil || kept.Decided.Outcome != "sending" {
|
|
t.Fatalf("the record changed despite the write failing: %v %v %+v (returned %+v)", kerr, ok, kept.Decided, d)
|
|
}
|
|
}
|
|
|
|
// Same finding, for a failed outcome (refused): the message is left to the bus, never terminated, and
|
|
// mail.send.failed is never emitted, when the write itself cannot be kept.
|
|
func TestRecordOutcomeOnRefusedDoesNotTerminateOrEmitWhenTheWriteCannotBeKept(t *testing.T) {
|
|
req, claimed, rev, sends := outcomeTestClaim(t, 1)
|
|
sends.failKey, sends.failErr = req.ID, errors.New("the store is down")
|
|
emit := &memEmit{}
|
|
|
|
_, action, _, err := recordOutcome(sends, emit.emit, req, claimed, rev, &SendError{Outcome: Refused, Err: errors.New("550 no such mailbox")},
|
|
judgedAt.Add(2*time.Minute))
|
|
if err == nil {
|
|
t.Fatalf("the write failure was not reported")
|
|
}
|
|
if action != actionLeave {
|
|
t.Fatalf("action: %v, want actionLeave", action)
|
|
}
|
|
if len(emit.calls) != 0 {
|
|
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)
|
|
}
|