admin/eagle0
1
0
Fork 0
mirror of https://github.com/nolen777/eagle0.git synced 2026-07-28 20:55:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid state replay for chronicle events (#8730)

Browse Source
This commit is contained in:
admin
2026-07-20 07:42:07 -07:00
committed by GitHub
parent bedc104d0b
commit 741bc690fb
7 changed files with 167 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/scala/net/eagle0/eagle/library/BUILD.bazel
+2
View File
@@ -164,12 +164,14 @@ scala_library(
],
exports = [
"//src/main/scala/net/eagle0/eagle/library/actions/applier:action_result_applier",
"//src/main/scala/net/eagle0/eagle/model/action_result:action_result_trait",
"//src/main/scala/net/eagle0/eagle/model/state/date",
"//src/main/scala/net/eagle0/eagle/model/state/game_state",
],
deps = [
"//src/main/scala/net/eagle0/eagle:eagle_pkg",
"//src/main/scala/net/eagle0/eagle/library/actions/applier:action_result_applier",
"//src/main/scala/net/eagle0/eagle/model/action_result:action_result_trait",
"//src/main/scala/net/eagle0/eagle/model/state/date",
"//src/main/scala/net/eagle0/eagle/model/state/game_state",
],
src/main/scala/net/eagle0/eagle/library/GameHistory.scala
+19 -1
View File
@@ -2,10 +2,19 @@ package net.eagle0.eagle.library
import scala.annotation.unused
import net.eagle0.eagle.{FactionId, RoundId}
import net.eagle0.eagle.library.actions.applier.ActionResultWithResultingState
import net.eagle0.eagle.model.action_result.types.ActionResultType
import net.eagle0.eagle.model.action_result.NotificationT
import net.eagle0.eagle.model.state.date.Date
import net.eagle0.eagle.model.state.game_state.GameState
import net.eagle0.eagle.RoundId
case class ChronicleEventSource(
date: Date,
actionResultType: ActionResultType,
newNotifications: Vector[NotificationT],
removedFactionIds: Vector[FactionId]
)
case class CappedResults(
results: Vector[ActionResultWithResultingState],
@@ -27,6 +36,15 @@ trait GameHistory {
)
def stateAfter(count: Int): GameState
def sinceDate(date: Date): Vector[ActionResultWithResultingState]
def chronicleEventSourcesSinceDate(date: Date): Vector[ChronicleEventSource] =
sinceDate(date).map { awrs =>
ChronicleEventSource(
date = awrs.resultingState.currentDate,
actionResultType = awrs.actionResult.actionResultType,
newNotifications = awrs.actionResult.newNotifications,
removedFactionIds = awrs.actionResult.removedFactionIds
)
}
def count: Int
def last: ActionResultWithResultingState
src/main/scala/net/eagle0/eagle/library/actions/impl/action/ChronicleEventGenerator.scala
+12 -14
View File
@@ -1,7 +1,6 @@
package net.eagle0.eagle.library.actions.impl.action
import net.eagle0.eagle.library.actions.applier.ActionResultWithResultingState
import net.eagle0.eagle.library.GameHistory
import net.eagle0.eagle.library.{ChronicleEventSource, GameHistory}
import net.eagle0.eagle.model.action_result.generated_text_request.chronicle_event.*
import net.eagle0.eagle.model.action_result.types.ActionResultType
import net.eagle0.eagle.model.action_result.NotificationDetails
@@ -9,18 +8,18 @@ import net.eagle0.eagle.model.state.date.Date
object ChronicleEventGenerator {
private def relevantNotificationDetails(
awrs: ActionResultWithResultingState
source: ChronicleEventSource
): Vector[NotificationDetails] =
awrs.actionResult.newNotifications
source.newNotifications
.map(_.details)
.distinct
private def notificationEntries(
awrs: ActionResultWithResultingState
source: ChronicleEventSource
): Vector[ChronicleEvent] = {
// Chronicle events are only generated during active gameplay when currentDate is present
val date = awrs.resultingState.currentDate
relevantNotificationDetails(awrs).collect {
val date = source.date
relevantNotificationDetails(source).collect {
case NotificationDetails.AllianceAccepted(
offeringFid,
targetFid,
@@ -319,14 +318,13 @@ object ChronicleEventGenerator {
}
private def basicActionResultEntries(
awrs: ActionResultWithResultingState
source: ChronicleEventSource
): Vector[ChronicleEvent] =
awrs.actionResult.actionResultType match {
source.actionResultType match {
case ActionResultType.FactionDestroyed =>
val date = awrs.resultingState.currentDate
awrs.actionResult.removedFactionIds.map { factionId =>
source.removedFactionIds.map { factionId =>
FactionDestroyedChronicleEvent(
date = date,
date = source.date,
factionId = factionId
)
}
@@ -338,7 +336,7 @@ object ChronicleEventGenerator {
gameHistory: GameHistory,
since: Date
): Vector[ChronicleEvent] =
gameHistory.sinceDate(since).flatMap { (awrs: ActionResultWithResultingState) =>
notificationEntries(awrs) ++ basicActionResultEntries(awrs)
gameHistory.chronicleEventSourcesSinceDate(since).flatMap { source =>
notificationEntries(source) ++ basicActionResultEntries(source)
}
}
src/main/scala/net/eagle0/eagle/service/BUILD.bazel
+2
View File
@@ -675,6 +675,7 @@ scala_library(
":action_result_enum_columns_writer",
":action_result_numeric_columns_writer",
":action_result_structured_blob_columns_writer",
":action_result_vector_tables_reader",
":action_result_vector_tables_writer",
":changed_battalion_table_writer",
":changed_faction_table_writer",
@@ -689,6 +690,7 @@ scala_library(
":new_faction_table_writer",
":new_hero_table_writer",
":new_province_table_writer",
":notification_table_reader",
":notification_table_writer",
":per_entity_write_dispatcher",
"//src/main/protobuf/net/eagle0/eagle/internal:game_state_scala_proto",
src/main/scala/net/eagle0/eagle/service/PostgresHistory.scala
+64
View File
@@ -10,7 +10,9 @@ import net.eagle0.common.{JfrEvents, SimpleTimedLogger}
import net.eagle0.eagle.{FactionId, GameId, RoundId, ShardokGameId}
import net.eagle0.eagle.internal.game_state.GameState as GameStateProto
import net.eagle0.eagle.library.actions.applier.ActionResultWithResultingState
import net.eagle0.eagle.library.ChronicleEventSource
import net.eagle0.eagle.library.EagleInternalException
import net.eagle0.eagle.model.action_result.types.ActionResultType
import net.eagle0.eagle.model.action_result.ActionResultT
import net.eagle0.eagle.model.proto_converters.game_state.GameStateConverter
import net.eagle0.eagle.model.state.date.Date
@@ -321,6 +323,10 @@ class PostgresHistory private[service] (
firstSeq.fold(Vector.empty)(seq => replayFrom(seq, cachedCount))
}
override def chronicleEventSourcesSinceDate(date: Date): Vector[ChronicleEventSource] = withDbLock {
PostgresHistory.readChronicleEventSources(connection, date, cachedCount)
}
override def recentResultsForRound(
roundId: RoundId,
predicate: ActionResultWithResultingState => Boolean
@@ -798,6 +804,18 @@ object PostgresHistory {
private[service] val ShardokWriteMaxAttempts = 3
private[service] val ShardokWriteRetryDelayMillis = 50L
private[service] val ChronicleEventSourcesSql: String =
"""SELECT action_seq, action_result_type, date_year, date_month
| FROM action_results
| WHERE action_seq >= (
| SELECT MIN(action_seq)
| FROM action_results
| WHERE date_year IS NOT NULL
| AND (date_year > ? OR (date_year = ? AND date_month >= ?))
| )
| AND action_seq < ?
| ORDER BY action_seq""".stripMargin
private val shardokWriterThreadCounter = new AtomicInteger(0)
private[service] val shardokWriteExecutor: Executor = Executors.newFixedThreadPool(
4,
@@ -835,6 +853,52 @@ object PostgresHistory {
private[service] def requireIdempotentBatch(table: String, results: Array[Int], expected: Int): Unit =
requireSuccessfulBatch(table, results, expected)
private[service] def readChronicleEventSources(
connection: Connection,
since: Date,
endExclusive: Int
): Vector[ChronicleEventSource] =
JfrEvents.postgresQuery("postgres-history", "PostgresHistory.readChronicleEventSources", ChronicleEventSourcesSql) {
event =>
val statement = connection.prepareStatement(ChronicleEventSourcesSql)
val rows =
try {
statement.setInt(1, since.year)
statement.setInt(2, since.year)
statement.setInt(3, since.month.value)
statement.setInt(4, endExclusive)
val resultSet = statement.executeQuery()
val builder = Vector.newBuilder[(Int, Date, ActionResultType)]
while resultSet.next() do
builder += ((
resultSet.getInt("action_seq"),
Date(
year = resultSet.getInt("date_year"),
month = Date.Month.fromInt(resultSet.getInt("date_month"))
),
ActionResultType.fromValue(resultSet.getInt("action_result_type"))
))
builder.result()
} finally statement.close()
event.rows = rows.size
rows.headOption.fold(Vector.empty) {
case (startInclusive, _, _) =>
val notifications = NotificationTableReader.readNewRange(connection, startInclusive, endExclusive)
val removedFactions =
ActionResultVectorTablesReader.readRemovedFactionIdsRange(connection, startInclusive, endExclusive)
rows.map {
case (actionSeq, date, actionResultType) =>
ChronicleEventSource(
date = date,
actionResultType = actionResultType,
newNotifications = notifications.getOrElse(actionSeq, Vector.empty),
removedFactionIds = removedFactions.getOrElse(actionSeq, Vector.empty)
)
}
}
}
val ActionResultInsertSql: String =
"""INSERT INTO action_results
| (action_seq, action_result_type, round_id, date_year, date_month,
src/test/scala/net/eagle0/eagle/library/actions/impl/action/NewRoundActionTest.scala
+1 -1
View File
@@ -121,7 +121,7 @@ class NewRoundActionTest
.randomResults(SeededRandom(0))
.newValue
mockHistory.sinceDate
mockHistory.chronicleEventSourcesSinceDate
.expects(Date(year = 0, month = Date.Month.January))
.returning(Vector.empty)
.anyNumberOfTimes(): Unit
src/test/scala/net/eagle0/eagle/service/PostgresHistorySchemaBatchTest.scala
+67
View File
@@ -7,6 +7,8 @@ import java.util.concurrent.atomic.AtomicReference
import scala.jdk.CollectionConverters.*
import com.google.protobuf.ByteString
import net.eagle0.eagle.library.ChronicleEventSource
import net.eagle0.eagle.model.action_result.types.ActionResultType
import net.eagle0.eagle.model.state.{GameType, RoundPhase}
import net.eagle0.eagle.model.state.date.Date
import net.eagle0.eagle.model.state.game_state.{EagleMapInfo, GameState}
@@ -341,6 +343,71 @@ class PostgresHistorySchemaBatchTest extends AnyFlatSpec with Matchers with Mock
history.sinceFromState(1, cachedState) shouldBe empty
}
"PostgresHistory chronicle event projection" should "read dates and chronicle fields without replaying game states" in {
val connection = mock[Connection]
val actionStatement = mock[PreparedStatement]
val notificationStatement = mock[PreparedStatement]
val removedStatement = mock[PreparedStatement]
val actionRows = mock[ResultSet]
val notificationRows = mock[ResultSet]
val removedRows = mock[ResultSet]
val since = Date(1501, Date.Month.March)
inSequence {
(connection
.prepareStatement(_: String))
.expects(PostgresHistory.ChronicleEventSourcesSql)
.returning(actionStatement)
.once(): Unit
(connection.prepareStatement(_: String)).expects(*).returning(notificationStatement).once(): Unit
(connection.prepareStatement(_: String)).expects(*).returning(removedStatement).once(): Unit
}
actionStatement.setInt.expects(1, 1501).once(): Unit
actionStatement.setInt.expects(2, 1501).once(): Unit
actionStatement.setInt.expects(3, Date.Month.March.value).once(): Unit
actionStatement.setInt.expects(4, 12).once(): Unit
(() => actionStatement.executeQuery()).expects().returning(actionRows).once(): Unit
inSequence {
(() => actionRows.next()).expects().returning(true).once(): Unit
(() => actionRows.next()).expects().returning(false).once(): Unit
}
(actionRows.getInt(_: String)).expects("action_seq").returning(10).once(): Unit
(actionRows.getInt(_: String)).expects("date_year").returning(1501).once(): Unit
(actionRows.getInt(_: String)).expects("date_month").returning(Date.Month.March.value).once(): Unit
(actionRows
.getInt(_: String))
.expects("action_result_type")
.returning(ActionResultType.FactionDestroyed.value)
.once(): Unit
(() => actionStatement.close()).expects().once(): Unit
notificationStatement.setInt.expects(1, 10).once(): Unit
notificationStatement.setInt.expects(2, 12).once(): Unit
(() => notificationStatement.executeQuery()).expects().returning(notificationRows).once(): Unit
(() => notificationRows.next()).expects().returning(false).once(): Unit
(() => notificationStatement.close()).expects().once(): Unit
removedStatement.setInt.expects(1, 10).once(): Unit
removedStatement.setInt.expects(2, 12).once(): Unit
(() => removedStatement.executeQuery()).expects().returning(removedRows).once(): Unit
inSequence {
(() => removedRows.next()).expects().returning(true).once(): Unit
(() => removedRows.next()).expects().returning(false).once(): Unit
}
(removedRows.getInt(_: String)).expects("action_seq").returning(10).once(): Unit
(removedRows.getInt(_: String)).expects("faction_id").returning(7).once(): Unit
(() => removedStatement.close()).expects().once(): Unit
PostgresHistory.readChronicleEventSources(connection, since, endExclusive = 12) shouldBe Vector(
ChronicleEventSource(
date = since,
actionResultType = ActionResultType.FactionDestroyed,
newNotifications = Vector.empty,
removedFactionIds = Vector(7)
)
)
}
"PostgresHistory Shardok write-behind" should "publish accepted updates before a blocked database write completes" in {
val writeStarted = new CountDownLatch(1)
val allowWrite = new CountDownLatch(1)