"""Integrity repair and dependency collection for SQL permanentization stores."""
from __future__ import annotations
from collections import defaultdict, deque
from collections.abc import Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING
import sqlalchemy
from httk.store.backend.schema import TableSchema, resolve_schema
from httk.store.backend.sql.graph import LogicalEdgeGraph
from httk.store.backend.sql.layout import METADATA_TABLE_NAME, actual_table_names
from httk.store.backend.sql.mapping import (
CONTENT_ID_COLUMN,
LOGICAL_ID_COLUMN,
RETRACTED_COLUMN,
ROLE_COLUMN,
SID_COLUMN,
SOURCE_LID_COLUMN,
STORE_TIMESTAMP_COLUMN,
TARGET_LID_COLUMN,
backing_dispatch_column_name,
entry_dispatch_table_name,
)
from httk.store.store_timestamp import FUTURE_TIMESTAMP_SLACK_NS
if TYPE_CHECKING:
from httk.store.backend.sql.store import SqlStore
__all__ = ["FsckSummary", "FsckTableSummary", "run_fsck"]
@dataclass(frozen=True)
[docs]
class FsckTableSummary:
"""Counters for one physical record, child, or dispatch table."""
@dataclass(frozen=True)
[docs]
class FsckSummary:
"""Immutable SQL fsck report, modeled after Mongo's public summary."""
[docs]
tables: Mapping[str, FsckTableSummary]
[docs]
violations: tuple[str, ...]
class _Counter:
def __init__(self) -> None:
self.examined = self.repaired = self.conflicts = self.deleted = 0
def freeze(self) -> FsckTableSummary:
return FsckTableSummary(self.examined, self.repaired, self.conflicts, self.deleted)
[docs]
def run_fsck(
store: SqlStore,
*,
repair: bool = True,
collect_garbage: bool = True,
repair_conflicts: bool = False,
clamp_future_timestamps: bool = False,
known_types: tuple[type, ...] = (),
exclusive: bool = False,
) -> FsckSummary:
"""Repair dispatches, sweep incomplete residue, and report logical dangling references.
DuckDB does not serialize a read-then-delete fsck against concurrent
writers. Callers must therefore pass ``exclusive=True`` there, explicitly
acknowledging that they have taken the database offline from all writers.
SQLite transactional stores instead issue ``BEGIN IMMEDIATE`` themselves.
"""
if store._database.engine.dialect.name == "duckdb" and not exclusive:
raise RuntimeError("DuckDB fsck requires exclusive=True and offline ownership from all writers")
verification_only = store.write_profile == "degraded" and not repair and not collect_garbage
schemas: dict[str, TableSchema] = {}
pending = [
*known_types,
*store._known_record_types,
*(record for family in store.layout.families for record in family.records),
]
while pending:
record = pending.pop()
schema = resolve_schema(record)
if schema.table_name in schemas:
continue
schemas[schema.table_name] = schema
pending.extend(schema.referenced_classes())
# Rebuild the in-memory mapping after a reopen; this is deliberately not
# DDL and therefore does not make missing tables appear.
store._register_tables(tuple(schema.cls for schema in schemas.values()))
graph = LogicalEdgeGraph.from_store(store, tuple(schemas.values()))
counters: defaultdict[str, _Counter] = defaultdict(_Counter)
violations: list[str] = []
timestamp_repaired = False
# A degraded verification has no repair, collection, lease, or dirty-row
# semantics. In particular it must not manufacture a metadata ``lease``
# row merely to inspect an otherwise untouched store.
connection_scope = store._read_connection() if verification_only else store._fsck_connection()
with connection_scope as connection:
present = actual_table_names(connection)
link_tables = {link.table_name for schema in schemas.values() for link in schema.links}
expected = set(graph.tables) | {METADATA_TABLE_NAME, "_httk_sid_counters"} | link_tables
unattributed = sorted(name for name in present if name not in expected and not name.startswith("_httk_"))
for name in unattributed:
counters[name].conflicts += 1
violations.append(f"table {name!r} cannot be attributed to a known schema and blocks fsck sweep")
# Attribution refusal happens before *any* mutation. A verification
# call is read-only by construction, and even a repair call does not
# touch a partially attributable database.
mutation_allowed = not unattributed
timestamp_repaired = _check_future_timestamps(
store,
connection,
present,
graph,
counters,
violations,
clamp_future_timestamps and repair and mutation_allowed,
)
if repair and mutation_allowed:
_repair_dispatches(store, connection, present, counters, violations, True, repair_conflicts)
else:
_repair_dispatches(store, connection, present, counters, violations, False, False)
marked = _mark(store, connection, graph, present, counters, violations, repair and mutation_allowed)
_check_links(store, connection, schemas, present, counters, violations)
if collect_garbage and not unattributed:
for table_name, schema in schemas.items():
if table_name not in present:
continue
table = store._table(table_name)
survivors = marked.get(table_name, set())
condition = table.c[ROLE_COLUMN] == 0
if survivors:
condition = sqlalchemy.and_(condition, table.c[SID_COLUMN].not_in(survivors))
if store.write_profile == "degraded" and schema.cls in store._entry_record_types:
store._touch_dirty_table(connection, table)
result = connection.execute(sqlalchemy.delete(table).where(condition))
counters[table_name].deleted += max(result.rowcount, 0)
# Deleting unreachable dependency parents can make their owned
# element rows ownerless. Sweep afterwards so one pass reaches a
# physical fixpoint.
_sweep_ownerless_children(connection, graph, present, counters)
elif collect_garbage:
violations.append("sweep aborted because unattributed application tables exist")
if mutation_allowed and (repair or collect_garbage):
for table_name in schemas:
if table_name in present:
store._release_unused_identity_claims(connection, table_name)
store._sync_identity_ownership(connection, store.layout)
store._clear_identity_caches()
if timestamp_repaired:
with store._mutation_lock, store._read_connection() as connection:
store._initialize_store_timestamp_mark(connection)
return FsckSummary(
MappingProxyType({name: value.freeze() for name, value in sorted(counters.items())}), tuple(violations)
)
def _check_future_timestamps(
store: SqlStore,
connection: sqlalchemy.Connection,
present: set[str] | frozenset[str],
graph: LogicalEdgeGraph,
counters: defaultdict[str, _Counter],
violations: list[str],
clamp: bool,
) -> bool:
"""Report parent timestamps beyond the current clock plus writer/checker slack.
Clamping is destructive to historic-query fidelity and is a last-resort
repair for data written with a badly skewed clock.
"""
if not store.store_timestamps:
return False
now_ns = store._clock()
resolution = store.store_timestamp_resolution
# Invariant: store_timestamp_resolution is None only when store_timestamps is
# falsy, which the guard above already returned on.
assert resolution is not None
limit_units = (now_ns + FUTURE_TIMESTAMP_SLACK_NS) // resolution
now_units = now_ns // resolution
repaired_any = False
for name in graph.tables:
if name not in present or name not in store._metadata.tables:
continue
table = store._table(name)
if ROLE_COLUMN not in table.c or STORE_TIMESTAMP_COLUMN not in table.c:
continue
rows = connection.execute(
sqlalchemy.select(table.c[SID_COLUMN], table.c[STORE_TIMESTAMP_COLUMN]).where(
table.c[STORE_TIMESTAMP_COLUMN] > limit_units
)
).all()
for sid, value in rows:
counters[name].examined += 1
future_ns = int(value) * resolution
limit_ns = limit_units * resolution
if clamp:
connection.execute(
sqlalchemy.update(table)
.where(table.c[SID_COLUMN] == sid)
.values({STORE_TIMESTAMP_COLUMN: now_units})
)
repaired_any = True
counters[name].repaired += 1
violations.append(
f"table {name!r} sid {sid} store_timestamp {future_ns} ns exceeds {limit_ns} ns; "
f"clamped to {now_ns} ns"
)
else:
counters[name].conflicts += 1
violations.append(f"table {name!r} sid {sid} store_timestamp {future_ns} ns exceeds {limit_ns} ns")
return repaired_any
def _repair_dispatches(
store: SqlStore,
connection: sqlalchemy.Connection,
present: set[str] | frozenset[str],
counters,
violations,
repair: bool,
repair_conflicts: bool,
) -> None:
for family in store.layout.families:
if len(family.records) < 2:
continue
name = entry_dispatch_table_name(family.name)
backing_names = [resolve_schema(record).table_name for record in family.records]
if name not in present:
nonempty_backings = [
backing_name
for backing_name in backing_names
if backing_name in present
and connection.execute(
sqlalchemy.select(sqlalchemy.func.count()).select_from(store._table(backing_name))
).scalar_one()
]
if nonempty_backings:
counters[name].conflicts += 1
violations.append(
f"dispatch {name!r} is missing while backing rows exist in {tuple(nonempty_backings)!r}"
)
continue
dispatch = store._table(name)
for row in connection.execute(sqlalchemy.select(dispatch)).mappings():
counters[name].examined += 1
populated = [
(record, int(row[backing_dispatch_column_name(record_name)]))
for record_name, record in zip(family.record_names, family.records, strict=True)
if row[backing_dispatch_column_name(record_name)] is not None
]
valid = len(populated) == 1
if valid:
record, sid = populated[0]
backing_name = resolve_schema(record).table_name
if backing_name not in present:
valid = False
else:
backing = store._table(backing_name)
valid = (
connection.execute(
sqlalchemy.select(backing.c[CONTENT_ID_COLUMN]).where(backing.c[SID_COLUMN] == sid)
).scalar_one_or_none()
== row[CONTENT_ID_COLUMN]
)
if valid:
continue
counters[name].conflicts += 1
violations.append(f"dispatch {name!r} has an invalid backing association")
if repair and repair_conflicts:
connection.execute(
sqlalchemy.delete(dispatch).where(dispatch.c[CONTENT_ID_COLUMN] == row[CONTENT_ID_COLUMN])
)
counters[name].deleted += 1
if not repair:
continue
for record_name, record in zip(family.record_names, family.records, strict=True):
backing = store._table(resolve_schema(record).table_name)
if backing.name not in present:
continue
column = backing_dispatch_column_name(record_name)
for sid, content in connection.execute(
sqlalchemy.select(backing.c[SID_COLUMN], backing.c[CONTENT_ID_COLUMN]).where(
backing.c[ROLE_COLUMN] == 1
)
):
if (
connection.execute(
sqlalchemy.select(dispatch.c[CONTENT_ID_COLUMN]).where(dispatch.c[CONTENT_ID_COLUMN] == content)
).first()
is not None
):
continue
values = {dispatch_column.name: None for dispatch_column in dispatch.columns}
values[CONTENT_ID_COLUMN] = content
values[column] = sid
connection.execute(sqlalchemy.insert(dispatch).values(values))
counters[name].repaired += 1
def _lineage_ids(
store: SqlStore,
connection: sqlalchemy.Connection,
table_name: str,
present: set[str] | frozenset[str],
) -> set[int] | None:
"""The distinct ``logical_id`` values of a parent table, or None when it cannot be read."""
if table_name not in present or table_name not in store._metadata.tables:
return None
table = store._table(table_name)
if LOGICAL_ID_COLUMN not in table.c:
return None
return {
int(value) for value in connection.execute(sqlalchemy.select(table.c[LOGICAL_ID_COLUMN]).distinct()).scalars()
}
def _check_links(
store: SqlStore,
connection: sqlalchemy.Connection,
schemas: dict[str, TableSchema],
present: set[str] | frozenset[str],
counters: defaultdict[str, _Counter],
violations: list[str],
) -> None:
"""Verify weak-link tables: valid ``retracted``, lineage integrity, no dangling endpoints.
Weak links are not ownership/reachability edges: this only reports, retains
no rows, and never affects the garbage sweep. A pair carrying more than one
lineage (a tolerated concurrency outcome) is a REPAIRABLE note, not
corruption, and is not counted as a conflict.
"""
for schema in schemas.values():
source_lids: set[int] | None = None
for link in schema.links:
name = link.table_name
if name not in present or name not in store._metadata.tables:
continue
if source_lids is None:
source_lids = _lineage_ids(store, connection, schema.table_name, present)
target_lids = _lineage_ids(store, connection, resolve_schema(link.target).table_name, present)
table = store._table(name)
lineage_min_sid: dict[int, int] = {}
pair_lineages: defaultdict[tuple[int, int], set[int]] = defaultdict(set)
for sid_v, logical_v, source_v, target_v, retracted_v in connection.execute(
sqlalchemy.select(
table.c[SID_COLUMN],
table.c[LOGICAL_ID_COLUMN],
table.c[SOURCE_LID_COLUMN],
table.c[TARGET_LID_COLUMN],
table.c[RETRACTED_COLUMN],
)
):
sid, logical_id = int(sid_v), int(logical_v)
source_lid, target_lid, retracted = int(source_v), int(target_v), int(retracted_v)
counters[name].examined += 1
if retracted not in (0, 1):
counters[name].conflicts += 1
violations.append(f"link table {name!r} sid {sid} has invalid retracted {retracted!r}")
previous = lineage_min_sid.get(logical_id)
if previous is None or sid < previous:
lineage_min_sid[logical_id] = sid
pair_lineages[(source_lid, target_lid)].add(logical_id)
if source_lids is not None and source_lid not in source_lids:
counters[name].conflicts += 1
violations.append(
f"link table {name!r} sid {sid} source_lid {source_lid} matches no "
f"{schema.table_name!r} logical_id"
)
if target_lids is not None and target_lid not in target_lids:
counters[name].conflicts += 1
violations.append(
f"link table {name!r} sid {sid} target_lid {target_lid} matches no "
f"{resolve_schema(link.target).table_name!r} logical_id"
)
for logical_id, min_sid in lineage_min_sid.items():
if logical_id != min_sid:
counters[name].conflicts += 1
violations.append(
f"link table {name!r} lineage logical_id {logical_id} does not equal its founder sid {min_sid}"
)
for (source_lid, target_lid), lineages in pair_lineages.items():
if len(lineages) > 1:
violations.append(
f"link table {name!r} pair ({source_lid}, {target_lid}) carries {len(lineages)} lineages; "
"a tolerated concurrency outcome (the pair stays live) — deduplicating is safe but is "
"NOT performed by repair"
)
def _sweep_ownerless_children(
connection: sqlalchemy.Connection, graph: LogicalEdgeGraph, present: set[str] | frozenset[str], counters
) -> None:
for edge in graph.ownership():
if edge.source_table not in present or edge.target_table not in present:
continue
assert edge.target_column is not None
child = sqlalchemy.table(edge.target_table, sqlalchemy.column(edge.target_column))
parent = sqlalchemy.table(edge.source_table, sqlalchemy.column(SID_COLUMN))
result = connection.execute(
sqlalchemy.delete(child).where(
~sqlalchemy.exists(sqlalchemy.select(1).where(parent.c[SID_COLUMN] == child.c[edge.target_column]))
)
)
counters[edge.target_table].deleted += max(result.rowcount, 0)
def _mark(
store: SqlStore,
connection: sqlalchemy.Connection,
graph: LogicalEdgeGraph,
present,
counters,
violations,
repair_roles: bool,
) -> dict[str, set[int]]:
marked: dict[str, set[int]] = defaultdict(set)
queue: deque[tuple[str, int]] = deque()
def add(name: str, sid: object) -> None:
if name in marked and isinstance(sid, int) and sid not in marked[name]:
marked[name].add(sid)
queue.append((name, sid))
# Initialize every parent table key so dependencies can be queued.
for name in graph.tables:
if name in present and name in store._metadata.tables and SID_COLUMN in store._table(name).c:
marked[name]
table = store._table(name)
for sid, role in connection.execute(sqlalchemy.select(table.c[SID_COLUMN], table.c[ROLE_COLUMN])):
if role in (0, 1):
continue
counters[name].conflicts += 1
violations.append(f"table {name!r} sid {sid} has invalid _httk_role {role!r}")
if repair_roles:
# Invalid roles are normalized to dependency. This is the
# conservative repair: corrupt data never becomes a root
# merely because fsck was asked to repair it.
connection.execute(
sqlalchemy.update(table).where(table.c[SID_COLUMN] == sid).values({ROLE_COLUMN: 0})
)
counters[name].repaired += 1
for sid in connection.execute(
sqlalchemy.select(table.c[SID_COLUMN]).where(table.c[ROLE_COLUMN] == 1)
).scalars():
counters[name].examined += 1
add(name, sid)
for edge in graph.edges:
if (
edge.kind != "dispatch"
or edge.source_table not in present
or edge.source_table not in store._metadata.tables
):
continue
table = store._table(edge.source_table)
assert edge.source_column is not None
for sid in connection.execute(
sqlalchemy.select(table.c[edge.source_column]).where(table.c[edge.source_column].is_not(None))
).scalars():
add(edge.target_table, sid)
outgoing: defaultdict[str, list] = defaultdict(list)
for edge in graph.edges:
outgoing[edge.source_table].append(edge)
while queue:
table_name, sid = queue.popleft()
if table_name not in present:
violations.append(f"dangling logical reference to absent table {table_name!r}/{sid}")
continue
table = store._table(table_name)
row = connection.execute(sqlalchemy.select(table).where(table.c[SID_COLUMN] == sid)).mappings().one_or_none()
if row is None:
violations.append(f"dangling logical reference to {table_name!r}/{sid}")
continue
for edge in outgoing[table_name]:
if edge.kind == "reference":
assert edge.source_column is not None
add(edge.target_table, row[edge.source_column])
elif edge.kind == "ownership":
assert edge.target_column is not None
child = store._table(edge.target_table)
for child_row in connection.execute(
sqlalchemy.select(child).where(child.c[edge.target_column] == sid)
).mappings():
for child_edge in outgoing[edge.target_table]:
if child_edge.kind == "child_element":
assert child_edge.source_column is not None
add(child_edge.target_table, child_row[child_edge.source_column])
return marked