Fix a tunnel-hostname race that could publish duplicate rows
DevPlace CI / test (push) Failing after 26m5s
DevPlace CI / test (push) Failing after 26m5s
tunnels.create() checked for an existing row by hostname and then inserted/updated in a separate statement, with no lock between the two. Concurrent publish calls for the same port (e.g. the editor's Ports extension re-POSTing on every onDidChangeTunnels event) could race past the check together and each insert their own row for the identical hostname. Since routers/tunnel.py resolves a tunnel by a plain find_one(hostname=...) with no ordering, whichever duplicate it happened to return decided whether a request reached the app or got a generic "no tunnel is published at this address" 404 - even while a sibling row for the same hostname was active and serving traffic. Replaced the check-then-write with a single atomic INSERT ... ON CONFLICT(hostname) DO UPDATE, which needs hostname to actually be unique: idx_tunnels_hostname is now a UNIQUE index instead of a plain one. Since existing databases likely already carry duplicate rows from this race, init_db() now runs a one-time _dedupe_tunnel_hostnames() pass (keeps the best row per hostname - active > provisioning > pending > failed, then newest) before dropping and recreating the index as unique - CREATE UNIQUE INDEX IF NOT EXISTS silently no-ops when an index with that name already exists, so the old non-unique index has to be dropped first or the constraint would never actually upgrade. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -30,6 +30,54 @@ def migrate_bug_tables_to_issue_tables() -> None:
|
|||||||
logger.info("Dropped table %s after migration", source_name)
|
logger.info("Dropped table %s after migration", source_name)
|
||||||
|
|
||||||
|
|
||||||
|
_TUNNEL_STATUS_RANK = {
|
||||||
|
"active": 4,
|
||||||
|
"provisioning": 3,
|
||||||
|
"pending": 2,
|
||||||
|
"failed": 1,
|
||||||
|
"suspended": 0,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _dedupe_tunnel_hostnames() -> None:
|
||||||
|
if "tunnels" not in db.tables:
|
||||||
|
return
|
||||||
|
table = get_table("tunnels")
|
||||||
|
if not table.has_column("hostname") or not table.has_column("uid"):
|
||||||
|
return
|
||||||
|
groups = list(
|
||||||
|
db.query(
|
||||||
|
"SELECT hostname FROM tunnels "
|
||||||
|
"WHERE hostname IS NOT NULL AND hostname != '' "
|
||||||
|
"GROUP BY hostname HAVING COUNT(*) > 1"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
for group in groups:
|
||||||
|
hostname = group["hostname"]
|
||||||
|
dupes = list(table.find(hostname=hostname))
|
||||||
|
dupes.sort(
|
||||||
|
key=lambda r: (
|
||||||
|
_TUNNEL_STATUS_RANK.get(r.get("status") or "", -1),
|
||||||
|
r.get("deleted_at") is None,
|
||||||
|
r.get("created_at") or "",
|
||||||
|
r.get("id") or 0,
|
||||||
|
),
|
||||||
|
reverse=True,
|
||||||
|
)
|
||||||
|
losers = [r["uid"] for r in dupes[1:]]
|
||||||
|
if not losers:
|
||||||
|
continue
|
||||||
|
with db:
|
||||||
|
for uid in losers:
|
||||||
|
db.query("DELETE FROM tunnels WHERE uid = :uid", uid=uid)
|
||||||
|
logger.warning(
|
||||||
|
"Removed %d duplicate tunnel row(s) for hostname %s, kept %s",
|
||||||
|
len(losers),
|
||||||
|
hostname,
|
||||||
|
dupes[0]["uid"],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def init_db():
|
def init_db():
|
||||||
tables = db.tables
|
tables = db.tables
|
||||||
_index(db, "users", "idx_users_username", ["username"])
|
_index(db, "users", "idx_users_username", ["username"])
|
||||||
@@ -785,7 +833,9 @@ def init_db():
|
|||||||
_index(db, "instances", "idx_instances_workspace", ["is_workspace", "status"])
|
_index(db, "instances", "idx_instances_workspace", ["is_workspace", "status"])
|
||||||
_index(db, "instances", "idx_instances_workspace_owner", ["workspace_owner_uid"])
|
_index(db, "instances", "idx_instances_workspace_owner", ["workspace_owner_uid"])
|
||||||
_index(db, "instances", "idx_instances_tunnel_name", ["tunnel_name"])
|
_index(db, "instances", "idx_instances_tunnel_name", ["tunnel_name"])
|
||||||
_index(db, "tunnels", "idx_tunnels_hostname", ["hostname"])
|
_dedupe_tunnel_hostnames()
|
||||||
|
_drop_index(db, "idx_tunnels_hostname")
|
||||||
|
_index(db, "tunnels", "idx_tunnels_hostname", ["hostname"], unique=True)
|
||||||
_index(db, "tunnels", "idx_tunnels_instance", ["instance_uid"])
|
_index(db, "tunnels", "idx_tunnels_instance", ["instance_uid"])
|
||||||
_index(db, "tunnels", "idx_tunnels_user", ["user_uid"])
|
_index(db, "tunnels", "idx_tunnels_user", ["user_uid"])
|
||||||
_index(db, "tunnels", "idx_tunnels_state", ["desired_state", "status"])
|
_index(db, "tunnels", "idx_tunnels_state", ["desired_state", "status"])
|
||||||
|
|||||||
@@ -69,52 +69,50 @@ def create(
|
|||||||
return None
|
return None
|
||||||
hostname = naming.hostname_for(name, container_port)
|
hostname = naming.hostname_for(name, container_port)
|
||||||
table = _table()
|
table = _table()
|
||||||
revived = table.find_one(hostname=hostname)
|
|
||||||
stamp = _now()
|
stamp = _now()
|
||||||
if revived:
|
with db:
|
||||||
changes = {
|
db.query(
|
||||||
"uid": revived["uid"],
|
"""
|
||||||
"instance_uid": instance["uid"],
|
INSERT INTO tunnels (
|
||||||
"project_uid": instance.get("project_uid", ""),
|
uid, instance_uid, project_uid, user_uid, hostname, label,
|
||||||
"user_uid": user_uid,
|
container_port, desired_state, status, cert_status,
|
||||||
"label": label or f"port {container_port}",
|
cert_checked_at, request_count, bytes_out, last_request_at,
|
||||||
"container_port": container_port,
|
last_error, last_synced_at, created_at, updated_at,
|
||||||
"desired_state": "present",
|
deleted_at, deleted_by
|
||||||
"updated_at": stamp,
|
) VALUES (
|
||||||
"deleted_at": None,
|
:uid, :instance_uid, :project_uid, :user_uid, :hostname, :label,
|
||||||
"deleted_by": None,
|
:container_port, 'present', :status_pending, '', '', 0, 0, '', '',
|
||||||
}
|
'', :stamp, :stamp, NULL, NULL
|
||||||
if not keeps_certificate(revived):
|
)
|
||||||
changes["status"] = STATUS_PENDING
|
ON CONFLICT(hostname) DO UPDATE SET
|
||||||
changes["last_error"] = ""
|
instance_uid = excluded.instance_uid,
|
||||||
table.update(changes, ["uid"])
|
project_uid = excluded.project_uid,
|
||||||
return table.find_one(uid=revived["uid"])
|
user_uid = excluded.user_uid,
|
||||||
uid = generate_uid()
|
label = excluded.label,
|
||||||
table.insert(
|
container_port = excluded.container_port,
|
||||||
{
|
desired_state = 'present',
|
||||||
"uid": uid,
|
updated_at = excluded.updated_at,
|
||||||
"instance_uid": instance["uid"],
|
deleted_at = NULL,
|
||||||
"project_uid": instance.get("project_uid", ""),
|
deleted_by = NULL,
|
||||||
"user_uid": user_uid,
|
status = CASE
|
||||||
"hostname": hostname,
|
WHEN tunnels.deleted_at IS NULL AND tunnels.status = :status_active
|
||||||
"label": label or f"port {container_port}",
|
THEN tunnels.status ELSE :status_pending END,
|
||||||
"container_port": container_port,
|
last_error = CASE
|
||||||
"desired_state": "present",
|
WHEN tunnels.deleted_at IS NULL AND tunnels.status = :status_active
|
||||||
"status": STATUS_PENDING,
|
THEN tunnels.last_error ELSE '' END
|
||||||
"cert_status": "",
|
""",
|
||||||
"cert_checked_at": "",
|
uid=generate_uid(),
|
||||||
"request_count": 0,
|
instance_uid=instance["uid"],
|
||||||
"bytes_out": 0,
|
project_uid=instance.get("project_uid", ""),
|
||||||
"last_request_at": "",
|
user_uid=user_uid,
|
||||||
"last_error": "",
|
hostname=hostname,
|
||||||
"last_synced_at": "",
|
label=label or f"port {container_port}",
|
||||||
"created_at": stamp,
|
container_port=container_port,
|
||||||
"updated_at": stamp,
|
stamp=stamp,
|
||||||
"deleted_at": None,
|
status_pending=STATUS_PENDING,
|
||||||
"deleted_by": None,
|
status_active=STATUS_ACTIVE,
|
||||||
}
|
)
|
||||||
)
|
return table.find_one(hostname=hostname)
|
||||||
return table.find_one(uid=uid)
|
|
||||||
|
|
||||||
|
|
||||||
def awaiting_certificate() -> list[dict]:
|
def awaiting_certificate() -> list[dict]:
|
||||||
|
|||||||
Reference in New Issue
Block a user