From 461542b53a633a18e74d3655e0445d29dfe5c948 Mon Sep 17 00:00:00 2001 From: Will Meister Date: Mon, 24 Aug 2020 14:48:05 -0500 Subject: [PATCH 1/4] Adding fix to block processing errors including partial state recovery --- docker-compose.microservices.example.yml | 8 +- .../src/app/queued-persisted-processor.ts | 24 +- .../app/queued-persisted-processor.spec.ts | 119 +++++++- .../rollup-core/src/app/data/data-service.ts | 56 +++- .../data/producers/l1-chain-data-persister.ts | 38 ++- .../src/types/data/l1-data-service.ts | 17 +- packages/rollup-core/src/types/types.ts | 7 + .../test/app/l1-chain-data-persister.spec.ts | 255 +++++++++++++++++- .../test/db/l2-data-service.dbspec.ts | 15 ++ packages/rollup-services/src/exec/services.ts | 30 ++- .../test/test-submit-to-l2.spec.ts | 15 +- 11 files changed, 533 insertions(+), 51 deletions(-) diff --git a/docker-compose.microservices.example.yml b/docker-compose.microservices.example.yml index 7a55821a7d6..4a78d932c13 100644 --- a/docker-compose.microservices.example.yml +++ b/docker-compose.microservices.example.yml @@ -58,13 +58,13 @@ services: - CANONICAL_CHAIN_MIN_BATCH_SIZE=50 # The minimum batch size to build -- if fewer than this number of transactions are ready, a batch will not be created (defaults to 10) - CANONICAL_CHAIN_MAX_BATCH_SIZE=50 # The maximum batch size to build -- if more than this number of transactions are ready, they will be split into multiple batches of at most this size (defaults to 100) - CANONICAL_CHAIN_BATCH_CREATOR_PERIOD_MILLIS=30000 # The period in millis at which the CanonicalChainBatchCreator should attempt to create Canonical Chain Batches (defaults to 10,000) - # Canonical Transaction Chain Batch Submitter (needs Postgres, L1 Node, L1 Submitters, and CANONICAL_TRANSACTION_CHAIN_CONTRACT_ADDRESS vars above) + # Canonical Transaction Chain Batch Submitter / Finalizer (needs Postgres, L1 Node, L1 Submitters, and CANONICAL_TRANSACTION_CHAIN_CONTRACT_ADDRESS vars above) - CANONICAL_CHAIN_BATCH_SUBMITTER_PERIOD_MILLIS # The period in millis at which the CanonicalChainBatchCreator should attempt to create Canonical Chain Batches (defaults to 10,000) # State Commitment Chain Batch Creator (needs Postgres vars above) - STATE_COMMITMENT_CHAIN_MIN_BATCH_SIZE=40 # The minimum batch size to build -- if fewer than this number of transactions are ready, a batch will not be created (defaults to 10) - STATE_COMMITMENT_CHAIN_MAX_BATCH_SIZE=500 # The maximum batch size to build -- if more than this number of transactions are ready, they will be split into multiple batches of at most this size (defaults to 100) - STATE_COMMITMENT_CHAIN_BATCH_CREATOR_PERIOD_MILLIS=30000 # The period in millis at which the StateCommitmentChainBatchCreator should attempt to create StateCommitmentChain Batches (defaults to 10,000) - # State Commitment Chain Batch Submitter (needs Postgres, L1 Node, L1 Submitters, STATE_COMMITMENT_CHAIN_CONTRACT_ADDRESS vars above) + # State Commitment Chain Batch Submitter / Finalizer (needs Postgres, L1 Node, L1 Submitters, STATE_COMMITMENT_CHAIN_CONTRACT_ADDRESS vars above) - STATE_COMMITMENT_CHAIN_BATCH_SUBMITTER_PERIOD_MILLIS # The period in millis at which the StateCommitmentChainBatchCreator should attempt to create StateCommitmentChain Batches (defaults to 10,000) # Fraud Detector - FRAUD_DETECTOR_PERIOD_MILLIS # The period in millis at which the FraudDetector should run (defaults to 10,000) @@ -75,9 +75,9 @@ services: - RUN_GETH_SUBMISSION_QUEUER=1 # Set to anything to run Geth Submission Queuer - RUN_QUEUED_GETH_SUBMITTER=1 # Set to anything to run Queued Geth Submitter - RUN_CANONICAL_CHAIN_BATCH_CREATOR=1 # Set to anything to run Canonical Chain Batch Creator - - RUN_CANONICAL_CHAIN_BATCH_SUBMITTER=1 # Set to anything to run Canonical Chain Batch Submitter + - RUN_CANONICAL_CHAIN_BATCH_SUBMITTER=1 # Set to anything to run Canonical Chain Batch Submitter & Finalizer - RUN_STATE_COMMITMENT_CHAIN_BATCH_CREATOR=1 # Set to anything to run State Commitment Chain Batch Creator - - RUN_STATE_COMMITMENT_CHAIN_BATCH_SUBMITTER=1 # Set to anything to run State Commitment Chain Batch Submitter + - RUN_STATE_COMMITMENT_CHAIN_BATCH_SUBMITTER=1 # Set to anything to run State Commitment Chain Batch Submitter & Finalizer - RUN_FRAUD_DETECTOR=1 # Set to anything to run Fraud Detector postgres: diff --git a/packages/core-db/src/app/queued-persisted-processor.ts b/packages/core-db/src/app/queued-persisted-processor.ts index 86611964703..522bd28f41b 100644 --- a/packages/core-db/src/app/queued-persisted-processor.ts +++ b/packages/core-db/src/app/queued-persisted-processor.ts @@ -28,7 +28,8 @@ export abstract class BaseQueuedPersistedProcessor protected constructor( private readonly db: DB, private readonly persistenceKey: string, - startIndex: number = 0 + startIndex: number = 0, + private readonly retrySleepDelayMillis: number = 1000 ) { this.initialized = false this.nextIndexToProcess = startIndex @@ -70,13 +71,25 @@ export abstract class BaseQueuedPersistedProcessor return } + try { + await this.setNextToProcess(index + 1) + } catch (e) { + this.log(`Error setting next to process to ${index + 1}!`, e) + throw e + } + this.setLastProcessed(index).then(async () => { + this.log( + `Attempting to fetch index ${this.nextIndexToProcess} from storage` + ) const nextItem = await this.fetchItem(this.nextIndexToProcess) if (!!nextItem) { this.log( `Index ${this.nextIndexToProcess} was already stored. Handling it now.` ) - await this.handleIfReady(this.nextIndexToProcess, nextItem) + setTimeout(() => { + this.handleIfReady(this.nextIndexToProcess, nextItem) + }, 0) } else { this.log( `Have not received index ${this.nextIndexToProcess} yet. Waiting...` @@ -222,13 +235,12 @@ export abstract class BaseQueuedPersistedProcessor (allowRetries && index === this.nextIndexToProcess - 1)) ) { try { - await this.setNextToProcess(index + 1) this.log(`Handling index ${index}.`) await this.handleNextItem(index, item) } catch (e) { logError(log, `Error handling item ${index}. Going to retry.`, e) - await sleep(1000) - return this.handleIfReady(index, item, allowRetries) + await sleep(this.retrySleepDelayMillis) + return this.handleIfReady(index, item, true) } } else { this.log( @@ -306,7 +318,7 @@ export abstract class BaseQueuedPersistedProcessor * Sets the next index to process, persisting the updated index in case of failure. * @param index The index to set Processed to. */ - private async setNextToProcess(index: number): Promise { + protected async setNextToProcess(index: number): Promise { await this.db.put( this.getStorageKey( BaseQueuedPersistedProcessor.NEXT_INDEX_TO_PROCESS_KEY diff --git a/packages/core-db/test/app/queued-persisted-processor.spec.ts b/packages/core-db/test/app/queued-persisted-processor.spec.ts index a405d65cf6f..2d448a9587e 100644 --- a/packages/core-db/test/app/queued-persisted-processor.spec.ts +++ b/packages/core-db/test/app/queued-persisted-processor.spec.ts @@ -10,22 +10,40 @@ import { DB } from '../../src/types/db' class DummyQueuedPersistedProcessor extends BaseQueuedPersistedProcessor< string > { + public throwOnceHandlingNextItem: boolean = false + public throwOnceOnSettingNextToProcess: boolean = false public handledQueue: string[] public static async create( db: DB, - persistenceKey: string + persistenceKey: string, + startIndex: number = 0, + retrySleepDelayMillis: number = 1000 ): Promise { - const processor = new DummyQueuedPersistedProcessor(db, persistenceKey) + const processor = new DummyQueuedPersistedProcessor( + db, + persistenceKey, + startIndex, + retrySleepDelayMillis + ) await processor.init() return processor } - private constructor(db: DB, persistenceKey: string) { - super(db, persistenceKey) + private constructor( + db: DB, + persistenceKey: string, + startIndex: number = 0, + retrySleepDelayMillis: number = 1000 + ) { + super(db, persistenceKey, startIndex, retrySleepDelayMillis) this.handledQueue = [] } protected async handleNextItem(index: number, item: string): Promise { + if (this.throwOnceHandlingNextItem) { + this.throwOnceHandlingNextItem = false + throw Error('you told me to throw in handleNextItem.') + } this.handledQueue.push(item) } @@ -36,6 +54,14 @@ class DummyQueuedPersistedProcessor extends BaseQueuedPersistedProcessor< protected async deserializeItem(itemBuffer: Buffer): Promise { return itemBuffer.toString('utf-8') } + + protected async setNextToProcess(index: number): Promise { + if (this.throwOnceOnSettingNextToProcess) { + this.throwOnceOnSettingNextToProcess = false + throw Error('you told me to throw in setNextToProcess.') + } + return super.setNextToProcess(index) + } } describe('Queued Persisted Processor', () => { @@ -45,7 +71,12 @@ describe('Queued Persisted Processor', () => { beforeEach(async () => { db = newInMemoryDB() - processor = await DummyQueuedPersistedProcessor.create(db, persistenceKey) + processor = await DummyQueuedPersistedProcessor.create( + db, + persistenceKey, + 0, + 100 + ) }) describe('Fresh start', () => { @@ -131,6 +162,84 @@ describe('Queued Persisted Processor', () => { `Incorrect item processed!` ) }) + + it('retries processing item if handleNextItemThrows', async () => { + const first = 'Number 0!' + const second = 'Number 1!' + await processor.add(0, first) + await processor.add(1, second) + await sleep(10) + processor.handledQueue.length.should.equal( + 1, + `Incorrect number processed!` + ) + processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) + + processor.throwOnceHandlingNextItem = true + await processor.markProcessed(0) + await sleep(10) + + processor.handledQueue.length.should.equal( + 1, + `There should still only be one item processed! Should fail and retry after 100 millis` + ) + + await sleep(200) + + processor.handledQueue.length.should.equal( + 2, + `Incorrect number processed!` + ) + processor.handledQueue[1].should.equal( + second, + `Incorrect item processed!` + ) + processor.throwOnceHandlingNextItem.should.equal( + false, + 'Throw once config should be reset!' + ) + }) + + it('replays item if setNextToProcess fails', async () => { + const first = 'Number 0!' + const second = 'Number 1!' + await processor.add(0, first) + await processor.add(1, second) + await sleep(10) + processor.handledQueue.length.should.equal( + 1, + `Incorrect number processed!` + ) + processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) + + processor.throwOnceOnSettingNextToProcess = true + await processor.markProcessed(0) + await sleep(10) + + processor.handledQueue.length.should.equal( + 2, + `There should be 2 items processed until item 2 is replayed!` + ) + + await sleep(200) + + processor.handledQueue.length.should.equal( + 3, + `Incorrect number processed!` + ) + processor.handledQueue[1].should.equal( + second, + `Incorrect item processed!` + ) + processor.handledQueue[2].should.equal( + second, + `Incorrect item re-processed!` + ) + processor.throwOnceOnSettingNextToProcess.should.equal( + false, + 'Throw once config should be reset!' + ) + }) }) describe('Start with existing state', () => { diff --git a/packages/rollup-core/src/app/data/data-service.ts b/packages/rollup-core/src/app/data/data-service.ts index 7ed7929ec85..0d02374f5ca 100644 --- a/packages/rollup-core/src/app/data/data-service.ts +++ b/packages/rollup-core/src/app/data/data-service.ts @@ -22,6 +22,7 @@ import { VerificationStatus, StateCommitmentBatchSubmission, BatchSubmission, + L1BlockPersistenceInfo, } from '../../types' import { getL1BlockInsertValue, @@ -44,6 +45,46 @@ export class DefaultDataService implements DataService { // TODO: All inserts below assume data is trusted and not malicious -- there is no SQL Injection protection. // If this is not a safe assumption, we have the bigger problem of not being able to trust our block data. + /** + * @inheritDoc + */ + public async getL1BlockPersistenceInfo( + blockNumber: number + ): Promise { + const res = await this.rdb.select( + `SELECT MAX(b.block_number) as block_number, MAX(tx.id) as l1_tx, MAX(rtx.id) as rollup_tx, MAX(rsb.id) as rollup_state_root_batch + FROM l1_block b + LEFT OUTER JOIN l1_tx tx ON tx.block_number = b.block_number + LEFT OUTER JOIN l1_rollup_tx rtx ON rtx.l1_tx_hash = tx.tx_hash + LEFT OUTER JOIN l1_rollup_state_root_batch rsb ON rsb.l1_tx_hash = tx.tx_hash + WHERE b.block_number = ${blockNumber} + GROUP BY b.block_number, tx.tx_hash, rtx.id, rsb.id + LIMIT 1` + ) + + const toReturn = { + blockPersisted: false, + txPersisted: false, + rollupTxsPersisted: false, + rollupStateRootsPersisted: false, + } + + if (!res || !res.length) { + return toReturn + } + + toReturn.blockPersisted = + res['block_number'] !== null && res['block_number'] !== undefined + toReturn.txPersisted = res['l1_tx'] !== null && res['l1_tx'] !== undefined + toReturn.rollupTxsPersisted = + res['rollup_tx'] !== null && res['rollup_tx'] !== undefined + toReturn.rollupStateRootsPersisted = + res['rollup_state_root_batch'] !== null && + res['rollup_state_root_batch'] !== undefined + + return toReturn + } + /** * @inheritDoc */ @@ -53,10 +94,8 @@ export class DefaultDataService implements DataService { txContext?: any ): Promise { return this.rdb.execute( - `${l1BlockInsertStatement} VALUES (${getL1BlockInsertValue( - block, - processed - )})`, + `${l1BlockInsertStatement} + VALUES (${getL1BlockInsertValue(block, processed)})`, txContext ) } @@ -75,7 +114,8 @@ export class DefaultDataService implements DataService { (tx, index) => `(${getL1TransactionInsertValue(tx, index)})` ) return this.rdb.execute( - `${l1TxInsertStatement} VALUES ${values.join(',')}`, + `${l1TxInsertStatement} + VALUES ${values.join(',')}`, txContext ) } @@ -321,9 +361,9 @@ export class DefaultDataService implements DataService { */ public async insertL2TransactionOutput(tx: TransactionOutput): Promise { return this.rdb.execute( - `${l2TransactionOutputInsertStatement} VALUES (${getL2TransactionOutputInsertValue( - tx - )})` + `${l2TransactionOutputInsertStatement} + VALUES (${getL2TransactionOutputInsertValue(tx)}) + ON CONFLICT (tx_hash) DO NOTHING` // makes it so if we're inserting data that already exists, it doesn't fail. If tx_hash is not unique, we have bigger problems =| ) } diff --git a/packages/rollup-core/src/app/data/producers/l1-chain-data-persister.ts b/packages/rollup-core/src/app/data/producers/l1-chain-data-persister.ts index 88db9421e8b..475d374e84f 100644 --- a/packages/rollup-core/src/app/data/producers/l1-chain-data-persister.ts +++ b/packages/rollup-core/src/app/data/producers/l1-chain-data-persister.ts @@ -6,7 +6,11 @@ import { Block, Provider, TransactionResponse } from 'ethers/providers' import { Log } from 'ethers/providers/abstract-provider' /* Internal Imports */ -import { L1DataService, LogHandlerContext } from '../../../types' +import { + L1BlockPersistenceInfo, + L1DataService, + LogHandlerContext, +} from '../../../types' import { ChainDataProcessor } from './chain-data-processor' const log: Logger = getLogger('l1-chain-data-persister') @@ -82,8 +86,27 @@ export class L1ChainDataPersister extends ChainDataProcessor { let relevantLogs: Log[] let txs: TransactionResponse[] - + let blockPersistenceInfo: L1BlockPersistenceInfo try { + blockPersistenceInfo = await this.l1DataService.getL1BlockPersistenceInfo( + block.number + ) + log.debug( + `Got block persistence info for block number ${ + block.number + }: ${JSON.stringify(blockPersistenceInfo)}.` + ) + + if ( + blockPersistenceInfo.rollupTxsPersisted || + blockPersistenceInfo.rollupStateRootsPersisted + ) { + log.info( + `block already had txs or state roots persisted. Marking processing as complete.` + ) + return this.markProcessed(index) + } + const logs: Log[] = await this.getLogsForBlock(block.hash) log.debug( @@ -107,9 +130,10 @@ export class L1ChainDataPersister extends ChainDataProcessor { log.debug( `No relevant logs found in block ${block.number}. Storing block and moving on.` ) - await this.l1DataService.insertL1Block(block, true) - await this.markProcessed(index) - return + if (!blockPersistenceInfo.blockPersisted) { + await this.l1DataService.insertL1Block(block, true) + } + return this.markProcessed(index) } log.debug( @@ -132,7 +156,9 @@ export class L1ChainDataPersister extends ChainDataProcessor { log.debug( `Inserting block ${block.number} and ${txs.length} transactions.` ) - await this.l1DataService.insertL1BlockAndTransactions(block, txs, false) + if (!blockPersistenceInfo.blockPersisted) { + await this.l1DataService.insertL1BlockAndTransactions(block, txs, false) + } log.debug( `Looping through ${relevantLogs.length} logs from block ${block.number} to insert rollup transactions & state roots` diff --git a/packages/rollup-core/src/types/data/l1-data-service.ts b/packages/rollup-core/src/types/data/l1-data-service.ts index eb14c093be8..720ad03ad53 100644 --- a/packages/rollup-core/src/types/data/l1-data-service.ts +++ b/packages/rollup-core/src/types/data/l1-data-service.ts @@ -2,10 +2,25 @@ import { Block, TransactionResponse } from 'ethers/providers' /* Internal Imports */ -import { GethSubmission, RollupTransaction } from '../types' +import { + GethSubmission, + L1BlockPersistenceInfo, + RollupTransaction, +} from '../types' import { GethSubmissionRecord } from './types' export interface L1DataService { + /** + * Gets Information regarding whether or not the block (and its related data) associated with + * the provided L1 block number is present in the DB. + * + * @param blockNumber The block number in question. + * @returns The L1BlockPersistenceInfo object containing booleans indicating what has been persisted. + */ + getL1BlockPersistenceInfo( + blockNumber: number + ): Promise + /** * Inserts the provided block into the associated RDB. * diff --git a/packages/rollup-core/src/types/types.ts b/packages/rollup-core/src/types/types.ts index dd4cb1b8ddb..3087dd42fdc 100644 --- a/packages/rollup-core/src/types/types.ts +++ b/packages/rollup-core/src/types/types.ts @@ -138,3 +138,10 @@ export interface TransactionResult { updatedStorage: StorageElement[] updatedContracts: ContractStorage[] } + +export interface L1BlockPersistenceInfo { + blockPersisted: boolean + txPersisted: boolean + rollupTxsPersisted: boolean + rollupStateRootsPersisted: boolean +} diff --git a/packages/rollup-core/test/app/l1-chain-data-persister.spec.ts b/packages/rollup-core/test/app/l1-chain-data-persister.spec.ts index cd9f907ae86..2669f8600df 100644 --- a/packages/rollup-core/test/app/l1-chain-data-persister.spec.ts +++ b/packages/rollup-core/test/app/l1-chain-data-persister.spec.ts @@ -25,9 +25,11 @@ import { RollupTransaction, L1DataService, GethSubmissionRecord, + L1BlockPersistenceInfo, } from '../../src/types' class MockDataService extends DefaultDataService { + public l1BlockPersistenceInfo: L1BlockPersistenceInfo public readonly blocks: Block[] = [] public readonly processedBlocks: Set = new Set() public readonly blockTransactions: Map @@ -41,6 +43,18 @@ class MockDataService extends DefaultDataService { this.blockTransactions = new Map() this.stateRoots = new Map() this.rollupTransactions = new Map() + this.l1BlockPersistenceInfo = { + blockPersisted: false, + txPersisted: false, + rollupTxsPersisted: false, + rollupStateRootsPersisted: false, + } + } + + public async getL1BlockPersistenceInfo( + blockNumber: number + ): Promise { + return this.l1BlockPersistenceInfo } public async insertL1Block(block: Block, processed: boolean): Promise { @@ -226,7 +240,7 @@ describe('L1 Chain Data Persister', () => { }) describe('Irrelevant logs', () => { - it('should not persist block without log handler', async () => { + it('should persist block but no txs without log handler', async () => { chainDataPersister = await L1ChainDataPersister.create( db, dataService, @@ -252,7 +266,7 @@ describe('L1 Chain Data Persister', () => { ) }) - it('should not persist block without logs relevant to log handler topic', async () => { + it('should persist block but no txs without logs relevant to log handler topic', async () => { const logHandlerContext: LogHandlerContext = { topic: 'not your topic', contractAddress: ZERO_ADDRESS, @@ -285,7 +299,7 @@ describe('L1 Chain Data Persister', () => { ) }) - it('should not persist block without logs relevant to log handler address', async () => { + it('should persist block but no txs without logs relevant to log handler address', async () => { chainDataPersister = await L1ChainDataPersister.create( db, dataService, @@ -692,4 +706,239 @@ describe('L1 Chain Data Persister', () => { }) }) }) + + describe('Partial state persisted', () => { + it('should not persist block if already persisted', async () => { + chainDataPersister = await L1ChainDataPersister.create( + db, + dataService, + provider, + [] + ) + + provider.topicToLogsToReturn.set('derp', [getLog(['derp'], ZERO_ADDRESS)]) + + dataService.l1BlockPersistenceInfo.blockPersisted = true + const block = getBlock(keccak256FromUtf8('derp')) + await chainDataPersister.handle(block) + + await sleep(1_000) + + dataService.blocks.length.should.equal(0, `Should not re-insert block!`) + dataService.blockTransactions.size.should.equal( + 0, + `Inserted transactions when shouldn't have!` + ) + dataService.stateRoots.size.should.equal( + 0, + `Inserted roots when shouldn't have!` + ) + }) + + describe('with logs', () => { + const configuredHandlerContext: LogHandlerContext = { + ...errorLogHandlerContext, + } + beforeEach(async () => { + chainDataPersister = await L1ChainDataPersister.create( + db, + dataService, + provider, + [configuredHandlerContext] + ) + }) + + it('should not persist block or l1 transaction if already persisted but should persist logs', async () => { + const rollupTxs = [getRollupTransaction()] + configuredHandlerContext.handleLog = async (ds, l, t) => { + await ds.insertL1RollupTransactions(t.hash, rollupTxs) + } + + const tx: TransactionResponse = getTransactionResponse() + provider.txsToReturn.set(tx.hash, tx) + provider.topicToLogsToReturn.set(topic, [ + getLog([topic], contractAddress, tx.hash), + ]) + + dataService.l1BlockPersistenceInfo.blockPersisted = true + + await chainDataPersister.handle(defaultBlock) + + await sleep(1_000) + + dataService.blocks.length.should.equal( + 0, + `Should not have inserted block because it already exists!` + ) + dataService.blockTransactions.size.should.equal( + 0, + `Should not have inserted transaction because it already exists!` + ) + + const rollupTxsExist: boolean = !!dataService.rollupTransactions.get( + tx.hash + ) + rollupTxsExist.should.equal( + true, + `Should have inserted rollup txs for the tx!` + ) + dataService.rollupTransactions + .get(tx.hash) + .length.should.equal(1, `Should have inserted 1 rollup tx!`) + dataService.rollupTransactions + .get(tx.hash)[0] + .should.deep.equal(rollupTxs[0], `Inserted rollup tx mismatch!`) + + dataService.processedBlocks.size.should.equal(1, `block not processed!`) + dataService.processedBlocks + .has(defaultBlock.hash) + .should.equal(true, `correct block not processed!`) + }) + + it('should not persist block, transaction or rollup transaction it is all already stored', async () => { + const rollupTxs = [getRollupTransaction()] + configuredHandlerContext.handleLog = async (ds, l, t) => { + await ds.insertL1RollupTransactions(t.hash, rollupTxs) + } + + const tx: TransactionResponse = getTransactionResponse() + provider.txsToReturn.set(tx.hash, tx) + provider.topicToLogsToReturn.set(topic, [ + getLog([topic], contractAddress, tx.hash), + ]) + + dataService.l1BlockPersistenceInfo.blockPersisted = true + dataService.l1BlockPersistenceInfo.rollupTxsPersisted = true + + await chainDataPersister.handle(defaultBlock) + + await sleep(1_000) + + dataService.blocks.length.should.equal( + 0, + `Should not have inserted block because it already exists!` + ) + dataService.blockTransactions.size.should.equal( + 0, + `Should not have inserted transaction because it already exists!` + ) + + const rollupTxsExist: boolean = !!dataService.rollupTransactions.get( + tx.hash + ) + rollupTxsExist.should.equal( + false, + `Should not have inserted rollup txs for the tx because they already exist!` + ) + dataService.processedBlocks.size.should.equal( + 0, + `block should not be marked processed because it already is!` + ) + }) + + it('should not persist block or transaction but should persist state roots', async () => { + const stateRoots = [keccak256FromUtf8('root')] + configuredHandlerContext.handleLog = async (ds, l, t) => { + await ds.insertL1RollupStateRoots(tx.hash, stateRoots) + } + chainDataPersister = await L1ChainDataPersister.create( + db, + dataService, + provider, + [configuredHandlerContext] + ) + + const tx: TransactionResponse = getTransactionResponse() + const tx2: TransactionResponse = getTransactionResponse( + keccak256FromUtf8('tx2') + ) + provider.txsToReturn.set(tx.hash, tx) + provider.txsToReturn.set(tx2.hash, tx2) + provider.topicToLogsToReturn.set(topic, [ + getLog([topic], contractAddress, tx.hash), + ]) + + dataService.l1BlockPersistenceInfo.blockPersisted = true + + await chainDataPersister.handle(defaultBlock) + + await sleep(1_000) + + dataService.blocks.length.should.equal( + 0, + `Should not have inserted block because it already exists!` + ) + dataService.blockTransactions.size.should.equal( + 0, + `Should not have inserted transactions for 1 block because it already exists!` + ) + + const stateRootsExist: boolean = !!dataService.stateRoots.get(tx.hash) + stateRootsExist.should.equal( + true, + `Should have inserted state roots for the tx!` + ) + dataService.stateRoots + .get(tx.hash) + .length.should.equal(1, `Should have inserted 1 state root!`) + dataService.stateRoots + .get(tx.hash)[0] + .should.deep.equal(stateRoots[0], `Inserted state Root mismatch!`) + + dataService.processedBlocks.size.should.equal(1, `block not processed!`) + dataService.processedBlocks + .has(defaultBlock.hash) + .should.equal(true, `correct block not processed!`) + }) + + it('should not persist block, transaction, or state roots if they are already persisted', async () => { + const stateRoots = [keccak256FromUtf8('root')] + configuredHandlerContext.handleLog = async (ds, l, t) => { + await ds.insertL1RollupStateRoots(tx.hash, stateRoots) + } + chainDataPersister = await L1ChainDataPersister.create( + db, + dataService, + provider, + [configuredHandlerContext] + ) + + const tx: TransactionResponse = getTransactionResponse() + const tx2: TransactionResponse = getTransactionResponse( + keccak256FromUtf8('tx2') + ) + provider.txsToReturn.set(tx.hash, tx) + provider.txsToReturn.set(tx2.hash, tx2) + provider.topicToLogsToReturn.set(topic, [ + getLog([topic], contractAddress, tx.hash), + ]) + + dataService.l1BlockPersistenceInfo.blockPersisted = true + dataService.l1BlockPersistenceInfo.rollupStateRootsPersisted = true + + await chainDataPersister.handle(defaultBlock) + + await sleep(1_000) + + dataService.blocks.length.should.equal( + 0, + `Should not have inserted block because it already exists!` + ) + dataService.blockTransactions.size.should.equal( + 0, + `Should not have inserted transactions for 1 block because it already exists!` + ) + + const stateRootsExist: boolean = !!dataService.stateRoots.get(tx.hash) + stateRootsExist.should.equal( + false, + `Should not have inserted state roots for the tx because they already exist!` + ) + dataService.processedBlocks.size.should.equal( + 0, + `block should not be marked processed because it already is!` + ) + }) + }) + }) }) diff --git a/packages/rollup-core/test/db/l2-data-service.dbspec.ts b/packages/rollup-core/test/db/l2-data-service.dbspec.ts index e428ea0fa29..3e7cc6be19e 100644 --- a/packages/rollup-core/test/db/l2-data-service.dbspec.ts +++ b/packages/rollup-core/test/db/l2-data-service.dbspec.ts @@ -48,6 +48,21 @@ describe('L2 Data Service (will fail if postgres is not running with expected sc res.length.should.equal(1, `No L2 Tx rows!`) verifyL2TxOutput(res[0], tx) }) + + it('Should not fail on duplicate insert into L2 Tx Output', async () => { + const tx = createTxOutput(keccak256FromUtf8('tx')) + await dataService.insertL2TransactionOutput(tx) + + let res = await postgres.select(`SELECT * FROM l2_tx_output`) + res.length.should.equal(1, `No L2 Tx rows!`) + verifyL2TxOutput(res[0], tx) + + await dataService.insertL2TransactionOutput(tx) + + res = await postgres.select(`SELECT * FROM l2_tx_output`) + res.length.should.equal(1, `No L2 Tx rows!`) + verifyL2TxOutput(res[0], tx) + }) }) describe('tryBuildCanonicalChainBatchNotPresentOnL1', () => { diff --git a/packages/rollup-services/src/exec/services.ts b/packages/rollup-services/src/exec/services.ts index edf9328313c..29ae40ea5f5 100644 --- a/packages/rollup-services/src/exec/services.ts +++ b/packages/rollup-services/src/exec/services.ts @@ -106,7 +106,10 @@ export const runServices = async (): Promise => { const subscriptions: Array> = [] if (!!l1ChainDataPersister) { services.push(l1ChainDataPersister) - const l1Processor: EthereumBlockProcessor = createL1BlockSubscriber() + const lastProcessedBlock = await l1ChainDataPersister.getLastIndexProcessed() + const l1Processor: EthereumBlockProcessor = createL1BlockSubscriber( + lastProcessedBlock + ) log.info(`Starting to sync L1 chain`) subscriptions.push( l1Processor.subscribe(getL1Provider(), l1ChainDataPersister) @@ -114,7 +117,10 @@ export const runServices = async (): Promise => { } if (!!l2ChainDataPersister) { services.push(l2ChainDataPersister) - const l2Processor: EthereumBlockProcessor = createL2BlockSubscriber() + const lastProcessedBlock = await l2ChainDataPersister.getLastIndexProcessed() + const l2Processor: EthereumBlockProcessor = createL2BlockSubscriber( + lastProcessedBlock + ) log.info(`Starting to sync L2 chain`) subscriptions.push( l2Processor.subscribe(getL2Provider(), l2ChainDataPersister) @@ -425,16 +431,28 @@ const createFraudDetector = (): FraudDetector => { ) } -const createL1BlockSubscriber = (): EthereumBlockProcessor => { +const createL1BlockSubscriber = ( + lastBlockProcessed: number = 0 +): EthereumBlockProcessor => { + const startBlock = Math.max( + lastBlockProcessed, + Environment.getOrThrow(Environment.l1EarliestBlock) + ) + log.info(`Starting subscription to L1 chain starting at block ${startBlock}`) return new EthereumBlockProcessor( getL1BlockProcessorDB(), - Environment.getOrThrow(Environment.l1EarliestBlock), + startBlock, Environment.getOrThrow(Environment.finalityDelayInBlocks) ) } -const createL2BlockSubscriber = (): EthereumBlockProcessor => { - return new EthereumBlockProcessor(getL2Db(), 0, 1) +const createL2BlockSubscriber = ( + lastBlockProcessed: number = 0 +): EthereumBlockProcessor => { + log.info( + `Starting subscription to L2 node starting at block ${lastBlockProcessed}` + ) + return new EthereumBlockProcessor(getL2Db(), lastBlockProcessed, 1) } /********************* diff --git a/packages/test-rollup-workflow/test/test-submit-to-l2.spec.ts b/packages/test-rollup-workflow/test/test-submit-to-l2.spec.ts index 51bf358d71f..65f63f99437 100644 --- a/packages/test-rollup-workflow/test/test-submit-to-l2.spec.ts +++ b/packages/test-rollup-workflow/test/test-submit-to-l2.spec.ts @@ -1,7 +1,7 @@ import './setup' /* External Imports */ -import { getLogger, keccak256FromUtf8 } from '@eth-optimism/core-utils' +import { getLogger, keccak256FromUtf8, sleep } from '@eth-optimism/core-utils' import { CHAIN_ID, GAS_LIMIT, @@ -62,19 +62,10 @@ describe('Test Sending Transactions Directly To L2', () => { it('Sets storage N times', async () => { const key: string = 'test' - for (let i = 0; i < 5; i++) { + for (let i = 0; i < 20; i++) { log.debug(`Sending tx to set storage key ${key}`) const res = await simpleStorage.setStorage(key, `${key}${i}`) - const receipt: TransactionReceipt = await provider.waitForTransaction( - res.hash - ) - receipt.status.should.equal( - 1, - `Transaction ${i} failed! ${JSON.stringify(receipt)}` - ) - - const setStorage = await simpleStorage.getStorage(key) - setStorage.should.equal(`${key}${i}`, `Storage not set to ${key}${i}`) + await sleep(1) } }).timeout(100_000) }) From dd56c60658622c51ed6fe3e12812781452ea0145 Mon Sep 17 00:00:00 2001 From: Will Meister Date: Mon, 24 Aug 2020 16:09:00 -0500 Subject: [PATCH 2/4] Upping timeout of test that is timing out consistently --- packages/solc-transpiler/test/libraries.spec.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/solc-transpiler/test/libraries.spec.ts b/packages/solc-transpiler/test/libraries.spec.ts index 477d6f8d5fa..4fac761e7df 100644 --- a/packages/solc-transpiler/test/libraries.spec.ts +++ b/packages/solc-transpiler/test/libraries.spec.ts @@ -80,5 +80,5 @@ describe('Library usage tests', () => { wrappedSolcJson.contracts['SafeMathUser.sol'][ 'SafeMathUser' ].should.not.equal(undefined, 'SafeMathUser contract not found!') - }).timeout(10_000) + }).timeout(20_000) }) From 572eeb8fbe37c6ac2fd46260f9e99c6ef8138513 Mon Sep 17 00:00:00 2001 From: Will Meister Date: Mon, 24 Aug 2020 16:33:27 -0500 Subject: [PATCH 3/4] fixing broken test --- .../app/queued-persisted-processor.spec.ts | 45 +++++++------------ 1 file changed, 16 insertions(+), 29 deletions(-) diff --git a/packages/core-db/test/app/queued-persisted-processor.spec.ts b/packages/core-db/test/app/queued-persisted-processor.spec.ts index 2d448a9587e..3d65e150781 100644 --- a/packages/core-db/test/app/queued-persisted-processor.spec.ts +++ b/packages/core-db/test/app/queued-persisted-processor.spec.ts @@ -10,6 +10,7 @@ import { DB } from '../../src/types/db' class DummyQueuedPersistedProcessor extends BaseQueuedPersistedProcessor< string > { + public callMarkProcessed: boolean = true public throwOnceHandlingNextItem: boolean = false public throwOnceOnSettingNextToProcess: boolean = false public handledQueue: string[] @@ -45,6 +46,9 @@ class DummyQueuedPersistedProcessor extends BaseQueuedPersistedProcessor< throw Error('you told me to throw in handleNextItem.') } this.handledQueue.push(item) + if (this.callMarkProcessed) { + return this.markProcessed(index) + } } protected async serializeItem(item: string): Promise { @@ -96,6 +100,8 @@ describe('Queued Persisted Processor', () => { }) it('does not handle next item if previous is not acknowledged', async () => { + processor.callMarkProcessed = false + const first = 'Number 0!' await processor.add(0, first) await processor.add(1, 'Number 1!') @@ -112,20 +118,12 @@ describe('Queued Persisted Processor', () => { const second = 'Number 1!' await processor.add(0, first) await processor.add(1, second) - await sleep(10) - processor.handledQueue.length.should.equal( - 1, - `Incorrect number processed!` - ) - processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) - - await processor.markProcessed(0) - await sleep(10) - + await sleep(20) processor.handledQueue.length.should.equal( 2, `Incorrect number processed!` ) + processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) processor.handledQueue[1].should.equal( second, `Incorrect item processed!` @@ -136,15 +134,7 @@ describe('Queued Persisted Processor', () => { const first = 'Number 0!' const second = 'Number 1!' await processor.add(0, first) - await sleep(10) - processor.handledQueue.length.should.equal( - 1, - `Incorrect number processed!` - ) - processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) - - await processor.markProcessed(0) - await sleep(10) + await sleep(20) processor.handledQueue.length.should.equal( 1, @@ -152,7 +142,7 @@ describe('Queued Persisted Processor', () => { ) await processor.add(1, second) - await sleep(10) + await sleep(20) processor.handledQueue.length.should.equal( 2, `Incorrect number processed!` @@ -167,7 +157,6 @@ describe('Queued Persisted Processor', () => { const first = 'Number 0!' const second = 'Number 1!' await processor.add(0, first) - await processor.add(1, second) await sleep(10) processor.handledQueue.length.should.equal( 1, @@ -176,8 +165,9 @@ describe('Queued Persisted Processor', () => { processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) processor.throwOnceHandlingNextItem = true - await processor.markProcessed(0) - await sleep(10) + await processor.add(1, second) + + await sleep(20) processor.handledQueue.length.should.equal( 1, @@ -204,7 +194,6 @@ describe('Queued Persisted Processor', () => { const first = 'Number 0!' const second = 'Number 1!' await processor.add(0, first) - await processor.add(1, second) await sleep(10) processor.handledQueue.length.should.equal( 1, @@ -213,7 +202,7 @@ describe('Queued Persisted Processor', () => { processor.handledQueue[0].should.equal(first, `Incorrect item processed!`) processor.throwOnceOnSettingNextToProcess = true - await processor.markProcessed(0) + await processor.add(1, second) await sleep(10) processor.handledQueue.length.should.equal( @@ -265,6 +254,8 @@ describe('Queued Persisted Processor', () => { }) it('restarts with existing state (1 added but not acknowledged)', async () => { + processor.callMarkProcessed = false + const item: string = 'Number 0!' await processor.add(0, item) await sleep(10) @@ -296,8 +287,6 @@ describe('Queued Persisted Processor', () => { ) processor.handledQueue[0].should.equal(item, `Incorrect item processed`) - await processor.markProcessed(0) - const secondProc = await DummyQueuedPersistedProcessor.create( db, persistenceKey @@ -319,8 +308,6 @@ describe('Queued Persisted Processor', () => { ) processor.handledQueue[0].should.equal(item, `Incorrect item processed`) - await processor.markProcessed(0) - const secondProc = await DummyQueuedPersistedProcessor.create( db, persistenceKey From 425b6bd19e9c1009f8a7ad40a9fa327f540d4ef6 Mon Sep 17 00:00:00 2001 From: Will Meister Date: Mon, 24 Aug 2020 16:39:08 -0500 Subject: [PATCH 4/4] adding configurable retry delay in queued-persisted-processor.spec.ts --- .../core-db/test/app/queued-persisted-processor.spec.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/packages/core-db/test/app/queued-persisted-processor.spec.ts b/packages/core-db/test/app/queued-persisted-processor.spec.ts index 3d65e150781..e211a1b617e 100644 --- a/packages/core-db/test/app/queued-persisted-processor.spec.ts +++ b/packages/core-db/test/app/queued-persisted-processor.spec.ts @@ -72,6 +72,7 @@ describe('Queued Persisted Processor', () => { let db: DB let processor: DummyQueuedPersistedProcessor const persistenceKey: string = 'derp' + const retrySleepDelayMillis: number = 100 beforeEach(async () => { db = newInMemoryDB() @@ -79,7 +80,7 @@ describe('Queued Persisted Processor', () => { db, persistenceKey, 0, - 100 + retrySleepDelayMillis ) }) @@ -171,10 +172,10 @@ describe('Queued Persisted Processor', () => { processor.handledQueue.length.should.equal( 1, - `There should still only be one item processed! Should fail and retry after 100 millis` + `There should still only be one item processed! Should fail and retry after ${retrySleepDelayMillis} millis` ) - await sleep(200) + await sleep(retrySleepDelayMillis * 2) processor.handledQueue.length.should.equal( 2, @@ -210,7 +211,7 @@ describe('Queued Persisted Processor', () => { `There should be 2 items processed until item 2 is replayed!` ) - await sleep(200) + await sleep(retrySleepDelayMillis * 2) processor.handledQueue.length.should.equal( 3,