Index: lib/Comms/CMakeLists.txt =================================================================== diff -u -rcaca75be9a284ac5f98c078ec47c2e826ac5e980 -refc0e8ccb0ee99f25834eb5a19dee1cf2e50c532 --- lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision caca75be9a284ac5f98c078ec47c2e826ac5e980) +++ lib/Comms/CMakeLists.txt (.../CMakeLists.txt) (revision efc0e8ccb0ee99f25834eb5a19dee1cf2e50c532) @@ -7,6 +7,7 @@ find_package(QT NAMES Qt6 Qt5 REQUIRED COMPONENTS Core) find_package(Qt${QT_VERSION_MAJOR} REQUIRED COMPONENTS Core Network SerialBus) +find_package(SQLite3 REQUIRED) set(INCLUDES include/CanInterface.h @@ -20,6 +21,7 @@ include/main.h include/MessageBuilder.h include/MessageDispatcher.h + include/MessageSpool.h include/types.h ) @@ -33,6 +35,7 @@ src/FrameInterface.cpp src/MessageBuilder.cpp src/MessageDispatcher.cpp + src/MessageSpool.cpp src/types.cpp ) @@ -55,6 +58,7 @@ Qt${QT_VERSION_MAJOR}::Core Qt${QT_VERSION_MAJOR}::Network Qt${QT_VERSION_MAJOR}::SerialBus + SQLite::SQLite3 ) target_include_directories(${PROJECT_NAME} PUBLIC $) Index: lib/Comms/Comms.pro =================================================================== diff -u -rcaca75be9a284ac5f98c078ec47c2e826ac5e980 -refc0e8ccb0ee99f25834eb5a19dee1cf2e50c532 --- lib/Comms/Comms.pro (.../Comms.pro) (revision caca75be9a284ac5f98c078ec47c2e826ac5e980) +++ lib/Comms/Comms.pro (.../Comms.pro) (revision efc0e8ccb0ee99f25834eb5a19dee1cf2e50c532) @@ -21,6 +21,7 @@ include/main.h \ include/MessageBuilder.h \ include/MessageDispatcher.h \ + include/MessageSpool.h \ include/types.h SOURCES = \ @@ -33,10 +34,15 @@ src/FrameInterface.cpp \ src/MessageBuilder.cpp \ src/MessageDispatcher.cpp \ + src/MessageSpool.cpp \ src/types.cpp INCLUDEPATH += $$PWD/include +# MessageSpool uses the SQLite C API directly (see docs/SDD/AgentMigrationPlan.md +# section 5.5). Requires libsqlite3-dev at build time. +LIBS += -lsqlite3 + isEmpty(PREFIX): PREFIX = $$PWD/../../install target.path = $$PREFIX/lib Index: lib/Comms/include/MessageSpool.h =================================================================== diff -u --- lib/Comms/include/MessageSpool.h (revision 0) +++ lib/Comms/include/MessageSpool.h (revision efc0e8ccb0ee99f25834eb5a19dee1cf2e50c532) @@ -0,0 +1,103 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MessageSpool.h + * \author (original) Stephen Quong + * \date (original) 28-Jul-2026 + * + */ +#pragma once + +#include +#include +#include +#include + +struct sqlite3; + +/*! + * \brief One message held in, or destined for, the durable spool. + * \details spoolId is the SQLite rowid; -1 means the message has not been + * persisted yet. Callers hold the id returned by insert() so they can + * remove() the row once delivery is confirmed. + */ +struct SpooledMessage +{ + qint64 spoolId = -1; ///< SQLite rowid, or -1 when not yet spooled. + qint64 timestampMs = 0; ///< Unix epoch ms; stamped by insert() when 0. + quint16 sequence = 0; ///< Sequence number carried from the source message. + QString topic; ///< Resolved outbound topic. + QString deviceId; ///< Device identifier. + QString msgId; ///< Source message identifier, e.g. "0x72A0". + QByteArray payload; ///< Serialised payload bytes. +}; + +/*! + * \brief Durable SQLite store providing the store-and-forward guarantee. + * \details Every outbound message is inserted here before any delivery is + * attempted, and its row is removed only once delivery is confirmed. + * A crash between the two causes redelivery, never loss. + * + * Not a QObject: this is a leaf persistence detail with no signals and + * no thread affinity of its own. All public methods are mutex-guarded + * and safe to call from any thread, including non-Qt threads such as + * an MQTT client's event loop. + * + * Schema: + * \code + * CREATE TABLE spool ( + * id INTEGER PRIMARY KEY AUTOINCREMENT, + * timestamp_ms INTEGER NOT NULL, + * topic TEXT NOT NULL, + * payload BLOB NOT NULL, + * sequence INTEGER NOT NULL DEFAULT 0, + * device_id TEXT NOT NULL DEFAULT '', + * msg_id TEXT NOT NULL DEFAULT '' + * ); + * \endcode + */ +class MessageSpool +{ +public: + /*! + * \brief Tuning applied when the database is opened. + * \details retentionHours defaults to 96 per the SRS minimum, not the 48 + * used by the legacy agent implementation. + */ + struct Config + { + QString dbPath; ///< Absolute path to the SQLite file. + quint32 retentionHours = 96; ///< Maximum message age before purge. + }; + + explicit MessageSpool(Config config); + ~MessageSpool(); + + MessageSpool(const MessageSpool &) = delete; + MessageSpool &operator=(const MessageSpool &) = delete; + + bool open(); + void close(); + bool isOpen() const; + + qint64 insert(const SpooledMessage &msg); + QVector fetchBatch(quint32 limit); + bool remove(qint64 spoolId); + + quint64 depth() const; + quint64 sizeBytes() const; + + int purgeExpired(); + +private: + bool execSql(const char *sql); + quint64 scalarQuery(const char *sql) const; + + Config _config; + sqlite3 *_db = nullptr; + mutable QMutex _mutex; +}; Index: lib/Comms/src/MessageSpool.cpp =================================================================== diff -u --- lib/Comms/src/MessageSpool.cpp (revision 0) +++ lib/Comms/src/MessageSpool.cpp (revision efc0e8ccb0ee99f25834eb5a19dee1cf2e50c532) @@ -0,0 +1,428 @@ +/*! + * + * Copyright (c) 2026 Diality Inc. - All Rights Reserved. + * \copyright + * THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN + * WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. + * + * \file MessageSpool.cpp + * \author (original) Stephen Quong + * \date (original) 28-Jul-2026 + * + */ +#include "MessageSpool.h" + +#include +#include +#include +#include +#include + +#include + +Q_LOGGING_CATEGORY(lcSpool, "cloudconnect.spool") + +namespace { + +/*! + * \brief Stmt + * \details RAII wrapper around a prepared statement. sqlite3_finalize() must + * run on every exit path; hand-written finalize calls are the source + * of the leaks seen in the legacy implementation, where an early + * return skipped them. + */ +class Stmt +{ +public: + Stmt(sqlite3 *db, const char *sql) + { + _rc = sqlite3_prepare_v2(db, sql, -1, &_stmt, nullptr); + } + + ~Stmt() + { + if (_stmt != nullptr) { + sqlite3_finalize(_stmt); + } + } + + Stmt(const Stmt &) = delete; + Stmt &operator=(const Stmt &) = delete; + + bool isValid() const { return _rc == SQLITE_OK && _stmt != nullptr; } + sqlite3_stmt *get() const { return _stmt; } + +private: + sqlite3_stmt *_stmt = nullptr; + int _rc = SQLITE_ERROR; +}; + +/*! + * \brief columnText + * \details Reads a TEXT column that may be NULL. sqlite3_column_text() returns + * nullptr for a NULL column, which would construct a QString from a + * null pointer if used unguarded. + * \param stmt Prepared statement positioned on a row. + * \param column Zero-based column index. + * \return The column value, or an empty QString when the column is NULL. + */ +QString columnText(sqlite3_stmt *stmt, int column) +{ + const auto *text = reinterpret_cast(sqlite3_column_text(stmt, column)); + return (text != nullptr) ? QString::fromUtf8(text) : QString(); +} + +/*! + * \brief bindText + * \details Binds a QString as UTF-8. SQLITE_TRANSIENT tells SQLite to copy the + * bytes immediately; the temporary QByteArray below dies at the end of + * this call, so SQLITE_STATIC would leave SQLite holding a dangling + * pointer. + * \param stmt Prepared statement to bind against. + * \param index One-based bind parameter index. + * \param value String to bind. + * \return The sqlite3_bind_text() result code. + */ +int bindText(sqlite3_stmt *stmt, int index, const QString &value) +{ + const QByteArray utf8 = value.toUtf8(); + return sqlite3_bind_text(stmt, index, utf8.constData(), utf8.size(), SQLITE_TRANSIENT); +} + +} // namespace + +/*! + * \brief MessageSpool::MessageSpool + * \details Constructor. Stores the configuration; the database is not touched + * until open() is called. + * \param config Database path and retention window. + */ +MessageSpool::MessageSpool(Config config) + : _config(std::move(config)) +{ +} + +/*! + * \brief MessageSpool::~MessageSpool + * \details Destructor. Closes the database if it is still open. + */ +MessageSpool::~MessageSpool() +{ + close(); +} + +/*! + * \brief MessageSpool::open + * \details Creates the parent directory if needed, opens the database, applies + * the durability pragmas, and creates the schema. + * \return true when the spool is ready for use. + */ +bool MessageSpool::open() +{ + QMutexLocker locker(&_mutex); + + if (_db != nullptr) { + return true; + } + + const QString parentDir = QFileInfo(_config.dbPath).absolutePath(); + if (!QDir().mkpath(parentDir)) { + qCCritical(lcSpool) << "Cannot create spool directory" << parentDir; + return false; + } + + if (sqlite3_open(_config.dbPath.toUtf8().constData(), &_db) != SQLITE_OK) { + qCCritical(lcSpool) << "Cannot open" << _config.dbPath + << ":" << sqlite3_errmsg(_db); + sqlite3_close(_db); + _db = nullptr; + return false; + } + + // WAL keeps readers non-blocking during a write; busy_timeout absorbs the + // contention between a drain and a concurrent insert. + const bool pragmasOk = execSql("PRAGMA journal_mode=WAL;") + && execSql("PRAGMA synchronous=NORMAL;") + && execSql("PRAGMA busy_timeout=5000;"); + if (!pragmasOk) { + qCCritical(lcSpool) << "Cannot apply pragmas to" << _config.dbPath; + sqlite3_close(_db); + _db = nullptr; + return false; + } + + const bool schemaOk = execSql( + "CREATE TABLE IF NOT EXISTS spool (" + " id INTEGER PRIMARY KEY AUTOINCREMENT," + " timestamp_ms INTEGER NOT NULL," + " topic TEXT NOT NULL," + " payload BLOB NOT NULL," + " sequence INTEGER NOT NULL DEFAULT 0," + " device_id TEXT NOT NULL DEFAULT ''," + " msg_id TEXT NOT NULL DEFAULT ''" + ");") + && execSql("CREATE INDEX IF NOT EXISTS idx_spool_timestamp ON spool(timestamp_ms);"); + if (!schemaOk) { + qCCritical(lcSpool) << "Cannot create schema in" << _config.dbPath; + sqlite3_close(_db); + _db = nullptr; + return false; + } + + qCInfo(lcSpool) << "Opened" << _config.dbPath + << "retention" << _config.retentionHours << "hours"; + return true; +} + +/*! + * \brief MessageSpool::close + * \details Closes the database connection. Safe to call when already closed. + */ +void MessageSpool::close() +{ + QMutexLocker locker(&_mutex); + + if (_db != nullptr) { + sqlite3_close(_db); + _db = nullptr; + } +} + +/*! + * \brief MessageSpool::isOpen + * \details Reports whether the database is currently open. + * \return true if open() has succeeded and close() has not run since. + */ +bool MessageSpool::isOpen() const +{ + QMutexLocker locker(&_mutex); + return _db != nullptr; +} + +/*! + * \brief MessageSpool::insert + * \details Persists one message. Stamps the current time when the caller left + * timestampMs at 0. + * \param msg Message to persist. + * \return The assigned rowid, or -1 on failure. + */ +qint64 MessageSpool::insert(const SpooledMessage &msg) +{ + QMutexLocker locker(&_mutex); + + if (_db == nullptr) { + qCWarning(lcSpool) << "Insert on a closed spool"; + return -1; + } + + Stmt stmt(_db, + "INSERT INTO spool(timestamp_ms, topic, payload, sequence, device_id, msg_id)" + " VALUES(?,?,?,?,?,?);"); + if (!stmt.isValid()) { + qCCritical(lcSpool) << "Insert prepare failed:" << sqlite3_errmsg(_db); + return -1; + } + + const qint64 timestampMs = + (msg.timestampMs != 0) ? msg.timestampMs : QDateTime::currentMSecsSinceEpoch(); + + sqlite3_bind_int64(stmt.get(), 1, timestampMs); + bindText(stmt.get(), 2, msg.topic); + sqlite3_bind_blob(stmt.get(), 3, msg.payload.constData(), msg.payload.size(), SQLITE_TRANSIENT); + sqlite3_bind_int64(stmt.get(), 4, static_cast(msg.sequence)); + bindText(stmt.get(), 5, msg.deviceId); + bindText(stmt.get(), 6, msg.msgId); + + if (sqlite3_step(stmt.get()) != SQLITE_DONE) { + qCCritical(lcSpool) << "Insert failed:" << sqlite3_errmsg(_db); + return -1; + } + + return sqlite3_last_insert_rowid(_db); +} + +/*! + * \brief MessageSpool::fetchBatch + * \details Returns the oldest messages first. Ordering is by the AUTOINCREMENT + * id rather than by timestamp, so insertion order is preserved even + * when two messages share a millisecond or the clock steps backwards. + * \param limit Maximum number of messages to return. + * \return Up to limit messages, oldest first, each with spoolId populated. + */ +QVector MessageSpool::fetchBatch(quint32 limit) +{ + QMutexLocker locker(&_mutex); + + QVector batch; + if (_db == nullptr || limit == 0) { + return batch; + } + + Stmt stmt(_db, + "SELECT id, timestamp_ms, topic, payload, sequence, device_id, msg_id" + " FROM spool ORDER BY id ASC LIMIT ?;"); + if (!stmt.isValid()) { + qCCritical(lcSpool) << "Fetch prepare failed:" << sqlite3_errmsg(_db); + return batch; + } + + sqlite3_bind_int(stmt.get(), 1, static_cast(limit)); + + while (sqlite3_step(stmt.get()) == SQLITE_ROW) { + SpooledMessage msg; + msg.spoolId = sqlite3_column_int64(stmt.get(), 0); + msg.timestampMs = sqlite3_column_int64(stmt.get(), 1); + msg.topic = columnText(stmt.get(), 2); + + const auto *blob = static_cast(sqlite3_column_blob(stmt.get(), 3)); + const int blobSize = sqlite3_column_bytes(stmt.get(), 3); + if (blob != nullptr && blobSize > 0) { + msg.payload = QByteArray(blob, blobSize); + } + + msg.sequence = static_cast(sqlite3_column_int64(stmt.get(), 4)); + msg.deviceId = columnText(stmt.get(), 5); + msg.msgId = columnText(stmt.get(), 6); + + batch.append(msg); + } + + return batch; +} + +/*! + * \brief MessageSpool::remove + * \details Deletes one row. Call only once delivery of that message has been + * confirmed; removing earlier forfeits the store-and-forward guarantee. + * \param spoolId Rowid returned by insert(). + * \return true when the row was deleted. + */ +bool MessageSpool::remove(qint64 spoolId) +{ + QMutexLocker locker(&_mutex); + + if (_db == nullptr) { + return false; + } + + Stmt stmt(_db, "DELETE FROM spool WHERE id = ?;"); + if (!stmt.isValid()) { + qCCritical(lcSpool) << "Remove prepare failed:" << sqlite3_errmsg(_db); + return false; + } + + sqlite3_bind_int64(stmt.get(), 1, spoolId); + + if (sqlite3_step(stmt.get()) != SQLITE_DONE) { + qCCritical(lcSpool) << "Remove failed for id" << spoolId + << ":" << sqlite3_errmsg(_db); + return false; + } + + return sqlite3_changes(_db) > 0; +} + +/*! + * \brief MessageSpool::depth + * \details Counts the messages currently held. + * \return Number of spooled messages. + */ +quint64 MessageSpool::depth() const +{ + return scalarQuery("SELECT COUNT(*) FROM spool;"); +} + +/*! + * \brief MessageSpool::sizeBytes + * \details Sums the stored payload bytes. + * \return Total payload size in bytes, excluding SQLite overhead. + */ +quint64 MessageSpool::sizeBytes() const +{ + return scalarQuery("SELECT COALESCE(SUM(LENGTH(payload)),0) FROM spool;"); +} + +/*! + * \brief MessageSpool::purgeExpired + * \details Deletes messages older than the configured retention window. Time + * is the only bound on spool growth, so this must be called + * periodically for the disk budget to hold. + * \return The number of messages purged, or -1 on failure. + */ +int MessageSpool::purgeExpired() +{ + QMutexLocker locker(&_mutex); + + if (_db == nullptr) { + return -1; + } + + const qint64 cutoffMs = QDateTime::currentMSecsSinceEpoch() + - static_cast(_config.retentionHours) * 3600LL * 1000LL; + + Stmt stmt(_db, "DELETE FROM spool WHERE timestamp_ms < ?;"); + if (!stmt.isValid()) { + qCCritical(lcSpool) << "Purge prepare failed:" << sqlite3_errmsg(_db); + return -1; + } + + sqlite3_bind_int64(stmt.get(), 1, cutoffMs); + + if (sqlite3_step(stmt.get()) != SQLITE_DONE) { + qCCritical(lcSpool) << "Purge failed:" << sqlite3_errmsg(_db); + return -1; + } + + const int purged = sqlite3_changes(_db); + if (purged > 0) { + qCInfo(lcSpool) << "Purged" << purged << "messages older than" + << _config.retentionHours << "hours"; + } + + return purged; +} + +/*! + * \brief MessageSpool::execSql + * \details Runs a statement that returns no rows, logging any error. + * \param sql Statement to execute. + * \return true when the statement succeeded. + * \note Callers must already hold the mutex. + */ +bool MessageSpool::execSql(const char *sql) +{ + char *error = nullptr; + if (sqlite3_exec(_db, sql, nullptr, nullptr, &error) != SQLITE_OK) { + qCCritical(lcSpool) << "SQL error:" << (error != nullptr ? error : "unknown"); + sqlite3_free(error); + return false; + } + return true; +} + +/*! + * \brief MessageSpool::scalarQuery + * \details Runs a statement whose first row and column hold a single count. + * \param sql Statement to execute. + * \return The scalar value, or 0 when the spool is closed or the query fails. + */ +quint64 MessageSpool::scalarQuery(const char *sql) const +{ + QMutexLocker locker(&_mutex); + + if (_db == nullptr) { + return 0; + } + + Stmt stmt(_db, sql); + if (!stmt.isValid()) { + return 0; + } + + quint64 value = 0; + if (sqlite3_step(stmt.get()) == SQLITE_ROW) { + value = static_cast(sqlite3_column_int64(stmt.get(), 0)); + } + + return value; +}