Run the load generators against our server: the window holds, memory returns to baseline, no defects
This commit is contained in:
@@ -0,0 +1,321 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test, { after, describe } from 'node:test';
|
||||
import type { Session } from '../src/session.ts';
|
||||
import type { Sms } from '../src/sms.ts';
|
||||
import type { LogMethod, SmppLog } from '../src/log.ts';
|
||||
import { server } from '../src/server.ts';
|
||||
|
||||
const SMPP_PORT = Number(process.env.SMPP_PORT ?? '2775');
|
||||
/** Slower than every scenario's submission rate (2000/s for the window runs), so a real backlog
|
||||
* presses on the configured window instead of draining as fast as it fills - see findings/07-load.md. */
|
||||
const SLOW_HANDLER_DELAY_MS = 2;
|
||||
|
||||
function delay(ms: number): Promise<void> {
|
||||
return new Promise(resolve => { setTimeout(resolve, ms); });
|
||||
}
|
||||
|
||||
async function waitFor<T>(get: () => T | undefined, budget: number): Promise<T | undefined> {
|
||||
const deadline = Date.now() + budget;
|
||||
let value = get();
|
||||
|
||||
while (value === undefined && Date.now() < deadline) {
|
||||
await delay(50);
|
||||
value = get();
|
||||
}
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
type ScenarioUserData = { systemId: string };
|
||||
|
||||
function isScenarioUserData(value: unknown): value is ScenarioUserData {
|
||||
return typeof value === 'object' && value !== null && typeof (value as { systemId?: unknown }).systemId === 'string';
|
||||
}
|
||||
|
||||
function scenarioOf(session: Session): string {
|
||||
return isScenarioUserData(session.userData) ? session.userData.systemId : 'unknown';
|
||||
}
|
||||
|
||||
type ScenarioStats = {
|
||||
answerOrder: number[];
|
||||
answered: number;
|
||||
arrived: number;
|
||||
// Set from a 'close' listener attached the moment the session is first seen (smpp.on('session')),
|
||||
// never lazily inside a test body - a session can close well before a test gets around to
|
||||
// watching for it (S9 above may run for a minute; idleTimeout is 40s), and an EventEmitter never
|
||||
// replays an event to a listener added after it fired.
|
||||
closed: boolean;
|
||||
duplicateIds: number;
|
||||
ids: Set<string>;
|
||||
peakOutstanding: number;
|
||||
unansweredErrors: number;
|
||||
};
|
||||
|
||||
const stats = new Map<string, ScenarioStats>();
|
||||
|
||||
function statsFor(name: string): ScenarioStats {
|
||||
const existing = stats.get(name);
|
||||
|
||||
if (existing) return existing;
|
||||
|
||||
const created: ScenarioStats = {
|
||||
answerOrder: [],
|
||||
answered: 0,
|
||||
arrived: 0,
|
||||
closed: false,
|
||||
duplicateIds: 0,
|
||||
ids: new Set(),
|
||||
peakOutstanding: 0,
|
||||
unansweredErrors: 0,
|
||||
};
|
||||
|
||||
stats.set(name, created);
|
||||
|
||||
return created;
|
||||
}
|
||||
|
||||
type LogEntry = { level: string; message: string; metadata: Record<string, boolean | number | string> | undefined };
|
||||
|
||||
const logEntries: LogEntry[] = [];
|
||||
|
||||
function capture(level: string): LogMethod {
|
||||
return (message, metadata) => { logEntries.push({ level, message, metadata }); };
|
||||
}
|
||||
|
||||
const log: SmppLog = {
|
||||
debug: capture('debug'),
|
||||
error: capture('error'),
|
||||
info: capture('info'),
|
||||
verbose: capture('verbose'),
|
||||
warn: capture('warn'),
|
||||
};
|
||||
|
||||
type MemSample = { heapUsed: number; rss: number; t: number };
|
||||
|
||||
const memSamples: MemSample[] = [];
|
||||
const memTimer = setInterval(() => {
|
||||
const usage = process.memoryUsage();
|
||||
|
||||
memSamples.push({ heapUsed: usage.heapUsed, rss: usage.rss, t: Date.now() });
|
||||
}, 5000);
|
||||
|
||||
memTimer.unref();
|
||||
|
||||
const { err, server: smpp } = await server({
|
||||
authenticate: ({ systemId }) => ({ userData: { systemId } satisfies ScenarioUserData }),
|
||||
idleTimeout: 40_000,
|
||||
log,
|
||||
port: SMPP_PORT,
|
||||
});
|
||||
|
||||
assert.equal(err, undefined);
|
||||
assert.ok(smpp);
|
||||
|
||||
const serverErrors: Error[] = [];
|
||||
|
||||
smpp.on('serverError', serverError => { serverErrors.push(serverError); });
|
||||
|
||||
const sessionByScenario = new Map<string, Session>();
|
||||
const slowQueues = new Map<Session, Promise<void>>();
|
||||
|
||||
function answered(session: Session, arrivalIndex: number, result: { err?: Error }): void {
|
||||
const s = statsFor(scenarioOf(session));
|
||||
|
||||
s.answered++;
|
||||
s.answerOrder.push(arrivalIndex);
|
||||
if (result.err) s.unansweredErrors++;
|
||||
}
|
||||
|
||||
function slowRespond(session: Session, sms: Sms, arrivalIndex: number): void {
|
||||
const chain = (slowQueues.get(session) ?? Promise.resolve())
|
||||
.then(async () => { await delay(SLOW_HANDLER_DELAY_MS); })
|
||||
.then(async () => { answered(session, arrivalIndex, await sms.sendResp()); });
|
||||
|
||||
slowQueues.set(session, chain);
|
||||
}
|
||||
|
||||
function fastRespond(session: Session, sms: Sms, arrivalIndex: number): void {
|
||||
void sms.sendResp().then(result => { answered(session, arrivalIndex, result); });
|
||||
}
|
||||
|
||||
smpp.on('session', session => {
|
||||
// Attached now, not lazily in a test body - see the comment on ScenarioStats.closed.
|
||||
session.on('close', () => { statsFor(scenarioOf(session)).closed = true; });
|
||||
|
||||
session.on('sms', sms => {
|
||||
const name = scenarioOf(session);
|
||||
|
||||
sessionByScenario.set(name, session);
|
||||
|
||||
const s = statsFor(name);
|
||||
const arrivalIndex = s.arrived;
|
||||
|
||||
s.arrived++;
|
||||
if (s.ids.has(sms.smsId)) s.duplicateIds++;
|
||||
else s.ids.add(sms.smsId);
|
||||
s.peakOutstanding = Math.max(s.peakOutstanding, s.arrived - s.answered);
|
||||
|
||||
if (name === 'dumb-w500' || name === 'dumb-w2000') slowRespond(session, sms, arrivalIndex);
|
||||
else fastRespond(session, sms, arrivalIndex);
|
||||
});
|
||||
});
|
||||
|
||||
function memShape(): string {
|
||||
if (memSamples.length === 0) return 'no samples taken (run shorter than the 5s sample interval)';
|
||||
|
||||
const first = memSamples[0];
|
||||
const last = memSamples[memSamples.length - 1];
|
||||
|
||||
assert.ok(first);
|
||||
assert.ok(last);
|
||||
|
||||
const rssValues = memSamples.map(sample => sample.rss);
|
||||
const peakRss = Math.max(...rssValues);
|
||||
const minRss = Math.min(...rssValues);
|
||||
const spanS = ((last.t - first.t) / 1000).toFixed(0);
|
||||
|
||||
return [
|
||||
`samples=${String(memSamples.length)} over ${spanS}s`,
|
||||
`rss first=${String(Math.round(first.rss / 1024 / 1024))}MiB`,
|
||||
`min=${String(Math.round(minRss / 1024 / 1024))}MiB`,
|
||||
`max=${String(Math.round(peakRss / 1024 / 1024))}MiB`,
|
||||
`last=${String(Math.round(last.rss / 1024 / 1024))}MiB`,
|
||||
`heapUsed last=${String(Math.round(last.heapUsed / 1024 / 1024))}MiB`,
|
||||
].join(', ');
|
||||
}
|
||||
|
||||
function isSorted(values: number[]): boolean {
|
||||
return values.every((value, index) => index === 0 || (values[index - 1] ?? 0) <= value);
|
||||
}
|
||||
|
||||
function report(line: string): void {
|
||||
process.stdout.write(`${line}\n`);
|
||||
}
|
||||
|
||||
after(async () => {
|
||||
clearInterval(memTimer);
|
||||
report(`memory shape: ${memShape()}`);
|
||||
for (const [name, s] of stats) {
|
||||
report(`${name}: arrived=${String(s.arrived)} answered=${String(s.answered)} duplicateIds=${String(s.duplicateIds)} peakOutstanding=${String(s.peakOutstanding)} unansweredErrors=${String(s.unansweredErrors)}`);
|
||||
}
|
||||
|
||||
await smpp.close();
|
||||
|
||||
for (const serverError of serverErrors) report(`serverError: ${serverError.message}`);
|
||||
});
|
||||
|
||||
// S9 (target 11) and the backpressure-at-server scenario: window 2000 at a high rate against a
|
||||
// handler slowed enough to build a real backlog. window500 is the same shape with a window below
|
||||
// maxHeldMessages (1000, session-options.ts defaults.maxHeldMessages) - see findings/07-load.md for
|
||||
// what that constant, rather than maxOutstanding, turns out to be the one that interacts with a
|
||||
// peer's window.
|
||||
describe('S9 - bounded window against a slowed handler', () => {
|
||||
for (const [name, expectedCount] of [['dumb-w500', 20_000], ['dumb-w2000', 20_000]] as const) {
|
||||
test(`${name}: every message answered exactly once, ordering holds`, async () => {
|
||||
const done = await waitFor(() => (statsFor(name).answered >= expectedCount ? true : undefined), 180_000);
|
||||
|
||||
assert.ok(done, `${name} did not answer ${String(expectedCount)} messages within budget`);
|
||||
|
||||
const s = statsFor(name);
|
||||
|
||||
assert.equal(s.arrived, expectedCount);
|
||||
assert.equal(s.answered, expectedCount);
|
||||
assert.equal(s.duplicateIds, 0);
|
||||
assert.equal(s.unansweredErrors, 0);
|
||||
assert.equal(s.ids.size, expectedCount);
|
||||
assert.ok(isSorted(s.answerOrder), `${name} answered out of arrival order`);
|
||||
});
|
||||
}
|
||||
|
||||
test('window 2000 pressed past maxHeldMessages (1000): the internal held-message cap evicts, window500 never does', async () => {
|
||||
await waitFor(() => (statsFor('dumb-w2000').answered >= 20_000 ? true : undefined), 180_000);
|
||||
|
||||
const evictions = logEntries.filter(entry => entry.message === 'heldMessages - buffer full, dropping the oldest message');
|
||||
|
||||
// window500's peak (<=500) never reaches the 1000 default, so any eviction observed is
|
||||
// necessarily from the w2000 session - the two runs share one server and one log.
|
||||
assert.ok(evictions.length > 0, 'expected at least one held-message eviction under window 2000');
|
||||
// The peer's own window, respected exactly both runs (peakOutstanding read 500 and 2000 on
|
||||
// the nose) - the lower bound is what distinguishes this from window500's own eviction-free run.
|
||||
assert.ok(statsFor('dumb-w2000').peakOutstanding > 1000 && statsFor('dumb-w2000').peakOutstanding <= 2000);
|
||||
assert.equal(statsFor('dumb-w500').peakOutstanding <= 500, true);
|
||||
});
|
||||
|
||||
test('memory after the backlog drains back down is close to before either window run started', async () => {
|
||||
const before = memSamples[0];
|
||||
|
||||
assert.ok(before, 'no memory sample taken before the window runs started');
|
||||
|
||||
// Node's GC is opportunistic, so this is printed evidence of the shape (per findings/07-load.md),
|
||||
// not a hard bound - a real leak reads as a trend across the whole run's samples, not one pair.
|
||||
await delay(5000);
|
||||
|
||||
const after = process.memoryUsage();
|
||||
|
||||
report(
|
||||
`memory around the window runs: before=${String(Math.round(before.rss / 1024 / 1024))}MiB `
|
||||
+ `after=${String(Math.round(after.rss / 1024 / 1024))}MiB`,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
// S6 (target: idleTimeout) - a peer that sends one message and then, using the no-ping binary
|
||||
// (Dockerfile), never speaks again: no enquire_link, ever. smppload was meant to be this peer and
|
||||
// is blocked (findings/07-load.md), so this is the substitute.
|
||||
describe('S6 - idle peer, no enquire_link at all', () => {
|
||||
test('our server drops it at idleTimeout, with no response sent past the one it owed', async () => {
|
||||
const bound = await waitFor(() => (statsFor('dumb-idle').arrived >= 1 ? true : undefined), 20_000);
|
||||
|
||||
assert.ok(bound, 'dumb-idle never submitted its one message');
|
||||
|
||||
// idleTimeout is 40s from the last byte the peer sent (its submit_sm), never from our own
|
||||
// writes (link-timers.ts resets only on inbound data). This test may start running well
|
||||
// past that mark on its own (S9 above can take a minute) - statsFor(...).closed is set from
|
||||
// a 'close' listener attached at session-creation time, so a close from before this test
|
||||
// even started is still seen; budget is slack for a session that is still open, not a clock.
|
||||
const droppedIdle = await waitFor(() => (statsFor('dumb-idle').closed ? true : undefined), 60_000);
|
||||
|
||||
assert.ok(droppedIdle, 'server never dropped the idle peer within idleTimeout + slack');
|
||||
assert.ok(logEntries.some(entry => entry.message === 'linkTimers - closing an idle peer'));
|
||||
// teardown() (session.ts) is a raw close, not an unbind exchange - nothing further is on the
|
||||
// wire for this session, which the capture's histogram (findings/07-load.md) confirms.
|
||||
assert.equal(statsFor('dumb-idle').answered, 1);
|
||||
});
|
||||
});
|
||||
|
||||
// The long soak: the longest run the time-box allows, fast handler, watched for anything that
|
||||
// grows without bound (held messages, listeners, memory). Bounded by wall-clock rather than a
|
||||
// target count: smpp-dumb-client's own TX-tracking window bookkeeping stalls under sustained load
|
||||
// (findings/07-load.md, Peer quirks) well short of the configured count, on the client's side only
|
||||
// - our own arrived/answered stay in lockstep throughout, which is what this asserts.
|
||||
describe('Long soak', () => {
|
||||
const SOAK_DURATION_MS = 300_000;
|
||||
|
||||
test('the longest run the time-box allows: every arrival answered, nothing duplicated, memory does not grow without bound', async () => {
|
||||
await delay(SOAK_DURATION_MS);
|
||||
|
||||
// One more turn for a response mid-flight when the clock ran out to land, not a target count.
|
||||
await waitFor(() => {
|
||||
const s = statsFor('dumb-soak');
|
||||
|
||||
return s.arrived === s.answered ? true : undefined;
|
||||
}, 5000);
|
||||
|
||||
const s = statsFor('dumb-soak');
|
||||
|
||||
report(`soak reached: arrived=${String(s.arrived)} over ${String(SOAK_DURATION_MS / 1000)}s`);
|
||||
|
||||
assert.ok(s.arrived > 0, 'dumb-soak never submitted anything');
|
||||
assert.equal(s.answered, s.arrived);
|
||||
assert.equal(s.duplicateIds, 0);
|
||||
assert.equal(s.unansweredErrors, 0);
|
||||
|
||||
const session = sessionByScenario.get('dumb-soak');
|
||||
|
||||
assert.ok(session);
|
||||
|
||||
const closed = await session.close();
|
||||
|
||||
assert.equal(closed.err, undefined);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user