Debug output

3 files changed, 62 insertions, 66 deletions

diff --git a/common/changereplay.cpp b/common/changereplay.cpp
index 8ed0532..532cca8 100644
--- a/common/changereplay.cpp
+++ b/common/changereplay.cpp
@@ -30,7 +30,7 @@ using namespace Sink;
 using namespace Sink::Storage;
 
 ChangeReplay::ChangeReplay(const ResourceContext &resourceContext)
-    : mStorage(storageLocation(), resourceContext.instanceId(), DataStore::ReadOnly), mChangeReplayStore(storageLocation(), resourceContext.instanceId() + ".changereplay", DataStore::ReadWrite), mReplayInProgress(false), mLogCtx{resourceContext.instanceId() + ".changereplay"}
+    : mStorage(storageLocation(), resourceContext.instanceId(), DataStore::ReadOnly), mChangeReplayStore(storageLocation(), resourceContext.instanceId() + ".changereplay", DataStore::ReadWrite), mReplayInProgress(false), mLogCtx{"changereplay"}
 {
     SinkTraceCtx(mLogCtx) << "Created change replay: " << resourceContext.instanceId();
 }
diff --git a/common/storage/entitystore.cpp b/common/storage/entitystore.cpp
index 96c8ccd..6c341e2 100644
--- a/common/storage/entitystore.cpp
+++ b/common/storage/entitystore.cpp
@@ -37,8 +37,6 @@
 using namespace Sink;
 using namespace Sink::Storage;
 
-SINK_DEBUG_AREA("entitystore");
-
 class EntityStore::Private {
 public:
     Private(const ResourceContext &context, const Sink::Log::Context &ctx) : resourceContext(context), logCtx(ctx.subContext("entitystore")) {}
@@ -112,7 +110,7 @@ EntityStore::EntityStore(const ResourceContext &context, const Log::Context &ctx
 
 void EntityStore::startTransaction(Sink::Storage::DataStore::AccessMode accessMode)
 {
-    SinkTraceCtx(d->logCtx) << "Starting transaction";
+    SinkTraceCtx(d->logCtx) << "Starting transaction: " << accessMode;
     Sink::Storage::DataStore store(Sink::storageLocation(), d->resourceContext.instanceId(), accessMode);
     d->transaction = store.createTransaction(accessMode);
     Q_ASSERT(d->transaction.validateNamedDatabases());
@@ -136,7 +134,7 @@ void EntityStore::copyBlobs(ApplicationDomain::ApplicationDomainType &entity, qi
 {
     const auto directory = d->entityBlobStoragePath(entity.identifier());
     if (!QDir().mkpath(directory)) {
-        SinkWarning() << "Failed to create the directory: " << directory;
+        SinkWarningCtx(d->logCtx) << "Failed to create the directory: " << directory;
     }
 
     for (const auto &property : entity.changedProperties()) {
@@ -151,10 +149,10 @@ void EntityStore::copyBlobs(ApplicationDomain::ApplicationDomainType &entity, qi
                 QFile::remove(filePath);
                 QFile origFile(oldPath);
                 if (!origFile.open(QIODevice::ReadWrite)) {
-                    SinkWarning() << "Failed to open the original file with write rights: " << origFile.errorString();
+                    SinkWarningCtx(d->logCtx) << "Failed to open the original file with write rights: " << origFile.errorString();
                 }
                 if (!origFile.rename(filePath)) {
-                    SinkWarning() << "Failed to move the file from: " << oldPath << " to " << filePath << ". " << origFile.errorString();
+                    SinkWarningCtx(d->logCtx) << "Failed to move the file from: " << oldPath << " to " << filePath << ". " << origFile.errorString();
                 }
                 origFile.close();
                 entity.setProperty(property, QVariant::fromValue(ApplicationDomain::BLOB{filePath}));
@@ -166,7 +164,7 @@ void EntityStore::copyBlobs(ApplicationDomain::ApplicationDomainType &entity, qi
 bool EntityStore::add(const QByteArray &type, const ApplicationDomain::ApplicationDomainType &entity_, bool replayToSource, const PreprocessCreation &preprocess)
 {
     if (entity_.identifier().isEmpty()) {
-        SinkWarning() << "Can't write entity with an empty identifier";
+        SinkWarningCtx(d->logCtx) << "Can't write entity with an empty identifier";
         return false;
     }
 
@@ -197,7 +195,7 @@ bool EntityStore::add(const QByteArray &type, const ApplicationDomain::Applicati
 
     DataStore::mainDatabase(d->transaction, type)
         .write(DataStore::assembleKey(entity.identifier(), newRevision), BufferUtils::extractBuffer(fbb),
-            [&](const DataStore::Error &error) { SinkWarning() << "Failed to write entity" << entity.identifier() << newRevision; });
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << entity.identifier() << newRevision; });
     DataStore::setMaxRevision(d->transaction, newRevision);
     DataStore::recordRevision(d->transaction, newRevision, entity.identifier(), type);
     SinkTraceCtx(d->logCtx) << "Wrote entity: " << entity.identifier() << type << newRevision;
@@ -209,7 +207,7 @@ bool EntityStore::modify(const QByteArray &type, const ApplicationDomain::Applic
     auto changeset = diff.changedProperties();
     const auto current = readLatest(type, diff.identifier());
     if (current.identifier().isEmpty()) {
-        SinkWarning() << "Failed to read current version: " << diff.identifier();
+        SinkWarningCtx(d->logCtx) << "Failed to read current version: " << diff.identifier();
         return false;
     }
 
@@ -263,7 +261,7 @@ bool EntityStore::modify(const QByteArray &type, const ApplicationDomain::Applic
 
     DataStore::mainDatabase(d->transaction, type)
         .write(DataStore::assembleKey(newEntity.identifier(), newRevision), BufferUtils::extractBuffer(fbb),
-            [&](const DataStore::Error &error) { SinkWarning() << "Failed to write entity" << newEntity.identifier() << newRevision; });
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << newEntity.identifier() << newRevision; });
     DataStore::setMaxRevision(d->transaction, newRevision);
     DataStore::recordRevision(d->transaction, newRevision, newEntity.identifier(), type);
     SinkTraceCtx(d->logCtx) << "Wrote modified entity: " << newEntity.identifier() << type << newRevision;
@@ -287,14 +285,14 @@ bool EntityStore::remove(const QByteArray &type, const QByteArray &uid, bool rep
                 }
                 return false;
             },
-            [](const DataStore::Error &error) { SinkWarning() << "Failed to read old revision from storage: " << error.message; });
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read old revision from storage: " << error.message; });
 
     if (!found) {
-        SinkWarning() << "Remove: Failed to find entity " << uid;
+        SinkWarningCtx(d->logCtx) << "Remove: Failed to find entity " << uid;
         return false;
     }
     if (alreadyRemoved) {
-        SinkWarning() << "Remove: Entity is already removed " << uid;
+        SinkWarningCtx(d->logCtx) << "Remove: Entity is already removed " << uid;
         return false;
     }
 
@@ -320,7 +318,7 @@ bool EntityStore::remove(const QByteArray &type, const QByteArray &uid, bool rep
 
     DataStore::mainDatabase(d->transaction, type)
         .write(DataStore::assembleKey(uid, newRevision), BufferUtils::extractBuffer(fbb),
-            [&](const DataStore::Error &error) { SinkWarning() << "Failed to write entity" << uid << newRevision; });
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to write entity" << uid << newRevision; });
     DataStore::setMaxRevision(d->transaction, newRevision);
     DataStore::recordRevision(d->transaction, newRevision, uid, type);
     return true;
@@ -336,7 +334,7 @@ void EntityStore::cleanupEntityRevisionsUntil(qint64 revision)
             [&](const QByteArray &key, const QByteArray &data) -> bool {
                 EntityBuffer buffer(const_cast<const char *>(data.data()), data.size());
                 if (!buffer.isValid()) {
-                    SinkWarning() << "Read invalid buffer from disk";
+                    SinkWarningCtx(d->logCtx) << "Read invalid buffer from disk";
                 } else {
                     const auto metadata = flatbuffers::GetRoot<Metadata>(buffer.metadataBuffer());
                     const qint64 rev = metadata->revision();
@@ -350,7 +348,7 @@ void EntityStore::cleanupEntityRevisionsUntil(qint64 revision)
                         const auto directory = d->entityBlobStoragePath(uid);
                         QDir dir(directory);
                         if (!dir.removeRecursively()) {
-                            SinkError() << "Failed to cleanup: " << directory;
+                            SinkErrorCtx(d->logCtx) << "Failed to cleanup: " << directory;
                         }
                     }
                     //Don't cleanup more than specified
@@ -361,7 +359,7 @@ void EntityStore::cleanupEntityRevisionsUntil(qint64 revision)
 
                 return true;
             },
-            [](const DataStore::Error &error) { SinkWarning() << "Error while reading: " << error.message; }, true);
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error while reading: " << error.message; }, true);
     DataStore::setCleanedUpRevision(d->transaction, revision);
 }
 
@@ -407,7 +405,7 @@ QVector<QByteArray> EntityStore::fullScan(const QByteArray &type)
                 keys << uid;
                 return true;
             },
-            [](const DataStore::Error &error) { SinkWarning() << "Error during query: " << error.message; });
+            [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during query: " << error.message; });
 
     SinkTraceCtx(d->logCtx) << "Full scan retrieved " << keys.size() << " results.";
     return keys.toList().toVector();
@@ -446,7 +444,7 @@ void EntityStore::indexLookup(const QByteArray &type, const QByteArray &property
     /*     callback(sinkId); */
     /* }, */
     /* [&](const Index::Error &error) { */
-    /*     SinkWarning() << "Error in index: " <<  error.message << property; */
+    /*     SinkWarningCtx(d->logCtx) << "Error in index: " <<  error.message << property; */
     /* }); */
 }
 
@@ -458,7 +456,7 @@ void EntityStore::readLatest(const QByteArray &type, const QByteArray &uid, cons
             callback(DataStore::uidFromKey(key), Sink::EntityBuffer(value.data(), value.size()));
             return false;
         },
-        [&](const DataStore::Error &error) { SinkWarning() << "Error during query: " << error.message << uid; });
+        [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during query: " << error.message << uid; });
 }
 
 void EntityStore::readLatest(const QByteArray &type, const QByteArray &uid, const std::function<void(const ApplicationDomain::ApplicationDomainType &)> callback)
@@ -494,7 +492,7 @@ void EntityStore::readEntity(const QByteArray &type, const QByteArray &key, cons
             callback(DataStore::uidFromKey(key), Sink::EntityBuffer(value.data(), value.size()));
             return false;
         },
-        [&](const DataStore::Error &error) { SinkWarning() << "Error during query: " << error.message << key; });
+        [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during query: " << error.message << key; });
 }
 
 void EntityStore::readEntity(const QByteArray &type, const QByteArray &uid, const std::function<void(const ApplicationDomain::ApplicationDomainType &)> callback)
@@ -524,7 +522,7 @@ void EntityStore::readAll(const QByteArray &type, const std::function<void(const
             callback(d->createApplicationDomainType(type, uid, DataStore::maxRevision(d->getTransaction()), buffer));
             return true;
         },
-        [&](const DataStore::Error &error) { SinkWarning() << "Error during query: " << error.message; });
+        [&](const DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during query: " << error.message; });
 }
 
 void EntityStore::readRevisions(qint64 baseRevision, const QByteArray &expectedType, const std::function<void(const QByteArray &key)> &callback)
@@ -561,7 +559,7 @@ void EntityStore::readPrevious(const QByteArray &type, const QByteArray &uid, qi
             }
             return true;
         },
-        [](const Sink::Storage::DataStore::Error &error) { SinkWarning() << "Failed to read current value from storage: " << error.message; }, true);
+        [&](const Sink::Storage::DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read current value from storage: " << error.message; }, true);
     return readEntity(type, Sink::Storage::DataStore::assembleKey(uid, latestRevision), callback);
 }
 
@@ -591,7 +589,7 @@ void EntityStore::readAllUids(const QByteArray &type, const std::function<void(c
             callback(Sink::Storage::DataStore::uidFromKey(key));
             return true;
         },
-        [](const Sink::Storage::DataStore::Error &error) { SinkWarning() << "Failed to read current value from storage: " << error.message; });
+        [&](const Sink::Storage::DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Failed to read current value from storage: " << error.message; });
 }
 
 bool EntityStore::contains(const QByteArray &type, const QByteArray &uid)
@@ -615,10 +613,10 @@ qint64 EntityStore::maxRevision()
 /*         Sink::Storage::DataStore storage(Sink::storageLocation(), mResourceInstanceIdentifier); */
 /*         if (!storage.exists()) { */
 /*             //This is not an error if the resource wasn't started before */
-/*             SinkLog() << "Store doesn't exist: " << mResourceInstanceIdentifier; */
+/*             SinkLogCtx(d->logCtx) << "Store doesn't exist: " << mResourceInstanceIdentifier; */
 /*             return Sink::Storage::DataStore::Transaction(); */
 /*         } */
-/*         storage.setDefaultErrorHandler([this](const Sink::Storage::DataStore::Error &error) { SinkWarning() << "Error during query: " << error.store << error.message; }); */
+/*         storage.setDefaultErrorHandler([this](const Sink::Storage::DataStore::Error &error) { SinkWarningCtx(d->logCtx) << "Error during query: " << error.store << error.message; }); */
 /*         transaction = storage.createTransaction(Sink::Storage::DataStore::ReadOnly); */
 /*     } */
 
diff --git a/common/synchronizer.cpp b/common/synchronizer.cpp
index 57e994e..f731b31 100644
--- a/common/synchronizer.cpp
+++ b/common/synchronizer.cpp
@@ -30,8 +30,6 @@
 #include "flush_generated.h"
 #include "notification_generated.h"
 
-SINK_DEBUG_AREA("synchronizer")
-
 using namespace Sink;
 
 Synchronizer::Synchronizer(const Sink::ResourceContext &context)
@@ -42,7 +40,7 @@ Synchronizer::Synchronizer(const Sink::ResourceContext &context)
     mSyncStorage(Sink::storageLocation(), mResourceContext.instanceId() + ".synchronization", Sink::Storage::DataStore::DataStore::ReadWrite),
     mSyncInProgress(false)
 {
-    SinkTrace() << "Starting synchronizer: " << mResourceContext.resourceType << mResourceContext.instanceId();
+    SinkTraceCtx(mLogCtx) << "Starting synchronizer: " << mResourceContext.resourceType << mResourceContext.instanceId();
 }
 
 Synchronizer::~Synchronizer()
@@ -128,11 +126,11 @@ void Synchronizer::scanForRemovals(const QByteArray &bufferType, const std::func
 {
     entryGenerator([this, bufferType, &exists](const QByteArray &sinkId) {
         const auto remoteId = syncStore().resolveLocalId(bufferType, sinkId);
-        SinkTrace() << "Checking for removal " << sinkId << remoteId;
+        SinkTraceCtx(mLogCtx) << "Checking for removal " << sinkId << remoteId;
         // If we have no remoteId, the entity hasn't been replayed to the source yet
         if (!remoteId.isEmpty()) {
             if (!exists(remoteId)) {
-                SinkTrace() << "Found a removed entity: " << sinkId;
+                SinkTraceCtx(mLogCtx) << "Found a removed entity: " << sinkId;
                 deleteEntity(sinkId, mEntityStore->maxRevision(), bufferType);
             }
         }
@@ -157,12 +155,12 @@ void Synchronizer::modifyIfChanged(Storage::EntityStore &store, const QByteArray
         bool changed = false;
         for (const auto &property : entity.changedProperties()) {
             if (entity.getProperty(property) != current.getProperty(property)) {
-                SinkTrace() << "Property changed " << sinkId << property;
+                SinkTraceCtx(mLogCtx) << "Property changed " << sinkId << property;
                 changed = true;
             }
         }
         if (changed) {
-            SinkTrace() << "Found a modified entity: " << sinkId;
+            SinkTraceCtx(mLogCtx) << "Found a modified entity: " << sinkId;
             modifyEntity(sinkId, store.maxRevision(), bufferType, entity);
         }
     });
@@ -177,12 +175,12 @@ void Synchronizer::modify(const QByteArray &bufferType, const QByteArray &remote
 
 void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const Sink::ApplicationDomain::ApplicationDomainType &entity)
 {
-    SinkTrace() << "Create or modify" << bufferType << remoteId;
+    SinkTraceCtx(mLogCtx) << "Create or modify" << bufferType << remoteId;
     Storage::EntityStore store(mResourceContext, mLogCtx);
     const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
     const auto found = store.contains(bufferType, sinkId);
     if (!found) {
-        SinkTrace() << "Found a new entity: " << remoteId;
+        SinkTraceCtx(mLogCtx) << "Found a new entity: " << remoteId;
         createEntity(sinkId, bufferType, entity);
     } else { // modification
         modify(bufferType, remoteId, entity);
@@ -193,7 +191,7 @@ template<typename DomainType>
 void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray &remoteId, const DomainType &entity, const QHash<QByteArray, Sink::Query::Comparator> &mergeCriteria)
 {
 
-    SinkTrace() << "Create or modify" << bufferType << remoteId;
+    SinkTraceCtx(mLogCtx) << "Create or modify" << bufferType << remoteId;
     const auto sinkId = syncStore().resolveRemoteId(bufferType, remoteId);
     Storage::EntityStore store(mResourceContext, mLogCtx);
     const auto found = store.contains(bufferType, sinkId);
@@ -209,16 +207,16 @@ void Synchronizer::createOrModify(const QByteArray &bufferType, const QByteArray
             auto resultSet = dataStoreQuery.execute();
             resultSet.replaySet(0, 1, [this, &merge, bufferType, remoteId](const ResultSet::Result &r) {
                 merge = true;
-                SinkTrace() << "Merging local entity with remote entity: " << r.entity.identifier() << remoteId;
+                SinkTraceCtx(mLogCtx) << "Merging local entity with remote entity: " << r.entity.identifier() << remoteId;
                 syncStore().recordRemoteId(bufferType, r.entity.identifier(), remoteId);
             });
 
             if (!merge) {
-                SinkTrace() << "Found a new entity: " << remoteId;
+                SinkTraceCtx(mLogCtx) << "Found a new entity: " << remoteId;
                 createEntity(sinkId, bufferType, entity);
             }
         } else {
-            SinkTrace() << "Found a new entity: " << remoteId;
+            SinkTraceCtx(mLogCtx) << "Found a new entity: " << remoteId;
             createEntity(sinkId, bufferType, entity);
         }
     } else { // modification
@@ -240,7 +238,7 @@ QByteArrayList Synchronizer::resolveFilter(const QueryBase::Comparator &filter)
             result << r.entity.identifier();
         });
     } else {
-        SinkWarning() << "unknown filter type: " << filter;
+        SinkWarningCtx(mLogCtx) << "unknown filter type: " << filter;
         Q_ASSERT(false);
     }
     return result;
@@ -261,7 +259,7 @@ QList<Synchronizer::SyncRequest> Synchronizer::getSyncRequests(const Sink::Query
 
 void Synchronizer::synchronize(const Sink::QueryBase &query)
 {
-    SinkTrace() << "Synchronizing";
+    SinkTraceCtx(mLogCtx) << "Synchronizing";
     mSyncRequestQueue << getSyncRequests(query);
     processSyncQueue().exec();
 }
@@ -269,7 +267,7 @@ void Synchronizer::synchronize(const Sink::QueryBase &query)
 void Synchronizer::flush(int commandId, const QByteArray &flushId)
 {
     Q_ASSERT(!flushId.isEmpty());
-    SinkTrace() << "Flushing the synchronization queue " << flushId;
+    SinkTraceCtx(mLogCtx) << "Flushing the synchronization queue " << flushId;
     mSyncRequestQueue << Synchronizer::SyncRequest{Synchronizer::SyncRequest::Flush, commandId, flushId};
     processSyncQueue().exec();
 }
@@ -299,7 +297,7 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
             if (modifiedRequest.requestId.isEmpty()) {
                 modifiedRequest.requestId = QUuid::createUuid().toRfc4122();
             }
-            SinkWarning() << "Enquing flush request " << modifiedRequest.requestId;
+            SinkWarningCtx(mLogCtx) << "Enquing flush request " << modifiedRequest.requestId;
 
             //The sync request will be executed once the flush has completed
             mPendingSyncRequests.insert(modifiedRequest.requestId, modifiedRequest);
@@ -349,7 +347,7 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
             Q_ASSERT(!request.requestId.isEmpty());
             //FIXME it looks like this is emitted before the replay actually finishes
             if (request.flushType == Flush::FlushReplayQueue) {
-                SinkTrace() << "Emitting flush completion.";
+                SinkTraceCtx(mLogCtx) << "Emitting flush completion.";
                 Sink::Notification n;
                 n.type = Sink::Notification::FlushCompletion;
                 n.id = request.requestId;
@@ -365,7 +363,7 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
     } else if (request.requestType == Synchronizer::SyncRequest::ChangeReplay) {
         return replayNextRevision();
     } else {
-        SinkWarning() << "Unknown request type: " << request.requestType;
+        SinkWarningCtx(mLogCtx) << "Unknown request type: " << request.requestType;
         return KAsync::error(KAsync::Error{"Unknown request type."});
     }
 
@@ -374,13 +372,13 @@ KAsync::Job<void> Synchronizer::processRequest(const SyncRequest &request)
 KAsync::Job<void> Synchronizer::processSyncQueue()
 {
     if (mSyncRequestQueue.isEmpty() || mSyncInProgress) {
-        SinkTrace() << "Sync still in progress or nothing to do.";
+        SinkTraceCtx(mLogCtx) << "Sync still in progress or nothing to do.";
         return KAsync::null<void>();
     }
     //Don't process any new requests until we're done with the pending ones.
     //Otherwise we might process a flush before the previous request actually completed.
     if (!mPendingSyncRequests.isEmpty()) {
-        SinkTrace() << "We still have pending sync requests. Not executing next request.";
+        SinkTraceCtx(mLogCtx) << "We still have pending sync requests. Not executing next request.";
         return KAsync::null<void>();
     }
 
@@ -391,7 +389,7 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
     })
     .then(processRequest(request))
     .then<void>([this](const KAsync::Error &error) {
-        SinkTrace() << "Sync request processed";
+        SinkTraceCtx(mLogCtx) << "Sync request processed";
         mSyncTransaction.abort();
         mMessageQueue->commit();
         mSyncStore.clear();
@@ -400,7 +398,7 @@ KAsync::Job<void> Synchronizer::processSyncQueue()
             emit changesReplayed();
         }
         if (error) {
-            SinkWarning() << "Error during sync: " << error.errorMessage;
+            SinkWarningCtx(mLogCtx) << "Error during sync: " << error.errorMessage;
             return KAsync::error(error);
         }
         //In case we got more requests meanwhile.
@@ -422,7 +420,7 @@ void Synchronizer::commit()
 Sink::Storage::DataStore::DataStore::Transaction &Synchronizer::syncTransaction()
 {
     if (!mSyncTransaction) {
-        SinkTrace() << "Starting transaction on sync store.";
+        SinkTraceCtx(mLogCtx) << "Starting transaction on sync store.";
         mSyncTransaction = mSyncStorage.createTransaction(Sink::Storage::DataStore::DataStore::ReadWrite);
     }
     return mSyncTransaction;
@@ -447,28 +445,28 @@ bool Synchronizer::canReplay(const QByteArray &type, const QByteArray &key, cons
     const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
     Q_ASSERT(metadataBuffer);
     if (!metadataBuffer->replayToSource()) {
-        SinkTrace() << "Change is coming from the source";
+        SinkTraceCtx(mLogCtx) << "Change is coming from the source";
     }
     return metadataBuffer->replayToSource();
 }
 
 KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray &key, const QByteArray &value)
 {
-    SinkTrace() << "Replaying" << type << key;
+    SinkTraceCtx(mLogCtx) << "Replaying" << type << key;
 
     Sink::EntityBuffer buffer(value);
     const Sink::Entity &entity = buffer.entity();
     const auto metadataBuffer = Sink::EntityBuffer::readBuffer<Sink::Metadata>(entity.metadata());
     if (!metadataBuffer) {
-        SinkError() << "No metadata buffer available.";
+        SinkErrorCtx(mLogCtx) << "No metadata buffer available.";
         return KAsync::error("No metadata buffer");
     }
     if (mSyncTransaction) {
-        SinkError() << "Leftover sync transaction.";
+        SinkErrorCtx(mLogCtx) << "Leftover sync transaction.";
         mSyncTransaction.abort();
     }
     if (mSyncStore) {
-        SinkError() << "Leftover sync store.";
+        SinkErrorCtx(mLogCtx) << "Leftover sync store.";
         mSyncStore.clear();
     }
     Q_ASSERT(metadataBuffer);
@@ -485,7 +483,7 @@ KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray
         oldRemoteId = syncStore().resolveLocalId(type, uid);
         //oldRemoteId can be empty if the resource implementation didn't return a remoteid
     }
-    SinkTrace() << "Replaying " << key << type << uid << oldRemoteId;
+    SinkTraceCtx(mLogCtx) << "Replaying " << key << type << uid << oldRemoteId;
 
     KAsync::Job<QByteArray> job = KAsync::null<QByteArray>();
     //TODO This requires supporting every domain type here as well. Can we solve this better so we can do the dispatch somewhere centrally?
@@ -496,34 +494,34 @@ KAsync::Job<void> Synchronizer::replay(const QByteArray &type, const QByteArray
         auto mail = store().readEntity<ApplicationDomain::Mail>(key);
         job = replay(mail, operation, oldRemoteId, modifiedProperties);
     } else {
-        SinkError() << "Replayed unknown type: " << type;
+        SinkErrorCtx(mLogCtx) << "Replayed unknown type: " << type;
     }
 
     return job.then([this, operation, type, uid, oldRemoteId](const QByteArray &remoteId) {
         if (operation == Sink::Operation_Creation) {
-            SinkTrace() << "Replayed creation with remote id: " << remoteId;
+            SinkTraceCtx(mLogCtx) << "Replayed creation with remote id: " << remoteId;
             if (remoteId.isEmpty()) {
-                SinkWarning() << "Returned an empty remoteId from the creation";
+                SinkWarningCtx(mLogCtx) << "Returned an empty remoteId from the creation";
             } else {
                 syncStore().recordRemoteId(type, uid, remoteId);
             }
         } else if (operation == Sink::Operation_Modification) {
-            SinkTrace() << "Replayed modification with remote id: " << remoteId;
+            SinkTraceCtx(mLogCtx) << "Replayed modification with remote id: " << remoteId;
             if (remoteId.isEmpty()) {
-                SinkWarning() << "Returned an empty remoteId from the modification";
+                SinkWarningCtx(mLogCtx) << "Returned an empty remoteId from the modification";
             } else {
                syncStore().updateRemoteId(type, uid, remoteId);
             }
         } else if (operation == Sink::Operation_Removal) {
-            SinkTrace() << "Replayed removal with remote id: " << oldRemoteId;
+            SinkTraceCtx(mLogCtx) << "Replayed removal with remote id: " << oldRemoteId;
             syncStore().removeRemoteId(type, uid, oldRemoteId);
         } else {
-            SinkError() << "Unkown operation" << operation;
+            SinkErrorCtx(mLogCtx) << "Unkown operation" << operation;
         }
     })
     .then([this](const KAsync::Error &error) {
         if (error) {
-            SinkWarning() << "Failed to replay change: " << error.errorMessage;
+            SinkWarningCtx(mLogCtx) << "Failed to replay change: " << error.errorMessage;
         }
         mSyncStore.clear();
         mSyncTransaction.commit();
@@ -544,7 +542,7 @@ KAsync::Job<QByteArray> Synchronizer::replay(const ApplicationDomain::Folder &,
 bool Synchronizer::allChangesReplayed()
 {
     if (!mSyncRequestQueue.isEmpty()) {
-        SinkTrace() << "Queue is not empty";
+        SinkTraceCtx(mLogCtx) << "Queue is not empty";
         return false;
     }
     return ChangeReplay::allChangesReplayed();