Merge branch 'release-1.10.0' into feat/extension-production-install

Author: erichare

src/backend/base/langflow/alembic/versions/b4c2f8e9a1d3_ensure_message_ingestion_record_message_fk.py

@@ -0,0 +1,81 @@
+"""Ensure message ingestion records keep their message FK
+
+Phase: EXPAND
+
+Revision ID: b4c2f8e9a1d3
+Revises: mb00a1b2c3d4
+Create Date: 2026-05-09 00:00:00.000000
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from langflow.utils import migration
+
+# revision identifiers, used by Alembic.
+revision: str = "b4c2f8e9a1d3"  # pragma: allowlist secret
+down_revision: str | None = "mb00a1b2c3d4"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+MIR_TABLE = "message_ingestion_record"
+MESSAGE_TABLE = "message"
+MESSAGE_FK_NAME = "fk_message_ingestion_record_message_id_message"
+
+
+def _message_fk_exists(conn) -> bool:
+    inspector = sa.inspect(conn)
+    for fk in inspector.get_foreign_keys(MIR_TABLE):
+        options = fk.get("options") or {}
+        ondelete = (options.get("ondelete") or "").upper()
+        if (
+            fk.get("constrained_columns") == ["message_id"]
+            and fk.get("referred_table") == MESSAGE_TABLE
+            and fk.get("referred_columns") == ["id"]
+            and ondelete == "CASCADE"
+        ):
+            return True
+    return False
+
+
+def _constraint_exists(conn, constraint_name: str) -> bool:
+    inspector = sa.inspect(conn)
+    return any(fk.get("name") == constraint_name for fk in inspector.get_foreign_keys(MIR_TABLE))
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+
+    # This repairs a PostgreSQL startup race where an older idempotent migration can
+    # replay DROP TABLE "message" CASCADE after the memory-base table already exists.
+    # That drops only the inbound message_id FK, while the mb00 migration then skips
+    # recreating the existing message_ingestion_record table.
+    if conn.dialect.name != "postgresql":
+        return
+
+    if not migration.table_exists(MIR_TABLE, conn) or not migration.table_exists(MESSAGE_TABLE, conn):
+        return
+
+    if _message_fk_exists(conn):
+        return
+
+    op.create_foreign_key(
+        MESSAGE_FK_NAME,
+        MIR_TABLE,
+        MESSAGE_TABLE,
+        ["message_id"],
+        ["id"],
+        ondelete="CASCADE",
+        postgresql_not_valid=True,
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+
+    if conn.dialect.name != "postgresql" or not migration.table_exists(MIR_TABLE, conn):
+        return
+
+    if _constraint_exists(conn, MESSAGE_FK_NAME):
+        op.drop_constraint(MESSAGE_FK_NAME, MIR_TABLE, type_="foreignkey")

src/backend/tests/unit/alembic/test_migration_execution.py

@@ -379,6 +379,76 @@ def test_no_phantom_migrations(db_url):
         )
 
 
+def test_message_ingestion_message_fk_repaired(db_url):
+    """Repair the Postgres-only state left by concurrent startup migrations."""
+    if not db_url.startswith("postgresql"):
+        pytest.skip("PostgreSQL-only migration repair")
+
+    alembic_cfg = _make_alembic_cfg(db_url)
+    command.upgrade(alembic_cfg, "mb00a1b2c3d4")
+
+    engine = create_engine(_engine_url(db_url))
+    try:
+        with engine.begin() as connection:
+            constraint_name = connection.execute(
+                text(
+                    """
+                    SELECT conname
+                    FROM pg_constraint
+                    WHERE contype = 'f'
+                      AND conrelid = 'message_ingestion_record'::regclass
+                      AND confrelid = 'message'::regclass
+                      AND conkey = ARRAY[
+                          (
+                              SELECT attnum
+                              FROM pg_attribute
+                              WHERE attrelid = 'message_ingestion_record'::regclass
+                                AND attname = 'message_id'
+                          )
+                      ]::smallint[]
+                    """
+                )
+            ).scalar_one()
+            connection.execute(text(f'ALTER TABLE message_ingestion_record DROP CONSTRAINT "{constraint_name}"'))
+    finally:
+        engine.dispose()
+
+    command.upgrade(alembic_cfg, "head")
+
+    engine = create_engine(_engine_url(db_url))
+    try:
+        with engine.connect() as connection:
+            restored_fk = connection.execute(
+                text(
+                    """
+                    SELECT conname, confdeltype
+                    FROM pg_constraint
+                    WHERE contype = 'f'
+                      AND conrelid = 'message_ingestion_record'::regclass
+                      AND confrelid = 'message'::regclass
+                      AND conkey = ARRAY[
+                          (
+                              SELECT attnum
+                              FROM pg_attribute
+                              WHERE attrelid = 'message_ingestion_record'::regclass
+                                AND attname = 'message_id'
+                          )
+                      ]::smallint[]
+                    """
+                )
+            ).one_or_none()
+            assert restored_fk is not None
+            assert restored_fk.confdeltype == "c"
+
+            migration_context = MigrationContext.configure(connection)
+            diffs = compare_metadata(migration_context, SQLModel.metadata)
+    finally:
+        engine.dispose()
+
+    significant_diffs = _filter_diffs(diffs, db_url)
+    assert significant_diffs == []
+
+
 def test_upgrade_from_main_branch(db_url):
     """Verify that a DB at main's head can upgrade to current head and downgrade back.