Implement verified weekly archives and SQLite inventory reporting
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Standalone verified weekly PostgreSQL archiver."""
|
||||
@@ -0,0 +1,3 @@
|
||||
from .cli import main
|
||||
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,91 @@
|
||||
import json
|
||||
import shutil
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
from itertools import islice
|
||||
|
||||
from . import sqlite_archive
|
||||
from .errors import SafetyError
|
||||
from .weeks import WEEK, weeks, timestamp
|
||||
|
||||
|
||||
def log(event, **fields):
|
||||
print(json.dumps({"event": event, **fields}, default=lambda value: timestamp(value) if isinstance(value, datetime) else str(value), sort_keys=True), flush=True)
|
||||
|
||||
|
||||
def require_space(directory, needed):
|
||||
if shutil.disk_usage(directory).free < needed + 64 * 1024 * 1024:
|
||||
raise SafetyError("insufficient working storage")
|
||||
|
||||
|
||||
def run(pg, store, config, cutoff, mode, max_weeks=None):
|
||||
pg.lock()
|
||||
pg.check_schema()
|
||||
store.require_versioning()
|
||||
config.work_dir.mkdir(parents=True, exist_ok=True)
|
||||
candidates = weeks(pg.first_week(cutoff), cutoff)
|
||||
if max_weeks is not None:
|
||||
candidates = islice(candidates, max_weeks)
|
||||
log("invocation", cutoff=cutoff, mode=mode)
|
||||
for start in candidates:
|
||||
key = config.prefix + start.date().isoformat() + ".sqlite"
|
||||
begun = time.monotonic()
|
||||
try:
|
||||
pg.alive()
|
||||
counts = pg.counts(start, start + WEEK)
|
||||
log("started", week_start=start, week_end=start + WEEK, cutoff=cutoff, counts=counts)
|
||||
pg.ledger(start, "exporting", key)
|
||||
existing = store.head(key)
|
||||
if existing is None and pg.archived_version(start):
|
||||
raise SafetyError("previously verified archive missing: operator recovery required")
|
||||
old_size = existing["ContentLength"] if existing else 0
|
||||
# Conservative estimate for text encoding, SQLite overhead, journal and verification copy.
|
||||
require_space(config.work_dir, old_size * 3 + int(pg.estimated_bytes(start)) * 8)
|
||||
with tempfile.TemporaryDirectory(prefix="btcdata-week-", dir=config.work_dir) as directory:
|
||||
local = Path(directory) / "archive.sqlite"
|
||||
downloaded = Path(directory) / "verified.sqlite"
|
||||
if existing:
|
||||
store.download(key, existing["VersionId"], local)
|
||||
sqlite_archive.validate(local, start, config.pg["dbname"], config.batch_rows)
|
||||
advertised = existing.get("Metadata", {}).get("sha256")
|
||||
if not advertised or sqlite_archive.file_hash(local) != advertised:
|
||||
raise SafetyError("existing object file checksum mismatch")
|
||||
else:
|
||||
sqlite_archive.create(local)
|
||||
pg.export(local, start)
|
||||
metadata = sqlite_archive.finalize(local, start, config.pg["dbname"], datetime.now(timezone.utc), config.batch_rows)
|
||||
checksum = sqlite_archive.file_hash(local)
|
||||
require_space(config.work_dir, local.stat().st_size)
|
||||
pg.alive()
|
||||
version = store.upload(key, local, checksum)
|
||||
store.download(key, version, downloaded)
|
||||
if sqlite_archive.file_hash(downloaded) != checksum:
|
||||
raise SafetyError("downloaded file SHA-256 mismatch")
|
||||
verified = sqlite_archive.validate(downloaded, start, config.pg["dbname"], config.batch_rows)
|
||||
if verified != metadata:
|
||||
raise SafetyError("downloaded archive metadata mismatch")
|
||||
archive_counts = {t: s["count"] for t, s in verified["tables"].items()}
|
||||
pg.ledger(start, "verified", key, version, checksum, archive_counts)
|
||||
log("verified", week_start=start, counts=archive_counts, version_id=version, sha256=checksum)
|
||||
if mode == "archive-and-prune":
|
||||
pg.ledger(start, "pruning", key)
|
||||
|
||||
def progress(deleted):
|
||||
pg.ledger(start, "pruning", key, deletions=deleted)
|
||||
log("deletion_progress", week_start=start, counts=archive_counts, deleted=deleted)
|
||||
|
||||
deleted = pg.prune(downloaded, start, cutoff, progress)
|
||||
log("pruned", week_start=start, counts=archive_counts, deleted=deleted)
|
||||
pg.ledger(start, "complete", key, deletions=deleted)
|
||||
log("complete", week_start=start, counts=archive_counts, duration_seconds=round(time.monotonic() - begun, 3))
|
||||
except BaseException as exc:
|
||||
# Do not serialize database/SDK exception messages: they may contain rows or secrets.
|
||||
reason = str(exc) if isinstance(exc, SafetyError) else type(exc).__name__
|
||||
try:
|
||||
pg.ledger(start, "failed", key, error=reason)
|
||||
except Exception:
|
||||
pass
|
||||
log("failed", week_start=start, error=reason)
|
||||
raise
|
||||
@@ -0,0 +1,65 @@
|
||||
import argparse
|
||||
import signal
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from .archive import log, run
|
||||
from .config import Config
|
||||
from .errors import SafetyError
|
||||
from .object_store import ObjectStore
|
||||
from .postgres import Postgres
|
||||
from .weeks import boundaries, weeks, WEEK
|
||||
|
||||
|
||||
def positive(value):
|
||||
number = int(value)
|
||||
if number <= 0:
|
||||
raise argparse.ArgumentTypeError("must be a positive integer")
|
||||
return number
|
||||
|
||||
|
||||
def parser():
|
||||
result = argparse.ArgumentParser(description="Verified weekly btcprod archives; no implicit pruning mode")
|
||||
commands = result.add_subparsers(dest="command", required=True)
|
||||
commands.add_parser("inventory")
|
||||
commands.add_parser("status")
|
||||
command = commands.add_parser("run")
|
||||
command.add_argument("--mode", required=True, choices=("verify-only", "archive-and-prune"))
|
||||
command.add_argument("--max-weeks", type=positive)
|
||||
return result
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
args = parser().parse_args(argv)
|
||||
# Capture once, before connecting; a catch-up run never slides its cutoff.
|
||||
anchor, cutoff = boundaries(datetime.now(timezone.utc))
|
||||
pg = None
|
||||
|
||||
def terminate(signum, frame):
|
||||
raise KeyboardInterrupt()
|
||||
|
||||
signal.signal(signal.SIGTERM, terminate)
|
||||
try:
|
||||
config = Config.from_env(need_s3=args.command == "run")
|
||||
pg = Postgres(config)
|
||||
if args.command == "run":
|
||||
run(pg, ObjectStore(config), config, cutoff, args.mode, args.max_weeks)
|
||||
elif args.command == "inventory":
|
||||
pg.check_schema()
|
||||
first = pg.first_week(cutoff)
|
||||
log("inventory", anchor=anchor, cutoff=cutoff, oldest_eligible_week=next(weeks(first, cutoff), None))
|
||||
for start in weeks(first, cutoff):
|
||||
log("candidate", week_start=start, counts=pg.counts(start, start + WEEK))
|
||||
else:
|
||||
pg.check_schema()
|
||||
log("status", anchor=anchor, cutoff=cutoff, **pg.status(cutoff))
|
||||
return 0
|
||||
except Exception as exc:
|
||||
# Connection errors, SQL errors, SDK responses can contain credentials or row data.
|
||||
log("error", error=str(exc) if isinstance(exc, SafetyError) else type(exc).__name__)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
log("interrupted")
|
||||
return 130
|
||||
finally:
|
||||
if pg is not None:
|
||||
pg.close()
|
||||
@@ -0,0 +1,75 @@
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from .errors import SafetyError
|
||||
|
||||
|
||||
def load_env_file(path: Path):
|
||||
"""Load literal KEY=value entries without shell execution or interpolation.
|
||||
|
||||
Existing environment variables take precedence. Never include values in errors.
|
||||
"""
|
||||
if not path.exists():
|
||||
return
|
||||
for number, line in enumerate(path.read_text().splitlines(), 1):
|
||||
line = line.strip()
|
||||
if not line or line.startswith('#'):
|
||||
continue
|
||||
match = re.fullmatch(r'(?:export\s+)?([A-Za-z_][A-Za-z_0-9]*)\s*=\s*(.*)', line)
|
||||
if not match:
|
||||
raise SafetyError(f"invalid environment file entry at line {number}")
|
||||
try:
|
||||
parts = shlex.split(match.group(2), comments=True, posix=True)
|
||||
except ValueError:
|
||||
raise SafetyError(f"invalid environment file quoting at line {number}") from None
|
||||
if len(parts) > 1:
|
||||
raise SafetyError(f"quote environment file values containing spaces at line {number}")
|
||||
os.environ.setdefault(match.group(1), parts[0] if parts else '')
|
||||
|
||||
|
||||
@dataclass(frozen=True, repr=False)
|
||||
class Config:
|
||||
pg: dict
|
||||
endpoint: str | None
|
||||
access_key: str | None
|
||||
secret_key: str | None
|
||||
region: str
|
||||
bucket: str
|
||||
prefix: str
|
||||
work_dir: Path
|
||||
batch_rows: int
|
||||
|
||||
@classmethod
|
||||
def from_env(cls, need_s3: bool = True, need_pg: bool = True):
|
||||
if env_file := os.environ.get('ARCHIVE_ENV_FILE'):
|
||||
load_env_file(Path(env_file))
|
||||
required = ["PGHOST", "PGUSER", "PGPASSWORD"] if need_pg else []
|
||||
if need_s3:
|
||||
required += ["S3_ENDPOINT_URL", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]
|
||||
missing = [name for name in required if not os.environ.get(name)]
|
||||
if missing:
|
||||
raise SafetyError("missing configuration: " + ", ".join(missing))
|
||||
endpoint = os.environ.get("S3_ENDPOINT_URL")
|
||||
if endpoint:
|
||||
url = urlsplit(endpoint)
|
||||
if url.scheme not in ("http", "https") or not url.hostname or url.username or url.password or url.query or url.fragment:
|
||||
raise SafetyError("S3_ENDPOINT_URL must be an explicit HTTP(S) endpoint without credentials")
|
||||
batch = int(os.environ.get("ARCHIVE_BATCH_ROWS", "5000"))
|
||||
if batch <= 0:
|
||||
raise SafetyError("ARCHIVE_BATCH_ROWS must be positive")
|
||||
prefix = os.environ.get("ARCHIVE_PREFIX", "btcprod/weeks/")
|
||||
if not prefix.endswith("/") or prefix.startswith("/"):
|
||||
raise SafetyError("ARCHIVE_PREFIX must be relative and end in /")
|
||||
return cls(
|
||||
pg={"host": os.environ["PGHOST"], "port": int(os.environ.get("PGPORT", "5432")),
|
||||
"dbname": os.environ.get("PGDATABASE", "btcprod"), "user": os.environ["PGUSER"],
|
||||
"password": os.environ["PGPASSWORD"], "sslmode": os.environ.get("PGSSLMODE", "prefer"),
|
||||
"connect_timeout": 15, "application_name": "btcdata-archive"} if need_pg else {},
|
||||
endpoint=endpoint, access_key=os.environ.get("AWS_ACCESS_KEY_ID"),
|
||||
secret_key=os.environ.get("AWS_SECRET_ACCESS_KEY"), region=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
|
||||
bucket=os.environ.get("ARCHIVE_BUCKET", "btcdata-archive"), prefix=prefix,
|
||||
work_dir=Path(os.environ.get("ARCHIVE_WORK_DIR", "/work")), batch_rows=batch)
|
||||
@@ -0,0 +1,2 @@
|
||||
class SafetyError(ValueError, RuntimeError):
|
||||
"""Operator-safe diagnostic containing no credentials or row payloads."""
|
||||
@@ -0,0 +1,72 @@
|
||||
import boto3
|
||||
from botocore.config import Config as S3Config
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
from .errors import SafetyError
|
||||
|
||||
|
||||
class ObjectStore:
|
||||
def __init__(self, config):
|
||||
if not config.endpoint:
|
||||
raise SafetyError("explicit S3 endpoint required")
|
||||
self.bucket = config.bucket
|
||||
self.client = boto3.client("s3", endpoint_url=config.endpoint,
|
||||
aws_access_key_id=config.access_key, aws_secret_access_key=config.secret_key,
|
||||
region_name=config.region,
|
||||
config=S3Config(s3={"addressing_style": "path"}, connect_timeout=15,
|
||||
read_timeout=120, retries={"max_attempts": 3, "mode": "standard"}))
|
||||
|
||||
def require_versioning(self):
|
||||
if self.client.get_bucket_versioning(Bucket=self.bucket).get("Status") != "Enabled":
|
||||
raise SafetyError("bucket versioning must be enabled by an administrator")
|
||||
|
||||
def head(self, key):
|
||||
try:
|
||||
info = self.client.head_object(Bucket=self.bucket, Key=key)
|
||||
except ClientError as exc:
|
||||
if exc.response["Error"]["Code"] in ("404", "NoSuchKey", "NotFound"):
|
||||
return None
|
||||
raise
|
||||
if info.get("VersionId") in (None, "", "null"):
|
||||
raise SafetyError("existing object lacks a durable version ID")
|
||||
return info
|
||||
|
||||
def download(self, key, version, path):
|
||||
if version in (None, "", "null"):
|
||||
raise SafetyError("exact object version required")
|
||||
response = self.client.get_object(Bucket=self.bucket, Key=key, VersionId=version)
|
||||
if response.get("VersionId") != version:
|
||||
response["Body"].close()
|
||||
raise SafetyError("object store returned wrong version")
|
||||
try:
|
||||
with path.open("wb") as out:
|
||||
while chunk := response["Body"].read(1024 * 1024):
|
||||
out.write(chunk)
|
||||
finally:
|
||||
response["Body"].close()
|
||||
|
||||
def upload(self, key, path, checksum):
|
||||
self.require_versioning()
|
||||
args = {"Bucket": self.bucket, "Key": key}
|
||||
response = self.client.create_multipart_upload(**args, ContentType="application/vnd.sqlite3", Metadata={"sha256": checksum})
|
||||
upload_id = response["UploadId"]
|
||||
try:
|
||||
parts = []
|
||||
# Stay below S3's 10,000-part limit, with bounded memory even for large weeks.
|
||||
part_size = max(8 * 1024 * 1024, (path.stat().st_size + 9998) // 9999)
|
||||
with path.open("rb") as source:
|
||||
while chunk := source.read(part_size):
|
||||
number = len(parts) + 1
|
||||
part = self.client.upload_part(**args, UploadId=upload_id, PartNumber=number, Body=chunk)
|
||||
parts.append({"PartNumber": number, "ETag": part["ETag"]})
|
||||
response = self.client.complete_multipart_upload(**args, UploadId=upload_id, MultipartUpload={"Parts": parts})
|
||||
except BaseException:
|
||||
try:
|
||||
self.client.abort_multipart_upload(**args, UploadId=upload_id)
|
||||
except Exception:
|
||||
pass
|
||||
raise
|
||||
version = response.get("VersionId")
|
||||
if version in (None, "", "null"):
|
||||
raise SafetyError("upload did not return a durable object version ID")
|
||||
return version
|
||||
@@ -0,0 +1,175 @@
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import psycopg
|
||||
from psycopg import sql
|
||||
from psycopg.types.json import Jsonb
|
||||
|
||||
from .schema import TABLES, FORMAT_VERSION, Digest, encode_row, decode_row
|
||||
from . import sqlite_archive
|
||||
from .errors import SafetyError
|
||||
from .weeks import WEEK
|
||||
|
||||
LOCK_ID = 728194107
|
||||
|
||||
|
||||
class Postgres:
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.conn = psycopg.connect(**config.pg, autocommit=True)
|
||||
self.conn.execute("SET TIME ZONE 'UTC'")
|
||||
self.locked = False
|
||||
|
||||
def close(self):
|
||||
self.conn.close()
|
||||
|
||||
def lock(self):
|
||||
if self.locked:
|
||||
self.alive()
|
||||
return
|
||||
if not self.conn.execute("SELECT pg_try_advisory_lock(%s)", (LOCK_ID,)).fetchone()[0]:
|
||||
raise SafetyError("another archiver holds the advisory lock")
|
||||
self.locked = True
|
||||
|
||||
def alive(self):
|
||||
if not self.locked:
|
||||
raise SafetyError("archiver lock required")
|
||||
self.conn.execute("SELECT 1").fetchone() # never reconnect this session
|
||||
|
||||
def check_schema(self):
|
||||
for table in TABLES:
|
||||
relation = self.conn.execute("SELECT relkind, relrowsecurity FROM pg_class WHERE oid=to_regclass(%s)", ("public." + table.name,)).fetchone()
|
||||
if relation != ('r', False):
|
||||
raise SafetyError(f"migration required: public.{table.name} must be an ordinary table without RLS")
|
||||
if self.conn.execute("SELECT count(*) FROM pg_trigger WHERE tgrelid=to_regclass(%s) AND NOT tgisinternal AND tgenabled <> 'D' AND (tgtype & 8) <> 0", ("public." + table.name,)).fetchone()[0]:
|
||||
raise SafetyError(f"migration required: public.{table.name} has unsupported DELETE triggers")
|
||||
if self.conn.execute("SELECT count(*) FROM pg_constraint WHERE confrelid=to_regclass(%s) AND contype='f' AND confdeltype IN ('c','n','d')", ("public." + table.name,)).fetchone()[0]:
|
||||
raise SafetyError(f"migration required: public.{table.name} has mutating inbound DELETE foreign keys")
|
||||
actual = self.conn.execute("""
|
||||
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod), NOT a.attnotnull
|
||||
FROM pg_catalog.pg_attribute a
|
||||
WHERE a.attrelid = to_regclass(%s) AND a.attnum > 0 AND NOT a.attisdropped
|
||||
ORDER BY a.attnum
|
||||
""", ("public." + table.name,)).fetchall()
|
||||
if sorted(actual) != sorted(table.columns):
|
||||
raise SafetyError(f"migration required: unsupported public.{table.name} columns/types/nullability")
|
||||
pk = self.conn.execute("""
|
||||
SELECT array_agg(a.attname ORDER BY k.ordinality)
|
||||
FROM pg_constraint c CROSS JOIN LATERAL unnest(c.conkey) WITH ORDINALITY k(attnum, ordinality)
|
||||
JOIN pg_attribute a ON a.attrelid=c.conrelid AND a.attnum=k.attnum
|
||||
WHERE c.conrelid=to_regclass(%s) AND c.contype='p'
|
||||
GROUP BY c.oid
|
||||
""", ("public." + table.name,)).fetchall()
|
||||
if pk != [([table.pk],)]:
|
||||
raise SafetyError(f"migration required: unsupported public.{table.name} primary key")
|
||||
|
||||
def first_week(self, cutoff):
|
||||
values = []
|
||||
for table in TABLES:
|
||||
_, end = table.bounds(cutoff, cutoff)
|
||||
value = self.conn.execute(f"SELECT min({table.retention}) FROM public.{table.name} WHERE {table.retention} < %s", (end,)).fetchone()[0]
|
||||
if value is not None:
|
||||
values.append(datetime.fromtimestamp(value, timezone.utc) if table.retention == "ts" else value)
|
||||
value = self.conn.execute("SELECT min(week_start) FROM btcdata_archive.weeks WHERE status <> 'complete' AND week_end <= %s", (cutoff,)).fetchone()[0]
|
||||
if value is not None:
|
||||
values.append(value)
|
||||
return min(values) if values else None
|
||||
|
||||
def counts(self, start, end):
|
||||
return {t.name: self.conn.execute(f"SELECT count(*) FROM public.{t.name} WHERE {t.retention} >= %s AND {t.retention} < %s", t.bounds(start, end)).fetchone()[0] for t in TABLES}
|
||||
|
||||
def archived_version(self, start):
|
||||
row = self.conn.execute("SELECT object_version_id FROM btcdata_archive.weeks WHERE week_start=%s", (start,)).fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
def estimated_bytes(self, start):
|
||||
return sum(self.conn.execute(f"SELECT coalesce(sum(pg_column_size(t)),0) FROM public.{t.name} t WHERE {t.retention} >= %s AND {t.retention} < %s", t.bounds(start, start + WEEK)).fetchone()[0] for t in TABLES)
|
||||
|
||||
def ledger(self, start, status, key, version=None, checksum=None, counts=None, deletions=None, error=None):
|
||||
self.alive()
|
||||
# PostgreSQL checks the proposed INSERT before ON CONFLICT runs. Resolve
|
||||
# durable verification fields first so progress-only transitions satisfy
|
||||
# the same constraints as the final updated row; never weaken the checks.
|
||||
self.conn.execute("""
|
||||
WITH previous AS (
|
||||
SELECT object_version_id,file_sha256 FROM btcdata_archive.weeks WHERE week_start=%s
|
||||
)
|
||||
INSERT INTO btcdata_archive.weeks
|
||||
(week_start,week_end,format_version,status,object_key,object_version_id,file_sha256,row_counts,deletion_counts,attempt_time,completion_time,last_error)
|
||||
VALUES (%s,%s,%s,%s,%s,
|
||||
coalesce(%s,(SELECT object_version_id FROM previous)),
|
||||
coalesce(%s,(SELECT file_sha256 FROM previous)),
|
||||
%s,%s,now(),CASE WHEN %s='complete' THEN now() ELSE NULL END,%s)
|
||||
ON CONFLICT (week_start) DO UPDATE SET status=excluded.status,object_key=excluded.object_key,
|
||||
object_version_id=coalesce(excluded.object_version_id,btcdata_archive.weeks.object_version_id),
|
||||
file_sha256=coalesce(excluded.file_sha256,btcdata_archive.weeks.file_sha256),
|
||||
row_counts=coalesce(excluded.row_counts,btcdata_archive.weeks.row_counts),
|
||||
deletion_counts=coalesce(excluded.deletion_counts,btcdata_archive.weeks.deletion_counts),
|
||||
attempt_time=excluded.attempt_time,completion_time=excluded.completion_time,last_error=excluded.last_error
|
||||
""", (start, start, start + WEEK, FORMAT_VERSION, status, key, version, checksum,
|
||||
Jsonb(counts) if counts is not None else None, Jsonb(deletions) if deletions is not None else None, status, error))
|
||||
|
||||
def export(self, path, start):
|
||||
db = sqlite_archive.connect(path)
|
||||
try:
|
||||
with self.conn.transaction():
|
||||
self.conn.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")
|
||||
for table in TABLES:
|
||||
db.execute("CREATE TEMP TABLE source_keys (pk INTEGER PRIMARY KEY)")
|
||||
source = Digest()
|
||||
projection = ",".join(name + "::text" if typ == "jsonb" else name for name, typ, _ in table.columns)
|
||||
with self.conn.cursor(name="archive_" + uuid.uuid4().hex) as cur:
|
||||
cur.execute(f"SELECT {projection} FROM public.{table.name} WHERE {table.retention} >= %s AND {table.retention} < %s ORDER BY {table.pk}", table.bounds(start, start + WEEK))
|
||||
while batch := cur.fetchmany(self.config.batch_rows):
|
||||
encoded = [encode_row(table, row) for row in batch]
|
||||
sqlite_archive.upsert(db, table, encoded)
|
||||
db.executemany("INSERT INTO source_keys VALUES (?)", [(r[0],) for r in encoded])
|
||||
for row in encoded:
|
||||
source.add(row)
|
||||
db.commit()
|
||||
fresh = Digest()
|
||||
cur = db.execute(f"SELECT {','.join('t.' + n for n in table.names)} FROM {table.name} t JOIN source_keys k ON t.{table.pk}=k.pk ORDER BY t.{table.pk}")
|
||||
while batch := cur.fetchmany(self.config.batch_rows):
|
||||
for row in batch:
|
||||
fresh.add(row)
|
||||
if source.result() != fresh.result():
|
||||
raise SafetyError("source export checksum mismatch")
|
||||
db.execute("DROP TABLE source_keys")
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
def prune(self, path, start, cutoff, progress):
|
||||
deleted = {t.name: 0 for t in TABLES}
|
||||
db = sqlite_archive.connect(path, readonly=True)
|
||||
try:
|
||||
for table in TABLES:
|
||||
for batch in sqlite_archive.rows(db, table, self.config.batch_rows):
|
||||
self.alive()
|
||||
with self.conn.transaction():
|
||||
definitions = sql.SQL(',').join(sql.SQL('{} {}').format(sql.Identifier(n), sql.SQL(typ)) for n, typ, _ in table.columns)
|
||||
self.conn.execute(sql.SQL("CREATE TEMP TABLE archive_stage ({}) ON COMMIT DROP").format(definitions))
|
||||
with self.conn.cursor().copy("COPY archive_stage FROM STDIN") as copy:
|
||||
for row in batch:
|
||||
copy.write_row(decode_row(table, row))
|
||||
comparisons = " AND ".join(f"s.{n} IS NOT DISTINCT FROM t.{n}" for n in table.names)
|
||||
begin, end = table.bounds(start, start + WEEK)
|
||||
_, limit = table.bounds(cutoff, cutoff)
|
||||
cur = self.conn.execute(f"DELETE FROM public.{table.name} t USING archive_stage s WHERE t.{table.pk}=s.{table.pk} AND {comparisons} AND t.{table.retention} >= %s AND t.{table.retention} < %s AND t.{table.retention} < %s", (begin, end, limit))
|
||||
deleted[table.name] += cur.rowcount
|
||||
progress(deleted.copy())
|
||||
finally:
|
||||
db.close()
|
||||
remaining = self.counts(start, start + WEEK)
|
||||
if any(remaining.values()):
|
||||
raise SafetyError("reconciliation required: changed or late rows remain")
|
||||
return deleted
|
||||
|
||||
def status(self, cutoff):
|
||||
complete = self.conn.execute("SELECT max(week_start) FROM btcdata_archive.weeks WHERE status='complete'").fetchone()[0]
|
||||
incomplete = self.conn.execute("SELECT week_start,status,attempt_time,last_error FROM btcdata_archive.weeks WHERE status<>'complete' ORDER BY week_start").fetchall()
|
||||
expired = {}
|
||||
for t in TABLES:
|
||||
_, end = t.bounds(cutoff, cutoff)
|
||||
expired[t.name] = self.conn.execute(f"SELECT count(*) FROM public.{t.name} WHERE {t.retention}<%s", (end,)).fetchone()[0]
|
||||
return {"last_completed_week": complete, "incomplete": incomplete, "expired_counts": expired}
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Read-only S3/SQLite file inventory. No PostgreSQL imports or connections."""
|
||||
import argparse
|
||||
from contextlib import closing
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
from .config import Config
|
||||
from .errors import SafetyError
|
||||
from .object_store import ObjectStore
|
||||
from .schema import TABLES
|
||||
from .sqlite_archive import file_hash
|
||||
|
||||
TABLE_NAMES = tuple(table.name for table in TABLES)
|
||||
|
||||
|
||||
def sqlite_counts(path: Path):
|
||||
# Do not trust metadata row counts: count the actual downloaded tables.
|
||||
with closing(sqlite3.connect(path.resolve().as_uri() + '?mode=ro', uri=True)) as db:
|
||||
db.execute('PRAGMA query_only=ON')
|
||||
if db.execute('PRAGMA integrity_check').fetchall() != [('ok',)]:
|
||||
raise SafetyError('SQLite integrity check failed')
|
||||
tables = {row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
||||
if not set(TABLE_NAMES) <= tables:
|
||||
raise SafetyError('SQLite archive is missing a required table')
|
||||
return {name: db.execute(f'SELECT COUNT(*) FROM {name}').fetchone()[0] for name in TABLE_NAMES}
|
||||
|
||||
|
||||
def files(store, config):
|
||||
"""Yield current object sizes and real table counts using at most one local file."""
|
||||
pages = store.client.get_paginator('list_objects_v2').paginate(Bucket=config.bucket, Prefix=config.prefix)
|
||||
for page in pages:
|
||||
for item in page.get('Contents', []):
|
||||
key = item['Key']
|
||||
if not key.lower().endswith('.sqlite'):
|
||||
yield key, item['Size'], None
|
||||
continue
|
||||
# Pin the version before downloading: a concurrent archiver may replace it.
|
||||
head = store.head(key)
|
||||
if head is None:
|
||||
raise SafetyError('Listed archive is no longer available: ' + key)
|
||||
size = head['ContentLength']
|
||||
config.work_dir.mkdir(parents=True, exist_ok=True)
|
||||
if shutil.disk_usage(config.work_dir).free < size + 64 * 1024 * 1024:
|
||||
raise SafetyError('Insufficient working storage to inspect: ' + key)
|
||||
with tempfile.TemporaryDirectory(prefix='btcdata-report-', dir=config.work_dir) as directory:
|
||||
path = Path(directory) / 'archive.sqlite'
|
||||
store.download(key, head['VersionId'], path)
|
||||
if path.stat().st_size != size:
|
||||
raise SafetyError('Downloaded archive size mismatch: ' + key)
|
||||
checksum = head.get('Metadata', {}).get('sha256')
|
||||
if checksum and file_hash(path) != checksum:
|
||||
raise SafetyError('Downloaded archive SHA-256 mismatch: ' + key)
|
||||
counts = sqlite_counts(path)
|
||||
# Clean up before yielding, so the next week cannot accumulate local files.
|
||||
yield key, size, counts
|
||||
|
||||
|
||||
def show(store, config):
|
||||
print(f'Current files in s3://{config.bucket}/{config.prefix} (latest versions only)', flush=True)
|
||||
print(f'{"FILE":<44} {"MiB":>10} ' + ' '.join(f'{name:>15}' for name in TABLE_NAMES), flush=True)
|
||||
total_bytes = file_count = 0
|
||||
totals = {name: 0 for name in TABLE_NAMES}
|
||||
for key, size, counts in files(store, config):
|
||||
values = ' '.join(f'{counts[name]:>15,}' if counts is not None else f'{"-":>15}' for name in TABLE_NAMES)
|
||||
print(f'{key:<44} {size / 1048576:>10.2f} {values}', flush=True)
|
||||
total_bytes += size
|
||||
file_count += 1
|
||||
if counts is not None:
|
||||
for name in TABLE_NAMES:
|
||||
totals[name] += counts[name]
|
||||
print(f'{"TOTAL":<44} {total_bytes / 1048576:>10.2f} ' + ' '.join(f'{totals[name]:>15,}' for name in TABLE_NAMES), flush=True)
|
||||
print(f'{file_count} files, {total_bytes:,} bytes ({total_bytes / 1073741824:.2f} GiB). Non-SQLite files show "-" counts.', flush=True)
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
argparse.ArgumentParser(description='List archive files, sizes and actual SQLite table counts; never query PostgreSQL.').parse_args(argv)
|
||||
try:
|
||||
config = Config.from_env(need_pg=False)
|
||||
show(ObjectStore(config), config)
|
||||
return 0
|
||||
except KeyboardInterrupt:
|
||||
print('Report interrupted.', file=sys.stderr)
|
||||
return 130
|
||||
except Exception as error:
|
||||
message = str(error) if isinstance(error, SafetyError) else type(error).__name__
|
||||
print('Archive report failed: ' + message, file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,95 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
import hashlib
|
||||
import json
|
||||
|
||||
from .weeks import timestamp
|
||||
|
||||
FORMAT_VERSION = 1
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Table:
|
||||
name: str
|
||||
columns: tuple[tuple[str, str, bool], ...] # PostgreSQL type, nullable
|
||||
retention: str
|
||||
|
||||
@property
|
||||
def pk(self):
|
||||
return self.columns[0][0]
|
||||
|
||||
@property
|
||||
def names(self):
|
||||
return tuple(c[0] for c in self.columns)
|
||||
|
||||
def bounds(self, start, end):
|
||||
if self.retention == "ts":
|
||||
return int(start.timestamp()), int(end.timestamp())
|
||||
return start, end
|
||||
|
||||
|
||||
TABLES = (
|
||||
Table("binance_price", (("id", "bigint", False), ("price", "numeric(18,8)", False),
|
||||
("recorded_at", "timestamp with time zone", False)), "recorded_at"),
|
||||
Table("strike_price", (("id", "bigint", False), ("rates", "jsonb", False),
|
||||
("recorded_at", "timestamp with time zone", False), ("quote_sats", "bigint", True)), "recorded_at"),
|
||||
Table("kraken_minute", (("ts", "bigint", False), *tuple((c, "numeric(18,8)", False) for c in ("open", "high", "low", "close", "vwap")),
|
||||
("volume", "numeric(24,8)", False), ("trade_count", "integer", False),
|
||||
("created_at", "timestamp with time zone", False)), "ts"),
|
||||
)
|
||||
|
||||
|
||||
def exact_json(value):
|
||||
"""Sorted JSON without any binary floating-point conversion."""
|
||||
if isinstance(value, dict):
|
||||
return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + exact_json(value[k]) for k in sorted(value)) + "}"
|
||||
if isinstance(value, list):
|
||||
return "[" + ",".join(exact_json(v) for v in value) + "]"
|
||||
if isinstance(value, Decimal):
|
||||
if not value.is_finite():
|
||||
raise ValueError("nonfinite JSON number")
|
||||
return format(value, "f")
|
||||
return json.dumps(value, ensure_ascii=False, allow_nan=False, separators=(",", ":"))
|
||||
|
||||
|
||||
def canonical_json(text):
|
||||
return exact_json(json.loads(text, parse_float=Decimal, parse_int=Decimal))
|
||||
|
||||
|
||||
def encode_row(table, row):
|
||||
result = []
|
||||
for value, (_, typ, nullable) in zip(row, table.columns, strict=True):
|
||||
if value is None:
|
||||
if not nullable:
|
||||
raise ValueError("unexpected SQL NULL")
|
||||
result.append(None)
|
||||
elif typ == "jsonb":
|
||||
result.append(canonical_json(value))
|
||||
elif typ.startswith("numeric"):
|
||||
result.append(str(value))
|
||||
elif typ == "timestamp with time zone":
|
||||
result.append(timestamp(value))
|
||||
else:
|
||||
result.append(int(value))
|
||||
return tuple(result)
|
||||
|
||||
|
||||
def decode_row(table, row):
|
||||
return tuple(datetime.fromisoformat(v) if typ == "timestamp with time zone" else
|
||||
Decimal(v) if typ.startswith("numeric") else v
|
||||
for v, (_, typ, _) in zip(row, table.columns, strict=True))
|
||||
|
||||
|
||||
class Digest:
|
||||
"""SHA-256 of UTF-8 compact JSON row arrays, each followed by LF, in PK order."""
|
||||
def __init__(self):
|
||||
self.count = 0
|
||||
self.hash = hashlib.sha256()
|
||||
|
||||
def add(self, row):
|
||||
self.hash.update((json.dumps(row, ensure_ascii=False, allow_nan=False, separators=(",", ":")) + "\n").encode())
|
||||
self.count += 1
|
||||
|
||||
def result(self):
|
||||
return {"count": self.count, "sha256": self.hash.hexdigest()}
|
||||
@@ -0,0 +1,138 @@
|
||||
import hashlib
|
||||
import json
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
|
||||
from .schema import TABLES, FORMAT_VERSION, Digest, canonical_json
|
||||
from .weeks import WEEK, timestamp
|
||||
from .errors import SafetyError
|
||||
|
||||
|
||||
def connect(path, readonly=False):
|
||||
if readonly:
|
||||
return sqlite3.connect(path.resolve().as_uri() + "?mode=ro", uri=True)
|
||||
db = sqlite3.connect(path)
|
||||
db.execute("PRAGMA journal_mode=DELETE")
|
||||
return db
|
||||
|
||||
|
||||
def sqlite_columns(table):
|
||||
return [(name, "INTEGER" if typ in ("bigint", "integer") else "TEXT", int(not nullable and name != table.pk), int(name == table.pk))
|
||||
for name, typ, nullable in table.columns]
|
||||
|
||||
|
||||
def create(path):
|
||||
with connect(path) as db:
|
||||
for table in TABLES:
|
||||
cols = [f'{name} {typ}' + (" PRIMARY KEY" if pk else " NOT NULL" if notnull else "")
|
||||
for name, typ, notnull, pk in sqlite_columns(table)]
|
||||
db.execute(f"CREATE TABLE {table.name} ({', '.join(cols)})")
|
||||
if table.retention == "recorded_at":
|
||||
db.execute(f"CREATE INDEX {table.name}_recorded_at_id ON {table.name}(recorded_at, id)")
|
||||
db.execute("CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
|
||||
db.close()
|
||||
|
||||
|
||||
def upsert(db, table, rows):
|
||||
placeholders = ",".join("?" for _ in table.columns)
|
||||
db.executemany(f"INSERT OR REPLACE INTO {table.name} VALUES ({placeholders})", rows)
|
||||
|
||||
|
||||
def rows(db, table, batch_rows):
|
||||
cur = db.execute(f"SELECT {', '.join(table.names)} FROM {table.name} ORDER BY {table.pk}")
|
||||
while batch := cur.fetchmany(batch_rows):
|
||||
yield batch
|
||||
|
||||
|
||||
def stats(db, start, batch_rows):
|
||||
result = {}
|
||||
for table in TABLES:
|
||||
digest = Digest()
|
||||
for batch in rows(db, table, batch_rows):
|
||||
for row in batch:
|
||||
for value, (_, typ, nullable) in zip(row, table.columns, strict=True):
|
||||
if value is None:
|
||||
if not nullable:
|
||||
raise ValueError("invalid archive NULL")
|
||||
elif typ in ("bigint", "integer"):
|
||||
if type(value) is not int or not -(2**63) <= value < 2**63:
|
||||
raise ValueError("invalid archive integer")
|
||||
if typ == "integer" and not -(2**31) <= value < 2**31:
|
||||
raise ValueError("invalid archive integer")
|
||||
elif not isinstance(value, str):
|
||||
raise ValueError("invalid archive text")
|
||||
elif typ.startswith("numeric"):
|
||||
number = Decimal(value)
|
||||
precision, scale = map(int, typ[8:-1].split(','))
|
||||
if (not number.is_finite() or number.as_tuple().exponent != -scale
|
||||
or abs(number) >= Decimal(10) ** (precision - scale)):
|
||||
raise SafetyError("invalid archive decimal precision/scale")
|
||||
elif typ == "jsonb":
|
||||
if canonical_json(value) != value:
|
||||
raise ValueError("noncanonical archive JSON")
|
||||
elif typ == "timestamp with time zone":
|
||||
if timestamp(datetime.fromisoformat(value)) != value:
|
||||
raise ValueError("noncanonical archive timestamp")
|
||||
value = row[table.names.index(table.retention)]
|
||||
instant = datetime.fromtimestamp(value, start.tzinfo) if table.retention == "ts" else datetime.fromisoformat(value)
|
||||
if not start <= instant < start + WEEK:
|
||||
raise ValueError("archive row outside week")
|
||||
digest.add(row)
|
||||
result[table.name] = digest.result()
|
||||
return result
|
||||
|
||||
|
||||
def finalize(path, start, database, created_at, batch_rows):
|
||||
with connect(path) as db:
|
||||
metadata = {"format_version": FORMAT_VERSION, "source_database": database, "source_schema": "public",
|
||||
"source_columns": {t.name: list(t.columns) for t in TABLES},
|
||||
"week_start": timestamp(start), "week_end": timestamp(start + WEEK),
|
||||
"created_at": timestamp(created_at), "tables": stats(db, start, batch_rows)}
|
||||
db.execute("DELETE FROM metadata")
|
||||
db.executemany("INSERT INTO metadata VALUES (?, ?)", [(k, json.dumps(v, sort_keys=True)) for k, v in metadata.items()])
|
||||
db.close()
|
||||
return validate(path, start, database, batch_rows)
|
||||
|
||||
|
||||
def validate(path, start, database, batch_rows):
|
||||
db = connect(path, readonly=True)
|
||||
try:
|
||||
if db.execute("PRAGMA integrity_check").fetchall() != [("ok",)]:
|
||||
raise SafetyError("SQLite integrity check failed")
|
||||
if db.execute("SELECT count(*) FROM sqlite_master WHERE type IN ('trigger','view')").fetchone()[0]:
|
||||
raise SafetyError("incompatible archive triggers/views")
|
||||
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
||||
if tables != {t.name for t in TABLES} | {"metadata"}:
|
||||
raise SafetyError("incompatible archive tables")
|
||||
for table in TABLES:
|
||||
actual = [(r[1], r[2], r[3], r[5]) for r in db.execute(f"PRAGMA table_info({table.name})")]
|
||||
if actual != sqlite_columns(table):
|
||||
raise SafetyError("incompatible archive schema")
|
||||
if table.retention == "recorded_at":
|
||||
indexes = [r[1] for r in db.execute(f"PRAGMA index_list({table.name})")]
|
||||
if not any([r[2] for r in db.execute(f'PRAGMA index_info("{idx}")')] == ["recorded_at", "id"] for idx in indexes):
|
||||
raise SafetyError("missing archive index")
|
||||
if [(r[1], r[2], r[3], r[5]) for r in db.execute("PRAGMA table_info(metadata)")] != [("key", "TEXT", 0, 1), ("value", "TEXT", 1, 0)]:
|
||||
raise SafetyError("incompatible archive metadata schema")
|
||||
meta = {k: json.loads(v) for k, v in db.execute("SELECT key, value FROM metadata")}
|
||||
expected = {"format_version": FORMAT_VERSION, "source_database": database, "source_schema": "public",
|
||||
"source_columns": {t.name: [list(c) for c in t.columns] for t in TABLES},
|
||||
"week_start": timestamp(start), "week_end": timestamp(start + WEEK)}
|
||||
if set(meta) != set(expected) | {"created_at", "tables"} or any(json.dumps(meta.get(k), sort_keys=True) != json.dumps(v, sort_keys=True) for k, v in expected.items()):
|
||||
raise SafetyError("incompatible archive metadata")
|
||||
if timestamp(datetime.fromisoformat(meta["created_at"])) != meta["created_at"]:
|
||||
raise ValueError("invalid archive creation time")
|
||||
if json.dumps(stats(db, start, batch_rows), sort_keys=True) != json.dumps(meta["tables"], sort_keys=True):
|
||||
raise SafetyError("archive content checksum mismatch")
|
||||
return meta
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def file_hash(path):
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as stream:
|
||||
while chunk := stream.read(1024 * 1024):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
@@ -0,0 +1,30 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
UTC = timezone.utc
|
||||
WEEK = timedelta(days=7)
|
||||
|
||||
|
||||
def week_start(instant: datetime) -> datetime:
|
||||
if instant.tzinfo is None:
|
||||
raise ValueError("timezone-aware timestamp required")
|
||||
instant = instant.astimezone(UTC)
|
||||
return instant.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=instant.weekday())
|
||||
|
||||
|
||||
def boundaries(now: datetime) -> tuple[datetime, datetime]:
|
||||
anchor = week_start(now)
|
||||
return anchor, anchor - timedelta(days=28)
|
||||
|
||||
|
||||
def weeks(first: datetime | None, cutoff: datetime):
|
||||
if first is not None:
|
||||
current = week_start(first)
|
||||
while current + WEEK <= cutoff:
|
||||
yield current
|
||||
current += WEEK
|
||||
|
||||
|
||||
def timestamp(value: datetime) -> str:
|
||||
if value.tzinfo is None:
|
||||
raise ValueError("timezone-aware timestamp required")
|
||||
return value.astimezone(UTC).isoformat(timespec="microseconds").replace("+00:00", "Z")
|
||||
Reference in New Issue
Block a user