Reshard logic #82
Reshard logic #82
Conversation
src/helpers/dbHelper.ts
Outdated
| @@ -39,7 +39,7 @@ const options = { | |||
| logger.info(`PG_USER is ${EnvLoader.getPropertyOrFail('PG_USER')}`) | |||
| const pg: pgPromise.IMain<{}, IClient> = pgPromise(options) | |||
| export const pgPool = pg( | |||
| `postgres://${EnvLoader.getPropertyOrFail('PG_USER')}:${EnvLoader.getPropertyOrFail('PG_PASS')}@${EnvLoader.getPropertyOrFail('PG_HOST')}:${EnvLoader.getPropertyOrDefault('PG_PORT', '5432')}/${EnvLoader.getPropertyOrFail('DB_NAME')}` | |||
| `postgres://${EnvLoader.getPropertyOrFail('PG_USER')}:${EnvLoader.getPropertyOrFail('PG_PASS')}@${EnvLoader.getPropertyOrFail('PG_HOST')}:${EnvLoader.getPropertyOrDefault('PG_PORT', '10007')}/${EnvLoader.getPropertyOrFail('DB_NAME')}` | |||
There was a problem hiding this comment.
Could we use the same 5432 as a default port?
src/services/block/block.ts
Outdated
| @@ -41,4 +49,366 @@ export class Block { | |||
| Block.log.debug('getBulkBlocksByHash() result: %s', statusArray) | |||
| return statusArray | |||
| } | |||
|
|
|||
| static async checkIfTableViewExists(viewName: string): Promise<boolean> { | |||
There was a problem hiding this comment.
Could you remove this method? I cannot find any real usages.
src/rpc/push_getBlockHashFromView.ts
Outdated
|
|
||
| public async execute(params: [string, number, number, number]) { | ||
| try { | ||
| console.log('Executing push_getBlockHashFromView', { params }) |
There was a problem hiding this comment.
Could we replace console log with this.log.debug everywhere
src/services/block/block.ts
Outdated
| shardIds | ||
| ) | ||
|
|
||
| console.log({ |
There was a problem hiding this comment.
Could we replace console log with this.log.debug everywhere
src/services/block/blockPolling.ts
Outdated
| } | ||
| } | ||
|
|
||
| this.log.info('Node shard map constructed successfully', { |
There was a problem hiding this comment.
This won't log anything. Please find all places with parameterized logging and so smth like
log.info('test %s', param);
valid params are %s, %o, %d
src/services/block/blockPolling.ts
Outdated
| for (const [shardId, clients] of this.nodeMap.entries()) { | ||
| this.log.info('Initiating pooling for shard:', { shardId }) | ||
| // Update the currently syncing shard and last synced page number | ||
| clients.map(async (client) => { |
There was a problem hiding this comment.
!!! This code looks fully async.
It might lead to the fact that
await this.validateAndStoreBlockHashes - won't see your db changes, such as syncing shard = shardId and pageNumber = 1 ???????
src/services/block/blockPolling.ts
Outdated
| const blockProcessPromises = fullBlocks.map((block) => { | ||
| const blockBytes = BitUtil.base16ToBytes(block) | ||
| const parsedBlock = BlockUtil.parseBlock(blockBytes) | ||
| return Container.get(StorageNode).handleBlock(parsedBlock, blockBytes) |
There was a problem hiding this comment.
Could you inject the field into BlockPolling like this
@Inject((type) => StorageNode)
private storageNode: StorageNode
or
@Injject() without params; whatever works
src/services/block/blockPolling.ts
Outdated
| } | ||
| } | ||
|
|
||
| public async validateAndStoreBlockHashesOld( |
There was a problem hiding this comment.
Could you remove this, unless the method is being used somewhere
src/services/block/blockPolling.ts
Outdated
| client.getViewName() | ||
| ) | ||
|
|
||
| const blockProcessPromises = fullBlocks.map((block) => { |
There was a problem hiding this comment.
I think you need a
try {
do block processing
} catch (e) {
log.error ...
}
I'd expect that some block processing might throw exceptions and this should not stop syncing the client itself.
| * 3. if the blockhashes are not present in the db, query the full block from the respective node | ||
| * 4. parse the block and store it in the db | ||
| */ | ||
| public async validateAndStoreBlockHashes( |
There was a problem hiding this comment.
!!! I'd expect to have some parallelization in one of two key points
Option1:
every shardId is processing in parallel (so 32 logical workloads max)
in this case we will call
let promises = [];
promises.push(validateAndStoreBlockHashes(1,...32))
Promise.allSettled(promises)
Option2:
every "for (const client of clients) {" body at line 246 gets executed in parallel
Let me know what you think.
No description provided.