From 6378f15ad0a442fe616141d0b15abb69f9c99719 Mon Sep 17 00:00:00 2001 From: thinkAfCod Date: Fri, 2 Feb 2024 23:23:05 +0800 Subject: [PATCH] fix: miss tx --- .../main/java/io/optimism/derive/State.java | 31 +++++++++- .../io/optimism/derive/stages/Batches.java | 59 ++++++++++++------- .../main/java/io/optimism/driver/Driver.java | 27 ++++++++- 3 files changed, 92 insertions(+), 25 deletions(-) diff --git a/hildr-node/src/main/java/io/optimism/derive/State.java b/hildr-node/src/main/java/io/optimism/derive/State.java index 07a64423..3c43687c 100644 --- a/hildr-node/src/main/java/io/optimism/derive/State.java +++ b/hildr-node/src/main/java/io/optimism/derive/State.java @@ -10,7 +10,10 @@ import java.util.TreeMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.StructuredTaskScope; +import java.util.function.Function; import org.apache.commons.lang3.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.web3j.protocol.Web3j; import org.web3j.protocol.core.DefaultBlockParameter; import org.web3j.protocol.core.methods.response.EthBlock; @@ -24,12 +27,16 @@ */ public class State { + private static final Logger LOGGER = LoggerFactory.getLogger(State.class); + private final TreeMap l1Info; private final TreeMap l1Hashes; private final TreeMap> l2Refs; + private final Function> l2Fetcher; + private BlockInfo safeHead; private Epoch safeEpoch; @@ -44,6 +51,7 @@ public class State { * @param l1Info the L1 info * @param l1Hashes the L1 hashes * @param l2Refs the L2 block info references + * @param l2Fetcher the L2 block info fetcher * @param safeHead the safe head * @param safeEpoch the safe epoch * @param currentEpochNum the current epoch num @@ -53,6 +61,7 @@ public State( TreeMap l1Info, TreeMap l1Hashes, TreeMap> l2Refs, + Function> l2Fetcher, BlockInfo safeHead, Epoch safeEpoch, BigInteger currentEpochNum, @@ -60,6 +69,7 @@ public State( this.l1Info = l1Info; this.l1Hashes = l1Hashes; this.l2Refs = l2Refs; + this.l2Fetcher = l2Fetcher; this.safeHead = safeHead; this.safeEpoch = safeEpoch; this.currentEpochNum = currentEpochNum; @@ -70,6 +80,7 @@ public State( * Create state. * * @param l2Refs the L2 block info references + * @param l2Fetcher the L2 block info fetcher * @param finalizedHead the finalized head * @param finalizedEpoch the finalized epoch * @param config the config @@ -77,11 +88,19 @@ public State( */ public static State create( TreeMap> l2Refs, + Function> l2Fetcher, BlockInfo finalizedHead, Epoch finalizedEpoch, Config config) { return new State( - new TreeMap<>(), new TreeMap<>(), l2Refs, finalizedHead, finalizedEpoch, BigInteger.ZERO, config); + new TreeMap<>(), + new TreeMap<>(), + l2Refs, + l2Fetcher, + finalizedHead, + finalizedEpoch, + BigInteger.ZERO, + config); } /** @@ -119,7 +138,14 @@ public Tuple2 l2Info(BigInteger timestamp) { .subtract(config.chainConfig().l2Genesis().timestamp()) .divide(config.chainConfig().blockTime()) .add(config.chainConfig().l2Genesis().number()); - return this.l2Refs.get(blockNum); + var cache = l2Refs.get(blockNum); + if (cache != null) { + return cache; + } + + var res = l2Fetcher.apply(blockNum); + this.l2Refs.put(res.component1().number(), res); + return res; } /** @@ -173,6 +199,7 @@ public void updateL1Info(L1Info l1Info) { * @param safeEpoch the safe epoch */ public void purge(BlockInfo safeHead, Epoch safeEpoch) { + LOGGER.info("purge state: safeHead.number={}, safeEpoch. ={}", safeHead.number(), safeEpoch.hash()); this.safeHead = safeHead; this.safeEpoch = safeEpoch; this.l1Info.clear(); diff --git a/hildr-node/src/main/java/io/optimism/derive/stages/Batches.java b/hildr-node/src/main/java/io/optimism/derive/stages/Batches.java index 7a81a46e..10448c32 100644 --- a/hildr-node/src/main/java/io/optimism/derive/stages/Batches.java +++ b/hildr-node/src/main/java/io/optimism/derive/stages/Batches.java @@ -75,6 +75,9 @@ public Batches(TreeMap batches, I channelIterator, AtomicRefe public void purge() { this.channelIterator.purge(); this.batches.clear(); + if (!this.nextSingularBatches.isEmpty()) { + LOGGER.warn("batches has element but will be discarded"); + } this.nextSingularBatches.clear(); } @@ -86,8 +89,13 @@ public Batch next() { } Channel channel = this.channelIterator.next(); if (channel != null) { - decodeBatches(this.config.chainConfig(), channel) - .forEach(batch -> this.batches.put(batch.batch().getTimestamp(), batch)); + decodeBatches(this.config.chainConfig(), channel).forEach(batch -> { + Batch prev = this.batches.put(batch.batch().getTimestamp(), batch); + if (prev != null) { + LOGGER.warn( + "batch was replaced: timestamp={}", batch.batch().getTimestamp()); + } + }); } Batch derivedBatch = null; @@ -121,24 +129,28 @@ public Batch next() { this.nextSingularBatches.addAll(singularBatches); return this.nextSingularBatches.poll(); } - } - - State state = this.state.get(); - - BigInteger currentL1Block = state.getCurrentEpochNum(); - BlockInfo safeHead = state.getSafeHead(); - Epoch epoch = state.getSafeEpoch(); - Epoch nextEpoch = state.epoch(epoch.number().add(BigInteger.ONE)); - BigInteger seqWindowSize = this.config.chainConfig().seqWindowSize(); - - if (nextEpoch != null) { - if (currentL1Block.compareTo(epoch.number().add(seqWindowSize)) > 0) { - BigInteger nextTimestamp = - safeHead.timestamp().add(this.config.chainConfig().blockTime()); - Epoch epochRes = nextTimestamp.compareTo(nextEpoch.timestamp()) < 0 ? epoch : nextEpoch; - var singularBatch = new SingularBatch( - safeHead.parentHash(), epochRes.number(), epochRes.hash(), nextTimestamp, Lists.newArrayList()); - batch = new Batch(singularBatch, currentL1Block); + } else { + State state = this.state.get(); + + BigInteger currentL1Block = state.getCurrentEpochNum(); + BlockInfo safeHead = state.getSafeHead(); + Epoch epoch = state.getSafeEpoch(); + Epoch nextEpoch = state.epoch(epoch.number().add(BigInteger.ONE)); + BigInteger seqWindowSize = this.config.chainConfig().seqWindowSize(); + + if (nextEpoch != null) { + if (currentL1Block.compareTo(epoch.number().add(seqWindowSize)) > 0) { + BigInteger nextTimestamp = + safeHead.timestamp().add(this.config.chainConfig().blockTime()); + Epoch epochRes = nextTimestamp.compareTo(nextEpoch.timestamp()) < 0 ? epoch : nextEpoch; + var singularBatch = new SingularBatch( + safeHead.parentHash(), + epochRes.number(), + epochRes.hash(), + nextTimestamp, + Lists.newArrayList()); + batch = new Batch(singularBatch, currentL1Block); + } } } return batch; @@ -147,6 +159,7 @@ public Batch next() { /** * Decode batches list. * + * @param chainConfig the chain config * @param channel the channel * @return the list */ @@ -485,6 +498,12 @@ private List toSingularBatches(final SpanBatch batch, final State List singularBatches = new ArrayList<>(); for (SpanBatchElement element : batch.getBatches()) { if (element.timestamp().compareTo(state.getSafeHead().timestamp()) <= 0) { + if (!element.transactions().isEmpty()) { + LOGGER.warn( + "past span batch element: timestamp{{}} <= safeHead.timestamp={{}}", + element.timestamp(), + state.getSafeHead().timestamp()); + } continue; } SingularBatch singularBatch = new SingularBatch(); diff --git a/hildr-node/src/main/java/io/optimism/driver/Driver.java b/hildr-node/src/main/java/io/optimism/driver/Driver.java index f7c6540b..27c6f54e 100644 --- a/hildr-node/src/main/java/io/optimism/driver/Driver.java +++ b/hildr-node/src/main/java/io/optimism/driver/Driver.java @@ -55,7 +55,9 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.web3j.protocol.Web3j; +import org.web3j.protocol.core.DefaultBlockParameter; import org.web3j.protocol.core.methods.response.EthBlock; +import org.web3j.tuples.generated.Tuple2; import org.web3j.utils.Numeric; /** @@ -166,7 +168,7 @@ public EngineDriver getEngineDriver() { */ public static Driver from(Config config, CountDownLatch latch) throws InterruptedException, ExecutionException { - Web3j provider = Web3jProvider.createClient(config.l2RpcUrl()); + final Web3j provider = Web3jProvider.createClient(config.l2RpcUrl()); EthBlock finalizedBlock; try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { @@ -207,8 +209,27 @@ public static Driver from(Config config, CountDownLatch latch) config); var l2Refs = io.optimism.derive.State.initL2Refs(finalizedHead.number(), config.chainConfig(), provider); - AtomicReference state = - new AtomicReference<>(io.optimism.derive.State.create(l2Refs, finalizedHead, finalizedEpoch, config)); + var l2Fetcher = (Function>) blockNum -> { + try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { + StructuredTaskScope.Subtask blockTask = scope.fork(TracerTaskWrapper.wrap( + () -> provider.ethGetBlockByNumber(DefaultBlockParameter.valueOf(blockNum), true) + .send())); + scope.join(); + scope.throwIfFailed(); + + var block = blockTask.get(); + if (block == null) { + return null; + } + final HeadInfo l2BlockInfo = HeadInfo.from(block.getBlock()); + return new Tuple2<>(l2BlockInfo.l2BlockInfo(), l2BlockInfo.l1Epoch()); + } catch (Exception e) { + LOGGER.error("failed to fetch L2 block", e); + return null; + } + }; + AtomicReference state = new AtomicReference<>( + io.optimism.derive.State.create(l2Refs, l2Fetcher, finalizedHead, finalizedEpoch, config)); EngineDriver engineDriver = new EngineDriver<>(finalizedHead, finalizedEpoch, provider, config);