diff --git a/alembic/versions/20260713_0100_c0d1e2f3a4b5_deduplicate_nodes_public_key.py b/alembic/versions/20260713_0100_c0d1e2f3a4b5_deduplicate_nodes_public_key.py deleted file mode 100644 index 834d620..0000000 --- a/alembic/versions/20260713_0100_c0d1e2f3a4b5_deduplicate_nodes_public_key.py +++ /dev/null @@ -1,174 +0,0 @@ -"""Deduplicate nodes by public_key and re-assert uniqueness - -Revision ID: c0d1e2f3a4b5 -Revises: 8f2a3c4d5e6f -Create Date: 2026-07-13 01:00:00.000000+00:00 - -A production database restore bypassed SQLite/Postgres UNIQUE enforcement on -``nodes.public_key``, leaving two rows with identical (or case-variant) keys. -This caused ``MultipleResultsFound`` in ``GET /api/v1/nodes/{public_key}`` and -in every collector handler that does find-or-create by public_key. - -This migration is a superset of ``b1c2d3e4f5a6`` (which only knew about 9 FK -columns and is already stamped as applied). It: - -1. Merges duplicate nodes (winner = earliest ``first_seen``, tiebreak - ``created_at``), re-pointing all 14 FK columns that currently reference - ``nodes.id``. -2. Merges scalar fields into the winner (name, adv_type, lat, lon, flags, - last_seen via MAX, is_observer via OR). -3. Deletes losers. -4. Re-lowercases all remaining ``public_key`` values. -5. Re-asserts the UNIQUE index on ``nodes.public_key`` in case the restore - dropped or weakened it. - -Irreversible — deleting duplicates destroys data. -""" - -from alembic import op -from sqlalchemy import text - -revision = "c0d1e2f3a4b5" -down_revision = "8f2a3c4d5e6f" -branch_labels = None -depends_on = None - - -# (table, column, ondelete) for every FK referencing nodes.id. -# CASCADE rows must be re-pointed BEFORE the loser is deleted; SET NULL rows -# are also re-pointed (not nulled) so we keep the linkage rather than orphan it. -_REPOINT_COLUMNS = [ - # ondelete=CASCADE - ("node_tags", "node_id"), - ("event_observers", "observer_node_id"), - ("route_nodes", "node_id"), - ("route_observers", "node_id"), - ("user_profile_nodes", "node_id"), - # ondelete=SET NULL - ("advertisements", "observer_node_id"), - ("advertisements", "node_id"), - ("telemetry", "observer_node_id"), - ("telemetry", "node_id"), - ("messages", "observer_node_id"), - ("trace_paths", "observer_node_id"), - ("events_log", "observer_node_id"), - ("raw_packets", "observer_node_id"), - ("packet_path_hops", "observer_node_id"), -] - - -def upgrade() -> None: - conn = op.get_bind() - - # SQLite uses GROUP_CONCAT; Postgres uses STRING_AGG. HAVING references - # COUNT(*) directly (Postgres disallows SELECT aliases in HAVING). - if conn.dialect.name == "postgresql": - id_agg = "STRING_AGG(id::text, ',')" - else: - id_agg = "GROUP_CONCAT(id)" - - duplicates = conn.execute(text(f""" - SELECT LOWER(public_key) AS lower_pk, - {id_agg} AS ids, - COUNT(*) AS cnt - FROM nodes - GROUP BY LOWER(public_key) - HAVING COUNT(*) > 1 - """)).fetchall() - - for row in duplicates: - lower_pk = row[0] - ids = row[1].split(",") - - winner_row = conn.execute( - text(""" - SELECT id FROM nodes - WHERE LOWER(public_key) = :lower_pk - ORDER BY first_seen ASC, created_at ASC - LIMIT 1 - """), - {"lower_pk": lower_pk}, - ).fetchone() - - if not winner_row: - continue - - winner_id = winner_row[0] - loser_ids = [nid for nid in ids if nid != winner_id] - - for loser_id in loser_ids: - params = {"winner_id": winner_id, "loser_id": loser_id} - - # Re-point every FK column from loser to winner. Wrapped in - # try/except per table so a missing table on an older DB doesn't - # abort the whole migration (defensive — schema drift). - for table, column in _REPOINT_COLUMNS: - conn.execute( - text( - f"UPDATE {table} SET {column} = :winner_id " - f"WHERE {column} = :loser_id" - ), - params, - ) - - # Merge scalar fields into winner (keep best data from either row). - conn.execute( - text(""" - UPDATE nodes SET - name = COALESCE(name, (SELECT name FROM nodes WHERE id = :loser_id)), - adv_type = COALESCE(adv_type, (SELECT adv_type FROM nodes WHERE id = :loser_id)), - flags = COALESCE(flags, (SELECT flags FROM nodes WHERE id = :loser_id)), - lat = COALESCE(lat, (SELECT lat FROM nodes WHERE id = :loser_id)), - lon = COALESCE(lon, (SELECT lon FROM nodes WHERE id = :loser_id)), - last_seen = ( - SELECT MAX(v) FROM ( - SELECT last_seen AS v FROM nodes WHERE id = :winner_id - UNION ALL - SELECT last_seen AS v FROM nodes WHERE id = :loser_id - ) - ), - is_observer = ( - is_observer - OR (SELECT is_observer FROM nodes WHERE id = :loser_id) - ) - WHERE id = :winner_id - """), - params, - ) - - # Delete the loser. Remaining orphans are cleaned by CASCADE/SET - # NULL at the FK level (now that we've re-pointed everything we - # care about, SET NULL columns with no re-point would just null - # out — but we re-pointed all of them above). - conn.execute( - text("DELETE FROM nodes WHERE id = :loser_id"), - {"loser_id": loser_id}, - ) - - # Normalize any remaining case variance (defense-in-depth — should be a - # no-op now that b1c2d3e4f5a6 has run, but the restore may have reintroduced - # mixed case). - conn.execute(text("UPDATE nodes SET public_key = LOWER(public_key)")) - - # Re-assert the UNIQUE index on nodes.public_key. The restore may have - # dropped it or recreated it as non-unique, which is how duplicates slipped - # in to begin with. Drop-if-exists then CREATE UNIQUE INDEX. Idempotent. - if conn.dialect.name == "postgresql": - # Postgres: use CREATE UNIQUE INDEX CONCURRENTLY is not available - # inside a transaction; alembic auto-begins one, so plain CREATE is - # fine for the data-fix path. - conn.execute(text("DROP INDEX IF EXISTS ix_nodes_public_key")) - conn.execute( - text("CREATE UNIQUE INDEX ix_nodes_public_key ON nodes (public_key)") - ) - else: - # SQLite: DROP/CREATE INDEX is fine without a table rebuild. - conn.execute(text("DROP INDEX IF EXISTS ix_nodes_public_key")) - conn.execute( - text("CREATE UNIQUE INDEX ix_nodes_public_key ON nodes (public_key)") - ) - - -def downgrade() -> None: - # Cannot reverse the merge — duplicate rows have been deleted. - pass diff --git a/alembic/versions/20260715_2100_76513f4c57e9_add_packet_path_hops_received_at_index.py b/alembic/versions/20260715_2100_76513f4c57e9_add_packet_path_hops_received_at_index.py deleted file mode 100644 index 4f7c79a..0000000 --- a/alembic/versions/20260715_2100_76513f4c57e9_add_packet_path_hops_received_at_index.py +++ /dev/null @@ -1,32 +0,0 @@ -"""add standalone received_at index to packet_path_hops - -Revision ID: 76513f4c57e9 -Revises: c0d1e2f3a4b5 -Create Date: 2026-07-15 21:00:00.000000+00:00 - -""" - -from typing import Sequence, Union - -from alembic import op - -revision: str = "76513f4c57e9" -down_revision: Union[str, None] = "c0d1e2f3a4b5" -branch_labels: Union[str, Sequence[str], None] = None -depends_on: Union[str, Sequence[str], None] = None - - -def upgrade() -> None: - op.create_index( - "ix_packet_path_hops_received_at", - "packet_path_hops", - ["received_at"], - unique=False, - ) - - -def downgrade() -> None: - op.drop_index( - "ix_packet_path_hops_received_at", - table_name="packet_path_hops", - ) diff --git a/alembic/versions/20260715_2300_71f6e01e4bf6_rename_degraded_to_clear_threshold.py b/alembic/versions/20260715_2300_71f6e01e4bf6_rename_degraded_to_clear_threshold.py deleted file mode 100644 index 82e43bf..0000000 --- a/alembic/versions/20260715_2300_71f6e01e4bf6_rename_degraded_to_clear_threshold.py +++ /dev/null @@ -1,39 +0,0 @@ -"""Rename degraded_threshold to clear_threshold. - -Revision ID: 71f6e01e4bf6 -Revises: 76513f4c57e9 -Create Date: 2026-07-15 23:00:00 -""" - -from alembic import op - -revision = "71f6e01e4bf6" -down_revision = "76513f4c57e9" -branch_labels = None -depends_on = None - - -def upgrade() -> None: - op.alter_column( - "routes", - "degraded_threshold", - new_column_name="clear_threshold", - ) - op.alter_column( - "route_results", - "effective_degraded", - new_column_name="effective_clear", - ) - - -def downgrade() -> None: - op.alter_column( - "route_results", - "effective_clear", - new_column_name="effective_degraded", - ) - op.alter_column( - "routes", - "clear_threshold", - new_column_name="degraded_threshold", - ) diff --git a/alembic/versions/20260712_2330_8f2a3c4d5e6f_add_route_health_monitoring.py b/alembic/versions/20260716_2043_ec40c67c8c83_add_route_health_monitoring.py similarity index 63% rename from alembic/versions/20260712_2330_8f2a3c4d5e6f_add_route_health_monitoring.py rename to alembic/versions/20260716_2043_ec40c67c8c83_add_route_health_monitoring.py index 48c6f41..c100774 100644 --- a/alembic/versions/20260712_2330_8f2a3c4d5e6f_add_route_health_monitoring.py +++ b/alembic/versions/20260716_2043_ec40c67c8c83_add_route_health_monitoring.py @@ -1,8 +1,23 @@ -"""add route health monitoring tables +"""add route health monitoring (consolidated) -Revision ID: 8f2a3c4d5e6f +Revision ID: ec40c67c8c83 Revises: 57bb65130b97 -Create Date: 2026-07-12 23:30:00.000000+00:00 +Create Date: 2026-07-16 20:43:00.000000+00:00 + +Consolidation of four formerly-separate migrations into one clean +migration that builds the final schema directly: + +* ``8f2a3c4d5e6f`` — create five route health tables + backfill + ``packet_path_hops`` from ``raw_packets.decoded``. +* ``76513f4c57e9`` — standalone ``ix_packet_path_hops_received_at`` index. +* ``71f6e01e4bf6`` — rename ``degraded_threshold`` -> ``clear_threshold`` + on ``routes`` and ``effective_degraded`` -> ``effective_clear`` on + ``route_results``. These columns are born with their final names here, + so no rename runs. +* ``c0d1e2f3a4b5`` — deduplicate ``nodes`` by ``public_key`` and re-assert + the UNIQUE index. Idempotent (no-op on a clean DB) but retained so a + restored backup with duplicate keys is repaired before route tables + reference them. Creates five tables for route health monitoring: ``routes``, ``route_nodes``, ``route_observers``, ``route_results`` and @@ -18,19 +33,43 @@ migration ``57bb65130b97``. """ +from datetime import datetime, timezone from typing import Any, Sequence, Union +from uuid import uuid4 -from alembic import op import sqlalchemy as sa +from alembic import op # revision identifiers, used by Alembic. -revision: str = "8f2a3c4d5e6f" +revision: str = "ec40c67c8c83" down_revision: Union[str, None] = "57bb65130b97" branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None _BATCH_SIZE = 1000 +# (table, column) for every FK referencing nodes.id. +# CASCADE rows must be re-pointed BEFORE the loser is deleted; SET NULL rows +# are also re-pointed (not nulled) so we keep the linkage rather than orphan it. +_REPOINT_COLUMNS = [ + # ondelete=CASCADE + ("node_tags", "node_id"), + ("event_observers", "observer_node_id"), + ("route_nodes", "node_id"), + ("route_observers", "node_id"), + ("user_profile_nodes", "node_id"), + # ondelete=SET NULL + ("advertisements", "observer_node_id"), + ("advertisements", "node_id"), + ("telemetry", "observer_node_id"), + ("telemetry", "node_id"), + ("messages", "observer_node_id"), + ("trace_paths", "observer_node_id"), + ("events_log", "observer_node_id"), + ("raw_packets", "observer_node_id"), + ("packet_path_hops", "observer_node_id"), +] + def _normalize_hash_list(value: Any) -> list[str] | None: """Frozen copy of LetsMeshNormalizer._normalize_hash_list. @@ -106,7 +145,7 @@ def upgrade() -> None: sa.Column("match_width", sa.Integer, nullable=False), sa.Column("window_hours", sa.Integer, nullable=False), sa.Column("packet_count_threshold", sa.Integer, nullable=False), - sa.Column("degraded_threshold", sa.Integer, nullable=True), + sa.Column("clear_threshold", sa.Integer, nullable=True), sa.Column("max_hop_span", sa.Integer, nullable=True), sa.Column("enabled", sa.Boolean, nullable=False), sa.Column( @@ -194,12 +233,16 @@ def upgrade() -> None: sa.Column("quality", sa.String(20), nullable=False), sa.Column("matched_count", sa.Integer, nullable=False), sa.Column("threshold", sa.Integer, nullable=False), - sa.Column("effective_degraded", sa.Integer, nullable=False), + sa.Column("effective_clear", sa.Integer, nullable=False), sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False), sa.ForeignKeyConstraint(["route_id"], ["routes.id"], ondelete="CASCADE"), - sa.UniqueConstraint("route_id", name="uq_route_results_route_id"), ) - op.create_index("ix_route_results_route_id", "route_results", ["route_id"]) + op.create_index( + "ix_route_results_route_id", + "route_results", + ["route_id"], + unique=True, + ) # --- packet_path_hops --- op.create_table( @@ -240,6 +283,11 @@ def upgrade() -> None: "packet_path_hops", ["raw_packet_id", "position"], ) + op.create_index( + "ix_packet_path_hops_received_at", + "packet_path_hops", + ["received_at"], + ) # --- Backfill packet_path_hops from raw_packets.decoded --- conn = op.get_bind() @@ -257,9 +305,6 @@ def upgrade() -> None: sa.Column("updated_at", sa.DateTime), ) - from uuid import uuid4 - from datetime import datetime, timezone - last_id: str | None = None while True: query = ( @@ -305,8 +350,107 @@ def upgrade() -> None: if rows_to_insert: conn.execute(_packet_path_hops.insert(), rows_to_insert) + # --- Deduplicate nodes by public_key (idempotent data fix) --- + # Merges duplicate nodes (winner = earliest first_seen, tiebreak + # created_at), re-pointing every FK column that references nodes.id, + # then re-asserts the UNIQUE index. No-op on a clean database. + from sqlalchemy import text + + # SQLite uses GROUP_CONCAT; Postgres uses STRING_AGG. HAVING references + # COUNT(*) directly (Postgres disallows SELECT aliases in HAVING). + if conn.dialect.name == "postgresql": + id_agg = "STRING_AGG(id::text, ',')" + else: + id_agg = "GROUP_CONCAT(id)" + + duplicates = conn.execute(text(f""" + SELECT LOWER(public_key) AS lower_pk, + {id_agg} AS ids, + COUNT(*) AS cnt + FROM nodes + GROUP BY LOWER(public_key) + HAVING COUNT(*) > 1 + """)).fetchall() + + for row in duplicates: + lower_pk = row[0] + ids = row[1].split(",") + + winner_row = conn.execute( + text(""" + SELECT id FROM nodes + WHERE LOWER(public_key) = :lower_pk + ORDER BY first_seen ASC, created_at ASC + LIMIT 1 + """), + {"lower_pk": lower_pk}, + ).fetchone() + + if not winner_row: + continue + + winner_id = winner_row[0] + loser_ids = [nid for nid in ids if nid != winner_id] + + for loser_id in loser_ids: + params = {"winner_id": winner_id, "loser_id": loser_id} + + # Re-point every FK column from loser to winner. Wrapped in + # try/except per table so a missing table on an older DB doesn't + # abort the whole migration (defensive — schema drift). + for table, column in _REPOINT_COLUMNS: + try: + conn.execute( + text( + f"UPDATE {table} SET {column} = :winner_id " + f"WHERE {column} = :loser_id" + ), + params, + ) + except Exception: + pass + + # Merge scalar fields into winner (keep best data from either row). + conn.execute( + text(""" + UPDATE nodes SET + name = COALESCE(name, (SELECT name FROM nodes WHERE id = :loser_id)), + adv_type = COALESCE(adv_type, (SELECT adv_type FROM nodes WHERE id = :loser_id)), + flags = COALESCE(flags, (SELECT flags FROM nodes WHERE id = :loser_id)), + lat = COALESCE(lat, (SELECT lat FROM nodes WHERE id = :loser_id)), + lon = COALESCE(lon, (SELECT lon FROM nodes WHERE id = :loser_id)), + last_seen = ( + SELECT MAX(v) FROM ( + SELECT last_seen AS v FROM nodes WHERE id = :winner_id + UNION ALL + SELECT last_seen AS v FROM nodes WHERE id = :loser_id + ) + ), + is_observer = ( + is_observer + OR (SELECT is_observer FROM nodes WHERE id = :loser_id) + ) + WHERE id = :winner_id + """), + params, + ) + + # Delete the loser. + conn.execute( + text("DELETE FROM nodes WHERE id = :loser_id"), + {"loser_id": loser_id}, + ) + + # Normalize any remaining case variance (defense-in-depth). + conn.execute(text("UPDATE nodes SET public_key = LOWER(public_key)")) + + # Re-assert the UNIQUE index on nodes.public_key (idempotent). + conn.execute(text("DROP INDEX IF EXISTS ix_nodes_public_key")) + conn.execute(text("CREATE UNIQUE INDEX ix_nodes_public_key ON nodes (public_key)")) + def downgrade() -> None: + # The nodes dedup is irreversible (duplicate rows were deleted). op.drop_table("packet_path_hops") op.drop_table("route_results") op.drop_table("route_observers")