diff --git a/package.json b/package.json index 05577256ea..1c1ac6b1d8 100644 --- a/package.json +++ b/package.json @@ -25,7 +25,9 @@ "bootstrap": "npm run build:submodules && lerna bootstrap", "watch": "lerna watch -- lerna run build --since", "start:qss": "cd ./3rd-party/qss && pnpm run run:app docker:up:complete && MIKRO_ORM_HOST=0.0.0.0 pnpm run run:app migrate:up:docker", - "stop:qss": "cd ./3rd-party/qss && pnpm run run:app docker:down:complete" + "stop:qss": "cd ./3rd-party/qss && pnpm run run:app docker:down:complete", + "start:qss:toxi": "docker compose -f packages/backend/src/nest/qss/stress/docker-compose.yml up -d", + "stop:qss:toxi": "docker compose -f packages/backend/src/nest/qss/stress/docker-compose.yml down" }, "engines": { "node": "20.20.1", diff --git a/packages/backend/jest.stress.config.js b/packages/backend/jest.stress.config.js new file mode 100644 index 0000000000..73ecd0d035 --- /dev/null +++ b/packages/backend/jest.stress.config.js @@ -0,0 +1,28 @@ +/** + * Jest config for the QSS stress harness. + * + * Inherits the package's default ts-jest ESM setup but disables type + * diagnostics. The worktree symlinks 3rd-party submodules from the main + * checkout, which makes ts-jest see two distinct TS module identities for + * the same physical file. Runtime resolution (Node's symlink-collapsing) + * still works correctly; we just need to bypass the static type check. + */ +export default { + preset: 'ts-jest/presets/default-esm', + clearMocks: true, + coverageProvider: 'v8', + transformIgnorePatterns: ['node_modules/(?!p-defer|peer-id)'], + testTimeout: 240_000, + setupFiles: ['./jestSetup.ts'], + testEnvironment: 'jest-environment-node', + testRegex: ['[\\\\/]src[\\\\/]nest[\\\\/]qss[\\\\/]stress[\\\\/].*\\.stress\\.spec\\.ts$'], + transform: { + '^.+\\.tsx?$': [ + 'ts-jest', + { + useESM: true, + diagnostics: false, + }, + ], + }, +} diff --git a/packages/backend/package.json b/packages/backend/package.json index d95da929ca..6cd0016316 100644 --- a/packages/backend/package.json +++ b/packages/backend/package.json @@ -32,6 +32,7 @@ "test-replication-no-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" ts-node -v && cross-env DEBUG='backend:dbSnap*,backend:localTest*' ts-node src/nodeTest/testReplicate.ts --nodesCount 1 --timeThreshold 200 --entriesCount 1000 --no-useTor", "test-replication-tor": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" cross-env DEBUG='backend:dbSnap*,backend:localTest*' ts-node src/nodeTest/testReplicate.ts --nodesCount 1 --timeThreshold 500 --entriesCount 1000 --useTor", "test-it": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" DEBUG=ipfs:*,backend:* node_modules/jest/bin/jest.js --runInBand --verbose --testPathIgnorePatterns=\".src/(!?nodeTest*)|(.node_modules*)\" --", + "test-stress-qss": "cross-env NODE_OPTIONS=\"--experimental-vm-modules\" jest --config jest.stress.config.js --colors --verbose --maxWorkers=4 --detectOpenHandles --forceExit", "rmDist": "rimraf lib/" }, "repository": { @@ -56,6 +57,10 @@ "testEnvironment": "jest-environment-node", "testRegex": [ "[\\\\/]src[\\\\/].*\\.(spec|test)\\.[tj]sx?$" + ], + "testPathIgnorePatterns": [ + "/node_modules/", + "[\\\\/]src[\\\\/]nest[\\\\/]qss[\\\\/]stress[\\\\/]" ] }, "devDependencies": { diff --git a/packages/backend/src/nest/qss/stress/README.md b/packages/backend/src/nest/qss/stress/README.md new file mode 100644 index 0000000000..e2e09d6860 --- /dev/null +++ b/packages/backend/src/nest/qss/stress/README.md @@ -0,0 +1,90 @@ +# QSS stress harness + +A scenario-driven test harness that runs the real backend QSS path against a +real QSS server with a [Toxiproxy](https://github.com/Shopify/toxiproxy) +sidecar in front, so each scenario can inject latency, jitter, packet loss, +and forced connection drops at the network layer. + +Primary focus: **initial setup** — the fresh-create-community sequence +(`connect` → `GET_CAPTCHA_SITE_KEY` → `VERIFY_CAPTCHA` → `GEN_PUB_KEYS` → +`CREATE_COMMUNITY` → `AUTH_SYNC` handshake). Every step is a real round-trip +against real QSS; the only synthetic piece is the renderer-side captcha solve, +which is replaced by the official hCaptcha test token. QSS in dev is +configured with the matching test secret, so `VERIFY_CAPTCHA` really runs and +really succeeds. + +The harness boots only the QSS path of the Nest app (`QSSModule` + the modules +it transitively needs). No tor, no peer-to-peer libp2p — single client talking +to a single QSS server. + +## Layout + +| File | Purpose | +| --- | --- | +| `docker-compose.yml` | Toxiproxy sidecar (host networking, admin :8474). | +| `toxiproxy.ts` | Tiny zero-dep client for the Toxiproxy admin API. | +| `harness.ts` | `bootQssHarness()` — Nest test module wired through the proxy. | +| `precise-faults.ts` | `installWireFaultHooks()` — monkey-patches `qssClient.sendMessage` for per-message faults (drop / delay / error / busy-wait pause) keyed by `WebsocketEvents`. | + +Scenario specs live under `scenarios/*.stress.spec.ts` and are landed +separately on top of this tooling. + +## Running + +Two Node versions are needed: + +- **Node 20.20.1** for the backend tests (the stress jest run). Newer Node + breaks `msgpackr@1.10.2` used inside `3rd-party/auth` with + `RangeError: length is outside of buffer bounds`. +- **Node 22+** for the QSS Postgres migrations (the QSS submodule's `app/` + package has `engines.node >= 22.14.0`). + +From the repo root: + +```bash +# 1. Start a real QSS server on host:3003 +npm run start:qss # uses Node 22 internally for the migrate step + +# 2. Start the toxiproxy sidecar on host:8474 (admin) / :3013 (proxy) +npm run start:qss:toxi + +# 3. Run stress scenarios from packages/backend, under Node 20 +cd packages/backend +PATH=/path/to/node-20/bin:$PATH npm run test-stress-qss +``` + +To stop: + +```bash +npm run stop:qss:toxi +npm run stop:qss +``` + +`*.stress.spec.ts` is excluded from `test` and `test-ci` so this never runs on +CI by default. + +## Configuration + +Environment variables (with their defaults): + +| Var | Default | Meaning | +| --- | --- | --- | +| `QSS_STRESS_ENDPOINT` | `ws://127.0.0.1:3013` | URL the QSSService sees (the proxy listener). | +| `QSS_STRESS_LISTEN` | `127.0.0.1:3013` | Toxiproxy proxy listener (host:port). | +| `QSS_STRESS_UPSTREAM` | `127.0.0.1:3003` | Real QSS server (host:port). | +| `TOXIPROXY_ADMIN` | `http://127.0.0.1:8474` | Toxiproxy admin API. | + +## Why this exists + +Every cluster of recent QSS bug-fix commits has the same shape: a state +machine misbehaves when an external event arrives during a particular timing +window. Mock-socket unit tests can't reach those windows; full e2e tests are +slow and hide internal state behind the renderer/socket bridge. This harness +sits between: real socket, real QSS, no app shell — so scenarios can drive +the timing windows directly and assert on backend internal state. + +## macOS / Windows note + +`docker-compose.yml` uses `network_mode: host`, which only works on Linux. +On other platforms, change the service to use port mappings and set +`QSS_STRESS_UPSTREAM=host.docker.internal:3003`. diff --git a/packages/backend/src/nest/qss/stress/docker-compose.yml b/packages/backend/src/nest/qss/stress/docker-compose.yml new file mode 100644 index 0000000000..3763558dc7 --- /dev/null +++ b/packages/backend/src/nest/qss/stress/docker-compose.yml @@ -0,0 +1,16 @@ +# Toxiproxy sidecar for the QSS stress harness. +# +# Brings up Toxiproxy on the host network so it can front a QSS server +# already running on host:3003 (started via `npm run start:qss`). +# +# Linux: works as-is via `network_mode: host`. +# macOS / Windows: switch to a port-mapped service and point upstream at +# `host.docker.internal:3003` from the test harness. +services: + toxiproxy: + image: ghcr.io/shopify/toxiproxy:2.9.0 + network_mode: host + command: + - "-host=0.0.0.0" + - "-port=8474" + restart: unless-stopped diff --git a/packages/backend/src/nest/qss/stress/harness.ts b/packages/backend/src/nest/qss/stress/harness.ts new file mode 100644 index 0000000000..493db101a2 --- /dev/null +++ b/packages/backend/src/nest/qss/stress/harness.ts @@ -0,0 +1,488 @@ +/** + * Boots a Nest test module with the QSS path wired against a real QSS server + * (front of which a Toxiproxy listener allows network fault injection per scenario). + * + * Mirrors the module graph used by qss.service.spec.ts but overrides + * QSS_ALLOWED/QSS_ENDPOINT so the QSSService talks through the proxy, and + * leaves the full graph (sigchain, identity, current community) primed for + * connect / signIn flows. + */ +import { Test, TestingModule } from '@nestjs/testing' +import * as fs from 'node:fs' +import * as net from 'node:net' + +import { TestModule } from '../../common/test.module' +import { QSSModule } from '../qss.module' +import { QSSService } from '../qss.service' +import { QSSClient } from '../qss.client' +import { QSSAuthConnectionManager } from '../qss-auth-conn-manager.service' +import { QSS_ALLOWED, QSS_ENDPOINT } from '../../const' +import { SigChainModule } from '../../auth/sigchain.service.module' +import { SigChainService } from '../../auth/sigchain.service' +import { OrbitDbService } from '../../storage/orbitDb/orbitDb.service' +import { OrbitDbModule } from '../../storage/orbitDb/orbitdb.module' +import { Libp2pService } from '../../libp2p/libp2p.service' +import { IpfsService } from '../../ipfs/ipfs.service' +import { IpfsModule } from '../../ipfs/ipfs.module' +import { IpfsFileManagerModule } from '../../ipfs-file-manager/ipfs-file-manager.module' +import { LocalDbService } from '../../local-db/local-db.service' +import { CaptchaService } from '../../captcha/captcha.service' +import { spawnLibp2pInstancesInMemory } from '../../common/test-utils' +import { ToxiproxyClient } from './toxiproxy' +import { Community, CommunityOwnership, Identity, InvitationDataVersion } from '@quiet/types' +import { getReduxStoreFactory, prepareStore, Store } from '@quiet/state-manager' +import { FactoryGirl } from 'factory-girl' + +export interface HarnessOptions { + /** URL the QSSService will see for QSS — typically the toxiproxy listener. */ + qssEndpoint?: string + /** Toxiproxy admin URL (defaults to http://127.0.0.1:8474). */ + toxiproxyAdmin?: string + /** Name of the proxy to create at the admin API. */ + proxyName?: string + /** Where the proxy listens (host:port form). Must match qssEndpoint host:port. */ + proxyListen?: string + /** Upstream — where the QSS server actually runs (host:port). */ + proxyUpstream?: string + /** Username for the test identity. */ + username?: string + /** Team name. */ + teamName?: string +} + +export interface QssHarness { + module: TestingModule + qssService: QSSService + qssClient: QSSClient + qssAuthConnManager: QSSAuthConnectionManager + sigchainService: SigChainService + orbitDbService: OrbitDbService + ipfsService: IpfsService + libp2pService: Libp2pService + localDbService: LocalDbService + captchaService: CaptchaService + store: Store + factory: FactoryGirl + toxiproxy: ToxiproxyClient + proxyName: string + qssEndpoint: string + community: Community + identity: Identity + /** + * Pre-stuffs the renderer-side captcha token. After this call, the next + * captcha-required QSS operation runs end-to-end over the wire — the real + * `GET_CAPTCHA_SITE_KEY` and `VERIFY_CAPTCHA` round-trips happen, and QSS + * really validates against hCaptcha (the dev secret accepts the test token). + */ + primeCaptcha: () => void + shutdown: () => Promise +} + +/** + * Official hCaptcha test response token. Always validates as success against + * the test site/secret key pair that QSS runs in dev. + * https://docs.hcaptcha.com/#integration-testing-test-keys + */ +export const HCAPTCHA_TEST_TOKEN = '10000000-aaaa-bbbb-cccc-000000000001' + +export interface OwnerInvite { + inviteId: string + seed: string + salt: string + teamId: string + teamName: string +} + +/** + * Generate a long-lived invite from an owner harness whose community is + * already created on QSS. The returned shape feeds bootMemberHarness. + * + * Also writes the invite lockbox onto the owner's chain so the member's + * later self-assign-MEMBER call has team-side role keys to derive against. + * Without this the LFA assertion `keysAllGenerations` fires when the member + * tries to self-assign. + */ +export function generateOwnerInvite(owner: QssHarness): OwnerInvite { + const sigchain = owner.sigchainService.activeChain + if (sigchain.team == null) { + throw new Error('Owner sigchain has no team — call after createCommunity completes') + } + const invite = sigchain.invites.createLongLivedUserInvite() + sigchain.lockbox.createInviteLockboxes(invite.seed, invite.salt) + return { + inviteId: invite.id, + seed: invite.seed, + salt: invite.salt, + teamId: sigchain.team.id, + teamName: sigchain.team.teamName, + } +} + +export interface MemberHarnessOptions { + /** Invite produced by `generateOwnerInvite(owner)` after the owner created the community. */ + invite: OwnerInvite + /** Username for the joining member. Must differ from owner's. */ + username: string + /** + * If provided, the member shares the owner's proxy listener — chaos applied + * to that proxy is symmetric across both peers. Useful for "the network + * between everyone and QSS" scenarios. + * + * If omitted (the default), the member allocates its own proxy on a free + * port forwarding to the same QSS upstream — chaos applied to either + * peer's proxy only affects that peer's link to QSS. This models real + * asymmetric conditions (owner on home WiFi vs. member on cellular, + * member's phone backgrounding, captive-portal bursts on one side, etc.). + * + * Per-peer chaos is the right default for multi-client scenarios: bugs + * around "owner gives up too soon when member's link is slow" or + * "member's reconnect doesn't catch the owner's latest team state" can + * only show up under independent proxies. + */ + qssEndpoint?: string + toxiproxyAdmin?: string + /** + * If set, reuse this proxy name (and its already-bound listen port) for + * the member. Otherwise a fresh proxy is allocated on a free port via + * {@link pickFreePort} with a unique name from {@link proxyId}. + */ + proxyName?: string + proxyListen?: string + proxyUpstream?: string +} + +export const DEFAULTS = { + qssEndpoint: process.env.QSS_STRESS_ENDPOINT ?? 'ws://127.0.0.1:3013', + toxiproxyAdmin: process.env.TOXIPROXY_ADMIN ?? 'http://127.0.0.1:8474', + proxyName: 'qss', + proxyListen: process.env.QSS_STRESS_LISTEN ?? '127.0.0.1:3013', + proxyUpstream: process.env.QSS_STRESS_UPSTREAM ?? '127.0.0.1:3003', + username: 'stress-user', + teamName: 'stress-community', +} + +/** Default upstream the QSS server itself listens on; per-peer proxies forward here. */ +export const QSS_UPSTREAM = process.env.QSS_STRESS_UPSTREAM ?? '127.0.0.1:3003' + +let _proxyCounter = 0 +/** + * Generate a unique proxy name. Each member harness in a multi-client + * scenario gets its own listener so toxics can be applied per-peer. + */ +export function proxyId(prefix = 'qss-peer'): string { + _proxyCounter += 1 + return `${prefix}-${process.pid}-${Date.now().toString(36)}-${_proxyCounter}` +} + +/** + * Bind to port 0 to let the OS pick a free TCP port, then immediately close + * the socket and return the port. Cheap and reliable on Linux. + * + * NOTE: races with anyone else allocating ports concurrently — fine in test + * harnesses where we allocate sequentially before binding to it via toxiproxy. + */ +export async function pickFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer() + server.unref() + server.on('error', reject) + server.listen(0, '127.0.0.1', () => { + const addr = server.address() + if (typeof addr === 'object' && addr != null) { + const port = addr.port + server.close(() => resolve(port)) + } else { + server.close() + reject(new Error('failed to read port from server.address()')) + } + }) + }) +} + +export async function bootQssHarness(opts: HarnessOptions = {}): Promise { + const proxyUpstream = opts.proxyUpstream ?? DEFAULTS.proxyUpstream + const toxiproxyAdmin = opts.toxiproxyAdmin ?? DEFAULTS.toxiproxyAdmin + const username = opts.username ?? DEFAULTS.username + const teamName = opts.teamName ?? DEFAULTS.teamName + + // Per-test proxy: unique name + free local port. Lets multiple harness + // instances run in parallel without colliding on a shared 'qss' proxy. + const proxyName = opts.proxyName ?? proxyId('qss-owner') + const proxyPort = + opts.proxyListen != null ? Number(opts.proxyListen.split(':').pop()) : await pickFreePort() + const proxyListen = opts.proxyListen ?? `127.0.0.1:${proxyPort}` + const qssEndpoint = opts.qssEndpoint ?? `ws://127.0.0.1:${proxyPort}` + + const adminUrl = new URL(toxiproxyAdmin) + const toxiproxy = new ToxiproxyClient(adminUrl.hostname, Number(adminUrl.port || 8474)) + // fail loudly if toxiproxy isn't up — better than vague socket errors later + await toxiproxy.ping() + await toxiproxy.ensureProxy({ + name: proxyName, + listen: proxyListen, + upstream: proxyUpstream, + enabled: true, + }) + await toxiproxy.clearToxics(proxyName) + + const store = prepareStore().store + const factory = await getReduxStoreFactory(store) + + const module = await Test.createTestingModule({ + imports: [TestModule, SigChainModule, IpfsFileManagerModule, IpfsModule, OrbitDbModule, QSSModule], + }) + .overrideProvider(QSS_ALLOWED) + .useValue(true) + .overrideProvider(QSS_ENDPOINT) + .useValue(qssEndpoint) + .compile() + + const qssService = module.get(QSSService) + const qssClient = module.get(QSSClient) + const qssAuthConnManager = module.get(QSSAuthConnectionManager) + const sigchainService = module.get(SigChainService) + const captchaService = module.get(CaptchaService) + const libp2pService = await module.resolve(Libp2pService) + await spawnLibp2pInstancesInMemory([module]) + const ipfsService = await module.resolve(IpfsService) + await ipfsService.createInstance() + const localDbService = await module.resolve(LocalDbService) + + const community: Community = await factory.create('Community', { name: teamName }) + const identity: Identity = await factory.create('Identity', { communityId: community.id, nickname: username }) + // Sigchain must be active before OrbitDb.create, because the LFA identity + // provider derives the orbitdb identity from the active sigchain. + await sigchainService.createChain(community.name!, username, true) + + const orbitDbService = await module.resolve(OrbitDbService) + await orbitDbService.create(ipfsService.ipfsInstance!) + + await localDbService.setCommunity({ ...community, qssEnabled: true, qssSetup: false } as any) + await localDbService.setCurrentCommunityId(community.id) + await localDbService.setIdentity(identity) + + const shutdown = async (): Promise => { + await toxiproxy.clearToxics(proxyName).catch(() => undefined) + // Per-test proxy cleanup so toxiproxy admin doesn't accumulate state + // across long sweeps. Idempotent if the proxy is already gone. + await toxiproxy.deleteProxy(proxyName).catch(() => undefined) + try { + qssService.close() + } catch { + // ignore + } + await orbitDbService?.stop().catch(() => undefined) + if (orbitDbService?.orbitDbDir != null && fs.existsSync(orbitDbService.orbitDbDir)) { + try { + fs.rmSync(orbitDbService.orbitDbDir, { recursive: true }) + } catch { + // ignore + } + } + await ipfsService?.stop().catch(() => undefined) + await libp2pService?.close(true).catch(() => undefined) + await localDbService?.close().catch(() => undefined) + await module?.close().catch(() => undefined) + } + + const primeCaptcha = (): void => { + captchaService.hcaptchaToken = HCAPTCHA_TEST_TOKEN + } + + return { + module, + qssService, + qssClient, + qssAuthConnManager, + sigchainService, + orbitDbService, + ipfsService, + libp2pService, + localDbService, + captchaService, + store, + factory, + toxiproxy, + proxyName, + qssEndpoint, + community: (await localDbService.getCurrentCommunity())!, + identity, + primeCaptcha, + shutdown, + } +} + +/** + * Boots a member harness — same Nest module graph as the owner, but with a + * sigchain primed from the owner's invite seed and a localDb community whose + * `inviteData` carries the teamId. The QSSService's auto-flow takes the + * `signInToCommunity` branch (since `sigChain.team == null`), drives the + * AUTH_SYNC handshake through QSS, and converges to JoinStatus.JOINED once + * the owner-side LFA accepts the join. + * + * The owner harness must be running and its auth connection active when the + * member is booted, otherwise the AUTH_SYNC routing has no peer. + */ +export async function bootMemberHarness(opts: MemberHarnessOptions): Promise { + const toxiproxyAdmin = opts.toxiproxyAdmin ?? DEFAULTS.toxiproxyAdmin + const proxyUpstream = opts.proxyUpstream ?? QSS_UPSTREAM + + // Default to per-peer proxy allocation: each member gets its own listener + // (unique name + free port). Sharing requires explicitly passing + // proxyName/proxyListen/qssEndpoint matching the owner's. + let proxyName: string + let proxyListen: string + let qssEndpoint: string + if (opts.proxyName != null && opts.proxyListen != null && opts.qssEndpoint != null) { + // Caller explicitly asked to share an existing proxy listener. + proxyName = opts.proxyName + proxyListen = opts.proxyListen + qssEndpoint = opts.qssEndpoint + } else { + proxyName = opts.proxyName ?? proxyId('qss-member') + const port = opts.proxyListen != null ? Number(opts.proxyListen.split(':').pop()) : await pickFreePort() + const host = opts.proxyListen != null ? opts.proxyListen.split(':')[0] : '127.0.0.1' + proxyListen = `${host}:${port}` + qssEndpoint = opts.qssEndpoint ?? `ws://${host}:${port}` + } + const username = opts.username + + const adminUrl = new URL(toxiproxyAdmin) + const toxiproxy = new ToxiproxyClient(adminUrl.hostname, Number(adminUrl.port || 8474)) + await toxiproxy.ping() + // ensureProxy is idempotent — fine if a previous run created this name. + await toxiproxy.ensureProxy({ + name: proxyName, + listen: proxyListen, + upstream: proxyUpstream, + enabled: true, + }) + await toxiproxy.clearToxics(proxyName).catch(() => undefined) + + const store = prepareStore().store + const factory = await getReduxStoreFactory(store) + + const module = await Test.createTestingModule({ + imports: [TestModule, SigChainModule, IpfsFileManagerModule, IpfsModule, OrbitDbModule, QSSModule], + }) + .overrideProvider(QSS_ALLOWED) + .useValue(true) + .overrideProvider(QSS_ENDPOINT) + .useValue(qssEndpoint) + .compile() + + const qssService = module.get(QSSService) + const qssClient = module.get(QSSClient) + const qssAuthConnManager = module.get(QSSAuthConnectionManager) + const sigchainService = module.get(SigChainService) + const captchaService = module.get(CaptchaService) + const libp2pService = await module.resolve(Libp2pService) + await spawnLibp2pInstancesInMemory([module]) + const ipfsService = await module.resolve(IpfsService) + await ipfsService.createInstance() + const localDbService = await module.resolve(LocalDbService) + + // Member's sigchain is built from the invite seed — no team yet. The team + // arrives via the AUTH_SYNC `joined` event during the QSS-routed handshake. + await sigchainService.createChainFromInvite( + username, + opts.invite.teamName, + opts.invite.seed, + opts.invite.teamId, + true + ) + + // OrbitDB is intentionally NOT initialised here. Its LFA identity provider + // reads `sigchain.team!.id` during construction, which throws while the + // member is pre-join (no team yet). Production defers OrbitDB init until + // after the auth handshake completes; we mirror that. + const orbitDbService = await module.resolve(OrbitDbService) + + // Build a Community record carrying the invite data the QSS_HANDLE_SIGN_IN + // handler needs to extract teamId for `signInToCommunity`. qssSetup is left + // false; on this branch the handler reads teamId from inviteData regardless. + const community: Community = await factory.create('Community', { name: opts.invite.teamName }) + const identity: Identity = await factory.create('Identity', { + communityId: community.id, + nickname: username, + }) + const inviteData = { + version: InvitationDataVersion.v3, + pairs: [], + psk: 'no-libp2p-in-this-harness', + qssEnabled: true, + qssEndpoint, + authData: { + communityName: opts.invite.teamName, + seed: opts.invite.seed, + teamId: opts.invite.teamId, + salt: opts.invite.salt, + }, + } + await localDbService.setCommunity({ + ...community, + name: opts.invite.teamName, + qssEnabled: true, + qssSetup: false, + inviteData, + } as Community) + await localDbService.setCurrentCommunityId(community.id) + await localDbService.setIdentity(identity) + + const primeCaptcha = (): void => { + captchaService.hcaptchaToken = HCAPTCHA_TEST_TOKEN + } + + // Only the harness that allocated the proxy should remove it on shutdown. + // If the member is sharing the owner's proxy (legacy opt-in), leave it. + const ownsProxy = opts.proxyName == null + + const shutdown = async (): Promise => { + await toxiproxy.clearToxics(proxyName).catch(() => undefined) + // Per-test proxy cleanup so toxiproxy admin doesn't accumulate state + // across long sweeps. Idempotent if the proxy is already gone. + await toxiproxy.deleteProxy(proxyName).catch(() => undefined) + try { + qssService.close() + } catch { + // ignore + } + await orbitDbService?.stop().catch(() => undefined) + if (orbitDbService?.orbitDbDir != null && fs.existsSync(orbitDbService.orbitDbDir)) { + try { + fs.rmSync(orbitDbService.orbitDbDir, { recursive: true }) + } catch { + // ignore + } + } + await ipfsService?.stop().catch(() => undefined) + await libp2pService?.close(true).catch(() => undefined) + await localDbService?.close().catch(() => undefined) + await module?.close().catch(() => undefined) + if (ownsProxy) { + await toxiproxy.deleteProxy(proxyName).catch(() => undefined) + } + } + + return { + module, + qssService, + qssClient, + qssAuthConnManager, + sigchainService, + orbitDbService, + ipfsService, + libp2pService, + localDbService, + captchaService, + store, + factory, + toxiproxy, + proxyName, + qssEndpoint, + community: (await localDbService.getCurrentCommunity())!, + identity, + primeCaptcha, + shutdown, + } +} diff --git a/packages/backend/src/nest/qss/stress/precise-faults.ts b/packages/backend/src/nest/qss/stress/precise-faults.ts new file mode 100644 index 0000000000..e5ec684f9e --- /dev/null +++ b/packages/backend/src/nest/qss/stress/precise-faults.ts @@ -0,0 +1,283 @@ +/** + * Per-message wire-event fault precision for the QSS stress harness. + * + * Phase-level chaos (e.g. "the network was bad during the captcha→create + * window") is blunt for races that overlap one specific QSS round-trip. For + * example, "the proxy died while the GEN_PUB_KEYS ack was in flight" needs to + * land between `socket.emit('generate-public-keys', ...)` and the moment the + * ack would otherwise arrive. + * + * This module exposes a `WireEvent` taxonomy spanning every QSSClient-driven + * round-trip and a `WireFault` shape that runs an arbitrary action (toxiproxy + * toggles, toxic add/remove, eventLoopPause, etc.) at that precise wire + * boundary. The implementation monkey-patches `QSSClient.sendMessage` from the + * harness — production code is untouched. + * + * Usage from a scenario: + * + * ```ts + * const restore = installWireFaultHooks(harness, [ + * { + * at: 'after-gen-pub-keys-sent', + * action: async h => { + * // Drop the proxy 500 ms after the GEN_PUB_KEYS emit, then bring it + * // back 2 s later. The in-flight ack should be lost; the QSSService + * // should retry / surface the failure cleanly. + * setTimeout(() => h.toxiproxy.setEnabled(h.proxyName, false).catch(() => undefined), 500) + * setTimeout(() => h.toxiproxy.setEnabled(h.proxyName, true).catch(() => undefined), 2500) + * }, + * }, + * ]) + * try { + * // ...drive the scenario... + * } finally { + * restore() + * } + * ``` + * + * `before-*` fires synchronously *before* the `socket.emit` call. `after-*-sent` + * fires synchronously *immediately after* the emit returns — for ack-bearing + * round-trips (`emitWithAck`), the ack is still in flight at that moment, so a + * fault dispatched here precisely overlaps the ack window. + */ +import { WebsocketEvents } from '../qss.types' +import { type QssHarness } from './harness' + +/** + * Every wire boundary we want to be able to slot a fault around. One pair + * (`before-X` / `after-X-sent`) per QSS round-trip surfaced from QSSClient. + * + * The names mirror the production message kinds (kebab-case, prefixed with + * `before-` or `after-X-sent`). Add new pairs here as new QSS round-trips + * become interesting. + */ +export type WireEvent = + // captcha exchange + | 'before-get-captcha-site-key' + | 'after-get-captcha-site-key-sent' + | 'before-verify-captcha' + | 'after-verify-captcha-sent' + // pre-create / sign-in + | 'before-gen-pub-keys' + | 'after-gen-pub-keys-sent' + | 'before-create-community' + | 'after-create-community-sent' + | 'before-sign-in-community' + | 'after-sign-in-community-sent' + // post-join data plane + | 'before-auth-sync' + | 'after-auth-sync-sent' + | 'before-log-entry-sync' + | 'after-log-entry-sync-sent' + | 'before-log-entry-pull' + | 'after-log-entry-pull-sent' + | 'before-log-entry-fanout' + | 'after-log-entry-fanout-sent' + +/** + * Map a {@link WebsocketEvents} value to the matching `before-*` / `after-*` + * boundary names. Returns `null` for events not in the precise-fault catalog + * yet (e.g. push registration). Keep in sync with {@link WireEvent}. + */ +function wireBoundariesFor(event: WebsocketEvents): { before: WireEvent; after: WireEvent } | null { + switch (event) { + case WebsocketEvents.GET_CAPTCHA_SITE_KEY: + return { before: 'before-get-captcha-site-key', after: 'after-get-captcha-site-key-sent' } + case WebsocketEvents.VERIFY_CAPTCHA: + return { before: 'before-verify-captcha', after: 'after-verify-captcha-sent' } + case WebsocketEvents.GEN_PUB_KEYS: + return { before: 'before-gen-pub-keys', after: 'after-gen-pub-keys-sent' } + case WebsocketEvents.CREATE_COMMUNITY: + return { before: 'before-create-community', after: 'after-create-community-sent' } + case WebsocketEvents.SIGN_IN_COMMUNITY: + return { before: 'before-sign-in-community', after: 'after-sign-in-community-sent' } + case WebsocketEvents.AUTH_SYNC: + return { before: 'before-auth-sync', after: 'after-auth-sync-sent' } + case WebsocketEvents.LOG_ENTRY_SYNC: + return { before: 'before-log-entry-sync', after: 'after-log-entry-sync-sent' } + case WebsocketEvents.LOG_ENTRY_PULL: + return { before: 'before-log-entry-pull', after: 'after-log-entry-pull-sent' } + case WebsocketEvents.LOG_ENTRY_FANOUT: + return { before: 'before-log-entry-fanout', after: 'after-log-entry-fanout-sent' } + default: + return null + } +} + +export interface WireFault { + /** When the fault fires relative to the wire emit. */ + at: WireEvent + /** + * Run a function — e.g. apply/clear toxics, toggle the proxy, await an + * eventLoopPause, etc. The harness is passed so the action has access to + * the toxiproxy client and proxy name without closing over them. + * + * The returned promise is awaited. For `after-*-sent` events on ack-bearing + * round-trips, awaiting *blocks* the ack callback path until your action + * resolves — which is usually what you want when you're trying to overlap + * a fault with an in-flight ack. If your action only needs to *kick off* + * timers (e.g. "drop the proxy 500 ms from now"), schedule them and + * return immediately. + */ + action: (harness: QssHarness) => Promise | void +} + +/** + * Restore handle returned from {@link installWireFaultHooks}. Idempotent. + */ +export type RestoreWireFaults = () => void + +/** + * Install per-message fault hooks by monkey-patching `qssClient.sendMessage`. + * + * Implementation choice: we wrap `sendMessage` rather than the underlying + * `socket.emit`. That gives us a stable signature (we know the WebsocketEvent + * the call is for, regardless of how QSS internals route it) and avoids + * having to chase the socket as it gets recreated on reconnect — the next + * `sendMessage` after a reconnect goes through our wrapper just like the first. + * + * Returns a function that restores the original `sendMessage`. Always call it + * in a `finally` so a failing scenario doesn't leave the wrapper installed. + * + * The wrapper: + * - Looks up the `before-*` / `after-*-sent` pair for the event. + * - Calls every matching `before-*` action (in registration order) and + * awaits them, *then* calls the real `sendMessage`. + * - After `sendMessage` returns (which for `withAck = true` means the ack + * has already arrived), calls every matching `after-*-sent` action. + * + * Note on `after-*-sent` semantics for ack-bearing round-trips: the original + * `sendMessage` awaits the ack internally, so by the time we fire `after-*-sent` + * the ack has already been received. To precisely overlap the in-flight ack + * window, your `before-*` action should *schedule* the disturbance with a + * timer (see the example in the file header) — not block on it. We document + * this trade-off rather than try to hook between emit and ack, because doing + * the latter requires patching socket.io-client internals that are not stable. + * + * That said, for non-ack `socket.emit` calls (`withAck = false`), `after-*-sent` + * fires immediately after the emit returns, which is genuinely "right after + * the bytes hit the wire." + */ +export function installWireFaultHooks(harness: QssHarness, wireFaults: readonly WireFault[]): RestoreWireFaults { + if (wireFaults.length === 0) { + return () => undefined + } + + const byEvent = new Map() + for (const fault of wireFaults) { + const arr = byEvent.get(fault.at) ?? [] + arr.push(fault) + byEvent.set(fault.at, arr) + } + + const client = harness.qssClient + const original = client.sendMessage.bind(client) + + const fireAll = async (event: WireEvent): Promise => { + const faults = byEvent.get(event) + if (faults == null || faults.length === 0) return + for (const fault of faults) { + try { + await fault.action(harness) + } catch (e) { + // A fault action throwing shouldn't blow up the production code path + // — the whole point of the harness is to keep service code unaware + // of test injection. Log to stderr instead. + // eslint-disable-next-line no-console + console.warn(`[precise-faults] action for ${event} threw:`, e) + } + } + } + + // Replace the method with a wrapping function. We keep the same `this` + // binding by writing to the instance property — Nest's QSSClient is not + // re-instantiated within a single harness lifetime, so this stays in place + // until restore(). + ;(client as unknown as { sendMessage: typeof client.sendMessage }).sendMessage = (async function patched( + event: WebsocketEvents, + payload: object | undefined, + withAck = false, + timeoutAck = 5000 + ) { + const boundaries = wireBoundariesFor(event) + if (boundaries == null) { + return original(event, payload, withAck, timeoutAck) as unknown + } + + await fireAll(boundaries.before) + try { + const result = await original(event, payload, withAck, timeoutAck) + // After the emit (and, for ack-bearing calls, the ack) — fire post-send + // hooks. For non-ack emits this is "right after bytes left the socket"; + // for ack-bearing emits it's "right after ack came back" (or the + // sendMessage timeout fired). Both are useful boundaries. + await fireAll(boundaries.after) + return result + } catch (e) { + // Still fire `after-*-sent` so cleanup actions run even on failure. + await fireAll(boundaries.after) + throw e + } + }) as typeof client.sendMessage + + let restored = false + return (): void => { + if (restored) return + restored = true + ;(client as unknown as { sendMessage: typeof client.sendMessage }).sendMessage = original + } +} + +/** + * Helper: schedule an outage starting `delayMs` after the action fires, + * lasting `outageMs`. Returns immediately so the action doesn't block the + * production code path. Errors from the toxiproxy calls are swallowed — + * they're noise during chaos and the harness asserts on final state, not + * on transient errors. + */ +export function scheduleOutageAfter(delayMs: number, outageMs: number): WireFault['action'] { + return (h: QssHarness): void => { + setTimeout(() => { + h.toxiproxy.setEnabled(h.proxyName, false).catch(() => undefined) + setTimeout(() => { + h.toxiproxy.setEnabled(h.proxyName, true).catch(() => undefined) + }, outageMs) + }, delayMs) + } +} + +/** + * Helper: wrap an action so it fires only once, on the first matching wire + * event. Subsequent matches are ignored. Use this for "drop the *first* + * SIGN_IN_COMMUNITY ack" scenarios — without it, every reconnect-driven + * retry would re-trigger the fault and the service would never escape. + * + * Returns a closure that captures local "armed" state, so each call to + * `oneShot()` produces an independent latch. + */ +export function oneShot(action: WireFault['action']): WireFault['action'] { + let armed = true + return (h: QssHarness): Promise | void => { + if (!armed) return + armed = false + return action(h) + } +} + +/** + * Helper: synchronously block the event loop for `pauseMs`. Useful for + * "what if the renderer GC'd for half a second right after this message + * went out?" — different from network chaos because the socket is fine but + * the JS path is stalled. + * + * Implemented as a busy wait — keep `pauseMs` small (< 1 s) or you'll + * starve other tests in the same worker. + */ +export function busyWaitPause(pauseMs: number): WireFault['action'] { + return (): void => { + const start = Date.now() + while (Date.now() - start < pauseMs) { + // intentionally empty: simulating event-loop stall + } + } +} diff --git a/packages/backend/src/nest/qss/stress/toxiproxy.ts b/packages/backend/src/nest/qss/stress/toxiproxy.ts new file mode 100644 index 0000000000..b7edb9adee --- /dev/null +++ b/packages/backend/src/nest/qss/stress/toxiproxy.ts @@ -0,0 +1,164 @@ +import http from 'node:http' + +export interface ToxiproxyProxy { + name: string + listen: string + upstream: string + enabled: boolean +} + +export type ToxicType = 'latency' | 'bandwidth' | 'slow_close' | 'timeout' | 'slicer' | 'limit_data' | 'reset_peer' + +export interface ToxicAttrs { + latency?: number + jitter?: number + rate?: number + bytes?: number + delay?: number + timeout?: number + average_size?: number + size_variation?: number +} + +export interface Toxic { + name: string + type: ToxicType + stream?: 'upstream' | 'downstream' + toxicity?: number + attributes?: ToxicAttrs +} + +export class ToxiproxyClient { + constructor( + private readonly host: string = '127.0.0.1', + private readonly port: number = 8474 + ) {} + + private rawRequest(method: string, path: string, body?: unknown): Promise { + return new Promise((resolve, reject) => { + const data = body == null ? undefined : JSON.stringify(body) + const req = http.request( + { + host: this.host, + port: this.port, + method, + path, + headers: + data != null + ? { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(data) } + : { 'Content-Type': 'application/json' }, + }, + res => { + const chunks: Buffer[] = [] + res.on('data', (c: Buffer) => chunks.push(c)) + res.on('end', () => { + const text = Buffer.concat(chunks).toString('utf8') + if (res.statusCode == null || res.statusCode >= 400) { + const err = new Error(`toxiproxy ${method} ${path} -> ${res.statusCode}: ${text}`) + ;(err as Error & { status?: number }).status = res.statusCode + reject(err) + return + } + try { + resolve(text.length === 0 ? (undefined as unknown as T) : (JSON.parse(text) as T)) + } catch (e) { + reject(e instanceof Error ? e : new Error(String(e))) + } + }) + } + ) + req.setTimeout(20_000, () => { + req.destroy(new Error(`toxiproxy ${method} ${path} -> client timeout`)) + }) + req.on('error', reject) + if (data != null) req.write(data) + req.end() + }) + } + + /** + * Retry on transient toxiproxy errors. The admin API is single-threaded; + * under concurrent load it returns 503 timeouts and connection resets that + * are safe to retry. + */ + private async request(method: string, path: string, body?: unknown): Promise { + const maxAttempts = 6 + let lastErr: unknown + for (let attempt = 0; attempt < maxAttempts; attempt++) { + try { + return await this.rawRequest(method, path, body) + } catch (e) { + lastErr = e + const msg = e instanceof Error ? e.message : String(e) + const status = (e as Error & { status?: number }).status + const transient = + status === 503 || + msg.includes('client timeout') || + msg.includes('ECONNRESET') || + msg.includes('socket hang up') + if (!transient || attempt === maxAttempts - 1) throw e + const backoffMs = Math.min(1000, 50 * 2 ** attempt) + Math.floor(Math.random() * 50) + await new Promise(r => setTimeout(r, backoffMs)) + } + } + throw lastErr + } + + /** Create the proxy if absent; otherwise update its definition. */ + async ensureProxy(proxy: ToxiproxyProxy): Promise { + try { + return await this.request('POST', '/proxies', proxy) + } catch (e) { + const msg = e instanceof Error ? e.message : String(e) + // 409 -> already exists; update it via /proxies/{name} + if (!msg.includes('409') && !msg.toLowerCase().includes('already')) throw e + return this.request('POST', `/proxies/${proxy.name}`, proxy) + } + } + + async listProxies(): Promise> { + return this.request('GET', '/proxies') + } + + async getProxy(name: string): Promise { + return this.request('GET', `/proxies/${name}`) + } + + async setEnabled(proxyName: string, enabled: boolean): Promise { + return this.request('POST', `/proxies/${proxyName}`, { enabled }) + } + + async addToxic(proxyName: string, toxic: Toxic): Promise { + return this.request('POST', `/proxies/${proxyName}/toxics`, toxic) + } + + async removeToxic(proxyName: string, toxicName: string): Promise { + await this.request('DELETE', `/proxies/${proxyName}/toxics/${toxicName}`) + } + + async clearToxics(proxyName: string): Promise { + const proxy = await this.getProxy(proxyName) + for (const t of proxy.toxics ?? []) { + try { + await this.removeToxic(proxyName, t.name) + } catch { + // ignore — best-effort cleanup + } + } + } + + /** Reset all proxies to a clean state. */ + async reset(): Promise { + await this.request('POST', '/reset', {}) + } + + /** Remove a single proxy. Used by per-peer harnesses on shutdown. */ + async deleteProxy(name: string): Promise { + await this.request('DELETE', `/proxies/${name}`) + } + + /** Verify the admin API is reachable. Throws otherwise. */ + async ping(): Promise { + await this.request('GET', '/version') + } +}