Bound hydration wait

INTEGRATION.md

@@ -15,6 +15,16 @@ The openvfs binary accepts the following command-line parameters:
 
 The FUSE layer must be unmounted when the desktop client shuts down or when a sync connection is removed.
 
+## Configuration File
+
+The file passed via `-i` is JSON. All keys are optional; a missing key falls back to the built-in default.
+
+* `ignoreApps.byName`: List of absolute executable paths that must not trigger a hydration. A matching caller receives `EPERM` from `open()` instead of the file being downloaded.
+* `ignoreApps.endsWith`: List of suffixes matched against the calling executable's path, for the same purpose. This is the distribution-independent way to name a binary, and `-thumbnailer` catches the freedesktop thumbnailer convention in one entry.
+* `hydrationTimeoutSeconds`: How long `open()` blocks waiting for the client to hydrate a file before giving up and returning `ETIMEDOUT`. Defaults to `300`. Raise it if large files are legitimately transferred over slow links; the point of the bound is to keep an unresponsive client from wedging the calling application indefinitely.
+
+The shipped `config.json` is a cross-desktop starting point, not an exhaustive list. Any indexer, thumbnailer, or antivirus that opens files as a matter of routine will otherwise download the whole sync root.
+
 ## Socket API Communication
 
 OpenVFS communicates with the desktop client over a local, unauthenticated Unix socket — the so-called socket API. This bidirectional channel is already used by both OpenCloud and Nextcloud clients for file manager integration.
@@ -49,3 +59,7 @@ Expected response fields:
 
 On a successful response, the file is considered hydrated and its content is available locally.
 
+Messages are newline-delimited on a stream socket, so a reply may be split across several writes and several replies may be batched into one. openVFS reassembles them; the client need not size its replies to any particular limit.
+
+If the client reports an error, `open()` fails with `EIO`. If it does not answer within `hydrationTimeoutSeconds`, `open()` fails with `ETIMEDOUT`.
+

src/openvfsfuse/config.json

@@ -34,5 +34,6 @@
             "clamscan",
             "freshclam"
         ]
-    }
+    },
+    "hydrationTimeoutSeconds": 300
 }

src/openvfsfuse/main.cpp

@@ -5,6 +5,7 @@
 #include "openvfsfuse.h"
 
 #include <cassert>
+#include <chrono>
 #include <filesystem>
 #include <fstream>
 #include <getopt.h>
@@ -24,6 +25,7 @@ constexpr int MaxFuseArgs = 32;
 const std::string ConfigIgnoreAppsStr = "ignoreApps";
 const std::string ConfigByNameStr = "byName";
 const std::string ConfigEndsWith = "endsWith";
+const std::string ConfigHydrationTimeoutSecondsStr = "hydrationTimeoutSeconds";
 
 void usage(char *name)
 {
@@ -72,10 +74,30 @@ std::optional<openVFSfuse_Args> processArgs(int argc, char *argv[])
             break;
         case 'i': {
             std::ifstream ifs(optarg);
-            json data = json::parse(ifs);
-
-            out.appsNoHydrateFull = data[ConfigIgnoreAppsStr][ConfigByNameStr].get<std::vector<std::string>>();
-            out.appsNoHydrateEndsWith = data[ConfigIgnoreAppsStr][ConfigEndsWith].get<std::vector<std::string>>();
+            if (!ifs) {
+                std::cerr << "Failed to open config file " << optarg << std::endl;
+                return {};
+            }
+            json data;
+            try {
+                data = json::parse(ifs);
+            } catch (const json::exception &e) {
+                std::cerr << "Failed to parse config file " << optarg << ": " << e.what() << std::endl;
+                return {};
+            }
+
+            // Every setting is optional so that a config file written for an
+            // older version keeps working.
+            const auto ignoreApps = data.value(ConfigIgnoreAppsStr, json::object());
+            out.appsNoHydrateFull = ignoreApps.value(ConfigByNameStr, std::vector<std::string>{});
+            out.appsNoHydrateEndsWith = ignoreApps.value(ConfigEndsWith, std::vector<std::string>{});
+
+            const auto timeoutSeconds = data.value(ConfigHydrationTimeoutSecondsStr, 0);
+            if (timeoutSeconds > 0) {
+                out.hydrationTimeout = std::chrono::seconds(timeoutSeconds);
+            } else if (timeoutSeconds < 0) {
+                std::cerr << ConfigHydrationTimeoutSecondsStr << " must be positive, keeping the default" << std::endl;
+            }
             break;
         }
         case 'o':

src/openvfsfuse/openvfsfuse.cpp

@@ -19,6 +19,7 @@
 #include "sharedmap.h"
 #include "socketthread.h"
 
+#include <atomic>
 #include <chrono>
 #include <cstdint>
 #include <cstring>
@@ -41,7 +42,6 @@
 #include <syslog.h>
 #endif
 
-using namespace std::chrono_literals;
 using json = nlohmann::json;
 
 namespace {
@@ -55,6 +55,7 @@ class VFSFuseContext
         , _rootHandle(open(_mountPoint.c_str(), 0))
         , _appsNoHydrateFull(args.appsNoHydrateFull)
         , _appsNoHydrateEndsWith(args.appsNoHydrateEndsWith)
+        , _hydrationTimeout(args.hydrationTimeout)
         , _debugEnabled(args.debugEnabled)
     {
         assert(!_instance);
@@ -103,13 +104,17 @@ class VFSFuseContext
 
     bool debugEnabled() const { return _debugEnabled; }
 
+    /// How long open() waits for the desktop client to hydrate a file.
+    auto hydrationTimeout() const { return _hydrationTimeout; }
+
 private:
     static VFSFuseContext *_instance;
     std::filesystem::path _mountPoint;
     std::string _owner;
     int _rootHandle;
     std::vector<std::string> _appsNoHydrateFull;
     std::vector<std::string> _appsNoHydrateEndsWith;
+    std::chrono::milliseconds _hydrationTimeout;
     bool _debugEnabled;
 };
 
@@ -179,7 +184,8 @@ std::string getcallername(fuse_context *context)
 static SharedMap _jobs;
 static SocketThread _socketThread("SocketThread", _jobs);
 
-static int _transfer_id{12};
+// FUSE dispatches requests from several threads, so the transfer id needs to be atomic
+static std::atomic<int> _transfer_id{12};
 
 
 void openvfsfuse_log(const std::string &path, const char *action, int returncode, const char *format, ...)
@@ -554,96 +560,61 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi)
     if (callerBelongsToProcess(fuse_get_context()->pid, _jobs.desktopClientPid())) {
         openvfsfuse_log(path, "open", res, "Desktop client tries to access, bypassing!");
     } else {
-        if (const auto attribs = OpenVFS::PlaceHolderAttributes::fromAttributes(path)) {
-            if (attribs->state == OpenVFS::Constants::States::DeHydrated) {
-                // the file is virtual. It will be hydrated if the calling instance
-                // is not on the ignore list
-
-                // ignore list of apps that must not cause a hydration
-                if (VFSFuseContext::instance().blockedToOpen(opener)) {
-                    openvfsfuse_log(path, "open", 0, "blocked hydration for not permitted app %s", opener.c_str());
-                    return -EPERM;
-                }
-
-                // Create message to send to worker thread 1
-                std::shared_ptr<MsgData> msgData(new MsgData());
-                msgData->msg = "V2/HYDRATE_FILE";
-                msgData->file = VFSFuseContext::instance().getExternalPath(attribs->absolutePath);
-                msgData->fileId = attribs->fileId;
-                msgData->requester = getcallername(fuse_get_context());
-                msgData->id = ++_transfer_id; // attention: _transfer_id is global and can change!
-
-                openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id);
-
-                // push hydration request to the thread that handles the communication to the client
-                _socketThread.PostMsg(msgData);
-
-                // the socketThread now talks to the client, which downloads the file for us.
-                // Here in this thread we enter a loop and wait for results
-                int cnt{0};
-                int state{1};
-                const auto MaxCnt{20};
-                std::chrono::duration waitTime{10ms};
-                std::chrono::duration dur{30ms};
-
-                HydJob hj;
-
-                while (state == 1 && cnt++ < MaxCnt) {
-                    std::this_thread::sleep_for(waitTime); // sleep for some time
-                    waitTime += dur;
-                    dur += waitTime;
-                    // first: waittime: 10ms, dur: 30ms
-                    // second:waittime: 40ms, dur: 70ms
-                    // third: waittime: 110ms, dur: 180ms
-                    // forth: waittime: 290ms, dur: 470ms
-                    // fifth: waittime: 760ms, dur: 1230ms
-                    // ...
-
-                    // check shared map and see if the id has changed to 0, which means success
-                    // the value is changed in the other thread and fetched here
-
-                    if (!_jobs.get(msgData->id, hj)) {
-                        // The job is no longer there :-/
-                        openvfsfuse_log(path, "open", 1, "Job queue does not have job %d", msgData->id);
-                        state = -1;
-                    } else {
-                        state = hj.state;
-                        openvfsfuse_log(path, "open", 1, "Found in job queue %d", state);
-
-                        // With all the state values except 1, the loop is left
-                        if (state == 0) {
-                            // success!
-                            openvfsfuse_log(path, "open", 1, "Sucessfully finished job %d", msgData->id);
-                        } else if (state == 1) {
-                            // still running
-                        } else if (state == -1) {
-                            // fail
-                            openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id);
-                        } else if (state == 2) {
-                            // timeout
-                            openvfsfuse_log(path, "open", 0, "Job %d timed out", msgData->id);
-                        }
-                    }
-                }
-
-                // remove the job regardless of the result
-                _jobs.remove(msgData->id);
+        const auto attribs = OpenVFS::PlaceHolderAttributes::fromAttributes(path);
+        if (attribs && attribs->state == OpenVFS::Constants::States::DeHydrated) {
+            // the file is virtual. It will be hydrated if the calling instance
+            // is not on the ignore list
+
+            // ignore list of apps that must not cause a hydration
+            if (VFSFuseContext::instance().blockedToOpen(opener)) {
+                openvfsfuse_log(path, "open", 0, "blocked hydration for not permitted app %s", opener.c_str());
+                return -EPERM;
+            }
 
-                if (state == -1 || state == 2) {
-                    // Fail, job with ID was errornous
-                    openvfsfuse_log(path, "open", 1, "ERROR while retrieving: %d", state);
-                    return -ENOENT;
-                }
+            // Create message to send to worker thread 1
+            std::shared_ptr<MsgData> msgData(new MsgData());
+            msgData->msg = "V2/HYDRATE_FILE";
+            msgData->file = VFSFuseContext::instance().getExternalPath(attribs->absolutePath);
+            msgData->fileId = attribs->fileId;
+            msgData->requester = getcallername(fuse_get_context());
+            msgData->id = ++_transfer_id; // attention: _transfer_id is global and can change!
 
-                if (cnt >= MaxCnt) {
-                    openvfsfuse_log(path, "open", MaxCnt, "TIMEOUT - no answer from client");
-                    return -ENOENT;
-                }
+            openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id);
 
-                openvfsfuse_log(path, "open", 0, "-- open finished");
+            // Register the job *before* handing the request over. PostMsg() only
+            // queues, so the socket thread may well have sent the request and
+            // received the answer before this thread runs again. A waiter must
+            // never be able to observe the absence of a job it just posted.
+            _jobs.insert(msgData->id, HydJob{._state = HydJobState::Running});
+
+            // push hydration request to the thread that handles the communication to the client
+            if (!_socketThread.PostMsg(msgData)) {
+                _jobs.remove(msgData->id);
+                openvfsfuse_log(path, "open", 0, "Could not queue hydration request %d, shutting down", msgData->id);
+                return -EIO;
+            }
+
+            // The socketThread now talks to the client, which downloads the file
+            // for us. Block here until the client answers or the timeout expires.
+            const auto result = _jobs.waitForJob(msgData->id, VFSFuseContext::instance().hydrationTimeout());
+
+            // remove the job regardless of the result
+            _jobs.remove(msgData->id);
+
+            switch (result) {
+            case HydJobResult::Succeeded:
+                openvfsfuse_log(path, "open", 0, "Successfully finished job %d", msgData->id);
+                break;
+            case HydJobResult::Failed:
+                openvfsfuse_log(path, "open", 0, "Failed job %d", msgData->id);
+                return -EIO;
+            case HydJobResult::TimedOut:
+                openvfsfuse_log(path, "open", 0, "TIMEOUT - no answer from client for job %d", msgData->id);
+                return -ETIMEDOUT;
+            case HydJobResult::Lost:
+                openvfsfuse_log(path, "open", 0, "Job queue does not have job %d", msgData->id);
+                return -EIO;
             }
-        } else {
-            openvfsfuse_log(path, "open", 0, "Not a placeholder file");
         }
     }
     // File is not dehydrated and it is just going to be opened
@@ -831,8 +802,10 @@ int initializeOpenVFSFuse(openVFSfuse_Args &openVFSArgs)
     openVFSfuse_oper.removexattr = openVFSfuse_removexattr;
 
     assert(!openVFSArgs.socketPath.empty());
-    _socketThread.CreateThread(openVFSArgs.socketPath);
-
+    if (!_socketThread.CreateThread(openVFSArgs.socketPath)) {
+        std::cerr << "Unable to create the socket thread" << std::endl;
+        return -1;
+    }
 
     std::cout << "openVFSfuse starting: PID:" << getpid();
     std::cout << ", Socket:" << openVFSArgs.socketPath << std::endl;

src/openvfsfuse/openvfsfuse.h

@@ -12,6 +12,7 @@
 #define _X_SOURCE 500
 #endif
 
+#include <chrono>
 #include <filesystem>
 #include <vector>
 
@@ -24,6 +25,10 @@ struct openVFSfuse_Args
     std::vector<std::string> fuseArgv;
     std::vector<std::string> appsNoHydrateFull; // these apps are not permitted to cause a dehydration
     std::vector<std::string> appsNoHydrateEndsWith;
+    // Upper bound for how long open() blocks waiting for the desktop client to
+    // hydrate a file. A large file over a slow link is legitimately slow, an
+    // unresponsive client is not -- so this is configurable.
+    std::chrono::milliseconds hydrationTimeout = std::chrono::seconds(300);
     bool debugEnabled = false; // checked in the central logging function if logging is enabled
 };