refactor(alerts): use bounded batch repository operations

litellm/models/issues.py

@@ -7,7 +7,6 @@ from litellm.types.proxy.issues import (
     IssueActivityResponse,
     IssueCohort,
     IssueDeliveryBinding,
-    IssueDeliveryInput,
     IssueFamily,
     IssueFamilyList,
     IssueFamilySetting,
@@ -20,7 +19,6 @@ from litellm.types.proxy.issues import (
     IssueProjectionSignal,
     IssueScope,
     IssueSignal,
-    IssueSignalInput,
     IssueSignalList,
 )
 from litellm.types.proxy.management_endpoints.common_daily_activity import (
@@ -58,13 +56,6 @@ class IssueStore(Protocol):
 
     async def status_scopes(self, issue_id: str) -> tuple[IssueCohort, ...]: ...
 
-    async def upsert_issue(
-        self,
-        data: IssueInput,
-        reopen: bool = False,
-        status: UsageAlertStatus | None = None,
-    ) -> Issue: ...
-
     async def upsert_issues(
         self,
         inputs: tuple[IssueInput, ...],
@@ -72,12 +63,8 @@ class IssueStore(Protocol):
         status: UsageAlertStatus | None = None,
     ) -> tuple[Issue, ...]: ...
 
-    async def find_active_issue(self, cohort_key: str) -> Issue | None: ...
-
     async def find_active_issues(self, cohort_keys: tuple[str, ...]) -> tuple[Issue, ...]: ...
 
-    async def add_signals(self, issue_id: str, signals: tuple[IssueSignalInput, ...]) -> None: ...
-
     async def add_signal_batch(self, signals: tuple[IssueSignal, ...]) -> None: ...
 
     async def list_signals(
@@ -112,14 +99,6 @@ class IssueStore(Protocol):
         self, family: PerformanceIssueFamily, data: IssueFamilyUpdate
     ) -> IssueFamilySetting | None: ...
 
-    async def queue_delivery(
-        self,
-        data: IssueDeliveryInput,
-        rule_id: str,
-        rule_revision: str,
-        source_id: str,
-    ) -> None: ...
-
     async def queue_deliveries(self, bindings: tuple[IssueDeliveryBinding, ...]) -> None: ...
 
     async def pending_deliveries(self, limit: int = 100) -> tuple[IssuePendingDelivery, ...]: ...

litellm/repositories/issues.py

@@ -17,7 +17,6 @@ from litellm.types.proxy.issues import (
     Issue,
     IssueCohort,
     IssueDeliveryBinding,
-    IssueDeliveryInput,
     IssueFamily,
     IssueFamilyList,
     IssueFamilySetting,
@@ -29,7 +28,6 @@ from litellm.types.proxy.issues import (
     IssueProjectionSignal,
     IssueScope,
     IssueSignal,
-    IssueSignalInput,
     IssueSignalList,
 )
 from litellm.types.proxy.usage_alert_defaults import DEFAULT_USAGE_RULE_IDS, PerformanceIssueFamily
@@ -110,12 +108,6 @@ class IssuesRepository:
     def from_prisma(cls, prisma: "PrismaClient") -> "IssuesRepository":
         return cls(cast(AlertDatabase, prisma.writer_db))  # cast-ok: generated Prisma delegate implements query_raw
 
-    async def upsert_issue(
-        self, data: IssueInput, reopen: bool = False, status: UsageAlertStatus | None = None
-    ) -> Issue:
-        rows: Final = await self.upsert_issues((data,), reopen=reopen, status=status)
-        return rows[0]
-
     async def upsert_issues(
         self,
         inputs: tuple[IssueInput, ...],
@@ -225,10 +217,6 @@ class IssuesRepository:
             for issue in issues
         )
 
-    async def find_active_issue(self, cohort_key: str) -> Issue | None:
-        rows: Final = await self.find_active_issues((cohort_key,))
-        return rows[0] if rows else None
-
     async def find_active_issues(self, cohort_keys: tuple[str, ...]) -> tuple[Issue, ...]:
         if len(cohort_keys) > 500:
             raise ValueError("Read at most 500 issue cohorts per batch")
@@ -302,11 +290,6 @@ class IssuesRepository:
             )
         )
 
-    async def add_signals(self, issue_id: str, signals: tuple[IssueSignalInput, ...]) -> None:
-        await self.add_signal_batch(
-            tuple(IssueSignal.model_validate({**signal.model_dump(), "issue_id": issue_id}) for signal in signals)
-        )
-
     async def add_signal_batch(self, signals: tuple[IssueSignal, ...]) -> None:
         if not signals:
             return
@@ -537,17 +520,6 @@ class IssuesRepository:
 
         return await update_family(self.database, family, data)
 
-    async def queue_delivery(
-        self,
-        data: IssueDeliveryInput,
-        rule_id: str,
-        rule_revision: str,
-        source_id: str,
-    ) -> None:
-        await self.queue_deliveries(
-            (IssueDeliveryBinding(delivery=data, rule_id=rule_id, rule_revision=rule_revision, source_id=source_id),)
-        )
-
     async def queue_deliveries(self, bindings: tuple[IssueDeliveryBinding, ...]) -> None:
         if not bindings:
             return

litellm/repositories/usage_alerts.py

@@ -297,16 +297,6 @@ class UsageAlertsRepository:
             default_summary=count[0].default_summary,
         )
 
-    async def list_enabled_rules(self, limit: int = 100, offset: int = 0) -> tuple[UsageAlertRule, ...]:
-        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
-            await self.database.query_raw(
-                'SELECT record FROM "LiteLLM_UsageAlertRule" WHERE enabled AND NOT deleted ORDER BY id LIMIT $1 OFFSET $2',
-                limit,
-                offset,
-            )
-        )
-        return tuple(UsageAlertRule.model_validate(row.record) for row in rows)
-
     async def get_alert(self, alert_id: str) -> UsageAlert | None:
         rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
             await self.database.query_raw('SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE id=$1', alert_id)
@@ -557,21 +547,6 @@ class UsageAlertsRepository:
     def _checkpoint(rule: UsageAlertRule, day: date) -> str:
         return f"{rule.id}:{rule.updated_at.isoformat()}:{day.isoformat()}"
 
-    async def is_day_complete(self, rule: UsageAlertRule, day: date) -> bool:
-        rows: Final = TypeAdapter(tuple[_Count, ...]).validate_python(
-            await self.database.query_raw(
-                'SELECT 1 AS count FROM "LiteLLM_UsageAlertCheckpoint" WHERE id=$1 AND completed',
-                self._checkpoint(rule, day),
-            )
-        )
-        return bool(rows)
-
-    async def finish_day(self, rule: UsageAlertRule, day: date) -> None:
-        await self.database.query_raw(
-            'INSERT INTO "LiteLLM_UsageAlertCheckpoint" (id,completed) VALUES ($1,true) ON CONFLICT(id) DO NOTHING RETURNING id',
-            self._checkpoint(rule, day),
-        )
-
     async def ensure_default_rules(self, rules: tuple[UsageAlertRuleInput, ...]) -> None:
         for rule in rules:
             await self._ensure_default_rule(rule)
@@ -593,17 +568,6 @@ class UsageAlertsRepository:
             record.model_dump_json(),
         )
 
-    async def list_breached_entity_ids(self, rule_id: str, limit: int = 500, offset: int = 0) -> tuple[str | None, ...]:
-        rows: Final = TypeAdapter(tuple[Mapping[str, str | None], ...]).validate_python(
-            await self.database.query_raw(
-                """SELECT DISTINCT i.scope_id FROM "LiteLLM_UsageAlertIncident" i JOIN "LiteLLM_UsageAlertRule" r ON r.id=i.rule_id WHERE i.rule_id=$1 AND i.breached AND i.scope_type=r.scope_type AND i.metric=r.record->>'metric' AND i.record->>'direction'=r.record->>'direction' AND (r.scope_id IS NULL OR i.scope_id=r.scope_id) ORDER BY i.scope_id NULLS FIRST LIMIT $2 OFFSET $3""",
-                rule_id,
-                limit,
-                offset,
-            )
-        )
-        return tuple(row["scope_id"] for row in rows)
-
     async def record_evaluation_status(
         self, rule: UsageAlertRule, day: date, status: UsageAlertEvaluationStatus
     ) -> None:
@@ -615,34 +579,6 @@ class UsageAlertsRepository:
             rule.updated_at.isoformat(),
         )
 
-    async def record_observation(
-        self,
-        rule: UsageAlertRule,
-        entity_id: str | None,
-        day: date,
-        observed: float,
-        baseline: float,
-        threshold: float,
-        breached: bool,
-    ) -> UsageAlert | None:
-        observation: Final = UsageAlertObservation(
-            entity_id=entity_id, day=day, observed=observed, baseline=baseline, threshold=threshold, breached=breached
-        )
-        changed: Final = await self.record_observations(rule, (observation,))
-        if changed:
-            return changed[0]
-        rows: Final = TypeAdapter(tuple[_Record, ...]).validate_python(
-            await self.database.query_raw(
-                """SELECT record FROM "LiteLLM_UsageAlertIncident" WHERE rule_id=$1 AND scope_key=$2 AND scope_type=$3 AND metric=$4 AND record->>'direction'=$5 ORDER BY evaluation_day DESC,id DESC LIMIT 1""",
-                rule.id,
-                entity_id or "",
-                rule.scope_type,
-                rule.metric,
-                rule.direction,
-            )
-        )
-        return UsageAlert.model_validate(rows[0].record) if rows else None
-
     async def record_observations(
         self, rule: UsageAlertRule, observations: tuple[UsageAlertObservation, ...]
     ) -> tuple[UsageAlert, ...]:

tests/integration/database/test_issues.py

@@ -27,17 +27,19 @@ from litellm.types.proxy.issues import (
     IssueActivityResponse,
     IssueCohort,
     IssueComparison,
+    IssueDeliveryBinding,
     IssueDeliveryInput,
     IssueFamilyPolicyUpdate,
     IssueFamilyUpdate,
     IssueInput,
     IssueList,
     IssueModelPair,
+    IssueSignal,
     IssueSignalInput,
     IssueSignalList,
 )
 from litellm.types.proxy.usage_alert_defaults import default_usage_rule_id, default_usage_rules
-from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertObservation, UsageAlertRuleInput
+from litellm.types.proxy.usage_alerts import UsageAlert, UsageAlertObservation, UsageAlertRule, UsageAlertRuleInput
 
 
 def _issue(cohort: IssueCohort) -> IssueInput:
@@ -71,6 +73,16 @@ def _issue(cohort: IssueCohort) -> IssueInput:
     )
 
 
+def _signals(issue_id: str, inputs: tuple[IssueSignalInput, ...]) -> tuple[IssueSignal, ...]:
+    return tuple(IssueSignal.model_validate({**signal.model_dump(), "issue_id": issue_id}) for signal in inputs)
+
+
+def _delivery(data: IssueDeliveryInput, rule: UsageAlertRule, source: UsageAlert) -> IssueDeliveryBinding:
+    return IssueDeliveryBinding(
+        delivery=data, rule_id=rule.id, rule_revision=rule.updated_at.isoformat(), source_id=source.id
+    )
+
+
 async def _query(db: ManagementDatabase, query: str, *args: object) -> object:
     return await IssuesRepository.from_prisma(db.prisma).database.query_raw(query, *args)
 
@@ -141,18 +153,20 @@ async def test_issue_http_scope_pagination_and_source_visibility(gateway: Gatewa
                     "baseline_value": 9999,
                 }
             )
-            await store.add_signals(
-                first.id,
-                (
-                    IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),
-                    IssueSignalInput(
-                        source="usage_alert",
-                        source_id=broader.id,
-                        usage_alert=broader,
-                        grouping_reason="dominant_key",
-                        explanation="Parent contribution evidence.",
+            await store.add_signal_batch(
+                _signals(
+                    first.id,
+                    (
+                        IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),
+                        IssueSignalInput(
+                            source="usage_alert",
+                            source_id=broader.id,
+                            usage_alert=broader,
+                            grouping_reason="dominant_key",
+                            explanation="Parent contribution evidence.",
+                        ),
                     ),
-                ),
+                )
             )
             path: Final = f"/issues/{first.id}"
             own: Final = gateway.request("GET", path, key=owner_key)
@@ -312,7 +326,7 @@ async def test_issue_activity_bounds_models_without_losing_totals(gateway: Gatew
         model_count,
     )
     try:
-        await store.upsert_issue(issue)
+        await store.upsert_issues((issue,))
         scope: Final = issue_activity_scope(cohort, day, day + timedelta(days=2))
         original: Final = await get_daily_activity_aggregated(repository.activity, scope, api_key_limit=20)
         response: Final = gateway.request("GET", f"/issues/{issue.id}/activity")
@@ -425,7 +439,7 @@ async def test_issue_provider_pairs_filter_activity_pages_counts_and_exports(
     store: Final = IssuesRepository.from_prisma(db.prisma)
     issue: Final = _issue(cohort).model_copy(update={"range_start": day, "range_end": day})
     try:
-        await store.upsert_issue(issue)
+        await store.upsert_issues((issue,))
         activity_http: Final = gateway.request("GET", f"/issues/{issue.id}/activity")
         assert activity_http.status_code == 200 and activity_http.json()["metadata"]["total_api_requests"] == 100
         keys_http: Final = gateway.request("GET", f"/issues/{issue.id}/keys", params={"limit": "1"})
@@ -503,7 +517,7 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
     next_issue_id: Final = str(uuid4())
     try:
         source: Final = (await alerts.record_observations(rule, (observation,)))[0]
-        initial: Final = await store.upsert_issue(original)
+        initial: Final = (await store.upsert_issues((original,)))[0]
         assert tuple(
             item.alert.id
             for item in await store.active_signals(
@@ -512,11 +526,11 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
         ) == (source.id,)
         assert await store.active_signals(original.observed_on, scope_ids=()) == ()
         assert await store.active_signals(original.observed_on, scope_ids=("other-example-user",)) == ()
-        await store.add_signals(
-            initial.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),)
+        await store.add_signal_batch(
+            _signals(initial.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),))
         )
-        await store.queue_delivery(delivery, rule.id, rule.updated_at.isoformat(), source.id)
-        await store.queue_delivery(invalid, rule.id, rule.updated_at.isoformat(), source.id)
+        await store.queue_deliveries((_delivery(delivery, rule, source),))
+        await store.queue_deliveries((_delivery(invalid, rule, source),))
         pending: Final = await store.pending_delivery(delivery.id)
         assert pending is not None and pending.alert.id == source.id and pending.issue.id == initial.id
         assert await store.pending_delivery(invalid.id) is None
@@ -526,7 +540,7 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
         assert enabled is not None and await store.pending_delivery(delivery.id) is None
         reevaluated: Final = await alerts.record_observations(enabled, (observation,))
         assert len(reevaluated) == 1 and reevaluated[0].id == source.id
-        await store.queue_delivery(delivery, enabled.id, enabled.updated_at.isoformat(), source.id)
+        await store.queue_deliveries((_delivery(delivery, enabled, source),))
         assert await store.pending_delivery(delivery.id) is not None
         await alerts.set_status(source.id, "dismissed")
         assert await store.pending_delivery(delivery.id) is None
@@ -541,7 +555,7 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
                 "comparisons": (original.comparisons[0].model_copy(update={"observed_value": 999}),),
             }
         )
-        continued: Final = await store.upsert_issue(later)
+        continued: Final = (await store.upsert_issues((later,)))[0]
         assert continued.status == "dismissed" and continued.comparisons == initial.comparisons
         assert continued.title == initial.title and continued.observed_on == initial.observed_on
         assert continued.last_seen > initial.last_seen
@@ -558,7 +572,7 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
             retried is not None and retried.delivery.attempts == 1 and retried.delivery.last_error == "delivery_failed"
         )
         await store.mark_delivery_sent(delivery.id)
-        await store.queue_delivery(delivery, enabled.id, enabled.updated_at.isoformat(), source.id)
+        await store.queue_deliveries((_delivery(delivery, enabled, source),))
         assert await store.pending_delivery(delivery.id) is None
         healthy: Final = observation.model_copy(
             update={"day": observation.day + timedelta(days=1), "breached": False, "observed": 120}
@@ -567,24 +581,28 @@ async def test_issue_delivery_guards_snapshot_freeze_and_delayed_recovery(db: Ma
         await store.recover_issues(observation.day + timedelta(days=2))
         recovered: Final = await store.get_issue(initial.id)
         assert recovered is not None and not recovered.active and recovered.status == "resolved"
-        assert await store.find_active_issue(original.cohort_key) is None
+        assert await store.find_active_issues((original.cohort_key,)) == ()
         repeated: Final = (
             await alerts.record_observations(
                 enabled, (observation.model_copy(update={"day": observation.day + timedelta(days=2)}),)
             )
         )[0]
         assert repeated.id != source.id
-        new_episode: Final = await store.upsert_issue(
-            original.model_copy(
-                update={
-                    "id": next_issue_id,
-                    "observed_on": repeated.day,
-                    "observed_at": later.observed_at + timedelta(minutes=1),
-                }
+        new_episode: Final = (
+            await store.upsert_issues(
+                (
+                    original.model_copy(
+                        update={
+                            "id": next_issue_id,
+                            "observed_on": repeated.day,
+                            "observed_at": later.observed_at + timedelta(minutes=1),
+                        }
+                    ),
+                )
             )
-        )
+        )[0]
         assert new_episode.active and new_episode.status == "open" and new_episode.id != initial.id
-        assert (await store.find_active_issue(original.cohort_key)) == new_episode
+        assert await store.find_active_issues((original.cohort_key,)) == (new_episode,)
     finally:
         await _remove_issues(db, (original.id, next_issue_id))
         await alerts.delete_rule(rule.id)
@@ -664,14 +682,11 @@ async def test_late_default_override_stops_queued_delivery_and_recovers_on_fresh
         await store.upsert_issues(issues)
         for issue in issues:
             source: UsageAlert = next(source for source in sources if source.scope_id == issue.cohort.scope_id)
-            await store.add_signals(
-                issue.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),)
+            await store.add_signal_batch(
+                _signals(issue.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),))
             )
-            await store.queue_delivery(
-                IssueDeliveryInput(id=issue.id, issue_id=issue.id, channel="slack"),
-                rule.id,
-                rule.updated_at.isoformat(),
-                source.id,
+            await store.queue_deliveries(
+                (_delivery(IssueDeliveryInput(id=issue.id, issue_id=issue.id, channel="slack"), rule, source),)
             )
             assert await store.pending_delivery(issue.id) is not None
         definition: Final = UsageAlertRuleInput(
@@ -826,7 +841,7 @@ async def test_deleted_key_owner_retains_historical_issue_access(gateway: Gatewa
             owner,
         )
         try:
-            await store.upsert_issue(issue)
+            await store.upsert_issues((issue,))
             path: Final = f"/issues/{issue.id}"
             detail: Final = gateway.request("GET", path, key=owner_key)
             assert detail.status_code == 200, "Deleted key owner must retain historical issue access"
@@ -884,23 +899,23 @@ async def test_issue_status_rechecks_changed_scope_set_and_caps_sources(db: Mana
                 ),
             )
         )[0]
-        await store.upsert_issue(issue)
-        await store.add_signals(
-            issue.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),)
+        await store.upsert_issues((issue,))
+        await store.add_signal_batch(
+            _signals(issue.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),))
         )
         authorized: Final = await store.status_scopes(issue.id)
         assert authorized == (issue.cohort,)
         assert await store.set_status(issue.id, "dismissed", "open", authorized_scopes=authorized) is not None
         context: Final = source.model_copy(update={"id": str(uuid4()), "metric": "spend"})
         same_scope: Final = (IssueSignalInput(source="usage_alert", source_id=context.id, usage_alert=context),)
-        await store.add_signals(issue.id, same_scope)
+        await store.add_signal_batch(_signals(issue.id, same_scope))
         unchanged: Final = await store.get_issue(issue.id)
         assert unchanged is not None and unchanged.status == "dismissed"
-        await store.add_signals(issue.id, same_scope)
+        await store.add_signal_batch(_signals(issue.id, same_scope))
         assert await store.get_issue(issue.id) == unchanged
         broader: Final = source.model_copy(update={"id": str(uuid4()), "scope_type": "deployment", "scope_id": None})
-        await store.add_signals(
-            issue.id, (IssueSignalInput(source="usage_alert", source_id=broader.id, usage_alert=broader),)
+        await store.add_signal_batch(
+            _signals(issue.id, (IssueSignalInput(source="usage_alert", source_id=broader.id, usage_alert=broader),))
         )
         assert await store.set_status(issue.id, "dismissed", "open", authorized_scopes=authorized) is None
         current: Final = await store.get_issue(issue.id)
@@ -908,9 +923,11 @@ async def test_issue_status_rechecks_changed_scope_set_and_caps_sources(db: Mana
         all_scopes: Final = await store.status_scopes(issue.id)
         assert await store.set_status(issue.id, "dismissed", "open", authorized_scopes=all_scopes) is not None
         broader_context: Final = broader.model_copy(update={"id": str(uuid4()), "metric": "spend"})
-        await store.add_signals(
-            issue.id,
-            (IssueSignalInput(source="usage_alert", source_id=broader_context.id, usage_alert=broader_context),),
+        await store.add_signal_batch(
+            _signals(
+                issue.id,
+                (IssueSignalInput(source="usage_alert", source_id=broader_context.id, usage_alert=broader_context),),
+            )
         )
         repeated_scope: Final = await store.get_issue(issue.id)
         assert repeated_scope is not None and repeated_scope.status == "dismissed"
@@ -943,22 +960,28 @@ async def test_issue_status_rechecks_changed_scope_set_and_caps_sources(db: Mana
                 locked: Final = IssuesRepository(
                     cast(AlertDatabase, transaction)  # cast-ok: generated Prisma transaction query delegate
                 )
-                await locked.add_signals(
-                    issue.id, (IssueSignalInput(source="usage_alert", source_id=racing.id, usage_alert=racing),)
+                await locked.add_signal_batch(
+                    _signals(
+                        issue.id, (IssueSignalInput(source="usage_alert", source_id=racing.id, usage_alert=racing),)
+                    )
                 )
         assert update.result() is None
         raced: Final = await store.get_issue(issue.id)
         assert raced is not None and raced.status == "open"
-        await store.add_signals(
-            issue.id,
-            tuple(
-                IssueSignalInput(
-                    source="usage_alert",
-                    source_id=f"scope-{index}",
-                    usage_alert=source.model_copy(update={"id": f"scope-{index}", "scope_id": f"other-user-{index}"}),
-                )
-                for index in range(20)
-            ),
+        await store.add_signal_batch(
+            _signals(
+                issue.id,
+                tuple(
+                    IssueSignalInput(
+                        source="usage_alert",
+                        source_id=f"scope-{index}",
+                        usage_alert=source.model_copy(
+                            update={"id": f"scope-{index}", "scope_id": f"other-user-{index}"}
+                        ),
+                    )
+                    for index in range(20)
+                ),
+            )
         )
         limited: Final = await store.status_scopes(issue.id)
         assert len(limited) == 11
@@ -1015,12 +1038,15 @@ async def test_resource_recovery_ignores_broader_and_traffic_context(db: Managem
                 if rule is not None
             ]
         )
-        await store.upsert_issue(issue)
-        await store.add_signals(
-            issue.id,
-            tuple(
-                IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source) for source in sources
-            ),
+        await store.upsert_issues((issue,))
+        await store.add_signal_batch(
+            _signals(
+                issue.id,
+                tuple(
+                    IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source)
+                    for source in sources
+                ),
+            )
         )
         primary: Final = rules[0]
         assert primary is not None
@@ -1127,12 +1153,11 @@ async def test_issue_prune_removes_expired_children_in_bounded_passes(
                 ),
             )
         )[0]
-        await store.add_signals(
-            expired.id,
-            (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),),
+        await store.add_signal_batch(
+            _signals(expired.id, (IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source),))
         )
         for delivery in deliveries:
-            await store.queue_delivery(delivery, rule.id, rule.updated_at.isoformat(), source.id)
+            await store.queue_deliveries((_delivery(delivery, rule, source),))
         assert await store.set_status(expired.id, "resolved", "open") is not None
         assert await store.set_status(recent.id, "resolved", "open") is not None
         await _query(
@@ -1231,7 +1256,7 @@ async def test_issue_resource_metadata_bounds_labels_and_deleted_history(
                 day.isoformat(),
                 reporting_key,
             )
-            await store.upsert_issue(issue)
+            await store.upsert_issues((issue,))
             keys: Final = await repository.keys(issue.cohort, day, day, 1, 0)
             assert keys.total_api_keys == 1 and len(keys.api_keys) == 1
             key: Final = keys.api_keys[0]
@@ -1293,12 +1318,15 @@ async def test_disabled_later_breach_requires_new_healthy_primary_evidence(db: M
     )
     try:
         sources: Final = tuple([(await alerts.record_observations(rule, (breach,)))[0] for rule in (first, second)])
-        await store.upsert_issue(issue)
-        await store.add_signals(
-            issue.id,
-            tuple(
-                IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source) for source in sources
-            ),
+        await store.upsert_issues((issue,))
+        await store.add_signal_batch(
+            _signals(
+                issue.id,
+                tuple(
+                    IssueSignalInput(source="usage_alert", source_id=source.id, usage_alert=source)
+                    for source in sources
+                ),
+            )
         )
         await alerts.record_observations(
             first,