#!/usr/bin/env node import { spawn } from "node:child_process"; import { createRequire } from "node:module"; import { mkdirSync, writeFileSync, appendFileSync, readFileSync } from "node:fs"; import { dirname, join, resolve } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; const repoRoot = resolve(new URL("..", import.meta.url).pathname); const require = createRequire(import.meta.url); const transit = require(require.resolve("transit-js", { paths: [resolve(repoRoot, "cli")] })); const transitWriter = transit.writer("json"); const transitReader = transit.reader("json"); export function parseArgs(argv) { const opts = { graph: "pi-memory", page: "CLI Concurrent Stress", extraPages: 2, journalPages: 2, journalStart: new Date().toISOString().slice(0, 10), todayDate: null, config: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "cli.edn"), syncBase: "http://127.0.0.1:18080", clients: 1, concurrency: 8, maxOps: 0, timeoutMs: 20000, settleAttempts: 30, settleMs: 1000, logFile: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "events.jsonl"), sync: false, offline: false, offlineMs: 2500, failFast: false, startSyncServer: true, graphE2ee: true, e2eePassword: "11111", authPath: "/Users/tiensonqin/logseq/auth.json", syncServerPidFile: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "db-sync-server.pid"), syncServerLogFile: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "db-sync-server.log"), syncServerDataDir: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "db-sync-server-data"), }; for (let i = 0; i < argv.length; i += 1) { const arg = argv[i]; const next = () => { i += 1; if (i >= argv.length) { throw new Error(`Missing value for ${arg}`); } return argv[i]; }; switch (arg) { case "--graph": opts.graph = next(); break; case "--page": opts.page = next(); break; case "--extra-pages": opts.extraPages = Number.parseInt(next(), 10); break; case "--journal-pages": opts.journalPages = Number.parseInt(next(), 10); break; case "--journal-start": opts.journalStart = next(); break; case "--today-date": opts.todayDate = next(); break; case "--config": opts.config = resolve(next()); break; case "--root-dir": opts.rootDir = resolve(next()); break; case "--sync-base": opts.syncBase = next(); break; case "--clients": opts.clients = Number.parseInt(next(), 10); break; case "--concurrency": opts.concurrency = Number.parseInt(next(), 10); break; case "--max-ops": opts.maxOps = Number.parseInt(next(), 10); break; case "--timeout-ms": opts.timeoutMs = Number.parseInt(next(), 10); break; case "--settle-attempts": opts.settleAttempts = Number.parseInt(next(), 10); break; case "--settle-ms": opts.settleMs = Number.parseInt(next(), 10); break; case "--log-file": opts.logFile = resolve(next()); break; case "--sync-server-pid-file": opts.syncServerPidFile = resolve(next()); break; case "--sync-server-log-file": opts.syncServerLogFile = resolve(next()); break; case "--sync-server-data-dir": opts.syncServerDataDir = resolve(next()); break; case "--sync": opts.sync = true; break; case "--offline": opts.offline = true; break; case "--offline-ms": opts.offlineMs = Number.parseInt(next(), 10); break; case "--no-start-sync-server": opts.startSyncServer = false; break; case "--no-e2ee": opts.graphE2ee = false; break; case "--auth-path": opts.authPath = resolve(next()); break; case "--e2ee-password": opts.e2eePassword = next(); break; case "--fail-fast": opts.failFast = true; break; case "--help": case "-h": printHelp(); process.exit(0); break; default: throw new Error(`Unknown argument: ${arg}`); } } if (!Number.isInteger(opts.concurrency) || opts.concurrency < 1) { throw new Error("--concurrency must be a positive integer"); } if (!Number.isInteger(opts.clients) || opts.clients < 1) { throw new Error("--clients must be a positive integer"); } if (opts.clients > 1 && !opts.sync) { throw new Error("--clients greater than 1 requires --sync"); } if (!Number.isInteger(opts.maxOps) || opts.maxOps < 0) { throw new Error("--max-ops must be 0 or a positive integer"); } if (!Number.isInteger(opts.timeoutMs) || opts.timeoutMs < 1000) { throw new Error("--timeout-ms must be at least 1000"); } if (!Number.isInteger(opts.settleAttempts) || opts.settleAttempts < 1) { throw new Error("--settle-attempts must be a positive integer"); } if (!Number.isInteger(opts.settleMs) || opts.settleMs < 100) { throw new Error("--settle-ms must be at least 100"); } if (!Number.isInteger(opts.extraPages) || opts.extraPages < 0) { throw new Error("--extra-pages must be 0 or a positive integer"); } if (!Number.isInteger(opts.journalPages) || opts.journalPages < 0) { throw new Error("--journal-pages must be 0 or a positive integer"); } if (!Number.isInteger(opts.offlineMs) || opts.offlineMs < 100) { throw new Error("--offline-ms must be at least 100"); } if (opts.todayDate && Number.isNaN(new Date(`${opts.todayDate}T00:00:00.000Z`).getTime())) { throw new Error("--today-date must be a valid YYYY-MM-DD date"); } opts.wsUrl = `${opts.syncBase.replace(/^http/, "ws").replace(/\/$/, "")}/sync/%s`; opts.httpBase = opts.syncBase.replace(/\/$/, ""); return opts; } function printHelp() { console.log(`Usage: node scripts/cli-concurrent-edit-stress.mjs [options] Runs parallel Logseq CLI writes against a dedicated stress page. Options: --graph NAME Graph to mutate. Default: pi-memory --page NAME Stress page. Default: CLI Concurrent Stress --extra-pages N Additional normal pages to mutate. Default: 2 --journal-pages N Auto-created journal-style pages to mutate. Default: 2 --journal-start DATE First journal date, YYYY-MM-DD. Default: today --today-date DATE Override today's journal date, YYYY-MM-DD --config PATH Dedicated CLI config path under tmp/ --root-dir PATH Dedicated root dir for the seed CLI client --sync-base URL Local db-sync base URL. Default: http://127.0.0.1:18080 --sync Start db-sync client after verifying local /health --clients N Independent local clients synced to one remote graph. Default: 1 --offline Simulate offline windows by stopping sync during writes --offline-ms N Duration of each offline window. Default: 2500 --no-start-sync-server Do not auto-start local db-sync when --sync is used --no-e2ee Initialize the stress graph as non-E2EE for local sync --auth-path PATH Auth JSON for local db-sync. Default: /Users/tiensonqin/logseq/auth.json --e2ee-password TEXT Password used for local sync upload initialization. Default: 11111 --concurrency N Parallel workers. Default: 8 --max-ops N Stop after N operations. Default: 0, run forever --timeout-ms N Per-CLI-command timeout. Default: 20000 --settle-attempts N Final sync status polling attempts. Default: 30 --settle-ms N Delay between sync status polls. Default: 1000 --log-file PATH JSONL event log path --sync-server-pid-file PATH Local db-sync server pid file --sync-server-log-file PATH Local db-sync server log file --sync-server-data-dir PATH Local db-sync server data dir --fail-fast Exit on first CLI error `); } function assertLocalUrl(url, label) { let parsed; try { parsed = new URL(url.replace("%s", "graph-id")); } catch { throw new Error(`${label} is not a valid URL: ${url}`); } const localHosts = new Set(["127.0.0.1", "localhost", "::1"]); if (!localHosts.has(parsed.hostname)) { throw new Error(`${label} must point to localhost/127.0.0.1, got ${url}`); } } function authConfigLines(opts) { const auth = JSON.parse(readFileSync(opts.authPath, "utf8")); const lines = [` :auth-path "${opts.authPath}"`]; if (typeof auth["refresh-token"] === "string" && auth["refresh-token"].length > 0) { lines.push(` :refresh-token ${JSON.stringify(auth["refresh-token"])}`); } if (typeof auth["id-token"] === "string" && auth["id-token"].length > 0) { lines.push(` :id-token ${JSON.stringify(auth["id-token"])}`); } if (typeof auth["access-token"] === "string" && auth["access-token"].length > 0) { lines.push(` :access-token ${JSON.stringify(auth["access-token"])}`); } return lines; } export function stressConfigText(opts) { return [ "{", " :output-format :json", ` :http-base "${opts.httpBase}"`, ` :ws-url "${opts.wsUrl}"`, ...authConfigLines(opts), "}", "", ].join("\n"); } function ensureStressConfig(opts) { assertLocalUrl(opts.httpBase, "http-base"); assertLocalUrl(opts.wsUrl, "ws-url"); if (opts.rootDir) { mkdirSync(opts.rootDir, { recursive: true }); } if (opts.homeDir) { mkdirSync(opts.homeDir, { recursive: true }); } mkdirSync(dirname(opts.config), { recursive: true }); writeFileSync(opts.config, stressConfigText(opts), "utf8"); } function logEvent(opts, event) { const payload = { ts: new Date().toISOString(), graph: opts.graph, page: opts.page, ...event, }; appendFileSync(opts.logFile, `${JSON.stringify(payload)}\n`, "utf8"); if (event.level === "error" || event.level === "warn" || event.event === "summary") { console.log(JSON.stringify(payload)); } } const expectedRaceErrorCodes = new Set([ "block-not-found", "page-not-found", "recycled-page", "source-not-found", "target-not-found", "upsert-id-not-found", ]); function expectedCliRace(parsed, context = {}) { return ( expectedRaceErrorCodes.has(parsed?.error?.code) || (context.op === "property-delete-recreate" && context.phase === "delete" && (parsed?.error?.code === "property-not-found" || parsed?.error?.code === "ambiguous-property-name")) || (context.op === "tag-delete-recreate" && context.phase === "delete" && parsed?.error?.code === "tag-not-found") ); } export function classifyCliResult(result, parsed, context = {}) { const ok = result.code === 0 && (!parsed || parsed.status === "ok"); const expectedRace = !ok && expectedCliRace(parsed, context); return { ok, level: ok || expectedRace ? "info" : "error", outcome: ok ? "ok" : expectedRace ? "race-conflict" : "failed", expectedRace, }; } const rawViewHistoryOps = new Set([ "http-create-property-history", "http-create-linked-references-view", "http-create-unlinked-references-view", "http-delete-view-target-with-related-entities", ]); function expectedHttpRace(parsed, context = {}) { const message = parsed?.error?.message || ""; return ( (Number.isInteger(context.id) && parsed?.error?.code === "exception" && message === "Set block property failed: block or property doesn't exist") || (Number.isInteger(context.id) && rawViewHistoryOps.has(context.op) && parsed?.error?.code === "exception" && (message.startsWith("Nothing found for entity id (:block/uuid") || message.startsWith("Nothing found for entity id [:block/uuid"))) ); } export function classifyHttpResult(result, parsed, context = {}) { const noOp = result.ok && context.requireTruthyResult && !parsed?.result; const ok = result.ok && !noOp; const expectedRace = !ok && !noOp && expectedHttpRace(parsed, context); return { ok, level: ok || expectedRace ? "info" : "error", outcome: ok ? "ok" : expectedRace ? "race-conflict" : noOp ? "no-op" : "failed", expectedRace, }; } export function staleIdsFromContext(context = {}) { const ids = []; for (const key of ["id", "targetId", "parentId"]) { if (Number.isInteger(context[key])) { ids.push(context[key]); } } if (Array.isArray(context.ids)) { for (const id of context.ids) { if (Number.isInteger(id)) { ids.push(id); } } } return ids; } function runProcess(command, args, opts) { return new Promise((resolvePromise) => { const startedAt = Date.now(); const child = spawn(command, args, { cwd: repoRoot, env: { ...process.env, ...(opts.env || {}), LOGSEQ_CLI_CONFIG: opts.config, LOGSEQ_CLI_GRAPH: opts.graph, LOGSEQ_CLI_TIMEOUT_MS: String(opts.timeoutMs), LOGSEQ_CLI_OUTPUT: "json", }, stdio: ["ignore", "pipe", "pipe"], }); let stdout = ""; let stderr = ""; const timer = setTimeout(() => { child.kill("SIGTERM"); setTimeout(() => child.kill("SIGKILL"), 1500).unref(); }, opts.timeoutMs + 3000); child.stdout.on("data", (chunk) => { stdout += chunk.toString(); }); child.stderr.on("data", (chunk) => { stderr += chunk.toString(); }); child.on("close", (code, signal) => { clearTimeout(timer); resolvePromise({ command, args, code, signal, stdout: stdout.trim(), stderr: stderr.trim(), elapsedMs: Date.now() - startedAt, }); }); }); } async function runCli(opts, args, context = {}) { const command = cliCommand(process.env.LOGSEQ_BIN); const result = await runProcess(command.command, [...command.args, ...args], opts); let parsed = null; if (result.stdout) { try { parsed = JSON.parse(result.stdout); } catch { parsed = null; } } const classification = classifyCliResult(result, parsed, context); const event = { event: "cli", level: classification.level, outcome: classification.outcome, context, args, code: result.code, signal: result.signal, elapsedMs: result.elapsedMs, status: parsed?.status, error: parsed?.error, stderr: result.stderr || undefined, stdout: parsed ? undefined : result.stdout, }; logEvent(opts, event); if (!classification.ok && !classification.expectedRace && opts.failFast) { throw new Error(`CLI command failed: logseq ${args.join(" ")}`); } return { ...classification, parsed, result }; } export function cliCommand(bin = null) { const command = bin || "logseq"; if (command.endsWith(".js")) { return { command: process.execPath, args: [command] }; } return { command, args: [] }; } function cliFetchFailed(response) { return ( !response.ok && (String(response.result?.stderr || "").includes("fetch failed") || String(response.result?.stdout || "").includes("fetch failed") || response.parsed?.error?.message === "fetch failed") ); } async function runSyncStart(opts, context = {}) { let response = await runCli(opts, ["sync", "start", "--graph", opts.graph], context); if (cliFetchFailed(response)) { await delay(500); await runCli(opts, ["server", "list"], { ...context, op: "server-list-before-sync-start-retry" }); response = await runCli(opts, ["sync", "start", "--graph", opts.graph], { ...context, retry: "fetch-failed", }); } return response; } async function invokeThreadApi(opts, state, method, args, context = {}) { const startedAt = Date.now(); let response = null; let text = ""; let parsed = null; let error = null; const invokeOnce = async () => { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), opts.timeoutMs + 3000); try { const nextResponse = await fetch(`${state.workerBaseUrl}/v1/invoke`, { method: "POST", headers: { "Content-Type": "application/json", Accept: "application/json", }, body: JSON.stringify({ method, argsTransit: transitWriter.write(args), }), signal: controller.signal, }); const nextText = await nextResponse.text(); let nextParsed = null; if (nextText) { nextParsed = JSON.parse(nextText); } if (nextParsed?.resultTransit) { nextParsed.result = transitReader.read(nextParsed.resultTransit); } return { response: nextResponse, text: nextText, parsed: nextParsed, error: null }; } catch (caught) { return { response: null, text: "", parsed: null, error: caught }; } finally { clearTimeout(timer); } }; ({ response, text, parsed, error } = await invokeOnce()); if (error && String(error instanceof Error ? error.message : error).includes("fetch failed")) { const refreshed = await refreshWorkerBaseUrl(opts, state, context); if (refreshed) { ({ response, text, parsed, error } = await invokeOnce()); } } const classification = classifyHttpResult({ ok: !error && response?.ok, status: response?.status }, parsed, context); if (classification.expectedRace) { for (const id of staleIdsFromContext(context)) { state.activeIds.delete(id); state.activeBlockUuids.delete(id); state.taskIds.delete(id); } } logEvent(opts, { event: "http", level: classification.level, outcome: classification.outcome, context, method, elapsedMs: Date.now() - startedAt, status: response?.status, error: error ? { message: error instanceof Error ? error.message : String(error) } : parsed?.error, stdout: classification.ok ? undefined : text, }); if (!classification.ok && !classification.expectedRace && opts.failFast) { throw new Error(`HTTP invoke failed: ${method}`); } return { ...classification, parsed, text, error }; } async function refreshWorkerBaseUrl(opts, state, context = {}) { const serverList = await runCli(opts, ["server", "list"], { ...context, op: "refresh-worker-base-url", client: opts.clientName, }); const servers = serverList.parsed?.data?.servers || []; let graphServer = servers.find((server) => server.graph === opts.graph); if (!graphServer) { graphServer = await startDbWorkerNode(opts); } const baseUrl = graphServer?.["base-url"]; if (!baseUrl) { return false; } if (baseUrl !== state.workerBaseUrl) { logEvent(opts, { event: "db-worker-node", level: "warn", status: "base-url-refreshed", client: opts.clientName, previousBaseUrl: state.workerBaseUrl, baseUrl, }); } state.workerBaseUrl = baseUrl; state.repo = graphServerRepo(opts, graphServer); return true; } async function applyOutlinerOps(opts, state, ops, options, context = {}) { return invokeThreadApi( opts, state, "thread-api/apply-outliner-ops", [state.repo, ops, options || transit.map()], context, ); } async function transactRaw(opts, state, txData, txMeta, context = {}) { return invokeThreadApi( opts, state, "thread-api/transact", [state.repo, txData, txMeta || tmap(), null], context, ); } function resultIds(response) { const result = response.parsed?.data?.result; if (Array.isArray(result)) { return result.filter((value) => Number.isInteger(value)); } if (Number.isInteger(result)) { return [result]; } return []; } async function fetchJson(url, timeoutMs) { const controller = new AbortController(); const timer = setTimeout(() => controller.abort(), timeoutMs); try { const response = await fetch(url, { signal: controller.signal }); return { ok: response.ok, status: response.status, text: await response.text() }; } finally { clearTimeout(timer); } } function syncBaseUrl(opts) { return new URL(opts.syncBase.replace(/\/$/, "")); } export function syncServerStartArgs(opts) { const base = syncBaseUrl(opts); return [ "cli-e2e/scripts/db_sync_server.py", "start", "--repo-root", repoRoot, "--pid-file", opts.syncServerPidFile, "--log-file", opts.syncServerLogFile, "--data-dir", opts.syncServerDataDir, "--host", base.hostname, "--port", base.port || (base.protocol === "https:" ? "443" : "80"), "--startup-timeout-s", "25", "--auth-path", opts.authPath, ]; } export function syncUploadArgs(opts) { const args = ["sync", "upload", "--graph", opts.graph]; if (opts.graphE2ee !== false) { args.push("--e2ee-password", opts.e2eePassword); } return args; } export function syncDownloadArgs(opts) { const args = ["sync", "download", "--graph", opts.graph]; if (opts.graphE2ee !== false) { args.push("--e2ee-password", opts.e2eePassword); } return args; } export function syncEnsureKeysArgs(opts) { return ["sync", "ensure-keys", "--upload-keys", "--e2ee-password", opts.e2eePassword]; } export function syncNeedsEnsureKeys(opts) { return opts.graphE2ee !== false; } export function graphValidateArgs(opts) { return ["graph", "validate", "--graph", opts.graph]; } export function syncStatusUninitialized(response) { const status = response.parsed?.data; return response.ok && status && status["graph-id"] == null; } function numberStatusValue(value) { return Number.isInteger(value) ? value : Number.parseInt(String(value), 10); } export function isSettledSyncStatus(response) { const status = response.parsed?.data; if (!response.ok || !status) { return false; } const pendingLocal = numberStatusValue(status["pending-local"]); const pendingServer = numberStatusValue(status["pending-server"]); const localChecksum = status["local-checksum"]; const remoteChecksum = status["remote-checksum"]; return ( pendingLocal === 0 && pendingServer === 0 && typeof localChecksum === "string" && localChecksum.length > 0 && localChecksum === remoteChecksum ); } export function isIdleSyncStatus(response) { const status = response.parsed?.data; if (!response.ok || !status || status["last-error"] != null) { return false; } const pendingLocal = numberStatusValue(status["pending-local"]); const pendingServer = numberStatusValue(status["pending-server"]); const localTx = numberStatusValue(status["local-tx"]); const remoteTx = numberStatusValue(status["remote-tx"]); return pendingLocal === 0 && pendingServer === 0 && Number.isInteger(localTx) && localTx === remoteTx; } function transitValueGet(value, key) { if (value && typeof value.get === "function") { return value.get(key) ?? value.get(kw(key)) ?? value.get(`:${key}`); } if (value && typeof value === "object") { return value[key] ?? value[`:${key}`]; } return undefined; } export function checksumDiagnosticsMatch(result) { const recomputed = transitValueGet(result, "recomputed-checksum"); const remote = transitValueGet(result, "remote-checksum"); return typeof recomputed === "string" && recomputed.length > 0 && recomputed === remote; } export function clientRuntimeOptions(opts, clientIndex) { const clientNumber = clientIndex + 1; const clientName = `client-${clientNumber}`; const baseDir = resolve(dirname(opts.config), "clients", clientName); const rootDir = clientIndex === 0 ? opts.rootDir || process.env.LOGSEQ_CLI_ROOT_DIR : join(baseDir, "root"); const homeDir = clientIndex === 0 ? opts.homeDir || process.env.HOME : join(baseDir, "home"); const config = clientIndex === 0 ? opts.config : join(baseDir, "cli.edn"); const env = { ...(opts.env || {}) }; if (rootDir) { env.LOGSEQ_CLI_ROOT_DIR = rootDir; } if (homeDir) { env.HOME = homeDir; } return { ...opts, clientIndex, clientName, config, rootDir, homeDir, env, }; } async function ensureLocalSyncServer(opts) { const healthUrl = `${opts.httpBase}/health`; const health = await fetchJson(healthUrl, 2500).catch((error) => ({ ok: false, status: "fetch-error", text: String(error), })); if (health.ok) { logEvent(opts, { event: "db-sync-server", level: "info", status: "healthy", baseUrl: opts.httpBase }); return; } if (!opts.startSyncServer) { throw new Error(`Local db-sync server is not healthy at ${healthUrl}; start it or remove --no-start-sync-server`); } logEvent(opts, { event: "db-sync-server", level: "info", status: "starting", baseUrl: opts.httpBase }); const result = await runProcess("python3", syncServerStartArgs(opts), opts); let parsed = null; if (result.stdout) { try { parsed = JSON.parse(result.stdout); } catch { parsed = null; } } if (result.code !== 0 || parsed?.status !== "ok") { throw new Error( [ `Failed to start local db-sync server at ${opts.httpBase}.`, result.stderr || "", result.stdout || "", ].filter(Boolean).join("\n"), ); } logEvent(opts, { event: "db-sync-server", level: "info", status: "started", pid: parsed.pid, baseUrl: parsed.base_url, logFile: parsed.log_file, }); } function graphServerRepo(opts, graphServer) { return graphServer.repo || `logseq_db_${opts.graph}`; } async function startDbWorkerNode(opts) { await runCli(opts, ["server", "start", "--graph", opts.graph], { op: "server-start", client: opts.clientName }); const serverList = await runCli(opts, ["server", "list"], { op: "server-list", client: opts.clientName }); const servers = serverList.parsed?.data?.servers || []; const graphServer = servers.find((server) => server.graph === opts.graph); if (!graphServer || !String(graphServer["base-url"] || "").startsWith("http://127.0.0.1:")) { throw new Error(`No local db-worker-node server found for graph ${opts.graph} on ${opts.clientName || "client-1"}`); } logEvent(opts, { event: "db-worker-node", level: "info", client: opts.clientName, server: graphServer }); return graphServer; } async function setLocalGraphE2ee(opts, graphServer, graphE2ee) { const state = { workerBaseUrl: graphServer["base-url"], repo: graphServerRepo(opts, graphServer), activeIds: new Set(), activeBlockUuids: new Map(), taskIds: new Set(), }; const response = await invokeThreadApi( opts, state, "thread-api/transact", [ state.repo, [ tmap([ kw("db/ident"), kw("logseq.kv/graph-rtc-e2ee?"), kw("kv/value"), graphE2ee, ]), ], tmap([kw("outliner-op"), kw("set-kvs")]), null, ], { op: "set-local-graph-e2ee", graphE2ee }, ); if (!response.ok) { throw new Error(`Failed to set local graph E2EE metadata for ${opts.graph}`); } } async function ensureLocalServers(opts) { const graphServer = await startDbWorkerNode(opts); if (opts.sync) { await ensureLocalSyncServer(opts); const status = await runCli(opts, ["sync", "status", "--graph", opts.graph], { op: "sync-status-before-start" }); if (syncStatusUninitialized(status)) { if (opts.graphE2ee === false) { await setLocalGraphE2ee(opts, graphServer, false); } if (syncNeedsEnsureKeys(opts)) { const keys = await runCli(opts, syncEnsureKeysArgs(opts), { op: "sync-ensure-keys" }); if (!keys.ok) { throw new Error(`Failed to initialize local sync keys for ${opts.graph}`); } } const upload = await runCli(opts, syncUploadArgs(opts), { op: "sync-upload-initialize" }); if (!upload.ok) { throw new Error(`Failed to initialize local sync upload for ${opts.graph}`); } } const start = await runSyncStart(opts, { op: "sync-start" }); if (!start.ok) { throw new Error(`Failed to start local sync for ${opts.graph}`); } } return graphServer; } async function ensureReplicaLocalServers(opts) { await ensureLocalSyncServer(opts); const download = await runCli(opts, syncDownloadArgs(opts), { op: "sync-download-replica", client: opts.clientName }); if (!download.ok) { throw new Error(`Failed to download remote sync graph ${opts.graph} for ${opts.clientName}`); } const graphServer = await startDbWorkerNode(opts); const start = await runSyncStart(opts, { op: "sync-start", client: opts.clientName, }); if (!start.ok) { throw new Error(`Failed to start local sync for ${opts.graph} on ${opts.clientName}`); } return graphServer; } function choose(set) { const values = Array.from(set); if (values.length === 0) { return null; } return values[Math.floor(Math.random() * values.length)]; } function chooseArray(values) { if (!Array.isArray(values) || values.length === 0) { return null; } return values[Math.floor(Math.random() * values.length)]; } function chooseMany(set, maxCount) { const values = Array.from(set); values.sort(() => Math.random() - 0.5); return values.slice(0, Math.max(1, Math.min(maxCount, values.length))); } function ednString(value) { return JSON.stringify(String(value)); } function ordinalDay(day) { if (day % 100 >= 11 && day % 100 <= 13) { return `${day}th`; } switch (day % 10) { case 1: return `${day}st`; case 2: return `${day}nd`; case 3: return `${day}rd`; default: return `${day}th`; } } function journalPageName(date) { return `${date.toLocaleString("en-US", { month: "short", timeZone: "UTC" })} ${ordinalDay(date.getUTCDate())}, ${date.getUTCFullYear()}`; } function journalPageNameForYmd(ymd) { const date = new Date(`${ymd}T00:00:00.000Z`); if (Number.isNaN(date.getTime())) { throw new Error(`Invalid journal date: ${ymd}`); } return journalPageName(date); } function localYmd(date) { const year = date.getFullYear(); const month = String(date.getMonth() + 1).padStart(2, "0"); const day = String(date.getDate()).padStart(2, "0"); return `${year}-${month}-${day}`; } export function todayJournalPageName(opts = {}, now = new Date()) { return journalPageNameForYmd(opts.todayDate || localYmd(now)); } export function stressPageNames(opts) { const pages = [opts.page]; for (let index = 0; index < opts.extraPages; index += 1) { pages.push(`${opts.page} page ${index + 2}`); } const start = new Date(`${opts.journalStart}T00:00:00.000Z`); if (Number.isNaN(start.getTime())) { throw new Error(`Invalid --journal-start date: ${opts.journalStart}`); } for (let index = 0; index < opts.journalPages; index += 1) { const date = new Date(start.getTime() + index * 24 * 60 * 60 * 1000); pages.push(journalPageName(date)); } return pages; } export function bootstrapStressPageNames(opts) { const pages = [opts.page]; for (let index = 0; index < opts.extraPages; index += 1) { pages.push(`${opts.page} page ${index + 2}`); } return pages; } export function mutableStressPageNames(opts) { const pages = []; for (let index = 0; index < opts.extraPages; index += 1) { pages.push(`${opts.page} page ${index + 2}`); } return pages.length > 0 ? pages : [opts.page]; } export function operationWeights(opts = {}) { const operations = [ ["create-root", 18], ["create-child", 18], ["create-tree", 10], ["update", 8], ["move-child", 14], ["move-page", 12], ["move-sibling", 12], ["block-tags-add", 3], ["block-tags-remove", 2], ["block-properties-add", 3], ["block-properties-remove", 2], ["task-create", 3], ["task-update", 2], ["page-upsert", 2], ["page-properties-add", 1], ["page-properties-remove", 1], ["page-delete-restore", 6], ["tag-upsert", 1], ["tag-delete-recreate", 1], ["property-upsert", 1], ["property-delete-recreate", 1], ["http-insert-blocks", 12], ["http-delete-blocks", 12], ["http-move-blocks", 12], ["http-save-block", 4], ["http-move-up-down", 8], ["http-indent-outdent", 8], ["http-apply-template", 1], ["http-create-page", 1], ["http-rename-page", 1], ["http-delete-restore-page", 8], ["http-recycle-delete-page", 4], ["http-upsert-property", 1], ["http-set-block-property", 2], ["http-remove-block-property", 1], ["http-batch-set-property", 2], ["http-batch-remove-property", 1], ["http-batch-delete-property-value", 1], ["http-create-property-history", 2], ["http-create-linked-references-view", 2], ["http-create-unlinked-references-view", 2], ["http-toggle-reaction", 1], ["http-delete-view-target-with-related-entities", 4], ["http-insert-undo-redo", 12], ["delete-single", 18], ["delete-batch", 12], ["delete-update-race", 8], ]; if (opts.sync && opts.offline) { operations.push(["offline-window", 2]); operations.push(["offline-today-journal-race", 8]); if ((opts.clients || 1) > 1) { operations.push(["multi-client-offline-rebase", 12]); operations.push(["multi-client-today-journal-race", 10]); } } return operations; } export function operationNames(opts = {}) { return operationWeights(opts).map(([name]) => name); } export function uniqueOperationPageTitle(prefix, state, context = {}) { const workerId = context.workerId ?? "unknown-worker"; const seq = context.seq ?? "unknown-seq"; return `${prefix} ${state.runId} worker-${workerId} seq-${seq} ${crypto.randomUUID()}`; } export function offlineTodayJournalClientCount(opts = {}) { return Math.max(2, opts.clients ?? 1); } function liveBlocksQuery(page) { return `[:find [(pull ?b [:db/id :block/uuid]) ...] :where [?p :block/name ${ednString(page.toLowerCase())}] [?b :block/page ?p] (not [?b :block/name])]`; } export function blockByUuidQuery(uuidValue) { return `[:find ?b . :where [?b :block/uuid #uuid ${ednString(uuidValue)}]]`; } function pageByNameQuery(page) { return `[:find (pull ?p [:db/id :block/uuid :block/name :block/title]) . :where [?p :block/name ${ednString(page.toLowerCase())}]]`; } function normalizedUuidString(value) { if (typeof value === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value)) { return value; } if (value && typeof value === "object" && typeof value.value === "string") { return normalizedUuidString(value.value); } return null; } function pageUuidFromQueryResult(response) { return normalizedUuidString(response.parsed?.data?.result?.["block/uuid"]); } export function blockExistsFromQueryResult(response) { return Number.isInteger(response.parsed?.data?.result); } function liveBlocksFromQueryResult(response) { const rows = response.parsed?.data?.result; if (!Array.isArray(rows)) { return []; } return rows .map((row) => { const id = row?.["db/id"]; const uuid = normalizedUuidString(row?.["block/uuid"]); return Number.isInteger(id) && uuid ? { id, uuid } : null; }) .filter(Boolean); } async function findPageUuid(opts, page, context = {}) { const response = await runCli( opts, ["query", "--graph", opts.graph, "--query", pageByNameQuery(page)], { op: "find-page-uuid", page, ...context }, ); return pageUuidFromQueryResult(response); } async function refreshLiveIds(opts, state, reason) { const ids = new Set(); const blockUuids = new Map(); for (const page of state.pageNames) { const response = await runCli( opts, ["query", "--graph", opts.graph, "--query", liveBlocksQuery(page)], { op: "refresh-live-ids", reason, page }, ); for (const block of liveBlocksFromQueryResult(response)) { ids.add(block.id); blockUuids.set(block.id, block.uuid); } } state.activeIds = ids; state.activeBlockUuids = blockUuids; state.lastRefreshSeq = state.nextSeq; logEvent(opts, { event: "refresh-live-ids", level: "info", reason, activeIds: state.activeIds.size, }); } async function refreshLiveIdsIfNeeded(opts, state, reason) { if (state.refreshing) { return; } if (state.activeIds.size >= opts.concurrency * 2 && state.nextSeq - state.lastRefreshSeq < 50) { return; } state.refreshing = true; try { await refreshLiveIds(opts, state, reason); } finally { state.refreshing = false; } } function pruneStaleIds(state, response, context) { if (!response.expectedRace) { return; } for (const id of staleIdsFromContext(context)) { state.activeIds.delete(id); } } function weightedOperation(opts) { const operations = operationWeights(opts); const total = operations.reduce((sum, [, weight]) => sum + weight, 0); let cursor = Math.random() * total; for (const [name, weight] of operations) { cursor -= weight; if (cursor <= 0) { return name; } } return "create-root"; } async function createBlock(opts, state, content, targetArgs, context) { const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, ...targetArgs, "--content", content], context, ); pruneStaleIds(state, response, context); for (const id of resultIds(response)) { state.activeIds.add(id); } } function randomPage(state) { return chooseArray(state.pageNames) || state.pageNames[0]; } function mutablePage(state) { return chooseArray(state.mutablePageNames) || state.pageNames[0]; } function stressTag(state) { return chooseArray(state.stableTags); } function stressProperty(state) { return chooseArray(state.stableProperties); } function stressTagId(state) { return chooseArray(Array.from(state.stableTagIds.values())); } function rawPropertyIdent(state) { return state.rawPropertyIdent || "user.property/cli-http-prop"; } function propertyMap(name, value) { return `{${ednString(name)} ${ednString(value)}}`; } function propertyVector(name) { return `[${ednString(name)}]`; } function kw(name) { return transit.keyword(name); } function uuid(value) { return transit.uuid(value); } function tmap(entries = []) { return transit.map(entries); } function localTxMeta(entries = []) { return tmap([kw("db-sync/tx-id"), uuid(crypto.randomUUID()), kw("local-tx?"), true, ...entries]); } function blockMap(entries) { return tmap(entries.flatMap(([key, value]) => [kw(key), value])); } function lookupRef(attr, value) { return [kw(attr), value]; } function blockUuidRef(uuidValue) { return lookupRef("block/uuid", uuid(uuidValue)); } function chooseActiveUuid(state) { const id = choose(state.activeIds); if (!id) { return null; } const uuidValue = state.activeBlockUuids.get(id); return uuidValue ? { id, uuid: uuidValue } : null; } async function blockStillExists(opts, block, context = {}) { const response = await runCli( opts, ["query", "--graph", opts.graph, "--query", blockByUuidQuery(block.uuid)], { ...context, op: "check-live-block", sourceOp: context.op, id: block.id, uuid: block.uuid }, ); return blockExistsFromQueryResult(response); } async function chooseLiveActiveUuid(opts, state, context = {}) { const candidates = Array.from(state.activeIds) .map((id) => { const uuidValue = state.activeBlockUuids.get(id); return uuidValue ? { id, uuid: uuidValue } : null; }) .filter(Boolean); candidates.sort(() => Math.random() - 0.5); for (const block of candidates.slice(0, 4)) { if (await blockStillExists(opts, block, context)) { return block; } state.activeIds.delete(block.id); state.activeBlockUuids.delete(block.id); logEvent(opts, { event: "skip-stale-raw-target", level: "info", context, id: block.id, uuid: block.uuid, }); } return null; } function chooseDistinctActiveUuids(state) { const first = chooseActiveUuid(state); if (!first) { return null; } const candidates = Array.from(state.activeBlockUuids.entries()) .filter(([id]) => id !== first.id) .map(([id, uuidValue]) => ({ id, uuid: uuidValue })); const second = chooseArray(candidates); return second ? [first, second] : null; } async function ensureStressMetadata(opts, state) { for (const page of state.bootstrapPageNames) { await runCli(opts, ["upsert", "page", "--graph", opts.graph, "--page", page], { op: "bootstrap-page", page, }); } for (const name of [...state.stableTags, state.volatileTag]) { await runCli(opts, ["upsert", "tag", "--graph", opts.graph, "--name", name], { op: "bootstrap-tag", name, }); } for (const name of [...state.stableProperties, state.volatileProperty]) { await runCli(opts, ["upsert", "property", "--graph", opts.graph, "--name", name, "--type", "default", "--cardinality", "one", "--public", "true"], { op: "bootstrap-property", name, }); } await refreshStressTagIds(opts, state, "bootstrap"); await applyRawUpsertProperty(opts, state, { op: "bootstrap-http-upsert-property" }); } async function refreshStressTagIds(opts, state, reason) { const response = await runCli(opts, ["list", "tag", "--graph", opts.graph], { op: "refresh-stress-tags", reason, }); const items = response.parsed?.data?.items || []; state.stableTagIds = new Map(); for (const name of state.stableTags) { const item = items.find((candidate) => candidate?.["block/title"] === name); const id = item?.["db/id"]; if (Number.isInteger(id)) { state.stableTagIds.set(name, id); } } } async function applyRawSaveBlock(opts, state, context, title) { const block = chooseActiveUuid(state); if (!block) { return; } const response = await applyOutlinerOps( opts, state, [ [ kw("save-block"), [ blockMap([ ["block/uuid", uuid(block.uuid)], ["block/title", title], ]), tmap(), ], ], ], tmap(), { ...context, id: block.id, uuid: block.uuid }, ); if (!response.ok) { state.activeIds.delete(block.id); state.activeBlockUuids.delete(block.id); } } async function applyRawInsertBlocks(opts, state, context, title) { const target = chooseActiveUuid(state); if (!target) { return; } const newUuid = crypto.randomUUID(); const block = blockMap([ ["block/uuid", uuid(newUuid)], ["block/title", title], ]); await applyOutlinerOps( opts, state, [[kw("insert-blocks"), [[block], uuid(target.uuid), tmap([kw("sibling?"), false, kw("keep-uuid?"), true])]]], tmap(), { ...context, targetId: target.id, targetUuid: target.uuid, uuid: newUuid }, ); await refreshLiveIds(opts, state, "http-insert-blocks"); } async function applyRawDeleteBlocks(opts, state, context) { const block = chooseActiveUuid(state); if (!block) { return; } await applyOutlinerOps( opts, state, [[kw("delete-blocks"), [[uuid(block.uuid)], tmap()]]], tmap(), { ...context, id: block.id, uuid: block.uuid }, ); state.activeIds.delete(block.id); state.activeBlockUuids.delete(block.id); } async function applyRawMoveBlocks(opts, state, context) { const pair = chooseDistinctActiveUuids(state); if (!pair) { return; } const [block, target] = pair; await applyOutlinerOps( opts, state, [[kw("move-blocks"), [[uuid(block.uuid)], uuid(target.uuid), tmap([kw("sibling?"), Math.random() < 0.5])]]], tmap(), { ...context, id: block.id, uuid: block.uuid, targetId: target.id, targetUuid: target.uuid }, ); } async function applyRawMoveUpDown(opts, state, context, up) { const block = chooseActiveUuid(state); if (!block) { return; } await applyOutlinerOps( opts, state, [[kw("move-blocks-up-down"), [[uuid(block.uuid)], up]]], tmap(), { ...context, id: block.id, uuid: block.uuid, up }, ); } async function applyRawIndentOutdent(opts, state, context, indent) { const block = chooseActiveUuid(state); if (!block) { return; } await applyOutlinerOps( opts, state, [[kw("indent-outdent-blocks"), [[uuid(block.uuid)], indent, tmap()]]], tmap(), { ...context, id: block.id, uuid: block.uuid, indent }, ); } async function applyRawApplyTemplate(opts, state, context, title) { const target = chooseActiveUuid(state); if (!target) { return; } const templateUuid = crypto.randomUUID(); const insertedUuid = crypto.randomUUID(); const templateBlock = blockMap([ ["block/uuid", uuid(insertedUuid)], ["block/title", title], ]); await applyOutlinerOps( opts, state, [ [ kw("apply-template"), [ uuid(templateUuid), uuid(target.uuid), tmap([kw("template-blocks"), [templateBlock], kw("sibling?"), false]), ], ], ], tmap(), { ...context, targetId: target.id, targetUuid: target.uuid, templateUuid, uuid: insertedUuid }, ); await refreshLiveIds(opts, state, "http-apply-template"); } async function applyRawCreatePage(opts, state, context, title) { await applyOutlinerOps( opts, state, [[kw("create-page"), [title, tmap()]]], tmap(), { ...context, page: title }, ); state.pageNames.push(title); state.mutablePageNames.push(title); } async function applyRawRenamePage(opts, state, context, title) { const page = mutablePage(state); const pageUuid = await findPageUuid(opts, page, context); if (!pageUuid) { return; } const temporaryTitle = `${title} ${Date.now()}`; await applyOutlinerOps( opts, state, [[kw("rename-page"), [uuid(pageUuid), temporaryTitle]]], tmap(), { ...context, page, pageUuid, phase: "rename-away", temporaryTitle }, ); await applyOutlinerOps( opts, state, [[kw("rename-page"), [uuid(pageUuid), page]]], tmap(), { ...context, page, pageUuid, phase: "rename-back", temporaryTitle }, ); } async function applyRawDeleteRestorePage(opts, state, context, title) { const page = title || uniqueOperationPageTitle("CLI HTTP Delete Restore", state, context); await applyOutlinerOps( opts, state, [[kw("create-page"), [page, tmap()]]], tmap(), { ...context, page, phase: "create", requireTruthyResult: true }, ); const pageUuid = await findPageUuid(opts, page, context); if (!pageUuid) { return; } await applyOutlinerOps( opts, state, [[kw("delete-page"), [uuid(pageUuid), tmap()]]], tmap(), { ...context, page, pageUuid, phase: "delete", requireTruthyResult: true }, ); await applyOutlinerOps( opts, state, [[kw("restore-recycled"), [uuid(pageUuid)]]], tmap(), { ...context, page, pageUuid, phase: "restore", requireTruthyResult: true }, ); await refreshLiveIds(opts, state, "http-delete-restore-page"); } async function applyRawRecycleDeletePage(opts, state, context, title) { await applyOutlinerOps( opts, state, [[kw("create-page"), [title, tmap()]]], tmap(), { ...context, page: title, phase: "create", requireTruthyResult: true }, ); const pageUuid = await findPageUuid(opts, title, context); if (!pageUuid) { return; } await applyOutlinerOps( opts, state, [[kw("delete-page"), [uuid(pageUuid), tmap()]]], tmap(), { ...context, page: title, pageUuid, phase: "delete", requireTruthyResult: true }, ); await applyOutlinerOps( opts, state, [[kw("recycle-delete-permanently"), [uuid(pageUuid)]]], tmap(), { ...context, page: title, pageUuid, phase: "permanent-delete", requireTruthyResult: true }, ); } async function applyRawUpsertProperty(opts, state, context) { await applyOutlinerOps( opts, state, [ [ kw("upsert-property"), [ kw(rawPropertyIdent(state)), tmap([kw("logseq.property/type"), kw("default"), kw("db/cardinality"), kw("db.cardinality/one")]), tmap([kw("property-name"), state.rawPropertyName || "cli-http-prop"]), ], ], ], tmap(), context, ); } async function applyRawSetBlockProperty(opts, state, context, value) { const block = chooseActiveUuid(state); if (!block) { return; } await applyOutlinerOps( opts, state, [[kw("set-block-property"), [uuid(block.uuid), kw(rawPropertyIdent(state)), value]]], tmap(), { ...context, id: block.id, uuid: block.uuid }, ); } async function applyRawRemoveBlockProperty(opts, state, context) { const block = chooseActiveUuid(state); if (!block) { return; } await applyOutlinerOps( opts, state, [[kw("remove-block-property"), [uuid(block.uuid), kw(rawPropertyIdent(state))]]], tmap(), { ...context, id: block.id, uuid: block.uuid }, ); } async function applyRawBatchSetProperty(opts, state, context, value) { const blocks = chooseMany(state.activeIds, 3) .map((id) => ({ id, uuid: state.activeBlockUuids.get(id) })) .filter((block) => block.uuid); if (blocks.length === 0) { return; } await applyOutlinerOps( opts, state, [[kw("batch-set-property"), [blocks.map((block) => uuid(block.uuid)), kw(rawPropertyIdent(state)), value, tmap()]]], tmap(), { ...context, ids: blocks.map((block) => block.id) }, ); } async function applyRawBatchRemoveProperty(opts, state, context) { const blocks = chooseMany(state.activeIds, 3) .map((id) => ({ id, uuid: state.activeBlockUuids.get(id) })) .filter((block) => block.uuid); if (blocks.length === 0) { return; } await applyOutlinerOps( opts, state, [[kw("batch-remove-property"), [blocks.map((block) => uuid(block.uuid)), kw(rawPropertyIdent(state))]]], tmap(), { ...context, ids: blocks.map((block) => block.id) }, ); } async function applyRawBatchDeletePropertyValue(opts, state, context) { const block = chooseActiveUuid(state); const tagId = stressTagId(state); if (!block || !tagId) { return; } await applyOutlinerOps( opts, state, [[kw("batch-set-property"), [[uuid(block.uuid)], kw("block/tags"), tagId, tmap()]]], tmap(), { ...context, id: block.id, uuid: block.uuid, tagId, phase: "set-tag" }, ); await applyOutlinerOps( opts, state, [[kw("batch-delete-property-value"), [[uuid(block.uuid)], kw("block/tags"), tagId]]], tmap(), { ...context, id: block.id, uuid: block.uuid, tagId, phase: "delete-tag-value" }, ); } function referenceViewBlock(viewUuid, viewsPageUuid, target, featureType, title, order, now) { return blockMap([ ["block/uuid", uuid(viewUuid)], ["block/title", title], ["block/page", blockUuidRef(viewsPageUuid)], ["block/parent", blockUuidRef(viewsPageUuid)], ["block/order", order], ["block/created-at", now], ["block/updated-at", now], ["block/refs", blockUuidRef(target.uuid)], ["logseq.property/view-for", blockUuidRef(target.uuid)], ["logseq.property.view/type", kw("logseq.property.view/type.list")], ["logseq.property.view/group-by-property", kw("block/page")], ["logseq.property.view/feature-type", kw(featureType)], ]); } function propertyHistoryBlock(historyUuid, targetRef, statusIdent, now) { return blockMap([ ["block/uuid", uuid(historyUuid)], ["block/created-at", now], ["block/updated-at", now], ["logseq.property.history/block", targetRef], ["logseq.property.history/property", kw("logseq.property/status")], ["logseq.property.history/ref-value", kw(statusIdent)], ]); } function statusHistoryValue(context, offset = 0) { const statuses = ["logseq.property/status.todo", "logseq.property/status.doing", "logseq.property/status.done"]; const seq = Number.isInteger(context.seq) ? context.seq : 0; return statuses[(seq + offset) % statuses.length]; } function reactionEmojiId(context) { const emojiIds = ["+1", "eyes", "tada"]; const seq = Number.isInteger(context.seq) ? context.seq : 0; return emojiIds[seq % emojiIds.length]; } export function stressReferenceViewOrder(context = {}, offset = 0) { const base62Digits = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"; const seq = Number.isInteger(context.seq) ? context.seq : 0; const index = Math.abs(seq * 2 + offset); const digit = base62Digits[index % base62Digits.length]; if (index < base62Digits.length) { return `a${digit}`; } const bucket = base62Digits[Math.floor((index - base62Digits.length) / base62Digits.length)]; return `b${bucket}${digit}`; } async function findViewsPageUuid(opts, context = {}) { return findPageUuid(opts, "$$$views", context); } async function createReferenceView(opts, state, context, featureType, title) { const target = await chooseLiveActiveUuid(opts, state, context); if (!target) { return null; } const viewsPageUuid = await findViewsPageUuid(opts, context); if (!viewsPageUuid) { return null; } const now = Date.now(); const viewUuid = crypto.randomUUID(); await transactRaw( opts, state, [referenceViewBlock(viewUuid, viewsPageUuid, target, featureType, title, stressReferenceViewOrder(context), now)], localTxMeta([kw("outliner-op"), kw("create-view")]), { ...context, id: target.id, uuid: target.uuid, viewUuid, featureType }, ); return { target, viewUuid }; } async function applyRawCreatePropertyHistory(opts, state, context) { const target = await chooseLiveActiveUuid(opts, state, context); if (!target) { return; } const viewsPageUuid = await findViewsPageUuid(opts, context); if (!viewsPageUuid) { return; } const now = Date.now(); const historyUuid = crypto.randomUUID(); await transactRaw( opts, state, [propertyHistoryBlock(historyUuid, blockUuidRef(target.uuid), statusHistoryValue(context), now)], localTxMeta([kw("outliner-op"), kw("record-property-history")]), { ...context, id: target.id, uuid: target.uuid, historyUuid }, ); } async function applyRawCreateReferenceView(opts, state, context, featureType, title) { await createReferenceView(opts, state, context, featureType, title); } async function applyRawToggleReactionForBlock(opts, state, block, context) { await applyOutlinerOps( opts, state, [[kw("toggle-reaction"), [uuid(block.uuid), reactionEmojiId(context), null]]], tmap(), { ...context, id: block.id, uuid: block.uuid }, ); } async function applyRawToggleReaction(opts, state, context) { const block = chooseActiveUuid(state); if (!block) { return; } await applyRawToggleReactionForBlock(opts, state, block, context); } async function applyRawDeleteViewTargetWithRelatedEntities(opts, state, context) { const target = await chooseLiveActiveUuid(opts, state, context); if (!target) { return; } const viewsPageUuid = await findViewsPageUuid(opts, context); if (!viewsPageUuid) { return; } const now = Date.now(); const linkedViewUuid = crypto.randomUUID(); const unlinkedViewUuid = crypto.randomUUID(); const targetHistoryUuid = crypto.randomUUID(); const linkedViewHistoryUuid = crypto.randomUUID(); const unlinkedViewHistoryUuid = crypto.randomUUID(); await transactRaw( opts, state, [ referenceViewBlock( linkedViewUuid, viewsPageUuid, target, "linked-references", "Linked references", stressReferenceViewOrder(context), now, ), referenceViewBlock( unlinkedViewUuid, viewsPageUuid, target, "unlinked-references", "Unlinked references", stressReferenceViewOrder(context, 1), now, ), propertyHistoryBlock(targetHistoryUuid, blockUuidRef(target.uuid), statusHistoryValue(context), now), propertyHistoryBlock(linkedViewHistoryUuid, blockUuidRef(linkedViewUuid), statusHistoryValue(context, 1), now), propertyHistoryBlock(unlinkedViewHistoryUuid, blockUuidRef(unlinkedViewUuid), statusHistoryValue(context, 2), now), ], localTxMeta([kw("outliner-op"), kw("create-view")]), { ...context, id: target.id, uuid: target.uuid, linkedViewUuid, unlinkedViewUuid, targetHistoryUuid, linkedViewHistoryUuid, unlinkedViewHistoryUuid, phase: "create-related-entities", }, ); await applyRawToggleReactionForBlock(opts, state, target, { ...context, phase: "create-reaction" }); await applyOutlinerOps( opts, state, [[kw("delete-blocks"), [[uuid(target.uuid)], tmap()]]], tmap(), { ...context, id: target.id, uuid: target.uuid, phase: "delete-target" }, ); state.activeIds.delete(target.id); state.activeBlockUuids.delete(target.id); } async function applyRawInsertUndoRedo(opts, state, context, title) { const target = chooseActiveUuid(state); if (!target) { return; } const newUuid = crypto.randomUUID(); const block = blockMap([ ["block/uuid", uuid(newUuid)], ["block/title", title], ]); const forward = [[kw("insert-blocks"), [[block], uuid(target.uuid), tmap([kw("sibling?"), false, kw("keep-uuid?"), true])]]]; const inverse = [[kw("delete-blocks"), [[uuid(newUuid)], tmap()]]]; await applyOutlinerOps(opts, state, forward, tmap(), { ...context, phase: "redo-initial", targetId: target.id, targetUuid: target.uuid, uuid: newUuid, }); await applyOutlinerOps(opts, state, inverse, tmap(), { ...context, phase: "undo", targetId: target.id, targetUuid: target.uuid, uuid: newUuid, }); await applyOutlinerOps(opts, state, forward, tmap(), { ...context, phase: "redo", targetId: target.id, targetUuid: target.uuid, uuid: newUuid, }); await refreshLiveIds(opts, state, "http-insert-undo-redo"); } async function offlineWindow(opts, state, context) { if (state.offlineRunning) { return; } state.offlineRunning = true; try { await runCli(opts, ["sync", "stop", "--graph", opts.graph], { ...context, phase: "stop" }); logEvent(opts, { event: "offline-window", level: "warn", status: "offline", context, durationMs: opts.offlineMs, }); await delay(opts.offlineMs); const start = await runSyncStart(opts, { ...context, phase: "start" }); if (!start.ok && opts.failFast) { throw new Error(`Failed to restart sync after offline window for ${opts.graph}`); } } finally { state.offlineRunning = false; } } async function offlineTodayJournalRace(opts, state, context) { if (state.offlineRunning) { return; } state.offlineRunning = true; const page = todayJournalPageName(opts); if (!state.pageNames.includes(page)) { state.pageNames.push(page); } const clientCount = offlineTodayJournalClientCount(opts); try { await runCli(opts, ["sync", "stop", "--graph", opts.graph], { ...context, phase: "stop", page }); logEvent(opts, { event: "offline-today-journal-race", level: "warn", status: "offline", context: { ...context, page, clientCount }, durationMs: opts.offlineMs, }); const responses = await Promise.all( Array.from({ length: clientCount }, (_, index) => { const clientId = index + 1; return runCli( opts, [ "upsert", "block", "--graph", opts.graph, "--target-page", page, "--content", `offline today journal ${state.runId} client=${clientId} seq=${context.seq}`, ], { ...context, phase: "write", page, clientId, clientCount }, ); }), ); for (const response of responses) { for (const id of resultIds(response)) { state.activeIds.add(id); } } await delay(opts.offlineMs); const start = await runSyncStart(opts, { ...context, phase: "start", page }); if (!start.ok && opts.failFast) { throw new Error(`Failed to restart sync after offline today journal race for ${opts.graph}`); } await refreshLiveIds(opts, state, "offline-today-journal-race"); } finally { state.offlineRunning = false; } } async function waitForSyncSettled(opts, context = {}, required = false) { if (!opts.sync) { return true; } let last = null; for (let attempt = 1; attempt <= opts.settleAttempts; attempt += 1) { const status = await runCli(opts, ["sync", "status", "--graph", opts.graph], { ...context, op: context.op || "sync-status-settle", attempt, client: opts.clientName, }); last = status; const settled = required ? isSettledSyncStatus(status) : isIdleSyncStatus(status); if (settled) { return true; } await delay(opts.settleMs); } logEvent(opts, { event: "sync-settle", level: required ? "error" : "warn", context, status: last?.parsed?.data, }); if (required) { throw new Error(`Sync did not settle for ${opts.graph} on ${opts.clientName || "client-1"}`); } return false; } async function validateClientChecksum(client) { if (!client.opts.sync) { return; } const response = await invokeThreadApi( client.opts, client.state, "thread-api/recompute-checksum-diagnostics", [client.state.repo], { op: "final-recompute-checksum", client: client.opts.clientName }, ); const diagnostics = response.parsed?.result; const ok = response.ok && checksumDiagnosticsMatch(diagnostics); logEvent(client.opts, { event: "checksum-diagnostics", level: ok ? "info" : "error", context: { op: "final-recompute-checksum", client: client.opts.clientName }, recomputedChecksum: transitValueGet(diagnostics, "recomputed-checksum"), localChecksum: transitValueGet(diagnostics, "local-checksum"), remoteChecksum: transitValueGet(diagnostics, "remote-checksum"), }); if (!ok) { throw new Error(`Final checksum validation failed for ${client.opts.graph} on ${client.opts.clientName}`); } } const localRebaseOps = [ "create-root", "create-child", "update", "move-child", "move-sibling", "delete-single", "http-insert-blocks", "http-delete-blocks", "http-move-blocks", "http-insert-undo-redo", ]; function chooseDistinctClients(clients) { if (!Array.isArray(clients) || clients.length < 2) { return null; } const first = chooseArray(clients); const candidates = clients.filter((client) => client !== first); return [first, chooseArray(candidates)]; } async function multiClientOfflineRebase(opts, cluster, context) { if (cluster.offlineRunning) { return; } const pair = chooseDistinctClients(cluster.clients); if (!pair) { return; } const [offlineClient, onlineClient] = pair; cluster.offlineRunning = true; try { await runCli(offlineClient.opts, ["sync", "stop", "--graph", offlineClient.opts.graph], { ...context, phase: "stop-offline-client", client: offlineClient.opts.clientName, }); logEvent(opts, { event: "multi-client-offline-rebase", level: "warn", context: { ...context, offlineClient: offlineClient.opts.clientName, onlineClient: onlineClient.opts.clientName, }, durationMs: opts.offlineMs, }); for (let index = 0; index < 3; index += 1) { const op = chooseArray(localRebaseOps); offlineClient.state.nextSeq += 1; await runOperation(offlineClient.opts, offlineClient.state, context.workerId, context.seq * 10 + index, op); } for (let index = 0; index < 3; index += 1) { const op = chooseArray(localRebaseOps); onlineClient.state.nextSeq += 1; await runOperation(onlineClient.opts, onlineClient.state, context.workerId, context.seq * 10 + index + 3, op); } await waitForSyncSettled( onlineClient.opts, { ...context, op: "multi-client-online-settle", onlineClient: onlineClient.opts.clientName }, false, ); await delay(opts.offlineMs); const start = await runSyncStart(offlineClient.opts, { ...context, phase: "start-offline-client", client: offlineClient.opts.clientName, }); if (!start.ok && opts.failFast) { throw new Error(`Failed to restart sync for ${offlineClient.opts.clientName}`); } await waitForSyncSettled( offlineClient.opts, { ...context, op: "multi-client-offline-settle", offlineClient: offlineClient.opts.clientName }, false, ); await Promise.all([ refreshLiveIds(offlineClient.opts, offlineClient.state, "multi-client-offline-rebase"), refreshLiveIds(onlineClient.opts, onlineClient.state, "multi-client-offline-rebase"), ]); } finally { cluster.offlineRunning = false; } } async function multiClientTodayJournalRace(opts, cluster, context) { if (cluster.offlineRunning) { return; } const pair = chooseDistinctClients(cluster.clients); if (!pair) { return; } const [first, second] = pair; const page = todayJournalPageName(opts); cluster.offlineRunning = true; try { for (const client of [first, second]) { if (!client.state.pageNames.includes(page)) { client.state.pageNames.push(page); } await runCli(client.opts, ["sync", "stop", "--graph", client.opts.graph], { ...context, phase: "stop", client: client.opts.clientName, page, }); } await Promise.all( [first, second].map((client, index) => runCli( client.opts, [ "upsert", "block", "--graph", client.opts.graph, "--target-page", page, "--content", `multi-client offline today journal ${client.state.runId} client=${client.opts.clientName} seq=${context.seq}`, ], { ...context, phase: "offline-write", client: client.opts.clientName, page, index }, ), ), ); const firstStart = await runSyncStart(first.opts, { ...context, phase: "start-first", client: first.opts.clientName, page, }); if (!firstStart.ok && opts.failFast) { throw new Error(`Failed to start sync for ${first.opts.clientName}`); } await waitForSyncSettled(first.opts, { ...context, op: "multi-client-today-first-settle", page }, false); await delay(opts.offlineMs); const secondStart = await runSyncStart(second.opts, { ...context, phase: "start-second", client: second.opts.clientName, page, }); if (!secondStart.ok && opts.failFast) { throw new Error(`Failed to start sync for ${second.opts.clientName}`); } await waitForSyncSettled(second.opts, { ...context, op: "multi-client-today-second-settle", page }, false); await Promise.all([ refreshLiveIds(first.opts, first.state, "multi-client-today-journal-race"), refreshLiveIds(second.opts, second.state, "multi-client-today-journal-race"), ]); } finally { cluster.offlineRunning = false; } } async function runOperation(opts, state, workerId, seq, forcedOp = null) { const op = forcedOp || weightedOperation(opts); const stamp = `${state.runId} worker=${workerId} seq=${seq} op=${op}`; await refreshLiveIdsIfNeeded(opts, state, "operation"); if (state.activeIds.size < opts.concurrency && !["create-root", "offline-window", "offline-today-journal-race"].includes(op)) { await createBlock(opts, state, `seed ${stamp}`, ["--target-page", randomPage(state)], { workerId, seq, op: "seed", }); return; } switch (op) { case "create-root": await createBlock(opts, state, `root ${stamp}`, ["--target-page", randomPage(state)], { workerId, seq, op }); return; case "create-child": { const parentId = choose(state.activeIds); await createBlock( opts, state, `child ${stamp} parent=${parentId}`, ["--target-id", String(parentId), "--pos", Math.random() < 0.5 ? "first-child" : "last-child"], { workerId, seq, op, parentId, }, ); return; } case "create-tree": { const page = randomPage(state); const blocks = `[ {:block/title ${ednString(`tree root ${stamp}`)} :block/children [{:block/title ${ednString(`tree child ${stamp}`)}}]} {:block/title ${ednString(`tree sibling ${stamp}`)}}]`; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--target-page", page, "--blocks", blocks], { workerId, seq, op, page }, ); for (const id of resultIds(response)) { state.activeIds.add(id); } return; } case "update": { const id = choose(state.activeIds); const context = { workerId, seq, op, id }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--content", `updated ${stamp} id=${id}`], context, ); pruneStaleIds(state, response, context); return; } case "move-child": { const id = choose(state.activeIds); const targetId = choose(state.activeIds); if (!id || !targetId || id === targetId) { return; } const context = { workerId, seq, op, id, targetId }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--target-id", String(targetId), "--pos", "last-child"], context, ); pruneStaleIds(state, response, context); return; } case "move-page": { const id = choose(state.activeIds); const page = randomPage(state); if (!id) { return; } const context = { workerId, seq, op, id, page }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--target-page", page], context, ); pruneStaleIds(state, response, context); return; } case "move-sibling": { const id = choose(state.activeIds); const targetId = choose(state.activeIds); if (!id || !targetId || id === targetId) { return; } const context = { workerId, seq, op, id, targetId }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--target-id", String(targetId), "--pos", "sibling"], context, ); pruneStaleIds(state, response, context); return; } case "block-tags-add": { const id = choose(state.activeIds); const tag = stressTag(state); if (!id || !tag) { return; } const context = { workerId, seq, op, id, tag }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--update-tags", `[${ednString(tag)}]`], context, ); pruneStaleIds(state, response, context); return; } case "block-tags-remove": { const id = choose(state.activeIds); const tag = stressTag(state); if (!id || !tag) { return; } const context = { workerId, seq, op, id, tag }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--remove-tags", `[${ednString(tag)}]`], context, ); pruneStaleIds(state, response, context); return; } case "block-properties-add": { const id = choose(state.activeIds); const property = stressProperty(state); if (!id || !property) { return; } const context = { workerId, seq, op, id, property }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--update-properties", propertyMap(property, `value ${stamp}`)], context, ); pruneStaleIds(state, response, context); return; } case "block-properties-remove": { const id = choose(state.activeIds); const property = stressProperty(state); if (!id || !property) { return; } const context = { workerId, seq, op, id, property }; const response = await runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--remove-properties", propertyVector(property)], context, ); pruneStaleIds(state, response, context); return; } case "task-create": { const page = randomPage(state); const response = await runCli( opts, [ "upsert", "task", "--graph", opts.graph, "--target-page", page, "--content", `task ${stamp}`, "--status", "todo", "--priority", "high", ], { workerId, seq, op, page }, ); for (const id of resultIds(response)) { state.activeIds.add(id); state.taskIds.add(id); } return; } case "task-update": { const id = choose(state.taskIds.size > 0 ? state.taskIds : state.activeIds); if (!id) { return; } const context = { workerId, seq, op, id }; const response = await runCli( opts, ["upsert", "task", "--graph", opts.graph, "--id", String(id), "--status", "doing", "--priority", "low"], context, ); pruneStaleIds(state, response, context); return; } case "page-upsert": { await runCli(opts, ["upsert", "page", "--graph", opts.graph, "--page", randomPage(state)], { workerId, seq, op, }); return; } case "page-properties-add": { const page = randomPage(state); const property = stressProperty(state); await runCli( opts, ["upsert", "page", "--graph", opts.graph, "--page", page, "--update-properties", propertyMap(property, `page ${stamp}`)], { workerId, seq, op, page, property }, ); return; } case "page-properties-remove": { const page = randomPage(state); const property = stressProperty(state); await runCli( opts, ["upsert", "page", "--graph", opts.graph, "--page", page, "--remove-properties", propertyVector(property)], { workerId, seq, op, page, property }, ); return; } case "page-delete-restore": { const page = mutablePage(state); await runCli(opts, ["remove", "page", "--graph", opts.graph, "--page", page], { workerId, seq, op, page, phase: "delete", }); await runCli(opts, ["upsert", "page", "--graph", opts.graph, "--page", page, "--restore"], { workerId, seq, op, page, phase: "restore", }); await refreshLiveIds(opts, state, "page-delete-restore"); return; } case "tag-upsert": { const tag = chooseArray([...state.stableTags, state.volatileTag]); await runCli(opts, ["upsert", "tag", "--graph", opts.graph, "--name", tag], { workerId, seq, op, tag, }); return; } case "tag-delete-recreate": { const tag = state.volatileTag; await runCli(opts, ["remove", "tag", "--graph", opts.graph, "--name", tag], { workerId, seq, op, tag, phase: "delete", }); await runCli(opts, ["upsert", "tag", "--graph", opts.graph, "--name", tag], { workerId, seq, op, tag, phase: "recreate", }); return; } case "property-upsert": { const property = chooseArray([...state.stableProperties, state.volatileProperty]); await runCli( opts, ["upsert", "property", "--graph", opts.graph, "--name", property, "--type", "default", "--cardinality", "one", "--public", "true"], { workerId, seq, op, property }, ); return; } case "property-delete-recreate": { const property = state.volatileProperty; await runCli(opts, ["remove", "property", "--graph", opts.graph, "--name", property], { workerId, seq, op, property, phase: "delete", }); await runCli( opts, ["upsert", "property", "--graph", opts.graph, "--name", property, "--type", "default", "--cardinality", "one", "--public", "true"], { workerId, seq, op, property, phase: "recreate" }, ); return; } case "http-insert-blocks": await applyRawInsertBlocks(opts, state, { workerId, seq, op }, `http insert ${stamp}`); return; case "http-delete-blocks": await applyRawDeleteBlocks(opts, state, { workerId, seq, op }); return; case "http-move-blocks": await applyRawMoveBlocks(opts, state, { workerId, seq, op }); return; case "http-save-block": await applyRawSaveBlock(opts, state, { workerId, seq, op }, `http save ${stamp}`); return; case "http-move-up-down": await applyRawMoveUpDown(opts, state, { workerId, seq, op }, Math.random() < 0.5); return; case "http-indent-outdent": await applyRawIndentOutdent(opts, state, { workerId, seq, op }, Math.random() < 0.5); return; case "http-apply-template": await applyRawApplyTemplate(opts, state, { workerId, seq, op }, `http template ${stamp}`); return; case "http-create-page": await applyRawCreatePage( opts, state, { workerId, seq, op }, uniqueOperationPageTitle("CLI HTTP Page", state, { workerId, seq, op }), ); return; case "http-rename-page": await applyRawRenamePage(opts, state, { workerId, seq, op }, `CLI HTTP Rename ${state.runId} ${seq}`); return; case "http-delete-restore-page": await applyRawDeleteRestorePage(opts, state, { workerId, seq, op }); return; case "http-recycle-delete-page": await applyRawRecycleDeletePage( opts, state, { workerId, seq, op }, uniqueOperationPageTitle("CLI HTTP Recycle", state, { workerId, seq, op }), ); return; case "http-upsert-property": await applyRawUpsertProperty(opts, state, { workerId, seq, op }); return; case "http-set-block-property": await applyRawSetBlockProperty(opts, state, { workerId, seq, op }, `http property ${stamp}`); return; case "http-remove-block-property": await applyRawRemoveBlockProperty(opts, state, { workerId, seq, op }); return; case "http-batch-set-property": await applyRawBatchSetProperty(opts, state, { workerId, seq, op }, `http batch ${stamp}`); return; case "http-batch-remove-property": await applyRawBatchRemoveProperty(opts, state, { workerId, seq, op }); return; case "http-batch-delete-property-value": await applyRawBatchDeletePropertyValue(opts, state, { workerId, seq, op }); return; case "http-create-property-history": await applyRawCreatePropertyHistory(opts, state, { workerId, seq, op }); return; case "http-create-linked-references-view": await applyRawCreateReferenceView(opts, state, { workerId, seq, op }, "linked-references", "Linked references"); return; case "http-create-unlinked-references-view": await applyRawCreateReferenceView(opts, state, { workerId, seq, op }, "unlinked-references", "Unlinked references"); return; case "http-toggle-reaction": await applyRawToggleReaction(opts, state, { workerId, seq, op }); return; case "http-delete-view-target-with-related-entities": await applyRawDeleteViewTargetWithRelatedEntities(opts, state, { workerId, seq, op }); return; case "http-insert-undo-redo": await applyRawInsertUndoRedo(opts, state, { workerId, seq, op }, `http undo redo ${stamp}`); return; case "offline-window": await offlineWindow(opts, state, { workerId, seq, op }); return; case "offline-today-journal-race": await offlineTodayJournalRace(opts, state, { workerId, seq, op }); return; case "delete-single": { const id = choose(state.activeIds); if (!id) { return; } const context = { workerId, seq, op, id }; const response = await runCli(opts, ["remove", "block", "--graph", opts.graph, "--id", String(id)], context); pruneStaleIds(state, response, context); state.activeIds.delete(id); return; } case "delete-batch": { const ids = chooseMany(state.activeIds, 4); if (ids.length === 0) { return; } const context = { workerId, seq, op, ids, }; const response = await runCli(opts, ["remove", "block", "--graph", opts.graph, "--id", JSON.stringify(ids)], context); pruneStaleIds(state, response, context); for (const id of ids) { state.activeIds.delete(id); } return; } case "delete-update-race": { const id = choose(state.activeIds); if (!id) { return; } const deleteContext = { workerId, seq, op: `${op}:delete`, id, }; const updateContext = { workerId, seq, op: `${op}:update`, id }; const responses = await Promise.all([ runCli(opts, ["remove", "block", "--graph", opts.graph, "--id", String(id)], deleteContext), runCli( opts, ["upsert", "block", "--graph", opts.graph, "--id", String(id), "--content", `race update ${stamp} id=${id}`], updateContext, ), ]); pruneStaleIds(state, responses[0], deleteContext); pruneStaleIds(state, responses[1], updateContext); state.activeIds.delete(id); return; } default: throw new Error(`Unhandled operation: ${op}`); } } async function worker(opts, state, workerId) { let localSeq = 0; while (!state.stop) { const seq = state.nextSeq; state.nextSeq += 1; if (opts.maxOps > 0 && seq >= opts.maxOps) { state.stop = true; break; } try { await runOperation(opts, state, workerId, localSeq); } catch (error) { logEvent(opts, { event: "worker-error", level: "error", workerId, seq: localSeq, message: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined, }); if (opts.failFast) { state.stop = true; throw error; } } localSeq += 1; if (localSeq % 25 === 0) { logEvent(opts, { event: "summary", level: "info", workerId, localSeq, totalOpsStarted: state.nextSeq, activeIds: state.activeIds.size, }); } await delay(Math.floor(Math.random() * 40)); } } async function clusterWorker(opts, cluster, workerId) { let localSeq = 0; while (!cluster.stop) { const seq = cluster.nextSeq; cluster.nextSeq += 1; if (opts.maxOps > 0 && seq >= opts.maxOps) { cluster.stop = true; break; } const op = weightedOperation(opts); try { if (op === "multi-client-offline-rebase") { await multiClientOfflineRebase(opts, cluster, { workerId, seq, op }); } else if (op === "multi-client-today-journal-race") { await multiClientTodayJournalRace(opts, cluster, { workerId, seq, op }); } else { const client = chooseArray(cluster.clients); client.state.nextSeq += 1; await runOperation(client.opts, client.state, workerId, localSeq, op); } } catch (error) { logEvent(opts, { event: "worker-error", level: "error", workerId, seq: localSeq, message: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined, }); if (opts.failFast) { cluster.stop = true; throw error; } } localSeq += 1; if (localSeq % 25 === 0) { logEvent(opts, { event: "summary", level: "info", workerId, localSeq, totalOpsStarted: cluster.nextSeq, activeIds: cluster.clients.reduce((sum, client) => sum + client.state.activeIds.size, 0), clients: cluster.clients.length, }); } await delay(Math.floor(Math.random() * 40)); } } function createStressState(opts, runId) { const state = { runId, pageNames: stressPageNames(opts), bootstrapPageNames: bootstrapStressPageNames(opts), mutablePageNames: [], stableTags: ["cli-stress-tag-a", "cli-stress-tag-b"], stableTagIds: new Map(), stableProperties: ["cli-stress-prop-a", "cli-stress-prop-b"], volatileTag: "cli-stress-volatile-tag", volatileProperty: "cli-stress-volatile-prop", rawPropertyIdent: "user.property/cli-http-prop", rawPropertyName: "cli-http-prop", activeIds: new Set(), activeBlockUuids: new Map(), taskIds: new Set(), nextSeq: 0, lastRefreshSeq: -Infinity, refreshing: false, offlineRunning: false, stop: false, }; state.mutablePageNames = mutableStressPageNames(opts); return state; } async function validateClientGraph(client, phase) { const validation = await runCli(client.opts, graphValidateArgs(client.opts), { op: `${phase}-graph-validate`, client: client.opts.clientName, }); if (!validation.ok) { throw new Error(`${phase} graph validation failed for ${client.opts.graph} on ${client.opts.clientName}`); } } async function buildSeedClient(opts, runId) { const clientOpts = clientRuntimeOptions(opts, 0); ensureStressConfig(clientOpts); const graphServer = await ensureLocalServers(clientOpts); const state = createStressState(clientOpts, runId); state.workerBaseUrl = graphServer["base-url"]; state.repo = graphServerRepo(clientOpts, graphServer); await validateClientGraph({ opts: clientOpts }, "preflight"); await ensureStressMetadata(clientOpts, state); await refreshLiveIds(clientOpts, state, "startup"); await waitForSyncSettled(clientOpts, { op: "seed-startup-settle", client: clientOpts.clientName }, false); return { opts: clientOpts, state, graphServer }; } async function buildReplicaClient(opts, clientIndex, runId) { const clientOpts = clientRuntimeOptions(opts, clientIndex); ensureStressConfig(clientOpts); const graphServer = await ensureReplicaLocalServers(clientOpts); const state = createStressState(clientOpts, runId); state.workerBaseUrl = graphServer["base-url"]; state.repo = graphServerRepo(clientOpts, graphServer); await validateClientGraph({ opts: clientOpts }, "preflight"); await waitForSyncSettled(clientOpts, { op: "replica-startup-settle", client: clientOpts.clientName }, false); await refreshStressTagIds(clientOpts, state, "replica-startup"); await refreshLiveIds(clientOpts, state, "replica-startup"); return { opts: clientOpts, state, graphServer }; } async function buildStressClients(opts, runId) { const clients = [await buildSeedClient(opts, runId)]; for (let index = 1; index < opts.clients; index += 1) { clients.push(await buildReplicaClient(opts, index, runId)); } return clients; } async function main() { const opts = parseArgs(process.argv.slice(2)); mkdirSync(dirname(opts.logFile), { recursive: true }); const runId = `stress-${Date.now()}`; logEvent(opts, { event: "start", level: "info", runId, config: opts.config, httpBase: opts.httpBase, wsUrl: opts.wsUrl, clients: opts.clients, concurrency: opts.concurrency, maxOps: opts.maxOps, sync: opts.sync, offline: opts.offline, graphE2ee: opts.graphE2ee, pageNames: stressPageNames(opts), }); const clients = await buildStressClients(opts, runId); let totalOpsStarted = clients[0]?.state.nextSeq || 0; if (clients.length === 1) { await Promise.all(Array.from({ length: opts.concurrency }, (_, index) => worker(clients[0].opts, clients[0].state, index + 1))); totalOpsStarted = clients[0].state.nextSeq; } else { const cluster = { clients, nextSeq: 0, stop: false, offlineRunning: false, }; await Promise.all(Array.from({ length: opts.concurrency }, (_, index) => clusterWorker(opts, cluster, index + 1))); totalOpsStarted = cluster.nextSeq; } for (const client of clients) { await waitForSyncSettled(client.opts, { op: "final-sync-settle", client: client.opts.clientName }, true); await validateClientChecksum(client); await validateClientGraph(client, "final"); } logEvent(opts, { event: "stop", level: "info", runId, totalOpsStarted, activeIds: clients.reduce((sum, client) => sum + client.state.activeIds.size, 0), clients: clients.length, }); } function isMainModule() { return process.argv[1] && resolve(process.argv[1]) === resolve(new URL(import.meta.url).pathname); } if (isMainModule()) { main().catch((error) => { const opts = (() => { try { return parseArgs(process.argv.slice(2)); } catch { return { graph: "pi-memory", page: "CLI Concurrent Stress", logFile: resolve(repoRoot, "tmp", "cli-concurrent-edit-stress", "events.jsonl"), }; } })(); mkdirSync(dirname(opts.logFile), { recursive: true }); logEvent(opts, { event: "fatal", level: "error", message: error instanceof Error ? error.message : String(error), stack: error instanceof Error ? error.stack : undefined, }); process.exitCode = 1; }); }