From 9f50147e96d7fe70176416e81cd861a64d8ed2f5 Mon Sep 17 00:00:00 2001 From: Nic Date: Wed, 5 Aug 2026 14:56:36 +0100 Subject: [PATCH 1/6] 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/6] 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/6] 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/6] 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) } From efa1c1a43ceb0fa4cef09bafcb172e5dc0c8c206 Mon Sep 17 00:00:00 2001 From: Nic Date: Thu, 6 Aug 2026 09:43:02 +0100 Subject: [PATCH 5/6] fix: admit already_stored to the uploads status constraint and delete path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The V2-399 dedup status has been written by the worker since it shipped, but both dialects' schema CHECK constraints never listed it — on any database built from the migrations, a duplicate re-upload failed at the status flip after the network store succeeded. It was also absent from the delete path's allowed statuses, leaving such uploads with no erasure path. Migration 014 extends the CHECK. SQLite needs a table rebuild; renaming uploads rewrites the FK clauses of file_tags, collection_files, and transactions to follow the old table, so those three are rebuilt against the new uploads (the 011 pattern), and the uploads indexes are recreated only after uploads_old is dropped, which is when their names free up. Co-Authored-By: Claude Fable 5 --- .../postgres/014_already_stored_status.sql | 21 ++ .../sqlite/014_already_stored_status.sql | 213 ++++++++++++++++++ internal/services/upload.go | 10 +- .../services/upload_already_stored_test.go | 65 ++++++ 4 files changed, 304 insertions(+), 5 deletions(-) create mode 100644 internal/database/migrations/postgres/014_already_stored_status.sql create mode 100644 internal/database/migrations/sqlite/014_already_stored_status.sql create mode 100644 internal/services/upload_already_stored_test.go diff --git a/internal/database/migrations/postgres/014_already_stored_status.sql b/internal/database/migrations/postgres/014_already_stored_status.sql new file mode 100644 index 0000000..35b26f2 --- /dev/null +++ b/internal/database/migrations/postgres/014_already_stored_status.sql @@ -0,0 +1,21 @@ +-- +goose Up + +-- V2-875: the 'already_stored' status (V2-399 content-addressed dedup — a +-- re-upload whose chunks were all already on the network) has been written by +-- the worker since V2-399 shipped, but the status CHECK constraint never +-- listed it, so the transition fails on any database built from these +-- migrations. Found by the #154 panel review. Postgres names the inline +-- column CHECK uploads_status_check. +ALTER TABLE uploads DROP CONSTRAINT uploads_status_check; +ALTER TABLE uploads ADD CONSTRAINT uploads_status_check + CHECK (status IN ('queued', 'processing', 'completed', 'failed', 'already_stored')); + +-- +goose Down + +-- Rows already in 'already_stored' would violate the original constraint, so +-- they are folded into 'completed' (both are terminal stored states; the +-- distinction is a UI nicety). +UPDATE uploads SET status = 'completed' WHERE status = 'already_stored'; +ALTER TABLE uploads DROP CONSTRAINT uploads_status_check; +ALTER TABLE uploads ADD CONSTRAINT uploads_status_check + CHECK (status IN ('queued', 'processing', 'completed', 'failed')); diff --git a/internal/database/migrations/sqlite/014_already_stored_status.sql b/internal/database/migrations/sqlite/014_already_stored_status.sql new file mode 100644 index 0000000..232ba7a --- /dev/null +++ b/internal/database/migrations/sqlite/014_already_stored_status.sql @@ -0,0 +1,213 @@ +-- +goose Up + +-- V2-875: the 'already_stored' status (V2-399 content-addressed dedup — a +-- re-upload whose chunks were all already on the network) has been written by +-- the worker since V2-399 shipped, but the status CHECK constraint never +-- listed it, so the transition fails on any database built from these +-- migrations. Found by the #154 panel review. +-- +-- SQLite cannot alter a CHECK in place, so uploads is rebuilt. Renaming a +-- table rewrites the FK clauses of every table that references it (and the +-- legacy_alter_table escape hatch is ignored inside goose's transaction), so +-- the three referencing tables — file_tags, collection_files, transactions — +-- are rebuilt afterwards against the new uploads, the 011 pattern. All child +-- rows are copied before the old tables are dropped; dropping a child table +-- deletes only child rows, which violates no FK, and uploads_old is dropped +-- last, once nothing references it. + +ALTER TABLE uploads RENAME TO uploads_old; + +CREATE TABLE uploads ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + user_id INTEGER NOT NULL REFERENCES users(id), + token_id INTEGER REFERENCES api_tokens(id), + filename TEXT NOT NULL, + original_filename TEXT NOT NULL, + file_size INTEGER NOT NULL, + content_type TEXT NOT NULL DEFAULT '', + visibility TEXT NOT NULL DEFAULT 'private' CHECK (visibility IN ('public', 'private')), + status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'processing', 'completed', 'failed', 'already_stored')), + status_detail TEXT, + datamap_address TEXT, + data_map TEXT, + estimated_cost TEXT, + actual_cost TEXT, + error_message TEXT, + temp_path TEXT, + backoff_until DATETIME, + backoff_attempt INTEGER NOT NULL DEFAULT 0, + last_quoted_cost TEXT, + queued_at DATETIME NOT NULL DEFAULT (datetime('now')), + processing_at DATETIME, + completed_at DATETIME, + failed_at DATETIME, + created_at DATETIME NOT NULL DEFAULT (datetime('now')), + cache_key TEXT +); + +INSERT INTO uploads (id, uuid, user_id, token_id, filename, original_filename, file_size, content_type, visibility, status, + status_detail, datamap_address, data_map, estimated_cost, actual_cost, error_message, temp_path, + backoff_until, backoff_attempt, last_quoted_cost, queued_at, processing_at, completed_at, failed_at, created_at, cache_key) +SELECT id, uuid, user_id, token_id, filename, original_filename, file_size, content_type, visibility, status, + status_detail, datamap_address, data_map, estimated_cost, actual_cost, error_message, temp_path, + backoff_until, backoff_attempt, last_quoted_cost, queued_at, processing_at, completed_at, failed_at, created_at, cache_key +FROM uploads_old; + +CREATE TABLE file_tags_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + upload_id INTEGER NOT NULL REFERENCES uploads(id), + tag_key TEXT NOT NULL, + tag_value TEXT NOT NULL, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO file_tags_new (id, upload_id, tag_key, tag_value, created_at) + SELECT id, upload_id, tag_key, tag_value, created_at FROM file_tags; +DROP TABLE file_tags; +ALTER TABLE file_tags_new RENAME TO file_tags; +CREATE INDEX idx_file_tags_upload_key ON file_tags(upload_id, tag_key); +CREATE INDEX idx_file_tags_key_value ON file_tags(tag_key, tag_value); +CREATE INDEX idx_file_tags_upload_id ON file_tags(upload_id); + +CREATE TABLE collection_files_new ( + collection_id INTEGER NOT NULL REFERENCES collections(id) ON DELETE CASCADE, + upload_id INTEGER NOT NULL REFERENCES uploads(id) ON DELETE CASCADE, + added_at DATETIME NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (collection_id, upload_id) +); +INSERT INTO collection_files_new (collection_id, upload_id, added_at) + SELECT collection_id, upload_id, added_at FROM collection_files; +DROP TABLE collection_files; +ALTER TABLE collection_files_new RENAME TO collection_files; +CREATE INDEX idx_collection_files_upload_id ON collection_files(upload_id); + +CREATE TABLE transactions_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER NOT NULL REFERENCES wallets(id), + upload_id INTEGER REFERENCES uploads(id), + tx_type TEXT NOT NULL, + amount TEXT NOT NULL, + balance_after TEXT NOT NULL, + tx_hash TEXT, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO transactions_new (id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at) + SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions; +DROP TABLE transactions; +ALTER TABLE transactions_new RENAME TO transactions; +CREATE INDEX idx_transactions_wallet_id ON transactions(wallet_id); + +DROP TABLE uploads_old; + +-- Index names are global and follow the renamed table until it is dropped, +-- so the new uploads indexes can only be created now. +CREATE INDEX idx_uploads_user_id ON uploads(user_id); +CREATE INDEX idx_uploads_status ON uploads(status); +CREATE INDEX idx_uploads_uuid ON uploads(uuid); +CREATE INDEX idx_uploads_user_status ON uploads(user_id, status); +CREATE INDEX idx_uploads_backoff ON uploads(status, backoff_until); +CREATE INDEX idx_uploads_status_processing ON uploads(status, processing_at); +CREATE INDEX idx_uploads_cache_key ON uploads(cache_key); + + +-- +goose Down + +-- Reverse rebuild with the original CHECK. Rows already in 'already_stored' +-- are folded into 'completed' (both are terminal stored states; the +-- distinction is a UI nicety). Same child-rebuild dance in reverse. + +ALTER TABLE uploads RENAME TO uploads_old; + +CREATE TABLE uploads ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + uuid TEXT NOT NULL UNIQUE, + user_id INTEGER NOT NULL REFERENCES users(id), + token_id INTEGER REFERENCES api_tokens(id), + filename TEXT NOT NULL, + original_filename TEXT NOT NULL, + file_size INTEGER NOT NULL, + content_type TEXT NOT NULL DEFAULT '', + visibility TEXT NOT NULL DEFAULT 'private' CHECK (visibility IN ('public', 'private')), + status TEXT NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'processing', 'completed', 'failed')), + status_detail TEXT, + datamap_address TEXT, + data_map TEXT, + estimated_cost TEXT, + actual_cost TEXT, + error_message TEXT, + temp_path TEXT, + backoff_until DATETIME, + backoff_attempt INTEGER NOT NULL DEFAULT 0, + last_quoted_cost TEXT, + queued_at DATETIME NOT NULL DEFAULT (datetime('now')), + processing_at DATETIME, + completed_at DATETIME, + failed_at DATETIME, + created_at DATETIME NOT NULL DEFAULT (datetime('now')), + cache_key TEXT +); + +INSERT INTO uploads (id, uuid, user_id, token_id, filename, original_filename, file_size, content_type, visibility, status, + status_detail, datamap_address, data_map, estimated_cost, actual_cost, error_message, temp_path, + backoff_until, backoff_attempt, last_quoted_cost, queued_at, processing_at, completed_at, failed_at, created_at, cache_key) +SELECT id, uuid, user_id, token_id, filename, original_filename, file_size, content_type, visibility, + CASE WHEN status = 'already_stored' THEN 'completed' ELSE status END, + status_detail, datamap_address, data_map, estimated_cost, actual_cost, error_message, temp_path, + backoff_until, backoff_attempt, last_quoted_cost, queued_at, processing_at, completed_at, failed_at, created_at, cache_key +FROM uploads_old; + +CREATE TABLE file_tags_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + upload_id INTEGER NOT NULL REFERENCES uploads(id), + tag_key TEXT NOT NULL, + tag_value TEXT NOT NULL, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO file_tags_new (id, upload_id, tag_key, tag_value, created_at) + SELECT id, upload_id, tag_key, tag_value, created_at FROM file_tags; +DROP TABLE file_tags; +ALTER TABLE file_tags_new RENAME TO file_tags; +CREATE INDEX idx_file_tags_upload_key ON file_tags(upload_id, tag_key); +CREATE INDEX idx_file_tags_key_value ON file_tags(tag_key, tag_value); +CREATE INDEX idx_file_tags_upload_id ON file_tags(upload_id); + +CREATE TABLE collection_files_new ( + collection_id INTEGER NOT NULL REFERENCES collections(id) ON DELETE CASCADE, + upload_id INTEGER NOT NULL REFERENCES uploads(id) ON DELETE CASCADE, + added_at DATETIME NOT NULL DEFAULT (datetime('now')), + PRIMARY KEY (collection_id, upload_id) +); +INSERT INTO collection_files_new (collection_id, upload_id, added_at) + SELECT collection_id, upload_id, added_at FROM collection_files; +DROP TABLE collection_files; +ALTER TABLE collection_files_new RENAME TO collection_files; +CREATE INDEX idx_collection_files_upload_id ON collection_files(upload_id); + +CREATE TABLE transactions_new ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + wallet_id INTEGER NOT NULL REFERENCES wallets(id), + upload_id INTEGER REFERENCES uploads(id), + tx_type TEXT NOT NULL, + amount TEXT NOT NULL, + balance_after TEXT NOT NULL, + tx_hash TEXT, + created_at DATETIME NOT NULL DEFAULT (datetime('now')) +); +INSERT INTO transactions_new (id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at) + SELECT id, wallet_id, upload_id, tx_type, amount, balance_after, tx_hash, created_at FROM transactions; +DROP TABLE transactions; +ALTER TABLE transactions_new RENAME TO transactions; +CREATE INDEX idx_transactions_wallet_id ON transactions(wallet_id); + +DROP TABLE uploads_old; + +-- Index names are global and follow the renamed table until it is dropped, +-- so the new uploads indexes can only be created now. +CREATE INDEX idx_uploads_user_id ON uploads(user_id); +CREATE INDEX idx_uploads_status ON uploads(status); +CREATE INDEX idx_uploads_uuid ON uploads(uuid); +CREATE INDEX idx_uploads_user_status ON uploads(user_id, status); +CREATE INDEX idx_uploads_backoff ON uploads(status, backoff_until); +CREATE INDEX idx_uploads_status_processing ON uploads(status, processing_at); +CREATE INDEX idx_uploads_cache_key ON uploads(cache_key); + diff --git a/internal/services/upload.go b/internal/services/upload.go index 98a347a..c90cc63 100644 --- a/internal/services/upload.go +++ b/internal/services/upload.go @@ -26,7 +26,7 @@ type Upload struct { FileSize int64 ContentType string Visibility string // "public" or "private" - Status string // "queued", "processing", "completed", "failed" + Status string // "queued", "processing", "completed", "failed", "already_stored" StatusDetail sql.NullString // substatus: "gas_backoff", etc. DatamapAddress sql.NullString EstimatedCost sql.NullString @@ -174,7 +174,7 @@ type UploadListOptions struct { Offset int SortBy string // "created_at" (default), "file_size", "filename", "status" SortOrder string // "desc" (default), "asc" - Status string // filter: "queued", "processing", "completed", "failed", "" (all) + Status string // filter: "queued", "processing", "completed", "failed", "already_stored", "" (all) From *time.Time To *time.Time } @@ -585,7 +585,7 @@ func (s *UploadService) Delete(id int64) error { 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") + return errors.New("only failed, completed, or already-stored uploads can be deleted") } return err } @@ -603,14 +603,14 @@ func (s *UploadService) Delete(id int64) error { } result, err := tx.Exec( - `DELETE FROM uploads WHERE id = ? AND status IN ('failed', 'completed')`, + `DELETE FROM uploads WHERE id = ? AND status IN ('failed', 'completed', 'already_stored')`, id, ) if err != nil { return err } if n, _ := result.RowsAffected(); n == 0 { - return errors.New("only failed or completed uploads can be deleted") + return errors.New("only failed, completed, or already-stored uploads can be deleted") } return tx.Commit() } diff --git a/internal/services/upload_already_stored_test.go b/internal/services/upload_already_stored_test.go new file mode 100644 index 0000000..6506deb --- /dev/null +++ b/internal/services/upload_already_stored_test.go @@ -0,0 +1,65 @@ +package services + +import ( + "testing" + + "github.com/WithAutonomi/indelible/internal/downloadcache" +) + +// V2-875: 'already_stored' (the V2-399 dedup status) was written by the +// worker but absent from the schema's status CHECK constraint — on any +// database built from the migrations, the transition itself failed. These +// tests run against the freshly-migrated schema in both CI dialects, so they +// are the direct regression: they fail on the pre-014 constraint. + +func TestMarkAlreadyStoredPersists(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "dedup@example.com", "D", "P") + svc := NewUploadService(db) + + private := createTestUpload(t, svc, user.ID, "dedup-private.bin", 10) + if err := svc.MarkAlreadyStored(private.ID, "dm-dedup-private", "0"); err != nil { + t.Fatalf("MarkAlreadyStored violated the schema: %v", err) + } + got, err := svc.GetByID(private.ID) + if err != nil || got.Status != "already_stored" { + t.Fatalf("status = %q (err=%v), want already_stored", got.Status, err) + } + if key := cacheKeyOf(t, svc, private.ID); key != downloadcache.KeyForIdentifier("dm-dedup-private") { + t.Fatalf("already_stored must stamp cache_key; got %q", key) + } + + public := createTestUpload(t, svc, user.ID, "dedup-public.bin", 10) + if err := svc.MarkAlreadyStoredPublic(public.ID, "addr-dedup-public", "0"); err != nil { + t.Fatalf("MarkAlreadyStoredPublic violated the schema: %v", err) + } + if got, err := svc.GetByID(public.ID); err != nil || got.Status != "already_stored" { + t.Fatalf("public status = %q (err=%v), want already_stored", got.Status, err) + } +} + +// The second half of V2-875: already_stored uploads were undeletable (the +// delete's status list omitted them), which since V2-824/873 also meant no +// erasure path for their DataMap row or cached bytes. +func TestDeleteAlreadyStoredUpload(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "dedup-del@example.com", "D", "D") + svc := NewUploadService(db) + + u := createTestUpload(t, svc, user.ID, "dedup-del.bin", 10) + if err := svc.MarkAlreadyStored(u.ID, "dm-dedup-del", "0"); err != nil { + t.Fatalf("MarkAlreadyStored: %v", err) + } + + if err := svc.Delete(u.ID); err != nil { + t.Fatalf("already_stored upload must be deletable: %v", err) + } + if _, err := svc.GetByID(u.ID); err != ErrUploadNotFound { + t.Fatalf("row survived the delete: %v", err) + } + // The purge fan-out ran like any other delete. + entries, err := svc.PurgeLogSince(0, 10) + if err != nil || len(entries) != 1 || entries[0].CacheKey != downloadcache.KeyForIdentifier("dm-dedup-del") { + t.Fatalf("purge log = %+v (err=%v), want the dedup key exactly once", entries, err) + } +} From 133e9bcc0ef679573d91794d3b9ff12e6228efec Mon Sep 17 00:00:00 2001 From: Nic Date: Thu, 6 Aug 2026 13:34:20 +0100 Subject: [PATCH 6/6] docs+test: changelog entry with the pre-014 retry note; public dedup delete test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups from #156: operators with pre-014 databases have dedup re-uploads stuck in failed state that migration 014 does not rewrite — one manual Retry completes them at zero cost. Plus the public-variant delete/purge-fan-out regression. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 6 +++++ .../services/upload_already_stored_test.go | 24 +++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 796d4be..fa28703 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ All notable changes to Indelible are documented in this file. This project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Fixed +- **Duplicate re-uploads no longer fail at the final step.** The `already_stored` status (content-addressed dedup — re-uploading a file whose chunks are all already on the network) was written by the upload worker but missing from the database schema's status constraint, so the transition was rejected after the network store succeeded and the upload surfaced as "Failed to save upload record". Migration 014 fixes the constraint in both dialects, and such uploads are now also deletable like any other stored upload. +- **Operator note:** uploads that hit this bug before the fix sit in `failed` state; migration 014 does not rewrite them. Retry them once (the re-Prepare is a zero-cost dedup) and they will complete as `already_stored`. + ## [0.11.0] - 2026-06-18 This release tracks **antd / ant-sdk v0.10.0** (bundled daemon, Go client module, and image). Note that `v0.10.0` is the *antd* version; the Indelible release is `v0.11.0`. diff --git a/internal/services/upload_already_stored_test.go b/internal/services/upload_already_stored_test.go index 6506deb..299d0aa 100644 --- a/internal/services/upload_already_stored_test.go +++ b/internal/services/upload_already_stored_test.go @@ -63,3 +63,27 @@ func TestDeleteAlreadyStoredUpload(t *testing.T) { t.Fatalf("purge log = %+v (err=%v), want the dedup key exactly once", entries, err) } } + +// Review follow-up: the public variant's delete → purge fan-out, keyed on the +// network-address derivation rather than the DataMap. +func TestDeleteAlreadyStoredPublicUpload(t *testing.T) { + db := setupTestDB(t) + user := createTestUser(t, NewUserService(db), "dedup-del-pub@example.com", "D", "P") + svc := NewUploadService(db) + + u := createTestUpload(t, svc, user.ID, "dedup-del-pub.bin", 10) + if err := svc.MarkAlreadyStoredPublic(u.ID, "addr-dedup-del-pub", "0"); err != nil { + t.Fatalf("MarkAlreadyStoredPublic: %v", err) + } + + if err := svc.Delete(u.ID); err != nil { + t.Fatalf("public already_stored upload must be deletable: %v", err) + } + if _, err := svc.GetByID(u.ID); err != ErrUploadNotFound { + t.Fatalf("row survived the delete: %v", err) + } + entries, err := svc.PurgeLogSince(0, 10) + if err != nil || len(entries) != 1 || entries[0].CacheKey != downloadcache.KeyForIdentifier("addr-dedup-del-pub") { + t.Fatalf("purge log = %+v (err=%v), want the address-derived key exactly once", entries, err) + } +}