Sync the data records for the given client
@param {String} datasetId
@param {Object} params the request body, it normally contain those fields:
@param {Object} params.query_params the query parameter for the dataset from the client
@param {Object} params.meta_data the meta data for the dataset from the client
@param {Object} params.__fh an object added by the client sdk. it should have the `cuid` of the client
@param {Object} params.clientRecs the client records
@param {Function} cb
function syncRecords(datasetId, params, cb) {
  debug('[%s] process syncRecords request', datasetId);
  var queryParams = params.query_params || {};
  var metaData = params.meta_data || {}; //NOTE: the client doesn't send in this value for syncRecords ATM
  var cuid = syncConfig.cuidProducer(params);
  var datasetClient = new DatasetClient(datasetId, {queryParams: queryParams, metaData: metaData});
  var clientRecords = params.clientRecs || {};
  async.waterfall([
    function checkDatasetclientStopped(callback) {
      syncStorage.readDatasetClient(datasetClient.getId(), function(err, datasetClientJson) {
        if (err) {
          return callback(err);
        }
        if (!datasetClientJson) {
          var errMsg = "unknown dataset client datasetId = " + datasetId + " :: queryParams = " + util.inspect(queryParams);
          debugError("[%s] %s", datasetId, errMsg);
          return callback(errMsg);
        }
        if (datasetClientJson.stopped === true) {
          return callback(new Error('sync stopped for dataset ' + datasetId));
        } else {
          return callback();
        }
      });
    },
    async.apply(listLocalDatasetClientData, datasetClient),
    function listOtherChanges(localDatasetData, callback) {
      if (localDatasetData) {
        listChangesNotInLocalDataset(datasetId, localDatasetData.syncLoopEnd, {cuid: cuid}, function(err, appliedUpdates, clientPendingChanges) {
          return callback(err, localDatasetData, appliedUpdates, clientPendingChanges);
        });
      } else {
        return callback();
      }
    }
  ], function(err, localDatasetData, appliedUpdatesSinceLastSync, pendingUpdates) {
    if (err) {
      return cb(err);
    }
    if (!localDatasetData) {
      return cb(null, {});
    }
    var localDatasetDataObj = convertToObject(localDatasetData.records);
    var appliedUpdatesSinceLastSyncObj = convertToObject(appliedUpdatesSinceLastSync);
    var pendingUpdatesObj = convertToObject(pendingUpdates);

    removeAppliedUpdates(clientRecords, localDatasetDataObj, appliedUpdatesSinceLastSyncObj);
    removePendingChanges(clientRecords, localDatasetDataObj, pendingUpdatesObj);

    var delta = computeDelta(datasetId, clientRecords, localDatasetDataObj);
    var res = delta;
    res.hash = localDatasetData.globalHash;
    debug('[%s] syncRecords API response %j', datasetId, res);
    return cb(null, res);
  });
}