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.

[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.

[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   88.03%   88.04%            
========================================
  Files       10337    10337            
  Lines      596922   597052   +130     
  Branches    23196    23196            
========================================
+ Hits       525525   525647   +122     
- Misses      70938    70946     +8     
  Partials      459      459