fix: admit already_stored to the uploads status constraint and delete path

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`.

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'));

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);
+

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()
 }

internal/services/upload_already_stored_test.go

@@ -0,0 +1,89 @@
+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)
+	}
+}
+
+// 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)
+	}
+}