fix: enforce durable migration ownership and readiness

hub_core/migrations/env.py

@@ -5,6 +5,7 @@ from alembic import context
 from sqlalchemy import engine_from_config, pool
 
 from hub_core.models import Base
+from hub_core.migrations.roles import migration_role_statement
 
 config = context.config
 
@@ -38,6 +39,8 @@ def run_migrations_online() -> None:
         poolclass=pool.NullPool,
     )
     with connectable.connect() as connection:
+        if statement := migration_role_statement(os.environ.get("HUB_CORE_MIGRATION_ROLE")):
+            connection.exec_driver_sql(statement)
         context.configure(connection=connection, target_metadata=target_metadata)
         with context.begin_transaction():
             context.run_migrations()

hub_core/migrations/roles.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import re
+
+
+def migration_role_statement(role: str | None) -> str | None:
+    """Return a safely quoted SET ROLE statement for an admitted owner role."""
+    if not role:
+        return None
+    if not re.fullmatch(r"[a-z_][a-z0-9_]{0,62}", role):
+        raise ValueError("HUB_CORE_MIGRATION_ROLE is not a safe PostgreSQL role name")
+    return f'SET ROLE "{role}"'

hub_core/runtime/postgres_store.py

@@ -22,6 +22,8 @@ from hub_core.runtime.models import (
     RegistryRegistration,
 )
 from hub_core.runtime.tables import (
+    compat_api_keys,
+    compat_hubs,
     runtime_audit_ledger,
     runtime_interaction_events,
     runtime_messages,
@@ -47,6 +49,13 @@ class PostgresPortStore:
         try:
             async with self.engine.connect() as connection:
                 await connection.execute(sa.text("SELECT 1"))
+                # A live connection alone can hide a mis-granted runtime lease.
+                # Exercise the durable port, compatibility, and authorization
+                # tables that protected traffic actually depends on.
+                for table in (runtime_audit_ledger, compat_hubs, compat_api_keys):
+                    await connection.execute(
+                        sa.select(sa.literal(1)).select_from(table).limit(1)
+                    )
         except Exception:
             return {"database": "unavailable"}
         return {"database": "ok"}