From c5a8e7a28c2d35151afb34f8593f9cd7e2429105 Mon Sep 17 00:00:00 2001 From: zhibei <785740487@qq.com> Date: Mon, 13 Jul 2026 15:13:29 +0800 Subject: [PATCH] =?UTF-8?q?fix(database):=20=E4=BF=AE=E5=A4=8D=E7=8E=A9?= =?UTF-8?q?=E5=AE=B6=E6=95=B0=E6=8D=AE=E5=B9=B6=E5=8F=91=E5=86=99=E5=85=A5?= =?UTF-8?q?=E4=B8=80=E8=87=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 通过复合唯一索引、兼容迁移和原子 upsert 阻止重复记录，并收敛缓存延迟写与调度竞态以确保最新值最终落库。 --- .../database/database-player/build.gradle.kts | 7 + .../taboolib/expansion/DataContainer.kt | 171 +++++++++- .../kotlin/taboolib/expansion/Database.kt | 287 +++++++++++++++- .../main/kotlin/taboolib/expansion/TypeSQL.kt | 8 +- .../kotlin/taboolib/expansion/TypeSQLite.kt | 13 +- .../com/zaxxer/hikari_4_0_3/HikariConfig.kt | 7 + .../zaxxer/hikari_4_0_3/HikariDataSource.kt | 7 + .../PlayerDatabaseConsistencyTest.kt | 317 ++++++++++++++++++ 8 files changed, 777 insertions(+), 40 deletions(-) create mode 100644 module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt create mode 100644 module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt create mode 100644 module/database/database-player/src/test/kotlin/taboolib/expansion/PlayerDatabaseConsistencyTest.kt diff --git a/module/database/database-player/build.gradle.kts b/module/database/database-player/build.gradle.kts index 360bd131d..8a639bcda 100644 --- a/module/database/database-player/build.gradle.kts +++ b/module/database/database-player/build.gradle.kts @@ -5,4 +5,11 @@ dependencies { compileOnly(project(":module:database")) compileOnly(project(":module:basic:basic-configuration")) compileOnly("ink.ptms.core:v11701:11701-minimize:universal") + testImplementation(project(":common")) + testImplementation(project(":common-util")) + testImplementation(project(":common-platform-api")) + testImplementation(project(":module:database")) + testImplementation("com.zaxxer:HikariCP:4.0.3") + testImplementation("org.junit.jupiter:junit-jupiter:5.10.2") + testImplementation("org.xerial:sqlite-jdbc:3.42.0.0") } \ No newline at end of file diff --git a/module/database/database-player/src/main/kotlin/taboolib/expansion/DataContainer.kt b/module/database/database-player/src/main/kotlin/taboolib/expansion/DataContainer.kt index 1f6420dc3..b4fa62560 100644 --- a/module/database/database-player/src/main/kotlin/taboolib/expansion/DataContainer.kt +++ b/module/database/database-player/src/main/kotlin/taboolib/expansion/DataContainer.kt @@ -23,6 +23,12 @@ class DataContainer(val user: String, val database: Database) { /** 存储需要更新的键值对及其更新时间 */ val updateMap = ConcurrentHashMap() + private val writeStates = ConcurrentHashMap() + + internal var asyncExecutor: ((() -> Unit) -> Unit) = { task -> + submitAsync { task() } + } + /** * 设置指定键的值并立即保存 * @@ -30,13 +36,8 @@ class DataContainer(val user: String, val database: Database) { * @param value 值 */ operator fun set(key: String, value: Any) { - source[key] = value.toString() - if (value.toString().isEmpty()) { - source.remove(key) - delete(key) - } else { - save(key) - } + val stringValue = value.toString() + updateValue(key, stringValue.takeUnless { it.isEmpty() }, deadline = null, updateSource = true) } /** @@ -48,11 +49,11 @@ class DataContainer(val user: String, val database: Database) { * @param sync 是否同步给内存，要求targetUser为UUID */ fun forcedSet(targetUser: String, key: String, value: Any, sync: Boolean = false) { - database[targetUser, key] = value.toString() - // 因为 targetUser 不一定是UUID + val stringValue = value.toString() + database[targetUser, key] = stringValue if (sync) { - UUID.fromString(targetUser)?.let { - playerDataContainer[it]?.source?.set(key, value.toString()) + runCatching { UUID.fromString(targetUser) }.getOrNull()?.let { uniqueId -> + playerDataContainer[uniqueId]?.set(key, stringValue) } } } @@ -66,8 +67,9 @@ class DataContainer(val user: String, val database: Database) { * @param timeUnit 时间单位 */ fun setDelayed(key: String, value: Any, delay: Long = 3L, timeUnit: TimeUnit = TimeUnit.SECONDS) { - source[key] = value.toString() - updateMap[key] = System.currentTimeMillis() - timeUnit.toMillis(delay) + val stringValue = value.toString() + val deadline = deadlineAfter(timeUnit.toMillis(delay)) + updateValue(key, stringValue.takeUnless { it.isEmpty() }, deadline, updateSource = true) } /** @@ -113,23 +115,136 @@ class DataContainer(val user: String, val database: Database) { * @param key 键 */ fun save(key: String) { - submitAsync { database[user, key] = source[key]!! } + val state = writeStates.computeIfAbsent(key) { WriteState() } + synchronized(state) { + state.revision++ + state.value = source[key] + state.deadline = null + state.ready = true + updateMap.remove(key) + if (state.startIfNeeded()) { + scheduleWrite(key, state) + } + } } /** * 从数据库执行删除指定的键操作 */ fun delete(key: String) { - submitAsync { database.remove(user, key) } + updateValue(key, value = null, deadline = null, updateSource = false) } /** * 检查并更新需要保存的键值对 */ fun checkUpdate() { - updateMap.filterValues { it < System.currentTimeMillis() }.forEach { (t, _) -> - updateMap.remove(t) - save(t) + val currentTime = System.currentTimeMillis() + writeStates.forEach { (key, state) -> + synchronized(state) { + val deadline = state.deadline + if (deadline != null && deadline <= currentTime) { + state.deadline = null + state.ready = true + updateMap.remove(key, deadline) + if (state.startIfNeeded()) { + scheduleWrite(key, state) + } + } + } + } + } + + private fun updateValue(key: String, value: String?, deadline: Long?, updateSource: Boolean) { + val state = writeStates.computeIfAbsent(key) { WriteState() } + synchronized(state) { + if (updateSource) { + if (value == null) { + source.remove(key) + } else { + source[key] = value + } + } + state.revision++ + state.value = value + state.deadline = deadline + state.ready = deadline == null + if (deadline == null) { + updateMap.remove(key) + } else { + updateMap[key] = deadline + } + if (state.startIfNeeded()) { + scheduleWrite(key, state) + } + } + } + + private fun scheduleWrite(key: String, state: WriteState) { + try { + asyncExecutor.invoke { + drainWrites(key, state) + } + } catch (ex: Throwable) { + synchronized(state) { + state.running = false + } + throw ex + } + } + + private fun drainWrites(key: String, state: WriteState) { + while (true) { + val snapshot = synchronized(state) { + if (!state.ready) { + state.running = false + return + } + WriteSnapshot(state.revision, state.value) + } + try { + if (snapshot.value == null) { + database.remove(user, key) + } else { + database[user, key] = snapshot.value + } + } catch (ex: Throwable) { + synchronized(state) { + val hasNewerValue = state.revision != snapshot.revision && state.ready + state.running = false + if (hasNewerValue) { + state.running = true + runCatching { scheduleWrite(key, state) }.exceptionOrNull()?.let(ex::addSuppressed) + } + } + throw ex + } + val shouldContinue = synchronized(state) { + when { + state.revision == snapshot.revision -> { + state.ready = false + state.running = false + false + } + state.ready -> true + else -> { + state.running = false + false + } + } + } + if (!shouldContinue) { + return + } + } + } + + private fun deadlineAfter(delayMillis: Long): Long { + val currentTime = System.currentTimeMillis() + return if (delayMillis > 0 && currentTime > Long.MAX_VALUE - delayMillis) { + Long.MAX_VALUE + } else { + currentTime + delayMillis } } @@ -142,6 +257,26 @@ class DataContainer(val user: String, val database: Database) { return "DataContainer(user='$user', source=$source)" } + private class WriteState { + + var revision = 0L + var value: String? = null + var deadline: Long? = null + var ready = false + var running = false + + fun startIfNeeded(): Boolean { + return if (ready && !running) { + running = true + true + } else { + false + } + } + } + + private data class WriteSnapshot(val revision: Long, val value: String?) + /** * 内部伴生对象，用于定期检查更新 */ diff --git a/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt b/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt index 369f29f36..dabdc5d34 100644 --- a/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt +++ b/module/database/database-player/src/main/kotlin/taboolib/expansion/Database.kt @@ -1,19 +1,32 @@ package taboolib.expansion +import taboolib.common.PrimitiveIO +import taboolib.module.database.asFormattedColumnName +import taboolib.module.database.setupQuoterForHost +import java.sql.Connection +import java.sql.SQLException +import java.sql.SQLIntegrityConstraintViolationException +import java.util.Locale +import java.util.TreeMap import java.util.concurrent.ConcurrentHashMap import javax.sql.DataSource class Database(val type: Type, val dataSource: DataSource = type.host().createDataSource()) { + private val table = type.tableVar() + private val uniqueIndexName = createUniqueIndexName(table.name) + private val migrationLock = migrationLocks.computeIfAbsent("${type.host().connectionUrl}|${table.name}") { Any() } + init { - type.tableVar().createTable(dataSource) + table.createTable(dataSource) + ensureUniqueKeyIndex() } /** * 根据用户获取用户所有的数据 */ operator fun get(user: String): MutableMap { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("key", "value") where("user" eq user) }.map { @@ -25,7 +38,7 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 根据用户和键获取数据 */ operator fun get(user: String, key: String): String? { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("value") where("user" eq user and ("key" eq key)) limit(1) @@ -43,15 +56,15 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa remove(user, key) return } - if (get(user, key) == null) { - type.tableVar().insert(dataSource, "user", "key", "value") { + when (type) { + is TypeSQL -> table.insert(dataSource, "user", "key", "value") { value(user, key, data) + onDuplicateKeyUpdate { + update("value", data) + } } - } else { - type.tableVar().update(dataSource) { - set("value", data) - where("user" eq user and ("key" eq key)) - } + is TypeSQLite -> upsertSQLite(user, key, data) + else -> upsertGeneric(user, key, data) } } @@ -60,7 +73,7 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 如果数据不存在则返回 null */ fun getValue(user: String, key: String): String? { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("key", "value") where("user" eq user and ("key" eq key)) }.firstOrNull { @@ -72,7 +85,7 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 返回所有满足 Key = Value 的用户 */ fun getUserList(key: String, value: String): List { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("user") where("key" eq key and ("value" eq value)) }.map { @@ -84,7 +97,7 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 根据 Key 来返回一个 的Map */ fun getListByKey(key: String): MutableMap { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("user", "value") where("key" eq key) }.map { @@ -97,7 +110,7 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 例如 key = "title-" 则会查询所有以 "title-" 开头的数据 */ fun getLikeKeyList(user: String, key: String): MutableMap { - return type.tableVar().select(dataSource) { + return table.select(dataSource) { rows("key", "value") where("user" eq user and ("key" like "${key}%")) }.map { @@ -109,8 +122,250 @@ class Database(val type: Type, val dataSource: DataSource = type.host().createDa * 删除符合条件的数据 */ fun remove(user: String, key: String) { - type.tableVar().delete(dataSource) { + table.delete(dataSource) { + where("user" eq user and ("key" eq key)) + } + } + + private fun upsertSQLite(user: String, key: String, data: String) { + setupQuoterForHost(type.host()) + val tableName = table.name.asFormattedColumnName() + val userColumn = "user".asFormattedColumnName() + val keyColumn = "key".asFormattedColumnName() + val valueColumn = "value".asFormattedColumnName() + val query = "INSERT OR REPLACE INTO $tableName ($userColumn, $keyColumn, $valueColumn) VALUES (?, ?, ?)" + dataSource.connection.use { connection -> + connection.prepareStatement(query).use { statement -> + statement.setString(1, user) + statement.setString(2, key) + statement.setString(3, data) + statement.executeUpdate() + } + } + } + + private fun upsertGeneric(user: String, key: String, data: String) { + if (updateValue(user, key, data) > 0) { + return + } + try { + table.insert(dataSource, "user", "key", "value") { + value(user, key, data) + } + } catch (ex: SQLException) { + if (!ex.isConstraintViolation() || updateValue(user, key, data) == 0) { + throw ex + } + } + } + + private fun updateValue(user: String, key: String, data: String): Int { + return table.update(dataSource) { + set("value", data) where("user" eq user and ("key" eq key)) } } -} \ No newline at end of file + + private fun ensureUniqueKeyIndex() { + synchronized(migrationLock) { + dataSource.connection.use { connection -> + if (findUniqueKeyIndex(connection) != null) { + return + } + if (connection.metaData.databaseProductName.orEmpty().contains("SQLite", ignoreCase = true)) { + migrateSQLite(connection) + } else { + migrateWithRetry(connection) + } + } + } + } + + private fun migrateSQLite(connection: Connection) { + val removedRows = inTransaction(connection) { + val removed = removeDuplicateRows(connection) + createUniqueIndex(connection, resolveUniqueIndexName(connection)) + removed + } + if (findUniqueKeyIndex(connection) == null) { + throw SQLException("Unable to create a unique player key index for table ${table.name}") + } + warnDuplicateRows(removedRows) + } + + private fun migrateWithRetry(connection: Connection) { + var removedRows = 0L + var lastFailure: SQLException? = null + repeat(MAX_INDEX_ATTEMPTS) { attempt -> + if (findUniqueKeyIndex(connection) != null) { + warnDuplicateRows(removedRows) + return + } + removedRows += inTransaction(connection) { + removeDuplicateRows(connection) + } + try { + createUniqueIndex(connection, resolveUniqueIndexName(connection)) + } catch (ex: SQLException) { + if (findUniqueKeyIndex(connection) != null) { + warnDuplicateRows(removedRows) + return + } + lastFailure = ex + if (attempt + 1 >= MAX_INDEX_ATTEMPTS || countDuplicateRows(connection) == 0L) { + throw ex + } + return@repeat + } + if (findUniqueKeyIndex(connection) != null) { + warnDuplicateRows(removedRows) + return + } + } + throw lastFailure ?: SQLException("Unable to create a unique player key index for table ${table.name}") + } + + private fun createUniqueIndex(connection: Connection, indexName: String) { + setupQuoterForHost(type.host()) + val tableName = table.name.asFormattedColumnName() + val formattedIndexName = indexName.asFormattedColumnName() + val userColumn = "user".asFormattedColumnName() + val keyColumn = "key".asFormattedColumnName() + val ifNotExists = if (connection.metaData.databaseProductName.orEmpty().contains("SQLite", ignoreCase = true)) " IF NOT EXISTS" else "" + val query = "CREATE UNIQUE INDEX$ifNotExists $formattedIndexName ON $tableName ($userColumn, $keyColumn)" + connection.prepareStatement(query).use { statement -> + statement.executeUpdate() + } + } + + private fun findUniqueKeyIndex(connection: Connection): String? { + return readIndices(connection).firstOrNull { index -> + !index.nonUnique && index.columns.values.map { it.lowercase(Locale.ROOT) } == UNIQUE_KEY_COLUMNS + }?.name + } + + private fun resolveUniqueIndexName(connection: Connection): String { + val existingNames = readIndices(connection).map { it.name.lowercase(Locale.ROOT) }.toHashSet() + if (uniqueIndexName.lowercase(Locale.ROOT) !in existingNames) { + return uniqueIndexName + } + for (suffix in 2..99) { + val candidate = "${uniqueIndexName}_$suffix" + if (candidate.lowercase(Locale.ROOT) !in existingNames) { + return candidate + } + } + throw SQLException("Unable to allocate a unique index name for table ${table.name}") + } + + private fun readIndices(connection: Connection): List { + val indices = LinkedHashMap() + val tableNames = linkedSetOf(table.name, table.name.substringAfterLast('.')) + tableNames.forEach { tableName -> + connection.metaData.getIndexInfo(connection.catalog, null, tableName, false, false).use { result -> + while (result.next()) { + val indexName = result.getString("INDEX_NAME") ?: continue + val columnName = result.getString("COLUMN_NAME") ?: continue + val index = indices.computeIfAbsent(indexName.lowercase(Locale.ROOT)) { + IndexMetadata(indexName, result.getBoolean("NON_UNIQUE")) + } + index.nonUnique = index.nonUnique || result.getBoolean("NON_UNIQUE") + index.columns[result.getShort("ORDINAL_POSITION").toInt()] = columnName + } + } + } + return indices.values.toList() + } + + private fun removeDuplicateRows(connection: Connection): Long { + val duplicateRows = countDuplicateRows(connection) + if (duplicateRows == 0L) { + return 0L + } + connection.prepareStatement(createDuplicateDeleteQuery(connection)).use { statement -> + statement.executeUpdate() + } + val remainingRows = countDuplicateRows(connection) + if (remainingRows > 0) { + throw SQLException("Unable to remove duplicate player database rows from table ${table.name}") + } + return duplicateRows + } + + private fun countDuplicateRows(connection: Connection): Long { + setupQuoterForHost(type.host()) + val tableName = table.name.asFormattedColumnName() + val userColumn = "user".asFormattedColumnName() + val keyColumn = "key".asFormattedColumnName() + val query = "SELECT COALESCE(SUM(group_size - 1), 0) FROM (" + + "SELECT COUNT(*) AS group_size FROM $tableName GROUP BY $userColumn, $keyColumn HAVING COUNT(*) > 1" + + ") duplicate_groups" + return connection.prepareStatement(query).use { statement -> + statement.executeQuery().use { result -> + if (result.next()) result.getLong(1) else 0L + } + } + } + + private fun createDuplicateDeleteQuery(connection: Connection): String { + setupQuoterForHost(type.host()) + val tableName = table.name.asFormattedColumnName() + val userColumn = "user".asFormattedColumnName() + val keyColumn = "key".asFormattedColumnName() + val databaseName = connection.metaData.databaseProductName.orEmpty() + return if (databaseName.contains("SQLite", ignoreCase = true)) { + "DELETE FROM $tableName WHERE rowid NOT IN (" + + "SELECT MAX(rowid) FROM $tableName GROUP BY $userColumn, $keyColumn)" + } else { + val idColumn = "id".asFormattedColumnName() + "DELETE FROM $tableName WHERE $idColumn NOT IN (" + + "SELECT retained_id FROM (SELECT MAX($idColumn) AS retained_id FROM $tableName " + + "GROUP BY $userColumn, $keyColumn) retained_rows)" + } + } + + private fun inTransaction(connection: Connection, block: () -> T): T { + val originalAutoCommit = connection.autoCommit + connection.autoCommit = false + return try { + block().also { connection.commit() } + } catch (ex: Throwable) { + runCatching { connection.rollback() }.exceptionOrNull()?.let(ex::addSuppressed) + throw ex + } finally { + runCatching { connection.autoCommit = originalAutoCommit } + } + } + + private fun warnDuplicateRows(removedRows: Long) { + if (removedRows > 0) { + PrimitiveIO.warning( + "Removed {0} duplicate rows from player database table {1} before creating its unique key index.", + removedRows, + table.name, + ) + } + } + + private fun createUniqueIndexName(tableName: String): String { + val normalizedName = tableName.replace(Regex("[^A-Za-z0-9_]"), "_").ifEmpty { "table" } + return "uk_${normalizedName.take(36)}_${Integer.toHexString(tableName.hashCode())}_user_key" + } + + private fun SQLException.isConstraintViolation(): Boolean { + return this is SQLIntegrityConstraintViolationException || sqlState?.startsWith("23") == true || errorCode == 19 + } + + private data class IndexMetadata( + val name: String, + var nonUnique: Boolean, + val columns: TreeMap = TreeMap(), + ) + + private companion object { + + const val MAX_INDEX_ATTEMPTS = 4 + val UNIQUE_KEY_COLUMNS = listOf("user", "key") + val migrationLocks = ConcurrentHashMap() + } +} diff --git a/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQL.kt b/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQL.kt index 5bb994ab6..7a611de2d 100644 --- a/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQL.kt +++ b/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQL.kt @@ -17,16 +17,18 @@ class TypeSQL(val host: Host, val table: String) : Type() { add { id() } add("user") { type(ColumnTypeSQL.VARCHAR, 36) { - options(ColumnOptionSQL.KEY) + options(ColumnOptionSQL.NOTNULL, ColumnOptionSQL.KEY) } } add("key") { type(ColumnTypeSQL.VARCHAR, 64) { - options(ColumnOptionSQL.KEY) + options(ColumnOptionSQL.NOTNULL, ColumnOptionSQL.KEY) } } add("value") { - type(ColumnTypeSQL.VARCHAR, 128) + type(ColumnTypeSQL.VARCHAR, 128) { + options(ColumnOptionSQL.NOTNULL) + } } } diff --git a/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQLite.kt b/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQLite.kt index a2cf1eabe..e1c54c42e 100644 --- a/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQLite.kt +++ b/module/database/database-player/src/main/kotlin/taboolib/expansion/TypeSQLite.kt @@ -2,6 +2,7 @@ package taboolib.expansion import taboolib.common.io.newFile import taboolib.common.platform.function.pluginId +import taboolib.module.database.ColumnOptionSQLite import taboolib.module.database.ColumnTypeSQLite import taboolib.module.database.Host import taboolib.module.database.Table @@ -26,13 +27,19 @@ class TypeSQLite(val file: File, val tableName: String? = null) : Type() { */ val tableVar = Table(tableName ?: pluginId, host) { add("user") { - type(ColumnTypeSQLite.TEXT, 64) + type(ColumnTypeSQLite.TEXT, 64) { + options(ColumnOptionSQLite.NOTNULL) + } } add("key") { - type(ColumnTypeSQLite.TEXT, 64) + type(ColumnTypeSQLite.TEXT, 64) { + options(ColumnOptionSQLite.NOTNULL) + } } add("value") { - type(ColumnTypeSQLite.TEXT) + type(ColumnTypeSQLite.TEXT) { + options(ColumnOptionSQLite.NOTNULL) + } } } diff --git a/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt new file mode 100644 index 000000000..134ac11c6 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariConfig.kt @@ -0,0 +1,7 @@ +package com.zaxxer.hikari_4_0_3 + +/** + * 测试环境使用 database 模块 shadow 产物，字节码会引用重定位后的 HikariConfig。 + * 生产环境由 TabooLib 运行时依赖提供，这里只在 test classpath 代理到原始 Hikari。 + */ +class HikariConfig : com.zaxxer.hikari.HikariConfig() diff --git a/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt new file mode 100644 index 000000000..b6ab0ba52 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/com/zaxxer/hikari_4_0_3/HikariDataSource.kt @@ -0,0 +1,7 @@ +package com.zaxxer.hikari_4_0_3 + +/** + * 测试环境使用 database 模块 shadow 产物，字节码会引用重定位后的 HikariDataSource。 + * 生产环境由 TabooLib 运行时依赖提供，这里只在 test classpath 代理到原始 Hikari。 + */ +class HikariDataSource(config: HikariConfig) : com.zaxxer.hikari.HikariDataSource(config) diff --git a/module/database/database-player/src/test/kotlin/taboolib/expansion/PlayerDatabaseConsistencyTest.kt b/module/database/database-player/src/test/kotlin/taboolib/expansion/PlayerDatabaseConsistencyTest.kt new file mode 100644 index 000000000..ec04445b3 --- /dev/null +++ b/module/database/database-player/src/test/kotlin/taboolib/expansion/PlayerDatabaseConsistencyTest.kt @@ -0,0 +1,317 @@ +package taboolib.expansion + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import org.sqlite.SQLiteConfig +import org.sqlite.SQLiteDataSource +import java.nio.file.Path +import java.sql.SQLException +import java.util.ArrayDeque +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.RejectedExecutionException +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +class PlayerDatabaseConsistencyTest { + + @TempDir + lateinit var tempDir: Path + + @Test + fun `concurrent first writes keep one row`() { + val fixture = createFixture("concurrent") + val executor = Executors.newFixedThreadPool(8) + val start = CountDownLatch(1) + val values = (0 until 32).map { "value-$it" } + val futures = values.map { value -> + executor.submit { + start.await() + fixture.database["player", "score"] = value + } + } + + try { + start.countDown() + futures.forEach { it.get(30, TimeUnit.SECONDS) } + } finally { + executor.shutdownNow() + } + + assertEquals(1, countRows(fixture.dataSource, fixture.table, "player", "score")) + assertTrue(fixture.database["player", "score"] in values) + } + + @Test + fun `legacy duplicate rows keep latest value before unique index creation`() { + val table = "legacy_player_data" + val file = tempDir.resolve("legacy.db").toFile() + val dataSource = createDataSource(file.toPath()) + dataSource.connection.use { connection -> + connection.createStatement().use { statement -> + statement.executeUpdate("CREATE TABLE `$table` (`user` TEXT, `key` TEXT, `value` TEXT)") + statement.executeUpdate("INSERT INTO `$table` (`user`, `key`, `value`) VALUES ('player', 'score', 'old')") + statement.executeUpdate("INSERT INTO `$table` (`user`, `key`, `value`) VALUES ('player', 'score', 'new')") + statement.executeUpdate("INSERT INTO `$table` (`user`, `key`, `value`) VALUES ('other', 'score', 'kept')") + } + } + + val database = Database(TypeSQLite(file, table), dataSource) + + assertEquals("new", database["player", "score"]) + assertEquals(1, countRows(dataSource, table, "player", "score")) + assertEquals("kept", database["other", "score"]) + assertThrows(SQLException::class.java) { + dataSource.connection.use { connection -> + connection.prepareStatement("INSERT INTO `$table` (`user`, `key`, `value`) VALUES (?, ?, ?)").use { statement -> + statement.setString(1, "player") + statement.setString(2, "score") + statement.setString(3, "duplicate") + statement.executeUpdate() + } + } + } + } + + @Test + fun `concurrent initialization migrates duplicates once`() { + val table = "concurrent_migration" + val file = tempDir.resolve("concurrent-migration.db").toFile() + val setupDataSource = createDataSource(file.toPath()) + setupDataSource.connection.use { connection -> + connection.createStatement().use { statement -> + statement.executeUpdate("CREATE TABLE `$table` (`user` TEXT, `key` TEXT, `value` TEXT)") + statement.executeUpdate("INSERT INTO `$table` (`user`, `key`, `value`) VALUES ('player', 'score', 'old')") + statement.executeUpdate("INSERT INTO `$table` (`user`, `key`, `value`) VALUES ('player', 'score', 'new')") + } + } + val executor = Executors.newFixedThreadPool(2) + val start = CountDownLatch(1) + val futures = (0 until 2).map { + executor.submit { + start.await() + val dataSource = createDataSource(file.toPath()) + Database(TypeSQLite(file, table), dataSource) + } + } + + val databases = try { + start.countDown() + futures.map { it.get(30, TimeUnit.SECONDS) } + } finally { + executor.shutdownNow() + } + + assertEquals("new", databases.first()["player", "score"]) + assertEquals(1, countRows(setupDataSource, table, "player", "score")) + } + + @Test + fun `same named index on wrong columns does not bypass key constraint`() { + val table = "wrong_index" + val file = tempDir.resolve("wrong-index.db").toFile() + val dataSource = createDataSource(file.toPath()) + val expectedIndexName = uniqueIndexName(table) + dataSource.connection.use { connection -> + connection.createStatement().use { statement -> + statement.executeUpdate("CREATE TABLE `$table` (`user` TEXT, `key` TEXT, `value` TEXT)") + statement.executeUpdate("CREATE UNIQUE INDEX `$expectedIndexName` ON `$table` (`value`)") + } + } + + val database = Database(TypeSQLite(file, table), dataSource) + database["player", "score"] = "one" + database["player", "score"] = "two" + + assertEquals("two", database["player", "score"]) + assertEquals(1, countRows(dataSource, table, "player", "score")) + } + + @Test + fun `special table names are quoted for unique index creation`() { + val fixture = createFixture("special", "player-data") + + fixture.database["player", "score"] = "one" + fixture.database["player", "score"] = "two" + + assertEquals("two", fixture.database["player", "score"]) + assertEquals(1, countRows(fixture.dataSource, fixture.table, "player", "score")) + } + + @Test + fun `queued writes coalesce to latest value`() { + val fixture = createFixture("coalesced") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + container.asyncExecutor = { tasks.addLast(it) } + + container["score"] = "one" + container["score"] = "two" + + assertEquals(1, tasks.size) + tasks.removeFirst().invoke() + assertEquals("two", fixture.database["player", "score"]) + } + + @Test + fun `scheduler rejection does not hide a concurrent newer write`() { + val fixture = createFixture("scheduler-rejection") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + val schedulingStarted = CountDownLatch(1) + val releaseFirstScheduling = CountDownLatch(1) + val schedulingAttempts = AtomicInteger() + container.asyncExecutor = { task -> + if (schedulingAttempts.getAndIncrement() == 0) { + schedulingStarted.countDown() + releaseFirstScheduling.await() + throw RejectedExecutionException("first scheduling attempt rejected") + } + tasks.addLast(task) + } + val executor = Executors.newFixedThreadPool(2) + val secondStarted = CountDownLatch(1) + val first = executor.submit { + runCatching { container["state"] = "old" }.exceptionOrNull() + } + assertTrue(schedulingStarted.await(10, TimeUnit.SECONDS)) + val second = executor.submit { + secondStarted.countDown() + container["state"] = "new" + } + assertTrue(secondStarted.await(10, TimeUnit.SECONDS)) + releaseFirstScheduling.countDown() + + try { + assertTrue(first.get(10, TimeUnit.SECONDS) is RejectedExecutionException) + second.get(10, TimeUnit.SECONDS) + } finally { + executor.shutdownNow() + } + + assertEquals(1, tasks.size) + tasks.removeFirst().invoke() + assertEquals("new", fixture.database["player", "state"]) + } + + @Test + fun `delete supersedes queued save without null assertion failure`() { + val fixture = createFixture("delete") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + container.asyncExecutor = { tasks.addLast(it) } + + container["state"] = "present" + container["state"] = "" + + assertEquals(1, tasks.size) + tasks.removeFirst().invoke() + assertNull(fixture.database["player", "state"]) + } + + @Test + fun `delayed value is not persisted before its deadline`() { + val fixture = createFixture("delayed") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + container.asyncExecutor = { tasks.addLast(it) } + + container["state"] = "old" + container.setDelayed("state", "new", 1, TimeUnit.DAYS) + tasks.removeFirst().invoke() + container.checkUpdate() + + assertTrue(tasks.isEmpty()) + assertNull(fixture.database["player", "state"]) + assertEquals("new", container["state"]) + } + + @Test + fun `concurrent delayed writes retain the latest deadline state`() { + val fixture = createFixture("concurrent-delayed") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + container.asyncExecutor = { tasks.addLast(it) } + val executor = Executors.newFixedThreadPool(8) + val start = CountDownLatch(1) + val values = (1..128).map { "value-$it" } + val futures = values.mapIndexed { index, value -> + executor.submit { + start.await() + container.setDelayed("state", value, -index.toLong(), TimeUnit.MILLISECONDS) + } + } + + try { + start.countDown() + futures.forEach { it.get(10, TimeUnit.SECONDS) } + } finally { + executor.shutdownNow() + } + + container.checkUpdate() + assertEquals(1, tasks.size) + tasks.removeFirst().invoke() + assertEquals(container["state"], fixture.database["player", "state"]) + } + + @Test + fun `elapsed delayed value schedules one write`() { + val fixture = createFixture("elapsed") + val container = DataContainer("player", fixture.database) + val tasks = ArrayDeque<() -> Unit>() + container.asyncExecutor = { tasks.addLast(it) } + + container.setDelayed("state", "ready", 0, TimeUnit.MILLISECONDS) + container.checkUpdate() + + assertEquals(1, tasks.size) + tasks.removeFirst().invoke() + assertEquals("ready", fixture.database["player", "state"]) + } + + private fun createFixture(name: String, table: String = "${name}_player_data"): Fixture { + val file = tempDir.resolve("$name.db").toFile() + val dataSource = createDataSource(file.toPath()) + return Fixture(Database(TypeSQLite(file, table), dataSource), dataSource, table) + } + + private fun uniqueIndexName(tableName: String): String { + val normalizedName = tableName.replace(Regex("[^A-Za-z0-9_]"), "_").ifEmpty { "table" } + return "uk_${normalizedName.take(36)}_${Integer.toHexString(tableName.hashCode())}_user_key" + } + + private fun createDataSource(path: Path): SQLiteDataSource { + val config = SQLiteConfig().apply { + setBusyTimeout(30_000) + setJournalMode(SQLiteConfig.JournalMode.WAL) + setSynchronous(SQLiteConfig.SynchronousMode.NORMAL) + } + return SQLiteDataSource(config).apply { + url = "jdbc:sqlite:${path.toAbsolutePath()}" + } + } + + private fun countRows(dataSource: SQLiteDataSource, table: String, user: String, key: String): Int { + return dataSource.connection.use { connection -> + connection.prepareStatement("SELECT COUNT(*) FROM `$table` WHERE `user` = ? AND `key` = ?").use { statement -> + statement.setString(1, user) + statement.setString(2, key) + statement.executeQuery().use { result -> + result.next() + result.getInt(1) + } + } + } + } + + private data class Fixture( + val database: Database, + val dataSource: SQLiteDataSource, + val table: String, + ) +}