From 232aef43651e85bac7634e9aa903cbe52dcd0caa Mon Sep 17 00:00:00 2001 From: xiaocui-big Date: Tue, 15 Sep 2026 23:16:05 +0800 Subject: [PATCH] fix(folders): clean up blob storage on recursive folder delete (#177) Recursive folder delete removed DB rows but left every underlying object in bucket storage. The code had an explicit TODO acknowledging this. Changes: - Add ObjectStore and logger options to folder.Service (WithStore, WithLogger) - Collect storage keys before deleting files, then delete blobs after the transaction commits (best-effort, matching file.Delete's shape) - Log failed blob deletions at WARN level so operators can see orphans - Update memd wiring to pass store and logger to folder service - Add integration tests verifying blob cleanup and backward compatibility - Document object retention behavior in DEPLOYMENT.md - Add CHANGELOG entry The crash window (process killed between commit and blob delete) is documented as a known gap. A reaper would need to record keys whose delete was never attempted, since storage.Store has no listing. --- CHANGELOG.md | 10 + docs/DEPLOYMENT.md | 23 ++ server/cmd/memd/main.go | 5 +- .../folder/delete_integration_test.go | 253 ++++++++++++++++++ server/internal/folder/folder.go | 104 ++++++- 5 files changed, 388 insertions(+), 7 deletions(-) create mode 100644 server/internal/folder/delete_integration_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bdac57..5daf1ee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,16 @@ The project publishes 0.x prerelease versions; a stable release line is not yet repository coordinate the installer itself uses, so the next rename cannot leave a stale identifier behind unnoticed. +### Fixed + +- Recursive folder delete now removes the corresponding objects from bucket + storage after the database transaction commits (`#177`). Previously, the DB + rows were deleted but the blobs remained orphaned in the bucket. The cleanup + is best-effort: a failed object delete does not roll back the folder delete, + and failures are logged at WARN level so the operator can see which keys + remain. The folder service now accepts an optional `ObjectStore` and logger + via `folder.WithStore` and `folder.WithLogger` options. + ### Security - Normalize the client-declared MIME type of a stored file before deciding how diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 25d0b57..698a94e 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -261,6 +261,29 @@ Redis AOF protects normal restarts but is not in the portable backup. A restore therefore starts with an empty queue/replay window. Requeue or reindex any file whose processing did not reach a terminal state before the backup. +### Object storage retention + +Object keys are per-file by construction: each key embeds the row's own file ID +(`users///`), so deleting one row's key cannot +remove another row's bytes. No reference counting is needed. + +When a file or folder is deleted, the database row is removed first, then the +corresponding object is deleted from bucket storage on a best-effort basis. A +failed object delete does not roll back the database change; the orphaned key +is logged at `WARN` level so the operator can see which keys remain. + +**Crash window**: if the process is killed after the database transaction +commits but before the blob delete lands, the object remains in the bucket +permanently. There is currently no reaper or garbage-collection pass to sweep +these residues. The server has no listing capability against the bucket (the +`storage.Store` interface exposes only `Put`/`Get`/`Delete`), so a reaper would +need to record keys whose delete was never attempted. This is a known gap; +operators should monitor bucket growth against expected database row counts. + +To manually reconcile, compare the bucket contents against the `files` table's +`storage_key` column. Objects present in the bucket but absent from the database +are safe to delete — they cannot be referenced by any live row. + ### Restore drill Restore only into an empty installation. The script verifies every checksum diff --git a/server/cmd/memd/main.go b/server/cmd/memd/main.go index 3e08444..af4d519 100644 --- a/server/cmd/memd/main.go +++ b/server/cmd/memd/main.go @@ -119,7 +119,10 @@ func run() error { "providers", cfg.ManagedEmbeddingProviders, ) } - folderSvc := folder.New(database.Pool) + folderSvc := folder.New(database.Pool, + folder.WithStore(store), + folder.WithLogger(logger), + ) fileSvc := file.New(database.Pool, store, folderSvc) memorySvc := memory.New(database.Pool) durableContextSvc := durablecontext.New(database.Pool, memorySvc) diff --git a/server/internal/folder/delete_integration_test.go b/server/internal/folder/delete_integration_test.go new file mode 100644 index 0000000..6630b6e --- /dev/null +++ b/server/internal/folder/delete_integration_test.go @@ -0,0 +1,253 @@ +package folder + +import ( + "bytes" + "context" + "io" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + + memdb "github.com/PeterGuy326/mem/server/internal/db" +) + +// trackingObjectStore records which keys are deleted so tests can assert that +// recursive folder delete cleans up blobs. +type trackingObjectStore struct { + mu sync.Mutex + objects map[string]bool + deleted []string +} + +func newTrackingObjectStore() *trackingObjectStore { + return &trackingObjectStore{objects: make(map[string]bool)} +} + +func (s *trackingObjectStore) Put(_ context.Context, key string, _ io.Reader, _ int64, _ string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.objects[key] = true + return nil +} + +func (s *trackingObjectStore) Get(_ context.Context, key string) (io.ReadCloser, error) { + s.mu.Lock() + defer s.mu.Unlock() + if !s.objects[key] { + return nil, &objectNotFoundError{key: key} + } + return io.NopCloser(bytes.NewReader(nil)), nil +} + +func (s *trackingObjectStore) Delete(_ context.Context, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.objects, key) + s.deleted = append(s.deleted, key) + return nil +} + +func (s *trackingObjectStore) has(key string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.objects[key] +} + +func (s *trackingObjectStore) deleteCount() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.deleted) +} + +type objectNotFoundError struct{ key string } + +func (e *objectNotFoundError) Error() string { return "object not found: " + e.key } + +// TestRecursiveDeleteCleansUpBlobs verifies that recursive folder delete +// removes objects from the store after the DB rows are deleted. +// +// MEM_TEST_DB=postgres://mem:mem@localhost:5432/mem_test?sslmode=disable \ +// go test ./internal/folder -run TestRecursiveDeleteCleansUpBlobs +func TestRecursiveDeleteCleansUpBlobs(t *testing.T) { + dsn := os.Getenv("MEM_TEST_DB") + if dsn == "" { + t.Skip("MEM_TEST_DB not set; skipping DB integration test") + } + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatalf("parse MEM_TEST_DB: %v", err) + } + if !strings.HasSuffix(config.ConnConfig.Database, "_test") { + t.Fatalf( + "refusing to modify non-test database %q; MEM_TEST_DB must end in _test", + config.ConnConfig.Database, + ) + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + database, err := memdb.Open(ctx, dsn) + if err != nil { + t.Fatalf("open test database: %v", err) + } + t.Cleanup(database.Close) + if err := database.Migrate(ctx); err != nil { + t.Fatalf("migrate test database: %v", err) + } + + userID := createFolderDeleteTenant(t, ctx, database.Pool) + store := newTrackingObjectStore() + service := New(database.Pool, WithStore(store)) + + // Create a folder hierarchy with files. + if _, err := service.Create(ctx, userID, "/Project/Sub"); err != nil { + t.Fatalf("create folders: %v", err) + } + + // Insert files with storage keys that the store tracks. + fileKeys := []string{ + "users/" + userID.String() + "/" + uuid.NewString() + "/a.txt", + "users/" + userID.String() + "/" + uuid.NewString() + "/b.txt", + "users/" + userID.String() + "/" + uuid.NewString() + "/c.txt", + } + for i, key := range fileKeys { + if err := store.Put(ctx, key, nil, 0, "text/plain"); err != nil { + t.Fatalf("put object: %v", err) + } + paths := []string{"/Project", "/Project", "/Project/Sub"} + names := []string{"a.txt", "b.txt", "c.txt"} + if _, err := database.Pool.Exec(ctx, ` + INSERT INTO files (id, user_id, name, path, size, sha256, mime, storage_key, index_status) + VALUES ($1, $2, $3, $4, 0, $5, 'text/plain', $6, 'ready') + `, uuid.New(), userID, names[i], paths[i], strings.Repeat("x", 64), key); err != nil { + t.Fatalf("insert file: %v", err) + } + } + + // Verify objects exist before delete. + for _, key := range fileKeys { + if !store.has(key) { + t.Fatalf("object %s should exist before delete", key) + } + } + + // Recursive delete. + if err := service.Delete(ctx, userID, "/Project", true); err != nil { + t.Fatalf("recursive delete: %v", err) + } + + // Verify all objects were deleted from the store. + for _, key := range fileKeys { + if store.has(key) { + t.Errorf("object %s should have been deleted from store", key) + } + } + if store.deleteCount() != len(fileKeys) { + t.Errorf("delete calls = %d, want %d", store.deleteCount(), len(fileKeys)) + } + + // Verify DB rows are gone. + var fileCount int + if err := database.Pool.QueryRow(ctx, ` + SELECT count(*) FROM files WHERE user_id = $1 + `, userID).Scan(&fileCount); err != nil { + t.Fatalf("count files: %v", err) + } + if fileCount != 0 { + t.Errorf("files remaining = %d, want 0", fileCount) + } + + var folderCount int + if err := database.Pool.QueryRow(ctx, ` + SELECT count(*) FROM folders WHERE user_id = $1 + `, userID).Scan(&folderCount); err != nil { + t.Fatalf("count folders: %v", err) + } + if folderCount != 0 { + t.Errorf("folders remaining = %d, want 0", folderCount) + } +} + +// TestRecursiveDeleteWithoutStore verifies that recursive delete works without +// a store configured (backward compatibility — DB rows are deleted but blobs +// are not cleaned up). +func TestRecursiveDeleteWithoutStore(t *testing.T) { + dsn := os.Getenv("MEM_TEST_DB") + if dsn == "" { + t.Skip("MEM_TEST_DB not set; skipping DB integration test") + } + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatalf("parse MEM_TEST_DB: %v", err) + } + if !strings.HasSuffix(config.ConnConfig.Database, "_test") { + t.Fatalf( + "refusing to modify non-test database %q; MEM_TEST_DB must end in _test", + config.ConnConfig.Database, + ) + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + database, err := memdb.Open(ctx, dsn) + if err != nil { + t.Fatalf("open test database: %v", err) + } + t.Cleanup(database.Close) + if err := database.Migrate(ctx); err != nil { + t.Fatalf("migrate test database: %v", err) + } + + userID := createFolderDeleteTenant(t, ctx, database.Pool) + service := New(database.Pool) // no store + + if _, err := service.Create(ctx, userID, "/Orphan"); err != nil { + t.Fatalf("create folder: %v", err) + } + key := "users/" + userID.String() + "/" + uuid.NewString() + "/orphan.txt" + if _, err := database.Pool.Exec(ctx, ` + INSERT INTO files (id, user_id, name, path, size, sha256, mime, storage_key, index_status) + VALUES ($1, $2, 'orphan.txt', '/Orphan', 0, $3, 'text/plain', $4, 'ready') + `, uuid.New(), userID, strings.Repeat("y", 64), key); err != nil { + t.Fatalf("insert file: %v", err) + } + + // Delete should succeed even without a store. + if err := service.Delete(ctx, userID, "/Orphan", true); err != nil { + t.Fatalf("recursive delete without store: %v", err) + } + + // DB row should be gone. + var fileCount int + if err := database.Pool.QueryRow(ctx, ` + SELECT count(*) FROM files WHERE user_id = $1 + `, userID).Scan(&fileCount); err != nil { + t.Fatalf("count files: %v", err) + } + if fileCount != 0 { + t.Errorf("files remaining = %d, want 0", fileCount) + } +} + +func createFolderDeleteTenant(t *testing.T, ctx context.Context, pool *pgxpool.Pool) uuid.UUID { + t.Helper() + var userID uuid.UUID + if err := pool.QueryRow(ctx, ` + INSERT INTO users (email, password_hash) + VALUES ($1, 'folder-delete-test') + RETURNING id + `, "folder-delete-"+uuid.NewString()+"@example.com").Scan(&userID); err != nil { + t.Fatalf("create user: %v", err) + } + t.Cleanup(func() { + cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, _ = pool.Exec(cleanupCtx, `DELETE FROM users WHERE id = $1`, userID) + }) + return userID +} diff --git a/server/internal/folder/folder.go b/server/internal/folder/folder.go index 0dd0469..f5571fb 100644 --- a/server/internal/folder/folder.go +++ b/server/internal/folder/folder.go @@ -13,6 +13,7 @@ import ( "context" "errors" "fmt" + "log/slog" "strings" "time" @@ -58,13 +59,40 @@ type Node struct { Children []*Node `json:"children,omitempty"` } +// ObjectStore is the subset of storage.Store that folder needs for blob cleanup. +type ObjectStore interface { + Delete(context.Context, string) error +} + // Service is the folder service. type Service struct { - pool *pgxpool.Pool + pool *pgxpool.Pool + store ObjectStore + logger *slog.Logger +} + +// Option configures a folder Service. +type Option func(*Service) + +// WithStore sets the object store used for blob cleanup on recursive delete. +// If not set, recursive delete removes DB rows but leaves orphan blobs. +func WithStore(store ObjectStore) Option { + return func(s *Service) { s.store = store } +} + +// WithLogger sets the logger for reporting best-effort blob cleanup failures. +func WithLogger(logger *slog.Logger) Option { + return func(s *Service) { s.logger = logger } } // New constructs a folder Service. -func New(pool *pgxpool.Pool) *Service { return &Service{pool: pool} } +func New(pool *pgxpool.Pool, opts ...Option) *Service { + s := &Service{pool: pool} + for _, opt := range opts { + opt(s) + } + return s +} // Sentinel errors. var ( @@ -586,9 +614,10 @@ func rewritePrefixTx(ctx context.Context, tx pgx.Tx, userID, srcID uuid.UUID, ol // // - recursive=false (default): folder must be empty (no subfolders, no files) // or ErrNotEmpty is returned. -// - recursive=true: subfolders + files are deleted from the DB. S3 cleanup -// is TODO — for now we only purge the DB rows; orphan blobs will be -// reaped by a future garbage-collection pass. +// - recursive=true: subfolders + files are deleted from the DB, and their +// blobs are removed from object storage on a best-effort basis after the +// transaction commits. A failed blob removal does not roll back the DB +// delete; orphaned keys are logged so the operator can see them. func (s *Service) Delete(ctx context.Context, userID uuid.UUID, path string, recursive bool) error { norm, err := pathx.Normalize(path) if err != nil { @@ -597,7 +626,8 @@ func (s *Service) Delete(ctx context.Context, userID uuid.UUID, path string, rec if norm == pathx.Root { return ErrRootOp } - return s.withPathMutationTx(ctx, userID, func(tx pgx.Tx) error { + var orphanKeys []string + err = s.withPathMutationTx(ctx, userID, func(tx pgx.Tx) error { src, err := selectFolderByPathTx(ctx, tx, userID, norm) if err != nil { return err @@ -628,6 +658,14 @@ func (s *Service) Delete(ctx context.Context, userID uuid.UUID, path string, rec // explicit forget operation first. return ErrContainsMemories } + // Collect storage keys before deleting rows so we can clean up + // blobs after the transaction commits. Keys are per-row by + // construction (see 0013_file_content_identity.sql), so deleting + // each key cannot destroy another row's bytes. + orphanKeys, err = collectStorageKeysTx(ctx, tx, userID, src.ID, src.Path) + if err != nil { + return fmt.Errorf("collect storage keys for recursive delete: %w", err) + } // Hard delete: remove all descendant files first (FKs cascade // from folders → files would only NULL out folder_id, so we have // to delete files explicitly). @@ -648,6 +686,60 @@ func (s *Service) Delete(ctx context.Context, userID uuid.UUID, path string, rec } return nil }) + if err != nil { + return err + } + // Blob cleanup happens after the transaction commits: a failed object + // delete must not roll back the user's folder delete. This matches the + // best-effort shape of file.Delete. + if len(orphanKeys) > 0 && s.store != nil { + s.cleanupBlobs(ctx, orphanKeys) + } + return nil +} + +// collectStorageKeysTx returns the storage_key values for all files that will +// be deleted by a recursive folder delete. This must be called before the +// DELETE FROM files so the keys are available for post-commit blob cleanup. +func collectStorageKeysTx(ctx context.Context, tx pgx.Tx, userID uuid.UUID, folderID uuid.UUID, folderPath string) ([]string, error) { + rows, err := tx.Query(ctx, + `SELECT storage_key FROM files + WHERE user_id = $1 + AND (folder_id = $2 OR path = $3 + OR left(path, length($3) + 1) = $3 || '/') + AND storage_key <> ''`, + userID, folderID, folderPath) + if err != nil { + return nil, err + } + defer rows.Close() + var keys []string + for rows.Next() { + var key string + if err := rows.Scan(&key); err != nil { + return nil, err + } + keys = append(keys, key) + } + return keys, rows.Err() +} + +// cleanupBlobs removes objects from the store on a best-effort basis. Failures +// are logged so the operator can see which keys remain; they do not propagate +// to the caller because the DB rows are already gone. +func (s *Service) cleanupBlobs(ctx context.Context, keys []string) { + cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) + defer cancel() + for _, key := range keys { + if err := s.store.Delete(cleanupCtx, key); err != nil { + if s.logger != nil { + s.logger.Warn("folder delete: blob cleanup failed", + "storage_key", key, + "error", err, + ) + } + } + } } func isEmptyTx(ctx context.Context, tx pgx.Tx, userID, folderID uuid.UUID, path string) (bool, error) {