diff --git a/scripts/create_shut_down_cycle.ts b/scripts/create_shut_down_cycle.ts index bf01283..f4a54f2 100644 --- a/scripts/create_shut_down_cycle.ts +++ b/scripts/create_shut_down_cycle.ts @@ -1,4 +1,4 @@ -import { readFileSync } from 'fs' +import { readFileSync, writeFileSync } from 'fs' import * as path from 'path' import { join } from 'path' import { overrideDefaultConfig, config } from '../src/Config' @@ -10,7 +10,7 @@ import * as Logger from '../src/Logger' import { P2P } from '@shardeum-foundation/lib-types' import { addSigListeners } from '../src/State' import { computeCycleMarker } from '../src/Data/Cycles' -import { Utils as StringUtils } from '@shardeum-foundation/lib-types' +import { Utils as StringUtils, P2P as P2PTypes } from '@shardeum-foundation/lib-types' import { initAjvSchemas } from '../src/types/ajv/Helpers' import { initializeSerialization } from '../src/utils/serialization/SchemaHelpers' @@ -32,18 +32,53 @@ const archiversAtShutdown = [ }, ] +export interface NodeInitTxData { + publicKey: string + nodeId: string + startTime: number +} + +export interface NodeRewardTxData { + publicKey: string + nodeId: string + endTime: number +} + +interface Tx { + cycle: number + hash: string + priority: number + subQueueKey?: string + txData: NodeInitTxData | NodeRewardTxData + type: string +} + +interface TransactionEntry { + hash: string + tx: Tx +} + +interface BuildTxListResult { + txList: TransactionEntry[] + newTxAdd: Tx[] +} + const runProgram = async (): Promise => { initAjvSchemas() initializeSerialization() + // Override default config params from config file, env vars, and cli args const file = join(process.cwd(), 'archiver-config.json') overrideDefaultConfig(file) + // Set crypto hash keys from config const hashKey = config.ARCHIVER_HASH_KEY if (!hashKey) { throw new Error('ARCHIVER_HASH_KEY is required') } Crypto.setCryptoHashKey(hashKey) + + // Initialize logger let logsConfig try { logsConfig = StringUtils.safeJsonParse(readFileSync(path.resolve(__dirname, '../archiver-log.json'), 'utf8')) @@ -57,21 +92,29 @@ const runProgram = async (): Promise => { if (logsConfig.saveConsoleOutput) { startSaving(join(baseDir, logsConfig.dir)) } + await dbstore.initializeDB(config) addSigListeners() - const txListPath = path.join(__dirname, '..', 'tx-list-restore.json') - const rawData = readFileSync(txListPath, 'utf8') - const ngtJson = JSON.parse(rawData) + let latestCycles = await CycleDB.queryLatestCycleRecords(1) + let latestCycleRecord = latestCycles[0] + console.log('latestCycleRecord before', latestCycleRecord) - const txListHash = Crypto.hashObj(ngtJson) + console.log(`Building txList by replaying cycles backwards from ${latestCycleRecord.counter}...`) - let latestCycle = await CycleDB.queryLatestCycleRecords(1) - let latestCycleRecord = latestCycle[0] - console.log('latestCycleRecord before', latestCycleRecord) - const newCycleRecord = { + const { txList, newTxAdd } = await buildTxList(latestCycleRecord) + const txListHash = Crypto.hashObj(txList) + + // Save the txList + const txListPath = path.join(__dirname, '..', 'tx-list-restore.json') + console.log(`Writing ${txList.length} entries to ${txListPath}...`) + writeFileSync(txListPath, JSON.stringify(txList, null, 2), 'utf8') + + // Create the shutdown cycle record + const shutdownCycleRecord = { ...latestCycleRecord, counter: latestCycleRecord.counter + 1, + start: latestCycleRecord.start + latestCycleRecord.duration, mode: 'shutdown' as P2P.ModesTypes.Record['mode'], removed: ['all'], archiversAtShutdown: archiversAtShutdown.map((archiver) => { @@ -84,21 +127,270 @@ const runProgram = async (): Promise => { standbyAdd: [], standbyRemove: [], txlisthash: txListHash, - txadd: [], + txadd: newTxAdd, txremove: [], } - delete newCycleRecord.marker - const marker = computeCycleMarker(newCycleRecord) - newCycleRecord.marker = marker - // console.log('newCycleRecord', newCycleRecord) + + // Remove the old marker and compute the new one + delete shutdownCycleRecord.marker + const marker = computeCycleMarker(shutdownCycleRecord) + shutdownCycleRecord.marker = marker + + // console.log('shutdownCycleRecord', shutdownCycleRecord) await CycleDB.insertCycle({ - counter: newCycleRecord.counter, - cycleMarker: newCycleRecord.marker, - cycleRecord: newCycleRecord, + counter: shutdownCycleRecord.counter, + cycleMarker: shutdownCycleRecord.marker, + cycleRecord: shutdownCycleRecord, }) - latestCycle = await CycleDB.queryLatestCycleRecords(1) - latestCycleRecord = latestCycle[0] + + latestCycles = await CycleDB.queryLatestCycleRecords(1) + latestCycleRecord = latestCycles[0] console.log('latestCycleRecord after', latestCycleRecord) + await dbstore.closeDatabase() } + +/** + * Rebuilds the transaction list by replaying cycles backwards and adding shutdown rewards. + * + * This function: + * 1. Replays cycles backwards from the given cycle + * 2. Reconstructs the txList by applying txadd/txremove operations + * 3. Verifies the txlisthash at each step + * 4. Adds nodeReward transactions for all active nodes in the shutdown cycle + * + * @param cycle - The cycle to start replaying from + * @returns Object containing the complete txList and the new txadd entries for shutdown + */ +async function buildTxList(cycle: P2PTypes.CycleCreatorTypes.CycleData): Promise { + const txList: TransactionEntry[] = [] + const txRemoveSet = new Set() + + const currentTxListHash = cycle.txlisthash + + // Replay cycles backwards to reconstruct the txList + let cycleCounter = cycle.counter + for (; cycleCounter >= 0; cycleCounter--) { + const cycleData = + cycleCounter === cycle.counter ? { cycleRecord: cycle } : await CycleDB.queryCycleByCounter(cycleCounter) + const cycleRecord: P2PTypes.CycleCreatorTypes.CycleData | undefined = cycleData?.cycleRecord + + if (!cycleRecord) { + console.warn(`No cycle record found for counter ${cycleCounter}, stopping replay.`) + break + } + + console.log( + `Processing cycle ${cycleCounter}: txlisthash=${cycleRecord.txlisthash}, txadd=${cycleRecord.txadd?.length || 0}, txremove=${cycleRecord.txremove?.length || 0}` + ) + + // Apply txadd: add transactions that were added in this cycle + if (Array.isArray(cycleRecord.txadd)) { + for (const tx of cycleRecord.txadd) { + // Skip if this transaction was removed in a later cycle + if (txRemoveSet.has(tx.hash)) { + continue + } + + // Remove signature from txData for consistency + const txDataWithoutSign: NodeInitTxData | NodeRewardTxData = { ...tx.txData } + if ('sign' in txDataWithoutSign) { + delete txDataWithoutSign.sign + } + + const entry: TransactionEntry = { + hash: tx.hash, + tx: { + cycle: tx.cycle, + hash: tx.hash, + priority: tx.priority, + ...(tx.subQueueKey && { subQueueKey: tx.subQueueKey }), + txData: txDataWithoutSign, + type: tx.type, + }, + } + + sortedInsert(txList, entry) + } + } + + // Record txremove hash; so that we don't add its entry in the txList while replaying + if (Array.isArray(cycleRecord.txremove)) { + for (const tx of cycleRecord.txremove) { + txRemoveSet.add(tx.txHash) + } + } + + // Verify hash against txlisthash + const computedHash = Crypto.hashObj(txList) + if (computedHash !== currentTxListHash) { + console.warn( + `txlisthash mismatch at cycle ${cycleCounter}: stored=${cycleRecord.txlisthash}, computed=${computedHash}` + ) + } else { + console.info(`Found matching txlisthash at cycle ${cycleCounter}: ${computedHash}`) + break + } + } + + console.info(`Replayed cycles from ${cycle.counter} down to ${cycleCounter}. txList length: ${txList.length}`) + console.dir(txList, { depth: null }) + + // Create nodeReward transactions for all active nodes at shutdown + const shutdownCycleNumber = cycle.counter + 1 + const shutdownTimestamp = cycle.start + cycle.duration + const activeNodes = await getActiveNodesAtCycle(cycle) + + const newTxAddEntries: TransactionEntry[] = [] + for (const [nodeId, publicKey] of activeNodes) { + const rewardTxData: NodeRewardTxData = { + publicKey, + nodeId, + endTime: shutdownTimestamp, + } + const txHash = Crypto.hashObj(rewardTxData) + + const entry: TransactionEntry = { + hash: txHash, + tx: { + cycle: shutdownCycleNumber, + hash: txHash, + priority: 0, + subQueueKey: publicKey, + txData: rewardTxData, + type: 'nodeReward', + }, + } + + sortedInsert(txList, entry) + sortedInsert(newTxAddEntries, entry) + } + + // Extract just the Tx objects for the cycle record + const newTxAdd: Tx[] = newTxAddEntries.map((entry) => entry.tx) + + console.info( + `Added ${newTxAdd.length} nodeReward transactions for shutdown cycle. Final txList length: ${txList.length}` + ) + console.dir(txList, { depth: null }) + + return { txList, newTxAdd } +} + +/** + * Insert into txList using exactly the same ordering as ServiceQueue.sortedInsert: + * - sort by cycle ASC + * - for same cycle: priority DESC + * - for same cycle & priority: hash ASC + */ +function sortedInsert(list: TransactionEntry[], entry: TransactionEntry): void { + const index = list.findIndex( + (item) => + item.tx.cycle > entry.tx.cycle || + (item.tx.cycle === entry.tx.cycle && item.tx.priority < entry.tx.priority) || + (item.tx.cycle === entry.tx.cycle && item.tx.priority === entry.tx.priority && item.hash > entry.hash) + ) + if (index === -1) { + list.push(entry) + } else { + list.splice(index, 0, entry) + } +} + +/** + * Returns all active nodes at the given cycle + * @param cycle - The cycle to compute active nodes for + * @returns Map of public keys to node IDs for all active nodes + */ +async function getActiveNodesAtCycle( + cycle: P2PTypes.CycleCreatorTypes.CycleData +): Promise> { + const activeNodesCount = cycle.active + console.log(`Total active nodes at cycle ${cycle.counter}: ${activeNodesCount}`) + + const activeNodesMap = new Map() + const removedNodeIds = new Set() + + // For about 5 cycles after current cycle, track the unrewarded nodes for removed nodes + const numUnrewardedCycles = 5 + const rewardedNodeIds = new Set() + const unRewardedNodeIds = new Set() + const unRewardedNodesMap = new Map() + + for (const tx of cycle.txadd) { + if (tx.type === 'nodeReward') { + rewardedNodeIds.add(tx.txData.nodeId) + } + } + + // Replay backwards to find all active nodes + let cycleCounter = cycle.counter - 1 + for (; cycleCounter >= 0; cycleCounter--) { + const cycleData = await CycleDB.queryCycleByCounter(cycleCounter) + const cycleRecord: P2PTypes.CycleCreatorTypes.CycleData | undefined = cycleData?.cycleRecord + + if (!cycleRecord) { + console.warn(`No cycle record found for counter ${cycleCounter}, stopping active node search.`) + break + } + + // Add activated nodes from this cycle to the active set + for (let i = 0; i < cycleRecord.activated.length; i++) { + const nodeId = cycleRecord.activated[i] + const nodePubKey = cycleRecord.activatedPublicKeys[i] + if (rewardedNodeIds.has(nodeId)) { + continue + } + if (unRewardedNodeIds.has(nodeId)) { + unRewardedNodesMap.set(nodeId, nodePubKey) + unRewardedNodeIds.delete(nodeId) + } + if (removedNodeIds.has(nodeId)) { + continue + } + if (activeNodesMap.size !== activeNodesCount) { + activeNodesMap.set(nodeId, nodePubKey) + } + } + + // Stop if we've found all expected active nodes + if (activeNodesMap.size === activeNodesCount && unRewardedNodeIds.size === 0) { + break + } + + // Add removed nodes from this cycle to the exclusion set + for (const nodeId of cycleRecord.removed) { + removedNodeIds.add(nodeId) + // If the removed node was unrewarded, add it to the unrewarded set + if (!rewardedNodeIds.add(nodeId) && cycle.counter - cycleCounter <= numUnrewardedCycles) { + unRewardedNodeIds.add(nodeId) + } + } + for (const nodeId of cycleRecord.apoptosized) { + removedNodeIds.add(nodeId) + } + for (const nodeId of cycleRecord.appRemoved) { + removedNodeIds.add(nodeId) + } + + for (const tx of cycleRecord.txadd) { + if (tx.type === 'nodeReward') { + rewardedNodeIds.add(tx.txData.nodeId) + } + } + } + + // Add unrewarded nodes to the active set + for (const [nodeId, publicKey] of unRewardedNodesMap) { + activeNodesMap.set(nodeId, publicKey) + } + + console.log(`Found ${activeNodesMap.size} active nodes at cycle ${cycle.counter}:`) + for (const [nodeId, publicKey] of activeNodesMap) { + console.log(` Node ID: ${nodeId}, Public Key: ${publicKey}`) + } + + return activeNodesMap +} + runProgram() diff --git a/scripts/update_network_account.ts b/scripts/update_network_account.ts index 55c065f..449bf63 100644 --- a/scripts/update_network_account.ts +++ b/scripts/update_network_account.ts @@ -56,12 +56,17 @@ const runProgram = async (): Promise => { } // If there is a validator config in the listOfChanges that need to be overridden at the network restart, we can add it here. eg: // networkAccount.data.listOfChanges.push({ change: { p2p: { minNodes: 150 } }, cycle: 55037 }) + // networkAccount.data.listOfChanges.push({ + // appData: { activeVersion, latestVersion, minVersion }, + // change: {}, + // cycle: 0, // Set shutdown cycle number for tracking of the config changes + // }) const calculatedAccountHash = accountSpecificHash(networkAccount.data) networkAccount.hash = calculatedAccountHash networkAccount.data.hash = calculatedAccountHash - await AccountDB.insertAccount(networkAccount) + await AccountDB.updateAccount(networkAccount) console.log('Network account after', networkAccount) await dbstore.closeDatabase() } diff --git a/src/Config.ts b/src/Config.ts index d3b0bb6..6539b5e 100644 --- a/src/Config.ts +++ b/src/Config.ts @@ -59,9 +59,9 @@ export interface Config { dataLogWriter: { dirName: string maxLogFiles: number - maxReceiptBytes: number - maxCycleBytes: number - maxOriginalTxBytes: number + maxReceiptEntries: number + maxCycleEntries: number + maxOriginalTxEntries: number } experimentalSnapshot: boolean failedBucketsDir: string @@ -179,10 +179,10 @@ let config: Config = { dataLogWrite: true, dataLogWriter: { dirName: 'data-logs', - maxLogFiles: 20, - maxReceiptBytes: 10 * 1024 * 1024, // 1MB - Should be >= max TPS * avg receipt size - maxCycleBytes: 10 * 1024 * 1024, // 50KB - cycles are smaller - maxOriginalTxBytes: 10 * 1024 * 1024, // 1MB - Should be >= max TPS * avg tx size + maxLogFiles: 10, + maxReceiptEntries: 10000, // Should be >= max TPS experienced by the network. + maxCycleEntries: 500, + maxOriginalTxEntries: 10000, // Should be >= max TPS experienced by the network. }, experimentalSnapshot: true, failedBucketsDir: 'failed-buckets', diff --git a/src/Data/Collector.ts b/src/Data/Collector.ts index 4f1f2d8..82da3b1 100644 --- a/src/Data/Collector.ts +++ b/src/Data/Collector.ts @@ -762,7 +762,6 @@ export const storeReceiptData = async ( const timestamp = receipt?.tx?.timestamp if (!txId || !timestamp) { - logReceiptData(receipt) continue } if ( @@ -771,7 +770,6 @@ export const storeReceiptData = async ( (receiptsInValidationMap.has(txId) && receiptsInValidationMap.get(txId) === timestamp)) ) { if (config.VERBOSE) console.log('RECEIPT', 'Skip', txId, timestamp, senderInfo) - logReceiptData(receipt, txId, timestamp) continue } if (config.VERBOSE) console.log('RECEIPT', 'Validate', txId, timestamp, senderInfo) @@ -783,7 +781,6 @@ export const storeReceiptData = async ( receiptsInValidationMap.delete(txId) if (nestedCountersInstance) nestedCountersInstance.countEvent('receipt', 'Invalid_receipt_validation_failed') if (profilerInstance) profilerInstance.profileSectionEnd('Validate_receipt') - logReceiptData(receipt, txId, timestamp) continue } @@ -791,7 +788,6 @@ export const storeReceiptData = async ( // only consider this for EVM txns and Non Global Internal Txns const result = await checkIfValidOverwrite(receipt, txId) if (!result && checkpoint) { - logReceiptData(receipt, txId, timestamp) continue // if the incoming receipt has a status of 0, do not allow it to overwrite a receipt of status 1 } } @@ -830,7 +826,6 @@ export const storeReceiptData = async ( if (nestedCountersInstance) nestedCountersInstance.countEvent('receipt', 'Invalid_receipt_verification_failed') if (profilerInstance) profilerInstance.profileSectionEnd('Validate_receipt') - logReceiptData(receipt, txId, timestamp) continue } @@ -847,7 +842,6 @@ export const storeReceiptData = async ( if (nestedCountersInstance) nestedCountersInstance.countEvent('receipt', 'Invalid_receipt_verification_failed') if (profilerInstance) profilerInstance.profileSectionEnd('Verify_archiver_receipt') - logReceiptData(receipt, txId, timestamp) continue } // console.log('offload receipt result', txId, timestamp, result) @@ -866,7 +860,6 @@ export const storeReceiptData = async ( if (result.success === false) { receiptsInValidationMap.delete(txId) if (profilerInstance) profilerInstance.profileSectionEnd('Validate_receipt') - logReceiptData(receipt, txId, timestamp) continue } } diff --git a/src/Data/Data.ts b/src/Data/Data.ts index c83a2fc..59fce47 100644 --- a/src/Data/Data.ts +++ b/src/Data/Data.ts @@ -969,10 +969,7 @@ async function syncFromNetworkConfig(): Promise { updateConfig({ minSigRequiredForArchiverWhitelist }) } if (!allowedArchiversManager.getCurrentConfig() || isAllowedArchiversUpdateNeeded) - allowedArchiversManager.setGlobalAccountConfig( - config.multisigKeys, - config.minSigRequiredForArchiverWhitelist - ) + allowedArchiversManager.setGlobalAccountConfig(config.multisigKeys, config.minSigRequiredForArchiverWhitelist) } return tallyItem } catch (error) { @@ -2971,9 +2968,11 @@ function validateCerts( return false } - if (NodeList.activeListByIdSorted.some((node) => node.publicKey === cleanCert.sign.owner) === false) { + // Check if the cert signer is a valid node in the network (must be in joined, syncing, or active lists) + // https://github.com/shardeum/core/blob/32d29a0a29ea610b9797ca98b929666eb9e20247/src/p2p/CycleCreator.ts#L1030 + if (NodeList.byPublicKey.has(cleanCert.sign.owner) === false) { nestedCountersInstance.countEvent('validateCerts', 'badOwner', 1) - Logger.mainLogger.warn(`validateCerts: bad owner ${cleanCert.sign.owner} not found in active nodes`) + Logger.mainLogger.warn(`validateCerts: bad owner ${cleanCert.sign.owner} not found in nodes list`) return false } if (certSigners.has(cert.sign.owner)) { diff --git a/src/Data/DataLogWriter.ts b/src/Data/DataLogWriter.ts index 391d0ae..326d432 100644 --- a/src/Data/DataLogWriter.ts +++ b/src/Data/DataLogWriter.ts @@ -19,7 +19,7 @@ class DataLogWriter { dataLogWriteStream: WriteStream | null dataWriteIndex: number dataLogFilePath: string - totalNumberOfBytes: number + totalNumberOfEntries: number activeLogFileName: string activeLogFilePath: string writeQueue: string[] @@ -28,7 +28,7 @@ class DataLogWriter { constructor( public dataName: string, public logCounter: number, - public maxNumberBytesPerLog: number + public maxNumberEntriesPerLog: number ) { this.logDir = `${LOG_WRITER_CONFIG.dirName}/${config.ARCHIVER_IP}_${config.ARCHIVER_PORT}` this.maxLogCounter = LOG_WRITER_CONFIG.maxLogFiles @@ -37,7 +37,7 @@ class DataLogWriter { this.activeLogFileName = `active-${dataName}-log.txt` this.activeLogFilePath = path.join(this.logDir, this.activeLogFileName) this.dataLogFilePath = path.join(this.logDir, `${dataName}-log${logCounter}.txt`) - this.totalNumberOfBytes = 0 + this.totalNumberOfEntries = 0 this.writeQueue = [] this.isWriting = false } @@ -58,16 +58,18 @@ class DataLogWriter { console.log(`> DataLogWriter: Active log file: ${this.dataName}-log${this.logCounter}.txt`) this.dataLogFilePath = path.join(this.logDir, `${this.dataName}-log${this.logCounter}.txt`) // eslint-disable-next-line security/detect-non-literal-fs-filename - const stats = await fs.stat(this.dataLogFilePath) - this.totalNumberOfBytes += stats.size - console.log(`> DataLogWriter: Total ${this.dataName} Bytes: ${this.totalNumberOfBytes}`) + const data = await fs.readFile(this.dataLogFilePath, { + encoding: 'utf8', + }) + this.totalNumberOfEntries += data.split('\n').length - 1 + console.log(`> DataLogWriter: Total ${this.dataName} Entries: ${this.totalNumberOfEntries}`) // eslint-disable-next-line security/detect-non-literal-fs-filename this.dataLogWriteStream = createWriteStream(this.dataLogFilePath, { flags: 'a' }) - if (this.totalNumberOfBytes >= this.maxNumberBytesPerLog) { - // Finish the log file with the total number of bytes. - await this.appendData(`End: Number of bytes: ${this.totalNumberOfBytes}\n`) + if (this.totalNumberOfEntries >= this.maxNumberEntriesPerLog) { + // Finish the log file with the total number of entries. + await this.appendData(`End: Number of entries: ${this.totalNumberOfEntries}\n`) await this.endStream() - this.totalNumberOfBytes = 0 + this.totalNumberOfEntries = 0 await this.rotateLogFile() await this.setActiveLog() } @@ -156,17 +158,17 @@ class DataLogWriter { while (this.writeQueue.length) { try { for (let i = 0; i < this.writeQueue.length; i++) { - if (this.totalNumberOfBytes >= this.maxNumberBytesPerLog) { - await this.appendData(`End: Number of bytes: ${this.totalNumberOfBytes}\n`) + if (this.totalNumberOfEntries === this.maxNumberEntriesPerLog) { + await this.appendData(`End: Number of entries: ${this.totalNumberOfEntries}\n`) await this.endStream() - this.totalNumberOfBytes = 0 + this.totalNumberOfEntries = 0 await this.rotateLogFile() await this.setActiveLog() } // eslint-disable-next-line security/detect-object-injection await this.appendData(this.writeQueue[i]) this.dataWriteIndex += 1 - this.totalNumberOfBytes += Buffer.byteLength(this.writeQueue[i], 'utf8') + this.totalNumberOfEntries += 1 } // console.log('-->> Write queue length: ', this.writeQueue.length) this.writeQueue.splice(0, this.dataWriteIndex) @@ -195,7 +197,7 @@ class DataLogWriter { appendData(data: string): Promise { // Check if we should continue writing - const canContinueToWrite = this.dataLogWriteStream!.write(data + '\n') + const canContinueToWrite = this.dataLogWriteStream!.write(data) if (!canContinueToWrite) { // Wait for drain event to continue writing @@ -210,7 +212,7 @@ class DataLogWriter { return new Promise((resolve, reject) => { try { this.dataLogWriteStream!.end(() => { - console.log(`✅ Finished writing ${this.totalNumberOfBytes} bytes.`) + console.log(`✅ Finished writing ${this.totalNumberOfEntries}.`) resolve() }) } catch (e) { @@ -236,10 +238,10 @@ export let ReceiptOverwriteLogWriter: DataLogWriter * @returns {Promise} A promise that resolves when all log writers are initialized. */ export async function initDataLogWriter(): Promise { - CycleLogWriter = new DataLogWriter('cycle', 1, LOG_WRITER_CONFIG.maxCycleBytes) - ReceiptLogWriter = new DataLogWriter('receipt', 1, LOG_WRITER_CONFIG.maxReceiptBytes) - OriginalTxDataLogWriter = new DataLogWriter('originalTx', 1, LOG_WRITER_CONFIG.maxOriginalTxBytes) - ReceiptOverwriteLogWriter = new DataLogWriter('receiptOverwrite', 1, LOG_WRITER_CONFIG.maxOriginalTxBytes) + CycleLogWriter = new DataLogWriter('cycle', 1, LOG_WRITER_CONFIG.maxCycleEntries) + ReceiptLogWriter = new DataLogWriter('receipt', 1, LOG_WRITER_CONFIG.maxReceiptEntries) + OriginalTxDataLogWriter = new DataLogWriter('originalTx', 1, LOG_WRITER_CONFIG.maxOriginalTxEntries) + ReceiptOverwriteLogWriter = new DataLogWriter('receiptOverwrite', 1, LOG_WRITER_CONFIG.maxOriginalTxEntries) await Promise.all([ CycleLogWriter.init(), ReceiptLogWriter.init(), diff --git a/src/dbstore/index.ts b/src/dbstore/index.ts index 5324841..37ea668 100644 --- a/src/dbstore/index.ts +++ b/src/dbstore/index.ts @@ -74,6 +74,11 @@ export const initializeDB = async (config: Config): Promise => { receiptDatabase, 'CREATE INDEX if not exists `receipts_cycle_timestamp` ON `receipts` (`cycle` ASC, `timestamp` ASC)' ) + // Composite index for cursor-based pagination (optimal for parallel sync) + await runCreate( + receiptDatabase, + 'CREATE INDEX if not exists `receipts_cycle_timestamp_receiptId` ON `receipts` (`cycle` ASC, `timestamp` ASC, `receiptId` ASC)' + ) await runCreate( originalTxDataDatabase, 'CREATE TABLE if not exists `originalTxsData` (`txId` TEXT NOT NULL, `timestamp` BIGINT NOT NULL, `cycle` NUMBER NOT NULL, `originalTxData` JSON NOT NULL, PRIMARY KEY (`txId`, `timestamp`))' @@ -90,6 +95,11 @@ export const initializeDB = async (config: Config): Promise => { originalTxDataDatabase, 'CREATE INDEX if not exists `originalTxsData_cycle_timestamp` ON `originalTxsData` (`cycle` ASC, `timestamp` ASC)' ) + // Composite index for cursor-based pagination (optimal for parallel sync) + await runCreate( + originalTxDataDatabase, + 'CREATE INDEX if not exists `originalTxsData_cycle_timestamp_txId` ON `originalTxsData` (`cycle` ASC, `timestamp` ASC, `txId` ASC)' + ) await runCreate( originalTxDataDatabase, 'CREATE INDEX if not exists `originalTxsData_txId` ON `originalTxsData` (`txId`)' diff --git a/src/dbstore/sqlite3storage.ts b/src/dbstore/sqlite3storage.ts index 388bc27..07f475a 100644 --- a/src/dbstore/sqlite3storage.ts +++ b/src/dbstore/sqlite3storage.ts @@ -10,10 +10,12 @@ export const createDB = async (dbPath: string, dbName: string): Promise { if (time > 500 && time < 1000) { console.log('SLOW QUERY', process.pid, sql, time)