678 lines
22 KiB
Go
678 lines
22 KiB
Go
package counting
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/alicebob/miniredis/v2"
|
|
"github.com/redis/go-redis/v9"
|
|
|
|
"migadu_counters/internal/domains"
|
|
"migadu_counters/internal/store"
|
|
)
|
|
|
|
// table is a fake domains lookup: domain -> record. Unknown domains return
|
|
// domains.ErrNotFound; the special domain "broken.tld" returns a read error.
|
|
type table map[string]*domains.Domain
|
|
|
|
func (t table) Lookup(domain string) (*domains.Domain, error) {
|
|
if domain == "broken.tld" {
|
|
return nil, errors.New("cdb read error")
|
|
}
|
|
if d, ok := t[domain]; ok {
|
|
return d, nil
|
|
}
|
|
return nil, domains.ErrNotFound
|
|
}
|
|
|
|
var (
|
|
fixedNow = time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)
|
|
day = "2026-10-13"
|
|
exempt = map[string]bool{"support@migadu.com": true, "contact@migadu.com": true}
|
|
)
|
|
|
|
func newTable() table {
|
|
return table{
|
|
"example.com": {Name: "example.com", OrgID: 7,
|
|
Limits: domains.ParseLimits("O/i=10000,O/o=2200,D/i=10000,D/o=2200,A/i=500,A/o=100,noreply@/o=5")},
|
|
"other.org": {Name: "other.org", OrgID: 9,
|
|
Limits: domains.ParseLimits("O/i=200,O/o=20,D/i=200,D/o=20,A/i=200,A/o=20")},
|
|
"nolimits.net": {Name: "nolimits.net", OrgID: 11, Limits: domains.Limits{}},
|
|
}
|
|
}
|
|
|
|
func newPlanner(countJunk bool) *Planner {
|
|
return &Planner{Lookup: newTable(), Exempt: exempt, CountJunk: countJunk, Now: func() time.Time { return fixedNow }}
|
|
}
|
|
|
|
type want struct {
|
|
level, id, flow, by string
|
|
limit int64
|
|
limited bool
|
|
}
|
|
|
|
func checkIncrements(t *testing.T, got []Increment, wants []want) {
|
|
t.Helper()
|
|
if len(got) != len(wants) {
|
|
t.Fatalf("got %d increments, want %d:\n%s", len(got), len(wants), dump(got))
|
|
}
|
|
for i, w := range wants {
|
|
g := got[i]
|
|
wantKey := store.Key(w.level, w.id, day, w.flow)
|
|
if g.Level != w.level || g.ID != w.id || g.Flow != w.flow || g.By != w.by || g.Limit != w.limit || g.Limited != w.limited || g.Key != wantKey {
|
|
t.Errorf("increment %d = %+v,\n want level=%s id=%s flow=%s by=%s limit=%d limited=%v key=%s",
|
|
i, g, w.level, w.id, w.flow, w.by, w.limit, w.limited, wantKey)
|
|
}
|
|
}
|
|
}
|
|
|
|
func dump(incs []Increment) string {
|
|
s := ""
|
|
for i, in := range incs {
|
|
s += fmt.Sprintf(" %d: %+v\n", i, in)
|
|
}
|
|
return s
|
|
}
|
|
|
|
// x@example.com -> y@example.com: sender outgoing on A/D/O, recipient incoming on A/D/O.
|
|
func TestPlanOut_LocalRecipient(t *testing.T) {
|
|
p := newPlanner(false)
|
|
plan, err := p.PlanOut("tr1", "x@example.com", "x@example.com", "y@example.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != "" || plan.Route != "out" || plan.TraceID != "tr1" || plan.Date != day {
|
|
t.Errorf("plan header = %+v", plan)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "x@example.com", "o", "x@example.com", 100, true},
|
|
{"D", "example.com", "o", "x@example.com", 2200, true},
|
|
{"O", "7", "o", "x@example.com", 2200, true},
|
|
{"A", "y@example.com", "i", "y@example.com", 500, true},
|
|
{"D", "example.com", "i", "y@example.com", 10000, true},
|
|
{"O", "7", "i", "y@example.com", 10000, true},
|
|
})
|
|
if len(plan.Lookups) != 2 || plan.Lookups[0].Role != "sender" || plan.Lookups[0].Domain == nil ||
|
|
plan.Lookups[1].Role != "recipient" || plan.Lookups[1].Domain == nil {
|
|
t.Errorf("lookups = %+v", plan.Lookups)
|
|
}
|
|
}
|
|
|
|
// x@example.com -> s@gmail.com: only the sender side; the recipient lookup is recorded as not found.
|
|
func TestPlanOut_ExternalRecipient(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", "x@example.com", "x@example.com", "s@gmail.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "x@example.com", "o", "x@example.com", 100, true},
|
|
{"D", "example.com", "o", "x@example.com", 2200, true},
|
|
{"O", "7", "o", "x@example.com", 2200, true},
|
|
})
|
|
if len(plan.Lookups) != 2 || plan.Lookups[1].Domain != nil || !errors.Is(plan.Lookups[1].Err, domains.ErrNotFound) {
|
|
t.Errorf("recipient lookup should be recorded as not found: %+v", plan.Lookups)
|
|
}
|
|
}
|
|
|
|
// Recipient in another hosted org: its own domain row decides D/O ids and limits.
|
|
func TestPlanOut_RecipientInOtherOrg(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", "x@example.com", "", "z@other.org")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments[3:], []want{
|
|
{"A", "z@other.org", "i", "z@other.org", 200, true},
|
|
{"D", "other.org", "i", "z@other.org", 200, true},
|
|
{"O", "9", "i", "z@other.org", 200, true},
|
|
})
|
|
}
|
|
|
|
func TestPlanOut_SenderDomainUnknown(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", "staff@unknown.tld", "", "y@example.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != "" {
|
|
t.Errorf("skipped = %q, want counting of the local recipient", plan.Skipped)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "y@example.com", "i", "y@example.com", 500, true},
|
|
{"D", "example.com", "i", "y@example.com", 10000, true},
|
|
{"O", "7", "i", "y@example.com", 10000, true},
|
|
})
|
|
}
|
|
|
|
func TestPlanOut_BothUnknown(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", "a@unknown.tld", "", "b@gmail.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != SkipUnknownDomain || len(plan.Increments) != 0 {
|
|
t.Errorf("plan = %+v", plan)
|
|
}
|
|
}
|
|
|
|
func TestPlanOut_Exempt(t *testing.T) {
|
|
cases := []struct{ authUser, mailFrom, recipient string }{
|
|
{"support@migadu.com", "support@migadu.com", "y@example.com"},
|
|
{"x@example.com", "x@example.com", "support@migadu.com"},
|
|
{"x@example.com", "contact@migadu.com", "y@example.com"},
|
|
{"x@example.com", "x@example.com", "Contact@Migadu.com"},
|
|
}
|
|
for _, c := range cases {
|
|
plan, err := newPlanner(false).PlanOut("tr", c.authUser, c.mailFrom, c.recipient)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != SkipExempt || len(plan.Increments) != 0 || len(plan.Lookups) != 0 {
|
|
t.Errorf("%+v: plan = %+v, want exempt with no lookups", c, plan)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPlanOut_AddressOverrideAndUnlimited(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", "noreply@example.com", "", "q@nolimits.net")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "noreply@example.com", "o", "noreply@example.com", 5, true}, // noreply@/o override
|
|
{"D", "example.com", "o", "noreply@example.com", 2200, true},
|
|
{"O", "7", "o", "noreply@example.com", 2200, true},
|
|
{"A", "q@nolimits.net", "i", "q@nolimits.net", 0, false},
|
|
{"D", "nolimits.net", "i", "q@nolimits.net", 0, false},
|
|
{"O", "11", "i", "q@nolimits.net", 0, false},
|
|
})
|
|
}
|
|
|
|
func TestPlanOut_NormalizesAddresses(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanOut("tr", " X@Example.COM ", "", "Y@EXAMPLE.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Increments[0].ID != "x@example.com" || plan.Increments[3].ID != "y@example.com" {
|
|
t.Errorf("ids not normalised: %s", dump(plan.Increments))
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_LocalRecipient(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanIn("tr9", "s@gmail.com", "y@example.com", false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Route != "in" || plan.TraceID != "tr9" || plan.Skipped != "" {
|
|
t.Errorf("plan header = %+v", plan)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "y@example.com", "i", "y@example.com", 500, true},
|
|
{"D", "example.com", "i", "y@example.com", 10000, true},
|
|
{"O", "7", "i", "y@example.com", 10000, true},
|
|
})
|
|
if len(plan.Lookups) != 1 || plan.Lookups[0].Role != "recipient" {
|
|
t.Errorf("lookups = %+v", plan.Lookups)
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_SenderNeverCounted(t *testing.T) {
|
|
// A hosted sender on /in (e.g. mail relayed back in) still only counts the recipient.
|
|
plan, err := newPlanner(false).PlanIn("tr", "x@example.com", "z@other.org", false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, in := range plan.Increments {
|
|
if in.Flow != "i" || in.By != "z@other.org" {
|
|
t.Errorf("unexpected increment %+v", in)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_UnknownDomain(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanIn("tr", "s@gmail.com", "nobody@gmail.com", false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != SkipUnknownDomain || len(plan.Increments) != 0 {
|
|
t.Errorf("plan = %+v", plan)
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_Junk(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanIn("tr", "s@gmail.com", "y@example.com", true)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != SkipJunk || len(plan.Increments) != 0 || len(plan.Lookups) != 0 {
|
|
t.Errorf("junk not skipped: %+v", plan)
|
|
}
|
|
plan, err = newPlanner(true).PlanIn("tr", "s@gmail.com", "y@example.com", true)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != "" || len(plan.Increments) != 3 {
|
|
t.Errorf("count_junk=true should count: %+v", plan)
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_Exempt(t *testing.T) {
|
|
for _, c := range []struct{ from, rcpt string }{{"support@migadu.com", "y@example.com"}, {"s@gmail.com", "contact@migadu.com"}} {
|
|
plan, err := newPlanner(false).PlanIn("tr", c.from, c.rcpt, false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != SkipExempt || len(plan.Increments) != 0 {
|
|
t.Errorf("%+v: plan = %+v", c, plan)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPlanIn_EmptyMailFromIsBounce(t *testing.T) {
|
|
plan, err := newPlanner(false).PlanIn("tr", "", "y@example.com", false)
|
|
if err != nil || plan.Skipped != "" || len(plan.Increments) != 3 {
|
|
t.Errorf("bounce (<>) must still count the recipient: %+v, %v", plan, err)
|
|
}
|
|
}
|
|
|
|
// A CDB read error (not "not found") must surface so the server answers 503.
|
|
func TestPlan_LookupError(t *testing.T) {
|
|
if _, err := newPlanner(false).PlanIn("tr", "s@gmail.com", "y@broken.tld", false); err == nil {
|
|
t.Error("PlanIn: expected error")
|
|
}
|
|
if _, err := newPlanner(false).PlanOut("tr", "x@broken.tld", "", "y@example.com"); err == nil {
|
|
t.Error("PlanOut: expected error")
|
|
}
|
|
}
|
|
|
|
// Against the real CDB: yagolivas.com is org 1049 with A/o=20, D/o=20, O/o=20.
|
|
func TestPlanOut_RealCDB(t *testing.T) {
|
|
db, err := domains.Open("../../test_files/rspamd_domains_db.cdb")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
p := &Planner{Lookup: db, Exempt: exempt, Now: func() time.Time { return fixedNow }}
|
|
plan, err := p.PlanOut("tr", "someone@yagolivas.com", "", "ext@gmail.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "someone@yagolivas.com", "o", "someone@yagolivas.com", 20, true},
|
|
{"D", "yagolivas.com", "o", "someone@yagolivas.com", 20, true},
|
|
{"O", "1049", "o", "someone@yagolivas.com", 20, true},
|
|
})
|
|
}
|
|
|
|
func TestCrossed(t *testing.T) {
|
|
tests := []struct {
|
|
newValue, amount, limit int64
|
|
want bool
|
|
}{
|
|
{1, 1, 1, true},
|
|
{5, 1, 5, true},
|
|
{4, 1, 5, false},
|
|
{6, 1, 5, false}, // already over: no second notification
|
|
{5, 3, 5, true}, // 2 -> 5 crosses
|
|
{7, 3, 5, true}, // 4 -> 7 crosses
|
|
{8, 3, 5, false}, // 5 -> 8 was already at the limit
|
|
{1, 1, 0, false}, // 0 / unlimited never crosses
|
|
{1, 1, -1, false},
|
|
}
|
|
for _, tt := range tests {
|
|
if got := Crossed(tt.newValue, tt.amount, tt.limit); got != tt.want {
|
|
t.Errorf("Crossed(%d, %d, %d) = %v, want %v", tt.newValue, tt.amount, tt.limit, got, tt.want)
|
|
}
|
|
}
|
|
}
|
|
|
|
func newCounter(t *testing.T) (*Counter, *miniredis.Miniredis) {
|
|
t.Helper()
|
|
m := miniredis.RunT(t)
|
|
st := store.NewWithClients(redis.NewClient(&redis.Options{Addr: m.Addr()}), nil)
|
|
t.Cleanup(func() { st.Close() })
|
|
return &Counter{Planner: newPlanner(false), Store: st}, m
|
|
}
|
|
|
|
// Executes the plan: counts come back from Redis and the crossing increment
|
|
// is flagged exactly once.
|
|
func TestCounter_Out(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
ctx := context.Background()
|
|
// noreply@example.com has A/o=5: the fifth message reaches it.
|
|
var last *Outcome
|
|
for i := 1; i <= 6; i++ {
|
|
out, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(out.Results) != 3 || out.Results[0].Count != int64(i) {
|
|
t.Fatalf("message %d: results = %+v", i, out.Results)
|
|
}
|
|
if reached := out.Results[0].Reached; reached != (i == 5) {
|
|
t.Errorf("message %d: A reached = %v", i, reached)
|
|
}
|
|
last = out
|
|
}
|
|
if v, _ := m.Get(store.Key("A", "noreply@example.com", day, "o")); v != "6" {
|
|
t.Errorf("redis A/o = %q, want 6", v)
|
|
}
|
|
if v, _ := m.Get(store.Key("O", "7", day, "o")); v != "6" {
|
|
t.Errorf("redis O/o = %q, want 6", v)
|
|
}
|
|
if last.Skipped != "" || last.Route != "out" {
|
|
t.Errorf("outcome = %+v", last)
|
|
}
|
|
}
|
|
|
|
// After the web app resets a counter (DEL), the next crossing fires again.
|
|
func TestCounter_ReachedAgainAfterReset(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
ctx := context.Background()
|
|
for i := 0; i < 5; i++ {
|
|
if _, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
m.Del(store.Key("A", "noreply@example.com", day, "o"))
|
|
var reachedAt int
|
|
for i := 1; i <= 5; i++ {
|
|
out, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if out.Results[0].Reached {
|
|
reachedAt = i
|
|
}
|
|
}
|
|
if reachedAt != 5 {
|
|
t.Errorf("after reset the limit was reached at message %d, want 5", reachedAt)
|
|
}
|
|
}
|
|
|
|
func TestCounter_In(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
out, err := c.CountIn(context.Background(), "tr", "s@gmail.com", "y@example.com", false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(out.Results) != 3 || out.Results[2].Count != 1 || out.Results[2].Level != "O" {
|
|
t.Errorf("results = %+v", out.Results)
|
|
}
|
|
if v, _ := m.Get(store.Key("D", "example.com", day, "i")); v != "1" {
|
|
t.Errorf("redis D/i = %q, want 1", v)
|
|
}
|
|
}
|
|
|
|
// A skipped plan must not touch Redis at all.
|
|
func TestCounter_SkippedDoesNotTouchRedis(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
out, err := c.CountIn(context.Background(), "tr", "support@migadu.com", "y@example.com", false)
|
|
if err != nil || out.Skipped != SkipExempt || len(out.Results) != 0 {
|
|
t.Fatalf("outcome = %+v, %v", out, err)
|
|
}
|
|
if keys := m.Keys(); len(keys) != 0 {
|
|
t.Errorf("redis keys = %v, want none", keys)
|
|
}
|
|
}
|
|
|
|
func TestCounter_RedisDown(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
m.Close()
|
|
if _, err := c.CountIn(context.Background(), "tr", "s@gmail.com", "y@example.com", false); err == nil {
|
|
t.Error("expected error when redis is down")
|
|
}
|
|
}
|
|
|
|
// --- Jobs with unusable sides (review #2, #3) --------------------------------
|
|
|
|
func TestPlan_OutSenderUnusableCountsRecipient(t *testing.T) {
|
|
job := Job{Route: RouteOut, TraceID: "u1", Recipient: "y@example.com",
|
|
SenderUnusable: Unusable{Role: "sender", Value: "nobody", Reason: "invalid e-mail address"}}
|
|
plan, err := newPlanner(false).Plan(job)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "y@example.com", "i", "y@example.com", 500, true},
|
|
{"D", "example.com", "i", "y@example.com", 10000, true},
|
|
{"O", "7", "i", "y@example.com", 10000, true},
|
|
})
|
|
if len(plan.Unusable) != 1 || plan.Unusable[0].Role != "sender" || plan.Unusable[0].Value != "nobody" {
|
|
t.Errorf("unusable = %+v", plan.Unusable)
|
|
}
|
|
if len(plan.Lookups) != 1 || plan.Lookups[0].Role != "recipient" {
|
|
t.Errorf("only the recipient should be looked up: %+v", plan.Lookups)
|
|
}
|
|
}
|
|
|
|
func TestPlan_OutRecipientUnusableCountsSender(t *testing.T) {
|
|
job := Job{Route: RouteOut, TraceID: "u2", AuthUser: "x@example.com", LoginSuffix: "SUPPORTTOKEN", MailFrom: "x@example.com",
|
|
RecipientUnusable: Unusable{Role: "recipient", Value: "u@[192.0.2.1]", Reason: "invalid e-mail address"}}
|
|
plan, err := newPlanner(false).Plan(job)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "x@example.com", "o", "x@example.com", 100, true},
|
|
{"D", "example.com", "o", "x@example.com", 2200, true},
|
|
{"O", "7", "o", "x@example.com", 2200, true},
|
|
})
|
|
if plan.LoginSuffix != "SUPPORTTOKEN" || plan.AuthUser != "x@example.com" {
|
|
t.Errorf("login suffix not carried: %+v", plan)
|
|
}
|
|
if len(plan.Unusable) != 1 || plan.Unusable[0].Role != "recipient" {
|
|
t.Errorf("unusable = %+v", plan.Unusable)
|
|
}
|
|
}
|
|
|
|
// An unusable mail_from on /in is logged, treated as non-exempt, and the
|
|
// recipient is counted.
|
|
func TestPlan_InMailFromUnusable(t *testing.T) {
|
|
job := Job{Route: RouteIn, TraceID: "u3", MailFrom: "john smith@x.com", Recipient: "y@example.com"}
|
|
plan, err := newPlanner(false).Plan(job)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if plan.Skipped != "" || len(plan.Increments) != 3 || plan.MailFrom != "john smith@x.com" {
|
|
t.Errorf("plan = %+v", plan)
|
|
}
|
|
}
|
|
|
|
// The exempt check still applies to usable sides of a Job.
|
|
func TestPlan_ExemptViaJob(t *testing.T) {
|
|
plan, err := newPlanner(false).Plan(Job{Route: RouteIn, MailFrom: "support@migadu.com", MailFromUsable: true, Recipient: "y@example.com"})
|
|
if err != nil || plan.Skipped != SkipExempt {
|
|
t.Errorf("plan = %+v, %v", plan, err)
|
|
}
|
|
}
|
|
|
|
// --- Execute: partial EXEC failure and lost replies (review #4) --------------
|
|
|
|
// One key holding garbage fails its own INCR; the others are counted and the
|
|
// crossing among them is still detected.
|
|
func TestCounter_PartialExecFailure(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
m.Set(store.Key("O", "7", day, "o"), "garbage")
|
|
m.Set(store.Key("A", "noreply@example.com", day, "o"), "4") // A/o=5: this increment crosses
|
|
out, err := c.CountOut(context.Background(), "p1", "noreply@example.com", "", "ext@gmail.com")
|
|
if err != nil {
|
|
t.Fatalf("a per-key failure must not fail the request: %v", err)
|
|
}
|
|
if len(out.Results) != 3 {
|
|
t.Fatalf("results = %+v", out.Results)
|
|
}
|
|
a, d, o := out.Results[0], out.Results[1], out.Results[2]
|
|
if a.Err != nil || a.Count != 5 || !a.Reached {
|
|
t.Errorf("A = %+v, want count 5 reached", a)
|
|
}
|
|
if d.Err != nil || d.Count != 1 {
|
|
t.Errorf("D = %+v", d)
|
|
}
|
|
if o.Err == nil || o.Reached {
|
|
t.Errorf("O = %+v, want its own error and no crossing", o)
|
|
}
|
|
if out.Unknown != nil {
|
|
t.Errorf("outcome is known: %v", out.Unknown)
|
|
}
|
|
}
|
|
|
|
// delayProxy forwards TCP to backend and, once it has seen EXEC in the
|
|
// request stream, holds back the next `delayed` replies by `delay`, so the
|
|
// client's read timeout fires after Redis has applied the transaction (the
|
|
// connection handshake replies pass undelayed).
|
|
func delayProxy(t *testing.T, backend string, delayed int, delay time.Duration) string {
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { ln.Close() })
|
|
var remaining atomic.Int64
|
|
remaining.Store(int64(delayed))
|
|
go func() {
|
|
for {
|
|
c, err := ln.Accept()
|
|
if err != nil {
|
|
return
|
|
}
|
|
b, err := net.Dial("tcp", backend)
|
|
if err != nil {
|
|
c.Close()
|
|
continue
|
|
}
|
|
var armed atomic.Bool
|
|
go func() {
|
|
defer b.Close()
|
|
buf := make([]byte, 32*1024)
|
|
for {
|
|
n, err := c.Read(buf)
|
|
if n > 0 {
|
|
if bytes.Contains(bytes.ToLower(buf[:n]), []byte("exec")) {
|
|
armed.Store(true)
|
|
}
|
|
if _, werr := b.Write(buf[:n]); werr != nil {
|
|
return
|
|
}
|
|
}
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
go func() {
|
|
defer c.Close()
|
|
buf := make([]byte, 32*1024)
|
|
for {
|
|
n, err := b.Read(buf)
|
|
if n > 0 {
|
|
if armed.Load() && remaining.Add(-1) >= 0 {
|
|
time.Sleep(delay)
|
|
}
|
|
if _, werr := c.Write(buf[:n]); werr != nil {
|
|
return
|
|
}
|
|
}
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
}()
|
|
return ln.Addr().String()
|
|
}
|
|
|
|
// The EXEC reply is lost after Redis applied the increments: the counter is
|
|
// re-read from the master and the crossing is still reported.
|
|
func TestCounter_LostReplyReReadsMaster(t *testing.T) {
|
|
m := miniredis.RunT(t)
|
|
proxy := delayProxy(t, m.Addr(), 1, 800*time.Millisecond)
|
|
st := store.NewWithClients(redis.NewClient(&redis.Options{Addr: proxy, ReadTimeout: 200 * time.Millisecond, PoolSize: 2}), nil)
|
|
t.Cleanup(func() { st.Close() })
|
|
c := &Counter{Planner: newPlanner(false), Store: st}
|
|
m.Set(store.Key("A", "noreply@example.com", day, "o"), "4") // A/o=5
|
|
|
|
out, err := c.CountOut(context.Background(), "lost", "noreply@example.com", "", "ext@gmail.com")
|
|
if err != nil {
|
|
t.Fatalf("lost reply must be resolved by a re-read, got error: %v", err)
|
|
}
|
|
if out.Unknown == nil {
|
|
t.Fatal("outcome should be flagged unknown (reply lost)")
|
|
}
|
|
if v, _ := m.Get(store.Key("A", "noreply@example.com", day, "o")); v != "5" {
|
|
t.Fatalf("redis A/o = %q, the increment should have been applied", v)
|
|
}
|
|
if a := out.Results[0]; a.Count != 5 || !a.Reached {
|
|
t.Errorf("re-read A = %+v, want count 5 reached", a)
|
|
}
|
|
if o := out.Results[2]; o.Count != 1 || o.Reached {
|
|
t.Errorf("re-read O = %+v", o)
|
|
}
|
|
}
|
|
|
|
// A definite failure (connection refused) is still an error: nothing counted.
|
|
func TestCounter_RefusedIsNotReRead(t *testing.T) {
|
|
c, m := newCounter(t)
|
|
m.Close()
|
|
out, err := c.CountIn(context.Background(), "tr", "s@gmail.com", "y@example.com", false)
|
|
if err == nil {
|
|
t.Errorf("expected error, got outcome %+v", out)
|
|
}
|
|
}
|
|
|
|
// --- review #7: plus-addressing counts against the mailbox -------------------
|
|
|
|
func TestPlan_PlusTagCountsMailbox(t *testing.T) {
|
|
p := newPlanner(false)
|
|
plan, err := p.PlanOut("pt", "noreply+news@example.com", "", "y+1@example.com")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "noreply@example.com", "o", "noreply@example.com", 5, true}, // override noreply@/o still applies
|
|
{"D", "example.com", "o", "noreply@example.com", 2200, true},
|
|
{"O", "7", "o", "noreply@example.com", 2200, true},
|
|
{"A", "y@example.com", "i", "y@example.com", 500, true},
|
|
{"D", "example.com", "i", "y@example.com", 10000, true},
|
|
{"O", "7", "i", "y@example.com", 10000, true},
|
|
})
|
|
if plan.Recipient != "y+1@example.com" || plan.AuthUser != "noreply+news@example.com" {
|
|
t.Errorf("the request line keeps the addresses as sent: %+v", plan)
|
|
}
|
|
// Exempt applies to the mailbox too.
|
|
plan, _ = p.PlanIn("pt", "s@gmail.com", "support+ticket@migadu.com", false)
|
|
if plan.Skipped != SkipExempt {
|
|
t.Errorf("support+ticket@ should be exempt: %+v", plan)
|
|
}
|
|
}
|
|
|
|
// --- review #8: a malformed CDB row is "not hosted", not a failure -----------
|
|
|
|
func TestPlan_MalformedRowIsNotHosted(t *testing.T) {
|
|
db, err := domains.Open("../../review/_proof_tests/b/bad.cdb")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
p := &Planner{Lookup: db, Now: func() time.Time { return fixedNow }}
|
|
plan, err := p.PlanOut("m", "a@good.com", "", "b@bad.com")
|
|
if err != nil {
|
|
t.Fatalf("malformed recipient row must not fail the request: %v", err)
|
|
}
|
|
checkIncrements(t, plan.Increments, []want{
|
|
{"A", "a@good.com", "o", "a@good.com", 5, true},
|
|
{"D", "good.com", "o", "a@good.com", 0, false},
|
|
{"O", "7", "o", "a@good.com", 0, false},
|
|
})
|
|
if len(plan.Lookups) != 2 || plan.Lookups[1].Domain != nil || !errors.Is(plan.Lookups[1].Err, domains.ErrMalformed) {
|
|
t.Errorf("recipient lookup should record the malformed row: %+v", plan.Lookups)
|
|
}
|
|
plan, err = p.PlanIn("m", "s@gmail.com", "b@bad.com", false)
|
|
if err != nil || plan.Skipped != SkipUnknownDomain || len(plan.Increments) != 0 {
|
|
t.Errorf("/in to a malformed row: plan=%+v err=%v, want skipped unknown_domain", plan, err)
|
|
}
|
|
}
|