todos-contra-el-fuego-web/imports/startup/server/calcUnionAsync.js

import path from 'path';
import { Worker } from 'worker_threads';
import { Meteor } from 'meteor/meteor';

const WORKER_PATH = path.resolve(process.cwd(), 'assets/app/workers/unionWorker.js');

// The app's own npm dependencies (installed by `meteor build`) live in
// programs/server/npm/node_modules — a directory that's a SIBLING of
// programs/server/assets, not an ancestor of it. Node's require() only walks
// up parent directories looking for node_modules, so a plain
// require('@turf/circle') from the worker script (a raw asset, not part of
// Meteor's own module system that knows about this layout) can never find it
// and fails with "Cannot find module '@turf/circle'". Resolving the absolute
// path here, where Meteor's own module system already knows how to find it,
// and handing it to the worker sidesteps that entirely.
const NPM_MODULES = path.resolve(process.cwd(), 'npm/node_modules');
const turfPaths = {
  circle: path.join(NPM_MODULES, '@turf/circle'),
  union: path.join(NPM_MODULES, '@turf/union'),
  truncate: path.join(NPM_MODULES, '@turf/truncate')
};

// A worker that never answers used to hang the promise forever, and with it the
// queue in subsUnionLogic: `busy` stays true and no union is ever computed again
// until the process restarts — silently, because nothing throws. Measured on
// this hardware (bench/union-bench.js), a full recreate of 10.000 subscriptions
// takes ~40 s, so ten minutes is a very wide margin over anything healthy while
// still being a bound.
const DEFAULT_TIMEOUT_MS = 10 * 60 * 1000;
export const timeoutMs = () => (
  (Meteor.settings.private && Meteor.settings.private.unionWorkerTimeoutMs) || DEFAULT_TIMEOUT_MS
);


class UnionWorkerTimeout extends Error {
  constructor(ms) {
    super(`union worker did not answer in ${ms} ms`);
    this.name = 'UnionWorkerTimeout';
    this.isTimeout = true;
  }
}

// Cota de heap del worker. No es para que quepa: es para que un caso patológico
// (miles de suscripciones que no se solapan) muera con ERR_WORKER_OUT_OF_MEMORY
// —que esta capa convierte en un rechazo y deja la unión anterior intacta— en
// vez de que el kernel se lleve por delante el proceso entero de Meteor. Medido:
// 10.000 suscripciones con 64 vértices por círculo son ~421 MB de RSS de todo el
// proceso, así que 1 GB de heap para el worker es holgado.
const DEFAULT_MAX_HEAP_MB = 1024;
const settingsOf = name => (Meteor.settings.private || {})[name];

const runOnce = (subs, baseUnion, ms) => new Promise((resolve, reject) => {
  const worker = new Worker(WORKER_PATH, {
    workerData: { subs, baseUnion, turfPaths, steps: settingsOf('unionCircleSteps') },
    resourceLimits: { maxOldGenerationSizeMb: settingsOf('unionWorkerMaxHeapMb') || DEFAULT_MAX_HEAP_MB }
  });
  let settled = false;
  const finish = (fn, arg) => {
    if (settled) return;
    settled = true;
    clearTimeout(timer);
    // terminate() is what actually stops the thread burning a core; without it a
    // timed-out union would keep computing forever next to its replacement.
    worker.terminate();
    fn(arg);
  };
  const timer = setTimeout(() => finish(reject, new UnionWorkerTimeout(ms)), ms);

  worker.once('message', (msg) => {
    if (msg.error) finish(reject, new Error(msg.error));
    else finish(resolve, msg.union);
  });
  worker.once('error', err => finish(reject, err));
});

/**
 * 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. If baseUnion is given, subs
 * are merged into it instead of a union computed from scratch.
 *
 * Fails after a timeout, and retries once on a genuine error — a failure to
 * spawn the thread, or a crash, is usually transient. A TIMEOUT is not retried:
 * whatever made it take longer than the bound will still be there, and a second
 * attempt only doubles the time the union stays stale.
 */
const calcUnionAsync = async (subs, baseUnion = null) => {
  const ms = timeoutMs();
  try {
    return await runOnce(subs, baseUnion, ms);
  } catch (e) {
    if (e.isTimeout) throw e;
    console.warn('union worker failed, retrying once', e.message);
    return runOnce(subs, baseUnion, ms);
  }
};

export default calcUnionAsync;