diff --git a/src/libs/msgapi/CMakeLists.txt b/src/libs/msgapi/CMakeLists.txt
index 8447c5c..dc5a384 100644
--- a/src/libs/msgapi/CMakeLists.txt
+++ b/src/libs/msgapi/CMakeLists.txt
@@ -21,4 +21,24 @@ target_link_libraries(bongomsgapi
install(TARGETS bongomsgapi DESTINATION ${LIB_INSTALL_DIR})
+if(BUILD_TESTING)
+ add_executable(msgapi-sqldb-transaction-test
+ tests/sqldb-transaction-test.c)
+ if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
+ target_link_options(msgapi-sqldb-transaction-test PRIVATE
+ "LINKER:--no-as-needed")
+ endif()
+ target_link_libraries(msgapi-sqldb-transaction-test PRIVATE
+ bongomsgapi
+ bongocal
+ bongojson
+ bongoutil
+ bongoxpl
+ SQLite3::SQLite3
+ Threads::Threads)
+ add_test(NAME msgapi-sqldb-transactions
+ COMMAND msgapi-sqldb-transaction-test)
+ set_tests_properties(msgapi-sqldb-transactions PROPERTIES TIMEOUT 30)
+endif()
+
add_subdirectory(auth-backends)
diff --git a/src/libs/msgapi/sqldb.c b/src/libs/msgapi/sqldb.c
index ef5b623..f91b5cf 100644
--- a/src/libs/msgapi/sqldb.c
+++ b/src/libs/msgapi/sqldb.c
@@ -138,38 +138,48 @@ MsgSQLSetLockTimeout(MsgSQLHandle *handle, int timeoutms)
int
MsgSQLBeginTransaction(MsgSQLHandle *handle)
{
+ MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
int count = 1 + handle->lockTimeoutMs / MSGSQL_STMT_SLEEP_MS;
- BOOL locked = FALSE;
+
+ memset(&statement, 0, sizeof(statement));
// acquire the transaction lock to prevent other people doing stuff
XplMutexLock(handle->transactionLock);
- if (locked) {
- Log(LOG_TRACE, "sql3: Acquired lock successfully");
- }
- stmt = MsgSQLPrepare(handle, "BEGIN TRANSACTION;", &handle->stmts.begin);
+ /*
+ * Store connections move between worker threads. Keeping transaction
+ * statements cached on the shared handle made SQLite reject a later BEGIN
+ * even though the previous transaction had completed. Transaction control
+ * statements are small, so prepare and finalize them within the lock.
+ */
+ stmt = MsgSQLPrepare(handle, "BEGIN TRANSACTION;", &statement);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to begin transaction");
+ XplMutexUnlock(handle->transactionLock);
return -1;
}
do {
result = sqlite3_step(stmt->stmt);
Log(LOG_TRACE, "sql3: BEGIN TRAN stmt %ld", stmt->stmt);
- sqlite3_reset(stmt->stmt);
- Log(LOG_TRACE, "sql: - bt reset stmt %ld", stmt->stmt);
} while (SQLITE_BUSY == result && --count && (XplDelay(MSGSQL_STMT_SLEEP_MS), 1));
switch (result) {
case SQLITE_DONE:
+ MsgSQLFinalize(&statement);
++(handle->transactionDepth);
return 0;
case SQLITE_BUSY:
+ Log(LOG_ERROR, "sql3: Database remained busy while beginning transaction");
+ MsgSQLFinalize(&statement);
+ XplMutexUnlock(handle->transactionLock);
return -2;
default:
Log(LOG_ERROR, "sql3: Database error %d : %s", result, sqlite3_errmsg(handle->db));
+ MsgSQLFinalize(&statement);
+ XplMutexUnlock(handle->transactionLock);
return -1;
}
}
@@ -178,11 +188,18 @@ MsgSQLBeginTransaction(MsgSQLHandle *handle)
int
MsgSQLCommitTransaction(MsgSQLHandle *handle)
{
+ MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
int count = 1 + handle->lockTimeoutMs / MSGSQL_STMT_SLEEP_MS;
- stmt = MsgSQLPrepare(handle, "END TRANSACTION;", &handle->stmts.end);
+ if (handle->transactionDepth <= 0) {
+ Log(LOG_ERROR, "sql3: Attempted to commit without an active transaction");
+ return -1;
+ }
+
+ memset(&statement, 0, sizeof(statement));
+ stmt = MsgSQLPrepare(handle, "END TRANSACTION;", &statement);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to prepare end of transaction statement");
return -1;
@@ -191,15 +208,15 @@ MsgSQLCommitTransaction(MsgSQLHandle *handle)
do {
result = sqlite3_step(stmt->stmt);
Log(LOG_TRACE, "sql3: COMMIT stmt %ld", stmt->stmt);
- sqlite3_reset(stmt->stmt);
- Log(LOG_TRACE, "sql3: - ct reset stmt %ld", stmt->stmt);
} while (SQLITE_BUSY == result && --count && (XplDelay(MSGSQL_STMT_SLEEP_MS), 1));
if (SQLITE_DONE != result) {
Log(LOG_ERROR, "sql3: Database commit error %d: %s", result, sqlite3_errmsg(handle->db));
+ MsgSQLFinalize(&statement);
return -1;
}
+ MsgSQLFinalize(&statement);
--handle->transactionDepth;
XplMutexUnlock(handle->transactionLock);
@@ -210,22 +227,29 @@ MsgSQLCommitTransaction(MsgSQLHandle *handle)
int
MsgSQLAbortTransaction(MsgSQLHandle *handle)
{
+ MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
- stmt = MsgSQLPrepare(handle, "ROLLBACK TRANSACTION;", &handle->stmts.abort);
- if (!stmt) {
- Log(LOG_ERROR, "sql3: Unable to rollback transaction");
+ if (handle->transactionDepth <= 0) {
+ Log(LOG_ERROR, "sql3: Attempted to rollback without an active transaction");
return -1;
}
- result = sqlite3_step(stmt->stmt);
- if (SQLITE_DONE != result) {
- Log(LOG_ERROR, "sql3: Transaction rollback failed");
+ memset(&statement, 0, sizeof(statement));
+ stmt = MsgSQLPrepare(handle, "ROLLBACK TRANSACTION;", &statement);
+ if (!stmt) {
+ Log(LOG_ERROR, "sql3: Unable to rollback transaction");
+ result = SQLITE_ERROR;
+ } else {
+ result = sqlite3_step(stmt->stmt);
+ if (SQLITE_DONE != result) {
+ Log(LOG_ERROR, "sql3: Transaction rollback failed: %s", sqlite3_errmsg(handle->db));
+ }
+ Log(LOG_TRACE, "sql3: ABORT stmt %ld\n", stmt->stmt);
+ MsgSQLFinalize(&statement);
}
- Log(LOG_TRACE, "sql3: ABORT stmt %ld\n", stmt->stmt);
- sqlite3_reset(stmt->stmt);
- Log(LOG_TRACE, "sql3: - at reset stmt %ld\n", stmt->stmt);
+
--handle->transactionDepth;
XplMutexUnlock(handle->transactionLock);
diff --git a/src/libs/msgapi/tests/sqldb-transaction-test.c b/src/libs/msgapi/tests/sqldb-transaction-test.c
new file mode 100644
index 0000000..0e59b66
--- /dev/null
+++ b/src/libs/msgapi/tests/sqldb-transaction-test.c
@@ -0,0 +1,116 @@
+/****************************************************************************
+ *
+ * Copyright (c) 2001 Novell, Inc. All Rights Reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of version 2 of the GNU General Public License
+ * as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, contact Novell, Inc.
+ *
+ ****************************************************************************/
+
+#include
+#include
+#include
+#include
+
+#include
+#include
+
+#define THREAD_COUNT 12
+#define TRANSACTIONS_PER_THREAD 100
+
+typedef struct {
+ MsgSQLHandle *handle;
+ int failed;
+} TransactionThread;
+
+static void *
+RunTransactions(void *argument)
+{
+ TransactionThread *thread = argument;
+ int i;
+
+ for (i = 0; i < TRANSACTIONS_PER_THREAD; i++) {
+ if (MsgSQLBeginTransaction(thread->handle) != 0) {
+ thread->failed = 1;
+ return NULL;
+ }
+ if (MsgSQLQuickExecute(thread->handle, "SELECT 1;") != 0) {
+ MsgSQLAbortTransaction(thread->handle);
+ thread->failed = 1;
+ return NULL;
+ }
+ if (MsgSQLCommitTransaction(thread->handle) != 0) {
+ MsgSQLAbortTransaction(thread->handle);
+ thread->failed = 1;
+ return NULL;
+ }
+ }
+ return NULL;
+}
+
+static void
+CheckBeginFailureReleasesLock(void)
+{
+ MsgSQLHandle *handle;
+ char *error = NULL;
+ int lockResult;
+
+ handle = MsgSQLOpen(":memory:", NULL, 0);
+ assert(handle != NULL);
+ assert(sqlite3_exec(handle->db, "BEGIN TRANSACTION;", NULL, NULL,
+ &error) == SQLITE_OK);
+ assert(error == NULL);
+ assert(MsgSQLBeginTransaction(handle) == -1);
+
+ lockResult = XplMutexTryLock(handle->transactionLock);
+ assert(lockResult == 0);
+ XplMutexUnlock(handle->transactionLock);
+
+ assert(sqlite3_exec(handle->db, "ROLLBACK TRANSACTION;", NULL, NULL,
+ &error) == SQLITE_OK);
+ assert(error == NULL);
+ assert(MsgSQLBeginTransaction(handle) == 0);
+ assert(MsgSQLCommitTransaction(handle) == 0);
+ MsgSQLClose(handle);
+}
+
+static void
+CheckConcurrentTransactions(void)
+{
+ MsgSQLHandle *handle;
+ TransactionThread threads[THREAD_COUNT];
+ pthread_t ids[THREAD_COUNT];
+ int i;
+
+ handle = MsgSQLOpen(":memory:", NULL, 1000);
+ assert(handle != NULL);
+ memset(threads, 0, sizeof(threads));
+ for (i = 0; i < THREAD_COUNT; i++) {
+ threads[i].handle = handle;
+ assert(pthread_create(&ids[i], NULL, RunTransactions,
+ &threads[i]) == 0);
+ }
+ for (i = 0; i < THREAD_COUNT; i++) {
+ assert(pthread_join(ids[i], NULL) == 0);
+ assert(threads[i].failed == 0);
+ }
+ assert(handle->transactionDepth == 0);
+ MsgSQLClose(handle);
+}
+
+int
+main(void)
+{
+ CheckBeginFailureReleasesLock();
+ CheckConcurrentTransactions();
+ return 0;
+}