STR-110: add balance reconciliation with event-derived internal ledger and drift detection#115
Open
Puneethkumarck wants to merge 2 commits into
Open
STR-110: add balance reconciliation with event-derived internal ledger and drift detection#115Puneethkumarck wants to merge 2 commits into
Puneethkumarck wants to merge 2 commits into
Conversation
…ft detection (#110) Implements two-way balance reconciliation between internal ledger and Fireblocks-reported balances. Debits internal balance on confirmed outgoing transactions, runs scheduled reconciliation via ShedLock, and publishes break events when drift exceeds configurable tolerance.
Puneethkumarck
added
phase:infrastructure
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (6)
📝 WalkthroughWalkthroughThis PR implements continuous balance reconciliation between an event-derived internal ledger and Fireblocks-reported balances. It adds domain models for reconciliation status and results, integrates internal balance debiting into transaction confirmation, persists reconciliation outcomes, runs a scheduled job to detect drift, and publishes events when mismatches occur. ChangesBalance Reconciliation Feature
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationBreakDetectedEvent.kt`:
- Around line 7-21: The ReconciliationBreakDetectedEvent payload needs runtime
validation: add an init block (or factory) in the
ReconciliationBreakDetectedEvent data class that throws IllegalArgumentException
when invariants fail — ensure currency and protocol are non-blank, toleranceUsed
is non-negative, absoluteDrift equals drift.abs(), and drift equals
(fireblocksBalance - internalBalance); include occurredAt non-null/valid check
if applicable; keep the TOPIC companion object unchanged and fail fast on
invalid inputs to prevent publishing bad events.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationCompletedEvent.kt`:
- Around line 7-20: ReconciliationCompletedEvent currently allows invalid
currency, protocol and occurredAt values; add an init block inside the
ReconciliationCompletedEvent data class that validates these inputs and throws
IllegalArgumentException on misuse: ensure currency and protocol are non-blank
(e.g., !currency.isBlank() and !protocol.isBlank(), optionally validate against
a permitted pattern), and ensure occurredAt is provided and sensible (e.g., not
in the future or otherwise invalid, e.g., !occurredAt.isAfter(Instant.now())).
Place the checks in the class init and reference the class name
ReconciliationCompletedEvent and the fields currency, protocol, and occurredAt
when implementing the validations.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/BalanceReconciliationService.kt`:
- Around line 10-17: Add an input validation guard at the start of reconcile to
reject negative tolerance values: in BalanceReconciliationService.reconcile(...)
check if tolerance < BigDecimal.ZERO and throw an IllegalArgumentException (or
IllegalArgumentException with a clear message) before computing drift or
returning a ReconciliationResult so negative tolerance cannot be used or
persisted; apply the same non-negative check to any other methods in this class
that accept a tolerance parameter (e.g., the overloaded reconciliation method
referenced around lines 34–40) to ensure toleranceUsed is never set from a
negative value.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/ReconciliationResult.kt`:
- Around line 7-19: The ReconciliationResult data class must enforce invariants:
add validation in ReconciliationResult (either via an init block or by making
absoluteDrift derived) to ensure absoluteDrift == drift.abs() and that
toleranceUsed is non-negative (toleranceUsed >= 0); if these checks fail throw
an IllegalArgumentException. Locate the ReconciliationResult declaration and
implement the init block (or remove absoluteDrift from the ctor and compute val
absoluteDrift = drift.abs()) and validate toleranceUsed to prevent inconsistent
reconciliation records.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandler.kt`:
- Around line 70-72: The current TransactionStatusHandler unconditionally calls
debitInternalBalance(result) when newStatus == TransactionStatus.CONFIRMED;
change this to only debit for outgoing transactions by adding a direction guard
(e.g., check result.transaction.direction == TransactionDirection.OUTGOING or
equivalent) so that debitInternalBalance(result) is invoked only when newStatus
== TransactionStatus.CONFIRMED && direction is OUTGOING; update any imports or
enum references to TransactionDirection.OUTGOING as needed.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/vault/VaultRepository.kt`:
- Line 17: The current VaultRepository.findAllActive() returns a full
List<Vault>, which can cause memory/DB pressure on recurring reconciliation;
change the repository contract to provide a paged or streaming variant (e.g.,
findActive(pageable: Pageable): Page<Vault> or streamActive(): Stream<Vault>)
and implement corresponding DAO/query methods and callers (reconciliation job)
to consume pages/stream rather than loading all Vaults at once; update
VaultRepository interface signature(s) and adjust usages to iterate pages/close
streams appropriately.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJob.kt`:
- Around line 127-160: When seeding fails the catch block only logs and drops
evidence; create and persist a failed ReconciliationResult and call
auditReconciliation before returning so every vault/asset run is auditable. In
the catch for BalanceReconciliationJob (the try that calls
fireblocksBalancePort.getBalance, internalBalanceRepository.save,
reconciliationService.reconcile, reconciliationResultRepository.save,
auditReconciliation) build a ReconciliationResult (or equivalent DTO used
elsewhere) populated with vaultId, asset identifiers
(asset.currency/asset.protocol or asset.fireblocksAssetId), a FAILED status, the
exception message/stack info, timestamp/Instant.now(), and any other required
reconciliation metadata, then save it via
reconciliationResultRepository.save(failedResult) and call
auditReconciliation(failedResult) prior to logging the error.
- Around line 92-111: The catch currently spans both the external fetch and
domain reconcile so any exception (including reconcile failures) becomes a
PARTIAL; change BalanceReconciliationJob to only catch errors from the
Fireblocks call: call fireblocksBalancePort.getBalance(...) inside its own
try/catch and on that catch log the Fireblocks-unavailable warning and call
reconciliationService.createPartialResult(...), but let exceptions from
reconciliationService.reconcile(...) propagate (or be handled separately).
Reference fireblocksBalancePort.getBalance, reconciliationService.reconcile, and
reconciliationService.createPartialResult when making this change.
In
`@custody-fireblocks/src/main/resources/db/migration/V13__create_reconciliation_results.sql`:
- Around line 15-16: The existing index idx_reconciliation_results_vault_created
on reconciliation_results(vault_id, created_at) doesn't support queries that
filter by vault_id + currency + protocol and order by created_at DESC; add a new
covering index that matches that query shape (e.g., include vault_id, currency,
protocol in that key order and created_at DESC) and include any additional
columns returned by the query to make it covering so the planner can use an
index-only scan; update the migration (instead of or alongside
idx_reconciliation_results_vault_created) to create this new index on
reconciliation_results.
In
`@custody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandlerTest.kt`:
- Around line 313-414: Add a regression test in TransactionStatusHandlerTest
that ensures CONFIRMED does not touch internal balances for non-outgoing
transactions: create a transaction via aTransaction with
status=TransactionStatus.CONFIRMING, fireblocksTransactionId like
"fb-tx-non-outgoing", and set its direction to a non-outgoing value (e.g.,
TransactionDirection.INCOMING or equivalent field on the test builder), mock
transactionRepository.findByFireblocksTransactionId to return it, mock
stateMachine.transition(transaction, TransactionStatus.CONFIRMED) to return
CONFIRMED and
transactionRepository.findByIdForUpdate/save/eventPublisher/auditLogRepository
as in other tests, call handler.handleStatusUpdate("fb-tx-non-outgoing",
"COMPLETED", null, "0xhash"), and verify that
internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(...) and
internalBalanceRepository.save(...) are NOT invoked (verify(exactly = 0){ ... })
to assert no internal-balance mutation for non-outgoing CONFIRMED transactions.
In
`@custody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJobTest.kt`:
- Around line 113-140: Add a new unit test in BalanceReconciliationJobTest that
covers the branch where there is no existing internal balance and the Fireblocks
API fails: mock vaultRepository.findAllActive() to return a vault,
walletAssetRepository.findByVaultId(...) to return the asset, mock
internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(...) to return
null (or Optional.empty equivalent), and mock
fireblocksBalancePort.getBalance(...) to throw FireblocksApiException; keep
reconciliationResultRepository.save(...) returningArgument 0 and
auditLogRepository.save(...) as before, call job.reconcileBalances(), and verify
reconciliationResultRepository.save(...) was called with a
ReconciliationStatus.PARTIAL result. Ensure you reference the same methods used
in the other test (reconcileBalances(), fireblocksBalancePort.getBalance,
internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol,
reconciliationResultRepository.save) so the new test exercises the "first-run
seeding failure" branch.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: a466ffe8-3374-4a5d-b277-e3fb586936d9
📒 Files selected for processing (35)
custody-fireblocks/src/integrationTest/kotlin/com/stablecoin/custody/fireblocks/FlywayMigrationIntegrationTest.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/audit/AuditOperation.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationBreakDetectedEvent.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationCompletedEvent.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/BalanceReconciliationService.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/InternalBalance.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/InternalBalanceRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/ReconciliationResult.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/ReconciliationResultRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/ReconciliationStatus.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/shared/ErrorCode.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandler.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/vault/VaultRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/wallet/WalletAssetRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/messaging/outbox/ReconciliationBreakEventOutboxPublisher.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/InternalBalanceEntity.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/InternalBalanceJpaRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/InternalBalanceRepositoryAdapter.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/ReconciliationResultEntity.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/ReconciliationResultJpaRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/ReconciliationResultRepositoryAdapter.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/VaultJpaRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/VaultRepositoryAdapter.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/WalletAssetJpaRepository.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/persistence/WalletAssetRepositoryAdapter.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJob.ktcustody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/ReconciliationProperties.ktcustody-fireblocks/src/main/resources/application.ymlcustody-fireblocks/src/main/resources/db/migration/V12__create_internal_balances.sqlcustody-fireblocks/src/main/resources/db/migration/V13__create_reconciliation_results.sqlcustody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/BalanceReconciliationServiceTest.ktcustody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandlerTest.ktcustody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJobTest.ktcustody-fireblocks/src/testFixtures/kotlin/com/stablecoin/custody/fireblocks/test/fixtures/InternalBalanceFixtures.ktcustody-fireblocks/src/testFixtures/kotlin/com/stablecoin/custody/fireblocks/test/fixtures/ReconciliationResultFixtures.kt
Comment on lines
+7
to
+21
| data class ReconciliationBreakDetectedEvent( | ||
| val vaultId: UUID, | ||
| val currency: String, | ||
| val protocol: String, | ||
| val internalBalance: BigDecimal, | ||
| val fireblocksBalance: BigDecimal, | ||
| val drift: BigDecimal, | ||
| val absoluteDrift: BigDecimal, | ||
| val toleranceUsed: BigDecimal, | ||
| val occurredAt: Instant, | ||
| ) { | ||
| companion object { | ||
| const val TOPIC = "custody.reconciliation.break-detected" | ||
| } | ||
| } |
There was a problem hiding this comment.
Guard event payload invariants before publishing.
This event can be constructed with invalid business data (blank asset identifiers, negative tolerance, or inconsistent drift fields), causing downstream false alarms.
Proposed fix
data class ReconciliationBreakDetectedEvent(
val vaultId: UUID,
val currency: String,
val protocol: String,
val internalBalance: BigDecimal,
val fireblocksBalance: BigDecimal,
val drift: BigDecimal,
val absoluteDrift: BigDecimal,
val toleranceUsed: BigDecimal,
val occurredAt: Instant,
) {
+ init {
+ require(currency.isNotBlank()) { "currency must not be blank" }
+ require(protocol.isNotBlank()) { "protocol must not be blank" }
+ require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" }
+ require(absoluteDrift.compareTo(drift.abs()) == 0) {
+ "absoluteDrift must equal abs(drift)"
+ }
+ }
+
companion object {
const val TOPIC = "custody.reconciliation.break-detected"
}
}As per coding guidelines "Validate inputs and check types; throw on misuse rather than silently producing wrong results".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| data class ReconciliationBreakDetectedEvent( | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val absoluteDrift: BigDecimal, | |
| val toleranceUsed: BigDecimal, | |
| val occurredAt: Instant, | |
| ) { | |
| companion object { | |
| const val TOPIC = "custody.reconciliation.break-detected" | |
| } | |
| } | |
| data class ReconciliationBreakDetectedEvent( | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val absoluteDrift: BigDecimal, | |
| val toleranceUsed: BigDecimal, | |
| val occurredAt: Instant, | |
| ) { | |
| init { | |
| require(currency.isNotBlank()) { "currency must not be blank" } | |
| require(protocol.isNotBlank()) { "protocol must not be blank" } | |
| require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" } | |
| require(absoluteDrift.compareTo(drift.abs()) == 0) { | |
| "absoluteDrift must equal abs(drift)" | |
| } | |
| } | |
| companion object { | |
| const val TOPIC = "custody.reconciliation.break-detected" | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationBreakDetectedEvent.kt`
around lines 7 - 21, The ReconciliationBreakDetectedEvent payload needs runtime
validation: add an init block (or factory) in the
ReconciliationBreakDetectedEvent data class that throws IllegalArgumentException
when invariants fail — ensure currency and protocol are non-blank, toleranceUsed
is non-negative, absoluteDrift equals drift.abs(), and drift equals
(fireblocksBalance - internalBalance); include occurredAt non-null/valid check
if applicable; keep the TOPIC companion object unchanged and fail fast on
invalid inputs to prevent publishing bad events.
Comment on lines
+7
to
+20
| data class ReconciliationCompletedEvent( | ||
| val vaultId: UUID, | ||
| val currency: String, | ||
| val protocol: String, | ||
| val internalBalance: BigDecimal, | ||
| val fireblocksBalance: BigDecimal, | ||
| val drift: BigDecimal, | ||
| val toleranceUsed: BigDecimal, | ||
| val occurredAt: Instant, | ||
| ) { | ||
| companion object { | ||
| const val TOPIC = "custody.reconciliation.completed" | ||
| } | ||
| } |
There was a problem hiding this comment.
Validate completed-event inputs at construction time.
Line 9, Line 10, and Line 14 accept invalid values today, which lets malformed reconciliation events propagate.
Proposed fix
data class ReconciliationCompletedEvent(
val vaultId: UUID,
val currency: String,
val protocol: String,
val internalBalance: BigDecimal,
val fireblocksBalance: BigDecimal,
val drift: BigDecimal,
val toleranceUsed: BigDecimal,
val occurredAt: Instant,
) {
+ init {
+ require(currency.isNotBlank()) { "currency must not be blank" }
+ require(protocol.isNotBlank()) { "protocol must not be blank" }
+ require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" }
+ }
+
companion object {
const val TOPIC = "custody.reconciliation.completed"
}
}As per coding guidelines "Validate inputs and check types; throw on misuse rather than silently producing wrong results".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| data class ReconciliationCompletedEvent( | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val toleranceUsed: BigDecimal, | |
| val occurredAt: Instant, | |
| ) { | |
| companion object { | |
| const val TOPIC = "custody.reconciliation.completed" | |
| } | |
| } | |
| data class ReconciliationCompletedEvent( | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val toleranceUsed: BigDecimal, | |
| val occurredAt: Instant, | |
| ) { | |
| init { | |
| require(currency.isNotBlank()) { "currency must not be blank" } | |
| require(protocol.isNotBlank()) { "protocol must not be blank" } | |
| require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" } | |
| } | |
| companion object { | |
| const val TOPIC = "custody.reconciliation.completed" | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/event/ReconciliationCompletedEvent.kt`
around lines 7 - 20, ReconciliationCompletedEvent currently allows invalid
currency, protocol and occurredAt values; add an init block inside the
ReconciliationCompletedEvent data class that validates these inputs and throws
IllegalArgumentException on misuse: ensure currency and protocol are non-blank
(e.g., !currency.isBlank() and !protocol.isBlank(), optionally validate against
a permitted pattern), and ensure occurredAt is provided and sensible (e.g., not
in the future or otherwise invalid, e.g., !occurredAt.isAfter(Instant.now())).
Place the checks in the class init and reference the class name
ReconciliationCompletedEvent and the fields currency, protocol, and occurredAt
when implementing the validations.
Comment on lines
+7
to
+19
| data class ReconciliationResult( | ||
| val id: UUID, | ||
| val vaultId: UUID, | ||
| val currency: String, | ||
| val protocol: String, | ||
| val internalBalance: BigDecimal, | ||
| val fireblocksBalance: BigDecimal, | ||
| val drift: BigDecimal, | ||
| val absoluteDrift: BigDecimal, | ||
| val status: ReconciliationStatus, | ||
| val toleranceUsed: BigDecimal, | ||
| val createdAt: Instant, | ||
| ) |
There was a problem hiding this comment.
Enforce reconciliation invariants in the domain model.
Line 15 and Line 17 can currently accept inconsistent values (e.g., absoluteDrift not matching abs(drift), negative tolerance), which can corrupt reconciliation records.
Proposed fix
data class ReconciliationResult(
val id: UUID,
val vaultId: UUID,
val currency: String,
val protocol: String,
val internalBalance: BigDecimal,
val fireblocksBalance: BigDecimal,
val drift: BigDecimal,
val absoluteDrift: BigDecimal,
val status: ReconciliationStatus,
val toleranceUsed: BigDecimal,
val createdAt: Instant,
-)
+) {
+ init {
+ require(currency.isNotBlank()) { "currency must not be blank" }
+ require(protocol.isNotBlank()) { "protocol must not be blank" }
+ require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" }
+ require(absoluteDrift.compareTo(drift.abs()) == 0) {
+ "absoluteDrift must equal abs(drift)"
+ }
+ }
+}As per coding guidelines "Validate inputs and check types; throw on misuse rather than silently producing wrong results".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| data class ReconciliationResult( | |
| val id: UUID, | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val absoluteDrift: BigDecimal, | |
| val status: ReconciliationStatus, | |
| val toleranceUsed: BigDecimal, | |
| val createdAt: Instant, | |
| ) | |
| data class ReconciliationResult( | |
| val id: UUID, | |
| val vaultId: UUID, | |
| val currency: String, | |
| val protocol: String, | |
| val internalBalance: BigDecimal, | |
| val fireblocksBalance: BigDecimal, | |
| val drift: BigDecimal, | |
| val absoluteDrift: BigDecimal, | |
| val status: ReconciliationStatus, | |
| val toleranceUsed: BigDecimal, | |
| val createdAt: Instant, | |
| ) { | |
| init { | |
| require(currency.isNotBlank()) { "currency must not be blank" } | |
| require(protocol.isNotBlank()) { "protocol must not be blank" } | |
| require(toleranceUsed >= BigDecimal.ZERO) { "tolerance must be non-negative" } | |
| require(absoluteDrift.compareTo(drift.abs()) == 0) { | |
| "absoluteDrift must equal abs(drift)" | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/reconciliation/ReconciliationResult.kt`
around lines 7 - 19, The ReconciliationResult data class must enforce
invariants: add validation in ReconciliationResult (either via an init block or
by making absoluteDrift derived) to ensure absoluteDrift == drift.abs() and that
toleranceUsed is non-negative (toleranceUsed >= 0); if these checks fail throw
an IllegalArgumentException. Locate the ReconciliationResult declaration and
implement the init block (or remove absoluteDrift from the ctor and compute val
absoluteDrift = drift.abs()) and validate toleranceUsed to prevent inconsistent
reconciliation records.
Comment on lines
+70
to
+72
| if (newStatus == TransactionStatus.CONFIRMED) { | ||
| debitInternalBalance(result) | ||
| } |
There was a problem hiding this comment.
Debit internal balance only for outgoing CONFIRMED transactions.
Line 70 debits on every CONFIRMED transition. The objective for this flow is scoped to outgoing transactions; without an outgoing-direction guard, non-outgoing confirmations can incorrectly reduce the internal ledger.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandler.kt`
around lines 70 - 72, The current TransactionStatusHandler unconditionally calls
debitInternalBalance(result) when newStatus == TransactionStatus.CONFIRMED;
change this to only debit for outgoing transactions by adding a direction guard
(e.g., check result.transaction.direction == TransactionDirection.OUTGOING or
equivalent) so that debitInternalBalance(result) is invoked only when newStatus
== TransactionStatus.CONFIRMED && direction is OUTGOING; update any imports or
enum references to TransactionDirection.OUTGOING as needed.
Comment on lines
+92
to
+111
| val result = | ||
| try { | ||
| val balanceResult = fireblocksBalancePort.getBalance(fireblocksVaultId, asset.fireblocksAssetId, true) | ||
| reconciliationService.reconcile(existing, balanceResult.available, tolerance) | ||
| } catch (e: Exception) { | ||
| log.warn( | ||
| "Fireblocks balance unavailable for vault={} asset={}/{}, creating PARTIAL result", | ||
| vaultId, | ||
| asset.currency, | ||
| asset.protocol, | ||
| e, | ||
| ) | ||
| reconciliationService.createPartialResult( | ||
| vaultId, | ||
| asset.currency, | ||
| asset.protocol, | ||
| existing.balance, | ||
| tolerance, | ||
| ) | ||
| } |
There was a problem hiding this comment.
Narrow PARTIAL fallback to Fireblocks-unavailable failures only
The catch block currently wraps both external fetch and domain reconcile logic, so non-Fireblocks failures can be silently misclassified as PARTIAL.
Proposed fix
+import com.stablecoin.custody.fireblocks.domain.exception.FireblocksApiException
...
- val result =
- try {
- val balanceResult = fireblocksBalancePort.getBalance(fireblocksVaultId, asset.fireblocksAssetId, true)
- reconciliationService.reconcile(existing, balanceResult.available, tolerance)
- } catch (e: Exception) {
- log.warn(
- "Fireblocks balance unavailable for vault={} asset={}/{}, creating PARTIAL result",
- vaultId,
- asset.currency,
- asset.protocol,
- e,
- )
- reconciliationService.createPartialResult(
- vaultId,
- asset.currency,
- asset.protocol,
- existing.balance,
- tolerance,
- )
- }
+ val balanceResult =
+ try {
+ fireblocksBalancePort.getBalance(fireblocksVaultId, asset.fireblocksAssetId, true)
+ } catch (e: FireblocksApiException) {
+ log.warn(
+ "Fireblocks balance unavailable for vault={} asset={}/{}, creating PARTIAL result",
+ vaultId,
+ asset.currency,
+ asset.protocol,
+ e,
+ )
+ val partial = reconciliationService.createPartialResult(
+ vaultId,
+ asset.currency,
+ asset.protocol,
+ existing.balance,
+ tolerance,
+ )
+ reconciliationResultRepository.save(partial)
+ auditReconciliation(partial)
+ return
+ }
+
+ val result = reconciliationService.reconcile(existing, balanceResult.available, tolerance)🧰 Tools
🪛 detekt (1.23.8)
[warning] 96-96: The caught exception is too generic. Prefer catching specific exceptions to the case that is currently handled.
(detekt.exceptions.TooGenericExceptionCaught)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJob.kt`
around lines 92 - 111, The catch currently spans both the external fetch and
domain reconcile so any exception (including reconcile failures) becomes a
PARTIAL; change BalanceReconciliationJob to only catch errors from the
Fireblocks call: call fireblocksBalancePort.getBalance(...) inside its own
try/catch and on that catch log the Fireblocks-unavailable warning and call
reconciliationService.createPartialResult(...), but let exceptions from
reconciliationService.reconcile(...) propagate (or be handled separately).
Reference fireblocksBalancePort.getBalance, reconciliationService.reconcile, and
reconciliationService.createPartialResult when making this change.
Comment on lines
+127
to
+160
| try { | ||
| val balanceResult = fireblocksBalancePort.getBalance(fireblocksVaultId, asset.fireblocksAssetId, true) | ||
| val seeded = | ||
| InternalBalance( | ||
| id = UUID.randomUUID(), | ||
| vaultId = vaultId, | ||
| currency = asset.currency, | ||
| protocol = asset.protocol, | ||
| balance = balanceResult.available, | ||
| lastTransactionId = null, | ||
| updatedAt = Instant.now(), | ||
| ) | ||
| internalBalanceRepository.save(seeded) | ||
|
|
||
| val matchedResult = reconciliationService.reconcile(seeded, balanceResult.available, tolerance) | ||
| reconciliationResultRepository.save(matchedResult) | ||
| auditReconciliation(matchedResult) | ||
|
|
||
| log.info( | ||
| "Seeded internal balance for vault={} asset={}/{} balance={}", | ||
| vaultId, | ||
| asset.currency, | ||
| asset.protocol, | ||
| balanceResult.available, | ||
| ) | ||
| } catch (e: Exception) { | ||
| log.error( | ||
| "Failed to seed internal balance for vault={} asset={}/{}", | ||
| vaultId, | ||
| asset.currency, | ||
| asset.protocol, | ||
| e, | ||
| ) | ||
| } |
There was a problem hiding this comment.
Seed failure path drops reconciliation evidence
When first-run seeding fails, the code only logs and exits. No ReconciliationResult is persisted, so this asset/vault pair leaves no auditable reconciliation outcome for that run.
Proposed fix
+import com.stablecoin.custody.fireblocks.domain.exception.FireblocksApiException
...
- } catch (e: Exception) {
- log.error(
- "Failed to seed internal balance for vault={} asset={}/{}",
- vaultId,
- asset.currency,
- asset.protocol,
- e,
- )
- }
+ } catch (e: FireblocksApiException) {
+ log.warn(
+ "Failed to seed internal balance due to Fireblocks unavailability for vault={} asset={}/{}; creating PARTIAL result",
+ vaultId,
+ asset.currency,
+ asset.protocol,
+ e,
+ )
+ val partial =
+ reconciliationService.createPartialResult(
+ vaultId = vaultId,
+ currency = asset.currency,
+ protocol = asset.protocol,
+ internalBalance = BigDecimal.ZERO,
+ toleranceUsed = tolerance,
+ )
+ reconciliationResultRepository.save(partial)
+ auditReconciliation(partial)
+ }🧰 Tools
🪛 detekt (1.23.8)
[warning] 152-152: The caught exception is too generic. Prefer catching specific exceptions to the case that is currently handled.
(detekt.exceptions.TooGenericExceptionCaught)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/kotlin/com/stablecoin/custody/fireblocks/infrastructure/scheduling/BalanceReconciliationJob.kt`
around lines 127 - 160, When seeding fails the catch block only logs and drops
evidence; create and persist a failed ReconciliationResult and call
auditReconciliation before returning so every vault/asset run is auditable. In
the catch for BalanceReconciliationJob (the try that calls
fireblocksBalancePort.getBalance, internalBalanceRepository.save,
reconciliationService.reconcile, reconciliationResultRepository.save,
auditReconciliation) build a ReconciliationResult (or equivalent DTO used
elsewhere) populated with vaultId, asset identifiers
(asset.currency/asset.protocol or asset.fireblocksAssetId), a FAILED status, the
exception message/stack info, timestamp/Instant.now(), and any other required
reconciliation metadata, then save it via
reconciliationResultRepository.save(failedResult) and call
auditReconciliation(failedResult) prior to logging the error.
Comment on lines
+15
to
+16
| CREATE INDEX idx_reconciliation_results_vault_created ON reconciliation_results(vault_id, created_at); | ||
| CREATE INDEX idx_reconciliation_results_mismatched ON reconciliation_results(status) WHERE status = 'MISMATCHED'; |
There was a problem hiding this comment.
Add a covering index for latest-by-asset lookups.
The current index on (vault_id, created_at) does not match the query shape filtering by vault_id + currency + protocol and ordering by created_at DESC, which will degrade as history grows.
Suggested migration adjustment
CREATE INDEX idx_reconciliation_results_vault_created ON reconciliation_results(vault_id, created_at);
CREATE INDEX idx_reconciliation_results_mismatched ON reconciliation_results(status) WHERE status = 'MISMATCHED';
+CREATE INDEX idx_reconciliation_results_vault_asset_latest
+ ON reconciliation_results(vault_id, currency, protocol, created_at DESC);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| CREATE INDEX idx_reconciliation_results_vault_created ON reconciliation_results(vault_id, created_at); | |
| CREATE INDEX idx_reconciliation_results_mismatched ON reconciliation_results(status) WHERE status = 'MISMATCHED'; | |
| CREATE INDEX idx_reconciliation_results_vault_created ON reconciliation_results(vault_id, created_at); | |
| CREATE INDEX idx_reconciliation_results_mismatched ON reconciliation_results(status) WHERE status = 'MISMATCHED'; | |
| CREATE INDEX idx_reconciliation_results_vault_asset_latest | |
| ON reconciliation_results(vault_id, currency, protocol, created_at DESC); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/main/resources/db/migration/V13__create_reconciliation_results.sql`
around lines 15 - 16, The existing index
idx_reconciliation_results_vault_created on reconciliation_results(vault_id,
created_at) doesn't support queries that filter by vault_id + currency +
protocol and order by created_at DESC; add a new covering index that matches
that query shape (e.g., include vault_id, currency, protocol in that key order
and created_at DESC) and include any additional columns returned by the query to
make it covering so the planner can use an index-only scan; update the migration
(instead of or alongside idx_reconciliation_results_vault_created) to create
this new index on reconciliation_results.
Comment on lines
+313
to
414
| @Test | ||
| fun `should debit internal balance on CONFIRMED status`() { | ||
| // given | ||
| val vaultId = UUID.randomUUID() | ||
| val transaction = | ||
| aTransaction( | ||
| status = TransactionStatus.CONFIRMING, | ||
| fireblocksTransactionId = "fb-tx-debit", | ||
| sourceVaultId = vaultId.toString(), | ||
| amount = BigDecimal("50.00"), | ||
| currency = "EURC", | ||
| protocol = "ETH", | ||
| ) | ||
| val existingBalance = | ||
| anInternalBalance( | ||
| vaultId = vaultId, | ||
| currency = "EURC", | ||
| protocol = "ETH", | ||
| balance = BigDecimal("1000.00"), | ||
| ) | ||
| every { transactionRepository.findByFireblocksTransactionId("fb-tx-debit") } returns transaction | ||
| every { stateMachine.transition(transaction, TransactionStatus.CONFIRMED) } returns TransactionStatus.CONFIRMED | ||
| every { transactionRepository.findByIdForUpdate(transaction.id) } returns transaction | ||
| every { transactionRepository.save(any()) } returnsArgument 0 | ||
| every { eventPublisher.publish(any()) } just runs | ||
| every { auditLogRepository.save(any()) } returnsArgument 0 | ||
| every { | ||
| internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(vaultId, "EURC", "ETH") | ||
| } returns existingBalance | ||
| every { internalBalanceRepository.save(any()) } returnsArgument 0 | ||
|
|
||
| // when | ||
| handler.handleStatusUpdate("fb-tx-debit", "COMPLETED", null, "0xhash") | ||
|
|
||
| // then | ||
| verify { | ||
| internalBalanceRepository.save( | ||
| match { it.balance == BigDecimal("950.00") && it.lastTransactionId == transaction.id.value }, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| fun `should create internal balance with negative amount if none exists on CONFIRMED status`() { | ||
| // given | ||
| val vaultId = UUID.randomUUID() | ||
| val transaction = | ||
| aTransaction( | ||
| status = TransactionStatus.CONFIRMING, | ||
| fireblocksTransactionId = "fb-tx-new-bal", | ||
| sourceVaultId = vaultId.toString(), | ||
| amount = BigDecimal("100.00"), | ||
| currency = "BTC", | ||
| protocol = "BTC", | ||
| ) | ||
| every { transactionRepository.findByFireblocksTransactionId("fb-tx-new-bal") } returns transaction | ||
| every { stateMachine.transition(transaction, TransactionStatus.CONFIRMED) } returns TransactionStatus.CONFIRMED | ||
| every { transactionRepository.findByIdForUpdate(transaction.id) } returns transaction | ||
| every { transactionRepository.save(any()) } returnsArgument 0 | ||
| every { eventPublisher.publish(any()) } just runs | ||
| every { auditLogRepository.save(any()) } returnsArgument 0 | ||
| every { | ||
| internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(vaultId, "BTC", "BTC") | ||
| } returns null | ||
| every { internalBalanceRepository.save(any()) } returnsArgument 0 | ||
|
|
||
| // when | ||
| handler.handleStatusUpdate("fb-tx-new-bal", "COMPLETED", null, "0xhash") | ||
|
|
||
| // then | ||
| verify { | ||
| internalBalanceRepository.save( | ||
| match { | ||
| it.balance.compareTo(BigDecimal("-100.00")) == 0 && | ||
| it.vaultId == vaultId && | ||
| it.currency == "BTC" && | ||
| it.protocol == "BTC" | ||
| }, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| @Test | ||
| fun `should not modify internal balance on FAILED status`() { | ||
| // given | ||
| val transaction = aTransaction(status = TransactionStatus.PROCESSING, fireblocksTransactionId = "fb-tx-fail-bal") | ||
| every { transactionRepository.findByFireblocksTransactionId("fb-tx-fail-bal") } returns transaction | ||
| every { stateMachine.transition(transaction, TransactionStatus.FAILED) } returns TransactionStatus.FAILED | ||
| every { transactionRepository.findByIdForUpdate(transaction.id) } returns transaction | ||
| every { transactionRepository.save(any()) } returnsArgument 0 | ||
| every { eventPublisher.publish(any()) } just runs | ||
| every { auditLogRepository.save(any()) } returnsArgument 0 | ||
| every { fundAllocationService.releaseByTransactionId(any()) } just runs | ||
|
|
||
| // when | ||
| handler.handleStatusUpdate("fb-tx-fail-bal", "FAILED", null, null) | ||
|
|
||
| // then | ||
| verify(exactly = 0) { internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(any(), any(), any()) } | ||
| verify(exactly = 0) { internalBalanceRepository.save(any()) } | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Add a regression test for CONFIRMED non-outgoing transactions.
Current tests validate debit/create behavior on CONFIRMED but do not protect the expected “outgoing-only” constraint. Add one case asserting no internal-balance mutation when a non-outgoing transaction reaches CONFIRMED.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@custody-fireblocks/src/test/kotlin/com/stablecoin/custody/fireblocks/domain/transaction/TransactionStatusHandlerTest.kt`
around lines 313 - 414, Add a regression test in TransactionStatusHandlerTest
that ensures CONFIRMED does not touch internal balances for non-outgoing
transactions: create a transaction via aTransaction with
status=TransactionStatus.CONFIRMING, fireblocksTransactionId like
"fb-tx-non-outgoing", and set its direction to a non-outgoing value (e.g.,
TransactionDirection.INCOMING or equivalent field on the test builder), mock
transactionRepository.findByFireblocksTransactionId to return it, mock
stateMachine.transition(transaction, TransactionStatus.CONFIRMED) to return
CONFIRMED and
transactionRepository.findByIdForUpdate/save/eventPublisher/auditLogRepository
as in other tests, call handler.handleStatusUpdate("fb-tx-non-outgoing",
"COMPLETED", null, "0xhash"), and verify that
internalBalanceRepository.findByVaultIdAndCurrencyAndProtocol(...) and
internalBalanceRepository.save(...) are NOT invoked (verify(exactly = 0){ ... })
to assert no internal-balance mutation for non-outgoing CONFIRMED transactions.
Move audit/event-publish logic from BalanceReconciliationJob into BalanceReconciliationService @transactional methods (persistResult, seedAndPersist) so outbox writes share the same transaction boundary. Change ReconciliationBreakEventOutboxPublisher propagation from REQUIRED to MANDATORY to enforce caller-provided transaction. Replace any() matchers with specific values in TransactionStatusHandlerTest.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
STR Issue
Closes #110
Changes
InternalBalancedomain model with event-derived debit tracking per vault/currency/protocol, seeded from Fireblocks API on first reconciliation runReconciliationResultdomain model with signed drift, absolute drift, tolerance, and MATCHED/MISMATCHED/PARTIAL statusBalanceReconciliationServicewith pure domain reconciliation logic comparing internal vs Fireblocks balancesTransactionStatusHandlerto debitInternalBalanceon CONFIRMED outgoing transactions (same DB transaction as status change)BalanceReconciliationJobscheduled via ShedLock (default 15min) iterating all active vaults/assets with per-pair error isolationReconciliationBreakDetectedEventandReconciliationCompletedEventdomain events, with outbox publishing for break eventsReconciliationPropertiesconfiguration with per-currency tolerance overrides (BTC, ETH, EURC)RECONCILIATION_FAILEDerror code and 3 reconciliation audit operationsfindAllActive()toVaultRepositoryandfindByVaultId()toWalletAssetRepositoryanInternalBalance()andaReconciliationResult()Checklist
./gradlew buildpassesSummary by CodeRabbit
Release Notes
New Features
Bug Fixes