import { createChild , nodeLogger } from '@neo-one/logger' ; import { BinaryReader , Block , Blockchain , DeserializeWireContext } from '@neo-one/node-core' ; import { makeErrorWithCode } from '@neo-one/utils' ; import * as fs from 'fs' ; import _ from 'lodash' ; import { Readable , Transform , Writable } from 'stream' ; import * as zlib from 'zlib' ; const logger = createChild ( nodeLogger , { service : 'chain-file' } ) ; export const InvalidBlockTransformEncodingError = makeErrorWithCode ( 'INVALID_BLOCK_TRANSFORM_ENCODING' , ( message ) => message ) ; const SIZE_OF_INT32 = 4 ; class BlockTransform extends Transform { public readonly context : DeserializeWireContext ; private mutableBuffers : Buffer [ ] ; private mutableLength : number ; public constructor ( context ) { super ( { readableObjectMode : true } ) ; this . context = context ; this . mutableBuffers = [ ] ; this . mutableLength = 0 ; } public _transform ( chunk , encoding , callback ) { if ( typeof chunk === 'string' ) { throw new InvalidBlockTransformEncodingError ( `Invalid Block Transform Chunk Type. Expected chunk type to be 'string', found: ${ typeof chunk }` ) ; } if ( encoding !== 'buffer' ) { throw new InvalidBlockTransformEncodingError ( `Invalid Block Transform Encoding. Expected: 'buffer', found: ${ encoding }` ) ; } this . mutableBuffers . push ( chunk ) ; this . mutableLength += chunk . length ; if ( this . mutableLength < 100000 ) { callback ( undefined ) ; return ; } try { const { remainingBuffer , mutableBlocks } = this . processBuffer ( new BinaryReader ( Buffer . concat ( this . mutableBuffers ) ) ) ; this . mutableBuffers = [ remainingBuffer ] ; this . mutableLength = remainingBuffer . length ; mutableBlocks . forEach ( ( block ) => this . push ( block ) ) ; callback ( undefined ) ; } catch ( error ) { callback ( error ) ; } } private processBuffer ( reader ) { const mutableBlocks = [ ] ; while ( true ) { if ( reader . remaining < SIZE_OF_INT32 ) { return { remainingBuffer : reader . remainingBuffer , mutableBlocks } ; } const length = reader . clone ( ) . readInt32LE ( ) ; if ( reader . remaining < length + SIZE_OF_INT32 ) { return { remainingBuffer : reader . remainingBuffer , mutableBlocks } ; } reader . readInt32LE ( ) ; const block = Block . deserializeWireBase ( { context : this . context , reader } ) ; mutableBlocks . push ( block ) ; } } } export interface Chain { readonly format : 'raw' | 'zip' ; readonly path : string ; } const getStream = ( chain ) : Readable => { const stream = fs . createReadStream ( chain . path ) ; if ( chain . format === 'zip' ) { return stream . pipe ( zlib . createUnzip ( ) ) ; } return stream ; } ; const getCount = async ( stream , readStart ? ) : Promise < number > => { let startBlock = readStart ? stream . read ( 4 ) : undefined ; while ( startBlock == undefined ) { await new Promise < void > ( ( resolve ) => setTimeout ( resolve , 250 ) ) ; startBlock = readStart ? stream . read ( 4 ) : Buffer . from ( [ 0 , 0 , 0 , 0 ] ) ; } const endBlock = stream . read ( 4 ) ; const startBlockIndex = startBlock . readUInt32LE ( 0 ) ; const endBlockIndex = endBlock . readUInt32LE ( 0 ) ; return readStart ? endBlockIndex - startBlockIndex : endBlockIndex ; } ; export const loadChain = async ( { blockchain , chain , readStart , } ) : Promise < void > => new Promise < void > ( ( resolve , reject ) => { logger . info ( { title : 'chain_file_load_start' , message : `attempting to sync from chain-file at: ${ chain . path }` } ) ; const stream = getStream ( chain ) ; const transform = new BlockTransform ( blockchain . deserializeWireContext ) ; const cleanup = ( ) => { stream . unpipe ( transform ) ; } ; let resolved = false ; let rejected = false ; const onError = ( error ) => { if ( ! resolved && ! rejected ) { rejected = true ; logger . info ( { title : 'chain_file_load_error' , error } ) ; cleanup ( ) ; reject ( error ) ; } } ; const onEnd = ( ) => { if ( ! resolved && ! rejected ) { resolved = true ; cleanup ( ) ; logger . info ( { title : 'chain_file_load_end' } ) ; resolve ( ) ; } } ; stream . once ( 'error' , onError ) ; transform . once ( 'finish' , onEnd ) ; transform . once ( 'end' , onEnd ) ; transform . once ( 'close' , onEnd ) ; transform . once ( 'error' , onError ) ; let pending = 0 ; let processed = 0 ; let start = Date . now ( ) ; let paused = false ; const trackIndex = blockchain . currentBlockIndex ; transform . on ( 'data' , ( block ) => { if ( block . index < trackIndex ) { return ; } pending += 1 ; blockchain . persistBlock ( { block , verify : false } ) . then ( ( ) => { pending -= 1 ; if ( pending < 500 && paused ) { paused = false ; transform . resume ( ) ; } if ( block . index === trackIndex ) { logger . info ( { title : 'chain_file_load' , message : `Loaded chain to current index in ${ Date . now ( ) - start } ms` } ) ; start = Date . now ( ) ; } else if ( block . index > trackIndex ) { processed += 1 ; if ( processed >= 100 ) { logger . info ( { title : 'chain_file_read' , message : `Processed ${ processed } blocks in ${ Date . now ( ) - start } ms (${ block . index })` } ) ; processed = 0 ; start = Date . now ( ) ; } } } ) . catch ( onError ) ; if ( pending > 1000 ) { paused = true ; transform . pause ( ) ; } } ) ; getCount ( stream , readStart ) . then ( ( count ) => { logger . info ( { title : 'chain_file_count' , count , currentIndex : blockchain . currentBlockIndex } ) ; if ( count > blockchain . currentBlockIndex ) { stream . pipe ( transform ) ; } else { resolved = true ; stream . destroy ( ) ; resolve ( ) ; } } ) . catch ( reject ) ; } ) ; const writeOut = async ( blockchain , out , height ) : Promise < void > => { let processed = 0 ; let start = Date . now ( ) ; for ( const chunk of _ . chunk ( _ . range ( 0 , height ) , 1000 ) ) { const blocks = await Promise . all ( chunk . map ( async ( index ) => blockchain . getBlock ( index ) ) ) ; const definedBlocks = blocks . filter ( ( b ) : b is Block => b !== undefined ) ; for ( const block of definedBlocks ) { const buffer = block . serializeWire ( ) ; const length = Buffer . alloc ( 4 , 0 ) ; length . writeInt32LE ( buffer . length , 0 ) ; await new Promise < void > ( ( resolve ) => out . write ( length , ( ) => resolve ( ) ) ) ; await new Promise < void > ( ( resolve ) => out . write ( buffer , ( ) => resolve ( ) ) ) ; processed += 1 ; if ( processed >= 100000 ) { logger . info ( { title : 'chain_file_write' , message : `Processed ${ processed } blocks in ${ Date . now ( ) - start } ms` } ) ; processed = 0 ; start = Date . now ( ) ; } } } out . end ( ) ; } ; export const dumpChain = async ( { blockchain , path , writeStart , } ) : Promise < void > => { const height = blockchain . currentBlockIndex ; await new Promise < void > ( ( resolve , reject ) => { const out = fs . createWriteStream ( path ) ; out . once ( 'open' , ( ) => { if ( writeStart ) { const start = Buffer . alloc ( 4 , 0 ) ; start . writeUInt32LE ( 0 , 0 ) ; out . write ( start ) ; } const count = Buffer . alloc ( 4 , 0 ) ; count . writeUInt32LE ( height , 0 ) ; out . write ( count ) ; writeOut ( blockchain , out , height ) . then ( resolve , reject ) ; } ) ; } ) ; } ;