From 110ad66e86bd0e70c0d1d4f939420780edfae337 Mon Sep 17 00:00:00 2001 From: vjrj Date: Thu, 30 Jul 2026 08:07:00 +0200 Subject: [PATCH] feat(subsUnion): incremental union merge for new subscriptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A full recreate() over 7k+ subscriptions takes ~20 minutes even in a worker thread — fine for not blocking the site, but means a new zone takes ages to show up on the map. Since union only grows when adding a circle, a new subscription can just be merged into the union already stored (one turf.union call) instead of rebuilding the whole chain. changed/removed still trigger a full recreate() (union can't be "subtracted" from safely), and incrementalAdd falls back to a full recreate whenever the stored count isn't exactly one behind what's expected, rather than risk merging into stale state. Both paths now share one serialized queue (FIFO for adds, coalesced for recomputes) so they never race on the same stored union. --- imports/startup/server/calcUnionAsync.js | 7 +- imports/startup/server/subsUnion.js | 137 +++++++++++++++-------- private/workers/unionWorker.js | 19 +++- 3 files changed, 111 insertions(+), 52 deletions(-) diff --git a/imports/startup/server/calcUnionAsync.js b/imports/startup/server/calcUnionAsync.js index 1cb1b24..6e00c03 100644 --- a/imports/startup/server/calcUnionAsync.js +++ b/imports/startup/server/calcUnionAsync.js @@ -7,10 +7,11 @@ const WORKER_PATH = path.resolve(process.cwd(), 'assets/app/workers/unionWorker. * Circle-union of already-validated, already-decorated subscriptions, run in a * worker thread so the main event loop (DDP/HTTP) stays free regardless of * how long any single turf.union call takes. subs must be an array of plain - * { location: { lat, lon }, distance } objects. + * { location: { lat, lon }, distance } objects. If baseUnion is given, subs + * are merged into it instead of a union computed from scratch. */ -const calcUnionAsync = subs => new Promise((resolve, reject) => { - const worker = new Worker(WORKER_PATH, { workerData: { subs } }); +const calcUnionAsync = (subs, baseUnion = null) => new Promise((resolve, reject) => { + const worker = new Worker(WORKER_PATH, { workerData: { subs, baseUnion } }); worker.once('message', (msg) => { worker.terminate(); if (msg.error) { diff --git a/imports/startup/server/subsUnion.js b/imports/startup/server/subsUnion.js index 9fe55ec..5828dc2 100644 --- a/imports/startup/server/subsUnion.js +++ b/imports/startup/server/subsUnion.js @@ -35,28 +35,18 @@ Meteor.startup(async () => { const noNoisy = sub => sub; - const process = async (isPublic) => { - const subscribers = await Subscriptions.find().fetchAsync(); - const decorate = isPublic ? addNoisy : noNoisy; - const validSubs = subscribers.filter((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; - }); - const decorated = validSubs.map(decorate); - - let union = null; - try { - union = await calcUnionAsync(decorated); - } catch (e) { - console.error('subsUnion worker failed', e); - } - const bounds = union === null ? null : L.geoJSON(union).getBounds(); + 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; + }; + const storeUnion = async (isPublic, union, count) => { const publicl = isPublic ? 'public' : 'private'; const Publicl = publicl.replace(/\b\w/g, l => l.toUpperCase()); if (typeof union === 'object') { + const bounds = union === null ? null : L.geoJSON(union).getBounds(); const unionSet = { $set: { name: `subs-${publicl}-union`, @@ -78,7 +68,7 @@ Meteor.startup(async () => { const sizeSet = { $set: { name: 'subs-union-count', - value: subscribers.length, + value: count, isPublic: false, description: 'Subscriptions count', type: 'number' @@ -95,40 +85,99 @@ Meteor.startup(async () => { } }; - const recreate = async () => { await process(true); await process(false); }; + const process = async (isPublic) => { + const subscribers = await Subscriptions.find().fetchAsync(); + const decorate = isPublic ? addNoisy : noNoisy; + const decorated = subscribers.filter(isValidSub).map(decorate); - 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; + let union = null; try { - await recreate(); + union = await calcUnionAsync(decorated); } catch (e) { - console.error('subsUnion recompute failed', e); - } finally { - recomputeRunning = false; - if (recomputePending) { - recomputePending = false; - runRecreate(); - } + console.error('subsUnion worker failed', e); } + await storeUnion(isPublic, union, subscribers.length); }; - // 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) { + 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; } - runRecreate(); + + // 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; + + let union = null; + try { + // eslint-disable-next-line no-await-in-loop + union = await calcUnionAsync([decorated], baseUnion); + } catch (e) { + console.error('subsUnion incremental worker failed, falling back to a full recreate', e); + recomputePending = true; + return; + } + // eslint-disable-next-line no-await-in-loop + await storeUnion(isPublic, union, countSubs); + } + 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 } }); @@ -146,9 +195,9 @@ Meteor.startup(async () => { } await Subscriptions.find({ createdAt: { $gt: new Date() } }).observeAsync({ - added: function newSubAdded() { // doc) { - if (debug) console.log('Subs added so recreate union'); - scheduleRecreate(); + added: function newSubAdded(doc) { + if (debug) console.log('Subs added, merging into union incrementally'); + enqueueAdd(doc); } }); diff --git a/private/workers/unionWorker.js b/private/workers/unionWorker.js index 9f6bce2..f6fc9ec 100644 --- a/private/workers/unionWorker.js +++ b/private/workers/unionWorker.js @@ -11,6 +11,10 @@ // // workerData.subs must already be validated (location.lat/lon/distance // present) and decorated (addNoisy/noNoisy already applied) by the caller. +// workerData.baseUnion, if given, is an already-computed union GeoJSON that +// subs gets merged into (the incremental-add fast path) instead of starting +// from scratch — for a single new subscription this is one tunion() call +// instead of redoing the whole chain over every subscription again. const { parentPort, workerData } = require('worker_threads'); const tcircle = require('@turf/circle').default; const tunion = require('@turf/union').default; @@ -19,16 +23,21 @@ const ttrunc = require('@turf/truncate').default; const truncOptions = { precision: 6, coordinates: 2 }; const run = () => { - const { subs } = workerData; - const unionGroup = subs.map(s => tcircle( + const { subs, baseUnion } = workerData; + const circles = subs.map(s => tcircle( [s.location.lon, s.location.lat], s.distance, { units: 'kilometers', steps: 144 } )); - 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); + let unionTemp = baseUnion || null; + let start = 0; + if (!unionTemp && circles.length > 0) { + unionTemp = ttrunc(circles[0], truncOptions); + start = 1; + } + for (let i = start; i < circles.length; i += 1) { + unionTemp = ttrunc(tunion(unionTemp, circles[i]), truncOptions); } return unionTemp; };