Buffered insertions #61
Conversation
# Conflicts: # src/clickhouse/handleSinkRequest.ts # src/schemas.ts
# Conflicts: # bun.lockb # src/clickhouse/handleSinkRequest.ts # src/fetch/POST.ts
There was a problem hiding this comment.
Overall looks good, but your Promise handling looks a bit wonky
You are trying to recreate your Promise based Queue batching
You should simply just bring back in PQueue for that section alone
Have a look at:
https://github.com/sindresorhus/p-queue#advanced-example
await queue.onEmpty()
// or
await queue.onIdle();To know when all the promises have been resolved to continue to the next step
src/clickhouse/handleSinkRequest.ts
Outdated
| const queue = new PQueue({ concurrency: config.queueConcurrency }); | ||
|
|
||
| let nextUpdateTime: number = 0; | ||
| let promise: Promise<unknown> = Promise.resolve(); |
There was a problem hiding this comment.
Should be replaced with PQueue
src/clickhouse/handleSinkRequest.ts
Outdated
| prometheus.queue_size.set(queue.size); | ||
| if (queue.size > config.queueLimit) await setTimeout(1000); | ||
| if (new Date().getTime() > nextUpdateTime) { | ||
| await promise; |
src/clickhouse/handleSinkRequest.ts
Outdated
| blocks: [], | ||
| }; | ||
|
|
||
| promise = new Promise<void>(async (resolve) => { |
src/clickhouse/handleSinkRequest.ts
Outdated
| } | ||
| } | ||
|
|
||
| resolve(); |
|
Edited the workflow with The timeout promise keeps track of the next insertion time. It has to be in the pqueue to block the process until it is the correct insertion time later on. The insertion promise is the call to ClickHouse. It will delay everything if it is slow. Here is a summary of the process:
|
# Conflicts: # bun.lockb
|
👍 update looks good |
Adds in-memory batching to the sink.
The pqueue has been removed in favor of a single promise that inserts data while more data is stored locally asynchronously.
The behavior can be configured in
.envor with the CLI usingMAX_BUFFER_SIZEandINSERTION_DELAY.