From 12e0c9985040b55f4d393d3ab438a721169bc5d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mateusz=20Gruszczy=C5=84ski?= Date: Mon, 2 Nov 2026 14:23:13 +0100 Subject: [PATCH] v0.2.13 --- CHANGELOG.md | 9 + Cargo.lock | 2 +- Cargo.toml | 2 +- NOTICE.md | 2 +- README.md | 10 +- RELEASE-MANIFEST.txt | 15 +- RELEASE-VALIDATION.txt | 16 +- config.example.toml | 2 +- docs/IMPLEMENTATION_STATUS.md | 11 +- docs/RELEASING.md | 2 +- docs/STORAGE_SEARCH_OPERATIONS.md | 10 +- migrations/README.md | 15 +- migrations/mysql/V049__blob_delete_outbox.sql | 11 ++ migrations/mysql/schema.sql | 15 +- migrations/pgsql/V049__blob_delete_outbox.sql | 13 ++ migrations/pgsql/schema.sql | 16 +- .../sqlite/V049__blob_delete_outbox.sql | 13 ++ migrations/sqlite/schema.sql | 16 +- packaging/rustbox.toml | 2 +- scripts/validate-release-static.py | 83 +++++++--- src/db/blob_delete_outbox.rs | 155 ++++++++++++++++++ src/db/drafts.rs | 61 +++++++ src/db/mod.rs | 1 + src/db/pool.rs | 14 ++ src/main.rs | 2 + src/queries/mysql/queries.rs | 11 ++ src/queries/pgsql/queries.rs | 11 ++ src/queries/sqlite/queries.rs | 11 ++ src/routes/attachments.rs | 81 ++++++--- src/routes/drafts.rs | 43 +++-- src/routes/send.rs | 59 +++++-- src/storage/mod.rs | 124 +++++++++++++- 32 files changed, 729 insertions(+), 109 deletions(-) create mode 100644 migrations/mysql/V049__blob_delete_outbox.sql create mode 100644 migrations/pgsql/V049__blob_delete_outbox.sql create mode 100644 migrations/sqlite/V049__blob_delete_outbox.sql create mode 100644 src/db/blob_delete_outbox.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index c95fad0..e2e87da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # RustBox changelog +### 0.2.13 + +- Added durable per-tenant blob deletion outbox for every configured blob backend (`filesystem` including shared NFS/POSIX, and S3-compatible storage). Attachment metadata deletion and cleanup enqueue now commit atomically, so a temporary storage failure cannot silently orphan blobs. +- Blob cleanup runs immediately when possible and is also retried by a background worker with claim leases and exponential backoff; missing objects are treated as successful idempotent deletes and retries do not dead-letter permanently. +- Deleting one attachment, deleting a whole draft, and post-send draft cleanup all use the same durable deletion path. Failed/finalized uploads and failed storage-migration destination objects are queued for cleanup as well. +- Storage health probes reuse a fixed probe key so a failed cleanup cannot accumulate unbounded health-check objects. +- Added runtime schema migration V049 for SQLite, MySQL and PostgreSQL (`blob_delete_outbox`). V050 is the next migration number. +- Bumped the application/source release to 0.2.13. + ### 0.2.12 - Saving or updating a draft no longer raises a realtime “new message” toast/system notification for the Drafts folder; the folder/list still refreshes normally, and IMAP snapshots remain `\Draft` + `\Seen`. diff --git a/Cargo.lock b/Cargo.lock index 8279c46..829e3a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3506,7 +3506,7 @@ dependencies = [ [[package]] name = "rustbox" -version = "0.2.12" +version = "0.2.13" dependencies = [ "aes-gcm", "async-imap", diff --git a/Cargo.toml b/Cargo.toml index b632e06..cb17099 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rustbox" -version = "0.2.12" +version = "0.2.13" edition = "2024" rust-version = "1.99.1" default-run = "rustbox" diff --git a/NOTICE.md b/NOTICE.md index b639b69..76dcf51 100644 --- a/NOTICE.md +++ b/NOTICE.md @@ -1,6 +1,6 @@ # RustBox attribution / notice -RustBox 0.2.12 is derived from the Oxi webmail project: +RustBox 0.2.13 is derived from the Oxi webmail project: - Upstream: `c0h1b4/oxi` - Base snapshot: `7f1507c486b78cf33e54beb701dd460d3bd2bd5d` diff --git a/README.md b/README.md index ea78155..218a00e 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# RustBox 0.2.12 +# RustBox 0.2.13 RustBox is a webmail derived from `c0h1b4/oxi`. The consumed Oxi revision is recorded in `UPSTREAM_BASE`. @@ -34,7 +34,7 @@ rustbox/ `build.rs` combines the CSS modules from `src/web/css/manifest.txt` and JavaScript modules from `src/web/js/manifest.txt`. Cargo embeds the generated CSS/JS bundles, HTML and SVG sprite in the RustBox executable. There is no standalone source `src/web/app.js`; `src/web/` is the single source of truth and Cargo generates the embedded bundles in `OUT_DIR`. -## Storage and search in 0.2.12 +## Storage and search in 0.2.13 RustBox separates three concerns: SQL metadata, blob storage and search. The SQL engine does **not** decide whether `data_dir` may be used. A single-node PostgreSQL/MySQL deployment may still use filesystem blobs and local Tantivy. In cluster mode, local node state is rejected unless the configured backend is explicitly shared. @@ -43,6 +43,8 @@ Blob backends: - `storage.backend="filesystem"` — local disk in single-node mode; set `storage.shared=true` only for a real shared POSIX mount such as NFS, CephFS or GlusterFS. - `storage.backend="s3"` — AWS S3 or compatible APIs such as MinIO/Ceph RGW. +Physical blob deletion is durable across all blob backends. RustBox records deletion work in the per-account SQL `blob_delete_outbox` in the same transaction that removes attachment metadata, attempts the delete immediately, and retries temporary failures in the background with leases/backoff. This covers local disk, shared POSIX/NFS and S3-compatible storage without permanently dead-lettering failed cleanup. + Search backends: - `search.backend="local_tantivy"` — single-node. @@ -166,7 +168,7 @@ rustbox search reindex Ready Docker examples are in `docker/`: standalone SQLite, PostgreSQL + SILO (S3-compatible) + OpenSearch, and MySQL + NFS + OpenSearch. -All executable application SQL is centralized by dialect under `src/queries/{sqlite,mysql,pgsql}/queries.rs`. The squashed compatibility baseline remains V046; RustBox 0.2.1 adds runtime migration V047 for attachment checksums/logical blob keys and the durable search outbox, while RustBox 0.2.8 adds V048 for monotonic folder-counter generations used by race-safe IMAP counter reconciliation. The next migration number is V049. See `migrations/README.md`. +All executable application SQL is centralized by dialect under `src/queries/{sqlite,mysql,pgsql}/queries.rs`. The squashed compatibility baseline remains V046; RustBox 0.2.1 adds runtime migration V047 for attachment checksums/logical blob keys and the durable search outbox, RustBox 0.2.8 adds V048 for monotonic folder-counter generations used by race-safe IMAP counter reconciliation, and RustBox 0.2.13 adds V049 for durable blob deletion retries. The next migration number is V050. See `migrations/README.md`. ## Development @@ -235,7 +237,7 @@ Initialize Git after extracting the release archive: ```bash git init git add . -git commit -m "RustBox 0.2.12 release" +git commit -m "RustBox 0.2.13 release" ./scripts/setup-upstream.sh ``` diff --git a/RELEASE-MANIFEST.txt b/RELEASE-MANIFEST.txt index 5ee94a0..361814e 100644 --- a/RELEASE-MANIFEST.txt +++ b/RELEASE-MANIFEST.txt @@ -1,6 +1,8 @@ -RustBox 0.2.12 release manifest +RustBox 0.2.13 release manifest -0.2.12 feature/configuration changes: +0.2.13 feature/configuration changes: +- adds a durable SQL blob-deletion outbox used by filesystem/shared POSIX/NFS and S3 backends; attachment/draft cleanup is transactionally queued and retried until successful +- failed finalized uploads and failed destination objects from storage migration are also queued for cleanup; storage health probes reuse a fixed key to avoid unbounded probe-object buildup - suppresses realtime “new message” toast/system notifications for Drafts saves while preserving normal folder/list refresh and `\Draft` + `\Seen` IMAP flags - adds `Delete draft` in the composer; IMAP deletion is UID/UIDVALIDITY-checked and local draft attachment rows/blobs are cascaded after the server draft is removed - adds an explicit `Save draft` action before `Cancel`; it persists the current message and attachments as a real IMAP Drafts snapshot without changing background autosave behavior @@ -24,7 +26,7 @@ RustBox 0.2.12 release manifest - previews PDFs from the authenticated inline attachment endpoint instead of a blob iframe for Safari/iPhone compatibility - presents compose attachment/message size limits as compact raw values without implementation-detail descriptions - retains the 0.2.9 realtime notification, shortcut and 300-message mailbox behavior -- requires no runtime schema migration; V049 remains next +- adds runtime schema migration V049 for durable `blob_delete_outbox`; V050 remains next Architecture: - database_backend = sqlite | mysql | pgsql selects authoritative SQL only - blob storage and search are independent from the SQL engine @@ -51,7 +53,8 @@ Database/schema: - compatibility baseline remains V046 - runtime migration V047 adds attachment checksum metadata and search_outbox - runtime migration V048 adds `folders.counter_generation` for race-safe authoritative IMAP counter writes -- next migration number: V049 +- runtime migration V049 adds `blob_delete_outbox` for durable physical blob cleanup +- next migration number: V050 Docker examples: - docker/compose-single-sqlite.yml @@ -61,14 +64,14 @@ Docker examples: Operations/docs: - standalone `scripts/migrate_roundcube.py` imports Roundcube contacts/identities using RustBox config.toml and supports SQLite/MySQL/PostgreSQL targets -- README.md documents 0.2.12 configuration +- README.md documents 0.2.13 configuration - docker/README.md documents Compose variants - docs/STORAGE_SEARCH_OPERATIONS.md documents failure modes, shared-storage requirements and migration behavior - TODO.md leaves live SILO/MinIO/AWS/Ceph and real multi-node acceptance validation explicitly pending Build/release metadata: - `scripts/package-release.sh` runs the full Rust release gate before creating ZIP/SHA256 artifacts -- Cargo.toml version = 0.2.12 +- Cargo.toml version = 0.2.13 - rust-toolchain.toml pins Rust 1.99.1 - Cargo.lock is retained for reproducible application builds; the release gate generates it only when missing - CI and scripts/build-release.sh use the retained Cargo.lock (generating it only if missing), then run locked tests/Clippy/release build diff --git a/RELEASE-VALIDATION.txt b/RELEASE-VALIDATION.txt index 07515d7..924a5e7 100644 --- a/RELEASE-VALIDATION.txt +++ b/RELEASE-VALIDATION.txt @@ -1,6 +1,8 @@ -RustBox 0.2.12 - final source validation +RustBox 0.2.13 - final source validation -0.2.12 scope: +0.2.13 scope: +- durable blob deletion outbox covers filesystem/shared POSIX/NFS and S3, pairing metadata deletion with cleanup enqueue in one SQL transaction and retrying storage failures without permanent dead-lettering +- failed finalized uploads and failed storage-migration destination objects are queued for cleanup - saving/updating a draft stays silent in realtime: Drafts still refreshes, but no “new message” toast/system notification is emitted - composer `Delete draft` removes the current verified IMAP draft and cascades local draft attachment metadata/blob cleanup - explicit Save draft action before Cancel forces a real IMAP Drafts snapshot while background autosave behavior remains unchanged @@ -20,17 +22,17 @@ RustBox 0.2.12 - final source validation - PDF attachment preview uses the authenticated inline URL for Safari/iPhone compatibility - Composing shows only compact attachment/message size-limit values on one line - 0.2.9 realtime notification gating, browser shortcut preservation and 300-message defaults remain intact -- no runtime schema migration is added; V049 remains next -- version metadata updated to 0.2.12 +- runtime schema migration V049 adds `blob_delete_outbox`; V050 remains next +- version metadata updated to 0.2.13 Completed in this environment: -- offline release validator: scripts/validate-release-static.py (374/374 checks passed) +- offline release validator: scripts/validate-release-static.py (396/396 checks passed) - browser JavaScript assembled in manifest order passes `node --check` - all JSON files parse successfully - Python helper scripts compile successfully - shell scripts pass `bash -n` -- release metadata consistently identifies 0.2.12 -- V048 migration files/catalog registration and SQLite upgrade execution are validated offline +- release metadata consistently identifies 0.2.13 +- V048/V049 migration files/catalog registration and SQLite upgrade execution are validated offline Not executable in this packaging environment: diff --git a/config.example.toml b/config.example.toml index 82c404a..84dfff6 100644 --- a/config.example.toml +++ b/config.example.toml @@ -1,4 +1,4 @@ -# RustBox 0.2.12 +# RustBox 0.2.13 host = "127.0.0.1" port = 3001 environment = "production" diff --git a/docs/IMPLEMENTATION_STATUS.md b/docs/IMPLEMENTATION_STATUS.md index b0b9ac9..5a14127 100644 --- a/docs/IMPLEMENTATION_STATUS.md +++ b/docs/IMPLEMENTATION_STATUS.md @@ -1,9 +1,9 @@ -# RustBox 0.2.12 - implementation status +# RustBox 0.2.13 - implementation status - Draft saves are realtime-silent: Drafts folder/list updates continue, but saving a draft does not surface as a new-mail notification. -This file tracks the `0.2.12` source snapshot. Storage/search scope and outstanding acceptance work are tracked in `TODO.md`. A feature is only listed as implemented when working code exists in this source tree; partial items are called out explicitly. +This file tracks the `0.2.13` source snapshot. Storage/search scope and outstanding acceptance work are tracked in `TODO.md`. A feature is only listed as implemented when working code exists in this source tree; partial items are called out explicitly. -## 0.2.12 browser mail UX +## 0.2.13 browser mail UX - Large To/Cc lists collapse to a compact recipient summary and expand on demand. - Message pagination exposes both `Load more` and `Load all`; both actions stretch evenly across the full list width, and the reader menu label is shortened to `Details`. @@ -23,7 +23,8 @@ This file tracks the `0.2.12` source snapshot. Storage/search scope and outstand - Composing displays attachment/message size limits as one compact line without configuration/MIME implementation details. - Initial synchronization remains silent for new-mail notifications; bulk operations retain one summary toast; browser/system modifier shortcuts remain preserved. - Mailbox pages default to 300 messages and the API/preferences accept values up to 300. -- No runtime schema migration is added; V049 remains next. +- Blob deletion uses a durable SQL outbox for filesystem/shared POSIX/NFS and S3 backends; attachment/draft metadata deletion is transactionally paired with cleanup enqueue and failed deletes retry in the background. +- Runtime migration V049 adds `blob_delete_outbox` on SQLite, MySQL and PostgreSQL; V050 is next. ## 0.2.8 browser security headers @@ -43,7 +44,7 @@ This file tracks the `0.2.12` source snapshot. Storage/search scope and outstand - Draft attachment upload/read/delete/snapshot/send paths use the `BlobStorage` abstraction and logical object keys instead of assuming node-local paths. - OpenSearch is available as the cluster-capable indexer. Local Tantivy remains single-node only; search may also be disabled. - Search mutations use a durable SQL outbox with claim/lease, retry/backoff and dead-letter state; the index remains rebuildable derived state. -- V047 adds attachment checksum metadata and the search outbox across SQLite, MySQL and PostgreSQL. V048 adds monotonic folder-counter generations used by compare-and-set reconciliation of authoritative IMAP counts; V049 is reserved as the next migration number. +- V047 adds attachment checksum metadata and the search outbox across SQLite, MySQL and PostgreSQL. V048 adds monotonic folder-counter generations used by compare-and-set reconciliation of authoritative IMAP counts. V049 adds the durable blob deletion outbox; V050 is reserved as the next migration number. - SQL global-schema initialization is centralized in `db::pool::configure()`; database-backed sessions do not repeat startup DDL for SQLite, MySQL or PostgreSQL. - Docker Compose examples cover standalone SQLite, PostgreSQL + SILO (S3-compatible) + OpenSearch, and MySQL + NFS + OpenSearch. - Live SILO/MinIO/AWS S3/Ceph RGW, live NFS multi-host, live OpenSearch and full multi-node failover acceptance remain release-environment validation items and are not represented as completed here. diff --git a/docs/RELEASING.md b/docs/RELEASING.md index 1253254..3338f1c 100644 --- a/docs/RELEASING.md +++ b/docs/RELEASING.md @@ -26,7 +26,7 @@ RUSTBOX_CONFIG=./config.dev.toml ./target/release/rustbox --check-config Browser JavaScript source lives in functional modules under `src/web/js/`, ordered by `src/web/js/manifest.txt`. `build.rs` concatenates those modules into the embedded `rustbox-ui.js` asset during every Cargo build. The repository has no duplicate `ui/` snapshot; `src/web/` is the only browser UI source. -## Updating an existing checkout to 0.2.12 +## Updating an existing checkout to 0.2.13 Do not overlay the release with `cp *` or another command that skips dotfiles. That can retain stale `.github`, `.env.example`, or `Cargo.lock` files. diff --git a/docs/STORAGE_SEARCH_OPERATIONS.md b/docs/STORAGE_SEARCH_OPERATIONS.md index 61f5a92..d9cc5f1 100644 --- a/docs/STORAGE_SEARCH_OPERATIONS.md +++ b/docs/STORAGE_SEARCH_OPERATIONS.md @@ -1,6 +1,6 @@ -# RustBox 0.2.12 storage/search operations +# RustBox 0.2.13 storage/search operations -RustBox 0.2.12 treats SQL, blob storage and search as independent capabilities. The database engine does not decide whether `data_dir` is allowed. +RustBox 0.2.13 treats SQL, blob storage and search as independent capabilities. The database engine does not decide whether `data_dir` is allowed. ## Deployment matrix @@ -28,6 +28,12 @@ For NFS/CephFS/GlusterFS: Attachment SQL rows store logical object keys. This keeps SQL independent from a node-specific filesystem path. +## Durable blob deletion + +Attachment deletion is not best-effort. RustBox writes a `blob_delete_outbox` entry in the same SQL transaction that removes the attachment row or draft row, then attempts the physical delete immediately. A background worker retries failures with claim leases and exponential backoff until the object is gone; object-store `NotFound` is treated as success. The same path is used for local filesystem, shared POSIX/NFS and S3-compatible backends. Failed finalized uploads and failed migration destination objects are also queued for cleanup. + +This avoids orphan accumulation when storage is temporarily unavailable while also avoiding the inverse failure mode where a blob is deleted first and the SQL transaction later fails. + ## Search and outbox OpenSearch is the cluster-capable indexer. Every indexed document is scoped by `user_hash`; queries add a server-side tenant filter and deterministic document IDs are routed by tenant. diff --git a/migrations/README.md b/migrations/README.md index bab6ac4..bfda70a 100644 --- a/migrations/README.md +++ b/migrations/README.md @@ -1,6 +1,6 @@ # Database schemas and migrations -RustBox 0.2.12 keeps SQL, blob storage and search as independent concerns: +RustBox 0.2.13 keeps SQL, blob storage and search as independent concerns: ```text database_backend = sqlite | mysql | pgsql @@ -18,18 +18,21 @@ migrations/ ├── sqlite/ │ ├── schema.sql │ ├── V047__blob_keys_and_search_outbox.sql -│ └── V048__folder_counter_generation.sql +│ ├── V048__folder_counter_generation.sql +│ └── V049__blob_delete_outbox.sql ├── mysql/ │ ├── schema.sql │ ├── V047__blob_keys_and_search_outbox.sql -│ └── V048__folder_counter_generation.sql +│ ├── V048__folder_counter_generation.sql +│ └── V049__blob_delete_outbox.sql └── pgsql/ ├── schema.sql ├── V047__blob_keys_and_search_outbox.sql - └── V048__folder_counter_generation.sql + ├── V048__folder_counter_generation.sql + └── V049__blob_delete_outbox.sql ``` -Each `schema.sql` is the current complete schema for fresh installations. The historical compatibility baseline remains **V046**. RustBox 0.2.1 adds **V047**, which adds attachment SHA-256 metadata and the durable search outbox used by external indexing. RustBox 0.2.8 adds **V048**, which adds the monotonic `folders.counter_generation` used to reject stale IMAP counter writes even when the numeric counts temporarily return to an earlier value. +Each `schema.sql` is the current complete schema for fresh installations. The historical compatibility baseline remains **V046**. RustBox 0.2.1 adds **V047**, which adds attachment SHA-256 metadata and the durable search outbox used by external indexing. RustBox 0.2.8 adds **V048**, which adds the monotonic `folders.counter_generation` used to reject stale IMAP counter writes even when the numeric counts temporarily return to an earlier value. RustBox 0.2.13 adds **V049**, the durable `blob_delete_outbox` used to retry physical object deletion after metadata is removed. Account data is isolated inside the selected physical database with a deterministic, collision-checked table prefix derived from the account hash. The `{{p}}` marker in tenant SQL is replaced by RustBox at provisioning/migration time. Sessions are global and therefore are not prefixed. @@ -37,7 +40,7 @@ Account data is isolated inside the selected physical database with a determinis Every schema-changing release must add the same logical migration for all supported SQL engines and register it in `src/db/pool.rs`. Do not update only `schema.sql`: fresh installs use the baseline while existing installations use the numbered migration path. -RustBox 0.2.12 does not add a runtime schema migration; **V049** remains the next migration number. +RustBox 0.2.13 adds runtime migration **V049** on SQLite, MySQL and PostgreSQL. **V050** is the next migration number. Redis and Redis Sentinel do not use SQL migrations. diff --git a/migrations/mysql/V049__blob_delete_outbox.sql b/migrations/mysql/V049__blob_delete_outbox.sql new file mode 100644 index 0000000..2be60f1 --- /dev/null +++ b/migrations/mysql/V049__blob_delete_outbox.sql @@ -0,0 +1,11 @@ +CREATE TABLE {{p}}blob_delete_outbox ( + id VARCHAR(64) PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts BIGINT NOT NULL DEFAULT 0, + available_at BIGINT NOT NULL DEFAULT 0, + claimed_until BIGINT NOT NULL DEFAULT 0, + last_error TEXT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + INDEX idx_blob_delete_outbox_ready (available_at, claimed_until, updated_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/migrations/mysql/schema.sql b/migrations/mysql/schema.sql index 87b7c11..f4c4e67 100644 --- a/migrations/mysql/schema.sql +++ b/migrations/mysql/schema.sql @@ -1,7 +1,7 @@ -- RustBox 0.2.0 MySQL schema. -- One physical database contains global sessions plus isolated per-account tables. -- Application tables are materialized by replacing {{p}} with a collision-checked prefix. --- Squashed compatibility baseline: V046. Current runtime schema: V048. +-- Squashed compatibility baseline: V046. Current runtime schema: V049. -- RUSTBOX:GLOBAL-BEGIN CREATE TABLE IF NOT EXISTS rustbox_accounts ( @@ -190,6 +190,19 @@ CREATE TABLE {{p}}draft_attachments ( created_at VARCHAR(64) NOT NULL DEFAULT (CAST(CURRENT_TIMESTAMP AS CHAR)) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE {{p}}blob_delete_outbox ( + id VARCHAR(64) PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts BIGINT NOT NULL DEFAULT 0, + available_at BIGINT NOT NULL DEFAULT 0, + claimed_until BIGINT NOT NULL DEFAULT 0, + last_error TEXT NULL, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL, + INDEX idx_blob_delete_outbox_ready (available_at, claimed_until, updated_at) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE {{p}}search_outbox ( id VARCHAR(64) PRIMARY KEY, generation VARCHAR(64) NOT NULL, diff --git a/migrations/pgsql/V049__blob_delete_outbox.sql b/migrations/pgsql/V049__blob_delete_outbox.sql new file mode 100644 index 0000000..194ca8a --- /dev/null +++ b/migrations/pgsql/V049__blob_delete_outbox.sql @@ -0,0 +1,13 @@ +CREATE TABLE {{p}}blob_delete_outbox ( + id TEXT PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts BIGINT NOT NULL DEFAULT 0, + available_at BIGINT NOT NULL DEFAULT 0, + claimed_until BIGINT NOT NULL DEFAULT 0, + last_error TEXT, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL +); + +CREATE INDEX {{p}}idx_blob_delete_outbox_ready + ON {{p}}blob_delete_outbox(available_at, claimed_until, updated_at); diff --git a/migrations/pgsql/schema.sql b/migrations/pgsql/schema.sql index 3f29bc9..dba981f 100644 --- a/migrations/pgsql/schema.sql +++ b/migrations/pgsql/schema.sql @@ -1,7 +1,7 @@ -- RustBox 0.2.0 PostgreSQL schema. -- One physical database contains global sessions plus isolated per-account tables. -- Application tables are materialized by replacing {{p}} with a collision-checked prefix. --- Squashed compatibility baseline: V046. Current runtime schema: V048. +-- Squashed compatibility baseline: V046. Current runtime schema: V049. -- RUSTBOX:GLOBAL-BEGIN CREATE TABLE IF NOT EXISTS rustbox_accounts ( @@ -190,6 +190,20 @@ CREATE TABLE {{p}}draft_attachments ( created_at TEXT NOT NULL DEFAULT (CURRENT_TIMESTAMP::text) ); + +CREATE TABLE {{p}}blob_delete_outbox ( + id TEXT PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts BIGINT NOT NULL DEFAULT 0, + available_at BIGINT NOT NULL DEFAULT 0, + claimed_until BIGINT NOT NULL DEFAULT 0, + last_error TEXT, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL +); + +CREATE INDEX {{p}}idx_blob_delete_outbox_ready ON {{p}}blob_delete_outbox(available_at, claimed_until, updated_at); + CREATE TABLE {{p}}search_outbox ( id TEXT PRIMARY KEY, generation TEXT NOT NULL, diff --git a/migrations/sqlite/V049__blob_delete_outbox.sql b/migrations/sqlite/V049__blob_delete_outbox.sql new file mode 100644 index 0000000..816e264 --- /dev/null +++ b/migrations/sqlite/V049__blob_delete_outbox.sql @@ -0,0 +1,13 @@ +CREATE TABLE {{p}}blob_delete_outbox ( + id TEXT PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + available_at INTEGER NOT NULL DEFAULT 0, + claimed_until INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE INDEX {{p}}idx_blob_delete_outbox_ready + ON {{p}}blob_delete_outbox(available_at, claimed_until, updated_at); diff --git a/migrations/sqlite/schema.sql b/migrations/sqlite/schema.sql index 15be6ae..c37b994 100644 --- a/migrations/sqlite/schema.sql +++ b/migrations/sqlite/schema.sql @@ -1,7 +1,7 @@ -- RustBox 0.2.0 SQLite schema. -- One physical database contains global sessions plus isolated per-account tables. -- Application tables are materialized from the tenant section by replacing {{p}}. --- Squashed compatibility baseline: V046. Current runtime schema: V048. +-- Squashed compatibility baseline: V046. Current runtime schema: V049. -- RUSTBOX:GLOBAL-BEGIN CREATE TABLE IF NOT EXISTS rustbox_accounts ( @@ -136,6 +136,20 @@ CREATE TABLE {{p}}draft_attachments ( FOREIGN KEY (draft_id) REFERENCES {{p}}drafts(id) ON DELETE CASCADE ); + +CREATE TABLE {{p}}blob_delete_outbox ( + id TEXT PRIMARY KEY, + storage_key TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + available_at INTEGER NOT NULL DEFAULT 0, + claimed_until INTEGER NOT NULL DEFAULT 0, + last_error TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE INDEX {{p}}idx_blob_delete_outbox_ready ON {{p}}blob_delete_outbox(available_at, claimed_until, updated_at); + CREATE TABLE {{p}}search_outbox ( id TEXT PRIMARY KEY, generation TEXT NOT NULL, diff --git a/packaging/rustbox.toml b/packaging/rustbox.toml index a3cf1e5..b3c40bc 100644 --- a/packaging/rustbox.toml +++ b/packaging/rustbox.toml @@ -1,4 +1,4 @@ -# RustBox 0.2.12 - packaged production configuration (DEB/RPM) +# RustBox 0.2.13 - packaged production configuration (DEB/RPM) # # This file contains all supported RustBox configuration options. # Commented options are optional and do not change the packaged defaults. diff --git a/scripts/validate-release-static.py b/scripts/validate-release-static.py index 87a5e89..e8cb95f 100755 --- a/scripts/validate-release-static.py +++ b/scripts/validate-release-static.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Static/offline release checks for RustBox 0.2.12. +"""Static/offline release checks for RustBox 0.2.13. These checks intentionally avoid pretending to replace `cargo test`: they validate configuration, SQL migrations, Docker examples, query-catalog parity and release @@ -49,14 +49,14 @@ def extract_section(text: str, begin: str, end: str) -> str: # Release/toolchain metadata. cargo = tomllib.loads(read("Cargo.toml")) -require(cargo["package"]["version"] == "0.2.12", "Cargo package version is 0.2.12") +require(cargo["package"]["version"] == "0.2.13", "Cargo package version is 0.2.13") require(cargo["package"]["rust-version"] == "1.99.1", "Cargo rust-version is 1.99.1") toolchain = tomllib.loads(read("rust-toolchain.toml")) require(toolchain["toolchain"]["channel"] == "1.99.1", "rust-toolchain pins Rust 1.99.1") require((ROOT / "Cargo.lock").is_file(), "Cargo.lock is retained for reproducible application builds") lock = tomllib.loads(read("Cargo.lock")) lock_rustbox = next((pkg for pkg in lock.get("package", []) if pkg.get("name") == "rustbox"), {}) -require(lock_rustbox.get("version") == "0.2.12", "Cargo.lock RustBox package version is 0.2.12") +require(lock_rustbox.get("version") == "0.2.13", "Cargo.lock RustBox package version is 0.2.13") require(cargo["dependencies"].get("encoding_rs") == "0.8", "direct charset decoder dependency is pinned") require("encoding_rs" in lock_rustbox.get("dependencies", []), "Cargo.lock records direct charset decoder dependency") require("cargo generate-lockfile" in read("scripts/build-release.sh"), "release script generates Cargo.lock when missing") @@ -71,10 +71,10 @@ require("make_zip.py" in package_script and "sha256sum" in package_script, "rele require("cargo generate-lockfile" in read(".github/workflows/ci.yml"), "CI regenerates Cargo.lock before --locked checks") require("validate-release-static.py" in read("scripts/build-release.sh"), "release script runs offline/static release gate") require("validate-release-static.py" in read(".github/workflows/ci.yml"), "CI runs offline/static release gate") -require(read("README.md").startswith("# RustBox 0.2.12"), "README identifies RustBox 0.2.12") -require(read("RELEASE-MANIFEST.txt").startswith("RustBox 0.2.12"), "release manifest identifies RustBox 0.2.12") -require(read("RELEASE-VALIDATION.txt").startswith("RustBox 0.2.12"), "release validation identifies RustBox 0.2.12") -require(read("docs/IMPLEMENTATION_STATUS.md").startswith("# RustBox 0.2.12"), "implementation status identifies RustBox 0.2.12") +require(read("README.md").startswith("# RustBox 0.2.13"), "README identifies RustBox 0.2.13") +require(read("RELEASE-MANIFEST.txt").startswith("RustBox 0.2.13"), "release manifest identifies RustBox 0.2.13") +require(read("RELEASE-VALIDATION.txt").startswith("RustBox 0.2.13"), "release validation identifies RustBox 0.2.13") +require(read("docs/IMPLEMENTATION_STATUS.md").startswith("# RustBox 0.2.13"), "implementation status identifies RustBox 0.2.13") bootstrap_js = read("src/web/js/00-bootstrap.js") bindings_js = read("src/web/js/95-bindings-boot.js") realtime_js = read("src/web/js/80-settings-realtime.js") @@ -124,8 +124,8 @@ require('encoding_rs::Encoding::for_label' in read("src/imap/parse.rs") and 'dec require('let is_pdf = response_content_type_is(&response, "application/pdf");' in read("src/routes/mod.rs") and 'HeaderValue::from_static("SAMEORIGIN")' in read("src/routes/mod.rs") and "HeaderValue::from_static(\"frame-ancestors 'self'\")" in read("src/routes/mod.rs"), "security headers allow same-origin framing for inline PDF previews only") index_html = read("src/web/index.html") require('class="compose-limit-summary"' in index_html and 'Per-file upload limit set with' not in index_html and 'Hard limit for the complete RFC822/MIME message' not in index_html, "Composing shows compact raw size limits without implementation details") -require("V048" in read("RELEASE-MANIFEST.txt") and "V049" in read("RELEASE-MANIFEST.txt"), "release manifest documents V048 and reserves V049") -require("V048" in read("migrations/README.md") and "V049" in read("migrations/README.md"), "migration guide documents V048 and reserves V049") +require("V049" in read("RELEASE-MANIFEST.txt") and "V050" in read("RELEASE-MANIFEST.txt"), "release manifest documents V049 and reserves V050") +require("V049" in read("migrations/README.md") and "V050" in read("migrations/README.md"), "migration guide documents V049 and reserves V050") require((ROOT / "docs/STORAGE_SEARCH_OPERATIONS.md").is_file(), "storage/search operations guide exists") require('imap_host = ""' not in read("config.dev.toml"), "development config does not override IMAP host with an invalid empty value") require(not re.search(r"(?m)^IMAP_HOST=\s*$", read(".env.example")), "environment example does not set an invalid empty IMAP host") @@ -521,6 +521,10 @@ try: require(outbox is not None, "fresh SQLite schema has search_outbox") outbox_columns = {row[1] for row in conn.execute("PRAGMA table_info(u_test__search_outbox)")} require("claimed_until" in outbox_columns, "fresh SQLite search_outbox separates worker lease from retry availability") + blob_outbox = conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='u_test__blob_delete_outbox'").fetchone() + require(blob_outbox is not None, "fresh SQLite schema has blob_delete_outbox") + blob_outbox_columns = {row[1] for row in conn.execute("PRAGMA table_info(u_test__blob_delete_outbox)")} + require({"storage_key", "attempts", "available_at", "claimed_until", "last_error"}.issubset(blob_outbox_columns), "fresh SQLite blob deletion outbox has retry/lease state") folder_columns = {row[1] for row in conn.execute("PRAGMA table_info(u_test__folders)")} require("counter_generation" in folder_columns, "fresh SQLite schema has V048 folder counter generation") sqlite_header_sync = "INSERT INTO u_test__user_meta (key, value) VALUES ('header_sync_generation', '1') ON CONFLICT(key) DO UPDATE SET value = CAST(value AS INTEGER) + 1" @@ -595,12 +599,26 @@ try: except Exception as exc: errors.append(f"SQLite V048 migration execution failed: {exc}") +# V049 creates the durable blob deletion outbox. +try: + conn = sqlite3.connect(":memory:") + migration = read("migrations/sqlite/V049__blob_delete_outbox.sql").replace("{{p}}", "u_test__") + conn.executescript(migration) + columns = {row[1] for row in conn.execute("PRAGMA table_info(u_test__blob_delete_outbox)")} + require({"id", "storage_key", "attempts", "available_at", "claimed_until", "last_error", "created_at", "updated_at"}.issubset(columns), "V049 creates SQLite blob_delete_outbox with durable retry state") + indexes = {row[1] for row in conn.execute("PRAGMA index_list(u_test__blob_delete_outbox)")} + require(any("blob_delete_outbox_ready" in name for name in indexes), "V049 creates SQLite blob cleanup readiness index") + conn.close() +except Exception as exc: + errors.append(f"SQLite V049 migration execution failed: {exc}") + # Migration/query parity across SQL engines. for dialect in ("sqlite", "mysql", "pgsql"): require((ROOT / f"migrations/{dialect}/V047__blob_keys_and_search_outbox.sql").is_file(), f"V047 migration exists for {dialect}") require((ROOT / f"migrations/{dialect}/V048__folder_counter_generation.sql").is_file(), f"V048 migration exists for {dialect}") + require((ROOT / f"migrations/{dialect}/V049__blob_delete_outbox.sql").is_file(), f"V049 migration exists for {dialect}") schema_text = read(f"migrations/{dialect}/schema.sql") - require("Current runtime schema: V048" in schema_text and "counter_generation" in schema_text, f"fresh {dialect} schema includes V048 folder counter generation") + require("Current runtime schema: V049" in schema_text and "counter_generation" in schema_text and "blob_delete_outbox" in schema_text, f"fresh {dialect} schema includes V049 blob deletion outbox") queries = read(f"src/queries/{dialect}/queries.rs") folder_block = re.search(r"pub mod db_folders\s*\{(.*?)\n\}", queries, re.S) require(folder_block is not None and "Q_032" in folder_block.group(1) and "counter_generation = counter_generation + 1" in folder_block.group(1), f"folder counter CAS increments generation for {dialect}") @@ -609,11 +627,17 @@ for dialect in ("sqlite", "mysql", "pgsql"): if block_match: qnames = set(re.findall(r"pub const (Q_\d{3})", block_match.group(1))) require(qnames == {f"Q_{i:03d}" for i in range(1, 8)}, f"search outbox Q001-Q007 parity for {dialect}") + blob_block = re.search(r"pub mod db_blob_delete_outbox\s*\{(.*?)\n\}", queries, re.S) + require(blob_block is not None, f"blob deletion outbox query module exists for {dialect}") + if blob_block: + blob_qnames = set(re.findall(r"pub const (Q_\d{3})", blob_block.group(1))) + require(blob_qnames == {f"Q_{i:03d}" for i in range(1, 6)}, f"blob deletion outbox Q001-Q005 parity for {dialect}") pool = read("src/db/pool.rs") for dialect in ("SQLITE", "MYSQL", "PGSQL"): require(f"{dialect}_MIGRATIONS" in pool and "V047__blob_keys_and_search_outbox.sql" in pool, f"runtime migration catalog includes V047 ({dialect.lower()})") require(f"{dialect}_MIGRATIONS" in pool and "V048__folder_counter_generation.sql" in pool, f"runtime migration catalog includes V048 ({dialect.lower()})") + require(f"{dialect}_MIGRATIONS" in pool and "V049__blob_delete_outbox.sql" in pool, f"runtime migration catalog includes V049 ({dialect.lower()})") require("const BASE_SCHEMA_VERSION: u32 = 46" in pool, "legacy squashed baseline stays V046") require("latest_schema_version" in pool, "runtime schema version is computed from migrations") require( @@ -706,9 +730,9 @@ require("storage.shared" in config_rs, "cluster-safe shared POSIX storage is cap require("AmazonS3Builder" in storage_rs and "BufWriter" in storage_rs, "S3 backend uses streaming object-store writer") require("with_retry" in storage_rs and "with_checksum_algorithm" in storage_rs, "S3 backend configures retries and checksums") require("attachment_key" in storage_rs and "users/{user_hash}/drafts/{draft_id}/attachments/{attachment_id}" in storage_rs, "attachments use logical storage keys") -require("destination.delete(&target_key).await" in storage_rs, "attachment migration removes a failed destination blob after verification failure") +require("enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key).await" in storage_rs, "attachment migration durably queues failed destination blob cleanup") require("let _ = upload.abort().await" in storage_rs and "failed reading source" in storage_rs, "attachment migration aborts a destination upload if source streaming fails") -require("if let Err(error) = crate::db::drafts::update_attachment_storage" in storage_rs, "attachment migration removes destination object when SQL pointer update fails") +require("if let Err(error) = crate::db::drafts::update_attachment_storage" in storage_rs and "Failed to queue storage migration cleanup" in storage_rs, "attachment migration queues destination cleanup when SQL pointer update fails") require("OpenSearchBackend" in search_rs, "OpenSearch backend is implemented") require("reqwest::blocking" not in search_rs and 'features = ["json", "rustls-tls"]' in read("Cargo.toml"), "OpenSearch uses the async reqwest client without blocking feature") require("async fn request(" in search_rs and "request.send().await" in search_rs, "OpenSearch HTTP requests are asynchronous") @@ -717,6 +741,15 @@ require("process_search_outbox" in search_rs and "claim(&conn" in search_rs, "Op require("fn event_lease_secs(&self) -> u64" in search_rs and "let lease_secs = backend.event_lease_secs();" in search_rs and "claim(&conn, &event, 60)" not in search_rs, "OpenSearch outbox lease is derived from HTTP timeout/failover budget instead of a fixed 60 seconds") require("opensearch_event_lease_covers_timeout_and_failover_budget" in search_rs, "OpenSearch lease budget has a source-level unit regression test") require("claimed_until" in read("migrations/sqlite/V047__blob_keys_and_search_outbox.sql") and "reenqueue_preserves_active_claim_lease" in read("src/db/search_outbox.rs"), "outbox re-enqueue cannot steal an active worker lease") +blob_delete_rs = read("src/db/blob_delete_outbox.rs") +attachments_rs = read("src/routes/attachments.rs") +send_rs = read("src/routes/send.rs") +require("mark_failure" in blob_delete_rs and "min(3600)" in blob_delete_rs and "max_attempts" not in blob_delete_rs and "dead_letter" not in read("migrations/sqlite/V049__blob_delete_outbox.sql"), "blob deletion outbox retries indefinitely with capped backoff and no permanent dead-letter") +require("unchecked_transaction()" in attachments_rs and "blob_delete_outbox::enqueue(&transaction" in attachments_rs and "delete_draft_attachment(&transaction" in attachments_rs, "single attachment deletion atomically queues blob cleanup with metadata deletion") +require("blob_delete_outbox::enqueue(&transaction" in drafts_rs and "delete_draft(&transaction" in drafts_rs, "whole-draft deletion atomically queues every attachment blob before metadata cascade") +require("blob_delete_outbox::enqueue(&transaction" in send_rs and "delete_draft(&transaction" in send_rs, "post-send draft cleanup atomically queues attachment blobs") +require("start_blob_delete_worker" in storage_rs and "process_blob_delete_outbox" in storage_rs and "start_blob_delete_worker(config.clone(), storage.clone())" in read("src/main.rs"), "durable blob deletion worker is wired at startup") +require('let key = "_health/probe".to_string();' in storage_rs, "storage health probe uses a fixed key and cannot accumulate unique orphan probe objects") require("index_messages_batch_with_connection" in search_rs and "enqueue_upsert_on" in search_rs, "OpenSearch outbox supports caller-owned SQL transactions") require("enqueue_participates_in_caller_transaction" in read("src/db/search_outbox.rs"), "transactional outbox rollback has a unit test") require("search reindex" in read("README.md") and "run_search_reindex" in read("src/main.rs"), "full search reindex command is documented and wired") @@ -734,19 +767,25 @@ require("docs/RUSTBOX_PLAN.md" not in read("docs/IMPLEMENTATION_STATUS.md"), "im require("intentionally run one RustBox application container" in read("docker/README.md"), "Docker docs distinguish shared backing topology from multi-app load balancing") attachments_rs = read("src/routes/attachments.rs") -require(".begin_upload(&key)" in attachments_rs and ".get(&attachment.storage_key)" in attachments_rs and (".delete(&attachment.storage_key)" in attachments_rs or ".delete(&storage_key)" in attachments_rs), "attachment HTTP routes use BlobStorage") +require(".begin_upload(&key)" in attachments_rs and ".get(&attachment.storage_key)" in attachments_rs and "storage::process_queued_delete" in attachments_rs, "attachment HTTP routes use BlobStorage") require("tokio::fs::" not in attachments_rs, "attachment HTTP routes no longer access local filesystem directly") -attachment_db_delete = attachments_rs.find("db::drafts::delete_draft_attachment") -attachment_blob_delete = attachments_rs.find("storage.delete(&storage_key)", attachment_db_delete) -require(attachment_db_delete >= 0 and attachment_blob_delete > attachment_db_delete, "attachment delete commits DB removal before best-effort blob cleanup") +attachment_enqueue = attachments_rs.find("blob_delete_outbox::enqueue(&transaction") +attachment_db_delete = attachments_rs.find("delete_draft_attachment(&transaction", attachment_enqueue) +attachment_commit = attachments_rs.find(".commit()", attachment_db_delete) +attachment_blob_delete = attachments_rs.find("storage::process_queued_delete", attachment_commit) +require(0 <= attachment_enqueue < attachment_db_delete < attachment_commit < attachment_blob_delete, "attachment delete commits metadata removal and cleanup enqueue atomically before physical blob deletion") drafts_rs = read("src/routes/drafts.rs") -draft_db_delete = drafts_rs.find("db::drafts::delete_draft(&conn, &id)") -draft_blob_delete = drafts_rs.find("storage.delete(&attachment.storage_key)", draft_db_delete) -require(draft_db_delete >= 0 and draft_blob_delete > draft_db_delete, "draft delete commits DB removal before best-effort blob cleanup") +draft_enqueue = drafts_rs.find("blob_delete_outbox::enqueue(&transaction") +draft_db_delete = drafts_rs.find("delete_draft(&transaction", draft_enqueue) +draft_commit = drafts_rs.find(".commit()", draft_db_delete) +draft_blob_delete = drafts_rs.find("storage::process_queued_delete", draft_commit) +require(0 <= draft_enqueue < draft_db_delete < draft_commit < draft_blob_delete, "draft delete commits metadata cascade and cleanup enqueue atomically before physical blob deletion") send_rs = read("src/routes/send.rs") -send_db_delete = send_rs.find("db::drafts::delete_draft(&conn, draft_id)") -send_blob_delete = send_rs.find("storage.delete(&attachment.storage_key)", send_db_delete) -require(send_db_delete >= 0 and send_blob_delete > send_db_delete, "post-send cleanup commits DB removal before best-effort blob cleanup") +send_enqueue = send_rs.find("blob_delete_outbox::enqueue(&transaction") +send_db_delete = send_rs.find("delete_draft(&transaction", send_enqueue) +send_commit = send_rs.find("transaction.commit()", send_db_delete) +send_blob_delete = send_rs.find("storage::process_queued_delete", send_commit) +require(0 <= send_enqueue < send_db_delete < send_commit < send_blob_delete, "post-send cleanup commits metadata removal and cleanup enqueue atomically before physical blob deletion") # Repository/source-package hygiene independent of Cargo. source_files = [path for path in ROOT.rglob("*") if path.is_file() and ".git" not in path.parts and "target" not in path.parts] diff --git a/src/db/blob_delete_outbox.rs b/src/db/blob_delete_outbox.rs new file mode 100644 index 0000000..5de5e96 --- /dev/null +++ b/src/db/blob_delete_outbox.rs @@ -0,0 +1,155 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::db::pool::{Connection, params}; + +#[derive(Debug, Clone)] +pub struct BlobDeleteEvent { + pub id: String, + pub storage_key: String, + pub attempts: u32, +} + +pub fn now_epoch() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + .min(i64::MAX as u64) as i64 +} + +pub fn enqueue(conn: &Connection, storage_key: &str) -> Result { + if storage_key.trim().is_empty() { + return Err("cannot enqueue an empty blob storage key".to_string()); + } + let id = uuid::Uuid::new_v4().to_string(); + let now = now_epoch(); + let result = match crate::db::pool::configured_backend() { + crate::db::pool::DatabaseBackend::Mysql => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_001), + params![&id, storage_key, now, now], + ), + _ => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_001), + params![&id, storage_key, now], + ), + }; + result.map_err(|e| format!("failed to enqueue blob deletion: {e}"))?; + Ok(BlobDeleteEvent { + id, + storage_key: storage_key.to_string(), + attempts: 0, + }) +} + +pub fn list_ready(conn: &Connection, limit: usize) -> Result, String> { + let now = now_epoch(); + let mut stmt = conn + .prepare(crate::dbq!(db_blob_delete_outbox::Q_002)) + .map_err(|e| format!("failed to prepare blob deletion outbox query: {e}"))?; + let query_params = match crate::db::pool::configured_backend() { + crate::db::pool::DatabaseBackend::Mysql => params![now, now, limit], + _ => params![now, limit], + }; + let rows = stmt + .query_map(query_params, |row| { + Ok(BlobDeleteEvent { + id: row.get(0)?, + storage_key: row.get(1)?, + attempts: row.get::<_, i64>(2)?.max(0) as u32, + }) + }) + .map_err(|e| format!("failed to query blob deletion outbox: {e}"))?; + rows.map(|row| row.map_err(|e| format!("failed to read blob deletion outbox row: {e}"))) + .collect() +} + +pub fn claim(conn: &Connection, event: &BlobDeleteEvent, lease_secs: u64) -> Result { + let now = now_epoch(); + let lease_until = now.saturating_add(lease_secs.max(1) as i64); + let changed = match crate::db::pool::configured_backend() { + crate::db::pool::DatabaseBackend::Mysql => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_003), + params![lease_until, &event.id, now, now], + ), + _ => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_003), + params![&event.id, lease_until, now], + ), + } + .map_err(|e| format!("failed to claim blob deletion: {e}"))?; + Ok(changed > 0) +} + +pub fn mark_success(conn: &Connection, event: &BlobDeleteEvent) -> Result { + let changed = conn + .execute( + crate::dbq!(db_blob_delete_outbox::Q_004), + params![&event.id], + ) + .map_err(|e| format!("failed to acknowledge blob deletion: {e}"))?; + Ok(changed > 0) +} + +pub fn mark_failure( + conn: &Connection, + event: &BlobDeleteEvent, + error: &str, +) -> Result { + let attempts = event.attempts.saturating_add(1); + let backoff = 2u64.saturating_pow(attempts.min(12)).min(3600); + let available_at = now_epoch().saturating_add(backoff as i64); + let now = now_epoch(); + let truncated: String = error.chars().take(2048).collect(); + let changed = match crate::db::pool::configured_backend() { + crate::db::pool::DatabaseBackend::Mysql => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_005), + params![attempts, available_at, &truncated, now, &event.id], + ), + _ => conn.execute( + crate::dbq!(db_blob_delete_outbox::Q_005), + params![&event.id, attempts, available_at, &truncated, now], + ), + } + .map_err(|e| format!("failed to update blob deletion retry state: {e}"))?; + Ok(changed > 0) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::pool::open_test_db; + + #[test] + fn enqueue_participates_in_caller_transaction() { + let conn = open_test_db(); + { + let transaction = conn.unchecked_transaction().unwrap(); + enqueue(&transaction, "users/test/drafts/d/attachments/a").unwrap(); + } + assert!(list_ready(&conn, 10).unwrap().is_empty()); + } + + #[test] + fn failed_deletions_are_retried_instead_of_dead_lettered() { + let conn = open_test_db(); + let event = enqueue(&conn, "users/test/drafts/d/attachments/a").unwrap(); + assert!(claim(&conn, &event, 60).unwrap()); + assert!(mark_failure(&conn, &event, "storage unavailable").unwrap()); + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM blob_delete_outbox WHERE id=?1", + params![&event.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(count, 1); + } + + #[test] + fn successful_delete_is_acknowledged() { + let conn = open_test_db(); + let event = enqueue(&conn, "users/test/drafts/d/attachments/a").unwrap(); + assert!(mark_success(&conn, &event).unwrap()); + assert!(list_ready(&conn, 10).unwrap().is_empty()); + } +} diff --git a/src/db/drafts.rs b/src/db/drafts.rs index f1ae892..7dd1bce 100644 --- a/src/db/drafts.rs +++ b/src/db/drafts.rs @@ -257,6 +257,67 @@ pub fn get_draft_attachments( Ok(attachments) } +pub fn lock_draft_for_update(conn: &Connection, id: &str) -> Result { + match conn.query_row(crate::dbq!(db_drafts::Q_010), params![id], |_row| Ok(())) { + Ok(()) => Ok(true), + Err(crate::db::pool::DbError::QueryReturnedNoRows) => Ok(false), + Err(e) => Err(format!("Failed to lock draft: {e}")), + } +} + +pub fn get_draft_attachment_for_update( + conn: &Connection, + draft_id: &str, + attachment_id: &str, +) -> Result, String> { + let result = conn.query_row( + crate::dbq!(db_drafts::Q_011), + params![draft_id, attachment_id], + |row| { + Ok(DraftAttachment { + id: row.get(0)?, + draft_id: row.get(1)?, + filename: row.get(2)?, + content_type: row.get(3)?, + size: row.get(4)?, + storage_key: row.get(5)?, + checksum_sha256: row.get(6)?, + created_at: row.get(7)?, + }) + }, + ); + match result { + Ok(attachment) => Ok(Some(attachment)), + Err(crate::db::pool::DbError::QueryReturnedNoRows) => Ok(None), + Err(e) => Err(format!("Failed to lock draft attachment: {e}")), + } +} + +pub fn get_draft_attachments_for_update( + conn: &Connection, + draft_id: &str, +) -> Result, String> { + let mut stmt = conn + .prepare(crate::dbq!(db_drafts::Q_012)) + .map_err(|e| format!("Failed to prepare locked draft attachments query: {e}"))?; + let rows = stmt + .query_map(params![draft_id], |row| { + Ok(DraftAttachment { + id: row.get(0)?, + draft_id: row.get(1)?, + filename: row.get(2)?, + content_type: row.get(3)?, + size: row.get(4)?, + storage_key: row.get(5)?, + checksum_sha256: row.get(6)?, + created_at: row.get(7)?, + }) + }) + .map_err(|e| format!("Failed to query locked draft attachments: {e}"))?; + rows.map(|row| row.map_err(|e| format!("Failed to read locked attachment row: {e}"))) + .collect() +} + pub fn delete_draft_attachment(conn: &Connection, id: &str) -> Result { let deleted = conn .execute(crate::dbq!(db_drafts::Q_008), params![id]) diff --git a/src/db/mod.rs b/src/db/mod.rs index be1fa85..34ac2fc 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -1,3 +1,4 @@ +pub mod blob_delete_outbox; pub mod calendar; pub mod contact_groups; pub mod contacts; diff --git a/src/db/pool.rs b/src/db/pool.rs index 059ccb1..bca51da 100644 --- a/src/db/pool.rs +++ b/src/db/pool.rs @@ -30,6 +30,10 @@ const SQLITE_MIGRATIONS: &[(u32, &str)] = &[ 48, include_str!("../../migrations/sqlite/V048__folder_counter_generation.sql"), ), + ( + 49, + include_str!("../../migrations/sqlite/V049__blob_delete_outbox.sql"), + ), ]; const MYSQL_MIGRATIONS: &[(u32, &str)] = &[ ( @@ -40,6 +44,10 @@ const MYSQL_MIGRATIONS: &[(u32, &str)] = &[ 48, include_str!("../../migrations/mysql/V048__folder_counter_generation.sql"), ), + ( + 49, + include_str!("../../migrations/mysql/V049__blob_delete_outbox.sql"), + ), ]; const PGSQL_MIGRATIONS: &[(u32, &str)] = &[ ( @@ -50,9 +58,14 @@ const PGSQL_MIGRATIONS: &[(u32, &str)] = &[ 48, include_str!("../../migrations/pgsql/V048__folder_counter_generation.sql"), ), + ( + 49, + include_str!("../../migrations/pgsql/V049__blob_delete_outbox.sql"), + ), ]; const TENANT_TABLES: &[&str] = &[ + "blob_delete_outbox", "calendar_events", "calendar_settings", "contact_group_members", @@ -83,6 +96,7 @@ const TENANT_TABLES: &[&str] = &[ ]; const IMPORT_TABLES: &[&str] = &[ + "blob_delete_outbox", "calendar_events", "calendar_settings", "contact_groups", diff --git a/src/main.rs b/src/main.rs index bc0c5f1..85cdccb 100644 --- a/src/main.rs +++ b/src/main.rs @@ -121,6 +121,8 @@ async fn main() -> Result<(), Box> { .map_err(std::io::Error::other)?; } let _search_worker = search_engine.start_worker(); + let _blob_delete_worker = + crate::storage::start_blob_delete_worker(config.clone(), storage.clone()); let event_bus = Arc::new(realtime::events::EventBus::new()); let idle_manager = Arc::new(realtime::idle::IdleManager::new()); diff --git a/src/queries/mysql/queries.rs b/src/queries/mysql/queries.rs index 25d21e1..b8a3a22 100644 --- a/src/queries/mysql/queries.rs +++ b/src/queries/mysql/queries.rs @@ -167,6 +167,9 @@ pub mod db_drafts { pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = ?"; pub const Q_009: &str = "UPDATE draft_attachments SET file_path=?, checksum_sha256=? WHERE id=?"; + pub const Q_010: &str = "SELECT id FROM drafts WHERE id=? FOR UPDATE"; + pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=? AND id=? FOR UPDATE"; + pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=? ORDER BY created_at ASC FOR UPDATE"; } pub mod db_folder_preferences { @@ -527,3 +530,11 @@ pub mod db_search_outbox { pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=? WHERE id=? AND generation=? AND dead_letter=0 AND available_at <= ? AND claimed_until <= ?"; pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < ?"; } + +pub mod db_blob_delete_outbox { + pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES (?, ?, 0, 0, 0, NULL, ?, ?)"; + pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= ? AND claimed_until <= ? ORDER BY updated_at ASC LIMIT ?"; + pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=? WHERE id=? AND available_at <= ? AND claimed_until <= ?"; + pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=?"; + pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=?, available_at=?, claimed_until=0, last_error=?, updated_at=? WHERE id=?"; +} diff --git a/src/queries/pgsql/queries.rs b/src/queries/pgsql/queries.rs index 0e9a39d..453b27b 100644 --- a/src/queries/pgsql/queries.rs +++ b/src/queries/pgsql/queries.rs @@ -168,6 +168,9 @@ pub mod db_drafts { pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = $1"; pub const Q_009: &str = "UPDATE draft_attachments SET file_path=$1, checksum_sha256=$2 WHERE id=$3"; + pub const Q_010: &str = "SELECT id FROM drafts WHERE id=$1 FOR UPDATE"; + pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=$1 AND id=$2 FOR UPDATE"; + pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=$1 ORDER BY created_at ASC FOR UPDATE"; } pub mod db_folder_preferences { @@ -528,3 +531,11 @@ pub mod db_search_outbox { pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=$3 WHERE id=$1 AND generation=$2 AND dead_letter=0 AND available_at <= $4 AND claimed_until <= $4"; pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < $1"; } + +pub mod db_blob_delete_outbox { + pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES ($1, $2, 0, 0, 0, NULL, $3, $3)"; + pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= $1 AND claimed_until <= $1 ORDER BY updated_at ASC LIMIT $2"; + pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=$2 WHERE id=$1 AND available_at <= $3 AND claimed_until <= $3"; + pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=$1"; + pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=$2, available_at=$3, claimed_until=0, last_error=$4, updated_at=$5 WHERE id=$1"; +} diff --git a/src/queries/sqlite/queries.rs b/src/queries/sqlite/queries.rs index 71b8572..7957d95 100644 --- a/src/queries/sqlite/queries.rs +++ b/src/queries/sqlite/queries.rs @@ -165,6 +165,9 @@ pub mod db_drafts { pub const Q_008: &str = "DELETE FROM draft_attachments WHERE id = ?1"; pub const Q_009: &str = "UPDATE draft_attachments SET file_path=?1, checksum_sha256=?2 WHERE id=?3"; + pub const Q_010: &str = "SELECT id FROM drafts WHERE id=?1"; + pub const Q_011: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=?1 AND id=?2"; + pub const Q_012: &str = "SELECT id, draft_id, filename, content_type, size, file_path, checksum_sha256, created_at FROM draft_attachments WHERE draft_id=?1 ORDER BY created_at ASC"; } pub mod db_folder_preferences { @@ -782,3 +785,11 @@ pub mod db_search_outbox { pub const Q_006: &str = "UPDATE search_outbox SET claimed_until=?3 WHERE id=?1 AND generation=?2 AND dead_letter=0 AND available_at <= ?4 AND claimed_until <= ?4"; pub const Q_007: &str = "DELETE FROM search_outbox WHERE updated_at < ?1"; } + +pub mod db_blob_delete_outbox { + pub const Q_001: &str = "INSERT INTO blob_delete_outbox (id, storage_key, attempts, available_at, claimed_until, last_error, created_at, updated_at) VALUES (?1, ?2, 0, 0, 0, NULL, ?3, ?3)"; + pub const Q_002: &str = "SELECT id, storage_key, attempts FROM blob_delete_outbox WHERE available_at <= ?1 AND claimed_until <= ?1 ORDER BY updated_at ASC LIMIT ?2"; + pub const Q_003: &str = "UPDATE blob_delete_outbox SET claimed_until=?2 WHERE id=?1 AND available_at <= ?3 AND claimed_until <= ?3"; + pub const Q_004: &str = "DELETE FROM blob_delete_outbox WHERE id=?1"; + pub const Q_005: &str = "UPDATE blob_delete_outbox SET attempts=?2, available_at=?3, claimed_until=0, last_error=?4, updated_at=?5 WHERE id=?1"; +} diff --git a/src/routes/attachments.rs b/src/routes/attachments.rs index b60bb00..4f14515 100644 --- a/src/routes/attachments.rs +++ b/src/routes/attachments.rs @@ -11,7 +11,7 @@ use crate::auth::session::SessionState; use crate::config::AppConfig; use crate::db; use crate::error::AppError; -use crate::storage::{BlobStorage, attachment_key}; +use crate::storage::{self, BlobStorage, attachment_key}; #[derive(Debug, Serialize)] struct UploadResponse { @@ -73,7 +73,13 @@ pub async fn upload_attachment( let chunk = match field.chunk().await { Ok(chunk) => chunk, Err(e) => { - let _ = writer.abort().await; + if writer.abort().await.is_err() + && let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key) + .await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue aborted attachment cleanup"); + } return Err(AppError::BadRequest(format!( "Failed to read field data: {e}" ))); @@ -82,7 +88,12 @@ pub async fn upload_attachment( let Some(chunk) = chunk else { break }; size = size.saturating_add(chunk.len()); if size > max_attachment_size { - let _ = writer.abort().await; + if writer.abort().await.is_err() + && let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue oversized attachment cleanup"); + } return Err(AppError::BadRequest(format!( "File '{}' exceeds maximum size of {} MiB", filename, config.max_attachment_size_mb @@ -90,18 +101,37 @@ pub async fn upload_attachment( } checksum.update(&chunk); if let Err(e) = writer.write(chunk).await { - let _ = writer.abort().await; + if writer.abort().await.is_err() + && let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue interrupted attachment cleanup"); + } return Err(AppError::ServiceUnavailable(format!( "Failed to upload attachment: {e}" ))); } } - let stat = writer.finish().await.map_err(|e| { - AppError::ServiceUnavailable(format!("Failed to finalize attachment upload: {e}")) - })?; + let stat = match writer.finish().await { + Ok(stat) => stat, + Err(e) => { + if let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue failed-finalize attachment cleanup"); + } + return Err(AppError::ServiceUnavailable(format!( + "Failed to finalize attachment upload: {e}" + ))); + } + }; if stat.size != size as u64 { - let _ = storage.delete(&key).await; + if let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue invalid attachment cleanup"); + } return Err(AppError::InternalError(format!( "Attachment size verification failed for '{filename}'" ))); @@ -118,7 +148,11 @@ pub async fn upload_attachment( checksum_sha256: Some(&checksum), }; if let Err(e) = db::drafts::add_draft_attachment(&conn, &attachment) { - let _ = storage.delete(&key).await; + if let Err(cleanup_error) = + storage::enqueue_unreferenced_delete(&conn, storage.as_ref(), &key).await + { + tracing::error!(error = %cleanup_error, storage_key = %key, "Failed to queue unattached blob cleanup"); + } return Err(AppError::InternalError(e)); } @@ -180,21 +214,30 @@ pub async fn delete_attachment( let conn = db::pool::open_user_db(&config.data_dir, &session.user_hash) .map_err(|e| AppError::InternalError(format!("Failed to open database: {e}")))?; - let attachments = - db::drafts::get_draft_attachments(&conn, &draft_id).map_err(AppError::InternalError)?; - let attachment = attachments - .iter() - .find(|a| a.id == attachment_id) - .ok_or_else(|| AppError::NotFound("Attachment not found".to_string()))?; - + let transaction = conn.unchecked_transaction().map_err(|e| { + AppError::InternalError(format!( + "Failed to start attachment delete transaction: {e}" + )) + })?; + let attachment = + db::drafts::get_draft_attachment_for_update(&transaction, &draft_id, &attachment_id) + .map_err(AppError::InternalError)? + .ok_or_else(|| AppError::NotFound("Attachment not found".to_string()))?; let storage_key = attachment.storage_key.clone(); - let deleted = db::drafts::delete_draft_attachment(&conn, &attachment_id) + let cleanup_event = db::blob_delete_outbox::enqueue(&transaction, &storage_key) + .map_err(AppError::InternalError)?; + let deleted = db::drafts::delete_draft_attachment(&transaction, &attachment_id) .map_err(AppError::InternalError)?; if !deleted { return Err(AppError::NotFound("Attachment not found".to_string())); } - if let Err(error) = storage.delete(&storage_key).await { - tracing::warn!(error = %error, storage_key = %storage_key, "Failed to remove deleted attachment blob"); + transaction.commit().map_err(|e| { + AppError::InternalError(format!("Failed to commit attachment deletion: {e}")) + })?; + if let Err(error) = + storage::process_queued_delete(&conn, storage.as_ref(), &cleanup_event).await + { + tracing::warn!(error = %error, storage_key = %storage_key, "Failed to process queued attachment blob deletion"); } Ok(Json(DeleteResponse { diff --git a/src/routes/drafts.rs b/src/routes/drafts.rs index 625dd35..4eb5e8d 100644 --- a/src/routes/drafts.rs +++ b/src/routes/drafts.rs @@ -1179,25 +1179,44 @@ pub async fn delete_draft_handler( delete_imap_draft_source(imap_client.as_ref(), &creds, &drafts_folder, source).await?; } - let attachments = if existing.is_some() { - db::drafts::get_draft_attachments(&conn, &id).map_err(AppError::InternalError)? - } else { - Vec::new() - }; - if existing.is_some() { - let deleted = db::drafts::delete_draft(&conn, &id).map_err(AppError::InternalError)?; + let (attachments, cleanup_events) = if existing.is_some() { + let transaction = conn.unchecked_transaction().map_err(|e| { + AppError::InternalError(format!("Failed to start draft delete transaction: {e}")) + })?; + if !db::drafts::lock_draft_for_update(&transaction, &id).map_err(AppError::InternalError)? { + return Err(AppError::NotFound("Draft not found".to_string())); + } + let attachments = db::drafts::get_draft_attachments_for_update(&transaction, &id) + .map_err(AppError::InternalError)?; + let mut events = Vec::with_capacity(attachments.len()); + for attachment in &attachments { + events.push( + db::blob_delete_outbox::enqueue(&transaction, &attachment.storage_key) + .map_err(AppError::InternalError)?, + ); + } + let deleted = + db::drafts::delete_draft(&transaction, &id).map_err(AppError::InternalError)?; if !deleted { return Err(AppError::NotFound("Draft not found".to_string())); } - } + transaction.commit().map_err(|e| { + AppError::InternalError(format!("Failed to commit draft deletion: {e}")) + })?; + (attachments, events) + } else { + (Vec::new(), Vec::new()) + }; + let mut attachments_deleted = 0usize; - for attachment in &attachments { - match storage.delete(&attachment.storage_key).await { - Ok(()) => attachments_deleted += 1, + for (attachment, event) in attachments.iter().zip(cleanup_events.iter()) { + match storage::process_queued_delete(&conn, storage.as_ref(), event).await { + Ok(true) => attachments_deleted += 1, + Ok(false) => {} Err(error) => tracing::warn!( error = %error, storage_key = %attachment.storage_key, - "Failed to remove deleted draft attachment blob" + "Failed to process queued draft attachment deletion" ), } } diff --git a/src/routes/send.rs b/src/routes/send.rs index 192c985..ca88421 100644 --- a/src/routes/send.rs +++ b/src/routes/send.rs @@ -457,26 +457,65 @@ async fn load_draft_attachments( } async fn cleanup_draft(data_dir: &str, user_hash: &str, draft_id: &str, storage: &dyn BlobStorage) { - let Ok(conn) = db::pool::open_user_db(data_dir, user_hash) else { - return; + let conn = match db::pool::open_user_db(data_dir, user_hash) { + Ok(conn) => conn, + Err(error) => { + tracing::warn!(error = %error, "Failed to open database for draft cleanup after send"); + return; + } }; - let attachments = db::drafts::get_draft_attachments(&conn, draft_id).unwrap_or_default(); - let deleted = match db::drafts::delete_draft(&conn, draft_id) { + let transaction = match conn.unchecked_transaction() { + Ok(transaction) => transaction, + Err(error) => { + tracing::warn!(error = %error, "Failed to start draft cleanup transaction after send"); + return; + } + }; + match db::drafts::lock_draft_for_update(&transaction, draft_id) { + Ok(true) => {} + Ok(false) => return, + Err(error) => { + tracing::warn!(error = %error, "Failed to lock draft for cleanup after send"); + return; + } + } + let attachments = match db::drafts::get_draft_attachments_for_update(&transaction, draft_id) { + Ok(attachments) => attachments, + Err(error) => { + tracing::warn!(error = %error, "Failed to load draft attachments for cleanup after send"); + return; + } + }; + let mut cleanup_events = Vec::with_capacity(attachments.len()); + for attachment in &attachments { + match db::blob_delete_outbox::enqueue(&transaction, &attachment.storage_key) { + Ok(event) => cleanup_events.push(event), + Err(error) => { + tracing::warn!(error = %error, storage_key = %attachment.storage_key, "Failed to queue draft attachment cleanup after send"); + return; + } + } + } + let deleted = match db::drafts::delete_draft(&transaction, draft_id) { Ok(deleted) => deleted, - Err(e) => { - tracing::warn!(error = %e, "Failed to delete draft after send"); + Err(error) => { + tracing::warn!(error = %error, "Failed to delete draft after send"); return; } }; if !deleted { return; } - for attachment in attachments { - if let Err(e) = storage.delete(&attachment.storage_key).await { + if let Err(error) = transaction.commit() { + tracing::warn!(error = %error, "Failed to commit draft cleanup after send"); + return; + } + for (attachment, event) in attachments.iter().zip(cleanup_events.iter()) { + if let Err(error) = storage::process_queued_delete(&conn, storage, event).await { tracing::warn!( - error = %e, + error = %error, storage_key = %attachment.storage_key, - "Failed to delete draft attachment after send" + "Failed to process queued draft attachment cleanup after send" ); } } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index f503091..d6913f1 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -160,13 +160,17 @@ impl BlobStorage for ObjectStorage { } async fn health(&self) -> Result<(), String> { - let key = format!("_health/{}", uuid::Uuid::new_v4()); + let key = "_health/probe".to_string(); let mut upload = self.begin_upload(&key).await?; upload.write(Bytes::from_static(b"ok")).await?; let stat = upload.finish().await?; if stat.size != 2 { - let _ = self.delete(&key).await; - return Err("storage health probe returned unexpected size".to_string()); + return match self.delete(&key).await { + Ok(()) => Err("storage health probe returned unexpected size".to_string()), + Err(cleanup_error) => Err(format!( + "storage health probe returned unexpected size and cleanup failed: {cleanup_error}" + )), + }; } self.delete(&key).await?; Ok(()) @@ -311,6 +315,89 @@ pub fn from_config(config: &AppConfig) -> Result, String> { } } +const BLOB_DELETE_BATCH_SIZE: usize = 100; +const BLOB_DELETE_LEASE_SECS: u64 = 60; +const BLOB_DELETE_WORKER_INTERVAL_SECS: u64 = 30; + +pub async fn process_queued_delete( + conn: &crate::db::pool::Connection, + storage: &dyn BlobStorage, + event: &crate::db::blob_delete_outbox::BlobDeleteEvent, +) -> Result { + if !crate::db::blob_delete_outbox::claim(conn, event, BLOB_DELETE_LEASE_SECS)? { + return Ok(false); + } + match storage.delete(&event.storage_key).await { + Ok(()) => { + crate::db::blob_delete_outbox::mark_success(conn, event)?; + Ok(true) + } + Err(error) => { + crate::db::blob_delete_outbox::mark_failure(conn, event, &error)?; + tracing::warn!( + error = %error, + storage_key = %event.storage_key, + attempts = event.attempts.saturating_add(1), + "Blob deletion failed; cleanup remains queued" + ); + Ok(false) + } + } +} + +pub async fn enqueue_unreferenced_delete( + conn: &crate::db::pool::Connection, + storage: &dyn BlobStorage, + storage_key: &str, +) -> Result { + match crate::db::blob_delete_outbox::enqueue(conn, storage_key) { + Ok(event) => process_queued_delete(conn, storage, &event).await, + Err(queue_error) => match storage.delete(storage_key).await { + Ok(()) => Ok(true), + Err(delete_error) => Err(format!( + "failed to persist blob cleanup ({queue_error}); direct delete also failed: {delete_error}" + )), + }, + } +} + +pub async fn process_blob_delete_outbox( + config: &AppConfig, + storage: &dyn BlobStorage, +) -> Result { + let mut processed = 0usize; + for user_hash in crate::db::pool::list_account_user_hashes()? { + let conn = crate::db::pool::open_user_db(&config.data_dir, &user_hash)?; + let events = crate::db::blob_delete_outbox::list_ready(&conn, BLOB_DELETE_BATCH_SIZE)?; + for event in events { + if process_queued_delete(&conn, storage, &event).await? { + processed = processed.saturating_add(1); + } + } + } + Ok(processed) +} + +pub fn start_blob_delete_worker( + config: Arc, + storage: Arc, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + loop { + match process_blob_delete_outbox(&config, storage.as_ref()).await { + Ok(processed) if processed > 0 => { + tracing::debug!(processed, "Processed queued blob deletions"); + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "Blob deletion worker pass failed"); + } + } + tokio::time::sleep(Duration::from_secs(BLOB_DELETE_WORKER_INTERVAL_SECS)).await; + } + }) +} + pub async fn read_all(storage: &dyn BlobStorage, key: &str) -> Result, String> { let mut stream = storage.get(key).await?; let mut data = Vec::new(); @@ -537,7 +624,11 @@ pub async fn migrate_attachments( let source_checksum = format!("{:x}", hasher.finalize()); let target_stat = upload.finish().await?; if target_stat.size != source_meta.len() { - let _ = destination.delete(&target_key).await; + if let Err(cleanup_error) = + enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key).await + { + eprintln!("fail {target_key}: cleanup queue failed: {cleanup_error}"); + } report.failed += 1; eprintln!("fail {target_key}: size verification failed"); continue; @@ -548,14 +639,29 @@ pub async fn migrate_attachments( { Ok(checksum) => checksum, Err(error) => { - let _ = destination.delete(&target_key).await; + if let Err(cleanup_error) = enqueue_unreferenced_delete( + &conn, + destination.as_ref(), + &target_key, + ) + .await + { + eprintln!( + "fail {target_key}: cleanup queue failed: {cleanup_error}" + ); + } report.failed += 1; eprintln!("fail {target_key}: checksum verification failed: {error}"); continue; } }; if target_checksum != source_checksum { - let _ = destination.delete(&target_key).await; + if let Err(cleanup_error) = + enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key) + .await + { + eprintln!("fail {target_key}: cleanup queue failed: {cleanup_error}"); + } report.failed += 1; eprintln!("fail {target_key}: checksum verification failed"); continue; @@ -567,7 +673,11 @@ pub async fn migrate_attachments( &target_key, Some(&source_checksum), ) { - let _ = destination.delete(&target_key).await; + if let Err(cleanup_error) = + enqueue_unreferenced_delete(&conn, destination.as_ref(), &target_key).await + { + tracing::error!(error = %cleanup_error, storage_key = %target_key, "Failed to queue storage migration cleanup"); + } return Err(error); } report.migrated += 1;