Fix bottlenecks in incoming activities queue

alembic/versions/2024_12_08_1122-aa8e72de7bb6_create_index_for_incoming_activity_next_.py

@@ -0,0 +1,28 @@
+"""create index for incoming activity next try
+
+Revision ID: aa8e72de7bb6
+Revises: a209f0333f5a
+Create Date: 2024-12-09 11:22:57.917748+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'aa8e72de7bb6'
+down_revision = 'a209f0333f5a'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('incoming_activity', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_incoming_activity_next_try'), ['next_try'], unique=False)
+        batch_op.create_index(batch_op.f('ix_incoming_activity_status'), ['is_processed', 'is_errored'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('incoming_activity', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_incoming_activity_next_try'))
+        batch_op.drop_index(batch_op.f('ix_incoming_status'))

alembic/versions/2025_01_10_2021-49774a0f5911_unpack_ap_in_reply_to.py

@@ -0,0 +1,39 @@
+"""unpack ap_in_reply_to
+
+Revision ID: 49774a0f5911
+Revises: aa8e72de7bb6
+Create Date: 2025-01-10 12:21:31.055099+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '49774a0f5911'
+down_revision = 'aa8e72de7bb6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table('inbox', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('in_reply_to', sa.Integer()))
+        batch_op.create_index(batch_op.f('ix_inbox_in_reply_to'), ['in_reply_to'], unique=False)
+
+    with op.batch_alter_table('outbox', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('in_reply_to', sa.Integer()))
+        batch_op.create_index(batch_op.f('ix_outbox_in_reply_to'), ['in_reply_to'], unique=False)
+
+    op.execute("UPDATE inbox SET in_reply_to = json_extract(ap_object, '$.inReplyTo')")
+    op.execute("UPDATE outbox SET in_reply_to = json_extract(ap_object, '$.inReplyTo')")
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('inbox', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_inbox_in_reply_to'))
+        batch_op.drop_column('in_reply_to')
+
+    with op.batch_alter_table('outbox', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_outbox_in_reply_to'))
+        batch_op.drop_column('in_reply_to')

app/ap_object.py

@@ -211,10 +211,6 @@ def activity_object_ap_id(self) -> str | None:
 
         return None
 
-    @property
-    def in_reply_to(self) -> str | None:
-        return self.ap_object.get("inReplyTo")
-
     @property
     def is_local_reply(self) -> bool:
         if not self.in_reply_to:
@@ -349,3 +345,7 @@ def ap_object(self) -> ap.RawObject:
     @property
     def actor(self) -> Actor:
         return self._actor
+
+    @property
+    def in_reply_to(self) -> str | None:
+        return self._raw_object.get("inReplyTo")

app/boxes.py

@@ -101,6 +101,7 @@ async def save_outbox_object(
         is_transient=is_transient,
         conversation=conversation,
         slug=slug,
+        in_reply_to=ro.in_reply_to,
     )
     db_session.add(outbox_object)
     await db_session.flush()
@@ -1264,16 +1265,14 @@ async def _get_replies_count(
     return (
         await db_session.scalar(
             select(func.count(models.InboxObject.id)).where(
-                func.json_extract(models.InboxObject.ap_object, "$.inReplyTo")
-                == replied_object_ap_id,
+                models.InboxObject.in_reply_to == replied_object_ap_id,
                 models.InboxObject.is_deleted.is_(False),
             )
         )
     ) + (
         await db_session.scalar(
             select(func.count(models.OutboxObject.id)).where(
-                func.json_extract(models.OutboxObject.ap_object, "$.inReplyTo")
-                == replied_object_ap_id,
+                models.OutboxObject.in_reply_to == replied_object_ap_id,
                 models.OutboxObject.is_deleted.is_(False),
             )
         )
@@ -1968,6 +1967,7 @@ async def _process_note_object(
         is_hidden_from_stream=not stream_visibility_callback(object_info),
         # We may already have some replies in DB
         replies_count=await _get_replies_count(db_session, ro.ap_id),
+        in_reply_to=ro.in_reply_to,
     )
 
     db_session.add(inbox_object)
@@ -2218,6 +2218,7 @@ async def _handle_announce_activity(
                         db_session, announced_object
                     ),
                     is_hidden_from_stream=True,
+                    in_reply_to=announced_object.in_reply_to,
                 )
                 db_session.add(announced_inbox_object)
                 await db_session.flush()
@@ -2430,6 +2431,7 @@ async def save_to_inbox(
         ),
         activity_object_ap_id=activity_ro.activity_object_ap_id,
         is_hidden_from_stream=True,
+        in_reply_to=activity_ro.in_reply_to,
     )
 
     db_session.add(inbox_object)
@@ -2619,6 +2621,7 @@ async def save_object_to_inbox(
         activity_object_ap_id=ro.activity_object_ap_id,
         og_meta=await opengraph.og_meta_from_note(db_session, ro),
         is_hidden_from_stream=True,
+        in_reply_to=ro.in_reply_to,
     )
 
     db_session.add(inbox_object)

app/models.py

@@ -131,6 +131,7 @@ class InboxObject(Base, BaseObject):
     is_deleted = Column(Boolean, nullable=False, default=False)
     is_transient = Column(Boolean, nullable=False, default=False, server_default="0")
 
+    in_reply_to: Mapped[str] = Column(String, index=True)
     replies_count: Mapped[int] = Column(Integer, nullable=False, default=0)
 
     og_meta: Mapped[list[dict[str, Any]] | None] = Column(JSON, nullable=True)
@@ -179,6 +180,7 @@ class OutboxObject(Base, BaseObject):
     ap_published_at = Column(DateTime(timezone=True), nullable=False, default=now)
     visibility = Column(Enum(ap.VisibilityEnum), nullable=False)
     conversation = Column(String, nullable=True)
+    in_reply_to: Mapped[str] = Column(String, index=True)
 
     likes_count = Column(Integer, nullable=False, default=0)
     announces_count = Column(Integer, nullable=False, default=0)