367 lines
19 KiB
Python
367 lines
19 KiB
Python
# #region Tooling.LabReset.Database [C:5] [TYPE Module] [SEMANTICS maintenance,postgresql,ownership,fk]
|
|
# @BRIEF Lock, inventory and delete only reviewed exact laboratory primary keys.
|
|
# @PRE Both applications stopped; explicit connection; no other client sessions; READ COMMITTED.
|
|
# @INVARIANT All database locks and reference probes precede the storage root lock; preserved rows never become delete targets.
|
|
from collections import defaultdict
|
|
from contextlib import contextmanager
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import MetaData, and_, create_engine, inspect, select, text
|
|
from sqlalchemy.pool import NullPool
|
|
from sqlalchemy.sql.elements import quoted_name
|
|
|
|
from dashboard_testing_lab_reset_archive import digest, encode, require
|
|
|
|
|
|
# #region Tooling.LabReset.Database.PrimaryKey [C:2] [TYPE Function]
|
|
def primary_key(row, columns):
|
|
result = {name: row[name] for name in columns}
|
|
return {key: str(value) if isinstance(value, UUID) else value for key, value in result.items()}
|
|
# #endregion Tooling.LabReset.Database.PrimaryKey
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Node [C:1] [TYPE Function]
|
|
def node(table, row, columns):
|
|
return table, digest(primary_key(row, columns))
|
|
# #endregion Tooling.LabReset.Database.Node
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Order [C:3] [TYPE Function]
|
|
# @POST Child-before-parent row order includes self-FKs; every genuine cycle refuses.
|
|
def child_first(nodes, edges):
|
|
remaining = set(nodes)
|
|
parents = defaultdict(set)
|
|
children = defaultdict(set)
|
|
for child, parent in edges:
|
|
require(child in remaining and parent in remaining, "LAB_RESET_FK_PLAN_INVALID")
|
|
parents[child].add(parent)
|
|
children[parent].add(child)
|
|
result = []
|
|
ready = sorted(item for item in remaining if not children[item])
|
|
while ready:
|
|
current = ready.pop(0)
|
|
remaining.remove(current)
|
|
result.append(current)
|
|
for parent in sorted(parents[current]):
|
|
children[parent].discard(current)
|
|
if not children[parent] and parent in remaining and parent not in ready:
|
|
ready.append(parent)
|
|
ready.sort()
|
|
require(not remaining, "LAB_RESET_FK_CYCLE")
|
|
return result
|
|
# #endregion Tooling.LabReset.Database.Order
|
|
|
|
|
|
# #region Tooling.LabReset.Database.CatalogOrder [C:3] [TYPE Function]
|
|
# @POST Ignore only the reviewed deferred head.current_revision_id edge for ordering; keep it in FK ownership checks.
|
|
def catalog_child_first(rows, keys, selected, foreign_keys):
|
|
edges = foreign_edges(rows, keys, selected, foreign_keys)
|
|
heads = [row for row in rows.get("baseline_catalog_heads", [])
|
|
if node("baseline_catalog_heads", row, keys["baseline_catalog_heads"]) in selected]
|
|
deferred = set()
|
|
if heads:
|
|
require(len(heads) == 1, "LAB_RESET_CATALOG_SCOPE_CONFLICT")
|
|
matches = [fk for fk in foreign_keys.get("baseline_catalog_heads", [])
|
|
if fk.get("constrained_columns") == ["id", "current_revision_id"]
|
|
and fk.get("referred_table") == "baseline_catalog_revisions"
|
|
and fk.get("referred_columns") == ["catalog_id", "id"]]
|
|
require(len(matches) == 1, "LAB_RESET_CATALOG_FK_SCHEMA_CONFLICT")
|
|
options = matches[0].get("options", {})
|
|
require(options.get("deferrable") is True
|
|
and str(options.get("initially", "")).upper() == "DEFERRED",
|
|
"LAB_RESET_CATALOG_FK_SCHEMA_CONFLICT")
|
|
head = heads[0]
|
|
revisions = [row for row in rows.get("baseline_catalog_revisions", [])
|
|
if row.get("catalog_id") == head.get("id")
|
|
and row.get("id") == head.get("current_revision_id")]
|
|
require(len(revisions) == 1, "LAB_RESET_CATALOG_POINTER_CONFLICT")
|
|
parent = revisions[0]
|
|
head_node = node("baseline_catalog_heads", head, keys["baseline_catalog_heads"])
|
|
revision_node = node("baseline_catalog_revisions", parent, keys["baseline_catalog_revisions"])
|
|
if head_node in selected:
|
|
require(revision_node in selected, "LAB_RESET_CATALOG_POINTER_CONFLICT")
|
|
deferred.add((head_node, revision_node))
|
|
require(deferred.issubset(set(edges)), "LAB_RESET_CATALOG_FK_SCHEMA_CONFLICT")
|
|
return child_first(selected, [edge for edge in edges if edge not in deferred])
|
|
# #endregion Tooling.LabReset.Database.CatalogOrder
|
|
|
|
|
|
# #region Tooling.LabReset.Database.ForeignEdges [C:4] [TYPE Function]
|
|
# @POST Actual preserved-child references refuse even for CASCADE/SET NULL/deferrable constraints.
|
|
def foreign_edges(rows, keys, selected, foreign_keys):
|
|
edges = []
|
|
for table, fks in foreign_keys.items():
|
|
for fk in fks:
|
|
require(fk.get("referred_schema") in (None, "public"), "LAB_RESET_FOREIGN_SCHEMA")
|
|
parent = fk["referred_table"]
|
|
require(parent in rows, "LAB_RESET_FOREIGN_TABLE")
|
|
lookup = {tuple(row[col] for col in fk["referred_columns"]): row for row in rows[parent]}
|
|
for row in rows[table]:
|
|
values = tuple(row[col] for col in fk["constrained_columns"])
|
|
if any(value is None for value in values) or values not in lookup:
|
|
continue
|
|
target = node(parent, lookup[values], keys[parent])
|
|
if target not in selected:
|
|
continue
|
|
child = node(table, row, keys[table])
|
|
require(child in selected, "LAB_RESET_PRESERVED_CHILD_REFERENCE")
|
|
edges.append((child, target))
|
|
return edges
|
|
# #endregion Tooling.LabReset.Database.ForeignEdges
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Quiescence [C:2] [TYPE Function]
|
|
def no_foreign_sessions(connection):
|
|
count = connection.execute(text("SELECT count(*) FROM pg_stat_activity "
|
|
"WHERE datname=current_database() AND pid<>pg_backend_pid()")).scalar_one()
|
|
require(count == 0, "LAB_RESET_FOREIGN_DATABASE_SESSION")
|
|
# #endregion Tooling.LabReset.Database.Quiescence
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Identity [C:2] [TYPE Function]
|
|
# @INVARIANT Use the producer's native inet address carrier; text casts change the reviewed digest representation.
|
|
def identity(connection, expected):
|
|
row = connection.execute(text("SELECT current_database() AS database, "
|
|
"inet_server_addr() AS address, inet_server_port() AS port, "
|
|
"(SELECT oid FROM pg_database WHERE datname=current_database()) AS oid")).mappings().one()
|
|
require(digest(dict(row)) == expected, "LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
# #endregion Tooling.LabReset.Database.Identity
|
|
|
|
|
|
# #region Tooling.LabReset.Database.TableNames [C:2] [TYPE Function]
|
|
def table_names(connection, scope):
|
|
expected = {item["table"] for item in scope["database_tables"]} | set(scope["protected_tables"])
|
|
inspector = inspect(connection)
|
|
require(set(inspector.get_table_names(schema="public")) == expected,
|
|
"LAB_RESET_TABLE_SET_CHANGED")
|
|
require(not inspector.get_view_names(schema="public"), "LAB_RESET_UNREVIEWED_VIEW")
|
|
return sorted(expected)
|
|
# #endregion Tooling.LabReset.Database.TableNames
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Lock [C:3] [TYPE Function]
|
|
# @POST Every known table is write-locked in deterministic order before any root lock; no catalog/global lock.
|
|
def lock_tables(connection, scope):
|
|
names = table_names(connection, scope)
|
|
connection.execute(text("SET LOCAL lock_timeout = '5s'"))
|
|
connection.execute(text("SET LOCAL statement_timeout = '60s'"))
|
|
quote = connection.dialect.identifier_preparer.quote
|
|
for name in names:
|
|
connection.execute(text(f"LOCK TABLE public.{quote(name)} IN SHARE ROW EXCLUSIVE MODE"))
|
|
no_foreign_sessions(connection)
|
|
require(table_names(connection, scope) == names, "LAB_RESET_TABLE_SET_CHANGED")
|
|
# #endregion Tooling.LabReset.Database.Lock
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Transaction [C:3] [TYPE Function]
|
|
@contextmanager
|
|
def locked_database(url, scope, expected_identity):
|
|
engine = create_engine(url, poolclass=NullPool, echo=False, hide_parameters=True)
|
|
try:
|
|
require(engine.dialect.name == "postgresql", "LAB_RESET_POSTGRESQL_REQUIRED")
|
|
with engine.connect().execution_options(isolation_level="READ COMMITTED") as connection:
|
|
with connection.begin():
|
|
identity(connection, expected_identity)
|
|
no_foreign_sessions(connection)
|
|
lock_tables(connection, scope)
|
|
yield connection
|
|
finally:
|
|
engine.dispose()
|
|
# #endregion Tooling.LabReset.Database.Transaction
|
|
|
|
|
|
# #region Tooling.LabReset.Database.RowFields [C:3] [TYPE Function]
|
|
# @PRE Exact reflected columns and only plain str/SQLAlchemy quoted_name keys; no duplicate/foreign/missing field.
|
|
# @POST Plain identifier keys preserve every native value by identity; nested values and recursive codec remain unchanged.
|
|
def row_fields(row, table):
|
|
identifiers = (str, quoted_name)
|
|
names = [column.name for column in table.columns]
|
|
require(all(type(name) in identifiers for name in names), "LAB_RESET_ROW_FIELD_CONFLICT")
|
|
expected = [str(name) for name in names]
|
|
items = list(row.items())
|
|
require(len(set(expected)) == len(expected) == len(items), "LAB_RESET_ROW_FIELD_CONFLICT")
|
|
result = {}
|
|
for key, value in items:
|
|
require(type(key) in identifiers, "LAB_RESET_ROW_FIELD_CONFLICT")
|
|
name = str(key)
|
|
require(name in expected and name not in result, "LAB_RESET_ROW_FIELD_CONFLICT")
|
|
result[name] = value
|
|
require(set(result) == set(expected), "LAB_RESET_ROW_FIELD_CONFLICT")
|
|
return result
|
|
# #endregion Tooling.LabReset.Database.RowFields
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Reflection [C:3] [TYPE Function]
|
|
def reflect(connection, scope):
|
|
metadata = MetaData(schema="public")
|
|
metadata.reflect(bind=connection, only=table_names(connection, scope), resolve_fks=False)
|
|
tables = {table.name: table for table in metadata.tables.values()}
|
|
inspector = inspect(connection)
|
|
keys = {name: [column.name for column in table.primary_key] for name, table in tables.items()}
|
|
require(all(keys.values()), "LAB_RESET_PRIMARY_KEY_REQUIRED")
|
|
fks = {name: inspector.get_foreign_keys(name, schema="public") for name in tables}
|
|
rows = {name: [row_fields(row, table) for row in connection.execute(select(table)).mappings()]
|
|
for name, table in tables.items()}
|
|
return tables, keys, fks, rows
|
|
# #endregion Tooling.LabReset.Database.Reflection
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Selected [C:3] [TYPE Function]
|
|
def selected_nodes(scope, rows, keys):
|
|
selected = set()
|
|
for item in scope["database_tables"]:
|
|
name = item["table"]
|
|
require(keys[name] == item["primary_key_columns"], "LAB_RESET_PRIMARY_KEY_CHANGED")
|
|
expected = item["selected_primary_keys"] + item["preserve_primary_keys"]
|
|
actual = [primary_key(row, keys[name]) for row in rows[name]]
|
|
require(len(expected) == len(set(map(digest, expected))), "LAB_RESET_ALLOWLIST_INVALID")
|
|
require(len(actual) == item["total"] and sorted(map(digest, actual)) == sorted(map(digest, expected)),
|
|
"LAB_RESET_PRIMARY_KEY_SET_CHANGED")
|
|
require(len(item["selected_primary_keys"]) == item["selected_count"]
|
|
and len(item["preserve_primary_keys"]) == item["preserved_count"],
|
|
"LAB_RESET_ALLOWLIST_INVALID")
|
|
for key in item["selected_primary_keys"]:
|
|
selected.add((name, digest(key)))
|
|
return selected
|
|
# #endregion Tooling.LabReset.Database.Selected
|
|
|
|
|
|
# #region Tooling.LabReset.Database.ModuleRunOwners [C:3] [TYPE Function]
|
|
# @POST Revalidate the producer's exact run PK sets and AgentRun intents; never discover or expand selection.
|
|
def module_run_owners(scope, rows, keys, selected):
|
|
for table, field in (("agent_runs", "module_agent_run_ids"), ("scenario_runs", "module_scenario_run_ids")):
|
|
expected = scope[field]
|
|
chosen = [row for row in rows[table] if node(table, row, keys[table]) in selected]
|
|
require(len(expected) == len(set(expected)) and set(expected) == {str(row["id"]) for row in chosen},
|
|
"LAB_RESET_OWNER_CONFLICT")
|
|
if table == "agent_runs":
|
|
require(all(row["intent"] in ("manual_reference_capture", "build_dashboard_test_scenario")
|
|
for row in chosen), "LAB_RESET_OWNER_CONFLICT")
|
|
# #endregion Tooling.LabReset.Database.ModuleRunOwners
|
|
|
|
|
|
# #region Tooling.LabReset.Database.SharedRowOwner [C:3] [TYPE Function]
|
|
# @INVARIANT linked_module_ids names row PKs; run-owner identity is independently checked against the producer run sets.
|
|
def shared_row_owner(scope, name, row, keys):
|
|
owner = scope["shared_ownership"][name]
|
|
require(len(keys[name]) == 1, "LAB_RESET_OWNER_CONFLICT")
|
|
identity = str(row[keys[name][0]])
|
|
agents, scenarios = set(scope["module_agent_run_ids"]), set(scope["module_scenario_run_ids"])
|
|
if name == "verification_runs":
|
|
categories = row["categories_run"]
|
|
require(isinstance(categories, list) and categories and all(type(value) is str for value in categories)
|
|
and set(categories).issubset({"metric", "structure"})
|
|
and row["trigger"] in ("release_create", "release_publish", "scheduled"), "LAB_RESET_OWNER_CONFLICT")
|
|
# Reviewed database_tables selection explicitly includes standalone NULL-owner verification rows.
|
|
if row["agent_run_id"] is None:
|
|
return
|
|
require(identity in owner["linked_module_ids"], "LAB_RESET_OWNER_CONFLICT")
|
|
if name == "scenario_artifacts":
|
|
require(row["owner_type"] == "scenario_run" and row["owner_id"] in scenarios, "LAB_RESET_OWNER_CONFLICT")
|
|
return
|
|
allowed = agents | scenarios if name in ("provider_capacity_leases", "provider_operation_receipts") else agents
|
|
require(row[owner["owner_column"]] in allowed, "LAB_RESET_OWNER_CONFLICT")
|
|
# #endregion Tooling.LabReset.Database.SharedRowOwner
|
|
|
|
|
|
# #region Tooling.LabReset.Database.SharedOwnership [C:3] [TYPE Function]
|
|
def shared_ownership(scope, rows, keys, selected):
|
|
module_run_owners(scope, rows, keys, selected)
|
|
for name in scope["shared_ownership"]:
|
|
for row in rows[name]:
|
|
if node(name, row, keys[name]) not in selected:
|
|
continue
|
|
shared_row_owner(scope, name, row, keys)
|
|
# #endregion Tooling.LabReset.Database.SharedOwnership
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Protected [C:3] [TYPE Function]
|
|
def protected_checkpoint(scope, rows, keys, tables):
|
|
for name, expected in scope["protected_tables"].items():
|
|
actual = [primary_key(row, keys[name]) for row in rows[name]]
|
|
require([column.name for column in tables[name].columns] == expected["columns"]
|
|
and keys[name] == expected["primary_key_columns"] and len(actual) == expected["count"],
|
|
"LAB_RESET_PROTECTED_SCHEMA_CHANGED")
|
|
# Historical inventory sorts compact JSON primary-key dictionaries, then hashes the list.
|
|
from dashboard_testing_lab_reset_archive import json_bytes
|
|
ordered = sorted(actual, key=lambda value: json_bytes(value).decode())
|
|
require(digest(ordered) == expected["primary_key_set_sha256"],
|
|
"LAB_RESET_PROTECTED_PRIMARY_KEYS_CHANGED")
|
|
# #endregion Tooling.LabReset.Database.Protected
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Snapshot [C:4] [TYPE Function]
|
|
# @POST Public snapshot contains hashes/counts/schema only; selected row bodies remain in memory for private archive.
|
|
def snapshot(connection, scope):
|
|
tables, keys, fks, rows = reflect(connection, scope)
|
|
selected = selected_nodes(scope, rows, keys)
|
|
shared_ownership(scope, rows, keys, selected)
|
|
protected_checkpoint(scope, rows, keys, tables)
|
|
order = catalog_child_first(rows, keys, selected, fks)
|
|
summary = {}
|
|
for name, table in tables.items():
|
|
keep = [encode(row) for row in rows[name] if node(name, row, keys[name]) not in selected]
|
|
chosen = [encode(row) for row in rows[name] if node(name, row, keys[name]) in selected]
|
|
summary[name] = {"columns": [[column.name, str(column.type)] for column in table.columns],
|
|
"keys": keys[name], "foreign_keys": fks[name], "preserved_count": len(keep),
|
|
"preserved_sha256": digest(sorted(keep, key=digest)), "selected_count": len(chosen),
|
|
"selected_sha256": digest(sorted(chosen, key=digest))}
|
|
return {"tables": summary, "delete_order_sha256": digest(order)}, (tables, keys, rows, selected, order)
|
|
# #endregion Tooling.LabReset.Database.Snapshot
|
|
|
|
|
|
# #region Tooling.LabReset.Database.ArchiveRows [C:2] [TYPE Function]
|
|
def archive_rows(state):
|
|
tables, keys, rows, selected, _ = state
|
|
return {name: {"columns": [[column.name, str(column.type)] for column in table.columns],
|
|
"primary_key_columns": keys[name],
|
|
"rows": [encode(row) for row in sorted(rows[name], key=lambda row: node(name, row, keys[name]))
|
|
if node(name, row, keys[name]) in selected]}
|
|
for name, table in tables.items() if any(node(name, row, keys[name]) in selected for row in rows[name])}
|
|
# #endregion Tooling.LabReset.Database.ArchiveRows
|
|
|
|
|
|
# #region Tooling.LabReset.Database.LockSelected [C:3] [TYPE Function]
|
|
# @POST Selected row locks are acquired before root; delete later cannot introduce a new selected row lock.
|
|
def lock_selected(connection, state):
|
|
tables, keys, rows, selected, _ = state
|
|
for name in sorted(tables):
|
|
ordered = sorted(rows[name], key=lambda row: node(name, row, keys[name]))
|
|
for row in ordered:
|
|
if node(name, row, keys[name]) not in selected:
|
|
continue
|
|
predicate = and_(*(tables[name].c[key] == row[key] for key in keys[name]))
|
|
require(connection.execute(select(tables[name]).where(predicate).with_for_update()).first() is not None,
|
|
"LAB_RESET_PRIMARY_KEY_SET_CHANGED")
|
|
# #endregion Tooling.LabReset.Database.LockSelected
|
|
|
|
|
|
# #region Tooling.LabReset.Database.Delete [C:3] [TYPE Function]
|
|
# @POST Every exact selected row deleted once, child-first; unexpected affected counts roll back the whole transaction.
|
|
def delete_rows(connection, state):
|
|
tables, keys, rows, selected, order = state
|
|
indexed = {node(name, row, keys[name]): row for name in tables for row in rows[name]
|
|
if node(name, row, keys[name]) in selected}
|
|
for item in order:
|
|
name, _ = item
|
|
row = indexed[item]
|
|
predicate = and_(*(tables[name].c[key] == row[key] for key in keys[name]))
|
|
result = connection.execute(tables[name].delete().where(predicate))
|
|
require(result.rowcount == 1, "LAB_RESET_DELETE_COUNT_CONFLICT")
|
|
# #endregion Tooling.LabReset.Database.Delete
|
|
|
|
|
|
# #region Tooling.LabReset.Database.AfterDelete [C:3] [TYPE Function]
|
|
def verify_deleted(connection, scope, before, state):
|
|
tables, keys, original, selected, _ = state
|
|
table_names(connection, scope)
|
|
no_foreign_sessions(connection)
|
|
for name, table in tables.items():
|
|
actual = [row_fields(row, table) for row in connection.execute(select(table)).mappings()]
|
|
expected = [row for row in original[name] if node(name, row, keys[name]) not in selected]
|
|
require(digest(sorted(map(encode, actual), key=digest)) == digest(sorted(map(encode, expected), key=digest)),
|
|
"LAB_RESET_POST_DELETE_STATE_CONFLICT")
|
|
require(len(actual) == before["tables"][name]["preserved_count"], "LAB_RESET_DELETE_COUNT_CONFLICT")
|
|
# #endregion Tooling.LabReset.Database.AfterDelete
|
|
# #endregion Tooling.LabReset.Database
|