Upload: asynchronious operations

Implements https://github.com/owncloud/core/pull/31851
This commit is contained in:
Olivier Goffart 2018-06-25 17:47:52 +02:00 committed by Kevin Ottens
parent 32c60c2f5d
commit 46bf3ed31a
No known key found for this signature in database
GPG key ID: 074BBBCB8DECC9E2
9 changed files with 349 additions and 46 deletions

View file

@ -111,9 +111,9 @@ public:
struct PollInfo
{
QString _file;
QString _url;
qint64 _modtime;
QString _file; // The relative path of a file
QString _url; // the poll url. (This pollinfo is invalid if _url is empty)
qint64 _modtime; // The modtime of the file being uploaded
};
DownloadInfo getDownloadInfo(const QString &file);

View file

@ -1005,13 +1005,12 @@ void CleanupPollsJob::start()
auto info = _pollInfos.first();
_pollInfos.pop_front();
SyncJournalFileRecord record;
if (_journal->getFileRecord(info._file, &record) && record.isValid()) {
SyncFileItemPtr item = SyncFileItem::fromSyncJournalFileRecord(record);
auto *job = new PollJob(_account, info._url, item, _journal, _localPath, this);
connect(job, &PollJob::finishedSignal, this, &CleanupPollsJob::slotPollFinished);
job->start();
}
SyncFileItemPtr item(new SyncFileItem);
item->_file = info._file;
item->_modtime = info._modtime;
auto *job = new PollJob(_account, info._url, item, _journal, _localPath, this);
connect(job, &PollJob::finishedSignal, this, &CleanupPollsJob::slotPollFinished);
job->start();
}
void CleanupPollsJob::slotPollFinished()
@ -1033,7 +1032,7 @@ void CleanupPollsJob::slotPollFinished()
deleteLater();
return;
}
// TODO: Is syncfilestatustracker notified somehow?
_journal->setUploadInfo(job->_item->_file, SyncJournalDb::UploadInfo());
}
// Continue with the next entry, or finish
start();

View file

@ -104,7 +104,7 @@ void PollJob::start()
QUrl finalUrl = QUrl::fromUserInput(accountUrl.scheme() + QLatin1String("://") + accountUrl.authority()
+ (path().startsWith('/') ? QLatin1String("") : QLatin1String("/")) + path());
sendRequest("GET", finalUrl);
connect(reply(), &QNetworkReply::downloadProgress, this, &AbstractNetworkJob::resetTimeout);
connect(reply(), &QNetworkReply::downloadProgress, this, &AbstractNetworkJob::resetTimeout, Qt::UniqueConnection);
AbstractNetworkJob::start();
}
@ -129,14 +129,14 @@ bool PollJob::finished()
emit finishedSignal();
return true;
}
start();
QTimer::singleShot(8 * 1000, this, &PollJob::start);
return false;
}
QByteArray jsonData = reply()->readAll().trimmed();
qCInfo(lcPollJob) << ">" << jsonData << "<" << reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt();
QJsonParseError jsonParseError;
QJsonObject status = QJsonDocument::fromJson(jsonData, &jsonParseError).object();
QJsonObject json = QJsonDocument::fromJson(jsonData, &jsonParseError).object();
qCInfo(lcPollJob) << ">" << jsonData << "<" << reply()->attribute(QNetworkRequest::HttpStatusCodeAttribute).toInt() << json << jsonParseError.errorString();
if (jsonParseError.error != QJsonParseError::NoError) {
_item->_errorString = tr("Invalid JSON reply from the poll URL");
_item->_status = SyncFileItem::NormalError;
@ -144,16 +144,23 @@ bool PollJob::finished()
return true;
}
if (status["unfinished"].toBool()) {
start();
auto status = json["status"].toString();
if (status == QLatin1String("init") || status == QLatin1String("started")) {
QTimer::singleShot(5 * 1000, this, &PollJob::start);
return false;
}
_item->_errorString = status["error"].toString();
_item->_status = _item->_errorString.isEmpty() ? SyncFileItem::Success : SyncFileItem::NormalError;
_item->_fileId = status["fileid"].toString().toUtf8();
_item->_etag = status["etag"].toString().toUtf8();
_item->_responseTimeStamp = responseTimestamp();
_item->_httpErrorCode = json["errorCode"].toInt();
if (status == QLatin1String("finished")) {
_item->_status = SyncFileItem::Success;
_item->_fileId = json["fileId"].toString().toUtf8();
_item->_etag = parseEtag(json["ETag"].toString().toUtf8());
} else { // error
_item->_status = classifyError(QNetworkReply::UnknownContentError, _item->_httpErrorCode);
_item->_errorString = json["errorMessage"].toString();
}
SyncJournalDb::PollInfo info;
info._file = _item->_file;
@ -705,9 +712,10 @@ void PropagateUploadFileCommon::abortWithError(SyncFileItem::Status status, cons
QMap<QByteArray, QByteArray> PropagateUploadFileCommon::headers()
{
QMap<QByteArray, QByteArray> headers;
headers[QByteArrayLiteral("OC-Async")] = QByteArrayLiteral("1");
headers[QByteArrayLiteral("Content-Type")] = QByteArrayLiteral("application/octet-stream");
headers[QByteArrayLiteral("X-OC-Mtime")] = QByteArray::number(qint64(_item->_modtime));
if (qEnvironmentVariableIntValue("OWNCLOUD_LAZYOPS"))
headers[QByteArrayLiteral("OC-LazyOps")] = QByteArrayLiteral("true");
if (_item->_file.contains(QLatin1String(".sys.admin#recall#"))) {
// This is a file recall triggered by the admin. Note: the

View file

@ -152,8 +152,8 @@ signals:
/**
* @brief This job implements the asynchronous PUT
*
* If the server replies to a PUT with a OC-Finish-Poll url, we will query this url until the server
* replies with an etag. https://github.com/owncloud/core/issues/12097
* If the server replies to a PUT with a OC-JobStatus-Location path, we will query this url until the server
* replies with an etag.
* @ingroup libsync
*/
class PollJob : public AbstractNetworkJob
@ -310,7 +310,7 @@ protected:
*/
static void adjustLastJobTimeout(AbstractNetworkJob *job, qint64 fileSize);
// Bases headers that need to be sent with every chunk
/** Bases headers that need to be sent on the PUT, or in the MOVE for chunking-ng */
QMap<QByteArray, QByteArray> headers();
private:
PropagateUploadEncrypted *_uploadEncryptedHelper;

View file

@ -459,6 +459,18 @@ void PropagateUploadFileNG::slotMoveJobFinished()
commonErrorHandling(job);
return;
}
if (_item->_httpErrorCode == 202) {
QString path = QString::fromUtf8(job->reply()->rawHeader("OC-JobStatus-Location"));
if (path.isEmpty()) {
done(SyncFileItem::NormalError, tr("Poll URL missing"));
return;
}
_finished = true;
startPollJob(path);
return;
}
if (_item->_httpErrorCode != 201 && _item->_httpErrorCode != 204) {
abortWithError(SyncFileItem::NormalError, tr("Unexpected return code from server (%1)").arg(_item->_httpErrorCode));
return;

View file

@ -211,7 +211,7 @@ void PropagateUploadFileV1::slotPutFinished()
// The server needs some time to process the request and provide us with a poll URL
if (_item->_httpErrorCode == 202) {
QString path = QString::fromUtf8(job->reply()->rawHeader("OC-Finish-Poll"));
QString path = QString::fromUtf8(job->reply()->rawHeader("OC-JobStatus-Location"));
if (path.isEmpty()) {
done(SyncFileItem::NormalError, tr("Poll URL missing"));
return;

View file

@ -52,6 +52,7 @@ nextcloud_add_test(SyncConflict "syncenginetestutils.h")
nextcloud_add_test(SyncFileStatusTracker "syncenginetestutils.h")
nextcloud_add_test(Download "syncenginetestutils.h")
nextcloud_add_test(ChunkingNg "syncenginetestutils.h")
nextcloud_add_test(AsyncOp "syncenginetestutils.h")
nextcloud_add_test(UploadReset "syncenginetestutils.h")
nextcloud_add_test(AllFilesDeleted "syncenginetestutils.h")
nextcloud_add_test(Blacklist "syncenginetestutils.h")

View file

@ -46,7 +46,7 @@ inline QString getFilePathFromUrl(const QUrl &url) {
inline QString generateEtag() {
return QString::number(QDateTime::currentDateTimeUtc().toMSecsSinceEpoch(), 16);
return QString::number(QDateTime::currentDateTimeUtc().toMSecsSinceEpoch(), 16) + QByteArray::number(qrand(), 16);
}
inline QByteArray generateFileId() {
return QByteArray::number(qrand(), 16);
@ -240,7 +240,7 @@ public:
auto file = it->find(std::move(pathComponents).subComponents(), invalidateEtags);
if (file && invalidateEtags) {
// Update parents on the way back
etag = file->etag;
etag = generateEtag();
}
return file;
}
@ -310,7 +310,6 @@ public:
QMap<QString, FileInfo> children;
QString parentPath;
private:
FileInfo *findInvalidatingEtags(PathComponents pathComponents) {
return find(std::move(pathComponents), true);
}
@ -432,24 +431,25 @@ public:
setUrl(request.url());
setOperation(op);
open(QIODevice::ReadOnly);
fileInfo = perform(remoteRootFileInfo, request, putPayload);
QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection);
}
static FileInfo *perform(FileInfo &remoteRootFileInfo, const QNetworkRequest &request, const QByteArray &putPayload)
{
QString fileName = getFilePathFromUrl(request.url());
Q_ASSERT(!fileName.isEmpty());
if ((fileInfo = remoteRootFileInfo.find(fileName))) {
FileInfo *fileInfo = remoteRootFileInfo.find(fileName);
if (fileInfo) {
fileInfo->size = putPayload.size();
fileInfo->contentChar = putPayload.at(0);
} else {
// Assume that the file is filled with the same character
fileInfo = remoteRootFileInfo.create(fileName, putPayload.size(), putPayload.at(0));
}
if (!fileInfo) {
abort();
return;
}
fileInfo->lastModified = OCC::Utility::qDateTimeFromTime_t(request.rawHeader("X-OC-Mtime").toLongLong());
remoteRootFileInfo.find(fileName, /*invalidateEtags=*/true);
QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection);
return fileInfo;
}
Q_INVOKABLE virtual void respond()
@ -639,7 +639,16 @@ public:
setUrl(request.url());
setOperation(op);
open(QIODevice::ReadOnly);
fileInfo = perform(uploadsFileInfo, remoteRootFileInfo, request);
if (!fileInfo) {
QTimer::singleShot(0, this, &FakeChunkMoveReply::respondPreconditionFailed);
} else {
QTimer::singleShot(0, this, &FakeChunkMoveReply::respond);
}
}
static FileInfo *perform(FileInfo &uploadsFileInfo, FileInfo &remoteRootFileInfo, const QNetworkRequest &request)
{
QString source = getFilePathFromUrl(request.url());
Q_ASSERT(!source.isEmpty());
Q_ASSERT(source.endsWith("/.file"));
@ -665,17 +674,17 @@ public:
} while(true);
Q_ASSERT(count > 1); // There should be at least two chunks, otherwise why would we use chunking?
QCOMPARE(sourceFolder->children.count(), count); // There should not be holes or extra files
Q_ASSERT(sourceFolder->children.count() == count); // There should not be holes or extra files
QString fileName = getFilePathFromUrl(QUrl::fromEncoded(request.rawHeader("Destination")));
Q_ASSERT(!fileName.isEmpty());
if ((fileInfo = remoteRootFileInfo.find(fileName))) {
QVERIFY(request.hasRawHeader("If")); // The client should put this header
FileInfo *fileInfo = remoteRootFileInfo.find(fileName);
if (fileInfo) {
Q_ASSERT(request.hasRawHeader("If")); // The client should put this header
if (request.rawHeader("If") != QByteArray("<" + request.rawHeader("Destination") +
"> ([\"" + fileInfo->etag.toLatin1() + "\"])")) {
QMetaObject::invokeMethod(this, "respondPreconditionFailed", Qt::QueuedConnection);
return;
return nullptr;
}
fileInfo->size = size;
fileInfo->contentChar = payload;
@ -685,14 +694,10 @@ public:
fileInfo = remoteRootFileInfo.create(fileName, size, payload);
}
if (!fileInfo) {
abort();
return;
}
fileInfo->lastModified = OCC::Utility::qDateTimeFromTime_t(request.rawHeader("X-OC-Mtime").toLongLong());
remoteRootFileInfo.find(fileName, /*invalidateEtags=*/true);
QTimer::singleShot(0, this, &FakeChunkMoveReply::respond);
return fileInfo;
}
Q_INVOKABLE virtual void respond()
@ -722,6 +727,48 @@ public:
};
class FakePayloadReply : public QNetworkReply
{
Q_OBJECT
public:
FakePayloadReply(QNetworkAccessManager::Operation op, const QNetworkRequest &request,
const QByteArray &body, QObject *parent)
: QNetworkReply{ parent }
, _body(body)
{
setRequest(request);
setUrl(request.url());
setOperation(op);
open(QIODevice::ReadOnly);
QTimer::singleShot(10, this, &FakePayloadReply::respond);
}
void respond()
{
setAttribute(QNetworkRequest::HttpStatusCodeAttribute, 200);
setHeader(QNetworkRequest::ContentLengthHeader, _body.size());
emit metaDataChanged();
emit readyRead();
setFinished(true);
emit finished();
}
void abort() override {}
qint64 readData(char *buf, qint64 max) override
{
max = qMin<qint64>(max, _body.size());
memcpy(buf, _body.constData(), max);
_body = _body.mid(max);
return max;
}
qint64 bytesAvailable() const override
{
return _body.size();
}
QByteArray _body;
};
class FakeErrorReply : public QNetworkReply
{
Q_OBJECT

236
test/testasyncop.cpp Normal file
View file

@ -0,0 +1,236 @@
/*
* This software is in the public domain, furnished "as is", without technical
* support, and with no warranty, express or implied, as to its usefulness for
* any purpose.
*
*/
#include <QtTest>
#include "syncenginetestutils.h"
#include <syncengine.h>
using namespace OCC;
class FakeAsyncReply : public QNetworkReply
{
Q_OBJECT
QByteArray _pollLocation;
public:
FakeAsyncReply(const QByteArray &pollLocation, QNetworkAccessManager::Operation op, const QNetworkRequest &request, QObject *parent)
: QNetworkReply{ parent }
, _pollLocation(pollLocation)
{
setRequest(request);
setUrl(request.url());
setOperation(op);
open(QIODevice::ReadOnly);
QMetaObject::invokeMethod(this, "respond", Qt::QueuedConnection);
}
Q_INVOKABLE void respond()
{
setAttribute(QNetworkRequest::HttpStatusCodeAttribute, 202);
setRawHeader("OC-JobStatus-Location", _pollLocation);
emit metaDataChanged();
emit finished();
}
void abort() override {}
qint64 readData(char *, qint64) override { return 0; }
};
class TestAsyncOp : public QObject
{
Q_OBJECT
private slots:
void asyncUploadOperations()
{
FakeFolder fakeFolder{ FileInfo::A12_B12_C12_S12() };
fakeFolder.syncEngine().account()->setCapabilities({ { "dav", QVariantMap{ { "chunking", "1.0" } } } });
// Reduce max chunk size a bit so we get more chunks
SyncOptions options;
options._maxChunkSize = 20 * 1000;
fakeFolder.syncEngine().setSyncOptions(options);
int nGET = 0;
// This test is made of several testcases.
// the testCases maps a filename to a couple of callback.
// When a file is uploaded, the fake server will always return the 202 code, and will set
// the `perform` functor to what needs to be done to complete the transaction.
// The testcase consist of the `pollRequest` which will be called when the sync engine
// calls the poll url.
struct TestCase
{
using PollRequest_t = std::function<QNetworkReply *(TestCase *, const QNetworkRequest &request)>;
PollRequest_t pollRequest;
std::function<FileInfo *()> perform = nullptr;
};
QHash<QString, TestCase> testCases;
fakeFolder.setServerOverride([&](QNetworkAccessManager::Operation op, const QNetworkRequest &request, QIODevice *outgoingData) -> QNetworkReply * {
auto path = request.url().path();
if (op == QNetworkAccessManager::GetOperation && path.startsWith("/async-poll/")) {
auto file = path.mid(sizeof("/async-poll/") - 1);
Q_ASSERT(testCases.contains(file));
auto &testCase = testCases[file];
return testCase.pollRequest(&testCase, request);
}
if (op == QNetworkAccessManager::PutOperation && !path.contains("/uploads/")) {
// Not chunking
auto file = getFilePathFromUrl(request.url());
Q_ASSERT(testCases.contains(file));
auto &testCase = testCases[file];
Q_ASSERT(!testCase.perform);
auto putPayload = outgoingData->readAll();
testCase.perform = [putPayload, request, &fakeFolder] {
return FakePutReply::perform(fakeFolder.remoteModifier(), request, putPayload);
};
return new FakeAsyncReply("/async-poll/" + file.toUtf8(), op, request, &fakeFolder.syncEngine());
} else if (request.attribute(QNetworkRequest::CustomVerbAttribute) == "MOVE") {
QString file = getFilePathFromUrl(QUrl::fromEncoded(request.rawHeader("Destination")));
Q_ASSERT(testCases.contains(file));
auto &testCase = testCases[file];
Q_ASSERT(!testCase.perform);
testCase.perform = [request, &fakeFolder] {
return FakeChunkMoveReply::perform(fakeFolder.uploadState(), fakeFolder.remoteModifier(), request);
};
return new FakeAsyncReply("/async-poll/" + file.toUtf8(), op, request, &fakeFolder.syncEngine());
} else if (op == QNetworkAccessManager::GetOperation) {
nGET++;
}
return nullptr;
});
// Callback to be used to finalize the transaction and return the success
auto successCallback = [](TestCase *tc, const QNetworkRequest &request) {
tc->pollRequest = [](auto...) -> QNetworkReply * { std::abort(); }; // shall no longer be called
FileInfo *info = tc->perform();
QByteArray body = "{ \"status\":\"finished\", \"ETag\":\"\\\"" + info->etag.toUtf8() + "\\\"\", \"fileId\":\"" + info->fileId + "\"}\n";
return new FakePayloadReply(QNetworkAccessManager::GetOperation, request, body, nullptr);
};
// Callback that never finishes
auto waitForeverCallback = [](TestCase *, const QNetworkRequest &request) {
QByteArray body = "{\"status\":\"started\"}\n";
return new FakePayloadReply(QNetworkAccessManager::GetOperation, request, body, nullptr);
};
// Callback that simulate an error.
auto errorCallback = [](TestCase *tc, const QNetworkRequest &request) {
tc->pollRequest = [](auto...) -> QNetworkReply * { std::abort(); }; // shall no longer be called;
QByteArray body = "{\"status\":\"error\",\"errorCode\":500,\"errorMessage\":\"TestingErrors\"}\n";
return new FakePayloadReply(QNetworkAccessManager::GetOperation, request, body, nullptr);
};
// This lambda takes another functor as a parameter, and returns a callback that will
// tell the client needs to poll again, and further call to the poll url will call the
// given callback
auto waitAndChain = [](const TestCase::PollRequest_t &chain) {
return [chain](TestCase *tc, const QNetworkRequest &request) {
tc->pollRequest = chain;
QByteArray body = "{\"status\":\"started\"}\n";
return new FakePayloadReply(QNetworkAccessManager::GetOperation, request, body, nullptr);
};
};
// Create a testcase by creating a file of a given size locally and assigning it a callback
auto insertFile = [&](const QString &file, int size, TestCase::PollRequest_t cb) {
fakeFolder.localModifier().insert(file, size);
testCases[file] = { std::move(cb) };
};
fakeFolder.localModifier().mkdir("success");
insertFile("success/chunked_success", options._maxChunkSize * 3, successCallback);
insertFile("success/single_success", 300, successCallback);
insertFile("success/chunked_patience", options._maxChunkSize * 3,
waitAndChain(waitAndChain(successCallback)));
insertFile("success/single_patience", 300,
waitAndChain(waitAndChain(successCallback)));
fakeFolder.localModifier().mkdir("err");
insertFile("err/chunked_error", options._maxChunkSize * 3, errorCallback);
insertFile("err/single_error", 300, errorCallback);
insertFile("err/chunked_error2", options._maxChunkSize * 3, waitAndChain(errorCallback));
insertFile("err/single_error2", 300, waitAndChain(errorCallback));
// First sync should finish by itself.
// All the things in "success/" should be transfered, the things in "err/" not
QVERIFY(!fakeFolder.syncOnce());
QCOMPARE(nGET, 0);
QCOMPARE(*fakeFolder.currentLocalState().find("success"),
*fakeFolder.currentRemoteState().find("success"));
testCases.clear();
testCases["err/chunked_error"] = { successCallback };
testCases["err/chunked_error2"] = { successCallback };
testCases["err/single_error"] = { successCallback };
testCases["err/single_error2"] = { successCallback };
fakeFolder.localModifier().mkdir("waiting");
insertFile("waiting/small", 300, waitForeverCallback);
insertFile("waiting/willNotConflict", 300, waitForeverCallback);
insertFile("waiting/big", options._maxChunkSize * 3,
waitAndChain(waitAndChain([&](TestCase *tc, const QNetworkRequest &request) {
QTimer::singleShot(0, &fakeFolder.syncEngine(), &SyncEngine::abort);
return waitAndChain(waitForeverCallback)(tc, request);
})));
fakeFolder.syncJournal().wipeErrorBlacklist();
// This second sync will redo the files that had errors
// But the waiting folder will not complete before it is aborted.
QVERIFY(!fakeFolder.syncOnce());
QCOMPARE(nGET, 0);
QCOMPARE(*fakeFolder.currentLocalState().find("err"),
*fakeFolder.currentRemoteState().find("err"));
testCases["waiting/small"].pollRequest = waitAndChain(waitAndChain(successCallback));
testCases["waiting/big"].pollRequest = waitAndChain(successCallback);
testCases["waiting/willNotConflict"].pollRequest =
[&fakeFolder, &successCallback](TestCase *tc, const QNetworkRequest &request) {
auto &remoteModifier = fakeFolder.remoteModifier(); // successCallback destroys the capture
auto reply = successCallback(tc, request);
// This is going to succeed, and after we just change the file.
// This should not be a conflict, but this should be downloaded in the
// next sync
remoteModifier.appendByte("waiting/willNotConflict");
return reply;
};
int nPUT = 0;
int nMOVE = 0;
int nDELETE = 0;
fakeFolder.setServerOverride([&](QNetworkAccessManager::Operation op, const QNetworkRequest &request, QIODevice *) -> QNetworkReply * {
auto path = request.url().path();
if (op == QNetworkAccessManager::GetOperation && path.startsWith("/async-poll/")) {
auto file = path.mid(sizeof("/async-poll/") - 1);
Q_ASSERT(testCases.contains(file));
auto &testCase = testCases[file];
return testCase.pollRequest(&testCase, request);
} else if (op == QNetworkAccessManager::PutOperation) {
nPUT++;
} else if (op == QNetworkAccessManager::GetOperation) {
nGET++;
} else if (op == QNetworkAccessManager::DeleteOperation) {
nDELETE++;
} else if (request.attribute(QNetworkRequest::CustomVerbAttribute) == "MOVE") {
nMOVE++;
}
return nullptr;
});
// This last sync will do the waiting stuff
QVERIFY(fakeFolder.syncOnce());
QCOMPARE(nGET, 1); // "waiting/willNotConflict"
QCOMPARE(nPUT, 0);
QCOMPARE(nMOVE, 0);
QCOMPARE(nDELETE, 0);
QCOMPARE(fakeFolder.currentLocalState(), fakeFolder.currentRemoteState());
}
};
QTEST_GUILESS_MAIN(TestAsyncOp)
#include "testasyncop.moc"