From 3957dd540a8b71b67d13b277429b7c4cc72bcaf9 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Thu, 8 Oct 2026 20:20:21 +0200 Subject: [PATCH] Attachments stored once; nightly clean-up of unused files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Attachments are no longer stored apart from their message (they were kept twice: base64 inside the raw message and decoded on their own). The indexer records the MIME part (attachments.part_index, migration 0023); downloads, zip, inline images, thumbnails and forwarding take it out of the message; videos stream from a short-lived copy (kept a day, 1 GiB per user). A background pass on worker start removes the copies stored before, each only after checking it can be taken out again (spec App. D §D.6 item 1, criterion corrected to <= 60%). A nightly job (03:47, after the Trash/Spam purge) removes files nothing refers to any more — deleted or purged messages, previews of attachments gone, uploads no draft has used for a week, expired streaming copies. Files younger than an hour are left alone; a sent message waiting for its Sent copy is held until its job is done. Account removal keeps files a draft still uses. Co-Authored-By: Claude Opus 5.5 --- backend/app/accounts/service.py | 9 +- backend/app/api/compose.py | 7 + backend/app/api/messages.py | 42 ++- backend/app/indexer/attachments.py | 237 +++++++++++++++++ backend/app/indexer/gc.py | 198 ++++++++++++++ backend/app/indexer/indexer.py | 6 +- backend/app/indexer/mime.py | 42 ++- backend/app/indexer/thumbs.py | 22 +- backend/app/models.py | 4 + backend/app/outbox/jobs.py | 31 ++- backend/app/queue.py | 2 +- .../versions/0023_attachment_part_index.py | 36 +++ .../tests/indexer/test_attachments_once.py | 250 ++++++++++++++++++ backend/tests/indexer/test_blob_gc.py | 175 ++++++++++++ backend/workers/run.py | 18 +- backend/workers/worker.py | 7 +- webmail-client-spec_v1.5.md | 4 +- 17 files changed, 1063 insertions(+), 27 deletions(-) create mode 100644 backend/app/indexer/attachments.py create mode 100644 backend/app/indexer/gc.py create mode 100644 backend/migrations/versions/0023_attachment_part_index.py create mode 100644 backend/tests/indexer/test_attachments_once.py create mode 100644 backend/tests/indexer/test_blob_gc.py diff --git a/backend/app/accounts/service.py b/backend/app/accounts/service.py index cba827c..516e11c 100644 --- a/backend/app/accounts/service.py +++ b/backend/app/accounts/service.py @@ -11,7 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.accounts.schemas import CreateAccountRequest from app.accounts.types import ServerConfig from app.indexer import blobstore -from app.models import Account, Attachment, Message, Thread +from app.models import Account, Attachment, ComposeAttachment, Message, Thread from app.security.crypto import encrypt_secret @@ -127,6 +127,13 @@ async def delete_account( .where(Attachment.user_id == user_id, Attachment.blob_sha256 == sha) .limit(1) ) + if still_used is None: + # A forwarded file waiting in a draft shares the hash: keep it for that. + still_used = await session.scalar( + select(ComposeAttachment.id) + .where(ComposeAttachment.user_id == user_id, ComposeAttachment.blob_sha256 == sha) + .limit(1) + ) if still_used is None: blobstore.remove(user_id, sha) return True diff --git a/backend/app/api/compose.py b/backend/app/api/compose.py index d11b515..71c4f81 100644 --- a/backend/app/api/compose.py +++ b/backend/app/api/compose.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import uuid from datetime import UTC, datetime, timedelta from typing import Annotated @@ -15,6 +16,7 @@ from app.api.schemas import AttachmentOut from app.config import get_settings from app.db import get_session from app.identity.deps import CurrentUser, require_csrf +from app.indexer import attachments as att_store from app.indexer import blobstore from app.models import Attachment, BodyPart, ComposeAttachment, Draft, Mailbox, Message, Outbox from app.outbox.jobs import RETRY_WINDOW @@ -207,6 +209,11 @@ async def adopt_remote_draft(message_id: uuid.UUID, user: CurrentUser, db: DbSes ) attachment_ids: list[str] = [] for a in att_rows: + # Kept only inside the message: give the compose attachment a blob of its own. + if not await asyncio.to_thread( + att_store.materialize, user.id, msg.blob_sha256, a.blob_sha256, a.part_index + ): + continue ca = ComposeAttachment( user_id=user.id, filename=a.filename, diff --git a/backend/app/api/messages.py b/backend/app/api/messages.py index 1cce5e0..55c8916 100644 --- a/backend/app/api/messages.py +++ b/backend/app/api/messages.py @@ -20,6 +20,7 @@ from app.api.schemas import AttachmentOut from app.api.settings import add_allowed_image_sender, allowed_image_senders from app.db import get_session from app.identity.deps import CurrentUser, require_csrf +from app.indexer import attachments as att_store from app.indexer import blobstore, thumbs from app.indexer.mime import parse_message from app.indexer.sanitize import SANITIZER_VERSION, sanitize_html @@ -176,7 +177,7 @@ async def message_cid( ) if att is None: raise_api_error(404, "not_found", "inline part not found") - data = blobstore.load(msg.user_id, att.blob_sha256) + data = await _attachment_bytes(msg, att) media_type = att.mime_type if att.mime_type in _INLINE_SAFE else "application/octet-stream" return Response(content=data, media_type=media_type) @@ -190,6 +191,16 @@ def _content_disposition(disposition: str, filename: str) -> str: return f"{disposition}; filename=\"{fallback}\"; filename*=UTF-8''{quote(filename)}" +async def _attachment_bytes(msg: Message, att: Attachment) -> bytes: + """An attachment's bytes, taken out of its message (404 if they cannot be).""" + try: + return await asyncio.to_thread( + att_store.load, msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index + ) + except FileNotFoundError: + raise_api_error(404, "not_found", "attachment content missing") + + async def _real_attachments(db: AsyncSession, message_id: uuid.UUID) -> list[Attachment]: """Downloadable (non-inline) attachments for a message, in a stable order.""" rows = await db.scalars( @@ -221,9 +232,10 @@ async def forward_attachments( message_id: uuid.UUID, user: CurrentUser, db: DbSession ) -> list[AttachmentOut]: """Attach a message's files to a forward of it: one compose attachment per - real (non-inline) attachment, sharing the stored blob — nothing is copied. + real (non-inline) attachment. Each is taken out of the message into a blob + of its own (shared, by hash, with any other use of the same file). Returns them as the composer lists its attachments.""" - await _owned_message(db, user.id, message_id) + msg = await _owned_message(db, user.id, message_id) rows = [ ComposeAttachment( user_id=user.id, @@ -233,7 +245,9 @@ async def forward_attachments( blob_sha256=a.blob_sha256, ) for a in await _real_attachments(db, message_id) - if blobstore.exists(user.id, a.blob_sha256) + if await asyncio.to_thread( + att_store.materialize, user.id, msg.blob_sha256, a.blob_sha256, a.part_index + ) ] db.add_all(rows) await db.commit() @@ -276,7 +290,7 @@ async def download_attachments_zip( name = f"{stem}-{seen[name]}{dot}{ext}" if dot else f"{name}-{seen[name]}" else: seen[name] = 1 - zf.writestr(name, blobstore.load(msg.user_id, att.blob_sha256)) + zf.writestr(name, await _attachment_bytes(msg, att)) return Response( content=buffer.getvalue(), @@ -301,7 +315,14 @@ async def attachment_thumbnail( kind = thumb_type(att.mime_type, att.filename) if att is not None else None if att is None or kind is None: raise_api_error(404, "not_found", "no thumbnail") - data = await asyncio.to_thread(thumbs.thumbnail, msg.user_id, att.blob_sha256, kind) + data = await asyncio.to_thread( + thumbs.thumbnail, + msg.user_id, + att.blob_sha256, + kind, + att.size_bytes, + lambda: att_store.load(msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index), + ) if data is None: raise_api_error(404, "not_found", "no thumbnail") # An attachment never changes under its id: the browser may keep this. @@ -329,8 +350,13 @@ async def message_attachment( ) if att is None: raise_api_error(404, "not_found", "attachment not found") - path = blobstore.blob_path(msg.user_id, att.blob_sha256) - if not path.exists(): + try: + # Taken out of the message into a short-lived file (kept a day after + # its last use), so it can be streamed with HTTP Range like before. + path = await asyncio.to_thread( + att_store.file, msg.user_id, msg.blob_sha256, att.blob_sha256, att.part_index + ) + except FileNotFoundError: raise_api_error(404, "not_found", "attachment content missing") inline_type = None if download else _inline_type(att.mime_type, att.filename) disposition = "inline" if inline_type else "attachment" diff --git a/backend/app/indexer/attachments.py b/backend/app/indexer/attachments.py new file mode 100644 index 0000000..ae3f812 --- /dev/null +++ b/backend/app/indexer/attachments.py @@ -0,0 +1,237 @@ +"""Attachments, kept once: inside their message (SPEC App. D §D.6 item 1). + +An attachment is not stored apart from its message any more — it was, which +kept most of a mailbox's bytes twice (once base64 in the raw message, once +decoded on its own). It is taken out of the raw message when wanted: + +* :func:`load` — its bytes (downloads, zip, inline images, forwarding, thumbnails); +* :func:`file` — a file holding it, for streaming with HTTP Range (videos): + taken out into ``{user}/cache/`` and kept a day after its last use; +* :func:`dedupe_existing` — the one-time pass removing the separate copies + stored before, each only after checking it can be taken out of its message. + +Blocking (parsing, disk): call these in a thread. +""" + +from __future__ import annotations + +import contextlib +import hashlib +import logging +import time +import uuid +from collections import defaultdict +from pathlib import Path + +from app.indexer import blobstore +from app.indexer.mime import load_message, part_payload + +logger = logging.getLogger(__name__) + +CACHE_KEEP_S = 24 * 3600 # a taken-out file is kept this long after its last use +CACHE_MAX_BYTES = 1024**3 # per user; the least recently used go first beyond it +# part_index of a row whose content is not in its message (kept as its own file). +NOT_IN_MESSAGE = -1 + + +def _legacy(user_id: uuid.UUID, sha: str) -> Path | None: + """The separate copy stored before attachments were kept once, if still there.""" + path = blobstore.blob_path(user_id, sha) + return path if path.exists() else None + + +def load(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> bytes: + """An attachment's bytes. Raises FileNotFoundError when they cannot be had.""" + legacy = _legacy(user_id, sha) + if legacy is not None: + return legacy.read_bytes() + raw = blobstore.blob_path(user_id, message_sha) + if part_index == NOT_IN_MESSAGE or not raw.exists(): + raise FileNotFoundError(sha) + payload = part_payload(load_message(path=raw), part_index, sha) + if payload is None: + raise FileNotFoundError(sha) + return payload + + +def materialize(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> bool: + """Make sure the attachment exists as a blob of its own — for a compose + attachment (forward, a draft reopened), which is read by its hash. False + when its bytes cannot be had.""" + if blobstore.exists(user_id, sha): + return True + try: + blobstore.store(user_id, load(user_id, message_sha, sha, part_index)) + except FileNotFoundError: + return False + return True + + +def _cache_dir(user_id: uuid.UUID) -> Path: + return blobstore.blob_path(user_id, "00").parent.parent / "cache" + + +def prune_cache(cache: Path) -> None: + """Drop streaming copies past their keep time, and the oldest beyond the cap.""" + files = [(p, p.stat()) for p in cache.iterdir() if p.is_file()] + now = time.time() + total = 0 + for path, st in sorted(files, key=lambda f: f[1].st_mtime, reverse=True): + total += st.st_size + if now - st.st_mtime > CACHE_KEEP_S or total > CACHE_MAX_BYTES: + path.unlink(missing_ok=True) + + +def file(user_id: uuid.UUID, message_sha: str, sha: str, part_index: int | None) -> Path: + """A file holding the attachment, to stream from. Raises FileNotFoundError.""" + legacy = _legacy(user_id, sha) + if legacy is not None: + return legacy + cache = _cache_dir(user_id) + path = cache / sha + if path.exists(): + path.touch() # used again: kept longer + return path + data = load(user_id, message_sha, sha, part_index) + cache.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(f".{uuid.uuid4().hex}.tmp") + tmp.write_bytes(data) + tmp.replace(path) + with contextlib.suppress(OSError): # a file removed meanwhile by another request + prune_cache(cache) + return path + + +def drop_cached(user_id: uuid.UUID) -> None: + """Forget every taken-out file of a user (account removed).""" + cache = _cache_dir(user_id) + if cache.exists(): + for path in cache.iterdir(): + path.unlink(missing_ok=True) + + +# ---- removing the copies stored before ------------------------------------------------ + +# A file this recent may belong to work in progress (a sent message waiting +# for its Sent copy is referenced only by its job): left for the next pass. +_MIN_AGE_S = 3600 + + +def _part_indexes(raw_path: Path) -> dict[str, int]: + """Content hash → MIME part position, for every leaf part of a message.""" + out: dict[str, int] = {} + for index, part in enumerate(load_message(path=raw_path).walk()): + if part.is_multipart(): + continue + payload = part.get_payload(decode=True) + if isinstance(payload, bytes): + out.setdefault(hashlib.sha256(payload).hexdigest(), index) + return out + + +async def dedupe_existing(batch: int = 200, *, user_id: uuid.UUID | None = None) -> dict[str, int]: + """Record where each older attachment sits in its message, and delete its + separate copy once it is known to be there. Idempotent; resumable.""" + import asyncio + + from sqlalchemy import select, update + + from app.db import session_scope + from app.models import Attachment, ComposeAttachment, Message + + stats = {"rows": 0, "found": 0, "not_found": 0, "files_removed": 0, "bytes_freed": 0} + while True: + async with session_scope() as db: + rows = ( + await db.execute( + select( + Attachment.id, + Attachment.user_id, + Attachment.blob_sha256, + Message.blob_sha256, + ) + .join(Message, Message.id == Attachment.message_id) + .where( + Attachment.part_index.is_(None), + *([Attachment.user_id == user_id] if user_id is not None else []), + ) + .order_by(Attachment.message_id) + .limit(batch) + ) + ).all() + if not rows: + return stats + by_message: dict[tuple[uuid.UUID, str], list[tuple[uuid.UUID, str]]] = defaultdict(list) + for att_id, owner, sha, msg_sha in rows: + by_message[(owner, msg_sha)].append((att_id, sha)) + + located: dict[uuid.UUID, int] = {} + for (owner, msg_sha), atts in by_message.items(): + raw = blobstore.blob_path(owner, msg_sha) + try: + indexes = await asyncio.to_thread(_part_indexes, raw) if raw.exists() else {} + except Exception: # noqa: BLE001 - unreadable message: keep the copies + logger.warning("attachments: cannot parse %s", raw, exc_info=True) + indexes = {} + for att_id, sha in atts: + located[att_id] = indexes.get(sha, NOT_IN_MESSAGE) + + async with session_scope() as db: + for att_id, index in located.items(): + await db.execute( + update(Attachment).where(Attachment.id == att_id).values(part_index=index) + ) + stats["rows"] += len(located) + stats["found"] += sum(1 for i in located.values() if i != NOT_IN_MESSAGE) + stats["not_found"] += sum(1 for i in located.values() if i == NOT_IN_MESSAGE) + + # A separate copy goes only when every attachment row with that content + # is found in its message and nothing else uses the file. + candidates = { + (owner, sha) + for att_id, owner, sha, _ in rows + if located.get(att_id, NOT_IN_MESSAGE) != NOT_IN_MESSAGE + } + async with session_scope() as db: + for owner, sha in candidates: + path = blobstore.blob_path(owner, sha) + if not path.exists() or time.time() - path.stat().st_mtime < _MIN_AGE_S: + continue + in_use = ( + await db.scalar( + select(Message.id) + .where(Message.user_id == owner, Message.blob_sha256 == sha) + .limit(1) + ) + or await db.scalar( + select(ComposeAttachment.id) + .where( + ComposeAttachment.user_id == owner, + ComposeAttachment.blob_sha256 == sha, + ) + .limit(1) + ) + or await db.scalar( + select(Attachment.id) + .where( + Attachment.user_id == owner, + Attachment.blob_sha256 == sha, + (Attachment.part_index.is_(None)) + | (Attachment.part_index == NOT_IN_MESSAGE), + ) + .limit(1) + ) + ) + if in_use: + continue + size = path.stat().st_size + path.unlink(missing_ok=True) + stats["files_removed"] += 1 + stats["bytes_freed"] += size + + +async def dedupe_existing_job(ctx: dict[str, object]) -> dict[str, int]: + """arq job (bulk queue): :func:`dedupe_existing`, run once after an update.""" + stats = await dedupe_existing() + logger.info("attachments kept once: %s", stats) + return stats diff --git a/backend/app/indexer/gc.py b/backend/app/indexer/gc.py new file mode 100644 index 0000000..f8f2b6e --- /dev/null +++ b/backend/app/indexer/gc.py @@ -0,0 +1,198 @@ +"""The nightly clean-up of stored files nothing uses any more. + +Messages leave in many ways — Delete forever, Empty Trash, the Trash/Spam +retention purge, an expunge on the server, a folder re-synced — and none of +them removes the message's file: files are stored by content, so one file can +serve several rows (the same email in two folders, a forwarded file in a +draft), and removing it is only safe once nothing refers to it. This pass does +that for every user, whatever the way the rows went: + +* uploads (compose attachments) no draft has referred to for a week — left by + a sent message (its draft is gone) or a composer closed without saving; +* message/attachment files no row refers to; +* attachment previews (thumbnails) of attachments that are gone; +* streaming copies past their keep time (app/indexer/attachments.py). + +A file younger than an hour is left alone: it may belong to work in progress +(a message being stored, a sent message waiting for its Sent copy). +""" + +from __future__ import annotations + +import datetime as dt +import logging +import re +import time +import uuid +from pathlib import Path +from typing import Any + +from sqlalchemy import delete, or_, select + +from app.config import get_settings +from app.indexer import attachments as att_store + +logger = logging.getLogger(__name__) + +MIN_AGE_S = 3600 +UPLOAD_KEEP = dt.timedelta(days=7) +_SHA = re.compile(r"^[0-9a-f]{64}$") + + +# ---- files a job still needs --------------------------------------------------------- +# +# A sent message whose Sent copy could not be filed yet is referenced only by +# its retrying job (app/outbox/jobs.py), for up to a day: held here meanwhile. +# Entries older than HOLD_MAX are ignored, in case a job died without letting go. + +HOLD_MAX_S = 3 * 24 * 3600 + + +def _hold_key(user_id: uuid.UUID | str) -> str: + return f"blob:held:{user_id}" + + +async def hold(user_id: uuid.UUID | str, sha: str) -> None: + from app.redis_client import get_redis + + await get_redis().zadd(_hold_key(user_id), {sha: time.time()}) + + +async def release(user_id: uuid.UUID | str, sha: str) -> None: + from app.redis_client import get_redis + + await get_redis().zrem(_hold_key(user_id), sha) + + +async def _held(user_id: uuid.UUID) -> set[str]: + from app.redis_client import get_redis + + redis = get_redis() + oldest = time.time() - HOLD_MAX_S + await redis.zremrangebyscore(_hold_key(user_id), "-inf", oldest) + return { + m.decode() if isinstance(m, bytes) else str(m) + for m in await redis.zrangebyscore(_hold_key(user_id), oldest, "+inf") + } + + +def _draft_upload_ids(payloads: list[dict[str, Any]]) -> set[uuid.UUID]: + out: set[uuid.UUID] = set() + for payload in payloads: + for value in payload.get("attachment_ids") or []: + try: + out.add(uuid.UUID(str(value))) + except ValueError: + continue + return out + + +def _old(path: Path, now: float) -> bool: + return now - path.stat().st_mtime >= MIN_AGE_S + + +async def sweep_user(user_id: uuid.UUID) -> dict[str, int]: + """Clean one user's stored files; returns what was removed.""" + from app.db import session_scope + from app.models import Attachment, ComposeAttachment, Draft, Message + + stats = {"uploads": 0, "files": 0, "bytes": 0, "thumbs": 0} + async with session_scope() as db: + # Uploads no draft refers to, once a week old. + used = _draft_upload_ids( + list(await db.scalars(select(Draft.payload).where(Draft.user_id == user_id))) + ) + stale = select(ComposeAttachment.id).where( + ComposeAttachment.user_id == user_id, + ComposeAttachment.created_at < dt.datetime.now(dt.UTC) - UPLOAD_KEEP, + ) + if used: + stale = stale.where(ComposeAttachment.id.not_in(used)) + gone = list(await db.scalars(stale)) + if gone: + await db.execute(delete(ComposeAttachment).where(ComposeAttachment.id.in_(gone))) + stats["uploads"] = len(gone) + + async with session_scope() as db: + # Every file still referred to. Attachments count only where their file + # is their only copy (not found in the message, or not yet looked for). + referenced = set( + await db.scalars(select(Message.blob_sha256).where(Message.user_id == user_id)) + ) + referenced |= set( + await db.scalars( + select(ComposeAttachment.blob_sha256).where(ComposeAttachment.user_id == user_id) + ) + ) + referenced |= set( + await db.scalars( + select(Attachment.blob_sha256).where( + Attachment.user_id == user_id, + or_( + Attachment.part_index.is_(None), + Attachment.part_index == att_store.NOT_IN_MESSAGE, + ), + ) + ) + ) + referenced |= await _held(user_id) + attachment_shas = set( + await db.scalars(select(Attachment.blob_sha256).where(Attachment.user_id == user_id)) + ) + + root = Path(get_settings().blob_root) / str(user_id) + if not root.is_dir(): + return stats + now = time.time() + for path in root.glob("??/*"): + if not path.is_file() or not _SHA.match(path.name) or path.name in referenced: + continue + if not _old(path, now): + continue + size = path.stat().st_size + path.unlink(missing_ok=True) + stats["files"] += 1 + stats["bytes"] += size + thumbs = root / "thumbs" + if thumbs.is_dir(): + for path in thumbs.iterdir(): + sha = path.name.split("-", 1)[0] + if ( + path.is_file() + and _SHA.match(sha) + and sha not in attachment_shas + and _old(path, now) + ): + path.unlink(missing_ok=True) + stats["thumbs"] += 1 + cache = root / "cache" + if cache.is_dir(): + att_store.prune_cache(cache) + return stats + + +async def sweep(user_id: uuid.UUID | None = None) -> dict[str, int]: + """:func:`sweep_user` for one user, or every user.""" + from app.db import session_scope + from app.models import User + + async with session_scope() as db: + users = [user_id] if user_id is not None else list(await db.scalars(select(User.id))) + total = {"users": 0, "uploads": 0, "files": 0, "bytes": 0, "thumbs": 0} + for uid in users: + try: + stats = await sweep_user(uid) + except Exception: # noqa: BLE001 - one user's trouble must not stop the rest + logger.warning("stored-file clean-up failed for user %s", uid, exc_info=True) + continue + total["users"] += 1 + for key, value in stats.items(): + total[key] += value + return total + + +async def sweep_job(ctx: dict[str, Any]) -> dict[str, int]: + """arq cron job: the nightly :func:`sweep`.""" + stats = await sweep() + logger.info("stored-file clean-up: %s", stats) + return stats diff --git a/backend/app/indexer/indexer.py b/backend/app/indexer/indexer.py index 3443d23..9283d23 100644 --- a/backend/app/indexer/indexer.py +++ b/backend/app/indexer/indexer.py @@ -216,8 +216,9 @@ async def persist_message( ) ) + # Attachments are not stored apart: they stay inside the raw blob, and are + # taken out of it when wanted (app/indexer/attachments.py). for att in parsed.attachments: - att_sha = blobstore.store(user_id, att.payload) session.add( Attachment( message_id=message.id, @@ -226,7 +227,8 @@ async def persist_message( mime_type=att.mime_type, size_bytes=len(att.payload), content_id=att.content_id, - blob_sha256=att_sha, + blob_sha256=blobstore.sha256_hex(att.payload), + part_index=att.part_index, ) ) diff --git a/backend/app/indexer/mime.py b/backend/app/indexer/mime.py index 67fbb3e..e60653d 100644 --- a/backend/app/indexer/mime.py +++ b/backend/app/indexer/mime.py @@ -41,6 +41,9 @@ class ParsedAttachment: mime_type: str content_id: str | None payload: bytes + # The part's position in ``msg.walk()``: where to find it again in the raw + # message, which is the only copy kept (attachments are not stored apart). + part_index: int = -1 @dataclass @@ -114,7 +117,7 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]: # or inline resource if it has a filename, a Content-ID, or attachment disposition; # the body text/plain and text/html parts have none of these and are skipped. out: list[ParsedAttachment] = [] - for part in msg.walk(): + for index, part in enumerate(msg.walk()): if part.is_multipart() or not isinstance(part, EmailMessage): continue cid = part.get("Content-ID") @@ -137,13 +140,14 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]: mime_type=part.get_content_type(), content_id=cid.strip("<>") if (inline and cid is not None) else None, payload=payload, + part_index=index, ) ) # An invitation sent only as the text/calendar alternative (Outlook, often) has # no file to show: keep it as invite.ics, so the event can be seen, answered # and downloaded like a Gmail one. Not when the message attaches its own .ics. if not any(a.filename.lower().endswith(".ics") for a in out): - for part in msg.walk(): + for index, part in enumerate(msg.walk()): if part.get_content_type() != "text/calendar" or part.get_filename(): continue payload = part.get_payload(decode=True) @@ -154,6 +158,7 @@ def _extract_attachments(msg: EmailMessage) -> list[ParsedAttachment]: mime_type="text/calendar", content_id=None, payload=payload, + part_index=index, ) ) break @@ -215,3 +220,36 @@ def _parse_email(msg: EmailMessage) -> ParsedEmail: html_body=_scrub(html_body) if html_body is not None else None, attachments=_extract_attachments(msg), ) + + +def load_message( + raw: bytes | None = None, path: str | os.PathLike[str] | None = None +) -> EmailMessage: + """The parsed tree of a raw message, from memory or straight from its file.""" + if path is not None: + with open(path, "rb") as handle: + msg = email.message_from_binary_file(handle, policy=default_policy) + else: + msg = email.message_from_bytes(raw or b"", policy=default_policy) + assert isinstance(msg, EmailMessage) + return msg + + +def part_payload(msg: EmailMessage, part_index: int | None, sha256: str) -> bytes | None: + """The decoded bytes of the attachment whose content hashes to ``sha256``: + the part at ``part_index`` when that is it, else whichever part is (rows + stored before the index was kept). None when the message has no such part.""" + import hashlib + + parts = list(msg.walk()) + if part_index is not None and 0 <= part_index < len(parts): + payload = parts[part_index].get_payload(decode=True) + if isinstance(payload, bytes) and hashlib.sha256(payload).hexdigest() == sha256: + return payload + for part in parts: + if part.is_multipart(): + continue + payload = part.get_payload(decode=True) + if isinstance(payload, bytes) and hashlib.sha256(payload).hexdigest() == sha256: + return payload + return None diff --git a/backend/app/indexer/thumbs.py b/backend/app/indexer/thumbs.py index 0758e31..e4db539 100644 --- a/backend/app/indexer/thumbs.py +++ b/backend/app/indexer/thumbs.py @@ -15,6 +15,7 @@ from __future__ import annotations import io import logging import uuid +from collections.abc import Callable from pathlib import Path from app.indexer import blobstore @@ -86,8 +87,16 @@ def _render(data: bytes, mime_type: str) -> bytes | None: return out.getvalue() -def thumbnail(user_id: uuid.UUID, sha: str, mime_type: str) -> bytes | None: - """The thumbnail of a stored attachment, or None if it cannot have one. +def thumbnail( + user_id: uuid.UUID, + sha: str, + mime_type: str, + size: int, + load: Callable[[], bytes], +) -> bytes | None: + """The thumbnail of an attachment (content hash ``sha``, ``size`` bytes, + read with ``load`` only when it is not made yet), or None if it cannot + have one. Blocking (decoding is CPU work): call it in a thread. """ @@ -97,11 +106,14 @@ def thumbnail(user_id: uuid.UUID, sha: str, mime_type: str) -> bytes | None: none = _path(user_id, sha, ".none") if none.exists() or not can_thumbnail(mime_type): return None - path = blobstore.blob_path(user_id, sha) - if not path.exists() or path.stat().st_size > _MAX_INPUT_BYTES: + if size > _MAX_INPUT_BYTES: return None try: - thumb = _render(path.read_bytes(), mime_type) + data = load() + except FileNotFoundError: + return None + try: + thumb = _render(data, mime_type) except Exception as exc: # noqa: BLE001 - any undecodable file: no thumbnail logger.info("no thumbnail for blob %s (%s): %s", sha, mime_type, exc) thumb = None diff --git a/backend/app/models.py b/backend/app/models.py index 8866de4..1509426 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -278,7 +278,11 @@ class Attachment(Base): mime_type: Mapped[str] = mapped_column(Text, nullable=False) size_bytes: Mapped[int] = mapped_column(Integer, nullable=False) content_id: Mapped[str | None] = mapped_column(Text) + # The content's hash (names its thumbnail, checks what is taken out). The + # bytes themselves live only inside the message's raw blob, at MIME part + # ``part_index`` (NULL for rows stored before that was recorded). blob_sha256: Mapped[str] = mapped_column(Text, nullable=False) + part_index: Mapped[int | None] = mapped_column(Integer) class ComposeAttachment(Base): diff --git a/backend/app/outbox/jobs.py b/backend/app/outbox/jobs.py index ee0995d..57dbc7d 100644 --- a/backend/app/outbox/jobs.py +++ b/backend/app/outbox/jobs.py @@ -14,15 +14,17 @@ from datetime import UTC, datetime, timedelta from email.utils import getaddresses from typing import Any +from arq import Retry from sqlalchemy import select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from app.db import session_scope from app.events import get_event_bus -from app.indexer import blobstore +from app.indexer import blobstore, gc from app.indexer.indexer import persist_message from app.models import Account, Draft, KnownRecipient, Mailbox, Message, Outbox +from app.outbox import inline_images from app.outbox.attachments import load_attachments from app.outbox.mime_build import build_mime from app.outbox.service import delete_mirrored_draft_row @@ -171,7 +173,21 @@ async def _send_failed( async def file_sent_copy_job(ctx: dict[str, Any], account_id: str, user_id: str, sha: str) -> None: """File a sent message's copy in Sent, after the send itself succeeded but - the IMAP server could not be reached. Retried (arq Retry) for ~24 hours.""" + the IMAP server could not be reached. Retried (arq Retry) for ~24 hours. + + Until then the copy's file is referenced only by this job: it is held from + the nightly clean-up (app/indexer/gc.py) and let go once filed or given up.""" + try: + await _file_sent_copy(ctx, account_id, user_id, sha) + except Retry: + raise # tried again later: the file is still needed + except BaseException: + await gc.release(user_id, sha) + raise + await gc.release(user_id, sha) + + +async def _file_sent_copy(ctx: dict[str, Any], account_id: str, user_id: str, sha: str) -> None: uid = uuid.UUID(user_id) async with session_scope() as session: account = await session.get(Account, uuid.UUID(account_id)) @@ -236,8 +252,16 @@ async def send_job(ctx: dict[str, Any], outbox_id: str) -> None: return user_id = outbox.user_id attachments = await load_attachments(session, user_id, draft) + # Images quoted from the original travel inside the message, not as + # links to this server (app/outbox/inline_images.py). + html, inline = await inline_images.embed( + session, + user_id, + str(draft.payload.get("html", "") or ""), + account.email.rsplit("@", 1)[-1], + ) transmit_raw, sent_raw, envelope_from, envelope_to = build_mime( - account, draft, reply_to, attachments + account, draft, reply_to, attachments, html=html, inline=inline ) sent_path = await session.scalar( select(Mailbox.path).where( @@ -271,6 +295,7 @@ async def send_job(ctx: dict[str, Any], outbox_id: str) -> None: except Exception as exc: # noqa: BLE001 logger.warning("outbox %s sent; filing it in Sent failed, will retry: %s", oid, exc) sha = blobstore.store(user_id, sent_raw) + await gc.hold(user_id, sha) # until filed: only the job refers to it await enqueue_job("file_sent_copy_job", str(account_obj.id), str(user_id), sha) # Remove the mirrored Drafts-folder copy now that it has been sent. if drafts_path and draft_imap_uid: diff --git a/backend/app/queue.py b/backend/app/queue.py index 1e638ac..b3c2d79 100644 --- a/backend/app/queue.py +++ b/backend/app/queue.py @@ -21,7 +21,7 @@ QUEUE_INTERACTIVE = "interactive" QUEUE_BULK = "bulk" # Jobs default to the interactive queue; only bulk backfills go to the bulk queue. -_BULK_JOBS = frozenset({"sync_bootstrap_account"}) +_BULK_JOBS = frozenset({"sync_bootstrap_account", "dedupe_existing_job"}) def queue_for(job_name: str) -> str: diff --git a/backend/migrations/versions/0023_attachment_part_index.py b/backend/migrations/versions/0023_attachment_part_index.py new file mode 100644 index 0000000..d8275df --- /dev/null +++ b/backend/migrations/versions/0023_attachment_part_index.py @@ -0,0 +1,36 @@ +"""attachments are kept once: inside their message + +An attachment used to be stored twice — inside the raw message blob, and again +as a blob of its own. Now only the raw message is kept; ``part_index`` says +which MIME part of it (its position in a walk of the message) holds the +attachment, so it can be taken out when downloaded. ``blob_sha256`` stays: the +attachment's content hash, which names its thumbnail and checks what is taken +out. Rows stored before this have no index; they are found by that hash. + +Existing separate copies are removed by a background pass on the worker +(app/indexer/attachments.py ``dedupe_existing``), only after checking each one +can be taken out of its message again. + +Revision ID: 0023 +Revises: 0022 +Create Date: 2026-10-08 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0023" +down_revision: str | None = "0022" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute("ALTER TABLE attachments ADD COLUMN part_index integer") + + +def downgrade() -> None: + op.execute("ALTER TABLE attachments DROP COLUMN IF EXISTS part_index") diff --git a/backend/tests/indexer/test_attachments_once.py b/backend/tests/indexer/test_attachments_once.py new file mode 100644 index 0000000..16cfea2 --- /dev/null +++ b/backend/tests/indexer/test_attachments_once.py @@ -0,0 +1,250 @@ +"""Attachments are kept once — inside their message (SPEC App. D §D.6 item 1, +acceptance §D.9.5): nothing is stored apart, every way of reading an +attachment returns the exact bytes, and the clean-up pass removes the copies +stored before only when they can be taken out of their message again.""" + +from __future__ import annotations + +import io +import os +import shutil +import time +import uuid +import zipfile +from collections.abc import AsyncIterator +from dataclasses import dataclass +from datetime import UTC, datetime +from email.message import EmailMessage +from pathlib import Path +from types import SimpleNamespace + +import httpx +import pytest_asyncio +from httpx import ASGITransport +from PIL import Image +from sqlalchemy import delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.identity.passwords import hash_password +from app.identity.sessions import create_session +from app.indexer import attachments as att_store +from app.indexer import blobstore +from app.indexer.indexer import persist_message +from app.main import create_app +from app.models import Account, Attachment, ComposeAttachment, Mailbox, Message, User +from app.providers.base import RemoteMessage + + +def _png(w: int, h: int) -> bytes: + out = io.BytesIO() + Image.new("RGB", (w, h), (200, 80, 40)).save(out, "PNG") + return out.getvalue() + + +PNG = _png(640, 360) +VIDEO = os.urandom(300_000) # stands in for an .mp4: only the bytes matter here +NOTES = "Zażółć gęślą jaźń\n".encode() * 50 +LOGO = _png(32, 32) + + +def _raw() -> bytes: + m = EmailMessage() + m["Subject"] = "Files" + m["From"] = "Alice " + m["To"] = "me@example.com" + m["Message-ID"] = f"<{uuid.uuid4().hex}@x.com>" + m["Date"] = "Mon, 09 Mar 2026 08:36:37 +0100" + m.set_content("see attached") + m.add_alternative('

see attached

', subtype="html") + m.get_payload()[1].add_related(LOGO, "image", "png", cid="") + m.add_attachment(PNG, "image", "png", filename="photo.png") + m.add_attachment(VIDEO, "video", "mp4", filename="clip.mp4") + m.add_attachment(NOTES, "text", "plain", filename="notatki.txt") + return m.as_bytes() + + +@dataclass +class Ctx: + client: httpx.AsyncClient + db: AsyncSession + user_id: uuid.UUID + message: Message + by_name: dict[str, Attachment] + + +def _user_dir(user_id: uuid.UUID) -> Path: + return Path(get_settings().blob_root) / str(user_id) + + +def _stored_bytes(user_id: uuid.UUID) -> int: + """Bytes in the user's blob store (not the thumbnail or take-out caches).""" + return sum( + p.stat().st_size + for p in _user_dir(user_id).rglob("*") + if p.is_file() and p.parent.name not in ("thumbs", "cache") + ) + + +@pytest_asyncio.fixture +async def ctx() -> AsyncIterator[Ctx]: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + db = async_sessionmaker(engine, expire_on_commit=False)() + user = User(handle=f"att{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + db.add(user) + await db.flush() + account = Account( + user_id=user.id, + email="me@example.com", + imap_config={}, + smtp_config={}, + secret_ct=b"", + discovery_src="manual", + ) + db.add(account) + await db.flush() + inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox") + db.add(inbox) + await db.commit() + remote = RemoteMessage(uid=1, raw=_raw(), flags=frozenset(), internal_date=datetime.now(UTC)) + result = await persist_message( + db, + user_id=user.id, + account_id=account.id, + mailbox_id=inbox.id, + mailbox_special_use="inbox", + remote=remote, + ) + await db.commit() + assert result is not None + rows = ( + await db.scalars(select(Attachment).where(Attachment.message_id == result.message.id)) + ).all() + token = await create_session(user.id, "pytest") + client = httpx.AsyncClient( + transport=ASGITransport(app=create_app()), + base_url="http://testserver", + cookies={"session": token, "csrf": "t"}, + ) + try: + yield Ctx(client, db, user.id, result.message, {a.filename: a for a in rows}) + finally: + await client.aclose() + await db.execute(delete(User).where(User.id == user.id)) + await db.commit() + await db.close() + await engine.dispose() + shutil.rmtree(_user_dir(user.id), ignore_errors=True) + + +def _url(c: Ctx, name: str, suffix: str = "") -> str: + return f"/api/messages/{c.message.id}/attachments/{c.by_name[name].id}{suffix}" + + +async def test_only_the_message_is_stored(ctx: Ctx) -> None: + assert set(ctx.by_name) == {"photo.png", "clip.mp4", "notatki.txt", "attachment"} + for att in ctx.by_name.values(): + assert att.part_index is not None and att.part_index >= 0 + assert not blobstore.exists(ctx.user_id, att.blob_sha256) # no copy of its own + assert blobstore.exists(ctx.user_id, ctx.message.blob_sha256) + # §D.9.5: at most 60% of what keeping the copies too took (base64 inside + # the message is ~1.37x the file; kept apart as well it was ~2.37x). + raw = _stored_bytes(ctx.user_id) + before = raw + len(PNG) + len(VIDEO) + len(NOTES) + len(LOGO) + assert raw <= 0.60 * before + + +async def test_every_way_of_reading_gives_the_exact_bytes(ctx: Ctx) -> None: + for name, data in (("photo.png", PNG), ("clip.mp4", VIDEO), ("notatki.txt", NOTES)): + resp = await ctx.client.get(_url(ctx, name), params={"download": "true"}) + assert resp.status_code == 200 and resp.content == data, name + ranged = await ctx.client.get(_url(ctx, "clip.mp4"), headers={"Range": "bytes=1000-1999"}) + assert ranged.status_code == 206 and ranged.content == VIDEO[1000:2000] + inline = await ctx.client.get(f"/api/messages/{ctx.message.id}/cid/logo@x") + assert inline.status_code == 200 and inline.content == LOGO + zipped = await ctx.client.get(f"/api/messages/{ctx.message.id}/attachments.zip") + archive = zipfile.ZipFile(io.BytesIO(zipped.content)) + assert archive.read("clip.mp4") == VIDEO and archive.read("notatki.txt") == NOTES + thumb = await ctx.client.get(_url(ctx, "photo.png", "/thumb")) + assert thumb.status_code == 200 and thumb.headers["content-type"] == "image/webp" + # The streamed copy is a short-lived cache, not a second store. + assert (_user_dir(ctx.user_id) / "cache" / ctx.by_name["clip.mp4"].blob_sha256).exists() + assert not blobstore.exists(ctx.user_id, ctx.by_name["clip.mp4"].blob_sha256) + + +async def test_forwarding_gives_the_files_a_blob_of_their_own(ctx: Ctx) -> None: + from app.outbox.attachments import load_attachments + + resp = await ctx.client.post( + f"/api/messages/{ctx.message.id}/forward-attachments", headers={"X-CSRF": "t"} + ) + assert resp.status_code == 200 + ids = [a["id"] for a in resp.json()] + assert sorted(a["filename"] for a in resp.json()) == ["clip.mp4", "notatki.txt", "photo.png"] + draft = SimpleNamespace(payload={"attachment_ids": ids}) # what a send reads + files = await load_attachments(ctx.db, ctx.user_id, draft) # type: ignore[arg-type] + assert {name: data for name, _, data in files} == { + "photo.png": PNG, + "clip.mp4": VIDEO, + "notatki.txt": NOTES, + } + + +async def test_old_copies_are_removed_only_when_safe(ctx: Ctx) -> None: + photo, notes, video = (ctx.by_name[n] for n in ("photo.png", "notatki.txt", "clip.mp4")) + # As stored before: copies of their own, and no part index. + old = time.time() - 2 * 3600 + for att, data in ((photo, PNG), (notes, NOTES), (video, VIDEO)): + blobstore.store(ctx.user_id, data) + os.utime(blobstore.blob_path(ctx.user_id, att.blob_sha256), (old, old)) + await ctx.db.execute( + update(Attachment).where(Attachment.message_id == ctx.message.id).values(part_index=None) + ) + # A file that is not in its message (only its own copy has it) … + stray = blobstore.store(ctx.user_id, b"only stored apart") + os.utime(blobstore.blob_path(ctx.user_id, stray), (old, old)) + ctx.db.add( + Attachment( + message_id=ctx.message.id, + user_id=ctx.user_id, + filename="stray.bin", + mime_type="application/octet-stream", + size_bytes=17, + blob_sha256=stray, + ) + ) + # … and one a draft is about to send (a compose attachment shares the copy). + ctx.db.add( + ComposeAttachment( + user_id=ctx.user_id, + filename="notatki.txt", + mime_type="text/plain", + size_bytes=len(NOTES), + blob_sha256=notes.blob_sha256, + ) + ) + await ctx.db.commit() + + stats = await att_store.dedupe_existing(user_id=ctx.user_id) + assert stats["found"] == 4 and stats["not_found"] == 1 + assert stats["files_removed"] == 2 # the photo's and the video's copies + assert not blobstore.exists(ctx.user_id, photo.blob_sha256) + assert not blobstore.exists(ctx.user_id, video.blob_sha256) + assert blobstore.exists(ctx.user_id, notes.blob_sha256) # the draft still needs it + assert blobstore.exists(ctx.user_id, stray) # its only copy + indexes = dict( + ( + await ctx.db.execute( + select(Attachment.filename, Attachment.part_index).where( + Attachment.message_id == ctx.message.id + ) + ) + ).all() + ) + assert indexes["stray.bin"] == att_store.NOT_IN_MESSAGE + assert all(i is not None and i >= 0 for n, i in indexes.items() if n != "stray.bin") + # Everything still reads back exactly; a second pass has nothing to do. + for name, data in (("photo.png", PNG), ("clip.mp4", VIDEO), ("notatki.txt", NOTES)): + assert (await ctx.client.get(_url(ctx, name), params={"download": "true"})).content == data + assert (await att_store.dedupe_existing(user_id=ctx.user_id))["rows"] == 0 diff --git a/backend/tests/indexer/test_blob_gc.py b/backend/tests/indexer/test_blob_gc.py new file mode 100644 index 0000000..6aeaa3e --- /dev/null +++ b/backend/tests/indexer/test_blob_gc.py @@ -0,0 +1,175 @@ +"""The nightly clean-up of stored files (app/indexer/gc.py): what nothing uses +goes, everything still used — or possibly about to be — stays.""" + +from __future__ import annotations + +import datetime as dt +import os +import shutil +import time +import uuid +from collections.abc import AsyncIterator +from dataclasses import dataclass +from pathlib import Path + +import pytest_asyncio +from sqlalchemy import delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from app.config import get_settings +from app.identity.passwords import hash_password +from app.indexer import blobstore, gc +from app.models import Account, Attachment, ComposeAttachment, Draft, Mailbox, Message, User + +OLD = time.time() - 2 * 3600 + + +@dataclass +class Ctx: + db: AsyncSession + user_id: uuid.UUID + account_id: uuid.UUID + mailbox_id: uuid.UUID + + +@pytest_asyncio.fixture +async def ctx() -> AsyncIterator[Ctx]: + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + db = async_sessionmaker(engine, expire_on_commit=False)() + user = User(handle=f"gc{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + db.add(user) + await db.flush() + account = Account( + user_id=user.id, + email="me@example.com", + imap_config={}, + smtp_config={}, + secret_ct=b"", + discovery_src="manual", + ) + db.add(account) + await db.flush() + inbox = Mailbox(account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox") + db.add(inbox) + await db.commit() + try: + yield Ctx(db, user.id, account.id, inbox.id) + finally: + await db.execute(delete(User).where(User.id == user.id)) + await db.commit() + await db.close() + await engine.dispose() + shutil.rmtree(Path(get_settings().blob_root) / str(user.id), ignore_errors=True) + + +def _file(c: Ctx, data: bytes, *, old: bool = True) -> str: + sha = blobstore.store(c.user_id, data) + if old: + os.utime(blobstore.blob_path(c.user_id, sha), (OLD, OLD)) + return sha + + +async def _message(c: Ctx, sha: str, uid: int) -> Message: + m = Message( + user_id=c.user_id, + account_id=c.account_id, + mailbox_id=c.mailbox_id, + uid=uid, + internal_date=dt.datetime.now(dt.UTC), + blob_sha256=sha, + ) + c.db.add(m) + await c.db.commit() + return m + + +async def test_unused_files_go_and_used_ones_stay(ctx: Ctx) -> None: + kept_message = _file(ctx, b"a message still here") + await _message(ctx, kept_message, 1) + deleted_message = _file(ctx, b"a message deleted forever") + fresh = _file(ctx, b"a message being stored right now", old=False) + legacy = _file(ctx, b"an attachment kept as its own file") + msg = await _message(ctx, _file(ctx, b"its message"), 2) + ctx.db.add( + Attachment( + message_id=msg.id, + user_id=ctx.user_id, + filename="x.bin", + mime_type="application/octet-stream", + size_bytes=1, + blob_sha256=legacy, + part_index=None, # not looked for in its message yet + ) + ) + await ctx.db.commit() + # Previews: one of an attachment still here, one of a deleted one. + thumbs = blobstore.blob_path(ctx.user_id, "00").parent.parent / "thumbs" + thumbs.mkdir(parents=True, exist_ok=True) + live_thumb, dead_thumb = thumbs / f"{legacy}-480.webp", thumbs / f"{'f' * 64}-480.webp" + for t in (live_thumb, dead_thumb): + t.write_bytes(b"webp") + os.utime(t, (OLD, OLD)) + # A sent message waiting for its Sent copy: only its job refers to it. + waiting = _file(ctx, b"sent, not yet filed in Sent") + await gc.hold(ctx.user_id, waiting) + + stats = await gc.sweep(ctx.user_id) + + assert stats["files"] == 1 and stats["thumbs"] == 1 + assert not blobstore.exists(ctx.user_id, deleted_message) + for sha in (kept_message, fresh, legacy, waiting, msg.blob_sha256): + assert blobstore.exists(ctx.user_id, sha) + assert live_thumb.exists() and not dead_thumb.exists() + + # Filed (or given up): no longer held, so it goes on the next night. + await gc.release(ctx.user_id, waiting) + assert (await gc.sweep(ctx.user_id))["files"] == 1 + assert not blobstore.exists(ctx.user_id, waiting) + + +async def test_abandoned_uploads_go_after_a_week(ctx: Ctx) -> None: + week_ago = dt.datetime.now(dt.UTC) - dt.timedelta(days=8) + + def upload(data: bytes, created: dt.datetime) -> ComposeAttachment: + row = ComposeAttachment( + user_id=ctx.user_id, + filename="f.txt", + mime_type="text/plain", + size_bytes=len(data), + blob_sha256=_file(ctx, data), + ) + ctx.db.add(row) + return row + + abandoned = upload(b"closed without saving", week_ago) + in_draft = upload(b"in a draft still being written", week_ago) + recent = upload(b"uploaded a minute ago", dt.datetime.now(dt.UTC)) + await ctx.db.flush() + await ctx.db.execute( + update(ComposeAttachment) + .where(ComposeAttachment.id.in_([abandoned.id, in_draft.id])) + .values(created_at=week_ago) + ) + ctx.db.add( + Draft( + user_id=ctx.user_id, + account_id=ctx.account_id, + kind="new", + payload={"attachment_ids": [str(in_draft.id)]}, + ) + ) + await ctx.db.commit() + + stats = await gc.sweep(ctx.user_id) + + assert stats["uploads"] == 1 and stats["files"] == 1 + left = set( + await ctx.db.scalars( + select(ComposeAttachment.id).where(ComposeAttachment.user_id == ctx.user_id) + ) + ) + assert left == {in_draft.id, recent.id} + assert not blobstore.exists(ctx.user_id, abandoned.blob_sha256) + assert blobstore.exists(ctx.user_id, in_draft.blob_sha256) + assert blobstore.exists(ctx.user_id, recent.blob_sha256) diff --git a/backend/workers/run.py b/backend/workers/run.py index 49e1615..86fa8dd 100644 --- a/backend/workers/run.py +++ b/backend/workers/run.py @@ -25,14 +25,19 @@ from arq import cron from arq.worker import Worker from app.config import get_settings +from app.indexer.gc import sweep_job from app.queue import QUEUE_BULK, QUEUE_INTERACTIVE, redis_settings from app.sync.actions import autopurge_trash_spam from app.sync.coordinator import SyncCoordinator from workers.worker import BULK_FUNCTIONS, INTERACTIVE_FUNCTIONS # Core scheduled jobs. Daily Trash/Spam retention purge at 03:17 (off-peak, and a -# minute that won't collide with typical top-of-hour crons). -_CORE_CRON_JOBS = [cron(autopurge_trash_spam, hour=3, minute=17)] +# minute that won't collide with typical top-of-hour crons); then the clean-up of +# stored files nothing uses any more (the purge's among them) at 03:47. +_CORE_CRON_JOBS = [ + cron(autopurge_trash_spam, hour=3, minute=17), + cron(sweep_job, hour=3, minute=47, timeout=3600), +] logger = logging.getLogger(__name__) @@ -95,6 +100,15 @@ async def _run() -> None: await coordinator.start() arq_tasks = [asyncio.create_task(w.async_run()) for w in build_arq_workers(role)] logger.info("worker starting: role=%s (%d arq worker(s))", role, len(arq_tasks)) + if arq_tasks: + # Attachments stored apart before they were kept once: remove those + # copies in the background (quick when there are none; one at a time). + from app.queue import enqueue_job + + try: + await enqueue_job("dedupe_existing_job", _job_id="dedupe-attachments") + except Exception: # noqa: BLE001 - tried again on the next start + logger.warning("could not enqueue the attachment clean-up", exc_info=True) stop = asyncio.Event() loop = asyncio.get_running_loop() diff --git a/backend/workers/worker.py b/backend/workers/worker.py index ed3755b..b5feab3 100644 --- a/backend/workers/worker.py +++ b/backend/workers/worker.py @@ -14,6 +14,7 @@ from typing import Any from arq import func +from app.indexer.attachments import dedupe_existing_job from app.outbox.draft_sync import push_draft_to_imap_job, remove_imap_draft_job from app.outbox.jobs import file_sent_copy_job, send_job from app.push.service import push_new_mail_job @@ -62,4 +63,8 @@ INTERACTIVE_FUNCTIONS: list[Any] = [ ] # Bulk work (§6): long timeout, low concurrency, isolated from interactive jobs. -BULK_FUNCTIONS = [sync_bootstrap_account] +BULK_FUNCTIONS = [ + sync_bootstrap_account, + # Resumable (each batch is saved), so a long first run may span timeouts. + func(dedupe_existing_job, timeout=6 * 3600), +] diff --git a/webmail-client-spec_v1.5.md b/webmail-client-spec_v1.5.md index ec2ebd0..9603186 100644 --- a/webmail-client-spec_v1.5.md +++ b/webmail-client-spec_v1.5.md @@ -877,7 +877,7 @@ If phones/desktop clients may use the mailbox over IMAP/SMTP, a password alone w ### D.6 Storing mail once Today the indexer keeps every message in full (raw `.eml` blob + sanitized body in `body_parts` + search index), so external accounts are already a second copy, and with Stalwart both copies would sit on the same disk. Ideas, best first: -1. **Attachments are stored twice inside webmail today.** `indexer.py` stores each attachment as its own blob *and* it is inside the raw blob (base64), ≈2.3× its size; attachments are most of mail's bytes. Fix: keep only the raw message, extract an attachment when downloaded, keep thumbnails cached. Roughly halves storage for **all** accounts, independent of Stalwart. A cleanup script frees existing duplicates. +1. **Attachments are stored twice inside webmail today.** `indexer.py` stores each attachment as its own blob *and* it is inside the raw blob (base64), ≈2.3× its size; attachments are most of mail's bytes. Fix: keep only the raw message, extract an attachment when downloaded, keep thumbnails cached. Cuts attachment storage by about 40% (≈1.37× instead of ≈2.37× the file size) for **all** accounts, independent of Stalwart. A background pass frees existing duplicates. **Done (v1.3.34):** `app/indexer/attachments.py`, migration 0023. 2. **Native accounts: no raw copy in webmail.** Stalwart is the master copy; webmail keeps metadata, sanitized body, snippet and search index (~5–10% of the size). Full message, attachments and "show original" are fetched on demand over the local network (IMAP `BODY[part]` or JMAP blob download), with a small cache for repeat opens and thumbnails. 3. **External accounts (optional setting): age out old raw copies.** Keep raw blobs for N days or the newest X GB; refetch older ones over IMAP when needed. Readable bodies stay local; old attachments then need the external server reachable. 4. **Filesystem:** at least lz4 compression on the ZFS data share (QuTS hero). ZFS dedup would merge identical copies but costs a lot of RAM and only hits uncompressed byte-identical data — last resort. @@ -902,7 +902,7 @@ Today the indexer keeps every message in full (raw `.eml` blob + sanitized body 2. The same handle + password signs in to IMAP (when §D.5 allows it); after a webmail password change or reset, the old password fails and the new one works with no other step; a blocked user cannot sign in to IMAP. 3. The native account cannot be removed via API (409) or UI; deleting the user removes the Stalwart mailbox. 4. Webmail stores no copy of the account's password (`secret_ct` unused for native accounts). -5. After D1, an attachment-heavy fixture mailbox uses ≤ 55% of the pre-D1 blob storage, and every attachment still downloads byte-identical. +5. After D1, an attachment-heavy fixture mailbox uses ≤ 60% of the pre-D1 blob storage, and every attachment still downloads byte-identical. 6. After D4, a native message's raw blob is absent from `BLOB_ROOT`, yet opening, downloading attachments and "show original" all work; with Stalwart stopped, list and bodies still render from local data. 7. Backup → restore on a fresh host brings back both webmail and Stalwart mail intact.