Keep pending uploads and server-side renames when an item's local ID is replaced

fs/cache.go

@@ -313,9 +313,16 @@ func (f *Filesystem) InsertChild(parentID string, child *Inode) uint64 {
 	return f.InsertID(id, child)
 }
 
-// DeleteID deletes an item from the cache, and removes it from its parent. Must
-// be called before InsertID if being used to rename/move an item.
+// DeleteID deletes an item from the cache, removes it from its parent and
+// cancels any pending upload of it. Must be called before InsertID if being
+// used to rename/move an item.
 func (f *Filesystem) DeleteID(id string) {
+	f.deleteID(id)
+	f.uploads.CancelUpload(id)
+}
+
+// deleteID is DeleteID without cancelling the item's uploads.
+func (f *Filesystem) deleteID(id string) {
 	if inode := f.GetID(id); inode != nil {
 		parent := f.GetID(inode.ParentID())
 		parent.Lock()
@@ -331,7 +338,6 @@ func (f *Filesystem) DeleteID(id string) {
 		parent.Unlock()
 	}
 	f.metadata.Delete(id)
-	f.uploads.CancelUpload(id)
 }
 
 // GetChild fetches a named child of an item. Wraps GetChildrenID.
@@ -506,8 +512,18 @@ func (f *Filesystem) InsertPath(key string, auth *graph.Auth, inode *Inode) (uin
 }
 
 // MoveID moves an item to a new ID name. Also responsible for handling the
-// actual overwrite of the item's IDInternal field
+// actual overwrite of the item's IDInternal field. A pending upload of the item
+// is moved to the new ID as well, it must not be cancelled: it can contain
+// local changes that have not been uploaded yet.
 func (f *Filesystem) MoveID(oldID string, newID string) error {
+	if f.uploads != nil && oldID != newID {
+		f.uploads.MoveUpload(oldID, newID)
+	}
+	return f.moveID(oldID, newID)
+}
+
+// moveID is MoveID without moving the item's uploads.
+func (f *Filesystem) moveID(oldID string, newID string) error {
 	inode := f.GetID(oldID)
 	if inode == nil {
 		// It may have already been renamed. This is not an error. We assume
@@ -518,19 +534,33 @@ func (f *Filesystem) MoveID(oldID string, newID string) error {
 		}
 	}
 
-	// need to rename the child under the parent
-	parent := f.GetID(inode.ParentID())
-	parent.Lock()
-	for i, child := range parent.children {
-		if child == oldID {
-			parent.children[i] = newID
-			break
+	f.deleteID(oldID)
+
+	// A delta may already have added a second inode for the new ID, if the
+	// item was renamed or moved on the server before we learned its new ID
+	// and so could not be matched by name. That inode has the item's current
+	// name and parent on the server: apply them to ours and drop it.
+	if entry, exists := f.metadata.Load(newID); exists && entry.(*Inode) != inode {
+		duplicate := entry.(*Inode)
+		name := duplicate.Name()
+		parentID := duplicate.ParentID()
+		f.deleteID(newID)
+
+		inode.Lock()
+		inode.DriveItem.Name = name
+		if inode.DriveItem.Parent != nil && parentID != "" {
+			inode.DriveItem.Parent.ID = parentID
 		}
+		inode.Unlock()
+		log.Info().
+			Str("oldID", oldID).
+			Str("id", newID).
+			Str("name", name).
+			Str("parentID", parentID).
+			Msg("Item already existed under its new ID, keeping its server-side name and parent.")
 	}
-	parent.Unlock()
 
 	// now actually perform the metadata+content move
-	f.DeleteID(oldID)
 	f.InsertID(newID, inode)
 	if inode.IsDir() {
 		return nil

fs/move_id_offline_test.go

@@ -0,0 +1,185 @@
+package fs
+
+import (
+	"encoding/json"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/hanwen/go-fuse/v2/fuse"
+	"github.com/jstaf/onedriver/fs/graph"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	bolt "go.etcd.io/bbolt"
+)
+
+// newOfflineFilesystem creates a Filesystem with only a root folder, backed by
+// a temporary cache directory. Its UploadManager ticks at the given interval
+// and has an empty auth, so any upload it starts fails without making network
+// requests.
+func newOfflineFilesystem(t *testing.T, tick time.Duration) (*Filesystem, *Inode) {
+	t.Helper()
+	dir := t.TempDir()
+	db, err := bolt.Open(filepath.Join(dir, "offline.db"), 0600, nil)
+	require.NoError(t, err)
+	t.Cleanup(func() { db.Close() })
+	require.NoError(t, db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(bucketMetadata)
+		return err
+	}))
+
+	f := &Filesystem{
+		RawFileSystem: fuse.NewDefaultRawFileSystem(),
+		content:       NewLoopbackCache(filepath.Join(dir, "content")),
+		db:            db,
+		auth:          &graph.Auth{},
+		opendirs:      make(map[uint64][]*Inode),
+	}
+	root := NewInodeDriveItem(&graph.DriveItem{
+		ID:     "offline-root",
+		Name:   "root",
+		Parent: &graph.DriveItemParent{},
+		Folder: &graph.Folder{},
+	})
+	f.root = root.ID()
+	f.InsertID(f.root, root)
+	f.uploads = NewUploadManager(tick, db, f, f.auth)
+	return f, root
+}
+
+// persistedUpload returns the upload session stored in the db under an ID.
+func persistedUpload(t *testing.T, f *Filesystem, id string) *UploadSession {
+	t.Helper()
+	var session *UploadSession
+	require.NoError(t, f.db.View(func(tx *bolt.Tx) error {
+		b := tx.Bucket(bucketUploads)
+		if b == nil {
+			return nil
+		}
+		if data := b.Get([]byte(id)); data != nil {
+			session = &UploadSession{}
+			return json.Unmarshal(data, session)
+		}
+		return nil
+	}))
+	return session
+}
+
+// A delta can report a new file's server ID before the upload manager has
+// noticed that the upload which created it completed. MoveID used to cancel the
+// item's pending upload in that case, losing a rewrite of the file that was
+// queued while the first upload was running.
+func TestMoveIDKeepsPendingUpload(t *testing.T) {
+	t.Parallel()
+	f, root := newOfflineFilesystem(t, time.Hour)
+
+	inode := NewInode("rewritten.txt", 0644|fuse.S_IFREG, root)
+	localID := inode.ID()
+	f.InsertID(localID, inode)
+	data := []byte("content written during the first upload")
+	require.NoError(t, f.content.Insert(localID, data))
+	session, err := NewUploadSession(inode, &data)
+	require.NoError(t, err)
+	f.uploads.queue <- session
+
+	const serverID = "offline-server-id-rewrite"
+	require.NoError(t, f.MoveID(localID, serverID))
+
+	assert.Equal(t, inode, f.GetID(serverID))
+	assert.Equal(t, serverID, inode.ID())
+	assert.Nil(t, persistedUpload(t, f, localID), "Upload still stored under the old ID.")
+	moved := persistedUpload(t, f, serverID)
+	if assert.NotNil(t, moved, "Pending upload was not moved to the new ID.") {
+		assert.Equal(t, serverID, moved.ID)
+		assert.Equal(t, serverID, moved.OldID)
+		assert.Equal(t, data, moved.Data)
+	}
+}
+
+// If a file is renamed or moved on the server before we learned its server ID,
+// a delta cannot match it by name and adds a second inode for it. MoveID used
+// to overwrite that inode with ours, losing the new name and parent, and left
+// the item listed twice in its parent.
+func TestMoveIDMergesInodeAddedByDelta(t *testing.T) {
+	t.Parallel()
+	f, root := newOfflineFilesystem(t, time.Hour)
+
+	dir := NewInodeDriveItem(&graph.DriveItem{
+		ID:     "offline-dir",
+		Name:   "dir",
+		Parent: &graph.DriveItemParent{ID: root.ID()},
+		Folder: &graph.Folder{},
+	})
+	f.InsertChild(root.ID(), dir)
+
+	inode := NewInode("before.txt", 0644|fuse.S_IFREG, root)
+	localID := inode.ID()
+	f.InsertID(localID, inode)
+	require.NoError(t, f.content.Insert(localID, []byte("uploaded content")))
+
+	// what applyDelta adds for the renamed and moved item
+	const serverID = "offline-server-id-rename"
+	f.InsertChild(dir.ID(), NewInodeDriveItem(&graph.DriveItem{
+		ID:     serverID,
+		Name:   "after.txt",
+		Parent: &graph.DriveItemParent{ID: dir.ID()},
+		File:   &graph.File{},
+	}))
+
+	require.NoError(t, f.MoveID(localID, serverID))
+
+	assert.Equal(t, inode, f.GetID(serverID))
+	assert.Equal(t, "after.txt", inode.Name())
+	assert.Equal(t, dir.ID(), inode.ParentID())
+	assert.Equal(t, []byte("uploaded content"), f.content.Get(serverID))
+
+	root.RLock()
+	assert.NotContains(t, root.children, localID)
+	assert.NotContains(t, root.children, serverID)
+	root.RUnlock()
+	dir.RLock()
+	assert.Equal(t, []string{serverID}, dir.children)
+	dir.RUnlock()
+}
+
+// When the upload loop notices that an upload produced a new ID, it moves the
+// item itself. It cannot use MoveID for this, which waits for the upload loop.
+func TestUploadCompletionMovesID(t *testing.T) {
+	t.Parallel()
+	f, root := newOfflineFilesystem(t, 10*time.Millisecond)
+
+	inode := NewInode("completed.txt", 0644|fuse.S_IFREG, root)
+	localID := inode.ID()
+	f.InsertID(localID, inode)
+	data := []byte("uploaded content")
+	require.NoError(t, f.content.Insert(localID, data))
+	session, err := NewUploadSession(inode, &data)
+	require.NoError(t, err)
+
+	// pretend the upload is running, so the upload loop does not start it
+	session.setState(uploadStarted, nil)
+	f.uploads.queue <- session
+	require.Eventually(t, func() bool {
+		return persistedUpload(t, f, localID) != nil
+	}, 10*time.Second, 10*time.Millisecond)
+
+	// what a successful Upload leaves behind
+	const serverID = "offline-server-id-completed"
+	session.Lock()
+	session.ID = serverID
+	session.ETag = "offline-etag"
+	session.Unlock()
+	session.setState(uploadComplete, nil)
+
+	require.Eventually(t, func() bool {
+		inode.RLock()
+		etag := inode.DriveItem.ETag
+		inode.RUnlock()
+		// the session is removed after the ETag is set
+		return f.GetID(serverID) == inode && etag == "offline-etag" &&
+			persistedUpload(t, f, localID) == nil &&
+			persistedUpload(t, f, serverID) == nil
+	}, 10*time.Second, 10*time.Millisecond, "Upload completion was not handled.")
+	assert.Equal(t, serverID, inode.ID())
+	assert.Equal(t, data, f.content.Get(serverID))
+}

fs/upload_manager.go

@@ -17,18 +17,26 @@ var bucketUploads = []byte("uploads")
 type UploadManager struct {
 	queue         chan *UploadSession
 	deletionQueue chan string
+	moveQueue     chan moveRequest
 	sessions      map[string]*UploadSession
 	inFlight      uint8 // number of sessions in flight
 	auth          *graph.Auth
 	fs            *Filesystem
 	db            *bolt.DB
 }
 
+// moveRequest asks the upload loop to move an item's upload to a new ID.
+type moveRequest struct {
+	oldID, newID string
+	done         chan struct{}
+}
+
 // NewUploadManager creates a new queue/thread for uploads
 func NewUploadManager(duration time.Duration, db *bolt.DB, fs *Filesystem, auth *graph.Auth) *UploadManager {
 	manager := UploadManager{
 		queue:         make(chan *UploadSession),
 		deletionQueue: make(chan string, 1000), // FIXME - why does this chan need to be buffered now???
+		moveQueue:     make(chan moveRequest),
 		sessions:      make(map[string]*UploadSession),
 		auth:          auth,
 		db:            db,
@@ -84,6 +92,10 @@ func (u *UploadManager) uploadLoop(duration time.Duration) {
 		case cancelID := <-u.deletionQueue: // remove uploads for deleted items
 			u.finishUpload(cancelID)
 
+		case req := <-u.moveQueue: // an item's ID changed
+			u.moveUpload(req.oldID, req.newID)
+			close(req.done)
+
 		case <-ticker.C: // periodically start uploads, or remove them if done/failed
 			for _, session := range u.sessions {
 				switch session.getState() {
@@ -125,11 +137,16 @@ func (u *UploadManager) uploadLoop(duration time.Duration) {
 
 					// ID changed during upload, move to new ID
 					if session.OldID != session.ID {
-						err := u.fs.MoveID(session.OldID, session.ID)
+						// Filesystem.MoveID would wait for this goroutine,
+						// move the session here. This also sets its OldID to
+						// the new ID.
+						oldID := session.OldID
+						u.moveUpload(oldID, session.ID)
+						err := u.fs.moveID(oldID, session.ID)
 						if err != nil {
 							log.Error().
 								Str("id", session.ID).
-								Str("oldID", session.OldID).
+								Str("oldID", oldID).
 								Str("name", session.Name).
 								Err(err).
 								Msg("Could not move inode to new ID!")
@@ -168,6 +185,43 @@ func (u *UploadManager) CancelUpload(id string) {
 	u.deletionQueue <- id
 }
 
+// MoveUpload moves any pending upload of an item to a new ID, for instance
+// when a local ID is replaced with the ID the item got on the server. The
+// upload is kept rather than cancelled, and targets the item by its new ID
+// from then on. Blocks until the upload loop has moved it, must not be called
+// from the upload loop itself.
+func (u *UploadManager) MoveUpload(oldID, newID string) {
+	req := moveRequest{oldID: oldID, newID: newID, done: make(chan struct{})}
+	u.moveQueue <- req
+	<-req.done
+}
+
+// moveUpload implements MoveUpload, only called from the upload loop.
+func (u *UploadManager) moveUpload(oldID, newID string) {
+	session, exists := u.sessions[oldID]
+	if !exists || oldID == newID {
+		return
+	}
+	if _, taken := u.sessions[newID]; taken {
+		// a session queued under the new ID is newer than this one
+		u.finishUpload(oldID)
+		return
+	}
+
+	delete(u.sessions, oldID)
+	session.Lock()
+	session.ID = newID
+	session.OldID = newID
+	session.Unlock()
+	u.sessions[newID] = session
+	contents, _ := json.Marshal(session)
+	u.db.Batch(func(tx *bolt.Tx) error {
+		b, _ := tx.CreateBucketIfNotExists(bucketUploads)
+		b.Delete([]byte(oldID))
+		return b.Put([]byte(newID), contents)
+	})
+}
+
 // finishUpload is an internal method that gets called when a session is
 // completed. It cancels the session if one was in progress, and then deletes
 // it from both memory and disk.