From b4e5511cd036a3e811c79734df13f19ebcc7a469 Mon Sep 17 00:00:00 2001 From: vjrj Date: Thu, 30 Jul 2026 06:36:37 +0200 Subject: [PATCH] fix(subsUnion): stop blocking DDP on subscription union recompute Every subscription add/change/remove recomputed the full geo union over all 7k+ subscriptions with a synchronous turf.union chain, freezing the single Node event loop (and thus DDP/HTTP) for minutes. The same recompute also runs at startup ("Subs union outdated"), so every restart froze the site too. calcUnionAsync yields to the event loop periodically during the union chain, and subsUnion.js now fires recomputes without blocking Meteor.startup or the observer callbacks, serializing overlapping triggers instead of stacking them. --- imports/startup/server/calcUnionAsync.js | 52 +++++++++++++++++++++ imports/startup/server/subsUnion.js | 58 ++++++++++++++++++------ package-lock.json | 3 ++ package.json | 3 ++ 4 files changed, 102 insertions(+), 14 deletions(-) create mode 100644 imports/startup/server/calcUnionAsync.js diff --git a/imports/startup/server/calcUnionAsync.js b/imports/startup/server/calcUnionAsync.js new file mode 100644 index 0000000..5101e5e --- /dev/null +++ b/imports/startup/server/calcUnionAsync.js @@ -0,0 +1,52 @@ +import tcircle from '@turf/circle'; +import tunion from '@turf/union'; +import ttrunc from '@turf/truncate'; + +const truncOptions = { precision: 6, coordinates: 2 }; +// How many polygons to process before yielding to the event loop once. +const YIELD_EVERY = 100; + +const yieldToEventLoop = () => new Promise((resolve) => { setImmediate(resolve); }); + +/** + * Circle-union of subscriptions, equivalent to map-common-utils' calcUnion(typeCircle=true), + * but split into a `for` loop that yields to the event loop every YIELD_EVERY polygons. + * The vendored calcUnion runs the same turf.union chain fully synchronously, which for + * thousands of subscriptions blocks Meteor's single-threaded DDP/HTTP handling for minutes + * (see plan-modernizacion notes on the subsUnion hang). This keeps the exact same geometry + * output while letting other requests interleave between chunks. + */ +const calcUnionAsync = async (subs, decorated) => { + const unionGroup = []; + for (let i = 0; i < subs.length; i += 1) { + const osub = subs[i]; + try { + if (osub.location && osub.location.lat && osub.location.lon && osub.distance) { + const dsub = decorated(osub); + unionGroup.push(tcircle( + [dsub.location.lon, dsub.location.lat], + dsub.distance, + { units: 'kilometers', steps: 144 } + )); + } else { + console.info(`Wrong element to do union ${JSON.stringify(osub)}`); + } + } catch (e) { + console.error(e, `Wrong element trying to make union ${JSON.stringify(osub)}`); + } + // Deliberate: yield to the event loop periodically instead of blocking it. + // eslint-disable-next-line no-await-in-loop + if (i % YIELD_EVERY === 0) await yieldToEventLoop(); + } + + let unionTemp = unionGroup.length > 0 ? ttrunc(unionGroup[0], truncOptions) : null; + for (let i = 1; i < unionGroup.length; i += 1) { + unionTemp = ttrunc(tunion(unionTemp, unionGroup[i]), truncOptions); + // Deliberate: yield to the event loop periodically instead of blocking it. + // eslint-disable-next-line no-await-in-loop + if (i % YIELD_EVERY === 0) await yieldToEventLoop(); + } + return unionTemp; +}; + +export default calcUnionAsync; diff --git a/imports/startup/server/subsUnion.js b/imports/startup/server/subsUnion.js index 0c78331..087a1e5 100644 --- a/imports/startup/server/subsUnion.js +++ b/imports/startup/server/subsUnion.js @@ -6,7 +6,7 @@ import SiteSettings from '/imports/api/SiteSettings/SiteSettings'; import Perlin from 'loms.perlin'; import './leaflet-workaround'; import L from 'leaflet'; -import calcUnion from '/imports/ui/components/Maps/SubsUnion/Unify'; +import calcUnionAsync from '/imports/startup/server/calcUnionAsync'; import { isMailServerMaster } from '/imports/startup/server/email'; // sudo apt-get install libcairo2-dev libjpeg-dev libgif-dev @@ -37,9 +37,8 @@ Meteor.startup(async () => { const process = async (isPublic) => { const subscribers = await Subscriptions.find().fetchAsync(); - const result = calcUnion(L, subscribers, isPublic ? addNoisy : noNoisy, true); - const union = result[0]; - const bounds = result[1]; + const union = await calcUnionAsync(subscribers, isPublic ? addNoisy : noNoisy); + const bounds = union === null ? null : L.geoJSON(union).getBounds(); const publicl = isPublic ? 'public' : 'private'; const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase()); @@ -83,6 +82,40 @@ Meteor.startup(async () => { } }; + const recreate = async () => { await process(true); await process(false); }; + + let recomputeRunning = false; + let recomputePending = false; + + // Runs recreate() serially: a recompute already in flight is never overlapped by + // another one, bursts of subscription changes just mark it pending and get a single + // extra pass once the current one finishes. + const runRecreate = async () => { + recomputeRunning = true; + try { + await recreate(); + } catch (e) { + console.error('subsUnion recompute failed', e); + } finally { + recomputeRunning = false; + if (recomputePending) { + recomputePending = false; + runRecreate(); + } + } + }; + + // Fire-and-forget on purpose: recreate() can take a while over thousands of + // subscriptions, and callers (Meteor.startup, observeAsync callbacks) must not + // block on it. + const scheduleRecreate = () => { + if (recomputeRunning) { + recomputePending = true; + return; + } + runRecreate(); + }; + // 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 } }); @@ -93,30 +126,27 @@ Meteor.startup(async () => { const lastSubsUpdated = lastSubs.updatedAt; if (lastUnionUpdated > lastSubsUpdated || !countUnionSubs || countSubs !== countUnionSubs.value) { console.log('Subs union outdated'); - await process(true); - await process(false); + scheduleRecreate(); } else { console.log('Subs union up-to-date'); } } - const recreate = async () => { await process(true); await process(false); }; - await Subscriptions.find({ createdAt: { $gt: new Date() } }).observeAsync({ - added: async function newSubAdded() { // doc) { + added: function newSubAdded() { // doc) { if (debug) console.log('Subs added so recreate union'); - await recreate(); + scheduleRecreate(); } }); await Subscriptions.find().observeAsync({ - changed: async function subsChanged() { // updatedDoc, oldDoc) { + changed: function subsChanged() { // updatedDoc, oldDoc) { if (debug) console.log('Subs changed so recreate union'); - await recreate(); + scheduleRecreate(); }, - removed: async function subsRemoved() { // oldDoc) { + removed: function subsRemoved() { // oldDoc) { if (debug) console.log('Subs removed so recreate union'); - await recreate(); + scheduleRecreate(); } }); }); diff --git a/package-lock.json b/package-lock.json index 0fbb382..2713045 100644 --- a/package-lock.json +++ b/package-lock.json @@ -11,6 +11,9 @@ "@sentry/browser": "^8.55.2", "@sentry/node": "^8.55.2", "@turf/buffer": "^5.1.5", + "@turf/circle": "^6.0.1", + "@turf/truncate": "^6.0.1", + "@turf/union": "^6.0.3", "babel-runtime": "^6.26.0", "bcrypt": "^5.1.1", "bootstrap": "^5.3.8", diff --git a/package.json b/package.json index 1391bd8..b7c8533 100644 --- a/package.json +++ b/package.json @@ -12,6 +12,9 @@ "@sentry/browser": "^8.55.2", "@sentry/node": "^8.55.2", "@turf/buffer": "^5.1.5", + "@turf/circle": "^6.0.1", + "@turf/truncate": "^6.0.1", + "@turf/union": "^6.0.3", "babel-runtime": "^6.26.0", "bcrypt": "^5.1.1", "bootstrap": "^5.3.8",