ref(aci): use enqueue time to query Snuba in delayed workflow

[cathteng]

For a slow alert condition (e.g. "> 10 events in the past 1 minute"), we always use the time the query is being made as the time to start looking backwards from. This is not optimal in the case where:

• A task needs to be rerun
• There is a backlog in the delayed workflow queue
• Because we flush the buffer every minute, here is an edge case where the time period is 1 minute: events come in at 11:00:02 am, are dispatched in a task at ~11:01:00am, the query is run at 11:01:03am or later, which would mean those events aren't picked up in the result

A better solution would be to use the time the event is enqueued into the delayed workflow buffer!

However, we also batch queries such that all alerts that end up making the same queries are grouped together, and we run a single query for multiple groups. For example:

• Two events for two different groups come in a different times within the same minute, are both processed by alert A
• We will make a single Snuba query for group A with the two groups

In the above case, we'll need to decide which enqueue time to use. We should use the latest enqueue time out of a list of groups as it will cover all groups being queried for in the Snuba query.

[sentry-io]

🔍 Existing Issues For Review

Your pull request is modifying functions with the following pre-existing issues:

📄 File: src/sentry/workflow_engine/processors/workflow.py

Function	Unhandled Issue
evaluate_workflows_action_filters	SoftTimeLimitExceeded: SoftTimeLimitExceeded() se... Event Count: 3

---

_{Did you find this useful? React with a 👍 or 👎}

[kcons]

if timestamp is not None:
self.timestamp = timestamp if self.timestamp is None else max(timestamp, self.timestamp)

perhaps.

[cathteng]

sgtm

[kcons]

Is BulkQueryParameters is more accurate name? Or perhaps GroupQueryParameters? There's a distinction between the unique queries and what we're associating them with that I'm not sure I'm capturing accurately, but TimeAndGroups strikes me as a bit too literal.

[kcons]

I feel like I should know what this means, but I don't.

[kcons]

another option is to make groups be a dict[GroupId, datetime | None] and
do time = max(ts for ts in groups.values() if ts, default=current_time).
Stores a bit more data, but lets the summarizing happen where it is being forced, which has a certain appeal.

[cathteng]

i need to refactor dcg_to_timestamp for this, if it comes to it we can do a refactor

[kcons]

worth explaining what this timestamp is and what it should correspond to. Or, rename the field to make the comment pointless.

[kcons]

TIL we can dumps a datetime.

[kcons]

You should probably add a test that requires us to parse this value correctly from the expected format. I strongly suspect we don't.

[kcons]

Need test to verify we can parse it and pydantic won't freak out.

[codecov]

Codecov Report

Attention: Patch coverage is 86.11111% with 10 lines in your changes missing coverage. Please review.

✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...try/workflow_engine/processors/delayed_workflow.py	65.51%	10 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           master   #93882       +/-   ##
===========================================
+ Coverage   37.94%   81.96%   +44.01%     
===========================================
  Files        9784    10351      +567     
  Lines      553763   598883    +45120     
  Branches    23268    23267        -1     
===========================================
+ Hits       210143   490884   +280741     
+ Misses     343151   106950   -236201     
- Partials      469     1049      +580     

[kcons]

Seems simpler and safer to have update(self, group_id: GroupId, timestamp: datetime | None).

[kcons]

maybe

# Used to pick the end of the time window in snuba querying. 
# Should be close to when fast conditions were evaluated to try to be consistent.

What you have is fine, though.