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

refactor(subsUnion): extract the logic out of Meteor.startup, cap the union size

Browse source 
Everything the union does lived inside a Meteor.startup callback with no exports:
the queue, the incremental fast path, the worker-failure handling. None of it was
reachable from a test, which is how three separate incidents shipped — a failed
worker stored as an empty union (50ca9cc), turf deps unresolvable from the worker
(3249362/570cb49) and a recompute that blocked the event loop and froze DDP
(b4e5511/fbb746f). subsUnionLogic.js now takes its collaborators as arguments and
subsUnion.js is only the wiring.

The FIXME at the old subsUnion.js:82 goes with it, because it lives inside the
extracted storeUnion: a union over 16 MiB is simply rejected by Mongo, so the map
would freeze at the last union that happened to fit, silently. unionSizeGuard.js
degrades the geometry instead — coordinate precision, then vertex decimation,
then holes, then the smallest polygons — until it fits, and reports what it did
so the setting document records it. The cap is 8 MiB rather than 16: the same
string is pushed to every browser over DDP.

addNoisy no longer mutates the document it is given, so the public pass cannot
leak its fuzzing into the private one.
This commit is contained in:
vjrj 2026-08-01 18:02:04 +02:00
parent e68340931c
commit 124ed4e135
4 changed files with 499 additions and 187 deletions
Download patch file Download diff file Expand all files Collapse all files

211
imports/startup/server/subsUnion.js
View file

@ -1,214 +1,51 @@
/* eslint-disable import/no-absolute-path */
// Wiring only: what the union actually does lives in subsUnionLogic.js, where it
// can be tested without booting a server (see test/server/subsUnion.test.js).
import { Meteor } from 'meteor/meteor';
import Subscriptions from '/imports/api/Subscriptions/Subscriptions';
import SiteSettings from '/imports/api/SiteSettings/SiteSettings';
import Perlin from 'loms.perlin';
import './leaflet-workaround';
import L from 'leaflet';
import calcUnionAsync from '/imports/startup/server/calcUnionAsync';
import unionBounds from '/imports/startup/server/unionBounds';
import createSubsUnion from '/imports/startup/server/subsUnionLogic';
import { isMailServerMaster } from '/imports/startup/server/email';
// sudo apt-get install libcairo2-dev libjpeg-dev libgif-dev
Meteor.startup(async () => {
if (!isMailServerMaster) {
console.log('We only process subsUnion in master');
return;
}
const debug = true; // Meteor.isDevelopment;
Perlin.seed(Math.random());
const addNoisy = (osub) => {
const sub = osub;
let lat = Math.round(sub.location.lat * 10) / 10;
let lon = Math.round(sub.location.lon * 10) / 10;
const noiseBase = Perlin.perlin2(lat, lon);
const noise = Math.abs(noiseBase / 3);
lat += noise;
lon += noise;
sub.location.lat = lat;
sub.location.lon = lon;
sub.distance += noiseBase;
return sub;
};
const subsUnion = createSubsUnion({
calcUnion: calcUnionAsync,
SiteSettings,
Subscriptions,
boundsOf: unionBounds
});
const noNoisy = sub => sub;
const isValidSub = (osub) => {
const valid = osub.location && osub.location.lat && osub.location.lon && osub.distance;
if (!valid) console.info(`Wrong element to do union ${JSON.stringify(osub)}`);
return valid;
};
// Only called once calcUnionAsync has actually resolved — union === null here
// means "zero valid subscriptions", a real result, not a failed computation.
// (A prior version guarded this with `typeof union === 'object'`, which is
// true for null too, so a failed worker call — caught upstream, leaving
// union as null — looked identical to a legitimate empty union and got
// silently stored as "null". Callers must catch calcUnionAsync failures
// themselves and skip calling this, instead of relying on this check.)
const storeUnion = async (isPublic, union, count) => {
const publicl = isPublic ? 'public' : 'private';
const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase());
const bounds = union === null ? null : L.geoJSON(union).getBounds();
const unionSet = {
$set: {
name: `subs-${publicl}-union`,
value: JSON.stringify(union),
isPublic,
description: `${Publicl} subscriptions union`,
type: 'string'
}
};
const boundsSet = {
$set: {
name: `subs-${publicl}-union-bounds`,
value: JSON.stringify(bounds),
isPublic,
description: `${Publicl} subscriptions union bounds`,
type: 'string'
}
};
const sizeSet = {
$set: {
name: 'subs-union-count',
value: count,
isPublic: false,
description: 'Subscriptions count',
type: 'number'
}
};
// FIXME, take care of object size:
// https://stackoverflow.com/questions/10827812/what-is-the-length-maximum-for-a-string-data-type-in-mongodb-used-with-ruby
await SiteSettings.upsertAsync({ name: `subs-${publicl}-union` }, unionSet, { multi: false });
await SiteSettings.upsertAsync({ name: `subs-${publicl}-union-bounds` }, boundsSet, { multi: false });
await SiteSettings.upsertAsync({ name: 'subs-union-count' }, sizeSet, { multi: false });
if (debug) console.log(`${Publicl} subscription union calculated`);
};
const process = async (isPublic) => {
const subscribers = await Subscriptions.find().fetchAsync();
const decorate = isPublic ? addNoisy : noNoisy;
const decorated = subscribers.filter(isValidSub).map(decorate);
try {
const union = await calcUnionAsync(decorated);
await storeUnion(isPublic, union, subscribers.length);
} catch (e) {
console.error('subsUnion worker failed, union left unchanged', e);
}
};
const recreate = async () => { await process(true); await process(false); };
// Single serialized worker over two kinds of work: incremental adds (a real
// FIFO queue — each must see the previous one's result, none get dropped)
// and full recomputes (coalesced — a burst of changed/removed events only
// needs one pass once things settle). Both touch the same stored union, so
// they share this one lock rather than racing each other.
let busy = false;
const addQueue = [];
let recomputePending = false;
// Fast path for the common case (a new subscription): merge just the new
// circle into the union already stored instead of redoing the full turf.union
// chain over every subscription again (that chain is what made a single
// recompute take ~20+ minutes with thousands of subscriptions). Falls back to
// a full recreate() if the stored state isn't exactly "one behind" what we
// expect — safer than merging into stale/inconsistent data. Sets
// recomputePending directly (rather than calling a schedule helper) since
// runLoop already re-checks it right after this returns.
const incrementalAdd = async (doc) => {
if (!isValidSub(doc)) return;
const countSetting = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
const countSubs = await Subscriptions.find({}).countAsync();
if (!countSetting || countSetting.value !== countSubs - 1) {
if (debug) console.log('Subs union state not exactly one behind, doing a full recreate instead');
recomputePending = true;
return;
}
// eslint-disable-next-line no-restricted-syntax
for (const isPublic of [true, false]) {
const decorate = isPublic ? addNoisy : noNoisy;
const decorated = decorate({ ...doc, location: { ...doc.location } });
const publicl = isPublic ? 'public' : 'private';
// eslint-disable-next-line no-await-in-loop
const current = await SiteSettings.findOneAsync({ name: `subs-${publicl}-union` });
const baseUnion = current && current.value ? JSON.parse(current.value) : null;
try {
// eslint-disable-next-line no-await-in-loop
const union = await calcUnionAsync([decorated], baseUnion);
// eslint-disable-next-line no-await-in-loop
await storeUnion(isPublic, union, countSubs);
} catch (e) {
console.error('subsUnion incremental worker failed, falling back to a full recreate', e);
recomputePending = true;
return;
}
}
if (debug) console.log('Subs union updated incrementally');
};
const runLoop = () => {
if (busy) return;
if (addQueue.length > 0) {
busy = true;
const doc = addQueue.shift();
incrementalAdd(doc)
.catch(e => console.error('subsUnion incremental add failed', e))
.finally(() => { busy = false; runLoop(); });
return;
}
if (recomputePending) {
recomputePending = false;
busy = true;
recreate()
.catch(e => console.error('subsUnion recompute failed', e))
.finally(() => { busy = false; runLoop(); });
}
};
const enqueueAdd = (doc) => { addQueue.push(doc); runLoop(); };
// Fire-and-forget on purpose: both recreate() and incrementalAdd() can take a
// while, and callers (Meteor.startup, observeAsync callbacks) must not block on it.
const scheduleRecreate = () => { recomputePending = true; runLoop(); };
// At startup, we check if it's necessary to calc subscriptions union again
const currentUnion = await SiteSettings.findOneAsync({ name: 'subs-public-union' });
const lastSubs = await Subscriptions.findOneAsync({}, { sort: { updatedAt: -1 } });
const countUnionSubs = await SiteSettings.findOneAsync({ name: 'subs-union-count' });
const countSubs = await Subscriptions.find({}).countAsync();
if (currentUnion && lastSubs) {
const lastUnionUpdated = currentUnion.updatedAt;
const lastSubsUpdated = lastSubs.updatedAt;
if (lastUnionUpdated > lastSubsUpdated || !countUnionSubs || countSubs !== countUnionSubs.value) {
console.log('Subs union outdated');
scheduleRecreate();
} else {
console.log('Subs union up-to-date');
}
if (await subsUnion.isUnionOutdated()) {
console.log('Subs union outdated');
subsUnion.scheduleRecreate();
} else {
console.log('Subs union up-to-date');
}
await Subscriptions.find({ createdAt: { $gt: new Date() } }).observeAsync({
added: function newSubAdded(doc) {
if (debug) console.log('Subs added, merging into union incrementally');
enqueueAdd(doc);
console.log('Subs added, merging into union incrementally');
subsUnion.enqueueAdd(doc);
}
});
await Subscriptions.find().observeAsync({
changed: function subsChanged() { // updatedDoc, oldDoc) {
if (debug) console.log('Subs changed so recreate union');
scheduleRecreate();
changed: function subsChanged() {
console.log('Subs changed so recreate union');
subsUnion.scheduleRecreate();
},
removed: function subsRemoved() { // oldDoc) {
if (debug) console.log('Subs removed so recreate union');
scheduleRecreate();
removed: function subsRemoved() {
console.log('Subs removed so recreate union');
subsUnion.scheduleRecreate();
}
});
});