diff --git a/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/ElasticSearchHandler.java b/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/ElasticSearchHandler.java index f786325..e462680 100644 --- a/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/ElasticSearchHandler.java +++ b/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/ElasticSearchHandler.java @@ -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; @@ -147,8 +148,11 @@ public BlockOffset getRemoteOffset() { Object meta = data.getSourceAsMap().get("_meta"); if (meta != null) { Map tip = (Map) ((LinkedHashMap) 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) { @@ -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 { diff --git a/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/IndexerHandle.java b/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/IndexerHandle.java index ea80f4a..c2cca6a 100644 --- a/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/IndexerHandle.java +++ b/starcoin-indexer/src/main/java/org/starcoin/indexer/handler/IndexerHandle.java @@ -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; @@ -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) { @@ -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!!"); @@ -180,4 +187,66 @@ protected void executeInternal(JobExecutionContext jobExecutionContext) { logger.error("chain header error:", e); } } -} \ No newline at end of file + + 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); + } +} diff --git a/starcoin-indexer/src/test/java/org/starcoin/indexer/handler/IndexerHandleTest.java b/starcoin-indexer/src/test/java/org/starcoin/indexer/handler/IndexerHandleTest.java index 08c2648..9e0c66b 100644 --- a/starcoin-indexer/src/test/java/org/starcoin/indexer/handler/IndexerHandleTest.java +++ b/starcoin-indexer/src/test/java/org/starcoin/indexer/handler/IndexerHandleTest.java @@ -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(); @@ -245,4 +233,27 @@ public void testNonMainNetworkNullOffsetHandling() throws JSONRPC2SessionExcepti verify(blockRPCClient, times(1)).getBlockByHeight(0L); verify(elasticSearchHandler, times(1)).setRemoteOffset(any(BlockOffset.class)); } -} \ No newline at end of file + + @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); + } +}