Corda 4.11 enterpriseにて追加されたLedger RecoveryのFinality Recovery Flowの詳細について把握できます。
はじめに
Corda 4.11 enterpriseでは、それまでのversionから大きくアップデートされ、複数の新機能が追加されました。
前回に引き続き、今回は第二弾として、新規Ledger Recovery手法の一部であるFinality Recovery Flowをご紹介したいと思います。
Finality Recovery Flowの導入により、障害によって中断されたファイナリティ処理をリカバリーすることが可能になり、ファイナリティ処理のリトライが柔軟に実行できるようになりました。このflowは、Corda enterprise Nodeを起動した時点で既に利用可能です。
Corda 4.11 enterpriseで新たに導入されたLedger Recoveryは、Finality Recovery FlowとLedger Recovery Flowの２つで構成されています。
詳しくは次回の記事でご紹介しますが、Ledger Recovery FlowがCorda 4.10 enterprise以前で使用されていたCollaborative Recoveryの上位互換に相当する部分であるため、Corda 4.11 enterpriseからCollaborative Recoveryは非推奨、Corda 4.12 enterpriseでは廃止となっています。
前提
前提として大事な箇所を、前回の記事から持ってきました。
Two Phase Finalityでは、IN_FLIGHTとい新しいトランザクションのstatusが導入され、ファイナリティ処理が開始した段階で、statusはUNVERIFIEDからIN_FLIGHTへ遷移します。また、ファイナリティ処理の終了後には、statusはVERIFIEDとなります。
Finality Recovery Flowが、リカバリーと対象とするのは、障害（ノードダウン・通信エラー等）により滞留するIN_FLIGHTのトランザクションとなります。（参考URL）
Finality Recovery Flowとは？
Finality Recovery Flowとは、Corda Nodeのクラッシュや通信エラーによって、失敗したファイナリティ処理を回復させ、トランザクションのNotarization, Sharing, Recordingを確定させるFlowです。Finality Recovery Flowのリカバリー対象は、IN_FLIGHT状態のトランザクションになります。
リカバリの手法は、前回の記事で紹介したTwo Phase Finalityのプロセスで共有されるDistribution record (トランザクションのメタデータ)が使用されます。
Cordaに詳しい方々は、Flow Hosiptalとの差異について気になっておられるかもしれません。（R3のドキュメントでは、Finality Recovery FlowとFlow Hospitalの棲み分けについて言及はありませんが）筆者はFlow Hospitalで回復できないflow（flow status: FAILED）を、Finality Recoveryで回復させるという理解をしております。これまでは、flow status: FAILEDになった場合は、障害のパターンに関わらず、再度flowを実行する必要がありました。一方、Finality Recovery Flowの導入により、Nodeのクラッシュや通信エラーなど回復可能な要素を含む場合、flowを一から実行しなおす必要がなくなりました。
整理するとFinality Recovery Flowの対象は、transaction status: IN_FLIGHTかつflow status: FAILEDのflowということになります。
つまり、「トランザクションがまだファイナリティ処理を終えていない」かつ「ファイナリティ処理中に失敗したflow」をリカバリーします。
以上がFinality Recovery Flowの概要と意義についてでした。
Finality Recovery Flowで回復できないflowについて
Cordappが原因等のflow実行エラーは、ファイナリティ処理前かつ根本のコードの改善が求められるため、Finality Recovery Flowの対象外になります。
Finality Recovery Flowの始点
Two Phase Finalityの導入によって、Distribution Record (トランザクションのメタデータ)が共有され、flowの送信者・受信者のどちらからでもFinality Recovery Flowの実行が可能になります。
※以下の説明は、片方のNodeで障害が起きた場合と両方のNodeで障害が起きた場合を含みます。どちらの場合でも、回復した方からリカバリーの始点になることができます。
送信者が始点になり得るのは・・
- ファイナリティ処理にてNotarization前に、障害が発生した場合
- ファイナリティ処理にてNotarization後に、障害が発生した場合
※送信者が先にリカバリ処理を行えば、受信側では必要ありません
※flowの送信者側にてファイナリティ処理を終えた場合のみ、トランザクションIDを指定して、リカバリーを行うことで、障害が発生した受信者のNodeを回復させることが可能
受信側が始点になり得るのは・・
- ファイナリティ処理にてNotarization後に、障害が発生した場合
※ ただし、送信側でのNotarization後に障害が起こった場合は、送信者側でもFinality Recovery Flowを実行する必要がある
以上が、送信側・受信側が始点となりFinality Recovery Flowを実行する場合になります。
Finality Recovery Flowの使い方
Corda 4.11 enterpriseのNodeを起動し、Node shellにてFinalityRecoveryFlowが存在することを確認します。
>>> flow list
com.template.flows.TemplateFlow$TemplateFlowInitiator
net.corda.core.flows.ContractUpgradeFlow$Authorise
net.corda.core.flows.ContractUpgradeFlow$Deauthorise
net.corda.core.flows.ContractUpgradeFlow$Initiate
net.corda.core.flows.FinalityRecoveryFlow
net.corda.core.flows.LedgerRecoveryFlowFinalityReocveryFlowの実行結果は次の３つのいずれかです。
Status | Description |
true | 特定のトランザクションのファイナリティの再実行が成功 |
false | 特定のトランザクションのファイナリティの再実行が必要でない場合 |
FlowRecoveryException | 何らかの原因でファイナリティ処理のリカバリーが失敗した |
Finality Recovery Flowの実行方法は３つほど存在しますので、それぞれ解説します。
- FlowRPCOps APIの拡張機能として利用する手法
- Corda node shellから実行する手法
- 3. RPC clientを立てる場合の手法
手法1. FlowRPCOps APIの拡張機能として実行
FlowRPCOps APIは、例としてnetworkMapSnapshot, nodeInfo, vaultQuery等の関数がAPIとして使用できます。
この拡張機能として、FinalityRecoveryFlowが実行でき、引数次第で６つの実行方法が存在します。関数の名前が微妙に異なるので、注意してください。
// 1. ファイナリティリカバリーしたいFlowのID (StateMachineRunId)を指定して、実行する場合
fun recoverFinalityFlow(id: StateMachineRunId, forceRecover: Boolean = false): Boolean
// 2. ファイナリティリカバリーしたいトランザクションのIDを複数まとめて、実行する場合
// 結果は、mapとしてリターンされます
fun recoverFinalityFlows(ids: Set<StateMachineRunId>, forceRecover: Boolean = false): Map<StateMachineRunId, Boolean>
// 3. flowStatus: FAILED & Transaction status: IN_FLIGHTのトランザクションを全てファイナリティリカバリーする場合
// 結果は、mapとしてリターンされます
fun recoverAllFinalityFlows(forceRecover: Boolean = false): Map<StateMachineRunId, Boolean>
// 4. トランザクションのIDを用いて、ファイナリティリカバリーする場合
fun recoverFinalityFlowByTxnId(txnId: SecureHash, forceRecover: Boolean = false): Boolean
// 5. 複数のトランザクションIDを用いて、ファイナリティリカバリーする場合
fun recoverFinalityFlowByTxnIds(txnIds: Set<SecureHash>, forceRecover: Boolean = false): Map<SecureHash, Boolean>
// 6. 特定のクエリに合致する、トランザクションのファイナリティリカバリーを行う場合
fun recoverFinalityFlowsMatching(query: FlowRecoveryQuery, forceRecover: Boolean = false): Map<StateMachineRunId, Boolean>参考
①FinalityRecoveryFlow実行時の引数：forceRecoverについて
通常flow status: FAILEDをリカバリーの対象としますが、forceRecover: trueによって、RUNNABLE, PAUSED, HOSPITALIZEDステータスのトランザクションを、リカバリーの対象にすることができます。ただし、PAUSED や HOSPITALIZED のflowは、ノードの再起動時に自動的にリトライされるため、forceRecover を使用しなくてもファイナリティ処理は再度行われる可能性があります。
②FinalityRecoveryFlow実行時の引数：FlowRecoveryQueryについて
クエリの詳細は以下の通りタイムフレームやFinalityFlowを実行したCordaX500nameとその相手を指定することが可能になります。
data class FlowRecoveryQuery(
val timeframe: FlowTimeWindow? = null,
val initiatedBy: CordaX500Name? = null,
val counterParties: List<CordaX500Name>? = null)③FinalityRecoveryFlowの対象になるトランザクションを把握する方法
NodeFlowStatusRpcOps RPC APIが新たに拡張され、ファイナリティリカバリーの対象になるトランザクションの発見が容易になりました。
//flow IDを元に、flowのstatusを確認するAPI
@RpcPermissionGroup(READ_ONLY)
fun getFlowTransaction(flowId: String): FlowTransactionInfo?
//tx idを元に、flowのstatusを確認するAPI
@RpcPermissionGroup(READ_ONLY)
fun getFlowTransactionByTxnId(txnId: String): FlowTransactionInfo?
//上２つがreturnするオブジェクト
data class FlowTransactionInfo(
val stateMachineRunId: StateMachineRunId,
val txId: String,
val status: TransactionStatus,
val timestamp: Instant,
val initiator: CordaX500Name? = null,
val peers: Set<CordaX500Name>? = null
)手法2. Corda Node shellから実行
続いて、Corda Node shellからFinality Recovery Flowを実行する手法の紹介です。
ファイナリティリカバリーを行うflow idやtransactionのidは、次のコマンドにて確認できます。
// StateMachineRunId (flow id)を用いて、IN_FLIGHTのtxを発見
flowStatus queryFinalityById e0d781be-b4ab-43e0-b694-e97cc4eaa6ee
FlowTransactionInfo(stateMachineRunId=[e0d781be-b4ab-43e0-b694-e97cc4eaa6ee], txId=19BE64484D3CBF532A8FB2ACA1AEACA38B1FBA3C38B0518B7F5316AC9E79432F, status=IN_FLIGHT, timestamp=2023-03-29T11:16:29.477Z, initiator=O=Alice Corp, L=Madrid, C=ES, peers=[O=Bob Plc, L=Rome, C=IT])
---
- stateMachineRunId:
uuid: "e0d781be-b4ab-43e0-b694-e97cc4eaa6ee"
txId: "19BE64484D3CBF532A8FB2ACA1AEACA38B1FBA3C38B0518B7F5316AC9E79432F"
status: "IN_FLIGHT"
initiator:
x500Principal:
name: "O=Alice Corp,L=Madrid,C=ES"
peers:
x500Principal:
name: "O=Bob Plc,L=Rome,C=IT"続いて、Finality Recovery Flowの実行方法を５つほど紹介します。明記したid群は、あくまで一例です。
// 1. ファイナリティリカバリーしたいFlowのID (StateMachineRunId)を指定
flow recoverFinality 821884be-8e9f-486d-8228-70d97e215218
// 成功時出力
Recovered finality flow [821884be-8e9f-486d-8228-70d97e215218]
// 失敗時出力
Failed to recover finality flow 821884be-8e9f-486d-8228-70d97e215218
// 2. ファイナリティリカバリーしたいtxのidを指定
flow recoverFinalityByTxnId 7E7EE31CA6371D73CDDB1A7992E239CE222606EA845F1ABC87995898017904A4
// 成功時出力
Recovered finality flow [821884be-8e9f-486d-8228-70d97e215218]
// 失敗時出力
Failed to recover finality flow 7E7EE31CA6371D73CDDB1A7992E239CE222606EA845F1ABC87995898017904A4
// 3. 対象となる全てのトランザクションを、ファイナリティリカバリーする場合
flow recoverAllFinality
// 成功時出力
Recovered finality flow(s)
Results: [[4cbfa031-90de-4564-a375-30141a18bbba]=true]
// 4. force-recoverオプションを付け、flowStatus: PAUSED, HOSPITALIZEDを含むflowを対象にする場合
flow recoverAllFinality --force-recover
// 成功時出力
Recovered finality flow(s)
Results: [[358a7b4e-074a-4da8-b6d7-64f1d923f9a8]=true, [c3cf2d33-6a36-4266-a9cb-f488ac3194cc]=true]
// 5. クエリを使用する場合
flow recoverFinalityMatching \
flowStartFromTime: "2023-12-04T10:15:30.00", \
flowStartUntilTime: "2023-12-05T10:15:30.00Z", \
initiatedBy: "O=PartyA,L=London,C=GB", \
counterParties: ["O=PartyA,L=London,C=GB", "O=PartyB,L=London,C=GB"]
手法３. RPC Client経由での実行
最後に、RPC clientを立て、flow呼び出しコールを活用することで、外部からflowを呼び出すことが可能です。こちらの機能を用いて、FinalityRecoveryFlowを実行することが可能です。
※ 詳細は、flow呼び出しコール一覧を参照ください。
RPC Client経由でflowを外部呼出しする際は、コードの記述が必要となります。参考例として、RPC Clientを立ち上げflow実行リクエストをRESTとして受信しRPCへ変換する処理を記述した、弊社が提供するCorda 4 trainingのサンプルコードをご参考ください。
以下にRPC client経由で、Finality Recovery Flowを呼び出すコードを解説します。
- まず複数のCorda nodeに対して、同一のインターフェースで、RPC接続を確立・管理するために、MultiRPCClientをインスタンス化します
val username = "testuser"
val password = "password"
val rpcHostAndPort = NetworkHostAndPort("localhost", 10006)
val flowClient = MultiRPCClient(rpcHostAndPort, FlowRPCOps::class.java, username, password).start().getOrThrow()- 続いて、Finality Recovery Flowの実行部分になります。こちらも手法１，２と同様、複数の呼び出し方法が存在します。
// 1. 単一のStateMachineRunId (flow id)で、ファイナリティリカバリーを行う場合
val status = flowClient.proxy.recoverFinalityFlow(flowHandle.id)
// 2. 複数のflow idでファイナリティリカバリーを行う場合
val resultMap = flowClient.proxy.recoverFinalityFlows(setOf(flowHandle1.id, flowHandle2.id))
// 3. 単一のflow idを用いるかつ、flow status: HOSPITALIZED or PAUSEDを含める場合
val status = flowClient.proxy.recoverFinalityFlow(flowHandle.id, forceRecover = true)
// 4. 単一のトランザクションidで、ファイナリティリカバリーを行う場合
val status = flowClient.proxy.recoverFinalityFlowByTxnId(stx.id)
// 5. 複数のトランザクションidで、ファイナリティリカバリーを行う場合
val resultMap = flowClient.proxy.recoverFinalityFlowByTxnIds(setOf(stx1.id, stx2,id))
// 6. 対象となるトランザクションを全てファイナリティリカバリーする場合
val resultMap = flowClient.proxy.recoverAllFinalityFlows()
// 7-1. クエリを使用する場合 (タイムフレームの指定)
val resultMap = flowRPC.proxy.recoverFinalityFlowsMatching(
FlowRecoveryQuery(timeframe = FlowTimeWindow(
fromTime = startTime,
untilTime = endTime
)
)
)
// 7-2. クエリを使用する場合 (flowの送信者の指定)
val resultMap = flowRPC.proxy.recoverFinalityFlowsMatching(
FlowRecoveryQuery(initiatedBy = CHARLIE_NAME))
// 7-3. クエリを使用する場合 (flowの受信者の指定)
val resultMap = flowRPC.proxy.recoverFinalityFlowsMatching(
FlowRecoveryQuery(counterParties = listOf(CHARLIE_NAME)))
手法３を使用する場合は、サーバー側のリソースリーク（不要なリソースが解放されずに残り続けること）を防ぐために、処理が終了した段階で flowClient.close() を呼び出し、RPC clientを適切にクローズしてください。
備考
Ledger RecoveryのFinality Recovery Flowは、Corda enterpriseの機能がゆえ、コードは一般公開されていません。しかしながら、ドキュメントには限定的ですが、いくつか公開されている情報が存在します。
- Corda nodeによって、FinalityRecoveryFlowが呼び出されると、Flowの送信者か受信者によって呼び出されるflowが異なります。
FinalityPeerRecoveryFlowの補助機能として、TransactionNotarizationCheckFlowが導入
Initiator: `net.corda.node.internal.recovery.FinalityInitiatorRecoveryFlow`
Receiver: `net.corda.node.internal.recovery.FinalityPeerRecoveryFlow`Flowの受信者側は、送信側から送られてきたSignedTransactionが、Notarizationのプロセスを経たかどうかを確認することはできません。そこで、TransactionNotarizationCheckFlowを使用することで、目的のstateをReference stateとしてNotary側に問い合わせることで、トランザクション内のstateの消費状態を確認することができます。
まとめ
本記事では、Finality Recovery Flowの導入によって、双方向のリカバリーや手法の拡充をはじめ、リカバリーの柔軟性が大幅に向上しました。リカバリーの３種類の手法に関しては、テスト目的であればNode shellからの実行を、運用を意識するのであれば、RPC Client経由での実行をお勧めします。
＜ご質問・ご要望の例＞
- Corda Portalの記事について質問したい
- ブロックチェーンを活用した新規事業を相談したい
- 企業でのブロックチェーン活用方法を教えて欲しい 等々
SBI R3 Japanインターン エンジニアリング部所属
Cordaの技術調査・検証や記事執筆など多岐に渡ります
趣味：英語学習（まだまだですが…）・テニス・映画鑑賞
イギリス大学院進学 2025年9月～🇬🇧