From 9f50147e96d7fe70176416e81cd861a64d8ed2f5 Mon Sep 17 00:00:00 2001 From: Nic Date: Wed, 5 Aug 2026 14:56:36 +0100 Subject: [PATCH 1/4] feat: private-content caching opt-in with fleet-wide delete purge MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit download_cache_private (default off) admits private uploads to the download cache — read-through and seeding, keyed on the serve path's DataMap derivation. Deletes now propagate to every instance: the delete appends the upload's cache keys to cache_purge_log before any row is removed, each instance's 1-minute sweep tick consumes the tail and never advances past a failed unlink, and boot reconciliation validates the whole cache directory against live rows (uploads.cache_key, stamped at terminal statuses and backfilled at writer boot) for instances that were down past log retention. Purge window: synchronous on the handling instance, ~one tick everywhere else. Co-Authored-By: Claude Fable 5 --- USER-GUIDE.md | 3 +- cmd/indelible/main.go | 9 + docs/guides/download-cache.md | 5 +- docs/guides/scaling.md | 1 + .../postgres/013_download_cache_purge.sql | 33 +++ .../sqlite/013_download_cache_purge.sql | 33 +++ internal/handlers/uploads.go | 30 +-- .../handlers/uploads_delete_purge_test.go | 42 ++++ internal/services/settings_cache.go | 11 + internal/services/settings_validation.go | 1 + internal/services/upload.go | 197 +++++++++++++++++- internal/services/upload_purge_test.go | 147 +++++++++++++ internal/worker/cache_purge_test.go | 174 ++++++++++++++++ internal/worker/cache_sweep.go | 121 +++++++++++ internal/worker/upload.go | 10 + web/src/views/admin/SettingsView.vue | 20 +- 16 files changed, 803 insertions(+), 34 deletions(-) create mode 100644 internal/database/migrations/postgres/013_download_cache_purge.sql create mode 100644 internal/database/migrations/sqlite/013_download_cache_purge.sql create mode 100644 internal/services/upload_purge_test.go create mode 100644 internal/worker/cache_purge_test.go diff --git a/USER-GUIDE.md b/USER-GUIDE.md index 0022445..108e07f 100644 --- a/USER-GUIDE.md +++ b/USER-GUIDE.md @@ -557,7 +557,8 @@ Runtime settings are stored in the database and take effect immediately without | `download_cache_max_object_bytes` | `67108864` (64 MiB) | Per-object ceiling for the download cache: larger files are never cached and always stream through the temp path. Bounds: **1–2^40** | | `download_cache_min_uses` | `1` | How many times an object must be requested (per instance) before its bytes are cached. `1` caches on first download; raising it keeps one-hit-wonders from displacing hot content. Bounds: **1–100** | | `download_cache_inactive_secs` | `0` (off) | Inactivity expiry for cached objects: entries not **accessed** within this window are deleted regardless of the size budget. Independent pruning axis — also bounds how long cached plaintext can linger on an instance's disk. Example: `604800` = 7 days. Bounds: **0–315360000** | -| `download_cache_seed_on_upload` | `true` | Seed the download cache from a **public** upload's staged bytes the moment its network store succeeds, so publish-then-read is served from local disk with no network fetch (a rename — zero extra I/O). Applies where uploads are processed — the writer (or the whole instance in an all-in-one deployment); on a reader fleet each reader still warms by read-through. Seeds respect the size ceiling and byte budget but deliberately skip `download_cache_min_uses`; the sweeper evicts wrong bets. Disable for archive-shaped workloads (upload-heavy, rarely re-read) | +| `download_cache_seed_on_upload` | `true` | Seed the download cache from an upload's staged bytes (public; private too when `download_cache_private` is on) the moment its network store succeeds, so publish-then-read is served from local disk with no network fetch (a rename — zero extra I/O). Applies where uploads are processed — the writer (or the whole instance in an all-in-one deployment); on a reader fleet each reader still warms by read-through. Seeds respect the size ceiling and byte budget but deliberately skip `download_cache_min_uses`; the sweeper evicts wrong bets. Disable for archive-shaped workloads (upload-heavy, rarely re-read) | +| `download_cache_private` | `false` | Allow **private** uploads into the download cache (read-through and seeding). Off by default because it keeps decrypted private content on every caching instance's disk — a new at-rest surface. With it on, deleting an upload purges the handling instance synchronously and every other instance within about one sweep tick (~60s) via the purge log; see the [download-cache deployment guide](docs/guides/download-cache.md) for the exact contract | For the `download_cache_*` family, the [download-cache deployment guide](docs/guides/download-cache.md) covers sizing, the stats telemetry, per-setting fleet scope, and the privacy posture. diff --git a/cmd/indelible/main.go b/cmd/indelible/main.go index 69e039a..d173ed0 100644 --- a/cmd/indelible/main.go +++ b/cmd/indelible/main.go @@ -188,6 +188,15 @@ func main() { // upload-queue dequeue, hash-chain writes). Defers here are function-scoped, // so they still fire on shutdown. if cfg.WorkersEnabled { + // One-time Go-side backfill of uploads.cache_key (V2-873 migration + // 013): the key is a Go-side digest, so SQL couldn't fill it. Writer + // singleton, idempotent, cheap once caught up. + if n, err := services.NewUploadService(db).BackfillCacheKeys(); err != nil { + slog.Warn("cache_key backfill incomplete; boot reconciliation may re-warm some entries", "stamped", n, "error", err) + } else if n > 0 { + slog.Info("backfilled uploads.cache_key", "rows", n) + } + uploadWorker := worker.NewUploadWorker(db, cfg, dlCache) uploadWorker.Start() defer uploadWorker.Stop() diff --git a/docs/guides/download-cache.md b/docs/guides/download-cache.md index 1b2539a..339914e 100644 --- a/docs/guides/download-cache.md +++ b/docs/guides/download-cache.md @@ -67,7 +67,8 @@ Every instance owns its cache — there is no shared cache tier, by design (shar Cached entries are **plaintext bytes on the instance's disk**, held outside the audit surface of the upload store. That is why: -- **Only public uploads are cached.** Private content would put plaintext where a disk snapshot, misconfigured backup, or shared volume could expose it, and a deleted-and-shredded upload must not survive in a cache copy. (Whether a private-content opt-in should ever exist — with the fleet purge propagation it would need — is tracked separately as V2-873.) +- **Public uploads are cached unconditionally; private uploads only behind `download_cache_private`** (default off). Private caching puts decrypted plaintext where a disk snapshot, misconfigured backup, or shared volume could expose it — enabling it is an explicit operator decision, and cache volumes then deserve the same care as the database and temp-upload disks. Authorization is unaffected: the cache is only consulted after the row/owner/visibility checks, so a cache hit never bypasses the token gate. - Cache files are owner-only (`0600`) under `/cache/objects`, named by content digest — the name never reveals the DataMap (which is the retrieval capability), and the digest is domain-separated from the plaintext hash. - `download_cache_inactive_secs` doubles as the bound on how long unread cached plaintext can linger; budget `0` (or the env override `0`) drains an instance completely. -- **Deleting an upload purges its cached copy synchronously on the instance that handles the delete** (V2-824): the purge runs before the record is removed, and a purge that cannot unlink fails the delete rather than reporting a deletion while the plaintext remains readable. A download already in flight when the delete lands finishes streaming (its descriptor outlives the unlink) and any promotion it makes is taken back out. That take-back is best-effort: in the doubly-degraded case (a promotion raced the delete AND the final unlink failed) the leftover bytes are not servable — the deleted record 404s before the cache is consulted — and fall to eviction/inactivity; durable purge retry arrives with the fleet purge log (V2-873). On a reader fleet, other instances' cached copies of a deleted *public* upload are unreachable immediately (the record is gone) and age out by eviction/inactivity — acceptable for public bytes, and exactly why private content is not cached. +- **Deleting an upload purges its cached copy synchronously on the instance that handles the delete** (V2-824): the purge runs before the record is removed, and a purge that cannot unlink fails the delete rather than reporting a deletion while the plaintext remains readable. A download already in flight when the delete lands finishes streaming (its descriptor outlives the unlink) and any promotion it makes is taken back out. That take-back is best-effort in the instant, but the purge log below retries it durably. +- **Deletes propagate to the whole fleet through the purge log.** The delete appends the upload's cache keys to `cache_purge_log` before any row is removed; every instance's sweep worker consumes the log tail each tick and unlinks local copies, never advancing past a key whose unlink failed (it retries next tick). At boot, each instance reconciles its entire cache directory against live upload rows (`uploads.cache_key`), which covers deletes that happened while it was down — even beyond the log's retention window (7 days, pruned by the writer). **The purge-window contract:** the instance handling the delete purges synchronously (the API fails rather than report a deletion with the plaintext still readable); every other instance purges within about one sweep tick (~60s). If database read-replicas are ever introduced, replica lag adds to that window. On a reader fleet, other instances' cached copies of a deleted *public* upload are unreachable immediately (the record is gone) and age out by eviction/inactivity — acceptable for public bytes, and exactly why private content is not cached. diff --git a/docs/guides/scaling.md b/docs/guides/scaling.md index 98ed4ac..b82ab80 100644 --- a/docs/guides/scaling.md +++ b/docs/guides/scaling.md @@ -103,6 +103,7 @@ Fleet notes: - The cache is **per instance** — every reader warms and evicts its own copy from its own traffic. With no load-balancer affinity, popular objects end up replicated on every reader; at typical working-set sizes (small hot objects) that is the right trade, and it is what sizing the budget per instance assumes. - The `download_cache_*` settings are **fleet-global values applied per instance** (readers share the writer's database). For a fleet with **different disk sizes per reader**, set the env override `INDELIBLE_DOWNLOAD_CACHE_MAX_BYTES` on the instances that differ — env beats the DB setting on that instance only, and `0` disables its cache outright. - The cache can never be the reason uploads pause: a background sweeper evicts it — aggressively, toward empty if necessary — as soon as the data volume approaches the disk-alert worker's critical threshold, and otherwise keeps it under budget by LRU with an optional inactivity window (`download_cache_inactive_secs`). +- **Deletes reach every reader's cache within about a minute.** The delete appends purge keys to a shared `cache_purge_log`; each instance's sweep worker applies them on its 1-minute tick, and boot reconciliation covers instances that were down. Relevant mainly with `download_cache_private` on — a deleted *public* upload's cached bytes are unreachable immediately either way (the record is gone). - **Upload seeding is writer-local.** `download_cache_seed_on_upload` (default on) promotes a public upload's staged bytes into the cache when the store succeeds — but uploads only pass through the writer, so on a role-split fleet this warms the **writer's** cache, not the readers'. Readers warm by read-through: the first download per reader is still a cold fetch. The full publish-then-read win applies to all-in-one deployments, where writer and reader are the same process. ## Load balancer notes diff --git a/internal/database/migrations/postgres/013_download_cache_purge.sql b/internal/database/migrations/postgres/013_download_cache_purge.sql new file mode 100644 index 0000000..2bf6da2 --- /dev/null +++ b/internal/database/migrations/postgres/013_download_cache_purge.sql @@ -0,0 +1,33 @@ +-- +goose Up + +-- V2-873: fleet-wide download-cache purge propagation. +-- +-- uploads.cache_key is the serve-path cache key (KeyForIdentifier over +-- data_map when present, else datamap_address — the same preference order +-- downloadETag uses), set when an upload reaches a terminal stored status and +-- backfilled at writer boot for pre-existing rows. Cache keys are one-way +-- digests, so without this column no instance can map its cached files back +-- to upload rows (needed by boot reconciliation). +ALTER TABLE uploads ADD COLUMN cache_key TEXT; +CREATE INDEX idx_uploads_cache_key ON uploads(cache_key); + +-- cache_purge_log is the delete fan-out: DeleteUpload's service half appends +-- the deleted upload's cache keys here (insert-first, so a failed delete can +-- at worst cause a spurious purge that re-warms), and every instance's cache +-- sweep worker consumes the tail each tick, dropping local entries. Rows are +-- pruned after a retention window by writer-role instances only (reader +-- discipline, V2-514); instances offline longer than the retention are +-- covered by boot reconciliation against uploads.cache_key. +CREATE TABLE cache_purge_log ( + id BIGSERIAL PRIMARY KEY, + cache_key TEXT NOT NULL, + deleted_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX idx_cache_purge_log_deleted_at ON cache_purge_log(deleted_at); + +-- +goose Down + +DROP TABLE cache_purge_log; +DROP INDEX idx_uploads_cache_key; +ALTER TABLE uploads DROP COLUMN cache_key; diff --git a/internal/database/migrations/sqlite/013_download_cache_purge.sql b/internal/database/migrations/sqlite/013_download_cache_purge.sql new file mode 100644 index 0000000..9ef64e2 --- /dev/null +++ b/internal/database/migrations/sqlite/013_download_cache_purge.sql @@ -0,0 +1,33 @@ +-- +goose Up + +-- V2-873: fleet-wide download-cache purge propagation. +-- +-- uploads.cache_key is the serve-path cache key (KeyForIdentifier over +-- data_map when present, else datamap_address — the same preference order +-- downloadETag uses), set when an upload reaches a terminal stored status and +-- backfilled at writer boot for pre-existing rows. Cache keys are one-way +-- digests, so without this column no instance can map its cached files back +-- to upload rows (needed by boot reconciliation). +ALTER TABLE uploads ADD COLUMN cache_key TEXT; +CREATE INDEX idx_uploads_cache_key ON uploads(cache_key); + +-- cache_purge_log is the delete fan-out: DeleteUpload's service half appends +-- the deleted upload's cache keys here (insert-first, so a failed delete can +-- at worst cause a spurious purge that re-warms), and every instance's cache +-- sweep worker consumes the tail each tick, dropping local entries. Rows are +-- pruned after a retention window by writer-role instances only (reader +-- discipline, V2-514); instances offline longer than the retention are +-- covered by boot reconciliation against uploads.cache_key. +CREATE TABLE cache_purge_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + cache_key TEXT NOT NULL, + deleted_at DATETIME NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX idx_cache_purge_log_deleted_at ON cache_purge_log(deleted_at); + +-- +goose Down + +DROP TABLE cache_purge_log; +DROP INDEX idx_uploads_cache_key; +ALTER TABLE uploads DROP COLUMN cache_key; diff --git a/internal/handlers/uploads.go b/internal/handlers/uploads.go index e003f29..2c59f78 100644 --- a/internal/handlers/uploads.go +++ b/internal/handlers/uploads.go @@ -2,7 +2,6 @@ package handlers import ( "context" - "database/sql" "encoding/json" "errors" "fmt" @@ -818,9 +817,16 @@ func DownloadUpload(db *database.DB, cfg *config.Config, cache *downloadcache.St queueWait := time.Duration(settingsSvc.GetIntWithBounds( "download_queue_wait_secs", 30, 0, 600, )) * time.Second + // Private content is cacheable only behind the explicit opt-in + // (V2-873): it puts decrypted plaintext on this instance's disk, so + // the operator owns that call. Public stays cacheable uncondition- + // ally. Auth is unaffected either way — the cache is consulted only + // after the row/owner/visibility checks above. var cacheKey string - if cacheBudget > 0 && etag != "" && upload.Visibility == "public" { - cacheKey = strings.Trim(etag, `"`) + if cacheBudget > 0 && etag != "" { + if upload.Visibility == "public" || settingsSvc.GetBool("download_cache_private", false) { + cacheKey = strings.Trim(etag, `"`) + } } // The coalesce loop clears cacheKey on wait-timeout; remember the // original eligibility so the hit/miss accounting (V2-825) still @@ -1237,7 +1243,7 @@ func DeleteUpload(db *database.DB, cache *downloadcache.Store) http.HandlerFunc // while a cached copy of the content remains locally readable. For a // private upload the row delete destroys the DataMap, so this // ordering is what makes the shred honest. - keys := cachePurgeKeys(upload) + keys := upload.CacheKeys() if cache != nil { for _, k := range keys { if err := cache.Drop(k); err != nil { @@ -1280,22 +1286,6 @@ func DeleteUpload(db *database.DB, cache *downloadcache.Store) http.HandlerFunc } } -// cachePurgeKeys returns every cache key an upload's bytes could live under: -// the serve path keys on the local DataMap when one exists while seeding keys -// on the network address, so a purge must cover both derivations. -func cachePurgeKeys(u *services.Upload) []string { - var keys []string - for _, id := range []sql.NullString{u.DataMap, u.DatamapAddress} { - if !id.Valid || id.String == "" { - continue - } - k := downloadcache.KeyForIdentifier(id.String) - if len(keys) == 0 || keys[0] != k { - keys = append(keys, k) - } - } - return keys -} // effectiveAllowlist resolves the content-type allowlist for an upload using // the override chain: token > user > system setting > built-in default. diff --git a/internal/handlers/uploads_delete_purge_test.go b/internal/handlers/uploads_delete_purge_test.go index 1d4cf5a..82aaa64 100644 --- a/internal/handlers/uploads_delete_purge_test.go +++ b/internal/handlers/uploads_delete_purge_test.go @@ -212,3 +212,45 @@ func TestDeleteResidual_OrphanedCacheBytesNotServable(t *testing.T) { t.Fatalf("orphaned cache bytes were served: %d", w.Code) } } + +// V2-873: private uploads enter the cache only behind download_cache_private, +// and their deletion purges the plaintext like any other entry. + +func TestDownloadUpload_PrivateCachedWhenOptedIn(t *testing.T) { + const content = "private but cached by choice" + fake := newCacheFakeAntd(t, content) + router, token, cfg, db, _ := newCacheTestEnvWithStore(t, fake.srv.URL, map[string]string{ + "download_cache_max_bytes": "1048576", + "download_cache_private": "true", + }) + uuid := makeUpload(t, router, db, token, "opted-in.txt", "private", "dm-opted-in") + + warmCache(t, router, token, uuid, cfg.DataDir, content) + if w := doDownload(router, token, uuid, ""); w.Code != http.StatusOK || w.Body.String() != content { + t.Fatalf("repeat download: %d %q", w.Code, w.Body.String()) + } + if got := fake.privateHits.Load(); got != 1 { + t.Fatalf("antd private fetches = %d, want 1 (repeat must be a cache hit)", got) + } +} + +func TestDeleteUpload_PurgesPrivateCachedCopy(t *testing.T) { + const content = "private plaintext must die with the DataMap" + fake := newCacheFakeAntd(t, content) + router, token, cfg, db, store := newCacheTestEnvWithStore(t, fake.srv.URL, map[string]string{ + "download_cache_max_bytes": "1048576", + "download_cache_private": "true", + }) + uuid := makeUpload(t, router, db, token, "shred-me.txt", "private", "dm-shred-me") + cached := warmCache(t, router, token, uuid, cfg.DataDir, content) + + if w := doDelete(router, token, uuid); w.Code != http.StatusOK { + t.Fatalf("delete: %d %s", w.Code, w.Body.String()) + } + if _, err := os.Stat(cached); !os.IsNotExist(err) { + t.Fatalf("private plaintext outlived the delete: %v", err) + } + if count, _ := store.Stats(); count != 0 { + t.Fatalf("cache still indexes %d entries after private delete", count) + } +} diff --git a/internal/services/settings_cache.go b/internal/services/settings_cache.go index b0e064f..9971af3 100644 --- a/internal/services/settings_cache.go +++ b/internal/services/settings_cache.go @@ -85,3 +85,14 @@ func (c *CachedSettingsService) GetIntWithBounds(key string, fallback, min, max } return n } + +// GetBool returns the setting as a bool ("true"/"false"), or fallback when +// missing or empty. Validation constrains stored values to that pair, so any +// other value is treated as fallback. +func (c *CachedSettingsService) GetBool(key string, fallback bool) bool { + v, err := c.Get(key) + if err != nil || v == "" { + return fallback + } + return v == "true" +} diff --git a/internal/services/settings_validation.go b/internal/services/settings_validation.go index 8381c7b..c999585 100644 --- a/internal/services/settings_validation.go +++ b/internal/services/settings_validation.go @@ -36,6 +36,7 @@ var typedValidators = map[string]func(string) error{ "download_cache_min_uses": optionalIntInRange(1, 100), "download_cache_inactive_secs": optionalIntInRange(0, 315360000), // 0 = off; max 10 years "download_cache_seed_on_upload": oneOf("true", "false"), + "download_cache_private": oneOf("true", "false"), } // oneOf builds a validator that requires the value to be in the allowed set. diff --git a/internal/services/upload.go b/internal/services/upload.go index f49e83d..8b135fc 100644 --- a/internal/services/upload.go +++ b/internal/services/upload.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/WithAutonomi/indelible/internal/database" + "github.com/WithAutonomi/indelible/internal/downloadcache" ) var ( @@ -340,8 +341,8 @@ func (s *UploadService) DequeueNext() (*Upload, error) { // The dataMap is the hex-encoded serialized DataMap returned by antd's finalize endpoint. func (s *UploadService) MarkCompleted(id int64, dataMap, actualCost string) error { _, err := s.db.Exec( - `UPDATE uploads SET status = 'completed', data_map = ?, actual_cost = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, - dataMap, actualCost, id, + `UPDATE uploads SET status = 'completed', data_map = ?, actual_cost = ?, cache_key = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, + dataMap, actualCost, downloadcache.KeyForIdentifier(dataMap), id, ) return err } @@ -353,8 +354,8 @@ func (s *UploadService) MarkCompleted(id int64, dataMap, actualCost string) erro // payment batch — no separate daemon-wallet payment. func (s *UploadService) MarkCompletedPublic(id int64, datamapAddress, actualCost string) error { _, err := s.db.Exec( - `UPDATE uploads SET status = 'completed', datamap_address = ?, actual_cost = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, - datamapAddress, actualCost, id, + `UPDATE uploads SET status = 'completed', datamap_address = ?, actual_cost = ?, cache_key = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, + datamapAddress, actualCost, downloadcache.KeyForIdentifier(datamapAddress), id, ) return err } @@ -366,8 +367,8 @@ func (s *UploadService) MarkCompletedPublic(id int64, datamapAddress, actualCost // fresh store (V2-399). func (s *UploadService) MarkAlreadyStored(id int64, dataMap, actualCost string) error { _, err := s.db.Exec( - `UPDATE uploads SET status = 'already_stored', data_map = ?, actual_cost = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, - dataMap, actualCost, id, + `UPDATE uploads SET status = 'already_stored', data_map = ?, actual_cost = ?, cache_key = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, + dataMap, actualCost, downloadcache.KeyForIdentifier(dataMap), id, ) return err } @@ -376,8 +377,8 @@ func (s *UploadService) MarkAlreadyStored(id int64, dataMap, actualCost string) // counterpart (V2-399). func (s *UploadService) MarkAlreadyStoredPublic(id int64, datamapAddress, actualCost string) error { _, err := s.db.Exec( - `UPDATE uploads SET status = 'already_stored', datamap_address = ?, actual_cost = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, - datamapAddress, actualCost, id, + `UPDATE uploads SET status = 'already_stored', datamap_address = ?, actual_cost = ?, cache_key = ?, completed_at = CURRENT_TIMESTAMP, temp_path = NULL WHERE id = ?`, + datamapAddress, actualCost, downloadcache.KeyForIdentifier(datamapAddress), id, ) return err } @@ -569,6 +570,22 @@ func (s *UploadService) ForceRetry(id int64) error { // Delete permanently removes an upload record. Only allowed for failed or completed uploads. func (s *UploadService) Delete(id int64) error { + // Fan the purge out to the fleet FIRST (V2-873): append the upload's + // cache keys to cache_purge_log before anything is deleted, so a failure + // later in this sequence can at worst cause a spurious purge (an instance + // drops a live entry and re-warms) — never a missed one. Every instance's + // cache sweep worker consumes this log each tick; the handling instance + // additionally purges synchronously in the handler (V2-824). + if u, err := s.GetByID(id); err == nil { + for _, key := range u.CacheKeys() { + if _, err := s.db.Exec(`INSERT INTO cache_purge_log (cache_key) VALUES (?)`, key); err != nil { + return err + } + } + } else if !errors.Is(err, ErrUploadNotFound) { + return err + } + // Clean up related data first if _, err := s.db.Exec(`DELETE FROM file_tags WHERE upload_id = ?`, id); err != nil { return err @@ -591,6 +608,170 @@ func (s *UploadService) Delete(id int64) error { return nil } +// CacheKeys returns every download-cache key this upload's bytes could live +// under, serve-path derivation first: the serve path (downloadETag) prefers +// the local DataMap when one exists, while public upload seeding keys on the +// network address — a purge must cover both. +func (u *Upload) CacheKeys() []string { + var keys []string + for _, id := range []sql.NullString{u.DataMap, u.DatamapAddress} { + if !id.Valid || id.String == "" { + continue + } + k := downloadcache.KeyForIdentifier(id.String) + if len(keys) == 0 || keys[0] != k { + keys = append(keys, k) + } + } + return keys +} + +// PurgeLogEntry is one consumed row of cache_purge_log. +type PurgeLogEntry struct { + ID int64 + CacheKey string +} + +// PurgeLogSince returns up to limit purge-log rows with id > after, oldest +// first — the sweep worker's per-tick tail read. +func (s *UploadService) PurgeLogSince(after int64, limit int) ([]PurgeLogEntry, error) { + rows, err := s.db.Query( + `SELECT id, cache_key FROM cache_purge_log WHERE id > ? ORDER BY id LIMIT ?`, + after, limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var out []PurgeLogEntry + for rows.Next() { + var e PurgeLogEntry + if err := rows.Scan(&e.ID, &e.CacheKey); err != nil { + return nil, err + } + out = append(out, e) + } + return out, rows.Err() +} + +// MaxPurgeLogID returns the current high-water mark of cache_purge_log (0 on +// an empty log). Boot uses it to skip history: everything at or before it is +// covered by the boot reconciliation pass. +func (s *UploadService) MaxPurgeLogID() (int64, error) { + var id sql.NullInt64 + if err := s.db.QueryRow(`SELECT MAX(id) FROM cache_purge_log`).Scan(&id); err != nil { + return 0, err + } + return id.Int64, nil +} + +// PrunePurgeLog deletes purge-log rows older than before. Writer-role only +// (reader discipline, V2-514); instances offline past the retention window +// are covered by boot reconciliation. +func (s *UploadService) PrunePurgeLog(before time.Time) (int64, error) { + res, err := s.db.Exec( + `DELETE FROM cache_purge_log WHERE deleted_at < ?`, + before.UTC().Format("2006-01-02 15:04:05"), + ) + if err != nil { + return 0, err + } + n, _ := res.RowsAffected() + return n, nil +} + +// LiveCacheKeys reports which of the given cache keys belong to a live upload +// row — the boot-reconciliation lookup (uploads.cache_key is indexed). +func (s *UploadService) LiveCacheKeys(keys []string) (map[string]bool, error) { + live := make(map[string]bool, len(keys)) + const chunk = 500 + for start := 0; start < len(keys); start += chunk { + end := start + chunk + if end > len(keys) { + end = len(keys) + } + batch := keys[start:end] + placeholders := make([]byte, 0, len(batch)*2) + args := make([]any, 0, len(batch)) + for i, k := range batch { + if i > 0 { + placeholders = append(placeholders, ',') + } + placeholders = append(placeholders, '?') + args = append(args, k) + } + rows, err := s.db.Query( + `SELECT cache_key FROM uploads WHERE cache_key IN (`+string(placeholders)+`)`, args..., + ) + if err != nil { + return nil, err + } + for rows.Next() { + var k string + if err := rows.Scan(&k); err != nil { + rows.Close() + return nil, err + } + live[k] = true + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, err + } + rows.Close() + } + return live, nil +} + +// BackfillCacheKeys computes uploads.cache_key for rows stored before the +// column existed (V2-873 migration 013). Writer-boot singleton — the digest +// is Go-side (KeyForIdentifier), so SQL backfill in the migration was not +// possible. Idempotent; returns how many rows were stamped. +func (s *UploadService) BackfillCacheKeys() (int64, error) { + rows, err := s.db.Query( + `SELECT id, data_map, datamap_address FROM uploads + WHERE cache_key IS NULL + AND ((data_map IS NOT NULL AND data_map != '') OR (datamap_address IS NOT NULL AND datamap_address != ''))`, + ) + if err != nil { + return 0, err + } + type pending struct { + id int64 + key string + } + var todo []pending + for rows.Next() { + var id int64 + var dataMap, addr sql.NullString + if err := rows.Scan(&id, &dataMap, &addr); err != nil { + rows.Close() + return 0, err + } + identifier := dataMap.String + if identifier == "" { + identifier = addr.String + } + if identifier != "" { + todo = append(todo, pending{id: id, key: downloadcache.KeyForIdentifier(identifier)}) + } + } + if err := rows.Err(); err != nil { + rows.Close() + return 0, err + } + rows.Close() + + var n int64 + for _, p := range todo { + if _, err := s.db.Exec(`UPDATE uploads SET cache_key = ? WHERE id = ?`, p.key, p.id); err != nil { + return n, err + } + n++ + } + return n, nil +} + // ListActiveTempPaths returns all temp_path values for uploads still in queued or processing state. func (s *UploadService) ListActiveTempPaths() ([]string, error) { // Keep temp files for in-flight uploads AND for recoverable failures (a diff --git a/internal/services/upload_purge_test.go b/internal/services/upload_purge_test.go new file mode 100644 index 0000000..dc41a28 --- /dev/null +++ b/internal/services/upload_purge_test.go @@ -0,0 +1,147 @@ +package services + +import ( + "testing" + "time" + + "github.com/WithAutonomi/indelible/internal/downloadcache" +) + +// V2-873: cache_key stamping, purge-log fan-out, and the legacy backfill. + +func cacheKeyOf(t *testing.T, svc *UploadService, id int64) string { + t.Helper() + var key *string + if err := svc.db.QueryRow(`SELECT cache_key FROM uploads WHERE id = ?`, id).Scan(&key); err != nil { + t.Fatalf("read cache_key: %v", err) + } + if key == nil { + return "" + } + return *key +} + +func TestMarkTerminalStatusesStampCacheKey(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "stamp@example.com", "S", "T") + svc := NewUploadService(db) + + private := createTestUpload(t, svc, user.ID, "private.bin", 10) + if err := svc.MarkCompleted(private.ID, "dm-hex-private", "0"); err != nil { + t.Fatalf("MarkCompleted: %v", err) + } + if got, want := cacheKeyOf(t, svc, private.ID), downloadcache.KeyForIdentifier("dm-hex-private"); got != want { + t.Fatalf("private cache_key = %q, want %q", got, want) + } + + public := createTestUpload(t, svc, user.ID, "public.bin", 10) + if err := svc.MarkCompletedPublic(public.ID, "addr-public", "0"); err != nil { + t.Fatalf("MarkCompletedPublic: %v", err) + } + if got, want := cacheKeyOf(t, svc, public.ID), downloadcache.KeyForIdentifier("addr-public"); got != want { + t.Fatalf("public cache_key = %q, want %q", got, want) + } +} + +func TestDeleteAppendsPurgeLogBothDerivations(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "purgelog@example.com", "P", "L") + svc := NewUploadService(db) + + // A published-formerly-private row carries BOTH identifiers — the purge + // fan-out must log both derivations. + u := createTestUpload(t, svc, user.ID, "both.bin", 10) + if err := svc.MarkCompleted(u.ID, "dm-both", "0"); err != nil { + t.Fatalf("MarkCompleted: %v", err) + } + if err := svc.MarkPublished(u.ID, "addr-both"); err != nil { + t.Fatalf("MarkPublished: %v", err) + } + + if err := svc.Delete(u.ID); err != nil { + t.Fatalf("Delete: %v", err) + } + + entries, err := svc.PurgeLogSince(0, 10) + if err != nil { + t.Fatalf("PurgeLogSince: %v", err) + } + want := map[string]bool{ + downloadcache.KeyForIdentifier("dm-both"): true, + downloadcache.KeyForIdentifier("addr-both"): true, + } + if len(entries) != 2 || !want[entries[0].CacheKey] || !want[entries[1].CacheKey] || entries[0].CacheKey == entries[1].CacheKey { + t.Fatalf("purge log = %+v, want both derivations exactly once", entries) + } + + if max, err := svc.MaxPurgeLogID(); err != nil || max != entries[1].ID { + t.Fatalf("MaxPurgeLogID = %d (err=%v), want %d", max, err, entries[1].ID) + } +} + +func TestBackfillCacheKeysStampsLegacyRows(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "backfill@example.com", "B", "F") + svc := NewUploadService(db) + + // Simulate pre-013 rows: terminal status + identifiers but NULL cache_key. + private := createTestUpload(t, svc, user.ID, "legacy-private.bin", 10) + public := createTestUpload(t, svc, user.ID, "legacy-public.bin", 10) + if _, err := db.Exec(`UPDATE uploads SET status='completed', data_map='legacy-dm', cache_key=NULL WHERE id = ?`, private.ID); err != nil { + t.Fatalf("legacy private: %v", err) + } + if _, err := db.Exec(`UPDATE uploads SET status='completed', visibility='public', datamap_address='legacy-addr', cache_key=NULL WHERE id = ?`, public.ID); err != nil { + t.Fatalf("legacy public: %v", err) + } + + n, err := svc.BackfillCacheKeys() + if err != nil || n != 2 { + t.Fatalf("BackfillCacheKeys = %d, %v; want 2, nil", n, err) + } + if got, want := cacheKeyOf(t, svc, private.ID), downloadcache.KeyForIdentifier("legacy-dm"); got != want { + t.Fatalf("backfilled private key = %q, want %q", got, want) + } + if got, want := cacheKeyOf(t, svc, public.ID), downloadcache.KeyForIdentifier("legacy-addr"); got != want { + t.Fatalf("backfilled public key = %q, want %q", got, want) + } + + // Idempotent: nothing left to stamp. + if n, err := svc.BackfillCacheKeys(); err != nil || n != 0 { + t.Fatalf("second backfill = %d, %v; want 0, nil", n, err) + } +} + +func TestPrunePurgeLogAndLiveCacheKeys(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "prune@example.com", "P", "R") + svc := NewUploadService(db) + + u := createTestUpload(t, svc, user.ID, "live.bin", 10) + if err := svc.MarkCompleted(u.ID, "dm-live", "0"); err != nil { + t.Fatalf("MarkCompleted: %v", err) + } + liveKey := downloadcache.KeyForIdentifier("dm-live") + + live, err := svc.LiveCacheKeys([]string{liveKey, "0000000000000000000000000000000000000000000000000000000000000000"}) + if err != nil { + t.Fatalf("LiveCacheKeys: %v", err) + } + if !live[liveKey] || len(live) != 1 { + t.Fatalf("LiveCacheKeys = %v, want only the live key", live) + } + + if _, err := db.Exec(`INSERT INTO cache_purge_log (cache_key, deleted_at) VALUES (?, ?)`, "old-key", "2020-01-01 00:00:00"); err != nil { + t.Fatalf("insert old log row: %v", err) + } + if _, err := db.Exec(`INSERT INTO cache_purge_log (cache_key) VALUES (?)`, "fresh-key"); err != nil { + t.Fatalf("insert fresh log row: %v", err) + } + n, err := svc.PrunePurgeLog(time.Now().Add(-24 * time.Hour)) + if err != nil || n != 1 { + t.Fatalf("PrunePurgeLog = %d, %v; want 1, nil", n, err) + } + entries, err := svc.PurgeLogSince(0, 10) + if err != nil || len(entries) != 1 || entries[0].CacheKey != "fresh-key" { + t.Fatalf("post-prune log = %+v (err=%v), want only fresh-key", entries, err) + } +} diff --git a/internal/worker/cache_purge_test.go b/internal/worker/cache_purge_test.go new file mode 100644 index 0000000..3c230f4 --- /dev/null +++ b/internal/worker/cache_purge_test.go @@ -0,0 +1,174 @@ +package worker + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/WithAutonomi/indelible/internal/config" + "github.com/WithAutonomi/indelible/internal/database" + "github.com/WithAutonomi/indelible/internal/dbtest" + "github.com/WithAutonomi/indelible/internal/downloadcache" + "github.com/WithAutonomi/indelible/internal/services" +) + +// V2-873 fleet purge propagation: the sweep worker of a *remote* instance +// (any instance other than the one that handled the delete) must apply +// deletes via the purge log within one tick, and reconcile its whole cache +// against live rows at boot. + +// newPurgeEnv builds a db, a ready store standing in for a remote reader's +// cache, and a sweep worker over it (workers enabled so prune paths run). +func newPurgeEnv(t *testing.T) (*database.DB, *downloadcache.Store, *CacheSweepWorker, *services.UploadService) { + t.Helper() + db := dbtest.OpenDB(t) + dir := t.TempDir() + store := downloadcache.New(filepath.Join(dir, "objects")) + if err := store.Scan(context.Background()); err != nil { + t.Fatalf("scan: %v", err) + } + cfg := &config.Config{DataDir: dir, WorkersEnabled: true} + w := NewCacheSweepWorker(db, cfg, store) + if _, err := db.Exec(`INSERT INTO users (id, email) VALUES (8001, 'purge-test@example.com')`); err != nil { + t.Fatalf("insert user: %v", err) + } + return db, store, w, services.NewUploadService(db) +} + +// mkStoredUpload inserts a completed upload row whose cache_key matches +// KeyForIdentifier(identifier), and returns its id and key. +func mkStoredUpload(t *testing.T, db *database.DB, id int64, uuid, identifier string) string { + t.Helper() + key := downloadcache.KeyForIdentifier(identifier) + if _, err := db.Exec(`INSERT INTO uploads (id, uuid, user_id, filename, original_filename, file_size, content_type, visibility, status, data_map, cache_key) + VALUES (?, ?, 8001, 'f.bin', 'f.bin', 4, 'text/plain', 'private', 'completed', ?, ?)`, + id, uuid, identifier, key); err != nil { + t.Fatalf("insert upload: %v", err) + } + return key +} + +// cacheEntry promotes content into the store under key and returns its path. +func cacheEntry(t *testing.T, store *downloadcache.Store, dir, key, content string) string { + t.Helper() + temp := filepath.Join(dir, "stage-"+key[:8]) + if err := os.WriteFile(temp, []byte(content), 0600); err != nil { + t.Fatalf("stage: %v", err) + } + if _, err := store.PromoteIfFits(key, temp, 1<<30); err != nil { + t.Fatalf("promote: %v", err) + } + p, ok := store.Get(key) + if !ok { + t.Fatal("promoted entry missed") + } + return p +} + +func TestPropagatePurges_RemoteInstanceAppliesDelete(t *testing.T) { + db, store, w, svc := newPurgeEnv(t) + dir := t.TempDir() + key := mkStoredUpload(t, db, 8101, "fleet-uuid-1", "dm-fleet-1") + p := cacheEntry(t, store, dir, key, "fleet bytes") + + // Boot tick: the row is live, so reconciliation keeps the entry. + w.propagatePurges(context.Background()) + if _, ok := store.Get(key); !ok { + t.Fatal("boot reconciliation purged a live entry") + } + + // The delete lands on some other instance: service-level delete appends + // to the purge log; this instance's next tick must apply it. + if err := svc.Delete(8101); err != nil { + t.Fatalf("delete: %v", err) + } + w.propagatePurges(context.Background()) + + if _, ok := store.Get(key); ok { + t.Fatal("remote entry survived the propagated purge") + } + if _, err := os.Stat(p); !os.IsNotExist(err) { + t.Fatal("remote plaintext still on disk after propagated purge") + } +} + +func TestBootReconcile_DropsOrphansKeepsLive(t *testing.T) { + db, store, w, _ := newPurgeEnv(t) + dir := t.TempDir() + liveKey := mkStoredUpload(t, db, 8102, "fleet-uuid-2", "dm-live-2") + cacheEntry(t, store, dir, liveKey, "live bytes") + orphanKey := downloadcache.KeyForIdentifier("dm-orphan-2") + orphanPath := cacheEntry(t, store, dir, orphanKey, "orphan bytes") + + w.propagatePurges(context.Background()) + + if _, ok := store.Get(liveKey); !ok { + t.Fatal("reconciliation dropped an entry with a live row") + } + if _, ok := store.Get(orphanKey); ok { + t.Fatal("reconciliation kept an orphan") + } + if _, err := os.Stat(orphanPath); !os.IsNotExist(err) { + t.Fatal("orphan bytes still on disk") + } +} + +func TestPropagatePurges_FailedUnlinkHaltsHighWater(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root ignores directory permissions") + } + db, store, w, svc := newPurgeEnv(t) + dir := t.TempDir() + key := mkStoredUpload(t, db, 8103, "fleet-uuid-3", "dm-stuck-3") + p := cacheEntry(t, store, dir, key, "stuck bytes") + + w.propagatePurges(context.Background()) // boot with the row live + + if err := svc.Delete(8103); err != nil { + t.Fatalf("delete: %v", err) + } + parent := filepath.Dir(p) + if err := os.Chmod(parent, 0500); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { _ = os.Chmod(parent, 0700) }) + + before := w.lastPurgeID + w.propagatePurges(context.Background()) + if w.lastPurgeID != before { + t.Fatal("high-water mark advanced past a failed unlink") + } + if _, ok := store.Get(key); !ok { + t.Fatal("entry must stay indexed while its bytes are stuck") + } + + // Obstruction cleared: the next tick retries and completes the purge. + if err := os.Chmod(parent, 0700); err != nil { + t.Fatalf("chmod restore: %v", err) + } + w.propagatePurges(context.Background()) + if _, ok := store.Get(key); ok { + t.Fatal("retry tick did not purge") + } + if w.lastPurgeID == before { + t.Fatal("high-water mark must advance after the retry succeeds") + } +} + +func TestPropagatePurges_WriterPrunesOldRows(t *testing.T) { + db, _, w, svc := newPurgeEnv(t) + + if _, err := db.Exec(`INSERT INTO cache_purge_log (cache_key, deleted_at) VALUES ('ancient', '2020-01-01 00:00:00')`); err != nil { + t.Fatalf("insert old row: %v", err) + } + w.propagatePurges(context.Background()) + + entries, err := svc.PurgeLogSince(0, 10) + if err != nil { + t.Fatalf("PurgeLogSince: %v", err) + } + if len(entries) != 0 { + t.Fatalf("old purge-log rows survived the writer prune: %+v", entries) + } +} diff --git a/internal/worker/cache_sweep.go b/internal/worker/cache_sweep.go index 4c6a40f..26bb58f 100644 --- a/internal/worker/cache_sweep.go +++ b/internal/worker/cache_sweep.go @@ -42,6 +42,17 @@ type CacheSweepWorker struct { cfg *config.Config settingsSvc *services.CachedSettingsService store *downloadcache.Store + uploadSvc *services.UploadService + + // Fleet purge propagation (V2-873): lastPurgeID is this instance's + // high-water mark in cache_purge_log — in-memory only, because boot runs + // a full reconciliation against uploads.cache_key that subsumes any log + // history. pruneLog is writer-role only (readers stay DB-write-free, + // V2-514). booted flips after the boot reconciliation succeeds. + lastPurgeID int64 + pruneLog bool + lastPrune time.Time + booted bool // usage reports volume capacity for the disk-pressure trigger; injected // so tests can simulate a filling disk. Defaults to diskusage.Usage. @@ -79,6 +90,15 @@ const ( cacheSweepBatch = 100 cacheSweepBatchPause = 50 * time.Millisecond + + // cachePurgeLogRetention is how long consumed-or-not purge-log rows are + // kept before the writer prunes them. It bounds the log's size, not + // correctness: an instance offline longer than this reconciles its whole + // cache against uploads.cache_key at boot anyway. + cachePurgeLogRetention = 7 * 24 * time.Hour + + // cachePurgeBatch is the per-read tail size when consuming the log. + cachePurgeBatch = 500 ) // NewCacheSweepWorker creates the sweeper over the same store the download @@ -88,6 +108,8 @@ func NewCacheSweepWorker(db *database.DB, cfg *config.Config, store *downloadcac cfg: cfg, settingsSvc: services.NewCachedSettingsService(services.NewSettingsService(db)), store: store, + uploadSvc: services.NewUploadService(db), + pruneLog: cfg.WorkersEnabled, usage: diskusage.Usage, interval: cacheSweepInterval, batch: cacheSweepBatch, @@ -135,6 +157,8 @@ func (w *CacheSweepWorker) Stop() { // stats emission, which runs even over an empty cache (counters may have // moved since the last tick drained it). func (w *CacheSweepWorker) sweep(ctx context.Context) { + w.propagatePurges(ctx) + if count, _ := w.store.Stats(); count > 0 { w.sweepDiskPressure(ctx) @@ -148,6 +172,103 @@ func (w *CacheSweepWorker) sweep(ctx context.Context) { w.emitStats() } +// propagatePurges applies fleet-wide delete purges to this instance's cache +// (V2-873). Boot: one full reconciliation of every cached key against live +// uploads.cache_key rows — which subsumes all purge-log history, so the log +// high-water mark starts at the log's current tail. Steady state: consume the +// log tail each tick, never advancing past a key whose unlink failed (it is +// retried next tick — a delete's disk-level guarantee on remote instances is +// this loop). Writer-role instances also prune the log hourly. +func (w *CacheSweepWorker) propagatePurges(ctx context.Context) { + if !w.booted { + if err := w.bootReconcile(ctx); err != nil { + slog.Warn("download cache boot reconciliation failed; retrying next tick", "error", err) + return // never consume the tail from an unreconciled baseline + } + w.booted = true + } + + for { + entries, err := w.uploadSvc.PurgeLogSince(w.lastPurgeID, cachePurgeBatch) + if err != nil { + slog.Warn("download cache purge-log read failed", "error", err) + return + } + if len(entries) == 0 { + break + } + for _, e := range entries { + if ctx.Err() != nil { + return + } + if err := w.store.Drop(e.CacheKey); err != nil { + slog.Warn("download cache propagated purge failed; will retry next tick", + "key", e.CacheKey, "error", err) + return + } + w.lastPurgeID = e.ID + } + if len(entries) < cachePurgeBatch { + break + } + } + + if w.pruneLog && time.Since(w.lastPrune) >= time.Hour { + w.lastPrune = time.Now() + if n, err := w.uploadSvc.PrunePurgeLog(time.Now().Add(-cachePurgeLogRetention)); err != nil { + slog.Warn("cache purge-log prune failed", "error", err) + } else if n > 0 { + slog.Info("cache purge-log pruned", "rows", n) + } + } +} + +// bootReconcile validates every cached key against live upload rows and +// purges the orphans — deletes that happened while this instance was down. +// The log high-water mark is read BEFORE the cache snapshot: a delete landing +// during reconciliation either logs past that mark (caught by the tail) or +// its row is already gone (caught by the liveness check) — no gap. Drops are +// unconditional (Store.Drop): the only concurrent promotion of a non-live +// key is a resurrection, which the promote-site guard is already unwinding. +func (w *CacheSweepWorker) bootReconcile(ctx context.Context) error { + maxID, err := w.uploadSvc.MaxPurgeLogID() + if err != nil { + return err + } + count, _ := w.store.Stats() + if count == 0 { + w.lastPurgeID = maxID + return nil + } + victims := w.store.Oldest(count) + keys := make([]string, len(victims)) + for i, v := range victims { + keys[i] = v.Key + } + live, err := w.uploadSvc.LiveCacheKeys(keys) + if err != nil { + return err + } + purged := 0 + for _, v := range victims { + if err := ctx.Err(); err != nil { + return err + } + if live[v.Key] { + continue + } + if err := w.store.Drop(v.Key); err != nil { + return err + } + purged++ + } + w.lastPurgeID = maxID + if purged > 0 { + slog.Info("download cache reconciled at boot", "purged_orphans", purged) + } + return nil +} + // emitStats writes one cumulative "download cache stats" line to slog (V2-825) // when any counter moved since the last emission — the operator's sizing // telemetry (hit ratio, eviction churn, bytes saved) until the V2-767 traffic diff --git a/internal/worker/upload.go b/internal/worker/upload.go index 6ed33d4..f5b42cc 100644 --- a/internal/worker/upload.go +++ b/internal/worker/upload.go @@ -577,6 +577,11 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa return fmt.Errorf("Failed to save upload record") } } + // Private seeding (V2-873) mirrors the public branch above, keyed on + // the local DataMap — the serve path's derivation for private rows. + // seedDownloadCache itself refuses unless the operator opted in via + // download_cache_private. + w.seedDownloadCache(upload, result.DataMap) } slog.Info("upload completed", "uuid", upload.UUID, "payment_type", prepared.PaymentType, @@ -621,6 +626,11 @@ func (w *UploadWorker) seedDownloadCache(upload *services.Upload, contentID stri if !upload.TempPath.Valid || upload.TempPath.String == "" { return } + // Private plaintext enters the cache only behind the operator's explicit + // opt-in (V2-873); public content is cacheable unconditionally. + if upload.Visibility != "public" && !w.settingsSvc.GetBool("download_cache_private", false) { + return + } // Default true: seeding is the point of the cache for publish-then-read // workloads; the switch exists for archive-shaped instances (upload-heavy, // rarely re-read) where every seed is a zero-use admission. diff --git a/web/src/views/admin/SettingsView.vue b/web/src/views/admin/SettingsView.vue index b1e1ccc..e1b5d42 100644 --- a/web/src/views/admin/SettingsView.vue +++ b/web/src/views/admin/SettingsView.vue @@ -41,8 +41,8 @@ watch(general, () => { }, { deep: true }) // Transfer Limits card -const uploadsSaved = ref({ max_upload_gb: '0', max_concurrent_uploads: '', max_gas_fee: '', payment_confirmation_timeout_seconds: '', max_concurrent_downloads: '', download_queue_wait_secs: '', download_cache_gb: '0', download_cache_max_object_mb: '', download_cache_min_uses: '', download_cache_inactive_secs: '', download_cache_seed_on_upload: 'true' }) -const uploads = reactive({ max_upload_gb: '0', max_concurrent_uploads: '', max_gas_fee: '', payment_confirmation_timeout_seconds: '', max_concurrent_downloads: '', download_queue_wait_secs: '', download_cache_gb: '0', download_cache_max_object_mb: '', download_cache_min_uses: '', download_cache_inactive_secs: '', download_cache_seed_on_upload: 'true' }) +const uploadsSaved = ref({ max_upload_gb: '0', max_concurrent_uploads: '', max_gas_fee: '', payment_confirmation_timeout_seconds: '', max_concurrent_downloads: '', download_queue_wait_secs: '', download_cache_gb: '0', download_cache_max_object_mb: '', download_cache_min_uses: '', download_cache_inactive_secs: '', download_cache_seed_on_upload: 'true', download_cache_private: 'false' }) +const uploads = reactive({ max_upload_gb: '0', max_concurrent_uploads: '', max_gas_fee: '', payment_confirmation_timeout_seconds: '', max_concurrent_downloads: '', download_queue_wait_secs: '', download_cache_gb: '0', download_cache_max_object_mb: '', download_cache_min_uses: '', download_cache_inactive_secs: '', download_cache_seed_on_upload: 'true', download_cache_private: 'false' }) const uploadsDirty = ref(false) const uploadsSaving = ref(false) @@ -58,7 +58,8 @@ watch(uploads, () => { uploads.download_cache_max_object_mb !== uploadsSaved.value.download_cache_max_object_mb || uploads.download_cache_min_uses !== uploadsSaved.value.download_cache_min_uses || uploads.download_cache_inactive_secs !== uploadsSaved.value.download_cache_inactive_secs || - uploads.download_cache_seed_on_upload !== uploadsSaved.value.download_cache_seed_on_upload + uploads.download_cache_seed_on_upload !== uploadsSaved.value.download_cache_seed_on_upload || + uploads.download_cache_private !== uploadsSaved.value.download_cache_private }, { deep: true }) // Operations card @@ -144,6 +145,7 @@ async function fetchSettings() { uploads.download_cache_min_uses = s.download_cache_min_uses || '' uploads.download_cache_inactive_secs = s.download_cache_inactive_secs || '' uploads.download_cache_seed_on_upload = s.download_cache_seed_on_upload || 'true' + uploads.download_cache_private = s.download_cache_private || 'false' uploadsSaved.value = { ...uploads } // Operations @@ -187,6 +189,7 @@ async function saveCard(card: string) { download_cache_min_uses: uploads.download_cache_min_uses, download_cache_inactive_secs: uploads.download_cache_inactive_secs, download_cache_seed_on_upload: uploads.download_cache_seed_on_upload, + download_cache_private: uploads.download_cache_private, } } else if (card === 'ops') { opsSaving.value = true @@ -464,6 +467,17 @@ onMounted(async () => { +
+
+ +

Serve repeat reads of private files from this deployment's disk. This keeps decrypted private content on every caching instance's disk (off by default): deleting an upload purges the deleting instance immediately and every other instance within about a minute, but treat cache volumes with the same care as the database. Leave off unless private read performance matters.

+
+
+ + {{ uploads.download_cache_private === 'true' ? 'Enabled' : 'Disabled' }} +
+
From 699af6b36fe286af3544cf477c4f00eb287c4996 Mon Sep 17 00:00:00 2001 From: Nic Date: Wed, 5 Aug 2026 15:08:50 +0100 Subject: [PATCH 2/4] test: back sweep-test cache entries with live upload rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Boot reconciliation (correctly) purged the sweep tests' fixtures as orphans before the eviction assertions ran — the entries stood in for legitimately cached live content and now have matching rows, which is also what production looks like. Co-Authored-By: Claude Fable 5 --- internal/worker/cache_sweep_test.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/internal/worker/cache_sweep_test.go b/internal/worker/cache_sweep_test.go index 8459db7..c995a1f 100644 --- a/internal/worker/cache_sweep_test.go +++ b/internal/worker/cache_sweep_test.go @@ -43,6 +43,13 @@ func newSweepEnv(t *testing.T, n int, content string, settings map[string]string if err := store.Scan(context.Background()); err != nil { t.Fatalf("scan: %v", err) } + // Each promoted entry gets a backing upload row with a matching + // cache_key: the V2-873 boot reconciliation purges cached keys with no + // live row, and these fixtures are standing in for legitimately cached + // live content — not orphans. + if _, err := db.Exec(`INSERT INTO users (id, email) VALUES (9001, 'sweep-test@example.com')`); err != nil { + t.Fatalf("insert user: %v", err) + } for i := 0; i < n; i++ { src := filepath.Join(dir, fmt.Sprintf("tmp-%d", i)) if err := os.WriteFile(src, []byte(content), 0600); err != nil { @@ -51,6 +58,11 @@ func newSweepEnv(t *testing.T, n int, content string, settings map[string]string if _, err := store.PromoteIfFits(sweepKey(i), src, 1<<40); err != nil { t.Fatalf("promote %d: %v", i, err) } + if _, err := db.Exec(`INSERT INTO uploads (id, uuid, user_id, filename, original_filename, file_size, content_type, visibility, status, cache_key) + VALUES (?, ?, 9001, 'sweep.bin', 'sweep.bin', ?, 'text/plain', 'public', 'completed', ?)`, + 9100+i, fmt.Sprintf("sweep-uuid-%d", i), len(content), sweepKey(i)); err != nil { + t.Fatalf("insert upload %d: %v", i, err) + } } cfg := &config.Config{DataDir: dir} From 87c503e06929c50535bfa779cb5bdc2b2c6560f7 Mon Sep 17 00:00:00 2001 From: Nic Date: Wed, 5 Aug 2026 16:31:21 +0100 Subject: [PATCH 3/4] fix: close the fleet purge-loss windows from the panel review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Finding 1: the purge-log append and upload-row delete now commit in one transaction — no interleaving lets a reconciling instance record a log high-water mark, still see the live row, and miss the delete between. A refused delete rolls its log rows back (regression-tested). Finding 2: the full liveness reconciliation now re-runs every 12 hours (and on demand) as the guaranteed backstop, so a purge whose log row was pruned while its unlink was stuck is re-derived from live rows and retried until the bytes are gone; a stuck key no longer delays later purges — every entry is attempted per tick, only the high-water mark waits for the contiguous prefix. Both panel reproductions are now regression tests. Finding 3: reconciliation is policy-aware — turning download_cache_private off purges already-cached private plaintext (next tick online, boot reconciliation for instances that were down) instead of stranding it; no-op setting updates do no spurious work. Co-Authored-By: Claude Fable 5 --- USER-GUIDE.md | 2 +- docs/guides/download-cache.md | 4 +- internal/services/upload.go | 64 +++++---- internal/services/upload_purge_test.go | 33 ++++- internal/worker/cache_purge_test.go | 185 ++++++++++++++++++++++++- internal/worker/cache_sweep.go | 135 ++++++++++++------ web/src/views/admin/SettingsView.vue | 2 +- 7 files changed, 339 insertions(+), 86 deletions(-) diff --git a/USER-GUIDE.md b/USER-GUIDE.md index 108e07f..09bc1ce 100644 --- a/USER-GUIDE.md +++ b/USER-GUIDE.md @@ -558,7 +558,7 @@ Runtime settings are stored in the database and take effect immediately without | `download_cache_min_uses` | `1` | How many times an object must be requested (per instance) before its bytes are cached. `1` caches on first download; raising it keeps one-hit-wonders from displacing hot content. Bounds: **1–100** | | `download_cache_inactive_secs` | `0` (off) | Inactivity expiry for cached objects: entries not **accessed** within this window are deleted regardless of the size budget. Independent pruning axis — also bounds how long cached plaintext can linger on an instance's disk. Example: `604800` = 7 days. Bounds: **0–315360000** | | `download_cache_seed_on_upload` | `true` | Seed the download cache from an upload's staged bytes (public; private too when `download_cache_private` is on) the moment its network store succeeds, so publish-then-read is served from local disk with no network fetch (a rename — zero extra I/O). Applies where uploads are processed — the writer (or the whole instance in an all-in-one deployment); on a reader fleet each reader still warms by read-through. Seeds respect the size ceiling and byte budget but deliberately skip `download_cache_min_uses`; the sweeper evicts wrong bets. Disable for archive-shaped workloads (upload-heavy, rarely re-read) | -| `download_cache_private` | `false` | Allow **private** uploads into the download cache (read-through and seeding). Off by default because it keeps decrypted private content on every caching instance's disk — a new at-rest surface. With it on, deleting an upload purges the handling instance synchronously and every other instance within about one sweep tick (~60s) via the purge log; see the [download-cache deployment guide](docs/guides/download-cache.md) for the exact contract | +| `download_cache_private` | `false` | Allow **private** uploads into the download cache (read-through and seeding). Off by default because it keeps decrypted private content on every caching instance's disk — a new at-rest surface. With it on, deleting an upload purges the handling instance synchronously and every other instance within about one sweep tick (~60s) via the purge log. Turning it back off removes already-cached private files fleet-wide (next tick online, boot reconciliation for instances that were down); see the [download-cache deployment guide](docs/guides/download-cache.md) for the exact contract | For the `download_cache_*` family, the [download-cache deployment guide](docs/guides/download-cache.md) covers sizing, the stats telemetry, per-setting fleet scope, and the privacy posture. diff --git a/docs/guides/download-cache.md b/docs/guides/download-cache.md index 339914e..7387fed 100644 --- a/docs/guides/download-cache.md +++ b/docs/guides/download-cache.md @@ -67,8 +67,8 @@ Every instance owns its cache — there is no shared cache tier, by design (shar Cached entries are **plaintext bytes on the instance's disk**, held outside the audit surface of the upload store. That is why: -- **Public uploads are cached unconditionally; private uploads only behind `download_cache_private`** (default off). Private caching puts decrypted plaintext where a disk snapshot, misconfigured backup, or shared volume could expose it — enabling it is an explicit operator decision, and cache volumes then deserve the same care as the database and temp-upload disks. Authorization is unaffected: the cache is only consulted after the row/owner/visibility checks, so a cache hit never bypasses the token gate. +- **Public uploads are cached unconditionally; private uploads only behind `download_cache_private`** (default off). Private caching puts decrypted plaintext where a disk snapshot, misconfigured backup, or shared volume could expose it — enabling it is an explicit operator decision, and cache volumes then deserve the same care as the database and temp-upload disks. Authorization is unaffected: the cache is only consulted after the row/owner/visibility checks, so a cache hit never bypasses the token gate. **Turning the opt-in off removes already-cached private plaintext fleet-wide**, not just future admissions: online instances purge on their next sweep tick, and an instance that was down applies the policy during its boot reconciliation. - Cache files are owner-only (`0600`) under `/cache/objects`, named by content digest — the name never reveals the DataMap (which is the retrieval capability), and the digest is domain-separated from the plaintext hash. - `download_cache_inactive_secs` doubles as the bound on how long unread cached plaintext can linger; budget `0` (or the env override `0`) drains an instance completely. - **Deleting an upload purges its cached copy synchronously on the instance that handles the delete** (V2-824): the purge runs before the record is removed, and a purge that cannot unlink fails the delete rather than reporting a deletion while the plaintext remains readable. A download already in flight when the delete lands finishes streaming (its descriptor outlives the unlink) and any promotion it makes is taken back out. That take-back is best-effort in the instant, but the purge log below retries it durably. -- **Deletes propagate to the whole fleet through the purge log.** The delete appends the upload's cache keys to `cache_purge_log` before any row is removed; every instance's sweep worker consumes the log tail each tick and unlinks local copies, never advancing past a key whose unlink failed (it retries next tick). At boot, each instance reconciles its entire cache directory against live upload rows (`uploads.cache_key`), which covers deletes that happened while it was down — even beyond the log's retention window (7 days, pruned by the writer). **The purge-window contract:** the instance handling the delete purges synchronously (the API fails rather than report a deletion with the plaintext still readable); every other instance purges within about one sweep tick (~60s). If database read-replicas are ever introduced, replica lag adds to that window. On a reader fleet, other instances' cached copies of a deleted *public* upload are unreachable immediately (the record is gone) and age out by eviction/inactivity — acceptable for public bytes, and exactly why private content is not cached. +- **Deletes propagate to the whole fleet through the purge log.** The delete appends the upload's cache keys to `cache_purge_log` **in the same transaction** as the row removal (no interleaving can observe the row live with the log entry missing, or vice versa); every instance's sweep worker consumes the log tail each tick and unlinks local copies. A key whose unlink fails is retried from the log next tick without delaying later purges (they are applied the same tick; only the high-water mark waits). Each instance also runs a **full liveness reconciliation** of its cache directory against live upload rows (`uploads.cache_key`) at boot and every 12 hours — the guaranteed backstop that covers deletes that happened while it was down, purge-log rows pruned past the 7-day retention, and any unlink that stayed stuck across ticks. **The purge-window contract:** the instance handling the delete purges synchronously (the API fails rather than report a deletion with the plaintext still readable); every other instance purges within about one sweep tick (~60s). If database read-replicas are ever introduced, replica lag adds to that window. On a reader fleet, other instances' cached copies of a deleted *public* upload are unreachable immediately (the record is gone) and age out by eviction/inactivity — acceptable for public bytes, and exactly why private content is not cached. diff --git a/internal/services/upload.go b/internal/services/upload.go index 8b135fc..98a347a 100644 --- a/internal/services/upload.go +++ b/internal/services/upload.go @@ -570,42 +570,49 @@ func (s *UploadService) ForceRetry(id int64) error { // Delete permanently removes an upload record. Only allowed for failed or completed uploads. func (s *UploadService) Delete(id int64) error { - // Fan the purge out to the fleet FIRST (V2-873): append the upload's - // cache keys to cache_purge_log before anything is deleted, so a failure - // later in this sequence can at worst cause a spurious purge (an instance - // drops a live entry and re-warms) — never a missed one. Every instance's - // cache sweep worker consumes this log each tick; the handling instance - // additionally purges synchronously in the handler (V2-824). - if u, err := s.GetByID(id); err == nil { - for _, key := range u.CacheKeys() { - if _, err := s.db.Exec(`INSERT INTO cache_purge_log (cache_key) VALUES (?)`, key); err != nil { - return err - } + // One transaction for the purge-log fan-out and the row deletion + // (V2-873, #155 panel finding 1): any other connection sees either + // row-live/no-log or row-gone/log-present — never a state where a + // reconciling instance could read the log's high-water mark, still see + // the live row, and then miss the delete that committed between. A + // refused or failed delete rolls the log rows back with it. + tx, err := s.db.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() // no-op once committed + + var dataMap, addr sql.NullString + if err := tx.QueryRow(`SELECT data_map, datamap_address FROM uploads WHERE id = ?`, id).Scan(&dataMap, &addr); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return errors.New("only failed or completed uploads can be deleted") } - } else if !errors.Is(err, ErrUploadNotFound) { return err } + for _, key := range (&Upload{DataMap: dataMap, DatamapAddress: addr}).CacheKeys() { + if _, err := tx.Exec(`INSERT INTO cache_purge_log (cache_key) VALUES (?)`, key); err != nil { + return err + } + } - // Clean up related data first - if _, err := s.db.Exec(`DELETE FROM file_tags WHERE upload_id = ?`, id); err != nil { + if _, err := tx.Exec(`DELETE FROM file_tags WHERE upload_id = ?`, id); err != nil { return err } - if _, err := s.db.Exec(`DELETE FROM collection_files WHERE upload_id = ?`, id); err != nil { + if _, err := tx.Exec(`DELETE FROM collection_files WHERE upload_id = ?`, id); err != nil { return err } - result, err := s.db.Exec( + result, err := tx.Exec( `DELETE FROM uploads WHERE id = ? AND status IN ('failed', 'completed')`, id, ) if err != nil { return err } - n, _ := result.RowsAffected() - if n == 0 { + if n, _ := result.RowsAffected(); n == 0 { return errors.New("only failed or completed uploads can be deleted") } - return nil + return tx.Commit() } // CacheKeys returns every download-cache key this upload's bytes could live @@ -680,10 +687,13 @@ func (s *UploadService) PrunePurgeLog(before time.Time) (int64, error) { return n, nil } -// LiveCacheKeys reports which of the given cache keys belong to a live upload -// row — the boot-reconciliation lookup (uploads.cache_key is indexed). -func (s *UploadService) LiveCacheKeys(keys []string) (map[string]bool, error) { - live := make(map[string]bool, len(keys)) +// CacheKeyVisibility reports, for each given cache key that belongs to a live +// upload row, that row's visibility ("public"/"private") — the reconciliation +// lookup (uploads.cache_key is indexed). Keys absent from the result have no +// live row. Reconciliation is policy-aware (#155 panel finding 3): a private +// key is kept only while download_cache_private allows it. +func (s *UploadService) CacheKeyVisibility(keys []string) (map[string]string, error) { + live := make(map[string]string, len(keys)) const chunk = 500 for start := 0; start < len(keys); start += chunk { end := start + chunk @@ -701,18 +711,18 @@ func (s *UploadService) LiveCacheKeys(keys []string) (map[string]bool, error) { args = append(args, k) } rows, err := s.db.Query( - `SELECT cache_key FROM uploads WHERE cache_key IN (`+string(placeholders)+`)`, args..., + `SELECT cache_key, visibility FROM uploads WHERE cache_key IN (`+string(placeholders)+`)`, args..., ) if err != nil { return nil, err } for rows.Next() { - var k string - if err := rows.Scan(&k); err != nil { + var k, vis string + if err := rows.Scan(&k, &vis); err != nil { rows.Close() return nil, err } - live[k] = true + live[k] = vis } if err := rows.Err(); err != nil { rows.Close() diff --git a/internal/services/upload_purge_test.go b/internal/services/upload_purge_test.go index dc41a28..12cab5c 100644 --- a/internal/services/upload_purge_test.go +++ b/internal/services/upload_purge_test.go @@ -122,12 +122,12 @@ func TestPrunePurgeLogAndLiveCacheKeys(t *testing.T) { } liveKey := downloadcache.KeyForIdentifier("dm-live") - live, err := svc.LiveCacheKeys([]string{liveKey, "0000000000000000000000000000000000000000000000000000000000000000"}) + live, err := svc.CacheKeyVisibility([]string{liveKey, "0000000000000000000000000000000000000000000000000000000000000000"}) if err != nil { - t.Fatalf("LiveCacheKeys: %v", err) + t.Fatalf("CacheKeyVisibility: %v", err) } - if !live[liveKey] || len(live) != 1 { - t.Fatalf("LiveCacheKeys = %v, want only the live key", live) + if live[liveKey] != "private" || len(live) != 1 { + t.Fatalf("CacheKeyVisibility = %v, want only the live key mapped to private", live) } if _, err := db.Exec(`INSERT INTO cache_purge_log (cache_key, deleted_at) VALUES (?, ?)`, "old-key", "2020-01-01 00:00:00"); err != nil { @@ -145,3 +145,28 @@ func TestPrunePurgeLogAndLiveCacheKeys(t *testing.T) { t.Fatalf("post-prune log = %+v (err=%v), want only fresh-key", entries, err) } } + +// #155 panel finding 1: the purge-log append and the row delete are one +// transaction — a refused delete must leave no log rows behind (and other +// connections can never observe row-live + log-present). +func TestDeleteRefusedRollsBackPurgeLog(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "atomic@example.com", "A", "T") + svc := NewUploadService(db) + + u := createTestUpload(t, svc, user.ID, "queued.bin", 10) // status stays "queued" + if _, err := db.Exec(`UPDATE uploads SET data_map='dm-atomic' WHERE id = ?`, u.ID); err != nil { + t.Fatalf("set identifier: %v", err) + } + + if err := svc.Delete(u.ID); err == nil { + t.Fatal("delete of a queued upload must be refused") + } + entries, err := svc.PurgeLogSince(0, 10) + if err != nil || len(entries) != 0 { + t.Fatalf("refused delete leaked purge-log rows: %+v (err=%v)", entries, err) + } + if _, err := svc.GetByID(u.ID); err != nil { + t.Fatalf("row must survive a refused delete: %v", err) + } +} diff --git a/internal/worker/cache_purge_test.go b/internal/worker/cache_purge_test.go index 3c230f4..76d8e4d 100644 --- a/internal/worker/cache_purge_test.go +++ b/internal/worker/cache_purge_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/WithAutonomi/indelible/internal/config" "github.com/WithAutonomi/indelible/internal/database" @@ -37,13 +38,13 @@ func newPurgeEnv(t *testing.T) (*database.DB, *downloadcache.Store, *CacheSweepW } // mkStoredUpload inserts a completed upload row whose cache_key matches -// KeyForIdentifier(identifier), and returns its id and key. -func mkStoredUpload(t *testing.T, db *database.DB, id int64, uuid, identifier string) string { +// KeyForIdentifier(identifier), and returns its key. +func mkStoredUpload(t *testing.T, db *database.DB, id int64, uuid, identifier, visibility string) string { t.Helper() key := downloadcache.KeyForIdentifier(identifier) if _, err := db.Exec(`INSERT INTO uploads (id, uuid, user_id, filename, original_filename, file_size, content_type, visibility, status, data_map, cache_key) - VALUES (?, ?, 8001, 'f.bin', 'f.bin', 4, 'text/plain', 'private', 'completed', ?, ?)`, - id, uuid, identifier, key); err != nil { + VALUES (?, ?, 8001, 'f.bin', 'f.bin', 4, 'text/plain', ?, 'completed', ?, ?)`, + id, uuid, visibility, identifier, key); err != nil { t.Fatalf("insert upload: %v", err) } return key @@ -69,7 +70,7 @@ func cacheEntry(t *testing.T, store *downloadcache.Store, dir, key, content stri func TestPropagatePurges_RemoteInstanceAppliesDelete(t *testing.T) { db, store, w, svc := newPurgeEnv(t) dir := t.TempDir() - key := mkStoredUpload(t, db, 8101, "fleet-uuid-1", "dm-fleet-1") + key := mkStoredUpload(t, db, 8101, "fleet-uuid-1", "dm-fleet-1", "public") p := cacheEntry(t, store, dir, key, "fleet bytes") // Boot tick: the row is live, so reconciliation keeps the entry. @@ -96,7 +97,7 @@ func TestPropagatePurges_RemoteInstanceAppliesDelete(t *testing.T) { func TestBootReconcile_DropsOrphansKeepsLive(t *testing.T) { db, store, w, _ := newPurgeEnv(t) dir := t.TempDir() - liveKey := mkStoredUpload(t, db, 8102, "fleet-uuid-2", "dm-live-2") + liveKey := mkStoredUpload(t, db, 8102, "fleet-uuid-2", "dm-live-2", "public") cacheEntry(t, store, dir, liveKey, "live bytes") orphanKey := downloadcache.KeyForIdentifier("dm-orphan-2") orphanPath := cacheEntry(t, store, dir, orphanKey, "orphan bytes") @@ -120,7 +121,7 @@ func TestPropagatePurges_FailedUnlinkHaltsHighWater(t *testing.T) { } db, store, w, svc := newPurgeEnv(t) dir := t.TempDir() - key := mkStoredUpload(t, db, 8103, "fleet-uuid-3", "dm-stuck-3") + key := mkStoredUpload(t, db, 8103, "fleet-uuid-3", "dm-stuck-3", "public") p := cacheEntry(t, store, dir, key, "stuck bytes") w.propagatePurges(context.Background()) // boot with the row live @@ -172,3 +173,173 @@ func TestPropagatePurges_WriterPrunesOldRows(t *testing.T) { t.Fatalf("old purge-log rows survived the writer prune: %+v", entries) } } + +// #155 panel finding 2, starvation half: one stuck key must not delay later +// deletes — every entry's Drop is attempted per tick even while the +// high-water mark stalls at the failed key. +func TestPropagatePurges_FailedKeyDoesNotDelayLaterPurges(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root ignores directory permissions") + } + db, store, w, svc := newPurgeEnv(t) + dir := t.TempDir() + stuckKey := mkStoredUpload(t, db, 8104, "fleet-uuid-4", "dm-stuck-4", "public") + stuckPath := cacheEntry(t, store, dir, stuckKey, "stuck bytes") + laterKey := mkStoredUpload(t, db, 8105, "fleet-uuid-5", "dm-later-5", "public") + laterPath := cacheEntry(t, store, dir, laterKey, "later bytes") + if filepath.Dir(stuckPath) == filepath.Dir(laterPath) { + t.Fatal("fixture keys share a fanout dir; pick different identifiers") + } + + w.propagatePurges(context.Background()) // boot, both rows live + + if err := svc.Delete(8104); err != nil { + t.Fatalf("delete stuck: %v", err) + } + if err := svc.Delete(8105); err != nil { + t.Fatalf("delete later: %v", err) + } + parent := filepath.Dir(stuckPath) + if err := os.Chmod(parent, 0500); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { _ = os.Chmod(parent, 0700) }) + + before := w.lastPurgeID + w.propagatePurges(context.Background()) + if _, ok := store.Get(laterKey); ok { + t.Fatal("a later delete was delayed behind the stuck key") + } + if _, ok := store.Get(stuckKey); !ok { + t.Fatal("stuck entry must stay indexed while its bytes are stuck") + } + if w.lastPurgeID != before { + t.Fatal("high-water mark advanced past the failed key") + } + + if err := os.Chmod(parent, 0700); err != nil { + t.Fatalf("chmod restore: %v", err) + } + w.propagatePurges(context.Background()) + if _, ok := store.Get(stuckKey); ok { + t.Fatal("retry tick did not purge the stuck key") + } +} + +// #155 panel finding 2, prune-loss half — the panel's exact reproduction: +// a stuck unlink stalls the watermark, the writer prunes the only retry +// record, permissions recover... and the tail has nothing to retry. The +// periodic full reconciliation is the guaranteed backstop that recovers it. +func TestPropagatePurges_PruneLossRecoveredByReconciliation(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root ignores directory permissions") + } + db, store, w, svc := newPurgeEnv(t) + dir := t.TempDir() + key := mkStoredUpload(t, db, 8106, "fleet-uuid-6", "dm-pruned-6", "public") + p := cacheEntry(t, store, dir, key, "pruned-retry bytes") + + w.propagatePurges(context.Background()) // boot, row live + + if err := svc.Delete(8106); err != nil { + t.Fatalf("delete: %v", err) + } + parent := filepath.Dir(p) + if err := os.Chmod(parent, 0500); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { _ = os.Chmod(parent, 0700) }) + w.propagatePurges(context.Background()) // unlink fails, watermark stalls + + // The retention prune erases the retry record while the key is stuck. + if _, err := db.Exec(`DELETE FROM cache_purge_log`); err != nil { + t.Fatalf("prune: %v", err) + } + if err := os.Chmod(parent, 0700); err != nil { + t.Fatalf("chmod restore: %v", err) + } + + // Tail-only tick: nothing left in the log — the orphan survives. This is + // the loss the panel reproduced. + w.propagatePurges(context.Background()) + if _, ok := store.Get(key); !ok { + t.Fatal("precondition: orphan should still be cached after the log was pruned") + } + + // The periodic reconciliation re-derives the orphan from live rows. + w.lastReconcile = time.Now().Add(-cacheReconcileInterval - time.Minute) + w.propagatePurges(context.Background()) + if _, ok := store.Get(key); ok { + t.Fatal("reconciliation backstop did not purge the orphan") + } + if _, err := os.Stat(p); !os.IsNotExist(err) { + t.Fatal("orphan bytes still on disk after reconciliation") + } +} + +// #155 panel finding 3, online half: flipping download_cache_private off +// purges already-cached private plaintext on the next tick, keeps public +// entries, and no-op ticks afterwards do no spurious work. +func TestPropagatePurges_PrivateOptOutPurgesExisting(t *testing.T) { + db, store, w, _ := newPurgeEnv(t) + settings := services.NewSettingsService(db) + if err := settings.SetInternal("download_cache_private", "true"); err != nil { + t.Fatalf("set: %v", err) + } + dir := t.TempDir() + privKey := mkStoredUpload(t, db, 8107, "fleet-uuid-7", "dm-priv-7", "private") + privPath := cacheEntry(t, store, dir, privKey, "private bytes") + pubKey := mkStoredUpload(t, db, 8108, "fleet-uuid-8", "dm-pub-8", "public") + cacheEntry(t, store, dir, pubKey, "public bytes") + + w.propagatePurges(context.Background()) // boot with the opt-in on: both kept + if _, ok := store.Get(privKey); !ok { + t.Fatal("boot purged an allowed private entry") + } + + if err := settings.SetInternal("download_cache_private", "false"); err != nil { + t.Fatalf("flip: %v", err) + } + w.settingsSvc.InvalidateAll() + w.propagatePurges(context.Background()) // transition tick + + if _, ok := store.Get(privKey); ok { + t.Fatal("opt-out left private plaintext cached") + } + if _, err := os.Stat(privPath); !os.IsNotExist(err) { + t.Fatal("private bytes still on disk after opt-out") + } + if _, ok := store.Get(pubKey); !ok { + t.Fatal("opt-out must not touch public entries") + } + + purged := store.Metrics().Purged.Load() + w.propagatePurges(context.Background()) // no-op tick + if got := store.Metrics().Purged.Load(); got != purged { + t.Fatalf("no-op tick purged %d more entries", got-purged) + } +} + +// #155 panel finding 3, offline half: an instance that was down past log +// retention boots with the opt-in off — reconciliation purges its private +// entries even though their rows are live. +func TestPropagatePurges_BootPurgesPrivateWhenOptedOut(t *testing.T) { + db, store, w, _ := newPurgeEnv(t) // download_cache_private defaults off + dir := t.TempDir() + privKey := mkStoredUpload(t, db, 8109, "fleet-uuid-9", "dm-priv-9", "private") + privPath := cacheEntry(t, store, dir, privKey, "stranded private bytes") + pubKey := mkStoredUpload(t, db, 8110, "fleet-uuid-10", "dm-pub-10", "public") + cacheEntry(t, store, dir, pubKey, "public bytes") + + w.propagatePurges(context.Background()) + + if _, ok := store.Get(privKey); ok { + t.Fatal("boot kept private plaintext despite the opt-in being off") + } + if _, err := os.Stat(privPath); !os.IsNotExist(err) { + t.Fatal("private bytes still on disk after policy-aware boot") + } + if _, ok := store.Get(pubKey); !ok { + t.Fatal("policy-aware boot must keep live public entries") + } +} diff --git a/internal/worker/cache_sweep.go b/internal/worker/cache_sweep.go index 26bb58f..074f45a 100644 --- a/internal/worker/cache_sweep.go +++ b/internal/worker/cache_sweep.go @@ -45,14 +45,20 @@ type CacheSweepWorker struct { uploadSvc *services.UploadService // Fleet purge propagation (V2-873): lastPurgeID is this instance's - // high-water mark in cache_purge_log — in-memory only, because boot runs - // a full reconciliation against uploads.cache_key that subsumes any log - // history. pruneLog is writer-role only (readers stay DB-write-free, - // V2-514). booted flips after the boot reconciliation succeeds. - lastPurgeID int64 - pruneLog bool - lastPrune time.Time - booted bool + // high-water mark in cache_purge_log — in-memory only, because a full + // reconciliation against uploads.cache_key subsumes log history and runs + // at boot, every cacheReconcileInterval (the guaranteed backstop the + // #155 panel required: it recovers purges whose log rows were pruned + // while an unlink was stuck), and on a private-caching opt-out (which + // must remove already-cached private plaintext, not just stop admitting + // it). pruneLog is writer-role only (readers stay DB-write-free, V2-514). + // privateAllowed is the last observed opt-in value, for detecting the + // true→false transition. + lastPurgeID int64 + pruneLog bool + lastPrune time.Time + lastReconcile time.Time + privateAllowed bool // usage reports volume capacity for the disk-pressure trigger; injected // so tests can simulate a filling disk. Defaults to diskusage.Usage. @@ -99,6 +105,14 @@ const ( // cachePurgeBatch is the per-read tail size when consuming the log. cachePurgeBatch = 500 + + // cacheReconcileInterval is how often the full cache-vs-rows liveness + // reconciliation re-runs after boot. It is the guaranteed erasure + // backstop: even if a purge-log row is pruned while an instance's unlink + // of that key is stuck, the next reconciliation re-derives the orphan + // from uploads.cache_key and retries. Comfortably inside the 7-day log + // retention. + cacheReconcileInterval = 12 * time.Hour ) // NewCacheSweepWorker creates the sweeper over the same store the download @@ -173,21 +187,36 @@ func (w *CacheSweepWorker) sweep(ctx context.Context) { } // propagatePurges applies fleet-wide delete purges to this instance's cache -// (V2-873). Boot: one full reconciliation of every cached key against live -// uploads.cache_key rows — which subsumes all purge-log history, so the log -// high-water mark starts at the log's current tail. Steady state: consume the -// log tail each tick, never advancing past a key whose unlink failed (it is -// retried next tick — a delete's disk-level guarantee on remote instances is -// this loop). Writer-role instances also prune the log hourly. +// (V2-873). A full liveness reconciliation runs at boot, every +// cacheReconcileInterval, and on a private-caching opt-out — it subsumes log +// history, so it is the guaranteed backstop for purges whose log rows were +// pruned while this instance's unlink was stuck. Steady state consumes the +// log tail: every entry's Drop is attempted (one stuck key never delays +// later deletes), but the high-water mark advances only through the +// contiguous successful prefix, so failed keys are retried from the log next +// tick (re-drops of already-purged keys are no-ops). Writer-role instances +// also prune the log hourly. func (w *CacheSweepWorker) propagatePurges(ctx context.Context) { - if !w.booted { - if err := w.bootReconcile(ctx); err != nil { - slog.Warn("download cache boot reconciliation failed; retrying next tick", "error", err) - return // never consume the tail from an unreconciled baseline - } - w.booted = true + privateNow := w.settingsSvc.GetBool("download_cache_private", false) + switch { + case w.lastReconcile.IsZero(): // boot + case w.privateAllowed && !privateNow: // opt-out: purge existing private plaintext + case time.Since(w.lastReconcile) >= cacheReconcileInterval: // periodic backstop + default: + goto tail } + if clean, err := w.reconcile(ctx, privateNow); err != nil { + slog.Warn("download cache reconciliation failed; retrying next tick", "error", err) + return // never consume the tail from an unreconciled baseline + } else if clean { + w.lastReconcile = time.Now() + } + // A not-clean pass (some unlink failed) completed its scan — the tail + // may proceed — but lastReconcile stays put so the next tick re-runs the + // reconciliation until every orphan's bytes are actually gone. +tail: + w.privateAllowed = privateNow for { entries, err := w.uploadSvc.PurgeLogSince(w.lastPurgeID, cachePurgeBatch) if err != nil { @@ -197,18 +226,24 @@ func (w *CacheSweepWorker) propagatePurges(ctx context.Context) { if len(entries) == 0 { break } + stalled := false for _, e := range entries { if ctx.Err() != nil { return } if err := w.store.Drop(e.CacheKey); err != nil { - slog.Warn("download cache propagated purge failed; will retry next tick", - "key", e.CacheKey, "error", err) - return + if !stalled { + slog.Warn("download cache propagated purge failed; will retry next tick", + "key", e.CacheKey, "error", err) + } + stalled = true + continue // later purges still apply this tick + } + if !stalled { + w.lastPurgeID = e.ID } - w.lastPurgeID = e.ID } - if len(entries) < cachePurgeBatch { + if stalled || len(entries) < cachePurgeBatch { break } } @@ -223,50 +258,62 @@ func (w *CacheSweepWorker) propagatePurges(ctx context.Context) { } } -// bootReconcile validates every cached key against live upload rows and -// purges the orphans — deletes that happened while this instance was down. -// The log high-water mark is read BEFORE the cache snapshot: a delete landing -// during reconciliation either logs past that mark (caught by the tail) or -// its row is already gone (caught by the liveness check) — no gap. Drops are -// unconditional (Store.Drop): the only concurrent promotion of a non-live -// key is a resurrection, which the promote-site guard is already unwinding. -func (w *CacheSweepWorker) bootReconcile(ctx context.Context) error { +// reconcile validates every cached key against live upload rows and purges +// the orphans — deletes that happened while this instance was down or whose +// log rows are gone — plus, policy-awareness (#155 panel finding 3): private +// keys are orphans whenever download_cache_private is off, so an opt-out +// removes already-cached private plaintext instead of stranding it. +// +// The log high-water mark is read BEFORE the cache snapshot, and the +// row-delete + log-append commit atomically (UploadService.Delete), so a +// delete landing during reconciliation either logs past that mark (caught by +// the tail) or its row is already invisible (caught by the liveness check) — +// no interleaving skips it. Drops are unconditional (Store.Drop): the only +// concurrent promotion of a non-live key is a resurrection, which the +// promote-site guard is already unwinding. +// +// Returns clean=false when some unlink failed: the scan completed and the +// caller may consume the tail, but the reconciliation must re-run next tick +// until the bytes are gone. +func (w *CacheSweepWorker) reconcile(ctx context.Context, privateAllowed bool) (clean bool, err error) { maxID, err := w.uploadSvc.MaxPurgeLogID() if err != nil { - return err + return false, err } count, _ := w.store.Stats() if count == 0 { w.lastPurgeID = maxID - return nil + return true, nil } victims := w.store.Oldest(count) keys := make([]string, len(victims)) for i, v := range victims { keys[i] = v.Key } - live, err := w.uploadSvc.LiveCacheKeys(keys) + visibility, err := w.uploadSvc.CacheKeyVisibility(keys) if err != nil { - return err + return false, err } - purged := 0 + purged, stuck := 0, 0 for _, v := range victims { if err := ctx.Err(); err != nil { - return err + return false, err } - if live[v.Key] { + vis, live := visibility[v.Key] + if live && (vis == "public" || privateAllowed) { continue } if err := w.store.Drop(v.Key); err != nil { - return err + stuck++ + continue } purged++ } w.lastPurgeID = maxID - if purged > 0 { - slog.Info("download cache reconciled at boot", "purged_orphans", purged) + if purged > 0 || stuck > 0 { + slog.Info("download cache reconciled", "purged", purged, "unlink_failures", stuck) } - return nil + return stuck == 0, nil } // emitStats writes one cumulative "download cache stats" line to slog (V2-825) diff --git a/web/src/views/admin/SettingsView.vue b/web/src/views/admin/SettingsView.vue index e1b5d42..461b6cd 100644 --- a/web/src/views/admin/SettingsView.vue +++ b/web/src/views/admin/SettingsView.vue @@ -470,7 +470,7 @@ onMounted(async () => {
-

Serve repeat reads of private files from this deployment's disk. This keeps decrypted private content on every caching instance's disk (off by default): deleting an upload purges the deleting instance immediately and every other instance within about a minute, but treat cache volumes with the same care as the database. Leave off unless private read performance matters.

+

Serve repeat reads of private files from this deployment's disk. This keeps decrypted private content on every caching instance's disk (off by default): deleting an upload purges the deleting instance immediately and every other instance within about a minute, but treat cache volumes with the same care as the database. Turning it off also removes already-cached private files fleet-wide (within about a minute online; at next start for instances that were down). Leave off unless private read performance matters.

Date: Thu, 6 Aug 2026 09:25:23 +0100 Subject: [PATCH 4/4] test: assert the no-op true-to-true tick does no purge work MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the panel's no-op-update matrix — false-to-false was already asserted; both are the same non-transition branch. Co-Authored-By: Claude Fable 5 --- internal/worker/cache_purge_test.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/internal/worker/cache_purge_test.go b/internal/worker/cache_purge_test.go index 76d8e4d..51db68c 100644 --- a/internal/worker/cache_purge_test.go +++ b/internal/worker/cache_purge_test.go @@ -297,6 +297,13 @@ func TestPropagatePurges_PrivateOptOutPurgesExisting(t *testing.T) { t.Fatal("boot purged an allowed private entry") } + // No-op true→true tick: nothing purged, no transition work. + purgedAtBoot := store.Metrics().Purged.Load() + w.propagatePurges(context.Background()) + if got := store.Metrics().Purged.Load(); got != purgedAtBoot { + t.Fatalf("true→true tick purged %d entries", got-purgedAtBoot) + } + if err := settings.SetInternal("download_cache_private", "false"); err != nil { t.Fatalf("flip: %v", err) }