adam/meshcore-hub
1
0
Fork 0
mirror of https://github.com/ipnet-mesh/meshcore-hub.git synced 2026-08-07 17:33:20 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: consolidate route health migrations into one clean migration

Browse Source 
Squashes four formerly-separate tail migrations into a single migration
(ec40c67c8c83) that builds the final schema directly off 57bb65130b97:

- 8f2a3c4d5e6f route health tables + packet_path_hops backfill
- c0d1e2f3a4b5 deduplicate nodes by public_key (idempotent data fix)
- 76513f4c57e9 ix_packet_path_hops_received_at index
- 71f6e01e4bf6 rename degraded_threshold -> clear_threshold

The rename runs as a no-op: routes.clear_threshold and
route_results.effective_clear are born with their final names. The
received_at index is created with the table, and the node dedup is
preserved verbatim (no-op on a clean DB).

Also drops the redundant uq_route_results_route_id unique constraint:
route_results.route_id now has a single unique index matching the model,
eliminating a pre-existing autogenerate drift.

Net schema is identical to running all four migrations. Verified with a
fresh alembic upgrade head + alembic check (no drift) on SQLite.
This commit is contained in:
Louis King
2026-07-16 22:33:13 +01:00
parent ad07c1521c
commit 30f18561bf
4 changed files with 156 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alembic/versions/20260713_0100_c0d1e2f3a4b5_deduplicate_nodes_public_key.py
-174
View File
@@ -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
alembic/versions/20260715_2100_76513f4c57e9_add_packet_path_hops_received_at_index.py
-32
View File
@@ -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",
)
alembic/versions/20260715_2300_71f6e01e4bf6_rename_degraded_to_clear_threshold.py
-39
View File
@@ -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",
)
alembic/versions/20260712_2330_8f2a3c4d5e6f_add_route_health_monitoring.py → alembic/versions/20260716_2043_ec40c67c8c83_add_route_health_monitoring.py
+156 -12
View File
@@ -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")