ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2025-04-25 05:19:44 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

ENT-9927 Ledger Recovery: synchronise changes from ENT -> OS. (#7445)

Browse Source
This commit is contained in:
Jose Coll 2023-08-15 15:32:00 +01:00 committed by GitHub
parent e0e4f51ba2
commit 0130914c89
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 160 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
core-tests/src/test/kotlin/net/corda/coretests/flows/FinalityFlowTests.kt
View File

@ -354,12 +354,11 @@ class FinalityFlowTests : WithFinality {
getSenderRecoveryData(stx.id, aliceNode.database).apply { getSenderRecoveryData(stx.id, aliceNode.database).apply {
assertEquals(1, this.size) assertEquals(1, this.size)
assertEquals(StatesToRecord.ONLY_RELEVANT, this[0].statesToRecord) assertEquals(StatesToRecord.ALL_VISIBLE, this[0].statesToRecord)
assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId) assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId)
} }
getReceiverRecoveryData(stx.id, bobNode.database).apply { getReceiverRecoveryData(stx.id, bobNode.database).apply {
assertEquals(StatesToRecord.ALL_VISIBLE, this?.statesToRecord) assertEquals(StatesToRecord.ONLY_RELEVANT, this?.statesToRecord)
assertEquals(StatesToRecord.ONLY_RELEVANT, this?.senderStatesToRecord)
assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId) assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId)
assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ALL_VISIBLE), this?.peersToStatesToRecord) assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ALL_VISIBLE), this?.peersToStatesToRecord)
} }
@ -387,12 +386,10 @@ class FinalityFlowTests : WithFinality {
assertEquals(2, this.size) assertEquals(2, this.size)
assertEquals(StatesToRecord.ONLY_RELEVANT, this[0].statesToRecord) assertEquals(StatesToRecord.ONLY_RELEVANT, this[0].statesToRecord)
assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId) assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId)
assertEquals(StatesToRecord.ONLY_RELEVANT, this[1].statesToRecord) assertEquals(StatesToRecord.ALL_VISIBLE, this[1].statesToRecord)
assertEquals(CHARLIE_NAME.hashCode().toLong(), this[1].peerPartyId) assertEquals(CHARLIE_NAME.hashCode().toLong(), this[1].peerPartyId)
} }
getReceiverRecoveryData(stx.id, bobNode.database).apply { getReceiverRecoveryData(stx.id, bobNode.database).apply {
assertEquals(StatesToRecord.ONLY_RELEVANT, this?.statesToRecord)
assertEquals(StatesToRecord.ONLY_RELEVANT, this?.senderStatesToRecord)
assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId) assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId)
// note: Charlie assertion here is using the hinted StatesToRecord value passed to it from Alice // note: Charlie assertion here is using the hinted StatesToRecord value passed to it from Alice
assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ONLY_RELEVANT, assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ONLY_RELEVANT,
@ -434,8 +431,6 @@ class FinalityFlowTests : WithFinality {
assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId) assertEquals(BOB_NAME.hashCode().toLong(), this[0].peerPartyId)
} }
getReceiverRecoveryData(stx.id, bobNode.database).apply { getReceiverRecoveryData(stx.id, bobNode.database).apply {
assertEquals(StatesToRecord.ONLY_RELEVANT, this?.statesToRecord)
assertEquals(StatesToRecord.ONLY_RELEVANT, this?.senderStatesToRecord)
assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId) assertEquals(aliceNode.info.singleIdentity().name.hashCode().toLong(), this?.initiatorPartyId)
assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ONLY_RELEVANT), this?.peersToStatesToRecord) assertEquals(mapOf(BOB_NAME.hashCode().toLong() to StatesToRecord.ONLY_RELEVANT), this?.peersToStatesToRecord)
} }

174
node/src/main/kotlin/net/corda/node/services/persistence/DBTransactionStorageLedgerRecovery.kt
View File

@ -5,6 +5,7 @@ import net.corda.core.flows.RecoveryTimeWindow
import net.corda.core.flows.TransactionMetadata import net.corda.core.flows.TransactionMetadata
import net.corda.core.identity.CordaX500Name import net.corda.core.identity.CordaX500Name
import net.corda.core.internal.NamedCacheFactory import net.corda.core.internal.NamedCacheFactory
import net.corda.core.internal.VisibleForTesting
import net.corda.core.node.StatesToRecord import net.corda.core.node.StatesToRecord
import net.corda.core.node.services.vault.Sort import net.corda.core.node.services.vault.Sort
import net.corda.core.serialization.CordaSerializable import net.corda.core.serialization.CordaSerializable
@ -20,7 +21,7 @@ import java.io.DataInputStream
import java.io.DataOutputStream import java.io.DataOutputStream
import java.io.Serializable import java.io.Serializable
import java.time.Instant import java.time.Instant
import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicInteger
import javax.persistence.Column import javax.persistence.Column
import javax.persistence.Embeddable import javax.persistence.Embeddable
import javax.persistence.EmbeddedId import javax.persistence.EmbeddedId
@ -33,20 +34,26 @@ import kotlin.streams.toList
class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence, cacheFactory: NamedCacheFactory, class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence, cacheFactory: NamedCacheFactory,
val clock: CordaClock, val clock: CordaClock,
private val cryptoService: CryptoService, val cryptoService: CryptoService,
private val partyInfoCache: PersistentPartyInfoCache) : DBTransactionStorage(database, cacheFactory, clock) { private val partyInfoCache: PersistentPartyInfoCache) : DBTransactionStorage(database, cacheFactory, clock) {
@Embeddable @Embeddable
@Immutable @Immutable
data class PersistentKey( data class PersistentKey(
@Column(name = "sequence_number", nullable = false) /** PartyId of flow peer **/
var sequenceNumber: Long, @Column(name = "peer_party_id", nullable = false)
var peerPartyId: Long,
@Column(name = "timestamp", nullable = false) @Column(name = "timestamp", nullable = false)
var timestamp: Instant var timestamp: Instant,
@Column(name = "timestamp_discriminator", nullable = false)
var timestampDiscriminator: Int
) : Serializable { ) : Serializable {
constructor(key: Key) : this(key.sequenceNumber, key.timestamp) constructor(key: Key) : this(key.partyId, key.timestamp, key.timestampDiscriminator)
} }
@CordaSerializable
@Entity @Entity
@Table(name = "${NODE_DATABASE_PREFIX}sender_distribution_records") @Table(name = "${NODE_DATABASE_PREFIX}sender_distribution_records")
data class DBSenderDistributionRecord( data class DBSenderDistributionRecord(
@ -56,10 +63,6 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
@Column(name = "transaction_id", length = 144, nullable = false) @Column(name = "transaction_id", length = 144, nullable = false)
var txId: String, var txId: String,
/** PartyId of flow peer **/
@Column(name = "receiver_party_id", nullable = false)
val receiverPartyId: Long,
/** states to record: NONE, ALL_VISIBLE, ONLY_RELEVANT */ /** states to record: NONE, ALL_VISIBLE, ONLY_RELEVANT */
@Column(name = "states_to_record", nullable = false) @Column(name = "states_to_record", nullable = false)
var statesToRecord: StatesToRecord var statesToRecord: StatesToRecord
@ -68,12 +71,13 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
fun toSenderDistributionRecord() = fun toSenderDistributionRecord() =
SenderDistributionRecord( SenderDistributionRecord(
SecureHash.parse(this.txId), SecureHash.parse(this.txId),
this.receiverPartyId, this.compositeKey.peerPartyId,
this.statesToRecord, this.statesToRecord,
this.compositeKey.timestamp this.compositeKey.timestamp
) )
} }
@CordaSerializable
@Entity @Entity
@Table(name = "${NODE_DATABASE_PREFIX}receiver_distribution_records") @Table(name = "${NODE_DATABASE_PREFIX}receiver_distribution_records")
data class DBReceiverDistributionRecord( data class DBReceiverDistributionRecord(
@ -83,34 +87,23 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
@Column(name = "transaction_id", length = 144, nullable = false) @Column(name = "transaction_id", length = 144, nullable = false)
var txId: String, var txId: String,
/** PartyId of flow initiator **/
@Column(name = "sender_party_id", nullable = true)
val senderPartyId: Long,
/** Encrypted recovery information for sole use by Sender **/ /** Encrypted recovery information for sole use by Sender **/
@Lob @Lob
@Column(name = "distribution_list", nullable = false) @Column(name = "distribution_list", nullable = false)
val distributionList: ByteArray, val distributionList: ByteArray
/** states to record: NONE, ALL_VISIBLE, ONLY_RELEVANT */
@Column(name = "receiver_states_to_record", nullable = false)
val receiverStatesToRecord: StatesToRecord
) { ) {
constructor(key: Key, txId: SecureHash, initiatorPartyId: Long, encryptedDistributionList: ByteArray, receiverStatesToRecord: StatesToRecord) : constructor(key: Key, txId: SecureHash, encryptedDistributionList: ByteArray) :
this(PersistentKey(key), this(PersistentKey(key),
txId = txId.toString(), txId = txId.toString(),
senderPartyId = initiatorPartyId, distributionList = encryptedDistributionList
distributionList = encryptedDistributionList,
receiverStatesToRecord = receiverStatesToRecord
) )
fun toReceiverDistributionRecord(cryptoService: CryptoService): ReceiverDistributionRecord { fun toReceiverDistributionRecord(cryptoService: CryptoService): ReceiverDistributionRecord {
val hashedDL = HashedDistributionList.deserialize(cryptoService.decrypt(this.distributionList)) val hashedDL = HashedDistributionList.deserialize(cryptoService.decrypt(this.distributionList))
return ReceiverDistributionRecord( return ReceiverDistributionRecord(
SecureHash.parse(this.txId), SecureHash.parse(this.txId),
this.senderPartyId, this.compositeKey.peerPartyId,
hashedDL.peerHashToStatesToRecord, hashedDL.peerHashToStatesToRecord,
this.receiverStatesToRecord,
hashedDL.senderStatesToRecord, hashedDL.senderStatesToRecord,
this.compositeKey.timestamp this.compositeKey.timestamp
) )
@ -130,23 +123,29 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
val partyName: String val partyName: String
) )
data class TimestampKey(val timestamp: Instant, val timestampDiscriminator: Int)
class Key( class Key(
val partyId: Long,
val timestamp: Instant, val timestamp: Instant,
val sequenceNumber: Long = nextSequenceNumber.andIncrement val timestampDiscriminator: Int = nextDiscriminatorNumber.andIncrement
) { ) {
constructor(key: TimestampKey, partyId: Long): this(partyId = partyId, timestamp = key.timestamp, timestampDiscriminator = key.timestampDiscriminator)
companion object { companion object {
private val nextSequenceNumber = AtomicLong() val nextDiscriminatorNumber = AtomicInteger()
} }
} }
override fun addSenderTransactionRecoveryMetadata(id: SecureHash, metadata: TransactionMetadata): ByteArray { override fun addSenderTransactionRecoveryMetadata(id: SecureHash, metadata: TransactionMetadata): ByteArray {
return database.transaction {
val senderRecordingTimestamp = clock.instant() val senderRecordingTimestamp = clock.instant()
metadata.distributionList.peersToStatesToRecord.forEach { (peer, _) -> return database.transaction {
val senderDistributionRecord = DBSenderDistributionRecord(PersistentKey(Key(senderRecordingTimestamp)), // sender distribution records must be unique per txnId and timestamp
val timeDiscriminator = Key.nextDiscriminatorNumber.andIncrement
metadata.distributionList.peersToStatesToRecord.map { (peerCordaX500Name, peerStatesToRecord) ->
val senderDistributionRecord = DBSenderDistributionRecord(
PersistentKey(Key(TimestampKey(senderRecordingTimestamp, timeDiscriminator), partyInfoCache.getPartyIdByCordaX500Name(peerCordaX500Name))),
id.toString(), id.toString(),
partyInfoCache.getPartyIdByCordaX500Name(peer), peerStatesToRecord)
metadata.distributionList.senderStatesToRecord)
session.save(senderDistributionRecord) session.save(senderDistributionRecord)
} }
val hashedPeersToStatesToRecord = metadata.distributionList.peersToStatesToRecord.map { (peer, statesToRecord) -> val hashedPeersToStatesToRecord = metadata.distributionList.peersToStatesToRecord.map { (peer, statesToRecord) ->
@ -156,19 +155,48 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
} }
} }
fun createReceiverTransactionRecoverMetadata(txId: SecureHash,
senderPartyId: Long,
senderStatesToRecord: StatesToRecord,
senderRecords: List<DBSenderDistributionRecord>): List<DBReceiverDistributionRecord> {
val senderRecordsByTimestampKey = senderRecords.groupBy { TimestampKey(it.compositeKey.timestamp, it.compositeKey.timestampDiscriminator) }
return senderRecordsByTimestampKey.map {
val hashedDistributionList = HashedDistributionList(
senderStatesToRecord = senderStatesToRecord,
peerHashToStatesToRecord = senderRecords.map { it.compositeKey.peerPartyId to it.statesToRecord }.toMap(),
senderRecordedTimestamp = it.key.timestamp
)
DBReceiverDistributionRecord(
compositeKey = PersistentKey(Key(TimestampKey(it.key.timestamp, it.key.timestampDiscriminator), senderPartyId)),
txId = txId.toString(),
distributionList = cryptoService.encrypt(hashedDistributionList.serialize())
)
}
}
fun addSenderTransactionRecoveryMetadata(record: DBSenderDistributionRecord) {
return database.transaction {
session.save(record)
}
}
override fun addReceiverTransactionRecoveryMetadata(id: SecureHash, sender: CordaX500Name, receiver: CordaX500Name, receiverStatesToRecord: StatesToRecord, encryptedDistributionList: ByteArray) { override fun addReceiverTransactionRecoveryMetadata(id: SecureHash, sender: CordaX500Name, receiver: CordaX500Name, receiverStatesToRecord: StatesToRecord, encryptedDistributionList: ByteArray) {
val senderRecordedTimestamp = HashedDistributionList.deserialize(cryptoService.decrypt(encryptedDistributionList)).senderRecordedTimestamp val senderRecordedTimestamp = HashedDistributionList.deserialize(cryptoService.decrypt(encryptedDistributionList)).senderRecordedTimestamp
database.transaction { database.transaction {
val receiverDistributionRecord = val receiverDistributionRecord =
DBReceiverDistributionRecord(Key(senderRecordedTimestamp), DBReceiverDistributionRecord(Key(partyInfoCache.getPartyIdByCordaX500Name(sender), senderRecordedTimestamp),
id, id,
partyInfoCache.getPartyIdByCordaX500Name(sender), encryptedDistributionList)
encryptedDistributionList,
receiverStatesToRecord)
session.save(receiverDistributionRecord) session.save(receiverDistributionRecord)
} }
} }
fun addReceiverTransactionRecoveryMetadata(record: DBReceiverDistributionRecord) {
return database.transaction {
session.save(record)
}
}
override fun removeUnnotarisedTransaction(id: SecureHash): Boolean { override fun removeUnnotarisedTransaction(id: SecureHash): Boolean {
return database.transaction { return database.transaction {
super.removeUnnotarisedTransaction(id) super.removeUnnotarisedTransaction(id)
@ -187,27 +215,30 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
fun queryDistributionRecords(timeWindow: RecoveryTimeWindow, fun queryDistributionRecords(timeWindow: RecoveryTimeWindow,
recordType: DistributionRecordType = DistributionRecordType.ALL, recordType: DistributionRecordType = DistributionRecordType.ALL,
excludingTxnIds: Set<SecureHash>? = null, excludingTxnIds: Set<SecureHash> = emptySet(),
orderByTimestamp: Sort.Direction? = null orderByTimestamp: Sort.Direction? = null
): List<DistributionRecord> { ): DistributionRecords {
return when(recordType) { return when(recordType) {
DistributionRecordType.SENDER -> DistributionRecordType.SENDER ->
querySenderDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp) DistributionRecords(senderRecords =
querySenderDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp))
DistributionRecordType.RECEIVER -> DistributionRecordType.RECEIVER ->
queryReceiverDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp) DistributionRecords(receiverRecords =
queryReceiverDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp))
DistributionRecordType.ALL -> DistributionRecordType.ALL ->
querySenderDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp).plus( DistributionRecords(senderRecords =
queryReceiverDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp) querySenderDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp),
) receiverRecords =
queryReceiverDistributionRecords(timeWindow, excludingTxnIds = excludingTxnIds, orderByTimestamp = orderByTimestamp))
} }
} }
@Suppress("SpreadOperator") @Suppress("SpreadOperator")
fun querySenderDistributionRecords(timeWindow: RecoveryTimeWindow, fun querySenderDistributionRecords(timeWindow: RecoveryTimeWindow,
peers: Set<CordaX500Name> = emptySet(), peers: Set<CordaX500Name> = emptySet(),
excludingTxnIds: Set<SecureHash>? = null, excludingTxnIds: Set<SecureHash> = emptySet(),
orderByTimestamp: Sort.Direction? = null orderByTimestamp: Sort.Direction? = null
): List<SenderDistributionRecord> { ): List<DBSenderDistributionRecord> {
return database.transaction { return database.transaction {
val criteriaBuilder = session.criteriaBuilder val criteriaBuilder = session.criteriaBuilder
val criteriaQuery = criteriaBuilder.createQuery(DBSenderDistributionRecord::class.java) val criteriaQuery = criteriaBuilder.createQuery(DBSenderDistributionRecord::class.java)
@ -216,13 +247,13 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
val compositeKey = txnMetadata.get<PersistentKey>("compositeKey") val compositeKey = txnMetadata.get<PersistentKey>("compositeKey")
predicates.add(criteriaBuilder.greaterThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.fromTime)) predicates.add(criteriaBuilder.greaterThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.fromTime))
predicates.add(criteriaBuilder.and(criteriaBuilder.lessThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.untilTime))) predicates.add(criteriaBuilder.and(criteriaBuilder.lessThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.untilTime)))
excludingTxnIds?.let { excludingTxnIds -> if (excludingTxnIds.isNotEmpty()) {
predicates.add(criteriaBuilder.and(criteriaBuilder.notEqual(txnMetadata.get<String>(DBSenderDistributionRecord::txId.name), predicates.add(criteriaBuilder.and(criteriaBuilder.not(txnMetadata.get<String>(DBSenderDistributionRecord::txId.name).`in`(
excludingTxnIds.map { it.toString() }))) excludingTxnIds.map { it.toString() }))))
} }
if (peers.isNotEmpty()) { if (peers.isNotEmpty()) {
val peerPartyIds = peers.map { partyInfoCache.getPartyIdByCordaX500Name(it) } val peerPartyIds = peers.map { partyInfoCache.getPartyIdByCordaX500Name(it) }
predicates.add(criteriaBuilder.and(txnMetadata.get<Long>(DBSenderDistributionRecord::receiverPartyId.name).`in`(peerPartyIds))) predicates.add(criteriaBuilder.and(compositeKey.get<Long>(PersistentKey::peerPartyId.name).`in`(peerPartyIds)))
} }
criteriaQuery.where(*predicates.toTypedArray()) criteriaQuery.where(*predicates.toTypedArray())
// optionally order by timestamp // optionally order by timestamp
@ -236,16 +267,27 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
criteriaQuery.orderBy(orderCriteria) criteriaQuery.orderBy(orderCriteria)
} }
val results = session.createQuery(criteriaQuery).stream() val results = session.createQuery(criteriaQuery).stream()
results.map { it.toSenderDistributionRecord() }.toList() results.toList()
}
}
fun querySenderDistributionRecordsByTxId(txId: SecureHash): List<DBSenderDistributionRecord> {
return database.transaction {
val criteriaBuilder = session.criteriaBuilder
val criteriaQuery = criteriaBuilder.createQuery(DBSenderDistributionRecord::class.java)
val txnMetadata = criteriaQuery.from(DBSenderDistributionRecord::class.java)
criteriaQuery.where(criteriaBuilder.equal(txnMetadata.get<String>(DBSenderDistributionRecord::txId.name), txId.toString()))
val results = session.createQuery(criteriaQuery).stream()
results.toList()
} }
} }
@Suppress("SpreadOperator") @Suppress("SpreadOperator")
fun queryReceiverDistributionRecords(timeWindow: RecoveryTimeWindow, fun queryReceiverDistributionRecords(timeWindow: RecoveryTimeWindow,
initiators: Set<CordaX500Name> = emptySet(), initiators: Set<CordaX500Name> = emptySet(),
excludingTxnIds: Set<SecureHash>? = null, excludingTxnIds: Set<SecureHash> = emptySet(),
orderByTimestamp: Sort.Direction? = null orderByTimestamp: Sort.Direction? = null
): List<ReceiverDistributionRecord> { ): List<DBReceiverDistributionRecord> {
return database.transaction { return database.transaction {
val criteriaBuilder = session.criteriaBuilder val criteriaBuilder = session.criteriaBuilder
val criteriaQuery = criteriaBuilder.createQuery(DBReceiverDistributionRecord::class.java) val criteriaQuery = criteriaBuilder.createQuery(DBReceiverDistributionRecord::class.java)
@ -254,13 +296,13 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
val compositeKey = txnMetadata.get<PersistentKey>("compositeKey") val compositeKey = txnMetadata.get<PersistentKey>("compositeKey")
predicates.add(criteriaBuilder.greaterThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.fromTime)) predicates.add(criteriaBuilder.greaterThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.fromTime))
predicates.add(criteriaBuilder.and(criteriaBuilder.lessThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.untilTime))) predicates.add(criteriaBuilder.and(criteriaBuilder.lessThanOrEqualTo(compositeKey.get<Instant>(PersistentKey::timestamp.name), timeWindow.untilTime)))
excludingTxnIds?.let { excludingTxnIds -> if (excludingTxnIds.isNotEmpty()) {
predicates.add(criteriaBuilder.and(criteriaBuilder.notEqual(txnMetadata.get<String>(DBReceiverDistributionRecord::txId.name), predicates.add(criteriaBuilder.and(criteriaBuilder.not(txnMetadata.get<String>(DBSenderDistributionRecord::txId.name).`in`(
excludingTxnIds.map { it.toString() }))) excludingTxnIds.map { it.toString() }))))
} }
if (initiators.isNotEmpty()) { if (initiators.isNotEmpty()) {
val initiatorPartyIds = initiators.map { partyInfoCache.getPartyIdByCordaX500Name(it) } val initiatorPartyIds = initiators.map { partyInfoCache.getPartyIdByCordaX500Name(it) }
predicates.add(criteriaBuilder.and(txnMetadata.get<Long>(DBReceiverDistributionRecord::senderPartyId.name).`in`(initiatorPartyIds))) predicates.add(criteriaBuilder.and(compositeKey.get<Long>(PersistentKey::peerPartyId.name).`in`(initiatorPartyIds)))
} }
criteriaQuery.where(*predicates.toTypedArray()) criteriaQuery.where(*predicates.toTypedArray())
// optionally order by timestamp // optionally order by timestamp
@ -274,13 +316,14 @@ class DBTransactionStorageLedgerRecovery(private val database: CordaPersistence,
criteriaQuery.orderBy(orderCriteria) criteriaQuery.orderBy(orderCriteria)
} }
val results = session.createQuery(criteriaQuery).stream() val results = session.createQuery(criteriaQuery).stream()
results.map { it.toReceiverDistributionRecord(cryptoService) }.toList() results.toList()
} }
} }
} }
// TO DO: https://r3-cev.atlassian.net/browse/ENT-9876 // TO DO: https://r3-cev.atlassian.net/browse/ENT-9876
private fun CryptoService.decrypt(bytes: ByteArray): ByteArray { @VisibleForTesting
fun CryptoService.decrypt(bytes: ByteArray): ByteArray {
return bytes return bytes
} }
@ -289,6 +332,18 @@ fun CryptoService.encrypt(bytes: ByteArray): ByteArray {
return bytes return bytes
} }
@CordaSerializable
class DistributionRecords(
val senderRecords: List<DBTransactionStorageLedgerRecovery.DBSenderDistributionRecord> = emptyList(),
val receiverRecords: List<DBTransactionStorageLedgerRecovery.DBReceiverDistributionRecord> = emptyList()
) {
init {
assert(senderRecords.isNotEmpty() || receiverRecords.isNotEmpty()) { "Must set senderRecords or receiverRecords or both." }
}
val size = senderRecords.size + receiverRecords.size
}
@CordaSerializable @CordaSerializable
open class DistributionRecord( open class DistributionRecord(
open val txId: SecureHash, open val txId: SecureHash,
@ -310,7 +365,6 @@ data class ReceiverDistributionRecord(
val initiatorPartyId: Long, // CordaX500Name hashCode() val initiatorPartyId: Long, // CordaX500Name hashCode()
val peersToStatesToRecord: Map<Long, StatesToRecord>, // CordaX500Name hashCode() -> StatesToRecord val peersToStatesToRecord: Map<Long, StatesToRecord>, // CordaX500Name hashCode() -> StatesToRecord
override val statesToRecord: StatesToRecord, override val statesToRecord: StatesToRecord,
val senderStatesToRecord: StatesToRecord,
override val timestamp: Instant override val timestamp: Instant
) : DistributionRecord(txId, statesToRecord, timestamp) ) : DistributionRecord(txId, statesToRecord, timestamp)

47
node/src/main/resources/migration/node-core.changelog-v25.xml
View File

@ -12,16 +12,16 @@
<changeSet author="R3.Corda" id="create_sender_distribution_records_table"> <changeSet author="R3.Corda" id="create_sender_distribution_records_table">
<createTable tableName="node_sender_distribution_records"> <createTable tableName="node_sender_distribution_records">
<column name="sequence_number" type="BIGINT"> <column name="timestamp" type="TIMESTAMP">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="timestamp" type="TIMESTAMP"> <column name="timestamp_discriminator" type="INT">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="transaction_id" type="NVARCHAR(144)"> <column name="transaction_id" type="NVARCHAR(144)">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="receiver_party_id" type="BIGINT"> <column name="peer_party_id" type="BIGINT">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="states_to_record" type="INT"> <column name="states_to_record" type="INT">
@ -31,54 +31,35 @@
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="node_sender_distribution_records_pkey"> <changeSet author="R3.Corda" id="node_sender_distribution_records_pkey">
<addPrimaryKey columnNames="timestamp, sequence_number" constraintName="node_sender_distribution_records_pkey" <addPrimaryKey columnNames="peer_party_id, timestamp, timestamp_discriminator" constraintName="node_sender_distribution_records_pkey"
tableName="node_sender_distribution_records"/> tableName="node_sender_distribution_records"/>
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="node_sender_distribution_records_idx">
<createIndex indexName="node_sender_distribution_records_idx" tableName="node_sender_distribution_records">
<column name="timestamp"/>
<column name="sequence_number"/>
<column name="receiver_party_id"/>
</createIndex>
</changeSet>
<changeSet author="R3.Corda" id="create_receiver_distribution_records_table"> <changeSet author="R3.Corda" id="create_receiver_distribution_records_table">
<createTable tableName="node_receiver_distribution_records"> <createTable tableName="node_receiver_distribution_records">
<column name="sequence_number" type="BIGINT"> <column name="timestamp" type="TIMESTAMP">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="timestamp" type="TIMESTAMP"> <column name="timestamp_discriminator" type="INT">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="transaction_id" type="NVARCHAR(144)"> <column name="transaction_id" type="NVARCHAR(144)">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="sender_party_id" type="BIGINT"> <column name="peer_party_id" type="BIGINT">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="distribution_list" type="BLOB"> <column name="distribution_list" type="BLOB">
<constraints nullable="false"/> <constraints nullable="false"/>
</column> </column>
<column name="receiver_states_to_record" type="INT">
<constraints nullable="false"/>
</column>
</createTable> </createTable>
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="node_receiver_distribution_records_pkey"> <changeSet author="R3.Corda" id="node_receiver_distribution_records_pkey">
<addPrimaryKey columnNames="timestamp, sequence_number" constraintName="node_receiver_distribution_records_pkey" <addPrimaryKey columnNames="peer_party_id, timestamp, timestamp_discriminator" constraintName="node_receiver_distribution_records_pkey"
tableName="node_receiver_distribution_records"/> tableName="node_receiver_distribution_records"/>
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="node_receiver_distribution_records_idx">
<createIndex indexName="node_receiver_distribution_records_idx" tableName="node_receiver_distribution_records">
<column name="timestamp"/>
<column name="sequence_number"/>
<column name="sender_party_id"/>
</createIndex>
</changeSet>
<changeSet author="R3.Corda" id="create_recovery_party_info_table"> <changeSet author="R3.Corda" id="create_recovery_party_info_table">
<createTable tableName="node_recovery_party_info"> <createTable tableName="node_recovery_party_info">
<column name="party_id" type="BIGINT"> <column name="party_id" type="BIGINT">
@ -94,15 +75,15 @@
<addPrimaryKey columnNames="party_id" constraintName="node_recovery_party_info_pkey" tableName="node_recovery_party_info"/> <addPrimaryKey columnNames="party_id" constraintName="node_recovery_party_info_pkey" tableName="node_recovery_party_info"/>
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="FK__sender_distribution_records__receiver_party_id"> <changeSet author="R3.Corda" id="FK__sender_distribution_records__peer_party_id">
<addForeignKeyConstraint baseColumnNames="receiver_party_id" baseTableName="node_sender_distribution_records" <addForeignKeyConstraint baseColumnNames="peer_party_id" baseTableName="node_sender_distribution_records"
constraintName="FK__sender_distribution_records__receiver_party_id" constraintName="FK__sender_distribution_records__peer_party_id"
referencedColumnNames="party_id" referencedTableName="node_recovery_party_info"/> referencedColumnNames="party_id" referencedTableName="node_recovery_party_info"/>
</changeSet> </changeSet>
<changeSet author="R3.Corda" id="FK__receiver_distribution_records__initiator_party_id"> <changeSet author="R3.Corda" id="FK__receiver_distribution_records__peer_party_id">
<addForeignKeyConstraint baseColumnNames="sender_party_id" baseTableName="node_receiver_distribution_records" <addForeignKeyConstraint baseColumnNames="peer_party_id" baseTableName="node_receiver_distribution_records"
constraintName="FK__receiver_distribution_records__initiator_party_id" constraintName="FK__receiver_distribution_records__peer_party_id"
referencedColumnNames="party_id" referencedTableName="node_recovery_party_info"/> referencedColumnNames="party_id" referencedTableName="node_recovery_party_info"/>
</changeSet> </changeSet>

39
node/src/test/kotlin/net/corda/node/services/persistence/DBTransactionStorageLedgerRecoveryTests.kt
View File

@ -109,6 +109,21 @@ class DBTransactionStorageLedgerRecoveryTests {
val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS)) val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS))
val results = transactionRecovery.querySenderDistributionRecords(timeWindow, excludingTxnIds = setOf(transaction1.id)) val results = transactionRecovery.querySenderDistributionRecords(timeWindow, excludingTxnIds = setOf(transaction1.id))
assertEquals(1, results.size) assertEquals(1, results.size)
assertEquals(transaction2.id.toString(), results[0].txId)
}
@Test(timeout = 300_000)
fun `query local ledger for transactions within timeWindow and for given peers`() {
val transaction1 = newTransaction()
transactionRecovery.addUnnotarisedTransaction(transaction1)
transactionRecovery.addSenderTransactionRecoveryMetadata(transaction1.id, TransactionMetadata(ALICE_NAME, DistributionList(ALL_VISIBLE, mapOf(BOB_NAME to ONLY_RELEVANT))))
val transaction2 = newTransaction()
transactionRecovery.addUnnotarisedTransaction(transaction2)
transactionRecovery.addSenderTransactionRecoveryMetadata(transaction2.id, TransactionMetadata(ALICE_NAME, DistributionList(ALL_VISIBLE, mapOf(CHARLIE_NAME to ONLY_RELEVANT))))
val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS))
val results = transactionRecovery.querySenderDistributionRecords(timeWindow, peers = setOf(CHARLIE_NAME))
assertEquals(1, results.size)
assertEquals(transaction2.id.toString(), results[0].txId)
} }
@Test(timeout = 300_000) @Test(timeout = 300_000)
@ -125,13 +140,13 @@ class DBTransactionStorageLedgerRecoveryTests {
val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS)) val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS))
transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.SENDER).let { transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.SENDER).let {
assertEquals(1, it.size) assertEquals(1, it.size)
assertEquals(BOB_NAME.hashCode().toLong(), (it[0] as SenderDistributionRecord).peerPartyId) assertEquals(BOB_NAME.hashCode().toLong(), it.senderRecords[0].compositeKey.peerPartyId)
assertEquals(ALL_VISIBLE, (it[0] as SenderDistributionRecord).statesToRecord) assertEquals(ALL_VISIBLE, it.senderRecords[0].statesToRecord)
} }
transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.RECEIVER).let { transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.RECEIVER).let {
assertEquals(1, it.size) assertEquals(1, it.size)
assertEquals(BOB_NAME.hashCode().toLong(), (it[0] as ReceiverDistributionRecord).initiatorPartyId) assertEquals(BOB_NAME.hashCode().toLong(), it.receiverRecords[0].compositeKey.peerPartyId)
assertEquals(ALL_VISIBLE, (it[0] as ReceiverDistributionRecord).statesToRecord) assertEquals(ALL_VISIBLE, (transactionRecovery.decrypt(it.receiverRecords[0].distributionList).peerHashToStatesToRecord.map { it.value }[0]))
} }
val resultsAll = transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.ALL) val resultsAll = transactionRecovery.queryDistributionRecords(timeWindow, recordType = DistributionRecordType.ALL)
assertEquals(2, resultsAll.size) assertEquals(2, resultsAll.size)
@ -192,9 +207,9 @@ class DBTransactionStorageLedgerRecoveryTests {
val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS)) val timeWindow = RecoveryTimeWindow(fromTime = now().minus(1, ChronoUnit.DAYS))
transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(ALICE_NAME)).let { transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(ALICE_NAME)).let {
assertEquals(3, it.size) assertEquals(3, it.size)
assertEquals(it[0].statesToRecord, ALL_VISIBLE) assertEquals(transactionRecovery.decrypt(it[0].distributionList).peerHashToStatesToRecord.map { it.value }[0], ALL_VISIBLE)
assertEquals(it[1].statesToRecord, ONLY_RELEVANT) assertEquals(transactionRecovery.decrypt(it[1].distributionList).peerHashToStatesToRecord.map { it.value }[0], ONLY_RELEVANT)
assertEquals(it[2].statesToRecord, NONE) assertEquals(transactionRecovery.decrypt(it[2].distributionList).peerHashToStatesToRecord.map { it.value }[0], NONE)
} }
assertEquals(1, transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(BOB_NAME)).size) assertEquals(1, transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(BOB_NAME)).size)
assertEquals(1, transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(CHARLIE_NAME)).size) assertEquals(1, transactionRecovery.queryReceiverDistributionRecords(timeWindow, initiators = setOf(CHARLIE_NAME)).size)
@ -229,8 +244,8 @@ class DBTransactionStorageLedgerRecoveryTests {
DistributionList(ONLY_RELEVANT, mapOf(BOB_NAME to ALL_VISIBLE)).toWire()) DistributionList(ONLY_RELEVANT, mapOf(BOB_NAME to ALL_VISIBLE)).toWire())
assertEquals(IN_FLIGHT, readTransactionFromDB(receiverTransaction.id).status) assertEquals(IN_FLIGHT, readTransactionFromDB(receiverTransaction.id).status)
readReceiverDistributionRecordFromDB(receiverTransaction.id).let { readReceiverDistributionRecordFromDB(receiverTransaction.id).let {
assertEquals(ALL_VISIBLE, it.statesToRecord) assertEquals(ONLY_RELEVANT, it.statesToRecord)
assertEquals(ONLY_RELEVANT, it.senderStatesToRecord) assertEquals(ALL_VISIBLE, it.peersToStatesToRecord.map { it.value }[0])
assertEquals(ALICE_NAME, partyInfoCache.getCordaX500NameByPartyId(it.initiatorPartyId)) assertEquals(ALICE_NAME, partyInfoCache.getCordaX500NameByPartyId(it.initiatorPartyId))
assertEquals(setOf(BOB_NAME), it.peersToStatesToRecord.map { (peer, _) -> partyInfoCache.getCordaX500NameByPartyId(peer) }.toSet() ) assertEquals(setOf(BOB_NAME), it.peersToStatesToRecord.map { (peer, _) -> partyInfoCache.getCordaX500NameByPartyId(peer) }.toSet() )
} }
@ -245,7 +260,7 @@ class DBTransactionStorageLedgerRecoveryTests {
assertEquals(VERIFIED, readTransactionFromDB(transaction.id).status) assertEquals(VERIFIED, readTransactionFromDB(transaction.id).status)
readSenderDistributionRecordFromDB(transaction.id).apply { readSenderDistributionRecordFromDB(transaction.id).apply {
assertEquals(1, this.size) assertEquals(1, this.size)
assertEquals(ONLY_RELEVANT, this[0].statesToRecord) assertEquals(ALL_VISIBLE, this[0].statesToRecord)
} }
} }
@ -377,3 +392,7 @@ class DBTransactionStorageLedgerRecoveryTests {
return cryptoService.encrypt(hashedDistributionList.serialize()) return cryptoService.encrypt(hashedDistributionList.serialize())
} }
} }
internal fun DBTransactionStorageLedgerRecovery.decrypt(distributionList: ByteArray): HashedDistributionList {
return HashedDistributionList.deserialize(this.cryptoService.decrypt(distributionList))
}