Skip to content

Commit

Permalink
Cherry pick fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
@JamesSmartCell
JamesSmartCell committed Dec 7, 2022
1 parent bd08254 commit cda91d3
Show file tree
Hide file tree
Showing 13 changed files with 371 additions and 167 deletions.
102 changes: 96 additions & 6 deletions app/src/main/java/com/alphawallet/app/entity/EventSync.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,10 @@
public class EventSync
{
public static final long BLOCK_SEARCH_INTERVAL = 100000L;
public static final long POLYGON_BLOCK_SEARCH_INTERVAL = 10000L;

private static final String TAG = "EVENT_SYNC";
private static final boolean EVENT_SYNC_DEBUGGING = false;

private final Token token;

Expand Down Expand Up @@ -71,8 +75,17 @@ public SyncDef getSyncDef(Realm realm)
case DOWNWARD_SYNC_START: //Start event sync, optimistically try the whole current event range from 1 -> LATEST
eventReadStartBlock = BigInteger.ONE;
eventReadEndBlock = BigInteger.valueOf(-1L);
//write the start point here
writeStartSyncBlock(realm, currentBlock.longValue());
if (EthereumNetworkBase.isEventBlockLimitEnforced(token.tokenInfo.chainId))
{
syncState = EventSyncState.UPWARD_SYNC;
eventReadStartBlock = currentBlock.subtract(EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).multiply(BigInteger.valueOf(3)));
EVENT_DEBUG("Init Sync for restricted block RPC");
}
else
{
//write the start point here
writeStartSyncBlock(realm, currentBlock.longValue());
}
break;
case DOWNWARD_SYNC: //we needed to slow down the sync
eventReadStartBlock = lastBlockRead.subtract(BigInteger.valueOf(readBlockSize));
Expand All @@ -85,19 +98,85 @@ public SyncDef getSyncDef(Realm realm)
break;
case UPWARD_SYNC_MAX: //we are syncing from the point we started the downward sync
upwardSync = true;
if (EthereumNetworkBase.isEventBlockLimitEnforced(token.tokenInfo.chainId) && upwardSyncStateLost(lastBlockRead, currentBlock))
{
syncState = EventSyncState.UPWARD_SYNC;
EVENT_DEBUG("Switch back to sync scan");
}

eventReadStartBlock = lastBlockRead;
eventReadEndBlock = BigInteger.valueOf(-1L);
break;
case UPWARD_SYNC: //we encountered upward sync issues
upwardSync = true;
eventReadStartBlock = lastBlockRead;
eventReadEndBlock = lastBlockRead.add(BigInteger.valueOf(readBlockSize));
if (upwardSyncComplete(eventReadStartBlock, currentBlock)) //detect completion of upward sync and switch to sync_max
{
eventReadEndBlock = BigInteger.valueOf(-1L);
syncState = EventSyncState.UPWARD_SYNC_MAX;
EVENT_DEBUG("Sync complete");
}
else
{
eventReadEndBlock = lastBlockRead.add(BigInteger.valueOf(readBlockSize));
}
break;
}

// Finally adjust the event end read if required. This is placed outside the switch because it should affect
// a few different paths
eventReadEndBlock = adjustForLimitedBlockSize(eventReadStartBlock, eventReadEndBlock, currentBlock);

// detect edge condition - it's highly unlikely but acts as a stopper in case of unexpected results
// This edge condition is where the start block read is greater than the current block.
if (eventReadStartBlock.compareTo(currentBlock) >= 0)
{
eventReadStartBlock = currentBlock.subtract(BigInteger.ONE);
eventReadEndBlock = BigInteger.valueOf(-1L);
syncState = EventSyncState.UPWARD_SYNC_MAX;
}

return new SyncDef(eventReadStartBlock, eventReadEndBlock, syncState, upwardSync);
}

private void EVENT_DEBUG(String message)
{
if (EVENT_SYNC_DEBUGGING)
{
Timber.tag(TAG).i(token.tokenInfo.chainId + " " + token.tokenInfo.address + ": " + message);
}
}

private boolean upwardSyncStateLost(BigInteger lastBlockRead, BigInteger currentBlock)
{
return currentBlock.subtract(lastBlockRead).compareTo(EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId)) >= 0;
}

private boolean upwardSyncComplete(BigInteger eventReadStartBlock, BigInteger currentBlock)
{
BigInteger maxBlockRead = EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).subtract(BigInteger.ONE);
BigInteger diff = currentBlock.subtract(eventReadStartBlock);

return diff.compareTo(maxBlockRead) < 0;
}

private BigInteger adjustForLimitedBlockSize(BigInteger eventReadStartBlock, BigInteger eventReadEndBlock, BigInteger currentBlock)
{
if (EthereumNetworkBase.isEventBlockLimitEnforced(token.tokenInfo.chainId))
{
BigInteger maxBlockRead = EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId);

long diff = currentBlock.subtract(eventReadStartBlock).longValue();

if (diff >= maxBlockRead.longValue())
{
return eventReadStartBlock.add(maxBlockRead).subtract(BigInteger.ONE);
}
}

return eventReadEndBlock;
}

public boolean handleEthLogError(Response.Error error, DefaultBlockParameter startBlock, DefaultBlockParameter endBlock, SyncDef sync, Realm realm)
{
if (error.getCode() == -32005)
Expand Down Expand Up @@ -177,6 +256,11 @@ private long reduceBlockSearch(long currentBlock, BigInteger startBlock)

private long getCurrentEventBlockSize(Realm instance)
{
if (EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).equals(BigInteger.valueOf(3500L)))
{
return EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).longValue();
}

RealmAuxData rd = instance.where(RealmAuxData.class)
.equalTo("instanceKey", TokensRealmSource.databaseKey(token.tokenInfo.chainId, token.getAddress()))
.findFirst();
Expand Down Expand Up @@ -205,8 +289,9 @@ protected EventSyncState getCurrentTokenSyncState(Realm instance)
else
{
int state = rd.getTokenId().intValue();
if (state >= EventSyncState.DOWNWARD_SYNC_START.ordinal() || state < EventSyncState.TOP_LIMIT.ordinal())
if (state >= EventSyncState.DOWNWARD_SYNC_START.ordinal() && state < EventSyncState.TOP_LIMIT.ordinal())
{
EVENT_DEBUG("Read State: " + EventSyncState.values()[state]);
return EventSyncState.values()[state];
}
else
Expand Down Expand Up @@ -248,6 +333,7 @@ protected long getLastEventRead(Realm instance)
}
else
{
EVENT_DEBUG("ReadEventSync: " + rd.getResultTime());
return rd.getResultTime();
}
}
Expand Down Expand Up @@ -333,14 +419,16 @@ private void updateEventReads(Realm realm, long lastRead, long readInterval, Eve
rd.setResultReceivedTime(readInterval);
rd.setTokenId(String.valueOf(state.ordinal()));

EVENT_DEBUG("WriteState: " + state + " " + lastRead);

r.insertOrUpdate(rd);
});
}

// If we're syncing downwards, work out what event block size we should read next
private long calcNewIntervalSize(SyncDef sync, int evReads)
{
if (sync.upwardSync) return BLOCK_SEARCH_INTERVAL;
if (sync.upwardSync) return EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).longValue();
long endBlock = sync.eventReadEndBlock.longValue() == -1 ? TransactionsService.getCurrentBlock(token.tokenInfo.chainId).longValue()
: sync.eventReadEndBlock.longValue();
long currentReadSize = endBlock - sync.eventReadStartBlock.longValue();
Expand All @@ -357,14 +445,16 @@ else if (evReads < 1000)
}
else if ((maxLogReads - evReads) > maxLogReads*0.25)
{
currentReadSize += BLOCK_SEARCH_INTERVAL;
currentReadSize += EthereumNetworkBase.getMaxEventFetch(token.tokenInfo.chainId).longValue();
}

return currentReadSize;
}

/***
* Event Handling
*
* TODO: batch up catch-up calls
*/

public Pair<Integer, Pair<HashSet<BigInteger>, HashSet<BigInteger>>> processTransferEvents(Web3j web3j, Event transferEvent, DefaultBlockParameter startBlock,
Expand Down
40 changes: 26 additions & 14 deletions app/src/main/java/com/alphawallet/app/entity/tokens/ERC1155Token.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -276,27 +276,39 @@ public Function getTransferFunction(String to, List<BigInteger> tokenIds) throws
}

@Override
public List<BigInteger> getChangeList(Map<BigInteger, NFTAsset> assetMap)
public Map<BigInteger, NFTAsset> getAssetChange(Map<BigInteger, NFTAsset> oldAssetList)
{
//detect asset removal
List<BigInteger> oldAssetIdList = new ArrayList<>(assetMap.keySet());
oldAssetIdList.removeAll(assets.keySet());
//first see if there's no change; if this is the case we can skip
if (assetsUnchanged(oldAssetList)) return assets;

List<BigInteger> changeList = new ArrayList<>(oldAssetIdList);
//add all known tokens in
Map<BigInteger, NFTAsset> sum = new HashMap<>(oldAssetList);
sum.putAll(assets);
Set<BigInteger> tokenIds = sum.keySet();
Function balanceOfBatch = balanceOfBatch(getWallet(), tokenIds);
List<Uint256> balances = callSmartContractFunctionArray(tokenInfo.chainId, balanceOfBatch, getAddress(), getWallet());
Map<BigInteger, NFTAsset> updatedAssetMap;

//Now detect differences or new tokens
for (BigInteger tokenId : assets.keySet())
if (balances != null && balances.size() > 0)
{
NFTAsset newAsset = assets.get(tokenId);
NFTAsset oldAsset = assetMap.get(tokenId);

if (oldAsset == null || newAsset.hashCode() != oldAsset.hashCode())
updatedAssetMap = new HashMap<>();
int index = 0;
for (BigInteger tokenId : tokenIds)
{
changeList.add(tokenId);
NFTAsset thisAsset = new NFTAsset(sum.get(tokenId));
BigInteger balance = balances.get(index).getValue();
thisAsset.setBalance(new BigDecimal(balance));
updatedAssetMap.put(tokenId, thisAsset);

index++;
}
}
else
{
updatedAssetMap = assets;
}

return changeList;
return updatedAssetMap;
}

private List<Uint256> fetchBalances(Set<BigInteger> tokenIds)
Expand Down Expand Up @@ -643,7 +655,7 @@ public BigDecimal updateBalance(Realm realm)

try
{
final Web3j web3j = TokenRepository.getWeb3jService(tokenInfo.chainId);
final Web3j web3j = TokenRepository.getWeb3jServiceForEvents(tokenInfo.chainId);

Pair<Integer, Pair<HashSet<BigInteger>, HashSet<BigInteger>>> evRead = eventSync.processTransferEvents(web3j,
getBalanceUpdateEvents(), startBlock, endBlock, realm);
Expand Down
Loading

0 comments on commit cda91d3

Please sign in to comment.