Skip to content

Commit f306cf4

Browse files
Puneethkumarckclaude
Puneethkumarck
and
claude
authored
feat(s4): transfer monitor — block indexer + confirmation polling (STA-142) (#145)
* feat(s4): transfer monitor — block indexer + confirmation polling (STA-142) Add transfer monitoring system with two scheduled jobs and domain services: - TransferMonitorCommandHandler: monitors SUBMITTED/CONFIRMING/RESUBMITTING transfers, handles confirmation counting, stuck detection (>120s), resubmission via CustodyEngine, and max-attempt failure (3 attempts) - BalanceSyncCommandHandler: syncs all wallet balances from on-chain RPC data - TransferMonitorJob: @scheduled thin delegator polling every 15s - BalanceSyncJob: @scheduled thin delegator syncing every 30s - Domain ports: TransferMonitorProperties, ChainConfirmationProperties - Config: ChainMonitorConfig, TransferMonitorConfig, ClockConfig - 14 unit tests covering confirm, resubmit, max attempts, stuck detection, failed receipt, balance sync, and error resilience Closes STA-142 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(s4): address CodeRabbit review findings — crash safety, transaction isolation, fail-fast config (STA-142) Fixes 6 review findings from PR #145: 1. **Resubmission crash safety** — persist claim before calling custody engine (claimResubmission → signAndSubmit → confirmResubmission) to prevent double-submission on crash between custody call and save. 2. **Per-transfer transaction isolation** — remove class-level @transactional, wrap each processTransfer() in TransactionTemplate(REQUIRES_NEW) so failures in one transfer don't affect others. 3. **Null token contract** — add TokenContractResolver domain port to resolve token contract addresses per chain/stablecoin; BalanceSyncCommandHandler now passes the correct contract to getTokenBalance() instead of null. 4. **Fail-fast for unknown chains** — ChainConfirmationProperties throws IllegalStateException for unconfigured chains instead of silently defaulting to 1 confirmation. ChainProperties rejects non-positive minConfirmations. 5. **BalanceSync transaction scope** — remove class-level @transactional so RPC calls happen outside any DB transaction; each save runs in its own transaction. 6. **CONFIRMING stall protection** — if receipt disappears beyond a configurable grace window (confirmingTimeoutS, default 300s), transfer is marked for resubmission instead of stalling indefinitely. Added CONFIRMING → RESUBMITTING state transition for reorg recovery. Also gates fallback TransferEventPublisher behind @ConditionalOnProperty to prevent silent event loss in production. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(s4): remove unused TokenContractResolver import in BalanceSyncCommandHandlerTest Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent 3485557 commit f306cf4

19 files changed

Lines changed: 1585 additions & 2 deletions

File tree

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/application/config/ChainMonitorConfig.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
package com.stablecoin.payments.custody.application.config;
2+
3+
import com.stablecoin.payments.custody.domain.model.ChainId;
4+
import com.stablecoin.payments.custody.domain.model.StablecoinTicker;
5+
import com.stablecoin.payments.custody.domain.port.ChainConfirmationProperties;
6+
import com.stablecoin.payments.custody.domain.port.TokenContractResolver;
7+
import org.springframework.boot.context.properties.ConfigurationProperties;
8+
9+
import java.util.Map;
10+
11+
/**
12+
* Application-layer implementation of {@link ChainConfirmationProperties} and {@link TokenContractResolver}.
13+
* Binds to {@code app.chains.*} YAML properties.
14+
*/
15+
@ConfigurationProperties(prefix = "app")
16+
public record ChainMonitorConfig(
17+
Map<String, ChainProperties> chains
18+
) implements ChainConfirmationProperties, TokenContractResolver {
19+
20+
public ChainMonitorConfig {
21+
if (chains == null) {
22+
chains = Map.of();
23+
}
24+
}
25+
26+
@Override
27+
public int getMinConfirmations(String chainId) {
28+
var props = chains.get(chainId);
29+
if (props == null) {
30+
throw new IllegalStateException(
31+
"Missing chain confirmation config for chainId=%s. Configured chains: %s"
32+
.formatted(chainId, chains.keySet()));
33+
}
34+
return props.minConfirmations();
35+
}
36+
37+
@Override
38+
public String resolveContract(ChainId chainId, StablecoinTicker stablecoin) {
39+
var props = chains.get(chainId.value());
40+
if (props == null) {
41+
throw new IllegalStateException(
42+
"Missing chain config for chainId=%s. Configured chains: %s"
43+
.formatted(chainId.value(), chains.keySet()));
44+
}
45+
var contracts = props.tokenContracts();
46+
if (contracts == null || !contracts.containsKey(stablecoin.ticker())) {
47+
throw new IllegalStateException(
48+
"No token contract configured for chain=%s stablecoin=%s"
49+
.formatted(chainId.value(), stablecoin.ticker()));
50+
}
51+
return contracts.get(stablecoin.ticker());
52+
}
53+
54+
/**
55+
* Per-chain configuration properties.
56+
*/
57+
public record ChainProperties(
58+
int minConfirmations,
59+
int avgFinalityS,
60+
String rpcUrl,
61+
Map<String, String> tokenContracts
62+
) {
63+
64+
public ChainProperties {
65+
if (minConfirmations <= 0) {
66+
throw new IllegalArgumentException(
67+
"minConfirmations must be positive, got: " + minConfirmations);
68+
}
69+
if (avgFinalityS <= 0) {
70+
avgFinalityS = 12;
71+
}
72+
if (tokenContracts == null) {
73+
tokenContracts = Map.of();
74+
}
75+
}
76+
}
77+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/application/config/ClockConfig.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package com.stablecoin.payments.custody.application.config;
2+
3+
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
4+
import org.springframework.context.annotation.Bean;
5+
import org.springframework.context.annotation.Configuration;
6+
7+
import java.time.Clock;
8+
9+
/**
10+
* Provides a system UTC {@link Clock} bean for time-dependent domain logic.
11+
* <p>
12+
* Tests can override this with a fixed clock for deterministic behaviour.
13+
*/
14+
@Configuration
15+
public class ClockConfig {
16+
17+
@Bean
18+
@ConditionalOnMissingBean
19+
public Clock clock() {
20+
return Clock.systemUTC();
21+
}
22+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/application/config/TransferMonitorConfig.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package com.stablecoin.payments.custody.application.config;
2+
3+
import com.stablecoin.payments.custody.domain.port.TransferMonitorProperties;
4+
import org.springframework.boot.context.properties.ConfigurationProperties;
5+
6+
/**
7+
* Application-layer implementation of {@link TransferMonitorProperties}.
8+
* Binds to {@code app.transfer.*} YAML properties.
9+
*/
10+
@ConfigurationProperties(prefix = "app.transfer")
11+
public record TransferMonitorConfig(
12+
int resubmitTimeoutS,
13+
int maxAttempts,
14+
int confirmingTimeoutS
15+
) implements TransferMonitorProperties {
16+
17+
public TransferMonitorConfig {
18+
if (resubmitTimeoutS <= 0) {
19+
resubmitTimeoutS = 120;
20+
}
21+
if (maxAttempts <= 0) {
22+
maxAttempts = 3;
23+
}
24+
if (confirmingTimeoutS <= 0) {
25+
confirmingTimeoutS = 300;
26+
}
27+
}
28+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/application/scheduler/BalanceSyncJob.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package com.stablecoin.payments.custody.application.scheduler;
2+
3+
import com.stablecoin.payments.custody.domain.service.BalanceSyncCommandHandler;
4+
import lombok.RequiredArgsConstructor;
5+
import lombok.extern.slf4j.Slf4j;
6+
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
7+
import org.springframework.scheduling.annotation.Scheduled;
8+
import org.springframework.stereotype.Component;
9+
10+
/**
11+
* Scheduled job that syncs blockchain balances from RPC every 30 seconds.
12+
* <p>
13+
* Delegates all business logic to {@link BalanceSyncCommandHandler}.
14+
*/
15+
@Slf4j
16+
@Component
17+
@RequiredArgsConstructor
18+
@ConditionalOnProperty(name = "app.balance.sync.enabled", havingValue = "true", matchIfMissing = true)
19+
public class BalanceSyncJob {
20+
21+
private final BalanceSyncCommandHandler balanceSyncCommandHandler;
22+
23+
@Scheduled(fixedDelayString = "${app.balance.sync.interval-ms:30000}")
24+
public void syncBalances() {
25+
log.debug("Balance sync job triggered");
26+
balanceSyncCommandHandler.syncAllBalances();
27+
}
28+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/application/scheduler/TransferMonitorJob.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
package com.stablecoin.payments.custody.application.scheduler;
2+
3+
import com.stablecoin.payments.custody.domain.service.TransferMonitorCommandHandler;
4+
import lombok.RequiredArgsConstructor;
5+
import lombok.extern.slf4j.Slf4j;
6+
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
7+
import org.springframework.scheduling.annotation.Scheduled;
8+
import org.springframework.stereotype.Component;
9+
10+
/**
11+
* Scheduled job that polls in-flight chain transfers every 15 seconds.
12+
* <p>
13+
* Delegates all business logic to {@link TransferMonitorCommandHandler}.
14+
*/
15+
@Slf4j
16+
@Component
17+
@RequiredArgsConstructor
18+
@ConditionalOnProperty(name = "app.transfer.monitor.enabled", havingValue = "true", matchIfMissing = true)
19+
public class TransferMonitorJob {
20+
21+
private final TransferMonitorCommandHandler transferMonitorCommandHandler;
22+
23+
@Scheduled(fixedDelayString = "${app.transfer.monitor.interval-ms:15000}")
24+
public void pollTransfers() {
25+
log.debug("Transfer monitor job triggered");
26+
transferMonitorCommandHandler.monitorPendingTransfers();
27+
}
28+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/config/FallbackAdaptersConfig.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,10 @@
1111
import com.stablecoin.payments.custody.domain.port.SignResult;
1212
import com.stablecoin.payments.custody.domain.port.TransactionReceipt;
1313
import com.stablecoin.payments.custody.domain.port.TransactionStatus;
14+
import com.stablecoin.payments.custody.domain.port.TransferEventPublisher;
1415
import lombok.extern.slf4j.Slf4j;
1516
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
17+
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
1618
import org.springframework.context.annotation.Bean;
1719
import org.springframework.context.annotation.Configuration;
1820

@@ -132,6 +134,20 @@ public BigDecimal getTokenBalance(ChainId chainId, String address, String tokenC
132134
};
133135
}
134136

137+
/**
138+
* Fallback event publisher for dev/test environments without Kafka outbox.
139+
* Logs the event instead of publishing.
140+
* Gated by property to prevent silent event loss in production.
141+
*/
142+
@Bean
143+
@ConditionalOnMissingBean
144+
@ConditionalOnProperty(name = "app.transfer.event-publisher.fallback-enabled",
145+
havingValue = "true", matchIfMissing = true)
146+
public TransferEventPublisher fallbackTransferEventPublisher() {
147+
log.info("Using fallback TransferEventPublisher (log only)");
148+
return event -> log.warn("[FALLBACK-EVENT] Published event: {}", event);
149+
}
150+
135151
static class InMemoryNonceRepository implements NonceRepository {
136152

137153
private final ConcurrentHashMap<String, AtomicLong> nonces = new ConcurrentHashMap<>();

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/domain/model/ChainTransfer.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ public record ChainTransfer(
8181

8282
// -- Resubmission path ------------------------------------------
8383
new StateTransition<>(SUBMITTED, RESUBMIT, RESUBMITTING),
84+
new StateTransition<>(CONFIRMING, RESUBMIT, RESUBMITTING),
8485
new StateTransition<>(RESUBMITTING, CONFIRM_SUBMISSION, SUBMITTED),
8586

8687
// -- Failure from multiple states --------------------------------
@@ -254,6 +255,41 @@ public ChainTransfer resubmit(String newTxHash) {
254255
.build();
255256
}
256257

258+
/**
259+
* Claims a resubmission attempt by incrementing the attempt counter.
260+
* Must be persisted BEFORE calling the custody engine to ensure crash safety.
261+
* Stays in RESUBMITTING state.
262+
*/
263+
public ChainTransfer claimResubmission() {
264+
assertNotTerminal();
265+
if (status != RESUBMITTING) {
266+
throw new IllegalStateException(
267+
"Can only claim resubmission in RESUBMITTING state, current: " + status);
268+
}
269+
return toBuilder()
270+
.attemptCount(attemptCount + 1)
271+
.updatedAt(Instant.now())
272+
.build();
273+
}
274+
275+
/**
276+
* Completes a previously claimed resubmission with the new transaction hash.
277+
* Transitions RESUBMITTING -> SUBMITTED without incrementing attempt count
278+
* (already incremented by {@link #claimResubmission()}).
279+
*/
280+
public ChainTransfer confirmResubmission(String newTxHash) {
281+
assertNotTerminal();
282+
if (newTxHash == null || newTxHash.isBlank()) {
283+
throw new IllegalArgumentException("Transaction hash is required for resubmission");
284+
}
285+
var nextState = STATE_MACHINE.transition(status, CONFIRM_SUBMISSION);
286+
return toBuilder()
287+
.status(nextState)
288+
.txHash(newTxHash)
289+
.updatedAt(Instant.now())
290+
.build();
291+
}
292+
257293
/**
258294
* Fails the transfer. Can be triggered from multiple non-terminal states.
259295
*/

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/domain/port/ChainConfirmationProperties.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
package com.stablecoin.payments.custody.domain.port;
2+
3+
/**
4+
* Domain port for per-chain confirmation configuration.
5+
*/
6+
public interface ChainConfirmationProperties {
7+
8+
/**
9+
* Returns the minimum confirmations required for a given chain.
10+
*
11+
* @param chainId the chain identifier (e.g., "base", "ethereum", "solana")
12+
* @return minimum confirmations (always positive)
13+
* @throws IllegalStateException if chain is not configured
14+
*/
15+
int getMinConfirmations(String chainId);
16+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/domain/port/TokenContractResolver.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package com.stablecoin.payments.custody.domain.port;
2+
3+
import com.stablecoin.payments.custody.domain.model.ChainId;
4+
import com.stablecoin.payments.custody.domain.model.StablecoinTicker;
5+
6+
/**
7+
* Domain port for resolving token contract addresses per chain and stablecoin.
8+
*/
9+
public interface TokenContractResolver {
10+
11+
/**
12+
* Resolves the on-chain token contract address for the given chain and stablecoin.
13+
*
14+
* @param chainId the blockchain identifier
15+
* @param stablecoin the stablecoin ticker
16+
* @return the contract address (e.g., ERC-20 address or SPL mint)
17+
* @throws IllegalStateException if no contract is configured for the given chain/stablecoin
18+
*/
19+
String resolveContract(ChainId chainId, StablecoinTicker stablecoin);
20+
}

‎blockchain-custody/blockchain-custody/src/main/java/com/stablecoin/payments/custody/domain/port/TransferMonitorProperties.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package com.stablecoin.payments.custody.domain.port;
2+
3+
/**
4+
* Domain port for transfer monitor configuration.
5+
*/
6+
public interface TransferMonitorProperties {
7+
8+
/**
9+
* Returns the number of seconds before a SUBMITTED transfer is considered stuck.
10+
*/
11+
int resubmitTimeoutS();
12+
13+
/**
14+
* Returns the maximum number of submission attempts before a transfer is failed.
15+
*/
16+
int maxAttempts();
17+
18+
/**
19+
* Returns the number of seconds a CONFIRMING transfer can remain without a receipt
20+
* before being marked for resubmission (e.g., after a chain reorg).
21+
* Defaults to 300 seconds (5 minutes).
22+
*/
23+
default int confirmingTimeoutS() {
24+
return 300;
25+
}
26+
}

0 commit comments

Comments
 (0)