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) <noreply@anthropic.com>
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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())
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user