Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion gradle.properties
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
kotlin.stdlib.default.dependency=false
org.gradle.parallel=true
version=4.1.7
version=4.1.8
org.gradle.jvmargs=-Xmx4G
2 changes: 1 addition & 1 deletion gradle/wrapper/gradle-wrapper.properties
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-9.7.1-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-9.8.0-bin.zip
networkTimeout=10000
retries=0
retryBackOffMs=500
Expand Down
1 change: 1 addition & 0 deletions surf-transaction-api/api/surf-transaction-api.api
Original file line number Diff line number Diff line change
Expand Up @@ -1025,6 +1025,7 @@ public abstract interface class dev/slne/surf/transaction/api/user/TransactionUs
public static synthetic fun beginTransfer-SBKQj6I$default (Ldev/slne/surf/transaction/api/user/TransactionUser;Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;Ldev/slne/surf/transaction/api/account/Account;JZZLjava/util/Set;Ljava/util/Set;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
public fun beginWithdrawal-WPi__2c (Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;JZ[Ldev/slne/surf/transaction/api/transaction/data/TransactionData;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static synthetic fun beginWithdrawal-WPi__2c$default (Ldev/slne/surf/transaction/api/user/TransactionUser;Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;JZ[Ldev/slne/surf/transaction/api/transaction/data/TransactionData;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
public abstract fun cachedBalance (Ldev/slne/surf/transaction/api/currency/Currency;)Ljava/math/BigDecimal;
public fun deposit (Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;Z[Ldev/slne/surf/transaction/api/transaction/data/TransactionData;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static synthetic fun deposit$default (Ldev/slne/surf/transaction/api/user/TransactionUser;Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;Z[Ldev/slne/surf/transaction/api/transaction/data/TransactionData;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
public fun transfer (Ljava/math/BigDecimal;Ldev/slne/surf/transaction/api/currency/Currency;Ldev/slne/surf/transaction/api/account/Account;ZZLjava/util/Set;Ljava/util/Set;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -434,6 +434,20 @@ interface TransactionUser : Transactional, AccountAccess {
currency: Currency
): BigDecimal = balance(getDefaultAccount(), currency)

/**
* Returns the locally cached balance of the user's default account in the specified [currency]
* without suspending.
*
* The cached balance is re-queried from the database whenever a transaction changes it.
* If it is not cached yet, loading starts in the background and `null` is returned.
*
* @param currency the currency of the balance
* @return the cached balance, or `null` if it is not cached yet
*/
fun cachedBalance(
currency: Currency
): BigDecimal?

@OptIn(InternalTransactionApi::class)
companion object {
/**
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package dev.slne.surf.transaction.core.client

import dev.slne.surf.rabbitmq.api.ClientRabbitMQApi
import dev.slne.surf.transaction.core.client.balance.BalanceCache
import dev.slne.surf.transaction.core.client.currency.CurrencyServiceImpl
import dev.slne.surf.transaction.core.client.redis.RedisService
import dev.slne.surf.transaction.core.common.CoreTransactionSerializerModule
Expand Down Expand Up @@ -31,6 +32,7 @@ abstract class ClientTransactionalInstance : TransactionInstance() {
super.disable()

CurrencyServiceImpl.INSTANCE.disposeScope()
BalanceCache.disposeScope()
withContext(Dispatchers.IO) { RedisService.INSTANCE.disconnect() }
rabbitApi.disconnect()
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
package dev.slne.surf.transaction.core.client.balance

import com.github.benmanes.caffeine.cache.AsyncLoadingCache
import com.github.benmanes.caffeine.cache.Cache
import com.github.benmanes.caffeine.cache.Caffeine
import dev.slne.surf.api.core.util.logger
import dev.slne.surf.transaction.api.currency.Currency
import dev.slne.surf.transaction.api.transaction.Transaction
import dev.slne.surf.transaction.core.client.account.AccountServiceImpl
import dev.slne.surf.transaction.core.client.rabbitApi
import dev.slne.surf.transaction.core.client.redis.RedisService
import dev.slne.surf.transaction.core.client.redis.events.balance.BalanceChangedEvent
import dev.slne.surf.transaction.core.common.protocol.transaction.balance.GetTransactionBalanceRequestPacket
import kotlinx.coroutines.*
import kotlinx.coroutines.future.await
import kotlinx.coroutines.future.future
import java.math.BigDecimal
import java.util.*
import java.util.concurrent.CompletableFuture
import kotlin.time.Duration.Companion.minutes
import kotlin.time.toJavaDuration

/**
* Locally cached balances of default accounts, readable without suspending.
*
* Cached balances are never recalculated locally: whenever a transaction touches an account,
* the balance is queried from the microservice again (see [balancesChanged]).
*/
object BalanceCache {
private val log = logger()
private val expireAfterAccess = 10.minutes.toJavaDuration()

private val scope =
CoroutineScope(Dispatchers.IO + SupervisorJob() + CoroutineName("surf-transaction-balance-cache") + CoroutineExceptionHandler { context, throwable ->
log.atSevere()
.withCause(throwable)
.log("Unhandled exception in ${context[CoroutineName]}")
})

private val defaultAccountIds: AsyncLoadingCache<UUID, UUID> = Caffeine.newBuilder()
.expireAfterAccess(expireAfterAccess)
.buildAsync { playerUuid, _ ->
scope.future { AccountServiceImpl.INSTANCE.getDefaultAccount(playerUuid).accountId }
}

private val balances: Cache<BalanceKey, CachedBalance> = Caffeine.newBuilder()
.expireAfterAccess(expireAfterAccess)
.build()

/**
* Returns the cached balance of the default account of [playerUuid] in [currency].
*
* Returns `null` while the balance is not cached yet and starts loading it in the background.
*/
fun cachedBalance(playerUuid: UUID, currency: Currency): BigDecimal? {
val accountId = defaultAccountIds.get(playerUuid).getNowOrNull() ?: return null
val key = BalanceKey(accountId, currency.name)

return balances.get(key) { CachedBalance().also { refresh(key, it) } }.value
}

/**
* Re-queries the cached balances touched by [transactions] on this server and notifies all
* other servers to do the same.
*/
fun balancesChanged(transactions: Collection<Transaction>) {
if (transactions.isEmpty()) return

transactions.groupBy { it.currency.name }.forEach { (currencyName, changed) ->
val accountIds = changed
.flatMap { listOfNotNull(it.senderAccountId, it.receiverAccountId) }
.toSet()

refreshAccounts(accountIds, currencyName)

scope.launch {
RedisService.publish(BalanceChangedEvent(accountIds, currencyName)).await()
}
}
}

/**
* Re-queries the balances of [accountIds] in [currencyName] that are cached on this server.
*/
fun refreshAccounts(accountIds: Collection<UUID>, currencyName: String) {
for (accountId in accountIds) {
val key = BalanceKey(accountId, currencyName)
val entry = balances.policy().getIfPresentQuietly(key) ?: continue

refresh(key, entry)
}
}

fun disposeScope() {
scope.cancel("Disposing BalanceCache scope")
}

private fun refresh(key: BalanceKey, entry: CachedBalance) {
val version = entry.nextVersion()

scope.launch {
try {
val request = GetTransactionBalanceRequestPacket(key.accountId, key.currencyName)
val (balance) = rabbitApi.sendRequest(request)

entry.update(version, balance)
} catch (cause: CancellationException) {
throw cause
} catch (cause: Throwable) {
log.atWarning()
.withCause(cause)
.log("Failed to refresh balance of account ${key.accountId} in ${key.currencyName}")

// Never loaded: drop the entry so the next access retries instead of staying empty.
if (entry.value == null) {
balances.asMap().remove(key, entry)
}
}
}
}

private fun <T : Any> CompletableFuture<T>.getNowOrNull(): T? =
if (isDone && !isCompletedExceptionally) join() else null

private data class BalanceKey(val accountId: UUID, val currencyName: String)

private class CachedBalance {
@Volatile
var value: BigDecimal? = null
private set

private var requestedVersion = 0L
private var appliedVersion = 0L

@Synchronized
fun nextVersion() = ++requestedVersion

/** Applies [balance] unless a refresh that started later has already been applied. */
@Synchronized
fun update(version: Long, balance: BigDecimal) {
if (version < appliedVersion) return

appliedVersion = version
value = balance
}
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
package dev.slne.surf.transaction.core.client.balance

import dev.slne.surf.redis.event.OnRedisEvent
import dev.slne.surf.transaction.core.client.redis.events.balance.BalanceChangedEvent

class BalanceEventsListener {

@OnRedisEvent
fun onBalanceChanged(event: BalanceChangedEvent) {
if (event.originatesFromThisClient()) {
return
}

BalanceCache.refreshAccounts(event.accountIds, event.currencyName)
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ object TransactionPermissions {
const val TRANSACTION_ADMIN_ADD = "$TRANSACTION_ADMIN.add"
const val TRANSACTION_ADMIN_REMOVE = "$TRANSACTION_ADMIN.remove"

fun TRANSACTION_ADMIN_CURRENCY(currencyName: String) = "$TRANSACTION_ADMIN.$currencyName"

const val ACCOUNT = "$PREFIX.account"
const val ACCOUNT_INFO = "$ACCOUNT.info"
const val ACCOUNT_LIST = "$ACCOUNT.list"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import dev.slne.surf.api.core.util.requiredService
import dev.slne.surf.redis.RedisApi
import dev.slne.surf.redis.event.RedisEvent
import dev.slne.surf.transaction.api.account.AccountService
import dev.slne.surf.transaction.core.client.balance.BalanceEventsListener
import dev.slne.surf.transaction.core.client.currency.CurrencyEventsListener
import org.jetbrains.annotations.Blocking
import org.jetbrains.annotations.MustBeInvokedByOverriders
Expand All @@ -29,6 +30,7 @@ abstract class RedisService {
@MustBeInvokedByOverriders
protected open fun register() {
redisApi.subscribeToEvents(CurrencyEventsListener())
redisApi.subscribeToEvents(BalanceEventsListener())
}

companion object {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
package dev.slne.surf.transaction.core.client.redis.events.balance

import dev.slne.surf.api.core.serializer.java.uuid.SerializableUUID
import dev.slne.surf.redis.event.RedisEvent
import kotlinx.serialization.Serializable

/**
* Published after a transaction changed the balance of [accountIds] in the currency [currencyName],
* so every server re-queries the balances it has cached for those accounts.
*/
@Serializable
class BalanceChangedEvent(
val accountIds: Set<SerializableUUID>,
val currencyName: String
) : RedisEvent()
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import dev.slne.surf.transaction.api.transaction.TransactionRollbackResult
import dev.slne.surf.transaction.api.transaction.TransactionResult
import dev.slne.surf.transaction.api.transaction.TransactionService
import dev.slne.surf.transaction.api.transaction.data.TransactionData
import dev.slne.surf.transaction.core.client.balance.BalanceCache
import dev.slne.surf.transaction.core.client.rabbitApi
import dev.slne.surf.transaction.core.common.protocol.transaction.balance.GetTransactionBalanceRequestPacket
import dev.slne.surf.transaction.core.common.protocol.transaction.create.CreateTransactionRequestPacket
Expand Down Expand Up @@ -51,7 +52,7 @@ class TransactionServiceImpl : CoreTransactionService {

return pendingRpcRequest(setOf(transaction.identifier)) {
transactionRpc.beginTransaction(transaction, timeoutMillis)
}
}.also { BalanceCache.balancesChanged(it.transactions) }
}

override suspend fun beginWithdrawal(
Expand All @@ -76,7 +77,7 @@ class TransactionServiceImpl : CoreTransactionService {

return pendingRpcRequest(setOf(transaction.identifier)) {
transactionRpc.beginTransaction(transaction, timeoutMillis)
}
}.also { BalanceCache.balancesChanged(it.transactions) }
}

override suspend fun beginTransfer(
Expand Down Expand Up @@ -113,7 +114,7 @@ class TransactionServiceImpl : CoreTransactionService {
receiverTransaction,
timeoutMillis
)
}
}.also { BalanceCache.balancesChanged(it.transactions) }
}

override suspend fun deposit(
Expand All @@ -134,6 +135,7 @@ class TransactionServiceImpl : CoreTransactionService {
)

val (result) = rabbitApi.sendRequest(CreateTransactionRequestPacket(transaction))
BalanceCache.balancesChanged(result.transactions)
return result
}

Expand All @@ -155,6 +157,7 @@ class TransactionServiceImpl : CoreTransactionService {
)

val (result) = rabbitApi.sendRequest(CreateTransactionRequestPacket(transaction))
BalanceCache.balancesChanged(result.transactions)
return result
}

Expand Down Expand Up @@ -184,6 +187,7 @@ class TransactionServiceImpl : CoreTransactionService {
val (result) = rabbitApi.sendRequest(
TransferTransactionCreateRequestPacket(senderTransaction, receiverTransaction)
)
BalanceCache.balancesChanged(result.transactions)
return result
}

Expand All @@ -197,8 +201,11 @@ class TransactionServiceImpl : CoreTransactionService {
return balance
}

override suspend fun commit(identifier: UUID): TransactionCommitResult = transactionRpc.commit(identifier)
override suspend fun rollback(identifier: UUID): TransactionRollbackResult = transactionRpc.rollback(identifier)
override suspend fun commit(identifier: UUID): TransactionCommitResult =
transactionRpc.commit(identifier).also { BalanceCache.balancesChanged(it.transactions) }

override suspend fun rollback(identifier: UUID): TransactionRollbackResult =
transactionRpc.rollback(identifier).also { BalanceCache.balancesChanged(it.transactions) }
override suspend fun find(identifier: UUID): Transaction? = transactionRpc.find(identifier)

private fun createTransaction(
Expand Down Expand Up @@ -271,3 +278,35 @@ class TransactionServiceImpl : CoreTransactionService {
val INSTANCE get() = TransactionService.INSTANCE as TransactionServiceImpl
}
}

/** Transactions whose accounts' balances changed through this result. */
private val TransactionResult.transactions: List<Transaction>
get() = when (this) {
is TransactionResult.Success -> listOf(transaction)
is TransactionResult.TransferSuccess -> listOf(senderTransaction, receiverTransaction)
else -> emptyList()
}

/** Transactions whose accounts' available balances changed by reserving them. */
private val PendingTransactionResult.transactions: List<Transaction>
get() = when (this) {
is PendingTransactionResult.Created -> listOf(transaction)
is PendingTransactionResult.TransferCreated -> listOf(senderTransaction, receiverTransaction)
else -> emptyList()
}

/** Transactions whose accounts' balances changed by committing or expiring them. */
private val TransactionCommitResult.transactions: List<Transaction>
get() = when (this) {
is TransactionCommitResult.Committed -> listOf(transaction)
is TransactionCommitResult.Expired -> listOf(transaction)
else -> emptyList()
}

/** Transactions whose accounts' available balances changed by releasing their reservation. */
private val TransactionRollbackResult.transactions: List<Transaction>
get() = when (this) {
is TransactionRollbackResult.RolledBack -> listOf(transaction)
is TransactionRollbackResult.Expired -> listOf(transaction)
else -> emptyList()
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
package dev.slne.surf.transaction.core.client.user

import dev.slne.surf.transaction.api.currency.Currency
import dev.slne.surf.transaction.core.client.balance.BalanceCache
import dev.slne.surf.transaction.core.common.user.TransactionUserImpl
import java.util.*

class ClientTransactionUserImpl(userUuid: UUID) : TransactionUserImpl(userUuid) {
override fun cachedBalance(currency: Currency) = BalanceCache.cachedBalance(userUuid, currency)
}
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,9 @@ package dev.slne.surf.transaction.core.client.user

import com.google.auto.service.AutoService
import dev.slne.surf.transaction.api.user.TransactionUserService
import dev.slne.surf.transaction.core.common.user.TransactionUserImpl
import java.util.UUID

@AutoService(TransactionUserService::class)
class TransactionUserServiceImpl : TransactionUserService {
override fun byUuid(uuid: UUID) = TransactionUserImpl(uuid)
override fun byUuid(uuid: UUID) = ClientTransactionUserImpl(uuid)
}
Loading
Loading