From c363a5a6dfe58e20c869740f889809a000f60e0f Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 2 Jun 2026 18:55:23 -0400 Subject: [PATCH] fix(retention): add cleanup sweeps + CalDAV orphan reconciliation Drift-audit Group 4 (retention / unbounded growth): - CalDAV pull now reconciles deletions: a previously-synced event whose caldav_uid no longer appears remotely within the synced window is soft-deleted (one batch_id per run, restorable), so a remote delete propagates locally instead of orphaning forever. Guarded on a non-empty fetch so a spurious empty result can't wipe every local copy. Also wrap the blocking fetch in a 120s wait_for and log run duration. - Notifications: hourly loop now purges read notifications older than 30d (unread kept). Table no longer grows without bound. - Auth tokens: new daily sweep deletes password-reset / invitation tokens whose validity window ended >7d ago; wired via start_auth_token_retention_loop in app startup. Both tables previously only flipped used=True, never pruned. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/fabledassistant/app.py | 2 + src/fabledassistant/services/auth.py | 43 ++++++++++++++++ src/fabledassistant/services/caldav_sync.py | 50 ++++++++++++++++--- src/fabledassistant/services/notifications.py | 28 +++++++++++ 4 files changed, 116 insertions(+), 7 deletions(-) diff --git a/src/fabledassistant/app.py b/src/fabledassistant/app.py index 8feb60c..40a7a15 100644 --- a/src/fabledassistant/app.py +++ b/src/fabledassistant/app.py @@ -148,12 +148,14 @@ def create_app() -> Quart: async def startup(): import asyncio + from fabledassistant.services.auth import start_auth_token_retention_loop from fabledassistant.services.embeddings import backfill_note_embeddings from fabledassistant.services.logging import start_log_retention_loop from fabledassistant.services.notifications import start_notification_loop start_log_retention_loop() start_notification_loop() + start_auth_token_retention_loop() # Backfill embeddings for any notes that don't have one. Runs in the # background so it never blocks the server from accepting requests. diff --git a/src/fabledassistant/services/auth.py b/src/fabledassistant/services/auth.py index 0605186..c94bb04 100644 --- a/src/fabledassistant/services/auth.py +++ b/src/fabledassistant/services/auth.py @@ -356,3 +356,46 @@ async def revoke_invitation(invitation_id: int) -> bool: await session.commit() logger.info("Invitation %d revoked", invitation_id) return True + + +# ── Token retention ───────────────────────────────────────────────────── +# Password-reset and invitation tokens are only ever flipped used=True and are +# never pruned, so on a long-lived instance both tables grow without bound. +# A daily sweep deletes any token whose validity window ended over grace_days +# ago (covers both used and naturally-expired rows once they're cold). + +_auth_retention_task = None + + +async def purge_expired_auth_tokens(grace_days: int = 7) -> int: + """Delete password-reset / invitation tokens that expired > grace_days ago.""" + from sqlalchemy import delete + cutoff = datetime.now(timezone.utc) - timedelta(days=grace_days) + removed = 0 + async with async_session() as session: + for model in (PasswordResetToken, InvitationToken): + result = await session.execute( + delete(model).where(model.expires_at < cutoff) + ) + removed += result.rowcount or 0 + await session.commit() + return removed + + +async def _auth_token_retention_loop() -> None: + import asyncio + while True: + await asyncio.sleep(86400) # daily + try: + removed = await purge_expired_auth_tokens() + if removed: + logger.info("Auth token retention: deleted %d expired token(s)", removed) + except Exception: + logger.exception("Error in auth token retention cleanup") + + +def start_auth_token_retention_loop() -> None: + global _auth_retention_task + import asyncio + if _auth_retention_task is None or _auth_retention_task.done(): + _auth_retention_task = asyncio.create_task(_auth_token_retention_loop()) diff --git a/src/fabledassistant/services/caldav_sync.py b/src/fabledassistant/services/caldav_sync.py index 60341fd..0c04919 100644 --- a/src/fabledassistant/services/caldav_sync.py +++ b/src/fabledassistant/services/caldav_sync.py @@ -11,7 +11,7 @@ import uuid from datetime import datetime, timedelta, timezone from typing import Any -from sqlalchemy import select +from sqlalchemy import select, update from fabledassistant.models import async_session from fabledassistant.models.event import Event @@ -20,6 +20,9 @@ logger = logging.getLogger(__name__) _SYNC_PAST_DAYS = 30 _SYNC_FUTURE_DAYS = 180 +# Wall-clock cap on the blocking CalDAV fetch so a hung/slow server can't +# wedge the hourly sweep indefinitely. +_SYNC_TIMEOUT_SECONDS = 120 def _parse_dt(val: Any) -> datetime | None: @@ -116,16 +119,24 @@ async def sync_user_events(user_id: int) -> dict: config = await get_caldav_config(user_id) + started = datetime.now(timezone.utc) + range_start = started - timedelta(days=_SYNC_PAST_DAYS) + range_end = started + timedelta(days=_SYNC_FUTURE_DAYS) + loop = asyncio.get_running_loop() try: - remote_events: list[dict] = await loop.run_in_executor( - None, _sync_one_user, config, user_id + remote_events: list[dict] = await asyncio.wait_for( + loop.run_in_executor(None, _sync_one_user, config, user_id), + timeout=_SYNC_TIMEOUT_SECONDS, ) + except asyncio.TimeoutError: + logger.warning("CalDAV pull sync timed out for user %d after %ds", user_id, _SYNC_TIMEOUT_SECONDS) + return {"error": "CalDAV fetch timed out"} except Exception: logger.warning("CalDAV pull sync failed for user %d", user_id, exc_info=True) return {"error": "CalDAV fetch failed"} - created = updated = unchanged = skipped = 0 + created = updated = unchanged = skipped = deleted = 0 async with async_session() as session: for ev in remote_events: @@ -185,13 +196,38 @@ async def sync_user_events(user_id: int) -> dict: else: unchanged += 1 + # Reconcile deletions: a previously-synced event (has a caldav_uid) + # that no longer appears remotely within the synced window is + # soft-deleted, so a delete on the remote propagates locally instead + # of orphaning forever. Guarded on a non-empty fetch so a spurious + # empty result can't wipe every local copy. + if remote_events: + remote_uids = {e["caldav_uid"] for e in remote_events} + orphan_batch = str(uuid.uuid4()) + orphan_res = await session.execute( + update(Event) + .where( + Event.user_id == user_id, + Event.caldav_uid.isnot(None), + Event.caldav_uid.notin_(remote_uids), + Event.deleted_at.is_(None), + Event.start_dt >= range_start, + Event.start_dt <= range_end, + ) + .values(deleted_at=datetime.now(timezone.utc), deleted_batch_id=orphan_batch) + ) + deleted = orphan_res.rowcount or 0 + await session.commit() + elapsed = (datetime.now(timezone.utc) - started).total_seconds() logger.info( - "CalDAV sync user %d: %d created, %d updated, %d unchanged, %d skipped (trashed)", - user_id, created, updated, unchanged, skipped, + "CalDAV sync user %d: %d created, %d updated, %d unchanged, %d skipped (trashed), " + "%d deleted (orphaned) in %.1fs", + user_id, created, updated, unchanged, skipped, deleted, elapsed, ) - return {"created": created, "updated": updated, "unchanged": unchanged, "skipped": skipped} + return {"created": created, "updated": updated, "unchanged": unchanged, + "skipped": skipped, "deleted": deleted} async def sync_all_users() -> None: diff --git a/src/fabledassistant/services/notifications.py b/src/fabledassistant/services/notifications.py index ac178ef..20a79c2 100644 --- a/src/fabledassistant/services/notifications.py +++ b/src/fabledassistant/services/notifications.py @@ -236,6 +236,28 @@ async def check_due_tasks() -> None: logger.exception("Failed to send task reminder for user %d", user_id) +# Read notifications are kept this long before the hourly sweep deletes them; +# unread are kept regardless. Without a sweep the table grows without bound. +_NOTIFICATION_RETENTION_DAYS = 30 + + +async def purge_old_read_notifications(retention_days: int = _NOTIFICATION_RETENTION_DAYS) -> int: + """Delete already-read in-app notifications older than retention_days.""" + from datetime import timedelta + from sqlalchemy import delete + from fabledassistant.models.notification import Notification + cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days) + async with async_session() as session: + result = await session.execute( + delete(Notification).where( + Notification.read_at.isnot(None), + Notification.read_at < cutoff, + ) + ) + await session.commit() + return result.rowcount or 0 + + async def _notification_loop() -> None: while True: await asyncio.sleep(3600) # hourly @@ -243,6 +265,12 @@ async def _notification_loop() -> None: await check_due_tasks() except Exception: logger.exception("Error in notification loop") + try: + removed = await purge_old_read_notifications() + if removed: + logger.info("Notification retention: deleted %d read notification(s)", removed) + except Exception: + logger.exception("Error in notification retention cleanup") def start_notification_loop() -> None: