Merge branch 'develop'

1 files changed, 30 insertions, 27 deletions

diff --git a/common/changereplay.cpp b/common/changereplay.cpp
index 7895b66..0adbd78 100644
--- a/common/changereplay.cpp
+++ b/common/changereplay.cpp
@@ -30,8 +30,7 @@ using namespace Sink;
 using namespace Sink::Storage;
 
 ChangeReplay::ChangeReplay(const ResourceContext &resourceContext, const Sink::Log::Context &ctx)
-    : mStorage(storageLocation(), resourceContext.instanceId(), DataStore::ReadOnly), mChangeReplayStore(storageLocation(), resourceContext.instanceId() + ".changereplay", DataStore::ReadWrite), mReplayInProgress(false), mLogCtx{ctx.subContext("changereplay")},
-    mGuard{new QObject}
+    : mStorage(storageLocation(), resourceContext.instanceId(), DataStore::ReadOnly), mChangeReplayStore(storageLocation(), resourceContext.instanceId() + ".changereplay", DataStore::ReadWrite), mReplayInProgress(false), mLogCtx{ctx.subContext("changereplay")}
 {
 }
 
@@ -84,6 +83,8 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
             auto replayStoreTransaction = mChangeReplayStore.createTransaction(DataStore::ReadOnly, [this](const DataStore::Error &error) {
                 SinkWarningCtx(mLogCtx) << error.message;
             });
+            Q_ASSERT(mMainStoreTransaction);
+            Q_ASSERT(replayStoreTransaction);
             replayStoreTransaction.openDatabase().scan("lastReplayedRevision",
                 [lastReplayedRevision](const QByteArray &key, const QByteArray &value) -> bool {
                     *lastReplayedRevision = value.toLongLong();
@@ -98,16 +99,13 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
             SinkTraceCtx(mLogCtx) << "Changereplay from " << *lastReplayedRevision << " to " << *topRevision;
             return KAsync::doWhile(
                 [this, lastReplayedRevision, topRevision]() -> KAsync::Job<KAsync::ControlFlowFlag> {
-                    if (!mGuard) {
-                        SinkTraceCtx(mLogCtx) << "Exit due to guard";
-                        return KAsync::value(KAsync::Break);
-                    }
                     if (*lastReplayedRevision >= *topRevision) {
                         SinkTraceCtx(mLogCtx) << "Done replaying" << *lastReplayedRevision << *topRevision;
                         return KAsync::value(KAsync::Break);
                     }
+                    Q_ASSERT(mMainStoreTransaction);
 
-                    KAsync::Job<void> replayJob = KAsync::null<void>();
+                    auto replayJob = KAsync::null();
                     qint64 revision = *lastReplayedRevision + 1;
                     while (revision <= *topRevision) {
                         const auto uid = DataStore::getUidFromRevision(mMainStoreTransaction, revision);
@@ -131,14 +129,14 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
                                 if (canReplay(type, key, entityBuffer)) {
                                     SinkTraceCtx(mLogCtx) << "Replaying " << key;
                                     replayJob = replay(type, key, entityBuffer);
+                                    //Set the last revision we tried to replay
+                                    *lastReplayedRevision = revision;
+                                    //Execute replay job and commit
+                                    break;
                                 } else {
-                                    SinkTraceCtx(mLogCtx) << "Cannot replay " << key;
+                                    SinkTraceCtx(mLogCtx) << "Not replaying " << key;
                                     //We silently skip over revisions that cannot be replayed, as this is not an error.
-                                    replayJob = KAsync::null();
                                 }
-                                //Set the last revision we tried to replay
-                                *lastReplayedRevision = revision;
-                                break;
                             }
                         }
                         //Bump the revision if we failed to even attempt to replay. This will simply skip over those revisions, as we can't recover from those situations.
@@ -149,22 +147,22 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
                         if (error) {
                             SinkWarningCtx(mLogCtx) << "Change replay failed: " << error  << "Last replayed revision: "  << *lastReplayedRevision;
                             //We're probably not online or so, so postpone retrying
-                            return KAsync::value(KAsync::Break);
+                            return KAsync::value(KAsync::Break).then(KAsync::error<KAsync::ControlFlowFlag>(error));
+                        }
+                        SinkTraceCtx(mLogCtx) << "Replayed until: " << *lastReplayedRevision;
+
+                        recordReplayedRevision(*lastReplayedRevision);
+                        reportProgress(*lastReplayedRevision, *topRevision);
+
+                        const bool gotMoreToReplay = (*lastReplayedRevision < *topRevision);
+                        if (gotMoreToReplay) {
+                            SinkTraceCtx(mLogCtx) << "Replaying some more...";
+                            //Replay more if we have more
+                            return KAsync::wait(0).then(KAsync::value(KAsync::Continue));
                         } else {
-                            SinkTraceCtx(mLogCtx) << "Replayed until: " << *lastReplayedRevision;
-                            recordReplayedRevision(*lastReplayedRevision);
-                            if (*lastReplayedRevision < *topRevision) {
-                                SinkTraceCtx(mLogCtx) << "Replaying some more...";
-                                //Replay more if we have more
-                                return KAsync::wait(0).then(KAsync::value(KAsync::Continue));
-                            } else {
-                                return KAsync::value(KAsync::Break);
-                            }
+                            return KAsync::value(KAsync::Break);
                         }
-                        //We shouldn't ever get here
-                        Q_ASSERT(false);
-                        return KAsync::value(KAsync::Break);
-                    });
+                    }).guard(&mGuard);
             });
         })
         .then([this](const KAsync::Error &error) {
@@ -181,7 +179,12 @@ KAsync::Job<void> ChangeReplay::replayNextRevision()
                     emit changesReplayed();
                 }
             }
-        });
+            if (error) {
+                return KAsync::error(error);
+            } else {
+                return KAsync::null();
+            }
+        }).guard(&mGuard);
 }
 
 void ChangeReplay::revisionChanged()