test(txn): cover LockManagerTest, waitUntil and TransactionManagerTest

src/test/kotlin/kestrel/store/txn/LockManagerTest.kt

@@ -0,0 +1,117 @@
+package kestrel.store.txn
+
+import java.util.concurrent.TimeUnit
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertTrue
+
+class LockManagerTest {
+    private fun waitUntil(timeoutMs: Long = 2000, predicate: () -> Boolean) {
+        val deadline = System.currentTimeMillis() + timeoutMs
+        while (!predicate()) {
+            if (System.currentTimeMillis() > deadline) throw AssertionError("condition never became true")
+            Thread.sleep(5)
+        }
+    }
+
+    @Test
+    fun sharedLocksAreCompatibleAcrossTransactions() {
+        val locks = LockManager()
+        locks.acquireShared(1L, 100L)
+        locks.acquireShared(2L, 100L)
+        assertEquals(mapOf(1L to LockManager.LockMode.SHARED, 2L to LockManager.LockMode.SHARED), locks.holders(100L))
+    }
+
+    @Test
+    fun reacquiringTheSameOrWeakerModeIsANoop() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L)
+        locks.acquireShared(1L, 100L)
+        assertEquals(mapOf(1L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L))
+    }
+
+    @Test
+    fun upgradeFromSharedToExclusiveSucceedsWhenSoleHolder() {
+        val locks = LockManager()
+        locks.acquireShared(1L, 100L)
+        locks.acquireExclusive(1L, 100L)
+        assertEquals(mapOf(1L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L))
+    }
+
+    @Test
+    fun exclusiveLockBlocksAnotherTransactionUntilReleased() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L)
+
+        val acquired = java.util.concurrent.atomic.AtomicBoolean(false)
+        val thread = Thread {
+            locks.acquireExclusive(2L, 100L)
+            acquired.set(true)
+        }
+        thread.start()
+        waitUntil { locks.waiters(100L).contains(2L) }
+        assertTrue(!acquired.get(), "txn 2 must not acquire while txn 1 holds the exclusive lock")
+
+        locks.release(1L, 100L)
+        thread.join(2000)
+        assertTrue(acquired.get())
+        assertEquals(mapOf(2L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L))
+    }
+
+    @Test
+    fun releaseAllDropsEveryLockHeldByATransaction() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L)
+        locks.acquireShared(1L, 200L)
+        locks.releaseAll(1L)
+        assertEquals(emptyMap(), locks.holders(100L))
+        assertEquals(emptyMap(), locks.holders(200L))
+    }
+
+    @Test
+    fun circularWaitIsDetectedAndRejectedWithoutBlocking() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L) // txn 1 holds resource A
+        locks.acquireExclusive(2L, 200L) // txn 2 holds resource B
+
+        val blocked = java.util.concurrent.atomic.AtomicBoolean(false)
+        val thread = Thread {
+            locks.acquireExclusive(1L, 200L) // txn 1 wants B, held by txn 2: blocks
+            blocked.set(true)
+        }
+        thread.start()
+        waitUntil { locks.waiters(200L).contains(1L) }
+
+        // txn 2 now wants A, held by txn 1, which is already waiting on txn 2: a cycle.
+        assertFailsWith<LockManager.DeadlockException> { locks.acquireExclusive(2L, 100L) }
+
+        // Unblock the background waiter so the test does not leak a thread.
+        locks.release(2L, 200L)
+        thread.join(2000)
+        assertTrue(blocked.get())
+    }
+
+    @Test
+    fun unrelatedWaitersDoNotFalselyTriggerDeadlock() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L)
+        // txn 2 requests an unrelated resource txn 1 does not hold: no cycle, should not throw.
+        locks.acquireExclusive(2L, 200L)
+    }
+
+    @Test
+    fun exclusiveWaiterEventuallyRunsToCompletionAfterHolderReleases() {
+        val locks = LockManager()
+        locks.acquireExclusive(1L, 100L)
+        val latch = java.util.concurrent.CountDownLatch(1)
+        val thread = Thread {
+            locks.acquireExclusive(2L, 100L)
+            latch.countDown()
+        }
+        thread.start()
+        waitUntil { locks.waiters(100L).contains(2L) }
+        locks.releaseAll(1L)
+        assertTrue(latch.await(2, TimeUnit.SECONDS))
+    }
+}

src/test/kotlin/kestrel/store/txn/TransactionManagerTest.kt

@@ -0,0 +1,88 @@
+package kestrel.store.txn
+
+import kestrel.store.page.DiskManager
+import kestrel.store.page.Page
+import java.io.File
+import java.nio.file.Files
+import kestrel.store.wal.WriteAheadLog
+import kotlin.test.Test
+import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+
+class TransactionManagerTest {
+    private fun tempFile(prefix: String): File = Files.createTempFile(prefix, ".tmp").toFile().also { it.deleteOnExit() }
+
+    private fun pageOf(fill: Int): ByteArray = ByteArray(Page.PAGE_SIZE) { fill.toByte() }
+
+    private class Fixture {
+        val diskManager = DiskManager(Files.createTempFile("kestrel-txnmgr-data", ".db").toFile().also { it.deleteOnExit() })
+        val wal = WriteAheadLog(Files.createTempFile("kestrel-txnmgr-wal", ".log").toFile().also { it.deleteOnExit() })
+        val manager = TransactionManager(wal, diskManager)
+    }
+
+    @Test
+    fun beginStartsATransactionInActiveStatus() {
+        val fixture = Fixture()
+        val txn = fixture.manager.begin()
+        assertEquals(TransactionManager.Status.ACTIVE, fixture.manager.statusOf(txn))
+    }
+
+    @Test
+    fun commitAppliesLoggedUpdateAndReleasesLocks() {
+        val fixture = Fixture()
+        val pageId = fixture.diskManager.allocatePage()
+        val before = pageOf(0)
+        fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(before, 0, it.data, 0, before.size) })
+
+        val txn = fixture.manager.begin()
+        fixture.manager.acquireExclusive(txn, pageId)
+        val after = pageOf(3)
+        fixture.manager.logUpdate(txn, pageId, before, after)
+        fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(after, 0, it.data, 0, after.size) })
+        fixture.manager.commit(txn)
+
+        assertEquals(TransactionManager.Status.COMMITTED, fixture.manager.statusOf(txn))
+        assertContentEquals(after, fixture.diskManager.readPage(pageId).data)
+
+        val otherTxn = fixture.manager.begin()
+        fixture.manager.acquireExclusive(otherTxn, pageId) // must not block: txn's locks were released on commit
+    }
+
+    @Test
+    fun abortRollsBackLoggedUpdatesAndReleasesLocks() {
+        val fixture = Fixture()
+        val pageId = fixture.diskManager.allocatePage()
+        val before = pageOf(1)
+        fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(before, 0, it.data, 0, before.size) })
+
+        val txn = fixture.manager.begin()
+        fixture.manager.acquireExclusive(txn, pageId)
+        val after = pageOf(4)
+        fixture.manager.logUpdate(txn, pageId, before, after)
+        fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(after, 0, it.data, 0, after.size) })
+        fixture.manager.abort(txn)
+
+        assertEquals(TransactionManager.Status.ABORTED, fixture.manager.statusOf(txn))
+        assertContentEquals(before, fixture.diskManager.readPage(pageId).data)
+
+        val otherTxn = fixture.manager.begin()
+        fixture.manager.acquireExclusive(otherTxn, pageId) // must not block: aborted txn released its locks
+    }
+
+    @Test
+    fun operationsOnAResolvedTransactionFail() {
+        val fixture = Fixture()
+        val txn = fixture.manager.begin()
+        fixture.manager.commit(txn)
+        assertFailsWith<TransactionException> { fixture.manager.commit(txn) }
+        assertFailsWith<TransactionException> { fixture.manager.abort(txn) }
+        assertFailsWith<TransactionException> { fixture.manager.acquireExclusive(txn, 1L) }
+    }
+
+    @Test
+    fun statusOfUnknownTransactionFails() {
+        val fixture = Fixture()
+        assertFailsWith<TransactionException> { fixture.manager.statusOf(999L) }
+    }
+}