96 lines
3.7 KiB
Python
96 lines
3.7 KiB
Python
"""Inspect or maintain the Earth change outbox without touching pending events."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import asyncpg
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "backend"))
|
|
|
|
from app.core.config import settings # noqa: E402
|
|
from app.services.earth_db_change_listener import normalize_asyncpg_dsn # noqa: E402
|
|
from app.services.earth_outbox_maintenance import ( # noqa: E402
|
|
DEFAULT_BATCH_SIZE,
|
|
DEFAULT_PAUSE_SECONDS,
|
|
compact_consumed_events,
|
|
configured_retention_days,
|
|
ensure_retention_index,
|
|
purge_consumed_events,
|
|
release_maintenance_lock,
|
|
retention_index_state,
|
|
try_maintenance_lock,
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--apply", action="store_true", help="perform maintenance; default is read-only"
|
|
)
|
|
parser.add_argument(
|
|
"--compact",
|
|
action="store_true",
|
|
help="atomic one-time rewrite; blocks table access until commit",
|
|
)
|
|
parser.add_argument("--retention-days", type=int, default=configured_retention_days())
|
|
parser.add_argument("--batch-size", type=int, default=DEFAULT_BATCH_SIZE)
|
|
parser.add_argument("--max-batches", type=int, default=100)
|
|
parser.add_argument("--pause-seconds", type=float, default=DEFAULT_PAUSE_SECONDS)
|
|
parser.add_argument("--lock-timeout-seconds", type=int, default=10)
|
|
parser.add_argument("--statement-timeout-seconds", type=int, default=900)
|
|
args = parser.parse_args()
|
|
if not 1 <= args.retention_days <= 3650:
|
|
parser.error("--retention-days must be between 1 and 3650")
|
|
if args.batch_size < 1 or args.max_batches < 1 or args.pause_seconds < 0:
|
|
parser.error("batch-size/max-batches must be positive and pause-seconds non-negative")
|
|
if args.lock_timeout_seconds < 1 or args.statement_timeout_seconds < 1:
|
|
parser.error("timeouts must be positive")
|
|
return args
|
|
|
|
|
|
async def run(args: argparse.Namespace) -> None:
|
|
conn = await asyncpg.connect(normalize_asyncpg_dsn(settings.DATABASE_URL))
|
|
try:
|
|
state = await retention_index_state(conn)
|
|
if not args.apply:
|
|
mode = "compact" if args.compact else "batched delete"
|
|
print(f"dry run: mode={mode}, retention_days={args.retention_days}, index={state}")
|
|
print("pending rows are always retained; pass --apply to make changes")
|
|
return
|
|
if not await try_maintenance_lock(conn):
|
|
raise RuntimeError("Earth outbox maintenance is already running")
|
|
try:
|
|
if args.compact:
|
|
retained, pending = await compact_consumed_events(
|
|
conn,
|
|
retention_days=args.retention_days,
|
|
lock_timeout_seconds=args.lock_timeout_seconds,
|
|
statement_timeout_seconds=args.statement_timeout_seconds,
|
|
)
|
|
print(f"compact committed: retained={retained}, pending={pending}")
|
|
await conn.execute("ANALYZE earth_data_change_events")
|
|
await ensure_retention_index(conn)
|
|
if not args.compact:
|
|
deleted = await purge_consumed_events(
|
|
conn,
|
|
retention_days=args.retention_days,
|
|
batch_size=args.batch_size,
|
|
max_batches=args.max_batches,
|
|
pause_seconds=args.pause_seconds,
|
|
)
|
|
print(f"batched delete complete: deleted={deleted}")
|
|
print("retention index: ready")
|
|
finally:
|
|
await release_maintenance_lock(conn)
|
|
finally:
|
|
await conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run(parse_args()))
|