1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
// Run the SHIPPED `pipelinedDownload` over the SHIPPED `MeshBayTransport.lanes`
// against a fake clock and a fake node, and see which connection carried which
// chunk.
//
// What is modelled is the node and the network: `connect` (where an offer is
// made) and `fetchChunk` (a megabyte that takes time). The lanes, the pipeline,
// the retry path and the idle close are the shipped code, the first by
// executing transport.js whole and the second lifted out of file-utils.js as
// text.
//
// Usage: node download_lanes_harness.mjs <file-utils.js> <transport sources> <json>
import { readFileSync } from 'fs';
import { delimiter } from 'path';
const utils = readFileSync(process.argv[2], 'utf8');
const transportSrc = process.argv[3].split(delimiter)
.map((p) => readFileSync(p, 'utf8')).join('\n');
const {
chunks = 200,
tr = 'lease',
connectMs = 800,
chunkMs = 150,
refuseLanes = false, // every lane's connect fails
failLaneAfter = -1, // the first lane to open fails every request after this many
closeAt = -1, // close the main connection at this time
} = JSON.parse(process.argv[4] || '{}');
for (const level of ['log', 'warn', 'error', 'info', 'debug']) {
console[level] = (...args) => process.stderr.write(args.join(' ') + '\n');
}
// A download that fails leaves the chunks still in flight to reject with nobody
// waiting on them, which a browser logs and Node would die of.
process.on('unhandledRejection', () => {});
// ── The clock ────────────────────────────────────────────────────────────────
let now = 0;
let nextId = 1;
let timers = [];
globalThis.setTimeout = (fn, ms) => {
const t = { at: now + (ms || 0), fn, id: nextId++ };
timers.push(t);
return t.id;
};
globalThis.setInterval = (fn, ms) => {
const t = { at: now + ms, fn, id: nextId++, every: ms };
timers.push(t);
return t.id;
};
globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
globalThis.clearInterval = globalThis.clearTimeout;
Date.now = () => now;
const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
async function run(until) {
await flush();
while (timers.length) {
const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
if (due.at > until) break;
if (due.every) due.at += due.every;
else timers = timers.filter((t) => t !== due);
now = Math.max(now, due.every ? due.at - due.every : due.at);
due.fn();
await flush();
}
now = until;
}
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
// ── The shipped transport, with a node behind it ─────────────────────────────
globalThis.window = globalThis;
globalThis.addEventListener = () => {};
globalThis.removeEventListener = () => {};
globalThis.location = { hash: '' };
globalThis.document = {
addEventListener() {}, removeEventListener() {}, visibilityState: 'visible',
};
new Function(transportSrc)();
const Transport = window.MeshBayTransport;
let offers = 0;
let inFlight = 0;
let peakInFlight = 0;
let nextName = 1;
const servedBy = []; // chunk index -> the connections asked for it, in order
const connectArgs = [];
const opened = [];
Transport.prototype.connect = async function connect(...args) {
offers += 1;
this.name = `lane${nextName++}`;
connectArgs.push(args);
opened.push(this);
await sleep(connectMs);
if (refuseLanes) throw new Error('Node is at its peer-connection limit');
this._connected = true;
this.asked = 0;
};
Transport.prototype.fetchChunk = async function fetchChunk(fileId, index, lease) {
(servedBy[index] = servedBy[index] || []).push(this.name);
this.asked = (this.asked || 0) + 1;
inFlight += 1;
peakInFlight = Math.max(peakInFlight, inFlight);
try {
await sleep(chunkMs);
if (!this._connected) throw new Error('Transport closed');
if (this.name === 'lane1' && failLaneAfter >= 0 && this.asked > failLaneAfter) {
throw new Error('Response timeout');
}
return { ct: new Uint8Array([index & 255]), nonce: index, tr: lease };
} finally {
inFlight -= 1;
}
};
const main = new Transport('https://hub.invalid', 'token');
main.name = 'main';
main._connected = true;
main._lastToken = 'token';
main._gekRaw = new Uint8Array(32);
main._identity = { the: 'identity' };
main._connectArgs = {
nodeId: 'node', groupId: 'group', bundleKey: 'bundle', username: 'u',
userId: 'uid', joinCode: 'never-again', joinNodePk: '', nodePk: 'pk',
};
// ── The shipped pipeline ─────────────────────────────────────────────────────
const constant = (name) => utils.match(new RegExp(`^const ${name} = [^;]+;`, 'm'))[0];
const from = utils.indexOf('const CHUNK_RETRY_ATTEMPTS');
const to = utils.indexOf('async function downloadEntry');
const { pipelinedDownload } = new Function('window', 't', `
${constant('PIPELINE_WINDOW')}
${constant('LANE_MIN_CHUNKS')}
${utils.slice(from, utils.lastIndexOf('/**', to))}
return { pipelinedDownload };`)(
{ MeshBayCrypto: { decryptChunkBin: async (_k, _f, index) => new Uint8Array([index & 255]) } },
(key) => key);
const written = [];
const writable = { write: async (bytes) => { written.push(bytes[0]); } };
let outcome = 'pending';
pipelinedDownload(main, 'gek', 'f'.repeat(64), chunks, null, writable, null, tr)
.then(() => { outcome = 'done'; }, (err) => { outcome = `failed: ${err.message}`; });
let lanesLeftOpenByClose = null;
if (closeAt >= 0) {
await run(closeAt);
main.close();
await run(closeAt + 2 * connectMs);
lanesLeftOpenByClose = opened.filter((lane) => !lane._closed).length;
}
// Long enough for any download here, then past the idle close.
await run(240000);
const inOrder = written.every((b, i) => b === (i & 255));
const carried = {};
for (const names of servedBy) {
const last = names[names.length - 1];
carried[last] = (carried[last] || 0) + 1;
}
process.stdout.write(JSON.stringify({
outcome, offers, peakInFlight, carried,
written: written.length, inOrder,
retried: servedBy.filter((names) => names.length > 1).length,
lanesAfterIdle: main._lanes.length,
// What a lane was connected with: the group, the identity and the key of the
// main connection, and never its join code.
laneArgs: connectArgs.length ? {
groupId: connectArgs[0][2], sameGek: connectArgs[0][3] === main._gekRaw,
sameIdentity: connectArgs[0][4] === main._identity,
joinCode: connectArgs[0][8] === undefined ? null : connectArgs[0][8],
nodePk: connectArgs[0][11],
} : null,
lanesLeftOpenByClose,
lanesLeftOpen: opened.filter((lane) => !lane._closed).length,
timersLeft: timers.length,
}));
|