Reliable event filtering

The engineering question

A record update can change a customer’s plan, a synchronization timestamp or both. Forwarding every update creates unnecessary downstream work. Filtering too aggressively can lose changes a consumer needs.

I built this example to make the decision explicit and testable. It compares complete before-and-after records and only suppresses changes the policy can confidently classify as irrelevant.

The policy

• Forward changes to business fields, including mixed business and metadata updates.
• Skip unchanged records and changes limited to explicitly excluded metadata fields.
• Forward unfamiliar field changes or records with no previous snapshot.
• Skip an identical retry. Raise an error when an event ID is reused with different content.

The policy returns both an action and a reason, so a skipped update can still be accounted for.

Measured results

Where the filter runs

The consumer receives the event, checks its identity, compares the records and decides whether to forward it. Each decision records a reason.

Tests and production limits

The project has 18 behavioral tests covering business changes, mixed updates, unknown fields, nulls, deleted fields, retries, identity conflicts and failed downstream calls.

The event registry is in memory and processes events serially. Restarting the process loses its history. A crash after a downstream action can still cause a repeat. This is not an exactly-once production system.

Production use would need durable identity tracking, downstream idempotency, concurrency handling, a quarantine path and explicit replay rules. The included runbook describes those gaps and an operator handoff.

Run it locally

python3 -m unittest -v
python3 demo.py

Python 3.9 or newer. No external packages or credentials required.