diff --git a/docs/changes/4.4.2.md b/docs/changes/4.4.2.md
index a0e57d4fd..f4659bcb3 100644
--- a/docs/changes/4.4.2.md
+++ b/docs/changes/4.4.2.md
@@ -7,6 +7,7 @@ Release date: `20xx-xx-xx`
 - [NXDRIVE-1831](https://jira.nuxeo.com/browse/NXDRIVE-1831): [GNU/Linux] Set the root local folder icon
 - [NXDRIVE-1847](https://jira.nuxeo.com/browse/NXDRIVE-1847): [Windows] Fix endless synchronization on fast create-then-rename folder
 - [NXDRIVE-1893](https://jira.nuxeo.com/browse/NXDRIVE-1893): Remove hardcoded "Nuxeo Drive" strings
+- [NXDRIVE-1910](https://jira.nuxeo.com/browse/NXDRIVE-1910): [Direct Transfer] Allow the feature when the application is paused
 - [NXDRIVE-1918](https://jira.nuxeo.com/browse/NXDRIVE-1918): Use Amazon S3 Direct Upload when available
 - [NXDRIVE-1945](https://jira.nuxeo.com/browse/NXDRIVE-1945): Fix mypy issues following the update to mypy 0.740
 - [NXDRIVE-1958](https://jira.nuxeo.com/browse/NXDRIVE-1958): [Direct Transfer] Create a remote subfolder when uploaing a folder
@@ -22,9 +23,11 @@ Release date: `20xx-xx-xx`
 - [NXDRIVE-1987](https://jira.nuxeo.com/browse/NXDRIVE-1987): Inexistent database backups should not prevent backup
 - [NXDRIVE-2002](https://jira.nuxeo.com/browse/NXDRIVE-2002): Use a stricter mypy configuration
 - [NXDRIVE-2005](https://jira.nuxeo.com/browse/NXDRIVE-2005): Use stricter flake8 plugins in pre-commit
+- [NXDRIVE-2006](https://jira.nuxeo.com/browse/NXDRIVE-2006): [Direct Transfer] Isolate the component
 - [NXDRIVE-2008](https://jira.nuxeo.com/browse/NXDRIVE-2008): Run codespell on the entire code base
 - [NXDRIVE-2012](https://jira.nuxeo.com/browse/NXDRIVE-2012): Upgrade to sentry-sdk 0.14.1 to fix memory leaks
 - [NXDRIVE-2027](https://jira.nuxeo.com/browse/NXDRIVE-2027): Allow Direct Edit on custom blob metadata values
+- [NXDRIVE-2029](https://jira.nuxeo.com/browse/NXDRIVE-2029): [Direct Transfer] Allow the feature when the synchronization is disabled
 - [NXDRIVE-2040](https://jira.nuxeo.com/browse/NXDRIVE-2040): [Direct Transfer] Temporary disable folder uploads
 - [NXDRIVE-2047](https://jira.nuxeo.com/browse/NXDRIVE-2047): Unlock updates on the centralized channel when auto-update is disabled
 - [NXDRIVE-2048](https://jira.nuxeo.com/browse/NXDRIVE-2048): Reduce the amount of Process information retrieved in the auto-locker
@@ -109,15 +112,27 @@ Release date: `20xx-xx-xx`
 - Removed `uid` argument from `BlacklistQueue.push()`
 - Changed `name` argument of `BlacklistQueue.push()` from `str` to `pathlib.Path`
 - Added `CustomMemoryHandler.flush()`
+- Added `Engine.dt_manager`
+- Added `Engine.direct_transfer_duplicate_callback`
+- Renamed `Engine.directTranferDuplicateError` to `directTransferDuplicateError`
+- Added `EngineDAO.update_pair_state()`
 - Added `EngineDAO.update_upload()`
 - Changed `Manager.is_paused()` from a function to an attribute
 - Removed `Manager.refresh_update_status()`. Use `.updater.refresh_status()` instead.
 - Added `Remote.personal_space()`
+- Removed `Remote.direct_transfer()`
+- Added `Report.copy_direct_transfer_dbs()`
+- Added `TransferStatus.ABORTED`
 - Changed `Upload.batch` from `str` to `nuxeo.models.Batch`
 - Removed `Upload.idx`. Use `Upload.batch["upload_idx"]` instead.
 - Removed `idx` column from `Uploads` table
+- Added the `direct_transfer` folder
 - Removed constants.py::`FORBID_CHARS_ALL`
 - Removed constants.py::`FORBID_CHARS_UNIX`
+- Removed exceptions.py::`DirectTransferDuplicateFoundError`
+- Added features.py
+- Removed notification.py::`DirectTransferStatus()`
+- Added state.py
 - Added updater/constants.py::`AutoUpdateState`
 - Added updater/utils.py::`auto_updates_state()`
 - Removed utils.py::`safe_os_filename()`. Use `safe_filename()` instead.
diff --git a/nxdrive/client/remote_client.py b/nxdrive/client/remote_client.py
index c1b1f0f7e..448a229d8 100644
--- a/nxdrive/client/remote_client.py
+++ b/nxdrive/client/remote_client.py
@@ -35,13 +35,7 @@
     UploadAction,
     VerificationAction,
 )
-from ..exceptions import (
-    DirectTransferDuplicateFoundError,
-    DownloadPaused,
-    NotFound,
-    ScrollDescendantsError,
-    UploadPaused,
-)
+from ..exceptions import DownloadPaused, NotFound, ScrollDescendantsError, UploadPaused
 from ..objects import Download, NuxeoDocumentInfo, RemoteFileInfo, Upload
 from ..options import Options
 from ..utils import (
@@ -52,7 +46,6 @@
     unlock_path,
     version_le,
 )
-from .local import LocalClient
 from .proxy import Proxy
 
 if TYPE_CHECKING:
@@ -658,7 +651,7 @@ def link_blob_to_doc(
             LinkingAction.finish_action()
 
     def get_document_or_none(self, uid: str = "", path: str = "") -> Optional[Document]:
-        """Fetch a document base don given criteria or return None if not found on the server."""
+        """Fetch a document based on given criteria or return None if not found on the server."""
         doc: Optional[Document] = None
         try:
             doc = self.documents.get(uid=uid, path=path)
@@ -667,105 +660,6 @@ def get_document_or_none(self, uid: str = "", path: str = "") -> Optional[Docume
                 raise
         return doc
 
-    def direct_transfer(
-        self, file: Path, parent_path: str, engine_uid: str, replace_blob: bool = False
-    ) -> None:
-        """Upload a given file to the given folderish document on the server.
-
-        Note about possible duplicate creation via a race condition client <-> server.
-        Given the local *file* with the path "$HOME/some-folder/subfolder/file.odt",
-        the file name is "file.odt".
-
-        Scenario:
-            - Step 1: local, check for a doc with the path name "file.odt" => nothing returned, continuing;
-            - Step 2: server, a document with a path name set to "file.odt" is created;
-            - Step 3: local, create the document with the path name "file.odt".
-
-        Even if the elapsed time between steps 1 and 3 is really short, it may happen.
-
-        What can be done to prevent such scenario is not on the Nuxeo Drive side but on the server one.
-        For now, 2 options are possible but not qualified to be done in a near future:
-            - https://jira.nuxeo.com/browse/NXP-22959;
-            - create a new operation `Document.GetOrCreate` that ensures atomicity.
-        """
-        log.info(f"Direct Transfer of {file!r} into {parent_path!r}")
-
-        # The remote file, when created, is stored in the file xattrs.
-        # So retrieve it and if it is defined, the document creation should
-        # be skipped to prevent duplicate creations.
-        remote_ref = LocalClient.get_path_remote_id(file, name="remote")
-
-        doc: Optional[Document] = None
-
-        if remote_ref:
-            # The remote ref is set, so it means either the file has already been uploaded,
-            # either a previous upload failed: the document was created, or not, and it has
-            # a blob attached, or not. In any cases, we need to ensure the user can upload
-            # without headhache.
-            doc = self.get_document_or_none(uid=remote_ref)
-
-        if not doc:
-            # We need to handle possbile duplicates based on the file name and
-            # the destination folder on the server.
-            # Note: using this way may still result in duplicates:
-            #  - the user created 2 documents with the same name on Web-UI or another way
-            #  - the user then deleted the 1st document
-            #  - the other document has a path like "name.TIMESTAMP"
-            # So then Drive will not see that document as a duplicate because it will check
-            # a path with "name" only.
-
-            # If we really want to avoid that situation, we should use that commented code:
-            """
-            # Note that it would be too much effort for the server, we do not want that!
-            for child in self.documents.get_children(path=parent_path):
-                # It is OK to have a folder and a file with the same name,
-                # but not 2 files or 2 folders with the same name
-                if child.title == file.name:
-                    local_is_dir = file.isdir()
-                    remote_is_dir = "Folderish" in child["facets"]
-                    if local_is_dir is remote_is_dir:
-                        # Duplicate found!
-                        doc = child
-                        remote_ref = doc.uid
-                        break
-            """
-
-            doc = self.get_document_or_none(path=f"{parent_path}/{file.name}")
-            if doc:
-                remote_ref = doc.uid
-
-        if not replace_blob and doc and doc.properties.get("file:content"):
-            # The document already exists and has a blob attached. Ask the user what to do.
-            raise DirectTransferDuplicateFoundError(file, doc)
-
-        if not doc:
-            # Create the document on the server
-            nature = "File" if file.is_file() else "Folder"
-            doc = self.documents.create(
-                Document(
-                    name=file.name, type=nature, properties={"dc:title": file.name}
-                ),
-                parent_path=parent_path,
-            )
-            remote_ref = doc.uid
-
-        # If the path is a folder, there is no more work to do
-        if file.is_dir():
-            return
-
-        # Save the remote document's UID into the file xattrs, in case next steps fails
-        LocalClient.set_path_remote_id(file, remote_ref, name="remote")
-
-        # Upload the blob and attach it to the document
-        self.upload(
-            file,
-            engine_uid=engine_uid,
-            document=remote_ref,
-            command="Blob.AttachOnDocument",
-            xpath="file:content",
-            void_op=True,
-        )
-
     def get_fs_info(
         self, fs_item_id: str, parent_fs_item_id: str = None
     ) -> RemoteFileInfo:
diff --git a/nxdrive/constants.py b/nxdrive/constants.py
index f546c84a0..eeffe4d9d 100644
--- a/nxdrive/constants.py
+++ b/nxdrive/constants.py
@@ -96,3 +96,4 @@ class TransferStatus(Enum):
     PAUSED = auto()
     SUSPENDED = auto()
     DONE = auto()
+    ABORTED = auto()
diff --git a/nxdrive/data/i18n/i18n.json b/nxdrive/data/i18n/i18n.json
index 9ed3be160..81190b9d1 100644
--- a/nxdrive/data/i18n/i18n.json
+++ b/nxdrive/data/i18n/i18n.json
@@ -81,9 +81,7 @@
     "DIRECT_EDIT_UNLOCK_ERROR_DESCRIPTION": "The document „%1“ was not unlocked",
     "DIRECT_EDIT_UPDATED_FILE": "The file „%1“ has been updated",
     "DIRECT_EDIT_VERSION": "You are trying to perform a DirectEdit on version %1 of „%2“. It is not the current version of the document, so it is readonly and cannot be edited.",
-    "DIRECT_TRANSFER_END": "Transfer done: „%1“",
     "DIRECT_TRANSFER_ERROR": "Transfer error: „%1“",
-    "DIRECT_TRANSFER_START": "Transfer start: „%1“",
     "DIRECT_TRANSFER_WINDOW_TITLE": "Direct Transfer - %1",
     "DISCONNECT": "Remove account",
     "DISCONNECTING": "Disconnecting …",
diff --git a/nxdrive/data/qml/SystrayMenu.qml b/nxdrive/data/qml/SystrayMenu.qml
index 0a3fc7d0e..bf8f1ccf3 100755
--- a/nxdrive/data/qml/SystrayMenu.qml
+++ b/nxdrive/data/qml/SystrayMenu.qml
@@ -52,7 +52,7 @@ ShadowRectangle {
             text: qsTr("QUIT") + tl.tr
             onClicked: {
                 application.hide_systray()
-                application.quit()
+                application.exit_app()
             }
         }
 
diff --git a/nxdrive/direct_transfer/__init__.py b/nxdrive/direct_transfer/__init__.py
new file mode 100644
index 000000000..1be44055c
--- /dev/null
+++ b/nxdrive/direct_transfer/__init__.py
@@ -0,0 +1,7 @@
+"""
+The Direct Transfer feature.
+"""
+from .manager import DirectTransferManager
+from .uploader import DirectTransferDuplicateFoundError
+
+__all__ = ("DirectTransferDuplicateFoundError", "DirectTransferManager")
diff --git a/nxdrive/direct_transfer/manager.py b/nxdrive/direct_transfer/manager.py
new file mode 100644
index 000000000..84935b4ea
--- /dev/null
+++ b/nxdrive/direct_transfer/manager.py
@@ -0,0 +1,432 @@
+# coding: utf-8
+"""
+The Direct Transfer feature.
+
+What: the transfers manager.
+
+For now, transfers are done in sequence as it eases the implementation.
+For better performances, one should really use Nuxeo Drive.
+"""
+from collections import deque
+from datetime import datetime
+from logging import getLogger
+from operator import attrgetter
+from pathlib import Path
+from typing import TYPE_CHECKING, Callable, List, Optional, Set
+
+from PyQt5.QtCore import QCoreApplication
+
+from ..constants import CONNECTION_ERROR, TransferStatus
+from ..state import State
+from ..utils import get_tree_list
+from . import uploader
+from .models import DATABASE, MODELS, Session, Transfer
+from .runner import Runner
+
+if TYPE_CHECKING:
+    from nuxeo.handlers.default import Uploader  # noqa
+    from nuxeo.models import Document  # noqa
+
+    from .client.remote_client import Remote  # noqa
+
+
+log = getLogger(__name__)
+
+
+class DirectTransferManager:
+    """Direct Transfer manager.
+    This is the core of the feature, separated from Nuxeo Drive.
+    It is used to create a new Direct Transfer, keep track of the progression,
+    pause/resume transfers and it is connected to a (beautiful) custom QML component.
+    """
+
+    def __init__(
+        self,
+        db: str,
+        engine_uid: str,
+        remote: "Remote",
+        chunk_callback: Callable[["Uploader"], None] = lambda *_: None,
+        done_callback: Callable[[bool, int], None] = lambda *_: None,
+        dupe_callback: Callable[[Path, "Document"], None] = lambda *_: None,
+    ) -> None:
+        """
+        *chunk_callback* is triggered when a chunk is successfully uploaded.
+        *done_callback* is triggered when an upload was successfully done.
+        *dupe_callback* is triggered when a given upload would generate a duplicate on the server.
+        """
+        self.db = db
+        self.engine_uid = engine_uid
+        self.remote = remote
+        self.chunk_callback = chunk_callback
+        self.done_callback = done_callback
+        self.dupe_callback = dupe_callback
+
+        self.is_started = False
+        self.reinit()
+
+    def reinit(self) -> None:
+        """(Re)Initialize the database connection, refetch session and associated transfers."""
+        # Init the database
+        self.init_db(self.db)
+
+        # Retrieve the session or create a new one
+        self.session, _ = Session.get_or_create(finished=None)
+
+        # Retrieve associated transfers
+        transfers = Transfer.get_or_none(Transfer.session == self.session.id) or []
+        if not isinstance(transfers, list):
+            # 1 record in the database
+            transfers = [transfers]
+        self.transfers: List[Transfer] = transfers
+
+        # The thread-safe queue
+        self._iterator: deque = deque()
+
+        # Inject new methods
+        self.inject()
+
+    def __repr__(self) -> str:
+        return (
+            f"{type(self).__name__}<uid={self.engine_uid!r}"
+            f", db={self.db!r}"
+            f", paths={len(self.transfers)}"
+            f", completed={self.is_completed!r}"
+            f", started={self.is_started!r}"
+            f", progress={self.progress:.1f}"
+            f", size={self.size}"
+            f", uploaded={self.uploaded}"
+            ">"
+        )
+
+    def __iter__(self) -> "DirectTransferManager":
+        """The whole class is an iterator.
+        It is used to io iterate over transfers to handle:
+
+            >>> for transfer in self:
+            ...     upload(transfer)
+
+        """
+        self._iterator.clear()
+        self._iterator.extend(self.partitions)
+
+        return self
+
+    def __next__(self) -> Transfer:
+        """Get the next transfer to handle. See __iter__()."""
+        try:
+            return self._iterator.popleft()  # type: ignore
+        except IndexError:
+            raise StopIteration()
+
+    @property
+    def is_completed(self) -> bool:
+        """Return True when all transfers are done."""
+        return not self.uploading and self.progress >= 100.0
+
+    @property
+    def partitions(self) -> List[Transfer]:
+        """Return a list of paths to upload, sorted by remote path > file type > local path."""
+        return sorted(
+            self.transfers, key=attrgetter("remote_path", "is_file", "local_path")
+        )
+
+    @property
+    def progress(self) -> float:
+        """Overall progression."""
+        if not self.has_transfers_to_handle():
+            # All transfers are either done or aborted
+            return 100.0
+
+        return self.uploaded * 100.0 / self.size
+
+    @property
+    def size(self) -> int:
+        """Overall files size to upload."""
+        return sum(
+            transfer.file_size
+            for transfer in self.transfers
+            if transfer.status is not TransferStatus.ABORTED
+        )
+
+    @property
+    def uploaded(self) -> int:
+        """Overall uploaded size."""
+        return sum(
+            transfer.uploaded_size
+            for transfer in self.transfers
+            if transfer.status is not TransferStatus.ABORTED
+        )
+
+    @property
+    def uploading(self) -> List[Transfer]:
+        """Get the transfers being uploaded."""
+        return [
+            transfer
+            for transfer in self.transfers
+            if transfer.status is TransferStatus.ONGOING
+        ]
+
+    def add(self, local_path: Path, remote_path: str) -> Optional[Transfer]:
+        """Plan the *local_path* upload to the given *remote_path*.
+        Return the Transfer item.
+        """
+        if self.already_added(local_path):
+            log.debug(f"{local_path!r} is already part of the upload list")
+            return None
+
+        try:
+            is_file = local_path.is_file()
+            size = local_path.stat().st_size
+        except OSError:
+            log.warning(f"Skipping errored {local_path!r}", exc_info=True)
+            return None
+        else:
+            # Folders have no size
+            if not is_file:
+                size = 0
+
+        transfer: Transfer = Transfer.create(
+            session=self.session.id,
+            local_path=local_path,
+            remote_path=remote_path,
+            is_file=is_file,
+            file_size=size,
+        )
+        log.debug(
+            f"Added {local_path!r} to the upload list of session {self.session.id}"
+        )
+        self.transfers.append(transfer)
+        return transfer
+
+    def add_all(self, local_paths: Set[Path], remote_path: str) -> None:
+        """Recursively plan *local_paths* upload to the given *remote_path*."""
+        for local_path in sorted(local_paths):
+            if local_path.is_file():
+                self.add(local_path, remote_path)
+            elif local_path.is_dir():
+                tree = get_tree_list(local_path, remote_path)
+                for computed_remote_path, path in sorted(tree):
+                    self.add(path, computed_remote_path)
+
+    def already_added(self, local_path: Path) -> bool:
+        """Return True if a given *local_path* is already planned for upload."""
+        return any(transfer.local_path == local_path for transfer in self.transfers)
+
+    def cancel(self, file: Path) -> None:
+        """Cancel the transfer of the given local *file*.
+        This method is used when a duplicate creation would happen but
+        the user asked to cancel the upload.
+        """
+        for transfer in self.transfers:
+            if transfer.local_path != file:
+                continue
+
+            log.info(f"Transfer of {file!r} cancelled (user choice)")
+            transfer.status = TransferStatus.ABORTED  # type: ignore
+            transfer.save()
+            self.requeue(transfer)
+
+    def has_transfers_to_handle(self) -> bool:
+        """Check if transfers are all either completed or aborted."""
+        return not all(
+            transfer.status in (TransferStatus.DONE, TransferStatus.ABORTED)
+            for transfer in self.transfers
+        )
+
+    def init_db(self, db: str) -> None:
+        """Initialize the database."""
+        DATABASE.init(db)
+        DATABASE.connect()
+        DATABASE.create_tables(MODELS)
+
+    def inject(self) -> None:
+        """Inject new methods into the Remote client."""
+        self.remote.dt_upload = uploader.dt_upload
+        self.remote.dt_do_upload = uploader.dt_do_upload
+        self.remote.dt_link_blob_to_doc = uploader.dt_link_blob_to_doc
+        self.remote.dt_upload_chunks = uploader.dt_upload_chunks
+
+    def replace_blob(self, file: Path, doc_uid: str) -> None:
+        """Replace the document's blob on the server.
+        This method is used when a duplicate creation would happen and
+        the user asked to replace the attached blob.
+        """
+        for transfer in self.transfers:
+            if transfer.local_path != file:
+                continue
+
+            log.info(
+                "Replacing the document's attached file "
+                f"(UID is {doc_uid!r}) for {file!r} (user choice)"
+            )
+            transfer.replace_blob = True
+            transfer.save()
+            self.requeue(transfer)
+
+    def requeue(self, transfer: Transfer) -> None:
+        """Repush the given *transfer* into the queue, at the beginning."""
+        self._iterator.appendleft(transfer)
+        log.debug(f"Re-queued {transfer.local_path!r}")
+        if not self.is_started:
+            self.start()
+
+    def reset(self) -> None:
+        """Reset data and start a new transfers session."""
+        if self.is_started:
+            log.warning("Cannot reset as there are ongoing transfers.")
+            return
+
+        if not self.session.started:
+            log.warning("Cannot reset unstarted session transfers.")
+            return
+
+        if self.session.status is not TransferStatus.DONE:
+            self.session.finished = datetime.now()
+            self.session.status = TransferStatus.ABORTED
+            self.session.save()
+
+        self._iterator.clear()
+        self.transfers = []
+        self.session = Session.create()
+
+    def should_stop(self) -> bool:
+        """Method calls before uploading any local path to let the possibility to abort the session."""
+        # .stop() has been called
+        return not self.is_started
+
+    def start(self) -> None:
+        """Start transfers."""
+        msg = f"Direct Transfer session {self.session.id}"
+
+        if self.is_started:
+            log.warning(f"{msg} already running ... ")
+            return
+
+        if self.session.status is TransferStatus.DONE:
+            log.info(f"{msg} already completed.")
+            return
+
+        log.info(f"{msg} starting ...")
+
+        if not self.session.started:
+            self.session.started = datetime.now()
+            self.session.save()
+
+        self.is_started = True
+        try:
+            for transfer in self:
+                if self.should_stop():
+                    break
+
+                if transfer.status in (TransferStatus.DONE, TransferStatus.ABORTED):
+                    # That means the local path was either:
+                    # - already processed
+                    # - cancelled
+                    # - locally removed
+                    continue
+
+                if transfer.error_count >= 3:
+                    transfer.error_count = 0
+                    self.stop()
+
+                self.upload(transfer)
+        finally:
+            self.is_started = False
+
+        if not self.is_completed:
+            log.info(f"{msg} ended (progression: {self.progress}%).")
+            return
+
+        self.session.finished = datetime.now()
+        self.session.status = TransferStatus.DONE
+        self.session.save()
+        elapsed = self.session.finished - self.session.started
+        log.info(f"{msg} finished in {elapsed}!")
+
+    def shutdown(self) -> None:
+        """Stop everything."""
+        log.debug("Shutting down the Direct Transfer manager ...")
+        self.stop()
+        if not DATABASE.is_closed():
+            DATABASE.close()
+
+    def stop(self) -> None:
+        """Stop transfers."""
+        if not self.is_started:
+            return
+
+        log.info(f"Direct Transfer session {self.session.id} stopping ...")
+
+        self.is_started = False
+        for transfer in self.transfers:
+            if transfer.status is not TransferStatus.ONGOING:
+                continue
+
+            transfer.status = TransferStatus.SUSPENDED  # type: ignore
+            transfer.save()
+
+    def upload_impl(self, transfer: Transfer) -> None:
+        """Process a Transfer."""
+        # Pre-check: the local path must exist
+        transfer.local_path.stat()
+
+        # If no error, do the upload
+        transfer.status = TransferStatus.ONGOING  # type: ignore
+
+        runner = Runner(
+            target=self.remote.dt_upload,
+            args=(self.remote, transfer),
+            kwargs={"chunk_callback": self.chunk_callback},
+        )
+        runner.start()
+
+        while runner.is_alive():
+            # Do not block the GUI
+            QCoreApplication.processEvents()
+
+            # The application is being stopped
+            if State.about_to_quit:
+                self.stop()
+                break
+
+        runner.join()
+        if runner.error:
+            # An exception occurred, re-raise it
+            raise runner.error from None
+
+        # And trigger the "done" callback
+        self.done_callback(not transfer.is_file, transfer.file_size)
+
+    def upload(self, transfer: Transfer) -> None:
+        """Process a Transfer with error management."""
+        try:
+            self.upload_impl(transfer)
+        except FileNotFoundError:
+            log.info(
+                f"Removing {transfer.local_path!r} from transfers as it does not exist anymore"
+            )
+            transfer.status = TransferStatus.ABORTED  # type: ignore
+            transfer.save()
+        except uploader.DirectTransferPaused:
+            log.info(f"Pausing transfer of {transfer.local_path!r}")
+            transfer.status = TransferStatus.PAUSED  # type: ignore
+            transfer.save()
+            self.stop()
+        except CONNECTION_ERROR as exc:
+            log.warning(f"Network error for {transfer.local_path!r}", exc_info=True)
+            transfer.error = type(exc).__name__
+            transfer.error_count += 1
+            transfer.save()
+            self.requeue(transfer)
+        except uploader.DirectTransferDuplicateFoundError as exc:
+            log.info(str(exc))
+            log.debug(f"Calling {self.dupe_callback} to ask what action to do")
+            self.dupe_callback(exc.file, exc.doc)
+        except Exception as exc:
+            # On any error, skip and retry on the next call to .start()
+            # It has to be done manually.
+            log.warning(f"Oups, cannot upload {transfer.local_path!r}", exc_info=True)
+            transfer.error = type(exc).__name__
+            transfer.error_count += 1
+            transfer.save()
+            self.requeue(transfer)
diff --git a/nxdrive/direct_transfer/models.py b/nxdrive/direct_transfer/models.py
new file mode 100644
index 000000000..fb708070c
--- /dev/null
+++ b/nxdrive/direct_transfer/models.py
@@ -0,0 +1,136 @@
+# coding: utf-8
+"""
+The Direct Transfer feature.
+
+What: SQL models.
+"""
+import json
+from pathlib import Path
+
+from nuxeo.models import Batch
+from peewee import (
+    BooleanField,
+    CharField,
+    DateTimeField,
+    Field,
+    ForeignKeyField,
+    IntegerField,
+    Model,
+    SqliteDatabase,
+)
+
+from ..constants import TransferStatus
+
+# https://sqlite.org/pragma.html
+PRAGMAS = (
+    ("foreign_keys", 1),
+    ("journal_mode", "wal"),
+    ("synchronous", 3),
+    ("wal_checkpoint", 100),
+)
+
+# Note: using None instead of a the real database file, it will be set later
+#       via DATABASE.init(FILE) in DirectTransferManager.init_db().
+DATABASE = SqliteDatabase(None, check_same_thread=False, pragmas=PRAGMAS)
+
+
+def empty_batch() -> Batch:
+    """Empty batch for BatchField."""
+    return Batch()
+
+
+class BaseModel(Model):
+    """Base class for all models."""
+
+    class Meta:
+        """The database to use for all models."""
+
+        database = DATABASE
+        only_save_dirty = True
+
+
+class BatchField(Field):
+    """Custom field to handle Batch objects."""
+
+    field_type = "batch"
+    unserializable = {"blobs"}
+
+    def db_value(self, batch: Batch) -> str:
+        """Convert the Batch to a serialized dict for database storage.
+        Any value that cannot be serialized is wiped out.
+        """
+        obj = {k: v for k, v in batch.as_dict().items() if k not in self.unserializable}
+        return json.dumps(obj)
+
+    def python_value(self, batch: str) -> Batch:
+        """Get back the Batch from the stored serialized dict (string)."""
+        return Batch(**json.loads(batch))
+
+
+class PathField(Field):
+    """Custom field to handle Path objects."""
+
+    field_type = "path"
+
+    def db_value(self, value: Path) -> str:
+        """Convert the path to a string for database storage."""
+        return str(value)
+
+    def python_value(self, value: str) -> Path:
+        """Get back the path from the stored string."""
+        return Path(value)
+
+
+class StatusField(Field):
+    """Custom field to handle TransferStatus objects."""
+
+    field_type = "status"
+
+    def db_value(self, status: TransferStatus) -> int:
+        """Convert the path to a string for database storage."""
+        return int(status.value)
+
+    def python_value(self, status: int) -> TransferStatus:
+        """Get back the path from the stored string."""
+        return TransferStatus(status)
+
+
+class Session(BaseModel):
+    """Represent a Direct Transfer session.
+    A session is a batch of transfers: each time a new Direct Transfer is initiated, a new session is created.
+    """
+
+    started = DateTimeField(null=True)
+    finished = DateTimeField(null=True)
+    status = StatusField(default=TransferStatus.ONGOING)
+
+
+class Transfer(BaseModel):
+    """Represent a Direct Transfer item."""
+
+    session = ForeignKeyField(Session)
+    local_path = PathField()
+    remote_path = CharField()
+    is_file = BooleanField(default=False)
+    file_size = IntegerField(default=0)
+    status = StatusField(default=TransferStatus.ONGOING)
+    batch = BatchField(default=empty_batch)
+    remote_ref = CharField(null=True)
+    doctype = CharField(null=True)
+    uploaded_size = IntegerField(default=0)
+    uploaded = BooleanField(default=False)
+    chunk_size = IntegerField(default=0)
+    replace_blob = BooleanField(default=False)
+    error_count = IntegerField(default=0)
+    error = CharField(null=True)
+
+    class Meta:
+        """Create indexes."""
+
+        indexes = (
+            # Create a unique on session/local_path/remote_path combination
+            (("session", "local_path", "remote_path"), True),
+        )
+
+
+MODELS = [Session, Transfer]
diff --git a/nxdrive/direct_transfer/runner.py b/nxdrive/direct_transfer/runner.py
new file mode 100644
index 000000000..dc3e3f588
--- /dev/null
+++ b/nxdrive/direct_transfer/runner.py
@@ -0,0 +1,26 @@
+# coding: utf-8
+"""
+The Direct Transfer feature.
+
+What: threaded runner.
+"""
+from threading import Thread
+from typing import Any, Optional
+
+
+class Runner(Thread):
+    """Simple runner used for every single upload.
+    Any exception will be forwarded to the caller via the .error attribute.
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Simple init to declare .error that is used to communicate eventual exceptions to the caller."""
+        super().__init__(*args, **kwargs)
+        self.error: Optional[Exception] = None
+
+    def run(self) -> None:
+        """Catch any error and store it inside the .error attribute instead of throwing it."""
+        try:
+            super().run()
+        except Exception as exc:
+            self.error = exc
diff --git a/nxdrive/direct_transfer/uploader.py b/nxdrive/direct_transfer/uploader.py
new file mode 100644
index 000000000..b6f1f7151
--- /dev/null
+++ b/nxdrive/direct_transfer/uploader.py
@@ -0,0 +1,287 @@
+# coding: utf-8
+"""
+The Direct Transfer feature.
+
+What: upload methods and exceptions.
+"""
+from logging import getLogger
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Optional
+
+from nuxeo.exceptions import HTTPError
+from nuxeo.handlers.default import Uploader
+from nuxeo.models import Document, FileBlob
+
+from ..constants import TX_TIMEOUT, TransferStatus
+from ..options import Options
+from .models import Transfer
+
+if TYPE_CHECKING:
+    from ..client.remote_client import Remote  # noqa
+
+log = getLogger(__name__)
+
+
+class DirectTransferPaused(Exception):
+    """A transfer has been paused, the file's processing should stop."""
+
+
+class DirectTransferDuplicateFoundError(ValueError):
+    """
+    Exception raised when a duplicate file already exists on the server
+    and trying to Direct Transfer a local file with the same name.
+    """
+
+    def __init__(self, file: Path, doc: Document) -> None:
+        self.file = file
+        self.doc = doc
+
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}<file={self.file!r}, doc={self.doc!r}>"
+
+    def __str__(self) -> str:
+        return (
+            f"Document with the name {self.file.name!r} already found on the server: {self.doc}."
+            f"Direct Transfer of {self.file!r} postponed after the user decided what to do."
+        )
+
+
+def dt_upload(self: "Remote", transfer: Transfer, **kwargs: Any) -> None:
+    """Upload a given file to the given folderish document on the server.
+
+    Note about possible duplicate creation via a race condition client <-> server.
+    Given the local *file* with the path "$HOME/some-folder/subfolder/file.odt",
+    the file name is "file.odt".
+
+    Scenario:
+        - Step 1: local, check for a doc with the path name "file.odt" => nothing returned, continuing;
+        - Step 2: server, a document with a path name set to "file.odt" is created;
+        - Step 3: local, create the document with the path name "file.odt".
+
+    Even if the elapsed time between steps 1 and 3 is really short, it may happen.
+
+    What can be done to prevent such scenario is not on the Nuxeo Drive side but on the server one.
+    For now, 2 options are possible but not qualified to be done in a near future:
+        - https://jira.nuxeo.com/browse/NXP-22959;
+        - create a new operation `Document.GetOrCreate` that ensures atomicity.
+
+    About kwargs, it is typically used to forward the chunk callback.
+    """
+    local_path = transfer.local_path
+    remote_path = transfer.remote_path
+    name = local_path.name
+    doc: Optional[Document] = None
+
+    log.info(
+        f"Direct Transfer of {local_path!r} into {remote_path!r} (replace_blob={transfer.replace_blob})"
+    )
+
+    if transfer.remote_ref:
+        # The remote ref is set, so it means either the file has already been uploaded,
+        # either a previous upload failed: the document was created, or not, and it has
+        # a blob attached, or not. In any cases, we need to ensure the user can upload
+        # without headhache.
+        doc = self.get_document_or_none(uid=transfer.remote_ref)
+
+    if not doc:
+        # We need to handle possbile duplicates based on the file name and
+        # the destination folder on the server.
+        # Note: using this way may still result in duplicates:
+        #  - the user created 2 documents with the same name on Web-UI or another way
+        #  - the user then deleted the 1st document
+        #  - the other document has a path like "name.TIMESTAMP"
+        # So then Drive will not see that document as a duplicate because it will check
+        # a path with "name" only.
+
+        # If we really want to avoid that situation, we should use that commented code:
+        """
+        # Note that it would be too much effort for the server, we do not want that!
+        for child in self.documents.get_children(path=parent_path):
+            # It is OK to have a folder and a file with the same name,
+            # but not 2 files or 2 folders with the same name
+            if child.title == file.name:
+                local_is_dir = file.isdir()
+                remote_is_dir = "Folderish" in child["facets"]
+                if local_is_dir is remote_is_dir:
+                    # Duplicate found!
+                    doc = child
+                    transfer.remote_ref = doc.uid
+                    break
+        """
+
+        doc = self.get_document_or_none(path=f"{remote_path}/{name}")
+        if doc:
+            transfer.remote_ref = doc.uid
+
+    if not transfer.replace_blob and doc and doc.properties.get("file:content"):
+        # The document already exists and has a blob attached. Ask the user what to do.
+        raise DirectTransferDuplicateFoundError(local_path, doc)
+
+    if not doc:
+        # Create the document on the server
+        nature = "File" if transfer.is_file else "Folder"
+        transfer.doctype = nature
+        doc = self.documents.create(
+            Document(name=name, type=nature, properties={"dc:title": name}),
+            parent_path=remote_path,
+        )
+        transfer.remote_ref = doc.uid
+
+    if not transfer.is_file:
+        transfer.uploaded = True
+        transfer.status = TransferStatus.DONE  # type: ignore
+
+    transfer.save()
+
+    if transfer.is_file:
+        # Upload the blob and attach it to the document
+        self.dt_do_upload(self, transfer, **kwargs)
+
+
+def dt_do_upload(self: "Remote", transfer: Transfer, **kwargs: Any) -> None:
+    """Upload a file with a batch.
+
+    If an exception happens at step 1 or 2, the upload will be continued the next
+    time the Processor handle the document (it will be postponed due to the error).
+
+    If the error was raised at step 1, the upload will not start from zero: it will
+    resume from the next chunk based on what previously chunks were sent.
+    This is dependent of the chunk TTL configured on the server (it must be large enough
+    to handle big files).
+
+    If the error was raised at step 2, the step 1 will be checked to ensure the blob
+    was successfully uploaded. But it most cases, nothing will be uploaded twice.
+    Also, it the error is one of HTTP 502 or 503, the Processor will check for
+    the file existence to bypass errors happening *after* the operation was successful.
+    If it exists, the error is skipped and the upload is seen as a success.
+    """
+    # Step 1: upload the blob
+    blob = self.dt_upload_chunks(self, transfer, **kwargs)
+
+    # Step 2: link the uploaded blob to the document
+    self.dt_link_blob_to_doc(self, transfer, blob)
+
+    # Transfer is completed, delete the upload from the database
+    transfer.status = TransferStatus.DONE  # type: ignore
+    transfer.save()
+
+
+def dt_upload_chunks(self: "Remote", transfer: Transfer, **kwargs: Any) -> FileBlob:
+    """Upload a blob by chunks or in one go."""
+
+    blob = FileBlob(str(transfer.local_path))
+    blob.name = transfer.local_path.name
+
+    # The upload was already done, no need to make useless calls to the server
+    if transfer.uploaded:
+        # Set those attributes as FileBlob does not have them
+        # and they are required for the step 2 of .dt_do_upload()
+        blob.batch_id = transfer.batch.uid
+        blob.fileIdx = transfer.batch.upload_idx
+        return blob
+
+    chunk_size = 0
+
+    # Check if the associated batch still exists server-side
+    if transfer.batch.uid:
+        # When fetching for an eventual batch, specifying the file index
+        # is not possible for S3 as there is no blob at the current index
+        # until the S3 upload is done itself and the call to
+        # batch.complete() done.
+        file_idx = None if transfer.batch.is_s3() else transfer.batch.upload_idx
+
+        try:
+            self.uploads.get(transfer.batch.uid, file_idx=file_idx)
+        except Exception:
+            transfer.batch.uid = None
+            log.debug(f"No associated batch found, restarting from zero", exc_info=True)
+        else:
+            log.debug(f"Associated batch found, resuming the upload")
+            transfer.batch.service = self.uploads
+            chunk_size = transfer.chunk_size
+
+    if not transfer.batch.uid:
+        # .uploads.handlers() result is cached, so it is convenient to call it each time here
+        # in case the server did not answer correctly the previous time and thus S3 would
+        # be completely disabled because of a one-time server error.
+        handler = "s3" if self.uploads.has_s3() else ""
+
+        # Create a new batch and save it in the DB
+        transfer.batch = self.uploads.batch(handler=handler)
+        transfer.save()
+
+    # Set those attributes as FileBlob does not have them
+    # and they are required for the step 2 of .dt_do_upload()
+    blob.batch_id = transfer.batch.uid
+    blob.fileIdx = transfer.batch.upload_idx
+
+    # By default, Options.chunk_size is 20, so chunks will be 20MiB.
+    # It can be set to a value between 1 and 20 through the config.ini
+    chunk_size = chunk_size or (Options.chunk_size * 1024 * 1024)
+
+    # For the upload to be chunked, the Options.chunk_upload must be True
+    # and the blob must be bigger than Options.chunk_limit, which by default
+    # is equal to Options.chunk_size.
+    chunked = Options.chunk_upload and blob.size > Options.chunk_limit * 1024 * 1024
+
+    try:
+        uploader: Uploader = transfer.batch.get_uploader(
+            blob,
+            chunked=chunked,
+            chunk_size=chunk_size,
+            callback=kwargs.get("chunk_callback", None),
+        )
+        log.debug(f"Using {type(uploader).__name__!r} uploader")
+
+        transfer.chunk_size = uploader.chunk_size
+
+        if uploader.chunked:
+            for up in uploader.iter_upload():
+                # Save the progression
+                transfer.uploaded_size = up.blob.uploadedSize
+
+                # Handle status changes every time a chunk is sent
+                if transfer.status not in (TransferStatus.ONGOING, TransferStatus.DONE):
+                    raise DirectTransferPaused(transfer.id)
+
+        else:
+            uploader.upload()
+            transfer.uploaded_size = blob.size
+
+        if transfer.batch.is_s3():
+            # Complete the S3 upload
+            transfer.batch.complete()
+
+        transfer.uploaded = True
+        transfer.batch.upload_idx = 0
+        transfer.save()
+
+        return blob
+    finally:
+        if blob.fd:
+            blob.fd.close()
+
+
+def dt_link_blob_to_doc(self: "Remote", transfer: Transfer, blob: FileBlob) -> None:
+    """Link the given uploaded *blob* to the given document.
+    The method exists only to be able to test upload errors.
+    """
+    try:
+        self.execute(
+            command="Blob.AttachOnDocument",
+            input_obj=blob,
+            headers={"Nuxeo-Transaction-Timeout": str(TX_TIMEOUT)},
+            timeout=TX_TIMEOUT,
+            xpath="file:content",
+            void_op=True,
+            document=transfer.remote_ref,
+        )
+    except HTTPError as exc:
+        if exc.status in (502, 503):
+            # As seen with NXDRIVE-1753, an uploaded file may have worked
+            # but for some reason the final state is in error. So, let's
+            # check if the document is present on the server to bypass
+            # (infinite|useless) retries.
+            if self.get_document_or_none(uid=transfer.remote_ref):
+                return
+        raise
diff --git a/nxdrive/engine/dao/sqlite.py b/nxdrive/engine/dao/sqlite.py
index 5678fcded..aacf22797 100644
--- a/nxdrive/engine/dao/sqlite.py
+++ b/nxdrive/engine/dao/sqlite.py
@@ -112,9 +112,6 @@
     ("unsynchronized", "moved"): "unsynchronized",
     ("unsynchronized", "synchronized"): "unsynchronized",
     ("unsynchronized", "deleted"): "remotely_deleted",
-    # Direct Transfer
-    ("direct", "unknown"): "direct_transfer",
-    ("direct", "deleted"): "direct_transfer_replace_blob",
 }
 
 
@@ -717,7 +714,6 @@ def _migrate_db(self, cursor: Cursor, version: int) -> None:
 
             # Delete the table
             cursor.execute("DROP TABLE Uploads_backup;")
-
             self.store_int(SCHEMA_VERSION, 7)
 
     def _create_table(self, cursor: Cursor, name: str, force: bool = False) -> None:
@@ -1176,21 +1172,6 @@ def update_local_modification_time(self, row: DocPair, info: FileInfo) -> None:
                 (info.last_modification_time, row.id),
             )
 
-    def update_pair_state(self, row: DocPair) -> None:
-        """Update local, remote and pair states of a given *doc_pair*.
-        States should already be defined in the *doc_pair* attributes,
-        the goal is only to save them in the database.
-        """
-        row.pair_state = self._get_pair_state(row)
-
-        with self.lock:
-            con = self._get_write_connection()
-            c = con.cursor()
-            c.execute(
-                "UPDATE States SET local_state = ?, remote_state = ?, pair_state = ? WHERE id = ?",
-                (row.local_state, row.remote_state, row.pair_state, row.id),
-            )
-
     def get_valid_duplicate_file(self, digest: str) -> Optional[DocPair]:
         """Find a file already synced with the same digest as the given *digest*."""
         c = self._get_read_connection().cursor()
diff --git a/nxdrive/engine/engine.py b/nxdrive/engine/engine.py
index b6b3ac74e..cf4be9003 100644
--- a/nxdrive/engine/engine.py
+++ b/nxdrive/engine/engine.py
@@ -28,6 +28,7 @@
     DelAction,
     TransferStatus,
 )
+from ..direct_transfer import DirectTransferManager
 from ..exceptions import (
     EngineInitError,
     InvalidDriveException,
@@ -41,7 +42,6 @@
     current_thread_id,
     find_icon,
     find_suitable_tmp_dir,
-    get_tree_list,
     if_frozen,
     safe_filename,
     set_path_readonly,
@@ -56,6 +56,7 @@
 from .workers import Worker
 
 if TYPE_CHECKING:
+    from ..direct_transfer import DirectTransferManager  # noqa
     from ..manager import Manager  # noqa
 
 __all__ = ("Engine", "ServerBindingSettings")
@@ -101,9 +102,8 @@ class Engine(QObject):
     online = pyqtSignal()
 
     # Direct Transfer
-    directTranferDuplicateError = pyqtSignal(Path, Document)
+    directTransferDuplicateError = pyqtSignal(Path, Document)
     directTranferError = pyqtSignal(Path)
-    directTranferStatus = pyqtSignal(Path, bool)
 
     type = "NXDRIVE"
     # Folder locker - LocalFolder processor can prevent
@@ -216,6 +216,19 @@ def __init__(
         # Pause in case of no more space on the device
         self.noSpaceLeftOnDevice.connect(self.suspend)
 
+        # Set the Direct Transfer database file
+        db = self.manager.home / "direct_transfer" / f"{self.uid}.db"
+        db.parent.mkdir(exist_ok=True)
+
+        # And create the Direct Transfer manager
+        self.dt_manager = DirectTransferManager(
+            str(db),
+            self.uid,
+            self.remote,
+            done_callback=self.manager.directTransferStats.emit,
+            dupe_callback=self.directTransferDuplicateError.emit,
+        )
+
     def __repr__(self) -> str:
         return (
             f"<{type(self).__name__} "
@@ -388,70 +401,32 @@ def delete_doc(self, path: Path, mode: DelAction = None) -> None:
                     f"{doc_pair.remote_parent_path}/{doc_pair.remote_ref}"
                 )
 
-    def direct_transfer(self, local_paths: Set[Path], remote_ref: str) -> None:
+    def direct_transfer(
+        self, local_paths: Set[Path], remote_path: str, start: bool = True
+    ) -> None:
         """Plan the Direct Transfer."""
-        # self.directTranferStatus.emit(local_path[0], True)
-
-        def plan(path: Path, remote_uid: str) -> None:
-            """Actions to do (refactored in a function to prevent duplicate code between files and folders)."""
-            # Save the remote folder's reference into the file/folder xattrs
-            try:
-                self.local.set_remote_id(path, remote_uid)
-            except PermissionError:
-                log.warning(
-                    f"Cannot set the remote ID on {path!r}, skipping the upload"
-                )
-                return
-
-            # Add the path into the database to plan the upload
-            info = self.local.get_info(path, check=False)
-            self.dao.insert_local_state(info, parent_path=None, local_state="direct")
-
         # Save the remote location for next times
-        self.dao.update_config("dt_last_remote_location", remote_ref)
+        self.dao.update_config("dt_last_remote_location", remote_path)
 
-        for local_path in sorted(local_paths):
-            if local_path.is_file():
-                plan(local_path, remote_ref)
-            else:
-                tree = sorted(get_tree_list(local_path, remote_ref))
-                for remote_path, path in tree:
-                    plan(path, remote_path)
+        if self.dt_manager.is_completed:
+            # Create a new transfers session
+            self.dt_manager.reset()
+
+        # Add paths recursively and start uploading!
+        if local_paths:
+            self.dt_manager.add_all(local_paths, remote_path)
+            if start:
+                self.dt_manager.start()
 
     def direct_transfer_cancel(self, file: Path) -> None:
         """Cancel the Direct Transfer of the given local *file*."""
-        log.info(f"Direct Transfer of {file!r}, user choice: cancel the upload")
-
-        doc_pair = self.dao.get_state_from_local(file)
-        if not doc_pair:
-            # Magic teleportation?
-            log.warning("The doc pair disappeared?! Direct Transfer cancelled.")
-            return
-
-        # Cancel the upload, clean-up the database and local file
-        self.dao.remove_state(doc_pair)
-        self.local.remove_remote_id(file)
-        self.local.remove_remote_id(file, name="remote")
+        if self.dt_manager:
+            self.dt_manager.cancel(file)
 
     def direct_transfer_replace_blob(self, file: Path, doc: Document) -> None:
         """Replace the document's blob on the server."""
-        log.info(
-            f"Direct Transfer of {file!r}, user choice: "
-            f"replace the document's attached file (UID is {doc.uid!r})"
-        )
-
-        doc_pair = self.dao.get_state_from_local(file)
-        if not doc_pair:
-            # Magic teleportation?
-            log.warning("The doc pair disappeared?! Direct Transfer cancelled.")
-            return
-
-        # Plan the replacement of the document's blob on the server
-        doc_pair.remote_state = "deleted"
-        self.dao.update_pair_state(doc_pair)
-
-        # Repush the pair to be traited later
-        self.queue_manager.push(doc_pair)
+        if self.dt_manager:
+            self.dt_manager.replace_blob(file, doc.uid)
 
     def rollback_delete(self, path: Path) -> None:
         """ Re-synchronize a document when a deletion is cancelled. """
@@ -543,6 +518,8 @@ def resume(self) -> None:
             else:
                 thread.start()
         self.resume_suspended_transfers()
+        if self.dt_manager:
+            self.dt_manager.start()
         self.syncResumed.emit()
 
     def resume_transfer(self, nature: str, uid: int) -> None:
@@ -590,6 +567,8 @@ def suspend(self) -> None:
         log.info(f"Engine {self.uid} is suspending")
         self._pause = True
         self.dao.suspend_transfers()
+        if self.dt_manager:
+            self.dt_manager.stop()
         self.queue_manager.suspend()
         for thread in self._threads:
             thread.worker.suspend()
diff --git a/nxdrive/engine/processor.py b/nxdrive/engine/processor.py
index e859418c5..8f3b75f8a 100644
--- a/nxdrive/engine/processor.py
+++ b/nxdrive/engine/processor.py
@@ -31,7 +31,6 @@
     TransferStatus,
 )
 from ..exceptions import (
-    DirectTransferDuplicateFoundError,
     DownloadPaused,
     DuplicationDisabledError,
     NotFound,
@@ -42,7 +41,6 @@
     UploadPaused,
 )
 from ..objects import DocPair, RemoteFileInfo
-from ..options import Options
 from ..utils import is_generated_tmp_file, lock_path, safe_filename, unlock_path
 from .workers import EngineWorker
 
@@ -312,16 +310,11 @@ def _execute(self) -> None:
                 self._postpone_pair(doc_pair, "MAX_RETRY_ERROR")
             except HTTPError as exc:
                 if exc.status == 404:
-                    if doc_pair.local_state == "direct":
-                        # Happening while Direct Transfer'ring a document,
-                        # that means the parent folder is not yet created. Postpone.
-                        self._postpone_pair(doc_pair, "Parent not yet synced")
-                    else:
-                        # We saw it happened once a migration is done.
-                        # Nuxeo kept the document reference but it does
-                        # not exist physically anywhere.
-                        log.info(f"The document does not exist anymore: {doc_pair!r}")
-                        self.dao.remove_state(doc_pair)
+                    # We saw it happened once a migration is done.
+                    # Nuxeo kept the document reference but it does
+                    # not exist physically anywhere.
+                    log.info(f"The document does not exist anymore: {doc_pair!r}")
+                    self.dao.remove_state(doc_pair)
                 elif exc.status == 409:  # Conflict
                     # It could happen on multiple files drag'n drop
                     # starting with identical characters.
@@ -391,23 +384,9 @@ def _execute(self) -> None:
                     self._postpone_pair(doc_pair, "Trashing not possible")
                 else:
                     self._handle_pair_handler_exception(doc_pair, handler_name, exc)
-            except DirectTransferDuplicateFoundError as exc:
-                # Ask the user what to do when a possible duplicate can be created by a Direct Transfer call
-                log.info(str(exc))
-                self.engine.directTranferDuplicateError.emit(exc.file, exc.doc)
             except Exception as exc:
                 # Workaround to forward unhandled exceptions to sys.excepthook between all Qthreads
                 sys.excepthook(*sys.exc_info())
-
-                # Show a notification for Direct Transfer errors
-                if doc_pair.pair_state.startswith("direct_transfer"):
-                    file = (
-                        doc_pair.local_path
-                        if WINDOWS
-                        else Path(f"/{doc_pair.local_path}")
-                    )
-                    self.engine.directTranferError.emit(file)
-
                 self._handle_pair_handler_exception(doc_pair, handler_name, exc)
             finally:
                 if soft_lock:
@@ -474,56 +453,6 @@ def _handle_pair_handler_exception(
             log.exception("Unknown error")
             self.increase_error(doc_pair, f"SYNC_HANDLER_{handler_name}", exception=e)
 
-    def _synchronize_direct_transfer(
-        self, doc_pair: DocPair, replace_blob: bool = False
-    ) -> None:
-        """Direct Transfer of a local file."""
-        if WINDOWS:
-            file = doc_pair.local_path
-        else:
-            # The path retrieved from the database will have its starting slash trimmed, restore it
-            file = Path(f"/{doc_pair.local_path}")
-
-        if not file.exists():
-            self.engine.directTranferError.emit(file)
-            log.warning(
-                f"Cancelling Direct Transfer of {file!r} because it does not exist anymore"
-            )
-            self.dao.remove_state(doc_pair)
-            self.dao.remove_transfer("upload", file)
-            return
-
-        # The remote path is stored as the remote ref in xattrs of the file
-        parent_path = self.local.get_remote_id(file)
-        if not parent_path:
-            self.engine.directTranferError.emit(file)
-            log.warning(
-                f"Cancelling Direct Transfer of {file!r} because it has no remote path set"
-            )
-            self.dao.remove_state(doc_pair)
-            self.dao.remove_transfer("upload", file)
-            return
-
-        # Do the upload
-        self.remote.direct_transfer(
-            file, parent_path, self.engine.uid, replace_blob=replace_blob
-        )
-
-        # Clean-up
-        self.dao.remove_state(doc_pair)
-        self.local.remove_remote_id(file)
-        self.local.remove_remote_id(file, name="remote")
-
-        # Display a notification only for big files (to prevent notifications flood)
-        if not doc_pair.folderish and doc_pair.size >= Options.big_file * 1024 * 1024:
-            self.engine.directTranferStatus.emit(file, False)
-
-        self.engine.manager.directTransferStats.emit(doc_pair.folderish, doc_pair.size)
-
-    def _synchronize_direct_transfer_replace_blob(self, doc_pair: DocPair) -> None:
-        """Force the blob replacement of the remote document (choice done by the user)."""
-        self._synchronize_direct_transfer(doc_pair, replace_blob=True)
-
     def _synchronize_conflicted(self, doc_pair: DocPair) -> None:
         if doc_pair.local_state == "moved" and doc_pair.remote_state in (
             "moved",
diff --git a/nxdrive/engine/queue_manager.py b/nxdrive/engine/queue_manager.py
index 8a692c6b7..2b24553de 100644
--- a/nxdrive/engine/queue_manager.py
+++ b/nxdrive/engine/queue_manager.py
@@ -208,13 +208,6 @@ def push(self, state: Union[DocPair, QueueItem]) -> None:
                     f"{self._remote_file_queue.qsize()}"
                 )
             self.newItem.emit(row_id)
-        elif state.pair_state.startswith("direct_transfer"):
-            self._local_file_queue.put(state)
-            log.debug(
-                "Pushed to _local_file_queue, now of size: "
-                f"{self._local_file_queue.qsize()}"
-            )
-            self.newItem.emit(row_id)
         else:
             # deleted and conflicted
             log.info(f"Not processable state: {state!r}")
diff --git a/nxdrive/exceptions.py b/nxdrive/exceptions.py
index c50855a22..cd0aca37e 100644
--- a/nxdrive/exceptions.py
+++ b/nxdrive/exceptions.py
@@ -1,9 +1,6 @@
 # coding: utf-8
-from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
-from nuxeo.models import Document
-
 if TYPE_CHECKING:
     from .engine.engine import Engine  # noqa
 
@@ -14,26 +11,6 @@ class DriveError(Exception):
     pass
 
 
-class DirectTransferDuplicateFoundError(ValueError):
-    """
-    Exception raised when a duplicate file already exists on the server
-    and trying to Direct Transfer a local file with the same name.
-    """
-
-    def __init__(self, file: Path, doc: Document) -> None:
-        self.file = file
-        self.doc = doc
-
-    def __repr__(self) -> str:
-        return (
-            f"Document with the name {self.file.name!r} already found on the server: {self.doc}."
-            f"Direct Transfer of {self.file!r} postponed after the user decided what to do."
-        )
-
-    def __str__(self) -> str:
-        return repr(self)
-
-
 class DocumentAlreadyLocked(DriveError):
     """ In DirectEdit, a document is locked by someone else. """
 
diff --git a/nxdrive/gui/application.py b/nxdrive/gui/application.py
index 4ef6e4b88..d86afd665 100644
--- a/nxdrive/gui/application.py
+++ b/nxdrive/gui/application.py
@@ -45,6 +45,7 @@
 from ..gui.folders_dialog import DialogMixin, DocumentsDialog, FoldersDialog
 from ..notification import Notification
 from ..options import Options
+from ..state import State
 from ..translator import Translator
 from ..updater.constants import (
     UPDATE_STATUS_INCOMPATIBLE_SERVER,
@@ -202,6 +203,12 @@ def __init__(self, manager: "Manager", *args: Any) -> None:
         if Options.protocol_url:
             self._handle_nxdrive_url(Options.protocol_url)
 
+    @pyqtSlot()
+    def exit_app(self) -> None:
+        """Initiate the application exit."""
+        State.about_to_quit = True
+        self.quit()
+
     @if_frozen
     def add_qml_import_path(self, view: QQuickView) -> None:
         """
@@ -828,7 +835,7 @@ def auth() -> None:
 
     @pyqtSlot(object)
     def _connect_engine(self, engine: Engine) -> None:
-        engine.directTranferDuplicateError.connect(
+        engine.directTransferDuplicateError.connect(
             self._direct_transfer_duplicate_error
         )
         engine.syncStarted.connect(self.change_systray_icon)
diff --git a/nxdrive/gui/systray.py b/nxdrive/gui/systray.py
index c31e9790c..9493bfca9 100644
--- a/nxdrive/gui/systray.py
+++ b/nxdrive/gui/systray.py
@@ -79,7 +79,7 @@ def get_context_menu(self) -> QMenu:
         menu.addAction(
             style.standardIcon(QStyle.SP_DialogCloseButton),
             Translator.get("QUIT"),
-            self.application.quit,
+            self.application.exit_app,
         )
 
         return menu
diff --git a/nxdrive/logging_config.py b/nxdrive/logging_config.py
index d8c7c5885..549e61085 100644
--- a/nxdrive/logging_config.py
+++ b/nxdrive/logging_config.py
@@ -193,6 +193,9 @@ def configure(
     # NXDRIVE-1918: filter out botocore logging (too verbose)
     logging.getLogger("botocore").setLevel(logging.ERROR)
 
+    # NXDRIVE-2006: filter out peewee logging (too verbose)
+    logging.getLogger("peewee").setLevel(logging.ERROR)
+
 
 def get_handler(name: str) -> Optional[logging.Handler]:
     for handler in logging.getLogger().handlers:
diff --git a/nxdrive/manager.py b/nxdrive/manager.py
index 00f5f856c..85341a0de 100644
--- a/nxdrive/manager.py
+++ b/nxdrive/manager.py
@@ -369,6 +369,7 @@ def stop(self, euid: str = None) -> None:
                 continue
             if engine.is_started():
                 engine.stop()
+            engine.dt_manager.shutdown()
         self.osi.cleanup()
         self.dispose_db()
         self.stopped.emit()
diff --git a/nxdrive/notification.py b/nxdrive/notification.py
index 84f3409c4..68486ec60 100644
--- a/nxdrive/notification.py
+++ b/nxdrive/notification.py
@@ -435,21 +435,6 @@ def __init__(self, file: Path) -> None:
         )
 
 
-class DirectTransferStatus(Notification):
-    """Display a notification when a Direct Transfer is starting or finished."""
-
-    def __init__(self, file: Path, status: bool) -> None:
-        label = "DIRECT_TRANSFER_START" if status else "DIRECT_TRANSFER_END"
-        values = [file.name]
-        super().__init__(
-            label,
-            title="Direct Transfer",
-            description=Translator.get(label, values=values),
-            level=Notification.LEVEL_INFO,
-            flags=Notification.FLAG_VOLATILE | Notification.FLAG_BUBBLE,
-        )
-
-
 class ErrorOpenedFile(Notification):
     def __init__(self, path: str, is_folder: bool) -> None:
         values = [short_name(path), APP_NAME]
@@ -528,16 +513,11 @@ def _connect_engine(self, engine: "Engine") -> None:
         engine.errorOpenedFile.connect(self._errorOpenedFile)
         engine.longPathError.connect(self._longPathError)
         engine.directTranferError.connect(self._direct_transfer_error)
-        engine.directTranferStatus.connect(self._direct_transfer_status)
 
     def _direct_transfer_error(self, file: Path) -> None:
         """Display a notification when a Direct Transfer is in error."""
         self.send_notification(DirectTransferError(file))
 
-    def _direct_transfer_status(self, file: Path, status: bool) -> None:
-        """Display a notification when a Direct Transfer is starting or finished."""
-        self.send_notification(DirectTransferStatus(file, status))
-
     def _errorOpenedFile(self, doc: DocPair) -> None:
         self.send_notification(ErrorOpenedFile(str(doc.local_path), doc.folderish))
 
diff --git a/nxdrive/report.py b/nxdrive/report.py
index 0da4c4fed..138c0bb06 100644
--- a/nxdrive/report.py
+++ b/nxdrive/report.py
@@ -71,6 +71,28 @@ def copy_logs(self, myzip: ZipFile) -> None:
             except Exception:
                 log.exception(f"Impossible to copy the log {rel_path!r}")
 
+    def copy_direct_transfer_dbs(self, myzip: ZipFile) -> None:
+        """Copy all log Direct Transfer databases to the ZIP report."""
+
+        folder = self._manager.home / "direct_transfer"
+        if not folder.is_dir():
+            return
+
+        for path in folder.iterdir():
+            if not path.is_file():
+                continue
+            if path.suffix != ".db":
+                continue
+
+            comp = ZIP_DEFLATED
+            rel_path = path.relative_to(self._manager.home)
+            try:
+                myzip.write(str(path), str(rel_path), compress_type=comp)
+            except Exception:
+                log.exception(
+                    f"Impossible to copy the Direct Transfer database {rel_path!r}"
+                )
+
     @staticmethod
     def copy_db(myzip: ZipFile, dao: "EngineDAO") -> None:
         """
@@ -127,6 +149,9 @@ def generate(self) -> None:
                 log.info(f"Engine metrics: {engine.get_metrics()!r}")
                 self.copy_db(zip_, engine.dao)
 
+            # Direct Transfer databases
+            self.copy_direct_transfer_dbs(zip_)
+
             # Logs
             self.copy_logs(zip_)
 
diff --git a/nxdrive/state.py b/nxdrive/state.py
new file mode 100644
index 000000000..bc527af83
--- /dev/null
+++ b/nxdrive/state.py
@@ -0,0 +1,14 @@
+# coding: utf-8
+"""
+General application state.
+Introduced in Nuxeo Drive 4.5.0.
+
+Available parameters:
+
+- about_to_quit
+    This is set from the GUI when clicking on Quit.
+
+"""
+from types import SimpleNamespace
+
+State = SimpleNamespace(about_to_quit=False)
diff --git a/requirements.txt b/requirements.txt
index a3b4bd015..beff40f2e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -95,6 +95,8 @@ nuxeo[s3]==2.4.3 \
 pathtools==0.1.2 \
     --hash=sha256:7c35c5421a39bb82e58018febd90e3b6e5db34c5443aaaf742b3f33d4655f1c0
     # via watchdog
+peewee==3.13.1 \
+    --hash=sha256:9492af4d1f8e18a7fa0e930960315b38931286ea0f1659bbd5503456cffdacde
 psutil==5.7.0 \
     --hash=sha256:298af2f14b635c3c7118fd9183843f4e73e681bb6f01e12284d4d70d48a60953 \
     --hash=sha256:75e22717d4dbc7ca529ec5063000b2b294fc9a367f9c9ede1f65846c7955fd38 \
diff --git a/tests/conftest.py b/tests/conftest.py
index 4d413e5e1..9085d0a5f 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -74,6 +74,9 @@ def no_warnings(recwarn):
         elif "type=SocketKind.SOCK_STREAM" in message:
             # Socket leaks on macOS, may need investigations
             continue
+        elif "resolve package from __spec__ or __package__" in message:
+            # Not related to our stuff
+            continue
 
         warn = f"{warning.filename}:{warning.lineno} {message}"
         print(warn, file=sys.stderr)
diff --git a/tests/functional/conftest.py b/tests/functional/conftest.py
index 712fb2a46..f408b787c 100644
--- a/tests/functional/conftest.py
+++ b/tests/functional/conftest.py
@@ -1,22 +1,36 @@
 # coding: utf-8
 from logging import getLogger
 from pathlib import Path
-from random import randint
 from typing import Callable, Optional
 from uuid import uuid4
 
 import pytest
-
 from faker import Faker
 from nuxeo.documents import Document
 from nuxeo.users import User
 from nxdrive.manager import Manager
 
 from .. import env
+from ..utils import salt
 
 log = getLogger(__name__)
 
 
+@pytest.fixture(scope="session")
+def app():
+    """Fixture required to be able to process events and quit smoothly the application."""
+    from PyQt5.QtCore import QCoreApplication, QTimer
+
+    app = QCoreApplication([])
+
+    # Little trick here! See Application.__init__() for details.
+    timer = QTimer()
+    timer.timeout.connect(lambda: None)
+    timer.start(100)
+
+    yield app
+
+
 @pytest.fixture(scope="session")
 def faker() -> Callable[[], Faker]:
     """
@@ -84,7 +98,8 @@ def user_factory(request, server, faker):
 
     def _make_user(username: str = "", password: str = "Administrator"):
         first_name, last_name = fake.name().split(" ", 1)
-        username = username or f"{first_name.lower()}-{randint(1, 99_999)}"
+        username = username or first_name.lower()
+        username = salt(username)
         properties = {
             "lastName": last_name,
             "firstName": first_name,
@@ -118,6 +133,7 @@ def _make(
         enable_sync: bool = False,
     ):
         title = title or str(uuid4())
+        title = salt(title)
         new = Document(name=title, type=nature, properties={"dc:title": title})
         obj = server.documents.create(new, parent_path=env.WS_DIR)
         request.addfinalizer(obj.delete)
diff --git a/tests/functional/test_direct_edit.py b/tests/functional/test_direct_edit.py
index 0f5595ea4..372c66cb4 100644
--- a/tests/functional/test_direct_edit.py
+++ b/tests/functional/test_direct_edit.py
@@ -19,6 +19,7 @@ def __init__(self, url: str, user: str):
         self._user = user
         self._stopped = True
         self._invalid_credentials = False
+        self.dt_manager = None
 
     def get_binder(self):
         """
diff --git a/tests/functional/test_direct_transfer.py b/tests/functional/test_direct_transfer.py
new file mode 100644
index 000000000..5d9fd942b
--- /dev/null
+++ b/tests/functional/test_direct_transfer.py
@@ -0,0 +1,556 @@
+"""
+Test the Direct Transfer feature in different scenarii.
+"""
+from pathlib import Path
+from shutil import copyfile, copytree
+from time import sleep
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from nuxeo.exceptions import HTTPError
+from nuxeo.models import Document
+from nxdrive.constants import TransferStatus
+from nxdrive.direct_transfer import DirectTransferDuplicateFoundError
+from nxdrive.options import Options
+from nxdrive.utils import get_tree_list
+from requests.exceptions import ConnectionError
+
+from .. import ensure_no_exception
+from ..markers import not_windows
+
+LOCATION = Path(__file__).parent.parent
+
+
+def has_blob(engine, path: str) -> bool:
+    """Check that *self.file* exists on the server and has a blob attached.
+    As when doing a Direct Transfer, the document is first created on the server,
+    this is the only way to check if the blob upload has been finished successfully.
+    """
+    try:
+        doc = engine.remote.documents.get(path=path)
+    except Exception:
+        return False
+    return bool(doc.properties.get("file:content"))
+
+
+def sync_and_check(engine, should_have_blob: bool = True) -> None:
+    # Sync
+    wait_sync(engine)
+
+    # Check the file exists on the server and has a blob attached
+    path = f"{engine.ws_root.path}/{engine.file.name}"
+    blob = has_blob(engine, path)
+    assert blob is should_have_blob
+
+
+def wait_sync(engine):
+    """Wait for the Direct transfer session to finish."""
+    for _ in range(30):  # 30 sec maxi
+        if not engine.dt_manager.is_started:
+            break
+        sleep(1)
+
+
+@pytest.fixture
+def engine(tmp_path, app, server, manager_factory, obj_factory):
+    Options.synchronization_enabled = False
+
+    manager, engine = manager_factory()
+
+    # Attach the fake workspace folder
+    engine.ws_root = obj_factory()
+
+    # Grant RW ACLs for the current user
+    server.operations.execute(
+        command="Document.SetACE",
+        input_obj=engine.ws_root.uid,
+        user=engine.remote_user,
+        permission="ReadWrite",
+        grant=True,
+    )
+
+    # Start the engine!
+    engine.start()
+
+    # Start an empty DT session to create the .dt_manager attribute
+    engine.direct_transfer([], "")
+
+    # Lower chunk_* options to have chunked uploads without having to create big files
+    default_chunk_limit = Options.chunk_limit
+    default_chunk_size = Options.chunk_size
+    Options.chunk_limit = 1
+    Options.chunk_size = 1
+
+    # The file used for the Direct Transfer (10 MiB)
+    engine.file = tmp_path / f"{uuid4()}.bin"
+    engine.file.write_bytes(b"0" * 1024 * 1024 * 10)
+
+    try:
+        with manager:
+            yield engine
+    finally:
+        # Restore options
+        Options.chunk_limit = default_chunk_limit
+        Options.chunk_size = default_chunk_size
+
+
+def test_simple(engine):
+    """Simple Direct Transfer test."""
+    with ensure_no_exception():
+        engine.direct_transfer([engine.file], engine.ws_root.path)
+        sync_and_check(engine)
+
+    # Ensure the remote path is saved for next times
+    assert engine.dao.get_config("dt_last_remote_location") == engine.ws_root.path
+
+
+def test_with_engine_not_started(engine):
+    """A Direct Transfer should work even if engines are stopped."""
+    engine.stop()
+
+    with ensure_no_exception():
+        engine.direct_transfer([engine.file], engine.ws_root.path)
+        sync_and_check(engine)
+
+
+def test_duplicate_file_but_no_blob_attached(engine):
+    """The file already exists on the server but has no blob attached yet."""
+
+    def create(*_, **__):
+        """Patch Nuxeo.server.documents.create() to be able to check that nothing will be done."""
+        assert 0, "No twice creation should be done!"
+
+    # Create the document on the server, with no blob attached
+    new_doc = Document(
+        name=engine.file.name, type="File", properties={"dc:title": engine.file.name}
+    )
+    doc = engine.remote.documents.create(new_doc, parent_path=engine.ws_root.path)
+    assert doc.properties.get("file:content") is None
+
+    # The upload should work: the doc will be retrieved and the blob uploaded and attached
+    with patch.object(engine.dt_manager.remote.documents, "create", new=create):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+
+
+def test_duplicate_file_cancellation(engine):
+    """The file already exists on the server and has a blob attached.
+    Then, the user wants to cancel the transfer.
+    """
+
+    def cancel(file: Path, doc: Document) -> None:
+        """Mimic the user choosing to cancel the upload."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        engine.direct_transfer_cancel(file)
+
+    def upload(*_, **__):
+        """Patch Remote.upload() to be able to check that nothing will be uploaded."""
+        assert 0, "No twice upload should be done!"
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "dupe_callback", new=cancel):
+        with ensure_no_exception():
+            # 1st upload: OK
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # 2nd upload: it should be cancelled by the user
+            with patch.object(engine.dt_manager.remote, "upload", new=upload):
+                engine.direct_transfer([engine.file], engine.ws_root.path)
+                wait_sync(engine)
+
+    assert checkpoint
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+
+
+def test_duplicate_file_replace_blob(engine):
+    """The file already exists on the server and has a blob attached.
+    Then, the user wants to replace the blob.
+    """
+
+    def replace(file: Path, doc: Document) -> None:
+        """Mimic the user choosing to replace the blob on the server."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        engine.direct_transfer_replace_blob(file, doc)
+
+    checkpoint = False
+    content = b"blob changed!"
+
+    with patch.object(engine.dt_manager, "dupe_callback", new=replace):
+        with ensure_no_exception():
+            # 1st upload: OK
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # To ease testing, we change local file content
+            engine.file.write_bytes(content)
+
+            # 2nd upload: the blob should be replaced on the server
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # Manual upload to test the exception raised
+            dt_man = engine.dt_manager
+            engine.direct_transfer([engine.file], engine.ws_root.path, start=False)
+            with pytest.raises(DirectTransferDuplicateFoundError) as exc:
+                dt_man.remote.dt_upload(dt_man.remote, dt_man.transfers[0])
+            assert repr(exc)
+            assert str(exc)
+
+    assert checkpoint
+
+    # Ensure there is only 1 document on the server
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == engine.file.name
+    assert engine.remote.get_blob(children[0].uid) == content
+
+
+def test_pause_upload_manually(engine):
+    """
+    Pause the transfer by simulating a click on the pause/resume icon
+    on the current upload in the systray menu.
+    """
+
+    def callback(*_):
+        """
+        This will mimic what is done in SystrayTranfer.qml:
+            - call API.pause_transfer() that will call:
+                - engine.dao.pause_transfer(nature, transfer_uid)
+        Then the upload will be paused in Remote.upload().
+        """
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_pause_upload_automatically(engine):
+    """
+    Pause the transfer by simulating an application exit
+    or clicking on the Suspend menu entry from the systray.
+    """
+
+    def callback(*_):
+        """This will mimic what is done in SystrayMenu.qml: suspend the app."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        assert len(engine.dt_manager.uploading) == 1
+
+        # Suspend!
+        engine.suspend()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.resume()
+    sync_and_check(engine)
+
+
+def test_modifying_paused_upload(engine):
+    """Modifying a paused upload should discard the current upload."""
+
+    def callback(*_):
+        """Pause the upload and apply changes to the document."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+        # Apply changes to the file
+        engine.file.write_bytes(content)
+
+    checkpoint = False
+    content = b"locally changed"
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+
+    # Check the local content is correct
+    sync_and_check(engine)
+    assert engine.file.read_bytes() == content
+
+    # And check the remote content is correct too
+    blob = engine.remote.get_blob(engine.dt_manager.transfers[0].remote_ref)
+    assert blob == content
+
+
+@not_windows(
+    reason="Cannot test the behavior as the local deletion is blocked by the OS."
+)
+def test_deleting_paused_upload(engine):
+    """Deleting a paused upload should discard the current upload."""
+
+    def callback(*_):
+        """Pause the upload and delete the document."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        # Ensure we have 1 ongoing upload
+        transfers = engine.dt_manager.uploading
+        assert len(transfers) == 1
+
+        # Pause the upload
+        transfers[0].status = TransferStatus.PAUSED
+        transfers[0].save()
+
+        # Remove the document
+        # (this is the problematic part on Windows, because for the
+        #  file descriptor to be released we need to escape from
+        #  Remote.upload(), which is not possible from here)
+        engine.file.unlink()
+        assert not engine.file.exists()
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager, "chunk_callback", new=callback):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+    assert checkpoint
+
+    # Resume the upload
+    engine.dt_manager.start()
+    sync_and_check(engine, should_have_blob=False)
+
+
+def test_not_server_error_upload(engine):
+    """Test an error happening after chunks were uploaded."""
+
+    def bad(*args, **kwargs):
+        """Simulate an exception that is not handled by the Processor."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        raise ValueError("Mocked exception")
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager.remote, "dt_link_blob_to_doc", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Check thransfer error details
+    transfer = engine.dt_manager.uploading[0]
+    assert transfer.error_cont == 1
+    assert transfer.error_cont == "ValueError"
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_server_error_but_upload_ok(engine):
+    """
+    Test an error happening after chunks were uploaded and the Blob.AttachOnDocument operation call.
+    This could happen if a proxy does not understand well the final requests as seen in NXDRIVE-1753.
+    """
+
+    def bad(*args, **kwargs):
+        # Call the original method to effectively end the upload process
+        execute_orig(*args, **kwargs)
+
+        if kwargs.get("command", "") == "Blob.AttachOnDocument":
+            # The file should be present on the server
+            children = engine.remote.documents.get_children(path=engine.ws_root.path)
+            assert len(children) == 1
+            assert children[0].title == engine.file.name
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # And throw an error
+            stack = (
+                "The proxy server received an invalid response from an upstream server."
+            )
+            raise HTTPError(status=502, message="Mocked Proxy Error", stacktrace=stack)
+
+    execute_orig = engine.remote.execute
+
+    with patch.object(engine.dt_manager.remote, "execute", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+        assert len(engine.dt_manager.uploading) == 0
+
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_server_error_upload(engine):
+    """Test a server error happening after chunks were uploaded, at the Blob.AttachOnDocument operation call."""
+
+    def bad(*args, **kwargs):
+        """Simulate a server error."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        raise ConnectionError("Mocked exception")
+
+    checkpoint = False
+
+    with patch.object(engine.dt_manager.remote, "dt_link_blob_to_doc", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_chunk_upload_error(engine):
+    """Test a server error happening while uploading chunks."""
+
+    def bad(*args, **kwargs):
+        """Simulate an error after the upload of chunk n°1."""
+        nonlocal checkpoint
+        checkpoint = True
+
+        nonlocal number
+
+        if number == 0:
+            number += 1
+            return send_data_orig(*args, **kwargs)
+        else:
+            raise ConnectionError("Mocked error")
+
+    checkpoint = False
+    number = 0
+    send_data_orig = engine.dt_manager.remote.uploads.send_data
+
+    with patch.object(engine.dt_manager.remote.uploads, "send_data", new=bad):
+        with ensure_no_exception():
+            engine.direct_transfer([engine.file], engine.ws_root.path)
+            wait_sync(engine)
+
+            # There should be 1 upload with ONGOING transfer status
+            assert len(engine.dt_manager.uploading) == 1
+
+            # The file exists on the server but has no blob yet
+            path = f"{engine.ws_root.path}/{engine.file.name}"
+            assert not has_blob(engine, path)
+
+    assert checkpoint
+
+    # Restart the upload
+    engine.dt_manager.start()
+    sync_and_check(engine)
+
+
+def test_folder(tmp_path, engine):
+    """Test the Direct Transfer on a folder containing files and a sufolder."""
+
+    # We need more data for this test
+    folder = tmp_path / str(uuid4())
+    copytree(LOCATION / "resources", folder, copy_function=copyfile)
+
+    # Get folders and files tests will handle
+    tree = {path: rpath for rpath, path in get_tree_list(folder, engine.ws_root.path)}
+    files = [path for path in tree.keys() if path.is_file()]
+    folders = [path for path in tree.keys() if path.is_dir()]
+    engine.tree = tree
+
+    def check() -> None:
+        # Sync
+        wait_sync(engine)
+
+        # Check files exist on the server with their attached blob
+        for file in files:
+            path = f"{tree[file]}/{file.name}"
+            assert has_blob(engine, path)
+
+        # Check subfolders
+        for folder in folders:
+            assert engine.remote.documents.get(path=tree[folder])
+
+    with ensure_no_exception():
+        engine.direct_transfer([folder], engine.ws_root.path)
+        check()
+
+    # Ensure there is only 1 folder created at the workspace root
+    children = engine.remote.documents.get_children(path=engine.ws_root.path)
+    assert len(children) == 1
+    assert children[0].title == folder.name
+
+    # All has been uploaded
+    assert engine.dt_manager.is_completed
diff --git a/tests/functional/test_direct_transfer_manager.py b/tests/functional/test_direct_transfer_manager.py
new file mode 100644
index 000000000..f335ed866
--- /dev/null
+++ b/tests/functional/test_direct_transfer_manager.py
@@ -0,0 +1,282 @@
+# coding: utf-8
+"""
+Test the Direct Transfer manager.
+"""
+from datetime import datetime
+from pathlib import Path
+from typing import List
+from unittest.mock import patch
+from uuid import uuid4
+
+import pytest
+from nuxeo.models import Batch
+from nxdrive.constants import TransferStatus
+from nxdrive.direct_transfer import DirectTransferManager
+from nxdrive.direct_transfer.models import Session, Transfer
+from nxdrive.options import Options
+
+
+class FakeRemote:
+    """Fake Remote client."""
+
+
+@pytest.fixture
+def manager(tmp_path):
+    db = tmp_path / f"{uuid4()}.db"
+    remote = FakeRemote()
+    yield DirectTransferManager(str(db), "UID", remote)
+
+
+def create_tree(root: Path) -> List[Path]:
+    """Create the tree for tests."""
+    paths = []
+    size = 0
+
+    # Folders
+    folder = root / "one folder"
+    subfolder = folder / "subfolder"
+    subfolder.mkdir(parents=True)
+    paths.append(folder)
+    paths.append(subfolder)
+
+    # Files
+    file = root / "root-file.bin"
+    size += file.write_bytes(b"0" * Options.chunk_size * 1024 * 1024 + b"0" * 1024)
+    paths.append(file)
+
+    file = folder / "one file.txt"
+    size += file.write_text("azerty")
+    paths.append(file)
+    file = folder / "two files.txt"
+    size += file.write_text("azerty" * 2)
+    paths.append(file)
+
+    file = subfolder / "file1.txt"
+    size += file.write_text("azerty")
+    paths.append(file)
+    file = subfolder / "file2.txt"
+    size += file.write_text("azerty" * 2)
+    paths.append(file)
+
+    return paths, size
+
+
+def test_manager_empty(manager):
+    assert repr(manager)
+    assert manager.engine_uid == "UID"
+    assert len(manager.transfers) == 0
+    assert manager.is_completed
+    assert not manager.is_started
+    assert manager.progress == 100.0
+    assert manager.size == 0.0
+    assert manager.uploaded == 0.0
+
+    # Check the session
+    session = manager.session
+    assert session == Session.get(Session.id == 1)
+    assert session.started is None
+    assert session.finished is None
+    assert session.status is TransferStatus.ONGOING
+
+    # Check the transfers
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 0
+
+    # Check methods were injected into the remote client
+    for meth in (
+        "dt_upload",
+        "dt_do_upload",
+        "dt_link_blob_to_doc",
+        "dt_upload_chunks",
+    ):
+        assert hasattr(manager.remote, meth)
+
+
+def test_manager_session_resume(tmp_path, manager):
+    """Test resuming a session with 1 transfer."""
+    paths, _ = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    manager.add(paths[-1], remote_path)
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 1
+
+    manager.reinit()
+
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == 1
+    assert manager.transfers[0].local_path == paths[-1]
+    assert manager.session.id == 1
+
+    manager.add_all(paths, remote_path)
+    assert isinstance(manager.transfers, list)
+    assert len(manager.transfers) == len(paths)
+
+
+def test_manager_with_transfers(tmp_path, manager):
+    paths, size = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    # Starting an empty session is useless
+    assert not manager.transfers
+    manager.start()
+
+    # Add several paths
+    manager.add_all(paths, remote_path)
+    assert len(manager.transfers) == len(paths)
+
+    # Ensure the total size is well-computed
+    assert manager.size == size
+
+    # Add again all paths, this should not affect the actual paths list
+    for path in paths:
+        assert manager.add(path, remote_path) is None
+
+    # Add a problematic path, this should not affect the actual paths list
+    assert manager.add(Path("inexistent"), remote_path) is None
+
+    # Check transfers
+    transfer = manager.transfers[0]
+    assert isinstance(transfer, Transfer)
+    assert transfer.session == Session.get(Session.id == 1)
+    assert isinstance(transfer.local_path, Path)
+    assert transfer.remote_path == remote_path
+    assert not transfer.is_file
+    assert not transfer.file_size
+    assert transfer.status is TransferStatus.ONGOING
+    assert isinstance(transfer.batch, Batch)
+    assert not transfer.remote_ref
+    assert not transfer.uploaded
+    assert not transfer.uploaded_size
+    assert not transfer.chunk_size
+    assert not transfer.replace_blob
+
+
+def test_manager_reset_transfers(tmp_path, manager):
+    paths, _ = create_tree(tmp_path)
+    remote_path = "/default-domain/something"
+
+    # Add several paths
+    manager.add_all(paths, remote_path)
+    assert len(manager.transfers) == len(paths)
+
+    # Now, reset the manager
+    assert manager.session == Session.get(Session.id == 1)
+    manager.reset()
+    # It should have no effect as the session is not finished
+    assert manager.session == Session.get(Session.id == 1)
+
+    # Mimic a finish session
+    manager.session.started = datetime.now()
+    manager.session.finished = datetime.now()
+    manager.session.status is TransferStatus.DONE
+
+    # And reset for good!
+    manager.reset()
+    # A new session is created
+    assert manager.session == Session.get(Session.id == 2)
+    assert not manager.transfers
+
+
+def test_manager_start_stop_transfers(
+    tmp_path, app, server, manager_factory, obj_factory
+):
+    manager, engine = manager_factory()
+    engine.start()
+
+    dt_manager = DirectTransferManager(":memory:", "UID", engine.remote)
+
+    # Create a remote folder for our tests
+    remote_folder = obj_factory()
+    # Grant RW ACLs for the current user
+    server.operations.execute(
+        command="Document.SetACE",
+        input_obj=remote_folder.uid,
+        user=engine.remote_user,
+        permission="ReadWrite",
+        grant=True,
+    )
+
+    # Add several paths
+    paths, size = create_tree(tmp_path)
+    dt_manager.add_all(paths, remote_folder.path)
+
+    # Stopping now has no effect
+    assert not dt_manager.is_started
+    dt_manager.stop()
+    assert not dt_manager.is_started
+
+    # Ensure session times are not set
+    assert not dt_manager.session.started
+    assert not dt_manager.session.finished
+    assert dt_manager.session.status is TransferStatus.ONGOING
+
+    # Simulate a stop() while working
+
+    def should_stop_at_0() -> bool:
+        """Stop at the start."""
+        # Ensure the session is started
+        assert dt_manager.is_started
+        assert isinstance(dt_manager.session.started, datetime)
+
+        # Check we cannot start again
+        dt_manager.start()
+
+        # Check we cannot reset
+        dt_manager.reset()
+
+        # Mimic a call to stop() from the outside
+        dt_manager.stop()
+        assert not dt_manager.is_started
+        return True
+
+    count = 0
+
+    def should_stop_at_4() -> bool:
+        """Stop at the 4th upload."""
+        # Ensure the session is started
+        assert dt_manager.is_started
+        assert isinstance(dt_manager.session.started, datetime)
+
+        nonlocal count
+        count += 1
+        return count >= 4
+
+    with patch.object(dt_manager, "should_stop", new=should_stop_at_0):
+        dt_manager.start()
+    assert dt_manager.uploaded == 0
+    assert dt_manager.progress == 0.0
+
+    with patch.object(dt_manager, "should_stop", new=should_stop_at_4):
+        dt_manager.start()
+    assert dt_manager.uploaded > 0
+    assert dt_manager.progress > 0.0
+
+    # The finished time is still not yet set
+    assert not dt_manager.session.finished
+
+    # Now, really process transfers
+    date_original = dt_manager.session.started
+    dt_manager.start()
+
+    # Check sizes
+    assert dt_manager.uploaded == size
+    assert dt_manager.progress == 100.0
+
+    # Ensure the started time is not altered
+    assert dt_manager.session.started == date_original
+
+    # Ensure the session is completed
+    assert not dt_manager.is_started
+    assert dt_manager.is_completed
+    assert isinstance(dt_manager.session.finished, datetime)
+    assert dt_manager.session.status is TransferStatus.DONE
+
+    # Trying to start again is useless
+    dt_manager.start()
+
+    # Finally, check the latest transfer (to test SQL models completely)
+    transfer = dt_manager.transfers[-1]
+    assert transfer.is_file
+    assert transfer.batch.uid
+    assert transfer == Transfer.get(Transfer.id == transfer.id)
diff --git a/tests/functional/test_remote_client.py b/tests/functional/test_remote_client.py
index 4082e7e11..c7d5b6e35 100644
--- a/tests/functional/test_remote_client.py
+++ b/tests/functional/test_remote_client.py
@@ -1,16 +1,15 @@
 # coding: utf-8
 import pytest
-
 from nuxeo.models import Document
 
 
 @pytest.mark.parametrize(
     "username",
     [
-        "ndt-Alice",
-        "ndt-bob@bar.com",
-        # "ndt-éléonor",
-        # "ndt-東京スカイツリー",
+        "Alice",
+        "bob@bar.com",
+        # "éléonor",
+        # "東京スカイツリー",
     ],
 )
 def test_personal_space(manager_factory, tmp, nuxeo_url, user_factory, username):
diff --git a/tests/old_functional/test_direct_transfer.py b/tests/old_functional/test_direct_transfer.py
index bb0ef4a7a..20d8f9a68 100644
--- a/tests/old_functional/test_direct_transfer.py
+++ b/tests/old_functional/test_direct_transfer.py
@@ -1,30 +1,27 @@
 """
-Test the Direct Transfer feature in different scenarii.
+Test the Direct Transfer feature.
+This is not a simple copy of the same test file from functional test.
+Here, we test signals transition from within the application.
 """
 from contextlib import suppress
-from shutil import copyfile, copytree
+from shutil import copyfile
 from time import sleep
 from unittest.mock import patch
 from uuid import uuid4
 
-import pytest
-from nuxeo.exceptions import HTTPError
-from nuxeo.models import Document
-from nxdrive.constants import TransferStatus
 from nxdrive.options import Options
-from nxdrive.utils import get_tree_list
-from requests.exceptions import ConnectionError
 
 from .. import ensure_no_exception
-from ..markers import not_windows
-from .common import OneUserNoSync, OneUserTest
+from .common import OneUserTest
 
 
-class DirectTransfer:
+class TestDirectTransfer(OneUserTest):
     def setUp(self):
         # No sync root, to ease testing
         self.remote_1.unregister_as_root(self.workspace)
         self.engine_1.start()
+        self.engine_1.direct_transfer([], "")
+        self.dt_manager = self.engine_1.dt_manager
 
         # Lower chunk_* options to have chunked uploads without having to create big files
         self.default_chunk_limit = Options.chunk_limit
@@ -47,35 +44,23 @@ def tearDown(self):
 
         # Disconnect eventual signals to prevent failures when tests are run in parallel
         with suppress(TypeError):
-            self.engine_1.directTranferDuplicateError.disconnect(
+            self.engine_1.directTransferDuplicateError.disconnect(
                 self.app.user_choice_cancel
             )
         with suppress(TypeError):
-            self.engine_1.directTranferDuplicateError.disconnect(
+            self.engine_1.directTransferDuplicateError.disconnect(
                 self.app.user_choice_replace
             )
 
-    def has_blob(self) -> bool:
-        """Check that *self.file* exists on the server and has a blob attached.
-        As when doing a Direct Transfer, the document is first created on the server,
-        this is the only way to check if the blob upload has been finished successfully.
-        """
-        try:
-            doc = self.root_remote.documents.get(
-                path=f"{self.ws.path}/{self.file.name}"
-            )
-        except Exception:
-            return False
-        return bool(doc.properties.get("file:content"))
-
-    def no_uploads(self) -> bool:
-        """Check there is no ongoing uploads."""
-        assert not any(
-            u for u in self.engine_1.dao.get_uploads() if u.path == self.file
-        )
+    def wait_sync(self) -> None:
+        """Wait for the Direct transfer session to finish."""
+        for _ in range(30):  # 30 sec maxi
+            if not self.dt_manager.is_started:
+                break
+            sleep(1)
 
     def sync_and_check(self, should_have_blob: bool = True) -> None:
-        # Let time for uploads to be planned
+        # Let some time for the signal to transit (as this is async, the test would fail)
         sleep(3)
 
         # Sync
@@ -84,48 +69,22 @@ def sync_and_check(self, should_have_blob: bool = True) -> None:
         # Check the error count
         assert not self.engine_1.dao.get_errors(limit=0)
 
-        # Check the uploads count
-        assert not list(self.engine_1.dao.get_uploads())
-
         # Check the file exists on the server and has a blob attached
-        if should_have_blob:
-            assert self.has_blob()
-        else:
-            assert not self.has_blob()
-
-    def test_with_engine_not_started(self):
-        """A Direct Transfer should work even if engines are stopped."""
-        pytest.xfail("Waiting for NXDRIVE-1910")
-
-        engine = self.engine_1
-        engine.stop()
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with ensure_no_exception():
-            engine.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
-
-    def test_duplicate_file_but_no_blob_attached(self):
-        """The file already exists on the server but has no blob attached yet."""
-
-        # Create the document on the server, with no blob attached
-        new_doc = Document(
-            name=self.file.name, type="File", properties={"dc:title": self.file.name}
-        )
-        doc = self.root_remote.documents.create(new_doc, parent_path=self.ws.path)
-        assert doc.properties.get("file:content") is None
-
-        with ensure_no_exception():
-            # The upload should work: the doc will be retrieved and the blob uploaded and attached
-            self.engine_1.direct_transfer([self.file], self.ws.path)
-            self.sync_and_check()
+        blob = self.has_blob()
+        assert blob is should_have_blob
 
-        # Ensure there is only 1 document on the server
-        children = self.remote_document_client_1.get_children_info(self.workspace)
-        assert len(children) == 1
-        assert children[0].name == self.file.name
+    def has_blob(self) -> bool:
+        """Check that *self.file* exists on the server and has a blob attached.
+        As when doing a Direct Transfer, the document is first created on the server,
+        this is the only way to check if the blob upload has been finished successfully.
+        """
+        try:
+            doc = self.root_remote.documents.get(
+                path=f"{self.ws.path}/{self.file.name}"
+            )
+        except Exception:
+            return False
+        return bool(doc.properties.get("file:content"))
 
     def test_duplicate_file_cancellation(self):
         """The file already exists on the server and has a blob attached.
@@ -133,7 +92,7 @@ def test_duplicate_file_cancellation(self):
         """
 
         # Mimic the user clicking on "Cancel"
-        self.engine_1.directTranferDuplicateError.connect(self.app.user_choice_cancel)
+        self.engine_1.directTransferDuplicateError.connect(self.app.user_choice_cancel)
 
         def upload(*_, **__):
             """Patch Remote.upload() to be able to check that nothing will be uploaded."""
@@ -145,7 +104,7 @@ def upload(*_, **__):
             self.sync_and_check()
 
             # 2nd upload: it should be cancelled by the user
-            with patch.object(self.engine_1.remote, "upload", new=upload):
+            with patch.object(self.dt_manager.remote, "upload", new=upload):
                 self.engine_1.direct_transfer([self.file], self.ws.path)
                 self.sync_and_check()
 
@@ -163,7 +122,7 @@ def test_duplicate_file_replace_blob(self):
         """
 
         # Mimic the user clicking on "Replace"
-        self.engine_1.directTranferDuplicateError.connect(self.app.user_choice_replace)
+        self.engine_1.directTransferDuplicateError.connect(self.app.user_choice_replace)
 
         with ensure_no_exception():
             # 1st upload: OK
@@ -181,411 +140,8 @@ def test_duplicate_file_replace_blob(self):
         assert self.app.emitted
 
         # Ensure there is only 1 document on the server
-        children = self.remote_document_client_1.get_children_info(self.workspace)
+        children = self.engine_1.remote.documents.get_children(path=self.ws.path)
         assert len(children) == 1
-        assert children[0].name == self.file.name
-
+        assert children[0].title == self.file.name
         # Ensure the blob content was updated
-        doc = self.app.doc
-        assert isinstance(doc, Document)
-        assert doc.fetch_blob(xpath="file:content") == b"blob changed!"
-
-    def test_pause_upload_manually(self):
-        """
-        Pause the transfer by simulating a click on the pause/resume icon
-        on the current upload in the systray menu.
-        """
-
-        def callback(*_):
-            """
-            This will mimic what is done in SystrayTranfer.qml:
-                - call API.pause_transfer() that will call:
-                    - engine.dao.pause_transfer(nature, transfer_uid)
-            Then the upload will be paused in Remote.upload().
-            """
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-            assert dao.get_uploads_with_status(TransferStatus.PAUSED)
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check()
-
-    def test_pause_upload_automatically(self):
-        """
-        Pause the transfer by simulating an application exit
-        or clicking on the Suspend menu entry from the systray.
-        """
-
-        def callback(*_):
-            """This will mimic what is done in SystrayMenu.qml: suspend the app."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Suspend!
-            self.manager_1.suspend()
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-            assert dao.get_uploads_with_status(TransferStatus.SUSPENDED)
-
-        # Resume the upload
-        self.manager_1.resume()
-        self.sync_and_check()
-
-    def test_modifying_paused_upload(self):
-        """Modifying a paused upload should discard the current upload."""
-
-        def callback(*_):
-            """Pause the upload and apply changes to the document."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-            # Apply changes to the file
-            self.file.write_bytes(b"locally changed")
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check()
-        # Check the local content is correct
-        assert self.file.read_bytes() == b"locally changed"
-
-    @not_windows(
-        reason="Cannot test the behavior as the local deletion is blocked by the OS."
-    )
-    def test_deleting_paused_upload(self):
-        """Deleting a paused upload should discard the current upload."""
-
-        def callback(*_):
-            """Pause the upload and delete the document."""
-            # Ensure we have 1 ongoing upload
-            uploads = list(dao.get_uploads())
-            assert uploads
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # Pause the upload
-            dao.pause_transfer("upload", upload.uid, 50.0)
-
-            # Remove the document
-            # (this is the problematic part on Windows, because for the
-            #  file descriptor to be released we need to escape from
-            #  Remote.upload(), which is not possible from here)
-            self.file.unlink()
-            assert not self.file.exists()
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "upload_callback", new=callback):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-        # Resume the upload
-        engine.resume_transfer("upload", list(dao.get_uploads())[0].uid)
-        self.sync_and_check(should_have_blob=False)
-
-    def test_not_server_error_upload(self):
-        """Test an error happening after chunks were uploaded, at the NuxeoDrive.CreateFile operation call."""
-
-        def link_blob_to_doc(*args, **kwargs):
-            """Simulate an exception that is not handled by the Processor."""
-            raise ValueError("Mocked exception")
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "link_blob_to_doc", new=link_blob_to_doc):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.DONE
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-                # The doc should be in error
-                assert len(dao.get_errors(limit=0)) == 1
-
-        # Reset the error
-        for state in dao.get_errors():
-            dao.reset_error(state)
-
-        self.sync_and_check()
-
-    def test_server_error_but_upload_ok(self):
-        """
-        Test an error happening after chunks were uploaded and the Blob.AttachOnDocument operation call.
-        This could happen if a proxy does not understand well the final requests as seen in NXDRIVE-1753.
-        """
-        pytest.skip("Not yet implemented.")
-
-        def link_blob_to_doc(*args, **kwargs):
-            # Call the original method to effectively end the upload process
-            link_blob_to_doc_orig(*args, **kwargs)
-
-            # The file should be present on the server
-            assert self.remote_1.exists(f"/{self.file.name}")
-
-            # There should be 1 upload with ONGOING transfer status
-            uploads = list(dao.get_uploads())
-            assert len(uploads) == 1
-            upload = uploads[0]
-            assert upload.status == TransferStatus.ONGOING
-
-            # And throw an error
-            stack = (
-                "The proxy server received an invalid response from an upstream server."
-            )
-            raise HTTPError(status=502, message="Mocked Proxy Error", stacktrace=stack)
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-        link_blob_to_doc_orig = engine.remote.link_blob_to_doc
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "link_blob_to_doc", new=link_blob_to_doc):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be no upload as the Processor has checked the file existence
-                # on the server and so deleted the upload from the database
-                self.no_uploads()
-
-        self.sync_and_check()
-
-    def test_server_error_upload(self):
-        """Test a server error happening after chunks were uploaded, at the Blob.AttachOnDocument operation call."""
-
-        def link_blob_to_doc(*args, **kwargs):
-            """Simulate a server error."""
-            raise ConnectionError("Mocked exception")
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine.remote, "link_blob_to_doc", new=link_blob_to_doc):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.DONE
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-        self.sync_and_check()
-
-    def test_chunk_upload_error(self):
-        """Test a server error happening while uploading chunks."""
-
-        def send_data(*args, **kwargs):
-            """Simulate an error."""
-            raise ConnectionError("Mocked error")
-
-        def callback(upload):
-            """Patch send_data() after chunk 1 is sent."""
-            if len(upload.blob.uploadedChunkIds) == 1:
-                upload.service.send_data = send_data
-
-        engine = self.engine_1
-        dao = self.engine_1.dao
-        bad_remote = self.get_bad_remote()
-        bad_remote.upload_callback = callback
-
-        # There is no upload, right now
-        self.no_uploads()
-
-        with patch.object(engine, "remote", new=bad_remote):
-            with ensure_no_exception():
-                engine.direct_transfer([self.file], self.ws.path)
-                self.wait_sync()
-
-                # There should be 1 upload with ONGOING transfer status
-                uploads = list(dao.get_uploads())
-                assert len(uploads) == 1
-                upload = uploads[0]
-                assert upload.status == TransferStatus.ONGOING
-
-                # The file exists on the server but has no blob yet
-                assert not self.has_blob()
-
-        self.sync_and_check()
-
-
-class TestDirectTransfer(OneUserTest, DirectTransfer):
-    """Direct Transfer in "normal" mode, i.e.: when synchronization features are enabled."""
-
-    def setUp(self):
-        DirectTransfer.setUp(self)
-
-
-class TestDirectTransferNoSync(OneUserNoSync, DirectTransfer):
-    """Direct Transfer should work when synchronization features are not enabled."""
-
-    def setUp(self):
-        DirectTransfer.setUp(self)
-
-
-class DirectTransferFolder:
-    def setUp(self):
-        # No sync root, to ease testing
-        self.remote_1.unregister_as_root(self.workspace)
-        self.engine_1.start()
-
-        # The folder used for the Direct Transfer (must be > 1 MiB)
-        folder = self.location / "resources"
-        # Work with a copy of the folder to allow parallel testing
-        self.folder = self.tmpdir / str(uuid4())
-        copytree(folder, self.folder, copy_function=copyfile)
-
-        # Get folders and files tests will handle
-        tree = {path: rpath for rpath, path in get_tree_list(self.folder, self.ws.path)}
-        self.files = [path for path in tree.keys() if path.is_file()]
-        self.folders = [path for path in tree.keys() if path.is_dir()]
-        self.tree = tree
-
-        # Lower chunk_* options to have chunked uploads without having to create big files
-        self.default_chunk_limit = Options.chunk_limit
-        self.default_chunk_size = Options.chunk_size
-        Options.chunk_limit = 1
-        Options.chunk_size = 1
-
-    def tearDown(self):
-        # Restore options
-        Options.chunk_limit = self.default_chunk_limit
-        Options.chunk_size = self.default_chunk_size
-
-    def has_blob(self, file: str) -> bool:
-        """Check that *file* exists on the server and has a blob attached.
-        As when doing a Direct Transfer, the document is first created on the server,
-        this is the only way to check if the blob upload has been finished successfully.
-        """
-        try:
-            doc = self.root_remote.documents.get(path=f"{self.tree[file]}/{file.name}")
-        except Exception:
-            return False
-        return bool(doc.properties.get("file:content"))
-
-    def sync_and_check(self) -> None:
-        # Let time for uploads to be planned
-        sleep(3)
-
-        # Sync
-        self.wait_sync()
-
-        # Check the error count
-        assert not self.engine_1.dao.get_errors(limit=0)
-
-        # Check the uploads count
-        assert not list(self.engine_1.dao.get_uploads())
-
-        # Check files exist on the server with their attached blob
-        for file in self.files:
-            assert self.has_blob(file)
-
-        # Check subfolders
-        for folder in self.folders:
-            assert self.root_remote.documents.get(path=self.tree[folder])
-
-    def test_folder(self):
-        """Test the Direct Transfer on a folder containing files and a sufolder."""
-
-        # There is no upload, right now
-        assert not list(self.engine_1.dao.get_uploads())
-
-        with ensure_no_exception():
-            self.engine_1.direct_transfer([self.folder], self.ws.path)
-            self.sync_and_check()
-
-        # Ensure there is only 1 folder created at the workspace root
-        children = self.remote_1.get_children(self.ws.path)["entries"]
-        assert len(children) == 1
-        assert children[0]["title"] == self.folder.name
-
-        # All has been uploaded
-        assert not list(self.engine_1.dao.get_uploads())
-
-
-# NXDRIVE-2019
-
-
-class _TestDirectTransferFolder(OneUserTest, DirectTransferFolder):
-    """Direct Transfer in "normal" mode, i.e.: when synchronization features are enabled."""
-
-    def setUp(self):
-        DirectTransferFolder.setUp(self)
-
-
-class _TestDirectTransferFolderNoSync(OneUserNoSync, DirectTransferFolder):
-    """Direct Transfer should work when synchronization features are not enabled."""
-
-    def setUp(self):
-        DirectTransferFolder.setUp(self)
+        assert self.engine_1.remote.get_blob(children[0].uid) == b"blob changed!"
diff --git a/tests/resources/subfolder/README.md b/tests/resources/subfolder/README.md
new file mode 100644
index 000000000..084c7005a
--- /dev/null
+++ b/tests/resources/subfolder/README.md
@@ -0,0 +1 @@
+Used by Direct Transfer manager tests.
diff --git a/tests/resources/subfolder/subfile.txt b/tests/resources/subfolder/subfile.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/resources/subfolder/subfile2.txt b/tests/resources/subfolder/subfile2.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/resources/subfolder/subfile3.txt b/tests/resources/subfolder/subfile3.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/tools/whitelist.py b/tools/whitelist.py
index 8fb6c8e25..280a28697 100644
--- a/tools/whitelist.py
+++ b/tools/whitelist.py
@@ -14,6 +14,14 @@
 CliHandler.unbind_root  # Used by the arguments parser
 CliHandler.unbind_server  # Used by the arguments parser
 _.close_settings_too  # Used by Appiclation.show_filters()
+direct_transfer.models.BaseModel.Meta  # Used by Peewee
+direct_transfer.models.BaseModel.Meta.only_save_dirty  # Used by Peewee
+direct_transfer.models.BatchField.field_type  # Used by Peewee
+direct_transfer.models.BatchField.db_value  # Used by Peewee
+direct_transfer.models.BatchField.python_value  # Used by Peewee
+direct_transfer.models.DirectTransfer.doctype  # Used by Peewee
+direct_transfer.uploader.transfer.batch.service  # Used by the Nuxeo Python client
+direct_transfer.uploader.transfer.doctype  # Used by Peewee
 DocPair.last_sync_error_date  # Check NXDRIVE-1804
 Download.transfer_type  # Used in QML
 Engine.account  # Used in QML
@@ -34,8 +42,6 @@
 on_any_event  # Used by Watchdog
 _.row_factory  # Internal use of SQLite
 Processor._synchronize_conflicted  # Used by Processor._execute()
-Processor._synchronize_direct_transfer  # Used by Processor._execute()
-Procesor._synchronize_direct_transfer_replace_blob  # Used by Processor._execute()
 Processor._synchronize_deleted_unknown  # Used by Processor._execute()
 Processor._synchronize_locally_deleted  # Used by Processor._execute()
 Processor._synchronize_locally_resolved  # Used by Processor._execute()
