From e1622e100bfcc57085d458febd0cb3e983400988 Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 16 Sep 2026 15:23:57 +0800 Subject: [PATCH 1/2] feat(plugins): backfill section bloom Rebuild historical SectionBloom indexes from retained transaction results with engine detection, persisted-head bounds, and failure reporting. Share bloom encoding with the node while preserving the database format. Validate engine handling, index compatibility, reruns, and error paths in the consolidated backfill suite; document operation and known limitations. --- .../java/org/tron/common/bloom/Bloom.java | 58 +- .../org/tron/common/bloom/BloomUtils.java | 61 ++ .../org/tron/common/bloom/BloomUtilsTest.java | 129 ++++ plugins/README.md | 45 ++ .../main/java/common/org/tron/plugins/Db.java | 1 + .../org/tron/plugins/DbBackfillBloom.java | 573 +++++++++++++++++ .../org/tron/plugins/utils/db/DbTool.java | 2 +- .../org/tron/plugins/DbBackfillBloomTest.java | 597 ++++++++++++++++++ .../org/tron/plugins/utils/DbToolTest.java | 68 ++ 9 files changed, 1480 insertions(+), 54 deletions(-) create mode 100644 crypto/src/main/java/org/tron/common/bloom/BloomUtils.java create mode 100644 framework/src/test/java/org/tron/common/bloom/BloomUtilsTest.java create mode 100644 plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java create mode 100644 plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java create mode 100644 plugins/src/test/java/org/tron/plugins/utils/DbToolTest.java diff --git a/chainbase/src/main/java/org/tron/common/bloom/Bloom.java b/chainbase/src/main/java/org/tron/common/bloom/Bloom.java index 19d2bf53097..bca69577429 100644 --- a/chainbase/src/main/java/org/tron/common/bloom/Bloom.java +++ b/chainbase/src/main/java/org/tron/common/bloom/Bloom.java @@ -1,22 +1,14 @@ package org.tron.common.bloom; -import com.google.protobuf.ByteString; import java.util.Arrays; -import java.util.Iterator; -import org.tron.common.crypto.Hash; import org.tron.common.utils.ByteArray; -import org.tron.common.utils.ByteUtil; import org.tron.core.capsule.TransactionRetCapsule; -import org.tron.protos.Protocol.TransactionInfo; -import org.tron.protos.Protocol.TransactionInfo.Log; public class Bloom { - public static final int BLOOM_BIT_SIZE = 2048; - public static final int BLOOM_BYTE_SIZE = BLOOM_BIT_SIZE / 8; - private static final int STEPS_8 = 8; + public static final int BLOOM_BIT_SIZE = BloomUtils.BLOOM_BIT_SIZE; + public static final int BLOOM_BYTE_SIZE = BloomUtils.BLOOM_BYTE_SIZE; private static final int ENSURE_BYTE = 255; - private static final int LOW_3_BITS = getLowBits(BLOOM_BIT_SIZE); private byte[] data = new byte[BLOOM_BYTE_SIZE]; public Bloom() { @@ -37,55 +29,15 @@ public static int getLowBits(int bloomBitSize) { //only use first six byte public static Bloom create(byte[] toBloom) { - - int mov1 = - (((toBloom[0] & ENSURE_BYTE) & (LOW_3_BITS)) << STEPS_8) + ((toBloom[1]) & ENSURE_BYTE); - int mov2 = - (((toBloom[2] & ENSURE_BYTE) & (LOW_3_BITS)) << STEPS_8) + ((toBloom[3]) & ENSURE_BYTE); - int mov3 = - (((toBloom[4] & ENSURE_BYTE) & (LOW_3_BITS)) << STEPS_8) + ((toBloom[5]) & ENSURE_BYTE); - - byte[] data = new byte[BLOOM_BYTE_SIZE]; - Bloom bloom = new Bloom(data); - - ByteUtil.setBit(data, mov1, 1); - ByteUtil.setBit(data, mov2, 1); - ByteUtil.setBit(data, mov3, 1); - - return bloom; + return new Bloom(BloomUtils.create(toBloom)); } public static Bloom createBloom(TransactionRetCapsule transactionRetCapsule) { if (transactionRetCapsule == null) { return null; } - Iterator it = - transactionRetCapsule.getInstance().getTransactioninfoList().iterator(); - Bloom blockBloom = null; - - while (it.hasNext()) { - TransactionInfo transactionInfo = it.next(); - if (transactionInfo == null || transactionInfo.getLogCount() == 0) { - continue; - } - - if (blockBloom == null) { - blockBloom = new Bloom(); - } - - for (Log log : transactionInfo.getLogList()) { - //log.address doesn't have "41" ahead - Bloom bloom = Bloom.create(Hash.sha3(log.getAddress().toByteArray())); - blockBloom.or(bloom); - - for (ByteString topic : log.getTopicsList()) { - bloom = Bloom.create(Hash.sha3(topic.toByteArray())); - blockBloom.or(bloom); - } - } - } - - return blockBloom; + byte[] bloom = BloomUtils.createBloom(transactionRetCapsule.getInstance()); + return bloom == null ? null : new Bloom(bloom); } public void or(Bloom bloom) { diff --git a/crypto/src/main/java/org/tron/common/bloom/BloomUtils.java b/crypto/src/main/java/org/tron/common/bloom/BloomUtils.java new file mode 100644 index 00000000000..71f54cb3369 --- /dev/null +++ b/crypto/src/main/java/org/tron/common/bloom/BloomUtils.java @@ -0,0 +1,61 @@ +package org.tron.common.bloom; + +import com.google.protobuf.ByteString; +import org.tron.common.crypto.Hash; +import org.tron.common.utils.ByteUtil; +import org.tron.protos.Protocol.TransactionInfo; +import org.tron.protos.Protocol.TransactionInfo.Log; +import org.tron.protos.Protocol.TransactionRet; + +/** + * Shared log bloom encoding for the node and offline tools. + */ +public final class BloomUtils { + + public static final int BLOOM_BIT_SIZE = 2048; + public static final int BLOOM_BYTE_SIZE = BLOOM_BIT_SIZE / 8; + + private BloomUtils() { + } + + /** + * Creates a bloom from the first six bytes of a Keccak-256 hash. + */ + public static byte[] create(byte[] hash) { + byte[] bloom = new byte[BLOOM_BYTE_SIZE]; + setBits(bloom, hash); + return bloom; + } + + /** + * Returns the block's bloom, or null when there are no logs. + */ + public static byte[] createBloom(TransactionRet transactionRet) { + if (transactionRet == null) { + return null; + } + + byte[] bloom = null; + for (TransactionInfo transactionInfo : transactionRet.getTransactioninfoList()) { + for (Log log : transactionInfo.getLogList()) { + if (bloom == null) { + bloom = new byte[BLOOM_BYTE_SIZE]; + } + // Log addresses already omit the TRON address prefix (0x41). + setBits(bloom, Hash.sha3(log.getAddress().toByteArray())); + for (ByteString topic : log.getTopicsList()) { + setBits(bloom, Hash.sha3(topic.toByteArray())); + } + } + } + return bloom; + } + + private static void setBits(byte[] bloom, byte[] hash) { + //only use first six byte + for (int i = 0; i < 6; i += 2) { + int position = ((hash[i] & 0x07) << 8) | (hash[i + 1] & 0xff); + ByteUtil.setBit(bloom, position, 1); + } + } +} diff --git a/framework/src/test/java/org/tron/common/bloom/BloomUtilsTest.java b/framework/src/test/java/org/tron/common/bloom/BloomUtilsTest.java new file mode 100644 index 00000000000..4a7d08a04fe --- /dev/null +++ b/framework/src/test/java/org/tron/common/bloom/BloomUtilsTest.java @@ -0,0 +1,129 @@ +package org.tron.common.bloom; + +import com.google.protobuf.ByteString; +import java.math.BigInteger; +import java.util.Arrays; +import java.util.Random; +import org.bouncycastle.jcajce.provider.digest.Keccak; +import org.junit.Assert; +import org.junit.Test; +import org.tron.common.math.StrictMathWrapper; +import org.tron.core.capsule.TransactionRetCapsule; +import org.tron.protos.Protocol.TransactionInfo; +import org.tron.protos.Protocol.TransactionInfo.Log; +import org.tron.protos.Protocol.TransactionRet; + +public class BloomUtilsTest { + + @Test + public void testHashBitsUseElevenBitsAndBigEndianBytes() { + byte[] hash = new byte[] {(byte) 0xf8, 0, (byte) 0xf8, 8, (byte) 0xff, (byte) 0xff}; + byte[] originalHash = hash.clone(); + byte[] expected = new byte[256]; + expected[255] = 1; + expected[254] = 1; + expected[0] = (byte) 0x80; + + Assert.assertArrayEquals(expected, BloomUtils.create(hash)); + Assert.assertArrayEquals(expected, Bloom.create(hash).getData()); + Assert.assertArrayEquals(originalHash, hash); + } + + @Test + public void testCollidingBitsAndUnusedHashBytes() { + byte[] hash = new byte[32]; + Arrays.fill(hash, (byte) 0xff); + byte[] expected = new byte[256]; + expected[0] = (byte) 0x80; + Assert.assertArrayEquals(expected, BloomUtils.create(hash)); + + Arrays.fill(hash, 6, hash.length, (byte) 0); + Assert.assertArrayEquals(expected, BloomUtils.create(hash)); + } + + @Test + public void testAbsentLogsReturnNull() { + Assert.assertNull(BloomUtils.createBloom(null)); + Assert.assertNull(BloomUtils.createBloom(TransactionRet.getDefaultInstance())); + Assert.assertNull(BloomUtils.createBloom(TransactionRet.newBuilder() + .addTransactioninfo(TransactionInfo.getDefaultInstance()).build())); + Assert.assertNull(Bloom.createBloom(null)); + Assert.assertNull(Bloom.createBloom(new TransactionRetCapsule())); + } + + @Test + public void testMultipleTransactionsMatchIndependentEncoding() throws Exception { + Random random = new Random(81); + TransactionRet.Builder transactionRet = TransactionRet.newBuilder(); + for (int transaction = 0; transaction < 3; transaction++) { + TransactionInfo.Builder info = TransactionInfo.newBuilder(); + for (int logIndex = 0; logIndex < 4; logIndex++) { + Log.Builder log = Log.newBuilder().setAddress(randomBytes(random, 20)) + .setData(randomBytes(random, 64)); + for (int topic = 0; topic < logIndex; topic++) { + log.addTopics(randomBytes(random, 32)); + } + info.addLog(log); + } + transactionRet.addTransactioninfo(info); + } + TransactionRet input = transactionRet.build(); + byte[] expected = referenceBloom(input); + Assert.assertFalse(Arrays.equals(new byte[256], expected)); + Assert.assertArrayEquals(expected, BloomUtils.createBloom(input)); + Assert.assertArrayEquals(expected, + Bloom.createBloom(new TransactionRetCapsule(input.toByteArray())).getData()); + } + + @Test + public void testRepeatedLogsOrderAndLogDataDoNotChangeBloom() { + Random random = new Random(82); + Log first = Log.newBuilder().setAddress(randomBytes(random, 20)) + .addTopics(randomBytes(random, 32)).build(); + Log second = Log.newBuilder().setAddress(randomBytes(random, 20)) + .addTopics(randomBytes(random, 32)).build(); + TransactionRet original = TransactionRet.newBuilder().addTransactioninfo( + TransactionInfo.newBuilder().addLog(first).addLog(second)).build(); + TransactionRet repeated = TransactionRet.newBuilder().addTransactioninfo( + TransactionInfo.newBuilder().addLog(second).addLog(first) + .addLog(first.toBuilder().setData(randomBytes(random, 128)))) + .addTransactioninfo(TransactionInfo.getDefaultInstance()).build(); + + Assert.assertArrayEquals(referenceBloom(original), BloomUtils.createBloom(repeated)); + Assert.assertArrayEquals(BloomUtils.createBloom(original), BloomUtils.createBloom(repeated)); + } + + private ByteString randomBytes(Random random, int length) { + byte[] bytes = new byte[length]; + random.nextBytes(bytes); + return ByteString.copyFrom(bytes); + } + + private byte[] referenceBloom(TransactionRet transactionRet) { + // Use an independent digest and integer bit representation to catch encoding regressions. + BigInteger bits = BigInteger.ZERO; + for (TransactionInfo info : transactionRet.getTransactioninfoList()) { + for (Log log : info.getLogList()) { + bits = addReferenceBits(bits, log.getAddress()); + for (ByteString topic : log.getTopicsList()) { + bits = addReferenceBits(bits, topic); + } + } + } + byte[] integerBytes = bits.toByteArray(); + byte[] bloom = new byte[256]; + int length = StrictMathWrapper.min(integerBytes.length, bloom.length); + System.arraycopy(integerBytes, integerBytes.length - length, bloom, bloom.length - length, + length); + return bloom; + } + + private BigInteger addReferenceBits(BigInteger bits, ByteString value) { + byte[] hash = new Keccak.Digest256().digest(value.toByteArray()); + for (int offset = 0; offset < 6; offset += 2) { + int index = new BigInteger(1, Arrays.copyOfRange(hash, offset, offset + 2)).intValue() % 2048; + bits = bits.setBit(index); + } + return bits; + } +} diff --git a/plugins/README.md b/plugins/README.md index f14e070c01a..ee1f0c4acf5 100644 --- a/plugins/README.md +++ b/plugins/README.md @@ -148,6 +148,51 @@ NOTE: large db may GC overhead limit exceeded. - `--db`: db name. - `-h | --help`: provide the help info +## DB Backfill-Bloom + +DB backfill bloom rebuilds missing historical SectionBloom indexes from transaction results stored in `transactionRetStore`, enabling `eth_getLogs` to filter by address and topics. This is useful for historical blocks processed by versions before v4.8.1 while JSON-RPC filtering (`isJsonRpcFilterEnabled`) was disabled. Since v4.8.1, SectionBloom indexes are generated independently of this setting. + +### Prerequisites and behavior + +- Stop the node and any other process using the database before running the command. +- The database directory must contain the `properties` and `transactionRetStore` databases. `transactionRetStore` must contain at least one non-zero block. +- Ensure `storage.transHistory.switch` was enabled while the historical blocks were processed. Only blocks whose transaction results are still present in `transactionRetStore` can be backfilled; this tool cannot recover missing transaction results. +- The start and end block numbers are inclusive. +- The command creates or updates the `section-bloom` database in the specified database directory. +- An existing `section-bloom` directory uses its own engine. A new one inherits the engine of `transactionRetStore`. Missing `engine.properties` is treated as LevelDB for compatibility with older databases. +- On ARM64, only RocksDB is supported. LevelDB is rejected before any database is opened or created. +- The operation is idempotent. If it is interrupted, safely rerun the same block range. Existing SectionBloom bits are preserved and set again. Do not run multiple backfill processes concurrently. + +### Available parameters + +- `-d | --database-directory`: Parent directory containing the source databases and the destination `section-bloom` database. Default: `output-directory/database`. +- `-s | --start-block`: Inclusive start block. Omitted or `0` selects the earliest non-zero block in `transactionRetStore`. A lower value is automatically raised to the earliest available block. Negative values are rejected. +- `-e | --end-block`: Inclusive end block. Omitted or `0` selects the latest persisted block header number (`latest_block_header_number`) in `properties`. A higher value is automatically reduced to this height. Negative values are rejected. +- `-c | --max-concurrency`: Maximum number of processing threads, from 1 to 128. Default: 8. Use 4–8 for SATA SSD, 8–16 for NVMe SSD, or 1–2 for HDD. The actual concurrency does not exceed the number of sections being processed. +- `-h | --help`: Display the help message. + +### Examples + +```shell script +# Full command +java -jar Toolkit.jar db backfill-bloom [-d ] [-s ] [-e ] [-c ] [-h] + +# Backfill the complete available range in the default database directory +java -jar Toolkit.jar db backfill-bloom + +# Backfill blocks 1,000,000 through 2,000,000, inclusive +java -jar Toolkit.jar db backfill-bloom -s 1000000 -e 2000000 + +# Use a custom database directory and eight processing threads +java -jar Toolkit.jar db backfill-bloom -d /path/to/database -c 8 +``` + +### Progress and performance + +The terminal progress bar displays completed blocks, elapsed time, and estimated remaining time. `toolkit.log` records progress every 10,000 scanned blocks and includes the percentage, elapsed time, average rate, and estimated remaining time. The final summary reports scanned and successful blocks, blocks containing logs, errors, Bloom writes, duration, rates, and the concurrency used. + +Performance depends on the number of logs, storage engine, disk, CPU, and database compaction. Increase `--max-concurrency` gradually while monitoring disk latency and CPU usage. + ## Keystore Keystore provides commands for managing account keystore files (Web3 Secret Storage format). diff --git a/plugins/src/main/java/common/org/tron/plugins/Db.java b/plugins/src/main/java/common/org/tron/plugins/Db.java index 84654dca934..0918d939dc5 100644 --- a/plugins/src/main/java/common/org/tron/plugins/Db.java +++ b/plugins/src/main/java/common/org/tron/plugins/Db.java @@ -12,6 +12,7 @@ DbConvert.class, DbLite.class, DbCopy.class, + DbBackfillBloom.class, DbRoot.class }, commandListHeading = "%nCommands:%n%nThe most commonly used db commands are:%n" diff --git a/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java b/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java new file mode 100644 index 00000000000..2a2533430e3 --- /dev/null +++ b/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java @@ -0,0 +1,573 @@ +package org.tron.plugins; + +import com.google.protobuf.InvalidProtocolBufferException; +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.BitSet; +import java.util.List; +import java.util.Locale; +import java.util.Objects; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicLong; +import lombok.extern.slf4j.Slf4j; +import me.tongfei.progressbar.ProgressBar; +import org.apache.commons.lang3.ArrayUtils; +import org.tron.common.arch.Arch; +import org.tron.common.bloom.BloomUtils; +import org.tron.common.es.ExecutorServiceManager; +import org.tron.common.utils.ByteArray; +import org.tron.common.utils.ByteUtil; +import org.tron.core.exception.EventBloomException; +import org.tron.plugins.utils.db.DBInterface; +import org.tron.plugins.utils.db.DBIterator; +import org.tron.plugins.utils.db.DbTool; +import org.tron.plugins.utils.db.DbTool.DbType; +import org.tron.protos.Protocol.TransactionRet; +import picocli.CommandLine; + +@Slf4j(topic = "backfill-bloom") +@CommandLine.Command(name = "backfill-bloom", + description = { + "Backfill SectionBloom for historical blocks to enable eth_getLogs filtering.", + "The same block range can be safely rerun after interruption." + }, + exitCodeListHeading = "Exit Codes:%n", + exitCodeList = { + "0:Successful", + "1:Internal error: exception occurred, please check toolkit.log"}) +public class DbBackfillBloom implements Callable { + + @CommandLine.Spec + CommandLine.Model.CommandSpec spec; + + @CommandLine.Option(names = {"--database-directory", "-d"}, + defaultValue = "output-directory/database", + description = "Database directory path. Default: ${DEFAULT-VALUE}", order = 1) + private String databaseDirectory; + + @CommandLine.Option(names = {"--start-block", "-s"}, + description = "Start block number for backfill. " + + "Default or 0: earliest non-zero block", order = 2) + private long startBlock; + + @CommandLine.Option(names = {"--end-block", "-e"}, + description = "End block number for backfill, inclusive. " + + "Default or 0: latest persisted block header", + order = 3) + private long endBlock; + + // sames as SectionBloomStore.BLOCK_PER_SECTION + private static final int BLOCKS_PER_SECTION = 2048; + private static final long PROGRESS_LOG_INTERVAL = 10_000L; + private static final String PROPERTIES_DB_NAME = "properties"; + private static final String TRANSACTION_RET_DB_NAME = "transactionRetStore"; + private static final String SECTION_BLOOM_DB_NAME = "section-bloom"; + private static final String LATEST_BLOCK_HEADER_NUMBER = "latest_block_header_number"; + + @CommandLine.Option(names = {"--max-concurrency", "-c"}, defaultValue = "8", + description = "Maximum concurrency for processing. Default: ${DEFAULT-VALUE}. For SATA SSD " + + "use 4–8; for NVMe SSD use 8–16; for HDD use 1–2.", + order = 5) + private int maxConcurrency; + + @CommandLine.Option(names = {"--help", "-h"}, help = true, + description = "Display help message", order = 7) + private boolean help; + + // Statistics + // Number of blocks traversed (including failed ones) + private final AtomicLong processedBlocks = new AtomicLong(0); + // Number of successfully processed blocks + private final AtomicLong successfulBlocks = new AtomicLong(0); + // Number of blocks containing logs + private final AtomicLong blocksWithLogs = new AtomicLong(0); + // Number of block and task failures + private final AtomicLong errorCount = new AtomicLong(0); + // Total number of bloom writes + private final AtomicLong totalBloomWrites = new AtomicLong(0); + + private DBInterface transactionRetDb; + private DBInterface sectionBloomDb; + private DBInterface propertiesDb; + + private static class SectionRange { + + final long start; + final long end; + + SectionRange(long start, long end) { + this.start = start; + this.end = end; + } + } + + @Override + public Integer call() { + if (help) { + logger.info("Displaying backfill-bloom help"); + spec.commandLine().usage(System.out); + return 0; + } + + try { + // Validate parameters + if (!validateParameters()) { + return 1; + } + + // Initialize database connections + if (!initializeDatabase()) { + return 1; + } + + // Bound the requested range by the latest persisted block header. + long latestBlockHeaderNumber; + try { + latestBlockHeaderNumber = getLatestBlockHeaderNumber(); + } catch (Exception e) { + printError(e, "Failed to read latest persisted block header number"); + return 1; + } + if (latestBlockHeaderNumber < 0) { + printError("Latest persisted block header number does not exist"); + return 1; + } + if (endBlock == 0) { + endBlock = latestBlockHeaderNumber; + } else if (endBlock > latestBlockHeaderNumber) { + printInfo("End block %d is larger than latest persisted block header number %d; " + + "using %d instead.", + endBlock, latestBlockHeaderNumber, latestBlockHeaderNumber); + endBlock = latestBlockHeaderNumber; + } + + long minBlockNumber; + try { + minBlockNumber = getMinBlockNumber(); + } catch (Exception e) { + printError(e, "Failed to determine the first transaction result block"); + return 1; + } + if (minBlockNumber < 0) { + printError("Transaction result database does not contain any non-zero block"); + return 1; + } + if (startBlock == 0) { + startBlock = minBlockNumber; + } else if (startBlock < minBlockNumber) { + printInfo("Start block %d is earlier than the first available transaction result block %d; " + + "using %d instead.", startBlock, minBlockNumber, minBlockNumber); + startBlock = minBlockNumber; + } + + // Validate block range + if (endBlock < startBlock) { + printError("End block %d must be greater than or equal to start block %d", + endBlock, startBlock); + return 1; + } + + long totalBlocks = endBlock - startBlock + 1; + printInfo("Starting SectionBloom backfill for block number %d to %d (%d blocks)", + startBlock, endBlock, totalBlocks); + + // Process blocks with progress bar + long startTime = System.currentTimeMillis(); + int result = processBlocks(startTime); + long duration = (System.currentTimeMillis() - startTime) / 1000; + + // Print summary + printSummary(duration, result); + + return result; + + } catch (Exception e) { + printError(e, "Backfill failed"); + return 1; + } finally { + DbTool.close(); + } + } + + private boolean validateParameters() { + if (startBlock < 0) { + printError("Start block must be >= zero, it is %d", startBlock); + return false; + } + if (endBlock < 0) { + printError("End block must be >= zero, it is %d", endBlock); + return false; + } + + if (maxConcurrency <= 0 || maxConcurrency > 128) { + printError("Max concurrency %d must be between 1 and 128", maxConcurrency); + return false; + } + + File dbDir = new File(databaseDirectory); + if (!dbDir.exists() || !dbDir.isDirectory()) { + printError("Database directory does not exist or is not a directory"); + return false; + } + if (!isDatabaseDirectory(dbDir, PROPERTIES_DB_NAME)) { + printError("Required database '%s' does not exist", PROPERTIES_DB_NAME); + return false; + } + if (!isDatabaseDirectory(dbDir, TRANSACTION_RET_DB_NAME)) { + printError("Required database '%s' does not exist", TRANSACTION_RET_DB_NAME); + return false; + } + return true; + } + + private boolean isDatabaseDirectory(File databaseRoot, String databaseName) { + return new File(databaseRoot, databaseName).isDirectory(); + } + + private boolean initializeDatabase() { + DbType dbType; + try { + // Node databases share one engine. Resolve it before opening or creating any database. + String engineDbName = TRANSACTION_RET_DB_NAME; + File sectionBloomDirectory = new File(databaseDirectory, SECTION_BLOOM_DB_NAME); + if (sectionBloomDirectory.exists()) { + if (!sectionBloomDirectory.isDirectory()) { + throw new IllegalArgumentException("Database 'section-bloom' is not a directory"); + } + engineDbName = SECTION_BLOOM_DB_NAME; + } + dbType = getSupportedDbType(engineDbName); + } catch (IllegalArgumentException | UnsupportedOperationException e) { + printError(e, "Database engine validation failed: %s", e.getMessage()); + return false; + } + + try { + // Open all DBs here, single-threaded, before any worker thread starts. DbTool.getDB + // caches handles in a ConcurrentMap but its check-then-open is not atomic, so two + // threads opening the same LevelDB dir concurrently would hit the exclusive-lock error. + // Keep these handles for all worker threads instead of opening the same DB again. + transactionRetDb = DbTool.getDB(databaseDirectory, TRANSACTION_RET_DB_NAME, dbType); + sectionBloomDb = DbTool.getDB(databaseDirectory, SECTION_BLOOM_DB_NAME, dbType); + propertiesDb = DbTool.getDB(databaseDirectory, PROPERTIES_DB_NAME, dbType); + + printInfo("Database connections initialized successfully"); + return true; + } catch (Exception e) { + printError(e, "Failed to initialize database connections"); + return false; + } + } + + private DbType getSupportedDbType(String dbName) { + DbType type = DbTool.getDbType(databaseDirectory, dbName); + if (type == DbType.LevelDB) { + Arch.throwIfUnsupportedArm64Exception("LevelDB database '" + dbName + "'"); + } + return type; + } + + private long getLatestBlockHeaderNumber() { + byte[] latestBlockHeaderKey = LATEST_BLOCK_HEADER_NUMBER.getBytes(StandardCharsets.UTF_8); + byte[] latestBlockHeaderBytes = propertiesDb.get(latestBlockHeaderKey); + if (latestBlockHeaderBytes != null) { + return ByteArray.toLong(latestBlockHeaderBytes); + } + return -1; + } + + private long getMinBlockNumber() throws IOException { + try (DBIterator iterator = transactionRetDb.iterator()) { + iterator.seek(ByteArray.fromLong(1)); + if (iterator.hasNext()) { + return ByteArray.toLong(iterator.getKey()); + } + } + return -1; + } + + private int processBlocks(long startTime) { + long totalBlocks = endBlock - startBlock + 1; + // Calculate the section range to be processed + List sectionRanges = calculateSectionRanges(startBlock, endBlock); + + maxConcurrency = StrictMath.min(maxConcurrency, sectionRanges.size()); + ExecutorService executor = + ExecutorServiceManager.newFixedThreadPool("backfill-bloom", maxConcurrency); + List> futures = new ArrayList<>(); + + try (ProgressBar pb = new ProgressBar("Backfill section-bloom", totalBlocks)) { + printInfo("Processing %d sections with %d threads", sectionRanges.size(), maxConcurrency); + // Submit all section tasks to the thread pool + for (SectionRange range : sectionRanges) { + final long finalSectionStart = range.start; + final long finalSectionEnd = range.end; + + CompletableFuture future = CompletableFuture.runAsync( + () -> processSection(finalSectionStart, finalSectionEnd, totalBlocks, startTime, pb), + executor).whenComplete((unused, failure) -> { + if (failure != null) { + errorCount.incrementAndGet(); + printError(failure, "Error processing section %d to %d", + finalSectionStart, finalSectionEnd); + } + }); + + futures.add(future); + } + + // Wait for all tasks to complete + CompletableFuture allTasks = CompletableFuture.allOf(futures.toArray( + new CompletableFuture[0])); + + try { + allTasks.get(); + printInfo("All %d batch tasks completed", futures.size()); + } catch (Exception e) { + printError(e, "Error waiting for backfill tasks to complete"); + return 1; + } + + } catch (Exception e) { + printError(e, "Error in progress tracking"); + return 1; + } finally { + ExecutorServiceManager.shutdownAndAwaitTermination(executor, "backfill-bloom"); + } + + return errorCount.get() > 0 ? 1 : 0; + } + + /** + * Calculate the section range to be processed to ensure each thread processes a complete section. + * For example, startBlock=1000 and endBlock=4000 will generate: + * - SectionRange 0: [1000-2047] + * - SectionRange 1: [2048-4000] + */ + private List calculateSectionRanges(long startBlock, long endBlock) { + List ranges = new ArrayList<>(); + + long currentBlock = startBlock; + while (currentBlock <= endBlock) { + // Calculate the section to which the current block belongs + long sectionId = currentBlock / BLOCKS_PER_SECTION; + + // Calculate the boundaries of this section + long sectionStart = sectionId * BLOCKS_PER_SECTION; + long sectionEnd = sectionStart + BLOCKS_PER_SECTION - 1; + + // Adjust to the actual range that needs to be processed + long rangeStart = StrictMath.max(currentBlock, sectionStart); + long rangeEnd = StrictMath.min(endBlock, sectionEnd); + + ranges.add(new SectionRange(rangeStart, rangeEnd)); + currentBlock = sectionEnd + 1; + } + + return ranges; + } + + private void processSection(long sectionStart, long sectionEnd, long totalBlocks, long startTime, + ProgressBar pb) { + for (long blockNum = sectionStart; blockNum <= sectionEnd; blockNum++) { + try { + backfillBlockBloom(blockNum, transactionRetDb, sectionBloomDb); + successfulBlocks.incrementAndGet(); + } catch (Exception e) { + printError(e, "Error processing block %d", blockNum); + errorCount.incrementAndGet(); + } finally { + long processed = processedBlocks.incrementAndGet(); + logProgress(processed, totalBlocks, startTime); + pb.step(); + } + } + } + + private void logProgress(long processed, long totalBlocks, long startTime) { + if (processed % PROGRESS_LOG_INTERVAL != 0) { + return; + } + + long elapsedMillis = + StrictMath.max(System.currentTimeMillis() - startTime, 1L); + double progress = (double) processed / totalBlocks * 100; + double blocksPerSecond = (double) processed * 1000 / elapsedMillis; + long remainingSeconds = (long) ((totalBlocks - processed) / blocksPerSecond); + logger.info( + "Backfill progress: {}/{} blocks ({}%), elapsed={}, rate={} blocks/s, remaining={}", + processed, totalBlocks, String.format(Locale.ROOT, "%.2f", progress), + formatDuration(elapsedMillis / 1000), + String.format(Locale.ROOT, "%.2f", blocksPerSecond), + formatDuration(remainingSeconds)); + } + + private String formatDuration(long totalSeconds) { + long hours = totalSeconds / 3600; + long minutes = totalSeconds % 3600 / 60; + long seconds = totalSeconds % 60; + return String.format(Locale.ROOT, "%02d:%02d:%02d", hours, minutes, seconds); + } + + private void backfillBlockBloom(long blockNum, DBInterface transactionRetDb, + DBInterface sectionBloomDb) throws InvalidProtocolBufferException, EventBloomException { + + // Get transaction info for this block + byte[] blockKey = ByteArray.fromLong(blockNum); + byte[] transactionRetData = transactionRetDb.get(blockKey); + + if (transactionRetData == null) { + return; + } + + TransactionRet transactionRet = TransactionRet.parseFrom(transactionRetData); + + // Create bloom filter for this block using the same logic as SectionBloomStore + byte[] blockBloom = BloomUtils.createBloom(transactionRet); + + if (blockBloom != null) { + // Extract bit positions from bloom filter + List bitList = extractBitPositions(blockBloom); + + // A non-null bloom contains at least one set bit from a log address. + writeSectionBloom(blockNum, bitList, sectionBloomDb); + blocksWithLogs.incrementAndGet(); + } + } + + private List extractBitPositions(byte[] blockBloom) { + List bitList = new ArrayList<>(); + BitSet bs = BitSet.valueOf(blockBloom); + for (int i = bs.nextSetBit(0); i >= 0; i = bs.nextSetBit(i + 1)) { + // operate on index i here + if (i == Integer.MAX_VALUE) { + break; // or (i+1) would overflow + } + bitList.add(i); + } + return bitList; + } + + private void writeSectionBloom(long blockNum, List bitList, DBInterface sectionBloomDb) + throws EventBloomException { + + int section = (int) (blockNum / BLOCKS_PER_SECTION); + int blockNumOffset = (int) (blockNum % BLOCKS_PER_SECTION); + + for (int bitIndex : bitList) { + // Get existing BitSet from database + BitSet bitSet = getSectionBloomBitSet(section, bitIndex, sectionBloomDb); + if (Objects.isNull(bitSet)) { + bitSet = new BitSet(BLOCKS_PER_SECTION); + } + + // Update the bit for this block + bitSet.set(blockNumOffset); + + // Put back into database + putSectionBloomBitSet(section, bitIndex, bitSet, sectionBloomDb); + totalBloomWrites.incrementAndGet(); + } + } + + private long combineKey(int section, int bitIndex) { + return section * 1_000_000L + bitIndex; + } + + private BitSet getSectionBloomBitSet(int section, int bitIndex, DBInterface sectionBloomDb) + throws EventBloomException { + long keyLong = combineKey(section, bitIndex); + byte[] key = Long.toHexString(keyLong).getBytes(); + byte[] data = sectionBloomDb.get(key); + + if (ArrayUtils.isEmpty(data)) { + return null; + } + + try { + byte[] decompressedData = ByteUtil.decompress(data); + return BitSet.valueOf(decompressedData); + } catch (Exception e) { + throw new EventBloomException("decompress byte failed"); + } + } + + private void putSectionBloomBitSet(int section, int bitIndex, BitSet bitSet, + DBInterface sectionBloomDb) throws EventBloomException { + long keyLong = combineKey(section, bitIndex); + byte[] key = Long.toHexString(keyLong).getBytes(); + + byte[] compressedData = ByteUtil.compress(bitSet.toByteArray()); + sectionBloomDb.put(key, compressedData); + } + + private void printSummary(long duration, int result) { + spec.commandLine().getOut().println(); + printInfo("=== Backfill Summary ==="); + + printInfo("Total blocks scanned: %d", processedBlocks.get()); + printInfo("Successfully processed: %d", successfulBlocks.get()); + printInfo("Blocks with logs: %d", blocksWithLogs.get()); + printInfo("Errors encountered: %d", errorCount.get()); + printInfo("Duration: %d seconds", duration); + + // Success rate statistics + if (processedBlocks.get() > 0) { + double successRate = (double) successfulBlocks.get() / processedBlocks.get() * 100; + double logRate = (double) blocksWithLogs.get() / processedBlocks.get() * 100; + printInfo("Success rate: %.2f%% (%d/%d)", + successRate, successfulBlocks.get(), processedBlocks.get()); + printInfo("Blocks with logs rate: %.2f%% (%d/%d)", + logRate, blocksWithLogs.get(), processedBlocks.get()); + } + + // Performance statistics + printInfo("Total bloom writes: %d", totalBloomWrites.get()); + printInfo("Max concurrency used: %d threads", maxConcurrency); + printInfo("Section-based processing: No locks needed"); + + if (duration > 0) { + printInfo("Scanning rate: %.2f blocks/second", (double) processedBlocks.get() / duration); + printInfo("Processing rate: %.2f blocks/second", (double) successfulBlocks.get() / duration); + if (totalBloomWrites.get() > 0) { + printInfo("Bloom write rate: %.2f writes/second", + (double) totalBloomWrites.get() / duration); + } + } + + // Result judgment + if (result == 0) { + printInfo("✓ Backfill completed successfully!"); + } else { + printWarning("⚠ Backfill failed; check toolkit.log for details."); + } + } + + private void printInfo(String format, Object... args) { + String message = String.format(Locale.ROOT, format, args); + logger.info(message); + spec.commandLine().getOut().println(message); + } + + private void printWarning(String format, Object... args) { + String message = String.format(Locale.ROOT, format, args); + logger.warn(message); + spec.commandLine().getOut().println(message); + } + + private void printError(String format, Object... args) { + String message = String.format(Locale.ROOT, format, args); + logger.error(message); + spec.commandLine().getErr().println(message); + } + + private void printError(Throwable cause, String format, Object... args) { + String message = String.format(Locale.ROOT, format, args); + logger.error(message, cause); + spec.commandLine().getErr().println(message); + } +} diff --git a/plugins/src/main/java/common/org/tron/plugins/utils/db/DbTool.java b/plugins/src/main/java/common/org/tron/plugins/utils/db/DbTool.java index 127b8f97db5..373f46ca58a 100644 --- a/plugins/src/main/java/common/org/tron/plugins/utils/db/DbTool.java +++ b/plugins/src/main/java/common/org/tron/plugins/utils/db/DbTool.java @@ -161,7 +161,7 @@ public static void close() { } } - private static DbType getDbType(String sourceDir, String dbName) { + public static DbType getDbType(String sourceDir, String dbName) { String engineFile = Paths.get(sourceDir, dbName, ENGINE_FILE).toString(); if (!new File(engineFile).exists()) { return DbType.LevelDB; diff --git a/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java b/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java new file mode 100644 index 00000000000..ac8b6fe2660 --- /dev/null +++ b/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java @@ -0,0 +1,597 @@ +package org.tron.plugins; + +import static org.mockito.AdditionalMatchers.aryEq; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.classic.spi.ThrowableProxy; +import ch.qos.logback.core.read.ListAppender; +import com.google.protobuf.ByteString; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.PrintStream; +import java.io.PrintWriter; +import java.io.StringWriter; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.BitSet; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import me.tongfei.progressbar.ProgressBar; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.slf4j.LoggerFactory; +import org.tron.common.TestConstants; +import org.tron.common.arch.Arch; +import org.tron.common.bloom.BloomUtils; +import org.tron.common.utils.ByteArray; +import org.tron.common.utils.ByteUtil; +import org.tron.core.capsule.BytesCapsule; +import org.tron.core.capsule.TransactionRetCapsule; +import org.tron.core.config.args.Args; +import org.tron.core.store.SectionBloomStore; +import org.tron.plugins.utils.DBUtils; +import org.tron.plugins.utils.db.DBInterface; +import org.tron.plugins.utils.db.DBIterator; +import org.tron.plugins.utils.db.DbTool; +import org.tron.plugins.utils.db.DbTool.DbType; +import org.tron.protos.Protocol.TransactionInfo; +import org.tron.protos.Protocol.TransactionInfo.Log; +import org.tron.protos.Protocol.TransactionRet; +import picocli.CommandLine; + +public class DbBackfillBloomTest { + + private static final byte[] HEADER_KEY = + "latest_block_header_number".getBytes(StandardCharsets.UTF_8); + + @Rule + public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private File databaseRoot; + private DBInterface transactions; + private DBInterface bloom; + private DBInterface properties; + private DBIterator iterator; + private TransactionRet transactionRet; + private StringWriter output; + private StringWriter errors; + private Logger logger; + private ListAppender appender; + + @Before + public void setUp() throws Exception { + databaseRoot = temporaryFolder.newFolder(); + Files.createDirectories(new File(databaseRoot, "properties").toPath()); + Files.createDirectories(new File(databaseRoot, "transactionRetStore").toPath()); + transactions = mock(DBInterface.class); + bloom = mock(DBInterface.class); + properties = mock(DBInterface.class); + iterator = mock(DBIterator.class); + when(properties.get(aryEq(HEADER_KEY))).thenReturn(ByteArray.fromLong(40)); + when(transactions.iterator()).thenReturn(iterator); + when(iterator.hasNext()).thenReturn(true); + when(iterator.getKey()).thenReturn(ByteArray.fromLong(1)); + transactionRet = createTransactionRet(81); + logger = (Logger) LoggerFactory.getLogger("backfill-bloom"); + appender = new ListAppender<>(); + appender.start(); + logger.addAppender(appender); + } + + @After + public void tearDown() { + logger.detachAppender(appender); + appender.stop(); + DbTool.close(); + } + + @Test + public void testHelp() throws Exception { + PrintStream originalOut = System.out; + ByteArrayOutputStream help = new ByteArrayOutputStream(); + try (PrintStream capture = new PrintStream(help, true, "UTF-8")) { + System.setOut(capture); + Assert.assertEquals(0, execute("-h")); + String helpText = help.toString("UTF-8").replaceAll("\\s+", " "); + Assert.assertTrue(helpText + .contains("The same block range can be safely rerun after interruption.")); + Assert.assertTrue(helpText.contains("Default or 0: earliest non-zero block")); + Assert.assertTrue(helpText.contains("Default or 0: latest persisted block header")); + } finally { + System.setOut(originalOut); + } + } + + @Test + public void testInvalidParameters() { + try (MockedStatic dbTool = mockDatabases()) { + for (String[] options : new String[][] {{"-s", "-1"}, {"-e", "-1"}, + {"-c", "0"}, {"-c", "129"}}) { + Assert.assertEquals(Arrays.toString(options), 1, execute(options)); + Assert.assertFalse(errors.toString().isEmpty()); + if ("-e".equals(options[0])) { + Assert.assertTrue(errors.toString().contains("End block must be >= zero")); + } + assertNoDatabasesOpened(dbTool); + } + Assert.assertEquals(1, execute("-s", "41", "-e", "40")); + Assert.assertTrue(errors.toString().contains("End block")); + } + } + + @Test + public void testInvalidDatabasePaths() throws Exception { + File missingProperties = temporaryFolder.newFolder(); + Files.createDirectories(new File(missingProperties, "transactionRetStore").toPath()); + File missingTransactions = temporaryFolder.newFolder(); + Files.createDirectories(new File(missingTransactions, "properties").toPath()); + Assert.assertTrue(new File(databaseRoot, "section-bloom").createNewFile()); + File[] roots = {new File(temporaryFolder.getRoot(), "missing"), temporaryFolder.newFile(), + missingProperties, missingTransactions, databaseRoot}; + String[] messages = {"Database directory does not exist", "Database directory does not exist", + "Required database 'properties'", "Required database 'transactionRetStore'", + "Database 'section-bloom' is not a directory"}; + for (int i = 0; i < roots.length; i++) { + databaseRoot = roots[i]; + try (MockedStatic dbTool = mockDatabases()) { + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString(), errors.toString().contains(messages[i])); + assertNoDatabasesOpened(dbTool); + } + } + } + + @Test + public void testDatabaseInitializationFailure() { + try (MockedStatic dbTool = mockDatabases()) { + dbTool.when(() -> DbTool.getDB(anyString(), anyString(), any(DbType.class))) + .thenThrow(new RuntimeException("open failed")); + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains("Failed to initialize database connections")); + } + } + + @Test + public void testMissingOrUnreadableHeader() { + when(properties.get(aryEq("LATEST_SOLIDIFIED_BLOCK_NUM".getBytes(StandardCharsets.UTF_8)))) + .thenReturn(ByteArray.fromLong(22)); + when(properties.get(aryEq(HEADER_KEY))).thenReturn(null) + .thenThrow(new RuntimeException("read failed")); + try (MockedStatic ignored = mockDatabases()) { + for (String message : new String[] {"Latest persisted block header number does not exist", + "Failed to read latest persisted block header number"}) { + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains(message)); + Assert.assertFalse(output.toString().contains("Starting SectionBloom backfill")); + } + } + } + + @Test + public void testUnavailableTransactionResults() throws Exception { + try (MockedStatic ignored = mockDatabases()) { + when(iterator.hasNext()).thenReturn(false); + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains("does not contain any non-zero block")); + verify(iterator).seek(aryEq(ByteArray.fromLong(1))); + verify(iterator).close(); + when(transactions.iterator()).thenThrow(new RuntimeException("iterator failed")); + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains("Failed to determine the first transaction")); + } + } + + @Test(timeout = 30_000) + public void testProgressAndSummary() { + when(properties.get(aryEq(HEADER_KEY))).thenReturn(ByteArray.fromLong(10_000)); + try (MockedStatic ignored = mockDatabases()) { + Assert.assertEquals(0, execute("-c", "1")); + Assert.assertTrue(output.toString().contains("Total blocks scanned: 10000")); + Assert.assertTrue(output.toString().contains("Successfully processed: 10000")); + Assert.assertTrue(output.toString().contains("Success rate: 100.00%")); + Assert.assertTrue(output.toString().contains("Backfill completed successfully!")); + Assert.assertFalse(output.toString().contains("Backfill progress:")); + Assert.assertEquals(1L, appender.list.stream().map(ILoggingEvent::getFormattedMessage) + .filter(message -> message.startsWith("Backfill progress: 10000/10000 blocks (100.00%)")) + .count()); + } + } + + @Test(timeout = 30_000) + public void testWorkerErrorsFailSummaryAndPreserveCauses() { + when(properties.get(aryEq(HEADER_KEY))).thenReturn(ByteArray.fromLong(2050)); + Error[] failures = {new AssertionError("worker failed"), + new NoClassDefFoundError("missing worker dependency")}; + when(transactions.get(aryEq(ByteArray.fromLong(1)))).thenThrow(failures[0]); + when(transactions.get(aryEq(ByteArray.fromLong(2048)))).thenThrow(failures[1]); + try (MockedStatic ignored = mockDatabases()) { + Assert.assertEquals(1, execute("-c", "2")); + Assert.assertTrue(output.toString().contains("Errors encountered: 2")); + Assert.assertTrue(output.toString().contains("Total blocks scanned: 2")); + Assert.assertTrue(output.toString().contains("Successfully processed: 0")); + Assert.assertTrue(output.toString().contains("Backfill failed;")); + Assert.assertFalse(output.toString().contains("Backfill completed successfully!")); + verify(transactions, never()).get(aryEq(ByteArray.fromLong(2))); + verify(transactions, never()).get(aryEq(ByteArray.fromLong(2049))); + for (int i = 0; i < failures.length; i++) { + String message = "Error processing section " + (i == 0 ? "1 to 2047" : "2048 to 2050"); + Assert.assertTrue(errors.toString().contains(message)); + ILoggingEvent event = appender.list.stream() + .filter(entry -> message.equals(entry.getFormattedMessage())).findFirst().orElse(null); + Assert.assertNotNull(event); + ThrowableProxy throwable = (ThrowableProxy) event.getThrowableProxy(); + Assert.assertNotNull(throwable); + Assert.assertSame(failures[i], throwable.getThrowable().getCause()); + } + } + } + + @Test(timeout = 30_000) + public void testSummaryReportsProgressFailureWithoutBlockErrors() { + RuntimeException failure = new RuntimeException("progress close failed"); + try (MockedStatic ignored = mockDatabases(); + MockedConstruction progressBars = mockConstruction(ProgressBar.class, + (bar, context) -> doThrow(failure).when(bar).close())) { + Assert.assertEquals(1, execute("-e", "1")); + Assert.assertEquals(1, progressBars.constructed().size()); + Assert.assertTrue(output.toString().contains("Successfully processed: 1")); + Assert.assertTrue(output.toString().contains("Errors encountered: 0")); + Assert.assertTrue(output.toString().contains("Backfill failed;")); + Assert.assertFalse(output.toString().contains("Backfill completed successfully!")); + Assert.assertTrue(errors.toString().contains("Error in progress tracking")); + } + } + + @Test(timeout = 30_000) + public void testDatabaseWriteFailurePreservesOriginalCause() { + when(transactions.get(aryEq(ByteArray.fromLong(1)))).thenReturn(transactionRet.toByteArray()); + RuntimeException failure = new RuntimeException("section-bloom", new IOException("disk full")); + doThrow(failure).when(bloom).put(any(byte[].class), any(byte[].class)); + try (MockedStatic ignored = mockDatabases()) { + Assert.assertEquals(1, execute("-e", "1")); + verify(bloom).put(any(byte[].class), any(byte[].class)); + Assert.assertTrue(errors.toString().contains("Error processing block 1")); + Assert.assertFalse(output.toString().contains("Error processing block 1")); + Assert.assertTrue(output.toString().contains("Errors encountered: 1")); + Assert.assertTrue(output.toString().contains("Successfully processed: 0")); + Assert.assertFalse(output.toString().contains("Backfill completed successfully!")); + ILoggingEvent event = appender.list.stream() + .filter(entry -> "Error processing block 1".equals(entry.getFormattedMessage())) + .findFirst().orElse(null); + Assert.assertNotNull(event); + ThrowableProxy throwable = (ThrowableProxy) event.getThrowableProxy(); + Assert.assertNotNull(throwable); + Assert.assertSame(failure, throwable.getThrowable()); + Assert.assertEquals("disk full", throwable.getCause().getMessage()); + } + } + + @Test(timeout = 30_000) + public void testMalformedProtobufFailsWithoutWritingBloom() throws Exception { + writeSource(1, 1, 1); + openDb("transactionRetStore").put(ByteArray.fromLong(1), new byte[] {(byte) 0x80}); + DbTool.close(); + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains("Error processing block 1")); + Assert.assertTrue(readBloomEntries().isEmpty()); + } + + @Test(timeout = 30_000) + public void testEngineSelectionUsesExistingBloomOrTransactionStore() throws Exception { + for (boolean existingBloom : new boolean[] {false, true}) { + for (String engine : new String[] {"LEVELDB", "ROCKSDB", null}) { + databaseRoot = temporaryFolder.newFolder(); + writeEngine("transactionRetStore", engine); + writeEngine("properties", engine); + if (existingBloom) { + writeEngine("section-bloom", engine); + } + DbType type = "ROCKSDB".equals(engine) ? DbType.RocksDB : DbType.LevelDB; + String reference = existingBloom ? "section-bloom" : "transactionRetStore"; + try (MockedStatic arch = mockStatic(Arch.class, CALLS_REAL_METHODS); + MockedStatic dbTool = mockDatabases()) { + arch.when(Arch::getOsArch).thenReturn("amd64"); + dbTool.when(() -> DbTool.getDbType(anyString(), anyString())).thenCallRealMethod(); + Assert.assertEquals(0, execute("-e", "1")); + dbTool.verify(() -> DbTool.getDbType(databaseRoot.toString(), reference)); + dbTool.verify(() -> DbTool.getDbType(anyString(), anyString()), times(1)); + for (String name : new String[] {"transactionRetStore", "section-bloom", "properties"}) { + dbTool.verify(() -> DbTool.getDB(databaseRoot.toString(), name, type)); + } + arch.verify(() -> Arch.throwIfUnsupportedArm64Exception(anyString()), + type == DbType.LevelDB ? times(1) : never()); + } + } + } + } + + @Test(timeout = 30_000) + public void testArmRejectsLegacyEnginesBeforeOpeningDatabases() throws Exception { + for (boolean existingBloom : new boolean[] {false, true}) { + for (String engine : new String[] {"LEVELDB", null}) { + databaseRoot = temporaryFolder.newFolder(); + writeEngine("transactionRetStore", engine); + writeEngine("properties", engine); + if (existingBloom) { + writeEngine("section-bloom", engine); + } + String reference = existingBloom ? "section-bloom" : "transactionRetStore"; + try (MockedStatic arch = mockStatic(Arch.class, CALLS_REAL_METHODS); + MockedStatic dbTool = mockDatabases()) { + arch.when(Arch::getOsArch).thenReturn(existingBloom ? "arm64" : "aarch64"); + dbTool.when(() -> DbTool.getDbType(anyString(), anyString())).thenCallRealMethod(); + Assert.assertEquals(1, execute()); + Assert.assertTrue(errors.toString().contains("LevelDB database '" + reference + + "': unsupported")); + assertNoDatabasesOpened(dbTool); + Assert.assertFalse(new File(databaseRoot, reference + "/CURRENT").exists()); + Assert.assertEquals(existingBloom, new File(databaseRoot, "section-bloom").exists()); + } + } + } + } + + @Test(timeout = 60_000) + public void testRealBackfillMatchesNodeAcrossSectionBoundaryAndRerun() throws Exception { + TransactionRet second = createTransactionRet(82); + TransactionRet empty = TransactionRet.newBuilder() + .addTransactioninfo(TransactionInfo.getDefaultInstance()).build(); + openDb("transactionRetStore").put(ByteArray.fromLong(2047), transactionRet.toByteArray()); + openDb("transactionRetStore").put(ByteArray.fromLong(2048), second.toByteArray()); + openDb("transactionRetStore").put(ByteArray.fromLong(2049), empty.toByteArray()); + openDb("properties").put(HEADER_KEY, ByteArray.fromLong(2050)); + DbTool.close(); + Assert.assertFalse(new File(databaseRoot, "section-bloom").exists()); + + Args.setParam(new String[] {"--output-directory", temporaryFolder.newFolder().toString(), + "--storage-db-engine", "ROCKSDB"}, TestConstants.TEST_CONF); + SectionBloomStore nodeStore = null; + try { + nodeStore = new SectionBloomStore("section-bloom"); + nodeStore.initBlockSection(new TransactionRetCapsule(transactionRet.toByteArray())); + nodeStore.write(2047); + nodeStore.initBlockSection(new TransactionRetCapsule(second.toByteArray())); + nodeStore.write(2048); + nodeStore.initBlockSection(new TransactionRetCapsule(empty.toByteArray())); + nodeStore.write(2049); + for (int run = 0; run < 3; run++) { + if (run == 1) { + // Seed only the older bit so the existing store must also be backfilled. + nodeStore.initBlockSection(new TransactionRetCapsule(transactionRet.toByteArray())); + nodeStore.write(7); + BitSet existing = new BitSet(); + existing.set(7); + writeBloom(ByteUtil.compress(existing.toByteArray())); + } + Assert.assertEquals(0, execute("-c", "2")); + Assert.assertTrue(output.toString().contains("Blocks with logs: 2")); + Assert.assertTrue(output.toString().contains("Successfully processed: 4")); + Assert.assertTrue(output.toString().contains("Processing 2 sections with 2 threads")); + Assert.assertEquals(DbType.RocksDB, + DbTool.getDbType(databaseRoot.toString(), "section-bloom")); + Map expected = readNodeSections(nodeStore); + Map actual = readBloomEntries(); + Assert.assertFalse(expected.isEmpty()); + Assert.assertEquals(expected.keySet(), actual.keySet()); + for (Map.Entry entry : expected.entrySet()) { + Assert.assertArrayEquals(entry.getValue(), actual.get(entry.getKey())); + } + } + } finally { + try { + if (nodeStore != null) { + nodeStore.close(); + } + } finally { + Args.clearParam(); + } + } + } + + @Test(timeout = 30_000) + public void testEndBlockUsesPersistedHead() throws Exception { + String[][] options = {{}, {"-e", "30"}, {"-e", "40"}, {"-e", "60"}, {"-e", "0"}}; + int[] expectedEnds = {40, 30, 40, 40, 40}; + for (int i = 0; i < options.length; i++) { + databaseRoot = temporaryFolder.newFolder(); + writeSource(1, 41, 40); + Assert.assertEquals(0, execute(options[i])); + Assert.assertTrue(output.toString().contains("Total blocks scanned: " + expectedEnds[i])); + assertIndexedBlocks(1, expectedEnds[i]); + if (i == 3) { + Assert.assertTrue(output.toString().contains("number 40; using 40 instead.")); + } + } + } + + @Test(timeout = 30_000) + public void testStartBlockUsesFirstNonZeroTransactionResult() throws Exception { + writeSource(25, 41, 40); + openDb("transactionRetStore").put(ByteArray.fromLong(0), transactionRet.toByteArray()); + DbTool.close(); + for (String[] options : new String[][] {{"-e", "30"}, {"-s", "0", "-e", "30"}, + {"-s", "1", "-e", "30"}}) { + Assert.assertEquals(0, execute(options)); + Assert.assertTrue(output.toString().contains("Total blocks scanned: 6")); + assertIndexedBlocks(25, 30); + } + Assert.assertTrue(output.toString().contains( + "Start block 1 is earlier than the first available transaction result block 25")); + } + + @Test(timeout = 30_000) + public void testEmptyBloomValuesAreRebuilt() throws Exception { + writeSource(1, 1, 1); + for (byte[] value : new byte[][] {new byte[0], ByteUtil.compress(new byte[0])}) { + writeBloom(value); + Assert.assertEquals(0, execute()); + Assert.assertTrue(output.toString().contains("Errors encountered: 0")); + assertIndexedBlocks(1, 1); + } + } + + private int execute(String... options) { + output = new StringWriter(); + errors = new StringWriter(); + appender.list.clear(); + List args = new ArrayList<>(Arrays.asList("db", "backfill-bloom", "-d", + databaseRoot.toString())); + Collections.addAll(args, options); + return new CommandLine(new Toolkit()).setOut(new PrintWriter(output)) + .setErr(new PrintWriter(errors)).execute(args.toArray(new String[0])); + } + + private MockedStatic mockDatabases() { + MockedStatic dbTool = mockStatic(DbTool.class); + dbTool.when(() -> DbTool.getDbType(anyString(), anyString())).thenReturn(DbType.RocksDB); + dbTool.when(() -> DbTool.getDB(anyString(), anyString(), any(DbType.class))) + .thenAnswer(invocation -> { + switch (invocation.getArgument(1, String.class)) { + case "transactionRetStore": + return transactions; + case "section-bloom": + return bloom; + case "properties": + return properties; + default: + throw new AssertionError("Unexpected database"); + } + }); + return dbTool; + } + + private void assertNoDatabasesOpened(MockedStatic dbTool) { + dbTool.verify(() -> DbTool.getDB(anyString(), anyString()), never()); + dbTool.verify(() -> DbTool.getDB(anyString(), anyString(), any(DbType.class)), never()); + } + + private void writeEngine(String name, String engine) throws Exception { + File directory = new File(databaseRoot, name); + Files.createDirectories(directory.toPath()); + if (engine != null) { + Files.write(directory.toPath().resolve(DBUtils.FILE_ENGINE), + ("ENGINE=" + engine).getBytes(StandardCharsets.UTF_8)); + } + } + + private DBInterface openDb(String name) throws Exception { + return DbTool.getDB(databaseRoot.toString(), name, DbType.RocksDB); + } + + private void writeSource(int first, int last, long head) throws Exception { + try { + DBInterface source = openDb("transactionRetStore"); + for (int block = first; block <= last; block++) { + source.put(ByteArray.fromLong(block), transactionRet.toByteArray()); + } + openDb("properties").put(HEADER_KEY, ByteArray.fromLong(head)); + openDb("properties").put("LATEST_SOLIDIFIED_BLOCK_NUM".getBytes(StandardCharsets.UTF_8), + ByteArray.fromLong(StrictMath.min(22, head))); + } finally { + DbTool.close(); + } + } + + private void writeBloom(byte[] value) throws Exception { + try { + BitSet indexes = BitSet.valueOf(BloomUtils.createBloom(transactionRet)); + Assert.assertFalse(indexes.isEmpty()); + for (int bit = indexes.nextSetBit(0); bit >= 0; bit = indexes.nextSetBit(bit + 1)) { + openDb("section-bloom").put(bloomKey(0, bit), value); + } + } finally { + DbTool.close(); + } + } + + private void assertIndexedBlocks(int first, int last) throws Exception { + Map actual = readBloomEntries(); + BitSet indexes = BitSet.valueOf(BloomUtils.createBloom(transactionRet)); + Assert.assertFalse(indexes.isEmpty()); + Assert.assertEquals(indexes.cardinality(), actual.size()); + BitSet expected = new BitSet(); + expected.set(first, last + 1); + for (int bit = indexes.nextSetBit(0); bit >= 0; bit = indexes.nextSetBit(bit + 1)) { + byte[] value = actual.get(ByteString.copyFrom(bloomKey(0, bit))); + Assert.assertNotNull(value); + Assert.assertEquals(expected, BitSet.valueOf(ByteUtil.decompress(value))); + } + } + + private Map readBloomEntries() throws Exception { + Map result = new HashMap<>(); + try { + DBInterface database = DbTool.getDB(databaseRoot.toString(), "section-bloom"); + try (DBIterator entries = database.iterator()) { + entries.seekToFirst(); + while (entries.hasNext()) { + result.put(ByteString.copyFrom(entries.getKey()), entries.getValue()); + entries.next(); + } + } + } finally { + DbTool.close(); + } + return result; + } + + private Map readNodeSections(SectionBloomStore store) { + Map result = new HashMap<>(); + for (int section = 0; section < 2; section++) { + for (int bit = 0; bit < 2048; bit++) { + byte[] key = bloomKey(section, bit); + BytesCapsule value = store.get(key); + if (value != null) { + result.put(ByteString.copyFrom(key), value.getData()); + } + } + } + return result; + } + + private byte[] bloomKey(int section, int bit) { + return Long.toHexString(section * 1_000_000L + bit).getBytes(StandardCharsets.UTF_8); + } + + private TransactionRet createTransactionRet(long seed) { + Random random = new Random(seed); + TransactionRet.Builder result = TransactionRet.newBuilder(); + for (int transaction = 0; transaction < 2; transaction++) { + TransactionInfo.Builder info = TransactionInfo.newBuilder(); + for (int log = 0; log < 2; log++) { + byte[] address = new byte[20]; + byte[] topic = new byte[32]; + random.nextBytes(address); + random.nextBytes(topic); + info.addLog(Log.newBuilder().setAddress(ByteString.copyFrom(address)) + .addTopics(ByteString.copyFrom(topic))); + } + result.addTransactioninfo(info); + } + return result.build(); + } +} diff --git a/plugins/src/test/java/org/tron/plugins/utils/DbToolTest.java b/plugins/src/test/java/org/tron/plugins/utils/DbToolTest.java new file mode 100644 index 00000000000..a649f97d7e5 --- /dev/null +++ b/plugins/src/test/java/org/tron/plugins/utils/DbToolTest.java @@ -0,0 +1,68 @@ +package org.tron.plugins.utils; + +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import org.junit.Assert; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.tron.plugins.utils.db.DbTool; +import org.tron.plugins.utils.db.DbTool.DbType; + +public class DbToolTest { + + @Rule + public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void testMissingMetadataDefaultsToLegacyEngineWithoutWriting() throws Exception { + File directory = temporaryFolder.newFolder("database"); + File database = new File(directory, "legacy"); + Assert.assertTrue(database.mkdir()); + + Assert.assertEquals(DbType.LevelDB, DbTool.getDbType(directory.toString(), "legacy")); + Assert.assertEquals(0, database.list().length); + Assert.assertEquals(DbType.LevelDB, DbTool.getDbType(directory.toString(), "missing")); + Assert.assertFalse(new File(directory, "missing").exists()); + } + + @Test + public void testRecognizesBothEnginesIgnoringCase() throws Exception { + File directory = temporaryFolder.newFolder("database"); + Path database = Files.createDirectory(directory.toPath().resolve("store")); + Path metadata = database.resolve(DBUtils.FILE_ENGINE); + for (DbType type : DbType.values()) { + byte[] content = ("ENGINE=" + type.name()).getBytes(StandardCharsets.UTF_8); + Files.write(metadata, content); + + Assert.assertEquals(type, DbTool.getDbType(directory.toString(), "store")); + Assert.assertArrayEquals(content, Files.readAllBytes(metadata)); + } + } + + @Test + public void testEmptyMissingAndUnknownEngineValuesDefaultToLegacyEngine() throws Exception { + File directory = temporaryFolder.newFolder("database"); + Path database = Files.createDirectory(directory.toPath().resolve("store")); + Path metadata = database.resolve(DBUtils.FILE_ENGINE); + for (String content : new String[] {"", "OTHER=ROCKSDB", "ENGINE=", "ENGINE=UNKNOWN"}) { + byte[] bytes = content.getBytes(StandardCharsets.UTF_8); + Files.write(metadata, bytes); + + Assert.assertEquals(DbType.LevelDB, DbTool.getDbType(directory.toString(), "store")); + Assert.assertArrayEquals(bytes, Files.readAllBytes(metadata)); + } + } + + @Test + public void testUnreadableMetadataDefaultsToLegacyEngine() throws Exception { + File directory = temporaryFolder.newFolder("database"); + Path database = Files.createDirectory(directory.toPath().resolve("store")); + Files.createDirectory(database.resolve(DBUtils.FILE_ENGINE)); + + Assert.assertEquals(DbType.LevelDB, DbTool.getDbType(directory.toString(), "store")); + Assert.assertFalse(Files.exists(database.resolve("CURRENT"))); + } +} From ddbd87850ccc808b374807d4b2ab0643608cb88f Mon Sep 17 00:00:00 2001 From: 317787106 <317787106@qq.com> Date: Wed, 16 Sep 2026 16:23:55 +0800 Subject: [PATCH 2/2] read and write section-bloom less --- plugins/README.md | 6 +- .../org/tron/plugins/DbBackfillBloom.java | 85 +++++----- .../org/tron/plugins/DbBackfillBloomTest.java | 153 ++++++++++++++++-- 3 files changed, 185 insertions(+), 59 deletions(-) diff --git a/plugins/README.md b/plugins/README.md index ee1f0c4acf5..3b3b26383ae 100644 --- a/plugins/README.md +++ b/plugins/README.md @@ -161,7 +161,7 @@ DB backfill bloom rebuilds missing historical SectionBloom indexes from transact - The command creates or updates the `section-bloom` database in the specified database directory. - An existing `section-bloom` directory uses its own engine. A new one inherits the engine of `transactionRetStore`. Missing `engine.properties` is treated as LevelDB for compatibility with older databases. - On ARM64, only RocksDB is supported. LevelDB is rejected before any database is opened or created. -- The operation is idempotent. If it is interrupted, safely rerun the same block range. Existing SectionBloom bits are preserved and set again. Do not run multiple backfill processes concurrently. +- The operation is idempotent. If it is interrupted, safely rerun the same block range. Existing SectionBloom bits are preserved, and unchanged index records are not rewritten. Do not run multiple backfill processes concurrently. ### Available parameters @@ -189,7 +189,9 @@ java -jar Toolkit.jar db backfill-bloom -d /path/to/database -c 8 ### Progress and performance -The terminal progress bar displays completed blocks, elapsed time, and estimated remaining time. `toolkit.log` records progress every 10,000 scanned blocks and includes the percentage, elapsed time, average rate, and estimated remaining time. The final summary reports scanned and successful blocks, blocks containing logs, errors, Bloom writes, duration, rates, and the concurrency used. +Each worker accumulates index bits for one section of up to 2,048 blocks. At the end of the section, each touched index record is read once, merged with existing bits, and written only if it changes. The Bloom write count reports actual index-record writes. + +The terminal progress bar displays scanned blocks, elapsed time, and estimated remaining time. `toolkit.log` records progress every 10,000 scanned blocks and includes the percentage, elapsed time, average rate, and estimated remaining time. A section's successful-block and log-block counts are added only after its required index writes finish. If a section write fails, none of its blocks are counted as successful; rerunning the same range completes any partially written section. The final summary reports scanned and successful blocks, blocks containing logs, block/task errors, Bloom writes, duration, rates, and the concurrency used. Performance depends on the number of logs, storage engine, disk, CPU, and database compaction. Increase `--max-concurrency` gradually while monitoring disk latency and CPU usage. diff --git a/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java b/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java index 2a2533430e3..01071e09460 100644 --- a/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java +++ b/plugins/src/main/java/common/org/tron/plugins/DbBackfillBloom.java @@ -8,7 +8,6 @@ import java.util.BitSet; import java.util.List; import java.util.Locale; -import java.util.Objects; import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; @@ -81,9 +80,9 @@ public class DbBackfillBloom implements Callable { // Statistics // Number of blocks traversed (including failed ones) private final AtomicLong processedBlocks = new AtomicLong(0); - // Number of successfully processed blocks + // Number of successfully processed blocks in fully written sections private final AtomicLong successfulBlocks = new AtomicLong(0); - // Number of blocks containing logs + // Number of blocks containing logs in fully written sections private final AtomicLong blocksWithLogs = new AtomicLong(0); // Number of block and task failures private final AtomicLong errorCount = new AtomicLong(0); @@ -373,10 +372,15 @@ private List calculateSectionRanges(long startBlock, long endBlock private void processSection(long sectionStart, long sectionEnd, long totalBlocks, long startTime, ProgressBar pb) { + BitSet[] sectionBloom = new BitSet[BloomUtils.BLOOM_BIT_SIZE]; + long sectionSuccessfulBlocks = 0; + long sectionBlocksWithLogs = 0; for (long blockNum = sectionStart; blockNum <= sectionEnd; blockNum++) { try { - backfillBlockBloom(blockNum, transactionRetDb, sectionBloomDb); - successfulBlocks.incrementAndGet(); + if (accumulateBlockBloom(blockNum, sectionBloom)) { + sectionBlocksWithLogs++; + } + sectionSuccessfulBlocks++; } catch (Exception e) { printError(e, "Error processing block %d", blockNum); errorCount.incrementAndGet(); @@ -386,6 +390,15 @@ private void processSection(long sectionStart, long sectionEnd, long totalBlocks pb.step(); } } + + try { + flushSectionBloom((int) (sectionStart / BLOCKS_PER_SECTION), sectionBloom); + successfulBlocks.addAndGet(sectionSuccessfulBlocks); + blocksWithLogs.addAndGet(sectionBlocksWithLogs); + } catch (Exception e) { + printError(e, "Error writing section %d to %d", sectionStart, sectionEnd); + errorCount.incrementAndGet(); + } } private void logProgress(long processed, long totalBlocks, long startTime) { @@ -413,15 +426,15 @@ private String formatDuration(long totalSeconds) { return String.format(Locale.ROOT, "%02d:%02d:%02d", hours, minutes, seconds); } - private void backfillBlockBloom(long blockNum, DBInterface transactionRetDb, - DBInterface sectionBloomDb) throws InvalidProtocolBufferException, EventBloomException { + private boolean accumulateBlockBloom(long blockNum, BitSet[] sectionBloom) + throws InvalidProtocolBufferException { // Get transaction info for this block byte[] blockKey = ByteArray.fromLong(blockNum); byte[] transactionRetData = transactionRetDb.get(blockKey); if (transactionRetData == null) { - return; + return false; } TransactionRet transactionRet = TransactionRet.parseFrom(transactionRetData); @@ -429,47 +442,39 @@ private void backfillBlockBloom(long blockNum, DBInterface transactionRetDb, // Create bloom filter for this block using the same logic as SectionBloomStore byte[] blockBloom = BloomUtils.createBloom(transactionRet); - if (blockBloom != null) { - // Extract bit positions from bloom filter - List bitList = extractBitPositions(blockBloom); - - // A non-null bloom contains at least one set bit from a log address. - writeSectionBloom(blockNum, bitList, sectionBloomDb); - blocksWithLogs.incrementAndGet(); + if (blockBloom == null) { + return false; } - } - private List extractBitPositions(byte[] blockBloom) { - List bitList = new ArrayList<>(); - BitSet bs = BitSet.valueOf(blockBloom); - for (int i = bs.nextSetBit(0); i >= 0; i = bs.nextSetBit(i + 1)) { - // operate on index i here - if (i == Integer.MAX_VALUE) { - break; // or (i+1) would overflow + int blockNumOffset = (int) (blockNum % BLOCKS_PER_SECTION); + BitSet bloomBits = BitSet.valueOf(blockBloom); + for (int bit = bloomBits.nextSetBit(0); bit >= 0; bit = bloomBits.nextSetBit(bit + 1)) { + if (sectionBloom[bit] == null) { + sectionBloom[bit] = new BitSet(BLOCKS_PER_SECTION); } - bitList.add(i); + sectionBloom[bit].set(blockNumOffset); } - return bitList; + return true; } - private void writeSectionBloom(long blockNum, List bitList, DBInterface sectionBloomDb) + private void flushSectionBloom(int section, BitSet[] sectionBloom) throws EventBloomException { - - int section = (int) (blockNum / BLOCKS_PER_SECTION); - int blockNumOffset = (int) (blockNum % BLOCKS_PER_SECTION); - - for (int bitIndex : bitList) { - // Get existing BitSet from database - BitSet bitSet = getSectionBloomBitSet(section, bitIndex, sectionBloomDb); - if (Objects.isNull(bitSet)) { - bitSet = new BitSet(BLOCKS_PER_SECTION); + for (int bitIndex = 0; bitIndex < sectionBloom.length; bitIndex++) { + BitSet additions = sectionBloom[bitIndex]; + if (additions == null) { + continue; + } + // Read each touched index once, preserving bits outside the requested block range. + BitSet existing = getSectionBloomBitSet(section, bitIndex, sectionBloomDb); + if (existing != null) { + additions.andNot(existing); + if (additions.isEmpty()) { + continue; + } + additions.or(existing); } - // Update the bit for this block - bitSet.set(blockNumOffset); - - // Put back into database - putSectionBloomBitSet(section, bitIndex, bitSet, sectionBloomDb); + putSectionBloomBitSet(section, bitIndex, additions, sectionBloomDb); totalBloomWrites.incrementAndGet(); } } diff --git a/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java b/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java index ac8b6fe2660..8376207a0e6 100644 --- a/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java +++ b/plugins/src/test/java/org/tron/plugins/DbBackfillBloomTest.java @@ -4,11 +4,13 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockConstruction; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -34,6 +36,7 @@ import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.atomic.AtomicInteger; import me.tongfei.progressbar.ProgressBar; import org.junit.After; import org.junit.Assert; @@ -43,6 +46,7 @@ import org.junit.rules.TemporaryFolder; import org.mockito.MockedConstruction; import org.mockito.MockedStatic; +import org.mockito.stubbing.Answer; import org.slf4j.LoggerFactory; import org.tron.common.TestConstants; import org.tron.common.arch.Arch; @@ -226,17 +230,22 @@ public void testWorkerErrorsFailSummaryAndPreserveCauses() { when(properties.get(aryEq(HEADER_KEY))).thenReturn(ByteArray.fromLong(2050)); Error[] failures = {new AssertionError("worker failed"), new NoClassDefFoundError("missing worker dependency")}; - when(transactions.get(aryEq(ByteArray.fromLong(1)))).thenThrow(failures[0]); - when(transactions.get(aryEq(ByteArray.fromLong(2048)))).thenThrow(failures[1]); + when(transactions.get(aryEq(ByteArray.fromLong(1)))).thenReturn(transactionRet.toByteArray()); + when(transactions.get(aryEq(ByteArray.fromLong(2)))).thenThrow(failures[0]); + when(transactions.get(aryEq(ByteArray.fromLong(2048)))) + .thenReturn(transactionRet.toByteArray()); + when(transactions.get(aryEq(ByteArray.fromLong(2049)))).thenThrow(failures[1]); try (MockedStatic ignored = mockDatabases()) { Assert.assertEquals(1, execute("-c", "2")); Assert.assertTrue(output.toString().contains("Errors encountered: 2")); - Assert.assertTrue(output.toString().contains("Total blocks scanned: 2")); + Assert.assertTrue(output.toString().contains("Total blocks scanned: 4")); Assert.assertTrue(output.toString().contains("Successfully processed: 0")); Assert.assertTrue(output.toString().contains("Backfill failed;")); Assert.assertFalse(output.toString().contains("Backfill completed successfully!")); - verify(transactions, never()).get(aryEq(ByteArray.fromLong(2))); - verify(transactions, never()).get(aryEq(ByteArray.fromLong(2049))); + verify(transactions, never()).get(aryEq(ByteArray.fromLong(3))); + verify(transactions, never()).get(aryEq(ByteArray.fromLong(2050))); + verify(bloom, never()).get(any(byte[].class)); + verify(bloom, never()).put(any(byte[].class), any(byte[].class)); for (int i = 0; i < failures.length; i++) { String message = "Error processing section " + (i == 0 ? "1 to 2047" : "2048 to 2050"); Assert.assertTrue(errors.toString().contains(message)); @@ -274,13 +283,13 @@ public void testDatabaseWriteFailurePreservesOriginalCause() { try (MockedStatic ignored = mockDatabases()) { Assert.assertEquals(1, execute("-e", "1")); verify(bloom).put(any(byte[].class), any(byte[].class)); - Assert.assertTrue(errors.toString().contains("Error processing block 1")); - Assert.assertFalse(output.toString().contains("Error processing block 1")); + Assert.assertTrue(errors.toString().contains("Error writing section 1 to 1")); + Assert.assertFalse(output.toString().contains("Error writing section 1 to 1")); Assert.assertTrue(output.toString().contains("Errors encountered: 1")); Assert.assertTrue(output.toString().contains("Successfully processed: 0")); Assert.assertFalse(output.toString().contains("Backfill completed successfully!")); ILoggingEvent event = appender.list.stream() - .filter(entry -> "Error processing block 1".equals(entry.getFormattedMessage())) + .filter(entry -> "Error writing section 1 to 1".equals(entry.getFormattedMessage())) .findFirst().orElse(null); Assert.assertNotNull(event); ThrowableProxy throwable = (ThrowableProxy) event.getThrowableProxy(); @@ -291,13 +300,104 @@ public void testDatabaseWriteFailurePreservesOriginalCause() { } @Test(timeout = 30_000) - public void testMalformedProtobufFailsWithoutWritingBloom() throws Exception { - writeSource(1, 1, 1); - openDb("transactionRetStore").put(ByteArray.fromLong(1), new byte[] {(byte) 0x80}); + public void testMalformedProtobufSkipsOnlyInvalidBlock() throws Exception { + writeSource(1, 3, 3); + openDb("transactionRetStore").put(ByteArray.fromLong(2), new byte[] {(byte) 0x80}); DbTool.close(); - Assert.assertEquals(1, execute()); - Assert.assertTrue(errors.toString().contains("Error processing block 1")); + Assert.assertEquals(1, execute("-s", "2", "-e", "2")); + Assert.assertTrue(errors.toString().contains("Error processing block 2")); Assert.assertTrue(readBloomEntries().isEmpty()); + Assert.assertEquals(1, execute()); + Assert.assertTrue(output.toString().contains("Errors encountered: 1")); + Assert.assertTrue(output.toString().contains("Successfully processed: 2")); + BitSet expected = new BitSet(); + expected.set(1); + expected.set(3); + assertIndexedBlocks(expected); + } + + @Test(timeout = 30_000) + public void testSectionReadsAndWritesEachChangedIndexOnce() throws Exception { + writeSource(1, 128, 128); + BitSet indexes = BitSet.valueOf(BloomUtils.createBloom(transactionRet)); + Assert.assertTrue(indexes.cardinality() > 1); + int unchangedIndex = indexes.nextSetBit(0); + BitSet existing = new BitSet(); + existing.set(512); + writeBloom(ByteUtil.compress(existing.toByteArray())); + BitSet expected = (BitSet) existing.clone(); + expected.set(1, 129); + openDb("section-bloom").put(bloomKey(0, unchangedIndex), + ByteUtil.compress(expected.toByteArray())); + DbTool.close(); + + for (int run = 0; run < 2; run++) { + int expectedWrites = run == 0 ? indexes.cardinality() - 1 : 0; + try (MockedStatic ignored = mockRealDatabases()) { + Assert.assertEquals(0, execute()); + Assert.assertTrue(output.toString().contains("Successfully processed: 128")); + Assert.assertTrue(output.toString().contains("Total bloom writes: " + expectedWrites)); + verify(bloom, times(indexes.cardinality())).get(any(byte[].class)); + verify(bloom, times(expectedWrites)).put(any(byte[].class), any(byte[].class)); + for (int bit = indexes.nextSetBit(0); bit >= 0; bit = indexes.nextSetBit(bit + 1)) { + verify(bloom).get(aryEq(bloomKey(0, bit))); + verify(bloom, times(run == 0 && bit != unchangedIndex ? 1 : 0)) + .put(aryEq(bloomKey(0, bit)), any(byte[].class)); + } + } + assertIndexedBlocks(expected); + } + } + + @Test(timeout = 30_000) + public void testPartialSectionFailureCanBeRerun() throws Exception { + for (boolean failOnRead : new boolean[] {false, true}) { + databaseRoot = temporaryFolder.newFolder(); + writeSource(1, 3, 3); + BitSet existing = new BitSet(); + existing.set(7); + writeBloom(ByteUtil.compress(existing.toByteArray())); + RuntimeException failure = new RuntimeException("index I/O failed"); + AtomicInteger attempts = new AtomicInteger(); + Answer failSecondOperation = invocation -> { + if (attempts.incrementAndGet() == 2) { + throw failure; + } + return invocation.callRealMethod(); + }; + try (MockedStatic ignored = mockRealDatabases()) { + if (failOnRead) { + doAnswer(failSecondOperation).when(bloom).get(any(byte[].class)); + } else { + doAnswer(failSecondOperation).when(bloom).put(any(byte[].class), any(byte[].class)); + } + Assert.assertEquals(1, execute()); + Assert.assertEquals(2, attempts.get()); + Assert.assertTrue(errors.toString().contains("Error writing section 1 to 3")); + Assert.assertTrue(output.toString().contains("Errors encountered: 1")); + Assert.assertTrue(output.toString().contains("Total blocks scanned: 3")); + Assert.assertTrue(output.toString().contains("Successfully processed: 0")); + Assert.assertTrue(output.toString().contains("Blocks with logs: 0")); + Assert.assertTrue(output.toString().contains("Total bloom writes: 1")); + } + BitSet indexes = BitSet.valueOf(BloomUtils.createBloom(transactionRet)); + Assert.assertTrue(indexes.cardinality() > 1); + BitSet expected = (BitSet) existing.clone(); + expected.set(1, 4); + Map partial = readBloomEntries(); + Assert.assertEquals(indexes.cardinality(), partial.size()); + for (int bit = indexes.nextSetBit(0); bit >= 0; bit = indexes.nextSetBit(bit + 1)) { + byte[] value = partial.get(ByteString.copyFrom(bloomKey(0, bit))); + Assert.assertNotNull(value); + Assert.assertEquals(bit == indexes.nextSetBit(0) ? expected : existing, + BitSet.valueOf(ByteUtil.decompress(value))); + } + Assert.assertEquals(0, execute()); + Assert.assertTrue(output.toString().contains("Successfully processed: 3")); + Assert.assertTrue(output.toString().contains( + "Total bloom writes: " + (indexes.cardinality() - 1))); + assertIndexedBlocks(expected); + } } @Test(timeout = 30_000) @@ -360,6 +460,7 @@ public void testRealBackfillMatchesNodeAcrossSectionBoundaryAndRerun() throws Ex TransactionRet second = createTransactionRet(82); TransactionRet empty = TransactionRet.newBuilder() .addTransactioninfo(TransactionInfo.getDefaultInstance()).build(); + openDb("transactionRetStore").put(ByteArray.fromLong(2046), second.toByteArray()); openDb("transactionRetStore").put(ByteArray.fromLong(2047), transactionRet.toByteArray()); openDb("transactionRetStore").put(ByteArray.fromLong(2048), second.toByteArray()); openDb("transactionRetStore").put(ByteArray.fromLong(2049), empty.toByteArray()); @@ -372,6 +473,8 @@ public void testRealBackfillMatchesNodeAcrossSectionBoundaryAndRerun() throws Ex SectionBloomStore nodeStore = null; try { nodeStore = new SectionBloomStore("section-bloom"); + nodeStore.initBlockSection(new TransactionRetCapsule(second.toByteArray())); + nodeStore.write(2046); nodeStore.initBlockSection(new TransactionRetCapsule(transactionRet.toByteArray())); nodeStore.write(2047); nodeStore.initBlockSection(new TransactionRetCapsule(second.toByteArray())); @@ -388,9 +491,12 @@ public void testRealBackfillMatchesNodeAcrossSectionBoundaryAndRerun() throws Ex writeBloom(ByteUtil.compress(existing.toByteArray())); } Assert.assertEquals(0, execute("-c", "2")); - Assert.assertTrue(output.toString().contains("Blocks with logs: 2")); - Assert.assertTrue(output.toString().contains("Successfully processed: 4")); + Assert.assertTrue(output.toString().contains("Blocks with logs: 3")); + Assert.assertTrue(output.toString().contains("Successfully processed: 5")); Assert.assertTrue(output.toString().contains("Processing 2 sections with 2 threads")); + if (run == 2) { + Assert.assertTrue(output.toString().contains("Total bloom writes: 0")); + } Assert.assertEquals(DbType.RocksDB, DbTool.getDbType(databaseRoot.toString(), "section-bloom")); Map expected = readNodeSections(nodeStore); @@ -484,6 +590,15 @@ private MockedStatic mockDatabases() { return dbTool; } + private MockedStatic mockRealDatabases() throws Exception { + transactions = openDb("transactionRetStore"); + properties = openDb("properties"); + bloom = spy(openDb("section-bloom")); + MockedStatic dbTool = mockDatabases(); + dbTool.when(DbTool::close).thenCallRealMethod(); + return dbTool; + } + private void assertNoDatabasesOpened(MockedStatic dbTool) { dbTool.verify(() -> DbTool.getDB(anyString(), anyString()), never()); dbTool.verify(() -> DbTool.getDB(anyString(), anyString(), any(DbType.class)), never()); @@ -529,12 +644,16 @@ private void writeBloom(byte[] value) throws Exception { } private void assertIndexedBlocks(int first, int last) throws Exception { + BitSet expected = new BitSet(); + expected.set(first, last + 1); + assertIndexedBlocks(expected); + } + + private void assertIndexedBlocks(BitSet expected) throws Exception { Map actual = readBloomEntries(); BitSet indexes = BitSet.valueOf(BloomUtils.createBloom(transactionRet)); Assert.assertFalse(indexes.isEmpty()); Assert.assertEquals(indexes.cardinality(), actual.size()); - BitSet expected = new BitSet(); - expected.set(first, last + 1); for (int bit = indexes.nextSetBit(0); bit >= 0; bit = indexes.nextSetBit(bit + 1)) { byte[] value = actual.get(ByteString.copyFrom(bloomKey(0, bit))); Assert.assertNotNull(value);