rotor: metrics performance improvements

Author: absorbb

libs/core-functions/package.json

@@ -44,10 +44,10 @@
     "juava": "workspace:*",
     "lodash": "^4.17.21",
     "mongodb": "^6.10.0",
-    "@clickhouse/client": "^1.9.1",
+    "@clickhouse/client": "^1.10.1",
     "node-cache": "^5.1.2",
     "parse-duration": "^1.1.0",
-    "axios": "^1.7.7",
+    "axios": "1.8.2",
     "posthog-node": "^4.2.1",
     "tslib": "^2.6.3",
     "zod": "^3.23.8"

libs/core-functions/src/functions/lib/clickhouse-logger.ts

@@ -4,6 +4,7 @@ const log = getLog("clickhouseLogger");
 
 import { createClient } from "@clickhouse/client";
 import { EventsStore } from "./index";
+import { Readable } from "stream";
 
 type LogEntry = {
   actorId: string;
@@ -34,6 +35,7 @@ export function createClickhouseLogger(): EventsStore {
       async_insert: 1,
       wait_for_async_insert: 0,
       async_insert_busy_timeout_ms: 10000,
+      async_insert_busy_timeout_max_ms: 10000,
       date_time_input_format: "best_effort",
     },
   });
@@ -42,18 +44,32 @@ export function createClickhouseLogger(): EventsStore {
     if (buffer.length === 0) {
       return;
     }
-    const copy = [...buffer];
+    const copy = buffer.slice();
     buffer.length = 0;
-    const res = await clickhouse.insert<LogEntry>({
+    const eventsStream = new Readable({ objectMode: true });
+    const res = clickhouse.insert<LogEntry>({
       table: metricsSchema + ".events_log",
       format: "JSONEachRow",
-      values: copy,
+      values: eventsStream,
     });
-    if (res.executed) {
-      log.atDebug().log(`Inserted ${copy.length} records.`);
-    } else {
-      log.atError().log(`Failed to insert ${copy.length} records: ${JSON.stringify(res)}`);
-    }
+    const asyncWrite = async () => {
+      for (let i = 0; i < copy.length; i++) {
+        eventsStream.push(copy[i]);
+      }
+      eventsStream.push(null);
+      return res;
+    };
+    return asyncWrite()
+      .then(res => {
+        if (res.executed) {
+          log.atDebug().log(`Inserted ${copy.length} records.`);
+        } else {
+          log.atError().log(`Failed to insert ${copy.length} records: ${JSON.stringify(res)}`);
+        }
+      })
+      .catch(e => {
+        log.atError().withCause(e).log(`Failed to insert ${copy.length} records`);
+      });
   };
 
   const interval = setInterval(async () => {

libs/core-functions/src/functions/lib/udf_wrapper.ts

@@ -365,6 +365,7 @@ export async function UDFTestRun(
   let realStore = false;
   try {
     const eventContext: EventContext = {
+      receivedAt: new Date(),
       geo: {
         country: {
           code: "US",