From 9a62bbc61be53d96bce3f4ddeffc1c55ad86090e Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:07:57 +0200 Subject: [PATCH 1/5] 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 | 179 +++++++++++++++++++------------ src/openvfsfuse/socketthread.h | 9 +- 2 files changed, 118 insertions(+), 70 deletions(-) diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index c253417..ff69463 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,14 @@ 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 * 100; +constexpr size_t SingleReadBufferSize = 1024 * 4; +} + using json = nlohmann::json; struct ThreadMsg @@ -159,86 +168,122 @@ 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[SingleReadBufferSize]; + 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; + } + + // n is -1 -> error condition, check the errno + if (errno == EINTR) { + // ignored, we try to get more... + 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, + // but is checked separately for portability to platforms where it isn't. +#if EAGAIN == EWOULDBLOCK + if (errno != EAGAIN) { +#else + if (errno != EAGAIN && errno != EWOULDBLOCK) { +#endif + // there is a serious error coming from read + perror("read"); + return; + } + // end the loop if we get here + 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 +447,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 625e9248df9380a5d56b581d32829fe10100d312 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Mon, 5 Oct 2026 15:07:02 +0200 Subject: [PATCH 2/5] Clear _rxBuffer in case of error condition --- src/openvfsfuse/socketthread.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index ff69463..dff6c9b 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -177,7 +177,7 @@ void SocketThread::processSocketInput() while (true) { const ssize_t n = read(_socket, buf, sizeof(buf)); if (n > 0) { - _rxBuffer.append(buf, static_cast(n)); + _rxBuffer.append(buf, n); continue; } if (n == 0) { @@ -205,6 +205,7 @@ void SocketThread::processSocketInput() #endif // there is a serious error coming from read perror("read"); + _rxBuffer.clear(); return; } // end the loop if we get here From 5449d9e8d09e110ae42ac2645541bb11863e7056 Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:14:47 +0200 Subject: [PATCH 3/5] Bound the hydration wait and close the open() registration race First, the job was registered by the socket thread only after the send succeeded. The job is now inserted before the message is posted, so no waiter can observe the absence of a job it just posted. Second, SharedMap now carries a condition variable and SharedMap::waitForJob() blocks on it with a deadline. Also made _transfer_id atomic. --- 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 | 25 ++++++--- src/openvfsfuse/socketthread.h | 3 +- 9 files changed, 188 insertions(+), 88 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 61e5191..3b77c3b 100644 --- a/src/openvfsfuse/config.json +++ b/src/openvfsfuse/config.json @@ -34,5 +34,6 @@ "clamscan", "freshclam" ] - } + }, + "hydrationTimeoutSeconds": 300 } diff --git a/src/openvfsfuse/main.cpp b/src/openvfsfuse/main.cpp index d414f01..cae0040 100644 --- a/src/openvfsfuse/main.cpp +++ b/src/openvfsfuse/main.cpp @@ -5,6 +5,7 @@ #include "openvfsfuse.h" #include +#include #include #include #include @@ -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 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 506a598..b820300 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, ...) @@ -571,69 +579,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 3c9bf5e..df63b4f 100644 --- a/src/openvfsfuse/sharedmap.cpp +++ b/src/openvfsfuse/sharedmap.cpp @@ -21,12 +21,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) @@ -42,25 +48,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 c43151c..660ba1a 100644 --- a/src/openvfsfuse/sharedmap.h +++ b/src/openvfsfuse/sharedmap.h @@ -22,18 +22,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: @@ -44,6 +66,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(); @@ -51,6 +82,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 dff6c9b..de106c8 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -160,6 +160,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; } @@ -170,6 +173,10 @@ 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. @@ -263,9 +270,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; } @@ -363,10 +370,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 @@ -376,6 +383,7 @@ void SocketThread::PostMsg(std::shared_ptr data) std::unique_lock lk(m_mutex); m_queue.push(threadMsg); m_cv.notify_one(); + return true; } //---------------------------------------------------------------------------- @@ -436,11 +444,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 35275ff04749cef37a473959155249f63da2b23b Mon Sep 17 00:00:00 2001 From: toxicphreAK Date: Tue, 18 Aug 2026 15:35:15 +0200 Subject: [PATCH 4/5] 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/openvfsfuse/tests/CMakeLists.txt | 13 ++ src/openvfsfuse/tests/socketthreadtest.cpp | 169 +++++++++++++++++++++ 2 files changed, 182 insertions(+) create mode 100644 src/openvfsfuse/tests/socketthreadtest.cpp diff --git a/src/openvfsfuse/tests/CMakeLists.txt b/src/openvfsfuse/tests/CMakeLists.txt index 9af17b0..b1e847c 100644 --- a/src/openvfsfuse/tests/CMakeLists.txt +++ b/src/openvfsfuse/tests/CMakeLists.txt @@ -3,3 +3,16 @@ add_executable(strtools_test ../strtools.cpp ) add_test(NAME strtools_test COMMAND strtools_test) + +# 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) +add_test(NAME socketthreadtest COMMAND socketthreadtest) + diff --git a/src/openvfsfuse/tests/socketthreadtest.cpp b/src/openvfsfuse/tests/socketthreadtest.cpp new file mode 100644 index 0000000..c0725ad --- /dev/null +++ b/src/openvfsfuse/tests/socketthreadtest.cpp @@ -0,0 +1,169 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// SPDX-FileCopyrightText: 2026 toxicphreAK + +/* + * 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 "socketthread.h" +#include "sharedmap.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; +} From fa9af2d0941c59a52bf144aa62ca73bb51d75fd3 Mon Sep 17 00:00:00 2001 From: Klaas Freitag Date: Fri, 9 Oct 2026 17:03:26 +0200 Subject: [PATCH 5/5] Rework the socket read loop again to make it easier to grasp Add more tests --- src/openvfsfuse/socketthread.cpp | 74 +++++++++++----------- src/openvfsfuse/tests/socketthreadtest.cpp | 22 +++++++ 2 files changed, 59 insertions(+), 37 deletions(-) diff --git a/src/openvfsfuse/socketthread.cpp b/src/openvfsfuse/socketthread.cpp index de106c8..3291573 100644 --- a/src/openvfsfuse/socketthread.cpp +++ b/src/openvfsfuse/socketthread.cpp @@ -181,54 +181,54 @@ void SocketThread::processSocketInput() // read may return a fragment of a message, several messages at once, or // both. Accumulate into _rxBuffer and only dispatch complete lines. char buf[SingleReadBufferSize]; - while (true) { - const ssize_t n = read(_socket, buf, sizeof(buf)); + + bool contRead{true}; + ssize_t n; + while (contRead) { + n = read(_socket, buf, sizeof(buf)); + if (n > 0) { + // all good, we read something and continue to do so _rxBuffer.append(buf, 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; - } + } else if (n == 0) { + // Peer closed the connection. Whatever is left in the buffer will + // be interpreted - and remaining incompleted reads will be erased. + std::cerr << "Socket closed by peer" << std::endl; + contRead = false; + } else if (n < 0) { + // error condition + if (errno == EINTR) { + // EINTR is just a interruption, continue reading + } else { + // stop reading in all other error conditions + contRead = false; - // n is -1 -> error condition, check the errno - if (errno == EINTR) { - // ignored, we try to get more... - 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, - // but is checked separately for portability to platforms where it isn't. -#if EAGAIN == EWOULDBLOCK - if (errno != EAGAIN) { -#else - if (errno != EAGAIN && errno != EWOULDBLOCK) { + // ...and if it is serious, log it + if (errno != EAGAIN +#if EAGAIN != EWOULDBLOCK + && errno != EWOULDBLOCK #endif - // there is a serious error coming from read - perror("read"); - _rxBuffer.clear(); - return; + ) { + // a serious errno was set + perror("socket read"); + } + } } - // end the loop if we get here - break; } - size_t pos; - while ((pos = _rxBuffer.find('\n')) != std::string::npos) { - handleReceivedMsg(_rxBuffer.substr(0, pos)); - _rxBuffer.erase(0, pos + 1); + // handle the contents of _rxBuffer here + size_t newpos, pos{0}; + while ((newpos = _rxBuffer.find('\n', pos)) != std::string::npos) { + handleReceivedMsg(_rxBuffer.substr(pos, newpos - pos)); + pos = newpos + 1; } + _rxBuffer.erase(0, pos); // clear everything that was processed // 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; + // Also, if n == 0 because that are useless fragments after connection closed + if (n == 0 || _rxBuffer.size() > MaxRxBufferSize) { + std::cerr << "Discarding " << _rxBuffer.size() << " bytes from the socket API" << std::endl; _rxBuffer.clear(); } } diff --git a/src/openvfsfuse/tests/socketthreadtest.cpp b/src/openvfsfuse/tests/socketthreadtest.cpp index c0725ad..2cd27a2 100644 --- a/src/openvfsfuse/tests/socketthreadtest.cpp +++ b/src/openvfsfuse/tests/socketthreadtest.cpp @@ -51,6 +51,11 @@ std::string makeSocketPath() int listenOn(const std::string &path) { const int fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) { + std::cerr << "Failed to create socket " << path << ": " << std::strerror(errno) << std::endl; + std::exit(2); + } + sockaddr_un addr{}; addr.sun_family = AF_UNIX; std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1); @@ -131,6 +136,23 @@ int main() writeAll(client, hydrateResult(104, "\"status\":\"OK\"")); check(jobs.waitForJob(104, 10s) == HydJobResult::Succeeded, "the stream stays in sync"); + // An unterminated fragment that exceeds the receive buffer cap (exercises + // looping over several SingleReadBufferSize-sized reads within a single + // processSocketInput() call, i.e. the "n > 0, keep reading" branch of the + // loop) must be discarded instead of growing _rxBuffer without bound, and + // the stream must still be usable for the next message afterwards. + writeAll(client, std::string(150 * 1024, 'X')); // no trailing '\n', so never dispatched + std::this_thread::sleep_for(750ms); + post(150); + writeAll(client, hydrateResult(150, "\"status\":\"OK\"")); + check(jobs.waitForJob(150, 10s) == HydJobResult::Succeeded, "an oversized unterminated fragment is discarded and the stream recovers"); + + // An empty line (two consecutive newlines) must be handled by the + // buffer-splitting loop without desynchronising the stream. + post(151); + writeAll(client, "\n" + hydrateResult(151, "\"status\":\"OK\"")); + check(jobs.waitForJob(151, 10s) == HydJobResult::Succeeded, "an empty line ahead of a message does not desync the stream"); + // A silent client must time out, bounded by wall-clock time. post(105); const auto beforeTimeout = std::chrono::steady_clock::now();