libsql-sqlite3: Add conflict detection to libsql_wal_insert_frame()

Author: penberg

libsql-sqlite3/doc/libsql_extensions.md

@@ -379,7 +379,8 @@ static void sync_db(sqlite3 *db_primary, sqlite3 *db_backup){
   for(int i=1; i<=max_frame; i++){
     char frame[4096+24];
     libsql_wal_get_frame(db_primary, i, frame, sizeof(frame));
-    libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame));
+    int conflict;
+    libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame), &conflict);
   }
   libsql_wal_end_commit(db_backup);
 }

libsql-sqlite3/src/main.c

@@ -2557,7 +2557,8 @@ int libsql_wal_insert_frame(
   sqlite3* db,
   unsigned int iFrame,
   void *pBuf,
-  unsigned int nBuf
+  unsigned int nBuf,
+  int *pConflict
 ){
   int rc = SQLITE_OK;
   Pager *pPager;
@@ -2572,7 +2573,7 @@ int libsql_wal_insert_frame(
 
   sqlite3_mutex_enter(db->mutex);
   pPager = sqlite3BtreePager(db->aDb[0].pBt);
-  rc = sqlite3PagerWalInsert(pPager, iFrame, pBuf, nBuf);
+  rc = sqlite3PagerWalInsert(pPager, iFrame, pBuf, nBuf, pConflict);
   if (rc != SQLITE_OK) {
     goto out_unlock;
   }

libsql-sqlite3/src/pager.c

@@ -7826,9 +7826,12 @@ int sqlite3PagerWalEndCommit(Pager *pPager) {
   return rc;
 }
 
-int sqlite3PagerWalInsert(Pager *pPager, unsigned int iFrame, void *pBuf, unsigned int nBuf) {
+int sqlite3PagerWalInsert(Pager *pPager, unsigned int iFrame, void *pBuf, unsigned int nBuf, int *pConflict) {
   int rc = SQLITE_OK;
 
+  if( pConflict ) {
+    *pConflict = 0;
+  }
   if (!pagerUseWal(pPager)) {
     return SQLITE_ERROR;
   }
@@ -7838,6 +7841,22 @@ int sqlite3PagerWalInsert(Pager *pPager, unsigned int iFrame, void *pBuf, unsign
     return rc;
   }
   if (iFrame <= mxFrame) {
+    unsigned long frame_len = nBuf-24;
+    unsigned char current[frame_len];
+    rc = pPager->wal->methods.xReadFrame(pPager->wal->pData, iFrame, frame_len, current);
+    if (rc != SQLITE_OK) {
+      return rc;
+    }
+    int conflict = 0;
+    if (memcmp(pBuf+24, current, frame_len) != 0) {
+      conflict = 1;
+    }
+    if (pConflict) {
+      *pConflict = conflict;
+    }
+    if (conflict) {
+      return SQLITE_ERROR;
+    }
     return SQLITE_OK;
   }
   u8 *aFrame = (u8*)pBuf;

libsql-sqlite3/src/pager.h

@@ -137,7 +137,7 @@ int sqlite3PagerWalFrameCount(Pager *, unsigned int *);
 int sqlite3PagerWalReadFrame(Pager *, unsigned int, void *, unsigned int);
 int sqlite3PagerWalBeginCommit(Pager*);
 int sqlite3PagerWalEndCommit(Pager*);
-int sqlite3PagerWalInsert(Pager*, unsigned int, void *, unsigned int);
+int sqlite3PagerWalInsert(Pager*, unsigned int, void *, unsigned int, int *);
 
 void sqlite3PagerSetCachesize(Pager*, int);
 int sqlite3PagerSetSpillsize(Pager*, int);

libsql-sqlite3/src/sqlite.h.in

@@ -10602,7 +10602,7 @@ int libsql_wal_insert_end(sqlite3*);
 ** CAPI3REF: Insert a frame into the WAL
 ** METHOD: sqlite3
 */
-int libsql_wal_insert_frame(sqlite3*, unsigned int, void *, unsigned int);
+int libsql_wal_insert_frame(sqlite3*, unsigned int, void *, unsigned int, int *);
 
 /*
 ** CAPI3REF: Low-level system error code

libsql-sqlite3/src/test_walapi.c

@@ -56,7 +56,9 @@ static void sync_db(sqlite3 *db_primary, sqlite3 *db_backup){
   for(int i=1; i<=max_frame; i++){
     char frame[4096+24];
     ensure(libsql_wal_get_frame(db_primary, i, frame, sizeof(frame)) == SQLITE_OK, "can't get frame: %s\n", sqlite3_errmsg(db_primary));
-    ensure(libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame)) == SQLITE_OK, "can't inject frame: %s\n", sqlite3_errmsg(db_backup));
+    int conflict;
+    ensure(libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame), &conflict) == SQLITE_OK, "can't inject frame: %s\n", sqlite3_errmsg(db_backup));
+    ensure(conflict == 0, "conflict at frame %d\n", i);
   }
   ensure(libsql_wal_insert_end(db_backup) == SQLITE_OK, "can't end commit: %s\n", sqlite3_errmsg(db_backup));
 }
@@ -110,7 +112,9 @@ void test_sync_by_parts() {
         in_commit = 1;
         ensure(libsql_wal_insert_begin(db_backup) == SQLITE_OK, "can't begin commit: %s\n", sqlite3_errmsg(db_backup));
       }
-      ensure(libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame)) == SQLITE_OK, "can't inject frame: %s\n", sqlite3_errmsg(db_backup));
+      int conflict; 
+      ensure(libsql_wal_insert_frame(db_backup, i, frame, sizeof(frame), &conflict) == SQLITE_OK, "can't inject frame: %s\n", sqlite3_errmsg(db_backup));
+      ensure(conflict == 0, "conflict at frame %d\n", i);
       if (is_commit) {
         ensure(libsql_wal_insert_end(db_backup) == SQLITE_OK, "can't end commit: %s\n", sqlite3_errmsg(db_backup));
         in_commit = 0;
@@ -145,6 +149,44 @@ void test_sync_while_reading() {
     cmp_data(db_primary, db_backup); 
 }
 
+// This test case writes to two different databases and then attempts to sync them to a third database.
+// Only the first database should be synced, the second database sync should return a conflict error
+void test_conflict() {    
+  sqlite3 *db1, *db2, *db_synced;
+  open_db("test_conflict_1.db", &db1);
+  open_db("test_conflict_2.db", &db2);
+  open_db("test_conflict_synced.db", &db_synced);
+
+  ensure(sqlite3_exec(db1, "CREATE TABLE t (x)", 0, 0, 0) == SQLITE_OK, "failed to insert data\n");
+  ensure(sqlite3_exec(db1, "INSERT INTO t VALUES (randomblob(4 * 1024))", 0, 0, 0) == SQLITE_OK, "failed to insert data\n");
+
+  sync_db(db1, db_synced);
+
+  ensure(sqlite3_exec(db2, "CREATE TABLE t (x)", 0, 0, 0) == SQLITE_OK, "failed to insert data\n");
+  ensure(sqlite3_exec(db2, "INSERT INTO t VALUES (randomblob(4 * 1024))", 0, 0, 0) == SQLITE_OK, "failed to insert data\n");
+
+  unsigned int max_frame;
+  ensure(libsql_wal_frame_count(db2, &max_frame) == SQLITE_OK, "can't get frame count: %s\n", sqlite3_errmsg(db2));
+  ensure(libsql_wal_insert_begin(db_synced) == SQLITE_OK, "can't begin commit: %s\n", sqlite3_errmsg(db_synced));
+  // First 3 frames should not conflict.
+  for(int i=1; i<=3; i++){
+    char frame[4096+24];
+    ensure(libsql_wal_get_frame(db2, i, frame, sizeof(frame)) == SQLITE_OK, "can't get frame: %s\n", sqlite3_errmsg(db2));
+    int conflict;
+    ensure(libsql_wal_insert_frame(db_synced, i, frame, sizeof(frame), &conflict) == SQLITE_OK, "conflict detected: %s\n", sqlite3_errmsg(db_synced));
+    ensure(conflict == 0, "conflict at frame %d\n", i);
+  }
+  // The rest should conflict.
+  for(int i=4; i<=max_frame; i++){
+    char frame[4096+24];
+    ensure(libsql_wal_get_frame(db2, i, frame, sizeof(frame)) == SQLITE_OK, "can't get frame: %s\n", sqlite3_errmsg(db2));
+    int conflict;
+    ensure(libsql_wal_insert_frame(db_synced, i, frame, sizeof(frame), &conflict) == SQLITE_ERROR, "conflict not detected: %s\n", sqlite3_errmsg(db_synced));
+    ensure(conflict == 1, "no conflict at frame %d\n", i);
+  }
+  ensure(libsql_wal_insert_end(db_synced) == SQLITE_OK, "can't end commit: %s\n", sqlite3_errmsg(db_synced));
+}
+
 int main(int argc, char *argv[])
 {
     test_huge_payload();
@@ -156,5 +198,8 @@ int main(int argc, char *argv[])
     test_sync_while_reading();
     printf("============= OK test_sync_while_reading\n");
 
+    test_conflict();
+    printf("============= OK test_conflict\n");
+
     return 0;
 }