Skip to content

Commit

Permalink
KDESKTOP-1232 - EOF is now expected at the end of CSV file
Browse files Browse the repository at this point in the history
  • Loading branch information
@ClementKunz
ClementKunz committed Oct 22, 2024
1 parent 94d04b7 commit 1f22f10
Show file tree
Hide file tree
Showing 6 changed files with 179 additions and 41 deletions.
37 changes: 16 additions & 21 deletions src/libsyncengine/jobs/network/API_v2/csvfullfilelistwithcursorjob.cpp
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@

/*
* Infomaniak kDrive - Desktop
* Copyright (C) 2023-2024 Infomaniak Network SA
Expand Down Expand Up @@ -29,9 +28,11 @@

#define API_TIMEOUT 900

constexpr std::string endOfFileDelimiter = "#EOF";

namespace KDC {

SnapshotItemHandler::SnapshotItemHandler(log4cplus::Logger logger) : _logger(logger){};
SnapshotItemHandler::SnapshotItemHandler(log4cplus::Logger logger) : _logger(logger) {};

void SnapshotItemHandler::logError(const std::wstring &methodName, const std::wstring &stdErrorType, const std::string &str,
const std::exception &exc) {
Expand Down Expand Up @@ -188,7 +189,7 @@ void SnapshotItemHandler::readSnapshotItemFields(SnapshotItem &item, const std::
}
}

bool SnapshotItemHandler::getItem(SnapshotItem &item, std::stringstream &ss, bool &error, bool &ignore) {
bool SnapshotItemHandler::getItem(SnapshotItem &item, std::stringstream &ss, bool &error, bool &ignore, bool &eof) {
error = false;
ignore = false;

Expand All @@ -209,8 +210,13 @@ bool SnapshotItemHandler::getItem(SnapshotItem &item, std::stringstream &ss, boo
}
}

ParsingState state;
if (line == endOfFileDelimiter) {
LOG_INFO(_logger, "End of file reached");
eof = true;
return false;
}

ParsingState state;
while (state.readNextLine) {
state.readNextLine = false;

Expand Down Expand Up @@ -256,8 +262,8 @@ bool SnapshotItemHandler::getItem(SnapshotItem &item, std::stringstream &ss, boo

CsvFullFileListWithCursorJob::CsvFullFileListWithCursorJob(int driveDbId, const NodeId &dirId,
std::unordered_set<NodeId> blacklist /*= {}*/, bool zip /*= true*/) :
AbstractTokenNetworkJob(ApiType::Drive, 0, 0, driveDbId, 0),
_dirId(dirId), _blacklist(blacklist), _zip(zip), _snapshotItemHandler(_logger) {
AbstractTokenNetworkJob(ApiType::Drive, 0, 0, driveDbId, 0), _dirId(dirId), _blacklist(blacklist), _zip(zip),
_snapshotItemHandler(_logger) {
_httpMethod = Poco::Net::HTTPRequest::HTTP_GET;
_customTimeout = API_TIMEOUT + 15;

Expand All @@ -266,11 +272,11 @@ CsvFullFileListWithCursorJob::CsvFullFileListWithCursorJob(int driveDbId, const
}
}

bool CsvFullFileListWithCursorJob::getItem(SnapshotItem &item, bool &error, bool &ignore) {
bool CsvFullFileListWithCursorJob::getItem(SnapshotItem &item, bool &error, bool &ignore, bool &eof) {
error = false;
ignore = false;

return _snapshotItemHandler.getItem(item, _ss, error, ignore);
return _snapshotItemHandler.getItem(item, _ss, error, ignore, eof);
}

std::string CsvFullFileListWithCursorJob::getCursor() {
Expand All @@ -286,7 +292,7 @@ std::string CsvFullFileListWithCursorJob::getSpecificUrl() {
void CsvFullFileListWithCursorJob::setQueryParameters(Poco::URI &uri, bool &canceled) {
uri.addQueryParameter("directory_id", _dirId);
uri.addQueryParameter("recursive", "true");
uri.addQueryParameter("format", "csv");
uri.addQueryParameter("format", "safe_csv");
if (!_blacklist.empty()) {
std::string str = Utility::list2str(_blacklist);
if (!str.empty()) {
Expand All @@ -307,28 +313,17 @@ bool CsvFullFileListWithCursorJob::handleResponse(std::istream &is) {

// Check that the stringstream is not empty (network issues)
_ss.seekg(0, std::ios_base::end);
int length = _ss.tellg();
const auto length = _ss.tellg();
if (length == 0) {
LOG_ERROR(_logger, "Reply " << jobId() << " received with empty content.");
return false;
}

// Check that the stringstream ends by a LF (0x0A)
_ss.seekg(length - 1, std::ios_base::beg);
char lastChar = 0x00;
_ss.read(&lastChar, 1);
_ss.seekg(0, std::ios_base::beg);
if (lastChar != 0x0A) {
LOGW_WARN(_logger, L"Reply " << jobId() << L" received with bad content - length=" << length << L" value="
<< Utility::s2ws(_ss.str()).c_str());
return false;
}

if (isExtendedLog()) {
LOGW_DEBUG(_logger,
L"Reply " << jobId() << L" received - length=" << length << L" value=" << Utility::s2ws(_ss.str()).c_str());
}

return true;
}

Expand Down
6 changes: 4 additions & 2 deletions src/libsyncengine/jobs/network/API_v2/csvfullfilelistwithcursorjob.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,10 @@ class SnapshotItemHandler {
* @param ss stringstream containg the CSV file
* @param error `true` if parsing fails
* @param ignore `true` if a line is ignored due to a non critical parsing issue
* @param eof `true` if the end of file (i.e. a line containg the end of file delimiter) has been reached
* @return `true` if there are more lines to be read
*/
bool getItem(SnapshotItem &item, std::stringstream &ss, bool &error, bool &ignore);
bool getItem(SnapshotItem &item, std::stringstream &ss, bool &error, bool &ignore, bool &eof);

private:
bool _ignoreFirstLine = true;
Expand All @@ -81,9 +82,10 @@ class CsvFullFileListWithCursorJob : public AbstractTokenNetworkJob {
* @param item : item extracted from line of the CSV file
* @param error : blocking error, stop the process
* @param ignore : parsing issue, non blocking, just ignore the item
* @param eof : whether the end of file has been reached or not
* @return if return == true, continue parsing
*/
bool getItem(SnapshotItem &item, bool &error, bool &ignore);
bool getItem(SnapshotItem &item, bool &error, bool &ignore, bool &eof);
std::string getCursor();

private:
Expand Down
16 changes: 13 additions & 3 deletions src/libsyncengine/update_detection/file_system_observer/remotefilesystemobserverworker.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,7 @@ namespace KDC {

RemoteFileSystemObserverWorker::RemoteFileSystemObserverWorker(std::shared_ptr<SyncPal> syncPal, const std::string &name,
const std::string &shortName) :
FileSystemObserverWorker(syncPal, name, shortName, ReplicaSide::Remote),
_driveDbId(syncPal->driveDbId()) {}
FileSystemObserverWorker(syncPal, name, shortName, ReplicaSide::Remote), _driveDbId(syncPal->driveDbId()) {}

RemoteFileSystemObserverWorker::~RemoteFileSystemObserverWorker() {
LOG_SYNCPAL_DEBUG(_logger, "~RemoteFileSystemObserverWorker");
Expand Down Expand Up @@ -317,13 +316,16 @@ ExitCode RemoteFileSystemObserverWorker::getItemsInDir(const NodeId &dirId, cons
SnapshotItem item;
bool error = false;
bool ignore = false;
bool eof = false;
std::unordered_set<SyncName> existingFiles;
uint64_t itemCount = 0;
while (job->getItem(item, error, ignore)) {
while (job->getItem(item, error, ignore, eof)) {
if (ignore) {
continue;
}

if (eof) break;

itemCount++;

if (error) {
Expand Down Expand Up @@ -374,6 +376,14 @@ ExitCode RemoteFileSystemObserverWorker::getItemsInDir(const NodeId &dirId, cons
}
}

if (!eof) {
const std::string msg = "Failed to parse CSV reply: missing EOF delimiter";
LOG_SYNCPAL_WARN(_logger, msg.c_str());
SentryHandler::instance()->captureMessage(SentryLevel::Warning, "RemoteFileSystemObserverWorker::getItemsInDir", "msg");
setExitCause(ExitCause::FullListParsingError);
return ExitCode::NetworkError;
}

// Delete orphans
std::unordered_set<NodeId> nodeIds;
_snapshot->ids(nodeIds);
Expand Down
34 changes: 31 additions & 3 deletions test/libsyncengine/jobs/network/testnetworkjobs.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,8 @@ void TestNetworkJobs::testFullFileListWithCursorCsv() {
SnapshotItem item;
bool error = false;
bool ignore = false;
while (job.getItem(item, error, ignore)) {
bool eof = false;
while (job.getItem(item, error, ignore, eof)) {
if (ignore) {
continue;
}
Expand All @@ -463,6 +464,7 @@ void TestNetworkJobs::testFullFileListWithCursorCsv() {

CPPUNIT_ASSERT(!cursor.empty());
CPPUNIT_ASSERT(counter == 5);
CPPUNIT_ASSERT(eof);
}

void TestNetworkJobs::testFullFileListWithCursorCsvZip() {
Expand All @@ -475,7 +477,8 @@ void TestNetworkJobs::testFullFileListWithCursorCsvZip() {
SnapshotItem item;
bool error = false;
bool ignore = false;
while (job.getItem(item, error, ignore)) {
bool eof = false;
while (job.getItem(item, error, ignore, eof)) {
if (ignore) {
continue;
}
Expand All @@ -487,6 +490,7 @@ void TestNetworkJobs::testFullFileListWithCursorCsvZip() {

CPPUNIT_ASSERT(!cursor.empty());
CPPUNIT_ASSERT(counter == 5);
CPPUNIT_ASSERT(eof);
}

void TestNetworkJobs::testFullFileListWithCursorJsonBlacklist() {
Expand Down Expand Up @@ -525,7 +529,8 @@ void TestNetworkJobs::testFullFileListWithCursorCsvBlacklist() {
SnapshotItem item;
bool error = false;
bool ignore = false;
while (job.getItem(item, error, ignore)) {
bool eof = false;
while (job.getItem(item, error, ignore, eof)) {
if (ignore) {
continue;
}
Expand All @@ -537,6 +542,29 @@ void TestNetworkJobs::testFullFileListWithCursorCsvBlacklist() {

CPPUNIT_ASSERT(!cursor.empty());
CPPUNIT_ASSERT(counter == 0);
CPPUNIT_ASSERT(eof);
}

void TestNetworkJobs::testFullFileListWithCursorMissingEof() {
CsvFullFileListWithCursorJob job(_driveDbId, "1");
ExitCode exitCode = job.runSynchronously();
CPPUNIT_ASSERT(exitCode == ExitCode::Ok);

int counter = 0;
const std::string cursor = job.getCursor();
SnapshotItem item;
bool error = false;
bool ignore = false;
bool eof = false;
// Call getItem only once to simulate a troncated CSV file
job.getItem(item, error, ignore, eof);
if (item.parentId() == pictureDirRemoteId) {
counter++;
}

CPPUNIT_ASSERT(!cursor.empty());
CPPUNIT_ASSERT_LESS(5, counter);
CPPUNIT_ASSERT(!eof);
}

void TestNetworkJobs::testGetInfoUser() {
Expand Down
2 changes: 2 additions & 0 deletions test/libsyncengine/jobs/network/testnetworkjobs.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ class TestNetworkJobs : public CppUnit::TestFixture {
CPPUNIT_TEST(testFullFileListWithCursorCsvZip);
CPPUNIT_TEST(testFullFileListWithCursorJsonBlacklist);
CPPUNIT_TEST(testFullFileListWithCursorCsvBlacklist);
CPPUNIT_TEST(testFullFileListWithCursorMissingEof);
CPPUNIT_TEST(testGetInfoUser);
CPPUNIT_TEST(testGetInfoDrive);
CPPUNIT_TEST(testThumbnail);
Expand Down Expand Up @@ -78,6 +79,7 @@ class TestNetworkJobs : public CppUnit::TestFixture {
void testFullFileListWithCursorCsvZip();
void testFullFileListWithCursorJsonBlacklist();
void testFullFileListWithCursorCsvBlacklist();
void testFullFileListWithCursorMissingEof();
void testGetInfoUser();
void testGetInfoDrive();
void testThumbnail();
Expand Down
Loading

0 comments on commit 1f22f10

Please sign in to comment.