Skip to content

Support log reordring from reorgs in LogSubscriptionService #86

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RatanRSur merged 41 commits into from
Oct 24, 2019
Merged

Support log reordring from reorgs in LogSubscriptionService #86

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
41 commits
Select commit Hold shift + click to select a range
42601c4
rework filters to use less mocking
RatanRSur Sep 16, 2019
7bac2c4
use checkArgument
RatanRSur Sep 16, 2019
fd1cb7b
move LogWithMetadata to core
RatanRSur Sep 16, 2019
5169788
use getOrThrow
RatanRSur Sep 17, 2019
4ad7d1f
move BlockWithReceipts to core
RatanRSur Sep 17, 2019
b87cc5c
thread logs with metadata through `DefaultBlockchain` into the `Block…
RatanRSur Sep 18, 2019
a561e25
reorder
RatanRSur Oct 1, 2019
8639ed4
make tests compile
RatanRSur Oct 1, 2019
8293dc2
wip: wire `LogSubscriptionService to use new event
RatanRSur Oct 1, 2019
57ad75a
cont: thread BlockWithReceipts further into call stack because new ch…
RatanRSur Oct 1, 2019
bd2c76b
singleton list
RatanRSur Oct 1, 2019
45be5ec
assertThat
RatanRSur Oct 2, 2019
bc28ce1
implement the chronological ordering of `LogWithMetadata` with `Compa…
RatanRSur Oct 2, 2019
7d26d11
clean up reorg
RatanRSur Oct 2, 2019
ca86acc
fix bug in test
RatanRSur Oct 2, 2019
ad6d114
remove unnecessary separation between (added|removed)LogsWithMetadata
RatanRSur Oct 3, 2019
64d14d9
use Deque instead of sorting
RatanRSur Oct 3, 2019
271a2ae
fix typo and remove abbrev.
RatanRSur Oct 3, 2019
fa09780
typo
RatanRSur Oct 3, 2019
33c9403
remove var
RatanRSur Oct 3, 2019
a955f84
override `getChainHeadBlock`
RatanRSur Oct 3, 2019
b7d4f15
base reorg iteration off of `BlockWithReceipt`s
RatanRSur Oct 3, 2019
733e396
move `List<LogWithMetadata>` logic from `BlockWithReceipts` to `LogWi…
RatanRSur Oct 3, 2019
2a853a2
imports and stuff
RatanRSur Oct 3, 2019
575fac6
functional style
RatanRSur Oct 14, 2019
08d7c1c
cleanup
RatanRSur Oct 14, 2019
7cd9ddd
add tests for log ordering
RatanRSur Oct 14, 2019
f37f627
imports
RatanRSur Oct 14, 2019
3b6f613
make it compile after rebase
RatanRSur Oct 21, 2019
de24d3e
consolidate LogWithMetadata creation
RatanRSur Oct 21, 2019
4ceefca
functional style
RatanRSur Oct 21, 2019
31e2fe7
remove parens
RatanRSur Oct 21, 2019
e4b0266
wip: more tests
RatanRSur Oct 22, 2019
8b76d6e
testing and bugfix
RatanRSur Oct 22, 2019
257e1d6
remove comparison
RatanRSur Oct 22, 2019
e02f37e
findAny
RatanRSur Oct 23, 2019
d481a0b
final
RatanRSur Oct 23, 2019
0e2a402
bytesValue size range
RatanRSur Oct 23, 2019
64675cd
return up to to block in query matching
RatanRSur Oct 24, 2019
3aa15ce
Merge branch 'master' into rework-filters
RatanRSur Oct 24, 2019
ac95246
fix toBlock spec test
RatanRSur Oct 24, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 3 additions & 7 deletions besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,8 +453,7 @@ public Runner build() {
final SubscriptionManager subscriptionManager =
createSubscriptionManager(vertx, transactionPool);

createLogsSubscriptionService(
context.getBlockchain(), context.getWorldStateArchive(), subscriptionManager);
createLogsSubscriptionService(context.getBlockchain(), subscriptionManager);

createNewBlockHeadersSubscriptionService(
context.getBlockchain(), context.getWorldStateArchive(), subscriptionManager);
Expand Down Expand Up @@ -621,12 +620,9 @@ private SubscriptionManager createSubscriptionManager(
}

private void createLogsSubscriptionService(
final Blockchain blockchain,
final WorldStateArchive worldStateArchive,
final SubscriptionManager subscriptionManager) {
final Blockchain blockchain, final SubscriptionManager subscriptionManager) {
final LogsSubscriptionService logsSubscriptionService =
new LogsSubscriptionService(
subscriptionManager, new BlockchainQueries(blockchain, worldStateArchive));
new LogsSubscriptionService(subscriptionManager);

blockchain.observeBlockAdded(logsSubscriptionService);
}
Expand Down
4 changes: 3 additions & 1 deletion ...est/java/org/hyperledger/besu/consensus/ibft/blockcreation/IbftMiningCoordinatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.hyperledger.besu.ethereum.core.Wei;
import org.hyperledger.besu.util.bytes.BytesValue;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

import org.assertj.core.util.Lists;
Expand Down Expand Up @@ -85,7 +86,8 @@ public void setsTheExtraData() {

@Test
public void addsNewChainHeadEventWhenNewCanonicalHeadBlockEventReceived() throws Exception {
BlockAddedEvent headAdvancement = BlockAddedEvent.createForHeadAdvancement(block);
BlockAddedEvent headAdvancement =
BlockAddedEvent.createForHeadAdvancement(block, Collections.emptyList());
ibftMiningCoordinator.onBlockAdded(headAdvancement, blockChain);

assertThat(eventQueue.size()).isEqualTo(1);
Expand Down
2 changes: 1 addition & 1 deletion ...reum/api/src/main/java/org/hyperledger/besu/ethereum/api/graphql/GraphQLDataFetchers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,14 +25,14 @@
import org.hyperledger.besu.ethereum.api.graphql.internal.response.GraphQLError;
import org.hyperledger.besu.ethereum.api.query.BlockWithMetadata;
import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.LogsQuery;
import org.hyperledger.besu.ethereum.api.query.TransactionWithMetadata;
import org.hyperledger.besu.ethereum.blockcreation.MiningCoordinator;
import org.hyperledger.besu.ethereum.core.Account;
import org.hyperledger.besu.ethereum.core.Address;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogTopic;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.MutableWorldState;
import org.hyperledger.besu.ethereum.core.Synchronizer;
import org.hyperledger.besu.ethereum.core.Transaction;
Expand Down
2 changes: 1 addition & 1 deletion ...java/org/hyperledger/besu/ethereum/api/graphql/internal/pojoadapter/BlockAdapterBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,13 @@
import org.hyperledger.besu.ethereum.api.graphql.GraphQLDataFetcherContext;
import org.hyperledger.besu.ethereum.api.query.BlockWithMetadata;
import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.LogsQuery;
import org.hyperledger.besu.ethereum.api.query.TransactionWithMetadata;
import org.hyperledger.besu.ethereum.core.Address;
import org.hyperledger.besu.ethereum.core.BlockHeader;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogTopic;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.Wei;
import org.hyperledger.besu.ethereum.core.WorldState;
import org.hyperledger.besu.ethereum.mainnet.ProtocolSchedule;
Expand Down
4 changes: 2 additions & 2 deletions .../main/java/org/hyperledger/besu/ethereum/api/graphql/internal/pojoadapter/LogAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,10 @@
package org.hyperledger.besu.ethereum.api.graphql.internal.pojoadapter;

import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.TransactionWithMetadata;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogTopic;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.util.bytes.Bytes32;
import org.hyperledger.besu.util.bytes.BytesValue;

Expand Down Expand Up @@ -70,6 +70,6 @@ public Optional<AccountAdapter> getAccount(final DataFetchingEnvironment environ

return query
.getWorldState(blockNumber)
.map(ws -> new AccountAdapter(ws.get(logWithMetadata.getAddress())));
.map(ws -> new AccountAdapter(ws.get(logWithMetadata.getLogger())));
}
}
10 changes: 5 additions & 5 deletions ...va/org/hyperledger/besu/ethereum/api/graphql/internal/pojoadapter/TransactionAdapter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,11 +15,11 @@
package org.hyperledger.besu.ethereum.api.graphql.internal.pojoadapter;

import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.TransactionReceiptWithMetadata;
import org.hyperledger.besu.ethereum.api.query.TransactionWithMetadata;
import org.hyperledger.besu.ethereum.core.Address;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.MutableWorldState;
import org.hyperledger.besu.ethereum.core.Transaction;
import org.hyperledger.besu.ethereum.core.TransactionReceipt;
Expand Down Expand Up @@ -167,13 +167,13 @@ public Optional<AccountAdapter> getCreatedContract(final DataFetchingEnvironment
public List<LogAdapter> getLogs(final DataFetchingEnvironment environment) {
final BlockchainQueries query = getBlockchainQueries(environment);
final Hash hash = transactionWithMetadata.getTransaction().getHash();
final Optional<TransactionReceiptWithMetadata> tranRpt =
final Optional<TransactionReceiptWithMetadata> maybeTransactionReceiptWithMetadata =
query.transactionReceiptByTransactionHash(hash);
final List<LogAdapter> results = new ArrayList<>();
if (tranRpt.isPresent()) {
if (maybeTransactionReceiptWithMetadata.isPresent()) {
final List<LogWithMetadata> logs =
BlockchainQueries.generateLogWithMetadataForTransaction(
tranRpt.get().getReceipt(),
LogWithMetadata.generate(
maybeTransactionReceiptWithMetadata.get().getReceipt(),
transactionWithMetadata.getBlockNumber().get(),
transactionWithMetadata.getBlockHash().get(),
hash,
Expand Down
2 changes: 1 addition & 1 deletion ...rc/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/filter/FilterManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,11 @@

import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.BlockParameter;
import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.LogsQuery;
import org.hyperledger.besu.ethereum.chain.BlockAddedEvent;
import org.hyperledger.besu.ethereum.chain.Blockchain;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.Transaction;
import org.hyperledger.besu.ethereum.eth.transactions.TransactionPool;

Expand Down
2 changes: 1 addition & 1 deletion ...pi/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/filter/LogFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,8 @@
package org.hyperledger.besu.ethereum.api.jsonrpc.internal.filter;

import org.hyperledger.besu.ethereum.api.jsonrpc.internal.parameters.BlockParameter;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.LogsQuery;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;

import java.util.ArrayList;
import java.util.List;
Expand Down
2 changes: 1 addition & 1 deletion .../java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/EthGetFilterChanges.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse;
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.LogsResult;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.Hash;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;

import java.util.List;
import java.util.stream.Collectors;
Expand Down
2 changes: 1 addition & 1 deletion ...ain/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/methods/EthGetFilterLogs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcResponse;
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.response.JsonRpcSuccessResponse;
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.LogsResult;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;

import java.util.List;

Expand Down
4 changes: 2 additions & 2 deletions ...i/src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/LogResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@
*/
package org.hyperledger.besu.ethereum.api.jsonrpc.internal.results;

import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.LogTopic;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;

import java.util.ArrayList;
import java.util.List;
Expand Down Expand Up @@ -53,7 +53,7 @@ public LogResult(final LogWithMetadata logWithMetadata) {
this.blockHash = logWithMetadata.getBlockHash().toString();
this.transactionHash = logWithMetadata.getTransactionHash().toString();
this.transactionIndex = Quantity.create(logWithMetadata.getTransactionIndex());
this.address = logWithMetadata.getAddress().toString();
this.address = logWithMetadata.getLogger().toString();
this.data = logWithMetadata.getData().toString();
this.topics = new ArrayList<>(logWithMetadata.getTopics().size());
this.removed = logWithMetadata.isRemoved();
Expand Down
2 changes: 1 addition & 1 deletion .../src/main/java/org/hyperledger/besu/ethereum/api/jsonrpc/internal/results/LogsResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
*/
package org.hyperledger.besu.ethereum.api.jsonrpc.internal.results;

import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.core.LogWithMetadata;

import java.util.ArrayList;
import java.util.List;
Expand Down
90 changes: 15 additions & 75 deletions ...ledger/besu/ethereum/api/jsonrpc/websocket/subscription/logs/LogsSubscriptionService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,97 +17,37 @@
import org.hyperledger.besu.ethereum.api.jsonrpc.internal.results.LogResult;
import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.SubscriptionManager;
import org.hyperledger.besu.ethereum.api.jsonrpc.websocket.subscription.request.SubscriptionType;
import org.hyperledger.besu.ethereum.api.query.BlockchainQueries;
import org.hyperledger.besu.ethereum.api.query.LogWithMetadata;
import org.hyperledger.besu.ethereum.api.query.TransactionReceiptWithMetadata;
import org.hyperledger.besu.ethereum.chain.BlockAddedEvent;
import org.hyperledger.besu.ethereum.chain.BlockAddedObserver;
import org.hyperledger.besu.ethereum.chain.Blockchain;
import org.hyperledger.besu.ethereum.core.Log;

import java.util.List;
import java.util.Optional;

public class LogsSubscriptionService implements BlockAddedObserver {

private final SubscriptionManager subscriptionManager;
private final BlockchainQueries blockchainQueries;

public LogsSubscriptionService(
final SubscriptionManager subscriptionManager, final BlockchainQueries blockchainQueries) {
public LogsSubscriptionService(final SubscriptionManager subscriptionManager) {
this.subscriptionManager = subscriptionManager;
this.blockchainQueries = blockchainQueries;
}

@Override
public void onBlockAdded(final BlockAddedEvent event, final Blockchain blockchain) {
public void onBlockAdded(final BlockAddedEvent event, final Blockchain __) {
Copy link
Copy Markdown
Contributor

@CjHare CjHare Oct 14, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What is the motivation for the underscore (_) as the input parameter name?

Copy link
Copy Markdown
Contributor Author

@RatanRSur RatanRSur Oct 14, 2019 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's to signify that the argument is not used

final List<LogsSubscription> logsSubscriptions =
subscriptionManager.subscriptionsOfType(SubscriptionType.LOGS, LogsSubscription.class);

if (logsSubscriptions.isEmpty()) {
return;
}

event.getAddedTransactions().stream()
.map(tx -> blockchainQueries.transactionReceiptByTransactionHash(tx.getHash()))
.filter(Optional::isPresent)
.map(Optional::get)
.forEachOrdered(
receiptWithMetadata -> {
final List<Log> logs = receiptWithMetadata.getReceipt().getLogs();
sendLogsToMatchingSubscriptions(logs, logsSubscriptions, receiptWithMetadata, false);
});

event.getRemovedTransactions().stream()
.map(tx -> blockchainQueries.transactionReceiptByTransactionHash(tx.getHash()))
.filter(Optional::isPresent)
.map(Optional::get)
.forEachOrdered(
receiptWithMetadata -> {
final List<Log> logs = receiptWithMetadata.getReceipt().getLogs();
sendLogsToMatchingSubscriptions(logs, logsSubscriptions, receiptWithMetadata, true);
});
}

private void sendLogsToMatchingSubscriptions(
final List<Log> logs,
final List<LogsSubscription> logsSubscriptions,
final TransactionReceiptWithMetadata receiptWithMetadata,
final boolean removed) {
for (int logIndex = 0; logIndex < logs.size(); logIndex++) {
for (final LogsSubscription subscription : logsSubscriptions) {
if (subscription.getLogsQuery().matches(logs.get(logIndex))) {
sendLogToSubscription(receiptWithMetadata, removed, logIndex, subscription);
}
}
}
}

private void sendLogToSubscription(
final TransactionReceiptWithMetadata receiptWithMetadata,
final boolean removed,
final int logIndex,
final LogsSubscription subscription) {
final LogWithMetadata logWithMetaData = logWithMetadata(logIndex, receiptWithMetadata, removed);
subscriptionManager.sendMessage(
subscription.getSubscriptionId(), new LogResult(logWithMetaData));
}

// @formatter:off
private LogWithMetadata logWithMetadata(
final int logIndex,
final TransactionReceiptWithMetadata transactionReceiptWithMetadata,
final boolean removed) {
return new LogWithMetadata(
logIndex,
transactionReceiptWithMetadata.getBlockNumber(),
transactionReceiptWithMetadata.getBlockHash(),
transactionReceiptWithMetadata.getTransactionHash(),
transactionReceiptWithMetadata.getTransactionIndex(),
transactionReceiptWithMetadata.getReceipt().getLogs().get(logIndex).getLogger(),
transactionReceiptWithMetadata.getReceipt().getLogs().get(logIndex).getData(),
transactionReceiptWithMetadata.getReceipt().getLogs().get(logIndex).getTopics(),
removed);
event
.getLogsWithMetadata()
.forEach(
logWithMetadata ->
logsSubscriptions.stream()
.filter(
logsSubscription ->
logsSubscription.getLogsQuery().matches(logWithMetadata))
.forEach(
logsSubscription ->
subscriptionManager.sendMessage(
logsSubscription.getSubscriptionId(),
new LogResult(logWithMetadata))));
}
// @formatter:on
}
Loading