Skip to content

Commit a627645

Browse files
TonisOrmissonTonisOrmisson
committed
Serialize MySQL-wire catalog initialization
1 parent 04f86fa commit a627645

1 file changed

Lines changed: 183 additions & 150 deletions

File tree

‎src/openstatspec/sql/wide.py‎

Lines changed: 183 additions & 150 deletions
Original file line numberDiff line numberDiff line change
@@ -1622,6 +1622,36 @@ def dolt_state_snapshot(
16221622
}
16231623

16241624

1625+
@contextmanager
1626+
def _catalog_initialization_serialization(
1627+
connection: Any, *, profile_name: str,
1628+
):
1629+
"""Serialize non-transactional catalog DDL on one database server."""
1630+
if profile_name not in MYSQL_WIRE_PROFILES:
1631+
yield
1632+
return
1633+
lock_name = "openstatspec.catalog-initialize.v1"
1634+
acquired = connection.execute(
1635+
text("SELECT GET_LOCK(:lock_name, 30)"), {"lock_name": lock_name},
1636+
).scalar_one()
1637+
connection.commit()
1638+
if acquired != 1:
1639+
raise UnsupportedOperationError(
1640+
"Could not acquire the catalog initialization lock."
1641+
)
1642+
try:
1643+
yield
1644+
finally:
1645+
try:
1646+
connection.execute(
1647+
text("SELECT RELEASE_LOCK(:lock_name)"),
1648+
{"lock_name": lock_name},
1649+
)
1650+
connection.commit()
1651+
except Exception:
1652+
connection.invalidate()
1653+
1654+
16251655
def initialize_wide_catalog(
16261656
*,
16271657
database_url: str,
@@ -1645,48 +1675,99 @@ def initialize_wide_catalog(
16451675
legacy, normative = _catalog_layout(metadata)
16461676
datasets, variables, multiple_response = legacy[:3]
16471677
with engine.connect() as connection:
1648-
state = _catalog_state(connection, normative, legacy)
1649-
if state not in {"absent", "verified", "migration_required"}:
1650-
raise UnsupportedOperationError(
1651-
f"The selected database catalog is {state}; initialization is not permitted."
1652-
)
1653-
before_tables, before_columns = _catalog_snapshot(connection)
1654-
pre_dolt_state = _capture_dolt_state(
1655-
connection, profile_name=profile.name, audit_relations=set(),
1656-
)
1657-
_require_dolt_working_set_binding(
1658-
pre_dolt_state, active, phase="catalog initialization preflight",
1659-
)
1660-
connection.rollback()
1661-
try:
1662-
with connection.begin():
1663-
create_normative_catalog(connection, normative)
1664-
metadata.create_all(connection, tables=list(legacy))
1665-
_migrate_catalog_columns(
1666-
connection, datasets, variables, multiple_response,
1678+
with _catalog_initialization_serialization(
1679+
connection, profile_name=profile.name,
1680+
):
1681+
state = _catalog_state(connection, normative, legacy)
1682+
if state not in {"absent", "verified", "migration_required"}:
1683+
raise UnsupportedOperationError(
1684+
f"The selected database catalog is {state}; initialization is not permitted."
16671685
)
1668-
_require_verified_catalog(connection, normative, legacy)
1669-
post_dolt_state = _capture_dolt_state(
1686+
before_tables, before_columns = _catalog_snapshot(connection)
1687+
pre_dolt_state = _capture_dolt_state(
16701688
connection, profile_name=profile.name, audit_relations=set(),
16711689
)
16721690
_require_dolt_working_set_binding(
1673-
post_dolt_state, active, phase="catalog initialization completion",
1691+
pre_dolt_state, active, phase="catalog initialization preflight",
16741692
)
1675-
_require_dolt_success_identity(
1676-
pre_dolt_state, post_dolt_state, phase="catalog initialization",
1677-
)
1678-
except Exception as install_error:
1693+
connection.rollback()
16791694
try:
16801695
with connection.begin():
1681-
_compensate_catalog_initialization(
1682-
connection, metadata=metadata, before_tables=before_tables,
1683-
before_columns=before_columns, normative=normative,
1684-
legacy=legacy,
1696+
create_normative_catalog(connection, normative)
1697+
metadata.create_all(connection, tables=list(legacy))
1698+
_migrate_catalog_columns(
1699+
connection, datasets, variables, multiple_response,
16851700
)
1686-
except Exception as cleanup_error:
1687-
inventory = _catalog_residual_inventory(
1688-
engine, before_tables=before_tables, before_columns=before_columns,
1701+
_require_verified_catalog(connection, normative, legacy)
1702+
post_dolt_state = _capture_dolt_state(
1703+
connection, profile_name=profile.name, audit_relations=set(),
1704+
)
1705+
_require_dolt_working_set_binding(
1706+
post_dolt_state, active, phase="catalog initialization completion",
1707+
)
1708+
_require_dolt_success_identity(
1709+
pre_dolt_state, post_dolt_state, phase="catalog initialization",
16891710
)
1711+
except Exception as install_error:
1712+
try:
1713+
with connection.begin():
1714+
_compensate_catalog_initialization(
1715+
connection, metadata=metadata, before_tables=before_tables,
1716+
before_columns=before_columns, normative=normative,
1717+
legacy=legacy,
1718+
)
1719+
except Exception as cleanup_error:
1720+
inventory = _catalog_residual_inventory(
1721+
engine, before_tables=before_tables, before_columns=before_columns,
1722+
)
1723+
try:
1724+
after_dolt_state = _capture_dolt_state(
1725+
connection, profile_name=profile.name, audit_relations=set(),
1726+
)
1727+
dolt_boundary = _dolt_failure_boundary_evidence(
1728+
pre_dolt_state, after_dolt_state,
1729+
)
1730+
except Exception as snapshot_error:
1731+
dolt_boundary = {
1732+
"applicable": profile.name == "dolt",
1733+
"verified": False,
1734+
"snapshot_fault": _safe_error_identity(
1735+
snapshot_error, phase="post_catalog_cleanup_dolt_state_capture",
1736+
),
1737+
}
1738+
raise ImportRecoveryError(
1739+
"cleanup_failed",
1740+
"Catalog initialization failed and its DDL compensation also failed.",
1741+
details={
1742+
"subcode": "catalog_install_cleanup_failed",
1743+
"original_cause": _safe_error_identity(
1744+
install_error, phase="catalog_initialization",
1745+
),
1746+
"cleanup_fault": _safe_error_identity(
1747+
cleanup_error, phase="catalog_compensation",
1748+
),
1749+
"residual_object_inventory": inventory,
1750+
"deterministic_recovery_evidence": {
1751+
"procedure_id": "openstatspec.catalog-init-compensation.v1",
1752+
"action_id": _canonical_sha256({
1753+
"namespace": active["catalog_binding"]["namespace"],
1754+
"before_tables": sorted(before_tables),
1755+
}),
1756+
"targets": {
1757+
"namespace": active["catalog_binding"]["namespace"],
1758+
"catalog_relations": sorted(
1759+
table.name for table in metadata.tables.values()
1760+
),
1761+
},
1762+
"residual_inventory_sha256": _canonical_sha256(inventory),
1763+
"cleanup_attempted": True,
1764+
"cleanup_succeeded": False,
1765+
"preexisting_unverified_catalog_mutation_forbidden": True,
1766+
"dolt_failure_boundary": dolt_boundary,
1767+
},
1768+
"success_forbidden": True,
1769+
},
1770+
) from cleanup_error
16901771
try:
16911772
after_dolt_state = _capture_dolt_state(
16921773
connection, profile_name=profile.name, audit_relations=set(),
@@ -1695,125 +1776,77 @@ def initialize_wide_catalog(
16951776
pre_dolt_state, after_dolt_state,
16961777
)
16971778
except Exception as snapshot_error:
1698-
dolt_boundary = {
1699-
"applicable": profile.name == "dolt",
1700-
"verified": False,
1701-
"snapshot_fault": _safe_error_identity(
1702-
snapshot_error, phase="post_catalog_cleanup_dolt_state_capture",
1703-
),
1779+
inventory = _catalog_residual_inventory(
1780+
engine, before_tables=before_tables, before_columns=before_columns,
1781+
)
1782+
recovery = {
1783+
"procedure_id": "openstatspec.dolt-failure-boundary.v1",
1784+
"action_id": _canonical_sha256({
1785+
"namespace": active["catalog_binding"]["namespace"],
1786+
"before_tables": sorted(before_tables),
1787+
}),
1788+
"targets": {
1789+
"namespace": active["catalog_binding"]["namespace"],
1790+
"catalog_relations": sorted(metadata.tables),
1791+
},
1792+
"residual_inventory_sha256": _canonical_sha256(inventory),
1793+
"dolt_failure_boundary": {
1794+
"applicable": profile.name == "dolt",
1795+
"verified": False,
1796+
},
17041797
}
1705-
raise ImportRecoveryError(
1706-
"cleanup_failed",
1707-
"Catalog initialization failed and its DDL compensation also failed.",
1708-
details={
1709-
"subcode": "catalog_install_cleanup_failed",
1710-
"original_cause": _safe_error_identity(
1711-
install_error, phase="catalog_initialization",
1712-
),
1713-
"cleanup_fault": _safe_error_identity(
1714-
cleanup_error, phase="catalog_compensation",
1715-
),
1716-
"residual_object_inventory": inventory,
1717-
"deterministic_recovery_evidence": {
1718-
"procedure_id": "openstatspec.catalog-init-compensation.v1",
1719-
"action_id": _canonical_sha256({
1720-
"namespace": active["catalog_binding"]["namespace"],
1721-
"before_tables": sorted(before_tables),
1722-
}),
1723-
"targets": {
1724-
"namespace": active["catalog_binding"]["namespace"],
1725-
"catalog_relations": sorted(
1726-
table.name for table in metadata.tables.values()
1727-
),
1728-
},
1729-
"residual_inventory_sha256": _canonical_sha256(inventory),
1730-
"cleanup_attempted": True,
1731-
"cleanup_succeeded": False,
1732-
"preexisting_unverified_catalog_mutation_forbidden": True,
1733-
"dolt_failure_boundary": dolt_boundary,
1798+
raise ImportRecoveryError(
1799+
"cleanup_failed",
1800+
"Catalog compensation completed but Dolt state could not be verified.",
1801+
details={
1802+
"subcode": "dolt_state_capture_failed",
1803+
"original_cause": _safe_error_identity(
1804+
install_error, phase="catalog_initialization",
1805+
),
1806+
"cleanup_fault": _safe_error_identity(
1807+
snapshot_error, phase="post_catalog_cleanup_dolt_state_capture",
1808+
),
1809+
"residual_object_inventory": inventory,
1810+
"deterministic_recovery_evidence": recovery,
1811+
"success_forbidden": True,
17341812
},
1735-
"success_forbidden": True,
1736-
},
1737-
) from cleanup_error
1738-
try:
1739-
after_dolt_state = _capture_dolt_state(
1740-
connection, profile_name=profile.name, audit_relations=set(),
1741-
)
1742-
dolt_boundary = _dolt_failure_boundary_evidence(
1743-
pre_dolt_state, after_dolt_state,
1744-
)
1745-
except Exception as snapshot_error:
1746-
inventory = _catalog_residual_inventory(
1747-
engine, before_tables=before_tables, before_columns=before_columns,
1748-
)
1749-
recovery = {
1750-
"procedure_id": "openstatspec.dolt-failure-boundary.v1",
1751-
"action_id": _canonical_sha256({
1752-
"namespace": active["catalog_binding"]["namespace"],
1753-
"before_tables": sorted(before_tables),
1754-
}),
1755-
"targets": {
1756-
"namespace": active["catalog_binding"]["namespace"],
1757-
"catalog_relations": sorted(metadata.tables),
1758-
},
1759-
"residual_inventory_sha256": _canonical_sha256(inventory),
1760-
"dolt_failure_boundary": {
1761-
"applicable": profile.name == "dolt",
1762-
"verified": False,
1763-
},
1764-
}
1765-
raise ImportRecoveryError(
1766-
"cleanup_failed",
1767-
"Catalog compensation completed but Dolt state could not be verified.",
1768-
details={
1769-
"subcode": "dolt_state_capture_failed",
1770-
"original_cause": _safe_error_identity(
1771-
install_error, phase="catalog_initialization",
1772-
),
1773-
"cleanup_fault": _safe_error_identity(
1774-
snapshot_error, phase="post_catalog_cleanup_dolt_state_capture",
1775-
),
1776-
"residual_object_inventory": inventory,
1777-
"deterministic_recovery_evidence": recovery,
1778-
"success_forbidden": True,
1779-
},
1780-
) from snapshot_error
1781-
if dolt_boundary.get("applicable") and not dolt_boundary.get("verified"):
1782-
inventory = _catalog_residual_inventory(
1783-
engine, before_tables=before_tables, before_columns=before_columns,
1784-
)
1785-
recovery = {
1786-
"procedure_id": "openstatspec.dolt-failure-boundary.v1",
1787-
"action_id": _canonical_sha256({
1788-
"namespace": active["catalog_binding"]["namespace"],
1789-
"before_tables": sorted(before_tables),
1790-
}),
1791-
"targets": {
1792-
"namespace": active["catalog_binding"]["namespace"],
1793-
"catalog_relations": sorted(metadata.tables),
1794-
},
1795-
"residual_inventory_sha256": _canonical_sha256(inventory),
1796-
"dolt_failure_boundary": dolt_boundary,
1797-
}
1798-
raise ImportRecoveryError(
1799-
"cleanup_failed",
1800-
"Catalog compensation did not preserve Dolt failure-boundary invariants.",
1801-
details={
1802-
"subcode": "dolt_state_invariant_failed",
1803-
"original_cause": _safe_error_identity(
1804-
install_error, phase="catalog_initialization",
1805-
),
1806-
"cleanup_fault": _verification_fault_identity(
1807-
"dolt_state_invariant_failed",
1808-
phase="post_catalog_cleanup_dolt_state_verification",
1809-
evidence=dolt_boundary,
1810-
),
1811-
"residual_object_inventory": inventory,
1812-
"deterministic_recovery_evidence": recovery,
1813-
"success_forbidden": True,
1814-
},
1815-
) from install_error
1816-
raise
1813+
) from snapshot_error
1814+
if dolt_boundary.get("applicable") and not dolt_boundary.get("verified"):
1815+
inventory = _catalog_residual_inventory(
1816+
engine, before_tables=before_tables, before_columns=before_columns,
1817+
)
1818+
recovery = {
1819+
"procedure_id": "openstatspec.dolt-failure-boundary.v1",
1820+
"action_id": _canonical_sha256({
1821+
"namespace": active["catalog_binding"]["namespace"],
1822+
"before_tables": sorted(before_tables),
1823+
}),
1824+
"targets": {
1825+
"namespace": active["catalog_binding"]["namespace"],
1826+
"catalog_relations": sorted(metadata.tables),
1827+
},
1828+
"residual_inventory_sha256": _canonical_sha256(inventory),
1829+
"dolt_failure_boundary": dolt_boundary,
1830+
}
1831+
raise ImportRecoveryError(
1832+
"cleanup_failed",
1833+
"Catalog compensation did not preserve Dolt failure-boundary invariants.",
1834+
details={
1835+
"subcode": "dolt_state_invariant_failed",
1836+
"original_cause": _safe_error_identity(
1837+
install_error, phase="catalog_initialization",
1838+
),
1839+
"cleanup_fault": _verification_fault_identity(
1840+
"dolt_state_invariant_failed",
1841+
phase="post_catalog_cleanup_dolt_state_verification",
1842+
evidence=dolt_boundary,
1843+
),
1844+
"residual_object_inventory": inventory,
1845+
"deterministic_recovery_evidence": recovery,
1846+
"success_forbidden": True,
1847+
},
1848+
) from install_error
1849+
raise
18171850
return {
18181851
"profile": profile.name,
18191852
"server_version": active["server_version"],

0 commit comments

Comments
 (0)