Test concurrent Store quota enforcement

src/agents/store/CMakeLists.txt

@@ -69,7 +69,9 @@ if(BUILD_TESTING)
 		${CMAKE_CURRENT_SOURCE_DIR})
 	target_compile_definitions(store-quota-test PRIVATE
 		STORE_SCHEMA_PATH="${CMAKE_CURRENT_SOURCE_DIR}/sql/create-store.sql")
-	target_link_libraries(store-quota-test PRIVATE SQLite3::SQLite3)
+	target_link_libraries(store-quota-test PRIVATE
+		SQLite3::SQLite3
+		Threads::Threads)
 	add_test(NAME store-quota COMMAND store-quota-test)
 
 	add_executable(store-initial-layout-test

src/agents/store/tests/quota-test.c

@@ -20,11 +20,14 @@
  ****************************************************************************/
 
 #include <assert.h>
+#include <inttypes.h>
+#include <pthread.h>
 #include <stdio.h>
 #include <sqlite3.h>
 #include <stdint.h>
 #include <stdlib.h>
 #include <string.h>
+#include <unistd.h>
 
 #include "quota-schema.h"
 
@@ -84,6 +87,104 @@ MailUsage(sqlite3 *database)
     return usage;
 }
 
+typedef struct {
+    const char *path;
+    int64_t guid;
+    pthread_mutex_t *mutex;
+    pthread_cond_t *condition;
+    int *ready;
+    int *start;
+    int result;
+    char *error;
+} ConcurrentInsert;
+
+static void *
+InsertMailConcurrently(void *opaque)
+{
+    ConcurrentInsert *insert = opaque;
+    sqlite3 *database = NULL;
+    char query[160];
+
+    assert(sqlite3_open(insert->path, &database) == SQLITE_OK);
+    assert(sqlite3_busy_timeout(database, 5000) == SQLITE_OK);
+
+    assert(pthread_mutex_lock(insert->mutex) == 0);
+    ++*insert->ready;
+    assert(pthread_cond_broadcast(insert->condition) == 0);
+    while (!*insert->start)
+        assert(pthread_cond_wait(insert->condition, insert->mutex) == 0);
+    assert(pthread_mutex_unlock(insert->mutex) == 0);
+
+    snprintf(query, sizeof(query),
+             "INSERT INTO storeobject(guid,type,size) "
+             "VALUES(%" PRId64 ",2,60)",
+             insert->guid);
+    insert->result = sqlite3_exec(database, query, NULL, NULL, &insert->error);
+    assert(sqlite3_close(database) == SQLITE_OK);
+    return NULL;
+}
+
+static void
+TestConcurrentDelivery(const char *schema)
+{
+    char path[] = "/tmp/bongo-store-quota-XXXXXX";
+    sqlite3 *database = NULL;
+    pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+    pthread_cond_t condition = PTHREAD_COND_INITIALIZER;
+    pthread_t threads[2];
+    int ready = 0;
+    int start = 0;
+    int successes = 0;
+    int over_quota = 0;
+    int file;
+    int index;
+    ConcurrentInsert inserts[2] = {
+        { path, 101, &mutex, &condition, &ready, &start, 0, NULL },
+        { path, 102, &mutex, &condition, &ready, &start, 0, NULL },
+    };
+
+    file = mkstemp(path);
+    assert(file >= 0);
+    assert(close(file) == 0);
+    assert(sqlite3_open(path, &database) == SQLITE_OK);
+    ExecOk(database, schema);
+    ExecOk(database, "UPDATE storequota SET limit_bytes = 100");
+    assert(sqlite3_close(database) == SQLITE_OK);
+
+    for (index = 0; index < 2; ++index)
+        assert(pthread_create(&threads[index], NULL, InsertMailConcurrently,
+                              &inserts[index]) == 0);
+
+    assert(pthread_mutex_lock(&mutex) == 0);
+    while (ready != 2)
+        assert(pthread_cond_wait(&condition, &mutex) == 0);
+    start = 1;
+    assert(pthread_cond_broadcast(&condition) == 0);
+    assert(pthread_mutex_unlock(&mutex) == 0);
+
+    for (index = 0; index < 2; ++index) {
+        assert(pthread_join(threads[index], NULL) == 0);
+        if (inserts[index].result == SQLITE_OK) {
+            ++successes;
+        } else {
+            assert(inserts[index].result == SQLITE_CONSTRAINT);
+            assert(inserts[index].error);
+            assert(strstr(inserts[index].error, "quota exceeded"));
+            ++over_quota;
+        }
+        sqlite3_free(inserts[index].error);
+    }
+    assert(successes == 1);
+    assert(over_quota == 1);
+
+    assert(sqlite3_open(path, &database) == SQLITE_OK);
+    assert(MailUsage(database) == 60);
+    assert(sqlite3_close(database) == SQLITE_OK);
+    assert(unlink(path) == 0);
+    assert(pthread_cond_destroy(&condition) == 0);
+    assert(pthread_mutex_destroy(&mutex) == 0);
+}
+
 static void
 TestMigration(void)
 {
@@ -119,7 +220,6 @@ main(void)
 
     assert(sqlite3_open(":memory:", &database) == SQLITE_OK);
     ExecOk(database, schema);
-    free(schema);
     ExecOk(database, "UPDATE storequota SET limit_bytes = 100");
 
     ExecOk(database,
@@ -164,5 +264,7 @@ main(void)
 
     assert(sqlite3_close(database) == SQLITE_OK);
     TestMigration();
+    TestConcurrentDelivery(schema);
+    free(schema);
     return 0;
 }