diff --git a/src/sentry/issues/action_log/publish.py b/src/sentry/issues/action_log/publish.py index 9ac34386436f..3e4bb0dfe558 100644 --- a/src/sentry/issues/action_log/publish.py +++ b/src/sentry/issues/action_log/publish.py @@ -10,7 +10,7 @@ from contextlib import contextmanager from contextvars import ContextVar from dataclasses import dataclass -from typing import TYPE_CHECKING, Sequence +from typing import TYPE_CHECKING, Optional, Sequence from sentry.hybridcloud.models.outbox import outbox_context from sentry.issues.action_log.types import ( @@ -80,6 +80,7 @@ def publish_action( project: Project, actor: GroupActionActor = SYSTEM_ACTOR, force_async_derived: bool = False, + idempotency_key: str | None = None, ) -> None: """ Record an issue action. @@ -91,6 +92,9 @@ def publish_action( If *force_async_derived* is True, derived data processing is deferred entirely to the async task. Useful for latency-sensitive paths. + If *idempotency_key* is set, the GroupActionLogEntry is created if and only if there + does not already exist a GALE with that group id & idempotency key; else it's a no-op. + Log publishing is managed by an outbox that flushes on commit by default. Wrap in ``outbox_context(flush=False)`` to defer the drain. """ @@ -145,6 +149,9 @@ def publish_action( "force_async_derived": force_async_derived, } + if idempotency_key is not None: + payload["idempotency_key"] = idempotency_key + outbox = CellOutbox( shard_scope=OutboxScope.GROUP_SCOPE, shard_identifier=group_id, @@ -163,6 +170,7 @@ def publish_action_from_context( group_id: int, project: Project, force_async_derived: bool = False, + idempotency_key: Optional[str] = None, ) -> None: """ Record an issue action using the current ActionContext. This is the primary API @@ -188,6 +196,7 @@ def publish_action_from_context( project=project, actor=actor, force_async_derived=force_async_derived, + idempotency_key=idempotency_key, ) diff --git a/src/sentry/issues/action_log/types.py b/src/sentry/issues/action_log/types.py index b58b286e38bf..2b848f0735b4 100644 --- a/src/sentry/issues/action_log/types.py +++ b/src/sentry/issues/action_log/types.py @@ -7,7 +7,7 @@ import abc import dataclasses from enum import IntEnum, StrEnum -from typing import Any, Literal, Optional, TypedDict +from typing import Any, Literal, NotRequired, Optional, TypedDict from pydantic import BaseModel @@ -430,6 +430,7 @@ class GroupActionLogPayload(TypedDict): source: str data: dict[str, Any] force_async_derived: bool + idempotency_key: NotRequired[str] class SetPublicAction(GroupAction): diff --git a/src/sentry/receivers/outbox/cell.py b/src/sentry/receivers/outbox/cell.py index 1bc234fe7d5f..66081f50199c 100644 --- a/src/sentry/receivers/outbox/cell.py +++ b/src/sentry/receivers/outbox/cell.py @@ -12,7 +12,7 @@ import logging from typing import Any, assert_never, cast -from django.db import router, transaction +from django.db import IntegrityError, router, transaction from django.dispatch import receiver from sentry.audit_log.services.log import AuditLogEvent, UserIpEvent, log_rpc_service @@ -346,24 +346,31 @@ def process_group_action_log_event(payload: GroupActionLogPayload, **kwds: Any) """Write a GroupActionLogEntry from the outbox payload, then trigger derived data processing.""" try: + using = router.db_for_write(GroupActionLogEntry) + group_id = payload["group_id"] force_async_derived = payload["force_async_derived"] - GroupActionLogEntry.objects.create( - group_id=group_id, - project_id=payload["project_id"], - type=payload["type"], - actor_type=payload["actor_type"], - actor_id=payload["actor_id"], - source=payload["source"], - data=payload["data"], - ) + try: + with transaction.atomic(using=using): + GroupActionLogEntry.objects.create( + group_id=group_id, + project_id=payload["project_id"], + type=payload["type"], + actor_type=payload["actor_type"], + actor_id=payload["actor_id"], + source=payload["source"], + data=payload["data"], + idempotency_key=payload.get("idempotency_key"), + ) + except IntegrityError: + # Idempotency conflict; we treat this as a no-op. + pass # This receiver runs inside the outbox drain transaction # (process_shard → transaction.atomic), so the GALE is not yet committed. # Defer to on_commit so the GALE is visible to readers on other connections. strategy = ProcessingStrategy.ASYNC if force_async_derived else ProcessingStrategy.INLINE - using = router.db_for_write(GroupActionLogEntry) transaction.on_commit( lambda: trigger_group_log_processing(group_id, strategy=strategy), using=using ) diff --git a/tests/sentry/issues/test_action_log.py b/tests/sentry/issues/test_action_log.py index 2ad9dbee9256..69d8dc5795d4 100644 --- a/tests/sentry/issues/test_action_log.py +++ b/tests/sentry/issues/test_action_log.py @@ -698,6 +698,35 @@ def test_inline_derived_processes_without_task(self, mock_task: MagicMock) -> No # No async task needed (single entry = caught up) mock_task.delay.assert_not_called() + def test_idempotency_key_publish(self) -> None: + idempotency_key = "test_idempotency_key_publish" + with self.feature("projects:issue-action-log-write-to-db"), outbox_runner(): + publish_action( + ViewAction(), + source=ActionSource.API, + group_id=self.group.id, + project=self.group.project, + actor=GroupActionActor.user(self.user.id), + idempotency_key=idempotency_key, + ) + + entry = GroupActionLogEntry.objects.get(group_id=self.group.id) + assert entry.idempotency_key == idempotency_key + + with ( + self.feature("projects:issue-action-log-write-to-db"), + outbox_runner(), + ): + # Tacitly assert silent failure / no exception + publish_action( + ViewAction(), + source=ActionSource.API, + group_id=self.group.id, + project=self.group.project, + actor=GroupActionActor.user(self.user.id), + idempotency_key=idempotency_key, + ) + class TestCaptureActionLog(TestCase): def _publish(self, action: GroupAction, **kwargs: Any) -> None: