feat(GAL): Add idempotency key to publish functions

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,
     )
 
 

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):

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
         )

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: