comunes/todos-contra-el-fuego-web
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test(subsUnion): cover the queue, the fast path and the size guard

Browse source 
The cases are the incidents: a worker failure must leave the stored union alone
instead of overwriting it with "null" (50ca9cc), an incremental merge must refuse
to run on state that is not exactly one behind, recomputes must coalesce while
adds must not be dropped, and legacy telegram subscriptions (no radius, no
location) must be skipped without throwing.

Also the 16 MiB case the FIXME asked for: a 20 MB union comes back under the cap,
still parseable, with the degradation recorded.
This commit is contained in:
vjrj 2026-08-01 18:02:20 +02:00
parent 124ed4e135
commit 6e448bda55
2 changed files with 605 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

485
test/server/subsUnion.test.js Normal file
View file

@ -0,0 +1,485 @@
/* eslint-env mocha */
/* eslint-disable func-names, prefer-arrow-callback */
/* eslint-disable import/no-absolute-path */
import { expect } from 'chai';
import Subscriptions from '/imports/api/Subscriptions/Subscriptions';
import SiteSettings from '/imports/api/SiteSettings/SiteSettings';
import createSubsUnion, { isValidSub } from '/imports/startup/server/subsUnionLogic';
// Everything here runs against the real collections with a fake calcUnion: what
// broke in production was never the geometry, it was the orchestration around it
// (a failed worker stored as an empty union, a recompute that never got
// serialized, an incremental merge on top of stale state).
const quietLog = {
log: () => {}, info: () => {}, warn: () => {}, error: () => {}
};
const sub = (lat, lon, distance = 10, extra = {}) => ({
location: { lat, lon },
geo: { type: 'Point', coordinates: [lon, lat] },
distance,
owner: 'test-owner',
type: 'web',
...extra
});
// A union whose "shape" is just the list of circle centres that went into it, so
// tests can assert what was merged without doing any geometry.
const fakeUnionOf = (subs, baseUnion = null) => {
const centres = (baseUnion ? baseUnion.centres : []).concat(
subs.map(s => [Math.round(s.location.lon * 100) / 100, Math.round(s.location.lat * 100) / 100])
);
return {
type: 'Feature',
centres,
geometry: { type: 'Polygon', coordinates: [[[0, 0], [0, 1], [1, 1], [0, 0]]] }
};
};
const cleanUp = async () => {
await Subscriptions.removeAsync({});
await SiteSettings.removeAsync({ name: { $in: ['subs-public-union', 'subs-private-union', 'subs-public-union-bounds', 'subs-private-union-bounds', 'subs-union-count'] } });
};
const storedUnion = async (isPublic = true) => {
const doc = await SiteSettings.findOneAsync({ name: `subs-${isPublic ? 'public' : 'private'}-union` });
return doc ? { doc, value: JSON.parse(doc.value) } : null;
};
const waitFor = async (condition, what, timeout = 10000) => {
const deadline = Date.now() + timeout;
while (Date.now() < deadline) {
// eslint-disable-next-line no-await-in-loop
if (await condition()) return;
// eslint-disable-next-line no-await-in-loop
await new Promise(resolve => setTimeout(resolve, 20));
}
throw new Error(`Timed out waiting for ${what}`);
};
// A calcUnion double: records its calls, and can be made to fail or to hang until
// released, which is how the queue is observed from the outside.
const makeCalc = ({ fail = false, gate = null } = {}) => {
let currentGate = gate;
const calls = [];
const calc = async (subs, baseUnion = null) => {
calls.push({ subs, baseUnion });
if (currentGate) await currentGate.promise;
if (fail) throw new Error('worker exploded');
return fakeUnionOf(subs, baseUnion);
};
calc.calls = calls;
// Lets a test set up its baseline union at full speed and only then start
// holding calls back.
calc.setGate = (g) => { currentGate = g; };
return calc;
};
const makeGate = () => {
let release;
const promise = new Promise((resolve) => { release = resolve; });
return { promise, release: () => release() };
};
const build = (calcUnion, overrides = {}) => createSubsUnion({
calcUnion,
SiteSettings,
Subscriptions,
perlinSeed: 0.42,
debug: false,
log: quietLog,
...overrides
});
describe('subsUnion isValidSub', function () {
it('accepts a well-formed web subscription', function () {
expect(isValidSub(sub(43.5, -5.9), quietLog)).to.equal(true);
});
// Legacy telegram subscriptions predate the web ones and can be missing the
// radius or the location entirely. They must be skipped, never throw.
it('skips legacy telegram subscriptions instead of throwing', function () {
const legacy = [
{ chatId: -253600015, location: { lat: 40.2, lon: -3.3 } }, // no distance
{ chatId: -253600016, distance: 40 }, // no location
{ chatId: -253600017, location: {}, distance: 40 },
undefined,
null
];
legacy.forEach((l) => {
expect(() => isValidSub(l, quietLog)).to.not.throw();
expect(isValidSub(l, quietLog)).to.equal(false);
});
});
});
describe('subsUnion recreate', function () {
this.timeout(20000);
beforeEach(cleanUp);
after(cleanUp);
it('computes both unions from every valid subscription and stores the count', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
await Subscriptions.insertAsync(sub(40.4, -3.7));
const calc = makeCalc();
await build(calc).recreate();
expect(calc.calls).to.have.lengthOf(2); // public + private
expect(calc.calls[0].baseUnion).to.equal(null);
expect(calc.calls[0].subs).to.have.lengthOf(2);
const publicUnion = await storedUnion(true);
const privateUnion = await storedUnion(false);
expect(publicUnion.value.centres).to.have.lengthOf(2);
expect(privateUnion.value.centres).to.have.lengthOf(2);
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(2);
});
it('fuzzes the public union but never the private one', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
const calc = makeCalc();
await build(calc).recreate();
const [publicCall, privateCall] = calc.calls;
expect(privateCall.subs[0].location.lat).to.equal(43.5);
expect(privateCall.subs[0].location.lon).to.equal(-5.9);
expect(publicCall.subs[0].location.lat).to.not.equal(43.5);
});
it('does not let the public pass leak its noise into the private one', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9, 10));
const calc = makeCalc();
await build(calc).recreate();
const [publicCall, privateCall] = calc.calls;
expect(publicCall.subs[0].distance).to.not.equal(10);
expect(privateCall.subs[0].distance).to.equal(10);
});
it('ignores legacy telegram subscriptions without failing the union', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
// Bypasses the schema on purpose: this is what is actually in the production
// collection, written by the telegram bot years before the web schema existed.
await Subscriptions.rawCollection().insertOne({ chatId: -253600015, location: { lat: 40.2, lon: -3.3 } });
const calc = makeCalc();
await build(calc).recreate();
expect(calc.calls[0].subs).to.have.lengthOf(1);
// The count is of subscriptions, not of drawable ones — unchanged behaviour.
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(2);
});
// Regression 50ca9cc: a failed worker call left `union` as null, and the store
// guard (`typeof union === 'object'`, true for null) happily wrote "null" over
// the good union. The map then went blank until the next successful recompute.
it('leaves the stored union untouched when the worker fails', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
await build(makeCalc()).recreate();
const before = await storedUnion(true);
expect(before.value.centres).to.have.lengthOf(1);
await Subscriptions.insertAsync(sub(40.4, -3.7));
await build(makeCalc({ fail: true })).recreate();
const after = await storedUnion(true);
expect(after.doc.value).to.equal(before.doc.value);
expect(after.doc.value).to.not.equal('null');
});
it('stores a real empty union when there are no subscriptions at all', async function () {
const calc = async () => null;
await build(calc).recreate();
const stored = await storedUnion(true);
expect(stored.doc.value).to.equal('null');
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(0);
});
});
describe('subsUnion incrementalAdd', function () {
this.timeout(20000);
beforeEach(cleanUp);
after(cleanUp);
it('merges into the stored union instead of recomputing everything', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
const calc = makeCalc();
const subsUnion = build(calc);
await subsUnion.recreate();
calc.calls.length = 0;
const doc = sub(40.4, -3.7);
await Subscriptions.insertAsync(doc);
await subsUnion.incrementalAdd(doc);
// One call per union kind, each with only the new circle and the stored union
// as a base — not the full list of subscriptions again.
expect(calc.calls).to.have.lengthOf(2);
expect(calc.calls[0].subs).to.have.lengthOf(1);
expect(calc.calls[0].baseUnion).to.not.equal(null);
expect(calc.calls[0].baseUnion.centres).to.have.lengthOf(1);
const stored = await storedUnion(true);
expect(stored.value.centres).to.have.lengthOf(2);
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(2);
});
it('falls back to a full recreate when the stored state is not exactly one behind', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
const calc = makeCalc();
const subsUnion = build(calc);
await subsUnion.recreate();
calc.calls.length = 0;
// Two new subscriptions, only one of them announced: the stored count is now
// two behind, so merging would silently lose one.
await Subscriptions.insertAsync(sub(40.4, -3.7));
const doc = sub(41.6, -0.9);
await Subscriptions.insertAsync(doc);
await subsUnion.incrementalAdd(doc);
expect(calc.calls).to.have.lengthOf(0);
expect(subsUnion.stats().recomputePending).to.equal(true);
});
it('falls back to a full recreate when there is no stored union yet', async function () {
const doc = sub(43.5, -5.9);
await Subscriptions.insertAsync(doc);
const calc = makeCalc();
const subsUnion = build(calc);
await subsUnion.incrementalAdd(doc);
expect(calc.calls).to.have.lengthOf(0);
expect(subsUnion.stats().recomputePending).to.equal(true);
});
it('skips a legacy subscription without touching the union', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
const calc = makeCalc();
const subsUnion = build(calc);
await subsUnion.recreate();
const before = await storedUnion(true);
calc.calls.length = 0;
await subsUnion.incrementalAdd({ chatId: -253600015, location: { lat: 40.2, lon: -3.3 } });
expect(calc.calls).to.have.lengthOf(0);
expect(subsUnion.stats().recomputePending).to.equal(false);
expect((await storedUnion(true)).doc.value).to.equal(before.doc.value);
});
// Regression 50ca9cc, incremental half: a failed merge must not store anything
// and must ask for a full recompute rather than leaving the union half-updated
// (public merged, private not).
it('asks for a full recreate when the merge fails, without storing a partial union', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
await build(makeCalc()).recreate();
const before = await storedUnion(true);
const doc = sub(40.4, -3.7);
await Subscriptions.insertAsync(doc);
const subsUnion = build(makeCalc({ fail: true }));
await subsUnion.incrementalAdd(doc);
expect((await storedUnion(true)).doc.value).to.equal(before.doc.value);
expect(subsUnion.stats().recomputePending).to.equal(true);
});
});
describe('subsUnion queue', function () {
this.timeout(20000);
beforeEach(cleanUp);
after(cleanUp);
it('serializes adds: the second one sees what the first one stored', async function () {
const calc = makeCalc();
const subsUnion = build(calc);
// Stored state must start exactly one behind for the fast path to engage.
await subsUnion.recreate();
const gate = makeGate();
calc.setGate(gate);
calc.calls.length = 0;
const first = sub(43.5, -5.9);
await Subscriptions.insertAsync(first);
subsUnion.enqueueAdd(first);
// Only add the second subscription once the first job is already past its
// "am I exactly one behind?" check and waiting on the gate, so the test is
// about the queue and not about who won a race with Mongo.
await waitFor(async () => calc.calls.length > 0, 'the first add to reach the worker');
const second = sub(40.4, -3.7);
await Subscriptions.insertAsync(second);
subsUnion.enqueueAdd(second);
expect(subsUnion.stats().queued).to.equal(1);
expect(calc.calls).to.have.lengthOf(1);
gate.release();
await subsUnion.idle();
const stored = await storedUnion(true);
expect(stored.value.centres).to.have.lengthOf(2);
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(2);
});
it('never runs two jobs at once', async function () {
const gate = makeGate();
const calc = makeCalc({ gate });
const subsUnion = build(calc);
subsUnion.scheduleRecreate();
subsUnion.scheduleRecreate();
subsUnion.scheduleRecreate();
await waitFor(async () => calc.calls.length > 0, 'the first union call');
expect(calc.calls).to.have.lengthOf(1);
expect(subsUnion.stats().busy).to.equal(true);
gate.release();
await subsUnion.idle();
});
// A burst of changed/removed events only needs one pass once things settle:
// recomputes coalesce, adds do not.
it('coalesces a burst of recomputes but drops no add', async function () {
const gate = makeGate();
const calc = makeCalc({ gate });
const subsUnion = build(calc);
subsUnion.scheduleRecreate();
await waitFor(async () => calc.calls.length > 0, 'the running recompute');
subsUnion.scheduleRecreate();
subsUnion.scheduleRecreate();
subsUnion.scheduleRecreate();
const a = sub(43.5, -5.9);
const b = sub(40.4, -3.7);
subsUnion.enqueueAdd(a);
subsUnion.enqueueAdd(b);
expect(subsUnion.stats().queued).to.equal(2);
expect(subsUnion.stats().recomputePending).to.equal(true);
gate.release();
await subsUnion.idle();
// 2 calls for the running recompute + 2 per add (public/private) + 2 for the
// single coalesced recompute the burst asked for.
expect(calc.calls.length).to.be.at.most(8);
expect(subsUnion.stats().queued).to.equal(0);
expect(subsUnion.stats().recomputePending).to.equal(false);
});
it('keeps draining after a job throws', async function () {
const calls = [];
let shouldFail = true;
const calc = async (subs, baseUnion) => {
calls.push(subs);
if (shouldFail) { shouldFail = false; throw new Error('worker exploded'); }
return fakeUnionOf(subs, baseUnion);
};
const subsUnion = build(calc);
subsUnion.scheduleRecreate();
await subsUnion.idle();
expect(calls.length).to.be.above(0);
await Subscriptions.insertAsync(sub(43.5, -5.9));
subsUnion.scheduleRecreate();
await subsUnion.idle();
const stored = await storedUnion(true);
expect(stored.value.centres).to.have.lengthOf(1);
});
});
describe('subsUnion full flow through the collection observers', function () {
this.timeout(30000);
beforeEach(cleanUp);
after(cleanUp);
// Same wiring as imports/startup/server/subsUnion.js, which cannot be imported
// here (it is a Meteor.startup callback tied to the mail-master check).
const wire = async (subsUnion) => {
const added = await Subscriptions.find({ createdAt: { $gt: new Date() } }).observeAsync({
added: doc => subsUnion.enqueueAdd(doc)
});
const rest = await Subscriptions.find().observeAsync({
changed: () => subsUnion.scheduleRecreate(),
removed: () => subsUnion.scheduleRecreate()
});
return () => { added.stop(); rest.stop(); };
};
it('updates the union on insert and recalculates it on removal', async function () {
const calc = makeCalc();
const subsUnion = build(calc);
await subsUnion.recreate(); // baseline: count 0, union null
const stop = await wire(subsUnion);
try {
const id = await Subscriptions.insertAsync(sub(43.5, -5.9));
await waitFor(async () => {
const stored = await storedUnion(true);
return stored && stored.value && stored.value.centres.length === 1;
}, 'the union to pick up the new subscription');
await Subscriptions.removeAsync(id);
await waitFor(async () => {
const stored = await storedUnion(true);
return stored && stored.value && stored.value.centres.length === 0;
}, 'the union to be recalculated after the removal');
const count = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
expect(count.value).to.equal(0);
} finally {
stop();
}
});
});
describe('subsUnion size guard integration', function () {
this.timeout(60000);
beforeEach(cleanUp);
after(cleanUp);
it('stores a degraded union rather than a document Mongo would reject', async function () {
await Subscriptions.insertAsync(sub(43.5, -5.9));
// A union far bigger than the cap given to storeUnion.
const ring = [];
for (let i = 0; i < 2000; i += 1) {
const angle = (2 * Math.PI * i) / 2000;
ring.push([-5.9 + (0.3 * Math.cos(angle)), 43.5 + (0.3 * Math.sin(angle))]);
}
ring.push(ring[0]);
const big = {
type: 'Feature',
properties: {},
geometry: { type: 'MultiPolygon', coordinates: [[ring], [ring], [ring], [ring]] }
};
const subsUnion = build(async () => big, { maxUnionBytes: 20000 });
await subsUnion.recreate();
const stored = await storedUnion(true);
expect(Buffer.byteLength(stored.doc.value, 'utf8')).to.be.at.most(20000);
expect(stored.doc.degraded).to.be.a('string');
expect(stored.value.geometry.type).to.equal('MultiPolygon');
});
});

120
test/server/unionSizeGuard.test.js Normal file
View file

@ -0,0 +1,120 @@
/* eslint-env mocha */
/* eslint-disable func-names, prefer-arrow-callback */
/* eslint-disable import/no-absolute-path */
import { expect } from 'chai';
import limitUnionSize, { byteLength, MAX_UNION_BYTES } from '/imports/startup/server/unionSizeGuard';
// A closed ring of `points` vertices around (lon, lat). Coordinates are left at
// full double precision on purpose: that is what turf hands back, and it is what
// makes the JSON big.
const ring = (lon, lat, points, radius = 0.2) => {
const out = [];
for (let i = 0; i < points; i += 1) {
const angle = (2 * Math.PI * i) / points;
out.push([lon + (radius * Math.cos(angle)), lat + (radius * Math.sin(angle))]);
}
out.push(out[0]);
return out;
};
const multiPolygon = (count, points) => {
const coordinates = [];
for (let i = 0; i < count; i += 1) {
coordinates.push([ring(-10 + ((i % 100) * 0.5), 35 + (Math.floor(i / 100) * 0.5), points)]);
}
return { type: 'Feature', properties: {}, geometry: { type: 'MultiPolygon', coordinates } };
};
const polygonsOf = u => u.geometry.coordinates;
const pointCount = u => polygonsOf(u).reduce((acc, poly) => acc + poly.reduce((a, r) => a + r.length, 0), 0);
describe('unionSizeGuard', function () {
// Building and serializing a 20 MB fixture takes a couple of seconds.
this.timeout(120000);
it('leaves a union that already fits completely alone', function () {
const union = multiPolygon(2, 20);
const result = limitUnionSize(union, MAX_UNION_BYTES);
expect(result.degraded).to.equal(null);
expect(result.union).to.deep.equal(union);
expect(result.json).to.equal(JSON.stringify(union));
});
it('handles a null union (no valid subscriptions) without inventing geometry', function () {
const result = limitUnionSize(null, MAX_UNION_BYTES);
expect(result.union).to.equal(null);
expect(result.json).to.equal('null');
expect(result.degraded).to.equal(null);
});
it('drops coordinate precision first, keeping every polygon', function () {
const union = multiPolygon(40, 60);
const budget = byteLength(JSON.stringify(union)) - 1000;
const result = limitUnionSize(union, budget);
expect(result.bytes).to.be.at.most(budget);
expect(result.degraded.steps[0]).to.match(/^precision:/);
expect(polygonsOf(result.union)).to.have.lengthOf(40);
});
it('decimates vertices when rounding is not enough, still keeping every polygon', function () {
const union = multiPolygon(40, 144);
const budget = Math.round(byteLength(JSON.stringify(union)) / 4);
const result = limitUnionSize(union, budget);
expect(result.bytes).to.be.at.most(budget);
expect(result.degraded.steps.some(s => s.startsWith('decimate:'))).to.equal(true);
expect(polygonsOf(result.union)).to.have.lengthOf(40);
expect(pointCount(result.union)).to.be.below(pointCount(union));
});
it('drops the smallest polygons last, and only when nothing else fits', function () {
const union = multiPolygon(60, 20);
// A budget only a couple of polygons can fit into.
const budget = Math.round(byteLength(JSON.stringify(union)) / 25);
const result = limitUnionSize(union, budget);
expect(result.bytes).to.be.at.most(budget);
expect(result.degraded.steps[result.degraded.steps.length - 1]).to.match(/^polygons:/);
expect(polygonsOf(result.union).length).to.be.below(60);
expect(polygonsOf(result.union).length).to.be.at.least(1);
});
it('keeps every ring closed and valid after degrading', function () {
const union = multiPolygon(30, 144);
const result = limitUnionSize(union, Math.round(byteLength(JSON.stringify(union)) / 6));
polygonsOf(result.union).forEach((polygon) => {
polygon.forEach((r) => {
expect(r.length).to.be.at.least(4);
expect(r[0]).to.deep.equal(r[r.length - 1]);
});
});
});
// The actual FIXME this module exists for: subsUnion.js used to hand Mongo
// whatever came out of turf, and a document over 16 MiB is simply rejected —
// the union would then stay frozen at the last one that happened to fit.
it('brings a union past Mongo 16 MiB document limit back under the cap', function () {
// ~20 MB of GeoJSON: 4000 subscription-sized circles at 144 steps each.
const union = multiPolygon(4000, 144);
const original = byteLength(JSON.stringify(union));
expect(original).to.be.above(16 * 1024 * 1024);
const result = limitUnionSize(union, MAX_UNION_BYTES);
expect(result.bytes).to.be.at.most(MAX_UNION_BYTES);
expect(byteLength(result.json)).to.equal(result.bytes);
expect(result.degraded.originalBytes).to.equal(original);
expect(result.union).to.not.equal(null);
// Still parseable GeoJSON, not a truncated string.
expect(JSON.parse(result.json).geometry.type).to.equal('MultiPolygon');
});
it('never returns something bigger than the cap, whatever the input', function () {
const budget = 2000;
const result = limitUnionSize(multiPolygon(500, 144), budget);
expect(result.bytes).to.be.at.most(budget);
});
});