perf(chat): ingest a message batch in one transaction
Ingesting a scan's worth of chat messages persisted each new message with its own append() call, so a batch of fresh messages opened and committed one SQLite transaction per row. - A new appendBatch persists many messages in a single transaction, using the same INSERT OR IGNORE dedup key as append() and preserving input order so load()'s rowid ordering is unchanged. - Every message is serialized and encrypted before the transaction opens, so a message with a non-UTF-8 body is skipped on its own without leaving a transaction open; the valid siblings still persist. - Ingest collects the newly stored messages and writes them once after the loop instead of per message. It is inert (returns 0) while locked or on empty input. Tests cover a batch insert, dedup on re-insert, empty input, bad-UTF-8 isolation with a valid sibling and a following append(), insert order, and the locked no-op. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -132,6 +132,55 @@ bool ChatDatabase::append(const ChatMessage& message)
|
|||||||
return sqlite3_changes(db_) > 0;
|
return sqlite3_changes(db_) > 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
int ChatDatabase::appendBatch(const std::vector<ChatMessage>& messages)
|
||||||
|
{
|
||||||
|
if (!key_ready_ || !ensureOpen()) return 0;
|
||||||
|
if (messages.empty()) return 0;
|
||||||
|
|
||||||
|
struct Row { std::string dedup; std::vector<unsigned char> nonce, cipher; };
|
||||||
|
std::vector<Row> rows;
|
||||||
|
rows.reserve(messages.size());
|
||||||
|
for (const auto& m : messages) {
|
||||||
|
if (isTombstoned(m)) continue; // locally deleted — don't re-persist
|
||||||
|
try {
|
||||||
|
std::string plain = serialize(m); // strict json.dump() may throw on non-UTF-8 body
|
||||||
|
std::vector<unsigned char> nonce, cipher;
|
||||||
|
const bool ok = encrypt(plain, nonce, cipher);
|
||||||
|
if (!plain.empty()) sodium_memzero(&plain[0], plain.size());
|
||||||
|
if (!ok) continue;
|
||||||
|
rows.push_back({ dedupHash(m.txid, m.payload_position), std::move(nonce), std::move(cipher) });
|
||||||
|
} catch (const std::exception&) {
|
||||||
|
continue; // skip only this message; never open a txn with bad data
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (rows.empty()) return 0;
|
||||||
|
|
||||||
|
if (!exec("BEGIN")) return 0; // match deleteMessages() in-file style
|
||||||
|
sqlite3_stmt* stmt = nullptr;
|
||||||
|
if (sqlite3_prepare_v2(db_,
|
||||||
|
"INSERT OR IGNORE INTO chat_messages (wallet_tag, dedup_hash, nonce, payload) "
|
||||||
|
"VALUES (?, ?, ?, ?)",
|
||||||
|
-1, &stmt, nullptr) != SQLITE_OK) {
|
||||||
|
exec("ROLLBACK");
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
int inserted = 0;
|
||||||
|
bool ok = true;
|
||||||
|
for (const auto& r : rows) {
|
||||||
|
sqlite3_reset(stmt);
|
||||||
|
sqlite3_clear_bindings(stmt);
|
||||||
|
sqlite3_bind_text(stmt, 1, wallet_tag_.c_str(), -1, SQLITE_TRANSIENT);
|
||||||
|
sqlite3_bind_text(stmt, 2, r.dedup.c_str(), -1, SQLITE_TRANSIENT);
|
||||||
|
sqlite3_bind_blob(stmt, 3, r.nonce.data(), static_cast<int>(r.nonce.size()), SQLITE_TRANSIENT);
|
||||||
|
sqlite3_bind_blob(stmt, 4, r.cipher.data(), static_cast<int>(r.cipher.size()), SQLITE_TRANSIENT);
|
||||||
|
if (sqlite3_step(stmt) != SQLITE_DONE) { ok = false; break; }
|
||||||
|
if (sqlite3_changes(db_) > 0) ++inserted;
|
||||||
|
}
|
||||||
|
sqlite3_finalize(stmt);
|
||||||
|
if (!exec(ok ? "COMMIT" : "ROLLBACK")) return 0;
|
||||||
|
return ok ? inserted : 0;
|
||||||
|
}
|
||||||
|
|
||||||
bool ChatDatabase::upsert(const ChatMessage& message)
|
bool ChatDatabase::upsert(const ChatMessage& message)
|
||||||
{
|
{
|
||||||
if (!key_ready_ || !ensureOpen()) return false;
|
if (!key_ready_ || !ensureOpen()) return false;
|
||||||
|
|||||||
@@ -44,6 +44,12 @@ public:
|
|||||||
// Returns true if newly inserted; false on duplicate or while locked.
|
// Returns true if newly inserted; false on duplicate or while locked.
|
||||||
bool append(const ChatMessage& message);
|
bool append(const ChatMessage& message);
|
||||||
|
|
||||||
|
// Persist many messages in ONE transaction (INSERT OR IGNORE, same dedup key as append()).
|
||||||
|
// Serializes+encrypts every message BEFORE opening the transaction, so a non-UTF-8 body is
|
||||||
|
// skipped without leaving a transaction open. Preserves input order (load() relies on rowid).
|
||||||
|
// Returns the number of rows actually inserted. No-op (returns 0) while locked or on empty input.
|
||||||
|
int appendBatch(const std::vector<ChatMessage>& messages);
|
||||||
|
|
||||||
// Persist-or-overwrite one message by its (txid+position) dedup key. Unlike append(), this
|
// Persist-or-overwrite one message by its (txid+position) dedup key. Unlike append(), this
|
||||||
// updates an existing row's payload — used for outgoing echoes whose delivery status changes
|
// updates an existing row's payload — used for outgoing echoes whose delivery status changes
|
||||||
// (Sending → Sent/Failed). Returns true on success; false while locked / on error.
|
// (Sending → Sent/Failed). Returns true on success; false while locked / on error.
|
||||||
|
|||||||
@@ -38,6 +38,7 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
|
|||||||
const std::string myPubKey = chatIdentityPublicKeyHex(identity_);
|
const std::string myPubKey = chatIdentityPublicKeyHex(identity_);
|
||||||
|
|
||||||
int added = 0;
|
int added = 0;
|
||||||
|
std::vector<ChatMessage> fresh; // persisted once after the loop (one transaction)
|
||||||
for (const auto& meta : metadata) {
|
for (const auto& meta : metadata) {
|
||||||
// Already stored: append would refuse it. Every scan re-sends the whole history, so this skips the
|
// Already stored: append would refuse it. Every scan re-sends the whole history, so this skips the
|
||||||
// decrypt for all but the new messages.
|
// decrypt for all but the new messages.
|
||||||
@@ -107,13 +108,14 @@ int ChatService::ingest(const std::vector<HushChatTransactionMetadata>& metadata
|
|||||||
// session loadFromDatabase() repopulates the store, so re-scanning the chain re-ingests but
|
// session loadFromDatabase() repopulates the store, so re-scanning the chain re-ingests but
|
||||||
// the store dedup prevents a duplicate write.
|
// the store dedup prevents a duplicate write.
|
||||||
if (store_.append(message)) {
|
if (store_.append(message)) {
|
||||||
if (db_) db_->append(message);
|
fresh.push_back(message);
|
||||||
++added;
|
++added;
|
||||||
// Every ingested message is incoming — report its cid so the caller can notify without
|
// Every ingested message is incoming — report its cid so the caller can notify without
|
||||||
// relying on a seen-watermark delta (which block-time vs wall-clock skew can swallow).
|
// relying on a seen-watermark delta (which block-time vs wall-clock skew can swallow).
|
||||||
if (newIncomingCids) newIncomingCids->push_back(message.conversation_id);
|
if (newIncomingCids) newIncomingCids->push_back(message.conversation_id);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (db_ && !fresh.empty()) db_->appendBatch(fresh);
|
||||||
return added;
|
return added;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -14274,6 +14274,34 @@ void testHushChatDatabase()
|
|||||||
EXPECT_TRUE(loaded[0].delivery == ChatDelivery::Failed);
|
EXPECT_TRUE(loaded[0].delivery == ChatDelivery::Failed);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// appendBatch: one transaction, dedup, order, tombstone-skip, bad-UTF-8 isolation.
|
||||||
|
scrub(dbPath);
|
||||||
|
{
|
||||||
|
ChatDatabase db(dbPath);
|
||||||
|
EXPECT_TRUE(db.unlockWithSecret(seedA));
|
||||||
|
std::vector<ChatMessage> batch = {
|
||||||
|
makeMsg("b1", 0, "c", "one", ChatMessageKind::Message, 1),
|
||||||
|
makeMsg("b2", 0, "c", "two", ChatMessageKind::Message, 2),
|
||||||
|
makeMsg("b3", 0, "c", "three", ChatMessageKind::Message, 3),
|
||||||
|
};
|
||||||
|
EXPECT_EQ(db.appendBatch(batch), 3);
|
||||||
|
EXPECT_EQ(db.appendBatch(batch), 0); // dedup: all ignored
|
||||||
|
EXPECT_EQ((int)db.appendBatch({}), 0); // empty → no-op
|
||||||
|
// bad UTF-8 body is skipped, the valid sibling still persists, and append() still works after.
|
||||||
|
ChatMessage bad = makeMsg("b4", 0, "c", std::string("\xff\xfe", 2), ChatMessageKind::Message, 4);
|
||||||
|
ChatMessage good = makeMsg("b5", 0, "c", "five", ChatMessageKind::Message, 5);
|
||||||
|
EXPECT_EQ(db.appendBatch({bad, good}), 1); // only good persisted, no open txn left
|
||||||
|
EXPECT_TRUE(db.append(makeMsg("b6", 0, "c", "six", ChatMessageKind::Message, 6)));
|
||||||
|
std::vector<ChatMessage> all = db.load();
|
||||||
|
EXPECT_EQ((int)all.size(), 5); // b1,b2,b3,b5,b6
|
||||||
|
EXPECT_EQ(all[0].body, std::string("one")); // rowid/insert order
|
||||||
|
EXPECT_EQ(all[4].body, std::string("six"));
|
||||||
|
}
|
||||||
|
{
|
||||||
|
ChatDatabase locked(dbPath); // no unlock → inert
|
||||||
|
EXPECT_EQ(locked.appendBatch({makeMsg("x", 0, "c", "x", ChatMessageKind::Message, 0)}), 0);
|
||||||
|
}
|
||||||
|
|
||||||
scrub(dbPath); scrub(dbPath2);
|
scrub(dbPath); scrub(dbPath2);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user