lib/driver-continuation-pair-transfer.js

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

var continuationPair = require("./driver-continuation-pair");

function attachContinuationPairTransfer(settings) {
  var ctx = settings.ctx;

  function commit(lease) {
    var transfer = lease && lease.pairTransfer;
    if (!transfer || transfer.committed) return { ok: true };
    var result = ctx.splitStore.commitOwnedDriverTransfer(transfer);
    if (!result.ok) return { ok: false, error: "The Split Worker ownership transfer could not be persisted: " + result.error };
    return { ok: true };
  }

  function rollback(lease) {
    var transfer = lease && lease.pairTransfer;
    if (!transfer || !transfer.staged) return { ok: true };
    return ctx.splitStore.rollbackOwnedDriverTransfer(transfer);
  }

  function begin(source, target, proposal, lease) {
    var staged = continuationPair.stage(source, target, ctx);
    if (!staged.ok) return staged;
    lease.pairTransfer = staged.transaction || null;
    lease.onCancel = function (cancelError) {
      var rolledBack = rollback(lease);
      if (!rolledBack.ok) {
        lease.cleanupError = "The Split Worker ownership transfer could not be rolled back: " + rolledBack.error;
        return false;
      }
      var persisted = settings.update(source, proposal, { status: "superseded", targetSessionId: null,
        targetOriginId: null, error: cancelError });
      if (!persisted) {
        lease.cleanupError = "The cancelled continuation state could not be persisted; the successor was retained for recovery.";
        return false;
      }
      lease.cleanupComplete = true;
      ctx.sm.deleteSessionQuiet(target.localId);
      settings.releaseLease(source, lease);
      return true;
    };
    return staged;
  }

  function validationError(source, target, lease, validateSource) {
    var sourceError = validateSource();
    if (sourceError) return sourceError;
    if (!lease.pairTransfer) return "";
    var transferred = continuationPair.inspect(target, ctx);
    if (!transferred.ok) return transferred.error;
    if (!transferred.group || transferred.group !== lease.pairTransfer.runtimeGroup ||
        transferred.worker !== lease.pairTransfer.worker) {
      return "The staged Split Worker ownership changed during successor startup.";
    }
    return "";
  }

  function retryError(source, target, lease, validateSource) {
    if (!lease || !lease.pairTransfer || lease.pairTransfer.committed) return "";
    return validationError(source, target, lease, validateSource);
  }

  function commitOrCancel(lease) {
    var result = commit(lease);
    if (result.ok) return result;
    lease.cancel(result.error);
    return { ok: false, error: lease.cleanupError || result.error };
  }

  function recoverIfNeeded(source, target, proposal) {
    var current = continuationPair.inspect(source, ctx);
    if (!current.group) return { ok: true, lease: null };
    var lease = settings.beginLease(source, target, proposal.proposalId);
    var recovered = begin(source, target, proposal, lease);
    if (!recovered.ok) {
      settings.releaseLease(source, lease);
      return recovered;
    }
    lease.waitingCommit = true;
    return { ok: true, lease: lease };
  }

  return { begin: begin, commit: commit, commitOrCancel: commitOrCancel, recoverIfNeeded: recoverIfNeeded,
    retryError: retryError, rollback: rollback, validationError: validationError };
}

module.exports = { attachContinuationPairTransfer: attachContinuationPairTransfer };