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

Mirror SQLite history deltas to Postgres

Browse Source
This commit is contained in:
admin
2026-06-08 08:02:01 -07:00
parent 835d1b7a83
commit 047b9c850e
5 changed files with 170 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/docker_build.yml
+2
View File
@@ -269,6 +269,7 @@ jobs:
EAGLE_POSTGRES_DATABASE: ${{ secrets.EAGLE_POSTGRES_DATABASE }}
EAGLE_POSTGRES_USER: ${{ secrets.EAGLE_POSTGRES_USER }}
EAGLE_POSTGRES_PASSWORD: ${{ secrets.EAGLE_POSTGRES_PASSWORD }}
EAGLE_POSTGRES_HISTORY_MIRROR: ${{ secrets.EAGLE_POSTGRES_HISTORY_MIRROR }}
DO_DROPLET_IP: ${{ secrets.DO_DROPLET_IP }}
DO_REGISTRY_TOKEN: ${{ secrets.DO_REGISTRY_TOKEN }}
NOTIFY_SECRET: ${{ secrets.EAGLE_NOTIFY_SECRET }}
@@ -431,6 +432,7 @@ jobs:
export EAGLE_POSTGRES_DATABASE="${EAGLE_POSTGRES_DATABASE}"
export EAGLE_POSTGRES_USER="${EAGLE_POSTGRES_USER}"
export EAGLE_POSTGRES_PASSWORD="${EAGLE_POSTGRES_PASSWORD}"
export EAGLE_POSTGRES_HISTORY_MIRROR="${EAGLE_POSTGRES_HISTORY_MIRROR:-false}"
export DO_REGISTRY_TOKEN="${DO_REGISTRY_TOKEN}"
export NOTIFY_SECRET="${NOTIFY_SECRET}"
export GITHUB_TOKEN="${GITHUB_TOKEN_FOR_ADMIN}"
docker-compose.prod.yml
+2
View File
@@ -38,6 +38,7 @@ services:
EAGLE_POSTGRES_USER: "${EAGLE_POSTGRES_USER:-}"
EAGLE_POSTGRES_PASSWORD: "${EAGLE_POSTGRES_PASSWORD:-}"
EAGLE_POSTGRES_SSLMODE: "${EAGLE_POSTGRES_SSLMODE:-require}"
EAGLE_POSTGRES_HISTORY_MIRROR: "${EAGLE_POSTGRES_HISTORY_MIRROR:-false}"
# JWT public key for token validation (auth service handles signing)
# Reads from /etc/eagle0/keys/public.pem via shared volume
# Auth token for Shardok on Hetzner (required)
@@ -93,6 +94,7 @@ services:
EAGLE_POSTGRES_USER: "${EAGLE_POSTGRES_USER:-}"
EAGLE_POSTGRES_PASSWORD: "${EAGLE_POSTGRES_PASSWORD:-}"
EAGLE_POSTGRES_SSLMODE: "${EAGLE_POSTGRES_SSLMODE:-require}"
EAGLE_POSTGRES_HISTORY_MIRROR: "${EAGLE_POSTGRES_HISTORY_MIRROR:-false}"
# JWT public key for token validation (auth service handles signing)
# Reads from /etc/eagle0/keys/public.pem via shared volume
SHARDOK_AUTH_TOKEN: "${SHARDOK_AUTH_TOKEN:-}"
src/main/scala/net/eagle0/eagle/service/BUILD.bazel
+5 -1
View File
@@ -517,7 +517,10 @@ scala_library(
scala_library(
name = "sqlite_history",
srcs = ["SqliteHistory.scala"],
srcs = [
"PostgresHistoryMirror.scala",
"SqliteHistory.scala",
],
visibility = [
"//src/main/scala/net/eagle0/eagle:__subpackages__",
"//src/test/scala/net/eagle0/eagle:__subpackages__",
@@ -526,6 +529,7 @@ scala_library(
":full_game_history",
],
runtime_deps = [
"@maven//:org_postgresql_postgresql",
"@maven//:org_xerial_sqlite_jdbc",
],
deps = [
src/main/scala/net/eagle0/eagle/service/PostgresHistoryMirror.scala
+141
View File
@@ -0,0 +1,141 @@
package net.eagle0.eagle.service
import java.sql.{Connection, DriverManager}
import net.eagle0.eagle.GameId
/**
* Optional synchronous mirror for SQLite history deltas.
*
* This is a rollout bridge, not the final authoritative Postgres history backend. It writes the same compact delta
* payloads already used for remote SQLite recovery into Postgres, letting production exercise the managed database on
* real game write paths while keeping SQLite reads and S3 recovery unchanged.
*/
private[service] object PostgresHistoryMirror {
private case class Config(
host: String,
port: String,
database: String,
user: String,
password: String,
sslMode: String
) {
val jdbcUrl: String = s"jdbc:postgresql://$host:$port/$database?sslmode=$sslMode"
}
private lazy val config: Option[Config] =
if Option(System.getenv("EAGLE_POSTGRES_HISTORY_MIRROR")).exists(_.equalsIgnoreCase("true")) then
configFromEnv() match {
case Right(config) =>
Class.forName("org.postgresql.Driver")
Some(config)
case Left(error) =>
System.err.println(s"PostgresHistoryMirror disabled: $error")
None
}
else None
def enabled: Boolean = config.isDefined
def saveEagleDelta(gameId: GameId, startSeq: Int, endSeq: Int, key: String, payload: Array[Byte]): Unit =
config.foreach { c =>
try
withConnection(c) { connection =>
initializeSchema(connection)
val stmt = connection.prepareStatement(
"""INSERT INTO eagle_history_mirror_deltas(game_id, key, delta_type, start_seq, end_seq, payload)
|VALUES (?, ?, 'eagle', ?, ?, ?)
|ON CONFLICT (game_id, key) DO NOTHING""".stripMargin
)
try {
stmt.setLong(1, gameId)
stmt.setString(2, key)
stmt.setInt(3, startSeq)
stmt.setInt(4, endSeq)
stmt.setBytes(5, payload)
stmt.executeUpdate(): Unit
} finally stmt.close()
}
catch {
case e: Exception =>
System.err.println(
s"PostgresHistoryMirror: failed to mirror Eagle delta for game ${gameId.toHexString} " +
s"[$startSeq, $endSeq): ${e.getMessage}"
)
}
}
def saveShardokDelta(gameId: GameId, key: String, payload: Array[Byte]): Unit =
config.foreach { c =>
try
withConnection(c) { connection =>
initializeSchema(connection)
val stmt = connection.prepareStatement(
"""INSERT INTO eagle_history_mirror_deltas(game_id, key, delta_type, payload)
|VALUES (?, ?, 'shardok', ?)
|ON CONFLICT (game_id, key) DO NOTHING""".stripMargin
)
try {
stmt.setLong(1, gameId)
stmt.setString(2, key)
stmt.setBytes(3, payload)
stmt.executeUpdate(): Unit
} finally stmt.close()
}
catch {
case e: Exception =>
System.err.println(
s"PostgresHistoryMirror: failed to mirror Shardok delta for game ${gameId.toHexString}: ${e.getMessage}"
)
}
}
private def configFromEnv(): Either[String, Config] = {
def required(name: String): Either[String, String] =
Option(System.getenv(name)).map(_.trim).filter(_.nonEmpty).toRight(s"Missing required environment variable $name")
for
host <- required("EAGLE_POSTGRES_HOST")
port <- required("EAGLE_POSTGRES_PORT")
database <- required("EAGLE_POSTGRES_DATABASE")
user <- required("EAGLE_POSTGRES_USER")
password <- required("EAGLE_POSTGRES_PASSWORD")
yield Config(
host = host,
port = port,
database = database,
user = user,
password = password,
sslMode = Option(System.getenv("EAGLE_POSTGRES_SSLMODE")).map(_.trim).filter(_.nonEmpty).getOrElse("require")
)
}
private def withConnection[A](config: Config)(f: Connection => A): A = {
val connection = DriverManager.getConnection(config.jdbcUrl, config.user, config.password)
try f(connection)
finally connection.close()
}
private def initializeSchema(connection: Connection): Unit = {
val stmt = connection.createStatement()
try {
stmt.executeUpdate(
"""CREATE TABLE IF NOT EXISTS eagle_history_mirror_deltas (
| game_id BIGINT NOT NULL,
| key TEXT NOT NULL,
| delta_type TEXT NOT NULL,
| start_seq INTEGER,
| end_seq INTEGER,
| payload BYTEA NOT NULL,
| created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
| PRIMARY KEY (game_id, key)
|)""".stripMargin
): Unit
stmt.executeUpdate(
"""CREATE INDEX IF NOT EXISTS eagle_history_mirror_deltas_game_type_created
|ON eagle_history_mirror_deltas(game_id, delta_type, created_at)""".stripMargin
): Unit
} finally stmt.close()
end try
}
}
src/main/scala/net/eagle0/eagle/service/SqliteHistory.scala
+20 -5
View File
@@ -6,6 +6,8 @@ import java.sql.{Connection, DriverManager, Types}
import java.util.concurrent.{CopyOnWriteArrayList, Executors, TimeUnit}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
import scala.util.Try
import net.eagle0.common.JfrEvents
import net.eagle0.eagle.{FactionId, RoundId, ShardokGameId}
import net.eagle0.eagle.internal.game_state.GameState as GameStateProto
@@ -56,6 +58,8 @@ class SqliteHistory private[service] (
private val uploadRunning = new AtomicBoolean(false)
private val uploadMonitor = new Object
private val dbLock = new Object
private val gameIdForMirror = Option(dbFile.getParentFile)
.flatMap(parent => Try(java.lang.Long.parseUnsignedLong(parent.getName, 16)).toOption)
private def withDbLock[A](f: => A): A = dbLock.synchronized(f)
@@ -323,10 +327,15 @@ class SqliteHistory private[service] (
}
private def saveEagleDeltaToPersister(startSeq: Int, endSeq: Int): Unit =
persister.foreach { p =>
if persister.nonEmpty || PostgresHistoryMirror.enabled then {
val key = SqliteHistory.nextEagleDeltaKey(startSeq)
val bytes = withDbLock(SqliteHistory.serializeEagleDelta(startSeq, endSeq, eagleDeltaDbBytes(startSeq, endSeq)))
val _ = p.save(key, bytes)
gameIdForMirror.foreach { gameId =>
PostgresHistoryMirror.saveEagleDelta(gameId, startSeq, endSeq, key, bytes)
}
persister.foreach { p =>
val _ = p.save(key, bytes)
}
}
private def eagleDeltaDbBytes(startSeq: Int, endSeq: Int): Array[Byte] = {
@@ -949,10 +958,16 @@ class SqliteHistory private[service] (
}
private def saveShardokDeltaToPersister(delta: SqliteHistory.ShardokDelta): Unit =
persister.foreach { p =>
if persister.nonEmpty || PostgresHistoryMirror.enabled then {
val key = SqliteHistory.nextShardokDeltaKey()
val saved = p.save(key, SqliteHistory.serializeShardokDelta(delta))
if saved then recordAppliedShardokDelta(key)
val bytes = SqliteHistory.serializeShardokDelta(delta)
gameIdForMirror.foreach { gameId =>
PostgresHistoryMirror.saveShardokDelta(gameId, key, bytes)
}
persister.foreach { p =>
val saved = p.save(key, bytes)
if saved then recordAppliedShardokDelta(key)
}
}
private def hasAppliedShardokDelta(key: String): Boolean = withDbLock {