From 71ebc147190c5425f70ddbe7209ada7c77ce5ca8 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:06:08 +0200 Subject: [PATCH 1/6] Fix inverted pin state names reported by openvfs_stat name(PinStates) returned the strings belonging to name(States) for its first two cases, and inverted at that: OnlineOnly printed as "hydrated" and AlwaysLocal as "dehydrated". The only caller is the openvfs_stat diagnostic tool, which prints state and pin state side by side -- so the tool integrators reach for when pin state handling misbehaves reported the exact opposite of the truth. The two vocabularies are now distinguishable at a glance as well. Fixes #4 --- src/openvfs/openvfconstants.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/openvfs/openvfconstants.h b/src/openvfs/openvfconstants.h index 83d3f44..67fa869 100644 --- a/src/openvfs/openvfconstants.h +++ b/src/openvfs/openvfconstants.h @@ -84,9 +84,9 @@ namespace Constants { { switch (name) { case PinStates::OnlineOnly: - return "hydrated"; + return "onlineonly"; case PinStates::AlwaysLocal: - return "dehydrated"; + return "alwayslocal"; case PinStates::Inherited: return "inherited"; case PinStates::Excluded: From 0de6fdcdea0291ce3edfe990fb1d7c95a343f681 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:07:57 +0200 Subject: [PATCH 2/6] Reassemble socket API messages across reads readSocket() read at most 1023 bytes per call and left message reassembly to a FIXME. The socket is a SOCK_STREAM and carries no message boundaries, so any reply longer than the buffer -- or merely split across two reads by the kernel -- was parsed as two independent messages. Both halves then failed to parse, the original reply was lost, and the waiting open() sat out its full backoff. Worse, the effect is self-sustaining: once the stream desynchronises every subsequent message on the connection is misparsed, so a single long reply broke hydration until restart. A V2/HYDRATE_FILE_RESULT carrying a path plus a free-form arguments.error string clears 1 KB without trying. processSocketInput() now drains the socket into a persistent buffer, dispatches only complete newline terminated messages, and keeps the remainder for the next read. handleReceivedMsg() correspondingly handles a single message and no longer splits on newlines itself, which also retires its trailing-empty-fragment case. The buffer is bounded so a peer that never sends a newline cannot grow it without limit. Fixes #2 --- src/openvfsfuse/socketthread.cpp | 168 ++++++++++++++++++------------- src/openvfsfuse/socketthread.h | 9 +- 2 files changed, 107 insertions(+), 70 deletions(-) diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index c253417..7639989 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -24,6 +24,7 @@ #include "sharedmap.h" #include "strtools.h" +#include #include #include #include @@ -42,6 +43,13 @@ using namespace std; #define MSG_POST_USER_DATA 2 #define MSG_TIMER 3 +namespace { +/// Upper bound for the receive buffer. A message from the socket API is a +/// single JSON line and stays far below this; anything larger means the peer +/// is not speaking the protocol. +constexpr size_t MaxRxBufferSize = 1024 * 1024; +} + using json = nlohmann::json; struct ThreadMsg @@ -159,86 +167,112 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) // openvfsfuse_log(socket_path.c_str(), "socket send", value, "Message: %s", msg.c_str()); } -std::string SocketThread::readSocket() +void SocketThread::processSocketInput() { - // read answer FIXME: Split messages by \n and keep the rest - char buf[1024]; - ssize_t n = read(_socket, buf, sizeof(buf) - 1); - if (n <= 0) - return std::string(); - return std::string(buf, n); + // The socket is a SOCK_STREAM and carries no message boundaries: a single + // read may return a fragment of a message, several messages at once, or + // both. Accumulate into _rxBuffer and only dispatch complete lines. + char buf[4096]; + while (true) { + const ssize_t n = read(_socket, buf, sizeof(buf)); + if (n > 0) { + _rxBuffer.append(buf, static_cast(n)); + continue; + } + if (n == 0) { + // Peer closed the connection. Whatever is left in the buffer can + // never be completed, so drop it rather than misparsing it later. + if (!_rxBuffer.empty()) { + std::cerr << "Socket closed with " << _rxBuffer.size() << " bytes of incomplete message, discarding" << std::endl; + _rxBuffer.clear(); + } + return; + } + if (errno == EINTR) { + continue; + } + // EAGAIN on the non-blocking socket simply means there is nothing more + // to read right now. (EWOULDBLOCK is an alias for it on Linux and macOS.) + if (errno != EAGAIN) { + perror("read"); + return; + } + break; + } + + size_t pos; + while ((pos = _rxBuffer.find('\n')) != std::string::npos) { + handleReceivedMsg(_rxBuffer.substr(0, pos)); + _rxBuffer.erase(0, pos + 1); + } + + // A peer that never sends a newline must not be able to grow our buffer + // without bound. + if (_rxBuffer.size() > MaxRxBufferSize) { + std::cerr << "Discarding " << _rxBuffer.size() << " bytes of unterminated message from the socket API" << std::endl; + _rxBuffer.clear(); + } } -void SocketThread::handleReceivedMsg(const std::string &rawmsg) +void SocketThread::handleReceivedMsg(const std::string &msg) { - if (rawmsg.empty()) { - cout << "Received Message empty" << endl; + string msgType, msgAttr; + if (msg.empty()) { return; } - auto copies = StrTools::split(rawmsg, 0x000A); + cout << "Handle single message " << msg << endl; - for (const string &msg : copies) { - string msgType, msgAttr; - if (msg.empty()) { - continue; - } + size_t found = msg.find(':'); + if (found != string::npos) { + msgType = msg.substr(0, found); + msgAttr = msg.substr(found + 1, string::npos); + } else { + std::cerr << "Invalid message format: " << msg << std::endl; + return; + } - cout << "Handle single message " << msg << endl; + if (msgType == "V2/HYDRATE_FILE_RESULT") { + int id = -1; + std::string status; - size_t found = msg.find(':'); - if (found != string::npos) { - msgType = msg.substr(0, found); - msgAttr = msg.substr(found + 1, string::npos); - } else { - std::cerr << "Invalid message format: " << msg << std::endl; - continue; + try { + const auto j = json::parse(msgAttr); + id = std::stoi(j["id"].get()); + const auto arguments = j["arguments"].get(); + if (arguments.contains("error")) { + std::cerr << "Error from socket API for Id " << id << ": " << arguments["error"].get() << std::endl; + } else { + status = arguments["status"].get(); + } + } catch (json::exception &e) { + std::cerr << "Invalid JSON message: " << msgAttr << e.what() << std::endl; + return; } - // FIXME: Think if splitting by newline makes sense - - if (msgType == "V2/HYDRATE_FILE_RESULT") { - int id = -1; - std::string status; - - try { - const auto j = json::parse(msgAttr); - id = std::stoi(j["id"].get()); - const auto arguments = j["arguments"].get(); - if (arguments.contains("error")) { - std::cerr << "Error from socket API for Id " << id << ": " << arguments["error"].get() << std::endl; - } else { - status = arguments["status"].get(); - } - } catch (json::exception &e) { - std::cerr << "Invalid JSON message: " << msgAttr << e.what() << std::endl; - continue; + if (id > 0) { + int res{-1}; // Default set to fail + if (status == "OK") { + res = 0; // good! + } else { + cout << "ERROR from socket API for Id" << id << endl; } - if (id > 0) { - int res{-1}; // Default set to fail - if (status == "OK") { - res = 0; // good! - } else { - cout << "ERROR from socket API for Id" << id << endl; - } - - const HydJob hj{.state = res}; - bool ok = _sharedMap.set(id, hj); - if (!ok) { - // the id could not be set. That means, the job was not inserted. - cout << "Job not found:" << id << endl; - } else { - cout << "Setting Job ID " << id << " to result " << res << endl; - } - } - } else if (msgType == "VERSION") { - vector attribs = StrTools::split(msgAttr, ':'); - if (attribs.size() == 3) { - cout << "Got PID of the Desktop Client: " << attribs.at(2) << endl; - _sharedMap.setDesktopClientPid(std::stol(attribs.at(2))); + const HydJob hj{.state = res}; + bool ok = _sharedMap.set(id, hj); + if (!ok) { + // the id could not be set. That means, the job was not inserted. + cout << "Job not found:" << id << endl; + } else { + cout << "Setting Job ID " << id << " to result " << res << endl; } } + } else if (msgType == "VERSION") { + vector attribs = StrTools::split(msgAttr, ':'); + if (attribs.size() == 3) { + cout << "Got PID of the Desktop Client: " << attribs.at(2) << endl; + _sharedMap.setDesktopClientPid(std::stol(attribs.at(2))); + } } } @@ -402,11 +436,7 @@ void SocketThread::Process() case MSG_TIMER: { // cout << "Timer expired on " << THREAD_NAME << endl; - const std::string msg = readSocket(); - if (!msg.empty()) { - cout << "Message received: " << msg << endl; - handleReceivedMsg(msg); - } + processSocketInput(); break; } diff --git a/src/openvfsfuse/socketthread.h b/src/openvfsfuse/socketthread.h index ed4734c..93b81ed 100644 --- a/src/openvfsfuse/socketthread.h +++ b/src/openvfsfuse/socketthread.h @@ -89,7 +89,10 @@ class SocketThread int initSocket(const std::string& socketPath); bool socketSendMsg(std::shared_ptr); - std::string readSocket(); + + /// Drain everything readable from the socket into _rxBuffer and dispatch + /// every complete (newline terminated) message it contains. + void processSocketInput(); void handleReceivedMsg(const std::string &msg); /// Entry point for the worker thread @@ -114,6 +117,10 @@ class SocketThread std::atomic _socket; + /// Receive buffer holding the bytes read from the socket that do not form + /// a complete message yet. Only touched by the worker thread. + std::string _rxBuffer; + SharedMap &_sharedMap; }; From 4305cf3fbcfcdce3118856d39d88ecfa5b9aa812 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:14:47 +0200 Subject: [PATCH 3/6] Bound the hydration wait and close the open() registration race Two problems in the wait loop of openVFSfuse_open(), both on the hottest path in the project. First, the job was registered by the socket thread only after the send succeeded, while open() started polling the map 10 ms after PostMsg() -- which merely queues. If the socket thread had not been scheduled and had not completed its write() within that window, the lookup missed, and absence-from-map was read as failure: open() returned ENOENT for a file that plainly exists and was in the middle of being hydrated. That is the "the first open fails, the second works" flakiness. The job is now inserted before the message is posted, so no waiter can observe the absence of a job it just posted, and a send failure is published explicitly as a Failed state instead of being inferred from a missing entry -- two conditions the old code collapsed into one. Second, the backoff grew by roughly the golden ratio per step with no wall-clock ceiling: MaxCnt of 20 bounded the number of polls, not the time. By iteration 9 a single sleep was ~36 s, and a wedged client left open() blocked in uninterruptible sleep for what is effectively forever, with the calling application unkillable. Late replies also waited out a whole sleep interval before being noticed, so latency was dominated by poll granularity rather than by the download. SharedMap now carries a condition variable and SharedMap::waitForJob() blocks on it with a deadline: a result is observed the moment the socket thread publishes it, and the wait is bounded in seconds rather than in polls. The timeout is configurable via hydrationTimeoutSeconds, since a large file over a slow link is legitimately slow while an unresponsive client is not. The failure returns are now honest as well: EIO when the client reports an error, ETIMEDOUT when it does not answer. Applications and users both read ENOENT as "the file is gone", which sent us looking in the wrong place. Also made _transfer_id atomic. FUSE dispatches from several threads and the counter was incremented unguarded, so two concurrent opens could be handed the same id and share one job entry -- the same class of bug, noticed while fixing the above. Config parsing now tolerates missing keys so that an older config file keeps working. Fixes #1 Fixes #3 --- INTEGRATION.md | 14 +++++ src/openvfsfuse/config.json | 3 +- src/openvfsfuse/main.cpp | 30 ++++++++-- src/openvfsfuse/openvfsfuse.cpp | 96 ++++++++++++-------------------- src/openvfsfuse/openvfsfuse.h | 5 ++ src/openvfsfuse/sharedmap.cpp | 68 +++++++++++++++++----- src/openvfsfuse/sharedmap.h | 32 +++++++++++ src/openvfsfuse/socketthread.cpp | 28 +++++++--- src/openvfsfuse/socketthread.h | 3 +- 9 files changed, 190 insertions(+), 89 deletions(-) diff --git a/INTEGRATION.md b/INTEGRATION.md index 4c24364..f7d063b 100644 --- a/INTEGRATION.md +++ b/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`. + diff --git a/src/openvfsfuse/config.json b/src/openvfsfuse/config.json index e981705..c3cbee7 100644 --- a/src/openvfsfuse/config.json +++ b/src/openvfsfuse/config.json @@ -7,5 +7,6 @@ "kioworker", "baloo_file" ] - } + }, + "hydrationTimeoutSeconds": 300 } diff --git a/src/openvfsfuse/main.cpp b/src/openvfsfuse/main.cpp index 0b5577e..7b0e5ed 100644 --- a/src/openvfsfuse/main.cpp +++ b/src/openvfsfuse/main.cpp @@ -1,6 +1,7 @@ #include "openvfsfuse.h" #include +#include #include #include #include @@ -37,6 +38,7 @@ namespace { const std::string ConfigIgnoreAppsStr = "ignoreApps"; const std::string ConfigByNameStr = "byName"; const std::string ConfigEndsWith = "endsWith"; + const std::string ConfigHydrationTimeoutSecondsStr = "hydrationTimeoutSeconds"; } void usage(char *name) @@ -82,10 +84,30 @@ std::optional processArgs(int argc, char *argv[]) break; case 'i': { std::ifstream ifs(optarg); - json data = json::parse(ifs); - - out.appsNoHydrateFull = data[ConfigIgnoreAppsStr][ConfigByNameStr].get>(); - out.appsNoHydrateEndsWith = data[ConfigIgnoreAppsStr][ConfigEndsWith].get>(); + 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{}); + out.appsNoHydrateEndsWith = ignoreApps.value(ConfigEndsWith, std::vector{}); + + 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': diff --git a/src/openvfsfuse/openvfsfuse.cpp b/src/openvfsfuse/openvfsfuse.cpp index a53ae57..29f20a2 100644 --- a/src/openvfsfuse/openvfsfuse.cpp +++ b/src/openvfsfuse/openvfsfuse.cpp @@ -19,6 +19,7 @@ #include "sharedmap.h" #include "socketthread.h" +#include #include #include #include @@ -41,7 +42,6 @@ #include #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,6 +104,9 @@ 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; @@ -110,6 +114,7 @@ class VFSFuseContext int _rootHandle; std::vector _appsNoHydrateFull; std::vector _appsNoHydrateEndsWith; + std::chrono::milliseconds _hydrationTimeout; bool _debugEnabled; }; @@ -152,7 +157,10 @@ 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 handed to +// the client must be incremented atomically -- two opens racing for the same id +// would share a single job entry. +static std::atomic _transfer_id{12}; void openvfsfuse_log(const std::string &path, const char *action, int returncode, const char *format, ...) @@ -510,69 +518,39 @@ static int openVFSfuse_open(const char *orig_path, struct fuse_file_info *fi) openvfsfuse_log(path, "open", 0, "Requesting hydration %s: %d", path.c_str(), msgData->id); + // 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 - _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); - } - } + 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); - if (state == -1 || state == 2) { - // Fail, job with ID was errornous - openvfsfuse_log(path, "open", 1, "ERROR while retrieving: %d", state); - return -ENOENT; - } - - if (cnt >= MaxCnt) { - openvfsfuse_log(path, "open", MaxCnt, "TIMEOUT - no answer from client"); - return -ENOENT; + switch (result) { + case HydJobResult::Succeeded: + openvfsfuse_log(path, "open", 0, "Sucessfully 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; } openvfsfuse_log(path, "open", 0, "-- open finished"); diff --git a/src/openvfsfuse/openvfsfuse.h b/src/openvfsfuse/openvfsfuse.h index 78b3c67..28ec585 100644 --- a/src/openvfsfuse/openvfsfuse.h +++ b/src/openvfsfuse/openvfsfuse.h @@ -12,6 +12,7 @@ #define _X_SOURCE 500 #endif +#include #include #include @@ -24,6 +25,10 @@ struct openVFSfuse_Args std::vector fuseArgv; std::vector appsNoHydrateFull; // these apps are not permitted to cause a dehydration std::vector 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 }; diff --git a/src/openvfsfuse/sharedmap.cpp b/src/openvfsfuse/sharedmap.cpp index bf8163c..0a6b754 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -18,12 +18,18 @@ #include "sharedmap.h" -SharedMap::SharedMap() { } +SharedMap::SharedMap() + : _pid(0) +{ +} void SharedMap::insert(int key, const HydJob &value) { - std::lock_guard lock(_mutex); - _data[key] = value; + { + std::lock_guard lock(_mutex); + _data[key] = value; + } + _cv.notify_all(); } bool SharedMap::get(int key, HydJob &outValue) @@ -39,25 +45,57 @@ bool SharedMap::get(int key, HydJob &outValue) bool SharedMap::set(int key, const HydJob &value) { - std::lock_guard lock(_mutex); + bool found{false}; + { + std::lock_guard lock(_mutex); - auto it = _data.find(key); - if (it != _data.end()) { - _data[key] = value; - return true; + auto it = _data.find(key); + if (it != _data.end()) { + it->second = value; + found = true; + } } - return false; + if (found) { + _cv.notify_all(); + } + return found; +} + +HydJobResult SharedMap::waitForJob(int key, std::chrono::milliseconds timeout) +{ + std::unique_lock lock(_mutex); + + const bool settled = _cv.wait_for(lock, timeout, [this, key] { + const auto it = _data.find(key); + return it == _data.end() || it->second.state != HydJobState::Running; + }); + + if (!settled) { + return HydJobResult::TimedOut; + } + + const auto it = _data.find(key); + if (it == _data.end()) { + return HydJobResult::Lost; + } + return it->second.state == HydJobState::Success ? HydJobResult::Succeeded : HydJobResult::Failed; } bool SharedMap::remove(int id) { - std::lock_guard lock(_mutex); - auto it = _data.find(id); - if (it != _data.end()) { - _data.erase(id); - return true; + bool found{false}; + { + std::lock_guard lock(_mutex); + auto it = _data.find(id); + if (it != _data.end()) { + _data.erase(it); + found = true; + } } - return false; + if (found) { + _cv.notify_all(); + } + return found; } void SharedMap::setDesktopClientPid(long pid) diff --git a/src/openvfsfuse/sharedmap.h b/src/openvfsfuse/sharedmap.h index f761ed7..1ef2aa5 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -19,18 +19,40 @@ #ifndef SHAREDMAP_H #define SHAREDMAP_H +#include +#include #include #include #include #include #include +/** + * The values a hydration job can take. A job starts out Running and is moved to + * one of the other states by the socket thread. + */ +namespace HydJobState { +constexpr int Success = 0; +constexpr int Running = 1; +constexpr int Failed = -1; +} + struct HydJob { public: int state; }; +/** + * Outcome of waiting for a hydration job to leave the Running state. + */ +enum class HydJobResult { + Succeeded, ///< the client reported the file as hydrated + Failed, ///< the client reported an error, or the request could not be sent + TimedOut, ///< the client did not answer within the timeout + Lost, ///< the job vanished from the map, which should not happen +}; + class SharedMap { public: @@ -41,6 +63,15 @@ class SharedMap bool remove(int id); bool set(int key, const HydJob &value); + /** + * Block until the job identified by @p key leaves the Running state, or + * until @p timeout has elapsed. + * + * The wait is bounded by wall-clock time rather than by a number of polls, + * and a state change is observed as soon as the socket thread publishes it. + */ + HydJobResult waitForJob(int key, std::chrono::milliseconds timeout); + void printAll(); void setDesktopClientPid(long pid); long desktopClientPid(); @@ -48,6 +79,7 @@ class SharedMap private: std::map _data; std::mutex _mutex; + std::condition_variable _cv; long _pid; }; diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index 7639989..e77730b 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -159,6 +159,9 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) if (value < 0) { perror("write"); close(_socket); + // Mark the descriptor as gone so we do not keep reading from and + // writing to a closed fd on every subsequent tick. + _socket = -1; return false; } @@ -169,11 +172,16 @@ bool SocketThread::socketSendMsg(std::shared_ptr msgData) void SocketThread::processSocketInput() { + if (_socket < 0) { + return; + } + // The socket is a SOCK_STREAM and carries no message boundaries: a single // read may return a fragment of a message, several messages at once, or // both. Accumulate into _rxBuffer and only dispatch complete lines. char buf[4096]; - while (true) { + // The upper bound also stops a flooding peer from keeping us in this loop. + while (_rxBuffer.size() <= MaxRxBufferSize) { const ssize_t n = read(_socket, buf, sizeof(buf)); if (n > 0) { _rxBuffer.append(buf, static_cast(n)); @@ -251,9 +259,9 @@ void SocketThread::handleReceivedMsg(const std::string &msg) } if (id > 0) { - int res{-1}; // Default set to fail + int res{HydJobState::Failed}; // Default set to fail if (status == "OK") { - res = 0; // good! + res = HydJobState::Success; // good! } else { cout << "ERROR from socket API for Id" << id << endl; } @@ -351,10 +359,10 @@ void SocketThread::ExitThread() //---------------------------------------------------------------------------- // PostMsg //---------------------------------------------------------------------------- -void SocketThread::PostMsg(std::shared_ptr data) +bool SocketThread::PostMsg(std::shared_ptr data) { if (m_exit.load()) - return; + return false; assert(m_thread); // Create a new ThreadMsg @@ -364,6 +372,7 @@ void SocketThread::PostMsg(std::shared_ptr data) std::unique_lock lk(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); + return true; } //---------------------------------------------------------------------------- @@ -424,11 +433,12 @@ void SocketThread::Process() if (!socketSendMsg(msgData)) { cout << "Failed to send msg " << msgData->id << ": " << msgData->msg.c_str() << endl; - } else { + // The job was registered by the caller before it posted the + // message, so a failure to send has to be published as such. + // Otherwise the caller waits out its full timeout for a + // request that never left the machine. if (msgData->id > 0) { - const HydJob hj{.state = 1}; - _sharedMap.insert(msgData->id, hj); - cout << "Storing sent message ID" << msgData->id << endl; + _sharedMap.set(msgData->id, HydJob{.state = HydJobState::Failed}); } } break; diff --git a/src/openvfsfuse/socketthread.h b/src/openvfsfuse/socketthread.h index 93b81ed..7b3a6bf 100644 --- a/src/openvfsfuse/socketthread.h +++ b/src/openvfsfuse/socketthread.h @@ -72,7 +72,8 @@ class SocketThread /// Add a message to the thread queue /// @param[in] data - thread specific message information - void PostMsg(std::shared_ptr msg); + /// @return False if the thread is shutting down and the message was dropped. + bool PostMsg(std::shared_ptr msg); /// Get size of thread message queue. size_t GetQueueSize(); From 986cd0e2307c0e96ae72e4be0e919966b2ac039a Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:15:38 +0200 Subject: [PATCH 4/6] Ship a cross-desktop ignoreApps default The shipped config blocked only KDE components from triggering a hydration, so on GNOME, Cinnamon, MATE or Xfce nothing was blocked at all: nautilus and its thumbnailer helpers, tracker/localsearch, gvfsd and the shell search providers open dehydrated files as a matter of routine. Browsing a folder downloaded everything in it and indexing downloaded the entire sync root -- silently, and on what is probably the most common Linux desktop. The list now covers the common file managers, the freedesktop thumbnailer convention (a "thumbnailer" suffix catches the -thumbnailer binaries and ffmpegthumbnailer alike), the GNOME/tracker and localsearch indexers, gvfs helpers, locate's updatedb and ClamAV. /usr/bin/dolphin moved from byName to an endsWith entry: matching the binary name rather than an absolute path holds across distributions that install elsewhere. This is a starting point rather than an exhaustive list -- see INTEGRATION.md. Enumerating every indexer, thumbnailer and antivirus in existence does not converge, so a permit-list of the few applications that should be allowed to trigger a download may scale better; that is a design decision for the maintainers and is deliberately not attempted here. The errno returned to a blocked caller is likewise left at EPERM. Fixes #5 --- src/openvfsfuse/config.json | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/src/openvfsfuse/config.json b/src/openvfsfuse/config.json index c3cbee7..6e01b33 100644 --- a/src/openvfsfuse/config.json +++ b/src/openvfsfuse/config.json @@ -1,11 +1,37 @@ { "ignoreApps": { "byName": [ - "/usr/bin/dolphin" ], "endsWith": [ + "dolphin", "kioworker", - "baloo_file" + "baloo_file", + "baloo_file_extractor", + "nautilus", + "nemo", + "caja", + "thunar", + "pcmanfm", + "pcmanfm-qt", + "thumbnailer", + "gnome-thumbnail-font", + "gnome-thumbnail-factory", + "tumblerd", + "tracker-miner-fs", + "tracker-miner-fs-3", + "tracker-extract", + "tracker-extract-3", + "localsearch-3", + "localsearch-extractor-3", + "gvfsd", + "gvfsd-metadata", + "gvfsd-trash", + "updatedb", + "updatedb.mlocate", + "updatedb.plocate", + "clamd", + "clamscan", + "freshclam" ] }, "hydrationTimeoutSeconds": 300 From 1c270b20eb8c8f9c13c9d6cf48e7d4f7dd5a71d0 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:20:57 +0200 Subject: [PATCH 5/6] Make SPDX headers consistent and complete src/openvfs/ is one library in one directory but gave three different licensing answers: four files GPL-3.0-or-later, openvfs.cpp GPL-2.0-or-later, and three files in src/openvfsfuse/ with no SPDX tag at all. libopenvfs is what downstream sync clients link against, so its license determines the license of the combined binary they ship -- anyone doing that assessment got a different answer depending on which file they opened first. openvfs.cpp is aligned with the other four and with the top-level LICENSE, the missing tags are added to main.cpp and sharedmap.{h,cpp} (which already carried the full GPL-3 notice in prose), and the tag in strtools.h moves above the include guard like everywhere else. socketthread.{h,cpp} are deliberately left alone: they carry third-party MIT code by David Lafreniere alongside the GPL-3 notice, and pinning a single machine-readable expression on that combination is the maintainers' call, not a drive-by fix. A REUSE.toml would make the whole tree checkable with `reuse lint` and is worth doing as a follow-up. Fixes #6 --- src/openvfs/openvfs.cpp | 2 +- src/openvfsfuse/main.cpp | 4 ++++ src/openvfsfuse/sharedmap.cpp | 3 +++ src/openvfsfuse/sharedmap.h | 3 +++ src/openvfsfuse/strtools.h | 6 +++--- 5 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/openvfs/openvfs.cpp b/src/openvfs/openvfs.cpp index 0d0abe8..e1caabc 100644 --- a/src/openvfs/openvfs.cpp +++ b/src/openvfs/openvfs.cpp @@ -1,4 +1,4 @@ -// SPDX-License-Identifier: GPL-2.0-or-later +// SPDX-License-Identifier: GPL-3.0-or-later // SPDX-FileCopyrightText: 2025 Hannah von Reth #include "openvfs/openvfs.h" diff --git a/src/openvfsfuse/main.cpp b/src/openvfsfuse/main.cpp index 7b0e5ed..0658371 100644 --- a/src/openvfsfuse/main.cpp +++ b/src/openvfsfuse/main.cpp @@ -1,3 +1,7 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2025 Hannah von Reth +// SPDX-FileCopyrightText: 2025 Klaas Freitag + #include "openvfsfuse.h" #include diff --git a/src/openvfsfuse/sharedmap.cpp b/src/openvfsfuse/sharedmap.cpp index 0a6b754..df63b4f 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -1,3 +1,6 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2025 Klaas Freitag + /* * openvfsfuse - a Fuse layer to handle virtual filesystem items of cloud storage * Copyright (C) 2025 Klaas Freitag diff --git a/src/openvfsfuse/sharedmap.h b/src/openvfsfuse/sharedmap.h index 1ef2aa5..660ba1a 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -1,3 +1,6 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2025 Klaas Freitag + /* * openvfsfuse - a Fuse layer to handle virtual filesystem items of cloud storage * Copyright (C) 2025 Klaas Freitag diff --git a/src/openvfsfuse/strtools.h b/src/openvfsfuse/strtools.h index 7610994..0d6fa87 100644 --- a/src/openvfsfuse/strtools.h +++ b/src/openvfsfuse/strtools.h @@ -1,9 +1,9 @@ -#ifndef STRTOOLS_H -#define STRTOOLS_H - // SPDX-License-Identifier: GPL-3.0-or-later // SPDX-FileCopyrightText: 2026 Klaas Freitag +#ifndef STRTOOLS_H +#define STRTOOLS_H + #pragma once #include #include From 0096111e22346c7e9b580bc73701fba6cf2681a1 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:35:15 +0200 Subject: [PATCH 6/6] Add a test for the socket framing and the hydration wait The bugs fixed in this branch are timing and framing dependent, which is exactly the kind that comes back unnoticed. The test stands in for the desktop client on a real AF_UNIX socket and drives the actual SocketThread and SharedMap. It covers a message split across two writes, a reply far larger than any single read buffer, several replies batched into one write, the stream still being in sync afterwards, a silent client timing out within its deadline, a late reply being observed without a growing backoff, and PostMsg reporting a message dropped during shutdown. Verified to fail against the pre-fix framing code. The repository had ctest wired up but no tests, so this also gives the existing "Run tests" CI step something to run. --- src/CMakeLists.txt | 1 + src/openvfsfuse/CMakeLists.txt | 4 + src/openvfsfuse/autotests/CMakeLists.txt | 13 ++ .../autotests/socketthreadtest.cpp | 169 ++++++++++++++++++ 4 files changed, 187 insertions(+) create mode 100644 src/openvfsfuse/autotests/CMakeLists.txt create mode 100644 src/openvfsfuse/autotests/socketthreadtest.cpp diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 981d171..033715b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -1,4 +1,5 @@ include(CMakePackageConfigHelpers) +include(ECMMarkAsTest) include(ECMMarkNonGuiExecutable) include(ECMSetupVersion) diff --git a/src/openvfsfuse/CMakeLists.txt b/src/openvfsfuse/CMakeLists.txt index 014c847..84c81ca 100644 --- a/src/openvfsfuse/CMakeLists.txt +++ b/src/openvfsfuse/CMakeLists.txt @@ -12,5 +12,9 @@ target_link_libraries(openvfsfuse PRIVATE openvfs nlohmann_json::nlohmann_json F # disable darwin extensions for now target_compile_definitions(openvfsfuse PRIVATE FUSE_DARWIN_ENABLE_EXTENSIONS=0) +if (BUILD_TESTING) + add_subdirectory(autotests) +endif() + install(FILES config.json DESTINATION ${KDE_INSTALL_CONFDIR}/openvfs) install(TARGETS openvfsfuse EXPORT OpenVFSConfig ${KDE_INSTALL_TARGETS_DEFAULT_ARGS} ) diff --git a/src/openvfsfuse/autotests/CMakeLists.txt b/src/openvfsfuse/autotests/CMakeLists.txt new file mode 100644 index 0000000..9a383a7 --- /dev/null +++ b/src/openvfsfuse/autotests/CMakeLists.txt @@ -0,0 +1,13 @@ +# The socket and job handling live in the openvfsfuse executable, so the test +# compiles those sources directly rather than linking a library. +add_executable(socketthreadtest + socketthreadtest.cpp + ../socketthread.cpp + ../sharedmap.cpp + ../strtools.cpp +) +target_include_directories(socketthreadtest PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/..) +target_link_libraries(socketthreadtest PRIVATE nlohmann_json::nlohmann_json Threads::Threads) +ecm_mark_as_test(socketthreadtest) + +add_test(NAME socketthreadtest COMMAND socketthreadtest) diff --git a/src/openvfsfuse/autotests/socketthreadtest.cpp b/src/openvfsfuse/autotests/socketthreadtest.cpp new file mode 100644 index 0000000..8ae4ae3 --- /dev/null +++ b/src/openvfsfuse/autotests/socketthreadtest.cpp @@ -0,0 +1,169 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2026 Klaas Freitag + +/* + * Drives SocketThread and SharedMap over a real AF_UNIX socket, standing in for + * the desktop client. Covers the stream framing and the hydration wait, both of + * which are timing dependent and regress silently. + */ + +#include "sharedmap.h" +#include "socketthread.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std::chrono_literals; + +namespace { + +int failures = 0; + +void check(bool ok, const std::string &what) +{ + std::cerr << (ok ? "ok : " : "FAIL : ") << what << std::endl; + if (!ok) { + ++failures; + } +} + +/// A socket path has to fit into sockaddr_un::sun_path, so keep it short and +/// out of the (potentially deeply nested) build directory. +std::string makeSocketPath() +{ + char tmpl[] = "/tmp/openvfs-test-XXXXXX"; + const char *dir = mkdtemp(tmpl); + if (!dir) { + std::cerr << "Failed to create a temporary directory" << std::endl; + std::exit(2); + } + return std::string(dir) + "/s"; +} + +int listenOn(const std::string &path) +{ + const int fd = socket(AF_UNIX, SOCK_STREAM, 0); + sockaddr_un addr{}; + addr.sun_family = AF_UNIX; + std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1); + if (bind(fd, reinterpret_cast(&addr), sizeof(addr)) != 0 || listen(fd, 1) != 0) { + std::cerr << "Failed to listen on " << path << ": " << std::strerror(errno) << std::endl; + std::exit(2); + } + return fd; +} + +void writeAll(int fd, const std::string &data) +{ + size_t offset = 0; + while (offset < data.size()) { + const ssize_t n = write(fd, data.data() + offset, data.size() - offset); + if (n <= 0) { + return; + } + offset += static_cast(n); + } +} + +std::string hydrateResult(int id, const std::string &argumentsBody) +{ + return "V2/HYDRATE_FILE_RESULT:{\"id\":\"" + std::to_string(id) + "\",\"arguments\":{" + argumentsBody + "}}\n"; +} + +} + +int main() +{ + const std::string socketPath = makeSocketPath(); + const int server = listenOn(socketPath); + + SharedMap jobs; + SocketThread socketThread("TestSocketThread", jobs); + socketThread.CreateThread(socketPath); + + const int client = accept(server, nullptr, nullptr); + if (client < 0) { + std::cerr << "SocketThread did not connect" << std::endl; + return 2; + } + + // Register a job the way openVFSfuse_open() does, before posting it. + const auto post = [&](int id) { + auto data = std::make_shared(); + data->msg = "V2/HYDRATE_FILE"; + data->file = "/tmp/some/file"; + data->id = id; + jobs.insert(id, HydJob{.state = HydJobState::Running}); + socketThread.PostMsg(data); + }; + + // A message split across two writes must not be acted on until it is complete. + writeAll(client, "VERSION:1.2.3:2"); + std::this_thread::sleep_for(750ms); + check(jobs.desktopClientPid() == 0, "an incomplete message is not dispatched"); + writeAll(client, ":4242\n"); + std::this_thread::sleep_for(750ms); + check(jobs.desktopClientPid() == 4242, "a message split across two writes is reassembled"); + + // A reply well past any single read buffer must arrive as one message. An + // error string carrying a path and a description clears 1 KB easily. + post(101); + writeAll(client, hydrateResult(101, "\"error\":\"" + std::string(8000, 'E') + "\"")); + check(jobs.waitForJob(101, 10s) == HydJobResult::Failed, "an 8 KB reply is parsed as a single message"); + + // Several replies batched into one write must all be dispatched. + post(102); + post(103); + writeAll(client, hydrateResult(102, "\"status\":\"OK\"") + hydrateResult(103, "\"status\":\"FAILED\"")); + check(jobs.waitForJob(102, 10s) == HydJobResult::Succeeded, "first of two batched replies is dispatched"); + check(jobs.waitForJob(103, 10s) == HydJobResult::Failed, "second of two batched replies is dispatched"); + + // ... and the stream is still in sync after all of the above. + post(104); + writeAll(client, hydrateResult(104, "\"status\":\"OK\"")); + check(jobs.waitForJob(104, 10s) == HydJobResult::Succeeded, "the stream stays in sync"); + + // A silent client must time out, bounded by wall-clock time. + post(105); + const auto beforeTimeout = std::chrono::steady_clock::now(); + const auto timedOut = jobs.waitForJob(105, 500ms); + const auto waited = std::chrono::steady_clock::now() - beforeTimeout; + check(timedOut == HydJobResult::TimedOut, "a silent client is reported as a timeout"); + check(waited >= 450ms && waited < 30s, "the timeout is bounded by wall-clock time"); + + // A reply arriving later must be observed without waiting out a long backoff. + post(106); + std::thread late([&] { + std::this_thread::sleep_for(300ms); + writeAll(client, hydrateResult(106, "\"status\":\"OK\"")); + }); + const auto beforeLate = std::chrono::steady_clock::now(); + const auto lateResult = jobs.waitForJob(106, 60s); + const auto latency = std::chrono::steady_clock::now() - beforeLate; + late.join(); + check(lateResult == HydJobResult::Succeeded, "a late reply is picked up"); + check(latency < 10s, "a late reply is not delayed by a growing backoff"); + + // A caller must never be left waiting for a message that was never queued. + socketThread.ExitThread(); + auto dropped = std::make_shared(); + dropped->msg = "V2/HYDRATE_FILE"; + dropped->id = 107; + check(!socketThread.PostMsg(dropped), "PostMsg reports messages dropped during shutdown"); + + close(client); + close(server); + std::error_code ec; + std::filesystem::remove_all(std::filesystem::path(socketPath).parent_path(), ec); + + std::cerr << (failures == 0 ? "All checks passed" : std::to_string(failures) + " check(s) failed") << std::endl; + return failures == 0 ? 0 : 1; +}