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 b9cb05a3b14..5bea36bf632 100755 --- a/framework/src/main/java/org/tron/core/services/RpcApiService.java +++ b/framework/src/main/java/org/tron/core/services/RpcApiService.java @@ -181,6 +181,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(); @@ -362,249 +364,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 @@ -616,29 +507,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 @@ -661,198 +536,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 @@ -865,45 +620,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 @@ -915,23 +638,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); } } @@ -953,7 +666,9 @@ private TransactionListExtention transactionList2Extention(TransactionList trans } /** - * WalletApi. + * WalletApi is the full implementation of the {@code protocol.Wallet} gRPC service, including + * write and build endpoints. {@link WalletSolidityApi} is the read-only subset of this surface + * and delegates its handlers here. */ public class WalletApi extends WalletImplBase { @@ -1479,15 +1194,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(); } @@ -1594,7 +1320,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); @@ -1881,6 +1607,7 @@ public void getPaginatedNowWitnessList(PaginatedMessage request, wallet.getPaginatedNowWitnessList(request.getOffset(), request.getLimit())); } catch (MaintenanceUnavailableException e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1910,6 +1637,7 @@ public void getDelegatedResourceV2(DelegatedResourceMessage request, ); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1922,6 +1650,7 @@ public void getDelegatedResourceAccountIndex(BytesMessage request, .onNext(wallet.getDelegatedResourceAccountIndex(request.getValue())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1934,6 +1663,7 @@ public void getDelegatedResourceAccountIndexV2(BytesMessage request, .onNext(wallet.getDelegatedResourceAccountIndexV2(request.getValue())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1946,6 +1676,7 @@ public void getCanDelegatedMaxSize(GrpcAPI.CanDelegatedMaxSizeRequestMessage req request.getOwnerAddress(), request.getType())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -1959,6 +1690,7 @@ public void getAvailableUnfreezeCount(GrpcAPI.GetAvailableUnfreezeCountRequestMe request.getOwnerAddress())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -1974,6 +1706,7 @@ public void getCanWithdrawUnfreezeAmount(CanWithdrawUnfreezeAmountRequestMessage )); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1985,6 +1718,7 @@ public void getBandwidthPrices(EmptyMessage request, responseObserver.onNext(wallet.getBandwidthPrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -1996,6 +1730,7 @@ public void getEnergyPrices(EmptyMessage request, responseObserver.onNext(wallet.getEnergyPrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2007,6 +1742,7 @@ public void getMemoFee(EmptyMessage request, responseObserver.onNext(wallet.getMemoFeePrices()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2063,6 +1799,7 @@ public void getNodeInfo(EmptyMessage request, StreamObserver responseO responseObserver.onNext(nodeInfoService.getNodeInfo().transferToProtoEntity()); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2539,6 +2276,7 @@ public void getTransactionInfoByBlockNum(NumberMessage request, responseObserver.onNext(wallet.getTransactionInfoByBlockNum(request.getNum())); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); @@ -2568,6 +2306,7 @@ public void getMarketOrderByAccount(BytesMessage request, responseObserver.onNext(marketOrderList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2583,6 +2322,7 @@ public void getMarketOrderById(BytesMessage request, responseObserver.onNext(marketOrder); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2597,6 +2337,7 @@ public void getMarketPriceByPair(MarketOrderPair request, responseObserver.onNext(marketPriceList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2611,6 +2352,7 @@ public void getMarketOrderListByPair(org.tron.protos.Protocol.MarketOrderPair re responseObserver.onNext(orderPairList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2623,6 +2365,7 @@ public void getMarketPairList(EmptyMessage request, responseObserver.onNext(pairList); } catch (Exception e) { responseObserver.onError(getRunTimeException(e)); + return; } responseObserver.onCompleted(); } @@ -2672,6 +2415,7 @@ public void getRewardInfoCommon(BytesMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2685,6 +2429,7 @@ public void getBurnTrxCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2700,6 +2445,7 @@ public void getBrokerageInfoCommon(BytesMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2726,6 +2472,7 @@ public void getTransactionFromPendingCommon(BytesMessage request, responseObserver.onNext(transactionCapsule == null ? null : transactionCapsule.getInstance()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2738,6 +2485,7 @@ public void getTransactionListFromPendingCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2750,6 +2498,7 @@ public void getPendingSizeCommon(EmptyMessage request, responseObserver.onNext(builder.build()); } catch (Exception e) { responseObserver.onError(e); + return; } responseObserver.onCompleted(); } @@ -2765,6 +2514,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 315d70df8d6..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,70 +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.parameter.CommonParameter; -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(); @@ -72,417 +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(CommonParameter.getInstance().isECKeyCryptoEngine(), - 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 c3ac5800971..3df54b9c0f9 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()); + } +}