From 80c3c79a31ccbbfc17ab6fdda6eb2fa2726b50d7 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Fri, 11 Sep 2026 03:11:35 -0700 Subject: [PATCH] feat: add live coverage that the machine windows survive the settings document's jsonb round trip and reach the classifier, a guard that every shipped scanner CIDR is written as its own network address since the loader truncates host bits silently, and correct the comments that claimed an edit lands on the very next event when the consumer reads through a thirty second cache in its own process --- .../dashboard/configuration/SettingsTab.tsx | 4 +- cmd/consumer/main.go | 4 +- .../docs/development/configuration.mdx | 2 +- internal/app/consumer/event_tracking.go | 8 +- .../app/consumer/tracking_window_live_test.go | 151 ++++++++++++++++++ internal/app/instancesettings/service.go | 2 +- tracking/src/scanners.rs | 26 +++ 7 files changed, 190 insertions(+), 7 deletions(-) create mode 100644 internal/app/consumer/tracking_window_live_test.go diff --git a/admin/src/app/dashboard/configuration/SettingsTab.tsx b/admin/src/app/dashboard/configuration/SettingsTab.tsx index 8b8cac20..9e7bd8b8 100644 --- a/admin/src/app/dashboard/configuration/SettingsTab.tsx +++ b/admin/src/app/dashboard/configuration/SettingsTab.tsx @@ -608,7 +608,9 @@ export function SettingsTab({ onDirtyChange, onSwitchTab }: SettingsTabProps) { The clock starts when the send is handed to a worker, so the window also covers the provider's queue and the transit to the recipient. Known scanner networks are matched separately and - are not bounded by time. + are not bounded by time. A change applies within a minute and + only to events recorded after it: opens and clicks already + stored keep the label they were given when they arrived. diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 3d1485db..631d89d4 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -537,7 +537,9 @@ func main() { } else { // The engagement prune reads its window from the instance settings on // every pass, and the machine-window rule reads its windows per event, - // so editing either in the admin panel needs no restart. + // so editing either in the admin panel needs no restart. Both go + // through this process's own read cache, so an edit lands within its + // TTL rather than instantly. trackingSettings := instancesettings.NewService(instancesettings.NewStore(primaryDB.Pool)) trackingConsumer.WireRetention(trackingSettings) trackingConsumer.WireTrackingPolicy(trackingSettings) diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 5e59ac3e..70439be5 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -617,7 +617,7 @@ These are the only settings a browser can change, and no environment variable ow | `retention.engagement_event_days` | integer, 1 to 3650 | `365` | How long the per-event open and click logs (client, device, approximate location) are kept. Campaign progress keeps its own summary that outlives them, so counts, filters and branching never change | | `retention.form_event_days` | integer, 1 to 3650 | `180` | How long form funnel events (views, starts, field-level drop-off) are kept. Funnel reports range up to 90 days, so anything shorter shortens the report too | | `retention.audit_log_days` | integer, 1 to 3650 | `90` | How long the audit trail is kept. It carries IP addresses, user agents and change payloads, so this is also how long that data is held | -| `tracking.machine_window_open_seconds` | integer, 1 to 900 | `60` | How soon after a send was dispatched an open is recorded as automated rather than a person's. The clock starts when the send is handed to a worker, so this window also covers the sending provider's queue and the transit to the recipient, not just reading time. Raise it when delivery-time scanners are being counted as opens; lower it when recipients who read immediately are being missed | +| `tracking.machine_window_open_seconds` | integer, 1 to 900 | `60` | How soon after a send was dispatched an open is recorded as automated rather than a person's. The clock starts when the send is handed to a worker, so this window also covers the sending provider's queue and the transit to the recipient, not just reading time. Raise it when delivery-time scanners are being counted as opens; lower it when recipients who read immediately are being missed. A change applies within a minute and only to events recorded after it | | `tracking.machine_window_click_seconds` | integer, 1 to 900 | `30` | The same window for click tickets. Kept separate because the two mistakes cost different things: a misjudged open loses a metric, a misjudged click loses the automation behind an interested lead | | `deliverability.enforce_domain_auth` | boolean | `true` | Whether a sending domain that fails SPF or DMARC stops cold campaign sending and warmup sending from every mailbox on it. Off keeps the check running and still shows the state and the Advisor card, it just never blocks | | `deliverability.auth_grace_hours` | integer, 1 to 720 | `72` | How long a domain must stay failing before the gate applies. The clock starts when the background check first sees the failure, so this is also how much warning the owner gets | diff --git a/internal/app/consumer/event_tracking.go b/internal/app/consumer/event_tracking.go index e0802b80..56116135 100644 --- a/internal/app/consumer/event_tracking.go +++ b/internal/app/consumer/event_tracking.go @@ -78,9 +78,11 @@ func (tc *TrackingConsumer) engagementRetentionDays(ctx context.Context) int { } // TrackingPolicySource is the operator-editable engagement-classification -// section, satisfied by instancesettings.Service. Read per event, so an edit -// in the admin panel applies to the next open or click rather than at the -// next restart. +// section, satisfied by instancesettings.Service. Read per event rather than +// held from boot, so an edit in the admin panel needs no restart. The +// consumer's own service caches for instancesettings.cacheTTL and the backend +// that wrote the row is a different process, so an edit lands within that TTL, +// not on the very next event. type TrackingPolicySource interface { TrackingPolicy(ctx context.Context) instancesettings.Tracking } diff --git a/internal/app/consumer/tracking_window_live_test.go b/internal/app/consumer/tracking_window_live_test.go new file mode 100644 index 00000000..cee5c603 --- /dev/null +++ b/internal/app/consumer/tracking_window_live_test.go @@ -0,0 +1,151 @@ +package jobs + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/warmbly/warmbly/internal/app/instancesettings" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/repository" +) + +// Live checks that the operator-editable machine windows survive the trip +// through the settings document's jsonb column and reach the classifier. +// Skipped unless WARMBLY_TEST_DB is set: +// +// WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \ +// go test ./internal/app/consumer/ -run Live -v +// +// The unit tests cover the arithmetic. What they cannot cover is the storage: +// the whole section is one jsonb document, so a section that fails to marshal, +// or that an existing instance's document lacks entirely, is only visible +// against a real row. + +// An instance that upgrades has a settings row written before this section +// existed. Its document has no "tracking" key at all, and the windows have to +// come back as the shipped defaults rather than as zero, which would read as +// "no window" and let every delivery-time scan count as a person. +func TestLiveTrackingWindowsDefaultOnADocumentWithoutTheSection(t *testing.T) { + handle := liveDB(t) + ctx := context.Background() + store := instancesettings.NewStore(handle.Pool) + + // A document exactly as an older version would have written it. + old := instancesettings.Defaults() + raw, err := json.Marshal(old) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var stripped map[string]any + if err := json.Unmarshal(raw, &stripped); err != nil { + t.Fatalf("unmarshal: %v", err) + } + delete(stripped, "tracking") + if _, ok := stripped["tracking"]; ok { + t.Fatal("the fixture still carries a tracking section") + } + pruned, err := json.Marshal(stripped) + if err != nil { + t.Fatalf("marshal pruned: %v", err) + } + var doc instancesettings.Document + if err := json.Unmarshal(pruned, &doc); err != nil { + t.Fatalf("unmarshal pruned: %v", err) + } + if err := store.Put(ctx, doc, nil); err != nil { + t.Fatalf("put: %v", err) + } + t.Cleanup(func() { _ = store.Put(ctx, instancesettings.Defaults(), nil) }) + + got, err := store.Get(ctx) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.Tracking.MachineWindowOpenSeconds != config.TrackingMachineWindowOpenSecondsDefault { + t.Errorf("open window = %d, want the shipped %d on a document with no tracking section", + got.Tracking.MachineWindowOpenSeconds, config.TrackingMachineWindowOpenSecondsDefault) + } + if got.Tracking.MachineWindowClickSeconds != config.TrackingMachineWindowClickSecondsDefault { + t.Errorf("click window = %d, want the shipped %d on a document with no tracking section", + got.Tracking.MachineWindowClickSeconds, config.TrackingMachineWindowClickSecondsDefault) + } +} + +// The operator's saved value has to reach the classifier, which is the whole +// point of the setting. This walks the real path: a patch through the service, +// the jsonb row, and the consumer reading it back to classify an event. +func TestLiveTrackingWindowReachesTheClassifier(t *testing.T) { + handle := liveDB(t) + ctx := context.Background() + store := instancesettings.NewStore(handle.Pool) + t.Cleanup(func() { _ = store.Put(ctx, instancesettings.Defaults(), nil) }) + + widened := 300 + patch := instancesettings.Patch{Tracking: &struct { + MachineWindowOpenSeconds *int `json:"machine_window_open_seconds"` + MachineWindowClickSeconds *int `json:"machine_window_click_seconds"` + }{MachineWindowOpenSeconds: &widened}} + + if _, err := instancesettings.NewService(store).Put(ctx, patch, nil); err != nil { + t.Fatalf("put: %v", err) + } + + // A reader that has never cached, as the consumer process is on the next + // poll after an edit. + tc := &TrackingConsumer{} + tc.WireTrackingPolicy(instancesettings.NewService(store)) + + windows := tc.machineWindows(ctx) + if got, want := windows.OpenWindow(), time.Duration(widened)*time.Second; got != want { + t.Fatalf("open window = %v, want %v", got, want) + } + // The click window was not part of the patch and must keep its default + // rather than being cleared by a partial write. + if got, want := windows.ClickWindow(), time.Duration(config.TrackingMachineWindowClickSecondsDefault)*time.Second; got != want { + t.Fatalf("click window = %v, want the untouched default %v", got, want) + } + + sent := time.Now() + chrome := strp("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36") + at := sent.Add(4 * time.Minute) + + // Four minutes is well past the shipped 60s and inside the saved 300s, so + // this only passes if the stored value is what the classifier used. + if m, r := classifyOpen(chrome, nil, &sent, at, windows.OpenWindow()); !m || r != repository.EmailOpenReasonInstant { + t.Fatalf("an open inside the saved window is automated, got %v %q", m, r) + } + if m, _ := classifyOpen(chrome, nil, &sent, at, instancesettings.DefaultTracking().OpenWindow()); m { + t.Fatal("the same open is a person's under the shipped window; the test proves nothing otherwise") + } +} + +// An out-of-range value must be clamped on the way in, not stored and applied. +// Normalize runs on write and on read, so a hand-edited row is bounded too. +func TestLiveTrackingWindowClampsThroughStorage(t *testing.T) { + handle := liveDB(t) + ctx := context.Background() + store := instancesettings.NewStore(handle.Pool) + t.Cleanup(func() { _ = store.Put(ctx, instancesettings.Defaults(), nil) }) + + doc := instancesettings.Defaults() + doc.Tracking.MachineWindowOpenSeconds = config.TrackingMachineWindowSecondsMax + 10_000 + doc.Tracking.MachineWindowClickSeconds = -5 + if err := store.Put(ctx, doc, nil); err != nil { + t.Fatalf("put: %v", err) + } + + got, err := store.Get(ctx) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.Tracking.MachineWindowOpenSeconds != config.TrackingMachineWindowSecondsMax { + t.Errorf("open window = %d, want it clamped to %d", + got.Tracking.MachineWindowOpenSeconds, config.TrackingMachineWindowSecondsMax) + } + if got.Tracking.MachineWindowClickSeconds != config.TrackingMachineWindowClickSecondsDefault { + t.Errorf("click window = %d, want a negative to resolve to the default %d", + got.Tracking.MachineWindowClickSeconds, config.TrackingMachineWindowClickSecondsDefault) + } +} diff --git a/internal/app/instancesettings/service.go b/internal/app/instancesettings/service.go index 17582751..b3cd0023 100644 --- a/internal/app/instancesettings/service.go +++ b/internal/app/instancesettings/service.go @@ -46,7 +46,7 @@ type Service interface { RetentionWindows(ctx context.Context) Retention // TrackingPolicy is the engagement-classification section, already // normalized. The tracking consumer reads it per event, so an edit takes - // effect on the next open or click rather than at the next restart. + // effect without a restart, within the cacheTTL the read goes through. TrackingPolicy(ctx context.Context) Tracking // DomainAuth is the sending-domain authentication gate: whether it is // enforced at all, and how long a domain must stay failing first. diff --git a/tracking/src/scanners.rs b/tracking/src/scanners.rs index 43f5ce2e..b2348f69 100644 --- a/tracking/src/scanners.rs +++ b/tracking/src/scanners.rs @@ -282,6 +282,32 @@ mod tests { assert_eq!(s.classify("13.107.128.5", &h, true, Request::Open), None); } + // Every CIDR in the shipped catalogue must already be its own network + // address. `insert` calls `trunc()`, so `209.222.82.9/24` would silently + // become `209.222.82.0/24` and a typo in the host part of a block would + // widen or shift it with nothing to show for it. + #[test] + fn shipped_networks_are_written_in_canonical_form() { + for line in BUILTIN_CATALOGUE.lines() { + let line = line.split('#').next().unwrap_or("").trim(); + let Some(source) = line.split_whitespace().next() else { + continue; + }; + if source.starts_with("asn:") { + continue; + } + let net: IpNet = source + .parse() + .unwrap_or_else(|_| panic!("{source} must parse")); + assert_eq!( + net, + net.trunc(), + "{source} has host bits set; write it as {}", + net.trunc() + ); + } + } + // Barracuda's published filtering blocks are narrow, per-region and // documented by the vendor as its own mail tier, so they ship enabled on // both endpoints like the EOP ranges.