pipeline-lock.mjs

Rama main · réplica de lectura del repositorio original.

// pipeline-lock.mjs — a cross-process advisory lock for data/pipeline.md.
//
// appendToPipeline() (scan.mjs) is a plain read-modify-write: readFileSync,
// mutate the string, writeFileSync. It's exported and called from three
// places — scan.mjs itself, scan-ats-full.mjs, and plugins.mjs (pipeline
// mode) — so any two of them running concurrently (a scheduled scan
// overlapping a manual `/career-ops pipeline` run, or two plugin jobs) can
// silently drop one side's offers: whichever write lands second overwrites
// the first's in-memory read, with no error and no trace anything was lost.
//
// The protocol itself — the lock directory, owner.json, the recover guard,
// stale recovery and the token-checked release — is lib/dir-lock.mjs, shared
// with every other lock in the core. This module picks the lock path, the
// timing defaults and their env overrides, and the timeout error. The
// protocol's helpers are re-exported so existing importers keep working.

import { acquireDirLock, releaseOrWarn } from './lib/dir-lock.mjs';

export {
  OWNERLESS_GRACE_MS,
  sameLockDirectory,
  isMkdirContention,
  isRmContention,
  rmLockArtifactSync,
  RECOVER_STALE,
  RECOVER_VANISHED,
  RECOVER_LIVE,
  lockRecoveryVerdict,
  createLockWaitPolicy,
} from './lib/dir-lock.mjs';

const DEFAULT_STALE_MS = 30_000;
const DEFAULT_RETRY_MS = 80;
const DEFAULT_TIMEOUT_MS = 8_000;

export class LockTimeoutError extends Error {
  constructor(lockDir, timeoutMs) {
    super(`pipeline lock timeout: ${lockDir} held > ${timeoutMs}ms`);
    this.name = 'LockTimeoutError';
    this.lockDir = lockDir;
  }
}

export function lockDirFor(pipelinePath) {
  return `${pipelinePath}.lock`;
}

/**
 * Blocks until the lock on `pipelinePath` is held, then returns a handle whose
 * release() frees it. Throws LockTimeoutError if the lock stays busy.
 *
 * @param {string} pipelinePath - File the lock guards.
 * @param {object} [options]
 * @param {number} [options.timeoutMs=8000] - Max time a SINGLE holder may block this caller. Waiting is extended while the lock keeps changing hands (see createLockWaitPolicy).
 * @param {number} [options.retryMs=80] - Delay between acquisition attempts.
 * @param {number} [options.staleMs=30000] - Age threshold for a lock with no readable owner, floored at OWNERLESS_GRACE_MS.
 * @param {number} [options.maxWaitMs] - Absolute ceiling on waiting, however much progress the lock makes. Defaults to 10x timeoutMs.
 */
export async function acquirePipelineLock(pipelinePath, options = {}) {
  // Env overrides let a caller several frames up the stack (a test driving
  // appendToPipeline, say) tune contention timing without threading options
  // through every signature — same escape hatch the tracker lock provides.
  const timeoutMs = options.timeoutMs ?? (Number(process.env.CAREER_OPS_PIPELINE_LOCK_TIMEOUT_MS) || DEFAULT_TIMEOUT_MS);
  const retryMs = options.retryMs ?? (Number(process.env.CAREER_OPS_PIPELINE_LOCK_RETRY_MS) || DEFAULT_RETRY_MS);
  const staleMs = options.staleMs ?? (Number(process.env.CAREER_OPS_PIPELINE_LOCK_STALE_MS) || DEFAULT_STALE_MS);
  // A ceiling that is not a number is not a ceiling. NaN compares false against
  // everything, so a bad value would not fail loudly — it would silently remove
  // the only bound on waiting, which is the single thing this option exists to
  // provide. Only a usable ceiling is honoured; everything else is dropped, and
  // the protocol applies its default, which is at least bounded.
  //
  // Read WITHOUT the `|| default` idiom its three siblings above use, because
  // for this one 0 is a meaningful setting — "never wait past now" — and
  // `0 || default` would turn a caller who asked for no waiting into one who
  // waits the whole default ceiling, the exact opposite. timeoutMs and retryMs
  // have no such reading, which is why the idiom is fine there and wrong here.
  //
  // Infinity is kept as written: an explicit, legible "no ceiling" is a choice,
  // not a mistake. -Infinity and NaN are mistakes, so they take the default
  // rather than being clamped into a silently different behaviour.
  const envMaxWait = process.env.CAREER_OPS_PIPELINE_LOCK_MAX_WAIT_MS;
  const requestedMaxWait = options.maxWaitMs
    ?? (envMaxWait === undefined || envMaxWait.trim() === '' ? undefined : envMaxWait);
  const parsedMaxWait = requestedMaxWait === undefined ? undefined : Number(requestedMaxWait);
  const maxWaitMs = parsedMaxWait === Infinity || (Number.isFinite(parsedMaxWait) && parsedMaxWait >= 0)
    ? parsedMaxWait
    : undefined;

  const lock = await acquireDirLock(lockDirFor(pipelinePath), {
    timeoutMs,
    retryMs,
    staleMs,
    maxWaitMs,
    owner: { pipeline: pipelinePath },
    createParent: true,
    timeoutError: ({ lockDir, expiredMs, owner, lastMkdirError, diagnosis }) => {
      const err = new LockTimeoutError(lockDir, expiredMs);
      err.owner = owner;
      err.message += diagnosis;
      if (lastMkdirError) err.lastMkdirError = lastMkdirError;
      return err;
    },
  });
  return { lockDir: lock.lockDir, release: lock.release };
}

/** Acquires the lock on `pipelinePath`, runs fn, and always releases it. */
export async function withPipelineLock(pipelinePath, fn, options = {}) {
  const lock = await acquirePipelineLock(pipelinePath, options);
  try {
    return await fn();
  } finally {
    releaseOrWarn(lock);
  }
}