Issue platform may process messages out of order

The crons team has an issue with how status updates and occurrences are consumed. Here's the summary

When a monitor fails we create a `MonitorIncident` for that monitor. There can only be one monitor incident at a time, so when the incident recovers it is marked as complete. When an incident starts we also **produce an event occurance** in to the issue platform consumer. When the incident closes **we produce a `STATUS_CHANGE`** message to the incident consumer.

We're seeing some of our monitors with **issues still open after the incident has been resolved**. This seems to imply there's an issue somewhere in the issue status update logic.

I've narrowed it down and what I think is happening is that **status updates are being consumed before occurrences that create issue groups**.

The fix here is going to be to partition messages on the issue platform topic by group hash and then in the consumer itself guarantee that events with the same group hashes are processed in order.

---

Here's what I found

We added logging to `mark_ok`'s resolve here: https://github.com/getsentry/sentry/pull/67537

Looking at a monitor that has multiple unresolved issues ([`deliver-from-outbox`](https://sentry.sentry.io/crons/sentry/deliver-from-outbox/?environment=prod)) we can look at its logs as well as it's incident history (`monitor_id` 109268). Here are it's 10 most recent incidents:

```sql
select
to_char(starting_timestamp, 'YYYY-MM-DD HH12:MI:SS TZ') incident_started,
to_char(resolving_timestamp, 'YYYY-MM-DD HH12:MI:SS TZ') resolve_at,
grouphash incident_grouphash,
md5(grouphash) group_fingerprint
from
sentry_monitorincident
where
monitor_environment_id = 109268
and starting_timestamp <= DATE('2024-04-02 00:00')
order by
id desc
limit
10
```

|incident_started |resolve_at |incident_grouphash |group_fingerprint|
|----------------|----------------|--------------------------------|-----------------|
|2024-04-01 11:41:00 UTC|2024-04-01 11:43:00 UTC|440801ecba3a468c84c9724b8ef4493f|f058886fef6966d892d174638388dad4|
|2024-04-01 11:08:00 UTC|2024-04-01 11:10:00 UTC|bf00428880ed434ba7841b9b7318d587|2dbbfd0dc44c795e60cee8d86c2b190f|
|2024-04-01 10:37:00 UTC|2024-04-01 10:39:00 UTC|14377a4347ca43a0817da4cf3e206ce2|2912a8773d7008f2a51c9917cc5d092d|
|2024-04-01 10:04:00 UTC|2024-04-01 10:06:00 UTC|d39e4ea8ad974b288a414760d0e00e02|68f24ca6735005adbbe6fc295c134cfd|
|2024-04-01 08:58:00 UTC|2024-04-01 08:59:59 UTC|50d3d994ab5643a095bea7612d49107b|f280018d3f1eb269d60266eae0286918|
|2024-04-01 06:39:00 UTC|2024-04-01 06:41:00 UTC|095c9f8d1cfc46cbb60cd5a638063036|cca5c41216dfcbcdf949841eb7f75d05|
|2024-04-01 01:55:00 UTC|2024-04-01 01:54:00 UTC|2cb7abc8ba5346ed8952f3b96e3e08d1|c33d9c935e1f43029bd0112b6f2552be|
|2024-04-01 12:10:00 UTC|2024-04-01 12:11:00 UTC|9ae45c29039d42238f6cd7280583cf46|261298fce8a0ae06e119ce06d4a4b34e|
|2024-04-01 07:48:00 UTC|2024-04-01 07:48:00 UTC|44f1c019ef414d19a6fe76961b692a16|6e82eaf071116070a2583146d08f7f30|
|2024-03-30 01:30:00 UTC|2024-03-30 01:31:00 UTC|1b2bff2c029f4d369bffef551e46a2d1|7fbf807237043ac23ac6dd5c8cd0ddeb|

These incidents are all currently resolved. Let's take a look at the associated issues

> [!NOTE]
> Note that the `incident_grouphash` is different from the `group_fingerprint`. The fingerprint is computed from the md5 of the incident group hash.

```sql
select
to_char(first_seen, 'YYYY-MM-DD HH12:MI:SS TZ') first_seen,
hash,
status
from
sentry_grouphash sgh
inner join sentry_groupedmessage sgm on sgm.id = sgh.group_id
where
sgh.hash in (
'f058886fef6966d892d174638388dad4',
'2dbbfd0dc44c795e60cee8d86c2b190f',
'2912a8773d7008f2a51c9917cc5d092d',
'68f24ca6735005adbbe6fc295c134cfd',
'f280018d3f1eb269d60266eae0286918',
'cca5c41216dfcbcdf949841eb7f75d05',
'c33d9c935e1f43029bd0112b6f2552be',
'261298fce8a0ae06e119ce06d4a4b34e',
'6e82eaf071116070a2583146d08f7f30',
'7fbf807237043ac23ac6dd5c8cd0ddeb'
)
and sgh.project_id = 1
order by sgh.id desc
```

|first_seen |hash |status |
|----------------|----------------|--------------------------------|
|2024-04-01 11:42:07 UTC|f058886fef6966d892d174638388dad4|1 |
|2024-04-01 11:09:06 UTC|2dbbfd0dc44c795e60cee8d86c2b190f|1 |
|2024-04-01 10:38:05 UTC|2912a8773d7008f2a51c9917cc5d092d|1 |
|2024-04-01 10:05:08 UTC|68f24ca6735005adbbe6fc295c134cfd|1 |
|2024-04-01 08:59:05 UTC|f280018d3f1eb269d60266eae0286918|1 |
|2024-04-01 06:40:13 UTC|cca5c41216dfcbcdf949841eb7f75d05|1 |
|2024-04-01 01:54:03 UTC|c33d9c935e1f43029bd0112b6f2552be|0 |
|2024-04-01 12:11:01 UTC|261298fce8a0ae06e119ce06d4a4b34e|0 |
|2024-04-01 07:47:03 UTC|6e82eaf071116070a2583146d08f7f30|1 |
|2024-03-30 01:31:02 UTC|7fbf807237043ac23ac6dd5c8cd0ddeb|0 |

As we can see here **indeed some of these groups are not resolved**. Let's look at the logs for the `monitors.logic.mark_ok.resolving_incident` log we added and make sure it was called for all of these incident groups.

```
resource.type="k8s_container"
labels.name=~"sentry"
jsonPayload.event="monitors.logic.mark_ok.resolving_incident"
jsonPayload.monitor_env_id=109268
jsonPayload.grouphash=(
"440801ecba3a468c84c9724b8ef4493f" OR
"bf00428880ed434ba7841b9b7318d587" OR
"14377a4347ca43a0817da4cf3e206ce2" OR
"d39e4ea8ad974b288a414760d0e00e02" OR
"50d3d994ab5643a095bea7612d49107b" OR
"095c9f8d1cfc46cbb60cd5a638063036" OR
"2cb7abc8ba5346ed8952f3b96e3e08d1" OR
"9ae45c29039d42238f6cd7280583cf46" OR
"44f1c019ef414d19a6fe76961b692a16" OR
"1b2bff2c029f4d369bffef551e46a2d1"
)
```

|insertId |jsonPayload.event|jsonPayload.grouphash |jsonPayload.incident_id|jsonPayload.level|jsonPayload.monitor_env_id|jsonPayload.name |labels."compute.googleapis.com/resource_name" |labels."k8s-pod/app_feature"|labels."k8s-pod/app_function"|labels."k8s-pod/cogs_category"|labels."k8s-pod/component"|labels."k8s-pod/consumer"|labels."k8s-pod/env"|labels."k8s-pod/environment"|labels."k8s-pod/pod-template-hash"|labels."k8s-pod/security_istio_io/tlsMode"|labels."k8s-pod/service"|labels."k8s-pod/service_istio_io/canonical-name" |labels."k8s-pod/service_istio_io/canonical-revision"|labels."k8s-pod/system"|labels.name |logName |receiveLocation|receiveTimestamp |resource.labels.cluster_name|resource.labels.container_name|resource.labels.location|resource.labels.namespace_name|resource.labels.pod_name |resource.labels.project_id|resource.type|severity|timestamp |
|----------------|-----------------|--------------------------------|-----------------------|-----------------|--------------------------|-----------------------------|--------------------------------------------------|----------------------------|-----------------------------|------------------------------|--------------------------|-------------------------|--------------------|----------------------------|----------------------------------|------------------------------------------|------------------------|----------------------------------------------------|----------------------------------------------------|-----------------------|-----------------------------|------------------------------------|---------------|------------------------------|----------------------------|------------------------------|------------------------|------------------------------|---------------------------------------------------------------|--------------------------|-------------|--------|------------------------------|
|apirukxo2ptaowog|monitors.logic.mark_ok.resolving_incident|440801ecba3a468c84c9724b8ef4493f|4342919 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-vv4h|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |5484579cd7 |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T23:43:01.959416238Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-5484579cd7-mlsww |internal-sentry |k8s_container|INFO |2024-04-01T23:43:01.828589951Z|
|ydbt1xe8zjh7ctg8|monitors.logic.mark_ok.resolving_incident|bf00428880ed434ba7841b9b7318d587|4342423 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-b2jm|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |78ddccc5bf |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T23:10:08.342843949Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-78ddccc5bf-x6d5x |internal-sentry |k8s_container|INFO |2024-04-01T23:10:04.392472864Z|
|m848xkkqqvo3qp91|monitors.logic.mark_ok.resolving_incident|14377a4347ca43a0817da4cf3e206ce2|4341921 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-s48c|monitors |ingest |monitors |consumer |ingest-monitors |canary |production |7f45b6cc84 |istio |getsentry |getsentry-consumer-ingest-monitors-production-canary|latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T22:39:05.859960032Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-canary-7f45bm4c7s|internal-sentry |k8s_container|INFO |2024-04-01T22:39:02.332129044Z|
|f3kmblzz7y5qojgo|monitors.logic.mark_ok.resolving_incident|d39e4ea8ad974b288a414760d0e00e02|4341440 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-7lz7|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |6464895765 |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T22:06:04.625510187Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-6464895765-szcjd |internal-sentry |k8s_container|INFO |2024-04-01T22:06:02.073527142Z|
|sosoqx5ij1vxkx0k|monitors.logic.mark_ok.resolving_incident|50d3d994ab5643a095bea7612d49107b|4340282 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-p5gn|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |776c74cc9c |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T21:00:10.370587522Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-776c74cc9c-2mpqf |internal-sentry |k8s_container|INFO |2024-04-01T21:00:06.098563171Z|
|vi5h8dk2u1pqu17n|monitors.logic.mark_ok.resolving_incident|095c9f8d1cfc46cbb60cd5a638063036|4338090 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-nb4z|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |5bbb95c499 |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T18:41:06.017389159Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-5bbb95c499-7xxlg |internal-sentry |k8s_container|INFO |2024-04-01T18:41:05.078871391Z|
|s63477o9tmhimeht|monitors.logic.mark_ok.resolving_incident|2cb7abc8ba5346ed8952f3b96e3e08d1|4333726 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-qcqp|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |7f686b888d |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T13:54:07.153368920Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-7f686b888d-mtv5p |internal-sentry |k8s_container|INFO |2024-04-01T13:54:03.207864131Z|
|vr9syvq2bgw6xxbw|monitors.logic.mark_ok.resolving_incident|9ae45c29039d42238f6cd7280583cf46|4332234 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-qcqp|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |7f686b888d |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T12:11:07.139926168Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-7f686b888d-mtv5p |internal-sentry |k8s_container|INFO |2024-04-01T12:11:03.393002484Z|
|dqied6qq00ay8q57|monitors.logic.mark_ok.resolving_incident|44f1c019ef414d19a6fe76961b692a16|4328003 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-qcqp|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |7f686b888d |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-04-01T07:48:07.159065054Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-7f686b888d-mtv5p |internal-sentry |k8s_container|INFO |2024-04-01T07:48:02.989845744Z|
|e9zmp0f3w0u4qe1p|monitors.logic.mark_ok.resolving_incident|1b2bff2c029f4d369bffef551e46a2d1|4278867 |info |109268 |sentry.monitors.logic.mark_ok|gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-qcqp|monitors |ingest |monitors |consumer |ingest-monitors |primary |production |7f686b888d |istio |getsentry |getsentry-consumer-ingest-monitors-production |latest |kafka_consumer |sentry.monitors.logic.mark_ok|projects/internal-sentry/logs/stdout| |2024-03-30T01:31:07.177017261Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-monitors-production-7f686b888d-mtv5p |internal-sentry |k8s_container|INFO |2024-03-30T01:31:03.388075774Z|

**Yes, we attempted to resolve each of these issues when the incidents were closed**. You can see the code for this here https://github.com/getsentry/sentry/blob/5cdf1ec2e0e2c9ebcae6a3398e7050ad9f9715a5/src/sentry/monitors/logic/mark_ok.py#L49-L61

So we're producing a `STATUS_CHNAGE` into the issue consumer. We log when the group is not found. Let's look for those logs for the groups that **are not resolved**. ([that logging happens here](https://github.com/getsentry/sentry/blob/5cdf1ec2e0e2c9ebcae6a3398e7050ad9f9715a5/src/sentry/issues/status_change_consumer.py#L147-L157))

Querying for just the fingerprints where the issue is unresolved

```
resource.type="k8s_container"
labels.name=~"sentry"
jsonPayload.event="grouphash.not_found"
jsonPayload.name="sentry.issues.status_change_consumer"
jsonPayload.project_id=1
jsonPayload.fingerprint=(
"c33d9c935e1f43029bd0112b6f2552be" OR
"261298fce8a0ae06e119ce06d4a4b34e" OR
"7fbf807237043ac23ac6dd5c8cd0ddeb"
)
```

|errorGroups.id |insertId |jsonPayload.event |jsonPayload.exception|jsonPayload.fingerprint |jsonPayload.level|jsonPayload.name |jsonPayload.project_id|labels."compute.googleapis.com/resource_name" |labels."k8s-pod/app_feature"|labels."k8s-pod/app_function"|labels."k8s-pod/cogs_category"|labels."k8s-pod/component"|labels."k8s-pod/consumer"|labels."k8s-pod/env"|labels."k8s-pod/environment"|labels."k8s-pod/pod-template-hash"|labels."k8s-pod/security_istio_io/tlsMode"|labels."k8s-pod/service"|labels."k8s-pod/service_istio_io/canonical-name"|labels."k8s-pod/service_istio_io/canonical-revision"|labels."k8s-pod/system"|labels.name |logName |receiveLocation|receiveTimestamp |resource.labels.cluster_name|resource.labels.container_name|resource.labels.location|resource.labels.namespace_name|resource.labels.pod_name |resource.labels.project_id|resource.type|severity|timestamp |
|----------------|----------------|--------------------------------|---------------------|--------------------------------|-----------------|------------------------------------|----------------------|--------------------------------------------------|----------------------------|-----------------------------|------------------------------|--------------------------|-------------------------|--------------------|----------------------------|----------------------------------|------------------------------------------|------------------------|------------------------------------------------|----------------------------------------------------|-----------------------|------------------------------------|------------------------------------|---------------|------------------------------|----------------------------|------------------------------|------------------------|------------------------------|---------------------------------------------------------------|--------------------------|-------------|--------|------------------------------|
|["COPs4_e2hKaFzgE"]|2b5swelnden0bkkj|grouphash.not_found |Traceback (most recent call last): File "/usr/src/sentry/src/sentry/issues/status_change_consumer.py", line 149, in bulk_get_groups_from_fingerprints raise Exception("grouphash.not_found") Exception: grouphash.not_found|c33d9c935e1f43029bd0112b6f2552be|error |sentry.issues.status_change_consumer|1 |gke-zdpwkxst-c2-standard-30-f715d1d4-f892a95b-b2jm|issueplatform |ingest |issueplatform |consumer |ingest-occurrences |primary |production |65fb4948d |istio |getsentry |getsentry-consumer-ingest-occurrences-production|latest |kafka_consumer |sentry.issues.status_change_consumer|projects/internal-sentry/logs/stdout| |2024-04-01T13:54:13.297237614Z|zdpwkxst |sentry |us-central1-b |default |getsentry-consumer-ingest-occurrences-production-65fb4948ddc65d|internal-sentry |k8s_container|ERROR |2024-04-01T13:54:11.466503844Z|

> [!WARNING]
> It's not clear to me why we're missing two other log messages. Is the logging here not guaranteed?