CipherLabs
/
besu
mirror of https://github.com/hyperledger/besu
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Snap sync server StorageRange message limit to apply limit hash as post check (#7399)

Browse Source 
Signed-off-by: Jason Frame <jason.frame@consensys.net>
pull/7456/head
Jason Frame 3 months ago committed by GitHub
parent a55c331e21
commit dc336f48e9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 126 additions and 44 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 10
      ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/storage/DiffBasedWorldStateKeyValueStorage.java
  2. 39
      ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/storage/flat/FlatDbStrategy.java
  3. 80
      ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapServer.java
  4. 41
      ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapServerTest.java

10
ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/storage/DiffBasedWorldStateKeyValueStorage.java
Unescape View File

@ -129,11 +129,9 @@ public abstract class DiffBasedWorldStateKeyValueStorage
}
public NavigableMap<Bytes32, Bytes> streamFlatAccounts(
final Bytes startKeyHash,
final Bytes32 endKeyHash,
final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
final Bytes startKeyHash, final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
return getFlatDbStrategy()
.streamAccountFlatDatabase(composedWorldStateStorage, startKeyHash, endKeyHash, takeWhile);
.streamAccountFlatDatabase(composedWorldStateStorage, startKeyHash, takeWhile);
}
public NavigableMap<Bytes32, Bytes> streamFlatStorages(
@ -146,11 +144,9 @@ public abstract class DiffBasedWorldStateKeyValueStorage
public NavigableMap<Bytes32, Bytes> streamFlatStorages(
final Hash accountHash,
final Bytes startKeyHash,
final Bytes32 endKeyHash,
final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
return getFlatDbStrategy()
.streamStorageFlatDatabase(
composedWorldStateStorage, accountHash, startKeyHash, endKeyHash, takeWhile);
.streamStorageFlatDatabase(composedWorldStateStorage, accountHash, startKeyHash, takeWhile);
}
public boolean isWorldStateAvailable(final Bytes32 rootHash, final Hash blockHash) {

39
ethereum/core/src/main/java/org/hyperledger/besu/ethereum/trie/diffbased/common/storage/flat/FlatDbStrategy.java
Unescape View File

@ -207,11 +207,9 @@ public abstract class FlatDbStrategy {
public NavigableMap<Bytes32, Bytes> streamAccountFlatDatabase(
final SegmentedKeyValueStorage storage,
final Bytes startKeyHash,
final Bytes32 endKeyHash,
final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
return toNavigableMap(
accountsToPairStream(storage, startKeyHash, endKeyHash).takeWhile(takeWhile));
return toNavigableMap(accountsToPairStream(storage, startKeyHash).takeWhile(takeWhile));
}
/** streams RLP encoded storage values using a specified stream limit. */
@ -240,6 +238,34 @@ public abstract class FlatDbStrategy {
.takeWhile(takeWhile));
}
/** streams raw storage Bytes using a specified predicate filter and value mapper. */
public NavigableMap<Bytes32, Bytes> streamStorageFlatDatabase(
final SegmentedKeyValueStorage storage,
final Hash accountHash,
final Bytes startKeyHash,
final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
return toNavigableMap(
storageToPairStream(storage, accountHash, startKeyHash, RLP::encodeValue)
.takeWhile(takeWhile));
}
private static Stream<Pair<Bytes32, Bytes>> storageToPairStream(
final SegmentedKeyValueStorage storage,
final Hash accountHash,
final Bytes startKeyHash,
final Function<Bytes, Bytes> valueMapper) {
return storage
.streamFromKey(
ACCOUNT_STORAGE_STORAGE, Bytes.concatenate(accountHash, startKeyHash).toArrayUnsafe())
.takeWhile(pair -> Bytes.wrap(pair.getKey()).slice(0, Hash.SIZE).equals(accountHash))
.map(
pair ->
new Pair<>(
Bytes32.wrap(Bytes.wrap(pair.getKey()).slice(Hash.SIZE)),
valueMapper.apply(Bytes.wrap(pair.getValue()).trimLeadingZeros())));
}
private static Stream<Pair<Bytes32, Bytes>> storageToPairStream(
final SegmentedKeyValueStorage storage,
final Hash accountHash,
@ -266,6 +292,13 @@ public abstract class FlatDbStrategy {
.map(pair -> new Pair<>(Bytes32.wrap(pair.getKey()), Bytes.wrap(pair.getValue())));
}
private static Stream<Pair<Bytes32, Bytes>> accountsToPairStream(
final SegmentedKeyValueStorage storage, final Bytes startKeyHash) {
return storage
.streamFromKey(ACCOUNT_INFO_STATE, startKeyHash.toArrayUnsafe())
.map(pair -> new Pair<>(Bytes32.wrap(pair.getKey()), Bytes.wrap(pair.getValue())));
}
private static NavigableMap<Bytes32, Bytes> toNavigableMap(
final Stream<Pair<Bytes32, Bytes>> pairStream) {
final TreeMap<Bytes32, Bytes> collected =

80
ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapServer.java
Unescape View File

@ -234,8 +234,8 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
.map(
storage -> {
LOGGER.trace("obtained worldstate in {}", stopWatch);
StatefulPredicate shouldContinuePredicate =
new StatefulPredicate(
ResponseSizePredicate responseSizePredicate =
new ResponseSizePredicate(
"account",
stopWatch,
maxResponseBytes,
@ -248,9 +248,13 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
return rlpOutput.encodedSize();
});
final Bytes32 endKeyBytes = range.endKeyHash();
var shouldContinuePredicate =
new ExceedingPredicate(
new EndKeyExceedsPredicate(endKeyBytes).and(responseSizePredicate));
NavigableMap<Bytes32, Bytes> accounts =
storage.streamFlatAccounts(
range.startKeyHash(), range.endKeyHash(), shouldContinuePredicate);
storage.streamFlatAccounts(range.startKeyHash(), shouldContinuePredicate);
if (accounts.isEmpty() && shouldContinuePredicate.shouldContinue.get()) {
// fetch next account after range, if it exists
@ -331,8 +335,8 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
storage -> {
LOGGER.trace("obtained worldstate in {}", stopWatch);
// reusable predicate to limit by rec count and bytes:
var statefulPredicate =
new StatefulPredicate(
var responsePredicate =
new ResponseSizePredicate(
"storage",
stopWatch,
maxResponseBytes,
@ -364,9 +368,12 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
new WorldStateProofProvider(new WorldStateStorageCoordinator(storage));
for (var forAccountHash : range.hashes()) {
var predicate =
new ExceedingPredicate(
new EndKeyExceedsPredicate(endKeyBytes).and(responsePredicate));
var accountStorages =
storage.streamFlatStorages(
Hash.wrap(forAccountHash), startKeyBytes, endKeyBytes, statefulPredicate);
Hash.wrap(forAccountHash), startKeyBytes, predicate);
//// address partial range queries that return empty
if (accountStorages.isEmpty() && isPartialRange) {
@ -386,7 +393,7 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
// if a partial storage range was requested, or we interrupted storage due to
// request limits, send proofs:
if (isPartialRange || !statefulPredicate.shouldGetMore()) {
if (isPartialRange || !predicate.shouldGetMore()) {
// send a proof for the left side range origin
proofNodes.addAll(
worldStateProof.getStorageProofRelatedNodes(
@ -403,7 +410,7 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
}
}
if (!statefulPredicate.shouldGetMore()) {
if (!predicate.shouldGetMore()) {
break;
}
}
@ -462,7 +469,7 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
if (optCode.isPresent()) {
if (!codeBytes.isEmpty()
&& (sumListBytes(codeBytes) + optCode.get().size() > maxResponseBytes
|| stopWatch.getTime() > StatefulPredicate.MAX_MILLIS_PER_REQUEST)) {
|| stopWatch.getTime() > ResponseSizePredicate.MAX_MILLIS_PER_REQUEST)) {
break;
}
codeBytes.add(optCode.get());
@ -521,7 +528,8 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
var trieNode = optStorage.orElse(Bytes.EMPTY);
if (!trieNodes.isEmpty()
&& (sumListBytes(trieNodes) + trieNode.size() > maxResponseBytes
|| stopWatch.getTime() > StatefulPredicate.MAX_MILLIS_PER_REQUEST)) {
|| stopWatch.getTime()
> ResponseSizePredicate.MAX_MILLIS_PER_REQUEST)) {
break;
}
trieNodes.add(trieNode);
@ -578,7 +586,39 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
}
}
static class StatefulPredicate implements Predicate<Pair<Bytes32, Bytes>> {
/**
* Predicate that doesn't immediately stop when the delegate predicate returns false, but instead
* sets a flag to stop after the current element is processed.
*/
static class ExceedingPredicate implements Predicate<Pair<Bytes32, Bytes>> {
private final Predicate<Pair<Bytes32, Bytes>> delegate;
final AtomicBoolean shouldContinue = new AtomicBoolean(true);
public ExceedingPredicate(final Predicate<Pair<Bytes32, Bytes>> delegate) {
this.delegate = delegate;
}
@Override
public boolean test(final Pair<Bytes32, Bytes> pair) {
final boolean result = delegate.test(pair);
return shouldContinue.getAndSet(result);
}
public boolean shouldGetMore() {
return shouldContinue.get();
}
}
/** Predicate that stops when the end key is exceeded. */
record EndKeyExceedsPredicate(Bytes endKey) implements Predicate<Pair<Bytes32, Bytes>> {
@Override
public boolean test(final Pair<Bytes32, Bytes> pair) {
return endKey.compareTo(Bytes.wrap(pair.getFirst())) > 0;
}
}
static class ResponseSizePredicate implements Predicate<Pair<Bytes32, Bytes>> {
// default to a max of 4 seconds per request
static final long MAX_MILLIS_PER_REQUEST = 4000;
@ -588,26 +628,19 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
final Function<Pair<Bytes32, Bytes>, Integer> encodingSizeAccumulator;
final StopWatch stopWatch;
final int maxResponseBytes;
// TODO: remove this hack, 10% is a fudge factor to account for the proof node size
final int maxResponseBytesFudgeFactor;
final String forWhat;
StatefulPredicate(
ResponseSizePredicate(
final String forWhat,
final StopWatch stopWatch,
final int maxResponseBytes,
final Function<Pair<Bytes32, Bytes>, Integer> encodingSizeAccumulator) {
this.stopWatch = stopWatch;
this.maxResponseBytes = maxResponseBytes;
this.maxResponseBytesFudgeFactor = maxResponseBytes * 9 / 10;
this.forWhat = forWhat;
this.encodingSizeAccumulator = encodingSizeAccumulator;
}
public boolean shouldGetMore() {
return shouldContinue.get();
}
@Override
public boolean test(final Pair<Bytes32, Bytes> pair) {
LOGGER
@ -628,14 +661,11 @@ class SnapServer implements BesuEvents.InitialSyncCompletionListener {
return false;
}
var hasNoRecords = recordLimit.get() == 0;
var underRecordLimit = recordLimit.addAndGet(1) <= MAX_ENTRIES_PER_REQUEST;
var underByteLimit =
byteLimit.accumulateAndGet(0, (cur, __) -> cur + encodingSizeAccumulator.apply(pair))
< maxResponseBytesFudgeFactor;
// Only enforce limits when we have at least 1 record as the snapsync spec
// requires at least 1 record must be returned
if (hasNoRecords || (underRecordLimit && underByteLimit)) {
< maxResponseBytes;
if (underRecordLimit && underByteLimit) {
return true;
} else {
shouldContinue.set(false);

41
ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapServerTest.java
Unescape View File

@ -214,8 +214,7 @@ public class SnapServerTest {
var rangeData =
getAndVerifyAccountRangeData(
(AccountRangeMessage) snapServer.constructGetAccountRangeResponse(tinyRangeLimit),
// TODO: after sorting out the request fudge factor, adjust this assertion to match
acctCount * 90 / 100 - 1);
acctCount);
// assert proofs are valid for the requested range
assertThat(assertIsValidAccountRangeProof(Hash.ZERO, rangeData)).isTrue();
@ -315,6 +314,31 @@ public class SnapServerTest {
.isTrue();
}
@Test
public void assertPartialStorageLimitHashBetweenSlots() {
Bytes accountShortHash = Bytes.fromHexStringLenient("0x40");
Hash accountFullHash = Hash.wrap(Bytes32.leftPad(accountShortHash));
SnapTestAccount testAccount = createTestContractAccount(accountFullHash, 2, inMemoryStorage);
Hash startHash = Hash.wrap(Bytes32.rightPad(Bytes.fromHexString("12"))); // slot 2
Hash endHash = Hash.wrap(Bytes32.rightPad(Bytes.fromHexString("13"))); // between slots 2 and 3
var rangeData = requestStorageRange(List.of(testAccount.addressHash), startHash, endHash);
assertThat(rangeData).isNotNull();
var slotsData = rangeData.slotsData(false);
assertThat(slotsData).isNotNull();
assertThat(slotsData.slots()).isNotNull();
assertThat(slotsData.slots().size()).isEqualTo(1);
var firstAccountStorages = slotsData.slots().first();
// expecting to see 2 slots
assertThat(firstAccountStorages.size()).isEqualTo(2);
// assert proofs are valid for the requested range
assertThat(
assertIsValidStorageProof(
testAccount, startHash, firstAccountStorages, slotsData.proofs()))
.isTrue();
}
@Test
public void assertLastEmptyPartialStorageForSingleAccount() {
// When our final range request is empty, no next account is possible,
@ -343,7 +367,7 @@ public class SnapServerTest {
@Test
public void assertStorageLimitRangeResponse() {
// assert we limit the range response according to bytessize
final int storageSlotSize = 70;
final int storageSlotSize = 69;
final int storageSlotCount = 16;
insertTestAccounts(acct1, acct2, acct3, acct4);
@ -374,8 +398,7 @@ public class SnapServerTest {
assertThat(firstAccountStorages.size()).isEqualTo(10);
var secondAccountStorages = slotsData.slots().last();
// expecting to see only 6 since request was limited to 16 slots
// TODO: after sorting out the request fudge factor, adjust this assertion to match
assertThat(secondAccountStorages.size()).isEqualTo(6 * 90 / 100 - 1);
assertThat(secondAccountStorages.size()).isEqualTo(6);
// proofs required for interrupted storage range:
assertThat(slotsData.proofs().size()).isNotEqualTo(0);
@ -556,7 +579,7 @@ public class SnapServerTest {
public void assertStorageTrieShortAccountHashPathRequest() {
Bytes accountShortHash = Bytes.fromHexStringLenient("0x40");
Hash accountFullHash = Hash.wrap(Bytes32.leftPad(accountShortHash));
SnapTestAccount testAccount = createTestContractAccount(accountFullHash, inMemoryStorage);
SnapTestAccount testAccount = createTestContractAccount(accountFullHash, 1, inMemoryStorage);
insertTestAccounts(testAccount);
var pathToSlot11 = CompactEncoding.encode(Bytes.fromHexStringLenient("0x0101"));
var pathToSlot12 = CompactEncoding.encode(Bytes.fromHexStringLenient("0x0102"));
@ -707,11 +730,11 @@ public class SnapServerTest {
static SnapTestAccount createTestContractAccount(
final String hexAddr, final BonsaiWorldStateKeyValueStorage storage) {
final Hash acctHash = Hash.wrap(Bytes32.rightPad(Bytes.fromHexString(hexAddr)));
return createTestContractAccount(acctHash, storage);
return createTestContractAccount(acctHash, 1, storage);
}
static SnapTestAccount createTestContractAccount(
final Hash acctHash, final BonsaiWorldStateKeyValueStorage storage) {
final Hash acctHash, final int slotKeyGap, final BonsaiWorldStateKeyValueStorage storage) {
MerkleTrie<Bytes32, Bytes> trie =
new StoredMerklePatriciaTrie<>(
(loc, hash) -> storage.getAccountStorageTrieNode(acctHash, loc, hash),
@ -724,7 +747,7 @@ public class SnapServerTest {
var flatdb = storage.getFlatDbStrategy();
var updater = storage.updater();
updater.putCode(Hash.hash(mockCode), mockCode);
IntStream.range(10, 20)
IntStream.iterate(10, i -> i < 20, i -> i + slotKeyGap)
.boxed()
.forEach(
i -> {

Write Preview
Loading…
Cancel
Save