diff --git a/src/common/checksums.cpp b/src/common/checksums.cpp index 5787f2d26dd09..77291d09d3164 100644 --- a/src/common/checksums.cpp +++ b/src/common/checksums.cpp @@ -201,7 +201,7 @@ QByteArray ComputeChecksum::checksumType() const void ComputeChecksum::start(const QString &filePath) { - qCInfo(lcChecksums) << "Computing" << checksumType() << "checksum of" << filePath << "in a thread"; + qCDebug(lcChecksums) << "Computing" << checksumType() << "checksum of" << filePath << "in a thread"; startImpl(filePath); } diff --git a/src/common/syncjournaldb.cpp b/src/common/syncjournaldb.cpp index 870b22f8bbe83..0813c12a17dc4 100644 --- a/src/common/syncjournaldb.cpp +++ b/src/common/syncjournaldb.cpp @@ -1711,13 +1711,11 @@ static void toDownloadInfo(SqlQuery &query, SyncJournalDb::DownloadInfo *res) res->_valid = ok; } -static bool deleteBatch(SqlQuery &query, const QStringList &entries, const QString &name) +static bool deleteBatch(SqlQuery &query, const QStringList &entries) { if (entries.isEmpty()) return true; - qCDebug(lcDb) << "Removing stale" << name << "entries:" << entries.join(QStringLiteral(", ")); - // FIXME: Was ported from execBatch, check if correct! for (const auto &entry : entries) { query.reset_and_clear_bindings(); query.bindValue(1, entry); @@ -1831,7 +1829,7 @@ QVector SyncJournalDb::getAndDeleteStaleDownloadInf qCDebug(lcDb) << "database error:" << query->error(); return empty_result; } - if (!deleteBatch(*query, superfluousPaths, QStringLiteral("downloadinfo"))) { + if (!deleteBatch(*query, superfluousPaths)) { return empty_result; } } @@ -1965,7 +1963,7 @@ QVector SyncJournalDb::deleteStaleUploadInfos(const QSet &keep) } const auto deleteUploadInfoQuery = _queryManager.get(PreparedSqlQueryManager::DeleteUploadInfoQuery); - deleteBatch(*deleteUploadInfoQuery, superfluousPaths, QStringLiteral("uploadinfo")); + deleteBatch(*deleteUploadInfoQuery, superfluousPaths); return ids; } @@ -2033,7 +2031,7 @@ bool SyncJournalDb::deleteStaleErrorBlacklistEntries(const QSet &keep) SqlQuery delQuery(_db); delQuery.prepare("DELETE FROM blacklist WHERE path = ?"); - return deleteBatch(delQuery, superfluousPaths, QStringLiteral("blacklist")); + return deleteBatch(delQuery, superfluousPaths); } void SyncJournalDb::deleteStaleFlagsEntries() diff --git a/src/gui/folderwatcher.cpp b/src/gui/folderwatcher.cpp index fc7d05d685a45..e8bed199bdef6 100644 --- a/src/gui/folderwatcher.cpp +++ b/src/gui/folderwatcher.cpp @@ -223,8 +223,6 @@ void FolderWatcher::changeDetected(const QStringList &paths) _lockedFiles.insert(checkResult.path); } - qCDebug(lcFolderWatcher) << "Locked files:" << _lockedFiles.values(); - // ------- handle ignores: if (pathIsIgnored(path)) { continue; @@ -233,9 +231,6 @@ void FolderWatcher::changeDetected(const QStringList &paths) changedPaths.insert(path); } - qCDebug(lcFolderWatcher) << "Unlocked files:" << _unlockedFiles.values(); - qCDebug(lcFolderWatcher) << "Locked files:" << _lockedFiles; - if (!_lockedFiles.isEmpty() || !_unlockedFiles.isEmpty()) { if (_lockChangeDebouncingTimer.isActive()) { _lockChangeDebouncingTimer.stop(); diff --git a/src/libsync/bulkpropagatorjob.cpp b/src/libsync/bulkpropagatorjob.cpp index 7e2154198d88f..80d17d1c9c196 100644 --- a/src/libsync/bulkpropagatorjob.cpp +++ b/src/libsync/bulkpropagatorjob.cpp @@ -58,7 +58,6 @@ QByteArray getHeaderFromJsonReply(const QJsonObject &reply, const QByteArray &he return reply.value(headerName).toString().toLatin1(); } -constexpr auto batchSize = 100; constexpr auto parallelJobsMaximumCount = 1; } @@ -70,9 +69,10 @@ Q_LOGGING_CATEGORY(lcBulkPropagatorJob, "nextcloud.sync.propagator.bulkupload", BulkPropagatorJob::BulkPropagatorJob(OwncloudPropagator *propagator, const std::deque &items) : PropagatorJob(propagator) , _items(items) + , _currentBatchSize(_items.size()) { - _filesToUpload.reserve(batchSize); - _pendingChecksumFiles.reserve(batchSize); + _filesToUpload.reserve(_items.size()); + _pendingChecksumFiles.reserve(_items.size()); } bool BulkPropagatorJob::scheduleSelfOrChild() @@ -83,11 +83,15 @@ bool BulkPropagatorJob::scheduleSelfOrChild() _state = Running; - for(auto i = 0; i < batchSize && !_items.empty(); ++i) { + qCDebug(lcBulkPropagatorJob()) << "max chunk size" << PropagatorJob::propagator()->syncOptions().maxChunkSize(); + + for(auto batchDataSize = 0; batchDataSize <= PropagatorJob::propagator()->syncOptions().maxChunkSize() && !_items.empty(); ) { const auto currentItem = _items.front(); _items.pop_front(); _pendingChecksumFiles.insert(currentItem->_file); + batchDataSize += currentItem->_size; + QMetaObject::invokeMethod(this, [this, currentItem] { UploadFileInfo fileToUpload; fileToUpload._file = currentItem->_file; @@ -107,6 +111,29 @@ bool BulkPropagatorJob::scheduleSelfOrChild() return _items.empty() && _filesToUpload.empty(); } +bool BulkPropagatorJob::handleBatchSize() +{ + // no error, no batch size to change + if (_finalStatus == SyncFileItem::Success || _finalStatus == SyncFileItem::NoStatus) { + qCDebug(lcBulkPropagatorJob) << "No error, no need to change the bulk upload batch size!"; + return true; + } + + // change batch size before trying it again + const auto halfBatchSize = static_cast(_items.size() / 2); + + // we already tried to upload with half of the batch size + if(_currentBatchSize == halfBatchSize) { + qCDebug(lcBulkPropagatorJob) << "There was another error, stop syncing now!"; + return false; + } + + // try to upload with half of the batch size + _currentBatchSize = halfBatchSize; + qCDebug(lcBulkPropagatorJob) << "There was an error, sync again with bulk upload batch size cut to half!"; + return true; +} + PropagatorJob::JobParallelism BulkPropagatorJob::parallelism() const { return PropagatorJob::JobParallelism::FullParallelism; @@ -197,7 +224,6 @@ void BulkPropagatorJob::doStartUpload(SyncFileItemPtr item, remotePath, fileToUpload._path, fileToUpload._size, currentHeaders}; - qCInfo(lcBulkPropagatorJob) << remotePath << "transmission checksum" << transmissionChecksumHeader << fileToUpload._path; _filesToUpload.push_back(std::move(newUploadFile)); _pendingChecksumFiles.remove(item->_file); @@ -265,13 +291,16 @@ void BulkPropagatorJob::checkPropagationIsDone() // just wait for the other job to finish. return; } - - qCInfo(lcBulkPropagatorJob) << "final status" << _finalStatus; - emit finished(_finalStatus); - propagator()->scheduleNextJob(); } else { - scheduleSelfOrChild(); + if (handleBatchSize()) { + scheduleSelfOrChild(); + return; + } } + + qCInfo(lcBulkPropagatorJob) << "final status" << _finalStatus; + emit finished(_finalStatus); + propagator()->scheduleNextJob(); } void BulkPropagatorJob::slotComputeTransmissionChecksum(SyncFileItemPtr item, @@ -418,6 +447,9 @@ void BulkPropagatorJob::slotPutFinishedOneFile(const BulkUploadItem &singleFile, singleFile._item->_status = SyncFileItem::Success; + // upload succeeded, so remove from black list + propagator()->removeFromBulkUploadBlackList(singleFile._item->_file); + // Check the file again post upload. // Two cases must be considered separately: If the upload is finished, // the file is on the server and has a changed ETag. In that case, @@ -553,7 +585,7 @@ void BulkPropagatorJob::finalizeOneFile(const BulkUploadItem &oneFile) void BulkPropagatorJob::finalize(const QJsonObject &fullReply) { - qCDebug(lcBulkPropagatorJob) << "Received a full reply" << fullReply; + qCDebug(lcBulkPropagatorJob) << "Received a full reply" << QJsonDocument::fromVariant(fullReply).toJson(); for(auto singleFileIt = std::begin(_filesToUpload); singleFileIt != std::end(_filesToUpload); ) { const auto &singleFile = *singleFileIt; diff --git a/src/libsync/bulkpropagatorjob.h b/src/libsync/bulkpropagatorjob.h index 95189bf2670fd..611bf8a9e0854 100644 --- a/src/libsync/bulkpropagatorjob.h +++ b/src/libsync/bulkpropagatorjob.h @@ -162,6 +162,8 @@ private slots: void checkPropagationIsDone(); + bool handleBatchSize(); + std::deque _items; QVector _jobs; /// network jobs that are currently in transit @@ -173,6 +175,7 @@ private slots: qint64 _sentTotal = 0; SyncFileItem::Status _finalStatus = SyncFileItem::Status::NoStatus; + int _currentBatchSize = 0; }; } diff --git a/src/libsync/configfile.cpp b/src/libsync/configfile.cpp index 1efd01e5da355..8f875cc4d0f88 100644 --- a/src/libsync/configfile.cpp +++ b/src/libsync/configfile.cpp @@ -270,13 +270,13 @@ qint64 ConfigFile::chunkSize() const qint64 ConfigFile::maxChunkSize() const { QSettings settings(configFile(), QSettings::IniFormat); - return settings.value(QLatin1String(maxChunkSizeC), 5LL * 1000LL * 1000LL * 1000LL).toLongLong(); // default to 5000 MB + return settings.value(QLatin1String(maxChunkSizeC), 100LL * 1024LL * 1024LL).toLongLong(); // default to 100 MiB } qint64 ConfigFile::minChunkSize() const { QSettings settings(configFile(), QSettings::IniFormat); - return settings.value(QLatin1String(minChunkSizeC), 5LL * 1000LL * 1000LL).toLongLong(); // default to 5 MB + return settings.value(QLatin1String(minChunkSizeC), 5LL * 1024LL * 1024LL).toLongLong(); // default to 5 MiB } chrono::milliseconds ConfigFile::targetChunkUploadDuration() const diff --git a/src/libsync/owncloudpropagator.cpp b/src/libsync/owncloudpropagator.cpp index beec5b96610a6..969a8ba031afc 100644 --- a/src/libsync/owncloudpropagator.cpp +++ b/src/libsync/owncloudpropagator.cpp @@ -427,8 +427,6 @@ std::unique_ptr OwncloudPropagator::createUploadJob(S job->setDeleteExisting(deleteExisting); - removeFromBulkUploadBlackList(item->_file); - return job; } @@ -1269,7 +1267,9 @@ bool PropagatorCompositeJob::scheduleSelfOrChild() _tasksToDo.remove(0); PropagatorJob *job = propagator()->createJob(nextTask); if (!job) { - qCWarning(lcDirectory) << "Useless task found for file" << nextTask->destination() << "instruction" << nextTask->_instruction; + if (!propagator()->isDelayedUploadItem(nextTask)) { + qCWarning(lcDirectory) << "Useless task found for file" << nextTask->destination() << "instruction" << nextTask->_instruction; + } continue; } appendJob(job); @@ -1338,8 +1338,9 @@ void PropagatorCompositeJob::finalize() { // The propagator will do parallel scheduling and this could be posted // multiple times on the event loop, ignore the duplicate calls. - if (_state == Finished) + if (_state == Finished) { return; + } _state = Finished; emit finished(_hasError == SyncFileItem::NoStatus ? SyncFileItem::Success : _hasError); diff --git a/src/libsync/propagateupload.cpp b/src/libsync/propagateupload.cpp index 2bac80f172b89..ab7899c525c7d 100644 --- a/src/libsync/propagateupload.cpp +++ b/src/libsync/propagateupload.cpp @@ -64,18 +64,27 @@ void PUTFileJob::start() req.setPriority(QNetworkRequest::LowPriority); // Long uploads must not block non-propagation jobs. + auto requestID = QByteArray{}; + if (_url.isValid()) { - sendRequest("PUT", _url, req, _device); + const auto reply = sendRequest("PUT", _url, req, _device); + requestID = reply->request().rawHeader("X-Request-ID"); } else { - sendRequest("PUT", makeDavUrl(path()), req, _device); + const auto reply = sendRequest("PUT", makeDavUrl(path()), req, _device); + requestID = reply->request().rawHeader("X-Request-ID"); } if (reply()->error() != QNetworkReply::NoError) { qCWarning(lcPutJob) << " Network error: " << reply()->errorString(); } + connect(reply(), &QNetworkReply::uploadProgress, this, [requestID] (qint64 bytesSent, qint64 bytesTotal) { + qCDebug(lcPutJob()) << requestID << "upload progress" << bytesSent << bytesTotal; + }); + connect(reply(), &QNetworkReply::uploadProgress, this, &PUTFileJob::uploadProgress); connect(this, &AbstractNetworkJob::networkActivity, account().data(), &Account::propagatorNetworkActivity); + _requestTimer.start(); AbstractNetworkJob::start(); } diff --git a/src/libsync/putmultifilejob.cpp b/src/libsync/putmultifilejob.cpp index 46b15f8b58ffa..aa23e9b5ca316 100644 --- a/src/libsync/putmultifilejob.cpp +++ b/src/libsync/putmultifilejob.cpp @@ -32,8 +32,6 @@ PutMultiFileJob::PutMultiFileJob(AccountPtr account, for(const auto &singleDevice : _devices) { singleDevice._device->setParent(this); - connect(this, &PutMultiFileJob::uploadProgress, - singleDevice._device.get(), &UploadDevice::slotJobUploadProgress); } } @@ -56,7 +54,12 @@ void PutMultiFileJob::start() if (oneDevice._device->size() == 0) { onePart.setBody({}); } else { - onePart.setBodyDevice(oneDevice._device.get()); + const auto allData = oneDevice._device->readAll(); + onePart.setBody(allData); + } + + if (oneDevice._device->isOpen()) { + oneDevice._device->close(); } for (auto it = oneDevice._headers.begin(); it != oneDevice._headers.end(); ++it) { @@ -68,13 +71,17 @@ void PutMultiFileJob::start() _body.append(onePart); } - sendRequest("POST", _url, req, &_body); + const auto newReply = sendRequest("POST", _url, req, &_body); + const auto &requestID = newReply->request().rawHeader("X-Request-ID"); if (reply()->error() != QNetworkReply::NoError) { qCWarning(lcPutMultiFileJob) << " Network error: " << reply()->errorString(); } connect(reply(), &QNetworkReply::uploadProgress, this, &PutMultiFileJob::uploadProgress); + connect(reply(), &QNetworkReply::uploadProgress, this, [requestID] (qint64 bytesSent, qint64 bytesTotal) { + qCDebug(lcPutMultiFileJob()) << requestID << "upload progress" << bytesSent << bytesTotal; + }); connect(this, &AbstractNetworkJob::networkActivity, account().data(), &Account::propagatorNetworkActivity); _requestTimer.start(); AbstractNetworkJob::start(); @@ -90,15 +97,12 @@ bool PutMultiFileJob::finished() for(const auto &oneDevice : _devices) { Q_ASSERT(oneDevice._device); - if (!oneDevice._device->errorString().isEmpty()) { - qCWarning(lcPutMultiFileJob) << "oneDevice has error:" << oneDevice._device->errorString(); - } - if (oneDevice._device->isOpen()) { + if (!oneDevice._device->errorString().isEmpty()) { + qCWarning(lcPutMultiFileJob) << "oneDevice has error:" << oneDevice._device->errorString(); + } + oneDevice._device->close(); - } else { - qCWarning(lcPutMultiFileJob) << "Did not close device" << oneDevice._device.get() - << "as it was not open"; } } diff --git a/src/libsync/syncengine.cpp b/src/libsync/syncengine.cpp index 24c86a73b9a0d..9319116312b3b 100644 --- a/src/libsync/syncengine.cpp +++ b/src/libsync/syncengine.cpp @@ -228,7 +228,6 @@ void SyncEngine::deleteStaleDownloadInfos(const SyncFileItemVector &syncItems) _journal->getAndDeleteStaleDownloadInfos(download_file_paths); for (const SyncJournalDb::DownloadInfo &deleted_info : deleted_infos) { const QString tmppath = _propagator->fullLocalPath(deleted_info._tmpfile); - qCInfo(lcEngine) << "Deleting stale temporary file: " << tmppath; FileSystem::remove(tmppath); } } diff --git a/test/testsyncengine.cpp b/test/testsyncengine.cpp index ec603c17e81fe..21ca008f92582 100644 --- a/test/testsyncengine.cpp +++ b/test/testsyncengine.cpp @@ -1150,6 +1150,84 @@ private slots: QCOMPARE(fakeFolder.currentLocalState(), fakeFolder.currentRemoteState()); } + void testNetworkErrorsWithSmallerBatchSizes() + { + FakeFolder fakeFolder{ FileInfo::A12_B12_C12_S12() }; + fakeFolder.syncEngine().account()->setCapabilities({ { "dav", QVariantMap{ {"bulkupload", "1.0"} } } }); + + int nPUT = 0; + int nPOST = 0; + fakeFolder.setServerOverride([&](QNetworkAccessManager::Operation op, const QNetworkRequest &request, QIODevice *outgoingData) -> QNetworkReply * { + auto contentType = request.header(QNetworkRequest::ContentTypeHeader).toString(); + if (op == QNetworkAccessManager::PostOperation) { + ++nPOST; + if (contentType.startsWith(QStringLiteral("multipart/related; boundary="))) { + auto jsonReplyObject = fakeFolder.forEachReplyPart(outgoingData, contentType, [] (const QMap &allHeaders) -> QJsonObject { + auto reply = QJsonObject{}; + const auto fileName = allHeaders[QStringLiteral("X-File-Path")]; + if(fileName.endsWith("B/small30") || + fileName.endsWith("B/small60") || + fileName.endsWith("B/big30") || + fileName.endsWith("B/big60")) { + reply.insert(QStringLiteral("error"), true); + reply.insert(QStringLiteral("etag"), {}); + return reply; + } else { + reply.insert(QStringLiteral("error"), false); + reply.insert(QStringLiteral("etag"), {}); + } + return reply; + }); + if (jsonReplyObject.size()) { + auto jsonReply = QJsonDocument{}; + jsonReply.setObject(jsonReplyObject); + return new FakeJsonErrorReply{op, request, this, 200, jsonReply}; + } + return nullptr; + } + } else if (op == QNetworkAccessManager::PutOperation) { + ++nPUT; + const auto fileName = getFilePathFromUrl(request.url()); + if (fileName.endsWith("B/small30") || + fileName.endsWith("B/small60") || + fileName.endsWith("B/big30") || + fileName.endsWith("B/big60")) { + return new FakeErrorReply(op, request, this, 504); + } + return nullptr; + } + return nullptr; + }); + + const auto smallSize = 0.5 * 1000 * 1000; + const auto bigSize = 10 * 1000 * 1000; + + for(auto i = 0 ; i < 120; ++i) { + fakeFolder.localModifier().insert(QString("A/small%1").arg(i), smallSize); + } + + QVERIFY(fakeFolder.syncOnce()); + QCOMPARE(nPUT, 0); + QCOMPARE(nPOST, 1); + nPUT = 0; + nPOST = 0; + + for(auto i = 0 ; i < 120; ++i) { + fakeFolder.localModifier().insert(QString("B/small%1").arg(i), smallSize); + fakeFolder.localModifier().insert(QString("B/big%1").arg(i), bigSize); + } + + QVERIFY(!fakeFolder.syncOnce()); + QCOMPARE(nPUT, 120); + QCOMPARE(nPOST, 1); + nPUT = 0; + nPOST = 0; + + QVERIFY(!fakeFolder.syncOnce()); + QCOMPARE(nPUT, 0); + QCOMPARE(nPOST, 0); + } + void testRemoteMoveFailedInsufficientStorageLocalMoveRolledBack() { FakeFolder fakeFolder{FileInfo{}};