use a less CPU-intensive way of inserting task runs

Author: ericallam

apps/webapp/app/services/runsReplicationService.server.ts

@@ -772,7 +772,7 @@ export class RunsReplicationService {
 
   async #insertTaskRunInserts(taskRunInserts: TaskRunV2[], attempt: number) {
     return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
-      const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insert(
+      const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insertUnsafe(
         taskRunInserts,
         {
           params: {
@@ -797,7 +797,7 @@ export class RunsReplicationService {
 
   async #insertPayloadInserts(payloadInserts: RawTaskRunPayloadV1[], attempt: number) {
     return await startSpan(this._tracer, "insertPayloadInserts", async (span) => {
-      const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insertPayloads(
+      const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insertPayloadsUnsafe(
         payloadInserts,
         {
           params: {

apps/webapp/scripts/profile-runs-replication.ts

@@ -3,6 +3,7 @@
 import { program } from "commander";
 import path from "path";
 import fs from "fs/promises";
+import { spawn } from "child_process";
 import { RunsReplicationHarness } from "../test/performance/harness";
 import { getDefaultConfig, type HarnessConfig } from "../test/performance/config";
 
@@ -243,6 +244,48 @@ async function createSummaryReport(
   await fs.writeFile(outputPath, lines.join("\n"));
 }
 
+async function generateVisualization(config: HarnessConfig): Promise<void> {
+  console.log("\n🔥 Generating flamegraph visualization...");
+
+  // Find the clinic flame data file
+  const files = await fs.readdir(config.profiling.outputDir);
+  const flameDataFile = files.find((f) => f.endsWith(".clinic-flame"));
+
+  if (!flameDataFile) {
+    console.warn("⚠️  No clinic flame data file found. Skipping visualization.");
+    return;
+  }
+
+  const dataPath = path.join(config.profiling.outputDir, flameDataFile);
+  console.log(`   Data file: ${flameDataFile}`);
+
+  // Run clinic flame --visualize-only
+  const clinicPath = path.join(__dirname, "../node_modules/.bin/clinic");
+
+  await new Promise<void>((resolve, reject) => {
+    const proc = spawn(clinicPath, ["flame", "--visualize-only", dataPath], {
+      stdio: "inherit",
+      cwd: path.join(__dirname, ".."),
+    });
+
+    proc.on("exit", (code) => {
+      if (code === 0) {
+        console.log("✅ Flamegraph generated successfully!");
+        console.log(`   Open: ${dataPath.replace(".clinic-flame", ".clinic-flame.html")}\n`);
+        resolve();
+      } else {
+        console.error(`⚠️  Flamegraph generation exited with code ${code}`);
+        resolve(); // Don't fail the whole run
+      }
+    });
+
+    proc.on("error", (error) => {
+      console.error("⚠️  Error generating flamegraph:", error.message);
+      resolve(); // Don't fail the whole run
+    });
+  });
+}
+
 async function main() {
   const options = program.opts();
   const config = await loadConfig(options);
@@ -272,8 +315,8 @@ async function main() {
     console.log(`📊 Detailed metrics: ${config.output.metricsFile}\n`);
 
     if (config.profiling.enabled && config.profiling.tool !== "none") {
-      console.log("🔥 Profiling data:");
-      console.log(`   View flamegraph/analysis in: ${config.profiling.outputDir}\n`);
+      // Generate visualization from collected data
+      await generateVisualization(config);
     }
 
     process.exit(0);

apps/webapp/test/performance/config.ts

@@ -38,6 +38,7 @@ export interface ConsumerConfig {
   redisOptions: RedisOptions;
   slotName: string;
   publicationName: string;
+  outputDir?: string; // For shutdown signal file
 }
 
 export interface ProfilingConfig {

apps/webapp/test/performance/consumer-runner.ts

@@ -3,17 +3,24 @@
 import { RunsReplicationService } from "~/services/runsReplicationService.server";
 import { MockClickHouse } from "./clickhouse-mock";
 import type { ConsumerConfig } from "./config";
+import fs from "fs";
+import path from "path";
 
 async function main() {
   const hasIPC = !!process.send;
 
   if (!hasIPC) {
-    console.log("Warning: IPC not available (likely running under profiler) - metrics will not be sent to parent");
+    console.log(
+      "Warning: IPC not available (likely running under profiler) - metrics will not be sent to parent"
+    );
   }
 
   // Parse configuration from environment variable
   const config: ConsumerConfig = JSON.parse(process.env.CONSUMER_CONFIG!);
 
+  // Create shutdown signal file path
+  const shutdownFilePath = path.join(config.outputDir || "/tmp", ".shutdown-signal");
+
   console.log("Consumer process starting with config:", {
     useMockClickhouse: config.useMockClickhouse,
     flushBatchSize: config.flushBatchSize,
@@ -34,6 +41,7 @@ async function main() {
       compression: {
         request: true,
       },
+      logLevel: "info",
     });
   }
 
@@ -64,6 +72,20 @@ async function main() {
     });
   }
 
+  // Watch for shutdown signal file (works even when IPC is unavailable)
+  const shutdownCheckInterval = setInterval(() => {
+    if (fs.existsSync(shutdownFilePath)) {
+      console.log("Consumer: Shutdown signal file detected, exiting...");
+      clearInterval(shutdownCheckInterval);
+      clearInterval(metricsInterval);
+      // Clean up the signal file
+      try {
+        fs.unlinkSync(shutdownFilePath);
+      } catch (e) {}
+      process.exit(0);
+    }
+  }, 500);
+
   // Send periodic metrics to parent (if IPC available)
   const metricsInterval = setInterval(() => {
     const memUsage = process.memoryUsage();
@@ -82,14 +104,13 @@ async function main() {
     }
   }, 1000);
 
-  // Listen for shutdown signal from parent (if IPC available)
+  // Listen for shutdown signal from parent (if IPC available, for non-profiling mode)
   if (hasIPC) {
     process.on("message", async (msg: any) => {
       if (msg.type === "shutdown") {
-        console.log("Consumer: Received shutdown signal");
+        console.log("Consumer: Received IPC shutdown message");
+        clearInterval(shutdownCheckInterval);
         clearInterval(metricsInterval);
-        await service.shutdown();
-        await service.stop();
         process.exit(0);
       }
     });

apps/webapp/test/performance/consumer.ts

@@ -1,5 +1,6 @@
 import { ChildProcess, spawn } from "child_process";
 import path from "path";
+import fs from "fs";
 import type { ConsumerConfig, ProfilingConfig } from "./config";
 
 interface ConsumerMetrics {
@@ -37,6 +38,7 @@ export class ConsumerProcessManager {
     const isProfiling = this.profiling.enabled && this.profiling.tool !== "none";
 
     this.process = spawn(args[0], args.slice(1), {
+      cwd: path.join(__dirname, "../.."), // Run from webapp directory
       env: {
         ...process.env,
         CONSUMER_CONFIG: JSON.stringify(this.config),
@@ -99,18 +101,45 @@ export class ConsumerProcessManager {
     }
 
     console.log("Stopping consumer process");
-    this.process.send({ type: "shutdown" });
+
+    const isProfiling = this.profiling.enabled && this.profiling.tool !== "none";
+
+    if (isProfiling) {
+      // When profiling, IPC doesn't work - use a shutdown signal file instead
+      const outputDir = this.config.outputDir || "/tmp";
+      const shutdownFilePath = path.join(outputDir, ".shutdown-signal");
+      console.log("Creating shutdown signal file for consumer (profiling mode)");
+
+      // Ensure directory exists
+      if (!fs.existsSync(outputDir)) {
+        fs.mkdirSync(outputDir, { recursive: true });
+      }
+
+      fs.writeFileSync(shutdownFilePath, "shutdown");
+    } else {
+      // For non-profiling runs, use IPC message
+      try {
+        this.process.send({ type: "shutdown" });
+      } catch (error) {
+        console.warn("Could not send shutdown message, process may have already exited");
+      }
+    }
 
     // Wait for process to exit
     await new Promise<void>((resolve) => {
+      // With shutdown signal file, consumer-runner should exit within a few seconds
+      // With --collect-only, Clinic.js then quickly packages the data and exits
+      const timeoutMs = isProfiling ? 15000 : 30000;
+
       const timeout = setTimeout(() => {
-        console.warn("Consumer process did not exit gracefully, killing");
+        console.warn(`Consumer process did not exit after ${timeoutMs}ms, killing`);
         this.process?.kill("SIGKILL");
         resolve();
-      }, 30000); // 30 second timeout
+      }, timeoutMs);
 
-      this.process?.on("exit", () => {
+      this.process?.on("exit", (code, signal) => {
         clearTimeout(timeout);
+        console.log(`Consumer process exited with code ${code}, signal ${signal}`);
         resolve();
       });
     });
@@ -144,21 +173,30 @@ export class ConsumerProcessManager {
     const tool = this.profiling.tool === "both" ? "doctor" : this.profiling.tool;
     const runnerPath = path.join(__dirname, "consumer-runner.ts");
 
+    // Use clinic from node_modules/.bin directly (more reliable than pnpm exec)
+    const clinicPath = path.join(__dirname, "../../node_modules/.bin/clinic");
+
+    // Point --dest to the output directory itself
+    // Clinic.js will create PID.clinic-flame inside this directory
+    const destPath = path.resolve(this.profiling.outputDir);
+
     // Clinic.js requires node, so use node with tsx/register loader
     const args = [
-      "pnpm",
-      "exec",
-      "clinic",
+      clinicPath,
       tool,
+      "--collect-only", // Only collect data, don't generate visualization immediately
+      "--open=false", // Don't try to open in browser
       "--dest",
-      this.profiling.outputDir,
+      destPath,
       "--",
       "node",
       "--import",
       "tsx",
       runnerPath,
     ];
 
+    console.log(`Clinic.js will save profiling data to: ${destPath}`);
+
     return args;
   }
 

apps/webapp/test/performance/harness.ts

@@ -97,6 +97,9 @@ export class RunsReplicationHarness {
 
     // 9. Start consumer process
     console.log("\nStarting consumer process...");
+    // Set outputDir for shutdown signal file (needed when IPC isn't available under Clinic.js)
+    this.config.consumer.outputDir = this.config.profiling.outputDir;
+
     this.consumerProcess = new ConsumerProcessManager(
       this.config.consumer,
       this.config.profiling
@@ -282,15 +285,59 @@ export class RunsReplicationHarness {
     );
 
     // Drop existing replication slot if it exists (ensures clean slate)
-    const slotExists = await this.prisma!.$queryRaw<Array<{ slot_name: string }>>`
-      SELECT slot_name FROM pg_replication_slots WHERE slot_name = ${this.config.consumer.slotName};
+    const slotExists = await this.prisma!.$queryRaw<
+      Array<{ slot_name: string; active: boolean; active_pid: number | null }>
+    >`
+      SELECT slot_name, active, active_pid FROM pg_replication_slots WHERE slot_name = ${this.config.consumer.slotName};
     `;
 
     if (slotExists.length > 0) {
+      const slot = slotExists[0];
       console.log(`🧹 Dropping existing replication slot: ${this.config.consumer.slotName}`);
-      await this.prisma!.$executeRawUnsafe(
-        `SELECT pg_drop_replication_slot('${this.config.consumer.slotName}');`
-      );
+
+      // If the slot is active, terminate the backend process first
+      if (slot.active && slot.active_pid) {
+        console.log(
+          `⚠️  Replication slot is active for PID ${slot.active_pid}, terminating backend...`
+        );
+        try {
+          await this.prisma!.$executeRawUnsafe(
+            `SELECT pg_terminate_backend(${slot.active_pid});`
+          );
+          console.log(`✅ Terminated backend process ${slot.active_pid}`);
+
+          // Wait a bit for the termination to complete
+          await new Promise((resolve) => setTimeout(resolve, 1000));
+        } catch (error) {
+          console.warn(
+            `⚠️  Could not terminate backend ${slot.active_pid}, it may have already exited`
+          );
+        }
+      }
+
+      // Now try to drop the slot with retry logic
+      let attempts = 0;
+      const maxAttempts = 5;
+      while (attempts < maxAttempts) {
+        try {
+          await this.prisma!.$executeRawUnsafe(
+            `SELECT pg_drop_replication_slot('${this.config.consumer.slotName}');`
+          );
+          console.log(`✅ Dropped replication slot: ${this.config.consumer.slotName}`);
+          break;
+        } catch (error: any) {
+          attempts++;
+          if (attempts === maxAttempts) {
+            throw new Error(
+              `Failed to drop replication slot after ${maxAttempts} attempts: ${error.message}`
+            );
+          }
+          console.log(
+            `⚠️  Slot still active, waiting 2s before retry (attempt ${attempts}/${maxAttempts})...`
+          );
+          await new Promise((resolve) => setTimeout(resolve, 2000));
+        }
+      }
     }
 
     // Drop and recreate publication (ensures clean slate)

internal-packages/clickhouse/src/index.ts

@@ -4,7 +4,9 @@ import { ClickhouseReader, ClickhouseWriter } from "./client/types.js";
 import { NoopClient } from "./client/noop.js";
 import {
   insertTaskRuns,
+  insertTaskRunsUnsafe,
   insertRawTaskRunPayloads,
+  insertRawTaskRunPayloadsUnsafe,
   getTaskRunsQueryBuilder,
   getTaskActivityQueryBuilder,
   getCurrentRunningStats,
@@ -169,7 +171,9 @@ export class ClickHouse {
   get taskRuns() {
     return {
       insert: insertTaskRuns(this.writer),
+      insertUnsafe: insertTaskRunsUnsafe(this.writer),
       insertPayloads: insertRawTaskRunPayloads(this.writer),
+      insertPayloadsUnsafe: insertRawTaskRunPayloadsUnsafe(this.writer),
       queryBuilder: getTaskRunsQueryBuilder(this.reader),
       countQueryBuilder: getTaskRunsCountQueryBuilder(this.reader),
       tagQueryBuilder: getTaskRunTagsQueryBuilder(this.reader),