chore(hybridcloud) Remove dynamic default_flush on ApiToken

src/sentry/hybridcloud/outbox/base.py

@@ -226,7 +226,6 @@ class ControlOutboxProducingModel(Model):
 
     default_flush: bool | None = None
     replication_version: int = 1
-    enqueue_after_flush: bool = False
 
     class Meta:
         abstract = True
@@ -239,21 +238,13 @@ def _maybe_prepare_outboxes(self, *, outbox_before_super: bool) -> Generator[Non
             transaction.atomic(router.db_for_write(type(self))),
             flush=self.default_flush,
         ):
-            saved_outboxes = []
             if not outbox_before_super:
                 yield
             for outbox in self.outboxes_for_update():
                 outbox.save()
-                saved_outboxes.append(outbox.id)
             if outbox_before_super:
                 yield
 
-        if not self.default_flush and self.enqueue_after_flush:
-            transaction.on_commit(
-                lambda: self._schedule_async_replication(saved_outboxes),
-                using=router.db_for_write(type(self)),
-            )
-
     def save(self, *args: Any, **kwds: Any) -> None:
         with self._maybe_prepare_outboxes(outbox_before_super=False):
             super().save(*args, **kwds)
@@ -269,24 +260,6 @@ def delete(self, *args: Any, **kwds: Any) -> tuple[int, dict[str, Any]]:
     def outboxes_for_update(self, shard_identifier: int | None = None) -> list[ControlOutboxBase]:
         raise NotImplementedError
 
-    def _schedule_async_replication(self, saved_outboxes: list[int]) -> None:
-        from sentry.hybridcloud.tasks.deliver_from_outbox import drain_outbox_shards_control
-
-        if not saved_outboxes:
-            logger.error(
-                "missing-outboxes.async-replication",
-                extra={
-                    "model": self.__class__.__name__,
-                },
-            )
-            return
-
-        drain_outbox_shards_control.delay(
-            outbox_identifier_low=min(saved_outboxes),
-            outbox_identifier_hi=max(saved_outboxes) + 1,
-            outbox_name="sentry.ControlOutbox",
-        )
-
 
 _CM = TypeVar("_CM", bound=ControlOutboxProducingModel)
 

src/sentry/models/apitoken.py

@@ -185,10 +185,6 @@ class ApiToken(ReplicatedControlModel, HasApiScopes):
     category = OutboxCategory.API_TOKEN_UPDATE
     replication_version = 2
 
-    # Outbox settings
-    enqueue_after_flush = True
-    _default_flush: bool | None = None
-
     # users can generate tokens without being application-bound
     application = FlexibleForeignKey("sentry.ApiApplication", null=True)
     user = FlexibleForeignKey("sentry.User")
@@ -610,21 +606,6 @@ def organization_id(self) -> int | None:
 
         return installation.organization_id
 
-    @property
-    def default_flush(self) -> bool:
-        from sentry import options
-
-        has_async_flush = options.get("api-token-async-flush")
-
-        if self._default_flush is not None:
-            return self._default_flush
-
-        return not has_async_flush
-
-    @default_flush.setter
-    def default_flush(self, value: bool) -> None:
-        self._default_flush = value
-
 
 def is_api_token_auth(auth: object) -> TypeGuard[AuthenticatedToken | ApiToken | ApiTokenReplica]:
     """:returns True when an API token is hitting the API."""

src/sentry/options/defaults.py

@@ -3677,13 +3677,6 @@
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
 
-# Global flag to enable API token async flush
-register(
-    "api-token-async-flush",
-    default=False,
-    type=Bool,
-    flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
-)
 
 # Cells
 

tests/sentry/models/test_apitoken.py

@@ -231,90 +231,6 @@ def test_handle_async_deletion_called(self, mock_delete_replica: mock.MagicMock)
             cell_name=mock.ANY,
         )
 
-    @override_options({"api-token-async-flush": True})
-    def test_outboxes_created_with_default_flush_false(self) -> None:
-        user = self.create_user()
-
-        token = ApiToken.objects.create(user_id=user.id)
-
-        outboxes = ControlOutbox.objects.filter(
-            shard_scope=OutboxScope.API_TOKEN_SCOPE,
-            shard_identifier=token.id,
-            category=OutboxCategory.API_TOKEN_UPDATE,
-            object_identifier=token.id,
-        )
-        assert outboxes.exists()
-        assert outboxes.count() > 0
-
-        with assume_test_silo_mode(SiloMode.CELL):
-            assert not ApiTokenReplica.objects.filter(apitoken_id=token.id).exists()
-
-    @override_options({"api-token-async-flush": True})
-    def test_outboxes_created_on_update_with_async_flush(self) -> None:
-        user = self.create_user()
-
-        with outbox_runner():
-            token = ApiToken.objects.create(user_id=user.id)
-
-        updated_expires_at = timezone.now() + timedelta(days=30)
-        token.update(expires_at=updated_expires_at)
-
-        outboxes = ControlOutbox.objects.filter(
-            shard_scope=OutboxScope.API_TOKEN_SCOPE,
-            shard_identifier=token.id,
-            category=OutboxCategory.API_TOKEN_UPDATE,
-            object_identifier=token.id,
-        )
-        assert outboxes.exists()
-        assert outboxes.count() > 0
-
-        with assume_test_silo_mode(SiloMode.CELL):
-            replica = ApiTokenReplica.objects.get(apitoken_id=token.id)
-            assert replica.expires_at != updated_expires_at
-
-    @override_options({"api-token-async-flush": True})
-    def test_async_replication_creates_replica_after_processing(self) -> None:
-        user = self.create_user()
-
-        with self.tasks():
-            token = ApiToken.objects.create(user_id=user.id)
-
-        # Verify outboxes were processed (should be deleted after processing)
-        remaining_outboxes = ControlOutbox.objects.filter(
-            shard_scope=OutboxScope.API_TOKEN_SCOPE,
-            shard_identifier=token.id,
-            category=OutboxCategory.API_TOKEN_UPDATE,
-            object_identifier=token.id,
-        )
-        assert not remaining_outboxes.exists()
-
-        with assume_test_silo_mode(SiloMode.CELL):
-            replica = ApiTokenReplica.objects.get(apitoken_id=token.id)
-            assert replica.hashed_token == token.hashed_token
-            assert replica.user_id == user.id
-
-    @override_options({"api-token-async-flush": True})
-    def test_async_replication_updates_existing_replica(self) -> None:
-        user = self.create_user()
-        initial_expires_at = timezone.now() + timedelta(days=1)
-        updated_expires_at = timezone.now() + timedelta(days=30)
-
-        with self.tasks():
-            token = ApiToken.objects.create(user_id=user.id, expires_at=initial_expires_at)
-
-        with assume_test_silo_mode(SiloMode.CELL):
-            replica = ApiTokenReplica.objects.get(apitoken_id=token.id)
-            assert replica.expires_at is not None
-            assert abs((replica.expires_at - initial_expires_at).total_seconds()) < 1
-
-        with self.tasks():
-            token.update(expires_at=updated_expires_at)
-
-        with assume_test_silo_mode(SiloMode.CELL):
-            replica = ApiTokenReplica.objects.get(apitoken_id=token.id)
-            assert replica.expires_at is not None
-            assert abs((replica.expires_at - updated_expires_at).total_seconds()) < 1
-
     def convert_token_outboxes_to_user_scope(self, token_id: int, user: User) -> None:
         original_query = ControlOutbox.objects.filter(
             shard_scope=OutboxScope.API_TOKEN_SCOPE, shard_identifier=token_id