From a87666ee09004570d9d81e2d0218276fad53ff8f Mon Sep 17 00:00:00 2001 From: Leif Rossau Date: Tue, 13 Oct 2026 09:11:56 +0200 Subject: [PATCH 1/2] Keep pending uploads when an item's ID changes When a new file is uploaded for the first time, the server can report it in a delta before the upload loop has noticed that the upload completed. applyDelta then finds the item by name and calls MoveID to replace its local ID with the server ID. MoveID removed the old ID with DeleteID, which also cancels the item's upload. If the file was written again while the first upload was running, the upload queued for that rewrite was dropped: the new content stayed local only and the server kept the old content, with nothing left to retry the upload. MoveID now moves a pending upload to the new ID instead of cancelling it. The upload loop is the only goroutine that touches its sessions, so MoveID asks it to move the session and waits for it, before the inode is moved. A session created from the inode after that already has the new ID and replaces the moved one as usual. A moved session targets the item by its ID, so it does not create a copy of the file if the item was renamed on the server in the meantime. The upload loop moves inodes itself when an upload produced a new ID. It cannot wait for itself, so it moves the session directly and uses the inode-only part of MoveID. This also means it no longer queues a cancellation for the old ID on every first upload of a file. UploadSession.Upload now reads the session ID once under the lock, since it can change while an upload is running. Co-Authored-By: Claude Opus 5.5 --- fs/cache.go | 26 +++++-- fs/move_id_offline_test.go | 139 +++++++++++++++++++++++++++++++++++++ fs/upload_manager.go | 58 +++++++++++++++- fs/upload_session.go | 24 ++++--- 4 files changed, 231 insertions(+), 16 deletions(-) create mode 100644 fs/move_id_offline_test.go diff --git a/fs/cache.go b/fs/cache.go index 7920a60d..eb0fa265 100644 --- a/fs/cache.go +++ b/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 @@ -530,7 +546,7 @@ func (f *Filesystem) MoveID(oldID string, newID string) error { parent.Unlock() // now actually perform the metadata+content move - f.DeleteID(oldID) + f.deleteID(oldID) f.InsertID(newID, inode) if inode.IsDir() { return nil diff --git a/fs/move_id_offline_test.go b/fs/move_id_offline_test.go new file mode 100644 index 00000000..ee92b6c3 --- /dev/null +++ b/fs/move_id_offline_test.go @@ -0,0 +1,139 @@ +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) + } +} + +// 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)) +} diff --git a/fs/upload_manager.go b/fs/upload_manager.go index 24630006..1dbf2556 100644 --- a/fs/upload_manager.go +++ b/fs/upload_manager.go @@ -17,6 +17,7 @@ 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 @@ -24,11 +25,18 @@ type UploadManager struct { 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. diff --git a/fs/upload_session.go b/fs/upload_session.go index 5b4db620..551e97b0 100644 --- a/fs/upload_session.go +++ b/fs/upload_session.go @@ -144,6 +144,7 @@ func (u *UploadSession) cancel(auth *graph.Auth) { func (u *UploadSession) uploadChunk(auth *graph.Auth, offset uint64) ([]byte, int, error) { u.Lock() url := u.UploadURL + id := u.ID if url == "" { u.Unlock() return nil, -1, errors.New("UploadSession UploadURL cannot be empty") @@ -172,7 +173,7 @@ func (u *UploadSession) uploadChunk(auth *graph.Auth, offset uint64) ([]byte, in // no Authorization header - it will throw a 401 if present request.Header.Add("Content-Length", strconv.Itoa(int(reqChunkSize))) frags := fmt.Sprintf("bytes %d-%d/%d", offset, end-1, u.Size) - log.Info().Str("id", u.ID).Msg("Uploading " + frags) + log.Info().Str("id", id).Msg("Uploading " + frags) request.Header.Add("Content-Range", frags) resp, err := client.Do(request) @@ -189,7 +190,12 @@ func (u *UploadSession) uploadChunk(auth *graph.Auth, offset uint64) ([]byte, in // goroutine, or it can potentially block for a very long time. The uploadSession.error // field contains errors to be handled if called as a goroutine. func (u *UploadSession) Upload(auth *graph.Auth) error { - log.Info().Str("id", u.ID).Str("name", u.Name).Msg("Uploading file.") + // The upload manager can move the session to a new ID while it is + // running, read it only once. + u.Lock() + id := u.ID + u.Unlock() + log.Info().Str("id", id).Str("name", u.Name).Msg("Uploading file.") u.setState(uploadStarted, nil) var uploadPath string @@ -199,7 +205,7 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { // adding file modification times. We don't really care though, because // after some experimentation, the Microsoft API doesn't seem to properly // support these either (this is why we have to use etags). - if isLocalID(u.ID) { + if isLocalID(id) { uploadPath = fmt.Sprintf( "/me/drive/items/%s:/%s:/content", url.PathEscape(u.ParentID), @@ -208,7 +214,7 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { } else { uploadPath = fmt.Sprintf( "/me/drive/items/%s/content", - url.PathEscape(u.ID), + url.PathEscape(id), ) } // small files handled in this block @@ -223,7 +229,7 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { return u.setState(uploadErrored, fmt.Errorf("small upload failed: %w", err)) } } else { - if isLocalID(u.ID) { + if isLocalID(id) { uploadPath = fmt.Sprintf( "/me/drive/items/%s:/%s:/createUploadSession", url.PathEscape(u.ParentID), @@ -232,7 +238,7 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { } else { uploadPath = fmt.Sprintf( "/me/drive/items/%s/createUploadSession", - url.PathEscape(u.ID), + url.PathEscape(id), ) } sessionPostData, _ := json.Marshal(UploadSessionPost{ @@ -272,7 +278,7 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { // exit this loop unless it receives a non 5xx error or serious failure for backoff := 1; status >= 500; backoff *= 2 { log.Error(). - Str("id", u.ID). + Str("id", id). Str("name", u.Name). Int("chunk", i). Int("nchunks", nchunks). @@ -300,10 +306,10 @@ func (u *UploadSession) Upload(auth *graph.Auth) error { // the API frequently just returns a 0-byte response for completed // multipart uploads, so we manually fetch the newly updated item var remotePtr *graph.DriveItem - if isLocalID(u.ID) { + if isLocalID(id) { remotePtr, err = graph.GetItemChild(u.ParentID, u.Name, auth) } else { - remotePtr, err = graph.GetItem(u.ID, auth) + remotePtr, err = graph.GetItem(id, auth) } if err == nil { remote = *remotePtr From 6313f42f69ce34af1334ca6041a1614cd59afa50 Mon Sep 17 00:00:00 2001 From: Leif Rossau Date: Tue, 13 Oct 2026 09:12:30 +0200 Subject: [PATCH 2/2] Keep server-side renames of items that just got their server ID If a new file was renamed or moved on the server after its first upload, but before the upload loop noticed that the upload completed, the delta for it could not be matched to the local item by name. applyDelta then added a second inode for the server ID with the new name and parent. When the upload loop moved the local item to the server ID, MoveID overwrote that inode with the local one, so the file kept its old name and parent locally, and the server ID was listed twice in the parent. When an inode already exists under the new ID, MoveID now applies its name and parent to the item being moved and removes it, instead of overwriting it. The item is removed from its old parent before it is added under the new ID, so it ends up listed once, in the right parent. Co-Authored-By: Claude Opus 5.5 --- fs/cache.go | 32 ++++++++++++++++++-------- fs/move_id_offline_test.go | 46 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 9 deletions(-) diff --git a/fs/cache.go b/fs/cache.go index eb0fa265..318387d9 100644 --- a/fs/cache.go +++ b/fs/cache.go @@ -534,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 diff --git a/fs/move_id_offline_test.go b/fs/move_id_offline_test.go index ee92b6c3..730bd202 100644 --- a/fs/move_id_offline_test.go +++ b/fs/move_id_offline_test.go @@ -96,6 +96,52 @@ func TestMoveIDKeepsPendingUpload(t *testing.T) { } } +// 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) {