581 lines
22 KiB
TypeScript
581 lines
22 KiB
TypeScript
import crypto from 'node:crypto';
|
|
import net from 'node:net';
|
|
import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process';
|
|
import { isDeviceInterface, type NeighborObservation } from '../adapters/neighbors.js';
|
|
|
|
interface CommandOptions {
|
|
encoding: BufferEncoding;
|
|
timeout: number;
|
|
killSignal: NodeJS.Signals;
|
|
input?: string;
|
|
}
|
|
|
|
interface CommandResult {
|
|
status: number | null;
|
|
stdout: string;
|
|
stderr: string;
|
|
error?: unknown;
|
|
}
|
|
|
|
interface TrafficDevice {
|
|
ip: string;
|
|
mac: string;
|
|
interface: string;
|
|
key: string;
|
|
}
|
|
|
|
type CounterKind = 'upload' | 'download' | 'proxy-upload' | 'proxy-download' | 'direct-upload' | 'direct-download';
|
|
type CounterField = 'upload' | 'download' | 'proxyUpload' | 'proxyDownload' | 'directUpload' | 'directDownload';
|
|
type CounterOutput = 'uploadBytes' | 'downloadBytes' | 'proxyUploadBytes' | 'proxyDownloadBytes' | 'directUploadBytes' | 'directDownloadBytes';
|
|
type CounterValues = Record<CounterField, bigint>;
|
|
|
|
interface RetiredCounters {
|
|
slot: 'A' | 'B';
|
|
devices: TrafficDevice[];
|
|
counters: Map<string, string>;
|
|
}
|
|
|
|
interface TrafficSnapshot {
|
|
epoch: string;
|
|
generation: string;
|
|
observedAt: string | null;
|
|
source: { error: string | null };
|
|
direct: { uploadBytes: string; downloadBytes: string };
|
|
devices: Record<string, unknown>[];
|
|
}
|
|
|
|
type RunCommand = (command: string, args: string[], options?: CommandOptions) => Promise<CommandResult>;
|
|
|
|
const COMMAND_OPTIONS: CommandOptions = { encoding: 'utf8', timeout: 2_000, killSignal: 'SIGKILL' };
|
|
const MAC_PATTERN = /^[0-9a-f]{2}(?::[0-9a-f]{2}){5}$/i;
|
|
const CHAIN_PATTERN = /^[a-z0-9_]{1,24}$/i;
|
|
const COUNTERS = [
|
|
['upload', 'upload', 'uploadBytes'],
|
|
['download', 'download', 'downloadBytes'],
|
|
['proxy-upload', 'proxyUpload', 'proxyUploadBytes'],
|
|
['proxy-download', 'proxyDownload', 'proxyDownloadBytes'],
|
|
['direct-upload', 'directUpload', 'directUploadBytes'],
|
|
['direct-download', 'directDownload', 'directDownloadBytes'],
|
|
] as const satisfies readonly (readonly [CounterKind, CounterField, CounterOutput])[];
|
|
|
|
const childChain = (chain: string, slot: string) => `${chain}_${slot}`;
|
|
const proxyChildChain = (chain: string, slot: string) => `${childChain(chain, slot)}_P`;
|
|
const counterKey = ({ ip, mac, interface: deviceInterface }: Omit<TrafficDevice, 'key'>) => crypto
|
|
.createHash('sha256')
|
|
.update(`${ip}|${mac}|${deviceInterface}`)
|
|
.digest('hex')
|
|
.slice(0, 16);
|
|
|
|
function commandError(command: string, result: CommandResult) {
|
|
const cause = result.error instanceof Error ? result.error.message : result.error;
|
|
return new Error(String(
|
|
result.stderr || result.stdout || cause || `${command} завершился с ошибкой`,
|
|
).trim());
|
|
}
|
|
|
|
function runCommand(command: string, args: string[], options: CommandOptions = COMMAND_OPTIONS): Promise<CommandResult> {
|
|
return new Promise((resolve) => {
|
|
let child: ChildProcessWithoutNullStreams;
|
|
try {
|
|
child = spawn(command, args, { stdio: ['pipe', 'pipe', 'pipe'] });
|
|
} catch (error) {
|
|
resolve({ status: null, stdout: '', stderr: '', error });
|
|
return;
|
|
}
|
|
const stdout: Buffer[] = [];
|
|
const stderr: Buffer[] = [];
|
|
let settled = false;
|
|
let timedOut = false;
|
|
let timer: ReturnType<typeof setTimeout> | undefined;
|
|
const finish = (result: Pick<CommandResult, 'status'> & { error?: unknown }) => {
|
|
if (settled) return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
resolve({
|
|
stdout: Buffer.concat(stdout).toString(options.encoding || 'utf8'),
|
|
stderr: Buffer.concat(stderr).toString(options.encoding || 'utf8'),
|
|
...result,
|
|
});
|
|
};
|
|
child.stdout.on('data', (chunk: Buffer) => stdout.push(chunk));
|
|
child.stderr.on('data', (chunk: Buffer) => stderr.push(chunk));
|
|
child.on('error', (error) => finish({ status: null, error }));
|
|
child.on('close', (status) => finish({
|
|
status,
|
|
error: timedOut ? new Error(`${command} превысил ${options.timeout} мс`) : null,
|
|
}));
|
|
timer = setTimeout(() => {
|
|
timedOut = true;
|
|
child.kill(options.killSignal || 'SIGKILL');
|
|
}, options.timeout);
|
|
child.stdin.on('error', () => {});
|
|
child.stdin.end(options.input == null ? undefined : options.input);
|
|
});
|
|
}
|
|
|
|
function isIpv4Cidr(value: unknown) {
|
|
const [address, prefix, extra] = String(value).split('/');
|
|
const size = Number(prefix);
|
|
return extra === undefined && net.isIPv4(address)
|
|
&& Number.isInteger(size) && size >= 0 && size <= 32;
|
|
}
|
|
|
|
function cidrRange(cidr: string) {
|
|
const [address, prefix] = cidr.split('/');
|
|
const value = address.split('.').reduce((result, octet) => result * 256n + BigInt(octet), 0n);
|
|
const bits = BigInt(Number(prefix));
|
|
const mask = bits === 0n ? 0n : (0xffff_ffffn << (32n - bits)) & 0xffff_ffffn;
|
|
const first = value & mask;
|
|
return [first, first | (0xffff_ffffn ^ mask)] as const;
|
|
}
|
|
|
|
function hasOverlappingCidrs(cidrs: readonly string[]) {
|
|
const ranges = cidrs.map(cidrRange).sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0));
|
|
return ranges.some(([start], index) => index > 0 && start <= ranges[index - 1][1]);
|
|
}
|
|
|
|
const zeroCounters = (): CounterValues => ({
|
|
upload: 0n,
|
|
download: 0n,
|
|
proxyUpload: 0n,
|
|
proxyDownload: 0n,
|
|
directUpload: 0n,
|
|
directDownload: 0n,
|
|
});
|
|
|
|
function markValue(value: unknown) {
|
|
try {
|
|
const parsed = BigInt(String(value));
|
|
return parsed > 0n && parsed <= 0xffff_ffffn ? parsed : null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function record(value: unknown): Record<string, unknown> {
|
|
return value && typeof value === 'object' && !Array.isArray(value)
|
|
? value as Record<string, unknown>
|
|
: {};
|
|
}
|
|
|
|
export function selectTrafficDevices(observations: unknown): TrafficDevice[] {
|
|
const candidates = new Map<string, Omit<TrafficDevice, 'key'>>();
|
|
const ipsByMac = new Map<string, Set<string>>();
|
|
const locationsByIp = new Map<string, Set<string>>();
|
|
|
|
for (const value of Array.isArray(observations) ? observations : []) {
|
|
const observation = record(value);
|
|
const ip = String(observation.ip || '');
|
|
const mac = String(observation.mac || '').toLowerCase();
|
|
const deviceInterface = String(observation.interface || '');
|
|
if (!net.isIPv4(ip) || !MAC_PATTERN.test(mac) || !isDeviceInterface(deviceInterface)) continue;
|
|
|
|
const location = `${mac}|${deviceInterface}`;
|
|
candidates.set(`${ip}|${location}`, { ip, mac, interface: deviceInterface });
|
|
if (!ipsByMac.has(mac)) ipsByMac.set(mac, new Set());
|
|
ipsByMac.get(mac)?.add(ip);
|
|
if (!locationsByIp.has(ip)) locationsByIp.set(ip, new Set());
|
|
locationsByIp.get(ip)?.add(location);
|
|
}
|
|
|
|
return [...candidates.values()]
|
|
.filter(({ ip, mac }) => ipsByMac.get(mac)?.size === 1 && locationsByIp.get(ip)?.size === 1)
|
|
.map((device) => ({ ...device, key: counterKey(device) }))
|
|
.sort((left, right) => (
|
|
left.ip.localeCompare(right.ip)
|
|
|| left.mac.localeCompare(right.mac)
|
|
|| left.interface.localeCompare(right.interface)
|
|
));
|
|
}
|
|
|
|
export function buildTrafficRestore({
|
|
devices,
|
|
bypassCidrs,
|
|
uploadChain,
|
|
downloadChain,
|
|
directChain,
|
|
directMark,
|
|
tproxyMark,
|
|
gatewayClientCidrs,
|
|
slot,
|
|
proxyPort,
|
|
}: {
|
|
devices: readonly TrafficDevice[];
|
|
bypassCidrs: readonly string[];
|
|
uploadChain: string;
|
|
downloadChain: string;
|
|
directChain: string;
|
|
directMark: string;
|
|
tproxyMark: string;
|
|
gatewayClientCidrs: readonly string[];
|
|
slot: string;
|
|
proxyPort: number;
|
|
}) {
|
|
const parsedDirectMark = markValue(directMark);
|
|
const parsedTproxyMark = markValue(tproxyMark);
|
|
if (!CHAIN_PATTERN.test(uploadChain) || !CHAIN_PATTERN.test(downloadChain) || !CHAIN_PATTERN.test(directChain)
|
|
|| !['A', 'B'].includes(slot) || !Number.isInteger(proxyPort)
|
|
|| proxyPort < 1 || proxyPort > 65_535
|
|
|| parsedDirectMark == null || (parsedDirectMark & (parsedDirectMark - 1n)) !== 0n
|
|
|| parsedTproxyMark == null || (parsedDirectMark & parsedTproxyMark) !== 0n
|
|
|| !Array.isArray(bypassCidrs) || bypassCidrs.some((cidr) => !isIpv4Cidr(cidr))
|
|
|| !Array.isArray(gatewayClientCidrs) || gatewayClientCidrs.some((cidr) => !isIpv4Cidr(cidr))
|
|
|| hasOverlappingCidrs(gatewayClientCidrs)) {
|
|
throw new Error('Некорректная конфигурация traffic accounting');
|
|
}
|
|
const uploadChild = childChain(uploadChain, slot);
|
|
const downloadChild = childChain(downloadChain, slot);
|
|
const directChild = childChain(directChain, slot);
|
|
const proxyUploadChild = proxyChildChain(uploadChain, slot);
|
|
const proxyDownloadChild = proxyChildChain(downloadChain, slot);
|
|
const raw = [
|
|
'*raw',
|
|
`-F ${uploadChild}`,
|
|
`-F ${proxyUploadChild}`,
|
|
`-A ${uploadChild} -i br-+ -j RETURN`,
|
|
`-A ${uploadChild} -p tcp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j ${proxyUploadChild}`,
|
|
`-A ${uploadChild} -p tcp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j RETURN`,
|
|
`-A ${uploadChild} -p udp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j ${proxyUploadChild}`,
|
|
`-A ${uploadChild} -p udp --dport ${proxyPort} -m addrtype --dst-type LOCAL -j RETURN`,
|
|
];
|
|
const mangle = [
|
|
'*mangle',
|
|
`-F ${directChild}`,
|
|
`-F ${downloadChild}`,
|
|
`-F ${proxyDownloadChild}`,
|
|
`-A ${directChild} -m addrtype --dst-type LOCAL -j RETURN`,
|
|
`-A ${directChild} -m mark --mark ${tproxyMark}/${tproxyMark} -j RETURN`,
|
|
`-A ${directChild} -i br-+ -j RETURN`,
|
|
`-A ${downloadChild} -p tcp --sport ${proxyPort} -m addrtype --src-type LOCAL -j ${proxyDownloadChild}`,
|
|
`-A ${downloadChild} -p tcp --sport ${proxyPort} -m addrtype --src-type LOCAL -j RETURN`,
|
|
`-A ${downloadChild} -p udp --sport ${proxyPort} -m addrtype --src-type LOCAL -j ${proxyDownloadChild}`,
|
|
`-A ${downloadChild} -p udp --sport ${proxyPort} -m addrtype --src-type LOCAL -j RETURN`,
|
|
];
|
|
|
|
for (const device of devices) {
|
|
for (const protocol of ['tcp', 'udp']) {
|
|
raw.push(`-A ${proxyUploadChild} -i ${device.interface} -s ${device.ip} -m mac --mac-source ${device.mac} -p ${protocol} -m comment --comment harbor-traffic:${device.key}:proxy-upload -j RETURN`);
|
|
mangle.push(`-A ${proxyDownloadChild} -o ${device.interface} -d ${device.ip} -p ${protocol} -m comment --comment harbor-traffic:${device.key}:proxy-download -j RETURN`);
|
|
}
|
|
}
|
|
for (const cidr of bypassCidrs) {
|
|
raw.push(`-A ${uploadChild} -d ${cidr} -j RETURN`);
|
|
mangle.push(`-A ${directChild} -d ${cidr} -j RETURN`);
|
|
mangle.push(`-A ${downloadChild} -s ${cidr} -j RETURN`);
|
|
}
|
|
for (const cidr of gatewayClientCidrs) {
|
|
mangle.push(`-A ${directChild} -s ${cidr} -m comment --comment harbor-traffic:global:direct-upload -j CONNMARK --set-xmark ${directMark}/${directMark}`);
|
|
mangle.push(`-A ${downloadChild} -d ${cidr} -m connmark --mark ${directMark}/${directMark} -m comment --comment harbor-traffic:global:direct-download`);
|
|
}
|
|
for (const device of devices) {
|
|
mangle.push(`-A ${directChild} -i ${device.interface} -s ${device.ip} -m mac --mac-source ${device.mac} -m connmark --mark ${directMark}/${directMark} -m comment --comment harbor-traffic:${device.key}:direct-upload`);
|
|
mangle.push(`-A ${downloadChild} -o ${device.interface} -d ${device.ip} -m connmark --mark ${directMark}/${directMark} -m comment --comment harbor-traffic:${device.key}:direct-download`);
|
|
raw.push(`-A ${uploadChild} -i ${device.interface} -s ${device.ip} -m mac --mac-source ${device.mac} -m comment --comment harbor-traffic:${device.key}:upload -j RETURN`);
|
|
mangle.push(`-A ${downloadChild} -o ${device.interface} -d ${device.ip} -m comment --comment harbor-traffic:${device.key}:download -j RETURN`);
|
|
}
|
|
return [...raw, 'COMMIT', ...mangle, 'COMMIT', ''].join('\n');
|
|
}
|
|
|
|
export function parseTrafficCounters(text: unknown, chain: string): Map<string, string> {
|
|
const escapedChain = chain.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
|
|
const linePattern = new RegExp(
|
|
`^\\[(\\d+):(\\d+)\\] -A ${escapedChain} .*--comment "?harbor-traffic:([a-f0-9]{16}|global):(upload|download|proxy-upload|proxy-download|direct-upload|direct-download)"?`,
|
|
);
|
|
const counters = new Map<string, string>();
|
|
for (const line of String(text || '').split(/\r?\n/)) {
|
|
const match = line.match(linePattern);
|
|
if (!match) continue;
|
|
const key = `${match[3]}:${match[4]}`;
|
|
counters.set(key, (BigInt(counters.get(key) || '0') + BigInt(match[2])).toString());
|
|
}
|
|
return counters;
|
|
}
|
|
|
|
export function createDeviceTrafficService({
|
|
observe,
|
|
uploadChain,
|
|
downloadChain,
|
|
directChain,
|
|
directMark,
|
|
tproxyMark,
|
|
gatewayClientCidrs,
|
|
bypassCidrs,
|
|
proxyPort,
|
|
run = runCommand,
|
|
nextGeneration = () => crypto.randomUUID(),
|
|
}: {
|
|
observe: () => Promise<unknown> | unknown;
|
|
uploadChain: string;
|
|
downloadChain: string;
|
|
directChain: string;
|
|
directMark: string;
|
|
tproxyMark: string;
|
|
gatewayClientCidrs: string[];
|
|
bypassCidrs: string[];
|
|
proxyPort: number;
|
|
run?: RunCommand;
|
|
nextGeneration?: () => string;
|
|
}) {
|
|
const epoch = nextGeneration();
|
|
let activeSlot: 'A' | 'B' | null = null;
|
|
let activeDevices: TrafficDevice[] = [];
|
|
let activeSignature = '';
|
|
let activeCounters = new Map<string, string>();
|
|
let pendingRetired: RetiredCounters | null = null;
|
|
let refreshPromise: Promise<TrafficSnapshot> | null = null;
|
|
const finalized = new Map<string, CounterValues>();
|
|
const finalizedDirect = { upload: 0n, download: 0n };
|
|
const devicesByKey = new Map<string, TrafficDevice>();
|
|
let current: TrafficSnapshot = {
|
|
epoch,
|
|
generation: epoch,
|
|
observedAt: null,
|
|
source: { error: null },
|
|
direct: { uploadBytes: '0', downloadBytes: '0' },
|
|
devices: [],
|
|
};
|
|
|
|
async function execute(command: string, args: string[], options: CommandOptions = COMMAND_OPTIONS) {
|
|
const result = await run(command, args, options);
|
|
if (result.error || result.status !== 0) throw commandError(command, result);
|
|
return String(result.stdout || '');
|
|
}
|
|
|
|
async function prepare(slot: 'A' | 'B', devices: TrafficDevice[]) {
|
|
const input = buildTrafficRestore({
|
|
devices,
|
|
bypassCidrs,
|
|
uploadChain,
|
|
downloadChain,
|
|
directChain,
|
|
directMark,
|
|
tproxyMark,
|
|
gatewayClientCidrs,
|
|
slot,
|
|
proxyPort,
|
|
});
|
|
await execute('iptables-restore', ['-w', '1', '--noflush'], { ...COMMAND_OPTIONS, input });
|
|
}
|
|
|
|
async function switchTo(slot: 'A' | 'B') {
|
|
const uploadChild = childChain(uploadChain, slot);
|
|
const downloadChild = childChain(downloadChain, slot);
|
|
const directChild = childChain(directChain, slot);
|
|
const replace = activeSlot ? '-R' : '-A';
|
|
const uploadArgs = activeSlot
|
|
? ['-w', '1', '-t', 'raw', replace, uploadChain, '1', '-j', uploadChild]
|
|
: ['-w', '1', '-t', 'raw', replace, uploadChain, '-j', uploadChild];
|
|
const mangleInput = [
|
|
'*mangle',
|
|
activeSlot
|
|
? `-R ${downloadChain} 1 -j ${downloadChild}`
|
|
: `-A ${downloadChain} -j ${downloadChild}`,
|
|
activeSlot
|
|
? `-R ${directChain} 1 -j ${directChild}`
|
|
: `-A ${directChain} -j ${directChild}`,
|
|
'COMMIT',
|
|
'',
|
|
].join('\n');
|
|
|
|
await execute('iptables', uploadArgs);
|
|
try {
|
|
await execute('iptables-restore', ['-w', '1', '--noflush'], { ...COMMAND_OPTIONS, input: mangleInput });
|
|
} catch (error) {
|
|
const uploadRollback = activeSlot
|
|
? ['-w', '1', '-t', 'raw', '-R', uploadChain, '1', '-j', childChain(uploadChain, activeSlot)]
|
|
: ['-w', '1', '-t', 'raw', '-F', uploadChain];
|
|
try {
|
|
await execute('iptables', uploadRollback);
|
|
} catch (rollbackError) {
|
|
const original = error instanceof Error ? error.message : String(error);
|
|
const rollback = rollbackError instanceof Error ? rollbackError.message : String(rollbackError);
|
|
throw new Error(`${original}; rollback: ${rollback}`, { cause: error });
|
|
}
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function readCounters(devices: TrafficDevice[], slot: 'A' | 'B' | null): Promise<Map<string, string>> {
|
|
if (!slot) return new Map<string, string>();
|
|
const [raw, mangle] = await Promise.all([
|
|
execute('iptables-save', ['-c', '-t', 'raw']),
|
|
execute('iptables-save', ['-c', '-t', 'mangle']),
|
|
]);
|
|
const parsed = {
|
|
upload: parseTrafficCounters(raw, childChain(uploadChain, slot)),
|
|
download: parseTrafficCounters(mangle, childChain(downloadChain, slot)),
|
|
proxyUpload: parseTrafficCounters(raw, proxyChildChain(uploadChain, slot)),
|
|
proxyDownload: parseTrafficCounters(mangle, proxyChildChain(downloadChain, slot)),
|
|
directUpload: parseTrafficCounters(mangle, childChain(directChain, slot)),
|
|
directDownload: parseTrafficCounters(mangle, childChain(downloadChain, slot)),
|
|
};
|
|
const counters = new Map<string, string>();
|
|
for (const { key } of devices) {
|
|
for (const [kind, field] of COUNTERS) {
|
|
counters.set(`${key}:${kind}`, parsed[field].get(`${key}:${kind}`) || '0');
|
|
}
|
|
}
|
|
counters.set('global:direct-upload', parsed.directUpload.get('global:direct-upload') || '0');
|
|
counters.set('global:direct-download', parsed.directDownload.get('global:direct-download') || '0');
|
|
return counters;
|
|
}
|
|
|
|
function counter(counters: Map<string, string>, key: string, direction: CounterKind) {
|
|
return BigInt(counters.get(`${key}:${direction}`) || '0');
|
|
}
|
|
|
|
function remember(devices: TrafficDevice[]) {
|
|
for (const device of devices) devicesByKey.set(device.key, device);
|
|
}
|
|
|
|
async function finalizeRetired() {
|
|
if (!pendingRetired) return false;
|
|
const counters = await readCounters(pendingRetired.devices, pendingRetired.slot);
|
|
for (const { key } of pendingRetired.devices) {
|
|
const previous = finalized.get(key) || zeroCounters();
|
|
const next: CounterValues = { ...previous };
|
|
for (const [kind, field] of COUNTERS) next[field] += counter(counters, key, kind);
|
|
finalized.set(key, next);
|
|
}
|
|
finalizedDirect.upload += counter(counters, 'global', 'direct-upload');
|
|
finalizedDirect.download += counter(counters, 'global', 'direct-download');
|
|
pendingRetired = null;
|
|
return true;
|
|
}
|
|
|
|
function processTotals() {
|
|
const activeByMac = new Map(activeDevices.map((device) => [device.mac, device]));
|
|
const totalsByMac = new Map<string, TrafficDevice & CounterValues>();
|
|
for (const [key, remembered] of devicesByKey) {
|
|
const base = finalized.get(key) || zeroCounters();
|
|
const pending = pendingRetired?.counters || new Map();
|
|
const previous = totalsByMac.get(remembered.mac) || { ...remembered, ...zeroCounters() };
|
|
const total: TrafficDevice & CounterValues = {
|
|
...(activeByMac.get(remembered.mac) || remembered),
|
|
...zeroCounters(),
|
|
};
|
|
for (const [kind, field] of COUNTERS) {
|
|
total[field] = previous[field] + base[field]
|
|
+ counter(pending, key, kind) + counter(activeCounters, key, kind);
|
|
}
|
|
totalsByMac.set(remembered.mac, total);
|
|
}
|
|
return [...totalsByMac.values()]
|
|
.map((total) => {
|
|
const {
|
|
key: _key,
|
|
upload,
|
|
download,
|
|
proxyUpload,
|
|
proxyDownload,
|
|
directUpload,
|
|
directDownload,
|
|
...device
|
|
} = total;
|
|
return {
|
|
...device,
|
|
uploadBytes: upload.toString(),
|
|
downloadBytes: download.toString(),
|
|
proxyUploadBytes: proxyUpload.toString(),
|
|
proxyDownloadBytes: proxyDownload.toString(),
|
|
directUploadBytes: directUpload.toString(),
|
|
directDownloadBytes: directDownload.toString(),
|
|
};
|
|
})
|
|
.sort((left, right) => left.mac.localeCompare(right.mac));
|
|
}
|
|
|
|
function directTotals() {
|
|
const pending = pendingRetired?.counters || new Map();
|
|
return {
|
|
uploadBytes: (finalizedDirect.upload
|
|
+ counter(pending, 'global', 'direct-upload')
|
|
+ counter(activeCounters, 'global', 'direct-upload')).toString(),
|
|
downloadBytes: (finalizedDirect.download
|
|
+ counter(pending, 'global', 'direct-download')
|
|
+ counter(activeCounters, 'global', 'direct-download')).toString(),
|
|
};
|
|
}
|
|
|
|
async function performRefresh() {
|
|
let observed: Record<string, unknown>;
|
|
try {
|
|
observed = record(await observe());
|
|
} catch (error) {
|
|
observed = { observedAt: new Date().toISOString(), observations: [], error: error instanceof Error ? error.message : String(error) };
|
|
}
|
|
|
|
let sourceError = observed.error ? String(observed.error) : null;
|
|
const nextDevices = sourceError
|
|
? activeDevices
|
|
: selectTrafficDevices(observed?.observations);
|
|
const nextSignature = JSON.stringify(nextDevices);
|
|
let countersRead = false;
|
|
|
|
if (pendingRetired) {
|
|
try {
|
|
countersRead = await finalizeRetired() || countersRead;
|
|
} catch (error) {
|
|
sourceError = sourceError || (error instanceof Error ? error.message : String(error));
|
|
}
|
|
}
|
|
|
|
if (!pendingRetired && !sourceError && nextSignature !== activeSignature) {
|
|
const nextSlot = activeSlot === 'A' ? 'B' : 'A';
|
|
try {
|
|
await prepare(nextSlot, nextDevices);
|
|
await switchTo(nextSlot);
|
|
const retired = activeSlot ? {
|
|
slot: activeSlot,
|
|
devices: activeDevices,
|
|
counters: activeCounters,
|
|
} : null;
|
|
activeSlot = nextSlot;
|
|
activeDevices = nextDevices;
|
|
activeSignature = nextSignature;
|
|
activeCounters = new Map();
|
|
pendingRetired = retired;
|
|
remember(nextDevices);
|
|
current.generation = nextGeneration();
|
|
if (pendingRetired) countersRead = await finalizeRetired() || countersRead;
|
|
} catch (error) {
|
|
sourceError = error instanceof Error ? error.message : String(error);
|
|
}
|
|
}
|
|
|
|
if (activeSlot) {
|
|
try {
|
|
activeCounters = await readCounters(activeDevices, activeSlot);
|
|
countersRead = true;
|
|
} catch (error) {
|
|
sourceError = sourceError || (error instanceof Error ? error.message : String(error));
|
|
}
|
|
}
|
|
current = {
|
|
epoch,
|
|
generation: current.generation,
|
|
observedAt: countersRead && typeof observed.observedAt === 'string'
|
|
? observed.observedAt
|
|
: current.observedAt,
|
|
source: { error: sourceError },
|
|
direct: countersRead ? directTotals() : current.direct,
|
|
devices: countersRead ? processTotals() : current.devices,
|
|
};
|
|
return structuredClone(current);
|
|
}
|
|
|
|
function refresh() {
|
|
if (!refreshPromise) {
|
|
refreshPromise = performRefresh().finally(() => {
|
|
refreshPromise = null;
|
|
});
|
|
}
|
|
return refreshPromise;
|
|
}
|
|
|
|
return {
|
|
refresh,
|
|
snapshot: () => structuredClone(current),
|
|
};
|
|
}
|