Public commit record
98/100 · Strong
What the developer wrote
Authored by Bastien Teinturier
Stop storing channel errors in `AuditDb` (#3236)
We stop storing every `ChannelErrorOccured` event in the `AuditDb`.
This is very spammy because there are a lot of non-fatal events,
and they turned out to be unnecessary for auditing (they make more
sense in the logs for debugging).
Such events include for example notifying that we tried to relay a
payment to a channel that didn't have enough balance, or commands
sent to channels that are offline, closing or reconnecting.
Note that we stop creating the corresponding table, but we don't
delete old data: it is up to node operators to decide whether they
want to keep it or not. We still support migrating older databases
including data from this table.
We also remove tests from very old migrations (more than 2 years old)
which are spammy but not very interesting anymore.
✓ Descriptive subject✓ Names a concrete action or component✓ Provides detailed explanatory context✓ Explains rationale or failure mode✓ Mentions testing or verification✓ Links an issue, advisory, or supporting reference
What changed, and why it matters
This commit removes a database table that was used to store routine, mostly non-serious channel error events. It is a cleanup change: the events are still logged to normal log files, but they are no longer written to the audit database. There is no indication this fixes a security vulnerability; it is described by the project as a way to reduce database noise and storage use.
Recommended action
No security action required. Operators who relied on the audit database for channel error history should note that new events will only be available in application logs after this change. Reviewers may confirm that the removed table is not referenced elsewhere and that the retained metrics still meet operational needs.
Security signals we found
01
No security-relevant signals in commit message or diff
02
Data retention change: stops writing new channel error records to audit DB
03
Existing data not deleted; migration support preserved
04
Log message converted from string interpolation to structured logging (minor, not a security fix)
Evidence from the diff
The change stops persisting ChannelErrorOccurred events in AuditDb. It removes the add(channelErrorOccurred: ChannelErrorOccurred) method from the AuditDb trait and its implementations for PostgreSQL (PgAuditDb) and SQLite (SqliteAuditDb), drops creation of the channel_errors table/index in new databases, and removes the auditDb.add(e) call from the ChannelErrorOccurred handler in DbEventHandler. Metrics counters for local/remote/fatal errors are retained. Existing databases keep their old table and data; migrations still support older schemas that include the table. The commit also deletes several old migration tests (>2 years) that exercised the removed table.
Changed components
eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scalaeclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scalaeclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scalaeclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scalaeclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scalaInspect captured patch +9 / −651
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scala b/eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scala
index fee50e7..b708a8c 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/db/AuditDb.scala
@@ -38,8 +38,6 @@ trait AuditDb {
def add(txConfirmed: TransactionConfirmed): Unit
- def add(channelErrorOccurred: ChannelErrorOccurred): Unit
-
def addChannelUpdate(channelUpdateParametersChanged: ChannelUpdateParametersChanged): Unit
def addPathFindingExperimentMetrics(metrics: PathFindingExperimentMetrics): Unit
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scala b/eclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scala
index a7da4fb..116e70f 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/db/DbEventHandler.scala
@@ -100,7 +100,7 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
case e: ChannelLiquidityPurchased => liquidityDb.addPurchase(e)
case e: TransactionPublished =>
- log.info(s"paying mining fee=${e.miningFee} for txid=${e.tx.txid} desc=${e.desc}")
+ log.info("paying mining fee={} for txid={} desc={}", e.miningFee, e.tx.txid, e.desc)
auditDb.add(e)
case e: TransactionConfirmed =>
@@ -108,15 +108,13 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
auditDb.add(e)
case e: ChannelErrorOccurred =>
- // first pattern matching level is to ignore some errors, second level is to separate between different kind of errors
+ // The first pattern matching level is to ignore some errors, the second level is to separate between different kind of errors.
e.error match {
case LocalError(_: CannotAffordFees) => () // will be thrown at each new block if our balance is too low to update the commitment fee
- case _ =>
- e.error match {
- case LocalError(_) => ChannelMetrics.ChannelErrors.withTag(ChannelTags.Origin, ChannelTags.Origins.Local).withTag(ChannelTags.Fatal, value = e.isFatal).increment()
- case RemoteError(_) => ChannelMetrics.ChannelErrors.withTag(ChannelTags.Origin, ChannelTags.Origins.Remote).increment()
- }
- auditDb.add(e)
+ case _ => e.error match {
+ case LocalError(_) => ChannelMetrics.ChannelErrors.withTag(ChannelTags.Origin, ChannelTags.Origins.Local).withTag(ChannelTags.Fatal, value = e.isFatal).increment()
+ case RemoteError(_) => ChannelMetrics.ChannelErrors.withTag(ChannelTags.Origin, ChannelTags.Origins.Remote).increment()
+ }
}
case e: ChannelStateChanged =>
@@ -127,7 +125,6 @@ class DbEventHandler(nodeParams: NodeParams) extends Actor with DiagnosticActorL
val event = ChannelEvent.EventType.Created
auditDb.add(ChannelEvent(channelId, remoteNodeId, commitments.latest.capacity, commitments.localChannelParams.isChannelOpener, !commitments.announceChannel, event))
channelsDb.updateChannelMeta(channelId, event)
- case ChannelStateChanged(_, _, _, _, WAIT_FOR_INIT_INTERNAL, _, _) =>
case ChannelStateChanged(_, channelId, _, _, OFFLINE, SYNCING, _) =>
channelsDb.updateChannelMeta(channelId, ChannelEvent.EventType.Connected)
case ChannelStateChanged(_, _, _, _, _, CLOSING, _) =>
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scala b/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scala
index 67e1c59..8eceb18 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/db/pg/PgAuditDb.scala
@@ -131,7 +131,6 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
statement.executeUpdate("CREATE TABLE audit.transactions_published (tx_id TEXT NOT NULL PRIMARY KEY, channel_id TEXT NOT NULL, node_id TEXT NOT NULL, mining_fee_sat BIGINT NOT NULL, tx_type TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
statement.executeUpdate("CREATE TABLE audit.transactions_confirmed (tx_id TEXT NOT NULL PRIMARY KEY, channel_id TEXT NOT NULL, node_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.channel_errors (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal BOOLEAN NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
statement.executeUpdate("CREATE INDEX sent_timestamp_idx ON audit.sent(timestamp)")
statement.executeUpdate("CREATE INDEX received_timestamp_idx ON audit.received(timestamp)")
statement.executeUpdate("CREATE INDEX relayed_timestamp_idx ON audit.relayed(timestamp)")
@@ -140,7 +139,6 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
statement.executeUpdate("CREATE INDEX relayed_trampoline_payment_hash_idx ON audit.relayed_trampoline(payment_hash)")
statement.executeUpdate("CREATE INDEX relayed_channel_id_idx ON audit.relayed(channel_id)")
statement.executeUpdate("CREATE INDEX channel_events_timestamp_idx ON audit.channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_errors_timestamp_idx ON audit.channel_errors(timestamp)")
statement.executeUpdate("CREATE INDEX channel_updates_cid_idx ON audit.channel_updates(channel_id)")
statement.executeUpdate("CREATE INDEX channel_updates_nid_idx ON audit.channel_updates(node_id)")
statement.executeUpdate("CREATE INDEX channel_updates_timestamp_idx ON audit.channel_updates(timestamp)")
@@ -305,24 +303,6 @@ class PgAuditDb(implicit ds: DataSource) extends AuditDb with Logging {
}
}
- override def add(e: ChannelErrorOccurred): Unit = withMetrics("audit/add-channel-error", DbBackends.Postgres) {
- inTransaction { pg =>
- using(pg.prepareStatement("INSERT INTO audit.channel_errors VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- val (errorName, errorMessage) = e.error match {
- case LocalError(t) => (t.getClass.getSimpleName, t.getMessage)
- case RemoteError(error) => ("remote", error.toAscii)
- }
- statement.setString(1, e.channelId.toHex)
- statement.setString(2, e.remoteNodeId.value.toHex)
- statement.setString(3, errorName)
- statement.setString(4, errorMessage)
- statement.setBoolean(5, e.isFatal)
- statement.setTimestamp(6, Timestamp.from(Instant.now()))
- statement.executeUpdate()
- }
- }
- }
-
override def addChannelUpdate(u: ChannelUpdateParametersChanged): Unit = withMetrics("audit/add-channel-update", DbBackends.Postgres) {
inTransaction { pg =>
using(pg.prepareStatement("INSERT INTO audit.channel_updates VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) { statement =>
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scala b/eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scala
index f118766..e5cce5b 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/db/sqlite/SqliteAuditDb.scala
@@ -124,7 +124,6 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
statement.executeUpdate("CREATE TABLE relayed (payment_hash BLOB NOT NULL, amount_msat INTEGER NOT NULL, channel_id BLOB NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
statement.executeUpdate("CREATE TABLE relayed_trampoline (payment_hash BLOB NOT NULL, amount_msat INTEGER NOT NULL, next_node_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
statement.executeUpdate("CREATE TABLE channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE channel_errors (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
statement.executeUpdate("CREATE TABLE channel_updates (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, fee_base_msat INTEGER NOT NULL, fee_proportional_millionths INTEGER NOT NULL, cltv_expiry_delta INTEGER NOT NULL, htlc_minimum_msat INTEGER NOT NULL, htlc_maximum_msat INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
statement.executeUpdate("CREATE TABLE path_finding_metrics (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, status TEXT NOT NULL, duration_ms INTEGER NOT NULL, timestamp INTEGER NOT NULL, is_mpp INTEGER NOT NULL, experiment_name TEXT NOT NULL, recipient_node_id BLOB NOT NULL)")
statement.executeUpdate("CREATE TABLE transactions_published (tx_id BLOB NOT NULL PRIMARY KEY, channel_id BLOB NOT NULL, node_id BLOB NOT NULL, mining_fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
@@ -138,7 +137,6 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
statement.executeUpdate("CREATE INDEX relayed_trampoline_timestamp_idx ON relayed_trampoline(timestamp)")
statement.executeUpdate("CREATE INDEX relayed_trampoline_payment_hash_idx ON relayed_trampoline(payment_hash)")
statement.executeUpdate("CREATE INDEX channel_events_timestamp_idx ON channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_errors_timestamp_idx ON channel_errors(timestamp)")
statement.executeUpdate("CREATE INDEX channel_updates_cid_idx ON channel_updates(channel_id)")
statement.executeUpdate("CREATE INDEX channel_updates_nid_idx ON channel_updates(node_id)")
statement.executeUpdate("CREATE INDEX channel_updates_timestamp_idx ON channel_updates(timestamp)")
@@ -288,22 +286,6 @@ class SqliteAuditDb(val sqlite: Connection) extends AuditDb with Logging {
}
}
- override def add(e: ChannelErrorOccurred): Unit = withMetrics("audit/add-channel-error", DbBackends.Sqlite) {
- using(sqlite.prepareStatement("INSERT INTO channel_errors VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- val (errorName, errorMessage) = e.error match {
- case LocalError(t) => (t.getClass.getSimpleName, t.getMessage)
- case RemoteError(error) => ("remote", error.toAscii)
- }
- statement.setBytes(1, e.channelId.toArray)
- statement.setBytes(2, e.remoteNodeId.value.toArray)
- statement.setString(3, errorName)
- statement.setString(4, errorMessage)
- statement.setBoolean(5, e.isFatal)
- statement.setLong(6, TimestampMilli.now().toLong)
- statement.executeUpdate()
- }
- }
-
override def addChannelUpdate(u: ChannelUpdateParametersChanged): Unit = withMetrics("audit/add-channel-update", DbBackends.Sqlite) {
using(sqlite.prepareStatement("INSERT INTO channel_updates VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) { statement =>
statement.setBytes(1, u.channelId.toArray)
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scala
index aff010d..0fabf79 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/db/AuditDbSpec.scala
@@ -16,22 +16,20 @@
package fr.acinq.eclair.db
-import fr.acinq.bitcoin.scalacompat.Crypto.{PrivateKey, PublicKey}
+import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
import fr.acinq.bitcoin.scalacompat.{Block, ByteVector32, SatoshiLong, Script, Transaction, TxOut}
-import fr.acinq.eclair.TestDatabases.{TestPgDatabases, TestSqliteDatabases, migrationCheck}
+import fr.acinq.eclair.TestDatabases.{TestPgDatabases, TestSqliteDatabases}
import fr.acinq.eclair._
import fr.acinq.eclair.channel.Helpers.Closing.MutualClose
import fr.acinq.eclair.channel._
-import fr.acinq.eclair.db.AuditDb.{NetworkFee, Stats}
+import fr.acinq.eclair.db.AuditDb.Stats
import fr.acinq.eclair.db.DbEventHandler.ChannelEvent
import fr.acinq.eclair.db.jdbc.JdbcUtils.using
import fr.acinq.eclair.db.pg.PgAuditDb
-import fr.acinq.eclair.db.pg.PgUtils.{getVersion, setVersion}
import fr.acinq.eclair.db.sqlite.SqliteAuditDb
import fr.acinq.eclair.payment.Bolt11Invoice.ExtraHop
import fr.acinq.eclair.payment._
import fr.acinq.eclair.router.Announcements
-import fr.acinq.eclair.wire.protocol.Error
import org.scalatest.Tag
import org.scalatest.funsuite.AnyFunSuite
import scodec.bits.HexStringSyntax
@@ -78,8 +76,6 @@ class AuditDbSpec extends AnyFunSuite {
val pp6 = PaymentSent.PartialPayment(UUID.randomUUID(), 42000 msat, 1000 msat, randomBytes32(), None, timestamp = now + 10.minutes)
val e6 = PaymentSent(UUID.randomUUID(), randomBytes32(), randomBytes32(), 42000 msat, randomKey().publicKey, pp6 :: Nil, None)
val e7 = ChannelEvent(randomBytes32(), randomKey().publicKey, 456123000 sat, isChannelOpener = true, isPrivate = false, ChannelEvent.EventType.Closed(MutualClose(null)))
- val e8 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, LocalError(new RuntimeException("oops")), isFatal = true)
- val e9 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, RemoteError(Error(randomBytes32(), "remote oops")), isFatal = true)
val e10 = TrampolinePaymentRelayed(randomBytes32(),
Seq(
PaymentRelayed.IncomingPart(20000 msat, randomBytes32(), now - 7.seconds),
@@ -104,8 +100,6 @@ class AuditDbSpec extends AnyFunSuite {
db.add(e5)
db.add(e6)
db.add(e7)
- db.add(e8)
- db.add(e9)
db.add(e10)
db.add(e11)
db.add(e12)
@@ -230,599 +224,6 @@ class AuditDbSpec extends AnyFunSuite {
}
}
- test("migrate sqlite audit database v1 -> current") {
-
- val dbs = TestSqliteDatabases()
-
- val ps = PaymentSent(UUID.randomUUID(), randomBytes32(), randomBytes32(), 42000 msat, PrivateKey(ByteVector32.One).publicKey, PaymentSent.PartialPayment(UUID.randomUUID(), 42000 msat, 1000 msat, randomBytes32(), None) :: Nil, None)
- val pp1 = PaymentSent.PartialPayment(UUID.randomUUID(), 42001 msat, 1001 msat, randomBytes32(), None)
- val pp2 = PaymentSent.PartialPayment(UUID.randomUUID(), 42002 msat, 1002 msat, randomBytes32(), None)
- val ps1 = PaymentSent(UUID.randomUUID(), randomBytes32(), randomBytes32(), 84003 msat, PrivateKey(ByteVector32.One).publicKey, pp1 :: pp2 :: Nil, None)
- val e1 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, LocalError(new RuntimeException("oops")), isFatal = true)
- val e2 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, RemoteError(Error(randomBytes32(), "remote oops")), isFatal = true)
-
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS balance_updated (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, amount_msat INTEGER NOT NULL, capacity_sat INTEGER NOT NULL, reserve_sat INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS sent (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, payment_preimage BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS received (amount_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS relayed (amount_in_msat INTEGER NOT NULL, amount_out_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS network_fees (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, tx_id BLOB NOT NULL, fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event STRING NOT NULL, timestamp INTEGER NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS balance_updated_idx ON balance_updated(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_events_timestamp_idx ON channel_events(timestamp)")
-
- setVersion(statement, "audit", 1)
- }
-
- // add a row (no ID on sent)
- using(connection.prepareStatement("INSERT INTO sent VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setLong(1, ps.recipientAmount.toLong)
- statement.setLong(2, ps.feesPaid.toLong)
- statement.setBytes(3, ps.paymentHash.toArray)
- statement.setBytes(4, ps.paymentPreimage.toArray)
- statement.setBytes(5, ps.parts.head.toChannelId.toArray)
- statement.setLong(6, ps.timestamp.toLong)
- statement.executeUpdate()
- }
- },
- dbName = SqliteAuditDb.DB_NAME,
- targetVersion = SqliteAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- // existing rows in the 'sent' table will use id=00000000-0000-0000-0000-000000000000 as default
- assert(dbs.audit.listSent(0 unixms, TimestampMilli.now() + 1.minute) == Seq(ps.copy(id = ZERO_UUID, parts = Seq(ps.parts.head.copy(id = ZERO_UUID)))))
-
- val postMigrationDb = new SqliteAuditDb(connection)
-
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
-
- postMigrationDb.add(ps1)
- postMigrationDb.add(e1)
- postMigrationDb.add(e2)
-
- // the old record will have the UNKNOWN_UUID but the new ones will have their actual id
- val expected = Seq(ps.copy(id = ZERO_UUID, parts = Seq(ps.parts.head.copy(id = ZERO_UUID))), ps1)
- assert(postMigrationDb.listSent(0 unixms, TimestampMilli.now() + 1.minute) == expected)
- }
- )
- }
-
- test("migrate sqlite audit database v2 -> current") {
- val dbs = TestSqliteDatabases()
-
- val e1 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, LocalError(new RuntimeException("oops")), isFatal = true)
- val e2 = ChannelErrorOccurred(null, randomBytes32(), randomKey().publicKey, RemoteError(Error(randomBytes32(), "remote oops")), isFatal = true)
-
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS balance_updated (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, amount_msat INTEGER NOT NULL, capacity_sat INTEGER NOT NULL, reserve_sat INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS sent (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, payment_preimage BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL, id BLOB NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS received (amount_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS relayed (amount_in_msat INTEGER NOT NULL, amount_out_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS network_fees (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, tx_id BLOB NOT NULL, fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event STRING NOT NULL, timestamp INTEGER NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS balance_updated_idx ON balance_updated(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_events_timestamp_idx ON channel_events(timestamp)")
-
- setVersion(statement, "audit", 2)
- }
- },
- dbName = SqliteAuditDb.DB_NAME,
- targetVersion = SqliteAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- migratedDb.add(e1)
-
- val postMigrationDb = new SqliteAuditDb(connection)
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- postMigrationDb.add(e2)
- }
- )
- }
-
- test("migrate sqlite audit database v3 -> current") {
-
- val dbs = TestSqliteDatabases()
-
- val pp1 = PaymentSent.PartialPayment(UUID.randomUUID(), 500 msat, 10 msat, randomBytes32(), None, 100 unixms)
- val pp2 = PaymentSent.PartialPayment(UUID.randomUUID(), 600 msat, 5 msat, randomBytes32(), None, 110 unixms)
- val ps1 = PaymentSent(UUID.randomUUID(), randomBytes32(), randomBytes32(), 1100 msat, PrivateKey(ByteVector32.One).publicKey, pp1 :: pp2 :: Nil, None)
-
- val relayed1 = ChannelPaymentRelayed(600 msat, 500 msat, randomBytes32(), randomBytes32(), randomBytes32(), 105 unixms, 105 unixms)
- val relayed2 = ChannelPaymentRelayed(650 msat, 500 msat, randomBytes32(), randomBytes32(), randomBytes32(), 115 unixms, 115 unixms)
-
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS balance_updated (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, amount_msat INTEGER NOT NULL, capacity_sat INTEGER NOT NULL, reserve_sat INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS sent (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, payment_preimage BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL, id BLOB NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS received (amount_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS relayed (amount_in_msat INTEGER NOT NULL, amount_out_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS network_fees (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, tx_id BLOB NOT NULL, fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_errors (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS balance_updated_idx ON balance_updated(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_events_timestamp_idx ON channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_errors_timestamp_idx ON channel_errors(timestamp)")
-
- setVersion(statement, "audit", 3)
- }
-
- for (pp <- Seq(pp1, pp2)) {
- using(connection.prepareStatement("INSERT INTO sent (amount_msat, fees_msat, payment_hash, payment_preimage, to_channel_id, timestamp, id) VALUES (?, ?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setLong(1, pp.amount.toLong)
- statement.setLong(2, pp.feesPaid.toLong)
- statement.setBytes(3, ps1.paymentHash.toArray)
- statement.setBytes(4, ps1.paymentPreimage.toArray)
- statement.setBytes(5, pp.toChannelId.toArray)
- statement.setLong(6, pp.timestamp.toLong)
- statement.setBytes(7, pp.id.toString.getBytes)
- statement.executeUpdate()
- }
- }
-
- for (relayed <- Seq(relayed1, relayed2)) {
- using(connection.prepareStatement("INSERT INTO relayed (amount_in_msat, amount_out_msat, payment_hash, from_channel_id, to_channel_id, timestamp) VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setLong(1, relayed.amountIn.toLong)
- statement.setLong(2, relayed.amountOut.toLong)
- statement.setBytes(3, relayed.paymentHash.toArray)
- statement.setBytes(4, relayed.fromChannelId.toArray)
- statement.setBytes(5, relayed.toChannelId.toArray)
- statement.setLong(6, relayed.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- },
- dbName = SqliteAuditDb.DB_NAME,
- targetVersion = SqliteAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- assert(migratedDb.listSent(50 unixms, 150 unixms).toSet == Set(
- ps1.copy(id = pp1.id, recipientAmount = pp1.amount, parts = pp1 :: Nil),
- ps1.copy(id = pp2.id, recipientAmount = pp2.amount, parts = pp2 :: Nil)
- ))
- assert(migratedDb.listRelayed(100 unixms, 120 unixms) == Seq(relayed1, relayed2))
-
- val postMigrationDb = new SqliteAuditDb(connection)
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- val ps2 = PaymentSent(UUID.randomUUID(), randomBytes32(), randomBytes32(), 1100 msat, randomKey().publicKey, Seq(
- PaymentSent.PartialPayment(UUID.randomUUID(), 500 msat, 10 msat, randomBytes32(), None, 160 unixms),
- PaymentSent.PartialPayment(UUID.randomUUID(), 600 msat, 5 msat, randomBytes32(), None, 165 unixms)
- ), None)
- val relayed3 = TrampolinePaymentRelayed(randomBytes32(), Seq(PaymentRelayed.IncomingPart(450 msat, randomBytes32(), 150 unixms), PaymentRelayed.IncomingPart(500 msat, randomBytes32(), 150 unixms)), Seq(PaymentRelayed.OutgoingPart(800 msat, randomBytes32(), 150 unixms)), randomKey().publicKey, 700 msat)
- postMigrationDb.add(ps2)
- assert(postMigrationDb.listSent(155 unixms, 200 unixms) == Seq(ps2))
- postMigrationDb.add(relayed3)
- assert(postMigrationDb.listRelayed(100 unixms, 160 unixms) == Seq(relayed1, relayed2, relayed3))
- }
- )
- }
-
- test("migrate audit database v4 -> current") {
- val relayed1 = ChannelPaymentRelayed(600 msat, 500 msat, randomBytes32(), randomBytes32(), randomBytes32(), 105 unixms, 105 unixms)
- // We weren't properly storing the outgoing trampoline node, which makes this useless, so we'll skip it when migrating.
- val relayed2 = TrampolinePaymentRelayed(randomBytes32(), Seq(PaymentRelayed.IncomingPart(300 msat, randomBytes32(), 110 unixms), PaymentRelayed.IncomingPart(350 msat, randomBytes32(), 110 unixms)), Seq(PaymentRelayed.OutgoingPart(600 msat, randomBytes32(), 110 unixms)), randomKey().publicKey, 0 msat)
-
- forAllDbs {
- case dbs: TestPgDatabases =>
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS sent (amount_msat BIGINT NOT NULL, fees_msat BIGINT NOT NULL, recipient_amount_msat BIGINT NOT NULL, payment_id TEXT NOT NULL, parent_payment_id TEXT NOT NULL, payment_hash TEXT NOT NULL, payment_preimage TEXT NOT NULL, recipient_node_id TEXT NOT NULL, to_channel_id TEXT NOT NULL, timestamp BIGINT NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS received (amount_msat BIGINT NOT NULL, payment_hash TEXT NOT NULL, from_channel_id TEXT NOT NULL, timestamp BIGINT NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS relayed (payment_hash TEXT NOT NULL, amount_msat BIGINT NOT NULL, channel_id TEXT NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp BIGINT NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS network_fees (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, tx_id TEXT NOT NULL, fee_sat BIGINT NOT NULL, tx_type TEXT NOT NULL, timestamp BIGINT NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_events (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, capacity_sat BIGINT NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp BIGINT NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_errors (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal BOOLEAN NOT NULL, timestamp BIGINT NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_payment_hash_idx ON relayed(payment_hash)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_events_timestamp_idx ON channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_errors_timestamp_idx ON channel_errors(timestamp)")
-
- setVersion(statement, "audit", 4)
- }
-
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setString(1, relayed1.paymentHash.toHex)
- statement.setLong(2, relayed1.amountIn.toLong)
- statement.setString(3, relayed1.fromChannelId.toHex)
- statement.setString(4, "IN")
- statement.setString(5, "channel")
- statement.setLong(6, relayed1.timestamp.toLong)
- statement.executeUpdate()
- }
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setString(1, relayed1.paymentHash.toHex)
- statement.setLong(2, relayed1.amountOut.toLong)
- statement.setString(3, relayed1.toChannelId.toHex)
- statement.setString(4, "OUT")
- statement.setString(5, "channel")
- statement.setLong(6, relayed1.timestamp.toLong)
- statement.executeUpdate()
- }
- for (incoming <- relayed2.incoming) {
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setString(1, relayed2.paymentHash.toHex)
- statement.setLong(2, incoming.amount.toLong)
- statement.setString(3, incoming.channelId.toHex)
- statement.setString(4, "IN")
- statement.setString(5, "trampoline")
- statement.setLong(6, relayed2.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- for (outgoing <- relayed2.outgoing) {
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setString(1, relayed2.paymentHash.toHex)
- statement.setLong(2, outgoing.amount.toLong)
- statement.setString(3, outgoing.channelId.toHex)
- statement.setString(4, "OUT")
- statement.setString(5, "trampoline")
- statement.setLong(6, relayed2.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- },
- dbName = PgAuditDb.DB_NAME,
- targetVersion = PgAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
-
- assert(migratedDb.listRelayed(100 unixms, 120 unixms) == Seq(relayed1))
-
- val postMigrationDb = new PgAuditDb()(dbs.datasource)
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(PgAuditDb.CURRENT_VERSION))
- }
- val relayed3 = TrampolinePaymentRelayed(randomBytes32(), Seq(PaymentRelayed.IncomingPart(450 msat, randomBytes32(), 150 unixms), PaymentRelayed.IncomingPart(500 msat, randomBytes32(), 150 unixms)), Seq(PaymentRelayed.OutgoingPart(800 msat, randomBytes32(), 150 unixms)), randomKey().publicKey, 700 msat)
- postMigrationDb.add(relayed3)
- assert(postMigrationDb.listRelayed(100 unixms, 160 unixms) == Seq(relayed1, relayed3))
- }
- )
- case dbs: TestSqliteDatabases =>
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS sent (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, recipient_amount_msat INTEGER NOT NULL, payment_id TEXT NOT NULL, parent_payment_id TEXT NOT NULL, payment_hash BLOB NOT NULL, payment_preimage BLOB NOT NULL, recipient_node_id BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS received (amount_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS relayed (payment_hash BLOB NOT NULL, amount_msat INTEGER NOT NULL, channel_id BLOB NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS network_fees (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, tx_id BLOB NOT NULL, fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE IF NOT EXISTS channel_errors (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS relayed_payment_hash_idx ON relayed(payment_hash)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_events_timestamp_idx ON channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX IF NOT EXISTS channel_errors_timestamp_idx ON channel_errors(timestamp)")
-
- setVersion(statement, "audit", 4)
- }
-
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setBytes(1, relayed1.paymentHash.toArray)
- statement.setLong(2, relayed1.amountIn.toLong)
- statement.setBytes(3, relayed1.fromChannelId.toArray)
- statement.setString(4, "IN")
- statement.setString(5, "channel")
- statement.setLong(6, relayed1.timestamp.toLong)
- statement.executeUpdate()
- }
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setBytes(1, relayed1.paymentHash.toArray)
- statement.setLong(2, relayed1.amountOut.toLong)
- statement.setBytes(3, relayed1.toChannelId.toArray)
- statement.setString(4, "OUT")
- statement.setString(5, "channel")
- statement.setLong(6, relayed1.timestamp.toLong)
- statement.executeUpdate()
- }
- for (incoming <- relayed2.incoming) {
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setBytes(1, relayed2.paymentHash.toArray)
- statement.setLong(2, incoming.amount.toLong)
- statement.setBytes(3, incoming.channelId.toArray)
- statement.setString(4, "IN")
- statement.setString(5, "trampoline")
- statement.setLong(6, relayed2.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- for (outgoing <- relayed2.outgoing) {
- using(connection.prepareStatement("INSERT INTO relayed VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setBytes(1, relayed2.paymentHash.toArray)
- statement.setLong(2, outgoing.amount.toLong)
- statement.setBytes(3, outgoing.channelId.toArray)
- statement.setString(4, "OUT")
- statement.setString(5, "trampoline")
- statement.setLong(6, relayed2.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- },
- dbName = SqliteAuditDb.DB_NAME,
- targetVersion = SqliteAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- assert(migratedDb.listRelayed(100 unixms, 120 unixms) == Seq(relayed1))
-
- val postMigrationDb = new SqliteAuditDb(connection)
- using(connection.createStatement()) { statement =>
- assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION))
- }
- val relayed3 = TrampolinePaymentRelayed(randomBytes32(), Seq(PaymentRelayed.IncomingPart(450 msat, randomBytes32(), 150 unixms), PaymentRelayed.IncomingPart(500 msat, randomBytes32(), 150 unixms)), Seq(PaymentRelayed.OutgoingPart(800 msat, randomBytes32(), 150 unixms)), randomKey().publicKey, 700 msat)
- postMigrationDb.add(relayed3)
- assert(postMigrationDb.listRelayed(100 unixms, 160 unixms) == Seq(relayed1, relayed3))
- }
- )
- }
- }
-
- test("migrate audit database v7 -> current") {
- val networkFees = Seq(
- NetworkFee(randomKey().publicKey, randomBytes32(), randomBytes32(), 50 sat, "test-tx-1", 500 unixms),
- NetworkFee(randomKey().publicKey, randomBytes32(), randomBytes32(), 0 sat, "test-tx-2", 600 unixms),
- )
-
- forAllDbs {
- case dbs: TestPgDatabases =>
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE SCHEMA audit")
-
- statement.executeUpdate("CREATE TABLE audit.sent (amount_msat BIGINT NOT NULL, fees_msat BIGINT NOT NULL, recipient_amount_msat BIGINT NOT NULL, payment_id TEXT NOT NULL, parent_payment_id TEXT NOT NULL, payment_hash TEXT NOT NULL, payment_preimage TEXT NOT NULL, recipient_node_id TEXT NOT NULL, to_channel_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.received (amount_msat BIGINT NOT NULL, payment_hash TEXT NOT NULL, from_channel_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.relayed (payment_hash TEXT NOT NULL, amount_msat BIGINT NOT NULL, channel_id TEXT NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.relayed_trampoline (payment_hash TEXT NOT NULL, amount_msat BIGINT NOT NULL, next_node_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.network_fees (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, tx_id TEXT NOT NULL, fee_sat BIGINT NOT NULL, tx_type TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.channel_events (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, capacity_sat BIGINT NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.channel_updates (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, fee_base_msat BIGINT NOT NULL, fee_proportional_millionths BIGINT NOT NULL, cltv_expiry_delta BIGINT NOT NULL, htlc_minimum_msat BIGINT NOT NULL, htlc_maximum_msat BIGINT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.path_finding_metrics (amount_msat BIGINT NOT NULL, fees_msat BIGINT NOT NULL, status TEXT NOT NULL, duration_ms BIGINT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL, is_mpp BOOLEAN NOT NULL, experiment_name TEXT NOT NULL, recipient_node_id TEXT NOT NULL)")
-
- statement.executeUpdate("CREATE TABLE audit.channel_errors (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal BOOLEAN NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE INDEX sent_timestamp_idx ON audit.sent(timestamp)")
- statement.executeUpdate("CREATE INDEX received_timestamp_idx ON audit.received(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_timestamp_idx ON audit.relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_payment_hash_idx ON audit.relayed(payment_hash)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_timestamp_idx ON audit.relayed_trampoline(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_payment_hash_idx ON audit.relayed_trampoline(payment_hash)")
- statement.executeUpdate("CREATE INDEX network_fees_timestamp_idx ON audit.network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_events_timestamp_idx ON audit.channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_errors_timestamp_idx ON audit.channel_errors(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_updates_cid_idx ON audit.channel_updates(channel_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_nid_idx ON audit.channel_updates(node_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_timestamp_idx ON audit.channel_updates(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_status_idx ON audit.path_finding_metrics(status)")
- statement.executeUpdate("CREATE INDEX metrics_timestamp_idx ON audit.path_finding_metrics(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_mpp_idx ON audit.path_finding_metrics(is_mpp)")
- statement.executeUpdate("CREATE INDEX metrics_name_idx ON audit.path_finding_metrics(experiment_name)")
-
- setVersion(statement, "audit", 9)
- }
-
- // We insert some transactions in the table.
- // NB: the first transaction is explicitly duplicated to test the primary key addition.
- for (tx <- networkFees.head +: networkFees) {
- using(connection.prepareStatement("INSERT INTO audit.network_fees VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setString(1, tx.channelId.toHex)
- statement.setString(2, tx.remoteNodeId.value.toHex)
- statement.setString(3, tx.txId.toHex)
- statement.setLong(4, tx.fee.toLong)
- statement.setString(5, tx.txType)
- statement.setTimestamp(6, tx.timestamp.toSqlTimestamp)
- statement.executeUpdate()
- }
- }
- },
- dbName = PgAuditDb.DB_NAME,
- targetVersion = PgAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
- using(connection.createStatement()) { statement => assert(getVersion(statement, "audit").contains(PgAuditDb.CURRENT_VERSION)) }
- assert(migratedDb.listNetworkFees(0 unixms, 700 unixms) == networkFees)
- }
- )
- case dbs: TestSqliteDatabases =>
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing previous version db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE TABLE sent (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, recipient_amount_msat INTEGER NOT NULL, payment_id TEXT NOT NULL, parent_payment_id TEXT NOT NULL, payment_hash BLOB NOT NULL, payment_preimage BLOB NOT NULL, recipient_node_id BLOB NOT NULL, to_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE received (amount_msat INTEGER NOT NULL, payment_hash BLOB NOT NULL, from_channel_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE relayed (payment_hash BLOB NOT NULL, amount_msat INTEGER NOT NULL, channel_id BLOB NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE relayed_trampoline (payment_hash BLOB NOT NULL, amount_msat INTEGER NOT NULL, next_node_id BLOB NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE network_fees (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, tx_id BLOB NOT NULL, fee_sat INTEGER NOT NULL, tx_type TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE channel_events (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, capacity_sat INTEGER NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE channel_errors (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE channel_updates (channel_id BLOB NOT NULL, node_id BLOB NOT NULL, fee_base_msat INTEGER NOT NULL, fee_proportional_millionths INTEGER NOT NULL, cltv_expiry_delta INTEGER NOT NULL, htlc_minimum_msat INTEGER NOT NULL, htlc_maximum_msat INTEGER NOT NULL, timestamp INTEGER NOT NULL)")
- statement.executeUpdate("CREATE TABLE path_finding_metrics (amount_msat INTEGER NOT NULL, fees_msat INTEGER NOT NULL, status TEXT NOT NULL, duration_ms INTEGER NOT NULL, timestamp INTEGER NOT NULL, is_mpp INTEGER NOT NULL, experiment_name TEXT NOT NULL, recipient_node_id BLOB NOT NULL)")
-
- statement.executeUpdate("CREATE INDEX sent_timestamp_idx ON sent(timestamp)")
- statement.executeUpdate("CREATE INDEX received_timestamp_idx ON received(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_timestamp_idx ON relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_payment_hash_idx ON relayed(payment_hash)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_timestamp_idx ON relayed_trampoline(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_payment_hash_idx ON relayed_trampoline(payment_hash)")
- statement.executeUpdate("CREATE INDEX network_fees_timestamp_idx ON network_fees(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_events_timestamp_idx ON channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_errors_timestamp_idx ON channel_errors(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_updates_cid_idx ON channel_updates(channel_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_nid_idx ON channel_updates(node_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_timestamp_idx ON channel_updates(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_status_idx ON path_finding_metrics(status)")
- statement.executeUpdate("CREATE INDEX metrics_timestamp_idx ON path_finding_metrics(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_mpp_idx ON path_finding_metrics(is_mpp)")
- statement.executeUpdate("CREATE INDEX metrics_name_idx ON path_finding_metrics(experiment_name)")
-
- setVersion(statement, "audit", 7)
- }
-
- // We insert some transactions in the table.
- // NB: the first transaction is explicitly duplicated to test the primary key addition.
- for (tx <- networkFees.head +: networkFees) {
- using(connection.prepareStatement("INSERT INTO network_fees VALUES (?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setBytes(1, tx.channelId.toArray)
- statement.setBytes(2, tx.remoteNodeId.value.toArray)
- statement.setBytes(3, tx.txId.toArray)
- statement.setLong(4, tx.fee.toLong)
- statement.setString(5, tx.txType)
- statement.setLong(6, tx.timestamp.toLong)
- statement.executeUpdate()
- }
- }
- },
- dbName = SqliteAuditDb.DB_NAME,
- targetVersion = SqliteAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- val migratedDb = dbs.audit
- using(connection.createStatement()) { statement => assert(getVersion(statement, "audit").contains(SqliteAuditDb.CURRENT_VERSION)) }
- assert(migratedDb.listNetworkFees(0 unixms, 700 unixms) == networkFees)
- }
- )
- }
- }
-
- test("migrate postgres audit database v10 -> current") {
- forAllDbs {
- case dbs: TestPgDatabases =>
- migrationCheck(
- dbs = dbs,
- initializeTables = connection => {
- // simulate existing v10 db
- using(connection.createStatement()) { statement =>
- statement.executeUpdate("CREATE SCHEMA audit")
-
- statement.executeUpdate("CREATE TABLE audit.sent (amount_msat BIGINT NOT NULL, fees_msat BIGINT NOT NULL, recipient_amount_msat BIGINT NOT NULL, payment_id TEXT NOT NULL, parent_payment_id TEXT NOT NULL, payment_hash TEXT NOT NULL, payment_preimage TEXT NOT NULL, recipient_node_id TEXT NOT NULL, to_channel_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.received (amount_msat BIGINT NOT NULL, payment_hash TEXT NOT NULL, from_channel_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.relayed (payment_hash TEXT NOT NULL, amount_msat BIGINT NOT NULL, channel_id TEXT NOT NULL, direction TEXT NOT NULL, relay_type TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.relayed_trampoline (payment_hash TEXT NOT NULL, amount_msat BIGINT NOT NULL, next_node_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.channel_events (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, capacity_sat BIGINT NOT NULL, is_funder BOOLEAN NOT NULL, is_private BOOLEAN NOT NULL, event TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.channel_updates (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, fee_base_msat BIGINT NOT NULL, fee_proportional_millionths BIGINT NOT NULL, cltv_expiry_delta BIGINT NOT NULL, htlc_minimum_msat BIGINT NOT NULL, htlc_maximum_msat BIGINT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.path_finding_metrics (amount_msat BIGINT NOT NULL, fees_msat BIGINT NOT NULL, status TEXT NOT NULL, duration_ms BIGINT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL, is_mpp BOOLEAN NOT NULL, experiment_name TEXT NOT NULL, recipient_node_id TEXT NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.transactions_published (tx_id TEXT NOT NULL PRIMARY KEY, channel_id TEXT NOT NULL, node_id TEXT NOT NULL, mining_fee_sat BIGINT NOT NULL, tx_type TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE TABLE audit.transactions_confirmed (tx_id TEXT NOT NULL PRIMARY KEY, channel_id TEXT NOT NULL, node_id TEXT NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
-
- statement.executeUpdate("CREATE TABLE audit.channel_errors (channel_id TEXT NOT NULL, node_id TEXT NOT NULL, error_name TEXT NOT NULL, error_message TEXT NOT NULL, is_fatal BOOLEAN NOT NULL, timestamp TIMESTAMP WITH TIME ZONE NOT NULL)")
- statement.executeUpdate("CREATE INDEX sent_timestamp_idx ON audit.sent(timestamp)")
- statement.executeUpdate("CREATE INDEX received_timestamp_idx ON audit.received(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_timestamp_idx ON audit.relayed(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_payment_hash_idx ON audit.relayed(payment_hash)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_timestamp_idx ON audit.relayed_trampoline(timestamp)")
- statement.executeUpdate("CREATE INDEX relayed_trampoline_payment_hash_idx ON audit.relayed_trampoline(payment_hash)")
- statement.executeUpdate("CREATE INDEX channel_events_timestamp_idx ON audit.channel_events(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_errors_timestamp_idx ON audit.channel_errors(timestamp)")
- statement.executeUpdate("CREATE INDEX channel_updates_cid_idx ON audit.channel_updates(channel_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_nid_idx ON audit.channel_updates(node_id)")
- statement.executeUpdate("CREATE INDEX channel_updates_timestamp_idx ON audit.channel_updates(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_status_idx ON audit.path_finding_metrics(status)")
- statement.executeUpdate("CREATE INDEX metrics_timestamp_idx ON audit.path_finding_metrics(timestamp)")
- statement.executeUpdate("CREATE INDEX metrics_mpp_idx ON audit.path_finding_metrics(is_mpp)")
- statement.executeUpdate("CREATE INDEX metrics_name_idx ON audit.path_finding_metrics(experiment_name)")
- statement.executeUpdate("CREATE INDEX transactions_published_timestamp_idx ON audit.transactions_published(timestamp)")
- statement.executeUpdate("CREATE INDEX transactions_confirmed_timestamp_idx ON audit.transactions_confirmed(timestamp)")
-
- setVersion(statement, "audit", 10)
- }
-
- using(connection.prepareStatement("INSERT INTO audit.path_finding_metrics VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setLong(1, 214000)
- statement.setLong(2, 345)
- statement.setString(3, "FAILURE")
- statement.setLong(4, 520)
- statement.setTimestamp(5, TimestampSecond(1651053434L).toSqlTimestamp)
- statement.setBoolean(6, true)
- statement.setString(7, "experiment-a")
- statement.setString(8, "03271338633d2d37b285dae4df40b413d8c6c791fbee7797bc5dc70812196d7d5c")
- statement.executeUpdate()
- }
- using(connection.prepareStatement("INSERT INTO audit.path_finding_metrics VALUES (?, ?, ?, ?, ?, ?, ?, ?)")) { statement =>
- statement.setLong(1, 35000)
- statement.setLong(2, 43)
- statement.setString(3, "SUCCESS")
- statement.setLong(4, 2043)
- statement.setTimestamp(5, TimestampSecond(1651054567L).toSqlTimestamp)
- statement.setBoolean(6, false)
- statement.setString(7, "experiment-b")
- statement.setString(8, "030c3f19d742ca294a55c00376b3b355c3c90d61c6b6b39554dbc7ac19b141c14f")
- statement.executeUpdate()
- }
- },
- dbName = PgAuditDb.DB_NAME,
- targetVersion = PgAuditDb.CURRENT_VERSION,
- postCheck = connection => {
- using(connection.createStatement()) { statement => assert(getVersion(statement, "audit").contains(PgAuditDb.CURRENT_VERSION)) }
- using(connection.prepareStatement(s"SELECT amount_msat, status, experiment_name, recipient_node_id FROM audit.path_finding_metrics ORDER BY timestamp")) { statement =>
- val result = statement.executeQuery()
- assert(result.next())
- assert(result.getLong(1) == 214000)
- assert(result.getString(2) == "FAILURE")
- assert(result.getString(3) == "experiment-a")
- assert(result.getString(4) == "03271338633d2d37b285dae4df40b413d8c6c791fbee7797bc5dc70812196d7d5c")
- assert(result.next())
- assert(result.getLong(1) == 35000)
- assert(result.getString(2) == "SUCCESS")
- assert(result.getString(3) == "experiment-b")
- assert(result.getString(4) == "030c3f19d742ca294a55c00376b3b355c3c90d61c6b6b39554dbc7ac19b141c14f")
- assert(!result.next())
- }
- }
- )
- case _: TestSqliteDatabases => ()
- }
- }
-
test("ignore invalid values in the DB") {
forAllDbs { dbs =>
val db = dbs.audit
Risk score
Our methodology →Why this scored 18/100
Human-validated context
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
No validated notes yet.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.