feat(aci): Delayed workflows cohort sharding

[kcons]

Delayed workflows tasks are currently all scheduled every minute.
This frequency (60s) is what we want, but running them all at once makes our load bursty, and causes some rate limits to be frequently hit in Snuba querying.

This change establishes "cohorts" and schedules one cohort at a time, allowing us to smoothly transition to N groups of delayed_workflow tasks per minute, one every 1/Nth of a minute, smoothing out our workload.

---

Note

Introduce cohort-sharded scheduling for delayed workflows, persisting cohort run timestamps in Redis and refactoring processing/cleanup; add pydantic-based Redis helpers and a new num_cohorts option.

• Workflow Engine (Scheduling & Processing)
  • Introduce cohort-sharded scheduling via ProjectChooser and chosen_projects, selecting projects per run based on workflow_engine.num_cohorts.
  • Persist cohort run timestamps in Redis as CohortUpdates using DelayedWorkflowClient.fetch_updates/persist_updates.
  • Refactor cleanup into mark_projects_processed; compute max_project_timestamp and use conditional delete or range clear accordingly.
• Buffer Layer
  • Add RedisHashSortedSetBuffer.get_parsed_key/put_parsed_key for pydantic models; extend supported ops and expiry handling.
  • Extend DelayedWorkflowClient with _COHORT_UPDATES_KEY and cohort update helpers.
• Options
  • Add workflow_engine.num_cohorts (Int, default 1).
• Tests
  • Convert to pytest-style and add coverage for cohort selection, Redis parsed key helpers, and new scheduling/cleanup flows.

^{Written by Cursor Bugbot for commit 87a09d0. This will update automatically on new commits. Configure here.}

[kcons]

TODO: Make cohort size an option, spell out the transition, use precise cleanup, add a test for num_cohorts = 1.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

Additional details and impacted files

@@             Coverage Diff             @@
##           master   #100038      +/-   ##
===========================================
+ Coverage   78.80%    81.03%   +2.22%     
===========================================
  Files        8699      8701       +2     
  Lines      385940    386024      +84     
  Branches    24413     24413              
===========================================
+ Hits       304159    312799    +8640     
+ Misses      81430     72874    -8556     
  Partials      351       351              

[kcons]

Bug: Conditional Delete Fails, Incorrectly Clears Projects

In mark_projects_processed, the fallback path for conditional deletes calculates max_project_timestamp from all retrieved projects, not just the ones actually processed. This means if the conditional delete fails, clear_project_ids may incorrectly remove unprocessed projects from the buffer.

src/sentry/workflow_engine/processors/schedule.py#L205-L209

sentry/src/sentry/workflow_engine/processors/schedule.py

Lines 205 to 209 in 54c7d82

	)
	# Fallback.
	buffer_client.clear_project_ids(
	min=0,
	max=max_project_timestamp,

 

This is true, but an artifact of the use_conditional_delete being required for partial clean-up. Next week the fallback will be removed.

[cathteng]

This is cool!