Change even more env variables -> cli flags

Author: coolreader18

templates/keynote-2/bun/bun-server.ts

@@ -110,7 +110,11 @@ async function rpcTransfer(args: Record<string, unknown>) {
   const toId = Number(args.to_id ?? args.to);
   const amount = Number(args.amount);
 
-  if (!Number.isInteger(fromId) || !Number.isInteger(toId) || !Number.isFinite(amount)) {
+  if (
+    !Number.isInteger(fromId) ||
+    !Number.isInteger(toId) ||
+    !Number.isFinite(amount)
+  ) {
     throw new Error('invalid transfer args');
   }
   if (fromId === toId || amount <= 0) return;
@@ -174,7 +178,6 @@ async function rpcGetAccount(args: Record<string, unknown>) {
   };
 }
 
-
 async function rpcVerify() {
   const rawInitial = process.env.SEED_INITIAL_BALANCE;
   if (!rawInitial) {

templates/keynote-2/spacetimedb-rust-client/src/main.rs

@@ -57,7 +57,7 @@ async fn init_conn(cli: &Common, handle: &Handle) -> (JoinHandle<()>, Recv, Send
     let params = WsParams {
         compression: Compression::None,
         light: true,
-        confirmed: cli.confirmed_reads.into(),
+        confirmed: cli.confirmed_reads.unwrap_or(CONFIRMED_READS).into(),
     };
 
     let conn = websocket::WsConnection::connect(uri, &cli.module, None, None, params)
@@ -146,7 +146,7 @@ fn bench(cli: &Common, bench: &Bench) {
 
     // Initialize connections.
     let connections = bench.connections;
-    let confirmed_reads = cli.confirmed_reads;
+    let confirmed_reads = cli.confirmed_reads.unwrap_or(CONFIRMED_READS);
     if !cli.quiet {
         println!("initializing {connections} connections with confirmed-reads={confirmed_reads}");
     }
@@ -261,8 +261,8 @@ struct Common {
     #[arg(short, long, default_value = MODULE)]
     module: String,
 
-    #[arg(long, default_value_t = CONFIRMED_READS)]
-    confirmed_reads: bool,
+    #[arg(long)]
+    confirmed_reads: Option<bool>,
 
     #[arg(long, default_value_t = ACCOUNTS)]
     accounts: u32,

templates/keynote-2/src/connectors/bun.ts

@@ -1,8 +1,7 @@
 import type { RpcConnector } from '../core/connectors.ts';
+import { bunUrl } from '../opts.ts';
 
-export default function bun(
-  url = process.env.BUN_URL || 'http://127.0.0.1:4000',
-): RpcConnector {
+export default function bun(url = bunUrl): RpcConnector {
   if (!url) throw new Error('BUN_URL not set');
 
   const baseUrl = url.replace(/\/+$/, '');

templates/keynote-2/src/connectors/convex.ts

@@ -1,8 +1,7 @@
 import type { RpcConnector } from '../core/connectors.ts';
+import { convexUrl } from '../opts.ts';
 
-export default function convex(
-  url = process.env.CONVEX_URL || 'http://127.0.0.1:3210',
-): RpcConnector {
+export default function convex(url = convexUrl): RpcConnector {
   if (!url) throw new Error('CONVEX_URL not set');
 
   function isWriteConflict(msg: unknown): boolean {

templates/keynote-2/src/connectors/index.ts

@@ -6,6 +6,7 @@ import cockroach_rpc from './rpc/cockroach_rpc.ts';
 import sqlite_rpc from './rpc/sqlite_rpc.ts';
 import supabase_rpc from './rpc/supabase_rpc.ts';
 import planetscale_pg_rpc from './rpc/planetscale_pg_rpc.ts';
+import { ConnectorKey } from '../opts.ts';
 
 export const CONNECTORS = {
   convex,
@@ -17,5 +18,4 @@ export const CONNECTORS = {
   sqlite_rpc,
   supabase_rpc,
   planetscale_pg_rpc,
-};
-export type ConnectorKey = keyof typeof CONNECTORS;
+} satisfies Record<ConnectorKey, unknown>;

templates/keynote-2/src/connectors/spacetimedb.ts

@@ -1,68 +1,67 @@
 import type { ReducerConnector } from '../core/connectors';
 import * as mod from '../../module_bindings';
+import {
+  initialBalance,
+  stdbConfirmedReads,
+  stdbModule,
+  stdbUrl,
+} from '../opts';
 
 export function spacetimedb(
-  url = process.env.STDB_URL!,
-  moduleName = process.env.STDB_MODULE!,
+  url = stdbUrl,
+  moduleName = stdbModule,
 ): ReducerConnector {
-  let ready!: Promise<void>;
+  let ready: ReturnType<typeof Promise.withResolvers<void>>;
   let conn: mod.DbConnection;
-  let resolveReady!: () => void;
-  let rejectReady!: (e: unknown) => void;
-
-  function armReady() {
-    ready = new Promise<void>((res, rej) => {
-      resolveReady = res;
-      rejectReady = rej;
-    });
-  }
 
   async function connectWithBindings() {
     if (!url) throw new Error('STDB_URL not set');
     if (!moduleName) throw new Error('STDB_MODULE not set');
 
     const Db = mod.DbConnection;
 
-    armReady();
+    ready = Promise.withResolvers<void>();
 
     const subscriptions: string[] = [];
     if (process.env.VERIFY === '1') {
       console.log('[spacetimedb] subscribing to accounts');
       subscriptions.push('SELECT * FROM accounts');
     }
-    let subscribed = subscriptions.length === 0;
+    const subscribed = Promise.withResolvers<void>();
+    if (subscriptions.length === 0) subscribed.resolve();
 
     const builder = Db.builder()
       .withUri(url)
       .withDatabaseName(moduleName)
-      .withConfirmedReads(process.env.STDB_CONFIRMED_READS === '1')
+      .withConfirmedReads(stdbConfirmedReads)
       .onConnect((ctx) => {
         console.log('[stdb] connected');
         const conn = ctx;
 
-        resolveReady();
+        ready.resolve();
 
         if (subscriptions.length > 0) {
           conn
             .subscriptionBuilder()
             .onApplied((_sCtx) => {
-              subscribed = true;
+              subscribed.resolve();
             })
             .onError((ctx) => {
               console.error('[stdb] subscription failed', ctx.event?.message);
+              subscribed.reject(ctx.event);
             })
             .subscribe(subscriptions);
         }
       })
       .onConnectError((_ctx, err: any) => {
-        console.error('[stdb] onConnectError', err);
-
         if (err instanceof Error) {
-          rejectReady(err);
+          ready.reject(err);
+        } else if (err && err.error instanceof Error) {
+          ready.reject(err.error);
         } else if (err && typeof err.message === 'string') {
-          rejectReady(new Error(err.message));
+          ready.reject(new Error(err.message));
         } else {
-          rejectReady(
+          ready.reject(
             new Error(`Spacetime connection error: ${JSON.stringify(err)}`),
           );
         }
@@ -71,9 +70,8 @@ export function spacetimedb(
 
     conn = builder.build();
 
-    while (!subscribed) {
-      await new Promise((res) => setTimeout(res, 25));
-    }
+    await ready.promise;
+    await subscribed.promise;
   }
 
   return {
@@ -83,9 +81,9 @@ export function spacetimedb(
     async open() {
       try {
         await connectWithBindings();
-        await ready;
+        await ready.promise;
       } catch (err) {
-        console.error('[spacetimedb] open() failed', err);
+        console.error('[spacetimedb] open() failed:', err);
         throw err;
       }
     },
@@ -94,7 +92,7 @@ export function spacetimedb(
       try {
         conn.disconnect();
       } catch (e) {
-        console.error('[spacetimedb] close() failed', e);
+        console.error('[spacetimedb] close() failed:', e);
       }
     },
 
@@ -111,7 +109,7 @@ export function spacetimedb(
     },
 
     async call(fn: string, args: Record<string, any>) {
-      await ready;
+      await ready.promise;
 
       switch (fn) {
         case 'seed': {
@@ -156,7 +154,7 @@ export function spacetimedb(
     async verify() {
       if (!conn) throw new Error('SpacetimeDB not connected');
 
-      const rawInitial = process.env.SEED_INITIAL_BALANCE;
+      const rawInitial = initialBalance;
       if (!rawInitial) {
         console.warn(
           '[spacetimedb] SEED_INITIAL_BALANCE not set; skipping verification',

templates/keynote-2/src/core/runner.ts

@@ -5,11 +5,22 @@ import { getSpacetimeCommittedTransfers } from './spacetimeMetrics.ts';
 import { makeCollisionTracker } from './collision_tracker.ts';
 import { RunResult } from './types.ts';
 import { BaseConnector } from './connectors.ts';
-
-const OP_TIMEOUT_MS = Number(process.env.BENCH_OP_TIMEOUT_MS ?? '15000');
-const MIN_OP_TIMEOUT_MS = Number(process.env.MIN_OP_TIMEOUT_MS ?? '250');
-const TAIL_SLACK_MS = Number(process.env.TAIL_SLACK_MS ?? '1000');
-const DEFAULT_PRECOMPUTED_TRANSFER_PAIRS = 10_000_000;
+import {
+  benchPipelined,
+  logErrors,
+  maxInflightPerWorker,
+  minOpTimeoutMs,
+  opTimeoutMs,
+  precomputedTransferPairs,
+  tailSlackMs,
+  useSpacetimeMetricsEndpoint,
+  verifyTransactions,
+} from '../opts.ts';
+
+const OP_TIMEOUT_MS = opTimeoutMs;
+const MIN_OP_TIMEOUT_MS = minOpTimeoutMs;
+const TAIL_SLACK_MS = tailSlackMs;
+const PRECOMPUTED_TRANSFER_PAIRS = precomputedTransferPairs;
 
 function precomputeZipfTransferPairs(
   accounts: number,
@@ -101,8 +112,7 @@ export async function runOne({
   }
 
   const useSpacetimeMetrics =
-    process.env.USE_SPACETIME_METRICS_ENDPOINT === '1' &&
-    connector.name === 'spacetimedb';
+    useSpacetimeMetricsEndpoint && connector.name === 'spacetimedb';
   let beforeTransfers: bigint | null = null;
 
   if (useSpacetimeMetrics) {
@@ -126,13 +136,7 @@ export async function runOne({
     }
   }
 
-  const precomputedPairsRaw = Number(
-    process.env.BENCH_PRECOMPUTED_TRANSFER_PAIRS ??
-      DEFAULT_PRECOMPUTED_TRANSFER_PAIRS,
-  );
-  const precomputedPairs = Number.isFinite(precomputedPairsRaw)
-    ? Math.max(1, Math.floor(precomputedPairsRaw))
-    : DEFAULT_PRECOMPUTED_TRANSFER_PAIRS;
+  const precomputedPairs = PRECOMPUTED_TRANSFER_PAIRS;
 
   console.log(
     `[${connector.name}] precomputing ${precomputedPairs} Zipf transfer pairs...`,
@@ -148,27 +152,13 @@ export async function runOne({
     `[${connector.name}] precomputed ${transferPairs.count} pairs in ${(precomputeElapsedMs / 1000).toFixed(2)}s`,
   );
 
-  const getEnvTernary = (envVal: string | undefined) => {
-    switch (envVal) {
-      case '0':
-        return false;
-      case '1':
-        return true;
-      default:
-        return null;
-    }
-  };
-
-  const PIPELINED =
-    getEnvTernary(process.env.BENCH_PIPELINED) ??
-    !!connector.maxInflightPerWorker;
-  const MAX_INFLIGHT_ENV = process.env.MAX_INFLIGHT_PER_WORKER;
+  const PIPELINED = benchPipelined ?? !!connector.maxInflightPerWorker;
   const MAX_INFLIGHT_PER_WORKER =
-    MAX_INFLIGHT_ENV == null
+    maxInflightPerWorker === undefined
       ? (connector.maxInflightPerWorker ?? 8)
-      : MAX_INFLIGHT_ENV === '0'
+      : maxInflightPerWorker == 0
         ? Infinity
-        : Number(MAX_INFLIGHT_ENV);
+        : maxInflightPerWorker;
 
   console.log(
     `[${connector.name}] max inflight per worker: ${MAX_INFLIGHT_PER_WORKER}`,
@@ -240,7 +230,7 @@ export async function runOne({
             );
             ok = true;
           } catch (err) {
-            if (process.env.LOG_ERRORS === '1') {
+            if (logErrors) {
               const msg =
                 err instanceof Error
                   ? `${err.name}: ${err.message}`
@@ -292,7 +282,7 @@ export async function runOne({
               hist.recordValue(Math.max(1, Math.round((t1 - t0) * 1e3)));
             }
           } catch (err) {
-            if (process.env.LOG_ERRORS === '1') {
+            if (logErrors) {
               const msg =
                 err instanceof Error
                   ? `${err instanceof Error ? err.message : String(err)}`
@@ -383,10 +373,10 @@ export async function runOne({
     return { start, completedWithinWindow, completedTotal, committedDelta };
   };
 
-  const warmUpSeconds = 5;
-  console.log(`[${connector.name}] Warming up for ${warmUpSeconds}s...`);
-  await run(warmUpSeconds);
-  console.log(`[${connector.name}] Finished warmup.`);
+  // const warmUpSeconds = 5;
+  // console.log(`[${connector.name}] Warming up for ${warmUpSeconds}s...`);
+  // await run(warmUpSeconds);
+  // console.log(`[${connector.name}] Finished warmup.`);
 
   console.log(`[${connector.name}] Starting workers for ${seconds}s run...`);
 
@@ -397,7 +387,7 @@ export async function runOne({
     `[${connector.name}] All workers finished (including in-flight ops)`,
   );
 
-  if (process.env.VERIFY === '1') {
+  if (verifyTransactions) {
     console.log(`[${connector.name}] Running verification pass...`);
     try {
       await withOpTimeout(connector.verify(), `${connector.name} verify()`);