Fix bug when renaming multiple objects

internal/sql/migrations.go

@@ -13,11 +13,6 @@ import (
 )
 
 type (
- Migration struct {
- ID string
- Migrate func(tx Tx) error
- }
-
  // Migrator is an interface for defining database-specific helper methods
  // required during migrations
  Migrator interface {
@@ -29,8 +24,19 @@ type (
  MainMigrator interface {
  Migrator
  MakeDirsForPath(ctx context.Context, tx Tx, path string) (int64, error)
+ ObjectsWithCorruptedDirectoryID(ctx context.Context, tx Tx) ([]Object, error)
  UpdateSetting(ctx context.Context, tx Tx, key, value string) error
  }
+
+ Migration struct {
+ ID string
+ Migrate func(tx Tx) error
+ }
+
+ Object struct {
+ ID uint
+ ObjectID string
+ }
 )
 
 var (
@@ -255,6 +261,46 @@ var (
  return performMigration(ctx, tx, migrationsFs, dbIdentifier, "00023_key_prefix", log)
  },
  },
+ {
+ ID: "00024_fix_directories",
+ Migrate: func(tx Tx) error {
+ log.Info("performing main migration '00024_fix_directories'")
+
+ // fetch corrupted objects
+ objects, err := m.ObjectsWithCorruptedDirectoryID(ctx, tx)
+ if err != nil {
+ return err
+ }
+ log.Infof("found %d objects with a corrupted path", len(objects))
+
+ // prepare update stmt
+ updateStmt, err := tx.Prepare(ctx, "UPDATE objects SET db_directory_id = ? WHERE id = ?")
+ if err != nil {
+ return fmt.Errorf("failed to prepare update statement, %w", err)
+ }
+ defer updateStmt.Close()
+
+ // loop every object and re-insert its directory
+ for _, o := range objects {
+ log.Debugf("re-inserting directory for object %v (%v)", o.ObjectID, o.ID)
+
+ // recreate dirs
+ dirID, err := m.MakeDirsForPath(ctx, tx, o.ObjectID)
+ if err != nil {
+ return fmt.Errorf("failed to create directory %s: %w", o.ObjectID, err)
+ }
+
+ // update object
+ _, err = updateStmt.Exec(ctx, dirID, o.ID)
+ if err != nil {
+ return fmt.Errorf("failed to execute update statement, %w", err)
+ }
+ }
+
+ log.Info("migration '00024_fix_directories' complete")
+ return nil
+ },
+ },
  }
  }
  MetricsMigrations = func(ctx context.Context, migrationsFs embed.FS, log *zap.SugaredLogger) []Migration {

internal/test/e2e/cluster_test.go

@@ -454,7 +454,7 @@ func TestObjectsWithDelimiterSlash(t *testing.T) {
  // assert mime type
  isDir := strings.HasSuffix(entries[i].Key, "/") && entries[i].Key != "//double/" // double is a file
  if (isDir && entries[i].MimeType != "") || (!isDir && entries[i].MimeType == "") {
- t.Fatal("unexpected mime type", entries[i].MimeType)
+ t.Fatal("unexpected mime type", entries[i], entries[i].MimeType)
  }
  entries[i].MimeType = ""
 

object/path.go

@@ -0,0 +1,26 @@
+package object
+
+import "strings"
+
+// Directories returns the directories for the given path. When explicit is
+// true, the returned directories do not include the path itself should it be a
+// directory. The root path ('/') is always excluded.
+func Directories(path string, explicit bool) (dirs []string) {
+ if explicit {
+ path = strings.TrimSuffix(path, "/")
+ }
+ if path == "/" {
+ return nil
+ }
+ for i, r := range path {
+ if r != '/' {
+ continue
+ }
+ dir := path[:i+1]
+ if dir == "/" {
+ continue
+ }
+ dirs = append(dirs, dir)
+ }
+ return
+}

object/path_test.go

@@ -0,0 +1,28 @@
+package object
+
+import (
+ "reflect"
+ "testing"
+)
+
+func TestDirectories(t *testing.T) {
+ cases := []struct {
+ path string
+ explicit bool
+ dirs []string
+ }{
+ {"/", true, nil},
+ {"/", false, nil},
+ {"/foo", true, nil},
+ {"/foo", false, nil},
+ {"/foo/bar", true, []string{"/foo/"}},
+ {"/foo/bar", false, []string{"/foo/"}},
+ {"/foo/bar/", true, []string{"/foo/"}},
+ {"/foo/bar/", false, []string{"/foo/", "/foo/bar/"}},
+ }
+ for _, c := range cases {
+ if got := Directories(c.path, c.explicit); !reflect.DeepEqual(got, c.dirs) {
+ t.Fatalf("unexpected dirs for path %v (explicit %t), %v != %v", c.path, c.explicit, got, c.dirs)
+ }
+ }
+}

stores/metadata.go

@@ -400,7 +400,7 @@ func (s *SQLStore) RecordContractSpending(ctx context.Context, records []api.Con
 func (s *SQLStore) RenameObject(ctx context.Context, bucket, keyOld, keyNew string, force bool) error {
  return s.db.Transaction(ctx, func(tx sql.DatabaseTx) error {
  // create new dir
- dirID, err := tx.MakeDirsForPath(ctx, keyNew)
+ dirID, err := tx.InsertDirectories(ctx, object.Directories(keyNew, true))
  if err != nil {
  return err
  }
@@ -418,10 +418,10 @@ func (s *SQLStore) RenameObject(ctx context.Context, bucket, keyOld, keyNew stri
 func (s *SQLStore) RenameObjects(ctx context.Context, bucket, prefixOld, prefixNew string, force bool) error {
  return s.db.Transaction(ctx, func(tx sql.DatabaseTx) error {
  // create new dir
- dirID, err := tx.MakeDirsForPath(ctx, prefixNew)
+ dirID, renamedIDs, err := tx.InsertDirectoriesForRename(ctx, prefixOld, prefixNew)
  if err != nil {
  return fmt.Errorf("RenameObjects: failed to create new directory: %w", err)
- } else if err := tx.RenameObjects(ctx, bucket, prefixOld, prefixNew, dirID, force); err != nil {
+ } else if err := tx.RenameObjects(ctx, bucket, prefixOld, prefixNew, dirID, renamedIDs, force); err != nil {
  return err
  }
  // prune old dirs
@@ -491,7 +491,7 @@ func (s *SQLStore) UpdateObject(ctx context.Context, bucket, key, contractSet, e
  }
 
  // create the dir
- dirID, err := tx.MakeDirsForPath(ctx, key)
+ dirID, err := tx.InsertDirectories(ctx, object.Directories(key, true))
  if err != nil {
  return fmt.Errorf("failed to create directories for key '%s': %w", key, err)
  }