From 54f9626121ad82a3035ebe88bbb7c15eca16d13d Mon Sep 17 00:00:00 2001 From: Dimitrije Dimitrijevic Date: Mon, 9 Nov 2026 09:01:56 +0100 Subject: [PATCH] Add 80% limit warning, stop repeat notifications, fix review findings Notifications: - Warn once when a counter reaches warn_percent of its limit, through separate warn_*_url endpoints. - Write ntf/ markers so a lost Redis reply cannot re-send a breach. - Retry only when the request never left or the web app answered 429/5xx. Fixes: - Retry transactions Redis rejected; treat pool timeouts as definite. - Install the signal handler before startup work. - Stop logging the login suffix; redact notify URL credentials. - Reject corrupt or empty CDBs, truncated hash tables included. - Require a Sentinel password, refuse example placeholders, and default admin_addr to localhost. - Normalise /counters domains, count on the arrival day, log rejections, hide internals in errors and exit 1 on a failed drain. Co-Authored-By: Claude Opus 5.5 --- README.md | 70 +++++-- cmd/counterctl/main.go | 16 +- cmd/counterctl/main_test.go | 35 ++++ cmd/migadu_counters/main.go | 33 ++- cmd/migadu_counters/main_test.go | 47 +++++ config.toml.example | 16 +- internal/api/api_test.go | 33 ++- internal/api/validate.go | 45 +++-- internal/config/config.go | 77 ++++++- internal/config/config_test.go | 59 +++++- internal/counting/counting.go | 143 ++++++++++--- internal/counting/counting_test.go | 6 +- internal/counting/notifications_test.go | 255 ++++++++++++++++++++++++ internal/domains/domains.go | 46 ++++- internal/domains/domains_test.go | 58 ++++++ internal/metrics/metrics.go | 44 ++-- internal/metrics/metrics_test.go | 8 +- internal/notify/notify.go | 150 ++++++++++---- internal/notify/notify_test.go | 147 +++++++++++++- internal/server/middleware.go | 11 + internal/server/notifications_test.go | 173 ++++++++++++++++ internal/server/server.go | 68 +++++-- internal/server/server_test.go | 21 +- internal/store/markers_test.go | 132 ++++++++++++ internal/store/store.go | 57 +++++- internal/txlog/txlog.go | 63 +++++- internal/txlog/txlog_test.go | 24 ++- review/_proof_tests/README.md | 14 ++ todo.md | 67 ++----- 29 files changed, 1673 insertions(+), 245 deletions(-) create mode 100644 internal/counting/notifications_test.go create mode 100644 internal/server/notifications_test.go create mode 100644 internal/store/markers_test.go diff --git a/README.md b/README.md index b57fcc8..c8c40ba 100644 --- a/README.md +++ b/README.md @@ -90,20 +90,46 @@ Redis failures are handled per outcome: a key whose own `INCR` fails (it holds a non-number) is logged `redis error on key …` while the other keys are counted and checked for crossings; when the EXEC reply is lost after the write (timeout, reset) the keys are re-read once from the master and the block says `redis -outcome unknown … re-read from master`, with a crossing reported when -`count >= limit`; a definite failure (connection refused) logs `redis error: … -(nothing counted)`. +outcome unknown … re-read from master`; a transaction Redis rejected as a whole +(READONLY or EXECABORT during a failover) applied nothing and is retried once +after 200 ms; a definite failure (connection refused, or a rejection that +repeats) logs `redis error: … (nothing counted)`. The CDB is pulled from the agent host by `contrib/pull-domains.sh` (see `contrib/README.md`): rsync, compare, verify, atomic `mv`, every 15 s. A limit is "reached" by the increment that moves the counter from below the -limit to at or above it (`old < limit ≤ new`). Because INCRBY is atomic exactly +limit to at or above it (`old < limit ≤ new`). Because INCR is atomic exactly one increment crosses, so the web app is notified once per level/id/flow/day — `GET ` with `$LEVEL $ID $FLOW $PERIOD(=d) $ADDRESS $COUNT $LIMIT` substituted (URL-escaped), basic auth, in the -background, 5 attempts with exponential backoff; 4xx is not retried. An empty -URL disables notifications for that level. +background. An empty URL disables notifications for that level. + +**Early warning.** The same rule at `warn_percent` of the limit (default 80, +rounded up, `0` = off) calls `warn_org_url` / `warn_address_url` / +`warn_domain_url`, which also accept `$PERCENT`. No warning URL means the +feature is off. A limit too small to have a threshold below it (1 or 2 at 80%) +has no warning. The warning URLs must not use the breach URL's path, because +the web app treats every call there as a breach; the daemon refuses to start +if they do. The web app endpoint is `GET /api/limits/warning`. + +**No repeats.** A breach or warning is sent once: + +- A confirmed crossing writes a marker, `ntf///d///` + (TTL 48 h). After a lost Redis reply a notification goes out only if `SET NX` + on that marker succeeds, so a counter that already notified stays silent. If + Redis cannot answer, nothing is sent and `_redis_errors_total{op="marker"}` rises. +- An attempt is retried only when the request never left or the web app answered + 429 or 5xx, up to 5 attempts with exponential backoff. A request that was sent + but got no answer may have been acted on, so it is not retried; it is logged + as `outcome unknown`. +- The web app may `DEL` a day key to reset a customer; the next confirmed + crossing notifies again, and the web app's once-per-day guard decides whether + an email goes out. + +Delivery is therefore at most once. A notification that fails, or is in flight +when the daemon stops past its 30 s deadline, is not sent later that day; look +for `failed limit` in the counting log and `_notifications_total{result="failed"}`. ## Counting transaction log @@ -125,8 +151,10 @@ Every request writes one block in the shape of the old Lua plugin's log lines `grep ''` returns one transaction; `grep 'skipped:'` every request that counted nothing; `grep 'not usable'` every side that failed validation (`sender address not usable: nobody (invalid e-mail address, not counted)`); -`grep 'dropped:'` every request refused by a full queue. A token login shows as -`… login_suffix=SUPPORTTOKEN` on the request line. `counting_format = "json"` emits the same events as one +`grep 'dropped:'` every request refused by a full queue; `grep 'rejected:'` +every `/in` or `/out` body refused by validation. A `user@dom@SUFFIX` login +shows as `… token_login=true` on the request line; the suffix may be a +credential and is never logged. `counting_format = "json"` emits the same events as one JSON object per line. Domains with thousands of per-mailbox overrides print the base keys plus the address's own overrides and a count of the rest. @@ -140,16 +168,19 @@ migadu_counters -version Foreground process; two listeners: `addr` (API, default `:5555`) and `admin_addr` (`/health`, `/metrics`, no auth). Signals: `SIGHUP` reopens both log files and reloads the CDB; `SIGTERM`/`SIGINT` stop accepting, drain the -counting queue (every accepted request is counted), wait for in-flight -notifications (≤ 30 s) and exit 0. On FreeBSD run it like rcptd: an rc.d script +counting queue, wait for in-flight notifications and their retries (≤ 30 s) +and exit 0. If the deadline passes first, the remaining jobs are lost and the +exit code is 1. On FreeBSD run it like rcptd: an rc.d script with `/usr/sbin/daemon -f -H -p -o `; `service migadu_counters reload` sends SIGHUP. There is deliberately no self-daemonising -or pidfile option (daemon(8) owns both). Startup fails without a readable CDB; -an unreachable Redis is only warned about (`/health` answers 503 until it is +or pidfile option (daemon(8) owns both). Startup fails without a readable, non-empty +CDB; an unreachable Redis is only warned about (`/health` answers 503 until it is back). -See `config.toml.example` for every option. `auth_token` is mandatory and -unknown keys (typos such as `auth-token`) are fatal at startup. Secrets may +See `config.toml.example` for every option. `auth_token` is mandatory, the +example's `"…"` placeholders are refused, `admin_addr` defaults to +`127.0.0.1:9095`, `redis.password` is required with Sentinel (unless +`redis.password_optional = true`) and unknown keys (typos such as `auth-token`) are fatal at startup. Secrets may come from the environment: `MIGADU_COUNTERS_AUTH_TOKEN`, `MIGADU_COUNTERS_REDIS_PASSWORD`, `MIGADU_COUNTERS_NOTIFY_PASSWORD`. `redis.db` must stay 0. Production uses Sentinel (`sentinels`, `master_name = "mymaster"`): writes go to the master, @@ -158,8 +189,9 @@ reads to a replica (falling back to the master when none is known); without Metrics (`/metrics` on the admin port): `migadu_counters_requests_total{route,status}`, `_request_duration_seconds{route}`, `_increments_total{level,flow}`, -`_limit_reached_total{level,flow}`, `_notifications_total{level,result}`, -`_redis_errors_total{op}` (`incr`, `incr_unknown`, `mget`, `ping`), +`_limit_reached_total{level,flow}`, `_warning_reached_total{level,flow}`, +`_notifications_total{kind,level,result}` (`kind`: `limit`, `warning`), +`_redis_errors_total{op}` (`incr`, `incr_unknown`, `mget`, `ping`, `marker`), `_dropped_total{route}`, `_queue_depth`, `_queue_wait_seconds`, `_cdb_domains`, `_cdb_reloads_total{result}`. @@ -195,12 +227,16 @@ on any error. ``` make build # bin/migadu_counters, bin/counterctl -make test # go test -race ./... (tests use test_files/rspamd_domains_db.cdb and miniredis) +make test # go test -race ./... (tests use miniredis and test_files/rspamd_domains_db.cdb) make vet # go vet + gofmt make test-contrib # shell test for contrib/pull-domains.sh (needs tinycdb cdb + rsync) make freebsd # static FreeBSD amd64 binaries for the Ansible role ``` +`test_files/` is gitignored because the CDB holds real customer domains, so on +a fresh clone the tests fail until you copy `rspamd_domains_db` from an agent +host (`/usr/local/etc/agent/build/rspamd/`) to `test_files/rspamd_domains_db.cdb`. + Manual check with a local Redis: `redis-server --port 6379`, a config in static mode (`write_addr = "127.0.0.1:6379"`) pointing `domains_cdb` at `test_files/rspamd_domains_db.cdb`, then diff --git a/cmd/counterctl/main.go b/cmd/counterctl/main.go index 570fe0d..a54b650 100644 --- a/cmd/counterctl/main.go +++ b/cmd/counterctl/main.go @@ -78,7 +78,12 @@ func run(args []string, stdout, stderr io.Writer, getenv func(string) string, no } q.Set("address", addr) } else { - q.Set("domain", strings.ToLower(strings.TrimSpace(*domain))) + dom, err := api.NormalizeDomain(*domain) + if err != nil { + fmt.Fprintf(stderr, "invalid domain %q: %v\n", *domain, err) + return 1 + } + q.Set("domain", dom) } if *dateFlag != "" { day, err := api.ParseDate(*dateFlag, now()) @@ -87,7 +92,8 @@ func run(args []string, stdout, stderr io.Writer, getenv func(string) string, no return 1 } q.Set("date", day) - if t, _ := time.Parse("2006-01-02", day); now().UTC().Sub(t) > 31*24*time.Hour { + today, _ := api.ParseDate("", now()) + if d, t := mustDay(today), mustDay(day); d.Sub(t) > 31*24*time.Hour { fmt.Fprintf(stderr, "warning: %s is more than 31 days ago; counter keys expire after 31 days and will read as 0\n", day) } } @@ -154,6 +160,12 @@ func run(args []string, stdout, stderr io.Writer, getenv func(string) string, no return 0 } +// mustDay parses a date api.ParseDate already validated. +func mustDay(s string) time.Time { + t, _ := time.Parse("2006-01-02", s) + return t +} + // flows picks the table columns: --in or --out alone restrict the table to // that flow; neither or both show both flows. func flows(in, out bool) (showIn, showOut bool) { diff --git a/cmd/counterctl/main_test.go b/cmd/counterctl/main_test.go index fb7d1b0..26d2e8b 100644 --- a/cmd/counterctl/main_test.go +++ b/cmd/counterctl/main_test.go @@ -57,6 +57,8 @@ func (f *fakeAPI) last() *http.Request { return f.requests[len(f.requests)-1] } +func (f *fakeAPI) rawQuery() string { return f.last().URL.RawQuery } + func writeConfig(t *testing.T, body string) string { t.Helper() p := filepath.Join(t.TempDir(), "config.toml") @@ -333,3 +335,36 @@ func TestRun_Version(t *testing.T) { t.Errorf("exit %d output %q", code, out) } } + +// Keys live 31 days from their last increment, so a date exactly 31 days back +// can still hold data. +func TestDateWarning_Boundary(t *testing.T) { + f := &fakeAPI{body: sampleResponse()} + s := httptest.NewServer(f) + defer s.Close() + cfg := writeConfig(t, "auth_token = \"t\"\n") + exact := now.UTC().AddDate(0, 0, -31).Format("2006-01-02") + if _, _, errOut := runCLI(t, noEnv, "--config", cfg, "--url", s.URL, "--address", "me@example.com", "--date", exact); strings.Contains(errOut, "will read as 0") { + t.Errorf("31 days back must not warn: %q", errOut) + } + older := now.UTC().AddDate(0, 0, -32).Format("2006-01-02") + if _, _, errOut := runCLI(t, noEnv, "--config", cfg, "--url", s.URL, "--address", "me@example.com", "--date", older); !strings.Contains(errOut, "will read as 0") { + t.Errorf("32 days back must warn: %q", errOut) + } +} + +func TestDomainFlag_Normalised(t *testing.T) { + f := &fakeAPI{body: sampleResponse()} + s := httptest.NewServer(f) + defer s.Close() + cfg := writeConfig(t, "auth_token = \"t\"\n") + if code, _, errOut := runCLI(t, noEnv, "--config", cfg, "--url", s.URL, "--domain", "Bücher.de"); code != 0 { + t.Fatalf("exit %d: %s", code, errOut) + } + if !strings.Contains(f.rawQuery(), "domain=xn--bcher-kva.de") { + t.Errorf("query = %q", f.rawQuery()) + } + if code, _, _ := runCLI(t, noEnv, "--config", cfg, "--url", s.URL, "--domain", "a..b"); code == 0 { + t.Error("an invalid domain must exit non-zero") + } +} diff --git a/cmd/migadu_counters/main.go b/cmd/migadu_counters/main.go index 65ce426..fc2723f 100644 --- a/cmd/migadu_counters/main.go +++ b/cmd/migadu_counters/main.go @@ -44,6 +44,9 @@ var ( date = "unknown" ) +// beforeStartup lets tests act right after the signal handler is installed. +var beforeStartup func() + func main() { os.Exit(run(os.Args[1:], os.Stdout, os.Stderr, nil)) } @@ -63,6 +66,18 @@ func run(args []string, stdout, stderr io.Writer, sigs <-chan os.Signal) int { return 0 } + // Installed before anything slow: a SIGHUP's default action is to kill the + // process. Signals that arrive during startup wait in the channel. + if sigs == nil { + ch := make(chan os.Signal, 8) + signal.Notify(ch, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT) + defer signal.Stop(ch) + sigs = ch + } + if beforeStartup != nil { + beforeStartup() + } + path, err := config.Discover(*configPath) if err != nil { fmt.Fprintln(stderr, err) @@ -93,12 +108,6 @@ func run(args []string, stdout, stderr io.Writer, sigs <-chan os.Signal) int { } a.Serve(apiLn, adminLn) - if sigs == nil { - ch := make(chan os.Signal, 2) - signal.Notify(ch, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT) - defer signal.Stop(ch) - sigs = ch - } for { select { case sig := <-sigs: @@ -115,6 +124,7 @@ func run(args []string, stdout, stderr io.Writer, sigs <-chan os.Signal) int { cancel() if err != nil { fmt.Fprintf(stderr, "shutdown: %v\n", err) + return 1 } return 0 case err := <-a.fatal: @@ -166,6 +176,11 @@ func newApp(cfg *config.Config) (*app, error) { logWriter.Close() return nil, err } + if db.Count() == 0 { + db.Close() + logWriter.Close() + return nil, fmt.Errorf("domains db %s holds no records", cfg.DomainsCDB) + } a.db.Store(db) logger.Info("domains db loaded", "path", cfg.DomainsCDB, "domains", db.Count(), "max_cached_domains", cfg.MaxCachedDomains) @@ -209,6 +224,7 @@ func newApp(cfg *config.Config) (*app, error) { a.notifier = notify.New(notify.Config{ OrgURL: cfg.Notify.OrgURL, AddressURL: cfg.Notify.AddressURL, DomainURL: cfg.Notify.DomainURL, + WarnOrgURL: cfg.Notify.WarnOrgURL, WarnAddressURL: cfg.Notify.WarnAddressURL, WarnDomainURL: cfg.Notify.WarnDomainURL, User: cfg.Notify.User, Password: cfg.Notify.Password, Timeout: cfg.Notify.Timeout(), }, server.NotifyHooks(tx, a.m)) @@ -220,8 +236,9 @@ func newApp(cfg *config.Config) (*app, error) { } return db.Lookup(domain) }), - Exempt: cfg.ExemptSet(), - CountJunk: cfg.CountJunk, + Exempt: cfg.ExemptSet(), + CountJunk: cfg.CountJunk, + WarnPercent: cfg.Notify.EffectiveWarnPercent(), } a.srv = server.New(server.Deps{ AuthToken: cfg.AuthToken, diff --git a/cmd/migadu_counters/main_test.go b/cmd/migadu_counters/main_test.go index c130402..946262f 100644 --- a/cmd/migadu_counters/main_test.go +++ b/cmd/migadu_counters/main_test.go @@ -415,3 +415,50 @@ func TestRun_SignalLoop(t *testing.T) { t.Errorf("log:\n%s", e.logs()) } } + +// A SIGHUP while the daemon starts must not kill it, and the reload it asked +// for runs once the daemon is ready. +func TestRun_SIGHUPDuringStartupIsQueued(t *testing.T) { + e := newEnv(t, "") + beforeStartup = func() { syscall.Kill(os.Getpid(), syscall.SIGHUP) } + t.Cleanup(func() { beforeStartup = nil }) + done := make(chan int, 1) + var out, errb bytes.Buffer + go func() { done <- run([]string{"-config", e.cfgPath}, &out, &errb, nil) }() + + deadline := time.Now().Add(5 * time.Second) + for !strings.Contains(e.logs(), "domains db reloaded") { + if time.Now().After(deadline) { + t.Fatalf("the early SIGHUP was not honoured; log:\n%s", e.logs()) + } + time.Sleep(20 * time.Millisecond) + } + syscall.Kill(os.Getpid(), syscall.SIGTERM) + select { + case code := <-done: + if code != 0 { + t.Errorf("exit code %d, stderr %q", code, errb.String()) + } + case <-time.After(15 * time.Second): + t.Fatal("run did not return after SIGTERM") + } +} + +func TestNewApp_EmptyCDBIsFatal(t *testing.T) { + e := newEnv(t, "") + w, err := cdb.Create(e.cdbPath) + if err != nil { + t.Fatal(err) + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + cfg, err := config.Load(e.cfgPath) + if err != nil { + t.Fatal(err) + } + if a, err := newApp(cfg); err == nil { + a.Shutdown(context.Background()) + t.Fatal("an empty domains db must be fatal at startup") + } +} diff --git a/config.toml.example b/config.toml.example index f735af6..95c812c 100644 --- a/config.toml.example +++ b/config.toml.example @@ -2,10 +2,10 @@ # Install as /usr/local/etc/migadu_counters/config.toml (or pass -config PATH). addr = ":5555" # API: POST /in, POST /out, GET /counters -admin_addr = ":9095" # /health and /metrics, no auth +admin_addr = "127.0.0.1:9095" # /health and /metrics, no auth: keep it private auth_token = "…" # REQUIRED (or MIGADU_COUNTERS_AUTH_TOKEN); same value in mailqueuer's config allowed_cidrs = [] # optional source-IP allowlist for the API, e.g. ["10.0.0.0/8"] -domains_cdb = "/usr/local/etc/rspamd/db/rspamd_domains_db" +domains_cdb = "/usr/local/etc/migadu_counters/db/rspamd_domains_db" # where contrib/pull-domains.sh installs it exempt = ["support@migadu.com", "contact@migadu.com"] # mail to/from these never counts count_junk = false # count /in requests flagged junk by mizu @@ -33,7 +33,8 @@ counting_format = "text" # Lua-like lines (default) or "json" [redis] db = 0 # must stay 0: migadu_web reads message counters from DB 0 -password = "…" # or MIGADU_COUNTERS_REDIS_PASSWORD +password = "…" # or MIGADU_COUNTERS_REDIS_PASSWORD; required with sentinels +# password_optional = true # development only: allow Sentinel without a password pool_size = 32 timeout_ms = 500 # Sentinel mode (production): writes -> master, reads -> replicas @@ -50,6 +51,13 @@ sentinel_password = "" org_url = "https://admin.migadu.com/api/limits?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT" address_url = "https://admin.migadu.com/api/limits?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT" domain_url = "" # web app's D branch is still a TODO +# Early warning: one call when a counter reaches warn_percent of its limit +# (0 = off). Empty URLs disable a level. These must not be the /api/limits +# URLs above: the web app treats every call there as a breach. +warn_percent = 80 +warn_org_url = "" # e.g. https://admin.migadu.com/api/limits/warning?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT&limit=$LIMIT&percent=$PERCENT +warn_address_url = "" +warn_domain_url = "" user = "…" password = "…" # or MIGADU_COUNTERS_NOTIFY_PASSWORD -timeout_seconds = 5 +timeout_seconds = 15 # per attempt; a request sent but unanswered is not retried diff --git a/internal/api/api_test.go b/internal/api/api_test.go index a7fcfe2..c314e54 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -22,7 +22,7 @@ func TestValidateOut(t *testing.T) { // Support/token logins (review #3): mizu forwards the raw SASL login // user@dom@SUFFIX; the base address is counted, the suffix logged. {"login suffix", OutRequest{AuthUser: "x@yagolivas.com@SUPPORTTOKEN", Recipient: "y@example.com"}, "", - ValidatedOut{AuthUser: Address{Value: "x@yagolivas.com"}, LoginSuffix: "SUPPORTTOKEN", Recipient: Address{Value: "y@example.com"}}}, + ValidatedOut{AuthUser: Address{Value: "x@yagolivas.com"}, TokenLogin: true, Recipient: Address{Value: "y@example.com"}}}, // mail_from never rejects (review #2): kept normalised when possible, // otherwise as a sanitised string for the log. {"mail_from empty", OutRequest{AuthUser: "x@example.com", Recipient: "y@example.com", MailFrom: ""}, "", @@ -253,3 +253,34 @@ func TestErrorResponseJSON(t *testing.T) { t.Errorf("got %s", b) } } + +func TestNormalizeDomain(t *testing.T) { + tests := []struct{ in, want string }{ + {"Example.COM", "example.com"}, + {" example.com. ", "example.com"}, + {"bücher.de", "xn--bcher-kva.de"}, + {"xn--bcher-kva.de", "xn--bcher-kva.de"}, + } + for _, tt := range tests { + got, err := NormalizeDomain(tt.in) + if err != nil || got != tt.want { + t.Errorf("NormalizeDomain(%q) = %q, %v; want %q", tt.in, got, err, tt.want) + } + } + for _, bad := range []string{"", ".", "a..b", ".a", "host_name.example.com", "[192.0.2.1]", "a b.com"} { + if got, err := NormalizeDomain(bad); err == nil { + t.Errorf("NormalizeDomain(%q) = %q, want an error", bad, got) + } + } +} + +// An unusable login is logged, but never with what follows a second "@". +func TestValidateOut_UnusableLoginDropsSuffix(t *testing.T) { + v, err := ValidateOut(OutRequest{AuthUser: "x@bad..domain@SECRET", Recipient: "y@example.com"}) + if err != nil { + t.Fatal(err) + } + if v.AuthUser.Usable() || v.AuthUser.Raw != "x@bad..domain" { + t.Errorf("AuthUser = %+v, want the raw value without the suffix", v.AuthUser) + } +} diff --git a/internal/api/validate.go b/internal/api/validate.go index 74279b8..5c94e59 100644 --- a/internal/api/validate.go +++ b/internal/api/validate.go @@ -40,7 +40,7 @@ func (a Address) Usable() bool { return a.Value != "" } type ValidatedOut struct { TraceID string AuthUser Address - LoginSuffix string // "SUPPORTTOKEN" of a user@dom@SUPPORTTOKEN login + TokenLogin bool // a user@dom@SUFFIX login; the suffix may be a credential and is dropped MailFrom string // normalised, or the sanitised input when not an address MailFromUsable bool Recipient Address @@ -60,7 +60,7 @@ type ValidatedIn struct { // mail_from only feeds the exempt check and never rejects. func ValidateOut(r OutRequest) (ValidatedOut, error) { v := ValidatedOut{TraceID: SanitizeTraceID(r.TraceID)} - v.AuthUser, v.LoginSuffix = validateAuthUser(r.AuthUser) + v.AuthUser, v.TokenLogin = validateAuthUser(r.AuthUser) v.Recipient = validateAddress(r.Recipient) v.MailFrom, v.MailFromUsable = lenient(r.MailFrom) if !v.AuthUser.Usable() && !v.Recipient.Usable() { @@ -93,15 +93,26 @@ func validateAddress(s string) Address { return Address{Value: n} } -func validateAuthUser(s string) (Address, string) { +func validateAuthUser(s string) (Address, bool) { if strings.TrimSpace(s) == "" { - return Address{Reason: "missing"}, "" + return Address{Reason: "missing"}, false } n, suffix, err := NormalizeAuthUser(s) if err != nil { - return Address{Raw: SanitizeForLog(s), Reason: "invalid e-mail address"}, "" + return Address{Raw: redactSuffix(s), Reason: "invalid e-mail address"}, false } - return Address{Value: n}, suffix + return Address{Value: n}, suffix != "" +} + +// redactSuffix prepares an unusable login for the log: anything after a second +// "@" may be a credential, so it is cut off first. +func redactSuffix(s string) string { + if first := strings.Index(s, "@"); first >= 0 { + if second := strings.Index(s[first+1:], "@"); second >= 0 { + s = s[:first+1+second] + } + } + return SanitizeForLog(s) } // lenient normalises when possible and otherwise keeps a sanitised copy for @@ -159,7 +170,21 @@ func NormalizeEmail(s string) (string, error) { return "", errors.New("local part contains whitespace or control characters") } } - domain = strings.TrimSuffix(domain, ".") + ascii, err := NormalizeDomain(domain) + if err != nil { + return "", err + } + out := local + "@" + ascii + if len(out) > MaxEmailLen { + return "", errors.New("longer than 254 characters") + } + return out, nil +} + +// NormalizeDomain trims and lowercases a domain, drops one trailing dot and +// converts a Unicode domain to punycode, the form the CDB keys it under. +func NormalizeDomain(s string) (string, error) { + domain := strings.TrimSuffix(strings.ToLower(strings.TrimSpace(s)), ".") if domain == "" || strings.HasPrefix(domain, ".") || strings.HasSuffix(domain, ".") || strings.Contains(domain, "..") { return "", errors.New("domain has an empty label") } @@ -173,11 +198,7 @@ func NormalizeEmail(s string) (string, error) { return "", errors.New("domain contains characters other than a-z, 0-9, '-' and '.'") } } - out := local + "@" + ascii - if len(out) > MaxEmailLen { - return "", errors.New("longer than 254 characters") - } - return out, nil + return ascii, nil } // SanitizeTraceID trims the id, strips control characters (it ends up in log diff --git a/internal/config/config.go b/internal/config/config.go index 935ce06..a6d0add 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,6 +29,10 @@ const ( EnvNotifyPassword = "MIGADU_COUNTERS_NOTIFY_PASSWORD" ) +// examplePlaceholder marks the secrets in config.toml.example that must be +// replaced before use. +const examplePlaceholder = "…" + // NotifyPlaceholders are the template variables every notify URL is expected // to carry: the web app's /api/limits action pattern-matches all six query // parameters and answers 500 when one is missing. @@ -84,10 +88,12 @@ func (l LoggingConfig) CountingEnabled() bool { return l.CountingOutput != "off" // Sentinel mode is used when Sentinels is non-empty; otherwise the static // WriteAddr/ReadAddrs are used (development, tests). type RedisConfig struct { - DB int `toml:"db"` - Password string `toml:"password"` - PoolSize int `toml:"pool_size"` - TimeoutMS int `toml:"timeout_ms"` + DB int `toml:"db"` + Password string `toml:"password"` + // PasswordOptional allows Sentinel mode without a password (development). + PasswordOptional bool `toml:"password_optional"` + PoolSize int `toml:"pool_size"` + TimeoutMS int `toml:"timeout_ms"` MasterName string `toml:"master_name"` Sentinels []string `toml:"sentinels"` @@ -106,9 +112,15 @@ func (r RedisConfig) Timeout() time.Duration { return time.Duration(r.TimeoutMS) // NotifyConfig holds the per-level web app notification URLs. An empty URL // disables notifications for that level. type NotifyConfig struct { - OrgURL string `toml:"org_url"` - AddressURL string `toml:"address_url"` - DomainURL string `toml:"domain_url"` + OrgURL string `toml:"org_url"` + AddressURL string `toml:"address_url"` + DomainURL string `toml:"domain_url"` + // Early warning: WarnPercent of a limit triggers a call to the warn_* URL + // of that level (0 = off). An empty URL disables that level. + WarnPercent int64 `toml:"warn_percent"` + WarnOrgURL string `toml:"warn_org_url"` + WarnAddressURL string `toml:"warn_address_url"` + WarnDomainURL string `toml:"warn_domain_url"` User string `toml:"user"` Password string `toml:"password"` TimeoutSeconds int `toml:"timeout_seconds"` @@ -117,11 +129,20 @@ type NotifyConfig struct { // Timeout is the per-attempt HTTP timeout for a notification. func (n NotifyConfig) Timeout() time.Duration { return time.Duration(n.TimeoutSeconds) * time.Second } +// EffectiveWarnPercent is WarnPercent, or 0 when no warning URL is configured, +// so an unused feature adds no markers, metrics or log lines. +func (n NotifyConfig) EffectiveWarnPercent() int64 { + if n.WarnOrgURL == "" && n.WarnAddressURL == "" && n.WarnDomainURL == "" { + return 0 + } + return n.WarnPercent +} + // Default returns the configuration with every default applied. func Default() Config { return Config{ Addr: ":5555", - AdminAddr: ":9095", + AdminAddr: "127.0.0.1:9095", Exempt: []string{"support@migadu.com", "contact@migadu.com"}, QueueSize: 10000, Workers: 8, @@ -132,7 +153,7 @@ func Default() Config { HandlerTimeout: 4, Logging: LoggingConfig{Output: "stderr", Format: "console", Level: "info", CountingFormat: "text"}, Redis: RedisConfig{PoolSize: 32, TimeoutMS: 500}, - Notify: NotifyConfig{TimeoutSeconds: 5}, + Notify: NotifyConfig{TimeoutSeconds: 15, WarnPercent: 80}, } } @@ -225,6 +246,13 @@ func (c *Config) Validate() error { if strings.TrimSpace(c.AuthToken) == "" { fail("auth_token is required; set it in the config or %s", EnvAuthToken) } + for _, p := range []struct{ name, value string }{ + {"auth_token", c.AuthToken}, {"redis.password", c.Redis.Password}, {"notify.password", c.Notify.Password}, + } { + if strings.TrimSpace(p.value) == examplePlaceholder { + fail("%s still holds the example placeholder %q", p.name, examplePlaceholder) + } + } if c.QueueSize <= 0 { fail("queue_size must be > 0") } @@ -282,6 +310,9 @@ func (c *Config) Validate() error { fail("redis.timeout_ms must be > 0") } if c.Redis.SentinelMode() { + if c.Redis.Password == "" && !c.Redis.PasswordOptional { + fail("redis.password is required in Sentinel mode; set it or %s, or set redis.password_optional = true", EnvRedisPassword) + } if c.Redis.MasterName == "" { fail("redis.master_name is required when redis.sentinels is set") } @@ -315,6 +346,18 @@ func (c *Config) Validate() error { if c.Notify.TimeoutSeconds <= 0 { fail("notify.timeout_seconds must be > 0") } + if c.Notify.WarnPercent < 0 || c.Notify.WarnPercent > 99 { + fail("notify.warn_percent must be between 0 and 99 (got %d)", c.Notify.WarnPercent) + } + for _, p := range []struct{ name, warn, breach string }{ + {"warn_org_url", c.Notify.WarnOrgURL, c.Notify.OrgURL}, + {"warn_address_url", c.Notify.WarnAddressURL, c.Notify.AddressURL}, + {"warn_domain_url", c.Notify.WarnDomainURL, c.Notify.DomainURL}, + } { + if p.warn != "" && p.breach != "" && samePath(p.warn, p.breach) { + fail("notify.%s must not use the same path as the breach URL: the web app treats every call there as a breach", p.name) + } + } return errors.Join(errs...) } @@ -331,6 +374,9 @@ func (c *Config) Warnings() []string { w = append(w, fmt.Sprintf("notify.%s: missing %s placeholder (the web app requires all six query parameters)", u.name, ph)) } } + if c.Notify.User != "" && strings.HasPrefix(strings.ToLower(u.url), "http://") { + w = append(w, fmt.Sprintf("notify.%s: basic auth credentials would travel over plain http", u.name)) + } } return w } @@ -340,9 +386,22 @@ func (c *Config) notifyURLs() []struct{ name, url string } { {"org_url", c.Notify.OrgURL}, {"address_url", c.Notify.AddressURL}, {"domain_url", c.Notify.DomainURL}, + {"warn_org_url", c.Notify.WarnOrgURL}, + {"warn_address_url", c.Notify.WarnAddressURL}, + {"warn_domain_url", c.Notify.WarnDomainURL}, } } +// samePath reports whether two URLs differ at most in their query string. +func samePath(a, b string) bool { + ua, errA := url.Parse(a) + ub, errB := url.Parse(b) + if errA != nil || errB != nil { + return false + } + return ua.Scheme == ub.Scheme && ua.Host == ub.Host && strings.TrimRight(ua.Path, "/") == strings.TrimRight(ub.Path, "/") +} + // ExemptSet returns the normalised exempt addresses. func (c *Config) ExemptSet() map[string]bool { return c.exempt } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 5adc335..a2577eb 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -43,7 +43,7 @@ func TestLoad_Defaults(t *testing.T) { want any }{ {"addr", cfg.Addr, ":5555"}, - {"admin_addr", cfg.AdminAddr, ":9095"}, + {"admin_addr", cfg.AdminAddr, "127.0.0.1:9095"}, {"auth_token", cfg.AuthToken, "t"}, {"queue_size", cfg.QueueSize, 10000}, {"workers", cfg.Workers, 8}, @@ -62,9 +62,10 @@ func TestLoad_Defaults(t *testing.T) { {"redis.pool_size", cfg.Redis.PoolSize, 32}, {"redis.timeout_ms", cfg.Redis.TimeoutMS, 500}, {"redis.sentinel_mode", cfg.Redis.SentinelMode(), false}, - {"notify.timeout_seconds", cfg.Notify.TimeoutSeconds, 5}, + {"notify.timeout_seconds", cfg.Notify.TimeoutSeconds, 15}, + {"notify.warn_percent", cfg.Notify.WarnPercent, int64(80)}, {"redis timeout", cfg.Redis.Timeout(), 500 * time.Millisecond}, - {"notify timeout", cfg.Notify.Timeout(), 5 * time.Second}, + {"notify timeout", cfg.Notify.Timeout(), 15 * time.Second}, {"handler timeout", cfg.HandlerTimeoutDuration(), 4 * time.Second}, } for _, c := range checks { @@ -86,9 +87,17 @@ func TestLoad_Defaults(t *testing.T) { } } -// The shipped example must always load, validate and produce no warnings. +// The shipped example must load and validate once its placeholders are filled +// in, produce no warnings, and be refused while they are still there. func TestLoad_Example(t *testing.T) { - cfg, err := Load("../../config.toml.example") + if _, err := Load("../../config.toml.example"); err == nil || !strings.Contains(err.Error(), "placeholder") { + t.Fatalf("the unedited example must be refused, got %v", err) + } + raw, err := os.ReadFile("../../config.toml.example") + if err != nil { + t.Fatal(err) + } + cfg, err := Load(writeConfig(t, strings.ReplaceAll(string(raw), "\"…\"", "\"filled-in\""))) if err != nil { t.Fatalf("Load(config.toml.example): %v", err) } @@ -164,7 +173,7 @@ func TestLoad_AllowedCIDRs(t *testing.T) { } func TestLoad_SentinelMode(t *testing.T) { - cfg := mustLoad(t, top+"[redis]\nmaster_name = \"mymaster\"\nsentinels = [\"10.0.0.1:26379\", \"10.0.0.2:26379\"]\n") + cfg := mustLoad(t, top+"[redis]\npassword = \"s\"\nmaster_name = \"mymaster\"\nsentinels = [\"10.0.0.1:26379\", \"10.0.0.2:26379\"]\n") if !cfg.Redis.SentinelMode() || len(cfg.Redis.ReadAddrs) != 0 || cfg.Redis.WriteAddr != "" { t.Errorf("sentinel mode config = %+v", cfg.Redis) } @@ -202,6 +211,14 @@ func TestValidate_Errors(t *testing.T) { {"addr", top + "addr = \"no-port\"\n" + redisStatic, "addr"}, {"admin_addr", top + "admin_addr = \"no-port\"\n" + redisStatic, "admin_addr"}, {"notify timeout", minimal + "[notify]\ntimeout_seconds = 0\n", "timeout_seconds"}, + {"sentinel needs password", top + "[redis]\nmaster_name = \"m\"\nsentinels = [\"10.0.0.1:26379\"]\n", "redis.password"}, + {"placeholder token", "domains_cdb = \"/tmp/x\"\nauth_token = \"…\"\n" + redisStatic, "auth_token"}, + {"placeholder redis password", top + redisStatic + "password = \"…\"\n", "redis.password"}, + {"placeholder notify password", minimal + "[notify]\npassword = \"…\"\n", "notify.password"}, + {"warn percent high", minimal + "[notify]\nwarn_percent = 100\n", "warn_percent"}, + {"warn percent negative", minimal + "[notify]\nwarn_percent = -1\n", "warn_percent"}, + {"warn url scheme", minimal + "[notify]\nwarn_org_url = \"ftp://x/w\"\n", "warn_org_url"}, + {"warn url same as breach", minimal + "[notify]\norg_url = \"https://x/api/limits?level=$LEVEL\"\nwarn_org_url = \"https://x/api/limits?level=$LEVEL&percent=$PERCENT\"\n", "warn_org_url"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -286,3 +303,33 @@ func TestDiscover(t *testing.T) { t.Errorf("Discover(\"\") = %q, %v; want %q", got, err, found) } } + +// An explicit opt-out allows a passwordless Sentinel setup (development). +func TestLoad_SentinelPasswordOptional(t *testing.T) { + cfg := mustLoad(t, top+"[redis]\npassword_optional = true\nmaster_name = \"m\"\nsentinels = [\"10.0.0.1:26379\"]\n") + if !cfg.Redis.SentinelMode() { + t.Errorf("config = %+v", cfg.Redis) + } +} + +func TestWarnings_PlainHTTPWithBasicAuth(t *testing.T) { + const q = "?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT" + cfg := mustLoad(t, minimal+"[notify]\nuser = \"u\"\norg_url = \"http://x/l"+q+"\"\n") + w := strings.Join(cfg.Warnings(), "\n") + if !strings.Contains(w, "org_url") || !strings.Contains(w, "plain http") { + t.Errorf("Warnings() = %q, want a plain-http warning for org_url", w) + } + cfg = mustLoad(t, minimal+"[notify]\nuser = \"u\"\norg_url = \"https://x/l"+q+"\"\n") + if w := cfg.Warnings(); len(w) != 0 { + t.Errorf("https should not warn: %v", w) + } +} + +func TestEffectiveWarnPercent(t *testing.T) { + if got := (NotifyConfig{WarnPercent: 80}).EffectiveWarnPercent(); got != 0 { + t.Errorf("no warning URLs: %d, want 0", got) + } + if got := (NotifyConfig{WarnPercent: 80, WarnOrgURL: "https://x/w"}).EffectiveWarnPercent(); got != 80 { + t.Errorf("with a warning URL: %d, want 80", got) + } +} diff --git a/internal/counting/counting.go b/internal/counting/counting.go index 6bbb461..ea8d638 100644 --- a/internal/counting/counting.go +++ b/internal/counting/counting.go @@ -56,6 +56,8 @@ type Planner struct { Exempt map[string]bool // lowercased addresses that never count CountJunk bool // count /in requests flagged junk Now func() time.Time // nil = time.Now; the UTC day of Now() is the counter day + // WarnPercent is the share of a limit that triggers an early warning; 0 = off. + WarnPercent int64 } // Job is one validated request handed to the queue: the fields the counting @@ -66,9 +68,9 @@ type Job struct { TraceID string Received time.Time // when the request arrived (queue wait metric) - AuthUser string // /out: normalised SMTP AUTH user, "" when not usable - LoginSuffix string // /out: "SUPPORTTOKEN" of a user@dom@SUPPORTTOKEN login - MailFrom string // normalised, or the sanitised input when not an address + AuthUser string // /out: normalised SMTP AUTH user, "" when not usable + TokenLogin bool // /out: a user@dom@SUFFIX login + MailFrom string // normalised, or the sanitised input when not an address // MailFromUsable tells whether MailFrom is a normalised address (and may // match the exempt list); an unusable value is never exempt. MailFromUsable bool @@ -107,20 +109,21 @@ type Increment struct { By string // address that caused it (sender for o, recipient for i); $ADDRESS in notifications Limit int64 // configured limit, 0 when unlimited Limited bool // false when no limit is configured for this level/flow + WarnAt int64 // counter value that triggers the early warning, 0 = none Domain *domains.Domain } // Plan is the result of applying the rules to one request. type Plan struct { - Route string - TraceID string - Date string // UTC day used in the keys - Skipped string // "" or one of the Skip* reasons - AuthUser string - LoginSuffix string - MailFrom string - Recipient string - Junk bool // /in only: the message was flagged junk by mizu + Route string + TraceID string + Date string // UTC day used in the keys + Skipped string // "" or one of the Skip* reasons + AuthUser string + TokenLogin bool + MailFrom string + Recipient string + Junk bool // /in only: the message was flagged junk by mizu Unusable []Unusable // sides that were not counted because they failed validation Lookups []Lookup @@ -166,8 +169,12 @@ func (p *Planner) Plan(job Job) (*Plan, error) { if job.MailFromUsable { mailFrom = normalize(mailFrom) } - plan := &Plan{Route: job.Route, TraceID: job.TraceID, Date: store.DayKey(p.now()), - AuthUser: authUser, LoginSuffix: job.LoginSuffix, MailFrom: mailFrom, Recipient: recipient, Junk: job.Junk} + counted := job.Received + if counted.IsZero() { + counted = p.now() + } + plan := &Plan{Route: job.Route, TraceID: job.TraceID, Date: store.DayKey(counted), + AuthUser: authUser, TokenLogin: job.TokenLogin, MailFrom: mailFrom, Recipient: recipient, Junk: job.Junk} if job.SenderUnusable.Set() { plan.Unusable = append(plan.Unusable, job.SenderUnusable) } @@ -234,11 +241,33 @@ func (p *Planner) add(plan *Plan, role, address, flow string) error { Level: l.level, ID: l.id, Flow: flow, Key: store.Key(l.level, l.id, plan.Date, flow), By: address, Limit: limit, Limited: limited, Domain: d, + WarnAt: warnAt(limit, limited, p.WarnPercent), }) } return nil } +func warnAt(limit int64, limited bool, percent int64) int64 { + if !limited { + return 0 + } + return WarnThreshold(limit, percent) +} + +// WarnThreshold returns the counter value at which percent of limit is used, +// rounded up, or 0 when no warning applies: warnings are off, there is no +// positive limit, or the threshold is not below the limit itself. +func WarnThreshold(limit, percent int64) int64 { + if limit <= 0 || percent <= 0 { + return 0 + } + t := limit/100*percent + (limit%100*percent+99)/100 + if t >= limit { + return 0 + } + return t +} + // Crossed reports whether an increment of amount that produced newValue is // the one that reached limit: old < limit <= new. Because INCRBY is atomic // exactly one increment can cross, so this is the once-per-crossing trigger @@ -251,8 +280,12 @@ func Crossed(newValue, amount, limit int64) bool { type Result struct { Increment Count int64 // value after the increment (from EXEC, or re-read after a lost reply) - Reached bool // this increment crossed the limit + Reached bool // this increment crossed the limit and is the one to notify + Warned bool // this increment crossed the warning threshold and is the one to notify Err error // this key's own INCR failed (e.g. it holds a non-number); Count is 0 + // MarkerErr is set when the notification marker could not be read or + // written. After a lost reply the notification is then withheld. + MarkerErr error } // Outcome is an executed Plan. @@ -269,6 +302,9 @@ type Counter struct { Planner *Planner Store *store.Store ReReadTimeout time.Duration // budget for the master re-read after a lost reply (default 2s) + // AbortRetryDelay is the pause before the single retry of a transaction + // Redis rejected, e.g. during a failover (default 200ms). + AbortRetryDelay time.Duration } // CountOut plans and executes an /out request. @@ -293,12 +329,17 @@ func (c *Counter) CountIn(ctx context.Context, traceID, mailFrom, recipient stri // not touch Redis. // // Failure handling (review #4): a key whose own INCR failed is reported in -// its Result.Err while the other keys are counted normally. When the whole -// reply was lost (timeout, reset) after the write may have been applied, the -// keys are re-read once from the master and the outcome is flagged Unknown; -// crossing detection then uses count >= limit so a crossing that happened in -// the lost transaction is still notified. A definite failure (connection -// refused, cancelled) is returned as an error: nothing was counted. +// its Result.Err while the other keys are counted normally. A transaction +// Redis rejected as a whole applied nothing and is retried once. When the +// whole reply was lost (timeout, reset) after the write may have been applied, +// the keys are re-read once from the master and the outcome is flagged +// Unknown. A definite failure (connection refused, cancelled) is returned as +// an error: nothing was counted. +// +// Notifications go out once per counter: a confirmed crossing is exact, so it +// always fires and writes a marker. After a lost reply the crossing cannot be +// pinned to this increment, so it fires only for the caller that wins the +// marker, and a counter that already notified stays silent. func (c *Counter) Execute(ctx context.Context, plan *Plan) (*Outcome, error) { out := &Outcome{Plan: plan} if len(plan.Increments) == 0 { @@ -309,13 +350,15 @@ func (c *Counter) Execute(ctx context.Context, plan *Plan) (*Outcome, error) { keys[i] = inc.Key } out.Results = make([]Result, len(plan.Increments)) - results, err := c.Store.Incr(ctx, keys) + results, err := c.incr(ctx, keys) if err == nil { for i, inc := range plan.Increments { r := Result{Increment: inc, Err: results[i].Err} if r.Err == nil { r.Count = results[i].Count r.Reached = inc.Limited && Crossed(r.Count, 1, inc.Limit) + r.Warned = Crossed(r.Count, 1, inc.WarnAt) + c.mark(ctx, plan.Date, &r) } out.Results[i] = r } @@ -332,11 +375,65 @@ func (c *Counter) Execute(ctx context.Context, plan *Plan) (*Outcome, error) { } out.Unknown = err for i, inc := range plan.Increments { - out.Results[i] = Result{Increment: inc, Count: counts[i], Reached: inc.Limited && counts[i] >= inc.Limit} + r := Result{Increment: inc, Count: counts[i]} + switch { + case inc.Limited && inc.Limit > 0 && r.Count >= inc.Limit: + r.Reached = c.claim(rctx, plan.Date, &r, store.MarkerLimit) + case inc.WarnAt > 0 && r.Count >= inc.WarnAt: + r.Warned = c.claim(rctx, plan.Date, &r, store.MarkerWarn) + } + out.Results[i] = r } return out, nil } +// incr runs the transaction and retries it once when Redis rejected it. +func (c *Counter) incr(ctx context.Context, keys []string) ([]store.IncrResult, error) { + results, err := c.Store.Incr(ctx, keys) + if !errors.Is(err, store.ErrNotApplied) { + return results, err + } + select { + case <-time.After(c.abortRetryDelay()): + case <-ctx.Done(): + return nil, err + } + return c.Store.Incr(ctx, keys) +} + +// mark records a confirmed crossing. A failure is reported but the +// notification still goes out, because this crossing is certain. +func (c *Counter) mark(ctx context.Context, date string, r *Result) { + if !r.Reached && !r.Warned { + return + } + kind := store.MarkerLimit + if r.Warned { + kind = store.MarkerWarn + } + mctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), c.reReadTimeout()) + defer cancel() + r.MarkerErr = c.Store.MarkNotified(mctx, store.MarkerKey(r.Level, r.ID, date, r.Flow, kind)) +} + +// claim takes the marker after a lost reply. If that fails nothing is sent: +// a missed notification is preferred to a repeated one. +func (c *Counter) claim(ctx context.Context, date string, r *Result, kind string) bool { + won, err := c.Store.ClaimNotified(ctx, store.MarkerKey(r.Level, r.ID, date, r.Flow, kind)) + if err != nil { + r.MarkerErr = err + return false + } + return won +} + +func (c *Counter) abortRetryDelay() time.Duration { + if c.AbortRetryDelay > 0 { + return c.AbortRetryDelay + } + return 200 * time.Millisecond +} + func (c *Counter) reReadTimeout() time.Duration { if c.ReReadTimeout > 0 { return c.ReReadTimeout diff --git a/internal/counting/counting_test.go b/internal/counting/counting_test.go index 21ffb85..ebbb121 100644 --- a/internal/counting/counting_test.go +++ b/internal/counting/counting_test.go @@ -450,7 +450,7 @@ func TestPlan_OutSenderUnusableCountsRecipient(t *testing.T) { } func TestPlan_OutRecipientUnusableCountsSender(t *testing.T) { - job := Job{Route: RouteOut, TraceID: "u2", AuthUser: "x@example.com", LoginSuffix: "SUPPORTTOKEN", MailFrom: "x@example.com", + job := Job{Route: RouteOut, TraceID: "u2", AuthUser: "x@example.com", TokenLogin: true, 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 { @@ -461,8 +461,8 @@ func TestPlan_OutRecipientUnusableCountsSender(t *testing.T) { {"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 !plan.TokenLogin || plan.AuthUser != "x@example.com" { + t.Errorf("token login not carried: %+v", plan) } if len(plan.Unusable) != 1 || plan.Unusable[0].Role != "recipient" { t.Errorf("unusable = %+v", plan.Unusable) diff --git a/internal/counting/notifications_test.go b/internal/counting/notifications_test.go new file mode 100644 index 0000000..2e54c7f --- /dev/null +++ b/internal/counting/notifications_test.go @@ -0,0 +1,255 @@ +package counting + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + + "migadu_counters/internal/store" +) + +func TestWarnThreshold(t *testing.T) { + tests := []struct { + limit, percent, want int64 + }{ + {1, 80, 0}, // ceil(0.8) = 1 is the limit itself: no warning + {2, 80, 0}, // ceil(1.6) = 2 is the limit itself: no warning + {5, 80, 4}, + {100, 80, 80}, + {101, 80, 81}, + {100, 0, 0}, // disabled + {0, 80, 0}, // no limit, no warning + {-5, 80, 0}, + {1 << 62, 50, 1 << 61}, // no overflow + } + for _, tt := range tests { + if got := WarnThreshold(tt.limit, tt.percent); got != tt.want { + t.Errorf("WarnThreshold(%d, %d) = %d, want %d", tt.limit, tt.percent, got, tt.want) + } + } +} + +func warnCounter(t *testing.T) (*Counter, *miniredis.Miniredis) { + t.Helper() + c, m := newCounter(t) + c.Planner.WarnPercent = 80 + return c, m +} + +func TestPlan_SetsWarnAt(t *testing.T) { + p := newPlanner(false) + p.WarnPercent = 80 + plan, err := p.PlanOut("w", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + got := []int64{plan.Increments[0].WarnAt, plan.Increments[1].WarnAt, plan.Increments[2].WarnAt} + if got[0] != 4 || got[1] != 1760 || got[2] != 1760 { + t.Errorf("WarnAt = %v, want [4 1760 1760]", got) + } + p.WarnPercent = 0 + plan, _ = p.PlanOut("w", "noreply@example.com", "", "ext@gmail.com") + if plan.Increments[0].WarnAt != 0 { + t.Errorf("WarnAt with warnings off = %d, want 0", plan.Increments[0].WarnAt) + } +} + +// noreply@example.com has A/o=5, so 80% is the 4th message. +func TestCounter_WarnsOnceThenBreachesOnce(t *testing.T) { + c, _ := warnCounter(t) + ctx := context.Background() + for i := 1; i <= 7; i++ { + out, err := c.CountOut(ctx, "tr", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + a := out.Results[0] + if a.Warned != (i == 4) { + t.Errorf("message %d: warned = %v", i, a.Warned) + } + if a.Reached != (i == 5) { + t.Errorf("message %d: reached = %v", i, a.Reached) + } + } +} + +// Markers must stay out of the cnt/ namespace: the web app scans cnt/*
*. +func TestCounter_ConfirmedCrossingWritesMarker(t *testing.T) { + c, m := warnCounter(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) + } + } + for _, kind := range []string{store.MarkerLimit, store.MarkerWarn} { + key := store.MarkerKey("A", "noreply@example.com", day, "o", kind) + if !strings.HasPrefix(key, "ntf/") { + t.Errorf("marker key %q must not share the cnt/ namespace", key) + } + if !m.Exists(key) { + t.Errorf("marker %s not written; keys = %v", key, m.Keys()) + } + if ttl := m.TTL(key); ttl != store.MarkerTTL { + t.Errorf("marker %s TTL = %v, want %v", key, ttl, store.MarkerTTL) + } + } +} + +// lostReplyCounter counts through a proxy that delays the next EXEC reply past +// the client's read timeout, so Redis applies the transaction but the reply is lost. +func lostReplyCounter(t *testing.T, m *miniredis.Miniredis) *Counter { + t.Helper() + 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() }) + p := newPlanner(false) + p.WarnPercent = 80 + return &Counter{Planner: p, Store: st} +} + +// A counter that already crossed and notified must stay silent on every later +// lost reply. +func TestCounter_LostReplyAfterBreachDoesNotNotifyAgain(t *testing.T) { + c, m := warnCounter(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) + } + } + out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + if out.Unknown == nil { + t.Fatal("reply should have been lost") + } + if a := out.Results[0]; a.Count != 6 || a.Reached || a.Warned { + t.Errorf("A = %+v, want count 6, no reached, no warned", a) + } +} + +// When the crossing increment itself loses its reply it notifies exactly once, +// however many later replies are lost too. +func TestCounter_LostReplyOnCrossingNotifiesOnce(t *testing.T) { + m := miniredis.RunT(t) + key := store.Key("A", "noreply@example.com", day, "o") + m.Set(key, "4") // A/o=5: the next increment crosses + ctx := context.Background() + reached := 0 + for i := 0; i < 2; i++ { + out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + if out.Unknown == nil { + t.Fatalf("round %d: reply should have been lost", i) + } + if out.Results[0].Reached { + reached++ + } + } + if reached != 1 { + t.Errorf("reached %d times over two lost replies, want 1", reached) + } +} + +func TestCounter_LostReplyWarnsOnce(t *testing.T) { + m := miniredis.RunT(t) + key := store.Key("A", "noreply@example.com", day, "o") + m.Set(key, "3") // 80% of A/o=5 is 4: the next increment warns + ctx := context.Background() + out, err := lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + if a := out.Results[0]; a.Count != 4 || !a.Warned || a.Reached { + t.Errorf("A = %+v, want count 4 warned", a) + } + m.Set(key, "3") // second lost reply lands on the same value + out, err = lostReplyCounter(t, m).CountOut(ctx, "lost", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatal(err) + } + if a := out.Results[0]; a.Warned { + t.Errorf("A = %+v, must not warn twice", a) + } +} + +// Crossed says a limit of 0 never fires; the lost-reply path must agree. +func TestCounter_LostReplyZeroLimitNeverFires(t *testing.T) { + m := miniredis.RunT(t) + c := lostReplyCounter(t, m) + plan := &Plan{Route: RouteOut, Date: day, Increments: []Increment{ + {Level: "A", ID: "x@example.com", Flow: "o", Key: store.Key("A", "x@example.com", day, "o"), Limited: true, Limit: 0}, + }} + out, err := c.Execute(context.Background(), plan) + if err != nil { + t.Fatal(err) + } + if out.Unknown == nil { + t.Fatal("reply should have been lost") + } + if out.Results[0].Reached { + t.Errorf("limit 0 fired after a lost reply: %+v", out.Results[0]) + } +} + +// A transaction Redis aborted (READONLY during a failover) applied nothing, so +// it is safe to run it once more. +func TestCounter_AbortedTransactionIsRetried(t *testing.T) { + c, m := newCounter(t) + c.AbortRetryDelay = 150 * time.Millisecond + m.SetError("READONLY You can't write against a read only replica.") + go func() { + time.Sleep(30 * time.Millisecond) + m.SetError("") + }() + out, err := c.CountOut(context.Background(), "fo", "noreply@example.com", "", "ext@gmail.com") + if err != nil { + t.Fatalf("retry should have counted: %v", err) + } + if out.Unknown != nil || out.Results[0].Err != nil || out.Results[0].Count != 1 { + t.Errorf("outcome = %+v unknown=%v", out.Results[0], out.Unknown) + } + if v, _ := m.Get(store.Key("A", "noreply@example.com", day, "o")); v != "1" { + t.Errorf("redis A/o = %q, want exactly 1", v) + } +} + +func TestCounter_AbortedTwiceIsAWholeFailure(t *testing.T) { + c, m := newCounter(t) + c.AbortRetryDelay = time.Millisecond + m.SetError("READONLY You can't write against a read only replica.") + out, err := c.CountOut(context.Background(), "fo", "noreply@example.com", "", "ext@gmail.com") + if err == nil || !errors.Is(err, store.ErrNotApplied) { + t.Fatalf("err = %v, outcome = %+v; want ErrNotApplied", err, out) + } + m.SetError("") + if keys := m.Keys(); len(keys) != 0 { + t.Errorf("nothing should be written, keys = %v", keys) + } +} + +// A job that waited in the queue across midnight UTC counts on the day it arrived. +func TestPlan_DayIsWhenTheRequestArrived(t *testing.T) { + p := newPlanner(false) + received := fixedNow.Add(-24 * time.Hour) + plan, err := p.Plan(Job{Route: RouteIn, Recipient: "y@example.com", Received: received}) + if err != nil { + t.Fatal(err) + } + if want := store.DayKey(received); plan.Date != want { + t.Errorf("plan date = %s, want %s", plan.Date, want) + } + plan, _ = p.Plan(Job{Route: RouteIn, Recipient: "y@example.com"}) + if plan.Date != day { + t.Errorf("plan date without Received = %s, want %s", plan.Date, day) + } +} diff --git a/internal/domains/domains.go b/internal/domains/domains.go index 24b4d53..a742cff 100644 --- a/internal/domains/domains.go +++ b/internal/domains/domains.go @@ -15,8 +15,10 @@ package domains import ( + "encoding/binary" "errors" "fmt" + "os" "sort" "strconv" "strings" @@ -83,6 +85,9 @@ func ParseLimits(s string) Limits { if k == "" { continue } + if strings.Contains(k, "@") { // a local part; lookups use the lowercased mailbox + k = strings.ToLower(k) + } n, err := strconv.ParseInt(strings.TrimSpace(v), 10, 64) if err != nil { continue @@ -173,11 +178,23 @@ func Open(path string) (*DB, error) { return OpenWithCache(path, DefaultMaxCache // OpenWithCache opens the CDB at path and counts its records (a few // milliseconds for ~100k records; the count feeds the cdb_domains gauge and // the reload log). maxCached bounds the parsed-record cache; 0 disables it. -func OpenWithCache(path string, maxCached int) (*DB, error) { +// +// A corrupt file is an error, never a panic: reloads run on a goroutine with +// no recover, and the cdb library panics on record lengths that overflow. +func OpenWithCache(path string, maxCached int) (db *DB, err error) { + if err := checkTables(path); err != nil { + return nil, err + } c, err := cdb.Open(path) if err != nil { return nil, fmt.Errorf("open domains db %s: %w", path, err) } + defer func() { + if p := recover(); p != nil { + c.Close() + db, err = nil, fmt.Errorf("read domains db %s: corrupt file: %v", path, p) + } + }() n := 0 it := c.Iter() for it.Next() { @@ -193,6 +210,33 @@ func OpenWithCache(path string, maxCached int) (*DB, error) { return &DB{path: path, db: c, count: n, maxCached: maxCached}, nil } +// checkTables verifies that every hash table named in the CDB header lies +// inside the file. Walking the records never reads these tables, so a file cut +// off inside them would load and then fail lookups. +func checkTables(path string) error { + f, err := os.Open(path) + if err != nil { + return fmt.Errorf("open domains db %s: %w", path, err) + } + defer f.Close() + fi, err := f.Stat() + if err != nil { + return fmt.Errorf("open domains db %s: %w", path, err) + } + header := make([]byte, 256*8) + if _, err := f.ReadAt(header, 0); err != nil { + return fmt.Errorf("read domains db %s: header: %w", path, err) + } + for i := 0; i < 256; i++ { + offset := int64(binary.LittleEndian.Uint32(header[i*8:])) + slots := int64(binary.LittleEndian.Uint32(header[i*8+4:])) + if offset+8*slots > fi.Size() { + return fmt.Errorf("read domains db %s: hash table %d ends past the end of the file", path, i) + } + } + return nil +} + // Lookup returns the record for domain (trimmed and lowercased first), or // ErrNotFound when the domain is not hosted here. The returned *Domain may // be shared with other callers and must be treated as read-only. diff --git a/internal/domains/domains_test.go b/internal/domains/domains_test.go index d1f9742..724e08d 100644 --- a/internal/domains/domains_test.go +++ b/internal/domains/domains_test.go @@ -1,7 +1,9 @@ package domains import ( + "encoding/binary" "errors" + "os" "path/filepath" "strings" "testing" @@ -387,3 +389,59 @@ func TestMailbox(t *testing.T) { } } } + +func TestParseLimits_OverrideKeysLowercased(t *testing.T) { + l := ParseLimits("A/o=20,Info@/o=5,O/o=9") + if n, ok := l.Limit(AddressLevel, FlowOut, "info"); !ok || n != 5 { + t.Errorf("override for info = %d, %v; want 5", n, ok) + } + if n, ok := l.Limit(OrganizationLevel, FlowOut, ""); !ok || n != 9 { + t.Errorf("O/o = %d, %v; level keys must keep their case", n, ok) + } +} + +// A file cut off inside its hash tables passes a walk over the records, but +// lookups would then fail: it must be refused at open. +func TestOpen_RejectsTruncatedHashTables(t *testing.T) { + path := filepath.Join(t.TempDir(), "t.cdb") + w, err := cdb.Create(path) + if err != nil { + t.Fatal(err) + } + for _, k := range []string{"a.example", "b.example", "c.example", "d.example"} { + if err := w.Put([]byte(k), []byte("1|A/o=1")); err != nil { + t.Fatal(err) + } + } + if err := w.Close(); err != nil { + t.Fatal(err) + } + fi, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + if err := os.Truncate(path, fi.Size()-8); err != nil { + t.Fatal(err) + } + if db, err := Open(path); err == nil { + db.Close() + t.Fatal("truncated CDB was accepted") + } +} + +// Lengths that overflow uint32 make the cdb library panic; a bad file must be +// an error, because reloads run on a goroutine with no recover. +func TestOpen_CorruptLengthsAreAnError(t *testing.T) { + buf := make([]byte, 2048+32) + binary.LittleEndian.PutUint32(buf[0:], 2048+32) // table 0 starts after one bogus record + binary.LittleEndian.PutUint32(buf[2048:], 0xFFFFFFFF) + binary.LittleEndian.PutUint32(buf[2052:], 2) + path := filepath.Join(t.TempDir(), "bad.cdb") + if err := os.WriteFile(path, buf, 0o600); err != nil { + t.Fatal(err) + } + if db, err := Open(path); err == nil { + db.Close() + t.Fatal("corrupt CDB was accepted") + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 011a427..822b7b8 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -7,17 +7,18 @@ import ( // Metrics holds every collector; fields are exported for tests. type Metrics struct { - Requests *prometheus.CounterVec // route, status - Duration *prometheus.HistogramVec // route - Increments *prometheus.CounterVec // level, flow - LimitReached *prometheus.CounterVec // level, flow - Notifications *prometheus.CounterVec // level, result (sent|failed|disabled) - RedisErrors *prometheus.CounterVec // op - CDBDomains prometheus.Gauge - CDBReloads *prometheus.CounterVec // result (success|failure) - Dropped *prometheus.CounterVec // route: requests refused because the counting queue was full - QueueDepth prometheus.Gauge - QueueWait prometheus.Histogram + Requests *prometheus.CounterVec // route, status + Duration *prometheus.HistogramVec // route + Increments *prometheus.CounterVec // level, flow + LimitReached *prometheus.CounterVec // level, flow + WarningReached *prometheus.CounterVec // level, flow + Notifications *prometheus.CounterVec // kind (limit|warning), level, result (sent|failed|disabled) + RedisErrors *prometheus.CounterVec // op + CDBDomains prometheus.Gauge + CDBReloads *prometheus.CounterVec // result (success|failure) + Dropped *prometheus.CounterVec // route: requests refused because the counting queue was full + QueueDepth prometheus.Gauge + QueueWait prometheus.Histogram } // New creates the collectors and registers them with reg. @@ -40,13 +41,17 @@ func New(reg prometheus.Registerer) *Metrics { Name: "migadu_counters_limit_reached_total", Help: "Increments that crossed a configured limit, by level and flow.", }, []string{"level", "flow"}), + WarningReached: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "migadu_counters_warning_reached_total", + Help: "Increments that crossed the early-warning threshold, by level and flow.", + }, []string{"level", "flow"}), Notifications: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "migadu_counters_notifications_total", - Help: "Web app notifications by level and result (sent, failed, disabled).", - }, []string{"level", "result"}), + Help: "Web app notifications by kind (limit, warning), level and result (sent, failed, disabled).", + }, []string{"kind", "level", "result"}), RedisErrors: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "migadu_counters_redis_errors_total", - Help: "Failed Redis operations by op (incr, mget, ping).", + Help: "Failed Redis operations by op (incr, incr_unknown, mget, ping, marker).", }, []string{"op"}), CDBDomains: prometheus.NewGauge(prometheus.GaugeOpts{ Name: "migadu_counters_cdb_domains", @@ -70,7 +75,7 @@ func New(reg prometheus.Registerer) *Metrics { Buckets: []float64{.0001, .0005, .001, .0025, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5}, }), } - reg.MustRegister(m.Requests, m.Duration, m.Increments, m.LimitReached, m.Notifications, m.RedisErrors, m.CDBDomains, m.CDBReloads, + reg.MustRegister(m.Requests, m.Duration, m.Increments, m.LimitReached, m.WarningReached, m.Notifications, m.RedisErrors, m.CDBDomains, m.CDBReloads, m.Dropped, m.QueueDepth, m.QueueWait) return m } @@ -101,8 +106,13 @@ func (m *Metrics) RecordLimitReached(level, flow string) { } // RecordNotification counts one notification outcome. -func (m *Metrics) RecordNotification(level, result string) { - m.Notifications.WithLabelValues(level, result).Inc() +func (m *Metrics) RecordNotification(kind, level, result string) { + m.Notifications.WithLabelValues(kind, level, result).Inc() +} + +// RecordWarningReached counts one early-warning crossing. +func (m *Metrics) RecordWarningReached(level, flow string) { + m.WarningReached.WithLabelValues(level, flow).Inc() } // RecordRedisError counts one failed Redis operation. diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index 4432e35..e311eb0 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -18,8 +18,8 @@ func TestMetrics(t *testing.T) { m.RecordIncrement("A", "o") m.RecordIncrement("O", "i") m.RecordLimitReached("O", "o") - m.RecordNotification("O", "sent") - m.RecordNotification("A", "failed") + m.RecordNotification("limit", "O", "sent") + m.RecordNotification("warning", "A", "failed") m.RecordRedisError("incr") m.SetCDBDomains(93549) m.RecordReload(true) @@ -40,8 +40,8 @@ func TestMetrics(t *testing.T) { {"increments A o", m.Increments.WithLabelValues("A", "o"), 2}, {"increments O i", m.Increments.WithLabelValues("O", "i"), 1}, {"limit reached O o", m.LimitReached.WithLabelValues("O", "o"), 1}, - {"notifications O sent", m.Notifications.WithLabelValues("O", "sent"), 1}, - {"notifications A failed", m.Notifications.WithLabelValues("A", "failed"), 1}, + {"notifications O sent", m.Notifications.WithLabelValues("limit", "O", "sent"), 1}, + {"notifications A failed", m.Notifications.WithLabelValues("warning", "A", "failed"), 1}, {"redis errors incr", m.RedisErrors.WithLabelValues("incr"), 1}, {"cdb domains", m.CDBDomains, 93549}, {"reload success", m.CDBReloads.WithLabelValues("success"), 1}, diff --git a/internal/notify/notify.go b/internal/notify/notify.go index 209ab81..cbcf2f8 100644 --- a/internal/notify/notify.go +++ b/internal/notify/notify.go @@ -12,55 +12,85 @@ import ( "fmt" "io" "net/http" + "net/http/httptrace" "net/url" "strconv" "strings" "sync" + "sync/atomic" "time" ) +// Event kinds: a limit breach, or the early warning before it. +const ( + KindLimit = "limit" + KindWarning = "warning" +) + // Config holds the notification settings. type Config struct { OrgURL string // $LEVEL=O AddressURL string // $LEVEL=A DomainURL string // $LEVEL=D - User string // basic auth; empty = no Authorization header - Password string - Timeout time.Duration // per attempt (default 5s) + // Early-warning URLs, per level. They must not be the breach URLs: the + // web app treats every call to those as a breach. + WarnOrgURL string + WarnAddressURL string + WarnDomainURL string + + User string // basic auth; empty = no Authorization header + Password string + + Timeout time.Duration // per attempt (default 15s) Attempts int // total attempts (default 5) Backoff time.Duration // first retry delay, doubled each retry (default 1s) MaxConcurrent int // notifications in flight (default 8) } -// URLFor returns the template for a level, "" when that level is disabled. -func (c Config) URLFor(level string) string { +// URLFor returns the template for a kind and level, "" when disabled. +func (c Config) URLFor(kind, level string) string { + warn := kind == KindWarning switch level { case "O": + if warn { + return c.WarnOrgURL + } return c.OrgURL case "A": + if warn { + return c.WarnAddressURL + } return c.AddressURL case "D": + if warn { + return c.WarnDomainURL + } return c.DomainURL } return "" } -// URLName returns the config key that serves a level, for log lines. -func URLName(level string) string { +// URLName returns the config key that serves a kind and level, for log lines. +func URLName(kind, level string) string { + prefix := "" + if kind == KindWarning { + prefix = "warn_" + } switch level { case "O": - return "org_url" + return prefix + "org_url" case "A": - return "address_url" + return prefix + "address_url" case "D": - return "domain_url" + return prefix + "domain_url" } return "" } -// Event is one "limit reached" occurrence. +// Event is one "limit reached" or early-warning occurrence. type Event struct { + Kind string // KindLimit (default) or KindWarning TraceID string Route string // in / out, for log correlation Level string // A, D or O @@ -69,6 +99,15 @@ type Event struct { Address string // the address that triggered it ($ADDRESS / by=) Count int64 // counter value after the increment ($COUNT) Limit int64 // configured limit ($LIMIT) + Percent int64 // warning threshold in percent ($PERCENT) +} + +// KindName returns the event's kind, defaulting to a limit breach. +func (e Event) KindName() string { + if e.Kind == "" { + return KindLimit + } + return e.Kind } // BuildURL fills the template placeholders with URL-escaped event values. @@ -81,6 +120,7 @@ func BuildURL(tmpl string, ev Event) string { "$ID", esc(ev.ID), "$FLOW", esc(ev.Flow), "$PERIOD", "d", + "$PERCENT", strconv.FormatInt(ev.Percent, 10), "$ADDRESS", esc(ev.Address), "$COUNT", strconv.FormatInt(ev.Count, 10), ).Replace(tmpl) @@ -102,14 +142,15 @@ type Notifier struct { wg sync.WaitGroup quit chan struct{} - mu sync.Mutex - closed bool + mu sync.Mutex + closed bool + quitOnce sync.Once } // New applies defaults and returns a ready Notifier. func New(cfg Config, hooks Hooks) *Notifier { if cfg.Timeout <= 0 { - cfg.Timeout = 5 * time.Second + cfg.Timeout = 15 * time.Second } if cfg.Attempts <= 0 { cfg.Attempts = 5 @@ -132,7 +173,7 @@ func New(cfg Config, hooks Hooks) *Notifier { // Notify dispatches ev in the background. It returns false, without doing // anything, when no URL is configured for the level or after Shutdown. func (n *Notifier) Notify(ev Event) bool { - tmpl := n.cfg.URLFor(ev.Level) + tmpl := n.cfg.URLFor(ev.KindName(), ev.Level) if tmpl == "" { return false } @@ -147,57 +188,87 @@ func (n *Notifier) Notify(ev Event) bool { u := BuildURL(tmpl, ev) go func() { defer n.wg.Done() - n.sem <- struct{}{} - defer func() { <-n.sem }() n.send(ev, u) }() return true } +// send delivers ev, holding a concurrency slot only while a request is in +// flight, never during a backoff. func (n *Notifier) send(ev Event, u string) { var lastErr error - for attempt := 1; attempt <= n.cfg.Attempts; attempt++ { + attempts := 0 + for attempts < n.cfg.Attempts { + if attempts > 0 && !n.wait(n.cfg.Backoff<<(attempts-1)) { + lastErr = fmt.Errorf("shutting down after attempt %d: %w", attempts, lastErr) + break + } + select { + case n.sem <- struct{}{}: + case <-n.quit: + n.failed(ev, u, attempts, errors.New("shutting down before the request was sent")) + return + } err, retry := n.attempt(u) + <-n.sem + attempts++ if err == nil { if n.hooks.Sent != nil { - n.hooks.Sent(ev, u, attempt) + n.hooks.Sent(ev, u, attempts) } return } lastErr = err if !retry { - if n.hooks.Failed != nil { - n.hooks.Failed(ev, u, attempt, err) - } - return - } - if attempt < n.cfg.Attempts { - select { - case <-time.After(n.cfg.Backoff << (attempt - 1)): - case <-n.quit: - lastErr = fmt.Errorf("shutting down after attempt %d: %w", attempt, err) - attempt = n.cfg.Attempts // stop retrying - } + break } } + n.failed(ev, u, attempts, lastErr) +} + +func (n *Notifier) failed(ev Event, u string, attempts int, err error) { if n.hooks.Failed != nil { - n.hooks.Failed(ev, u, n.cfg.Attempts, lastErr) + n.hooks.Failed(ev, u, attempts, err) } } -// attempt performs one GET. retry is false for 4xx: wrong credentials or a -// wrong URL will not fix themselves. +// wait sleeps for d and reports false when the shutdown deadline passed first. +func (n *Notifier) wait(d time.Duration) bool { + select { + case <-time.After(d): + return true + case <-n.quit: + return false + } +} + +// attempt performs one GET. retry is true only when a repeat cannot duplicate +// the notification or the web app said it failed: the request never left, or +// the answer was 429 or 5xx. A request that was sent but got no answer may +// have been acted on, and 4xx (wrong credentials, wrong URL) will not fix +// itself. func (n *Notifier) attempt(u string) (err error, retry bool) { req, err := http.NewRequest(http.MethodGet, u, nil) if err != nil { return err, false } + var wrote atomic.Bool + req = req.WithContext(httptrace.WithClientTrace(req.Context(), &httptrace.ClientTrace{ + WroteRequest: func(info httptrace.WroteRequestInfo) { + if info.Err == nil { + wrote.Store(true) + } + }, + })) if n.cfg.User != "" { req.SetBasicAuth(n.cfg.User, n.cfg.Password) } req.Header.Set("User-Agent", "migadu_counters") resp, err := n.client.Do(req) if err != nil { + if wrote.Load() { + return fmt.Errorf("outcome unknown, request was sent: %w", err), false + } return err, true } defer resp.Body.Close() @@ -212,14 +283,12 @@ func (n *Notifier) attempt(u string) (err error, retry bool) { } } -// Shutdown refuses new events, stops pending retries and waits for in-flight -// requests until ctx expires. Safe to call more than once. +// Shutdown refuses new events and waits for in-flight notifications and their +// pending retries until ctx expires; then remaining retries are abandoned. +// Safe to call more than once. func (n *Notifier) Shutdown(ctx context.Context) error { n.mu.Lock() - if !n.closed { - n.closed = true - close(n.quit) - } + n.closed = true n.mu.Unlock() done := make(chan struct{}) @@ -231,6 +300,7 @@ func (n *Notifier) Shutdown(ctx context.Context) error { case <-done: return nil case <-ctx.Done(): + n.quitOnce.Do(func() { close(n.quit) }) return errors.Join(ctx.Err(), errors.New("notifications still in flight")) } } diff --git a/internal/notify/notify_test.go b/internal/notify/notify_test.go index e940f21..630df70 100644 --- a/internal/notify/notify_test.go +++ b/internal/notify/notify_test.go @@ -36,10 +36,15 @@ func TestBuildURL(t *testing.T) { } func TestURLFor(t *testing.T) { - cfg := Config{OrgURL: "o", AddressURL: "a", DomainURL: "d"} + cfg := Config{OrgURL: "o", AddressURL: "a", DomainURL: "d", WarnOrgURL: "wo", WarnAddressURL: "wa", WarnDomainURL: "wd"} for level, want := range map[string]string{"O": "o", "A": "a", "D": "d", "X": ""} { - if got := cfg.URLFor(level); got != want { - t.Errorf("URLFor(%s) = %q, want %q", level, got, want) + if got := cfg.URLFor(KindLimit, level); got != want { + t.Errorf("URLFor(limit, %s) = %q, want %q", level, got, want) + } + } + for level, want := range map[string]string{"O": "wo", "A": "wa", "D": "wd", "X": ""} { + if got := cfg.URLFor(KindWarning, level); got != want { + t.Errorf("URLFor(warning, %s) = %q, want %q", level, got, want) } } } @@ -260,20 +265,146 @@ func TestNotify_NoRetryOn4xx(t *testing.T) { } } -func TestNotify_TimeoutIsRetried(t *testing.T) { +// The request reached the web app and only the answer is missing: it may have +// acted, so a retry could send the customer a second email. +func TestNotify_TimeoutAfterWriteIsNotRetried(t *testing.T) { c := &capture{delay: 300 * time.Millisecond} s := httptest.NewServer(c) defer s.Close() h := newHooks() cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") cfg.Timeout = 50 * time.Millisecond - cfg.Attempts = 2 n := New(cfg, h.Hooks()) n.Notify(Event{Level: "O", ID: "7"}) h.wait(t, 1) n.Shutdown(context.Background()) - if len(h.failed) != 1 || h.failed[0].attempts != 2 { - t.Errorf("failed=%+v; want failure after 2 timed-out attempts", h.failed) + if len(h.failed) != 1 || h.failed[0].attempts != 1 { + t.Fatalf("failed=%+v; want one failure after a single attempt", h.failed) + } + if !strings.Contains(h.failed[0].err.Error(), "outcome unknown") { + t.Errorf("error should say the outcome is unknown: %v", h.failed[0].err) + } + waitCalls(t, c, 1) + time.Sleep(100 * time.Millisecond) + if c.calls.Load() != 1 { + t.Errorf("web app calls = %d, want 1", c.calls.Load()) + } +} + +// Nothing was delivered, so retrying cannot duplicate anything. +func TestNotify_RefusedConnectionIsRetried(t *testing.T) { + s := httptest.NewServer(&capture{}) + addr := s.URL + s.Close() + h := newHooks() + cfg := fastConfig(addr+"/l?level=$LEVEL", "", "") + cfg.Attempts = 3 + n := New(cfg, h.Hooks()) + n.Notify(Event{Level: "O", ID: "7"}) + h.wait(t, 1) + n.Shutdown(context.Background()) + if len(h.failed) != 1 || h.failed[0].attempts != 3 { + t.Errorf("failed=%+v; want failure after 3 attempts", h.failed) + } +} + +// A notification waiting out its backoff must not hold a concurrency slot. +func TestNotify_BackoffReleasesSlot(t *testing.T) { + c := &capture{status: []int{500, 200, 200}} + s := httptest.NewServer(c) + defer s.Close() + h := newHooks() + cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") + cfg.MaxConcurrent = 1 + cfg.Backoff = 400 * time.Millisecond + n := New(cfg, h.Hooks()) + defer n.Shutdown(context.Background()) + n.Notify(Event{Level: "O", ID: "retrying"}) + waitCalls(t, c, 1) + n.Notify(Event{Level: "O", ID: "second"}) + select { + case <-h.done: + case <-time.After(250 * time.Millisecond): + t.Fatal("the second notification waited for the first one's backoff") + } + h.done <- struct{}{} + h.wait(t, 2) +} + +func waitCalls(t *testing.T, c *capture, n int32) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for c.calls.Load() < n { + if time.Now().After(deadline) { + t.Fatalf("calls = %d, want %d", c.calls.Load(), n) + } + time.Sleep(time.Millisecond) + } +} + +// Shutdown lets a pending retry finish while its deadline allows. +func TestNotify_ShutdownFinishesPendingRetries(t *testing.T) { + c := &capture{status: []int{500, 200}} + s := httptest.NewServer(c) + defer s.Close() + h := newHooks() + cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") + cfg.Backoff = 100 * time.Millisecond + n := New(cfg, h.Hooks()) + n.Notify(Event{Level: "O", ID: "7"}) + waitCalls(t, c, 1) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := n.Shutdown(ctx); err != nil { + t.Fatal(err) + } + if len(h.sent) != 1 || h.sent[0].attempt != 2 || len(h.failed) != 0 { + t.Errorf("sent=%+v failed=%+v; want delivery on attempt 2", h.sent, h.failed) + } +} + +// When the deadline cuts a retry short the failure reports the real attempt count. +func TestNotify_ShutdownDeadlineReportsRealAttempts(t *testing.T) { + c := &capture{status: []int{500}} + s := httptest.NewServer(c) + defer s.Close() + h := newHooks() + cfg := fastConfig(s.URL+"/l?level=$LEVEL", "", "") + cfg.Backoff = 5 * time.Second + n := New(cfg, h.Hooks()) + n.Notify(Event{Level: "O", ID: "7"}) + waitCalls(t, c, 1) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + _ = n.Shutdown(ctx) + h.wait(t, 1) + if len(h.failed) != 1 || h.failed[0].attempts != 1 { + t.Errorf("failed=%+v; want 1 real attempt", h.failed) + } +} + +func TestNotify_WarningUsesWarningURL(t *testing.T) { + breach, warn := &capture{}, &capture{} + sb, sw := httptest.NewServer(breach), httptest.NewServer(warn) + defer sb.Close() + defer sw.Close() + h := newHooks() + cfg := fastConfig(sb.URL+"/l?level=$LEVEL", "", "") + cfg.WarnOrgURL = sw.URL + "/w?level=$LEVEL&limit=$LIMIT&percent=$PERCENT" + n := New(cfg, h.Hooks()) + defer n.Shutdown(context.Background()) + if !n.Notify(Event{Kind: KindWarning, Level: "O", ID: "7", Limit: 100, Count: 80, Percent: 80}) { + t.Fatal("warning refused") + } + h.wait(t, 1) + if breach.calls.Load() != 0 || warn.calls.Load() != 1 { + t.Fatalf("breach=%d warn=%d, want the warning URL only", breach.calls.Load(), warn.calls.Load()) + } + if q := warn.requests[0].URL.Query(); q.Get("percent") != "80" || q.Get("limit") != "100" { + t.Errorf("query = %v", q) + } + if n.Notify(Event{Kind: KindWarning, Level: "A"}) { + t.Error("a level with no warning URL must be refused") } } @@ -326,7 +457,7 @@ func TestNotify_ShutdownDrainsAndRespectsContext(t *testing.T) { func TestNew_Defaults(t *testing.T) { n := New(Config{OrgURL: "https://x/?level=$LEVEL"}, Hooks{}) defer n.Shutdown(context.Background()) - if n.cfg.Attempts != 5 || n.cfg.Backoff != time.Second || n.cfg.MaxConcurrent != 8 || n.cfg.Timeout != 5*time.Second { + if n.cfg.Attempts != 5 || n.cfg.Backoff != time.Second || n.cfg.MaxConcurrent != 8 || n.cfg.Timeout != 15*time.Second { t.Errorf("defaults not applied: %+v", n.cfg) } } diff --git a/internal/server/middleware.go b/internal/server/middleware.go index d09010c..e7d581b 100644 --- a/internal/server/middleware.go +++ b/internal/server/middleware.go @@ -48,6 +48,17 @@ func (s *Server) reject(w http.ResponseWriter, r *http.Request, route string, e s.d.Logger.Warn("request rejected", "route", route, "remote", r.RemoteAddr, "status", e.status, "reason", e.msg) } +// rejectBody is reject for a request whose body was read: the trace id, when +// there is one, ties the rejection to the message, in both logs. +func (s *Server) rejectBody(w http.ResponseWriter, r *http.Request, route, traceID string, e *httpError) { + writeError(w, e.status, e.msg) + id := api.SanitizeTraceID(traceID) + s.d.Logger.Warn("request rejected", "route", route, "trace_id", id, "remote", r.RemoteAddr, "status", e.status, "reason", e.msg) + if s.d.TxLog != nil { + s.d.TxLog.Rejected(strings.TrimPrefix(route, "/"), id, e.status, api.SanitizeForLog(e.msg)) + } +} + // statusWriter captures the status code for metrics. type statusWriter struct { http.ResponseWriter diff --git a/internal/server/notifications_test.go b/internal/server/notifications_test.go new file mode 100644 index 0000000..348b537 --- /dev/null +++ b/internal/server/notifications_test.go @@ -0,0 +1,173 @@ +package server + +import ( + "context" + "net/url" + "strings" + "sync" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + + "migadu_counters/internal/store" +) + +// Real CDB: yagolivas.com has A/o=20 and O/o=20, so 80% is 16. +func TestOut_WarnsAt80PercentOnceAndBreachesOnce(t *testing.T) { + h := newHarness(t, options{warnPercent: 80}) + for i := 0; i < 22; i++ { + h.accept("/out", `{"trace_id":"w","auth_user":"a@yagolivas.com","recipient":"x@gmail.com"}`) + } + if err := h.notifier.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + if h.warnAddr.count() != 1 || h.warnOrg.count() != 1 { + t.Errorf("warnings addr=%d org=%d, want 1 each", h.warnAddr.count(), h.warnOrg.count()) + } + if h.addrURL.count() != 1 || h.orgURL.count() != 1 { + t.Errorf("breaches addr=%d org=%d, want 1 each", h.addrURL.count(), h.orgURL.count()) + } + u, err := url.Parse(h.warnAddr.urls[0]) + if err != nil { + t.Fatal(err) + } + q := u.Query() + if u.Path != "/api/limits/warning" || q.Get("count") != "16" || q.Get("limit") != "20" || q.Get("percent") != "80" || q.Get("id") != "a@yagolivas.com" { + t.Errorf("warning url = %s", h.warnAddr.urls[0]) + } + if h.metric(h.m.WarningReached.WithLabelValues("A", "o")) != 1 { + t.Error("warning_reached_total not recorded") + } + if h.metric(h.m.Notifications.WithLabelValues("warning", "A", "sent")) != 1 { + t.Error("notifications_total{warning,A,sent} not recorded") + } + if !strings.Contains(h.tx.String(), "warning reached (level=[A] id=[a@yagolivas.com] flow=[o] by=[a@yagolivas.com]) => notifying warn_address_url") { + t.Errorf("transaction log:\n%s", h.tx.String()) + } +} + +func TestOut_ConcurrentIncrementsNotifyOnce(t *testing.T) { + h := newHarness(t, options{warnPercent: 80, workers: 8}) + var wg sync.WaitGroup + for i := 0; i < 40; i++ { + wg.Add(1) + go func() { + defer wg.Done() + h.post("/out", `{"trace_id":"c","auth_user":"a@yagolivas.com","recipient":"x@gmail.com"}`) + }() + } + wg.Wait() + h.settle() + if err := h.notifier.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + if h.redisInt(store.Key("A", "a@yagolivas.com", day, "o")) != 40 { + t.Fatalf("counter = %d, want 40", h.redisInt(store.Key("A", "a@yagolivas.com", day, "o"))) + } + if h.warnAddr.count() != 1 || h.addrURL.count() != 1 { + t.Errorf("warnings=%d breaches=%d, want 1 each", h.warnAddr.count(), h.addrURL.count()) + } +} + +func TestOut_WarningsOffSendsNone(t *testing.T) { + h := newHarness(t, options{}) + for i := 0; i < 20; i++ { + h.accept("/out", `{"trace_id":"w","auth_user":"a@yagolivas.com","recipient":"x@gmail.com"}`) + } + h.notifier.Shutdown(context.Background()) + if h.warnAddr.count() != 0 || h.addrURL.count() != 1 { + t.Errorf("warnings=%d breaches=%d, want 0 and 1", h.warnAddr.count(), h.addrURL.count()) + } + if strings.Contains(h.tx.String(), "warning reached") { + t.Errorf("transaction log mentions a warning:\n%s", h.tx.String()) + } +} + +// Once a breach was notified, a lost EXEC reply on a later message sends nothing. +func TestLostReplyAfterBreachSendsNothing(t *testing.T) { + backend := miniredis.RunT(t) + proxy := delayProxy(t, backend.Addr(), 3*time.Second) + h := newHarness(t, options{writeAddr: proxy}) + key := store.Key("A", "a@yagolivas.com", day, "o") // A/o=20 + backend.Set(key, "40") + backend.Set(store.MarkerKey("A", "a@yagolivas.com", day, "o", store.MarkerLimit), "1") + h.accept("/out", `{"trace_id":"lost2","auth_user":"a@yagolivas.com","recipient":"x@gmail.com"}`) + if err := h.notifier.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + if h.addrURL.count() != 0 { + t.Errorf("address notifications = %d, want 0", h.addrURL.count()) + } + if !strings.Contains(h.tx.String(), "count=41 limit=20 reached=false (re-read)") { + t.Errorf("transaction log:\n%s", h.tx.String()) + } +} + +// --- review round 2 ------------------------------------------------------------ + +func TestOut_LoginSuffixNeverLogged(t *testing.T) { + h := newHarness(t, options{}) + h.accept("/out", `{"trace_id":"m2","auth_user":"x@yagolivas.com@SECRET\\nFORGED LINE","recipient":"y@yagolivas.com"}`) + h.accept("/out", `{"trace_id":"m3","auth_user":"x@yagolivas.com@SECRET\nFORGED LINE","recipient":"y@yagolivas.com"}`) + if tx := h.tx.String(); strings.Contains(tx, "SECRET") || strings.Contains(tx, "FORGED") { + t.Errorf("transaction log contains the login suffix:\n%s", tx) + } +} + +func TestReject_LeavesTraceInBothLogs(t *testing.T) { + h := newHarness(t, options{}) + if st, _ := h.post("/in", `{"trace_id":"TRACE-XYZ","recipient":"nobody"}`); st != 400 { + t.Fatal(st) + } + if !strings.Contains(h.log.String(), "TRACE-XYZ") { + t.Errorf("app log lacks the trace id:\n%s", h.log.String()) + } + if !strings.Contains(h.tx.String(), " in; rejected: 400 ") { + t.Errorf("counting log lacks the rejection:\n%s", h.tx.String()) + } +} + +func TestErrors_DoNotExposeInternals(t *testing.T) { + h := newHarness(t, options{}) + readAddr, writeAddr := h.read.Addr(), h.write.Addr() + h.read.Close() + st, body := h.get("/counters?address=a@yagolivas.com") + if st != 503 || strings.Contains(body, readAddr) || strings.Contains(body, "dial") { + t.Errorf("/counters = %d %s", st, body) + } + if !strings.Contains(h.log.String(), readAddr) { + t.Errorf("detail should stay in the app log:\n%s", h.log.String()) + } + h.write.Close() + st, body = h.adminGet("/health") + if st != 503 || strings.Contains(body, writeAddr) || strings.Contains(body, "dial") { + t.Errorf("/health = %d %s", st, body) + } +} + +func TestPanic_CountedInRequestMetrics(t *testing.T) { + h := newHarness(t, options{}) + h.srv.panicHook = func() { panic("boom") } + if st, _ := h.post("/in", `{"recipient":"b@yagolivas.com"}`); st != 500 { + t.Fatal(st) + } + if got := h.metric(h.m.Requests.WithLabelValues("/in", "500")); got != 1 { + t.Errorf("requests_total{/in,500} = %v, want 1", got) + } +} + +// Real CDB: xn-- domains are hosted under their punycode key. +func TestCounters_UnicodeDomainIsNormalised(t *testing.T) { + h := newHarness(t, options{}) + if st, _ := h.get("/counters?domain=host_name.example.com"); st != 400 { + t.Errorf("invalid domain status = %d, want 400", st) + } + st, body := h.get("/counters?domain=" + url.QueryEscape("Bücher.example.")) + if st != 200 { + t.Fatalf("status %d: %s", st, body) + } + if r := decodeCounters(t, body); r.Query != "xn--bcher-kva.example" { + t.Errorf("query = %q, want the punycode form", r.Query) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index b8783ac..2a65857 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -92,7 +92,7 @@ func NotifyHooks(tx *txlog.Logger, m *metrics.Metrics) notify.Hooks { return notify.Hooks{ Sent: func(ev notify.Event, url string, attempt int) { if m != nil { - m.RecordNotification(ev.Level, "sent") + m.RecordNotification(ev.KindName(), ev.Level, "sent") } if tx != nil { tx.NotificationSent(ev, url, attempt) @@ -100,7 +100,7 @@ func NotifyHooks(tx *txlog.Logger, m *metrics.Metrics) notify.Hooks { }, Failed: func(ev notify.Event, url string, attempts int, err error) { if m != nil { - m.RecordNotification(ev.Level, "failed") + m.RecordNotification(ev.KindName(), ev.Level, "failed") } if tx != nil { tx.NotificationFailed(ev, url, attempts, err) @@ -109,7 +109,7 @@ func NotifyHooks(tx *txlog.Logger, m *metrics.Metrics) notify.Hooks { } } -// Handler returns the API handler: recover → metrics → timeout → routes. +// Handler returns the API handler: metrics → recover → timeout → routes. func (s *Server) Handler() http.Handler { mux := http.NewServeMux() mux.Handle("/out", s.route("/out", http.MethodPost, s.handleOut)) @@ -119,7 +119,7 @@ func (s *Server) Handler() http.Handler { writeError(w, http.StatusNotFound, "not found") }) timed := http.TimeoutHandler(mux, s.d.HandlerTimeout, `{"error":"request timeout"}`) - return s.recoverer(s.observe(timed)) + return s.observe(s.recoverer(timed)) } // AdminHandler returns /health and /metrics (no auth; bind admin_addr locally). @@ -135,17 +135,17 @@ func (s *Server) AdminHandler() http.Handler { func (s *Server) handleOut(w http.ResponseWriter, r *http.Request) { var req api.OutRequest if e := decodeJSON(r, &req); e != nil { - s.reject(w, r, "/out", e) + s.rejectBody(w, r, "/out", req.TraceID, e) return } v, err := api.ValidateOut(req) if err != nil { - s.reject(w, r, "/out", &httpError{http.StatusBadRequest, err.Error()}) + s.rejectBody(w, r, "/out", req.TraceID, &httpError{http.StatusBadRequest, err.Error()}) return } job := counting.Job{ Route: counting.RouteOut, TraceID: v.TraceID, Received: s.d.Now(), - AuthUser: v.AuthUser.Value, LoginSuffix: v.LoginSuffix, + AuthUser: v.AuthUser.Value, TokenLogin: v.TokenLogin, MailFrom: v.MailFrom, MailFromUsable: v.MailFromUsable, Recipient: v.Recipient.Value, } @@ -161,12 +161,12 @@ func (s *Server) handleOut(w http.ResponseWriter, r *http.Request) { func (s *Server) handleIn(w http.ResponseWriter, r *http.Request) { var req api.InRequest if e := decodeJSON(r, &req); e != nil { - s.reject(w, r, "/in", e) + s.rejectBody(w, r, "/in", req.TraceID, e) return } v, err := api.ValidateIn(req) if err != nil { - s.reject(w, r, "/in", &httpError{http.StatusBadRequest, err.Error()}) + s.rejectBody(w, r, "/in", req.TraceID, &httpError{http.StatusBadRequest, err.Error()}) return } s.enqueue(w, r, "/in", counting.Job{ @@ -242,16 +242,23 @@ func (s *Server) process(ctx context.Context, job counting.Job) { if s.d.Metrics != nil { s.d.Metrics.RecordIncrement(res.Level, res.Flow) } - if res.Reached { + if res.MarkerErr != nil { + if s.d.Metrics != nil { + s.d.Metrics.RecordRedisError("marker") + } + s.d.Logger.Error("notification marker failed", "route", plan.Route, "trace_id", plan.TraceID, "key", res.Key, "error", res.MarkerErr) + } + switch { + case res.Reached: if s.d.Metrics != nil { s.d.Metrics.RecordLimitReached(res.Level, res.Flow) } - ev := notify.Event{TraceID: plan.TraceID, Route: plan.Route, Level: res.Level, ID: res.ID, - Flow: res.Flow, Address: res.By, Count: res.Count, Limit: res.Limit} - dispatched[i] = s.d.Notifier != nil && s.d.Notifier.Notify(ev) - if !dispatched[i] && s.d.Metrics != nil { - s.d.Metrics.RecordNotification(res.Level, "disabled") + dispatched[i] = s.notify(plan, res, notify.KindLimit) + case res.Warned: + if s.d.Metrics != nil { + s.d.Metrics.RecordWarningReached(res.Level, res.Flow) } + dispatched[i] = s.notify(plan, res, notify.KindWarning) } } if s.d.TxLog != nil { @@ -259,11 +266,22 @@ func (s *Server) process(ctx context.Context, job counting.Job) { } } +// notify dispatches one notification and reports whether it was handed over. +func (s *Server) notify(plan *counting.Plan, res counting.Result, kind string) bool { + ev := notify.Event{Kind: kind, TraceID: plan.TraceID, Route: plan.Route, Level: res.Level, ID: res.ID, + Flow: res.Flow, Address: res.By, Count: res.Count, Limit: res.Limit, Percent: s.d.Counter.Planner.WarnPercent} + sent := s.d.Notifier != nil && s.d.Notifier.Notify(ev) + if !sent && s.d.Metrics != nil { + s.d.Metrics.RecordNotification(kind, res.Level, "disabled") + } + return sent +} + // handleCounters answers GET /counters?address=|domain=[&date=] from the // read side. func (s *Server) handleCounters(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() - address, domain := strings.TrimSpace(q.Get("address")), strings.ToLower(strings.TrimSpace(q.Get("domain"))) + address, domain := strings.TrimSpace(q.Get("address")), strings.TrimSpace(q.Get("domain")) if (address == "") == (domain == "") { s.reject(w, r, "/counters", &httpError{http.StatusBadRequest, "exactly one of address or domain is required"}) return @@ -287,6 +305,12 @@ func (s *Server) handleCounters(w http.ResponseWriter, r *http.Request) { at := strings.LastIndex(addr, "@") local, domain = addr[:at], addr[at+1:] } else { + dom, err := api.NormalizeDomain(domain) + if err != nil { + s.reject(w, r, "/counters", &httpError{http.StatusBadRequest, "domain: invalid domain"}) + return + } + domain = dom resp.Query, resp.Mode = domain, "domain" } @@ -299,7 +323,8 @@ func (s *Server) handleCounters(w http.ResponseWriter, r *http.Request) { if errors.Is(err, domains.ErrMalformed) { s.d.Logger.Warn("malformed row in domains db, domain treated as not hosted", "domain", domain, "route", "/counters", "error", err) } else if err != nil && !errors.Is(err, domains.ErrNotFound) { - writeError(w, http.StatusServiceUnavailable, "domain lookup failed: "+err.Error()) + s.d.Logger.Error("domain lookup failed", "domain", domain, "route", "/counters", "error", err) + writeError(w, http.StatusServiceUnavailable, "domain lookup failed") return } managed := err == nil @@ -326,7 +351,8 @@ func (s *Server) handleCounters(w http.ResponseWriter, r *http.Request) { if s.d.Metrics != nil { s.d.Metrics.RecordRedisError("mget") } - writeError(w, http.StatusServiceUnavailable, "redis unavailable: "+err.Error()) + s.d.Logger.Error("redis read failed", "route", "/counters", "error", err) + writeError(w, http.StatusServiceUnavailable, "redis unavailable") return } limit := func(level, flow string) *int64 { @@ -355,7 +381,8 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { return } if err := db.Ping(); err != nil { - writeError(w, http.StatusServiceUnavailable, "domains db not readable: "+err.Error()) + s.d.Logger.Warn("health: domains db not readable", "error", err) + writeError(w, http.StatusServiceUnavailable, "domains db not readable") return } ctx, cancel := context.WithTimeout(r.Context(), 200*time.Millisecond) @@ -364,7 +391,8 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { if s.d.Metrics != nil { s.d.Metrics.RecordRedisError("ping") } - writeError(w, http.StatusServiceUnavailable, fmt.Sprintf("redis: %v", err)) + s.d.Logger.Warn("health: redis not reachable", "error", err) + writeError(w, http.StatusServiceUnavailable, "redis unavailable") return } writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) diff --git a/internal/server/server_test.go b/internal/server/server_test.go index af247ff..f0b54d7 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -87,6 +87,7 @@ type options struct { exempt map[string]bool queueSize int workers int + warnPercent int64 // > 0: warnings on, sent to warnAddr / warnOrg } type harness struct { @@ -102,6 +103,8 @@ type harness struct { addrURL *capture orgURL *capture domURL *capture + warnAddr *capture + warnOrg *capture log *syncBuffer tx *syncBuffer reg *prometheus.Registry @@ -154,6 +157,14 @@ func newHarness(t *testing.T, o options) *harness { if o.domainURL { ncfg.DomainURL = sd.URL + q } + h.warnAddr, h.warnOrg = &capture{}, &capture{} + if o.warnPercent > 0 { + swa, swo := httptest.NewServer(h.warnAddr), httptest.NewServer(h.warnOrg) + t.Cleanup(swa.Close) + t.Cleanup(swo.Close) + wq := "/api/limits/warning?level=$LEVEL&id=$ID&flow=$FLOW&period=$PERIOD&by=$ADDRESS&count=$COUNT&limit=$LIMIT&percent=$PERCENT" + ncfg.WarnAddressURL, ncfg.WarnOrgURL = swa.URL+wq, swo.URL+wq + } h.reg = prometheus.NewRegistry() h.m = metrics.New(h.reg) @@ -164,7 +175,7 @@ func newHarness(t *testing.T, o options) *harness { logger := slog.New(slog.NewTextHandler(h.log, &slog.HandlerOptions{Level: slog.LevelDebug})) planner := &counting.Planner{ Lookup: counting.LookupFunc(func(d string) (*domains.Domain, error) { return (*h.lookup.Load())(d) }), - Exempt: o.exempt, Now: func() time.Time { return clock }, + Exempt: o.exempt, Now: func() time.Time { return clock }, WarnPercent: o.warnPercent, } s := New(Deps{ AuthToken: o.token, @@ -349,8 +360,8 @@ func TestOut_ReachedNotifiesConfiguredLevelsOnce(t *testing.T) { t.Errorf("transaction log lacks %q:\n%s", want, tx) } } - if h.metric(h.m.LimitReached.WithLabelValues("A", "o")) != 1 || h.metric(h.m.Notifications.WithLabelValues("A", "sent")) != 1 || - h.metric(h.m.Notifications.WithLabelValues("D", "disabled")) != 1 { + if h.metric(h.m.LimitReached.WithLabelValues("A", "o")) != 1 || h.metric(h.m.Notifications.WithLabelValues("limit", "A", "sent")) != 1 || + h.metric(h.m.Notifications.WithLabelValues("limit", "D", "disabled")) != 1 { t.Error("limit/notification metrics not recorded") } } @@ -569,7 +580,7 @@ func TestOut_LoginSuffixCountsBaseAddress(t *testing.T) { if h.redisInt(store.Key("A", "x@yagolivas.com", day, "o")) != 1 || h.redisInt(store.Key("A", "y@yagolivas.com", day, "i")) != 1 { t.Errorf("base address not counted: %v", h.write.Keys()) } - if !strings.Contains(h.tx.String(), " out; request auth_user=x@yagolivas.com recipient=y@yagolivas.com mail_from=x@yagolivas.com login_suffix=SUPPORTTOKEN") { + if !strings.Contains(h.tx.String(), " out; request auth_user=x@yagolivas.com recipient=y@yagolivas.com mail_from=x@yagolivas.com token_login=true") { t.Errorf("transaction log:\n%s", h.tx.String()) } } @@ -1046,7 +1057,7 @@ func TestNotifyHooks(t *testing.T) { ev := notify.Event{TraceID: "t", Route: "out", Level: "O"} hooks.Sent(ev, "http://x", 1) hooks.Failed(ev, "http://x", 5, io.EOF) - if testutil.ToFloat64(m.Notifications.WithLabelValues("O", "sent")) != 1 || testutil.ToFloat64(m.Notifications.WithLabelValues("O", "failed")) != 1 { + if testutil.ToFloat64(m.Notifications.WithLabelValues("limit", "O", "sent")) != 1 || testutil.ToFloat64(m.Notifications.WithLabelValues("limit", "O", "failed")) != 1 { t.Error("hooks did not record metrics") } if s := buf.String(); !strings.Contains(s, "sent limit exceeded") || !strings.Contains(s, "failed limit exceeded") { diff --git a/internal/store/markers_test.go b/internal/store/markers_test.go new file mode 100644 index 0000000..12908a4 --- /dev/null +++ b/internal/store/markers_test.go @@ -0,0 +1,132 @@ +package store + +import ( + "context" + "errors" + "net" + "testing" + "time" + + "migadu_counters/internal/config" +) + +func TestMarkerKey(t *testing.T) { + got := MarkerKey("A", "x@example.com", "2026-10-13", "o", MarkerLimit) + if want := "ntf/A/x@example.com/d/2026-10-13/o/limit"; got != want { + t.Errorf("MarkerKey = %q, want %q", got, want) + } + if got := MarkerKey("O", "7", "2026-10-13", "i", MarkerWarn); got != "ntf/O/7/d/2026-10-13/i/warn" { + t.Errorf("MarkerKey = %q", got) + } +} + +func TestMarkAndClaimNotified(t *testing.T) { + s, w, _ := testStore(t) + ctx := context.Background() + if err := s.MarkNotified(ctx, "ntf/a"); err != nil { + t.Fatal(err) + } + if ttl := w.TTL("ntf/a"); ttl != MarkerTTL { + t.Errorf("TTL = %v, want %v", ttl, MarkerTTL) + } + // Marking again is not an error: a reset counter may cross again. + if err := s.MarkNotified(ctx, "ntf/a"); err != nil { + t.Errorf("second mark: %v", err) + } + ok, err := s.ClaimNotified(ctx, "ntf/a") + if err != nil || ok { + t.Errorf("claim of a marked key = %v, %v; want false", ok, err) + } + ok, err = s.ClaimNotified(ctx, "ntf/b") + if err != nil || !ok { + t.Errorf("first claim = %v, %v; want true", ok, err) + } + if ttl := w.TTL("ntf/b"); ttl != MarkerTTL { + t.Errorf("claim TTL = %v, want %v", ttl, MarkerTTL) + } + ok, err = s.ClaimNotified(ctx, "ntf/b") + if err != nil || ok { + t.Errorf("second claim = %v, %v; want false", ok, err) + } +} + +func TestClaimNotified_RedisDown(t *testing.T) { + s, w, _ := testStore(t) + w.Close() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if ok, err := s.ClaimNotified(ctx, "ntf/a"); err == nil || ok { + t.Errorf("claim with redis down = %v, %v; want an error and false", ok, err) + } +} + +// No connection could be had from the pool: no command was ever sent. +func TestOutcomeUnknown_PoolTimeoutIsDefinite(t *testing.T) { + if OutcomeUnknown(errors.New("redis incr: redis: connection pool timeout")) { + t.Error("a pool timeout never reached Redis and must not be an unknown outcome") + } +} + +func TestOutcomeUnknown_NotAppliedIsDefinite(t *testing.T) { + if OutcomeUnknown(ErrNotApplied) { + t.Error("ErrNotApplied must not be an unknown outcome") + } +} + +// Transaction-level rejection: every command failed, nothing was applied. +func TestIncr_RejectedTransactionIsNotApplied(t *testing.T) { + s, w, _ := testStore(t) + w.SetError("READONLY You can't write against a read only replica.") + defer w.SetError("") + _, err := s.Incr(context.Background(), []string{"a", "b"}) + if !errors.Is(err, ErrNotApplied) { + t.Fatalf("err = %v, want ErrNotApplied", err) + } +} + +// All INCRs failing on non-numbers is still per-key: the EXPIREs succeeded and +// EXEC ran, so it must not be mistaken for a rejected transaction. +func TestIncr_AllKeysNonNumericIsPerKey(t *testing.T) { + s, w, _ := testStore(t) + w.Set("a", "x") + w.Set("b", "y") + res, err := s.Incr(context.Background(), []string{"a", "b"}) + if err != nil { + t.Fatalf("per-key failures must not fail the transaction: %v", err) + } + if res[0].Err == nil || res[1].Err == nil { + t.Errorf("results = %+v", res) + } +} + +// A context deadline must bound socket time, not only the configured timeout. +func TestPing_HonoursContextDeadline(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { // accepts and never answers + for { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() + } + }() + s, err := New(config.RedisConfig{WriteAddr: ln.Addr().String(), PoolSize: 2, TimeoutMS: 3000}) + if err != nil { + t.Fatal(err) + } + defer s.Close() + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + start := time.Now() + if err := s.Ping(ctx); err == nil { + t.Fatal("expected a timeout") + } + if d := time.Since(start); d > time.Second { + t.Errorf("Ping took %v with a 100ms deadline", d) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index f4e2948..7d19e47 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -32,13 +32,30 @@ const ( Prefix = "cnt" // TTL is refreshed on every increment, like the Lua plugin did. TTL = 31 * 24 * time.Hour + + // MarkerLimit and MarkerWarn name the two notification markers of a counter. + MarkerLimit = "limit" + MarkerWarn = "warn" + // MarkerTTL outlives the UTC day the marker belongs to. + MarkerTTL = 48 * time.Hour ) +// ErrNotApplied wraps an Incr failure where Redis answered and rejected the +// whole transaction (EXECABORT, READONLY, LOADING, ...): nothing was applied, +// so running it again cannot double count. +var ErrNotApplied = errors.New("transaction not applied") + // Key builds a counter key: Key("A", "x@example.com", "2026-10-13", "i"). func Key(level, id, date, flow string) string { return Prefix + "/" + level + "/" + id + "/d/" + date + "/" + flow } +// MarkerKey builds the key recording that a counter's notification went out. +// It uses its own prefix because the web app scans cnt/*
*. +func MarkerKey(level, id, date, flow, kind string) string { + return "ntf/" + level + "/" + id + "/d/" + date + "/" + flow + "/" + kind +} + // DayKey returns the UTC calendar day of t, as used in the keys. The web app // builds its keys with Date.utc_today(), so this must never use local time. func DayKey(t time.Time) string { return t.UTC().Format("2006-01-02") } @@ -69,6 +86,8 @@ func New(cfg config.RedisConfig) (*Store, error) { DialTimeout: timeout, ReadTimeout: timeout, WriteTimeout: timeout, + + ContextTimeoutEnabled: true, } write := redis.NewFailoverClient(&base) // ReplicaOnly picks a random replica and falls back to the master when @@ -91,6 +110,8 @@ func New(cfg config.RedisConfig) (*Store, error) { DialTimeout: timeout, ReadTimeout: timeout, WriteTimeout: timeout, + + ContextTimeoutEnabled: true, } } addrs := cfg.ReadAddrs @@ -137,7 +158,7 @@ func (s *Store) Incr(ctx context.Context, keys []string) ([]IncrResult, error) { return nil, nil } cmds := make([]*redis.IntCmd, len(keys)) - _, err := s.write.TxPipelined(ctx, func(pipe redis.Pipeliner) error { + all, err := s.write.TxPipelined(ctx, func(pipe redis.Pipeliner) error { for i, k := range keys { cmds[i] = pipe.Incr(ctx, k) pipe.Expire(ctx, k, TTL) @@ -154,6 +175,9 @@ func (s *Store) Incr(ctx context.Context, keys []string) ([]IncrResult, error) { if !errors.As(err, &rerr) { return nil, fmt.Errorf("redis incr: %w", err) } + if allFailed(all) { + return nil, fmt.Errorf("redis incr: %w: %w", ErrNotApplied, err) + } } for i, c := range cmds { if cerr := c.Err(); cerr != nil { @@ -165,12 +189,29 @@ func (s *Store) Incr(ctx context.Context, keys []string) ([]IncrResult, error) { return out, nil } +// allFailed tells a rejected transaction from per-key failures: an INCR on a +// non-number fails alone and its EXPIRE still succeeds, so only a transaction +// Redis refused as a whole leaves every command with an error. +func allFailed(cmds []redis.Cmder) bool { + for _, c := range cmds { + if c.Err() == nil { + return false + } + } + return len(cmds) > 0 +} + // OutcomeUnknown reports whether a failed Incr may still have been applied // by Redis: the request may have been written but the reply lost (timeout, // EOF, reset). A refused connection, a dial failure or a cancelled context // never reached Redis and is a definite failure. func OutcomeUnknown(err error) bool { - if err == nil || errors.Is(err, context.Canceled) { + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, ErrNotApplied) { + return false + } + // No connection could be had from the pool, so no command was sent. go-redis + // keeps this error in an internal package; only its text is available. + if strings.Contains(err.Error(), "connection pool timeout") { return false } var opErr *net.OpError @@ -188,6 +229,18 @@ func OutcomeUnknown(err error) bool { return strings.Contains(msg, "timeout") || strings.Contains(msg, "reset") || strings.Contains(msg, "broken pipe") || strings.Contains(msg, "EOF") } +// MarkNotified records that a counter's notification was sent. It overwrites +// any older marker: a counter the web app reset may legitimately cross again. +func (s *Store) MarkNotified(ctx context.Context, key string) error { + return s.write.Set(ctx, key, "1", MarkerTTL).Err() +} + +// ClaimNotified atomically takes the right to send a notification. It returns +// true for exactly one caller while the marker exists. +func (s *Store) ClaimNotified(ctx context.Context, key string) (bool, error) { + return s.write.SetNX(ctx, key, "1", MarkerTTL).Result() +} + // GetMaster reads the keys from the write side (the master) with one MGET; // used to establish the outcome of an Incr whose reply was lost, when a // replica might not have the increment yet. A missing key reads as 0. diff --git a/internal/txlog/txlog.go b/internal/txlog/txlog.go index 74601c8..79aedfc 100644 --- a/internal/txlog/txlog.go +++ b/internal/txlog/txlog.go @@ -17,6 +17,7 @@ import ( "errors" "fmt" "io" + "net/url" "sort" "strings" "time" @@ -102,8 +103,12 @@ func (l *Logger) Outcome(out *counting.Outcome, dispatched []bool) { b.tracked(r, out.Unknown != nil) } for i, r := range out.Results { + sent := i < len(dispatched) && dispatched[i] if r.Reached { - b.reached(r, i < len(dispatched) && dispatched[i]) + b.reached(r, sent) + } + if r.Warned { + b.warned(r, sent) } } if out.Skipped != "" { @@ -128,6 +133,13 @@ func (l *Logger) Dropped(route, traceID string) { b.flush() } +// Rejected logs a request refused by the validation layer. +func (l *Logger) Rejected(route, traceID string, status int, reason string) { + b := l.begin(traceID, route) + b.line("rejected", fmt.Sprintf("rejected: %d %s (not counted)", status, reason), map[string]any{"status": status, "reason": reason}) + b.flush() +} + // Error logs a request that failed before a plan existed (domain lookup error). func (l *Logger) Error(route, traceID string, err error) { b := l.begin(traceID, route) @@ -136,28 +148,39 @@ func (l *Logger) Error(route, traceID string, err error) { } // NotificationSent logs a delivered web app notification (notify hook). -func (l *Logger) NotificationSent(ev notify.Event, url string, attempt int) { +func (l *Logger) NotificationSent(ev notify.Event, rawURL string, attempt int) { + u := redact(rawURL) b := l.begin(ev.TraceID, ev.Route) - b.line("notification_sent", fmt.Sprintf("sent limit exceeded notification to url %q", url), - map[string]any{"url": url, "attempt": attempt, "level": ev.Level, "id": ev.ID, "flow": ev.Flow}) + b.line("notification_sent", fmt.Sprintf("sent %s notification to url %q", kindWords(ev), u), + map[string]any{"url": u, "attempt": attempt, "kind": ev.KindName(), "level": ev.Level, "id": ev.ID, "flow": ev.Flow}) b.flush() } // NotificationFailed logs a notification given up on (notify hook). -func (l *Logger) NotificationFailed(ev notify.Event, url string, attempts int, err error) { +func (l *Logger) NotificationFailed(ev notify.Event, rawURL string, attempts int, err error) { + u := redact(rawURL) b := l.begin(ev.TraceID, ev.Route) - b.line("notification_failed", fmt.Sprintf("failed limit exceeded notification to url %q after %d attempts: %v", url, attempts, err), - map[string]any{"url": url, "attempts": attempts, "error": err.Error(), "level": ev.Level, "id": ev.ID, "flow": ev.Flow}) + b.line("notification_failed", fmt.Sprintf("failed %s notification to url %q after %d attempts: %v", kindWords(ev), u, attempts, err), + map[string]any{"url": u, "attempts": attempts, "error": err.Error(), "kind": ev.KindName(), "level": ev.Level, "id": ev.ID, "flow": ev.Flow}) b.flush() } +// redact hides a password embedded in a notification URL. +func redact(raw string) string { + u, err := url.Parse(raw) + if err != nil { + return "" + } + return u.Redacted() +} + func (b *block) request(p *counting.Plan) { if p.Route == counting.RouteOut { msg := fmt.Sprintf("request auth_user=%s recipient=%s mail_from=%s", p.AuthUser, p.Recipient, p.MailFrom) fields := map[string]any{"auth_user": p.AuthUser, "recipient": p.Recipient, "mail_from": p.MailFrom} - if p.LoginSuffix != "" { - msg += " login_suffix=" + p.LoginSuffix - fields["login_suffix"] = p.LoginSuffix + if p.TokenLogin { + msg += " token_login=true" + fields["token_login"] = true } b.line("request", msg, fields) } else { @@ -235,7 +258,7 @@ func (b *block) tracked(r counting.Result, reread bool) { } func (b *block) reached(r counting.Result, dispatched bool) { - name := notify.URLName(r.Level) + name := notify.URLName(notify.KindLimit, r.Level) verdict := "notifying " + name if !dispatched { verdict = "no " + name + " configured, not notifying" @@ -245,6 +268,24 @@ func (b *block) reached(r counting.Result, dispatched bool) { map[string]any{"level": r.Level, "id": r.ID, "flow": r.Flow, "by": r.By, "count": r.Count, "limit": r.Limit, "notifying": dispatched, "url_name": name}) } +func (b *block) warned(r counting.Result, dispatched bool) { + name := notify.URLName(notify.KindWarning, r.Level) + verdict := "notifying " + name + if !dispatched { + verdict = "no " + name + " configured, not notifying" + } + b.line("warning_reached", + fmt.Sprintf("warning reached (level=[%s] id=[%s] flow=[%s] by=[%s]) => %s", r.Level, r.ID, r.Flow, r.By, verdict), + map[string]any{"level": r.Level, "id": r.ID, "flow": r.Flow, "by": r.By, "count": r.Count, "warn_at": r.WarnAt, "limit": r.Limit, "notifying": dispatched, "url_name": name}) +} + +func kindWords(ev notify.Event) string { + if ev.KindName() == notify.KindWarning { + return "limit warning" + } + return "limit exceeded" +} + // summarize picks the limit keys worth printing for one address: the base // keys and the address's own overrides. Some domains carry thousands of // per-mailbox overrides (boxmailhere.com: 9 798); dumping them all would make diff --git a/internal/txlog/txlog_test.go b/internal/txlog/txlog_test.go index 96915a1..1c6f577 100644 --- a/internal/txlog/txlog_test.go +++ b/internal/txlog/txlog_test.go @@ -295,7 +295,7 @@ func TestErrorLine(t *testing.T) { // --- review round 1 lines ----------------------------------------------------- func TestRequestLine_LoginSuffixAndUnusable(t *testing.T) { - plan, err := planner().Plan(counting.Job{Route: counting.RouteOut, TraceID: "m1", AuthUser: "x@wearemiraclesnow.com", LoginSuffix: "SUPPORTTOKEN", + plan, err := planner().Plan(counting.Job{Route: counting.RouteOut, TraceID: "m1", AuthUser: "x@wearemiraclesnow.com", TokenLogin: true, MailFrom: "x@wearemiraclesnow.com", MailFromUsable: true, RecipientUnusable: counting.Unusable{Role: "recipient", Value: "u@[192.0.2.1]", Reason: "invalid e-mail address"}}) if err != nil { @@ -304,7 +304,7 @@ func TestRequestLine_LoginSuffixAndUnusable(t *testing.T) { l, buf := newLogger("text") l.Outcome(outcome(plan, 1, 1, 1), nil) got := buf.String() - if !strings.Contains(got, ts+" out; request auth_user=x@wearemiraclesnow.com recipient= mail_from=x@wearemiraclesnow.com login_suffix=SUPPORTTOKEN\n") { + if !strings.Contains(got, ts+" out; request auth_user=x@wearemiraclesnow.com recipient= mail_from=x@wearemiraclesnow.com token_login=true\n") { t.Errorf("request line should show the login suffix:\n%s", got) } if !strings.Contains(got, ts+" out; recipient address not usable: u@[192.0.2.1] (invalid e-mail address, not counted)\n") { @@ -318,7 +318,7 @@ func TestRequestLine_LoginSuffixAndUnusable(t *testing.T) { SenderUnusable: counting.Unusable{Role: "sender", Reason: "missing"}}) l, buf = newLogger("text") l.Outcome(outcome(plan, 1, 1, 1), nil) - if got := buf.String(); !strings.Contains(got, " out; sender address not usable: (missing, not counted)\n") || strings.Contains(got, "login_suffix") { + if got := buf.String(); !strings.Contains(got, " out; sender address not usable: (missing, not counted)\n") || strings.Contains(got, "token_login") { t.Errorf("missing sender line / unexpected suffix:\n%s", got) } @@ -411,3 +411,21 @@ func (m malformedTable) Lookup(d string) (*domains.Domain, error) { } return m.table.Lookup(d) } + +func TestNotificationLines_RedactCredentials(t *testing.T) { + l, buf := newLogger("text") + ev := notify.Event{TraceID: "t", Route: "out", Level: "O"} + l.NotificationSent(ev, "https://api:s3cret@admin.example/api/limits?level=O", 1) + l.NotificationFailed(ev, "https://api:s3cret@admin.example/api/limits?level=O", 2, errors.New("boom")) + if got := buf.String(); strings.Contains(got, "s3cret") || !strings.Contains(got, "api:xxxxx@admin.example/api/limits?level=O") { + t.Errorf("log:\n%s", got) + } +} + +func TestRejected(t *testing.T) { + l, buf := newLogger("text") + l.Rejected("in", "tr9", 400, "recipient: missing") + if got, want := buf.String(), ts+" in; rejected: 400 recipient: missing (not counted)\n"; got != want { + t.Errorf("got %q, want %q", got, want) + } +} diff --git a/review/_proof_tests/README.md b/review/_proof_tests/README.md index 51d4be7..15555ad 100644 --- a/review/_proof_tests/README.md +++ b/review/_proof_tests/README.md @@ -67,5 +67,19 @@ Requirements and notes: | 8 | `internal/server` `TestMalformedRowCountsOtherSide`; `internal/counting` `TestPlan_MalformedRowIsNotHosted`; `internal/domains` `TestLookup_MalformedRow`; `internal/txlog` `TestTrail_MalformedRow` | | 6 | `internal/config` `TestValidate_Errors` (auth_token / unknown keys), `TestLoad_TokenFromEnvOnly`, `TestLoad_Defaults` (`:5555`); `cmd/migadu_counters` `TestRun_FlagsAndExitCodes` | +## Fixed in round 2 (2026-11-09) — regression tests now in the suite + +| Merged # | Regression test | +|---|---| +| 9 | `internal/notify` `TestNotify_TimeoutAfterWriteIsNotRetried`, `TestNotify_BackoffReleasesSlot`, `TestNotify_ShutdownFinishesPendingRetries`, `TestNotify_ShutdownDeadlineReportsRealAttempts` | +| 10 | `cmd/migadu_counters` `TestRun_SIGHUPDuringStartupIsQueued` | +| 11 | `internal/server` `TestReject_LeavesTraceInBothLogs` | +| 12 | `internal/domains` `TestParseLimits_OverrideKeysLowercased` | +| 13 | `internal/txlog` `TestNotificationLines_RedactCredentials`; `internal/config` `TestWarnings_PlainHTTPWithBasicAuth` | +| 14 | `internal/server` `TestErrors_DoNotExposeInternals`; `internal/config` `TestLoad_Defaults` | +| 15 | `internal/config` `TestValidate_Errors` (case "sentinel needs password"), `TestLoad_SentinelPasswordOptional` | +| 16 | `cmd/counterctl` `TestDateWarning_Boundary` | +| 17 | `internal/server` `TestPanic_CountedInRequestMetrics` | + The proof files stay here as the review record; they no longer compile against the current API (`CountResponse`, `store.Incr`) and are not meant to. diff --git a/todo.md b/todo.md index e283af5..e1f58f5 100644 --- a/todo.md +++ b/todo.md @@ -1,6 +1,6 @@ # TODO — migadu_counters -State as of 2026-10-15. Review findings refer to `review/README.md`; the proof tests live in +State as of 2026-11-09. Review findings refer to `review/README.md`; the proof tests live in `review/_proof_tests/` and should become regression tests as each item is fixed (round 1 and items 7–8 are mapped at the end of `review/_proof_tests/README.md`). @@ -12,58 +12,24 @@ State as of 2026-10-15. Review findings refer to `review/README.md`; the proof t - #7 plus-addressing counts against the mailbox (`domains.Mailbox`). - #8 malformed CDB row is "not hosted" + warning (`domains.ErrMalformed`). -## Review round 2 — remaining findings +## Review round 2 — done 2026-11-09 -Suggested order: #10, #15, #9, then the rest in one sweep. +#9-#17 fixed, plus the new findings from the 2026-11-09 review: repeat notifications after a lost Redis +reply, pool timeouts treated as unknown outcomes, login suffix in the counting log, no retry of rejected +transactions, `/counters` punycode, counter day taken from arrival time, CDB corruption checks, context +deadlines on Redis sockets, example config path and placeholders, non-zero exit on a failed drain. +Also added: the 80% early warning (`warn_percent`, `warn_*_url`) and per-counter notification markers. -### Can stop the daemon or lose counts +## Open -- [ ] **#10 SIGHUP during startup kills the daemon.** The signal handler is installed after the Redis - ping (up to 2 s). Install `signal.Notify` before anything slow in `cmd/migadu_counters/main.go` - (`run`), and queue a reload that arrives before the app is ready. Evidence: `review/_proof_tests/a/run/`. -- [ ] **#15 Sentinel mode starts without a Redis password.** Every write fails; only `/health` and - metrics show it. Make an empty `redis.password` fatal in `config.Validate` when `sentinels` is set - (allow an explicit opt-out such as `password_optional = true` for dev). Proof: - `b/internal/config/review_b_test.go`, case "sentinel without password". - [ ] **Persistent / recoverable counting on Redis outage** (deferred from round 1, #4). A Redis stall longer than a few seconds fills the queue and new requests are dropped (503, `dropped_total`). - Options: marker key per transaction for exact lost-reply recovery; a disk-backed queue. Decide - whether the drop-and-alert behaviour is acceptable before building anything. - -### Correctness - -- [ ] **#9 Notification retries can send duplicate customer emails.** The web app's mailbox-level - handler (`core/email.ex:3572-3584`) emails on every call; the org handler is guarded once per day. - In `internal/notify`: treat a slow-but-2xx attempt as success (do not retry after the client timeout - if the request was sent — or make the retry budget shorter than the web app's handling time), do - not keep the concurrency slot while backing off, let Shutdown finish pending retries up to its - deadline, and fix the "after N attempts" count in the failure line. -- [ ] **#12 Override keys are not lowercased.** `Info@/o=5` would be ignored and the domain default - used instead. Lowercase the key in `domains.ParseLimits`; add a test row. No such keys in today's - data. -- [ ] **#16 counterctl warns one day too early.** A date exactly 31 days back is reported as expired. - Compare against the key TTL in whole days (`now.UTC` day minus date > 31). Proof: - `b/cmd/counterctl/review_b_test.go` `TestReviewB_31DayWarningOffByOne`. -- [ ] **#18 Work keeps running after a handler timeout.** Not reproduced by the reviewer and largely - covered by the queue change (the synchronous part only validates). Re-check that - `http.TimeoutHandler` cannot leave a goroutine writing after the 503, then close. - -### Operability / security hygiene - -- [ ] **#11 Rejected requests leave no trace.** 4xx rejections write nothing to the counting log and no - `trace_id` to the app log. Decode `trace_id` best-effort before rejecting; add a - `rejected: ` line to the counting log and `trace_id` to the warn line. Proof: - `b/internal/server/review_b_test.go` `TestReviewB_RejectLogHasNoTraceID`. -- [ ] **#13 Credentials in the notify URL are written to the log.** Redact `user:pass@` from URLs in - the txlog `sent/failed notification` lines; warn at startup when a notify URL is plain `http://` - with basic auth. Proof: `b/internal/server/review_b3_test.go`. -- [ ] **#14 Error responses expose internals.** 503 bodies carry Redis addresses and file paths; keep - the detail in the log and answer a generic message. Default `admin_addr` to `127.0.0.1:9095` - (currently `:9095`, all interfaces, no auth). Proof: `TestReviewB_ErrorLeak`. -- [ ] **#17 A recovered panic is not counted in `requests_total`.** The recoverer sits outside - `observe`; swap the order or record the 500 explicitly. Proof: `TestReviewB_PanicNotInMetrics`. -- [ ] **Test gaps (B F13).** Watcher tests can occasionally pass on a stale signal — make them wait for - the specific reload they triggered. Add a test for `handler_timeout` vs queue timeout interplay. + Decide whether drop-and-alert is acceptable before building anything. +- [ ] **Exempt `mail_from` on `/in`.** An outside sender using `MAIL FROM:` avoids the + recipient's incoming count, unless mizu rejects such senders inbound (not confirmed). Matches the + old Lua rule; decide whether to keep it. +- [ ] **Test gap.** Watcher tests can occasionally pass on a stale signal: make them wait for the + specific reload they triggered. ## Outside this repo (deployment prerequisites) @@ -77,9 +43,12 @@ Suggested order: #10, #15, #9, then the rest in one sweep. `contrib/migadu_counters-pull.cron` (see `contrib/README.md`), agent host key acceptance, firewall: mailqueuer hosts → :5555, service → Sentinel :26379/:6379 and admin.migadu.com:443. - [ ] **migadu_web** `/api/limits` `D` branch (domain level); until then `notify.domain_url` stays empty. +- [ ] **migadu_web3 branch `limit-warning-notifications`**: review the warning email wording, merge and deploy, + then set `warn_org_url` and `warn_address_url`. Until then warnings stay off. +- [ ] **Firewall / monitoring**: `admin_addr` now defaults to localhost; set it explicitly if `/metrics` or + `/health` are scraped from another host. - [ ] **Limits re-tuning**: outgoing is counted per recipient (Lua counted per message), so existing `*/o` limits are stricter. Review before relying on `outgoing_limit_exceeded!`. - [ ] **Retire stats-merger's quota module** once this service is the source (only fernrohr reads it). - [ ] **Dependencies**: `golang.org/x/net` is pinned to v0.43.0 because the offline module cache had it; run `go get -u golang.org/x/net && go mod tidy` with network access. -- [ ] **git init** the repo (never asked for so far).