From 015b935f81a6dcba8d9c45e4cf187e6ae36305ba Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 2 Aug 2026 20:02:07 +0800 Subject: [PATCH] Fix Alembic brownfield patches aborting the Postgres transaction. Optional DDL used bare try/except, which left the migration txn aborted so alembic_version never stamped and config-sync APIs failed. Run best-effort statements inside SAVEPOINTs. Co-authored-by: Cursor --- netx_api/port_traffic_migrate.py | 59 ++++++++--------------------- netx_api/schema_patches.py | 64 +++++++++++++++++++++++++------- netx_api/topology_migrate.py | 61 ++++++++++++------------------ 3 files changed, 89 insertions(+), 95 deletions(-) diff --git a/netx_api/port_traffic_migrate.py b/netx_api/port_traffic_migrate.py index 50df6d7..2dbe8c5 100644 --- a/netx_api/port_traffic_migrate.py +++ b/netx_api/port_traffic_migrate.py @@ -9,6 +9,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from .models import PortTrafficDevice, PortTrafficSample, PortTrafficSeries, PortTrafficTarget +from .schema_patches import _run_sql _log = logging.getLogger("netx.port_traffic.migrate") @@ -65,7 +66,7 @@ def ensure_port_traffic_series_schema(conn) -> None: ) # Older installs created task_id as NOT NULL before the device-centric rename. # Keep both columns populated and relax NOT NULL so ORM inserts that only set - # device_id (or only task_id) do not 500. + # device_id (or only task_id) do not 500. Each statement uses SAVEPOINT. for sql in ( """ UPDATE port_traffic_series @@ -79,28 +80,11 @@ def ensure_port_traffic_series_schema(conn) -> None: """, "ALTER TABLE port_traffic_series ALTER COLUMN task_id DROP NOT NULL", "ALTER TABLE port_traffic_series ALTER COLUMN task_id SET DEFAULT ''", - ): - try: - conn.exec_driver_sql(sql) - except Exception: - pass - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_port_traffic_series_device_id ON port_traffic_series (device_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_port_traffic_series_status ON port_traffic_series (status)" - ) - - conn.exec_driver_sql( - "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS device_id VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS task_id VARCHAR(64) DEFAULT ''" - ) - for sql in ( + "CREATE INDEX IF NOT EXISTS ix_port_traffic_series_device_id ON port_traffic_series (device_id)", + "CREATE INDEX IF NOT EXISTS ix_port_traffic_series_status ON port_traffic_series (status)", + "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''", + "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS device_id VARCHAR(64) DEFAULT ''", + "ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS task_id VARCHAR(64) DEFAULT ''", """ UPDATE port_traffic_target SET device_id = task_id @@ -113,31 +97,18 @@ def ensure_port_traffic_series_schema(conn) -> None: """, "ALTER TABLE port_traffic_target ALTER COLUMN task_id DROP NOT NULL", "ALTER TABLE port_traffic_target ALTER COLUMN task_id SET DEFAULT ''", - ): - try: - conn.exec_driver_sql(sql) - except Exception: - pass - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_port_traffic_target_series_id ON port_traffic_target (series_id)" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_port_traffic_target_device_id ON port_traffic_target (device_id)" - ) - conn.exec_driver_sql( - "ALTER TABLE port_traffic_sample ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''" - ) - conn.exec_driver_sql( - "CREATE INDEX IF NOT EXISTS ix_port_traffic_sample_series_id ON port_traffic_sample (series_id)" - ) - # One active interface globally per physical port - conn.exec_driver_sql( + "CREATE INDEX IF NOT EXISTS ix_port_traffic_target_series_id ON port_traffic_target (series_id)", + "CREATE INDEX IF NOT EXISTS ix_port_traffic_target_device_id ON port_traffic_target (device_id)", + "ALTER TABLE port_traffic_sample ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''", + "CREATE INDEX IF NOT EXISTS ix_port_traffic_sample_series_id ON port_traffic_sample (series_id)", + # One active interface globally per physical port (may fail if duplicates exist) """ CREATE UNIQUE INDEX IF NOT EXISTS uq_port_traffic_target_active_if ON port_traffic_target (source, target_id, ifname) WHERE status = 'active' - """ - ) + """, + ): + _run_sql(conn, sql) conn.exec_driver_sql( """ diff --git a/netx_api/schema_patches.py b/netx_api/schema_patches.py index 2671711..1d965bb 100644 --- a/netx_api/schema_patches.py +++ b/netx_api/schema_patches.py @@ -24,19 +24,62 @@ def _dialect_name(conn: Connection) -> str: def _run_sql(conn: Connection, sql: str, *, quiet: bool = True) -> None: + """Run one DDL/DML statement. + + On PostgreSQL a failed statement aborts the whole transaction. Optional + patches must use a SAVEPOINT (``begin_nested``) so Alembic can still stamp + ``alembic_version`` after a best-effort ALTER fails. + """ stmt = str(sql or "").strip() if not stmt: return - try: + + def _exec() -> None: if hasattr(conn, "exec_driver_sql"): conn.exec_driver_sql(stmt) else: conn.execute(text(stmt)) + + if not quiet: + _exec() + return + + nested = None + try: + nested = conn.begin_nested() except Exception: - if quiet: - _log.debug("schema patch skipped/failed: %s", stmt[:120], exc_info=True) - else: - raise + nested = None + try: + _exec() + if nested is not None: + nested.commit() + except Exception: + if nested is not None: + try: + nested.rollback() + except Exception: + pass + _log.debug("schema patch skipped/failed: %s", stmt[:120], exc_info=True) + + +def _run_optional_block(conn: Connection, label: str, fn) -> None: + """Run a block of optional DDL inside a SAVEPOINT.""" + nested = None + try: + nested = conn.begin_nested() + except Exception: + nested = None + try: + fn(conn) + if nested is not None: + nested.commit() + except Exception: + if nested is not None: + try: + nested.rollback() + except Exception: + pass + _log.debug("schema block skipped/failed: %s", label, exc_info=True) def apply_auth_schema_patches(conn: Connection) -> None: @@ -128,14 +171,9 @@ def apply_domain_schema_patches(conn: Connection) -> None: from .port_traffic_migrate import ensure_port_traffic_series_schema from .topology_migrate import ensure_topology_schema - try: - ensure_port_traffic_series_schema(conn) - except Exception: - _log.debug("port_traffic schema ensure failed", exc_info=True) - try: - ensure_topology_schema(conn) - except Exception: - _log.debug("topology schema ensure failed", exc_info=True) + # Never let optional ensure_* poison the Alembic transaction. + _run_optional_block(conn, "port_traffic", ensure_port_traffic_series_schema) + _run_optional_block(conn, "topology", ensure_topology_schema) _run_sql( conn, diff --git a/netx_api/topology_migrate.py b/netx_api/topology_migrate.py index a89e485..a9db859 100644 --- a/netx_api/topology_migrate.py +++ b/netx_api/topology_migrate.py @@ -4,9 +4,10 @@ from __future__ import annotations import logging -from sqlalchemy import text from sqlalchemy.engine import Connection +from .schema_patches import _run_sql + _log = logging.getLogger("netx.topology.migrate") _LEGACY_TABLES = ("topology_edge", "topology_node", "topology_map") @@ -16,13 +17,10 @@ def drop_legacy_topology_tables(conn: Connection) -> None: """Remove document-style topology_* tables after cutover to fabric/view.""" dialect = str(getattr(conn.dialect, "name", "") or "").lower() for table in _LEGACY_TABLES: - try: - if dialect.startswith("postgres"): - conn.execute(text(f'DROP TABLE IF EXISTS "{table}" CASCADE')) - else: - conn.execute(text(f"DROP TABLE IF EXISTS {table}")) - except Exception: - _log.exception("drop legacy topology table failed: %s", table) + if dialect.startswith("postgres"): + _run_sql(conn, f'DROP TABLE IF EXISTS "{table}" CASCADE') + else: + _run_sql(conn, f"DROP TABLE IF EXISTS {table}") def ensure_topology_schema(conn: Connection) -> None: @@ -120,34 +118,24 @@ def ensure_topology_schema(conn: Connection) -> None: ] ) for sql in alter_stmts: - try: - conn.execute(text(sql)) - except Exception: - _log.debug("topology alter skipped/failed: %s", sql[:80], exc_info=True) + _run_sql(conn, sql) # Backfill hours from legacy days when hours unset/zero. - try: - conn.execute( - text( - "UPDATE lldp_collect_policy SET interval_hours = " - "CASE WHEN COALESCE(interval_hours, 0) <= 0 " - "THEN GREATEST(1, COALESCE(interval_days, 1)) * 24 " - "ELSE interval_hours END" - ) - ) - except Exception: - try: - # SQLite has no GREATEST in older builds — use MAX. - conn.execute( - text( - "UPDATE lldp_collect_policy SET interval_hours = " - "CASE WHEN COALESCE(interval_hours, 0) <= 0 " - "THEN MAX(1, COALESCE(interval_days, 1)) * 24 " - "ELSE interval_hours END" - ) - ) - except Exception: - _log.debug("interval_hours backfill skipped", exc_info=True) + _run_sql( + conn, + "UPDATE lldp_collect_policy SET interval_hours = " + "CASE WHEN COALESCE(interval_hours, 0) <= 0 " + "THEN GREATEST(1, COALESCE(interval_days, 1)) * 24 " + "ELSE interval_hours END", + ) + # SQLite has no GREATEST in older builds — use MAX (no-op if previous applied). + _run_sql( + conn, + "UPDATE lldp_collect_policy SET interval_hours = " + "CASE WHEN COALESCE(interval_hours, 0) <= 0 " + "THEN MAX(1, COALESCE(interval_days, 1)) * 24 " + "ELSE interval_hours END", + ) if not dialect.startswith("postgres"): return @@ -169,7 +157,4 @@ def ensure_topology_schema(conn: Connection) -> None: "CREATE INDEX IF NOT EXISTS ix_topo_discover_job_trigger ON topo_discover_job (trigger_mode)", ] for sql in stmts: - try: - conn.execute(text(sql)) - except Exception: - _log.exception("ensure topology index failed: %s", sql[:80]) + _run_sql(conn, sql)