Upload and Watch & Watch Rearchitecture

.gitmodules

@@ -30,7 +30,7 @@
 	url = https://github.com/gabime/spdlog.git
 [submodule "dependencies/efsw"]
 	path = dependencies/efsw
-	url = https://github.com/SpartanJ/efsw.git
+	url = https://github.com/5cript/efsw.git
 [submodule "dependencies/promise-cpp"]
 	path = dependencies/promise-cpp
 	url = https://github.com/5cript/promise-cpp.git

_cmake/dependencies/efsw.cmake

@@ -3,8 +3,8 @@ if (FETCH_EFSW)
     # If you edit here, update work_dependencies.json
     FetchContent_Declare(
         efsw
-        GIT_REPOSITORY https://github.com/SpartanJ/efsw.git
-        GIT_TAG        87abe599995d5646f5d83cf2e3a225bd73148b3a
+        GIT_REPOSITORY https://github.com/5cript/efsw.git
+        GIT_TAG        d71ed846811cc67ee8827614341e2777d200231a
     )
 
     FetchContent_MakeAvailable(efsw)

backend/include/backend/file_tracking/batching_remote.hpp

@@ -0,0 +1,123 @@
+#pragma once
+
+#include <backend/file_tracking/tracker_remote.hpp>
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/strand.hpp>
+
+#include <chrono>
+#include <cstddef>
+#include <cstdint>
+#include <expected>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+namespace FileTracking
+{
+    /**
+     * @brief A TrackerRemote that also queues many uploads or deletes as one operation.
+     */
+    class BulkTrackerRemote : public TrackerRemote
+    {
+      public:
+        /**
+         * @brief Called once on the strand for each entry of a bulk once the bulk ended, with the entry's index.
+         */
+        using OnEntryEnded = std::function<void(std::size_t index, SharedData::OperationCompletionReason reason)>;
+
+        /**
+         * @brief Called once on the strand for each upload of a bulk once the bulk ended, with the entry's index.
+         */
+        using OnUploadEntryEnded = std::function<void(std::size_t index, UploadEnd const& ended)>;
+
+        /**
+         * @brief Queues the uploads of @p requests, regular files that may create the directories above their
+         *        destination, as one operation.
+         */
+        virtual std::expected<void, std::string>
+        uploadAll(std::vector<UploadRequest> const& requests, OnUploadEntryEnded onEnded) = 0;
+
+        /**
+         * @brief Queues the deletes of @p paths as one operation, a directory only when it is empty.
+         */
+        virtual std::expected<void, std::string>
+        removeAll(std::vector<Paths::RemotePath> const& paths, OnEntryEnded onEnded) = 0;
+    };
+
+    /**
+     * @brief Hands the uploads and deletes a Tracker queues within a short time to the server as one bulk each, which
+     *        costs one operation, one card and one listing refresh instead of one per path. Everything else passes
+     *        through at once. The tracker orders its operations per path, so what is gathered here lies on unrelated
+     *        paths and may run side by side. Lives on the strand of the session.
+     */
+    class BatchingRemote : public TrackerRemote
+    {
+      public:
+        struct Limits
+        {
+            /** @brief A batch goes once nothing was added to it for this long. */
+            std::chrono::milliseconds quiet{100};
+            /** @brief A batch goes at the latest this long after its first entry. */
+            std::chrono::milliseconds longest{1000};
+            /** @brief A batch goes at once when it holds this many entries. */
+            std::size_t maxEntries{500};
+            /** @brief A file this large goes on its own, an upload that can be stopped, see cancelUpload(). */
+            std::uint64_t aloneFromBytes{8u * 1024u * 1024u};
+        };
+
+        BatchingRemote(
+            boost::asio::strand<boost::asio::any_io_executor> strand,
+            std::unique_ptr<BulkTrackerRemote> inner,
+            Limits limits
+        );
+        BatchingRemote(
+            boost::asio::strand<boost::asio::any_io_executor> strand,
+            std::unique_ptr<BulkTrackerRemote> inner
+        );
+        ~BatchingRemote() override;
+        BatchingRemote(BatchingRemote const&) = delete;
+        BatchingRemote& operator=(BatchingRemote const&) = delete;
+        BatchingRemote(BatchingRemote&&) = delete;
+        BatchingRemote& operator=(BatchingRemote&&) = delete;
+
+        std::expected<std::optional<ServerState>, std::string> stat(Paths::RemotePath const& path) override;
+        std::expected<std::optional<ServerState>, std::string> linkStat(Paths::RemotePath const& path) override;
+        std::expected<void, std::string> download(
+            Paths::RemotePath const& source,
+            Paths::LocalPath const& destination,
+            std::uint64_t sizeBytes,
+            OnEnded onEnded
+        ) override;
+
+        /**
+         * @brief Gathers @p request into the next batch. It goes on its own when it creates no missing directories,
+         *        refreshes the listing, is no regular file, is a large one, continues a partial file or is alone in
+         *        its batch.
+         */
+        std::expected<void, std::string> upload(UploadRequest const& request, OnUploadEnded onEnded) override;
+
+        /**
+         * @brief An upload still gathered ends as cancelled without being sent, one on its way is stopped by the inner
+         *        remote. One in a bulk goes on.
+         */
+        void cancelUpload(Paths::RemotePath const& destination, PartialFile partial) override;
+
+        std::expected<void, std::string>
+        discardPartialUpload(Paths::RemotePath const& destination, OnEnded onEnded) override;
+
+        std::expected<void, std::string>
+        rename(Paths::RemotePath const& source, Paths::RemotePath const& destination, OnEnded onEnded) override;
+
+        /**
+         * @brief Gathers the delete of @p path into the next batch, alone in it it goes on its own.
+         */
+        std::expected<void, std::string> remove(Paths::RemotePath const& path, OnEnded onEnded) override;
+
+      private:
+        struct Implementation;
+        std::shared_ptr<Implementation> impl_;
+    };
+}

backend/include/backend/file_tracking/change_filter.hpp

@@ -1,7 +1,8 @@
 #pragma once
 
-#include <backend/file_tracking/temp_dir_instancing.hpp>
+#include <backend/file_tracking/file_action.hpp>
 
+#include <filesystem>
 #include <optional>
 #include <string>
 
@@ -22,6 +23,18 @@ namespace FileTracking
      */
     bool isTransientFile(std::string const& filename, std::string const& downloadTempSuffix);
 
+    /**
+     * @brief Whether a change of @p filename, moved from @p oldFilename when that is not empty, is noise nobody
+     *        follows: it ends at a temporary and does not come from a real file. Cheap enough for the watcher's
+     *        thread, which drops it before anything else looks at it.
+     * @param downloadTempSuffix The suffix of downloads in progress, see Persistence::effectiveTempFileSuffix.
+     */
+    bool isTransientChange(
+        std::string const& filename,
+        std::string const& oldFilename,
+        std::string const& downloadTempSuffix
+    );
+
     /**
      * @brief Turns a raw watcher event into the change to mirror to the remote side.
      * @param change The event as the watcher reports it.
@@ -31,4 +44,19 @@ namespace FileTracking
      */
     std::optional<FileChange>
     filterFileChange(FileChange change, bool isDirectory, std::string const& downloadTempSuffix);
+
+    /**
+     * @brief Whether a change of @p path, moved from @p oldPath when that is not empty, touches @p item or anything
+     *        below it. All paths are spelled alike (canonical).
+     */
+    bool isChangeOfItem(
+        std::filesystem::path const& item,
+        std::filesystem::path const& path,
+        std::filesystem::path const& oldPath
+    );
+
+    /**
+     * @brief Whether @p path is @p base or lies below it, compared lexically. An empty @p path is neither.
+     */
+    bool isAtOrBelow(std::filesystem::path const& base, std::filesystem::path const& path);
 }

backend/include/backend/file_tracking/change_reporter.hpp

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <backend/file_tracking/directory_watcher.hpp>
+#include <backend/file_tracking/reported_change.hpp>
+
+#include <boost/asio/any_io_executor.hpp>
+#include <boost/asio/strand.hpp>
+
+#include <memory>
+#include <string>
+
+namespace FileTracking
+{
+    class SettledChanges;
+
+    /**
+     * @brief Reports the changes of one watch to its sink. Noise (see filterFileChange) is dropped, content changes of
+     *        files are reported once they settled (see SettledChanges). A Missed change goes after what is held back
+     *        at and below its directory. Lives on its strand, copies share what is held back.
+     */
+    class ChangeReporter
+    {
+      public:
+        /**
+         * @param strand The strand report() is called on.
+         * @param sink Receives the changes on @p strand.
+         * @param downloadTempSuffix The suffix downloads write under until complete, see
+         *        Persistence::effectiveTempFileSuffix.
+         */
+        ChangeReporter(
+            boost::asio::strand<boost::asio::any_io_executor> strand,
+            ChangeSink sink,
+            std::string const& downloadTempSuffix
+        );
+
+        void report(DirectoryChange const& change) const;
+
+        /**
+         * @brief Whether a change of @p filename is noise that report() drops, see isTransientFile.
+         */
+        bool isNoise(std::string const& filename) const;
+
+      private:
+        std::shared_ptr<SettledChanges> settledChanges_;
+        std::string downloadTempSuffix_;
+    };
+}

backend/include/backend/file_tracking/directory_watch_service.hpp

@@ -0,0 +1,88 @@
+#pragma once
+
+#include <backend/file_tracking/instance_watch.hpp>
+#include <backend/file_tracking/file_action.hpp>
+
+#include <cstdint>
+#include <filesystem>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+
+namespace FileTracking
+{
+    /**
+     * @brief One change as efsw reports it, the names UTF-8 on every platform.
+     */
+    struct DirectoryEvent
+    {
+        FileAction action;
+        /**
+         * @brief The directory of the change, efsw ends it with a separator. For Missed the directory whose changes
+         *        were lost, with or without one.
+         */
+        std::string directory;
+        /** @brief Empty for Missed, which concerns the directory. */
+        std::string filename;
+        /** @brief The previous name of a move, empty otherwise. */
+        std::string oldFilename;
+    };
+
+    /**
+     * @brief Receives the events of a subscription on efsw's thread, it must only hand them on (post to a strand).
+     *        Anything slower holds up the events of every watch, and on Windows loses the last changes of a watched
+     *        directory that is deleted.
+     */
+    using DirectoryEventSink = std::function<void(DirectoryEvent const& event)>;
+
+    /**
+     * @brief The efsw watchers of the process. Directories watched without recursion share one watcher and its
+     *        thread, except on macOS: there a watcher is one FSEvents stream, which efsw rebuilds whenever one of its
+     *        watches comes or goes and drops the events on their way meanwhile. A recursive watch gets a watcher of
+     *        its own: efsw holds its lock while it walks the tree to set the watch up and while it delivers an event,
+     *        so a large tree would hold up the events of every other watch. A directory is watched once however many
+     *        subscribe to it, all of them with or all without recursion. Thread-safe. Held in a std::shared_ptr, see
+     *        shared().
+     */
+    class DirectoryWatchService : public std::enable_shared_from_this<DirectoryWatchService>
+    {
+      public:
+        /**
+         * @brief The service, created by the first caller. It lives as long as someone holds it.
+         */
+        static std::shared_ptr<DirectoryWatchService> shared();
+
+        DirectoryWatchService();
+        ~DirectoryWatchService();
+        DirectoryWatchService(DirectoryWatchService const&) = delete;
+        DirectoryWatchService& operator=(DirectoryWatchService const&) = delete;
+        DirectoryWatchService(DirectoryWatchService&&) = delete;
+        DirectoryWatchService& operator=(DirectoryWatchService&&) = delete;
+
+        /**
+         * @brief Delivers the changes in @p directory, and below it when @p recursive, to @p sink.
+         * @param directory An existing absolute path spelled like the events should report it. A recursive watch
+         *        walks it before this returns, without blocking other callers.
+         * @return The handle that ends the subscription, nothing when efsw rejected the watch or @p directory is
+         *         watched the other way (with or without recursion) already.
+         */
+        std::optional<InstanceWatch> subscribe(std::filesystem::path const& directory, bool recursive, DirectoryEventSink sink);
+
+        /**
+         * @brief Ends the subscription @p subscriptionId, the directory is no longer watched once nobody subscribes.
+         */
+        void unsubscribe(std::int64_t subscriptionId);
+
+        /**
+         * @brief Tells the subscriptions of the watch of @p directory, or of the recursive one above it, that the
+         *        changes in @p directory were lost, the way efsw does. For tests, efsw cannot be made to lose changes
+         *        on every platform.
+         */
+        void reportMissed(std::filesystem::path const& directory);
+
+      private:
+        struct Implementation;
+        std::unique_ptr<Implementation> impl_;
+    };
+}