chore: use threaded worker util from toolkit for msg parsing #85
Conversation
|
Vercel deployment URL: https://signer-metrics-vuiwsbd41-hirosystems.vercel.app 🚀 |
| @@ -491,20 +490,18 @@ export class PgStore extends BasePgStore { | |||
| } | |||
|
|
|||
| async getSignerDataForBlock({ sql, blockId }: { sql: PgSqlClient; blockId: BlockIdParam }) { | |||
| let blockFilter: Fragment; | |||
There was a problem hiding this comment.
Work around for a type error due to the api-toolkit using an older version of the postgres library.
TODO: update postgres library in toolkit repo
Codecov ReportAttention: Patch coverage is
📢 Thoughts on this report? Let us know! |
| const parsed = await this.threadedParser.parseStackerDbChunk(msg); | ||
| await this.handleStackerDbMsg(messageId, parseInt(timestamp), parsed); | ||
| const parsed = await this.threadedParser.exec({ kind: 'chunk', msg }); | ||
| const result = parsed.result as ParsedStackerDbChunk[]; |
There was a problem hiding this comment.
In this case yes, and tests would catch any accidental breakages. However, it could be made more type safe if the WorkerThreadManager implementation could understand conditional return types when the type for this .exec(...) function are inferred. That's a bit tricky though. Right now the type inference code extracts the args and response type separately, e.g.:
type WorkerPoolModuleInterface<TArgs extends unknown[], TResp> = {
workerModule: NodeJS.Module;
processTask: (...args: TArgs) => Promise<TResp> | TResp; // <-- any type interface involving both TArgs and TResp are broken here
}If the worker module function was called directly, fully type safety could be achieved with:
type TaskArg =
| { kind: 'block'; msg: CoreNodeNakamotoBlockMessage }
| { kind: 'chunk'; msg: StackerDbChunk };
type TaskResult<T extends TaskArg> = T['kind'] extends 'block'
? { kind: 'block'; result: ReturnType<typeof parseNakamotoBlockMsg> }
: { kind: 'chunk'; result: ReturnType<typeof parseStackerDbChunk> };
export function processTask<T extends TaskArg>(args: T): TaskResult<T> {
if (args.kind === 'block') {
return {
kind: 'block',
result: parseNakamotoBlockMsg(args.msg),
} as TaskResult<T>;
} else {
return {
kind: 'chunk',
result: parseStackerDbChunk(args.msg),
} as TaskResult<T>;
}
}But right now that code is broken by the WorkerPoolModuleInterface type inference. I'll make it a TODO.
| @@ -71,8 +57,9 @@ export class EventStreamHandler { | |||
| ); | |||
| } | |||
| if ('signer_signature_hash' in blockMsg) { | |||
| const parsed = await this.threadedParser.parseNakamotoBlock(nakamotoBlockMsg); | |||
| await this.handleNakamotoBlockMsg(messageId, parseInt(timestamp), parsed); | |||
| const parsed = await this.threadedParser.exec({ kind: 'block', msg: nakamotoBlockMsg }); | |||
There was a problem hiding this comment.
Note: only a single worker thread is used at a time right now, which is helpful because it prevents the main thread/loop from blocking on cpu. However, the msg parsing could be performed faster if multiple workers were used in parallel. This would require some changes in the S&P client library to allow consumers (like this project) to process msgs in batch. Right now only one at a time is supported.
Use the WorkerThread code from the toolkit library (which was originally developed in this repo then extracted out).
Note: only a single worker thread is used at a time right now, which is helpful because it prevents the main thread/loop from blocking on cpu. However, the msg parsing could be performed faster if multiple workers were used in parallel. This would require some changes in the S&P client library to allow consumers (like this project) to process msgs in batch. Right now only one at a time is supported.