Firelink/scripts/engine-payload-promotion.js

import fs from 'node:fs';
import path from 'node:path';

const RETRY_DELAYS_MS = [100, 250, 500, 1_000, 2_000];

function pathExists(value) {
  try {
    fs.lstatSync(value);
    return true;
  } catch (error) {
    if (error?.code === 'ENOENT') return false;
    throw error;
  }
}

function previousPayloadPrefix(destination) {
  return `.${path.basename(destination)}.previous-`;
}

function previousPayloads(destination) {
  const parent = path.dirname(destination);
  const prefix = previousPayloadPrefix(destination);
  return fs.readdirSync(parent, { withFileTypes: true })
    .filter(entry => entry.name.startsWith(prefix))
    .map(entry => path.join(parent, entry.name));
}

function previousPayloadOwner(destination, candidate) {
  const suffix = path.basename(candidate).slice(previousPayloadPrefix(destination).length);
  const separator = suffix.indexOf('-');
  const pid = separator >= 0 ? suffix.slice(0, separator) : suffix;
  return /^\d+$/.test(pid) ? Number(pid) : null;
}

function isProcessAlive(pid) {
  if (!Number.isSafeInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return error?.code === 'EPERM';
  }
}

async function removeOrphanedPreviousPayloads(destination) {
  for (const candidate of previousPayloads(destination)) {
    const owner = previousPayloadOwner(destination, candidate);
    if (owner === null || isProcessAlive(owner)) continue;
    await removePathWithRetry(candidate);
  }
}

function provisioningTemporaryPrefix(target) {
  return `.firelink-engines-${target}-`;
}

function provisioningTemporaryOwner(target, candidate) {
  const suffix = path.basename(candidate).slice(provisioningTemporaryPrefix(target).length);
  const separator = suffix.indexOf('-');
  const pid = separator >= 0 ? suffix.slice(0, separator) : suffix;
  return /^\d+$/.test(pid) ? Number(pid) : null;
}

/**
 * Removes only staging directories created by a provisioner whose PID is no
 * longer alive. Legacy directories without an owner PID remain untouched so
 * a concurrent or ambiguous provisioner can never lose its staging tree.
 */
export async function removeOrphanedProvisioningDirectories(destinationParent, target) {
  const prefix = provisioningTemporaryPrefix(target);
  for (const entry of fs.readdirSync(destinationParent, { withFileTypes: true })) {
    if (!entry.isDirectory() || !entry.name.startsWith(prefix)) continue;
    const candidate = path.join(destinationParent, entry.name);
    const owner = provisioningTemporaryOwner(target, candidate);
    if (owner === null || owner === process.pid || isProcessAlive(owner)) continue;
    await removePathWithRetry(candidate);
  }
}

/**
 * Restores the only previous payload left by a process that died after moving
 * the destination aside but before publishing its replacement. Multiple
 * candidates are ambiguous and remain untouched for manual recovery.
 */
export function recoverInterruptedPromotion(destination) {
  if (pathExists(destination)) return;

  const candidates = previousPayloads(destination);
  if (candidates.length === 0) return;
  if (candidates.length > 1) {
    throw new Error(
      `Cannot recover engine payload at ${destination}: found ${candidates.length} previous payloads`
    );
  }

  const candidate = candidates[0];
  if (!fs.lstatSync(candidate).isDirectory()) {
    throw new Error(`Cannot recover engine payload from non-directory backup: ${candidate}`);
  }
  fs.renameSync(candidate, destination);
}

function sleep(milliseconds) {
  return new Promise(resolve => setTimeout(resolve, milliseconds));
}

export async function removePathWithRetry(value) {
  for (let attempt = 0; ; attempt += 1) {
    try {
      fs.rmSync(value, { recursive: true, force: true });
      return;
    } catch (error) {
      const retryable = process.platform === 'win32'
        && ['EACCES', 'EBUSY', 'EPERM'].includes(error?.code);
      if (!retryable || attempt >= RETRY_DELAYS_MS.length) throw error;
      await sleep(RETRY_DELAYS_MS[attempt]);
    }
  }
}

/**
 * Publishes a fully verified payload without exposing a partially written
 * directory. The staging directory must share a filesystem with destination.
 */
export async function promoteDirectory(staging, destination) {
  recoverInterruptedPromotion(destination);
  const stagingStats = fs.lstatSync(staging);
  if (!stagingStats.isDirectory()) {
    throw new Error(`Engine payload staging path is not a directory: ${staging}`);
  }

  const parent = path.dirname(destination);
  const backup = path.join(
    parent,
    `.${path.basename(destination)}.previous-${process.pid}-${process.hrtime.bigint()}`
  );
  let movedExisting = false;

  try {
    if (pathExists(destination)) {
      fs.renameSync(destination, backup);
      movedExisting = true;
    }
    fs.renameSync(staging, destination);
  } catch (error) {
    if (movedExisting && !pathExists(destination) && pathExists(backup)) {
      try {
        fs.renameSync(backup, destination);
      } catch (restoreError) {
        throw new AggregateError(
          [error, restoreError],
          `Failed to publish engine payload and restore the previous payload at ${destination}`
        );
      }
    } else if (movedExisting && pathExists(destination) && pathExists(backup)) {
      // Another provisioner won the promotion race; discard only our backup.
      await removePathWithRetry(backup);
    }
    throw error;
  }

  if (movedExisting) await removePathWithRetry(backup);
  await removeOrphanedPreviousPayloads(destination);
}