From 6a0e87e21920a64f3582bd6d750fa9c7e96c91ca Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:07:57 +0200 Subject: [PATCH 1/3] 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. --- 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 5c9b4568320fd49c652d69518cae6a9123437b23 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:14:47 +0200 Subject: [PATCH 2/3] 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. --- 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 1b4188f..57ccfe7 100644 --- a/src/openvfsfuse/config.json +++ b/src/openvfsfuse/config.json @@ -8,5 +8,6 @@ "baloo_file", "baloo_filemetadata_temp_extractor" ] - } + }, + "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 032a3aa..6edb114 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; }; @@ -179,7 +184,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, ...) @@ -537,69 +545,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 001473fb6d3e98067304f338df15b7e3488c0c49 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:35:15 +0200 Subject: [PATCH 3/3] 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; +}