lib/durable-scheduler-store.js

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

var fs = require("fs");
var path = require("path");
var crypto = require("crypto");

function clone(value) {
  return JSON.parse(JSON.stringify(value));
}

function makeError(message, code, cause) {
  var error = new Error(message);
  error.code = code;
  if (cause) error.cause = cause;
  return error;
}

function pidIsAlive(pid) {
  if (typeof pid !== "number" || !isFinite(pid) || Math.floor(pid) !== pid || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return !!(error && error.code !== "ESRCH");
  }
}

function attachDurableSchedulerStore(ctx) {
  var filePath = ctx.filePath;
  var namespace = ctx.namespace;
  var fsImpl = ctx.fs || fs;
  var lockPath = filePath + ".lock";
  var guardPath = lockPath + ".acquire";
  var lockToken = crypto.randomBytes(16).toString("hex");
  var lockFd = null;
  var closed = false;

  if (!filePath || !namespace) throw new Error("Scheduler store requires filePath and namespace");

  function acquireLockUnderGuard() {
    try {
      lockFd = fsImpl.openSync(lockPath, "wx", 0o600);
      fsImpl.writeFileSync(lockFd, JSON.stringify({ pid: process.pid, namespace: namespace, token: lockToken }) + "\n");
      return;
    } catch (error) {
      if (!error || error.code !== "EEXIST") {
        if (lockFd !== null) {
          try { fsImpl.closeSync(lockFd); } catch (closeError) {}
          lockFd = null;
          try { fsImpl.unlinkSync(lockPath); } catch (unlinkError) {}
        }
        throw makeError("Unable to lock scheduler store: " + error.message, "SCHEDULER_STORE_LOCK", error);
      }
    }

    var lock;
    try {
      lock = JSON.parse(fsImpl.readFileSync(lockPath, "utf8"));
    } catch (error) {
      throw makeError("Scheduler store lock is unreadable", "SCHEDULER_STORE_LOCK", error);
    }
    if (pidIsAlive(lock.pid)) {
      throw makeError("Scheduler store is already owned by process " + lock.pid, "SCHEDULER_STORE_LOCKED");
    }
    var stalePath = lockPath + ".stale-" + process.pid + "-" + lockToken;
    try {
      fsImpl.renameSync(lockPath, stalePath);
      lockFd = fsImpl.openSync(lockPath, "wx", 0o600);
      fsImpl.writeFileSync(lockFd, JSON.stringify({ pid: process.pid, namespace: namespace, token: lockToken }) + "\n");
      try { fsImpl.unlinkSync(stalePath); } catch (cleanupError) {}
    } catch (error) {
      if (lockFd !== null) {
        try { fsImpl.closeSync(lockFd); } catch (closeError) {}
        lockFd = null;
        try { fsImpl.unlinkSync(lockPath); } catch (unlinkError) {}
      }
      try { fsImpl.unlinkSync(stalePath); } catch (cleanupError) {}
      throw makeError("Unable to recover stale scheduler store lock", "SCHEDULER_STORE_LOCK", error);
    }
  }

  function acquireLock() {
    fsImpl.mkdirSync(path.dirname(filePath), { recursive: true, mode: 0o700 });
    try {
      fsImpl.mkdirSync(guardPath, { mode: 0o700 });
    } catch (error) {
      if (error && error.code === "EEXIST") {
        throw makeError("Scheduler store acquisition is already in progress or requires manual stale-guard recovery", "SCHEDULER_STORE_LOCKED", error);
      }
      throw makeError("Unable to guard scheduler store acquisition", "SCHEDULER_STORE_LOCK", error);
    }

    var acquisitionError = null;
    try {
      acquireLockUnderGuard();
    } catch (error) {
      acquisitionError = error;
    }
    try {
      fsImpl.rmdirSync(guardPath);
    } catch (error) {
      if (lockFd !== null) {
        try { fsImpl.closeSync(lockFd); } catch (closeError) {}
        lockFd = null;
        try { fsImpl.unlinkSync(lockPath); } catch (unlinkError) {}
      }
      throw makeError("Unable to release scheduler store acquisition guard", "SCHEDULER_STORE_LOCK", error);
    }
    if (acquisitionError) throw acquisitionError;
  }

  function emptyData() {
    return { version: 1, namespace: namespace, jobs: [] };
  }

  function load() {
    if (closed) throw makeError("Scheduler store is closed", "SCHEDULER_STORE_CLOSED");
    if (!fsImpl.existsSync(filePath)) return emptyData();
    var parsed;
    try {
      parsed = JSON.parse(fsImpl.readFileSync(filePath, "utf8"));
    } catch (error) {
      throw makeError("Scheduler store is corrupt or unreadable", "SCHEDULER_STORE_CORRUPT", error);
    }
    if (!parsed || parsed.version !== 1 || parsed.namespace !== namespace || !Array.isArray(parsed.jobs)) {
      throw makeError("Scheduler store has an invalid schema or namespace", "SCHEDULER_STORE_CORRUPT");
    }
    return clone(parsed);
  }

  function save(data) {
    if (closed) throw makeError("Scheduler store is closed", "SCHEDULER_STORE_CLOSED");
    var serialized;
    try {
      serialized = JSON.stringify(data, null, 2) + "\n";
    } catch (error) {
      throw makeError("Scheduler store data is not serializable", "SCHEDULER_STORE_WRITE", error);
    }
    var temporary = filePath + ".tmp-" + process.pid;
    var temporaryFd = null;
    try {
      temporaryFd = fsImpl.openSync(temporary, "w", 0o600);
      fsImpl.writeFileSync(temporaryFd, serialized, "utf8");
      if (typeof fsImpl.fsyncSync === "function") fsImpl.fsyncSync(temporaryFd);
      fsImpl.closeSync(temporaryFd);
      temporaryFd = null;
      fsImpl.renameSync(temporary, filePath);
      try { fsImpl.chmodSync(filePath, 0o600); } catch (error) {}
      var directoryFd = null;
      try {
        directoryFd = fsImpl.openSync(path.dirname(filePath), "r");
        if (typeof fsImpl.fsyncSync === "function") fsImpl.fsyncSync(directoryFd);
      } catch (error) {
        var unsupported = error && (error.code === "EINVAL" || error.code === "ENOTSUP" || error.code === "EISDIR" || error.code === "EBADF" || error.code === "EPERM" || error.code === "EACCES");
        if (!unsupported) throw error;
      } finally {
        if (directoryFd !== null) {
          try { fsImpl.closeSync(directoryFd); } catch (closeError) {}
        }
      }
    } catch (error) {
      if (temporaryFd !== null) {
        try { fsImpl.closeSync(temporaryFd); } catch (closeError) {}
      }
      try { fsImpl.unlinkSync(temporary); } catch (cleanupError) {}
      throw makeError("Scheduler store write failed", "SCHEDULER_STORE_WRITE", error);
    }
  }

  function close() {
    if (closed) return;
    closed = true;
    if (lockFd !== null) {
      try { fsImpl.closeSync(lockFd); } catch (error) {}
      lockFd = null;
    }
    try {
      var lock = JSON.parse(fsImpl.readFileSync(lockPath, "utf8"));
      if (lock.pid === process.pid && lock.namespace === namespace && lock.token === lockToken) fsImpl.unlinkSync(lockPath);
    } catch (error) {}
  }

  acquireLock();
  return { load: load, save: save, close: close, filePath: filePath, namespace: namespace };
}

module.exports = { attachDurableSchedulerStore: attachDurableSchedulerStore };