Skip to content

Commit

Permalink
[skip ci] skip ci ; initial account storage message handling, proving…
Browse files Browse the repository at this point in the history 
… broken

Signed-off-by: garyschulte <[email protected]>
  • Loading branch information
@garyschulte
garyschulte committed Sep 23, 2023
1 parent 8179a31 commit 30dfc08
Show file tree
Hide file tree
Showing 10 changed files with 240 additions and 82 deletions.
11 changes: 11 additions & 0 deletions ...in/java/org/hyperledger/besu/ethereum/bonsai/storage/BonsaiWorldStateKeyValueStorage.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,17 @@ public NavigableMap<Bytes32, Bytes> streamFlatStorages(
composedWorldStateStorage, accountHash, startKeyHash, endKeyHash, max);
}

@Override
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);
}

public NavigableMap<Bytes32, AccountStorageEntry> storageEntriesFrom(
final Hash addressHash, final Bytes32 startKeyHash, final int limit) {
throw new RuntimeException("Bonsai Tries does not currently support enumerating storage");
Expand Down
17 changes: 17 additions & 0 deletions ethereum/core/src/main/java/org/hyperledger/besu/ethereum/worldstate/WorldStateStorage.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,23 @@ default NavigableMap<Bytes32, Bytes> streamFlatStorages(
return new TreeMap<>();
}

/**
* Streams flat storages within a specified range.
*
* @param accountHash The account hash.
* @param startKeyHash The start key hash of the range.
* @param endKeyHash The end key hash of the range.
* @param takeWhile predicate to limit storage streaming
* @return A map of flat storages. (Empty map in this default implementation)
*/
default NavigableMap<Bytes32, Bytes> streamFlatStorages(
final Hash accountHash,
final Bytes startKeyHash,
final Bytes32 endKeyHash,
final Predicate<Pair<Bytes32, Bytes>> takeWhile) {
return new TreeMap<>();
}

DataStorageFormat getDataStorageFormat();

void clear();
Expand Down
138 changes: 85 additions & 53 deletions ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/snap/SnapServer.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,12 @@
import org.hyperledger.besu.ethereum.worldstate.WorldStateArchive;
import org.hyperledger.besu.ethereum.worldstate.WorldStateStorage;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;
Expand Down Expand Up @@ -105,7 +107,7 @@ MessageData constructGetAccountRangeResponse(final MessageData message) {
storage.streamFlatAccounts(
range.startKeyHash(),
range.endKeyHash(),
takeWhilePredicate(maxResponseBytes));
new StatefulPredicate(maxResponseBytes));

if (accounts.isEmpty()) {
// fetch next account after range, if it exists
Expand All @@ -127,7 +129,7 @@ MessageData constructGetAccountRangeResponse(final MessageData message) {
.orElse(EMPTY_ACCOUNT_RANGE);
}

private MessageData constructGetStorageRangeResponse(final MessageData message) {
MessageData constructGetStorageRangeResponse(final MessageData message) {
final GetStorageRangeMessage getStorageRangeMessage = GetStorageRangeMessage.readFrom(message);
final GetStorageRangeMessage.StorageRange range = getStorageRangeMessage.range(true);
final int maxResponseBytes = Math.min(range.responseBytes().intValue(), MAX_RESPONSE_SIZE);
Expand All @@ -145,43 +147,52 @@ private MessageData constructGetStorageRangeResponse(final MessageData message)
.collect(Collectors.joining(",", "[", "]")))
.log();

return EMPTY_STORAGE_RANGE;
// return worldStateStorageProvider
// .apply(range.worldStateRootHash())
// .map(
// storage -> {
// NavigableMap<Bytes32, Bytes> accounts =
// storage.streamFlatAccounts(
// range.startKeyHash(), range.endKeyHash(), MAX_ENTRIES_PER_REQUEST);
//
// // for the first account, honor startHash
// Bytes32 startKeyBytes = range.startKeyHash();
// Bytes32 endKeyBytes = range.endKeyHash();
// NavigableMap<Bytes32, Bytes> collectedStorages = new TreeMap<>();
// ArrayList<Bytes> collectedProofs = new ArrayList<>();
// for (var forAccountHash : range.hashes()) {
// var accountStorages =
// storage.streamFlatStorages(
// Hash.wrap(forAccountHash),
// startKeyBytes,
// endKeyBytes,
// MAX_ENTRIES_PER_REQUEST);
// boolean shouldGetMore = false;
//// visitCollectedStorage(collectedStorages, collectedProofs,
// accountStorages, maxResponseBytes);
// // todo proofs for this accountHash
//
// if (shouldGetMore) {
// // reset startkeyBytes for subsequent accounts
// startKeyBytes = Bytes32.ZERO;
// } else {
// break;
// }
// }
//
// return StorageRangeMessage.create(new ArrayDeque<>(), Collections.emptyList());
// })
// .orElse(EMPTY_STORAGE_RANGE);
return worldStateStorageProvider
.apply(range.worldStateRootHash())
.map(
storage -> {
// reusable predicate to limit by rec count and bytes:
var statefulPredicate = new StatefulPredicate(maxResponseBytes);
// for the first account, honor startHash
Bytes32 startKeyBytes = range.startKeyHash();
Bytes32 endKeyBytes = range.endKeyHash();

ArrayDeque<NavigableMap<Bytes32, Bytes>> collectedStorages = new ArrayDeque<>();
ArrayList<Bytes> collectedProofs = new ArrayList<>();
final var worldStateProof = new WorldStateProofProvider(storage);

for (var forAccountHash : range.hashes()) {
// start key proof

// collectedProofs.addAll(
// worldStateProof.getStorageProofRelatedNodes(
// /* TODO: need account root hash here, not worldstate root hash */
// range.worldStateRootHash(), forAccountHash, Hash.wrap(startKeyBytes)));

var accountStorages =
storage.streamFlatStorages(
Hash.wrap(forAccountHash), startKeyBytes, endKeyBytes, statefulPredicate);
collectedStorages.add(accountStorages);

// last key proof

// collectedProofs.addAll(
// worldStateProof.getStorageProofRelatedNodes(
// /* TODO: need account root hash here, not worldstate root hash */
// range.worldStateRootHash(), forAccountHash,
// Hash.wrap(accountStorages.lastKey())));

if (statefulPredicate.shouldGetMore()) {
// reset startkeyBytes for subsequent accounts
startKeyBytes = Bytes32.ZERO;
} else {
break;
}
}

return StorageRangeMessage.create(collectedStorages, collectedProofs);
})
.orElse(EMPTY_STORAGE_RANGE);
}

private MessageData constructGetBytecodesResponse(final MessageData message) {
Expand All @@ -194,21 +205,42 @@ private MessageData constructGetTrieNodesResponse(final MessageData message) {
return TrieNodesMessage.create(new ArrayDeque<>());
}

private static Predicate<Pair<Bytes32, Bytes>> takeWhilePredicate(final int maxResponseBytes) {
static class StatefulPredicate implements Predicate<Pair<Bytes32, Bytes>> {

final AtomicInteger byteLimit = new AtomicInteger(0);
final AtomicInteger recordLimit = new AtomicInteger(0);
return pair ->
recordLimit.addAndGet(1) < MAX_ENTRIES_PER_REQUEST
&& byteLimit.accumulateAndGet(
0,
(cur, __) -> {
var rlpOutput = new BytesValueRLPOutput();
rlpOutput.startList();
rlpOutput.writeBytes(pair.getFirst());
rlpOutput.writeRLPBytes(pair.getSecond());
rlpOutput.endList();
return cur + rlpOutput.encoded().size();
})
< maxResponseBytes;
final AtomicBoolean shouldContinue = new AtomicBoolean(true);
final int maxResponseBytes;

StatefulPredicate(final int maxResponseBytes) {
this.maxResponseBytes = maxResponseBytes;
}

public boolean shouldGetMore() {
return shouldContinue.get();
}

@Override
public boolean test(final Pair<Bytes32, Bytes> pair) {
var underRecordLimit = recordLimit.addAndGet(1) < MAX_ENTRIES_PER_REQUEST;
var underByteLimit =
byteLimit.accumulateAndGet(
0,
(cur, __) -> {
var rlpOutput = new BytesValueRLPOutput();
rlpOutput.startList();
rlpOutput.writeBytes(pair.getFirst());
rlpOutput.writeRLPBytes(pair.getSecond());
rlpOutput.endList();
return cur + rlpOutput.encoded().size();
})
< maxResponseBytes;
if (underRecordLimit && underByteLimit) {
return true;
} else {
shouldContinue.set(false);
return false;
}
}
}
}
9 changes: 5 additions & 4 deletions ...th/src/main/java/org/hyperledger/besu/ethereum/eth/messages/snap/StorageRangeMessage.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

import java.math.BigInteger;
import java.util.List;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.TreeMap;

Expand Down Expand Up @@ -49,13 +50,13 @@ public static StorageRangeMessage readFrom(final MessageData message) {
}

public static StorageRangeMessage create(
final ArrayDeque<TreeMap<Bytes32, Bytes>> slots, final List<Bytes> proof) {
final ArrayDeque<NavigableMap<Bytes32, Bytes>> slots, final List<Bytes> proof) {
return create(Optional.empty(), slots, proof);
}

public static StorageRangeMessage create(
final Optional<BigInteger> requestId,
final ArrayDeque<TreeMap<Bytes32, Bytes>> slots,
final ArrayDeque<NavigableMap<Bytes32, Bytes>> slots,
final List<Bytes> proof) {
final BytesValueRLPOutput tmp = new BytesValueRLPOutput();
tmp.startList();
Expand Down Expand Up @@ -88,7 +89,7 @@ public int getCode() {
}

public SlotRangeData slotsData(final boolean withRequestId) {
final ArrayDeque<TreeMap<Bytes32, Bytes>> slots = new ArrayDeque<>();
final ArrayDeque<NavigableMap<Bytes32, Bytes>> slots = new ArrayDeque<>();
final ArrayDeque<Bytes> proofs = new ArrayDeque<>();
final RLPInput input = new BytesValueRLPInput(data, false);
input.enterList();
Expand Down Expand Up @@ -120,7 +121,7 @@ public SlotRangeData slotsData(final boolean withRequestId) {
@Value.Immutable
public interface SlotRangeData {

ArrayDeque<TreeMap<Bytes32, Bytes>> slots();
ArrayDeque<NavigableMap<Bytes32, Bytes>> slots();

ArrayDeque<Bytes> proofs();
}
Expand Down
2 changes: 1 addition & 1 deletion ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/RangeManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ public static Map<Bytes32, Bytes32> generateRanges(
public static Optional<Bytes32> findNewBeginElementInRange(
final Bytes32 worldstateRootHash,
final List<Bytes> proofs,
final TreeMap<Bytes32, Bytes> receivedKeys,
final NavigableMap<Bytes32, Bytes> receivedKeys,
final Bytes32 endKeyHash) {
if (receivedKeys.isEmpty() || receivedKeys.lastKey().compareTo(endKeyHash) >= 0) {
return Optional.empty();
Expand Down
7 changes: 5 additions & 2 deletions ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/sync/snapsync/StackTrie.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -72,7 +73,9 @@ public StackTrie(
}

public void addElement(
final Bytes32 taskIdentifier, final List<Bytes> proofs, final TreeMap<Bytes32, Bytes> keys) {
final Bytes32 taskIdentifier,
final List<Bytes> proofs,
final NavigableMap<Bytes32, Bytes> keys) {
this.elementsCount.addAndGet(keys.size());
this.elements.put(
taskIdentifier, ImmutableTaskElement.builder().proofs(proofs).keys(keys).build());
Expand Down Expand Up @@ -169,7 +172,7 @@ public List<Bytes> proofs() {
}

@Value.Default
public TreeMap<Bytes32, Bytes> keys() {
public NavigableMap<Bytes32, Bytes> keys() {
return new TreeMap<>();
}
}
Expand Down
3 changes: 2 additions & 1 deletion ...java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/AccountRangeDataRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -212,7 +213,7 @@ public Bytes32 getEndKeyHash() {
}

@VisibleForTesting
public TreeMap<Bytes32, Bytes> getAccounts() {
public NavigableMap<Bytes32, Bytes> getAccounts() {
return stackTrie.getElement(startKeyHash).keys();
}

Expand Down
6 changes: 3 additions & 3 deletions ...java/org/hyperledger/besu/ethereum/eth/sync/snapsync/request/StorageRangeDataRequest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,8 @@

import java.util.ArrayList;
import java.util.List;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

Expand Down Expand Up @@ -119,7 +119,7 @@ protected int doPersist(
public void addResponse(
final SnapWorldDownloadState downloadState,
final WorldStateProofProvider worldStateProofProvider,
final TreeMap<Bytes32, Bytes> slots,
final NavigableMap<Bytes32, Bytes> slots,
final ArrayDeque<Bytes> proofs) {
if (!slots.isEmpty() || !proofs.isEmpty()) {
if (!worldStateProofProvider.isValidRangeProof(
Expand Down Expand Up @@ -193,7 +193,7 @@ public Bytes32 getStorageRoot() {
return storageRoot;
}

public TreeMap<Bytes32, Bytes> getSlots() {
public NavigableMap<Bytes32, Bytes> getSlots() {
return stackTrie.getElement(startKeyHash).keys();
}

Expand Down
Loading

0 comments on commit 30dfc08

Please sign in to comment.