Remove file from bulk upload black list only when upload succeeded.

src/libsync/bulkpropagatorjob.cpp

@@ -1,4 +1,4 @@
 /*
  * Copyright 2021 (c) Matthieu Gallien <[email protected]>
  *
  * This program is free software; you can redistribute it and/or modify
@@ -70,6 +70,7 @@
 BulkPropagatorJob::BulkPropagatorJob(OwncloudPropagator *propagator, const std::deque<SyncFileItemPtr> &items)
     : PropagatorJob(propagator)
     , _items(items)
+    , _currentBatchSize(batchSize)
 {
     _filesToUpload.reserve(batchSize);
     _pendingChecksumFiles.reserve(batchSize);
@@ -83,7 +84,7 @@
 
     _state = Running;
 
-    for(auto i = 0; i < batchSize && !_items.empty(); ++i) {
+    for(auto i = 0; i < _currentBatchSize && !_items.empty(); ++i) {
         const auto currentItem = _items.front();
         _items.pop_front();
         _pendingChecksumFiles.insert(currentItem->_file);
@@ -107,6 +108,29 @@
     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 = batchSize/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;
@@ -254,13 +278,16 @@
             // 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,
@@ -374,6 +401,9 @@
 
     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,

src/libsync/bulkpropagatorjob.h

@@ -155,6 +155,8 @@ private slots:
 
     void checkPropagationIsDone();
 
+    bool handleBatchSize();
+
     std::deque<SyncFileItemPtr> _items;
 
     QVector<AbstractNetworkJob *> _jobs; /// network jobs that are currently in transit
@@ -166,6 +168,7 @@ private slots:
     qint64 _sentTotal = 0;
 
     SyncFileItem::Status _finalStatus = SyncFileItem::Status::NoStatus;
+    int _currentBatchSize = 0;
 };
 
 }

src/libsync/owncloudpropagator.cpp

@@ -428,8 +428,6 @@ std::unique_ptr<PropagateUploadFileCommon> OwncloudPropagator::createUploadJob(S
 
     job->setDeleteExisting(deleteExisting);
 
-    removeFromBulkUploadBlackList(item->_file);
-
     return job;
 }
 
@@ -1329,8 +1327,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);

test/testsyncengine.cpp

@@ -1,4 +1,4 @@
 /*
  *    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.
@@ -1086,6 +1086,84 @@
         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<QString, QByteArray> &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, 2);
+        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, 2);
+        nPUT = 0;
+        nPOST = 0;
+
+        QVERIFY(!fakeFolder.syncOnce());
+        QCOMPARE(nPUT, 2);
+        QCOMPARE(nPOST, 0);
+    }
+
     void testRemoteMoveFailedInsufficientStorageLocalMoveRolledBack()
     {
         FakeFolder fakeFolder{FileInfo{}};