fix: reconcile share-link token allocation with main

Preserve all published token assignments and add regression coverage for
all-enabled URL restoration, longer token lists, and manual provisional
token recovery after competing merges.

Builds on daikaginza's digits-first and merge-time allocation proposal.
Thanks to KnottyDyes for the review and regression-test suggestions.
This commit is contained in:
manjunath22466 committed 2026-09-25 22:33:03 +05:30
commit d0105f1e9d
27 files changed
+1440 -39

No files matched your search

+116 -1
View File
@@ -8,8 +8,9 @@ import { test } from 'node:test';
import assert from 'node:assert/strict';
import http from 'node:http';
import net from 'node:net';
import { Readable } from 'node:stream';
import { Readable, Writable } from 'node:stream';
import { cctvProxy } from '../../server/providers/cctv.js';
import { proxyMediaResponse } from '../../server/providers/cctv/media.js';
import { sanitizeCctvRangeHeader } from '../../server/providers/cctv/range.js';
import { CCTV_MEDIA_MAX_BODY_BYTES as CAP } from '../../server/providers/cctv/constants.js';
@@ -586,3 +587,117 @@ test('two sequential seeks are served as two partial responses', async (t) => {
assert.equal(first.body, 'chunk-0');
assert.equal(second.body, 'chunk-100');
});
// ---------------------------------------------------------------------------
// The body idle deadline. proxyMediaResponse is driven directly here because
// the deadline is only injectable through its options; the route always passes
// the production value.
// ---------------------------------------------------------------------------
/**
* A response to pipe into, built on a real Writable so the backpressure the
* proxy reads is the stream's own rather than a flag the test sets. With
* `drains: false` the bytes are accepted and never flushed, which is what a
* viewer on a slow link looks like from the server side.
*/
function pipeTarget({ drains = true } = {}) {
const res = new Writable({
// One byte is enough to leave a target that never empties its buffer
// permanently in need of a drain.
highWaterMark: drains ? undefined : 1,
write(_chunk, _encoding, callback) {
if (drains) callback();
},
});
res.writeHead = (status, headers) => {
res.statusCode = status;
res.headers = headers || {};
};
return res;
}
/**
* An upstream that answers with headers, sends one chunk, and then either
* keeps producing every `everyMs` or goes silent. `released` reports the body's
* own cleanup: a cancelled fetch body lands in the web stream's cancel hook.
*/
function mediaUpstream({ everyMs = 0 } = {}) {
const state = { produced: 0, released: false };
let ticker = null;
const body = new ReadableStream({
start(controller) {
const send = () => {
state.produced += 1;
controller.enqueue(new Uint8Array([0xff, 0xd8, 0xff, 0xe0]));
};
send();
if (everyMs > 0) {
ticker = setInterval(send, everyMs);
ticker.unref();
}
},
cancel() {
state.released = true;
if (ticker) clearInterval(ticker);
},
});
state.upstream = {
ok: true,
status: 200,
headers: {
get: (name) =>
String(name).toLowerCase() === 'content-type' ? 'video/mp4' : null,
},
body,
arrayBuffer: async () => Buffer.alloc(0),
};
return state;
}
test('an upstream that goes silent after its headers is released at the idle deadline', async (t) => {
const feed = mediaUpstream();
const res = pipeTarget();
t.after(() => res.destroy());
await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 40 });
await new Promise((resolve) => setTimeout(resolve, 20));
assert.equal(feed.released, false, 'released before the deadline was due');
await new Promise((resolve) => setTimeout(resolve, 120));
// Without the deadline this stream is held open against the camera host for
// as long as that host will keep the socket, with no bytes ever arriving.
assert.equal(feed.released, true, 'the silent upstream was never released');
assert.equal(res.writableEnded, true, 'the client was left waiting');
});
test('a live feed that keeps producing is not cut off by the idle deadline', async (t) => {
const feed = mediaUpstream({ everyMs: 10 });
const res = pipeTarget();
t.after(() => res.destroy());
await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 40 });
await new Promise((resolve) => setTimeout(resolve, 250));
assert.equal(feed.released, false, 'a healthy feed was torn down');
assert.equal(res.writableEnded, false, 'a healthy feed was ended early');
assert.ok(
feed.produced > 5,
`the feed stopped producing (${feed.produced} chunks)`,
);
});
test('a client that cannot keep up is not mistaken for a stalled upstream', async (t) => {
// Nothing arrives from upstream while the pipe is paused, which looks exactly
// like silence unless the response is asked whether it is still draining.
const feed = mediaUpstream({ everyMs: 10 });
const res = pipeTarget({ drains: false });
t.after(() => res.destroy());
await proxyMediaResponse(res, feed.upstream, { idleTimeoutMs: 30 });
await new Promise((resolve) => setTimeout(resolve, 200));
assert.equal(
res.writableNeedDrain,
true,
'the client was not the bottleneck',
);
assert.equal(feed.released, false, 'a slow viewer tore down a healthy feed');
assert.equal(res.writableEnded, false, 'a slow viewer ended the response');
});
+19
View File
@@ -456,6 +456,25 @@ test('v2 codec distinguishes absent from empty and keeps canonical deterministic
);
});
test('all production layers and options round-trip through a v2 share URL', () => {
const allEnabled = normalizeLayerState({
enabledLayerIds: REGISTERED_LAYER_IDS,
options: {
cctv: { coverageMode: 'viewshed', showProjection: false, autoHop: true },
flights: { models3d: true, models3dMode: 'all' },
radio: { filter: 'news', volume: 0.37 },
},
});
const shareUrl = new URL('https://example.invalid/');
shareUrl.hash = encode(allEnabled);
const params = new URLSearchParams(shareUrl.hash.slice(1));
const restored = decodeLayerStateParams(params);
assert.equal(params.get('l')?.split('.').length, REGISTERED_LAYER_IDS.length);
assert.deepEqual(restored?.enabledLayerIds, REGISTERED_LAYER_IDS);
assert.deepEqual(restored?.options, allEnabled.options);
});
test('unknown enabled-layer tokens reject the payload instead of becoming an empty set', () => {
assert.equal(
decodeLayerStateParams(new URLSearchParams('v=2&l=unknown')),
+148 -1
View File
@@ -4,12 +4,13 @@ import {
mkdtempSync,
mkdirSync,
readFileSync,
realpathSync,
rmSync,
writeFileSync,
} from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { fileURLToPath, pathToFileURL } from 'node:url';
import test from 'node:test';
import {
LEGACY_LAYER_STATE_TOKENS,
@@ -27,9 +28,15 @@ import {
const LEDGER_PATH = 'src/data/layerStateTokenReservations.json';
const SOURCE_PATH = 'src/data/layerState.js';
const CODEC_SOURCE_PATH = fileURLToPath(
new URL('./layerState.js', import.meta.url),
);
const CHECKER_PATH = fileURLToPath(
new URL('../../scripts/check-layer-state-tokens.mjs', import.meta.url),
);
const NEXT_PATH = fileURLToPath(
new URL('../../scripts/next-layer-state-token.mjs', import.meta.url),
);
const reservationRows = JSON.parse(
readFileSync(
new URL('./layerStateTokenReservations.json', import.meta.url),
@@ -90,6 +97,26 @@ function withPublishedBase(
}
}
function writeCodecFixture(cwd, rows, entries) {
const source = readFileSync(CODEC_SOURCE_PATH, 'utf8');
const registryEnd = ']);\n\nexport const REGISTERED_LAYER_IDS';
assert.equal(source.split(registryEnd).length, 2);
const additions = entries
.map(
({ id, token }) =>
` Object.freeze({ id: '${id}', token: '${token}', disposition: 'enabled-only' }),`,
)
.join('\n');
mkdirSync(path.join(cwd, 'src/data'), { recursive: true });
writeFileSync(path.join(cwd, 'package.json'), '{"type":"module"}');
writeFileSync(
path.join(cwd, SOURCE_PATH),
source.replace(registryEnd, `${additions}\n${registryEnd}`),
);
writeFileSync(path.join(cwd, LEDGER_PATH), JSON.stringify(rows));
return path.join(cwd, SOURCE_PATH);
}
test('reservation ledger is complete, pinned, and rejects duplicate or malformed rows', () => {
assert.deepEqual(
{ ...parseLayerStateTokenReservations(reservationRows) },
@@ -169,6 +196,77 @@ test('published retired digits and competing PRs advance the merge-time allocati
);
});
test('PR B manually replaces provisional 0 with 3 after PR A publishes 0', () => {
withPublishedBase({ rows: [...reservationRows, ['pr-a', '0']] }, (cwd) => {
mkdirSync(path.join(cwd, 'scripts'));
const fixtureChecker = path.join(
cwd,
'scripts/check-layer-state-tokens.mjs',
);
writeFileSync(fixtureChecker, readFileSync(CHECKER_PATH, 'utf8'));
const fixtureNext = path.join(cwd, 'scripts/next-layer-state-token.mjs');
writeFileSync(fixtureNext, readFileSync(NEXT_PATH, 'utf8'));
const runCheck = () =>
spawnSync(
process.execPath,
[realpathSync(fixtureChecker), '--base-ref', 'HEAD'],
{
cwd,
encoding: 'utf8',
},
);
writeCodecFixture(
cwd,
[...reservationRows, ['pr-b', '0']],
[{ id: 'pr-b', token: '0' }],
);
const beforeRebase = runCheck();
assert.equal(beforeRebase.status, 1);
assert.match(beforeRebase.stderr, /changed or removed: pr-a/);
writeCodecFixture(
cwd,
[...reservationRows, ['pr-a', '0'], ['pr-b', '0']],
[
{ id: 'pr-a', token: '0' },
{ id: 'pr-b', token: '0' },
],
);
const staleRebase = runCheck();
assert.equal(staleRebase.status, 1);
assert.match(staleRebase.stderr, /Duplicate layer-state token reservation/);
const published = readPublishedLayerStateReservations('HEAD', cwd);
assert.equal(nextLayerStateToken(published), '3');
const validBaselineRows = [...reservationRows, ['pr-a', '0']];
writeCodecFixture(cwd, validBaselineRows, [{ id: 'pr-a', token: '0' }]);
const helper = spawnSync(
process.execPath,
[realpathSync(fixtureNext), 'pr-b'],
{ cwd, encoding: 'utf8' },
);
assert.equal(helper.status, 0, helper.stderr);
assert.match(helper.stdout, /pr-b: 3/);
assert.deepEqual(
JSON.parse(readFileSync(path.join(cwd, LEDGER_PATH), 'utf8')),
validBaselineRows,
'the helper reports a token without rewriting the ledger',
);
writeCodecFixture(
cwd,
[...reservationRows, ['pr-a', '0'], ['pr-b', '3']],
[
{ id: 'pr-a', token: '0' },
{ id: 'pr-b', token: '3' },
],
);
const corrected = runCheck();
assert.equal(corrected.status, 0, corrected.stderr);
assert.match(corrected.stdout, /29 published, 1 new/);
});
});
test('allocation batches cross the last digit and base-36 pair boundaries in order', () => {
const digitRows = [...'03456789'].map((digit) => [`retired-${digit}`, digit]);
const allDigits = parseLayerStateTokenReservations([
@@ -219,6 +317,55 @@ test('allocation batches cross the last digit and base-36 pair boundaries in ord
);
});
test('an isolated valid two-character fixture round-trips an l field beyond the old 64-character cap', async () => {
const cwd = mkdtempSync(path.join(tmpdir(), 'gev-layer-token-width-'));
try {
const priorDigits = [...'03456789'].map((digit) => [
`qa-prior-digit-${digit}`,
digit,
]);
const fixtureLayers = ['00', '01', '02', '03'].map((token) => ({
id: `qa-layer-${token}`,
token,
}));
const fixtureRows = [
...reservationRows,
...priorDigits,
...fixtureLayers.map(({ id, token }) => [id, token]),
];
const fixtureModule = writeCodecFixture(cwd, fixtureRows, fixtureLayers);
const codec = await import(pathToFileURL(fixtureModule));
assert.equal(codec.validateLayerStateRegistry(), true);
assert.equal(
nextLayerStateToken(
parseLayerStateTokenReservations([...reservationRows, ...priorDigits]),
),
'00',
);
assert.equal(
validateLayerStateAllocations(
LAYER_STATE_TOKEN_RESERVATIONS,
codec.LAYER_STATE_TOKEN_RESERVATIONS,
),
true,
);
const expectedLayerIds = codec.REGISTERED_LAYER_IDS;
const state = codec.normalizeLayerState({
enabledLayerIds: expectedLayerIds,
});
const params = new URLSearchParams([['v', '2']]);
codec.encodeLayerStateParams(params, state);
assert.ok(params.get('l').length > 64, 'fixture must cross the old cap');
assert.equal(params.get('l').length, 67);
const restored = codec.decodeLayerStateParams(params);
assert.deepEqual(restored?.enabledLayerIds, expectedLayerIds);
assert.deepEqual(restored?.options, state.options);
} finally {
rmSync(cwd, { recursive: true, force: true });
}
});
test('checker reads a complete future base ledger and rejects retired-token reuse', () => {
withPublishedBase(
{