Merge PR #233: Threaded log database

libnymea-core/jsonrpc/logginghandler.cpp

@@ -140,16 +140,27 @@ JsonReply* LoggingHandler::GetLogEntries(const QVariantMap &params) const
 {
     LogFilter filter = unpackLogFilter(params);
 
-    QVariantList entries;
+    LogEntriesFetchJob *job = NymeaCore::instance()->logEngine()->fetchLogEntries(filter);
-    foreach (const LogEntry &entry, NymeaCore::instance()->logEngine()->logEntries(filter)) {
+
-        entries.append(packLogEntry(entry));
+    JsonReply *reply = createAsyncReply("GetLogEntries");
-    }
+
-    QVariantMap returns;
+    connect(job, &LogEntriesFetchJob::finished, reply, [reply, job, filter](){
-    returns.insert("loggingError", enumValueName<Logging::LoggingError>(Logging::LoggingErrorNoError));
+
-    returns.insert("logEntries", entries);
+        QVariantList entries;
-    returns.insert("offset", filter.offset());
+        foreach (const LogEntry &entry, job->results()) {
-    returns.insert("count", entries.count());
+            entries.append(packLogEntry(entry));
-    return createReply(returns);
+        }
+        QVariantMap returns;
+        returns.insert("loggingError", enumValueName<Logging::LoggingError>(Logging::LoggingErrorNoError));
+        returns.insert("logEntries", entries);
+        returns.insert("offset", filter.offset());
+        returns.insert("count", entries.count());
+
+        reply->setData(returns);
+        reply->finished();
+    });
+
+    return reply;
 }
 
 QVariantMap LoggingHandler::packLogEntry(const LogEntry &logEntry)

libnymea-core/logging/logengine.cpp

@@ -128,6 +128,7 @@
 #include <QDateTime>
 #include <QFileInfo>
 #include <QTime>
+#include <QtConcurrent/QtConcurrent>
 
 #define DB_SCHEMA_VERSION 3
 
@@ -138,8 +139,17 @@ namespace nymeaserver {
     \a dbName is the name of the database. In case of SQLITE this should contain a file path. The Driver will create the file if required. In case of using a
     database server like MYSQL, the database must exist on the host given by \a hostname and be accessible with the given \a username and \a password.
 */
+
+// IMPORTANT:
+// DatabaseJobs run threaded, however, QSql is *not* threadsafe.
+// It is crucial to *not* access m_db while the job queue is being processed.
+// That is, entire setup of the DB must happen before processQueue() is called
+// and teardown must happen only after the job queue is empty.
+
 LogEngine::LogEngine(const QString &driver, const QString &dbName, const QString &hostname, const QString &username, const QString &password, int maxDBSize, QObject *parent):
     QObject(parent),
+    m_username(username),
+    m_password(password),
     m_dbMaxSize(maxDBSize)
 {
     m_db = QSqlDatabase::addDatabase(driver, "logs");
@@ -169,9 +179,7 @@ LogEngine::LogEngine(const QString &driver, const QString &dbName, const QString
         }
     }
 
-    connect(&m_housekeepingTimer, &QTimer::timeout, this, &LogEngine::checkDBSize);
+    connect(&m_jobWatcher, SIGNAL(finished()), this, SLOT(handleJobFinished()));
-    m_housekeepingTimer.setInterval(1); // Trigger on next idle event loop run
-    m_housekeepingTimer.setSingleShot(true);
 
     checkDBSize();
 }
@@ -179,17 +187,21 @@ LogEngine::LogEngine(const QString &driver, const QString &dbName, const QString
 /*! Destructs the \l{LogEngine}. */
 LogEngine::~LogEngine()
 {
+    // Process the job queue before allowing to shut down
+    while (m_currentJob) {
+        qCDebug(dcLogEngine()) << "Waiting for job to finish... (" << m_jobQueue.count() << "jobs left in queue)";
+        m_jobWatcher.waitForFinished();
+        // Make sure that the job queue is processes
+        // We can't call processQueue ourselves because thread synchronisation is done via queued connections
+        qApp->processEvents();
+    }
+    qCDebug(dcLogEngine()) << "Closing Database";
     m_db.close();
 }
 
-/*! Returns the list of \l{LogEntry}{LogEntries} of the database matching the given \a filter.
+LogEntriesFetchJob *LogEngine::fetchLogEntries(const LogFilter &filter)
-
-  \sa LogEntry, LogFilter
-*/
-QList<LogEntry> LogEngine::logEntries(const LogFilter &filter) const
 {
     QList<LogEntry> results;
-    QSqlQuery query(m_db);
 
     QString limitString;
     if (filter.limit() >= 0) {
@@ -205,37 +217,66 @@ QList<LogEntry> LogEngine::logEntries(const LogFilter &filter) const
     } else {
         queryString = QString("SELECT * FROM entries WHERE %1 ORDER BY timestamp DESC %2;").arg(filter.queryString()).arg(limitString);
     }
-    qCDebug(dcLogEngine()) << "Preparing query:" << queryString;
-    query.prepare(queryString);
 
-    foreach (const QString &value, filter.values()) {
+    DatabaseJob *job = new DatabaseJob(m_db, queryString, filter.values());
-        query.addBindValue(LogValueTool::serializeValue(value));
+    LogEntriesFetchJob *fetchJob = new LogEntriesFetchJob(this);
-        qCDebug(dcLogEngine()) << "Binding value to query:" << LogValueTool::serializeValue(value);
-    }
 
-    query.exec();
+    connect(job, &DatabaseJob::finished, this, [this, job, fetchJob](){
+        fetchJob->deleteLater();
+        if (job->error().isValid()) {
+            qCWarning(dcLogEngine) << "Error fetching log entries. Driver error:" << job->error().driverText() << "Database error:" << job->error().databaseText();
+            fetchJob->finished();
+            return;
+        }
 
-    if (m_db.lastError().isValid()) {
+        foreach (const QSqlRecord &result, job->results()) {
-        qCWarning(dcLogEngine) << "Error fetching log entries. Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
+            LogEntry entry(
-        return QList<LogEntry>();
+                        QDateTime::fromTime_t(result.value("timestamp").toUInt()),
-    }
+                        static_cast<Logging::LoggingLevel>(result.value("loggingLevel").toInt()),
+                        static_cast<Logging::LoggingSource>(result.value("sourceType").toInt()),
+                        result.value("errorCode").toInt());
+            entry.setTypeId(result.value("typeId").toUuid());
+            entry.setDeviceId(DeviceId(result.value("deviceId").toString()));
+            entry.setValue(LogValueTool::convertVariantToString(LogValueTool::deserializeValue(result.value("value").toString())));
+            entry.setEventType(static_cast<Logging::LoggingEventType>(result.value("loggingEventType").toInt()));
+            entry.setActive(result.value("active").toBool());
 
-    while (query.next()) {
+            fetchJob->m_results.append(entry);
-        LogEntry entry(
+        }
-                    QDateTime::fromTime_t(query.value("timestamp").toLongLong()),
+        qCDebug(dcLogEngine) << "Fetched" << fetchJob->results().count() << "entries for db query:" << job->executedQuery();
-                    (Logging::LoggingLevel)query.value("loggingLevel").toInt(),
+        fetchJob->finished();
-                    (Logging::LoggingSource)query.value("sourceType").toInt(),
+    });
-                    query.value("errorCode").toInt());
-        entry.setTypeId(query.value("typeId").toUuid());
-        entry.setDeviceId(DeviceId(query.value("deviceId").toString()));
-        entry.setValue(LogValueTool::convertVariantToString(LogValueTool::deserializeValue(query.value("value").toString())));
-        entry.setEventType((Logging::LoggingEventType)query.value("loggingEventType").toInt());
-        entry.setActive(query.value("active").toBool());
-        results.append(entry);
-    }
-//    qCDebug(dcLogEngine) << "Fetched" << results.count() << "entries for db query:" << query.executedQuery();
 
-    return results;
+    enqueJob(job);
+
+    return fetchJob;
+}
+
+DevicesFetchJob *LogEngine::fetchDevices()
+{
+    QString queryString = QString("SELECT deviceId FROM entries WHERE deviceId != \"%1\" GROUP BY deviceId;").arg(QUuid().toString());
+
+    DatabaseJob *job = new DatabaseJob(m_db, queryString);
+    DevicesFetchJob *fetchJob = new DevicesFetchJob(this);
+    connect(job, &DatabaseJob::finished, this, [this, job, fetchJob](){
+        fetchJob->deleteLater();
+        if (job->error().type() != QSqlError::NoError) {
+            qCWarning(dcLogEngine()) << "Error fetching device entries from log database:" << job->error().driverText() << job->error().databaseText();
+            fetchJob->finished();
+            return;
+        }
+
+        foreach (const QSqlRecord &result, job->results()) {
+            fetchJob->m_results.append(DeviceId::fromUuid(result.value("deviceId").toUuid()));
+        }
+        fetchJob->finished();
+    });
+    return fetchJob;
+}
+
+bool LogEngine::jobsRunning() const
+{
+    return !m_jobQueue.isEmpty() || m_currentJob;
 }
 
 void LogEngine::setMaxLogEntries(int maxLogEntries, int overflow)
@@ -251,11 +292,17 @@ void LogEngine::clearDatabase()
     qCWarning(dcLogEngine) << "Clear logging database.";
 
     QString queryDeleteString = QString("DELETE FROM entries;");
-    if (m_db.exec(queryDeleteString).lastError().type() != QSqlError::NoError) {
-        qCWarning(dcLogEngine) << "Could not clear logging database. Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
-    }
 
-    emit logDatabaseUpdated();
+    DatabaseJob *job = new DatabaseJob(m_db, queryDeleteString);
+
+    connect(job, &DatabaseJob::finished, this, [this, job](){
+        if (job->error().type() != QSqlError::NoError) {
+            qCWarning(dcLogEngine) << "Could not clear logging database. Driver error:" << job->error().driverText() << "Database error:" << job->error().databaseText();
+        }
+        emit logDatabaseUpdated();
+    });
+
+    enqueJob(job);
 }
 
 void LogEngine::logSystemEvent(const QDateTime &dateTime, bool active, Logging::LoggingLevel level)
@@ -378,11 +425,17 @@ void LogEngine::removeDeviceLogs(const DeviceId &deviceId)
     qCDebug(dcLogEngine) << "Deleting log entries from device" << deviceId.toString();
 
     QString queryDeleteString = QString("DELETE FROM entries WHERE deviceId = '%1';").arg(deviceId.toString());
-    if (m_db.exec(queryDeleteString).lastError().type() != QSqlError::NoError) {
+
-        qCWarning(dcLogEngine) << "Error deleting log entries from device" << deviceId.toString() << ". Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
+    DatabaseJob *job = new DatabaseJob(m_db, queryDeleteString);
-    } else {
+    connect(job, &DatabaseJob::finished, this, [this, job, deviceId](){
-        emit logDatabaseUpdated();
+        if (job->error().type() != QSqlError::NoError) {
-    }
+            qCWarning(dcLogEngine) << "Error deleting log entries from device" << deviceId.toString() << ". Driver error:" << job->error().driverText() << "Database error:" << job->error().databaseText();
+        } else {
+            emit logDatabaseUpdated();
+        }
+    });
+
+    enqueJob(job);
 }
 
 void LogEngine::removeRuleLogs(const RuleId &ruleId)
@@ -390,29 +443,19 @@ void LogEngine::removeRuleLogs(const RuleId &ruleId)
     qCDebug(dcLogEngine) << "Deleting log entries from rule" << ruleId.toString();
 
     QString queryDeleteString = QString("DELETE FROM entries WHERE typeId = '%1';").arg(ruleId.toString());
-    if (m_db.exec(queryDeleteString).lastError().type() != QSqlError::NoError) {
-        qCWarning(dcLogEngine) << "Error deleting log entries from rule" << ruleId.toString() << ". Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
-    } else {
-        emit logDatabaseUpdated();
-    }
-}
 
-QList<DeviceId> LogEngine::devicesInLogs() const
+    DatabaseJob *job = new DatabaseJob(m_db, queryDeleteString);
-{
+
-    QString queryString = QString("SELECT deviceId FROM entries WHERE deviceId != \"%1\" GROUP BY deviceId;").arg(QUuid().toString());
+    connect(job, &DatabaseJob::finished, this, [this, job, ruleId](){
-    QSqlQuery result = m_db.exec(queryString);
+
-    QList<DeviceId> ret;
+        if (job->error().type() != QSqlError::NoError) {
-    if (result.lastError().type() != QSqlError::NoError) {
+            qCWarning(dcLogEngine) << "Error deleting log entries from rule" << ruleId.toString() << ". Driver error:" << job->error().driverText() << "Database error:" << job->error().databaseText();
-        qCWarning(dcLogEngine()) << "Error fetching device entries from log database:" << m_db.lastError().driverText() << m_db.lastError().databaseText();
+        } else {
-        return ret;
+            emit logDatabaseUpdated();
-    }
+        }
-    if (!result.first()) {
+    });
-        return ret;
+
-    }
+    enqueJob(job);
-    do {
-        ret.append(DeviceId::fromUuid(result.value("deviceId").toUuid()));
-    } while (result.next());
-    return ret;
 }
 
 void LogEngine::appendLogEntry(const LogEntry &entry)
@@ -428,17 +471,25 @@ void LogEngine::appendLogEntry(const LogEntry &entry)
             .arg(entry.active())
             .arg(entry.errorCode());
 
-    if (m_db.exec(queryString).lastError().type() != QSqlError::NoError) {
+    DatabaseJob *job = new DatabaseJob(m_db, queryString);
-        qCWarning(dcLogEngine) << "Error writing log entry. Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
-        qCWarning(dcLogEngine) << entry;
-        return;
-    }
 
-    emit logEntryAdded(entry);
+    connect(job, &DatabaseJob::finished, this, [this, job, entry](){
 
-    if (++m_entryCount > m_dbMaxSize + m_overflow) {
+        if (job->error().type() != QSqlError::NoError) {
-        m_housekeepingTimer.start();
+            qCWarning(dcLogEngine) << "Error writing log entry. Driver error:" << job->error().driverText() << "Database error:" << job->error().number() << job->error().databaseText();
-    }
+            qCWarning(dcLogEngine) << entry;
+            m_dbMalformed = true;
+            return;
+        }
+
+        emit logEntryAdded(entry);
+
+        if (++m_entryCount > m_dbMaxSize + m_overflow) {
+            checkDBSize();
+        }
+    });
+
+    enqueJob(job);
 }
 
 void LogEngine::checkDBSize()
@@ -449,31 +500,110 @@ void LogEngine::checkDBSize()
     }
     QDateTime startTime = QDateTime::currentDateTime();
     QString queryString = "SELECT COUNT(*) FROM entries;";
-    QSqlQuery result = m_db.exec(queryString);
-    if (m_db.lastError().type() != QSqlError::NoError) {
-        qWarning(dcLogEngine()) << "Failed to query entry count in db:" << m_db.lastError().databaseText();
-        return;
-    }
-    if (!result.first()) {
-        qWarning(dcLogEngine()) << "Failed retrieving entry count.";
-        return;
-    }
-    m_entryCount = result.value(0).toInt();
 
-    if (m_entryCount >= m_dbMaxSize) {
+    DatabaseJob *job = new DatabaseJob(m_db, queryString);
+    connect(job, &DatabaseJob::finished, this, [this, job, startTime](){
+
+        if (job->error().type() != QSqlError::NoError) {
+            qCWarning(dcLogEngine()) << "Failed to query entry count in db:" << job->error().databaseText();
+            return;
+        }
+        if (job->results().isEmpty()) {
+            qCWarning(dcLogEngine()) << "Failed retrieving entry count.";
+            return;
+        }
+        m_entryCount = job->results().first().value(0).toInt();
+
+        if (m_entryCount <= m_dbMaxSize) {
+            return;
+        }
+
         // keep only the latest m_dbMaxSize entries
         if (!m_trimWarningPrinted) {
             qCDebug(dcLogEngine) << "Deleting oldest entries" << (m_entryCount - m_dbMaxSize) << "and keep only the latest" << m_dbMaxSize << "entries.";
             m_trimWarningPrinted = true;
         }
         QString queryDeleteString = QString("DELETE FROM entries WHERE ROWID IN (SELECT ROWID FROM entries ORDER BY timestamp DESC LIMIT -1 OFFSET %1);").arg(QString::number(m_dbMaxSize));
-        if (m_db.exec(queryDeleteString).lastError().type() != QSqlError::NoError) {
+
-            qCWarning(dcLogEngine) << "Error deleting oldest log entries to keep size. Driver error:" << m_db.lastError().driverText() << "Database error:" << m_db.lastError().databaseText();
+        DatabaseJob *deleteJob = new DatabaseJob(m_db, queryDeleteString);
-        }
+
-        m_entryCount = m_dbMaxSize;
+        connect(deleteJob, &DatabaseJob::finished, this, [this, deleteJob,startTime](){
-        emit logDatabaseUpdated();
+            if (deleteJob->error().type() != QSqlError::NoError) {
+                qCWarning(dcLogEngine) << "Error deleting oldest log entries to keep size. Driver error:" << deleteJob->error().driverText() << "Database error:" << deleteJob->error().databaseText();
+            }
+            m_entryCount = m_dbMaxSize;
+            qCDebug(dcLogEngine()) << "Ran housekeeping on log database in" << startTime.msecsTo(QDateTime::currentDateTime()) << "ms.";
+
+            emit logDatabaseUpdated();
+        });
+        enqueJob(deleteJob);
+    });
+    enqueJob(job);
+
+}
+
+void LogEngine::enqueJob(DatabaseJob *job)
+{
+    m_jobQueue.append(job);
+    processQueue();
+}
+
+void LogEngine::processQueue()
+{
+    if (m_jobQueue.isEmpty()) {
+        emit jobsRunningChanged();
+        return;
     }
-    qCDebug(dcLogEngine()) << "Ran housekeeping on log database in" << startTime.msecsTo(QDateTime::currentDateTime()) << "ms.";
+
+    if (m_currentJob) {
+        return;
+    }
+
+    emit jobsRunningChanged();
+
+    if (m_dbMalformed) {
+        qCWarning(dcLogEngine()) << "Database is malformed. Trying to recover...";
+        m_db.close();
+        rotate(m_db.databaseName());
+        initDB(m_username, m_password);
+        m_dbMalformed = false;
+    }
+
+    DatabaseJob *job = m_jobQueue.takeFirst();
+    m_currentJob = job;
+
+    QFuture<DatabaseJob*> future = QtConcurrent::run([job](){
+        QSqlQuery query(job->m_db);
+        query.prepare(job->m_queryString);
+
+        foreach (const QString &value, job->m_bindValues) {
+            query.addBindValue(LogValueTool::serializeValue(value));
+        }
+
+        query.exec();
+
+        job->m_error = query.lastError();
+        job->m_executedQuery = query.executedQuery();
+
+        if (!query.lastError().isValid()) {
+            while (query.next()) {
+                job->m_results.append(query.record());
+            }
+        }
+
+       return job;
+    });
+
+    m_jobWatcher.setFuture(future);
+}
+
+void LogEngine::handleJobFinished()
+{
+    DatabaseJob *job = m_jobWatcher.result();
+    job->finished();
+    job->deleteLater();
+    m_currentJob = nullptr;
+    processQueue();
 }
 
 void LogEngine::rotate(const QString &dbName)
@@ -639,6 +769,7 @@ bool LogEngine::initDB(const QString &username, const QString &password)
     }
 
     if (!m_db.tables().contains("entries")) {
+        qCDebug(dcLogEngine()) << "No \"entries\" table in database. Creating it.";
         m_db.exec("CREATE TABLE entries "
                   "("
                   "timestamp int,"

libnymea-core/logging/logengine.h

@@ -32,18 +32,29 @@
 
 #include <QObject>
 #include <QSqlDatabase>
+#include <QSqlQuery>
+#include <QSqlError>
+#include <QSqlRecord>
 #include <QTimer>
+#include <QFutureWatcher>
 
 namespace nymeaserver {
 
+class DatabaseJob;
+class LogEntriesFetchJob;
+class DevicesFetchJob;
+
 class LogEngine: public QObject
 {
     Q_OBJECT
 public:
-    LogEngine(const QString &driver, const QString &dbName, const QString &hostname = QString("127.0.0.1"), const QString &username = QString(), const QString &password = QString(), int maxDBSize = 50000, QObject *parent = 0);
+    LogEngine(const QString &driver, const QString &dbName, const QString &hostname = QString("127.0.0.1"), const QString &username = QString(), const QString &password = QString(), int maxDBSize = 50000, QObject *parent = nullptr);
     ~LogEngine();
 
-    QList<LogEntry> logEntries(const LogFilter &filter = LogFilter()) const;
+    LogEntriesFetchJob *fetchLogEntries(const LogFilter &filter = LogFilter());
+    DevicesFetchJob *fetchDevices();
+
+    bool jobsRunning() const;
 
     void setMaxLogEntries(int maxLogEntries, int overflow);
     void clearDatabase();
@@ -60,12 +71,13 @@ public:
     void logRuleExitActionsExecuted(const Rule &rule);
     void removeDeviceLogs(const DeviceId &deviceId);
     void removeRuleLogs(const RuleId &ruleId);
-    QList<DeviceId> devicesInLogs() const;
 
 signals:
     void logEntryAdded(const LogEntry &logEntry);
     void logDatabaseUpdated();
 
+    void jobsRunningChanged();
+
 private:
     bool initDB(const QString &username, const QString &password);
     void appendLogEntry(const LogEntry &entry);
@@ -77,13 +89,79 @@ private:
 private slots:
     void checkDBSize();
 
+    void enqueJob(DatabaseJob *job);
+    void processQueue();
+    void handleJobFinished();
+
 private:
     QSqlDatabase m_db;
+    QString m_username;
+    QString m_password;
     int m_dbMaxSize;
     int m_overflow;
     bool m_trimWarningPrinted = false;
     int m_entryCount = 0;
-    QTimer m_housekeepingTimer;
+    bool m_dbMalformed = false;
+
+    QList<DatabaseJob*> m_jobQueue;
+    DatabaseJob *m_currentJob = nullptr;
+    QFutureWatcher<DatabaseJob*> m_jobWatcher;
+};
+
+class DatabaseJob: public QObject
+{
+    Q_OBJECT
+public:
+    DatabaseJob(const QSqlDatabase &db, const QString &queryString, const QStringList &bindValues = QStringList()):
+        m_db(db),
+        m_queryString(queryString),
+        m_bindValues(bindValues)
+    {
+    }
+
+    QString executedQuery() const { return m_executedQuery; }
+    QSqlError error() const { return m_error; }
+    QList<QSqlRecord> results() const { return m_results; }
+
+signals:
+    void finished();
+
+private:
+    QSqlDatabase m_db;
+    QString m_queryString;
+    QStringList m_bindValues;
+
+    QString m_executedQuery;
+    QSqlError m_error;
+    QList<QSqlRecord> m_results;
+
+    friend class LogEngine;
+};
+
+class LogEntriesFetchJob: public QObject
+{
+    Q_OBJECT
+public:
+    LogEntriesFetchJob(QObject *parent): QObject(parent) {}
+    QList<LogEntry> results() { return m_results; }
+signals:
+    void finished();
+private:
+    QList<LogEntry> m_results;
+    friend class LogEngine;
+};
+
+class DevicesFetchJob: public QObject
+{
+    Q_OBJECT
+public:
+    DevicesFetchJob(QObject *parent): QObject(parent) {}
+    QList<DeviceId> results() { return m_results; }
+signals:
+    void finished();
+private:
+    QList<DeviceId> m_results;
+    friend class LogEngine;
 };
 
 }

libnymea-core/nymeacore.cpp

@@ -924,12 +924,16 @@ void NymeaCore::deviceManagerLoaded()
     // Do some houskeeping...
     qCDebug(dcApplication()) << "Starting housekeeping...";
     QDateTime startTime = QDateTime::currentDateTime();
-    foreach (const DeviceId &deviceId, m_logger->devicesInLogs()) {
+    DevicesFetchJob *job = m_logger->fetchDevices();
-        if (!m_deviceManager->findConfiguredDevice(deviceId)) {
+    connect(job, &DevicesFetchJob::finished, this, [this, job, startTime](){
-            qCDebug(dcApplication()) << "Cleaning stale device entries from log DB for device id" << deviceId;
+        foreach (const DeviceId &deviceId, job->results()) {
-            m_logger->removeDeviceLogs(deviceId);
+            if (!m_deviceManager->findConfiguredDevice(deviceId)) {
+                qCDebug(dcApplication()) << "Cleaning stale device entries from log DB for device id" << deviceId;
+                m_logger->removeDeviceLogs(deviceId);
+            }
         }
-    }
+        qCDebug(dcApplication()) << "Housekeeping done in" << startTime.msecsTo(QDateTime::currentDateTime()) << "ms.";
+    });
 
     foreach (const DeviceId &deviceId, m_ruleEngine->devicesInRules()) {
         if (!m_deviceManager->findConfiguredDevice(deviceId)) {
@@ -939,9 +943,6 @@ void NymeaCore::deviceManagerLoaded()
             }
         }
     }
-
-    qCDebug(dcApplication()) << "Housekeeping done in" << startTime.msecsTo(QDateTime::currentDateTime()) << "ms.";
-
 }
 
 }

nymea.pro

@@ -58,8 +58,9 @@ INSTALLS += translations
 
 QMAKE_EXTRA_TARGETS += lupdate lrelease
 
-test.depends = licensecheck lrelease
+test.depends = licensecheck
-test.commands = LD_LIBRARY_PATH=$$top_builddir/libnymea-core:$$top_builddir/libnymea:$$top_builddir/tests/testlib make check
+test.depends += lrelease
+test.commands = LD_LIBRARY_PATH=$$top_builddir/libnymea-core:$$top_builddir/libnymea:$$top_builddir/tests/testlib make check TESTRUNNER=\"dbus-test-runner --bus-type=system --task\"
 QMAKE_EXTRA_TARGETS += test
 
 # Show doc files in project tree

tests/auto/autotests.pri

@@ -13,3 +13,6 @@ LIBS += -L$$top_builddir/libnymea/ -lnymea \
 
 target.path = /usr/tests
 INSTALLS += target
+
+test.commands = LD_LIBRARY_PATH=../../../libnymea:../../../libnymea-core/:../../testlib/ make check TESTRUNNER=\"dbus-test-runner --bus-type=system --task\"
+QMAKE_EXTRA_TARGETS += test

tests/auto/jsonrpc/testjsonrpc.cpp

@@ -41,6 +41,7 @@ private:
 
 private slots:
     void initTestCase();
+    void cleanup();
 
     void testHandshake();
 
@@ -148,6 +149,13 @@ void TestJSONRPC::initTestCase()
                                      "Tests.debug=true");
 }
 
+void TestJSONRPC::cleanup()
+{
+    while (NymeaCore::instance()->logEngine()->jobsRunning()) {
+        qApp->processEvents();
+    }
+}
+
 void TestJSONRPC::testHandshake()
 {
     QUuid newClientId = QUuid::createUuid();
@@ -846,16 +854,56 @@ void TestJSONRPC::ruleActiveChangedNotifications()
     QCOMPARE(notificationVariant.toMap().value("params").toMap().value("ruleId").toUuid().toString(), ruleId.toString());
     QCOMPARE(notificationVariant.toMap().value("params").toMap().value("active").toBool(), true);
 
+    clientSpy.wait();
+
+    // Make sure the logg notification contains all the stuff we expect
+    QVariantList logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    QVERIFY2(!logEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
+    bool found = false;
+    foreach (const QVariant &loggEntryAddedVariant, logEntryAddedVariants) {
+        if (loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("typeId").toUuid() == mockIntStateTypeId) {
+            found = true;
+            QCOMPARE(loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("source").toString(), QString("LoggingSourceStates"));
+            QCOMPARE(loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("value").toInt(), 20);
+            break;
+        }
+    }
+    QVERIFY2(found, "LogEntryAdded notification not received");
+
     spy.clear(); clientSpy.clear();
 
     // set the rule inactive
     qDebug() << "setting mock int state to 42";
     QNetworkRequest request2(QUrl(QString("http://localhost:%1/setstate?%2=%3").arg(m_mockDevice1Port).arg(mockIntStateTypeId.toString()).arg(42)));
     QNetworkReply *reply2 = nam.get(request2);
-    if (spy.count() == 0) spy.wait();
-    QCOMPARE(spy.count(), 1);
     connect(reply2, SIGNAL(finished()), reply2, SLOT(deleteLater()));
 
+    // Waiting for notifications:
+    // Devices.StateChanged for the change we did
+    // Devices.EventTriggered
+    // Events.EventTriggered <-- deprecated
+    // Rules.RuleActiveChanged
+    // Logging.LogEntryAdded
+    // Devices.StateChanged for the change done by the rule
+    // Devices.EventTriggered
+    // Events.EventTriggered <-- deprecated
+    while (clientSpy.count() < 8) {
+        clientSpy.wait();
+    }
+
+    // Make sure the logg notification contains all the stuff we expect
+    logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    QVERIFY2(!logEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
+    found = false;
+    foreach (const QVariant &loggEntryAddedVariant, logEntryAddedVariants) {
+        if (loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("typeId").toUuid() == mockIntStateTypeId) {
+            found = true;
+            QCOMPARE(loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("source").toString(), QString("LoggingSourceStates"));
+            QCOMPARE(loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("value").toInt(), 42);
+            break;
+        }
+    }
+    QVERIFY2(found, "LogEntryAdded notification not received");
 
     if (clientSpy.count() == 0) clientSpy.wait();
     notificationVariant = checkNotification(clientSpy, "Rules.RuleActiveChanged");
@@ -991,18 +1039,23 @@ void TestJSONRPC::stateChangeEmitsNotifications()
             break;
         }
     }
-    if (!found)
-        qDebug() << QJsonDocument::fromVariant(stateChangedVariants).toJson();
 
     QVERIFY2(found, "Could not find the correct Devices.StateChanged notification");
 
+    // Waiting for:
+    // Devices.StateChanged
+    // Devices.EventTriggered
+    // Events.EventTriggered <-- deprecated
+    // Logging.LogEntryAdded
+    while (clientSpy.count() < 4) {
+        clientSpy.wait();
+    }
 
     // Make sure the logg notification contains all the stuff we expect
-    QVariantList loggEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    QVariantList logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
-    QVERIFY2(!loggEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
+    QVERIFY2(!logEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
-    qDebug() << "got" << loggEntryAddedVariants.count() << "Logging.LogEntryAdded notifications";
     found = false;
-    foreach (const QVariant &loggEntryAddedVariant, loggEntryAddedVariants) {
+    foreach (const QVariant &loggEntryAddedVariant, logEntryAddedVariants) {
         if (loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("typeId").toUuid() == stateTypeId) {
             found = true;
             QCOMPARE(loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap().value("source").toString(), QString("LoggingSourceStates"));
@@ -1010,17 +1063,13 @@ void TestJSONRPC::stateChangeEmitsNotifications()
             break;
         }
     }
-    if (!found)
-        qDebug() << QJsonDocument::fromVariant(loggEntryAddedVariants).toJson();
 
     QVERIFY2(found, "Could not find the corresponding Logging.LogEntryAdded notification");
 
-
     // Make sure the notification contains all the stuff we expect
     QVariantList eventTriggeredVariants = checkNotifications(clientSpy, "Events.EventTriggered");
     QVERIFY2(!eventTriggeredVariants.isEmpty(), "Did not get Events.EventTriggered notification.");
     found = false;
-    qDebug() << "got" << eventTriggeredVariants.count() << "Events.EventTriggered notifications";
     foreach (const QVariant &eventTriggeredVariant, eventTriggeredVariants) {
         if (eventTriggeredVariant.toMap().value("params").toMap().value("event").toMap().value("eventTypeId").toUuid() == stateTypeId) {
             found = true;
@@ -1028,8 +1077,6 @@ void TestJSONRPC::stateChangeEmitsNotifications()
             break;
         }
     }
-    if (!found)
-        qDebug() << QJsonDocument::fromVariant(eventTriggeredVariants).toJson();
 
     QVERIFY2(found, "Could not find the corresponding Events.EventTriggered notification");
 
@@ -1108,6 +1155,7 @@ void TestJSONRPC::testPushButtonAuth()
 
 void TestJSONRPC::testPushButtonAuthInterrupt()
 {
+    enableNotifications({});
     PushButtonAgent pushButtonAgent;
     pushButtonAgent.init();
 

tests/auto/logging/testlogging.cpp

@@ -77,6 +77,7 @@ void TestLogging::initTestCase()
 {
     NymeaTestBase::initTestCase();
     QLoggingCategory::setFilterRules("*.debug=false\n"
+                                     "LogEngine.debug=true\n"
                                      "Tests.debug=true\n"
                                      "MockDevice.debug=true\n"
                                      "DeviceManager.debug=true\n");
@@ -88,8 +89,9 @@ void TestLogging::initLogs()
     verifyLoggingError(response);
 
     QVariantList logEntries = response.toMap().value("params").toMap().value("logEntries").toList();
-    qDebug() << "Got" << logEntries.count() << "logs";
+    QVERIFY2(logEntries.count() > 0,
-    QVERIFY(logEntries.count() > 0);
+             QString("Expected at least one log entry.")
+             .toUtf8());
 
     clearLoggingDatabase();
 
@@ -153,16 +155,34 @@ void TestLogging::coverageCalls()
 
 void TestLogging::systemLogs()
 {
-    // check the active system log at boot
+    qWarning() << "Clearing logging DB";
+    clearLoggingDatabase();
+
     QVariantMap params;
     params.insert("loggingSources", QVariantList() << enumValueName(Logging::LoggingSourceSystem));
     params.insert("eventTypes", QVariantList() << enumValueName(Logging::LoggingEventTypeActiveChange));
 
-    // there should be 2 logs, one for shutdown, one for startup (from server restart)
+    // there should be 0 log entries
     QVariant response = injectAndWait("Logging.GetLogEntries", params);
-    verifyLoggingError(response);
     QVariantList logEntries = response.toMap().value("params").toMap().value("logEntries").toList();
-    QVERIFY(logEntries.count() == 2);
+    QVERIFY2(logEntries.count() == 0,
+             QString("Expected 0 log entries but got:\n%1")
+             .arg(QString(QJsonDocument::fromVariant(logEntries).toJson()))
+             .toUtf8());
+
+    // check the active system log at boot
+    qWarning() << "Restarting server";
+    restartServer();
+    qWarning() << "Restart done";
+
+    // there should be 2 log entries, one for shutdown, one for startup (from server restart)
+    response = injectAndWait("Logging.GetLogEntries", params);
+    verifyLoggingError(response);
+    logEntries = response.toMap().value("params").toMap().value("logEntries").toList();
+    QVERIFY2(logEntries.count() == 2,
+             QString("Expected 2 log entries but got:\n%1")
+             .arg(QString(QJsonDocument::fromVariant(logEntries).toJson()))
+             .toUtf8());
 
     QVariantMap logEntryStartup = logEntries.first().toMap();
     QVariantMap logEntryShutdown = logEntries.last().toMap();
@@ -309,16 +329,19 @@ void TestLogging::actionLog()
     QVariant response = injectAndWait("Actions.ExecuteAction", params);
     verifyDeviceError(response);
 
-    // Lets wait for the notification
+    // wait for the outgoing data
+    // 3 packets: ExecuteAction reply,  LogDatabaseUpdated signal and LogEntryAdded signal
     clientSpy.wait(500);
+    if (clientSpy.count() < 3) {
+        clientSpy.wait(500);
+    }
 
-    QVariantList loggEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    QVariantList logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
-    QVERIFY2(!loggEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
+    qWarning() << QJsonDocument::fromVariant(logEntryAddedVariants).toJson();
-    qDebug() << "got" << loggEntryAddedVariants.count() << "Logging.LogEntryAdded notifications";
+    QVERIFY2(!logEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
 
     bool found = false;
-    qDebug() << "got" << loggEntryAddedVariants.count() << "Logging.LogEntryAdded";
+    foreach (const QVariant &loggEntryAddedVariant, logEntryAddedVariants) {
-    foreach (const QVariant &loggEntryAddedVariant, loggEntryAddedVariants) {
         QVariantMap logEntry = loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap();
         if (logEntry.value("deviceId").toUuid() == m_mockDeviceId) {
             found = true;
@@ -330,8 +353,6 @@ void TestLogging::actionLog()
             break;
         }
     }
-    if (!found)
-        qDebug() << QJsonDocument::fromVariant(loggEntryAddedVariants).toJson();
 
     QVERIFY2(found, "Could not find the corresponding Logging.LogEntryAdded notification");
 
@@ -344,8 +365,8 @@ void TestLogging::actionLog()
 
     clientSpy.wait(200);
 
-    loggEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
-    QVERIFY(!loggEntryAddedVariants.isEmpty());
+    QVERIFY(!logEntryAddedVariants.isEmpty());
 
     // get this logentry with filter
     params.clear();
@@ -371,13 +392,11 @@ void TestLogging::actionLog()
 
     clientSpy.wait(200);
 
-    loggEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
+    logEntryAddedVariants = checkNotifications(clientSpy, "Logging.LogEntryAdded");
-    QVERIFY2(!loggEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
+    QVERIFY2(!logEntryAddedVariants.isEmpty(), "Did not get Logging.LogEntryAdded notification.");
-    qDebug() << "got" << loggEntryAddedVariants.count() << "Logging.LogEntryAdded notifications";
 
     found = false;
-    qDebug() << "got" << loggEntryAddedVariants.count() << "Logging.LogEntryAdded";
+    foreach (const QVariant &loggEntryAddedVariant, logEntryAddedVariants) {
-    foreach (const QVariant &loggEntryAddedVariant, loggEntryAddedVariants) {
         QVariantMap logEntry = loggEntryAddedVariant.toMap().value("params").toMap().value("logEntry").toMap();
         if (logEntry.value("deviceId").toUuid() == m_mockDeviceId) {
             found = true;
@@ -390,8 +409,6 @@ void TestLogging::actionLog()
             break;
         }
     }
-    if (!found)
-        qDebug() << QJsonDocument::fromVariant(loggEntryAddedVariants).toJson();
 
     QVERIFY2(found, "Could not find the corresponding Logging.LogEntryAdded notification");
 

tests/auto/loggingdirect/testloggingdirect.cpp

@@ -81,25 +81,43 @@ void TestLoggingDirect::benchmarkDB()
     qDebug() << "Flushing DB for test";
     engine->setMaxLogEntries(prefill, overflow);
     engine->setMaxLogEntries(maxSize, overflow);
-    qDebug() << "DB has" << engine->logEntries().count() << "entries";
+
+    LogEntriesFetchJob *job = engine->fetchLogEntries();
+    QSignalSpy fetchSpy(job, &LogEntriesFetchJob::finished);
+    fetchSpy.wait();
+    QList<LogEntry> entries = job->results();
+
+    qDebug() << "DB has" << entries.count() << "entries";
     qDebug() << "Prefilling DB for test";
-    for (int i = engine->logEntries().count(); i < prefill; i++) {
+    for (int i = entries.count(); i < prefill; i++) {
         engine->logSystemEvent(QDateTime::currentDateTime(), true);
     }
-    qDebug() << "DB has" << engine->logEntries().count() << "entries";
 
-    qDebug() << "Starting benchmark with" << engine->logEntries().count() << "entries in the db";
+    job = engine->fetchLogEntries();
+    QSignalSpy fetchSpy2(job, &LogEntriesFetchJob::finished);
+    fetchSpy2.wait();
+    entries = job->results();
+
+    qDebug() << "DB has" << entries.count() << "entries";
+
+    qDebug() << "Starting benchmark with" << entries.count() << "entries in the db";
     QBENCHMARK {
         engine->logSystemEvent(QDateTime::currentDateTime(), true);
     }
     QDateTime now = QDateTime::currentDateTime();
-    while (engine->logEntries().count() > maxSize + overflow) {
+
+    job = engine->fetchLogEntries();
+    QSignalSpy fetchSpy3(job, &LogEntriesFetchJob::finished);
+    fetchSpy3.wait();
+    entries = job->results();
+
+    while (entries.count() > maxSize + overflow) {
         qApp->processEvents();
         if (now.addSecs(5) < QDateTime::currentDateTime()) {
-            QVERIFY2(false, QString("Housekeeping didn't work. Have %1 entries but expected to have max %2").arg(engine->logEntries().count()).arg(QString::number(maxSize)).toLocal8Bit());
+            QVERIFY2(false, QString("Housekeeping didn't work. Have %1 entries but expected to have max %2").arg(entries.count()).arg(QString::number(maxSize)).toLocal8Bit());
         }
     }
-    qDebug() << "Ended benchmark with" << engine->logEntries().count() << "entries in the db";
+    qDebug() << "Ended benchmark with" << entries.count() << "entries in the db";
 }
 
 #include "testloggingdirect.moc"

tests/auto/rules/testrules.cpp

@@ -2312,7 +2312,11 @@ void TestRules::testStateBasedAction()
     LogFilter filter;
     filter.addDeviceId(m_mockDeviceId);
     filter.addTypeId(mockWithParamsActionTypeId);
-    QList<LogEntry> entries = NymeaCore::instance()->logEngine()->logEntries(filter);
+
+    LogEntriesFetchJob *job = NymeaCore::instance()->logEngine()->fetchLogEntries(filter);
+    QSignalSpy fetchSpy(job, &LogEntriesFetchJob::finished);
+    fetchSpy.wait();
+    QList<LogEntry> entries = job->results();
     qCDebug(dcTests()) << "Log entries:" << entries;
 
     // set bool state to false
@@ -2331,10 +2335,12 @@ void TestRules::testStateBasedAction()
     QCOMPARE(spy.count(), 1);
     reply->deleteLater();
 
-    entries = NymeaCore::instance()->logEngine()->logEntries(filter);
+    job = NymeaCore::instance()->logEngine()->fetchLogEntries(filter);
+    QSignalSpy fetchSpy2(job, &LogEntriesFetchJob::finished);
+    fetchSpy2.wait();
+    entries = job->results();
+
     qCDebug(dcTests()) << "Log entries:" << entries;
-
-
 }
 
 void TestRules::removePolicyUpdate()

tests/scripts/gettags.sh

@@ -2,6 +2,10 @@
 
 if [ -z $1 ]; then
   echo "usage: $0 host"
-else
+  exit 1
-  (echo '{"id":1, "method":"Tags.GetTags"}'; sleep 1) | nc $1 2222
 fi
+
+cat << EOD | nc $1 2222
+{"id":0, "method":"JSONRPC.Hello"}
+{"id":1, "method":"Tags.GetTags"}
+EOD

tests/testlib/nymeatestbase.cpp

@@ -190,8 +190,9 @@ QVariantList NymeaTestBase::checkNotifications(const QSignalSpy &spy, const QStr
         // Make sure the response it a valid JSON string
         QJsonParseError error;
         QJsonDocument jsonDoc = QJsonDocument::fromJson(spy.at(i).last().toByteArray(), &error);
+//        qCDebug(dcTests()) << "Got packet:" << qUtf8Printable(jsonDoc.toJson());
         if (error.error != QJsonParseError::NoError) {
-            qWarning() << "JSON parser error" << error.errorString();
+            qCWarning(dcTests()) << "JSON parser error" << error.errorString();
             return notificationList;
         }