race condition fix

This commit is contained in:
2026-06-13 14:40:31 +03:00
parent 9c3548c6f2
commit 804203c698
98 changed files with 96 additions and 289 deletions
Vendored
BIN
View File
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
-1
View File
@@ -1,4 +1,3 @@
Manifest-Version: 1.0
Main-Class: MainKt
Multi-Release: true
Binary file not shown.
@@ -1,2 +1,2 @@
30
43
0
Binary file not shown.
-1
View File
@@ -1,4 +1,3 @@
Manifest-Version: 1.0
Main-Class: MainKt
Multi-Release: true
+13 -28
View File
@@ -1,9 +1,7 @@
import auth.JWTManager
import collection.CollectionManager
import commands.*
import database.CacheInvalidator
import database.DatabaseManager
import database.Repository
import commands.*
import mu.KotlinLogging
import network.NetworkManager
import runner.CommandInvoker
@@ -24,28 +22,16 @@ fun main(args: Array<String>) {
database.start()
val repository = Repository(jwtManager)
val manager = CollectionManager()
manager.loadFromFile(repository.read())
// Синхронизация кэша между серверами через PG NOTIFY
val cacheInvalidator = CacheInvalidator(database.dataSource) {
manager.loadFromFile(repository.read())
}
cacheInvalidator.start()
val invoker = CommandInvoker(jwtManager)
registerServerCommands(invoker, manager, repository)
registerServerCommands(invoker, repository)
// (Ctrl+C or kill)
val server = NetworkManager(port = port, invoker = invoker, repository = repository)
Runtime.getRuntime().addShutdownHook(Thread {
logger.info{ "Exit..." }
cacheInvalidator.stop()
server.stop()
})
// Server-side console
Thread {
val scanner = java.util.Scanner(System.`in`)
logger.info{ "[Server-console] Server-side commands: exit" }
@@ -69,23 +55,22 @@ private fun loadDotEnv() {
fun registerServerCommands(
invoker: CommandInvoker,
manager: CollectionManager,
repository: Repository
) {
listOf(
InfoCommand(manager),
ShowCommand(manager),
AddCommand(manager, repository),
UpdateCommand(manager, repository),
RemoveByIdCommand(manager, repository),
ClearCommand(manager, repository),
AddIfMaxCommand(manager, repository),
AddIfMinCommand(manager, repository),
InfoCommand(repository),
ShowCommand(repository),
AddCommand(repository),
UpdateCommand(repository),
RemoveByIdCommand(repository),
ClearCommand(repository),
AddIfMaxCommand(repository),
AddIfMinCommand(repository),
HistoryCommand(invoker),
SyncCommand(invoker),
SumOfMinutesCommand(manager),
MinByNameCommand(manager),
PrintDescendingMinutesCommand(manager),
SumOfMinutesCommand(repository),
MinByNameCommand(repository),
PrintDescendingMinutesCommand(repository),
).forEach { invoker.register(it)}
logger.info{ " [startup] Registered commands: ${invoker.getCommands().keys}" }
}
@@ -1,106 +0,0 @@
package collection
import models.HumanBeing
import mu.KotlinLogging
import java.time.LocalDate
import java.util.Collections
import java.util.TreeSet
import java.util.UUID
private val logger = KotlinLogging.logger {}
class CollectionManager {
private val collection: MutableSet<HumanBeing> =
Collections.synchronizedSortedSet(TreeSet())
val initDate: LocalDate = LocalDate.now()
fun add(humanBeing: HumanBeing): Boolean {
logger.info { "[ADD] Element created ${humanBeing.id}" }
return collection.add(humanBeing)
}
fun update(id: UUID, updated: HumanBeing): Boolean {
synchronized(collection) {
val old = getById(id) ?: return false
collection.remove(old)
val replaced = updated.copy(id = old.id, creationDate = old.creationDate)
logger.info { "[UPDATE] Element updated ${old.id}" }
return collection.add(replaced)
}
}
fun removeById(id: UUID): Boolean {
val target = getById(id) ?: return false
logger.info { "[REMOVE] Element deleted ${target.id}" }
return collection.remove(target)
}
fun clear() {
collection.clear()
logger.info { "[CLEAR] Collection cleared" }
}
fun getAll(): TreeSet<HumanBeing> {
synchronized(collection) {
return TreeSet(collection)
}
}
fun getById(id: UUID): HumanBeing? {
synchronized(collection) {
return collection.firstOrNull { it.id == id }
}
}
fun size(): Int = collection.size
fun isEmpty(): Boolean = collection.isEmpty()
fun getMax(): HumanBeing? {
synchronized(collection) {
return collection.maxWithOrNull(Comparator.naturalOrder())
}
}
fun getMin(): HumanBeing? {
synchronized(collection) {
return collection.minWithOrNull(Comparator.naturalOrder())
}
}
fun sumOfMinutesOfWaiting(): Double {
synchronized(collection) {
return collection.sumOf { it.minutesOfWaiting.toDouble() }
}
}
fun minByName(): HumanBeing? {
synchronized(collection) {
return collection.minByOrNull { it.name }
}
}
fun getMinutesOfWaitingDescending(): List<Float> {
synchronized(collection) {
return collection.map { it.minutesOfWaiting }.sortedDescending()
}
}
fun loadFromFile(items: List<HumanBeing>) {
synchronized(collection) {
collection.clear()
collection.addAll(items)
}
logger.info { "[LOAD] loaded ${items.size} elements" }
}
fun getInfo(): String =
"""
|Тип коллекции : ${collection::class.simpleName}
|Тип элементов : ${HumanBeing::class.simpleName}
|Дата инициал. : $initDate
|Кол-во элемен.: ${collection.size}
""".trimMargin()
}
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import java.util.UUID
@Serializable
class AddCommand(
@Contextual private val manager: CollectionManager,
@Contextual private val repository: Repository
) : Command {
override val name = "add"
@@ -23,11 +21,6 @@ class AddCommand(
val secured = humanBeing.copy(ownerId = UUID.fromString(userId))
secured.id = repository.write(secured)
secured.id ?: return Response(false, "Не удалось добавить элемент")
return if (manager.add(secured))
Response(true, "Элемент добавлен: ${secured.name} [${secured.id}]")
else
Response(false, "Элемент уже существует в коллекции.")
return Response(true, "Элемент добавлен: ${secured.name} [${secured.id}]")
}
}
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import java.util.UUID
@Serializable
class AddIfMaxCommand(
@Contextual val manager: CollectionManager,
@Contextual private val repository: Repository
) : Command {
override val name = "add_if_max"
@@ -21,11 +19,10 @@ class AddIfMaxCommand(
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response {
humanBeing ?: return Response(false, "[Ошибка] Объект HumanBeing не передан.")
val secured = humanBeing.copy(ownerId = UUID.fromString(userId))
val max = manager.getMax()
val max = repository.getMax()
return if (max == null || secured > max) {
secured.id = repository.write(secured)
secured.id ?: return Response(false, "Не удалось добавить элемент")
manager.add(secured)
Response(true, "Элемент добавлен (превышает максимум).")
} else {
Response(false, "Элемент не добавлен: не превышает текущий максимум (${max.name}).")
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import models.HumanBeing
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import java.util.UUID
@Serializable
class AddIfMinCommand(
@Contextual private val manager: CollectionManager,
@Contextual private val repository: Repository
) : Command {
override val name = "add_if_min"
@@ -21,11 +19,10 @@ class AddIfMinCommand(
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response {
humanBeing ?: return Response(false, "[Ошибка] Объект HumanBeing не передан.")
val secured = humanBeing.copy(ownerId = UUID.fromString(userId))
val min = manager.getMin()
val min = repository.getMin()
return if (min == null || secured < min) {
secured.id = repository.write(secured)
secured.id ?: return Response(false, "Не удалось добавить элемент")
manager.add(secured)
Response(true, "Элемент добавлен (меньше минимума).")
} else {
Response(false, "Элемент не добавлен: не меньше текущего минимума (${min.name}).")
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import java.util.UUID
@Serializable
class ClearCommand(
@Contextual private val manager: CollectionManager,
@Contextual private var repository: Repository
) : Command {
override val name = "clear"
@@ -23,9 +21,6 @@ class ClearCommand(
val ownerId = UUID.fromString(userId)
val deletedCount = repository.deleteByOwner(ownerId)
val toRemove = manager.getAll().filter { it.ownerId == ownerId }
toRemove.forEach { manager.removeById(it.id!!) }
return Response(true, "Удалено объектов: $deletedCount.")
}
}
@@ -1,18 +1,18 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
import models.HumanBeing
import network.Response
@Serializable
class InfoCommand(@Contextual private val manager: CollectionManager) : Command {
class InfoCommand(@Contextual private val repository: Repository) : Command {
override val name = "info"
override val description = "вывести информацию о коллекции"
override val type = CommandType.SIMPLE
override val args = listOf(CommandArgType.NONE)
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response =
Response(true, manager.getInfo())
Response(true, repository.getInfo())
}
@@ -1,21 +1,18 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
import models.HumanBeing
import network.Response
/**
* Выводит любой элемент с минимальным значением поля [HumanBeing.name].
*/
@Serializable
class MinByNameCommand(@Contextual private val manager: CollectionManager) : Command {
class MinByNameCommand(@Contextual private val repository: Repository) : Command {
override val name = "min_by_name"
override val description = "вывести элемент с минимальным именем"
override val type = CommandType.SIMPLE
override val args = listOf(CommandArgType.NONE)
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response =
Response(true, manager.minByName()?.toString() ?: "Коллекция пуста.")
Response(true, repository.minByName()?.toString() ?: "Коллекция пуста.")
}
@@ -1,23 +1,20 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
import models.HumanBeing
import network.Response
/**
* Выводит значения поля [models.HumanBeing.minutesOfWaiting] в порядке убывания.
*/
@Serializable
class PrintDescendingMinutesCommand(@Contextual private val manager: CollectionManager) : Command {
class PrintDescendingMinutesCommand(@Contextual private val repository: Repository) : Command {
override val name = "print_field_descending_minutes_of_waiting"
override val description = "вывести minutesOfWaiting в порядке убывания"
override val type = CommandType.SIMPLE
override val args = listOf(CommandArgType.NONE)
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response {
val values = manager.getMinutesOfWaitingDescending()
val values = repository.getMinutesOfWaitingDescending()
return if (values.isEmpty()) Response(true, "Коллекция пуста.")
else Response(true, values.joinToString("\n"))
}
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import network.Response
@Serializable
class RemoveByIdCommand(
@Contextual private val manager: CollectionManager,
@Contextual private val repository: Repository
) : Command {
override val name = "remove_by_id"
@@ -27,17 +25,16 @@ class RemoveByIdCommand(
return Response(false, "[Ошибка] Некорректный UUID: '${args[0]}'")
}
val existing = manager.getById(id)
val existing = repository.getById(id)
?: return Response(false, "Элемент не найден.")
if (existing.ownerId.toString() != userId)
return Response(false, "Нет прав: объект принадлежит другому пользователю.")
if (repository.delete(id)) {
manager.removeById(id)
return Response(true, "Элемент удалён.")
return if (repository.delete(id)) {
Response(true, "Элемент удалён.")
} else {
return Response(false, "[Ошибка] Элемент с id '$id' не найден и не был удалён.")
Response(false, "[Ошибка] Элемент с id '$id' не найден и не был удалён.")
}
}
}
@@ -1,20 +1,21 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
import models.HumanBeing
import network.Response
@Serializable
class ShowCommand(@Contextual private val manager: CollectionManager) : Command {
class ShowCommand(@Contextual private val repository: Repository) : Command {
override val name = "show"
override val description = "вывести все элементы коллекции"
override val type = CommandType.SIMPLE
override val args = listOf(CommandArgType.NONE)
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response {
if (manager.isEmpty()) return Response(true, "Коллекция пуста.")
return Response(true, manager.getAll().joinToString("\n") { it.toString() })
val all = repository.getAll()
if (all.isEmpty()) return Response(true, "Коллекция пуста.")
return Response(true, all.joinToString("\n") { it.toString() })
}
}
@@ -1,21 +1,18 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
import models.HumanBeing
import network.Response
/**
* Выводит сумму значений поля [models.HumanBeing.minutesOfWaiting].
*/
@Serializable
class SumOfMinutesCommand(@Contextual private val manager: CollectionManager) : Command {
class SumOfMinutesCommand(@Contextual private val repository: Repository) : Command {
override val name = "sum_of_minutes_of_waiting"
override val description = "вывести сумму minutesOfWaiting всех элементов"
override val type = CommandType.SIMPLE
override val args = listOf(CommandArgType.NONE)
override fun execute(args: List<String>, humanBeing: HumanBeing?, userId: String): Response =
Response(true, "Сумма minutesOfWaiting: ${manager.sumOfMinutesOfWaiting()}")
Response(true, "Сумма minutesOfWaiting: ${repository.sumOfMinutesOfWaiting()}")
}
@@ -1,6 +1,5 @@
package commands
import collection.CollectionManager
import database.Repository
import kotlinx.serialization.Contextual
import kotlinx.serialization.Serializable
@@ -10,7 +9,6 @@ import java.util.UUID
@Serializable
class UpdateCommand(
@Contextual private val manager: CollectionManager,
@Contextual private val repository: Repository
) : Command {
override val name = "update"
@@ -28,17 +26,16 @@ class UpdateCommand(
return Response(false, "[Ошибка] Некорректный UUID: '${args[0]}'")
}
val existing = manager.getById(id)
val existing = repository.getById(id)
?: return Response(false, "Элемент не найден.")
if (existing.ownerId.toString() != userId)
return Response(false, "Нет прав: объект принадлежит другому пользователю.")
if (repository.update(id, humanBeing)) {
manager.update(id, humanBeing)
return Response(true, "Элемент обновлён.")
return if (repository.update(id, humanBeing)) {
Response(true, "Элемент обновлён.")
} else {
return Response(false, "[Ошибка] Элемент не найден и не был обновлён.")
Response(false, "[Ошибка] Элемент не найден и не был обновлён.")
}
}
}
@@ -1,57 +0,0 @@
package database
import io.github.oshai.kotlinlogging.KotlinLogging
import org.postgresql.PGConnection
import javax.sql.DataSource
private val logger = KotlinLogging.logger {}
class CacheInvalidator(
private val dataSource: DataSource,
private val onInvalidated: () -> Unit
) {
@Volatile
private var running = false
private var thread: Thread? = null
fun start() {
running = true
thread = Thread {
while (running) {
try {
dataSource.connection.use { conn ->
conn.autoCommit = true
val pgConn = conn.unwrap(PGConnection::class.java)
conn.createStatement().use { stmt ->
stmt.execute("LISTEN cache_changed")
}
logger.info { "Listening for cache changes via PG NOTIFY..." }
while (running && !conn.isClosed) {
val notifications = pgConn.notifications
if (notifications != null && notifications.isNotEmpty()) {
logger.debug { "Cache invalidation received" }
onInvalidated()
}
Thread.sleep(500)
}
}
} catch (e: Exception) {
if (!running) break
logger.warn(e) { "Cache listener error, reconnecting in 5s..." }
Thread.sleep(5000)
}
}
}.also { it.isDaemon = true }.apply {
name = "pg-notify-listener"
start()
}
}
fun stop() {
running = false
thread?.interrupt()
}
}
+51 -32
View File
@@ -12,6 +12,7 @@ import models.HumanBeing
import org.jetbrains.exposed.v1.core.*
import org.jetbrains.exposed.v1.jdbc.*
import org.jetbrains.exposed.v1.jdbc.transactions.transaction
import java.time.LocalDate
import java.util.UUID
private val logger = KotlinLogging.logger {}
@@ -19,48 +20,54 @@ private val logger = KotlinLogging.logger {}
class Repository(
private val jwtManager: JWTManager,
) {
val initDate: LocalDate = LocalDate.now()
fun read(): List<HumanBeing> {
fun getAll(): List<HumanBeing> {
val beings = mutableListOf<HumanBeing>()
transaction {
val humans = HumanBeingTable
.selectAll()
val humans = HumanBeingTable.selectAll()
for (human in humans) {
logger.trace { "read. $human" }
logger.trace { "getAll. $human" }
val cord = CoordinatesTable
.select(CoordinatesTable.x, CoordinatesTable.y)
.where(CoordinatesTable.id eq human[HumanBeingTable.coordinates])
.single()
val car = CarTable
.select(CarTable.cool)
.where(CarTable.id eq human[HumanBeingTable.car])
.single()
val newCords = Coordinates(cord[CoordinatesTable.x]!!, cord[CoordinatesTable.y]!!)
val newCar = Car(car[CarTable.cool]!!)
val newBeing = HumanBeing(
name = human[HumanBeingTable.name],
ownerId = human[HumanBeingTable.owner_id].value,
coordinates = newCords,
car = newCar,
id = human[HumanBeingTable.id].value,
creationDate = human[HumanBeingTable.creationDate],
realHero = human[HumanBeingTable.realHero],
hasToothpick = human[HumanBeingTable.hasToothpick],
impactSpeed = human[HumanBeingTable.impactSpeed],
soundtrackName = human[HumanBeingTable.soundtrackName],
minutesOfWaiting = human[HumanBeingTable.minutesOfWaiting],
weaponType = human[HumanBeingTable.weaponType]
)
beings.add(newBeing)
beings.add(toHumanBeing(human, cord, car))
}
}
return beings
}
fun getById(id: UUID): HumanBeing? = getAll().firstOrNull { it.id == id }
fun getMax(): HumanBeing? = getAll().maxWithOrNull(Comparator.naturalOrder())
fun getMin(): HumanBeing? = getAll().minWithOrNull(Comparator.naturalOrder())
fun sumOfMinutesOfWaiting(): Double =
getAll().sumOf { it.minutesOfWaiting.toDouble() }
fun minByName(): HumanBeing? = getAll().minByOrNull { it.name }
fun getMinutesOfWaitingDescending(): List<Float> =
getAll().map { it.minutesOfWaiting }.sortedDescending()
fun size(): Int = getAll().size
fun isEmpty(): Boolean = getAll().isEmpty()
fun getInfo(): String =
"""
|Тип коллекции : Реляционная БД (PostgreSQL)
|Тип элементов : ${HumanBeing::class.simpleName}
|Дата инициал. : $initDate
|Кол-во элемен.: ${size()}
""".trimMargin()
fun write(item: HumanBeing): UUID? {
var humanId: UUID? = null
try {
@@ -69,11 +76,9 @@ class Repository(
it[x] = item.coordinates.x
it[y] = item.coordinates.y
}
val cars = CarTable.insertAndGetId {
it[cool] = item.car.cool
}
humanId = HumanBeingTable.insertAndGetId {
it[owner_id] = item.ownerId
it[name] = item.name
@@ -87,9 +92,7 @@ class Repository(
it[car] = cars
it[weaponType] = item.weaponType
}.value
exec("NOTIFY cache_changed")
}
} catch (e: Exception) {
logger.error(e) { "write failed. Here what's going on: \n $e \n " }
return humanId
@@ -137,7 +140,6 @@ class Repository(
it[hasToothpick] = item.hasToothpick
it[weaponType] = item.weaponType
}
exec("NOTIFY cache_changed")
allGood = true
}
}
@@ -157,7 +159,6 @@ class Repository(
HumanBeingTable.deleteWhere { HumanBeingTable.id eq id }
CoordinatesTable.deleteWhere { CoordinatesTable.id eq refs[HumanBeingTable.coordinates] }
CarTable.deleteWhere { CarTable.id eq refs[HumanBeingTable.car] }
exec("NOTIFY cache_changed")
} else {
allGood = false
}
@@ -221,8 +222,26 @@ class Repository(
deletedCount = HumanBeingTable.deleteWhere { HumanBeingTable.owner_id eq ownerId }
CoordinatesTable.deleteWhere { CoordinatesTable.id inList coordIds }
CarTable.deleteWhere { CarTable.id inList carIds }
exec("NOTIFY cache_changed")
}
return deletedCount
}
private fun toHumanBeing(
human: ResultRow,
cord: ResultRow,
car: ResultRow
): HumanBeing = HumanBeing(
name = human[HumanBeingTable.name],
ownerId = human[HumanBeingTable.owner_id].value,
coordinates = Coordinates(cord[CoordinatesTable.x]!!, cord[CoordinatesTable.y]!!),
car = Car(car[CarTable.cool]!!),
id = human[HumanBeingTable.id].value,
creationDate = human[HumanBeingTable.creationDate],
realHero = human[HumanBeingTable.realHero],
hasToothpick = human[HumanBeingTable.hasToothpick],
impactSpeed = human[HumanBeingTable.impactSpeed],
soundtrackName = human[HumanBeingTable.soundtrackName],
minutesOfWaiting = human[HumanBeingTable.minutesOfWaiting],
weaponType = human[HumanBeingTable.weaponType]
)
}