Skip to content

Fix stcscan offset recovery #75

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

Draft
sanlee42 wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Fix stcscan offset recovery #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
46 changes: 45 additions & 1 deletion starcoin-indexer/src/main/java/org/starcoin/indexer/handler/ElasticSearchHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.query.TermQueryBuilder;
Expand Down Expand Up @@ -147,8 +148,11 @@ public BlockOffset getRemoteOffset() {
Object meta = data.getSourceAsMap().get("_meta");
if (meta != null) {
Map<String, Object> tip = (Map<String, Object>) ((LinkedHashMap<String, Object>) meta).get("tip");
if (tip == null) {
return null;
}
String blockHash = tip.get("block_hash").toString();
Integer blockHeight = (Integer) tip.get("block_number");
Number blockHeight = (Number) tip.get("block_number");
return new BlockOffset(blockHeight.longValue(), blockHash);
}
} catch (Exception e) {
Expand All @@ -157,6 +161,46 @@ public BlockOffset getRemoteOffset() {
return null;
}

public BlockOffset getLatestIndexedBlockOffset() {
BlockOffset blockIdOffset = getLatestBlockOffset(blockIdsIndex, QueryBuilders.matchAllQuery());
if (blockIdOffset != null) {
return blockIdOffset;
}

BoolQueryBuilder contentQuery = QueryBuilders.boolQuery();
contentQuery.must(QueryBuilders.matchAllQuery());
contentQuery.mustNot(QueryBuilders.termQuery("deleted", true));
return getLatestBlockOffset(blockContentIndex, contentQuery);
}

private BlockOffset getLatestBlockOffset(String indexName, QueryBuilder queryBuilder) {
SearchRequest searchRequest = new SearchRequest(indexName);
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(queryBuilder);
searchSourceBuilder.size(1);
searchSourceBuilder.timeout(TimeValue.timeValueSeconds(5));
searchSourceBuilder.sort("header.number", SortOrder.DESC);
searchSourceBuilder.fetchSource(new String[]{"header.number", "header.block_hash"}, null);
searchRequest.source(searchSourceBuilder);

try {
SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
SearchHit[] hits = searchResponse.getHits().getHits();
if (hits.length == 0) {
return null;
}
Block block = JSON.parseObject(hits[0].getSourceAsString(), Block.class);
if (block == null || block.getHeader() == null || block.getHeader().getBlockHash() == null) {
logger.warn("latest indexed block source is invalid, index: {}, id: {}", indexName, hits[0].getId());
return null;
}
return new BlockOffset(block.getHeader().getHeight(), block.getHeader().getBlockHash());
} catch (Exception e) {
logger.error("get latest indexed block offset error, index: {}", indexName, e);
}
return null;
}

public void setRemoteOffset(BlockOffset blockOffset) {
PutMappingRequest request = new PutMappingRequest(blockContentIndex);
try {
Expand Down
75 changes: 72 additions & 3 deletions starcoin-indexer/src/main/java/org/starcoin/indexer/handler/IndexerHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

public class IndexerHandle extends QuartzJobBean {
private static final Logger logger = LoggerFactory.getLogger(IndexerHandle.class);
private static final long MIN_OFFSET_RECOVERY_GAP = 10_000L;

private BlockOffset localBlockOffset;
private BlockHeader currentHandleHeader;
Expand All @@ -42,7 +43,7 @@ public class IndexerHandle extends QuartzJobBean {

@PostConstruct
public void initOffset() throws JSONRPC2SessionException {
localBlockOffset = elasticSearchHandler.getRemoteOffset();
localBlockOffset = getRecoverableRemoteOffset();

// Case 1: Obtained
if (localBlockOffset != null) {
Expand Down Expand Up @@ -84,7 +85,13 @@ protected void executeInternal(JobExecutionContext jobExecutionContext) {
}
}

BlockOffset remoteBlockOffset = elasticSearchHandler.getRemoteOffset();
BlockOffset remoteBlockOffset;
try {
remoteBlockOffset = getRecoverableRemoteOffset();
} catch (JSONRPC2SessionException e) {
logger.error("get recoverable remote offset error:", e);
return;
}
logger.info("current remote offset: {}", remoteBlockOffset);
if (remoteBlockOffset == null) {
logger.warn("offset must not null, please check blocks.mapping!!");
Expand Down Expand Up @@ -180,4 +187,66 @@ protected void executeInternal(JobExecutionContext jobExecutionContext) {
logger.error("chain header error:", e);
}
}
}

private BlockOffset getRecoverableRemoteOffset() throws JSONRPC2SessionException {
BlockOffset remoteBlockOffset = elasticSearchHandler.getRemoteOffset();
if (remoteBlockOffset == null) {
return recoverRemoteOffset(remoteBlockOffset, "missing remote offset");
}

BlockHeader chainHeader;
try {
chainHeader = blockRPCClient.getChainHeader();
} catch (JsonProcessingException e) {
logger.warn("skip remote offset recovery, chain header parse error: remote={}", remoteBlockOffset, e);
return remoteBlockOffset;
}
if (chainHeader == null) {
logger.warn("skip remote offset recovery, chain header is null: remote={}", remoteBlockOffset);
return remoteBlockOffset;
}
long chainGap = chainHeader.getHeight() - remoteBlockOffset.getBlockHeight();
if (chainGap <= offsetRecoveryGap()) {
return remoteBlockOffset;
}

BlockOffset indexedOffset = elasticSearchHandler.getLatestIndexedBlockOffset();
if (indexedOffset == null) {
return remoteBlockOffset;
}
long indexedGap = indexedOffset.getBlockHeight() - remoteBlockOffset.getBlockHeight();
if (indexedGap > offsetRecoveryGap()) {
return recoverRemoteOffset(remoteBlockOffset, "stale remote offset");
}
return remoteBlockOffset;
}

private BlockOffset recoverRemoteOffset(BlockOffset remoteBlockOffset, String reason) throws JSONRPC2SessionException {
BlockOffset indexedOffset = elasticSearchHandler.getLatestIndexedBlockOffset();
if (indexedOffset == null) {
return remoteBlockOffset;
}

Block block = blockRPCClient.getBlockByHeight(indexedOffset.getBlockHeight());
if (block == null || block.getHeader() == null) {
logger.warn("skip remote offset recovery, indexed block not found on chain: remote={}, indexed={}",
remoteBlockOffset, indexedOffset);
return remoteBlockOffset;
}
String chainBlockHash = block.getHeader().getBlockHash();
if (!indexedOffset.getBlockHash().equals(chainBlockHash)) {
logger.warn("skip remote offset recovery, indexed block hash mismatch: remote={}, indexed={}, chainHash={}",
remoteBlockOffset, indexedOffset, chainBlockHash);
return remoteBlockOffset;
}

logger.warn("recover remote offset from indexed block, reason={}, remote={}, indexed={}",
reason, remoteBlockOffset, indexedOffset);
elasticSearchHandler.setRemoteOffset(indexedOffset);
return indexedOffset;
}

private long offsetRecoveryGap() {
return Math.max(MIN_OFFSET_RECOVERY_GAP, bulkSize * 100);
}
}
37 changes: 24 additions & 13 deletions starcoin-indexer/src/test/java/org/starcoin/indexer/handler/IndexerHandleTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,24 +130,12 @@ public void testForkHandling() throws JSONRPC2SessionException, JsonProcessingEx
forkHeader.setParentHash("parentHash");
forkBlock.setHeader(forkHeader);

// Create a master block that will be found during rollback
Block masterBlock = new Block();
BlockHeader masterHeader = new BlockHeader();
masterHeader.setHeight(99L);
masterHeader.setBlockHash("parentHash");
masterHeader.setParentHash("grandParentHash");
masterBlock.setHeader(masterHeader);

// Setup mock behavior
when(elasticSearchHandler.getRemoteOffset()).thenReturn(currentOffset);
when(blockRPCClient.getBlockByHeight(100L)).thenReturn(forkBlock);
when(blockRPCClient.getBlockByHeight(101L)).thenReturn(newBlock);
when(blockRPCClient.getBlockByHeight(99L)).thenReturn(masterBlock);
when(blockRPCClient.getChainHeader()).thenReturn(chainHeader);
when(elasticSearchHandler.getBlockContent(eq("oldHash"))).thenReturn(forkBlock);
when(elasticSearchHandler.getBlockContent(eq("parentHash"))).thenReturn(masterBlock);
when(blockRPCClient.getBlockByHash(eq("oldHash"))).thenReturn(forkBlock);
when(blockRPCClient.getBlockByHash(eq("parentHash"))).thenReturn(masterBlock);

// Execute
indexerHandle.initOffset();
Expand Down Expand Up @@ -245,4 +233,27 @@ public void testNonMainNetworkNullOffsetHandling() throws JSONRPC2SessionExcepti
verify(blockRPCClient, times(1)).getBlockByHeight(0L);
verify(elasticSearchHandler, times(1)).setRemoteOffset(any(BlockOffset.class));
}
}

@Test
public void testRecoveredOffsetDoesNotResetToGenesis() throws JSONRPC2SessionException {
BlockOffset recoveredOffset = new BlockOffset(31445103L, "recoveredHash");
BlockHeader recoveredHeader = new BlockHeader();
recoveredHeader.setHeight(31445103L);
recoveredHeader.setBlockHash("recoveredHash");

Block recoveredBlock = new Block();
recoveredBlock.setHeader(recoveredHeader);

when(elasticSearchHandler.getRemoteOffset()).thenReturn(null);
when(elasticSearchHandler.getLatestIndexedBlockOffset()).thenReturn(recoveredOffset);
when(blockRPCClient.getBlockByHeight(31445103L)).thenReturn(recoveredBlock);

indexerHandle.initOffset();

verify(elasticSearchHandler, times(1)).getRemoteOffset();
verify(elasticSearchHandler, times(1)).getLatestIndexedBlockOffset();
verify(blockRPCClient, times(2)).getBlockByHeight(31445103L);
verify(blockRPCClient, never()).getBlockByHeight(0L);
verify(elasticSearchHandler, times(1)).setRemoteOffset(recoveredOffset);
}
}
Loading