New ManyIntersections Load Generator workload (#39273)

GitOrigin-RevId: 313b5a1acdc42150d1390345085d511aa0b97fb3

Author: thomasballinger

crates/load_generator/src/main.rs

@@ -133,6 +133,7 @@ enum Scenario {
     Search,
     VectorSearch,
     SnapshotExport,
+    ManyIntersections { num_subscriptions: i32 },
     CloudBackup,
     RunHttpAction { path: String, method: String },
 }
@@ -151,6 +152,7 @@ impl Scenario {
             | Scenario::SnapshotExport
             | Scenario::CloudBackup
             | Scenario::RunHttpAction { .. } => false,
+            Scenario::ManyIntersections { .. } => false,
         }
     }
 

crates/load_generator/src/metrics.rs

@@ -88,6 +88,15 @@ static ERROR_METRICS: LazyLock<BTreeMap<&str, BTreeMap<&str, IntCounterVec>>> =
                     )
                 }
             },
+            "ManyIntersections" => btreemap!{
+                "mutation" => {
+                    register_convex_counter_owned!(
+                        MANY_INTERSECTIONS_INSERT_ERROR_TOTAL,
+                        "Errors on the insert mutation in many intersections scenario",
+                        &["backend_version", "load_description"],
+                    )
+                }
+            },
             "SnapshotExport" => btreemap!{
                 "request_export_failed" => {
                     register_convex_counter_owned!(
@@ -250,6 +259,22 @@ static LATENCY_METRICS: LazyLock<BTreeMap<&str, BTreeMap<&str, VMHistogramVec>>>
                     )
                 }
             },
+            "ManyIntersections" => btreemap!{
+                "mutation_completed" => {
+                    register_convex_histogram_owned!(
+                        OBSERVE_INSERT_WITH_MANY_INTERSECTIONS_COMPLETED_LATENCY_SECONDS,
+                        "Latency on mutation completion for ManyIntersections",
+                        &["backend_version", "load_description"],
+                    )
+                },
+                "mutation_observed" => {
+                    register_convex_histogram_owned!(
+                        OBSERVE_INSERT_WITH_MANY_INTERSECTIONS_OBSERVED_LATENCY_SECONDS,
+                        "Latency on observing mutation for ManyIntersections",
+                        &["backend_version", "load_description"],
+                    )
+                },
+            },
             "Search" => btreemap!{
                 "search" => {
                     register_convex_histogram_owned!(
@@ -331,6 +356,22 @@ static COUNT_METRICS: LazyLock<BTreeMap<&str, BTreeMap<&str, IntCounterVec>>> =
                     )
                 }
             },
+            "ManyIntersections" => btreemap!{
+                "mutation_send_timeout" => {
+                    register_convex_counter_owned!(
+                        OBSERVE_INSERT_WITH_MANY_INTERSECTIONS_SEND_TIMEOUT_TOTAL,
+                        "Count of send timeouts for mutation in ManyIntersections",
+                        &["backend_version", "load_description"],
+                    )
+                },
+                "mutation_observed_timeout" => {
+                    register_convex_counter_owned!(
+                        OBSERVE_INSERT_WITH_MANY_INTERSECTIONS_OBSERVED_TIMEOUT_TOTAL,
+                        "Count of observed timeouts for mutation in ManyIntersections",
+                        &["backend_version", "load_description"],
+                    )
+                }
+            },
             "ObserveInsertWithSearch" => btreemap!{
                 "mutation_send_timeout" => {
                     register_convex_counter_owned!(

crates/load_generator/workloads/many_intersections.json

@@ -0,0 +1,10 @@
+{
+  "name": "many_intersections",
+  "scenarios": [
+    {
+      "name": "ManyIntersections",
+      "benchmark": 1,
+      "num_subscriptions": 20000
+    }
+  ]
+}

npm-packages/@convex-dev/platform/src/generatedApi.ts

@@ -144,13 +144,10 @@ export interface components {
         AdminKey: string;
         CreateProjectArgs: {
             deploymentType?: null | components["schemas"]["DeploymentType"];
-            partitionId?: null | components["schemas"]["PartitionId"];
             projectName: components["schemas"]["ProjectName"];
             team: components["schemas"]["TeamSlug"];
         };
         ProjectName: string;
-        /** Format: int64 */
-        PartitionId: number;
         /** @enum {string} */
         DeploymentType: "dev" | "prod" | "preview";
         InstanceAuthForDashboardInteractionsResponse: {
@@ -259,7 +256,6 @@ export type ProjectId = components['schemas']['ProjectId'];
 export type AdminKey = components['schemas']['AdminKey'];
 export type CreateProjectArgs = components['schemas']['CreateProjectArgs'];
 export type ProjectName = components['schemas']['ProjectName'];
-export type PartitionId = components['schemas']['PartitionId'];
 export type DeploymentType = components['schemas']['DeploymentType'];
 export type InstanceAuthForDashboardInteractionsResponse = components['schemas']['InstanceAuthForDashboardInteractionsResponse'];
 export type SerializedAccessToken = components['schemas']['SerializedAccessToken'];

npm-packages/scenario-runner/convex/insert.ts

@@ -97,3 +97,31 @@ export const insertMessageWithArgs = mutation({
     );
   },
 });
+
+export const insertMessagesWithArgs = mutation({
+  args: {
+    channel: v.string(),
+    timestamp: v.number(),
+    rand: v.number(),
+    ballastCount: v.number(),
+    count: v.number(),
+    table: v.string(),
+    n: v.number(),
+  },
+  handler: async (
+    ctx,
+    { channel, timestamp, rand, ballastCount, count, table, n },
+  ) => {
+    for (let i = 0; i < n; i++) {
+      await insertMessageHelper(
+        ctx.db,
+        channel,
+        timestamp,
+        rand,
+        ballastCount,
+        count,
+        table as TableNamesInDataModel<DataModel>,
+      );
+    }
+  },
+});

npm-packages/scenario-runner/index.ts

@@ -13,6 +13,7 @@ import { ConvexClient } from "convex/browser";
 import { ScenarioMessage } from "./types.js";
 import { RunHttpAction } from "./scenarios/run_http_action.js";
 import dns from "node:dns";
+import { ManyIntersections } from "./scenarios/many_intersections.js";
 
 Sentry.init({
   tracesSampleRate: 0.1,
@@ -115,6 +116,9 @@ async function runScenario(
     case "ObserveInsert":
       scenario = new ObserveInsert(config, scenarioSpec.search_indexes);
       break;
+    case "ManyIntersections":
+      scenario = new ManyIntersections(config, scenarioSpec.num_subscriptions);
+      break;
     case "SnapshotExport":
       scenario = new SnapshotExport(config, adminKey);
       break;

npm-packages/scenario-runner/metrics.ts

@@ -2,6 +2,7 @@
 // Metrics must first be added to crates/load_generator/src/metrics.rs
 export type ScenarioName =
   | "CloudBackup"
+  | "ManyIntersections"
   | "ObserveInsert"
   | "ObserveInsertWithSearch"
   | "RunFunction"
@@ -12,6 +13,9 @@ export type ScenarioName =
 export type ScenarioLatencyMetric =
   // CloudBackup
   | "backup"
+  // ManyIntersections
+  | "mutation_completed"
+  | "mutation_observed"
   // ObserveInsert
   | "mutation_completed"
   | "mutation_observed"
@@ -33,6 +37,9 @@ export type ScenarioCountMetric =
   | "backup_completed"
   | "backup_timeout"
   | "request_backup_succeeded"
+  // ManyIntersections
+  | "mutation_observed_timeout"
+  | "mutation_send_timeout"
   // ObserveInsert
   | "mutation_observed_timeout"
   | "mutation_send_timeout"
@@ -61,6 +68,8 @@ export type ScenarioError =
   | "backup_failure"
   | "get_backup_failed"
   | "request_backup_failed"
+  // ManyIntersections
+  | "mutation"
   // ObserveInsert
   | "mutation"
   // RunFunction

npm-packages/scenario-runner/scenario.ts

@@ -56,6 +56,12 @@ export abstract class Scenario {
     this.cleanUpFunctions = [];
   }
 
+  /**
+   * Subscribe to a query and return a promise for the first result
+   * received for which isReady(result) returns true.
+   *
+   * The subscription will remain active until the scenario ends but
+   */
   waitForQuery<Query extends FunctionReference<"query", "public">>(
     client: ConvexClient,
     query: Query,

npm-packages/scenario-runner/scenarios/many_intersections.ts

@@ -0,0 +1,110 @@
+import { MUTATION_TIMEOUT, QUERY_TIMEOUT } from "../types.js";
+import { Scenario, nowSeconds, IScenario, Config } from "../scenario.js";
+import { api } from "../convex/_generated/api.js";
+import { ScenarioError } from "../metrics.js";
+import { ConvexClient } from "convex/browser";
+import { MessagesTable, rand } from "../convex/common.js";
+import { writeFileSync } from "node:fs";
+
+/**
+ * Subscribe to many queries and invalidate a small number of them.
+ */
+export class ManyIntersections extends Scenario implements IScenario {
+  table: MessagesTable;
+  numSubscriptions: number;
+  numDocumentsPerMutation: number;
+
+  constructor(config: Config, numSubscriptions = 1000) {
+    const name = "ManyIntersections";
+    super(name as any, config);
+    this.table = "messages";
+    this.numSubscriptions = numSubscriptions;
+    this.numDocumentsPerMutation = 100;
+  }
+
+  async sendMutation(client: ConvexClient, startTime: number, rand: number) {
+    await client.mutation(api.insert.insertMessagesWithArgs, {
+      channel: "global",
+      timestamp: startTime,
+      rand,
+      ballastCount: 0,
+      count: 1,
+      table: this.table,
+      n: this.numDocumentsPerMutation,
+    });
+  }
+
+  async run(client: ConvexClient) {
+    writeFileSync(
+      "/Users/tomb/log.txt",
+      `Making ${this.numDocumentsPerMutation} document writes after ${this.numSubscriptions} subs at ${new Date().toTimeString().split(" ")[0]}...`,
+      {
+        encoding: "utf-8",
+      },
+    );
+    const startTime = nowSeconds();
+    const _ = range(this.numSubscriptions).map(() => {
+      void this.subscribe(client, startTime, rand(), `channel-${rand()}`);
+    });
+
+    // Wait a sec
+    await new Promise((r) => setTimeout(r, 1000));
+
+    const t0 = nowSeconds();
+    const randomNumber = rand();
+    const subscribe = this.subscribe(client, startTime, randomNumber);
+
+    // Send the replace mutation or timeout
+    await this.executeOrTimeoutWithLatency(
+      this.sendMutation(client, startTime, randomNumber),
+      MUTATION_TIMEOUT * 50,
+      "mutation_send_timeout",
+      "mutation_completed",
+      t0,
+    );
+
+    // Wait until the mutation is observed or timeout
+    await this.executeOrTimeoutWithLatency(
+      subscribe!,
+      QUERY_TIMEOUT * 50,
+      "mutation_observed_timeout",
+      "mutation_observed",
+      t0,
+    );
+  }
+
+  // Watch a query on the given offset of the table to observe the insert mutation
+  async subscribe(
+    client: ConvexClient,
+    startTime: number,
+    rand: number,
+    channel = "global",
+  ): Promise<void> {
+    await this.waitForQuery(
+      client,
+      api.query_index.queryMessagesWithArgs,
+      {
+        channel,
+        rand,
+        limit: 10,
+        table: this.table,
+      },
+      (result) => {
+        for (const doc of result || []) {
+          if (doc.rand === rand && doc.timestamp! >= startTime) {
+            return true;
+          }
+        }
+        return false;
+      },
+    );
+  }
+
+  defaultErrorName(): ScenarioError {
+    return "mutation";
+  }
+}
+
+function range(n: number): number[] {
+  return Array.from({ length: n }, (_, i) => i);
+}

npm-packages/scenario-runner/types.ts

@@ -73,6 +73,10 @@ export type ScenarioSpec =
       name: "ObserveInsert";
       search_indexes: boolean;
     }
+  | {
+      name: "ManyIntersections";
+      num_subscriptions: number;
+    }
   | {
       name: "Search";
     }