feature: Added Filtering and Rate Limiting#1073
Draft
hollyschilling-cio wants to merge 21 commits into
Draft
Conversation
added 9 commits
May 28, 2026 16:42
Sample app builds 📱Below you will find the list of the latest versions of the sample apps. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #1073 +/- ##
==========================================
+ Coverage 70.47% 70.50% +0.03%
==========================================
Files 224 231 +7
Lines 11542 11803 +261
==========================================
+ Hits 8134 8322 +188
- Misses 3408 3481 +73 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
SDK binary size reports 📊SDK binary size of this PRSDK binary size diff report vs. main branch |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 43b4501. Configure here.
| // Integration key was present but carried no rules — server explicitly removed them. | ||
| // Delete the cached config so stale rules aren't reloaded on next launch. | ||
| try? storage?.deleteAggregationConfig() | ||
| } |
There was a problem hiding this comment.
Incorrect else-if branch deletes valid cached config
Low Severity
The else if config != nil branch is intended to handle the case where the integration key is present but carries no rules (i.e., ruleset is nil). However, this branch also executes when ruleset is non-nil but JSONEncoder().encode(ruleset) or String(data:encoding:) fails. In that scenario, the engine already loaded the valid ruleset into memory via engine.load(ruleset:), but the fallthrough into else if config != nil deletes the previously cached config from the database. On next app restart, no cached rules would be available until a settings refresh arrives from the CDN. The condition could be tightened to else if config != nil, ruleset == nil to only delete when rules are genuinely absent.
Reviewed by Cursor Bugbot for commit 43b4501. Configure here.
matt-frizzell
left a comment
matt-frizzell
left a comment
Contributor
There was a problem hiding this comment.
I think adding the database layer should/could be its own reviewable change, separate from the introduction of filtering. It would cut down the lines of code to read and keep the focus on reviewing and testing one thing at a time.
mahmoud-elmorabea
left a comment
mahmoud-elmorabea
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the draft! 👍
I have two main points here:
- To me, the overhead of the DB: creation, tables, migrations and the extra dependency isn't justified by the data being stored here. They're mostly key/value pairs and group of rules that would be saved/loaded and replaced in their entirety. We can still encrypt it if we want even without the database
- I'd like us to better link this to the issues we've had reported, run a kind of simulation. "For that particular issue we'd apply that rule and it would have that effect" type of thing. A question that could answer is, should these rules apply to only internal events or also customer events? Do we need these rules to be applied server side also or client side is enough?
| let rows = try db.query( | ||
| Select(col("payload"), col("fetched_at")) | ||
| .from("aggregation_rules") | ||
| .where(col("id") == 1) |
Contributor
There was a problem hiding this comment.
I think the fact that we are using a single hard-coded row id might signal that we don't need a database for this?
We might also be getting ahead of ourselves by creating a database for the rules, I would imagine there would be a handful of rules per workspace and that they wouldn't change all that often.
Contributor
Author
There was a problem hiding this comment.
You're right that this specific data doesn't need to be in the database. This is also not the reason to have a database. The rate-limiting history does need to be in a database because that may be updated multiple times per second and rewriting an entire JSON file each write is unreasonable.
Even within the database, there is a Key-Value table that could be used. The reason I put it into a table to start is if we decide in the future to begin doing diffing or keeping previous versions for fallback, which will require additional rows.
| return false | ||
| } | ||
|
|
||
| if let rl = rs.rateLimitsByKey[exactKey] ?? rs.rateLimitsByKey[wildcardKey] { |
Contributor
There was a problem hiding this comment.
Does rate limiting wildcardKey something we'd want to do? It would apply a time based rate limit on all track events including ones sent by customers?
| /// `@unchecked Sendable` is safe here: all mutable state (`processed`) is | ||
| /// protected by `Synchronized<>`, and `StorageManager.db` is serialized | ||
| /// internally via a private DispatchQueue. | ||
| final class EventPolicyEngine: Sendable { |
Contributor
There was a problem hiding this comment.
I'm finding myself reverse understanding how this engine would work, it would be very helpful if we can create a spec that describes the intended behavior
Contributor
There was a problem hiding this comment.
Given the issues we've seen, what makes sense to me is something like:
- Filtering (exactly as you describe): Either filter a specific event or all track/screen events
- Rate limiting an exact event (specific track or screen view): only send once per certain time frame
- De-duplicate: for a specific event if the payload is the same (excluding timestamp), drop that event
Contributor
Author
There was a problem hiding this comment.
The code is pretty clear so I'm not sure what you are not unerstanding.
De-duplication is not in scope for this.
Segment has 5 basic event types: track, screen, identify, alias, and group. We don't use alias or group, and identify should never be filtered or rate limited by either of these mechanisms. That leaves us only concerning ourselves with track and screen. screen events are screen views where the event name is the name of the screen. track events have a name of whatever event name was given at the call site.
Filter rules have a event name and type. When the engine gets asked if an event should be passed on, it uses the name and type to check against a look up table (O(1)) in memory to see if that particular event type and name combination is filtered or rate limited. If the event is neither, it goes on. If it is filtered, the event is discarded immediately. If it is rate limited, we check and set (atomically) the event's database record to see when it was seen last. If it has not been seen within the window, the last seen time is updated and the event is sent on. If it has been seen within the time window, the record is not modified and the event is destroyed.
added 3 commits
June 10, 2026 09:05
# Conflicts: # .swiftformat # Sources/DataPipeline/DataPipelineImplementation.swift
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
MBL-1747
This PR adds Filtering and Rate Limiting through an aggregation engine. The aggregation engine pulls its config from the Segment config pulled from
https://{cdnHost}/projects/{writeKey}/settingsusing values passed asintegrations.aggregationRules.The largest change introduced here is the addition of a dependency to have a local database. This is not yet handled in the Cocoapods, but we will update this PR to accommodate that before merging.
Complete each step to get your pull request merged in. Learn more about the workflow this project uses.
Note
High Risk
Changes which analytics events are sent (server-controlled drops and rate limits) and adds encrypted on-disk persistence keyed by the CDP API key; misconfiguration or storage issues could alter customer data delivery until CocoaPods parity is done.
Overview
Adds server-driven event filtering and rate limiting for Data Pipelines via Segment/CDP settings (
aggregationRuleson the Customer.io Data Pipelines integration). Rules are cached in a new encrypted local SQLCipher database (SyncSqlCipher), registered atDataPipeline.initialize, and applied by anEventPolicyPluginthat can drop track, screen, identify, alias, and group events before they leave the client.StorageManageropens a per–CDP API key DB under Application Support, runs aggregation schema migrations, and stores rules plus rate-limit counters (profile vs device scope).EventPolicyEngineevaluates block lists and time windows (including*wildcards); storage errors fail open so events are not silently dropped. Settings refresh updates the engine and persists or deletes cached rules;clearIdentifyclears profile-scoped rate-limit state only.SPM now depends on
SyncSqlCipherand bumps the Swift tools version to 5.10; CocoaPods wiring for the DB is called out as follow-up in the PR description. Broad unit tests cover decoding, engine, plugin, storage, and identify reset.Reviewed by Cursor Bugbot for commit 43b4501. Bugbot is set up for automated code reviews on this repo. Configure here.