300 lines
16 KiB
JavaScript
300 lines
16 KiB
JavaScript
import assert from 'node:assert/strict';
|
|
import fs from 'node:fs';
|
|
import os from 'node:os';
|
|
import path from 'node:path';
|
|
import { DatabaseSync } from 'node:sqlite';
|
|
import { performance } from 'node:perf_hooks';
|
|
import test from 'node:test';
|
|
import { openTrafficHistoryStore } from '../../dist/server/services/trafficHistoryStore.js';
|
|
import { createTrafficHistoryService } from '../../dist/server/services/trafficHistoryService.js';
|
|
import { openHarborStorage } from '../../dist/server/services/harborStorage.js';
|
|
import { createDomainTrafficService } from '../../dist/server/services/domainTrafficService.js';
|
|
import { assertTrafficHistorySnapshot, parseTrafficHistoryQuery } from '../../dist/shared/trafficHistory.js';
|
|
|
|
const DAY = 86_400_000;
|
|
const base = Date.parse('2026-06-01T10:00:00.000Z');
|
|
const query = (fields = {}) => ({ ...parseTrafficHistoryQuery(new URLSearchParams('range=90d')), ...fields });
|
|
function connection(id, up, down, fields = {}) {
|
|
return {
|
|
id, startedAt: new Date(base + 1_000).toISOString(), closedAt: null,
|
|
inbound: { tag: 'tproxy-in', type: 'tproxy' }, network: 'tcp', protocol: 'tls',
|
|
source: { ip: '192.0.2.10', port: 50_000 },
|
|
destination: { domain: 'www.yandex.ru', ip: '203.0.113.10', port: 443, provenance: 'sing-box' },
|
|
origin: { kind: 'device', id: 'dev_0123456789abcdef', label: 'Laptop', provenance: 'source-ip' },
|
|
route: { kind: 'vpn', scope: 'local-sing-box', outbound: 'vpn-one', outboundType: 'vless', chain: [], rule: null },
|
|
traffic: { uploadBytes: String(up), downloadBytes: String(down), uploadBytesPerSecond: '0', downloadBytesPerSecond: '0' },
|
|
...fields,
|
|
};
|
|
}
|
|
const batch = (at, connections = [], reset = false, epoch = 'sing-box-1') => ({ epoch,
|
|
observedAt: new Date(at).toISOString(), connections, reset, closedIds: connections.filter((c) => c.closedAt).map((c) => c.id) });
|
|
function fixture(t) {
|
|
const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'harbor-traffic-sql-'));
|
|
const closers = new Set();
|
|
t.after(async () => {
|
|
for (const close of closers) await close();
|
|
fs.rmSync(directory, { recursive: true, force: true });
|
|
});
|
|
return { directory, file: path.join(directory, 'traffic.sqlite'), register(store) {
|
|
const original = store.close;
|
|
store.close = () => { closers.delete(store.close); return original(); };
|
|
closers.add(store.close);
|
|
return store;
|
|
} };
|
|
}
|
|
|
|
test('SQL checkpoints survive reopen, reset/replayed CLOSED and same-UUID new lifecycles without double count', (t) => {
|
|
const f = fixture(t);
|
|
let clock = base;
|
|
let store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live');
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [connection('a', 10, 20)])], 'live');
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [connection('a', 15, 27)])], 'live');
|
|
store.close();
|
|
store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [connection('a', 15, 27)], true)], 'live');
|
|
clock += 1_000;
|
|
const closed = connection('a', 18, 30, { closedAt: new Date(clock).toISOString() });
|
|
store.ingest([batch(clock, [closed])], 'live');
|
|
store.close();
|
|
store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [closed], true)], 'live');
|
|
clock += 60_000;
|
|
assert.deepEqual(store.query(query()).totals, { uploadBytes: '18', downloadBytes: '30' });
|
|
store.ingest([batch(clock, [connection('a', 7, 9, { startedAt: new Date(clock).toISOString() })])], 'live');
|
|
clock += 60_000;
|
|
assert.deepEqual(store.query(query()).totals, { uploadBytes: '25', downloadBytes: '39' });
|
|
const result = assertTrafficHistorySnapshot(store.query(query()));
|
|
assert.equal(result.rows[0].label, 'Яндекс');
|
|
});
|
|
|
|
test('full hostname, PSL registered domain, separate IP/device/routes and IP-only destinations remain queryable', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
store.ingest([batch(clock, [
|
|
connection('a', 10, 20),
|
|
connection('b', 3, 4, { destination: { domain: 'mail.yandex.com', ip: '203.0.113.20' } }),
|
|
connection('c', 5, 6, { destination: { domain: 'api.example.co.uk', ip: '203.0.113.30' } }),
|
|
connection('d', 7, 8, { destination: { domain: null, ip: '2001:db8::1' }, origin: { kind: 'unknown', id: null, label: 'Unknown' } }),
|
|
])], 'live');
|
|
clock += 60_000;
|
|
const roots = store.query(query());
|
|
assert.deepEqual(new Set(roots.rows.map((row) => row.key)), new Set(['Яндекс', 'example.co.uk', '']));
|
|
const domains = store.query(query({ level: 'domain', service: 'Яндекс' }));
|
|
assert.deepEqual(new Set(domains.rows.map((row) => row.key)), new Set(['yandex.ru', 'yandex.com']));
|
|
assert.equal(store.query(query({ level: 'hostname', service: 'Яндекс', domain: 'yandex.ru' })).rows[0].key, 'www.yandex.ru');
|
|
const ip = store.query(query({ level: 'ip', service: 'Яндекс', domain: 'yandex.ru', hostname: 'www.yandex.ru' }));
|
|
assert.equal(ip.rows[0].key, '203.0.113.10');
|
|
assert.equal(store.query(query({ level: 'ip', service: '', domain: '', hostname: '' })).rows[0].key, '2001:db8::1');
|
|
assert.equal(store.query(query({ originId: 'unknown:192.0.2.10' })).totals.downloadBytes, '8');
|
|
assert.equal(store.query(query({ search: "x' OR 1=1 --" })).rows.length, 0);
|
|
assert.equal(store.query(query({ search: 'яндекс' })).totals.downloadBytes, '24');
|
|
});
|
|
|
|
test('initial snapshot is a baseline and downtime is explicitly partial', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [connection('old', 1_000, 2_000, { startedAt: new Date(base - DAY).toISOString() })], true)], 'live');
|
|
clock += 1_000;
|
|
assert.equal(store.query(query()).totals.uploadBytes, '0');
|
|
store.ingest([batch(clock, [connection('old', 1_010, 2_020, { startedAt: new Date(base - DAY).toISOString() })])], 'live');
|
|
clock += 60_000;
|
|
store.ingest([batch(clock, [connection('old', 1_040, 2_050, { startedAt: new Date(base - DAY).toISOString() })], true)], 'live');
|
|
clock += 60_000;
|
|
assert.deepEqual(store.query(query()).totals, { uploadBytes: '40', downloadBytes: '50' });
|
|
assert.equal(store.query(query()).coverage.partial, true);
|
|
});
|
|
|
|
test('counter regression retains high watermarks and reconnect counts new post-collection lifecycles', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live');
|
|
for (const total of [100, 90, 100]) {
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [connection('a', total, total)], true)], 'live');
|
|
}
|
|
clock += 60_000;
|
|
const result = store.query(query());
|
|
assert.deepEqual(result.totals, { uploadBytes: '100', downloadBytes: '100' });
|
|
assert.equal(result.coverage.partial, true);
|
|
});
|
|
|
|
test('fast collector reopen marks missing coverage even if the next source state is live', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
let store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live');
|
|
store.close(); clock += 1_000;
|
|
store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [connection('a', 10, 20)], true)], 'live');
|
|
clock += 60_000;
|
|
assert.equal(store.query(query()).coverage.gapCount, 1);
|
|
});
|
|
|
|
test('frozen history cutoff excludes its incomplete terminal bucket before and after hourly rollup', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
store.ingest([batch(clock, [connection('a', 100, 200)])], 'live');
|
|
clock += 60_000;
|
|
store.ingest([batch(clock, [connection('a', 110, 220)])], 'live');
|
|
const frozen = store.query(query());
|
|
assert.equal(frozen.query.until, base + 60_000);
|
|
assert.equal(frozen.totals.uploadBytes, '100');
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [connection('a', 120, 240)])], 'live');
|
|
assert.deepEqual(store.query(frozen.query).totals, frozen.totals);
|
|
clock = base + 8 * DAY;
|
|
const hourly = store.query(query({ until: base + 3_600_000 + 30_000 }));
|
|
assert.equal(hourly.query.until, base + 3_600_000);
|
|
assert.equal(hourly.totals.uploadBytes, '120');
|
|
const expired = store.query(query({ until: base - 100 * DAY }));
|
|
assert.equal(expired.rows.length, 0);
|
|
assert.equal(expired.period.from, expired.period.to);
|
|
});
|
|
|
|
test('minute to hour rollup preserves exact bytes and identities; 90-day cleanup leaves settings and metrics untouched', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
const settings = f.register(openHarborStorage(f.directory));
|
|
settings.state.update((state) => ({ ...state, revision: 42 }));
|
|
const metrics = createDomainTrafficService({ observe: () => ({ connections: [] }), devices: () => [] });
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
const data = batch(clock, [connection('a', '9007199254740993', '500')]);
|
|
store.ingest([data], 'live'); metrics.ingestNative(data);
|
|
const metricsBefore = JSON.stringify(metrics.snapshot());
|
|
clock += 60_000;
|
|
store.ingest([batch(clock, [connection('a', '9007199254741000', '550')])], 'live');
|
|
clock = base + 8 * DAY;
|
|
store.maintain(); store.maintain();
|
|
assert.deepEqual(store.query(query()).totals, { uploadBytes: '9007199254741000', downloadBytes: '550' });
|
|
const inspect = new DatabaseSync(f.file);
|
|
assert.deepEqual(inspect.prepare('SELECT DISTINCT resolution FROM buckets').all().map((r) => r.resolution), [3_600_000]);
|
|
inspect.close();
|
|
clock = base + 91 * DAY;
|
|
store.maintain();
|
|
assert.equal(store.query(query()).rows.length, 0);
|
|
assert.equal(settings.state.read().revision, 42);
|
|
assert.equal(JSON.stringify(metrics.snapshot()), metricsBefore);
|
|
});
|
|
|
|
test('a failed bucket transaction does not advance checkpoints, and retry counts exactly once', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
const blocker = new DatabaseSync(f.file);
|
|
blocker.exec("CREATE TRIGGER reject_bucket BEFORE INSERT ON buckets BEGIN SELECT RAISE(ABORT, 'simulated disk failure'); END");
|
|
const data = batch(clock, [connection('a', 10, 20)]);
|
|
assert.throws(() => store.ingest([data], 'live'), /simulated/);
|
|
assert.equal(blocker.prepare('SELECT COUNT(*) AS n FROM checkpoints').get().n, 0);
|
|
blocker.exec('DROP TRIGGER reject_bucket'); blocker.close();
|
|
store.ingest([data], 'live'); store.ingest([data], 'live'); clock += 60_000;
|
|
assert.deepEqual(store.query(query()).totals, { uploadBytes: '10', downloadBytes: '20' });
|
|
});
|
|
|
|
test('idle active checkpoints survive retention; a new runtime epoch removes old checkpoints but not buckets', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
store.ingest([batch(clock, [connection('idle', 100, 200)])], 'live');
|
|
clock += 91 * DAY;
|
|
store.maintain();
|
|
store.ingest([batch(clock, [connection('idle', 110, 220)])], 'live');
|
|
clock += 60_000;
|
|
assert.equal(store.query(query()).totals.uploadBytes, '10');
|
|
store.ingest([batch(clock, [], true, 'sing-box-2')], 'live');
|
|
const db = new DatabaseSync(f.file);
|
|
assert.equal(db.prepare('SELECT COUNT(*) AS n FROM checkpoints').get().n, 0);
|
|
db.close();
|
|
assert.equal(store.query(query()).totals.uploadBytes, '10');
|
|
});
|
|
|
|
test('reset disappearance and terminal identities retire checkpoints without forgetting still-active baselines', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live'); clock += 1_000;
|
|
store.ingest([batch(clock, [connection('gone', 10, 20), connection('idle', 100, 200), connection('terminal', 7, 8)])], 'live');
|
|
clock += 1_000;
|
|
store.ingest([{ ...batch(clock), closedIds: ['terminal'] }], 'live');
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, [connection('idle', 100, 200)], true)], 'live');
|
|
clock += 91 * DAY;
|
|
store.maintain();
|
|
const db = new DatabaseSync(f.file);
|
|
assert.deepEqual(db.prepare("SELECT json_extract(identity, '$[1]') AS id, closed FROM checkpoints").all()
|
|
.map((row) => ({ ...row })), [{ id: 'idle', closed: 0 }]);
|
|
db.close();
|
|
store.ingest([batch(clock, [connection('idle', 110, 220)])], 'live');
|
|
clock += 60_000;
|
|
assert.equal(store.query(query()).totals.uploadBytes, '10');
|
|
});
|
|
|
|
test('worker ingestion is independent of Prometheus and errors remain isolated; many closed events are not UI-capped', async (t) => {
|
|
const f = fixture(t);
|
|
const service = f.register(createTrafficHistoryService({ filePath: f.file, source: () => 'live' }));
|
|
const at = Date.now() - 120_000;
|
|
service.enqueue(batch(at, [], true));
|
|
service.enqueue(batch(at + 1, Array.from({ length: 2_049 }, (_, index) => connection(`closed-${index}`, 1, 2, {
|
|
startedAt: new Date(at + 1).toISOString(), closedAt: new Date(at + 1).toISOString(),
|
|
}))));
|
|
let ticked = false;
|
|
setImmediate(() => { ticked = true; });
|
|
const result = await service.query(query());
|
|
assert.equal(ticked, true);
|
|
assert.equal(result.storage.status, 'ready');
|
|
assert.deepEqual(result.totals, { uploadBytes: '2049', downloadBytes: '4098' });
|
|
const invalid = f.register(createTrafficHistoryService({ filePath: f.directory, source: () => 'live' }));
|
|
assert.doesNotThrow(() => invalid.enqueue(batch(at, [])));
|
|
assert.equal((await invalid.query(query())).storage.status, 'error');
|
|
});
|
|
|
|
test('bounded 90-day history performance sample reports write/query size without claiming a hardware guarantee', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live');
|
|
const start = performance.now();
|
|
for (let day = 0; day < 90; day++) {
|
|
clock = base + day * DAY + 1_000;
|
|
store.ingest([batch(clock, Array.from({ length: 100 }, (_, index) => connection(`${day}-${index}`, 100, 1_000, {
|
|
startedAt: new Date(clock).toISOString(), destination: { domain: `host${index}.example.org`, ip: `203.0.113.${index + 1}` },
|
|
})))], 'live');
|
|
}
|
|
const writeMs = performance.now() - start;
|
|
clock += 60_000;
|
|
const queryStart = performance.now();
|
|
const result = store.query(query());
|
|
const queryMs = performance.now() - queryStart;
|
|
assert.equal(result.totals.downloadBytes, '9000000');
|
|
const sizes = Object.fromEntries(['', '-wal'].map((suffix) => [suffix || 'db', fs.statSync(f.file + suffix).size]));
|
|
t.diagnostic(JSON.stringify({ samples: 9_000, simulatedDays: 90, writeMs, queryMs, sizes }));
|
|
});
|
|
|
|
test('high-churn sample measures closed-lifecycle storage and epoch reclamation', (t) => {
|
|
const f = fixture(t); let clock = base;
|
|
const store = f.register(openTrafficHistoryStore(f.file, () => clock));
|
|
store.ingest([batch(clock, [], true)], 'live');
|
|
const start = performance.now();
|
|
for (let page = 0; page < 100; page++) {
|
|
clock += 1_000;
|
|
store.ingest([batch(clock, Array.from({ length: 1_000 }, (_, index) => connection(`${page}-${index}`, 100, 1_000, {
|
|
startedAt: new Date(clock).toISOString(), closedAt: new Date(clock).toISOString(),
|
|
})))], 'live');
|
|
}
|
|
const writeMs = performance.now() - start;
|
|
clock += 60_000;
|
|
const startQuery = performance.now();
|
|
assert.equal(store.query(query()).totals.downloadBytes, '100000000');
|
|
const queryMs = performance.now() - startQuery;
|
|
const db = new DatabaseSync(f.file);
|
|
assert.equal(db.prepare('SELECT COUNT(*) AS n FROM checkpoints').get().n, 100_000);
|
|
const bytes = db.prepare('PRAGMA page_count').get().page_count * db.prepare('PRAGMA page_size').get().page_size;
|
|
store.ingest([batch(clock, [], true, 'sing-box-next')], 'live');
|
|
assert.equal(db.prepare('SELECT COUNT(*) AS n FROM checkpoints').get().n, 0);
|
|
const reusableBytes = db.prepare('PRAGMA freelist_count').get().freelist_count * db.prepare('PRAGMA page_size').get().page_size;
|
|
db.close();
|
|
t.diagnostic(JSON.stringify({ closedLifecycles: 100_000, writeMs, queryMs, bytes, reusableBytes }));
|
|
});
|