From 0449395eeed219d807cc00dd420f1c7d243909d6 Mon Sep 17 00:00:00 2001 From: GrapeS Date: Thu, 17 Sep 2026 15:05:15 +0800 Subject: [PATCH 1/5] refactor(rpc): serve solidity and pbft grpc via shared service instances Remove the duplicated wallet-solidity gRPC stack by serving the solidity and PBFT surfaces from the shared service instances, with a cursor interceptor selecting the store each call reads from. - Add cursor server interceptors for the solidity and PBFT ports and bind them to the shared services. - Serve solidity and PBFT gRPC through the shared service instances instead of separate handler implementations. - Deduplicate the remaining wallet-solidity read handlers. - Return after onError so a failed call is closed exactly once. - Document that the wallet-solidity API is the read-only subset of wallet, and assert that subset relationship in tests. - Cover the error path, cursor wiring and PBFT reads; drop probe tests that guarded nothing. --- .../org/tron/core/services/RpcApiService.java | 422 ++++------------ .../filter/CursorServerInterceptor.java | 52 ++ .../filter/PbftCursorInterceptor.java | 15 + .../filter/SolidityCursorInterceptor.java | 15 + .../interfaceOnPBFT/RpcApiServiceOnPBFT.java | 477 +----------------- .../RpcApiServiceOnSolidity.java | 468 +---------------- .../services/RpcApiServiceErrorPathTest.java | 140 +++++ .../core/services/RpcApiServicesTest.java | 2 + .../filter/CursorInterceptorScopeTest.java | 136 +++++ .../filter/CursorInterceptorServerTest.java | 123 +++++ .../filter/CursorInterceptorWiringTest.java | 101 ++++ 11 files changed, 687 insertions(+), 1264 deletions(-) create mode 100644 framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java create mode 100644 framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java create mode 100644 framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java create mode 100644 framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java create mode 100644 framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java create mode 100644 framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java create mode 100644 framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java diff --git a/framework/src/main/java/org/tron/core/services/RpcApiService.java b/framework/src/main/java/org/tron/core/services/RpcApiService.java index c9e961099bf..d3ba4afda0e 100755 --- a/framework/src/main/java/org/tron/core/services/RpcApiService.java +++ b/framework/src/main/java/org/tron/core/services/RpcApiService.java @@ -178,6 +178,8 @@ public class RpcApiService extends RpcService { private MetricsApiService metricsApiService; @Getter private DatabaseApi databaseApi = new DatabaseApi(); + // WalletApi is the full protocol.Wallet impl (HEAD); WalletSolidityApi is its read-only subset, + // reused by the Solidity/PBFT cursor ports. private WalletApi walletApi = new WalletApi(); @Getter private WalletSolidityApi walletSolidityApi = new WalletSolidityApi(); @@ -355,249 +357,138 @@ public void getDynamicProperties(EmptyMessage request, } /** - * WalletSolidityApi. + * WalletSolidityApi is the full implementation of the {@code protocol.WalletSolidity} gRPC + * service. Every method here is read-only and also present on {@link WalletApi}, so each one + * delegates to the shared {@code WalletApi} singleton instead of repeating its body. */ public class WalletSolidityApi extends WalletSolidityImplBase { @Override public void getAccount(Account request, StreamObserver responseObserver) { - ByteString addressBs = request.getAddress(); - if (addressBs != null) { - Account reply = wallet.getAccount(request); - responseObserver.onNext(reply); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getAccount(request, responseObserver); } @Override public void getAccountById(Account request, StreamObserver responseObserver) { - ByteString id = request.getAccountId(); - if (id != null) { - Account reply = wallet.getAccountById(request); - responseObserver.onNext(reply); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getAccountById(request, responseObserver); } @Override public void listWitnesses(EmptyMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getWitnessList()); - responseObserver.onCompleted(); + walletApi.listWitnesses(request, responseObserver); } @Override public void getPaginatedNowWitnessList(PaginatedMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext( - wallet.getPaginatedNowWitnessList(request.getOffset(), request.getLimit())); - } catch (MaintenanceUnavailableException e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getPaginatedNowWitnessList(request, responseObserver); } @Override public void getAssetIssueList(EmptyMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getAssetIssueList()); - responseObserver.onCompleted(); + walletApi.getAssetIssueList(request, responseObserver); } @Override public void getPaginatedAssetIssueList(PaginatedMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getAssetIssueList(request.getOffset(), request.getLimit())); - responseObserver.onCompleted(); + walletApi.getPaginatedAssetIssueList(request, responseObserver); } @Override public void getAssetIssueByName(BytesMessage request, StreamObserver responseObserver) { - ByteString assetName = request.getValue(); - if (assetName != null) { - try { - responseObserver.onNext(wallet.getAssetIssueByName(assetName)); - } catch (NonUniqueObjectException e) { - responseObserver.onNext(null); - logger.debug("Solidity NonUniqueObjectException: {}", e.getMessage()); - } - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getAssetIssueByName(request, responseObserver); } @Override public void getAssetIssueListByName(BytesMessage request, StreamObserver responseObserver) { - ByteString assetName = request.getValue(); - - if (assetName != null) { - responseObserver.onNext(wallet.getAssetIssueListByName(assetName)); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getAssetIssueListByName(request, responseObserver); } @Override public void getAssetIssueById(BytesMessage request, StreamObserver responseObserver) { - ByteString assetId = request.getValue(); - - if (assetId != null) { - responseObserver.onNext(wallet.getAssetIssueById(assetId.toStringUtf8())); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getAssetIssueById(request, responseObserver); } @Override public void getNowBlock(EmptyMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getNowBlock()); - responseObserver.onCompleted(); + walletApi.getNowBlock(request, responseObserver); } @Override public void getNowBlock2(EmptyMessage request, StreamObserver responseObserver) { - responseObserver.onNext(block2Extention(wallet.getNowBlock())); - responseObserver.onCompleted(); + walletApi.getNowBlock2(request, responseObserver); } @Override public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - long num = request.getNum(); - if (num >= 0) { - Block reply = wallet.getBlockByNum(num); - responseObserver.onNext(reply); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getBlockByNum(request, responseObserver); } @Override public void getBlockByNum2(NumberMessage request, StreamObserver responseObserver) { - long num = request.getNum(); - if (num >= 0) { - Block reply = wallet.getBlockByNum(num); - responseObserver.onNext(block2Extention(reply)); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getBlockByNum2(request, responseObserver); } @Override public void getDelegatedResource(DelegatedResourceMessage request, StreamObserver responseObserver) { - responseObserver - .onNext(wallet.getDelegatedResource(request.getFromAddress(), request.getToAddress())); - responseObserver.onCompleted(); + walletApi.getDelegatedResource(request, responseObserver); } @Override public void getDelegatedResourceV2(DelegatedResourceMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getDelegatedResourceV2( - request.getFromAddress(), request.getToAddress()) - ); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getDelegatedResourceV2(request, responseObserver); } @Override public void getDelegatedResourceAccountIndex(BytesMessage request, StreamObserver responseObserver) { - try { - responseObserver - .onNext(wallet.getDelegatedResourceAccountIndex(request.getValue())); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getDelegatedResourceAccountIndex(request, responseObserver); } @Override public void getDelegatedResourceAccountIndexV2(BytesMessage request, StreamObserver responseObserver) { - try { - responseObserver - .onNext(wallet.getDelegatedResourceAccountIndexV2(request.getValue())); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getDelegatedResourceAccountIndexV2(request, responseObserver); } @Override public void getCanDelegatedMaxSize(GrpcAPI.CanDelegatedMaxSizeRequestMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getCanDelegatedMaxSize( - request.getOwnerAddress(),request.getType())); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getCanDelegatedMaxSize(request, responseObserver); } @Override public void getAvailableUnfreezeCount(GrpcAPI.GetAvailableUnfreezeCountRequestMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getAvailableUnfreezeCount( - request.getOwnerAddress())); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getAvailableUnfreezeCount(request, responseObserver); } @Override public void getCanWithdrawUnfreezeAmount(CanWithdrawUnfreezeAmountRequestMessage request, StreamObserver responseObserver) { - try { - responseObserver - .onNext(wallet.getCanWithdrawUnfreezeAmount( - request.getOwnerAddress(), request.getTimestamp()) - ); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getCanWithdrawUnfreezeAmount(request, responseObserver); } @Override public void getExchangeById(BytesMessage request, StreamObserver responseObserver) { - ByteString exchangeId = request.getValue(); - - if (Objects.nonNull(exchangeId)) { - responseObserver.onNext(wallet.getExchangeById(exchangeId)); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getExchangeById(request, responseObserver); } @Override public void listExchanges(EmptyMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getExchangeList()); - responseObserver.onCompleted(); + walletApi.listExchanges(request, responseObserver); } @Override @@ -609,29 +500,13 @@ public void getTransactionCountByBlockNum(NumberMessage request, @Override public void getTransactionById(BytesMessage request, StreamObserver responseObserver) { - ByteString id = request.getValue(); - if (null != id) { - Transaction reply = wallet.getTransactionById(id); - - responseObserver.onNext(reply); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getTransactionById(request, responseObserver); } @Override public void getTransactionInfoById(BytesMessage request, StreamObserver responseObserver) { - ByteString id = request.getValue(); - if (null != id) { - TransactionInfo reply = wallet.getTransactionInfoById(id); - - responseObserver.onNext(reply); - } else { - responseObserver.onNext(null); - } - responseObserver.onCompleted(); + walletApi.getTransactionInfoById(request, responseObserver); } @Override @@ -654,198 +529,78 @@ public void getBurnTrx(EmptyMessage request, StreamObserver respo @Override public void getMerkleTreeVoucherInfo(OutputPointInfo request, StreamObserver responseObserver) { - - try { - IncrementalMerkleVoucherInfo witnessInfo = wallet - .getMerkleTreeVoucherInfo(request); - responseObserver.onNext(witnessInfo); - } catch (Exception ex) { - responseObserver.onError(getRunTimeException(ex)); - } - responseObserver.onCompleted(); + walletApi.getMerkleTreeVoucherInfo(request, responseObserver); } @Override public void scanNoteByIvk(GrpcAPI.IvkDecryptParameters request, StreamObserver responseObserver) { - long startNum = request.getStartBlockIndex(); - long endNum = request.getEndBlockIndex(); - - try { - DecryptNotes decryptNotes = wallet - .scanNoteByIvk(startNum, endNum, request.getIvk().toByteArray()); - responseObserver.onNext(decryptNotes); - } catch (BadItemException | ZksnarkException e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.scanNoteByIvk(request, responseObserver); } @Override public void scanAndMarkNoteByIvk(GrpcAPI.IvkDecryptAndMarkParameters request, StreamObserver responseObserver) { - long startNum = request.getStartBlockIndex(); - long endNum = request.getEndBlockIndex(); - - try { - DecryptNotesMarked decryptNotes = wallet.scanAndMarkNoteByIvk(startNum, endNum, - request.getIvk().toByteArray(), - request.getAk().toByteArray(), - request.getNk().toByteArray()); - responseObserver.onNext(decryptNotes); - } catch (BadItemException | ZksnarkException | InvalidProtocolBufferException - | ItemNotFoundException e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.scanAndMarkNoteByIvk(request, responseObserver); } @Override public void scanNoteByOvk(GrpcAPI.OvkDecryptParameters request, StreamObserver responseObserver) { - long startNum = request.getStartBlockIndex(); - long endNum = request.getEndBlockIndex(); - try { - DecryptNotes decryptNotes = wallet - .scanNoteByOvk(startNum, endNum, request.getOvk().toByteArray()); - responseObserver.onNext(decryptNotes); - } catch (BadItemException | ZksnarkException e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.scanNoteByOvk(request, responseObserver); } @Override public void isSpend(NoteParameters request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.isSpend(request)); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.isSpend(request, responseObserver); } @Override public void scanShieldedTRC20NotesByIvk(IvkDecryptTRC20Parameters request, StreamObserver responseObserver) { - if (rejectIfEventsPresent(responseObserver, request.getEventsList())) { - return; - } - long startNum = request.getStartBlockIndex(); - long endNum = request.getEndBlockIndex(); - byte[] contractAddress = request.getShieldedTRC20ContractAddress().toByteArray(); - byte[] ivk = request.getIvk().toByteArray(); - byte[] ak = request.getAk().toByteArray(); - byte[] nk = request.getNk().toByteArray(); - - try { - responseObserver.onNext( - wallet.scanShieldedTRC20NotesByIvk(startNum, endNum, contractAddress, ivk, ak, nk)); - - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.scanShieldedTRC20NotesByIvk(request, responseObserver); } @Override public void scanShieldedTRC20NotesByOvk(OvkDecryptTRC20Parameters request, StreamObserver responseObserver) { - if (rejectIfEventsPresent(responseObserver, request.getEventsList())) { - return; - } - long startNum = request.getStartBlockIndex(); - long endNum = request.getEndBlockIndex(); - byte[] contractAddress = request.getShieldedTRC20ContractAddress().toByteArray(); - byte[] ovk = request.getOvk().toByteArray(); - try { - responseObserver - .onNext(wallet.scanShieldedTRC20NotesByOvk(startNum, endNum, ovk, contractAddress)); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.scanShieldedTRC20NotesByOvk(request, responseObserver); } @Override public void isShieldedTRC20ContractNoteSpent(NfTRC20Parameters request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.isShieldedTRC20ContractNoteSpent(request)); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.isShieldedTRC20ContractNoteSpent(request, responseObserver); } @Override public void getMarketOrderByAccount(BytesMessage request, StreamObserver responseObserver) { - try { - ByteString address = request.getValue(); - - MarketOrderList marketOrderList = wallet - .getMarketOrderByAccount(address); - responseObserver.onNext(marketOrderList); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getMarketOrderByAccount(request, responseObserver); } @Override public void getMarketOrderById(BytesMessage request, StreamObserver responseObserver) { - try { - ByteString address = request.getValue(); - - MarketOrder marketOrder = wallet - .getMarketOrderById(address); - responseObserver.onNext(marketOrder); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getMarketOrderById(request, responseObserver); } @Override public void getMarketPriceByPair(MarketOrderPair request, StreamObserver responseObserver) { - try { - MarketPriceList marketPriceList = wallet - .getMarketPriceByPair(request.getSellTokenId().toByteArray(), - request.getBuyTokenId().toByteArray()); - responseObserver.onNext(marketPriceList); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getMarketPriceByPair(request, responseObserver); } @Override public void getMarketOrderListByPair(org.tron.protos.Protocol.MarketOrderPair request, StreamObserver responseObserver) { - try { - MarketOrderList orderPairList = wallet - .getMarketOrderListByPair(request.getSellTokenId().toByteArray(), - request.getBuyTokenId().toByteArray()); - responseObserver.onNext(orderPairList); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getMarketOrderListByPair(request, responseObserver); } @Override public void getMarketPairList(EmptyMessage request, StreamObserver responseObserver) { - try { - MarketOrderPairList pairList = wallet.getMarketPairList(); - responseObserver.onNext(pairList); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getMarketPairList(request, responseObserver); } @Override @@ -858,45 +613,13 @@ public void triggerConstantContract(TriggerSmartContract request, @Override public void estimateEnergy(TriggerSmartContract request, StreamObserver responseObserver) { - TransactionExtention.Builder trxExtBuilder = TransactionExtention.newBuilder(); - Return.Builder retBuilder = Return.newBuilder(); - EstimateEnergyMessage.Builder estimateBuilder - = EstimateEnergyMessage.newBuilder(); - - try { - TransactionCapsule trxCap = createTransactionCapsule(request, - ContractType.TriggerSmartContract); - wallet.estimateEnergy(request, trxCap, trxExtBuilder, retBuilder, estimateBuilder); - } catch (ContractValidateException | VMIllegalException e) { - retBuilder.setResult(false).setCode(response_code.CONTRACT_VALIDATE_ERROR) - .setMessage(ByteString.copyFromUtf8(Wallet - .CONTRACT_VALIDATE_ERROR + e.getMessage())); - logger.warn(CONTRACT_VALIDATE_EXCEPTION, e.getMessage()); - } catch (RuntimeException e) { - retBuilder.setResult(false).setCode(response_code.CONTRACT_EXE_ERROR) - .setMessage(ByteString.copyFromUtf8(e.getClass() + " : " + e.getMessage())); - logger.warn("When run estimate energy in VM, have Runtime Exception: " + e.getMessage()); - } catch (Exception e) { - retBuilder.setResult(false).setCode(response_code.OTHER_ERROR) - .setMessage(ByteString.copyFromUtf8(e.getClass() + " : " + e.getMessage())); - logger.warn(UNKNOWN_EXCEPTION_CAUGHT + e.getMessage(), e); - } finally { - estimateBuilder.setResult(retBuilder); - responseObserver.onNext(estimateBuilder.build()); - responseObserver.onCompleted(); - } + walletApi.estimateEnergy(request, responseObserver); } @Override public void getTransactionInfoByBlockNum(NumberMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getTransactionInfoByBlockNum(request.getNum())); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - - responseObserver.onCompleted(); + walletApi.getTransactionInfoByBlockNum(request, responseObserver); } @Override @@ -908,23 +631,13 @@ public void getBlock(GrpcAPI.BlockReq request, @Override public void getBandwidthPrices(EmptyMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getBandwidthPrices()); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getBandwidthPrices(request, responseObserver); } @Override public void getEnergyPrices(EmptyMessage request, StreamObserver responseObserver) { - try { - responseObserver.onNext(wallet.getEnergyPrices()); - } catch (Exception e) { - responseObserver.onError(getRunTimeException(e)); - } - responseObserver.onCompleted(); + walletApi.getEnergyPrices(request, responseObserver); } } @@ -1455,15 +1168,26 @@ public void getNowBlock2(EmptyMessage request, @Override public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - responseObserver.onNext(wallet.getBlockByNum(request.getNum())); + long num = request.getNum(); + if (num >= 0) { + Block reply = wallet.getBlockByNum(num); + responseObserver.onNext(reply); + } else { + responseObserver.onNext(null); + } responseObserver.onCompleted(); } @Override public void getBlockByNum2(NumberMessage request, StreamObserver responseObserver) { - Block block = wallet.getBlockByNum(request.getNum()); - responseObserver.onNext(block2Extention(block)); + long num = request.getNum(); + if (num >= 0) { + Block reply = wallet.getBlockByNum(num); + responseObserver.onNext(block2Extention(reply)); + } else { + responseObserver.onNext(null); + } responseObserver.onCompleted(); } @@ -1570,7 +1294,7 @@ public void getAssetIssueByName(BytesMessage request, responseObserver.onNext(wallet.getAssetIssueByName(assetName)); } catch (NonUniqueObjectException e) { responseObserver.onNext(null); - logger.debug("FullNode NonUniqueObjectException: {}", e.getMessage()); + logger.debug("NonUniqueObjectException: {}", e.getMessage()); } } else { responseObserver.onNext(null); @@ -1857,6 +1581,7 @@ public void getPaginatedNowWitnessList(PaginatedMessage request, wallet.getPaginatedNowWitnessList(request.getOffset(), request.getLimit())); } catch (MaintenanceUnavailableException e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1886,6 +1611,7 @@ public void getDelegatedResourceV2(DelegatedResourceMessage request, ); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1898,6 +1624,7 @@ public void getDelegatedResourceAccountIndex(BytesMessage request, .onNext(wallet.getDelegatedResourceAccountIndex(request.getValue())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1910,6 +1637,7 @@ public void getDelegatedResourceAccountIndexV2(BytesMessage request, .onNext(wallet.getDelegatedResourceAccountIndexV2(request.getValue())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1922,6 +1650,7 @@ public void getCanDelegatedMaxSize(GrpcAPI.CanDelegatedMaxSizeRequestMessage req request.getOwnerAddress(), request.getType())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -1935,6 +1664,7 @@ public void getAvailableUnfreezeCount(GrpcAPI.GetAvailableUnfreezeCountRequestMe request.getOwnerAddress())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -1950,6 +1680,7 @@ public void getCanWithdrawUnfreezeAmount(CanWithdrawUnfreezeAmountRequestMessage )); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1961,6 +1692,7 @@ public void getBandwidthPrices(EmptyMessage request, responseObserver.onNext(wallet.getBandwidthPrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1972,6 +1704,7 @@ public void getEnergyPrices(EmptyMessage request, responseObserver.onNext(wallet.getEnergyPrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1983,6 +1716,7 @@ public void getMemoFee(EmptyMessage request, responseObserver.onNext(wallet.getMemoFeePrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2039,6 +1773,7 @@ public void getNodeInfo(EmptyMessage request, StreamObserver responseO responseObserver.onNext(nodeInfoService.getNodeInfo().transferToProtoEntity()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2515,6 +2250,7 @@ public void getTransactionInfoByBlockNum(NumberMessage request, responseObserver.onNext(wallet.getTransactionInfoByBlockNum(request.getNum())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -2544,6 +2280,7 @@ public void getMarketOrderByAccount(BytesMessage request, responseObserver.onNext(marketOrderList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2559,6 +2296,7 @@ public void getMarketOrderById(BytesMessage request, responseObserver.onNext(marketOrder); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2573,6 +2311,7 @@ public void getMarketPriceByPair(MarketOrderPair request, responseObserver.onNext(marketPriceList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2587,6 +2326,7 @@ public void getMarketOrderListByPair(org.tron.protos.Protocol.MarketOrderPair re responseObserver.onNext(orderPairList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2599,6 +2339,7 @@ public void getMarketPairList(EmptyMessage request, responseObserver.onNext(pairList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2648,6 +2389,7 @@ public void getRewardInfoCommon(BytesMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2661,6 +2403,7 @@ public void getBurnTrxCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2676,6 +2419,7 @@ public void getBrokerageInfoCommon(BytesMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2702,6 +2446,7 @@ public void getTransactionFromPendingCommon(BytesMessage request, responseObserver.onNext(transactionCapsule == null ? null : transactionCapsule.getInstance()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2714,6 +2459,7 @@ public void getTransactionListFromPendingCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2726,6 +2472,7 @@ public void getPendingSizeCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2741,6 +2488,7 @@ public void getBlockCommon(GrpcAPI.BlockReq request, } else { responseObserver.onError(getRunTimeException(e)); } + return; } responseObserver.onCompleted(); } diff --git a/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java new file mode 100644 index 00000000000..74d14daede9 --- /dev/null +++ b/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java @@ -0,0 +1,52 @@ +package org.tron.core.services.filter; + +import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener; +import io.grpc.Metadata; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import io.grpc.ServerInterceptor; +import org.springframework.beans.factory.annotation.Autowired; +import org.tron.core.db.Manager; +import org.tron.core.db2.core.Chainbase; + +/** + * Switches the current thread's read cursor around the synchronous handler callback and restores it + * afterwards, so the handler reads from the snapshot its subclass selects (HEAD / SOLIDITY / PBFT); + * the services behind it never touch the cursor. + * + *

Two invariants it relies on: + *

    + *
  • The bracket wraps {@code onHalfClose()} — where gRPC runs the unary handler inline — not + * {@code interceptCall}, which may land on another pool thread; the cursor is a + * {@link ThreadLocal}, so setting it elsewhere fails silently and the port serves HEAD. + *
  • Handlers must be synchronous: the cursor is reset when {@code onHalfClose} returns, so a read + * deferred to another thread would read HEAD. + *
+ * + *

The {@code finally} reset is mandatory: the fixed thread pool is reused, so a leftover cursor + * leaks into the next call on that thread. + */ +public abstract class CursorServerInterceptor implements ServerInterceptor { + + @Autowired + protected Manager dbManager; + + /** Snapshot every call on this server reads from; set by the subclass. */ + protected Chainbase.Cursor cursor; + + @Override + public ServerCall.Listener interceptCall( + ServerCall call, Metadata headers, ServerCallHandler next) { + return new SimpleForwardingServerCallListener(next.startCall(call, headers)) { + @Override + public void onHalfClose() { + try { + dbManager.setCursor(cursor); + super.onHalfClose(); + } finally { + dbManager.resetCursor(); + } + } + }; + } +} diff --git a/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java new file mode 100644 index 00000000000..3ea55501d35 --- /dev/null +++ b/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java @@ -0,0 +1,15 @@ +package org.tron.core.services.filter; + +import org.springframework.stereotype.Component; +import org.tron.core.db2.core.Chainbase; + +/** + * Makes every call on the PBFT gRPC server read the PBFT-confirmed state view. + */ +@Component +public class PbftCursorInterceptor extends CursorServerInterceptor { + + public PbftCursorInterceptor() { + this.cursor = Chainbase.Cursor.PBFT; + } +} diff --git a/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java new file mode 100644 index 00000000000..5cf95ca0db4 --- /dev/null +++ b/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java @@ -0,0 +1,15 @@ +package org.tron.core.services.filter; + +import org.springframework.stereotype.Component; +import org.tron.core.db2.core.Chainbase; + +/** + * Makes every call on the Solidity gRPC server read the solidified state view. + */ +@Component +public class SolidityCursorInterceptor extends CursorServerInterceptor { + + public SolidityCursorInterceptor() { + this.cursor = Chainbase.Cursor.SOLIDITY; + } +} diff --git a/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java b/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java index 54e7b69f7fc..2e6d1bd59bd 100755 --- a/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java +++ b/framework/src/main/java/org/tron/core/services/interfaceOnPBFT/RpcApiServiceOnPBFT.java @@ -1,67 +1,22 @@ package org.tron.core.services.interfaceOnPBFT; +import io.grpc.ServerInterceptors; import io.grpc.netty.NettyServerBuilder; -import io.grpc.stub.StreamObserver; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; -import org.tron.api.DatabaseGrpc.DatabaseImplBase; -import org.tron.api.GrpcAPI; -import org.tron.api.GrpcAPI.AssetIssueList; -import org.tron.api.GrpcAPI.BlockExtention; -import org.tron.api.GrpcAPI.BlockReference; -import org.tron.api.GrpcAPI.BytesMessage; -import org.tron.api.GrpcAPI.CanDelegatedMaxSizeRequestMessage; -import org.tron.api.GrpcAPI.CanDelegatedMaxSizeResponseMessage; -import org.tron.api.GrpcAPI.CanWithdrawUnfreezeAmountRequestMessage; -import org.tron.api.GrpcAPI.CanWithdrawUnfreezeAmountResponseMessage; -import org.tron.api.GrpcAPI.DecryptNotesTRC20; -import org.tron.api.GrpcAPI.DelegatedResourceList; -import org.tron.api.GrpcAPI.DelegatedResourceMessage; -import org.tron.api.GrpcAPI.EmptyMessage; -import org.tron.api.GrpcAPI.ExchangeList; -import org.tron.api.GrpcAPI.GetAvailableUnfreezeCountRequestMessage; -import org.tron.api.GrpcAPI.GetAvailableUnfreezeCountResponseMessage; -import org.tron.api.GrpcAPI.IvkDecryptTRC20Parameters; -import org.tron.api.GrpcAPI.NfTRC20Parameters; -import org.tron.api.GrpcAPI.NoteParameters; -import org.tron.api.GrpcAPI.NullifierResult; -import org.tron.api.GrpcAPI.NumberMessage; -import org.tron.api.GrpcAPI.OvkDecryptTRC20Parameters; -import org.tron.api.GrpcAPI.PaginatedMessage; -import org.tron.api.GrpcAPI.PricesResponseMessage; -import org.tron.api.GrpcAPI.SpendResult; -import org.tron.api.GrpcAPI.TransactionExtention; -import org.tron.api.GrpcAPI.WitnessList; -import org.tron.api.WalletSolidityGrpc.WalletSolidityImplBase; import org.tron.common.application.RpcService; import org.tron.core.config.args.Args; import org.tron.core.services.RpcApiService; -import org.tron.protos.Protocol.Account; -import org.tron.protos.Protocol.Block; -import org.tron.protos.Protocol.DelegatedResourceAccountIndex; -import org.tron.protos.Protocol.DynamicProperties; -import org.tron.protos.Protocol.Exchange; -import org.tron.protos.Protocol.MarketOrder; -import org.tron.protos.Protocol.MarketOrderList; -import org.tron.protos.Protocol.MarketOrderPair; -import org.tron.protos.Protocol.MarketOrderPairList; -import org.tron.protos.Protocol.MarketPriceList; -import org.tron.protos.Protocol.Transaction; -import org.tron.protos.Protocol.TransactionInfo; -import org.tron.protos.contract.AssetIssueContractOuterClass.AssetIssueContract; -import org.tron.protos.contract.ShieldContract.IncrementalMerkleVoucherInfo; -import org.tron.protos.contract.ShieldContract.OutputPointInfo; -import org.tron.protos.contract.SmartContractOuterClass.TriggerSmartContract; - +import org.tron.core.services.filter.PbftCursorInterceptor; @Slf4j(topic = "API") public class RpcApiServiceOnPBFT extends RpcService { @Autowired - private WalletOnPBFT walletOnPBFT; + private RpcApiService rpcApiService; @Autowired - private RpcApiService rpcApiService; + private PbftCursorInterceptor pbftCursorInterceptor; public RpcApiServiceOnPBFT() { port = Args.getInstance().getRpcOnPBFTPort(); @@ -69,427 +24,13 @@ public RpcApiServiceOnPBFT() { executorName = "rpc-pbft-executor"; } + /** PBFT cursor bound at the service level, so it brackets only these two read services. */ @Override protected void addService(NettyServerBuilder serverBuilder) { - serverBuilder.addService(new DatabaseApi()); - serverBuilder.addService(new WalletPBFTApi()); + serverBuilder.addService( + ServerInterceptors.intercept(rpcApiService.getDatabaseApi(), pbftCursorInterceptor)); + serverBuilder.addService(ServerInterceptors.intercept( + rpcApiService.getWalletSolidityApi(), pbftCursorInterceptor)); } - /** - * DatabaseApi. - */ - private class DatabaseApi extends DatabaseImplBase { - - @Override - public void getBlockReference(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getDatabaseApi().getBlockReference(request, responseObserver) - ); - } - - @Override - public void getNowBlock(EmptyMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getDatabaseApi().getNowBlock(request, responseObserver)); - } - - @Override - public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getDatabaseApi().getBlockByNum(request, responseObserver) - ); - } - - @Override - public void getDynamicProperties(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getDatabaseApi().getDynamicProperties(request, responseObserver) - ); - } - } - - /** - * WalletPBFTApi. - */ - private class WalletPBFTApi extends WalletSolidityImplBase { - - @Override - public void getAccount(Account request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAccount(request, responseObserver) - ); - } - - @Override - public void getAccountById(Account request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAccountById(request, responseObserver) - ); - } - - @Override - public void listWitnesses(EmptyMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().listWitnesses(request, responseObserver) - ); - } - - @Override - public void getAssetIssueById(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAssetIssueById(request, responseObserver) - ); - } - - @Override - public void getAssetIssueByName(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAssetIssueByName(request, responseObserver) - ); - } - - @Override - public void getAssetIssueList(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAssetIssueList(request, responseObserver) - ); - } - - @Override - public void getAssetIssueListByName(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getAssetIssueListByName(request, responseObserver) - ); - } - - @Override - public void getPaginatedAssetIssueList(PaginatedMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getPaginatedAssetIssueList(request, responseObserver) - ); - } - - @Override - public void getExchangeById(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getExchangeById( - request, responseObserver - ) - ); - } - - @Override - public void getNowBlock(EmptyMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getNowBlock(request, responseObserver) - ); - } - - @Override - public void getNowBlock2(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getNowBlock2(request, responseObserver) - ); - - } - - @Override - public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlockByNum(request, responseObserver) - ); - } - - @Override - public void getBlockByNum2(NumberMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlockByNum2(request, responseObserver) - ); - } - - @Override - public void getDelegatedResource(DelegatedResourceMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getDelegatedResource(request, responseObserver) - ); - } - - @Override - public void getDelegatedResourceV2(DelegatedResourceMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceV2(request, responseObserver) - ); - } - - @Override - public void getDelegatedResourceAccountIndex(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceAccountIndex(request, responseObserver) - ); - } - - @Override - public void getDelegatedResourceAccountIndexV2(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceAccountIndexV2(request, responseObserver) - ); - } - - @Override - public void getCanDelegatedMaxSize(CanDelegatedMaxSizeRequestMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getCanDelegatedMaxSize(request, responseObserver) - ); - } - - @Override - public void getAvailableUnfreezeCount(GetAvailableUnfreezeCountRequestMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getAvailableUnfreezeCount(request, responseObserver) - ); - } - - @Override - public void getCanWithdrawUnfreezeAmount(CanWithdrawUnfreezeAmountRequestMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getCanWithdrawUnfreezeAmount(request, responseObserver) - ); - } - - @Override - public void getTransactionCountByBlockNum(NumberMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getTransactionCountByBlockNum(request, responseObserver) - ); - } - - @Override - public void getTransactionById(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getTransactionById(request, responseObserver) - ); - - } - - @Override - public void getTransactionInfoById(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getTransactionInfoById(request, responseObserver) - ); - - } - - @Override - public void listExchanges(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().listExchanges(request, responseObserver) - ); - } - - @Override - public void triggerConstantContract(TriggerSmartContract request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .triggerConstantContract(request, responseObserver) - ); - } - - @Override - public void estimateEnergy(TriggerSmartContract request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .estimateEnergy(request, responseObserver) - ); - } - - @Override - public void getRewardInfo(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getRewardInfo(request, responseObserver) - ); - } - - @Override - public void getBrokerageInfo(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBrokerageInfo(request, responseObserver) - ); - } - - @Override - public void getMerkleTreeVoucherInfo(OutputPointInfo request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMerkleTreeVoucherInfo(request, responseObserver) - ); - } - - @Override - public void scanNoteByIvk(GrpcAPI.IvkDecryptParameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().scanNoteByIvk(request, responseObserver) - ); - } - - @Override - public void scanAndMarkNoteByIvk(GrpcAPI.IvkDecryptAndMarkParameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().scanAndMarkNoteByIvk(request, responseObserver) - ); - } - - @Override - public void scanNoteByOvk(GrpcAPI.OvkDecryptParameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().scanNoteByOvk(request, responseObserver) - ); - } - - @Override - public void isSpend(NoteParameters request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().isSpend(request, responseObserver) - ); - } - - @Override - public void getMarketOrderByAccount(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderByAccount(request, responseObserver) - ); - } - - @Override - public void getMarketOrderById(BytesMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderById(request, responseObserver) - ); - } - - @Override - public void getMarketPriceByPair(MarketOrderPair request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketPriceByPair(request, responseObserver) - ); - } - - @Override - public void getMarketOrderListByPair(MarketOrderPair request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderListByPair(request, responseObserver) - ); - } - - @Override - public void getMarketPairList(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketPairList(request, responseObserver) - ); - } - - @Override - public void scanShieldedTRC20NotesByIvk(IvkDecryptTRC20Parameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .scanShieldedTRC20NotesByIvk(request, responseObserver) - ); - } - - @Override - public void scanShieldedTRC20NotesByOvk(OvkDecryptTRC20Parameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .scanShieldedTRC20NotesByOvk(request, responseObserver) - ); - } - - @Override - public void isShieldedTRC20ContractNoteSpent(NfTRC20Parameters request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .isShieldedTRC20ContractNoteSpent(request, responseObserver) - ); - } - - @Override - public void getBurnTrx(EmptyMessage request, StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBurnTrx(request, responseObserver) - ); - } - - @Override - public void getBlock(GrpcAPI.BlockReq request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlock(request, responseObserver)); - } - - @Override - public void getBandwidthPrices(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBandwidthPrices(request, responseObserver)); - } - - @Override - public void getEnergyPrices(EmptyMessage request, - StreamObserver responseObserver) { - walletOnPBFT.futureGet( - () -> rpcApiService.getWalletSolidityApi().getEnergyPrices(request, responseObserver)); - } - - } } diff --git a/framework/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java b/framework/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java index 32c2268ba7f..f0c7b1468d2 100755 --- a/framework/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java +++ b/framework/src/main/java/org/tron/core/services/interfaceOnSolidity/RpcApiServiceOnSolidity.java @@ -1,69 +1,22 @@ package org.tron.core.services.interfaceOnSolidity; -import com.google.protobuf.ByteString; +import io.grpc.ServerInterceptors; import io.grpc.netty.NettyServerBuilder; -import io.grpc.stub.StreamObserver; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; -import org.tron.api.DatabaseGrpc.DatabaseImplBase; -import org.tron.api.GrpcAPI; -import org.tron.api.GrpcAPI.AssetIssueList; -import org.tron.api.GrpcAPI.BlockExtention; -import org.tron.api.GrpcAPI.BlockReference; -import org.tron.api.GrpcAPI.BytesMessage; -import org.tron.api.GrpcAPI.CanDelegatedMaxSizeRequestMessage; -import org.tron.api.GrpcAPI.CanDelegatedMaxSizeResponseMessage; -import org.tron.api.GrpcAPI.CanWithdrawUnfreezeAmountRequestMessage; -import org.tron.api.GrpcAPI.CanWithdrawUnfreezeAmountResponseMessage; -import org.tron.api.GrpcAPI.DelegatedResourceList; -import org.tron.api.GrpcAPI.DelegatedResourceMessage; -import org.tron.api.GrpcAPI.EmptyMessage; -import org.tron.api.GrpcAPI.ExchangeList; -import org.tron.api.GrpcAPI.GetAvailableUnfreezeCountRequestMessage; -import org.tron.api.GrpcAPI.GetAvailableUnfreezeCountResponseMessage; -import org.tron.api.GrpcAPI.NoteParameters; -import org.tron.api.GrpcAPI.NumberMessage; -import org.tron.api.GrpcAPI.PaginatedMessage; -import org.tron.api.GrpcAPI.PricesResponseMessage; -import org.tron.api.GrpcAPI.Return; -import org.tron.api.GrpcAPI.Return.response_code; -import org.tron.api.GrpcAPI.SpendResult; -import org.tron.api.GrpcAPI.TransactionExtention; -import org.tron.api.GrpcAPI.TransactionInfoList; -import org.tron.api.GrpcAPI.WitnessList; -import org.tron.api.WalletSolidityGrpc.WalletSolidityImplBase; import org.tron.common.application.RpcService; -import org.tron.common.utils.Sha256Hash; -import org.tron.core.capsule.BlockCapsule; import org.tron.core.config.args.Args; import org.tron.core.services.RpcApiService; -import org.tron.protos.Protocol.Account; -import org.tron.protos.Protocol.Block; -import org.tron.protos.Protocol.DelegatedResourceAccountIndex; -import org.tron.protos.Protocol.DynamicProperties; -import org.tron.protos.Protocol.Exchange; -import org.tron.protos.Protocol.MarketOrder; -import org.tron.protos.Protocol.MarketOrderList; -import org.tron.protos.Protocol.MarketOrderPair; -import org.tron.protos.Protocol.MarketOrderPairList; -import org.tron.protos.Protocol.MarketPriceList; -import org.tron.protos.Protocol.Transaction; -import org.tron.protos.Protocol.TransactionInfo; -import org.tron.protos.contract.AssetIssueContractOuterClass.AssetIssueContract; -import org.tron.protos.contract.ShieldContract.IncrementalMerkleVoucherInfo; -import org.tron.protos.contract.ShieldContract.OutputPointInfo; -import org.tron.protos.contract.SmartContractOuterClass.TriggerSmartContract; - +import org.tron.core.services.filter.SolidityCursorInterceptor; @Slf4j(topic = "API") public class RpcApiServiceOnSolidity extends RpcService { - @Autowired - private WalletOnSolidity walletOnSolidity; + private RpcApiService rpcApiService; @Autowired - private RpcApiService rpcApiService; + private SolidityCursorInterceptor solidityCursorInterceptor; public RpcApiServiceOnSolidity() { port = Args.getInstance().getRpcOnSolidityPort(); @@ -71,416 +24,13 @@ public RpcApiServiceOnSolidity() { executorName = "rpc-solidity-executor"; } + /** SOLIDITY cursor bound at the service level, so it brackets only these two read services. */ @Override protected void addService(NettyServerBuilder serverBuilder) { - serverBuilder.addService(new DatabaseApi()); - serverBuilder.addService(new WalletSolidityApi()); + serverBuilder.addService( + ServerInterceptors.intercept(rpcApiService.getDatabaseApi(), solidityCursorInterceptor)); + serverBuilder.addService(ServerInterceptors.intercept( + rpcApiService.getWalletSolidityApi(), solidityCursorInterceptor)); } - private TransactionExtention transaction2Extention(Transaction transaction) { - if (transaction == null) { - return null; - } - TransactionExtention.Builder trxExtBuilder = TransactionExtention.newBuilder(); - Return.Builder retBuilder = Return.newBuilder(); - trxExtBuilder.setTransaction(transaction); - trxExtBuilder.setTxid(Sha256Hash.of(transaction.getRawData().toByteArray()).getByteString()); - retBuilder.setResult(true).setCode(response_code.SUCCESS); - trxExtBuilder.setResult(retBuilder); - return trxExtBuilder.build(); - } - - private BlockExtention block2Extention(Block block) { - if (block == null) { - return null; - } - BlockExtention.Builder builder = BlockExtention.newBuilder(); - BlockCapsule blockCapsule = new BlockCapsule(block); - builder.setBlockHeader(block.getBlockHeader()); - builder.setBlockid(ByteString.copyFrom(blockCapsule.getBlockId().getBytes())); - for (int i = 0; i < block.getTransactionsCount(); i++) { - Transaction transaction = block.getTransactions(i); - builder.addTransactions(transaction2Extention(transaction)); - } - return builder.build(); - } - - /** - * DatabaseApi. - */ - private class DatabaseApi extends DatabaseImplBase { - - @Override - public void getBlockReference(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getDatabaseApi().getBlockReference(request, responseObserver)); - } - - @Override - public void getNowBlock(EmptyMessage request, StreamObserver responseObserver) { - walletOnSolidity - .futureGet(() -> rpcApiService.getDatabaseApi().getNowBlock(request, responseObserver)); - } - - @Override - public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - walletOnSolidity - .futureGet(() -> rpcApiService.getDatabaseApi().getBlockByNum(request, responseObserver)); - } - - @Override - public void getDynamicProperties(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getDatabaseApi().getDynamicProperties(request, responseObserver)); - } - } - - /** - * WalletSolidityApi. - */ - private class WalletSolidityApi extends WalletSolidityImplBase { - - @Override - public void getAccount(Account request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAccount(request, responseObserver)); - } - - @Override - public void getAccountById(Account request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAccountById(request, responseObserver)); - } - - @Override - public void listWitnesses(EmptyMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().listWitnesses(request, responseObserver)); - } - - public void getPaginatedNowWitnessList(PaginatedMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getPaginatedNowWitnessList(request, responseObserver)); - } - - @Override - public void getAssetIssueById(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAssetIssueById(request, responseObserver)); - } - - @Override - public void getAssetIssueByName(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getAssetIssueByName(request, responseObserver)); - } - - @Override - public void getAssetIssueList(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getAssetIssueList(request, responseObserver)); - } - - @Override - public void getAssetIssueListByName(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getAssetIssueListByName(request, responseObserver)); - } - - @Override - public void getPaginatedAssetIssueList(PaginatedMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getPaginatedAssetIssueList(request, responseObserver)); - } - - @Override - public void getExchangeById(BytesMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getExchangeById(request, responseObserver)); - } - - @Override - public void getNowBlock(EmptyMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getNowBlock(request, responseObserver)); - } - - @Override - public void getNowBlock2(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getNowBlock2(request, responseObserver)); - - } - - @Override - public void getBlockByNum(NumberMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlockByNum(request, responseObserver)); - } - - @Override - public void getBlockByNum2(NumberMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlockByNum2(request, responseObserver)); - } - - @Override - public void getDelegatedResource(DelegatedResourceMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getDelegatedResource(request, responseObserver)); - } - - @Override - public void getDelegatedResourceV2(DelegatedResourceMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceV2(request, responseObserver)); - } - - @Override - public void getDelegatedResourceAccountIndex(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceAccountIndex(request, responseObserver)); - } - - @Override - public void getDelegatedResourceAccountIndexV2(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getDelegatedResourceAccountIndexV2(request, responseObserver)); - } - - @Override - public void getCanDelegatedMaxSize(CanDelegatedMaxSizeRequestMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getCanDelegatedMaxSize(request, responseObserver)); - } - - @Override - public void getAvailableUnfreezeCount(GetAvailableUnfreezeCountRequestMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getAvailableUnfreezeCount(request, responseObserver)); - } - - @Override - public void getCanWithdrawUnfreezeAmount(CanWithdrawUnfreezeAmountRequestMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getCanWithdrawUnfreezeAmount(request, responseObserver)); - } - - @Override - public void getTransactionCountByBlockNum(NumberMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getTransactionCountByBlockNum(request, responseObserver)); - } - - @Override - public void getTransactionById(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getTransactionById(request, responseObserver)); - - } - - @Override - public void getTransactionInfoById(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getTransactionInfoById(request, responseObserver)); - - } - - @Override - public void listExchanges(EmptyMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().listExchanges(request, responseObserver)); - } - - @Override - public void triggerConstantContract(TriggerSmartContract request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .triggerConstantContract(request, responseObserver)); - } - - @Override - public void estimateEnergy(TriggerSmartContract request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .estimateEnergy(request, responseObserver)); - } - - @Override - public void getRewardInfo(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getRewardInfo(request, responseObserver)); - } - - @Override - public void getBrokerageInfo(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBrokerageInfo(request, responseObserver)); - } - - @Override - public void getMerkleTreeVoucherInfo(OutputPointInfo request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getMerkleTreeVoucherInfo(request, responseObserver)); - } - - @Override - public void scanNoteByIvk(GrpcAPI.IvkDecryptParameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().scanNoteByIvk(request, responseObserver)); - } - - @Override - public void scanAndMarkNoteByIvk(GrpcAPI.IvkDecryptAndMarkParameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .scanAndMarkNoteByIvk(request, responseObserver)); - } - - @Override - public void scanNoteByOvk(GrpcAPI.OvkDecryptParameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().scanNoteByOvk(request, responseObserver)); - } - - @Override - public void isSpend(NoteParameters request, StreamObserver responseObserver) { - walletOnSolidity - .futureGet(() -> rpcApiService.getWalletSolidityApi().isSpend(request, responseObserver)); - } - - @Override - public void getTransactionInfoByBlockNum(NumberMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet(() -> rpcApiService.getWalletSolidityApi() - .getTransactionInfoByBlockNum(request, responseObserver)); - } - - @Override - public void scanShieldedTRC20NotesByIvk(GrpcAPI.IvkDecryptTRC20Parameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .scanShieldedTRC20NotesByIvk(request, responseObserver) - ); - } - - @Override - public void scanShieldedTRC20NotesByOvk(GrpcAPI.OvkDecryptTRC20Parameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .scanShieldedTRC20NotesByOvk(request, responseObserver) - ); - } - - @Override - public void isShieldedTRC20ContractNoteSpent(GrpcAPI.NfTRC20Parameters request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .isShieldedTRC20ContractNoteSpent(request, responseObserver) - ); - } - - @Override - public void getMarketOrderByAccount(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderByAccount(request, responseObserver) - ); - } - - @Override - public void getMarketOrderById(BytesMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderById(request, responseObserver) - ); - } - - @Override - public void getMarketPriceByPair(MarketOrderPair request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketPriceByPair(request, responseObserver) - ); - } - - @Override - public void getMarketOrderListByPair(org.tron.protos.Protocol.MarketOrderPair request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketOrderListByPair(request, responseObserver) - ); - } - - @Override - public void getMarketPairList(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi() - .getMarketPairList(request, responseObserver) - ); - } - - @Override - public void getBurnTrx(EmptyMessage request, StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBurnTrx(request, responseObserver) - ); - } - - @Override - public void getBlock(GrpcAPI.BlockReq request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBlock(request, responseObserver)); - } - - @Override - public void getBandwidthPrices(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getBandwidthPrices(request, responseObserver)); - } - - @Override - public void getEnergyPrices(EmptyMessage request, - StreamObserver responseObserver) { - walletOnSolidity.futureGet( - () -> rpcApiService.getWalletSolidityApi().getEnergyPrices(request, responseObserver)); - } - - } } diff --git a/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java new file mode 100644 index 00000000000..16437f139e7 --- /dev/null +++ b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java @@ -0,0 +1,140 @@ +package org.tron.core.services; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.withSettings; + +import com.google.protobuf.Message; +import io.grpc.stub.StreamObserver; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.List; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Answers; +import org.mockito.stubbing.Answer; +import org.tron.core.Wallet; +import org.tron.core.metrics.MetricsApiService; +import org.tron.core.services.RpcApiService.WalletApi; +import org.tron.core.services.RpcApiService.WalletSolidityApi; +import org.tron.core.utils.TransactionUtil; + +/** + * Pins the one-terminal-event rule on the gRPC error path: a handler that reports a failure through + * {@code onError} must not fall through to {@code onCompleted}. gRPC rejects the second close with + * {@code IllegalStateException("call already closed")}, so a handler doing both costs a server-side + * exception on every failed call while the client sees nothing extra. + * + *

The rule is checked for every handler rather than for the ones that were fixed, because the + * shape is trivially reintroduced by copying a neighbouring handler. + */ +public class RpcApiServiceErrorPathTest { + + /** Minimum handlers that must actually fail, so the sweep cannot silently cover nothing. */ + private static final int MIN_EXERCISED = 20; + + @Test + public void testWalletApiTerminatesTheCallOnce() throws Exception { + assertSingleTerminalEvent(WalletApi.class); + } + + @Test + public void testWalletSolidityApiTerminatesTheCallOnce() throws Exception { + assertSingleTerminalEvent(WalletSolidityApi.class); + } + + /** + * Drives every unary handler of the given service class with collaborators that throw, and + * asserts none of them terminates the call more than once. + */ + private static void assertSingleTerminalEvent(Class apiClass) throws Exception { + RpcApiService service = mock(RpcApiService.class, + withSettings().defaultAnswer(Answers.CALLS_REAL_METHODS)); + injectThrowingCollaborators(service); + Object api = apiClass.getDeclaredConstructor(RpcApiService.class).newInstance(service); + + int exercised = 0; + for (Method method : apiClass.getDeclaredMethods()) { + if (!isUnaryHandler(method)) { + continue; + } + Message request = (Message) method.getParameterTypes()[0] + .getMethod("getDefaultInstance").invoke(null); + TerminalRecorder recorder = new TerminalRecorder(); + try { + method.invoke(api, request, recorder); + } catch (InvocationTargetException e) { + // a handler that lets the failure escape cannot have closed the call twice + continue; + } + Assert.assertTrue( + method.getName() + " terminated the call " + recorder.events.size() + " times " + + recorder.events + "; onError must be followed by return", + recorder.events.size() <= 1); + if (!recorder.events.isEmpty()) { + exercised++; + } + } + Assert.assertTrue( + apiClass.getSimpleName() + " exercised only " + exercised + " handlers, expected at least " + + MIN_EXERCISED + " — the sweep is no longer reaching the handler bodies", + exercised >= MIN_EXERCISED); + } + + private static boolean isUnaryHandler(Method method) { + Class[] params = method.getParameterTypes(); + return Modifier.isPublic(method.getModifiers()) + && method.getReturnType() == void.class + && params.length == 2 + && Message.class.isAssignableFrom(params[0]) + && params[1] == StreamObserver.class; + } + + /** + * Replaces the service's collaborators with mocks that throw on every call, so each handler takes + * its own error path, and binds a real {@code WalletApi} for the solidity handlers to delegate + * to. + */ + private static void injectThrowingCollaborators(RpcApiService service) throws Exception { + Answer throwing = invocation -> { + throw new RuntimeException("collaborator unavailable"); + }; + set(service, "wallet", mock(Wallet.class, withSettings().defaultAnswer(throwing))); + set(service, "transactionUtil", + mock(TransactionUtil.class, withSettings().defaultAnswer(throwing))); + set(service, "nodeInfoService", + mock(NodeInfoService.class, withSettings().defaultAnswer(throwing))); + set(service, "metricsApiService", + mock(MetricsApiService.class, withSettings().defaultAnswer(throwing))); + set(service, "walletApi", + WalletApi.class.getDeclaredConstructor(RpcApiService.class).newInstance(service)); + } + + private static void set(RpcApiService service, String name, Object value) throws Exception { + Field field = RpcApiService.class.getDeclaredField(name); + field.setAccessible(true); + field.set(service, value); + } + + /** Counts terminal events instead of closing a real call. */ + private static final class TerminalRecorder implements StreamObserver { + + private final List events = new ArrayList<>(); + + @Override + public void onNext(Object value) { + } + + @Override + public void onError(Throwable t) { + events.add("onError"); + } + + @Override + public void onCompleted() { + events.add("onCompleted"); + } + } +} diff --git a/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java b/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java index 6d20f72da19..88aa3b367cd 100644 --- a/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java +++ b/framework/src/test/java/org/tron/core/services/RpcApiServicesTest.java @@ -277,6 +277,7 @@ public void testGetPaginatedNowWitnessList() { .setOffset(0).setLimit(5).build(); assertNotNull(blockingStubFull.getPaginatedNowWitnessList(paginatedMessage)); assertNotNull(blockingStubSolidity.getPaginatedNowWitnessList(paginatedMessage)); + assertNotNull(blockingStubPBFT.getPaginatedNowWitnessList(paginatedMessage)); } @Test @@ -673,6 +674,7 @@ public void testGetTransactionInfoByBlockNum() { NumberMessage message = NumberMessage.newBuilder().setNum(1).build(); assertNotNull(blockingStubFull.getTransactionInfoByBlockNum(message)); assertNotNull(blockingStubSolidity.getTransactionInfoByBlockNum(message)); + assertNotNull(blockingStubPBFT.getTransactionInfoByBlockNum(message)); } @Test diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java new file mode 100644 index 00000000000..54e82426c0c --- /dev/null +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java @@ -0,0 +1,136 @@ +package org.tron.core.services.filter; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; + +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.ServerCall; +import io.grpc.ServerCallHandler; +import java.lang.reflect.Field; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.core.db.Manager; +import org.tron.core.db2.core.Chainbase; + +/** + * Pins the cursor's scope without depending on thread-pool scheduling. + * + *

{@code interceptCall()} is driven on thread A and the returned listener's + * {@code onHalfClose()} on a different thread B, which is what gRPC's {@code SerializingExecutor} + * is free to do. An implementation that scoped the cursor around {@code interceptCall} instead of + * {@code onHalfClose} then fails here by construction rather than by luck. + */ +public class CursorInterceptorScopeTest { + + private ExecutorService threadA; + private ExecutorService threadB; + + private Manager manager; + private Chainbase.Cursor cursorDuringHandler; + private Chainbase.Cursor cursorAfterCall; + private Chainbase.Cursor current; + + @Before + public void setUp() { + threadA = Executors.newSingleThreadExecutor(r -> new Thread(r, "cursor-thread-A")); + threadB = Executors.newSingleThreadExecutor(r -> new Thread(r, "cursor-thread-B")); + + // a Manager whose cursor state is observable, standing in for the ThreadLocal in Chainbase + current = Chainbase.Cursor.HEAD; + manager = mock(Manager.class); + doAnswer(inv -> current = inv.getArgument(0)) + .when(manager).setCursor(any(Chainbase.Cursor.class)); + doAnswer(inv -> current = Chainbase.Cursor.HEAD).when(manager).resetCursor(); + } + + @After + public void tearDown() throws Exception { + threadA.shutdownNow(); + threadB.shutdownNow(); + threadA.awaitTermination(5, TimeUnit.SECONDS); + threadB.awaitTermination(5, TimeUnit.SECONDS); + } + + @Test + public void testHandlerSeesTheCursorWhenInterceptCallRanOnAnotherThread() throws Exception { + ServerCall.Listener listener = startCallOnThreadA(false); + + // the handler runs from onHalfClose, on a different thread than interceptCall + runOn(threadB, () -> { + listener.onHalfClose(); + return null; + }); + + Assert.assertEquals("handler must observe the SOLIDITY cursor", + Chainbase.Cursor.SOLIDITY, cursorDuringHandler); + Assert.assertEquals("cursor must be back at HEAD once the handler returns", + Chainbase.Cursor.HEAD, cursorAfterCall); + } + + @Test + public void testCursorIsRestoredOnThreadBWhenTheHandlerThrows() throws Exception { + ServerCall.Listener listener = startCallOnThreadA(true); + + try { + runOn(threadB, () -> { + listener.onHalfClose(); + return null; + }); + Assert.fail("expected the handler failure to propagate"); + } catch (Exception expected) { + // what matters is the cursor state below + } + + Assert.assertEquals("a throwing handler must still leave the cursor at HEAD", + Chainbase.Cursor.HEAD, current); + } + + /** Runs interceptCall on thread A and returns the listener, with a handler that records state. */ + private ServerCall.Listener startCallOnThreadA(boolean handlerThrows) throws Exception { + SolidityCursorInterceptor interceptor = new SolidityCursorInterceptor(); + Field dbManager = CursorServerInterceptor.class.getDeclaredField("dbManager"); + dbManager.setAccessible(true); + dbManager.set(interceptor, manager); + + @SuppressWarnings("unchecked") + ServerCall call = mock(ServerCall.class); + @SuppressWarnings("unchecked") + MethodDescriptor descriptor = mock(MethodDescriptor.class); + doAnswer(inv -> descriptor).when(call).getMethodDescriptor(); + + ServerCallHandler handler = (c, h) -> new ServerCall.Listener() { + @Override + public void onHalfClose() { + cursorDuringHandler = current; + if (handlerThrows) { + throw new IllegalStateException("boom"); + } + } + }; + + return runOn(threadA, () -> { + ServerCall.Listener l = interceptor.interceptCall(call, new Metadata(), handler); + cursorAfterCall = current; + return l; + }); + } + + private static T runOn(ExecutorService executor, Callable task) throws Exception { + try { + return executor.submit(task).get(5, TimeUnit.SECONDS); + } catch (java.util.concurrent.ExecutionException e) { + if (e.getCause() instanceof Exception) { + throw (Exception) e.getCause(); + } + throw e; + } + } +} diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java new file mode 100644 index 00000000000..57babc52051 --- /dev/null +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java @@ -0,0 +1,123 @@ +package org.tron.core.services.filter; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Server; +import io.grpc.ServerBuilder; +import io.grpc.stub.StreamObserver; +import java.lang.reflect.Field; +import java.net.ServerSocket; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.tron.api.DatabaseGrpc; +import org.tron.api.DatabaseGrpc.DatabaseImplBase; +import org.tron.api.GrpcAPI.EmptyMessage; +import org.tron.core.db.Manager; +import org.tron.core.db2.core.Chainbase; +import org.tron.protos.Protocol.Block; + +/** + * Drives the production interceptor through a real gRPC server, which is the only place the + * assumption it rests on can be checked: that gRPC runs the handler inline from + * {@code onHalfClose}, on the same thread. The cursor is a {@link ThreadLocal}, so if that stops + * holding the cursor never reaches the read path and the port serves HEAD data with no error — + * responses stay well-formed, so nothing else notices. + * + *

CursorInterceptorScopeTest covers the interceptor's own logic on a synthetic harness; this is + * the end-to-end half. + */ +public class CursorInterceptorServerTest { + + private ExecutorService executor; + + @Before + public void setUp() { + // a fixed thread pool mirrors the production server configuration + executor = Executors.newFixedThreadPool(2, r -> { + Thread thread = new Thread(r); + thread.setName("cursor-rpc-executor-" + thread.getId()); + return thread; + }); + } + + @After + public void tearDown() throws Exception { + if (executor != null) { + executor.shutdown(); + if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { + executor.shutdownNow(); + } + } + } + + @Test + public void testCursorIsSetAndRestoredOnTheHandlerThread() throws Exception { + final List setOn = new CopyOnWriteArrayList<>(); + final List resetOn = new CopyOnWriteArrayList<>(); + final String[] handlerOn = new String[1]; + + Manager manager = mock(Manager.class); + doAnswer(inv -> setOn.add(Thread.currentThread().getName())) + .when(manager).setCursor(any(Chainbase.Cursor.class)); + doAnswer(inv -> resetOn.add(Thread.currentThread().getName())) + .when(manager).resetCursor(); + + SolidityCursorInterceptor interceptor = new SolidityCursorInterceptor(); + Field dbManager = CursorServerInterceptor.class.getDeclaredField("dbManager"); + dbManager.setAccessible(true); + dbManager.set(interceptor, manager); + + int port = freePort(); + Server server = ServerBuilder.forPort(port) + .executor(executor) + .addService(new DatabaseImplBase() { + @Override + public void getNowBlock(EmptyMessage request, StreamObserver observer) { + handlerOn[0] = Thread.currentThread().getName(); + observer.onNext(Block.getDefaultInstance()); + observer.onCompleted(); + } + }) + .intercept(interceptor) + .build() + .start(); + + ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", port) + .usePlaintext().directExecutor().build(); + try { + DatabaseGrpc.newBlockingStub(channel).getNowBlock(EmptyMessage.getDefaultInstance()); + + Assert.assertEquals("cursor must be set exactly once per call", 1, setOn.size()); + Assert.assertEquals("cursor must be restored exactly once per call", 1, resetOn.size()); + Assert.assertNotNull("handler did not run", handlerOn[0]); + // the ThreadLocal cursor only reaches the read path if it is set on the handler's thread + Assert.assertEquals("cursor was set on a thread other than the handler's", + handlerOn[0], setOn.get(0)); + Assert.assertEquals("cursor was restored on a thread other than the handler's", + handlerOn[0], resetOn.get(0)); + verify(manager).setCursor(Chainbase.Cursor.SOLIDITY); + } finally { + channel.shutdownNow(); + server.shutdownNow(); + server.awaitTermination(5, TimeUnit.SECONDS); + } + } + + private static int freePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } +} diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java new file mode 100644 index 00000000000..5f403b9d742 --- /dev/null +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java @@ -0,0 +1,101 @@ +package org.tron.core.services.filter; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.withSettings; + +import io.grpc.BindableService; +import io.grpc.ServerServiceDefinition; +import io.grpc.netty.NettyServerBuilder; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Answers; +import org.mockito.ArgumentCaptor; +import org.tron.api.DatabaseGrpc; +import org.tron.api.WalletSolidityGrpc; +import org.tron.core.services.RpcApiService; +import org.tron.core.services.interfaceOnPBFT.RpcApiServiceOnPBFT; +import org.tron.core.services.interfaceOnSolidity.RpcApiServiceOnSolidity; + +/** + * Guards that each cursor gRPC service registers the shared read services through its + * cursor interceptor. Nothing else catches a dropped interceptor: the services would still be + * served and every response would still look well-formed, only resolved against HEAD instead of the + * solidified or PBFT snapshot. This is the gRPC counterpart of CursorFilterInstallationTest. + * + *

Registering without the interceptor binds the {@code addService(BindableService)} overload + * rather than the {@code addService(ServerServiceDefinition)} one, so the two are distinguishable + * here. What the interceptor does once attached is pinned by CursorInterceptorScopeTest and + * CursorInterceptorServerTest. + */ +public class CursorInterceptorWiringTest { + + private static final Set SHARED_READ_SERVICES = new HashSet<>( + Arrays.asList(DatabaseGrpc.SERVICE_NAME, WalletSolidityGrpc.SERVICE_NAME)); + + @Test + public void testSolidityServiceRegistersBothReadServicesThroughTheCursor() throws Exception { + Assert.assertEquals(SHARED_READ_SERVICES, + interceptedServices(RpcApiServiceOnSolidity.class, new SolidityCursorInterceptor())); + } + + @Test + public void testPbftServiceRegistersBothReadServicesThroughTheCursor() throws Exception { + Assert.assertEquals(SHARED_READ_SERVICES, + interceptedServices(RpcApiServiceOnPBFT.class, new PbftCursorInterceptor())); + } + + /** + * Runs the service's real addService against a mock builder and returns the names of the services + * it registered as intercepted definitions, failing if any was registered unintercepted. + */ + private static Set interceptedServices(Class serviceClass, + CursorServerInterceptor interceptor) throws Exception { + RpcApiService rpcApiService = mock(RpcApiService.class); + given(rpcApiService.getDatabaseApi()).willReturn(RpcApiService.DatabaseApi.class + .getDeclaredConstructor(RpcApiService.class).newInstance(rpcApiService)); + given(rpcApiService.getWalletSolidityApi()).willReturn(RpcApiService.WalletSolidityApi.class + .getDeclaredConstructor(RpcApiService.class).newInstance(rpcApiService)); + + Object service = mock(serviceClass, withSettings().defaultAnswer(Answers.CALLS_REAL_METHODS)); + inject(serviceClass, service, rpcApiService); + inject(serviceClass, service, interceptor); + + NettyServerBuilder builder = mock(NettyServerBuilder.class); + Method addService = serviceClass.getDeclaredMethod("addService", NettyServerBuilder.class); + addService.setAccessible(true); + addService.invoke(service, builder); + + verify(builder, never()).addService(any(BindableService.class)); + ArgumentCaptor registered = + ArgumentCaptor.forClass(ServerServiceDefinition.class); + verify(builder, times(2)).addService(registered.capture()); + + Set names = new HashSet<>(); + for (ServerServiceDefinition definition : registered.getAllValues()) { + names.add(definition.getServiceDescriptor().getName()); + } + return names; + } + + /** Sets the one declared field the value fits; the two injected types are unrelated. */ + private static void inject(Class serviceClass, Object service, Object value) throws Exception { + for (Field field : serviceClass.getDeclaredFields()) { + if (field.getType().isInstance(value)) { + field.setAccessible(true); + field.set(service, value); + return; + } + } + Assert.fail(serviceClass.getSimpleName() + " has no field for " + value.getClass().getName()); + } +} From ebc664a61cf586158529b497c9f389fc6889ea44 Mon Sep 17 00:00:00 2001 From: GrapeS Date: Wed, 30 Sep 2026 12:41:32 +0800 Subject: [PATCH 2/5] test(rpc): check the cursor reset on the handler thread and wait for it - CursorInterceptorScopeTest read the cursor on thread A right after interceptCall, before thread B ran onHalfClose, so it passed even when the cursor was never reset. Model the cursor as a ThreadLocal, as Chainbase does, and read it on thread B after onHalfClose returns. - CursorInterceptorServerTest asserted the reset as soon as the blocking call returned, but the client can get the response before the server leaves onHalfClose's finally block. Wait for the reset on a latch. --- .../filter/CursorInterceptorScopeTest.java | 37 ++++++++++++------- .../filter/CursorInterceptorServerTest.java | 12 +++++- 2 files changed, 33 insertions(+), 16 deletions(-) diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java index 54e82426c0c..d1e01693fd6 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java @@ -35,20 +35,26 @@ public class CursorInterceptorScopeTest { private Manager manager; private Chainbase.Cursor cursorDuringHandler; - private Chainbase.Cursor cursorAfterCall; - private Chainbase.Cursor current; + private Chainbase.Cursor cursorOnThreadAAfterInterceptCall; + + /** Per-thread cursor, like the ThreadLocal in Chainbase; HEAD when nothing is set. */ + private final ThreadLocal cursor = + ThreadLocal.withInitial(() -> Chainbase.Cursor.HEAD); @Before public void setUp() { threadA = Executors.newSingleThreadExecutor(r -> new Thread(r, "cursor-thread-A")); threadB = Executors.newSingleThreadExecutor(r -> new Thread(r, "cursor-thread-B")); - // a Manager whose cursor state is observable, standing in for the ThreadLocal in Chainbase - current = Chainbase.Cursor.HEAD; manager = mock(Manager.class); - doAnswer(inv -> current = inv.getArgument(0)) - .when(manager).setCursor(any(Chainbase.Cursor.class)); - doAnswer(inv -> current = Chainbase.Cursor.HEAD).when(manager).resetCursor(); + doAnswer(inv -> { + cursor.set(inv.getArgument(0)); + return null; + }).when(manager).setCursor(any(Chainbase.Cursor.class)); + doAnswer(inv -> { + cursor.set(Chainbase.Cursor.HEAD); + return null; + }).when(manager).resetCursor(); } @After @@ -64,15 +70,17 @@ public void testHandlerSeesTheCursorWhenInterceptCallRanOnAnotherThread() throws ServerCall.Listener listener = startCallOnThreadA(false); // the handler runs from onHalfClose, on a different thread than interceptCall - runOn(threadB, () -> { + Chainbase.Cursor cursorOnThreadBAfterHalfClose = runOn(threadB, () -> { listener.onHalfClose(); - return null; + return cursor.get(); }); + Assert.assertEquals("interceptCall must not select the cursor on its own thread", + Chainbase.Cursor.HEAD, cursorOnThreadAAfterInterceptCall); Assert.assertEquals("handler must observe the SOLIDITY cursor", Chainbase.Cursor.SOLIDITY, cursorDuringHandler); - Assert.assertEquals("cursor must be back at HEAD once the handler returns", - Chainbase.Cursor.HEAD, cursorAfterCall); + Assert.assertEquals("cursor must be back at HEAD on the handler thread once it returns", + Chainbase.Cursor.HEAD, cursorOnThreadBAfterHalfClose); } @Test @@ -89,8 +97,9 @@ public void testCursorIsRestoredOnThreadBWhenTheHandlerThrows() throws Exception // what matters is the cursor state below } + // threadB is a single thread, so this reads the cursor of the thread the handler threw on Assert.assertEquals("a throwing handler must still leave the cursor at HEAD", - Chainbase.Cursor.HEAD, current); + Chainbase.Cursor.HEAD, runOn(threadB, cursor::get)); } /** Runs interceptCall on thread A and returns the listener, with a handler that records state. */ @@ -109,7 +118,7 @@ private ServerCall.Listener startCallOnThreadA(boolean handlerThrows) th ServerCallHandler handler = (c, h) -> new ServerCall.Listener() { @Override public void onHalfClose() { - cursorDuringHandler = current; + cursorDuringHandler = cursor.get(); if (handlerThrows) { throw new IllegalStateException("boom"); } @@ -118,7 +127,7 @@ public void onHalfClose() { return runOn(threadA, () -> { ServerCall.Listener l = interceptor.interceptCall(call, new Metadata(), handler); - cursorAfterCall = current; + cursorOnThreadAAfterInterceptCall = cursor.get(); return l; }); } diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java index 57babc52051..5e58e3d7bcc 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java @@ -14,6 +14,7 @@ import java.net.ServerSocket; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -67,12 +68,17 @@ public void testCursorIsSetAndRestoredOnTheHandlerThread() throws Exception { final List setOn = new CopyOnWriteArrayList<>(); final List resetOn = new CopyOnWriteArrayList<>(); final String[] handlerOn = new String[1]; + // the client can receive the response before the server leaves onHalfClose's finally block + final CountDownLatch resetDone = new CountDownLatch(1); Manager manager = mock(Manager.class); doAnswer(inv -> setOn.add(Thread.currentThread().getName())) .when(manager).setCursor(any(Chainbase.Cursor.class)); - doAnswer(inv -> resetOn.add(Thread.currentThread().getName())) - .when(manager).resetCursor(); + doAnswer(inv -> { + resetOn.add(Thread.currentThread().getName()); + resetDone.countDown(); + return null; + }).when(manager).resetCursor(); SolidityCursorInterceptor interceptor = new SolidityCursorInterceptor(); Field dbManager = CursorServerInterceptor.class.getDeclaredField("dbManager"); @@ -99,6 +105,8 @@ public void getNowBlock(EmptyMessage request, StreamObserver observer) { try { DatabaseGrpc.newBlockingStub(channel).getNowBlock(EmptyMessage.getDefaultInstance()); + Assert.assertTrue("cursor was not restored after the call", + resetDone.await(5, TimeUnit.SECONDS)); Assert.assertEquals("cursor must be set exactly once per call", 1, setOn.size()); Assert.assertEquals("cursor must be restored exactly once per call", 1, resetOn.size()); Assert.assertNotNull("handler did not run", handlerOn[0]); From 240f094021d57f56d9b3760a2dcbcb8a0123e0a3 Mon Sep 17 00:00:00 2001 From: GrapeS Date: Mon, 12 Oct 2026 14:39:37 +0800 Subject: [PATCH 3/5] refactor(rpc): fix the cursor per interceptor and document port parity --- .../java/org/tron/core/services/RpcApiService.java | 7 +++++-- .../services/filter/CursorServerInterceptor.java | 14 ++++++++++++-- .../services/filter/PbftCursorInterceptor.java | 2 +- .../services/filter/SolidityCursorInterceptor.java | 2 +- 4 files changed, 19 insertions(+), 6 deletions(-) diff --git a/framework/src/main/java/org/tron/core/services/RpcApiService.java b/framework/src/main/java/org/tron/core/services/RpcApiService.java index d3ba4afda0e..22b6dde3259 100755 --- a/framework/src/main/java/org/tron/core/services/RpcApiService.java +++ b/framework/src/main/java/org/tron/core/services/RpcApiService.java @@ -358,8 +358,11 @@ public void getDynamicProperties(EmptyMessage request, /** * WalletSolidityApi is the full implementation of the {@code protocol.WalletSolidity} gRPC - * service. Every method here is read-only and also present on {@link WalletApi}, so each one - * delegates to the shared {@code WalletApi} singleton instead of repeating its body. + * service. Every method here is read-only and also present on {@link WalletApi}, so none repeats + * its body: most delegate to the shared {@code WalletApi} singleton, and six + * ({@code getTransactionCountByBlockNum}, {@code getRewardInfo}, {@code getBrokerageInfo}, + * {@code getBurnTrx}, {@code triggerConstantContract}, {@code getBlock}) call the same + * {@code *Common} / {@code callContract} helpers {@code WalletApi} uses. */ public class WalletSolidityApi extends WalletSolidityImplBase { diff --git a/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java index 74d14daede9..d7d56b06f29 100644 --- a/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java +++ b/framework/src/main/java/org/tron/core/services/filter/CursorServerInterceptor.java @@ -25,14 +25,24 @@ * *

The {@code finally} reset is mandatory: the fixed thread pool is reused, so a leftover cursor * leaks into the next call on that thread. + * + *

The SOLIDITY and PBFT ports register the shared {@code protocol.Database} and + * {@code protocol.WalletSolidity} services through this interceptor, so their method sets are + * defined by those proto services alone: a method added to {@code protocol.Database} is served on + * the HEAD, SOLIDITY and PBFT ports, and one added to {@code protocol.WalletSolidity} on both the + * SOLIDITY and PBFT ports. Keeping a method off one of these ports must be implemented explicitly. */ public abstract class CursorServerInterceptor implements ServerInterceptor { @Autowired protected Manager dbManager; - /** Snapshot every call on this server reads from; set by the subclass. */ - protected Chainbase.Cursor cursor; + /** Snapshot every call on this server reads from; fixed for the interceptor's lifetime. */ + private final Chainbase.Cursor cursor; + + protected CursorServerInterceptor(Chainbase.Cursor cursor) { + this.cursor = cursor; + } @Override public ServerCall.Listener interceptCall( diff --git a/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java index 3ea55501d35..187c37272f6 100644 --- a/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java +++ b/framework/src/main/java/org/tron/core/services/filter/PbftCursorInterceptor.java @@ -10,6 +10,6 @@ public class PbftCursorInterceptor extends CursorServerInterceptor { public PbftCursorInterceptor() { - this.cursor = Chainbase.Cursor.PBFT; + super(Chainbase.Cursor.PBFT); } } diff --git a/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java b/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java index 5cf95ca0db4..20ed2b02a6f 100644 --- a/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java +++ b/framework/src/main/java/org/tron/core/services/filter/SolidityCursorInterceptor.java @@ -10,6 +10,6 @@ public class SolidityCursorInterceptor extends CursorServerInterceptor { public SolidityCursorInterceptor() { - this.cursor = Chainbase.Cursor.SOLIDITY; + super(Chainbase.Cursor.SOLIDITY); } } From d280440a138ac636f3a614a3f8e3a067a8ce0cf9 Mon Sep 17 00:00:00 2001 From: GrapeS Date: Mon, 12 Oct 2026 14:39:37 +0800 Subject: [PATCH 4/5] test(rpc): drive registered services and checked error paths --- .../services/RpcApiServiceErrorPathTest.java | 169 +++++++++++++++--- .../filter/CursorInterceptorScopeTest.java | 3 + .../filter/CursorInterceptorServerTest.java | 16 +- .../filter/CursorInterceptorWiringTest.java | 161 ++++++++++++++--- 4 files changed, 290 insertions(+), 59 deletions(-) diff --git a/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java index 16437f139e7..b612236d108 100644 --- a/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java +++ b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java @@ -10,10 +10,12 @@ import java.lang.reflect.Method; import java.lang.reflect.Modifier; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.junit.Assert; import org.junit.Test; import org.mockito.Answers; +import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; import org.tron.core.Wallet; import org.tron.core.metrics.MetricsApiService; @@ -29,30 +31,64 @@ * *

The rule is checked for every handler rather than for the ones that were fixed, because the * shape is trivially reintroduced by copying a neighbouring handler. + * + *

Collaborators fail in two ways. An unchecked exception reaches the handlers' generic catch + * blocks. The checked exceptions a collaborator method declares reach the typed ones, such as the + * {@code MaintenanceUnavailableException} catch in {@code getPaginatedNowWitnessList}; a method can + * declare several, so the checked sweep runs once for each declaration position. A handler that + * lets the failure escape is checked as well: what it recorded before throwing may not contain a + * second terminal event either. + * + *

Nothing here is a singleton: each sweep builds its own service and mocks, so no {@code @After} + * reset is needed. */ public class RpcApiServiceErrorPathTest { /** Minimum handlers that must actually fail, so the sweep cannot silently cover nothing. */ private static final int MIN_EXERCISED = 20; + /** Minimum handlers whose typed catch must run in each checked sweep, for the same reason. */ + private static final int MIN_EXERCISED_CHECKED = 10; + + /** Collaborators the handlers call; the exceptions their methods declare drive the sweeps. */ + private static final List> COLLABORATORS = Arrays.asList( + Wallet.class, TransactionUtil.class, NodeInfoService.class, MetricsApiService.class); + @Test public void testWalletApiTerminatesTheCallOnce() throws Exception { - assertSingleTerminalEvent(WalletApi.class); + assertSingleTerminalEvent(WalletApi.class, Failure.unchecked(), MIN_EXERCISED); } @Test public void testWalletSolidityApiTerminatesTheCallOnce() throws Exception { - assertSingleTerminalEvent(WalletSolidityApi.class); + assertSingleTerminalEvent(WalletSolidityApi.class, Failure.unchecked(), MIN_EXERCISED); + } + + @Test + public void testWalletApiTerminatesTheCallOnceOnCheckedExceptions() throws Exception { + for (int position = 0; position < declarationPositions(); position++) { + assertSingleTerminalEvent(WalletApi.class, Failure.checked(position), + MIN_EXERCISED_CHECKED); + } + } + + @Test + public void testWalletSolidityApiTerminatesTheCallOnceOnCheckedExceptions() throws Exception { + for (int position = 0; position < declarationPositions(); position++) { + assertSingleTerminalEvent(WalletSolidityApi.class, Failure.checked(position), + MIN_EXERCISED_CHECKED); + } } /** - * Drives every unary handler of the given service class with collaborators that throw, and - * asserts none of them terminates the call more than once. + * Drives every unary handler of the given service class with collaborators that fail as + * {@code failure} says, and asserts none of them terminates the call more than once. */ - private static void assertSingleTerminalEvent(Class apiClass) throws Exception { + private static void assertSingleTerminalEvent(Class apiClass, Failure failure, + int minExercised) throws Exception { RpcApiService service = mock(RpcApiService.class, withSettings().defaultAnswer(Answers.CALLS_REAL_METHODS)); - injectThrowingCollaborators(service); + injectFailingCollaborators(service, failure); Object api = apiClass.getDeclaredConstructor(RpcApiService.class).newInstance(service); int exercised = 0; @@ -63,24 +99,26 @@ private static void assertSingleTerminalEvent(Class apiClass) throws Exceptio Message request = (Message) method.getParameterTypes()[0] .getMethod("getDefaultInstance").invoke(null); TerminalRecorder recorder = new TerminalRecorder(); + failure.threwChecked = false; + String escaped = ""; try { method.invoke(api, request, recorder); } catch (InvocationTargetException e) { - // a handler that lets the failure escape cannot have closed the call twice - continue; + escaped = " before letting " + e.getCause() + " escape"; } Assert.assertTrue( method.getName() + " terminated the call " + recorder.events.size() + " times " - + recorder.events + "; onError must be followed by return", + + recorder.events + escaped + " " + failure + "; onError must be followed by return", recorder.events.size() <= 1); - if (!recorder.events.isEmpty()) { + if (!recorder.events.isEmpty() && failure.reachedTheIntendedCatch()) { exercised++; } } Assert.assertTrue( - apiClass.getSimpleName() + " exercised only " + exercised + " handlers, expected at least " - + MIN_EXERCISED + " — the sweep is no longer reaching the handler bodies", - exercised >= MIN_EXERCISED); + apiClass.getSimpleName() + " exercised only " + exercised + " handlers " + failure + + ", expected at least " + minExercised + + " — the sweep is no longer reaching the handler bodies", + exercised >= minExercised); } private static boolean isUnaryHandler(Method method) { @@ -93,21 +131,19 @@ private static boolean isUnaryHandler(Method method) { } /** - * Replaces the service's collaborators with mocks that throw on every call, so each handler takes + * Replaces the service's collaborators with mocks that fail on every call, so each handler takes * its own error path, and binds a real {@code WalletApi} for the solidity handlers to delegate * to. */ - private static void injectThrowingCollaborators(RpcApiService service) throws Exception { - Answer throwing = invocation -> { - throw new RuntimeException("collaborator unavailable"); - }; - set(service, "wallet", mock(Wallet.class, withSettings().defaultAnswer(throwing))); + private static void injectFailingCollaborators(RpcApiService service, Failure failure) + throws Exception { + set(service, "wallet", mock(Wallet.class, withSettings().defaultAnswer(failure))); set(service, "transactionUtil", - mock(TransactionUtil.class, withSettings().defaultAnswer(throwing))); + mock(TransactionUtil.class, withSettings().defaultAnswer(failure))); set(service, "nodeInfoService", - mock(NodeInfoService.class, withSettings().defaultAnswer(throwing))); + mock(NodeInfoService.class, withSettings().defaultAnswer(failure))); set(service, "metricsApiService", - mock(MetricsApiService.class, withSettings().defaultAnswer(throwing))); + mock(MetricsApiService.class, withSettings().defaultAnswer(failure))); set(service, "walletApi", WalletApi.class.getDeclaredConstructor(RpcApiService.class).newInstance(service)); } @@ -118,6 +154,95 @@ private static void set(RpcApiService service, String name, Object value) throws field.set(service, value); } + /** Most checked exceptions any collaborator method declares, i.e. the checked sweeps to run. */ + private static int declarationPositions() { + int positions = 0; + for (Class collaborator : COLLABORATORS) { + for (Method method : collaborator.getMethods()) { + int declared = checkedExceptions(method).size(); + positions = declared > positions ? declared : positions; + } + } + Assert.assertTrue("no collaborator method declares a checked exception", positions > 0); + return positions; + } + + private static List> checkedExceptions(Method method) { + List> checked = new ArrayList<>(); + for (Class type : method.getExceptionTypes()) { + if (Exception.class.isAssignableFrom(type) + && !RuntimeException.class.isAssignableFrom(type)) { + checked.add(type); + } + } + return checked; + } + + /** Builds the exception through its message or no-arg constructor; null if it has neither. */ + private static Exception newException(Class type) { + try { + return (Exception) type.getConstructor(String.class).newInstance("collaborator unavailable"); + } catch (ReflectiveOperationException noMessageConstructor) { + try { + return (Exception) type.getConstructor().newInstance(); + } catch (ReflectiveOperationException noDefaultConstructor) { + return null; + } + } + } + + /** + * How every collaborator call fails in one sweep: with an unchecked exception, or with the + * checked exception the called method declares at a given position (its last one when it + * declares fewer). + */ + private static final class Failure implements Answer { + + /** Declaration position to throw, or -1 for an unchecked exception. */ + private final int position; + /** Whether the current handler's collaborator call threw a checked exception. */ + private boolean threwChecked; + + private Failure(int position) { + this.position = position; + } + + static Failure unchecked() { + return new Failure(-1); + } + + static Failure checked(int position) { + return new Failure(position); + } + + /** A checked sweep only counts handlers whose collaborator really threw a checked exception. */ + boolean reachedTheIntendedCatch() { + return position < 0 || threwChecked; + } + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + if (position >= 0) { + List> declared = checkedExceptions(invocation.getMethod()); + if (!declared.isEmpty()) { + Exception checked = newException( + declared.get(position < declared.size() ? position : declared.size() - 1)); + if (checked != null) { + threwChecked = true; + throw checked; + } + } + } + throw new RuntimeException("collaborator unavailable"); + } + + @Override + public String toString() { + return position < 0 ? "(unchecked failures)" + : "(checked failures, declaration position " + position + ")"; + } + } + /** Counts terminal events instead of closing a real call. */ private static final class TerminalRecorder implements StreamObserver { diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java index d1e01693fd6..237f68aa8ff 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java @@ -27,6 +27,9 @@ * {@code onHalfClose()} on a different thread B, which is what gRPC's {@code SerializingExecutor} * is free to do. An implementation that scoped the cursor around {@code interceptCall} instead of * {@code onHalfClose} then fails here by construction rather than by luck. + * + *

The interceptor under test is not a singleton: each test builds its own instance, so nothing + * needs resetting between tests; {@code tearDown} only stops this test's threads. */ public class CursorInterceptorScopeTest { diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java index 5e58e3d7bcc..d2aee4247eb 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java @@ -11,7 +11,6 @@ import io.grpc.ServerBuilder; import io.grpc.stub.StreamObserver; import java.lang.reflect.Field; -import java.net.ServerSocket; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; @@ -38,6 +37,9 @@ * *

CursorInterceptorScopeTest covers the interceptor's own logic on a synthetic harness; this is * the end-to-end half. + * + *

The interceptor under test is not a singleton: each test builds its own instance and server, + * so nothing needs resetting between tests; {@code tearDown} only stops this test's executor. */ public class CursorInterceptorServerTest { @@ -85,8 +87,8 @@ public void testCursorIsSetAndRestoredOnTheHandlerThread() throws Exception { dbManager.setAccessible(true); dbManager.set(interceptor, manager); - int port = freePort(); - Server server = ServerBuilder.forPort(port) + // port 0 lets the server bind a free port itself; reserving one first would race other tests + Server server = ServerBuilder.forPort(0) .executor(executor) .addService(new DatabaseImplBase() { @Override @@ -100,7 +102,7 @@ public void getNowBlock(EmptyMessage request, StreamObserver observer) { .build() .start(); - ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", port) + ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", server.getPort()) .usePlaintext().directExecutor().build(); try { DatabaseGrpc.newBlockingStub(channel).getNowBlock(EmptyMessage.getDefaultInstance()); @@ -122,10 +124,4 @@ public void getNowBlock(EmptyMessage request, StreamObserver observer) { server.awaitTermination(5, TimeUnit.SECONDS); } } - - private static int freePort() throws Exception { - try (ServerSocket socket = new ServerSocket(0)) { - return socket.getLocalPort(); - } - } } diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java index 5f403b9d742..af071541df7 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java @@ -2,6 +2,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -9,19 +10,30 @@ import static org.mockito.Mockito.withSettings; import io.grpc.BindableService; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.ServerCall; +import io.grpc.ServerMethodDefinition; import io.grpc.ServerServiceDefinition; import io.grpc.netty.NettyServerBuilder; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.ArrayList; import java.util.Arrays; -import java.util.HashSet; -import java.util.Set; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import org.junit.Assert; import org.junit.Test; import org.mockito.Answers; import org.mockito.ArgumentCaptor; +import org.mockito.stubbing.Answer; import org.tron.api.DatabaseGrpc; import org.tron.api.WalletSolidityGrpc; +import org.tron.core.ChainBaseManager; +import org.tron.core.Wallet; +import org.tron.core.db.Manager; +import org.tron.core.db2.core.Chainbase; import org.tron.core.services.RpcApiService; import org.tron.core.services.interfaceOnPBFT.RpcApiServiceOnPBFT; import org.tron.core.services.interfaceOnSolidity.RpcApiServiceOnSolidity; @@ -30,41 +42,95 @@ * Guards that each cursor gRPC service registers the shared read services through its * cursor interceptor. Nothing else catches a dropped interceptor: the services would still be * served and every response would still look well-formed, only resolved against HEAD instead of the - * solidified or PBFT snapshot. This is the gRPC counterpart of CursorFilterInstallationTest. + * solidified or PBFT snapshot. * - *

Registering without the interceptor binds the {@code addService(BindableService)} overload - * rather than the {@code addService(ServerServiceDefinition)} one, so the two are distinguishable - * here. What the interceptor does once attached is pinned by CursorInterceptorScopeTest and - * CursorInterceptorServerTest. + *

The production {@code addService} runs against a mock builder, and one unary call is driven + * through each definition it registers the way gRPC does: {@code startCall}, the request, then + * {@code onHalfClose}. The handler's first read is recorded alongside the cursor calls, so the + * port's cursor has to be selected before that read and reset after the handler returns. A + * registration that bypasses the interceptor, whether through the {@code BindableService} + * overload, a plain {@code bindService()} definition or a different interceptor, reaches the + * handler with no cursor selected and fails here. What the interceptor does on a real server is + * pinned by CursorInterceptorScopeTest and CursorInterceptorServerTest. + * + *

Nothing here is a singleton: each test builds its own service, interceptor and mocks, so no + * {@code @After} reset is needed. */ public class CursorInterceptorWiringTest { - private static final Set SHARED_READ_SERVICES = new HashSet<>( - Arrays.asList(DatabaseGrpc.SERVICE_NAME, WalletSolidityGrpc.SERVICE_NAME)); + /** The unary method driven on each shared read service; both read through a collaborator. */ + private static final Map PROBE_METHODS = new HashMap<>(); + + static { + PROBE_METHODS.put(DatabaseGrpc.SERVICE_NAME, + DatabaseGrpc.getGetNowBlockMethod().getFullMethodName()); + PROBE_METHODS.put(WalletSolidityGrpc.SERVICE_NAME, + WalletSolidityGrpc.getGetNowBlockMethod().getFullMethodName()); + } @Test - public void testSolidityServiceRegistersBothReadServicesThroughTheCursor() throws Exception { - Assert.assertEquals(SHARED_READ_SERVICES, - interceptedServices(RpcApiServiceOnSolidity.class, new SolidityCursorInterceptor())); + public void testSolidityServiceReadsBothServicesUnderTheSolidityCursor() throws Exception { + assertEveryServiceReadsUnder(RpcApiServiceOnSolidity.class, new SolidityCursorInterceptor(), + Chainbase.Cursor.SOLIDITY); } @Test - public void testPbftServiceRegistersBothReadServicesThroughTheCursor() throws Exception { - Assert.assertEquals(SHARED_READ_SERVICES, - interceptedServices(RpcApiServiceOnPBFT.class, new PbftCursorInterceptor())); + public void testPbftServiceReadsBothServicesUnderThePbftCursor() throws Exception { + assertEveryServiceReadsUnder(RpcApiServiceOnPBFT.class, new PbftCursorInterceptor(), + Chainbase.Cursor.PBFT); + } + + /** + * Registers the services through the production addService, drives one call through each + * registered definition, and asserts every handler read ran between the port's setCursor and + * resetCursor. + */ + private static void assertEveryServiceReadsUnder(Class serviceClass, + CursorServerInterceptor interceptor, Chainbase.Cursor cursor) throws Exception { + List events = new ArrayList<>(); + Manager manager = mock(Manager.class); + doAnswer(inv -> events.add("setCursor " + inv.getArgument(0))) + .when(manager).setCursor(any(Chainbase.Cursor.class)); + doAnswer(inv -> events.add("resetCursor")).when(manager).resetCursor(); + setField(CursorServerInterceptor.class, interceptor, "dbManager", manager); + + Map> eventsByService = new HashMap<>(); + for (ServerServiceDefinition definition + : register(serviceClass, interceptor, recordRead(events))) { + String service = definition.getServiceDescriptor().getName(); + String probe = PROBE_METHODS.get(service); + Assert.assertNotNull(serviceClass.getSimpleName() + " registered " + service, probe); + events.clear(); + driveUnaryCall(definition.getMethod(probe)); + eventsByService.put(service, new ArrayList<>(events)); + } + + List bracketed = Arrays.asList("setCursor " + cursor, "read", "resetCursor"); + Map> expected = new HashMap<>(); + for (String service : PROBE_METHODS.keySet()) { + expected.put(service, bracketed); + } + Assert.assertEquals(serviceClass.getSimpleName() + " must serve every shared read service" + + " under the " + cursor + " cursor", expected, eventsByService); } /** - * Runs the service's real addService against a mock builder and returns the names of the services - * it registered as intercepted definitions, failing if any was registered unintercepted. + * Runs the service's real addService against a mock builder and returns the definitions it + * registered, failing if any went through the {@code BindableService} overload. */ - private static Set interceptedServices(Class serviceClass, - CursorServerInterceptor interceptor) throws Exception { + private static List register(Class serviceClass, + CursorServerInterceptor interceptor, Answer collaborator) throws Exception { RpcApiService rpcApiService = mock(RpcApiService.class); - given(rpcApiService.getDatabaseApi()).willReturn(RpcApiService.DatabaseApi.class - .getDeclaredConstructor(RpcApiService.class).newInstance(rpcApiService)); - given(rpcApiService.getWalletSolidityApi()).willReturn(RpcApiService.WalletSolidityApi.class - .getDeclaredConstructor(RpcApiService.class).newInstance(rpcApiService)); + setField(RpcApiService.class, rpcApiService, "wallet", + mock(Wallet.class, withSettings().defaultAnswer(collaborator))); + setField(RpcApiService.class, rpcApiService, "chainBaseManager", + mock(ChainBaseManager.class, withSettings().defaultAnswer(collaborator))); + setField(RpcApiService.class, rpcApiService, "walletApi", + newInner(RpcApiService.WalletApi.class, rpcApiService)); + given(rpcApiService.getDatabaseApi()) + .willReturn(newInner(RpcApiService.DatabaseApi.class, rpcApiService)); + given(rpcApiService.getWalletSolidityApi()) + .willReturn(newInner(RpcApiService.WalletSolidityApi.class, rpcApiService)); Object service = mock(serviceClass, withSettings().defaultAnswer(Answers.CALLS_REAL_METHODS)); inject(serviceClass, service, rpcApiService); @@ -79,12 +145,45 @@ private static Set interceptedServices(Class serviceClass, ArgumentCaptor registered = ArgumentCaptor.forClass(ServerServiceDefinition.class); verify(builder, times(2)).addService(registered.capture()); + return registered.getAllValues(); + } - Set names = new HashSet<>(); - for (ServerServiceDefinition definition : registered.getAllValues()) { - names.add(definition.getServiceDescriptor().getName()); + /** Drives one unary call through the registered handler, the way gRPC delivers it. */ + @SuppressWarnings("unchecked") + private static void driveUnaryCall(ServerMethodDefinition method) { + ServerCall call = mock(ServerCall.class); + given(call.getMethodDescriptor()).willReturn(method.getMethodDescriptor()); + ServerCall.Listener listener = + method.getServerCallHandler().startCall(call, new Metadata()); + listener.onMessage(((MethodDescriptor.PrototypeMarshaller) + method.getMethodDescriptor().getRequestMarshaller()).getMessagePrototype()); + try { + listener.onHalfClose(); + } catch (StopRead expected) { + // the recording collaborator ends the handler at its first read } - return names; + } + + /** Records the handler's first read, then ends the handler there. */ + private static Answer recordRead(List events) { + return invocation -> { + if (invocation.getMethod().getDeclaringClass() == Object.class) { + return Answers.RETURNS_DEFAULTS.answer(invocation); + } + events.add("read"); + throw new StopRead(); + }; + } + + private static T newInner(Class innerClass, RpcApiService outer) throws Exception { + return innerClass.getDeclaredConstructor(RpcApiService.class).newInstance(outer); + } + + private static void setField(Class owner, Object target, String name, Object value) + throws Exception { + Field field = owner.getDeclaredField(name); + field.setAccessible(true); + field.set(target, value); } /** Sets the one declared field the value fits; the two injected types are unrelated. */ @@ -98,4 +197,12 @@ private static void inject(Class serviceClass, Object service, Object value) } Assert.fail(serviceClass.getSimpleName() + " has no field for " + value.getClass().getName()); } + + /** Thrown by the recording collaborator to stop a handler after its first read. */ + private static final class StopRead extends RuntimeException { + + StopRead() { + super("stop the handler after its first read"); + } + } } From 082d99b99b807e208851bc14795cc5ea73524031 Mon Sep 17 00:00:00 2001 From: GrapeS Date: Mon, 12 Oct 2026 16:12:40 +0800 Subject: [PATCH 5/5] test(rpc): drop redundant cursor and error-path test cases --- .../services/RpcApiServiceErrorPathTest.java | 92 ++++++------------- .../filter/CursorInterceptorScopeTest.java | 26 +----- .../filter/CursorInterceptorServerTest.java | 4 +- .../filter/CursorInterceptorWiringTest.java | 6 +- 4 files changed, 36 insertions(+), 92 deletions(-) diff --git a/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java index b612236d108..769456ec1e0 100644 --- a/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java +++ b/framework/src/test/java/org/tron/core/services/RpcApiServiceErrorPathTest.java @@ -10,7 +10,6 @@ import java.lang.reflect.Method; import java.lang.reflect.Modifier; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; import org.junit.Assert; import org.junit.Test; @@ -33,11 +32,12 @@ * shape is trivially reintroduced by copying a neighbouring handler. * *

Collaborators fail in two ways. An unchecked exception reaches the handlers' generic catch - * blocks. The checked exceptions a collaborator method declares reach the typed ones, such as the - * {@code MaintenanceUnavailableException} catch in {@code getPaginatedNowWitnessList}; a method can - * declare several, so the checked sweep runs once for each declaration position. A handler that - * lets the failure escape is checked as well: what it recorded before throwing may not contain a - * second terminal event either. + * blocks. The first checked exception a collaborator method declares reaches the typed ones, such + * as the {@code MaintenanceUnavailableException} catch in {@code getPaginatedNowWitnessList}; no + * handler catches two checked types in separate blocks, so one checked sweep reaches them all. That + * sweep runs on {@code WalletApi} only, because {@code WalletSolidityApi} forwards to it or calls + * the same helpers. A handler that lets the failure escape is checked as well: what it recorded + * before throwing may not contain a second terminal event either. * *

Nothing here is a singleton: each sweep builds its own service and mocks, so no {@code @After} * reset is needed. @@ -47,13 +47,9 @@ public class RpcApiServiceErrorPathTest { /** Minimum handlers that must actually fail, so the sweep cannot silently cover nothing. */ private static final int MIN_EXERCISED = 20; - /** Minimum handlers whose typed catch must run in each checked sweep, for the same reason. */ + /** Minimum handlers whose typed catch must run in the checked sweep, for the same reason. */ private static final int MIN_EXERCISED_CHECKED = 10; - /** Collaborators the handlers call; the exceptions their methods declare drive the sweeps. */ - private static final List> COLLABORATORS = Arrays.asList( - Wallet.class, TransactionUtil.class, NodeInfoService.class, MetricsApiService.class); - @Test public void testWalletApiTerminatesTheCallOnce() throws Exception { assertSingleTerminalEvent(WalletApi.class, Failure.unchecked(), MIN_EXERCISED); @@ -66,18 +62,7 @@ public void testWalletSolidityApiTerminatesTheCallOnce() throws Exception { @Test public void testWalletApiTerminatesTheCallOnceOnCheckedExceptions() throws Exception { - for (int position = 0; position < declarationPositions(); position++) { - assertSingleTerminalEvent(WalletApi.class, Failure.checked(position), - MIN_EXERCISED_CHECKED); - } - } - - @Test - public void testWalletSolidityApiTerminatesTheCallOnceOnCheckedExceptions() throws Exception { - for (int position = 0; position < declarationPositions(); position++) { - assertSingleTerminalEvent(WalletSolidityApi.class, Failure.checked(position), - MIN_EXERCISED_CHECKED); - } + assertSingleTerminalEvent(WalletApi.class, Failure.checked(), MIN_EXERCISED_CHECKED); } /** @@ -154,28 +139,15 @@ private static void set(RpcApiService service, String name, Object value) throws field.set(service, value); } - /** Most checked exceptions any collaborator method declares, i.e. the checked sweeps to run. */ - private static int declarationPositions() { - int positions = 0; - for (Class collaborator : COLLABORATORS) { - for (Method method : collaborator.getMethods()) { - int declared = checkedExceptions(method).size(); - positions = declared > positions ? declared : positions; - } - } - Assert.assertTrue("no collaborator method declares a checked exception", positions > 0); - return positions; - } - - private static List> checkedExceptions(Method method) { - List> checked = new ArrayList<>(); + /** The first checked exception the method declares, or null if it declares none. */ + private static Class firstCheckedException(Method method) { for (Class type : method.getExceptionTypes()) { if (Exception.class.isAssignableFrom(type) && !RuntimeException.class.isAssignableFrom(type)) { - checked.add(type); + return type; } } - return checked; + return null; } /** Builds the exception through its message or no-arg constructor; null if it has neither. */ @@ -192,45 +164,40 @@ private static Exception newException(Class type) { } /** - * How every collaborator call fails in one sweep: with an unchecked exception, or with the - * checked exception the called method declares at a given position (its last one when it - * declares fewer). + * How every collaborator call fails in one sweep: with an unchecked exception, or with the first + * checked exception the called method declares. */ private static final class Failure implements Answer { - /** Declaration position to throw, or -1 for an unchecked exception. */ - private final int position; + private final boolean checked; /** Whether the current handler's collaborator call threw a checked exception. */ private boolean threwChecked; - private Failure(int position) { - this.position = position; + private Failure(boolean checked) { + this.checked = checked; } static Failure unchecked() { - return new Failure(-1); + return new Failure(false); } - static Failure checked(int position) { - return new Failure(position); + static Failure checked() { + return new Failure(true); } - /** A checked sweep only counts handlers whose collaborator really threw a checked exception. */ + /** The checked sweep only counts handlers whose collaborator really threw a checked one. */ boolean reachedTheIntendedCatch() { - return position < 0 || threwChecked; + return !checked || threwChecked; } @Override public Object answer(InvocationOnMock invocation) throws Throwable { - if (position >= 0) { - List> declared = checkedExceptions(invocation.getMethod()); - if (!declared.isEmpty()) { - Exception checked = newException( - declared.get(position < declared.size() ? position : declared.size() - 1)); - if (checked != null) { - threwChecked = true; - throw checked; - } + if (checked) { + Class declared = firstCheckedException(invocation.getMethod()); + Exception exception = declared == null ? null : newException(declared); + if (exception != null) { + threwChecked = true; + throw exception; } } throw new RuntimeException("collaborator unavailable"); @@ -238,8 +205,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable { @Override public String toString() { - return position < 0 ? "(unchecked failures)" - : "(checked failures, declaration position " + position + ")"; + return checked ? "(checked failures)" : "(unchecked failures)"; } } diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java index 237f68aa8ff..6afd8a15f58 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorScopeTest.java @@ -70,7 +70,7 @@ public void tearDown() throws Exception { @Test public void testHandlerSeesTheCursorWhenInterceptCallRanOnAnotherThread() throws Exception { - ServerCall.Listener listener = startCallOnThreadA(false); + ServerCall.Listener listener = startCallOnThreadA(); // the handler runs from onHalfClose, on a different thread than interceptCall Chainbase.Cursor cursorOnThreadBAfterHalfClose = runOn(threadB, () -> { @@ -86,27 +86,8 @@ public void testHandlerSeesTheCursorWhenInterceptCallRanOnAnotherThread() throws Chainbase.Cursor.HEAD, cursorOnThreadBAfterHalfClose); } - @Test - public void testCursorIsRestoredOnThreadBWhenTheHandlerThrows() throws Exception { - ServerCall.Listener listener = startCallOnThreadA(true); - - try { - runOn(threadB, () -> { - listener.onHalfClose(); - return null; - }); - Assert.fail("expected the handler failure to propagate"); - } catch (Exception expected) { - // what matters is the cursor state below - } - - // threadB is a single thread, so this reads the cursor of the thread the handler threw on - Assert.assertEquals("a throwing handler must still leave the cursor at HEAD", - Chainbase.Cursor.HEAD, runOn(threadB, cursor::get)); - } - /** Runs interceptCall on thread A and returns the listener, with a handler that records state. */ - private ServerCall.Listener startCallOnThreadA(boolean handlerThrows) throws Exception { + private ServerCall.Listener startCallOnThreadA() throws Exception { SolidityCursorInterceptor interceptor = new SolidityCursorInterceptor(); Field dbManager = CursorServerInterceptor.class.getDeclaredField("dbManager"); dbManager.setAccessible(true); @@ -122,9 +103,6 @@ private ServerCall.Listener startCallOnThreadA(boolean handlerThrows) th @Override public void onHalfClose() { cursorDuringHandler = cursor.get(); - if (handlerThrows) { - throw new IllegalStateException("boom"); - } } }; diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java index d2aee4247eb..854fa08b09b 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorServerTest.java @@ -105,7 +105,9 @@ public void getNowBlock(EmptyMessage request, StreamObserver observer) { ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", server.getPort()) .usePlaintext().directExecutor().build(); try { - DatabaseGrpc.newBlockingStub(channel).getNowBlock(EmptyMessage.getDefaultInstance()); + // the deadline keeps a server that never answers from hanging the test + DatabaseGrpc.newBlockingStub(channel).withDeadlineAfter(5, TimeUnit.SECONDS) + .getNowBlock(EmptyMessage.getDefaultInstance()); Assert.assertTrue("cursor was not restored after the call", resetDone.await(5, TimeUnit.SECONDS)); diff --git a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java index af071541df7..f7dbb5ea5e6 100644 --- a/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java +++ b/framework/src/test/java/org/tron/core/services/filter/CursorInterceptorWiringTest.java @@ -4,12 +4,10 @@ import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.withSettings; -import io.grpc.BindableService; import io.grpc.Metadata; import io.grpc.MethodDescriptor; import io.grpc.ServerCall; @@ -116,7 +114,8 @@ private static void assertEveryServiceReadsUnder(Class serviceClass, /** * Runs the service's real addService against a mock builder and returns the definitions it - * registered, failing if any went through the {@code BindableService} overload. + * registered. A service added through the {@code BindableService} overload is not among them, so + * the count check below fails for it. */ private static List register(Class serviceClass, CursorServerInterceptor interceptor, Answer collaborator) throws Exception { @@ -141,7 +140,6 @@ private static List register(Class serviceClass, addService.setAccessible(true); addService.invoke(service, builder); - verify(builder, never()).addService(any(BindableService.class)); ArgumentCaptor registered = ArgumentCaptor.forClass(ServerServiceDefinition.class); verify(builder, times(2)).addService(registered.capture());