scripts/cli-concurrent-edit-stress.mjs
Rama master · réplica de lectura del repositorio original.
#!/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 "--client-worker-script":
{
const [indexRaw, scriptPath] = next().split("=");
const index = Number.parseInt(indexRaw, 10);
opts.clientWorkerScripts = opts.clientWorkerScripts || {};
opts.clientWorkerScripts[index] = resolve(scriptPath);
}
break;
case "--client-cli":
{
const [indexRaw, cliPath] = next().split("=");
const index = Number.parseInt(indexRaw, 10);
opts.clientClis = opts.clientClis || {};
opts.clientClis[index] = resolve(cliPath);
}
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) {
if (opts.runSummary) {
recordStressOutcome(opts.runSummary, 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" || event.event === "operation-summary") {
console.log(JSON.stringify(payload));
}
}
function recordStressOutcome(summary, event) {
if (event.event === "worker-error") {
summary.workerErrors += 1;
summary.unexpectedErrors += 1;
return;
}
if (event.event !== "cli" && event.event !== "http") {
return;
}
const key = `${event.event}:${event.context.op}`;
const counts = (summary.operations[key] ||= { ok: 0, expectedRace: 0, unexpectedErrors: 0 });
if (event.outcome === "ok") {
counts.ok += 1;
} else if (event.outcome === "race-conflict") {
counts.expectedRace += 1;
} else {
counts.unexpectedErrors += 1;
summary.unexpectedErrors += 1;
}
}
function assertStressRunSucceeded(summary) {
if (summary.unexpectedErrors > 0) {
throw new Error(`Stress run completed with ${summary.unexpectedErrors} unexpected errors; see operation-summary`);
}
}
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 cliBin =
opts.clientIndex != null && opts.clientClis?.[opts.clientIndex]
? opts.clientClis[opts.clientIndex]
: process.env.LOGSEQ_BIN;
const command = cliCommand(cliBin);
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;
}
const workerScript = opts.clientWorkerScripts?.[clientIndex];
if (workerScript) {
env.LOGSEQ_DB_WORKER_NODE_SCRIPT = workerScript;
}
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);
}
}
async function createStressTree(opts, state, page, stamp, context) {
const blocks = `- tree root ${stamp}\n - tree child ${stamp}\n- tree sibling ${stamp}`;
const response = await runCli(
opts,
["upsert", "block", "--graph", opts.graph, "--target-page", page, "--blocks", blocks],
{ ...context, page },
);
for (const id of resultIds(response)) {
state.activeIds.add(id);
}
return response;
}
function assertStressTree(response, stamp, createdIds) {
const rows = response.parsed?.data?.result;
const fail = () => { throw new Error("create-tree preflight failed: expected a persisted root, nested child and ordered sibling"); };
if (!response.ok || !Array.isArray(rows) || rows.length !== 3 || createdIds.length !== 3 || new Set(createdIds).size !== 3) {
fail();
}
const [root, child, sibling] = ["root", "child", "sibling"].map((kind) => rows.find((row) => row["block/title"] === `tree ${kind} ${stamp}`));
const pageId = root?.["block/page"]?.["db/id"];
if (!root || !child || !sibling || !Number.isInteger(pageId) ||
!rows.every((row) => Number.isInteger(row["db/id"]) && createdIds.includes(row["db/id"]) && row["block/page"]?.["db/id"] === pageId) ||
new Set(rows.map((row) => row["db/id"])).size !== 3 ||
root["block/parent"]?.["db/id"] !== pageId ||
sibling["block/parent"]?.["db/id"] !== pageId ||
child["block/parent"]?.["db/id"] !== root["db/id"] ||
typeof root["block/order"] !== "string" || !root["block/order"] ||
typeof sibling["block/order"] !== "string" || !(root["block/order"] < sibling["block/order"])) {
fail();
}
}
async function preflightCreateTree(opts, state) {
const stamp = `${state.runId} ${opts.clientName} preflight`;
const context = { op: "preflight-create-tree" };
const created = await createStressTree(opts, state, opts.page, stamp, context);
if (!created.ok) {
throw new Error("create-tree preflight failed: CLI rejected the tree fixture");
}
const ids = resultIds(created);
const titles = ["root", "child", "sibling"].map((kind) => ednString(`tree ${kind} ${stamp}`));
const response = await runCli(opts, ["query", "--graph", opts.graph, "--query",
`[:find [(pull ?b [:db/id :block/title {:block/parent [:db/id]} {:block/page [:db/id]} :block/order]) ...] :where [?b :block/title ?title] [(contains? #{${titles.join(" ")}} ?title)]]`],
{ ...context, phase: "verify-persisted-tree" });
assertStressTree(response, stamp, ids);
logEvent(opts, { event: "create-tree-preflight", level: "info", client: opts.clientName, ids });
}
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);
await createStressTree(opts, state, page, stamp, { workerId, seq, op });
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 preflightCreateTree(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 preflightCreateTree(clientOpts, state);
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));
opts.runSummary = { unexpectedErrors: 0, workerErrors: 0, operations: {} };
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;
}
let finalError = null;
for (const client of clients) {
let settleError = null;
try {
await waitForSyncSettled(client.opts, { op: "final-sync-settle", client: client.opts.clientName }, true);
} catch (error) {
settleError = error;
}
await validateClientChecksum(client).catch((error) => {
settleError = settleError || error;
});
await validateClientGraph(client, "final").catch((error) => {
settleError = settleError || error;
});
finalError = finalError || settleError;
}
logEvent(opts, {
event: "operation-summary",
level: opts.runSummary.unexpectedErrors > 0 ? "error" : "info",
runId,
...opts.runSummary,
});
if (finalError) {
throw finalError;
}
assertStressRunSucceeded(opts.runSummary);
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;
});
}