In WAL mode, plain reads and read-only transactions take the primary connection and block writers #99

import android.content.Context
import android.database.Cursor
import android.os.Build
import android.util.Log
import androidx.test.core.app.ApplicationProvider
import androidx.test.ext.junit.runners.AndroidJUnit4
import java.io.File
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import org.junit.Assume.assumeTrue
import org.junit.Test
import org.junit.runner.RunWith
import android.database.sqlite.SQLiteDatabase as FrameworkDatabase
import net.zetetic.database.sqlcipher.SQLiteDatabase as CipherDatabase

@RunWith(AndroidJUnit4::class)
class PrimaryConnectionReproTest {

    private val context: Context = ApplicationProvider.getApplicationContext()
    private val executor = Executors.newCachedThreadPool()

    private interface Engine {
        fun query(sql: String, vararg args: Any): Cursor
        fun write(value: Long)
        fun beginTransaction()
        fun beginTransactionReadOnly()
        fun endTransaction()
        fun close()
    }

    @Test
    fun writerWaitsForReaders() {
        assumeTrue(Build.VERSION.SDK_INT >= Build.VERSION_CODES.VANILLA_ICE_CREAM)
        System.loadLibrary("sqlcipher")
        val engines = listOf("framework" to ::framework, "sqlcipher" to ::cipher)
        engines.forEach { (name, open) ->
            report(name, open, "coldSingleReader") { engine -> writerWait(engine, listOf { slowRead(engine) }) }
            report(name, open, "warmTwoReaders") { engine ->
                openNonPrimaryConnection(engine)
                writerWait(engine, listOf({ slowRead(engine) }, { slowRead(engine) }))
            }
            report(name, open, "readOnlyTransaction") { engine ->
                writerWait(engine, listOf { readOnlyTransaction(engine) })
            }
        }
        executor.shutdown()
        Log.i(TAG, "versions sqlcipher=${cipherVersion()} device=${Build.MODEL} api=${Build.VERSION.SDK_INT}")
    }

    private fun report(name: String, open: () -> Engine, scenario: String, block: (Engine) -> Pair<Long, Long>) {
        val runs = List(RUNS) {
            val engine = open()
            try {
                requireWal(engine)
                block(engine)
            } finally {
                engine.close()
            }
        }
        val reader = runs.map { it.first }.sorted()[RUNS / 2]
        val writer = runs.map { it.second }.sorted()[RUNS / 2]
        Log.i(TAG, "RESULT engine=$name scenario=$scenario " +
            "readerMs=${ms(reader)} writerWaitMs=${ms(writer)}")
    }

    private fun writerWait(engine: Engine, readers: List<() -> Unit>): Pair<Long, Long> {
        val started = CountDownLatch(readers.size)
        val futures = readers.map { reader ->
            executor.submit<Long> {
                started.countDown()
                val start = System.nanoTime()
                reader()
                System.nanoTime() - start
            }
        }
        started.await()
        Thread.sleep(START_DELAY_MS)
        val start = System.nanoTime()
        engine.write(System.nanoTime())
        val wait = System.nanoTime() - start
        val reader = futures.maxOf { it.get() }
        return reader to wait
    }

    private fun requireWal(engine: Engine) {
        engine.query("PRAGMA journal_mode").use { cursor ->
            check(cursor.moveToFirst())
            check(cursor.getString(0).equals("wal", ignoreCase = true)) { "journal_mode=${cursor.getString(0)}" }
        }
    }

    private fun slowRead(engine: Engine) {
        engine.query(SLOW_SELECT).use { cursor ->
            check(cursor.moveToFirst())
            check(cursor.getLong(0) == SLOW_ROWS)
        }
    }

    private fun readOnlyTransaction(engine: Engine) {
        engine.beginTransactionReadOnly()
        try {
            engine.query("SELECT count(*) FROM t").use { it.moveToFirst() }
            Thread.sleep(HOLD_MS)
        } finally {
            engine.endTransaction()
        }
    }

    private fun openNonPrimaryConnection(engine: Engine) {
        engine.beginTransaction()
        try {
            executor.submit { engine.query("SELECT count(*) FROM t").use { it.moveToFirst() } }.get()
        } finally {
            engine.endTransaction()
        }
    }

    private fun framework(): Engine {
        val file = freshFile("primary-framework.db")
        val db = FrameworkDatabase.openDatabase(
            file.path,
            null,
            FrameworkDatabase.CREATE_IF_NECESSARY or FrameworkDatabase.ENABLE_WRITE_AHEAD_LOGGING
        )
        db.execSQL(CREATE_TABLE)
        return object : Engine {
            override fun query(sql: String, vararg args: Any): Cursor =
                db.rawQuery(sql, args.map { it.toString() }.toTypedArray())
            override fun write(value: Long) = db.execSQL(INSERT, arrayOf(value))
            override fun beginTransaction() = db.beginTransaction()
            override fun beginTransactionReadOnly() {
                if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.VANILLA_ICE_CREAM) {
                    db.beginTransactionReadOnly()
                } else {
                    error("beginTransactionReadOnly requires API 35")
                }
            }
            override fun endTransaction() = db.endTransaction()
            override fun close() = db.close()
        }
    }

    private fun cipher(): Engine {
        val file = freshFile("primary-sqlcipher.db")
        val db = CipherDatabase.openDatabase(
            file.path,
            "",
            null,
            CipherDatabase.CREATE_IF_NECESSARY or CipherDatabase.ENABLE_WRITE_AHEAD_LOGGING,
            null
        )
        db.execSQL(CREATE_TABLE)
        return object : Engine {
            override fun query(sql: String, vararg args: Any): Cursor = db.rawQuery(sql, *args)
            override fun write(value: Long) = db.execSQL(INSERT, arrayOf(value))
            override fun beginTransaction() = db.beginTransaction()
            override fun beginTransactionReadOnly() = db.beginTransactionReadOnly()
            override fun endTransaction() = db.endTransaction()
            override fun close() = db.close()
        }
    }

    private fun freshFile(name: String): File {
        context.deleteDatabase(name)
        return context.getDatabasePath(name).also { it.parentFile?.mkdirs() }
    }

    private fun cipherVersion(): String {
        val db = CipherDatabase.openDatabase(":memory:", "", null, CipherDatabase.CREATE_IF_NECESSARY, null)
        return db.rawQuery("PRAGMA cipher_version").use { cursor ->
            if (cursor.moveToFirst()) cursor.getString(0) else "?"
        }.also { db.close() }
    }

    private fun ms(nanos: Long) = "%.1f".format(nanos / 1e6)

    private companion object {
        const val TAG = "PrimaryConnectionRepro"
        const val CREATE_TABLE = "CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER NOT NULL)"
        const val INSERT = "INSERT INTO t (v) VALUES (?)"
        const val SLOW_ROWS = 3_000_000L
        const val SLOW_SELECT =
            "WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c WHERE x < $SLOW_ROWS) " +
                "SELECT count(*) FROM c"
        const val HOLD_MS = 500L
        const val START_DELAY_MS = 50L
        const val RUNS = 3
    }
}