lib/pending-processor.js
function applyPendingChange(pendingChange, tries, callback) {
  var datasetId = pendingChange.datasetId;
  if (!datasetId || !pendingChange.action || !pendingChange.uid || !pendingChange.cuid || !pendingChange.hash) {
    debugError("[%s] invalid pendingChange request dropped :: item = %j", datasetId, pendingChange);
    return callback();
  }
  debug('[%s] processPending :: item = %j', datasetId, pendingChange);
  if (tries > retryLimit) {
    //the pendingChange has been processed before. Mark it as failed
    debugError('[%s] processPending failed :: tries = %d  :: item = %j', datasetId, tries, pendingChange);
    return saveUpdate(datasetId, pendingChange, SYNC_UPDATE_TYPES.FAILED, "crashed", callback);
  }

  pendingChange.tries = tries;

  var action = pendingChange.action.toLowerCase();

  var timer = metrics.startTimer();

  function onComplete(err) {
    metricsClient.gauge(metrics.KEYS.PENDING_CHANGE_PROCESS_TIME, {success: !err, action: action}, timer.stop());
    return callback(err);
  }

  switch (action) {
    case "create":
      doCreate(datasetId, pendingChange, onComplete);
      break;
    case "update":
      doUpdate(datasetId, pendingChange, onComplete);
      break;
    case "delete":
      doDelete(datasetId, pendingChange, onComplete);
      break;
    default:
      debugError("[%s] invalid pendingChange request dropped :: item = %j", datasetId, pendingChange);
      return onComplete();
  }
}
function applyPendingChange(pendingChange, tries, callback) {
  var datasetId = pendingChange.datasetId;
  if (!datasetId || !pendingChange.action || !pendingChange.uid || !pendingChange.cuid || !pendingChange.hash) {
    debugError("[%s] invalid pendingChange request dropped :: item = %j", datasetId, pendingChange);
    return callback();
  }
  debug('[%s] processPending :: item = %j', datasetId, pendingChange);
  if (tries > retryLimit) {
    //the pendingChange has been processed before. Mark it as failed
    debugError('[%s] processPending failed :: tries = %d  :: item = %j', datasetId, tries, pendingChange);
    return saveUpdate(datasetId, pendingChange, SYNC_UPDATE_TYPES.FAILED, "crashed", callback);
  }

  pendingChange.tries = tries;

  var action = pendingChange.action.toLowerCase();

  var timer = metrics.startTimer();

  function onComplete(err) {
    metricsClient.gauge(metrics.KEYS.PENDING_CHANGE_PROCESS_TIME, {success: !err, action: action}, timer.stop());
    return callback(err);
  }

  switch (action) {
    case "create":
      doCreate(datasetId, pendingChange, onComplete);
      break;
    case "update":
      doUpdate(datasetId, pendingChange, onComplete);
      break;
    case "delete":
      doDelete(datasetId, pendingChange, onComplete);
      break;
    default:
      debugError("[%s] invalid pendingChange request dropped :: item = %j", datasetId, pendingChange);
      return onComplete();
  }
}
apply the given pending change to the backend using the dataHandlers
@param {Object} pendingChange the pending change object
@param {String} pendingChange.datasetId the dataset id of the pending change
@param {Object} pendingChange.meta_data the meta_data of the dataset
@param {String} pendingChange.action the action of the pending change, should be one of "create", "update" or "delete"
@param {String} pendingChange.uid the unique id of the data record
@param {String} pendingChange.cuid the unique client id of the client device
@param {String} pendingChange.hash a unique id of the pending change, normally it is a hash value generated from the content of the pendingChange object
@param {Object} pendingChange.pre the record before change. Optional.
@param {Object} pendingChange.post the record after change. Optional.
@param {Number} tries a counter to record how many times the given pending change has been executed
@param {Function} callback the callback function
@returns