/* fs_fs.c --- filesystem operations specific to fs_fs
 *
 * ====================================================================
 * Copyright (c) 2000-2008 CollabNet.  All rights reserved.
 *
 * This software is licensed as described in the file COPYING, which
 * you should have received as part of this distribution.  The terms
 * are also available at http://subversion.tigris.org/license-1.html.
 * If newer versions of this license are posted there, you may use a
 * newer version instead, at your option.
 *
 * This software consists of voluntary contributions made by many
 * individuals.  For exact contribution history, see the revision
 * history and logs, available at http://subversion.tigris.org/.
 * ====================================================================
 */

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <assert.h>
#include <errno.h>

#include <apr_general.h>
#include <apr_pools.h>
#include <apr_file_io.h>
#include <apr_uuid.h>
#include <apr_lib.h>
#include <apr_md5.h>
#include <apr_thread_mutex.h>

#include "svn_pools.h"
#include "svn_fs.h"
#include "svn_path.h"
#include "svn_hash.h"
#include "svn_props.h"
#include "svn_sorts.h"
#include "svn_time.h"
#include "svn_mergeinfo.h"
#include "svn_config.h"

#include "fs.h"
#include "err.h"
#include "tree.h"
#include "lock.h"
#include "key-gen.h"
#include "fs_fs.h"
#include "id.h"
#include "rep-cache.h"

#include "private/svn_fs_util.h"
#include "../libsvn_fs/fs-loader.h"

#include "svn_private_config.h"

/* An arbitrary maximum path length, so clients can't run us out of memory
 * by giving us arbitrarily large paths. */
#define FSFS_MAX_PATH_LEN 4096

/* The default maximum number of files per directory to store in the
   rev and revprops directory.  The number below is somewhat arbitrary,
   and can be overriden by defining the macro while compiling; the
   figure of 1000 is reasonable for VFAT filesystems, which are by far
   the worst performers in this area. */
#ifndef SVN_FS_FS_DEFAULT_MAX_FILES_PER_DIR
#define SVN_FS_FS_DEFAULT_MAX_FILES_PER_DIR 1000
#endif

/* Following are defines that specify the textual elements of the
   native filesystem directories and revision files. */

/* Headers used to describe node-revision in the revision file. */
#define HEADER_ID          "id"
#define HEADER_TYPE        "type"
#define HEADER_COUNT       "count"
#define HEADER_PROPS       "props"
#define HEADER_TEXT        "text"
#define HEADER_CPATH       "cpath"
#define HEADER_PRED        "pred"
#define HEADER_COPYFROM    "copyfrom"
#define HEADER_COPYROOT    "copyroot"
#define HEADER_FRESHTXNRT  "is-fresh-txn-root"
#define HEADER_MINFO_HERE  "minfo-here"
#define HEADER_MINFO_CNT   "minfo-cnt"

/* Kinds that a change can be. */
#define ACTION_MODIFY      "modify"
#define ACTION_ADD         "add"
#define ACTION_DELETE      "delete"
#define ACTION_REPLACE     "replace"
#define ACTION_RESET       "reset"

/* True and False flags. */
#define FLAG_TRUE          "true"
#define FLAG_FALSE         "false"

/* Kinds that a node-rev can be. */
#define KIND_FILE          "file"
#define KIND_DIR           "dir"

/* Kinds of representation. */
#define REP_PLAIN          "PLAIN"
#define REP_DELTA          "DELTA"

/* Notes:

To avoid opening and closing the rev-files all the time, it would
probably be advantageous to keep each rev-file open for the
lifetime of the transaction object.  I'll leave that as a later
optimization for now.

I didn't keep track of pool lifetimes at all in this code.  There
are likely some errors because of that.

*/

/* The vtable associated with an open transaction object. */
static txn_vtable_t txn_vtable = {
  svn_fs_fs__commit_txn,
  svn_fs_fs__abort_txn,
  svn_fs_fs__txn_prop,
  svn_fs_fs__txn_proplist,
  svn_fs_fs__change_txn_prop,
  svn_fs_fs__txn_root,
  svn_fs_fs__change_txn_props
};

/* Pathname helper functions */

static const char *
path_format(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_FORMAT, pool);
}

static APR_INLINE const char *
path_uuid(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_UUID, pool);
}

const char *
svn_fs_fs__path_current(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_CURRENT, pool);
}

static APR_INLINE const char *
path_txn_current(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_TXN_CURRENT, pool);
}

static APR_INLINE const char *
path_txn_current_lock(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_TXN_CURRENT_LOCK, pool);
}

static APR_INLINE const char *
path_lock(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_LOCK_FILE, pool);
}

static const char *
path_rev_packed(svn_fs_t *fs, svn_revnum_t rev, const char *kind,
                apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  assert(ffd->max_files_per_dir);
  return svn_path_join_many(pool, fs->path, PATH_REVS_DIR,
                            apr_psprintf(pool, "%ld.%s",
                                         rev / ffd->max_files_per_dir, kind),
                            NULL);
}

static const char *
path_rev_shard(svn_fs_t *fs, svn_revnum_t rev, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  assert(ffd->max_files_per_dir);
  return svn_path_join_many(pool, fs->path, PATH_REVS_DIR,
                            apr_psprintf(pool, "%ld",
                                               rev / ffd->max_files_per_dir),
                            NULL);
}

const char *
svn_fs_fs__path_rev(svn_fs_t *fs, svn_revnum_t rev, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  if (ffd->max_files_per_dir)
    {
      return svn_path_join(path_rev_shard(fs, rev, pool),
                           apr_psprintf(pool, "%ld", rev),
                           pool);
    }

  return svn_path_join_many(pool, fs->path, PATH_REVS_DIR,
                            apr_psprintf(pool, "%ld", rev), NULL);
}

static const char *
path_revprops_shard(svn_fs_t *fs, svn_revnum_t rev, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  assert(ffd->max_files_per_dir);
  return svn_path_join_many(pool, fs->path, PATH_REVPROPS_DIR,
                            apr_psprintf(pool, "%ld",
                                               rev / ffd->max_files_per_dir),
                            NULL);
}

static const char *
path_revprops(svn_fs_t *fs, svn_revnum_t rev, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  if (ffd->max_files_per_dir)
    {
      return svn_path_join(path_revprops_shard(fs, rev, pool),
                           apr_psprintf(pool, "%ld", rev),
                           pool);
    }

  return svn_path_join_many(pool, fs->path, PATH_REVPROPS_DIR,
                            apr_psprintf(pool, "%ld", rev), NULL);
}

static APR_INLINE const char *
path_txn_dir(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  return svn_path_join_many(pool, fs->path, PATH_TXNS_DIR,
                            apr_pstrcat(pool, txn_id, PATH_EXT_TXN, NULL),
                            NULL);
}

static APR_INLINE const char *
path_txn_changes(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  return svn_path_join(path_txn_dir(fs, txn_id, pool), PATH_CHANGES, pool);
}

static APR_INLINE const char *
path_txn_props(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  return svn_path_join(path_txn_dir(fs, txn_id, pool), PATH_TXN_PROPS, pool);
}

static APR_INLINE const char *
path_txn_next_ids(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  return svn_path_join(path_txn_dir(fs, txn_id, pool), PATH_NEXT_IDS, pool);
}

static APR_INLINE const char *
path_min_unpacked_rev(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_path_join(fs->path, PATH_MIN_UNPACKED_REV, pool);
}

static APR_INLINE const char *
path_txn_proto_rev(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  if (ffd->format >= SVN_FS_FS__MIN_PROTOREVS_DIR_FORMAT)
    return svn_path_join_many(pool, fs->path, PATH_TXN_PROTOS_DIR,
                              apr_pstrcat(pool, txn_id, PATH_EXT_REV, NULL),
                              NULL);
  else
    return svn_path_join(path_txn_dir(fs, txn_id, pool), PATH_REV, pool);
}

static APR_INLINE const char *
path_txn_proto_rev_lock(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  if (ffd->format >= SVN_FS_FS__MIN_PROTOREVS_DIR_FORMAT)
    return svn_path_join_many(pool, fs->path, PATH_TXN_PROTOS_DIR,
                              apr_pstrcat(pool, txn_id, PATH_EXT_REV_LOCK,
                                          NULL),
                              NULL);
  else
    return svn_path_join(path_txn_dir(fs, txn_id, pool), PATH_REV_LOCK, pool);
}

static const char *
path_txn_node_rev(svn_fs_t *fs, const svn_fs_id_t *id, apr_pool_t *pool)
{
  const char *txn_id = svn_fs_fs__id_txn_id(id);
  const char *node_id = svn_fs_fs__id_node_id(id);
  const char *copy_id = svn_fs_fs__id_copy_id(id);
  const char *name = apr_psprintf(pool, PATH_PREFIX_NODE "%s.%s",
                                  node_id, copy_id);

  return svn_path_join(path_txn_dir(fs, txn_id, pool), name, pool);
}

static APR_INLINE const char *
path_txn_node_props(svn_fs_t *fs, const svn_fs_id_t *id, apr_pool_t *pool)
{
  return apr_pstrcat(pool, path_txn_node_rev(fs, id, pool), PATH_EXT_PROPS,
                     NULL);
}

static APR_INLINE const char *
path_txn_node_children(svn_fs_t *fs, const svn_fs_id_t *id, apr_pool_t *pool)
{
  return apr_pstrcat(pool, path_txn_node_rev(fs, id, pool),
                     PATH_EXT_CHILDREN, NULL);
}

static APR_INLINE const char *
path_node_origin(svn_fs_t *fs, const char *node_id, apr_pool_t *pool)
{
  int len = strlen(node_id);
  const char *node_id_minus_last_char =
    (len == 1) ? "0" : apr_pstrmemdup(pool, node_id, len - 1);
  return svn_path_join_many(pool, fs->path, PATH_NODE_ORIGINS_DIR,
                            node_id_minus_last_char, NULL);
}


/* Functions for working with shared transaction data. */

/* Return the transaction object for transaction TXN_ID from the
   transaction list of filesystem FS (which must already be locked via the
   txn_list_lock mutex).  If the transaction does not exist in the list,
   then create a new transaction object and return it (if CREATE_NEW is
   true) or return NULL (otherwise). */
static fs_fs_shared_txn_data_t *
get_shared_txn(svn_fs_t *fs, const char *txn_id, svn_boolean_t create_new)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  fs_fs_shared_data_t *ffsd = ffd->shared;
  fs_fs_shared_txn_data_t *txn;

  for (txn = ffsd->txns; txn; txn = txn->next)
    if (strcmp(txn->txn_id, txn_id) == 0)
      break;

  if (txn || !create_new)
    return txn;

  /* Use the transaction object from the (single-object) freelist,
     if one is available, or otherwise create a new object. */
  if (ffsd->free_txn)
    {
      txn = ffsd->free_txn;
      ffsd->free_txn = NULL;
    }
  else
    {
      apr_pool_t *subpool = svn_pool_create(ffsd->common_pool);
      txn = apr_palloc(subpool, sizeof(*txn));
      txn->pool = subpool;
    }

  assert(strlen(txn_id) < sizeof(txn->txn_id));
  strcpy(txn->txn_id, txn_id);
  txn->being_written = FALSE;

  /* Link this transaction into the head of the list.  We will typically
     be dealing with only one active transaction at a time, so it makes
     sense for searches through the transaction list to look at the
     newest transactions first.  */
  txn->next = ffsd->txns;
  ffsd->txns = txn;

  return txn;
}

/* Free the transaction object for transaction TXN_ID, and remove it
   from the transaction list of filesystem FS (which must already be
   locked via the txn_list_lock mutex).  Do nothing if the transaction
   does not exist. */
static void
free_shared_txn(svn_fs_t *fs, const char *txn_id)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  fs_fs_shared_data_t *ffsd = ffd->shared;
  fs_fs_shared_txn_data_t *txn, *prev = NULL;

  for (txn = ffsd->txns; txn; prev = txn, txn = txn->next)
    if (strcmp(txn->txn_id, txn_id) == 0)
      break;

  if (!txn)
    return;

  if (prev)
    prev->next = txn->next;
  else
    ffsd->txns = txn->next;

  /* As we typically will be dealing with one transaction after another,
     we will maintain a single-object free list so that we can hopefully
     keep reusing the same transaction object. */
  if (!ffsd->free_txn)
    ffsd->free_txn = txn;
  else
    svn_pool_destroy(txn->pool);
}


/* Obtain a lock on the transaction list of filesystem FS, call BODY
   with FS, BATON, and POOL, and then unlock the transaction list.
   Return what BODY returned. */
static svn_error_t *
with_txnlist_lock(svn_fs_t *fs,
                  svn_error_t *(*body)(svn_fs_t *fs,
                                       const void *baton,
                                       apr_pool_t *pool),
                  void *baton,
                  apr_pool_t *pool)
{
  svn_error_t *err;
#if APR_HAS_THREADS
  fs_fs_data_t *ffd = fs->fsap_data;
  fs_fs_shared_data_t *ffsd = ffd->shared;
  apr_status_t apr_err;

  apr_err = apr_thread_mutex_lock(ffsd->txn_list_lock);
  if (apr_err)
    return svn_error_wrap_apr(apr_err, _("Can't grab FSFS txn list mutex"));
#endif

  err = body(fs, baton, pool);

#if APR_HAS_THREADS
  apr_err = apr_thread_mutex_unlock(ffsd->txn_list_lock);
  if (apr_err && !err)
    return svn_error_wrap_apr(apr_err, _("Can't ungrab FSFS txn list mutex"));
#endif

  return err;
}


/* Get a lock on empty file LOCK_FILENAME, creating it in POOL. */
static svn_error_t *
get_lock_on_filesystem(const char *lock_filename,
                       apr_pool_t *pool)
{
  svn_error_t *err = svn_io_file_lock2(lock_filename, TRUE, FALSE, pool);

  if (err && APR_STATUS_IS_ENOENT(err->apr_err))
    {
      /* No lock file?  No big deal; these are just empty files
         anyway.  Create it and try again. */
      svn_error_clear(err);
      err = NULL;

      SVN_ERR(svn_io_file_create(lock_filename, "", pool));
      SVN_ERR(svn_io_file_lock2(lock_filename, TRUE, FALSE, pool));
    }

  return err;
}

/* Obtain a write lock on the file LOCK_FILENAME (protecting with
   LOCK_MUTEX if APR is threaded) in a subpool of POOL, call BODY with
   BATON and that subpool, destroy the subpool (releasing the write
   lock) and return what BODY returned. */
static svn_error_t *
with_some_lock(svn_error_t *(*body)(void *baton,
                                    apr_pool_t *pool),
               void *baton,
               const char *lock_filename,
#if APR_HAS_THREADS
               apr_thread_mutex_t *lock_mutex,
#endif
               apr_pool_t *pool)
{
  apr_pool_t *subpool = svn_pool_create(pool);
  svn_error_t *err;

#if APR_HAS_THREADS
  apr_status_t status;

  /* POSIX fcntl locks are per-process, so we need to serialize locks
     within the process. */
  status = apr_thread_mutex_lock(lock_mutex);
  if (status)
    return svn_error_wrap_apr(status,
                              _("Can't grab FSFS mutex for '%s'"),
                              lock_filename);
#endif

  err = get_lock_on_filesystem(lock_filename, subpool);

  if (!err)
    err = body(baton, subpool);

  svn_pool_destroy(subpool);

#if APR_HAS_THREADS
  status = apr_thread_mutex_unlock(lock_mutex);
  if (status && !err)
    return svn_error_wrap_apr(status,
                              _("Can't ungrab FSFS mutex for '%s'"),
                              lock_filename);
#endif

  return err;
}

svn_error_t *
svn_fs_fs__with_write_lock(svn_fs_t *fs,
                           svn_error_t *(*body)(void *baton,
                                                apr_pool_t *pool),
                           void *baton,
                           apr_pool_t *pool)
{
#if APR_HAS_THREADS
  fs_fs_data_t *ffd = fs->fsap_data;
  fs_fs_shared_data_t *ffsd = ffd->shared;
  apr_thread_mutex_t *mutex = ffsd->fs_write_lock;
#endif

  return with_some_lock(body, baton,
                        path_lock(fs, pool),
#if APR_HAS_THREADS
                        mutex,
#endif
                        pool);
}

/* Run BODY (with BATON and POOL) while the txn-current file
   of FS is locked. */
static svn_error_t *
with_txn_current_lock(svn_fs_t *fs,
                      svn_error_t *(*body)(void *baton,
                                           apr_pool_t *pool),
                      void *baton,
                      apr_pool_t *pool)
{
#if APR_HAS_THREADS
  fs_fs_data_t *ffd = fs->fsap_data;
  fs_fs_shared_data_t *ffsd = ffd->shared;
  apr_thread_mutex_t *mutex = ffsd->txn_current_lock;
#endif

  return with_some_lock(body, baton,
                        path_txn_current_lock(fs, pool),
#if APR_HAS_THREADS
                        mutex,
#endif
                        pool);
}

/* A structure used by unlock_proto_rev() and unlock_proto_rev_body(),
   which see. */
struct unlock_proto_rev_baton
{
  const char *txn_id;
  void *lockcookie;
};

/* Callback used in the implementation of unlock_proto_rev(). */
static svn_error_t *
unlock_proto_rev_body(svn_fs_t *fs, const void *baton, apr_pool_t *pool)
{
  const struct unlock_proto_rev_baton *b = baton;
  const char *txn_id = b->txn_id;
  apr_file_t *lockfile = b->lockcookie;
  fs_fs_shared_txn_data_t *txn = get_shared_txn(fs, txn_id, FALSE);
  apr_status_t apr_err;

  if (!txn)
    return svn_error_createf(SVN_ERR_FS_CORRUPT, NULL,
                             _("Can't unlock unknown transaction '%s'"),
                             txn_id);
  if (!txn->being_written)
    return svn_error_createf(SVN_ERR_FS_CORRUPT, NULL,
                             _("Can't unlock nonlocked transaction '%s'"),
                             txn_id);

  apr_err = apr_file_unlock(lockfile);
  if (apr_err)
    return svn_error_wrap_apr
      (apr_err,
       _("Can't unlock prototype revision lockfile for transaction '%s'"),
       txn_id);
  apr_err = apr_file_close(lockfile);
  if (apr_err)
    return svn_error_wrap_apr
      (apr_err,
       _("Can't close prototype revision lockfile for transaction '%s'"),
       txn_id);

  txn->being_written = FALSE;

  return SVN_NO_ERROR;
}

/* Unlock the prototype revision file for transaction TXN_ID in filesystem
   FS using cookie LOCKCOOKIE.  The original prototype revision file must
   have been closed _before_ calling this function.

   Perform temporary allocations in POOL. */
static svn_error_t *
unlock_proto_rev(svn_fs_t *fs, const char *txn_id, void *lockcookie,
                 apr_pool_t *pool)
{
  struct unlock_proto_rev_baton b;

  b.txn_id = txn_id;
  b.lockcookie = lockcookie;
  return with_txnlist_lock(fs, unlock_proto_rev_body, &b, pool);
}

/* Same as unlock_proto_rev(), but requires that the transaction list
   lock is already held. */
static svn_error_t *
unlock_proto_rev_list_locked(svn_fs_t *fs, const char *txn_id,
                             void *lockcookie,
                             apr_pool_t *pool)
{
  struct unlock_proto_rev_baton b;

  b.txn_id = txn_id;
  b.lockcookie = lockcookie;
  return unlock_proto_rev_body(fs, &b, pool);
}

/* A structure used by get_writable_proto_rev() and
   get_writable_proto_rev_body(), which see. */
struct get_writable_proto_rev_baton
{
  apr_file_t **file;
  void **lockcookie;
  const char *txn_id;
};

/* Callback used in the implementation of get_writable_proto_rev(). */
static svn_error_t *
get_writable_proto_rev_body(svn_fs_t *fs, const void *baton, apr_pool_t *pool)
{
  const struct get_writable_proto_rev_baton *b = baton;
  apr_file_t **file = b->file;
  void **lockcookie = b->lockcookie;
  const char *txn_id = b->txn_id;
  svn_error_t *err;
  fs_fs_shared_txn_data_t *txn = get_shared_txn(fs, txn_id, TRUE);

  /* First, ensure that no thread in this process (including this one)
     is currently writing to this transaction's proto-rev file. */
  if (txn->being_written)
    return svn_error_createf(SVN_ERR_FS_REP_BEING_WRITTEN, NULL,
                             _("Cannot write to the prototype revision file "
                               "of transaction '%s' because a previous "
                               "representation is currently being written by "
                               "this process"),
                             txn_id);


  /* We know that no thread in this process is writing to the proto-rev
     file, and by extension, that no thread in this process is holding a
     lock on the prototype revision lock file.  It is therefore safe
     for us to attempt to lock this file, to see if any other process
     is holding a lock. */

  {
    apr_file_t *lockfile;
    apr_status_t apr_err;
    const char *lockfile_path = path_txn_proto_rev_lock(fs, txn_id, pool);

    /* Open the proto-rev lockfile, creating it if necessary, as it may
       not exist if the transaction dates from before the lockfiles were
       introduced.

       ### We'd also like to use something like svn_io_file_lock2(), but
           that forces us to create a subpool just to be able to unlock
           the file, which seems a waste. */
    SVN_ERR(svn_io_file_open(&lockfile, lockfile_path,
                             APR_WRITE | APR_CREATE, APR_OS_DEFAULT, pool));

    apr_err = apr_file_lock(lockfile,
                            APR_FLOCK_EXCLUSIVE | APR_FLOCK_NONBLOCK);
    if (apr_err)
      {
        svn_error_clear(svn_io_file_close(lockfile, pool));

        if (APR_STATUS_IS_EAGAIN(apr_err))
          return svn_error_createf(SVN_ERR_FS_REP_BEING_WRITTEN, NULL,
                                   _("Cannot write to the prototype revision "
                                     "file of transaction '%s' because a "
                                     "previous representation is currently "
                                     "being written by another process"),
                                   txn_id);

        return svn_error_wrap_apr(apr_err,
                                  _("Can't get exclusive lock on file '%s'"),
                                  svn_path_local_style(lockfile_path, pool));
      }

    *lockcookie = lockfile;
  }

  /* We've successfully locked the transaction; mark it as such. */
  txn->being_written = TRUE;


  /* Now open the prototype revision file and seek to the end. */
  err = svn_io_file_open(file, path_txn_proto_rev(fs, txn_id, pool),
                         APR_WRITE | APR_BUFFERED, APR_OS_DEFAULT, pool);

  /* You might expect that we could dispense with the following seek
     and achieve the same thing by opening the file using APR_APPEND.
     Unfortunately, APR's buffered file implementation unconditionally
     places its initial file pointer at the start of the file (even for
     files opened with APR_APPEND), so we need this seek to reconcile
     the APR file pointer to the OS file pointer (since we need to be
     able to read the current file position later). */
  if (!err)
    {
      apr_off_t offset = 0;
      err = svn_io_file_seek(*file, APR_END, &offset, 0);
    }

  if (err)
    {
      svn_error_clear(unlock_proto_rev_list_locked(fs, txn_id, *lockcookie,
                                                   pool));
      *lockcookie = NULL;
    }

  return err;
}

/* Get a handle to the prototype revision file for transaction TXN_ID in
   filesystem FS, and lock it for writing.  Return FILE, a file handle
   positioned at the end of the file, and LOCKCOOKIE, a cookie that
   should be passed to unlock_proto_rev() to unlock the file once FILE
   has been closed.

   If the prototype revision file is already locked, return error
   SVN_ERR_FS_REP_BEING_WRITTEN.

   Perform all allocations in POOL. */
static svn_error_t *
get_writable_proto_rev(apr_file_t **file,
                       void **lockcookie,
                       svn_fs_t *fs, const char *txn_id,
                       apr_pool_t *pool)
{
  struct get_writable_proto_rev_baton b;

  b.file = file;
  b.lockcookie = lockcookie;
  b.txn_id = txn_id;

  return with_txnlist_lock(fs, get_writable_proto_rev_body, &b, pool);
}

/* Callback used in the implementation of purge_shared_txn(). */
static svn_error_t *
purge_shared_txn_body(svn_fs_t *fs, const void *baton, apr_pool_t *pool)
{
  const char *txn_id = *(const char **)baton;

  free_shared_txn(fs, txn_id);
  return SVN_NO_ERROR;
}

/* Purge the shared data for transaction TXN_ID in filesystem FS.
   Perform all allocations in POOL. */
static svn_error_t *
purge_shared_txn(svn_fs_t *fs, const char *txn_id, apr_pool_t *pool)
{
  return with_txnlist_lock(fs, purge_shared_txn_body, (char **) &txn_id, pool);
}



/* Fetch the current offset of FILE into *OFFSET_P. */
static svn_error_t *
get_file_offset(apr_off_t *offset_p, apr_file_t *file, apr_pool_t *pool)
{
  apr_off_t offset;

  /* Note that, for buffered files, one (possibly surprising) side-effect
     of this call is to flush any unwritten data to disk. */
  offset = 0;
  SVN_ERR(svn_io_file_seek(file, APR_CUR, &offset, pool));
  *offset_p = offset;

  return SVN_NO_ERROR;
}


/* Check that BUF, a buffer of text from format file PATH, contains
   only digits, raising error SVN_ERR_BAD_VERSION_FILE_FORMAT if not.

   Uses POOL for temporary allocation. */
static svn_error_t *
check_format_file_buffer_numeric(const char *buf, const char *path,
                                 apr_pool_t *pool)
{
  const char *p;

  for (p = buf; *p; p++)
    if (!apr_isdigit(*p))
      return svn_error_createf(SVN_ERR_BAD_VERSION_FILE_FORMAT, NULL,
        _("Format file '%s' contains an unexpected non-digit"),
        svn_path_local_style(path, pool));

  return SVN_NO_ERROR;
}

/* Read the format number and maximum number of files per directory
   from PATH and return them in *PFORMAT and *MAX_FILES_PER_DIR
   respectively.

   *MAX_FILES_PER_DIR is obtained from the 'layout' format option, and
   will be set to zero if a linear scheme should be used.

   Use POOL for temporary allocation. */
static svn_error_t *
read_format(int *pformat, int *max_files_per_dir,
            const char *path, apr_pool_t *pool)
{
  svn_error_t *err;
  apr_file_t *file;
  char buf[80];
  apr_size_t len;

  err = svn_io_file_open(&file, path, APR_READ | APR_BUFFERED,
                         APR_OS_DEFAULT, pool);
  if (err && APR_STATUS_IS_ENOENT(err->apr_err))
    {
      /* Treat an absent format file as format 1.  Do not try to
         create the format file on the fly, because the repository
         might be read-only for us, or this might be a read-only
         operation, and the spirit of FSFS is to make no changes
         whatseover in read-only operations.  See thread starting at
         http://subversion.tigris.org/servlets/ReadMsg?list=dev&msgNo=97600
         for more. */
      svn_error_clear(err);
      *pformat = 1;
      *max_files_per_dir = 0;

      return SVN_NO_ERROR;
    }

  len = sizeof(buf);
  err = svn_io_read_length_line(file, buf, &len, pool);
  if (err && APR_STATUS_IS_EOF(err->apr_err))
    {
      /* Return a more useful error message. */
      svn_error_clear(err);
      return svn_error_createf(SVN_ERR_BAD_VERSION_FILE_FORMAT, NULL,
                               _("Can't read first line of format file '%s'"),
                               svn_path_local_style(path, pool));
    }
  SVN_ERR(err);

  /* Check that the first line contains only digits. */
  SVN_ERR(check_format_file_buffer_numeric(buf, path, pool));
  *pformat = atoi(buf);

  /* Set the default values for anything that can be set via an option. */
  *max_files_per_dir = 0;

  /* Read any options. */
  while (1)
    {
      len = sizeof(buf);
      err = svn_io_read_length_line(file, buf, &len, pool);
      if (err && APR_STATUS_IS_EOF(err->apr_err))
        {
          /* No more options; that's okay. */
          svn_error_clear(err);
          break;
        }
      SVN_ERR(err);

      if (*pformat >= SVN_FS_FS__MIN_LAYOUT_FORMAT_OPTION_FORMAT &&
          strncmp(buf, "layout ", 7) == 0)
        {
          if (strcmp(buf+7, "linear") == 0)
            {
              *max_files_per_dir = 0;
              continue;
            }

          if (strncmp(buf+7, "sharded ", 8) == 0)
            {
              /* Check that the argument is numeric. */
              SVN_ERR(check_format_file_buffer_numeric(buf+15, path, pool));
              *max_files_per_dir = atoi(buf+15);
              continue;
            }
        }

      return svn_error_createf(SVN_ERR_BAD_VERSION_FILE_FORMAT, NULL,
         _("'%s' contains invalid filesystem format option '%s'"),
         svn_path_local_style(path, pool), buf);
    }

  return svn_io_file_close(file, pool);
}

/* Write the format number and maximum number of files per directory
   to a new format file in PATH, possibly expecting to overwrite a
   previously existing file.

   Use POOL for temporary allocation. */
static svn_error_t *
write_format(const char *path, int format, int max_files_per_dir,
             svn_boolean_t overwrite, apr_pool_t *pool)
{
  svn_stringbuf_t *sb;
  svn_string_t *contents;

  SVN_ERR_ASSERT(1 <= format && format <= SVN_FS_FS__FORMAT_NUMBER);

  sb = svn_stringbuf_createf(pool, "%d\n", format);

  if (format >= SVN_FS_FS__MIN_LAYOUT_FORMAT_OPTION_FORMAT)
    {
      if (max_files_per_dir)
        svn_stringbuf_appendcstr(sb, apr_psprintf(pool, "layout sharded %d\n",
                                                  max_files_per_dir));
      else
        svn_stringbuf_appendcstr(sb, "layout linear\n");
    }

  contents = svn_string_create_from_buf(sb, pool);

  /* svn_io_write_version_file() does a load of magic to allow it to
     replace version files that already exist.  We only need to do
     that when we're allowed to overwrite an existing file. */
  if (! overwrite)
    {
      /* Create the file */
      SVN_ERR(svn_io_file_create(path, contents->data, pool));
    }
  else
    {
      const char *path_tmp;

      SVN_ERR(svn_io_write_unique(&path_tmp,
                                  svn_path_dirname(path, pool),
                                  contents->data, contents->len,
                                  svn_io_file_del_none, pool));

#ifdef WIN32
      /* make the destination writable, but only on Windows, because
         Windows does not let us replace read-only files. */
      SVN_ERR(svn_io_set_file_read_write(path, TRUE, pool));
#endif /* WIN32 */

      /* rename the temp file as the real destination */
      SVN_ERR(svn_io_file_rename(path_tmp, path, pool));
    }

  /* And set the perms to make it read only */
  return svn_io_set_file_read_only(path, FALSE, pool);
}

/* Return the error SVN_ERR_FS_UNSUPPORTED_FORMAT if FS's format
   number is not the same as a format number supported by this
   Subversion. */
static svn_error_t *
check_format(int format)
{
  /* We support all formats from 1-current simultaneously */
  if (1 <= format && format <= SVN_FS_FS__FORMAT_NUMBER)
    return SVN_NO_ERROR;

  return svn_error_createf(SVN_ERR_FS_UNSUPPORTED_FORMAT, NULL,
     _("Expected FS format between '1' and '%d'; found format '%d'"),
     SVN_FS_FS__FORMAT_NUMBER, format);
}

svn_boolean_t
svn_fs_fs__fs_supports_mergeinfo(svn_fs_t *fs)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  return ffd->format >= SVN_FS_FS__MIN_MERGEINFO_FORMAT;
}

static svn_error_t *
write_config(svn_fs_t *fs,
             apr_pool_t *pool)
{
#define NL APR_EOL_STR
  static const char * const fsfs_conf_contents =
"### This file controls the configuration of the FSFS filesystem."           NL
""                                                                           NL
"[" SVN_CACHE_CONFIG_CATEGORY_MEMCACHED_SERVERS "]"                          NL
"### These options name memcached servers used to cache internal FSFS"       NL
"### data.  See http://www.danga.com/memcached/ for more information on"     NL
"### memcached.  To use memcached with FSFS, run one or more memcached"      NL
"### servers, and specify each of them as an option like so:"                NL
"# first-server = 127.0.0.1:11211"                                           NL
"# remote-memcached = mymemcached.corp.example.com:11212"                    NL
"### The option name is ignored; the value is of the form HOST:PORT."        NL
"### memcached servers can be shared between multiple repositories;"         NL
"### however, if you do this, you *must* ensure that repositories have"      NL
"### distinct UUIDs and paths, or else cached data from one repository"      NL
"### might be used by another accidentally.  Note also that memcached has"   NL
"### no authentication for reads or writes, so you must ensure that your"    NL
"### memcached servers are only accessible by trusted users."                NL
""                                                                           NL
"[" CONFIG_SECTION_CACHES "]"                                                NL
"### When a cache-related error occurs, normally Subversion ignores it"      NL
"### and continues, logging an error if the server is appropriately"         NL
"### configured (and ignoring it with file:// access).  To make"             NL
"### Subversion never ignore cache errors, uncomment this line."             NL
"# " CONFIG_OPTION_FAIL_STOP " = true"                                       NL
""                                                                           NL
"[" CONFIG_SECTION_REP_SHARING "]"                                           NL
"### To conserve space, the filesystem can optionally avoid storing"         NL
"### duplicate representations.  This comes at a slight cost in performace," NL
"### as maintaining a database of shared representations can increase"       NL
"### commit times.  The space savings are dependent upon the size of the"    NL
"### repository, the number of objects it contains and the amount of"        NL
"### duplication between them, usually a function of the branching and"      NL
"### merging process."                                                       NL
"###"                                                                        NL
"### The following parameter enables rep-sharing in the repository.  It can" NL
"### be switched on and off at will, but for best space-saving results"      NL
"### should be enabled consistently over the life of the repository."        NL
"### " CONFIG_OPTION_ENABLE_REP_SHARING " = true"                            NL

;
#undef NL
  return svn_io_file_create(svn_path_join(fs->path, PATH_CONFIG, pool),
                            fsfs_conf_contents, pool);
}

static svn_error_t *
read_min_unpacked_rev(svn_revnum_t *min_unpacked_rev,
                      const char *path,
                      apr_pool_t *pool)
{
  char buf[80];
  apr_file_t *file;
  apr_size_t len;
      
  SVN_ERR(svn_io_file_open(&file, path, APR_READ | APR_BUFFERED,
                           APR_OS_DEFAULT, pool));
  len = sizeof(buf);
  SVN_ERR(svn_io_read_length_line(file, buf, &len, pool));
  SVN_ERR(svn_io_file_close(file, pool));

  *min_unpacked_rev = SVN_STR_TO_REV(buf);
  return SVN_NO_ERROR;
}

static svn_error_t *
get_youngest(svn_revnum_t *youngest_p, const char *fs_path, apr_pool_t *pool);

svn_error_t *
svn_fs_fs__open(svn_fs_t *fs, const char *path, apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  apr_file_t *uuid_file;
  int format, max_files_per_dir;
  char buf[APR_UUID_FORMATTED_LENGTH + 2];
  apr_size_t limit;
  svn_boolean_t rep_sharing_allowed;

  fs->path = apr_pstrdup(fs->pool, path);

  /* Read the FS format number. */
  SVN_ERR(read_format(&format, &max_files_per_dir,
                      path_format(fs, pool), pool));

  /* Now we've got a format number no matter what. */
  ffd->format = format;
  ffd->max_files_per_dir = max_files_per_dir;
  SVN_ERR(check_format(format));

  /* Read in and cache the repository uuid. */
  SVN_ERR(svn_io_file_open(&uuid_file, path_uuid(fs, pool),
                           APR_READ | APR_BUFFERED, APR_OS_DEFAULT, pool));

  limit = sizeof(buf);
  SVN_ERR(svn_io_read_length_line(uuid_file, buf, &limit, pool));
  ffd->uuid = apr_pstrdup(fs->pool, buf);

  SVN_ERR(svn_io_file_close(uuid_file, pool));

  /* Read the min unpacked revision. */
  if (ffd->format >= SVN_FS_FS__MIN_PACKED_FORMAT)
    SVN_ERR(read_min_unpacked_rev(&ffd->min_unpacked_rev,
                                  path_min_unpacked_rev(fs, pool), pool));

  /* Read the configuration file. */
  SVN_ERR(svn_config_read(&ffd->config,
                          svn_path_join(fs->path, PATH_CONFIG, pool),
                          FALSE, fs->pool));
  SVN_ERR(svn_config_get_bool(ffd->config, &rep_sharing_allowed,
                              CONFIG_SECTION_REP_SHARING,
                              CONFIG_OPTION_ENABLE_REP_SHARING, TRUE));

  /* Open the rep cache. */
  if (ffd->format >= SVN_FS_FS__MIN_REP_SHARING_FORMAT
        && rep_sharing_allowed)
    SVN_ERR(svn_fs_fs__open_rep_cache(fs, fs->pool));

  return get_youngest(&(ffd->youngest_rev_cache), path, pool);
}

/* Wrapper around svn_io_file_create which ignores EEXIST. */
static svn_error_t *
create_file_ignore_eexist(const char *file,
                          const char *contents,
                          apr_pool_t *pool)
{
  svn_error_t *err = svn_io_file_create(file, contents, pool);
  if (err && APR_STATUS_IS_EEXIST(err->apr_err))
    {
      svn_error_clear(err);
      err = SVN_NO_ERROR;
    }
  return err;
}

static svn_error_t *
upgrade_body(void *baton, apr_pool_t *pool)
{
  svn_fs_t *fs = baton;
  int format, max_files_per_dir;
  const char *format_path = path_format(fs, pool);

  /* Read the FS format number and max-files-per-dir setting. */
  SVN_ERR(read_format(&format, &max_files_per_dir, format_path, pool));

  /* If we're already up-to-date, there's nothing to be done here. */
  if (format == SVN_FS_FS__FORMAT_NUMBER)
    return SVN_NO_ERROR;

  /* If our filesystem predates the existance of the 'txn-current
     file', make that file and its corresponding lock file. */
  if (format < SVN_FS_FS__MIN_TXN_CURRENT_FORMAT)
    {
      SVN_ERR(create_file_ignore_eexist(path_txn_current(fs, pool), "0\n",
                                        pool));
      SVN_ERR(create_file_ignore_eexist(path_txn_current_lock(fs, pool), "",
                                        pool));
    }

  /* If our filesystem predates the existance of the 'txn-protorevs'
     dir, make that directory.  */
  if (format < SVN_FS_FS__MIN_PROTOREVS_DIR_FORMAT)
    {
      /* We don't use path_txn_proto_rev() here because it expects
         we've already bumped our format. */
      SVN_ERR(svn_io_make_dir_recursively
              (svn_path_join(fs->path, PATH_TXN_PROTOS_DIR, pool), pool));
    }

  /* If our filesystem is new enough, write the min unpacked rev file. */
  if (format < SVN_FS_FS__MIN_PACKED_FORMAT)
    SVN_ERR(svn_io_file_create(path_min_unpacked_rev(fs, pool), "0\n", pool));

  /* Bump the format file. */
  return write_format(format_path, SVN_FS_FS__FORMAT_NUMBER, max_files_per_dir,
                      TRUE, pool);
}


svn_error_t *
svn_fs_fs__upgrade(svn_fs_t *fs, apr_pool_t *pool)
{
  return svn_fs_fs__with_write_lock(fs, upgrade_body, (void *)fs, pool);
}


/* SVN_ERR-like macros for dealing with recoverable errors on mutable files
 *
 * Revprops, current, and txn-current files are mutable; that is, they
 * change as part of normal fsfs operation, in constrat to revs files, or
 * the format file, which are written once at create (or upgrade) time.
 * When more than one host writes to the same repository, we will
 * sometimes see these recoverable errors when accesssing these files.
 *
 * These errors all relate to NFS, and thus we only use this retry code if
 * ESTALE is defined.
 *
 ** ESTALE
 *
 * In NFS v3 and under, the server doesn't track opened files.  If you
 * unlink(2) or rename(2) a file held open by another process *on the
 * same host*, that host's kernel typically renames the file to
 * .nfsXXXX and automatically deletes that when it's no longer open,
 * but this behavior is not required.
 *
 * For obvious reasons, this does not work *across hosts*.  No one
 * knows about the opened file; not the server, and not the deleting
 * client.  So the file vanishes, and the reader gets stale NFS file
 * handle.
 *
 ** EIO, ENOENT
 *
 * Some client implementations (at least the 2.6.18.5 kernel that ships
 * with Ubuntu Dapper) sometimes give spurious ENOENT (only on open) or
 * even EIO errors when trying to read these files that have been renamed
 * over on some other host.
 *
 ** Solution
 *
 * Wrap opens and reads of such files with RETRY_RECOVERABLE and
 * closes with IGNORE_RECOVERABLE.  Call these macros within a loop of
 * RECOVERABLE_RETRY_COUNT iterations (though, realistically, the
 * second try will succeed).  Make sure you put a break statement
 * after the close, at the end of your loop.  Immediately after your
 * loop, return err if err.
 *
 * You must initialize err to SVN_NO_ERROR and filehandle to NULL, as
 * these macros do not.
 */

#define RECOVERABLE_RETRY_COUNT 10

#ifdef ESTALE
#define RETRY_RECOVERABLE(err, filehandle, expr)                \
  {                                                             \
    svn_error_clear(err);                                       \
    err = (expr);                                               \
    if (err)                                                    \
      {                                                         \
        apr_status_t _e = APR_TO_OS_ERROR(err->apr_err);        \
        if ((_e == ESTALE) || (_e == EIO) || (_e == ENOENT)) {  \
          if (NULL != filehandle)                               \
            (void)apr_file_close(filehandle);                   \
          continue;                                             \
        }                                                       \
        return err;                                             \
      }                                                         \
  }
#define IGNORE_RECOVERABLE(err, expr)                           \
  {                                                             \
    svn_error_clear(err);                                       \
    err = (expr);                                               \
    if (err)                                                    \
      {                                                         \
        apr_status_t _e = APR_TO_OS_ERROR(err->apr_err);        \
        if ((_e != ESTALE) && (_e != EIO))                      \
          return err;                                           \
      }                                                         \
  }
#else
#define RETRY_RECOVERABLE(err, filehandle, expr)  SVN_ERR(expr)
#define IGNORE_RECOVERABLE(err, expr) SVN_ERR(expr)
#endif

/* Long enough to hold: "<svn_revnum_t> <node id> <copy id>\0"
 * 19 bytes for svn_revnum_t (room for 32 or 64 bit values)
 * + 2 spaces
 * + 26 bytes for each id (these are actually unbounded, so we just
 *   have to pick something; 2^64 is 13 bytes in base-36)
 * + 1 terminating null
 */
#define CURRENT_BUF_LEN 48

/* Read the 'current' file FNAME and store the contents in *BUF.
   Allocations are performed in POOL. */
static svn_error_t *
read_current(const char *fname, char **buf, apr_pool_t *pool)
{
  apr_file_t *revision_file = NULL;
  apr_size_t len;
  int i;
  svn_error_t *err = SVN_NO_ERROR;
  apr_pool_t *iterpool;

  *buf = apr_palloc(pool, CURRENT_BUF_LEN);
  iterpool = svn_pool_create(pool);
  for (i = 0; i < RECOVERABLE_RETRY_COUNT; i++)
    {
      svn_pool_clear(iterpool);

      RETRY_RECOVERABLE(err, revision_file,
                        svn_io_file_open(&revision_file, fname,
                                         APR_READ | APR_BUFFERED,
                                         APR_OS_DEFAULT, iterpool));

      len = CURRENT_BUF_LEN;
      RETRY_RECOVERABLE(err, revision_file,
                        svn_io_read_length_line(revision_file,
                                                *buf, &len, iterpool));
      IGNORE_RECOVERABLE(err, svn_io_file_close(revision_file, iterpool));

      break;
    }
  svn_pool_destroy(iterpool);

  return err;
}

/* Find the youngest revision in a repository at path FS_PATH and
   return it in *YOUNGEST_P.  Perform temporary allocations in
   POOL. */
static svn_error_t *
get_youngest(svn_revnum_t *youngest_p,
             const char *fs_path,
             apr_pool_t *pool)
{
  char *buf;

  SVN_ERR(read_current(svn_path_join(fs_path, PATH_CURRENT, pool),
                       &buf, pool));

  *youngest_p = SVN_STR_TO_REV(buf);

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__hotcopy(const char *src_path,
                   const char *dst_path,
                   apr_pool_t *pool)
{
  const char *src_subdir, *dst_subdir;
  svn_revnum_t youngest, rev;
  apr_pool_t *iterpool;
  svn_node_kind_t kind;
  int format, max_files_per_dir;

  /* Check format to be sure we know how to hotcopy this FS. */
  SVN_ERR(read_format(&format, &max_files_per_dir,
                      svn_path_join(src_path, PATH_FORMAT, pool),
                      pool));
  SVN_ERR(check_format(format));

  /* Copy the current file. */
  SVN_ERR(svn_io_dir_file_copy(src_path, dst_path, PATH_CURRENT, pool));

  /* Copy the uuid. */
  SVN_ERR(svn_io_dir_file_copy(src_path, dst_path, PATH_UUID, pool));

  /* Copy the min unpacked rev. */
  if (format >= SVN_FS_FS__MIN_PACKED_FORMAT)
    SVN_ERR(svn_io_dir_file_copy(src_path, dst_path, PATH_MIN_UNPACKED_REV,
                                 pool));

  /* Find the youngest revision from this current file. */
  SVN_ERR(get_youngest(&youngest, dst_path, pool));

  /* Copy the necessary rev files. */
  src_subdir = svn_path_join(src_path, PATH_REVS_DIR, pool);
  dst_subdir = svn_path_join(dst_path, PATH_REVS_DIR, pool);

  SVN_ERR(svn_io_make_dir_recursively(dst_subdir, pool));

  iterpool = svn_pool_create(pool);
  for (rev = 0; rev <= youngest; rev++)
    {
      const char *src_subdir_shard = src_subdir,
                 *dst_subdir_shard = dst_subdir;

      if (max_files_per_dir)
        {
          const char *shard = apr_psprintf(iterpool, "%ld",
                                           rev / max_files_per_dir);
          src_subdir_shard = svn_path_join(src_subdir, shard, iterpool);
          dst_subdir_shard = svn_path_join(dst_subdir, shard, iterpool);

          if (rev % max_files_per_dir == 0)
            {
              SVN_ERR(svn_io_dir_make(dst_subdir_shard, APR_OS_DEFAULT,
                                      iterpool));
              SVN_ERR(svn_fs_fs__dup_perms(dst_subdir_shard, dst_subdir,
                                           iterpool));
            }
        }

      SVN_ERR(svn_io_dir_file_copy(src_subdir_shard, dst_subdir_shard,
                                   apr_psprintf(iterpool, "%ld", rev),
                                   iterpool));
      svn_pool_clear(iterpool);
    }

  /* Copy the necessary revprop files. */
  src_subdir = svn_path_join(src_path, PATH_REVPROPS_DIR, pool);
  dst_subdir = svn_path_join(dst_path, PATH_REVPROPS_DIR, pool);

  SVN_ERR(svn_io_make_dir_recursively(dst_subdir, pool));

  for (rev = 0; rev <= youngest; rev++)
    {
      const char *src_subdir_shard = src_subdir,
                 *dst_subdir_shard = dst_subdir;

      svn_pool_clear(iterpool);

      if (max_files_per_dir)
        {
          const char *shard = apr_psprintf(iterpool, "%ld",
                                           rev / max_files_per_dir);
          src_subdir_shard = svn_path_join(src_subdir, shard, iterpool);
          dst_subdir_shard = svn_path_join(dst_subdir, shard, iterpool);

          if (rev % max_files_per_dir == 0)
            {
              SVN_ERR(svn_io_dir_make(dst_subdir_shard, APR_OS_DEFAULT,
                                      iterpool));
              SVN_ERR(svn_fs_fs__dup_perms(dst_subdir_shard, dst_subdir,
                                           iterpool));
            }
        }

      SVN_ERR(svn_io_dir_file_copy(src_subdir_shard, dst_subdir_shard,
                                   apr_psprintf(iterpool, "%ld", rev),
                                   iterpool));
    }

  svn_pool_destroy(iterpool);

  /* Make an empty transactions directory for now.  Eventually some
     method of copying in progress transactions will need to be
     developed.*/
  dst_subdir = svn_path_join(dst_path, PATH_TXNS_DIR, pool);
  SVN_ERR(svn_io_make_dir_recursively(dst_subdir, pool));
  if (format >= SVN_FS_FS__MIN_PROTOREVS_DIR_FORMAT)
    {
      dst_subdir = svn_path_join(dst_path, PATH_TXN_PROTOS_DIR, pool);
      SVN_ERR(svn_io_make_dir_recursively(dst_subdir, pool));
    }

  /* Now copy the locks tree. */
  src_subdir = svn_path_join(src_path, PATH_LOCKS_DIR, pool);
  SVN_ERR(svn_io_check_path(src_subdir, &kind, pool));
  if (kind == svn_node_dir)
    SVN_ERR(svn_io_copy_dir_recursively(src_subdir, dst_path,
                                        PATH_LOCKS_DIR, TRUE, NULL,
                                        NULL, pool));

  /* Now copy the node-origins cache tree. */
  src_subdir = svn_path_join(src_path, PATH_NODE_ORIGINS_DIR, pool);
  SVN_ERR(svn_io_check_path(src_subdir, &kind, pool));
  if (kind == svn_node_dir)
    SVN_ERR(svn_io_copy_dir_recursively(src_subdir, dst_path,
                                        PATH_NODE_ORIGINS_DIR, TRUE, NULL,
                                        NULL, pool));

  /* Now copy the rep cache. */
  src_subdir = svn_path_join(src_path, REP_CACHE_DB_NAME, pool);
  SVN_ERR(svn_io_check_path(src_subdir, &kind, pool));
  if (kind == svn_node_file)
    SVN_ERR(svn_io_dir_file_copy(src_path, dst_path, REP_CACHE_DB_NAME, pool));

  /* Copy the txn-current file. */
  if (format >= SVN_FS_FS__MIN_TXN_CURRENT_FORMAT)
    SVN_ERR(svn_io_dir_file_copy(src_path, dst_path, PATH_TXN_CURRENT, pool));

  /* Hotcopied FS is complete. Stamp it with a format file. */
  return write_format(svn_path_join(dst_path, PATH_FORMAT, pool),
                      format, max_files_per_dir, FALSE, pool);
}

svn_error_t *
svn_fs_fs__youngest_rev(svn_revnum_t *youngest_p,
                        svn_fs_t *fs,
                        apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  SVN_ERR(get_youngest(youngest_p, fs->path, pool));
  ffd->youngest_rev_cache = *youngest_p;

  return SVN_NO_ERROR;
}

/* Given a revision file FILE that has been pre-positioned at the
   beginning of a Node-Rev header block, read in that header block and
   store it in the apr_hash_t HEADERS.  All allocations will be from
   POOL. */
static svn_error_t * read_header_block(apr_hash_t **headers,
                                       svn_stream_t *stream,
                                       apr_pool_t *pool)
{
  *headers = apr_hash_make(pool);

  while (1)
    {
      svn_stringbuf_t *header_str;
      const char *name, *value;
      apr_size_t i = 0;
      svn_boolean_t eof;

      SVN_ERR(svn_stream_readline(stream, &header_str, "\n", &eof, pool));

      if (eof || header_str->len == 0)
        break; /* end of header block */

      while (header_str->data[i] != ':')
        {
          if (header_str->data[i] == '\0')
            return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                    _("Found malformed header in "
                                      "revision file"));
          i++;
        }

      /* Create a 'name' string and point to it. */
      header_str->data[i] = '\0';
      name = header_str->data;

      /* Skip over the NULL byte and the space following it. */
      i += 2;

      if (i > header_str->len)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Found malformed header in "
                                  "revision file"));

      value = header_str->data + i;

      /* header_str is safely in our pool, so we can use bits of it as
         key and value. */
      apr_hash_set(*headers, name, APR_HASH_KEY_STRING, value);
    }

  return SVN_NO_ERROR;
}

/* Return SVN_ERR_FS_NO_SUCH_REVISION if the given revision is newer
   than the current youngest revision or is simply not a valid
   revision number, else return success.

   FSFS is based around the concept that commits only take effect when
   the number in "current" is bumped.  Thus if there happens to be a rev
   or revprops file installed for a revision higher than the one recorded
   in "current" (because a commit failed between installing the rev file
   and bumping "current", or because an administrator rolled back the
   repository by resetting "current" without deleting rev files, etc), it
   ought to be completely ignored.  This function provides the check
   by which callers can make that decision. */
static svn_error_t *
ensure_revision_exists(svn_fs_t *fs,
                       svn_revnum_t rev,
                       apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;

  if (! SVN_IS_VALID_REVNUM(rev))
    return svn_error_createf(SVN_ERR_FS_NO_SUCH_REVISION, NULL,
                             _("Invalid revision number '%ld'"), rev);


  /* Did the revision exist the last time we checked the current
     file? */
  if (rev <= ffd->youngest_rev_cache)
    return SVN_NO_ERROR;

  SVN_ERR(get_youngest(&(ffd->youngest_rev_cache), fs->path, pool));

  /* Check again. */
  if (rev <= ffd->youngest_rev_cache)
    return SVN_NO_ERROR;

  return svn_error_createf(SVN_ERR_FS_NO_SUCH_REVISION, NULL,
                           _("No such revision %ld"), rev);
}

/* Open the correct revision file for REV.  If the filesystem FS has
   been packed, *FILE will be set to the packed file; otherwise, set *FILE
   to the revision file for REV.  Use POOL for allocations. */
static svn_error_t *
open_pack_or_rev_file(apr_file_t **file,
                      svn_fs_t *fs,
                      svn_revnum_t rev,
                      apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  svn_error_t *err;

   err = svn_io_file_open(file,
                          rev < ffd->min_unpacked_rev
                              ? path_rev_packed(fs, rev, "pack", pool)
                              : svn_fs_fs__path_rev(fs, rev, pool),
                          APR_READ | APR_BUFFERED, APR_OS_DEFAULT, pool);

  if (err && APR_STATUS_IS_ENOENT(err->apr_err))
    {
      svn_error_clear(err);
      return svn_error_createf(SVN_ERR_FS_NO_SUCH_REVISION, NULL,
                               _("No such revision %ld"), rev);
    }

  return err;
}

/* Given REV in FS, set *REV_OFFSET to REV's offset in the packed file.
   Use POOL for temporary allocations. */
static svn_error_t *
get_packed_offset(apr_off_t *rev_offset,
                  svn_fs_t *fs,
                  svn_revnum_t rev,
                  apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  svn_stream_t *manifest_stream;
  svn_boolean_t is_cached;
  apr_off_t *cached_rev_offset;
  svn_revnum_t tmp_rev, start_rev, end_rev;
  apr_pool_t *iterpool;

  SVN_ERR(svn_cache__get((void **) &cached_rev_offset, &is_cached,
                         ffd->packed_offset_cache, &rev, pool));

  if (is_cached)
    {
      *rev_offset = *cached_rev_offset;
      return SVN_NO_ERROR;
    }

  /* Open the manifest file. */
  SVN_ERR(svn_stream_open_readonly(&manifest_stream,
                                   path_rev_packed(fs, rev, "manifest", pool),
                                   pool, pool));

  /* While we're here, let's just read the entire manifest into our cache. */
  start_rev = rev - (rev % ffd->max_files_per_dir);
  end_rev = start_rev + ffd->max_files_per_dir - 1;
  iterpool = svn_pool_create(pool);
  for (tmp_rev = start_rev; tmp_rev <= end_rev; tmp_rev++)
    {
      svn_stringbuf_t *sb;
      apr_off_t offset;
      svn_boolean_t eof;

      svn_pool_clear(iterpool);
      SVN_ERR(svn_stream_readline(manifest_stream, &sb, "\n", &eof, iterpool));
      offset = apr_atoi64(svn_string_create_from_buf(sb, iterpool)->data);
      SVN_ERR(svn_cache__set(ffd->packed_offset_cache, &tmp_rev, &offset,
                             iterpool));
    }
  svn_pool_destroy(iterpool);

  /* Close everything up, and get the value we're interested in from the
     cache. */
  SVN_ERR(svn_stream_close(manifest_stream));

  SVN_ERR(svn_cache__get((void **) &cached_rev_offset, &is_cached,
                         ffd->packed_offset_cache, &rev, pool));
  *rev_offset = *cached_rev_offset;

  return SVN_NO_ERROR;
}

/* Open the revision file for revision REV in filesystem FS and store
   the newly opened file in FILE.  Seek to location OFFSET before
   returning.  Perform temporary allocations in POOL. */
static svn_error_t *
open_and_seek_revision(apr_file_t **file,
                       svn_fs_t *fs,
                       svn_revnum_t rev,
                       apr_off_t offset,
                       apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  apr_file_t *rev_file;

  SVN_ERR(ensure_revision_exists(fs, rev, pool));

  SVN_ERR(open_pack_or_rev_file(&rev_file, fs, rev, pool));

  if (rev < ffd->min_unpacked_rev)
    {
      apr_off_t rev_offset;

      SVN_ERR(get_packed_offset(&rev_offset, fs, rev, pool));
      offset += rev_offset;
    }

  SVN_ERR(svn_io_file_seek(rev_file, APR_SET, &offset, pool));

  *file = rev_file;

  return SVN_NO_ERROR;
}

/* Open the representation for a node-revision in transaction TXN_ID
   in filesystem FS and store the newly opened file in FILE.  Seek to
   location OFFSET before returning.  Perform temporary allocations in
   POOL.  Only appropriate for file contents, nor props or directory
   contents. */
static svn_error_t *
open_and_seek_transaction(apr_file_t **file,
                          svn_fs_t *fs,
                          const char *txn_id,
                          representation_t *rep,
                          apr_pool_t *pool)
{
  apr_file_t *rev_file;
  apr_off_t offset;

  SVN_ERR(svn_io_file_open(&rev_file, path_txn_proto_rev(fs, txn_id, pool),
                           APR_READ | APR_BUFFERED, APR_OS_DEFAULT, pool));

  offset = rep->offset;
  SVN_ERR(svn_io_file_seek(rev_file, APR_SET, &offset, pool));

  *file = rev_file;

  return SVN_NO_ERROR;
}

/* Given a node-id ID, and a representation REP in filesystem FS, open
   the correct file and seek to the correction location.  Store this
   file in *FILE_P.  Perform any allocations in POOL. */
static svn_error_t *
open_and_seek_representation(apr_file_t **file_p,
                             svn_fs_t *fs,
                             representation_t *rep,
                             apr_pool_t *pool)
{
  if (! rep->txn_id)
    return open_and_seek_revision(file_p, fs, rep->revision, rep->offset,
                                  pool);
  else
    return open_and_seek_transaction(file_p, fs, rep->txn_id, rep, pool);
}

/* Parse the description of a representation from STRING and store it
   into *REP_P.  If the representation is mutable (the revision is
   given as -1), then use TXN_ID for the representation's txn_id
   field.  If MUTABLE_REP_TRUNCATED is true, then this representation
   is for property or directory contents, and no information will be
   expected except the "-1" revision number for a mutable
   representation.  Allocate *REP_P in POOL. */
static svn_error_t *
read_rep_offsets(representation_t **rep_p,
                 char *string,
                 const char *txn_id,
                 svn_boolean_t mutable_rep_truncated,
                 apr_pool_t *pool)
{
  representation_t *rep;
  char *str, *last_str;

  rep = apr_pcalloc(pool, sizeof(*rep));
  *rep_p = rep;

  str = apr_strtok(string, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));


  rep->revision = SVN_STR_TO_REV(str);
  if (rep->revision == SVN_INVALID_REVNUM)
    {
      rep->txn_id = txn_id;
      if (mutable_rep_truncated)
        return SVN_NO_ERROR;
    }

  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));

  rep->offset = apr_atoi64(str);

  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));

  rep->size = apr_atoi64(str);

  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));

  rep->expanded_size = apr_atoi64(str);

  /* Read in the MD5 hash. */
  str = apr_strtok(NULL, " ", &last_str);
  if ((str == NULL) || (strlen(str) != (APR_MD5_DIGESTSIZE * 2)))
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));

  SVN_ERR(svn_checksum_parse_hex(&rep->md5_checksum, svn_checksum_md5, str,
                                 pool));

  /* The remaining fields are only used for formats >= 4, so check that. */
  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return SVN_NO_ERROR;

  /* Read the SHA1 hash. */
  if (strlen(str) != (APR_SHA1_DIGESTSIZE * 2))
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Malformed text rep offset line in node-rev"));

  SVN_ERR(svn_checksum_parse_hex(&rep->sha1_checksum, svn_checksum_sha1, str,
                                 pool));

  /* Read the reuse count. */
  str = apr_strtok(NULL, " ", &last_str);
  rep->reuse_count = apr_atoi64(str);

  return SVN_NO_ERROR;
}

/* See svn_fs_fs__get_node_revision, which wraps this and adds another
   error. */
static svn_error_t *
get_node_revision_body(node_revision_t **noderev_p,
                       svn_fs_t *fs,
                       const svn_fs_id_t *id,
                       apr_pool_t *pool)
{
  apr_file_t *revision_file;
  svn_error_t *err;

  if (svn_fs_fs__id_txn_id(id))
    {
      /* This is a transaction node-rev. */
      err = svn_io_file_open(&revision_file, path_txn_node_rev(fs, id, pool),
                             APR_READ | APR_BUFFERED, APR_OS_DEFAULT, pool);
    }
  else
    {
      /* This is a revision node-rev. */
      err = open_and_seek_revision(&revision_file, fs,
                                   svn_fs_fs__id_rev(id),
                                   svn_fs_fs__id_offset(id),
                                   pool);
    }

  if (err)
    {
      if (APR_STATUS_IS_ENOENT(err->apr_err))
        {
          svn_error_clear(err);
          return svn_fs_fs__err_dangling_id(fs, id);
        }

      return err;
    }

  return svn_fs_fs__read_noderev(noderev_p,
                                 svn_stream_from_aprfile2(revision_file, FALSE,
                                                          pool),
                                 pool);
}

svn_error_t *
svn_fs_fs__read_noderev(node_revision_t **noderev_p,
                        svn_stream_t *stream,
                        apr_pool_t *pool)
{
  apr_hash_t *headers;
  node_revision_t *noderev;
  char *value;

  SVN_ERR(read_header_block(&headers, stream, pool));

  noderev = apr_pcalloc(pool, sizeof(*noderev));

  /* Read the node-rev id. */
  value = apr_hash_get(headers, HEADER_ID, APR_HASH_KEY_STRING);
  if (value == NULL)
      return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                              _("Missing id field in node-rev"));

  SVN_ERR(svn_stream_close(stream));

  noderev->id = svn_fs_fs__id_parse(value, strlen(value), pool);

  /* Read the type. */
  value = apr_hash_get(headers, HEADER_TYPE, APR_HASH_KEY_STRING);

  if ((value == NULL) ||
      (strcmp(value, KIND_FILE) != 0 && strcmp(value, KIND_DIR)))
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Missing kind field in node-rev"));

  noderev->kind = (strcmp(value, KIND_FILE) == 0) ? svn_node_file
    : svn_node_dir;

  /* Read the 'count' field. */
  value = apr_hash_get(headers, HEADER_COUNT, APR_HASH_KEY_STRING);
  noderev->predecessor_count = (value == NULL) ? 0 : atoi(value);

  /* Get the properties location. */
  value = apr_hash_get(headers, HEADER_PROPS, APR_HASH_KEY_STRING);
  if (value)
    {
      SVN_ERR(read_rep_offsets(&noderev->prop_rep, value,
                               svn_fs_fs__id_txn_id(noderev->id), TRUE, pool));
    }

  /* Get the data location. */
  value = apr_hash_get(headers, HEADER_TEXT, APR_HASH_KEY_STRING);
  if (value)
    {
      SVN_ERR(read_rep_offsets(&noderev->data_rep, value,
                               svn_fs_fs__id_txn_id(noderev->id),
                               (noderev->kind == svn_node_dir), pool));
    }

  /* Get the created path. */
  value = apr_hash_get(headers, HEADER_CPATH, APR_HASH_KEY_STRING);
  if (value == NULL)
    {
      return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                              _("Missing cpath in node-rev"));
    }
  else
    {
      noderev->created_path = apr_pstrdup(pool, value);
    }

  /* Get the predecessor ID. */
  value = apr_hash_get(headers, HEADER_PRED, APR_HASH_KEY_STRING);
  if (value)
    noderev->predecessor_id = svn_fs_fs__id_parse(value, strlen(value),
                                                  pool);

  /* Get the copyroot. */
  value = apr_hash_get(headers, HEADER_COPYROOT, APR_HASH_KEY_STRING);
  if (value == NULL)
    {
      noderev->copyroot_path = apr_pstrdup(pool, noderev->created_path);
      noderev->copyroot_rev = svn_fs_fs__id_rev(noderev->id);
    }
  else
    {
      char *str, *last_str;

      str = apr_strtok(value, " ", &last_str);
      if (str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Malformed copyroot line in node-rev"));

      noderev->copyroot_rev = atoi(str);

      if (last_str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Malformed copyroot line in node-rev"));
      noderev->copyroot_path = apr_pstrdup(pool, last_str);
    }

  /* Get the copyfrom. */
  value = apr_hash_get(headers, HEADER_COPYFROM, APR_HASH_KEY_STRING);
  if (value == NULL)
    {
      noderev->copyfrom_path = NULL;
      noderev->copyfrom_rev = SVN_INVALID_REVNUM;
    }
  else
    {
      char *str, *last_str;

      str = apr_strtok(value, " ", &last_str);
      if (str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Malformed copyfrom line in node-rev"));

      noderev->copyfrom_rev = atoi(str);

      if (last_str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Malformed copyfrom line in node-rev"));
      noderev->copyfrom_path = apr_pstrdup(pool, last_str);
    }

  /* Get whether this is a fresh txn root. */
  value = apr_hash_get(headers, HEADER_FRESHTXNRT, APR_HASH_KEY_STRING);
  noderev->is_fresh_txn_root = (value != NULL);

  /* Get the mergeinfo count. */
  value = apr_hash_get(headers, HEADER_MINFO_CNT, APR_HASH_KEY_STRING);
  noderev->mergeinfo_count = (value == NULL) ? 0 : apr_atoi64(value);

  /* Get whether *this* node has mergeinfo. */
  value = apr_hash_get(headers, HEADER_MINFO_HERE, APR_HASH_KEY_STRING);
  noderev->has_mergeinfo = (value != NULL);

  *noderev_p = noderev;

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__get_node_revision(node_revision_t **noderev_p,
                             svn_fs_t *fs,
                             const svn_fs_id_t *id,
                             apr_pool_t *pool)
{
  svn_error_t *err = get_node_revision_body(noderev_p, fs, id, pool);
  if (err && err->apr_err == SVN_ERR_FS_CORRUPT)
    {
      svn_string_t *id_string = svn_fs_fs__id_unparse(id, pool);
      return svn_error_createf(SVN_ERR_FS_CORRUPT, err,
                               "Corrupt node-revision '%s'",
                               id_string->data);
    }
  return err;
}


/* Return a formatted string, compatible with filesystem format FORMAT,
   that represents the location of representation REP.  If
   MUTABLE_REP_TRUNCATED is given, the rep is for props or dir contents,
   and only a "-1" revision number will be given for a mutable rep.
   Perform the allocation from POOL.  */
static const char *
representation_string(representation_t *rep,
                      int format,
                      svn_boolean_t mutable_rep_truncated,
                      apr_pool_t *pool)
{
  if (rep->txn_id && mutable_rep_truncated)
    return "-1";

  if (format < SVN_FS_FS__MIN_REP_SHARING_FORMAT)
    return apr_psprintf(pool, "%ld %" APR_OFF_T_FMT " %" SVN_FILESIZE_T_FMT
                        " %" SVN_FILESIZE_T_FMT " %s",
                        rep->revision, rep->offset, rep->size,
                        rep->expanded_size,
                        svn_checksum_to_cstring_display(rep->md5_checksum,
                                                        pool));

  return apr_psprintf(pool, "%ld %" APR_OFF_T_FMT " %" SVN_FILESIZE_T_FMT
                      " %" SVN_FILESIZE_T_FMT " %s %s %" APR_INT64_T_FMT,
                      rep->revision, rep->offset, rep->size,
                      rep->expanded_size,
                      svn_checksum_to_cstring_display(rep->md5_checksum,
                                                      pool),
                      rep->sha1_checksum ?
                          svn_checksum_to_cstring_display(rep->sha1_checksum,
                                                          pool) :
                          "0000000000000000000000000000000000000000",
                      rep->reuse_count);
}


svn_error_t *
svn_fs_fs__write_noderev(svn_stream_t *outfile,
                         node_revision_t *noderev,
                         int format,
                         svn_boolean_t include_mergeinfo,
                         apr_pool_t *pool)
{
  SVN_ERR(svn_stream_printf(outfile, pool, HEADER_ID ": %s\n",
                            svn_fs_fs__id_unparse(noderev->id,
                                                  pool)->data));

  SVN_ERR(svn_stream_printf(outfile, pool, HEADER_TYPE ": %s\n",
                            (noderev->kind == svn_node_file) ?
                            KIND_FILE : KIND_DIR));

  if (noderev->predecessor_id)
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_PRED ": %s\n",
                              svn_fs_fs__id_unparse(noderev->predecessor_id,
                                                    pool)->data));

  SVN_ERR(svn_stream_printf(outfile, pool, HEADER_COUNT ": %d\n",
                            noderev->predecessor_count));

  if (noderev->data_rep)
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_TEXT ": %s\n",
                              representation_string(noderev->data_rep,
                                                    format,
                                                    (noderev->kind
                                                     == svn_node_dir),
                                                    pool)));

  if (noderev->prop_rep)
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_PROPS ": %s\n",
                              representation_string(noderev->prop_rep, format,
                                                    TRUE, pool)));

  SVN_ERR(svn_stream_printf(outfile, pool, HEADER_CPATH ": %s\n",
                            noderev->created_path));

  if (noderev->copyfrom_path)
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_COPYFROM ": %ld"
                              " %s\n",
                              noderev->copyfrom_rev,
                              noderev->copyfrom_path));

  if ((noderev->copyroot_rev != svn_fs_fs__id_rev(noderev->id)) ||
      (strcmp(noderev->copyroot_path, noderev->created_path) != 0))
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_COPYROOT ": %ld"
                              " %s\n",
                              noderev->copyroot_rev,
                              noderev->copyroot_path));

  if (noderev->is_fresh_txn_root)
    SVN_ERR(svn_stream_printf(outfile, pool, HEADER_FRESHTXNRT ": y\n"));

  if (include_mergeinfo)
    {
      if (noderev->mergeinfo_count > 0)
        SVN_ERR(svn_stream_printf(outfile, pool, HEADER_MINFO_CNT ": %"
                                  APR_INT64_T_FMT "\n",
                                  noderev->mergeinfo_count));

      if (noderev->has_mergeinfo)
        SVN_ERR(svn_stream_printf(outfile, pool, HEADER_MINFO_HERE ": y\n"));
    }

  return svn_stream_printf(outfile, pool, "\n");
}

svn_error_t *
svn_fs_fs__put_node_revision(svn_fs_t *fs,
                             const svn_fs_id_t *id,
                             node_revision_t *noderev,
                             svn_boolean_t fresh_txn_root,
                             apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  apr_file_t *noderev_file;
  const char *txn_id = svn_fs_fs__id_txn_id(id);

  noderev->is_fresh_txn_root = fresh_txn_root;

  if (! txn_id)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Attempted to write to non-transaction"));

  SVN_ERR(svn_io_file_open(&noderev_file, path_txn_node_rev(fs, id, pool),
                           APR_WRITE | APR_CREATE | APR_TRUNCATE
                           | APR_BUFFERED, APR_OS_DEFAULT, pool));

  SVN_ERR(svn_fs_fs__write_noderev(svn_stream_from_aprfile2(noderev_file, TRUE,
                                                            pool),
                                   noderev, ffd->format,
                                   svn_fs_fs__fs_supports_mergeinfo(fs),
                                   pool));

  return svn_io_file_close(noderev_file, pool);
}


/* This structure is used to hold the information associated with a
   REP line. */
struct rep_args
{
  svn_boolean_t is_delta;
  svn_boolean_t is_delta_vs_empty;

  svn_revnum_t base_revision;
  apr_off_t base_offset;
  apr_size_t base_length;
};

/* Read the next line from file FILE and parse it as a text
   representation entry.  Return the parsed entry in *REP_ARGS_P.
   Perform all allocations in POOL. */
static svn_error_t *
read_rep_line(struct rep_args **rep_args_p,
              apr_file_t *file,
              apr_pool_t *pool)
{
  char buffer[160];
  apr_size_t limit;
  struct rep_args *rep_args;
  char *str, *last_str;

  limit = sizeof(buffer);
  SVN_ERR(svn_io_read_length_line(file, buffer, &limit, pool));

  rep_args = apr_pcalloc(pool, sizeof(*rep_args));
  rep_args->is_delta = FALSE;

  if (strcmp(buffer, REP_PLAIN) == 0)
    {
      *rep_args_p = rep_args;
      return SVN_NO_ERROR;
    }

  if (strcmp(buffer, REP_DELTA) == 0)
    {
      /* This is a delta against the empty stream. */
      rep_args->is_delta = TRUE;
      rep_args->is_delta_vs_empty = TRUE;
      *rep_args_p = rep_args;
      return SVN_NO_ERROR;
    }

  rep_args->is_delta = TRUE;
  rep_args->is_delta_vs_empty = FALSE;

  /* We have hopefully a DELTA vs. a non-empty base revision. */
  str = apr_strtok(buffer, " ", &last_str);
  if (! str || (strcmp(str, REP_DELTA) != 0)) goto err;

  str = apr_strtok(NULL, " ", &last_str);
  if (! str) goto err;
  rep_args->base_revision = atol(str);

  str = apr_strtok(NULL, " ", &last_str);
  if (! str) goto err;
  rep_args->base_offset = (apr_off_t) apr_atoi64(str);

  str = apr_strtok(NULL, " ", &last_str);
  if (! str) goto err;
  rep_args->base_length = (apr_size_t) apr_atoi64(str);

  *rep_args_p = rep_args;
  return SVN_NO_ERROR;

 err:
  return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                          _("Malformed representation header"));
}

/* Given a revision file REV_FILE, opened to REV in FS, find the Node-ID
   of the header located at OFFSET and store it in *ID_P.  Allocate
   temporary variables from POOL. */
static svn_error_t *
get_fs_id_at_offset(svn_fs_id_t **id_p,
                    apr_file_t *rev_file,
                    svn_fs_t *fs,
                    svn_revnum_t rev,
                    apr_off_t offset,
                    apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  svn_fs_id_t *id;
  apr_hash_t *headers;
  const char *node_id_str;

  if (rev < ffd->min_unpacked_rev)
    {
      apr_off_t rev_offset;

      SVN_ERR(get_packed_offset(&rev_offset, fs, rev, pool));
      offset += rev_offset;
    }

  SVN_ERR(svn_io_file_seek(rev_file, APR_SET, &offset, pool));

  SVN_ERR(read_header_block(&headers,
                            svn_stream_from_aprfile2(rev_file, TRUE, pool),
                            pool));

  node_id_str = apr_hash_get(headers, HEADER_ID, APR_HASH_KEY_STRING);

  if (node_id_str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Missing node-id in node-rev"));

  id = svn_fs_fs__id_parse(node_id_str, strlen(node_id_str), pool);

  if (id == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Corrupt node-id in node-rev"));

  *id_p = id;

  return SVN_NO_ERROR;
}


/* Given an open revision file REV_FILE in FS for REV, locate the trailer that
   specifies the offset to the root node-id and to the changed path
   information.  Store the root node offset in *ROOT_OFFSET and the
   changed path offset in *CHANGES_OFFSET.  If either of these
   pointers is NULL, do nothing with it.  If PACKED is true, REV_FILE
   should be a packed shard file.  Allocate temporary variables from POOL. */
static svn_error_t *
get_root_changes_offset(apr_off_t *root_offset,
                        apr_off_t *changes_offset,
                        apr_file_t *rev_file,
                        svn_fs_t *fs,
                        svn_revnum_t rev,
                        apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  apr_off_t offset;
  char buf[64];
  int i, num_bytes;
  apr_size_t len;
  apr_seek_where_t seek_relative;

  /* Determine where to seek to in the file.

     If we've got a pack file, we want to seek to the end of the desired
     revision.  But we don't track that, so we seek to the beginning of the
     next revision.

     Unless the next revision is in a different file, in which case, we can
     just seek to the end of the pack file -- just like we do in the
     non-packed case. */
  if ((rev < ffd->min_unpacked_rev) && ((rev + 1) % ffd->max_files_per_dir != 0))
    {
      SVN_ERR(get_packed_offset(&offset, fs, rev + 1, pool));
      seek_relative = APR_SET;
    }
  else
    {
      seek_relative = APR_END;
      offset = 0;
    }

  /* We will assume that the last line containing the two offsets
     will never be longer than 64 characters. */
  SVN_ERR(svn_io_file_seek(rev_file, seek_relative, &offset, pool));

  offset -= sizeof(buf);
  SVN_ERR(svn_io_file_seek(rev_file, APR_SET, &offset, pool));

  /* Read in this last block, from which we will identify the last line. */
  len = sizeof(buf);
  SVN_ERR(svn_io_file_read(rev_file, buf, &len, pool));

  /* This cast should be safe since the maximum amount read, 64, will
     never be bigger than the size of an int. */
  num_bytes = (int) len;

  /* The last byte should be a newline. */
  if (buf[num_bytes - 1] != '\n')
    {
      return svn_error_createf(SVN_ERR_FS_CORRUPT, NULL,
                               _("Revision file lacks trailing newline"));
    }

  /* Look for the next previous newline. */
  for (i = num_bytes - 2; i >= 0; i--)
    {
      if (buf[i] == '\n') break;
    }

  if (i < 0)
    {
      return svn_error_createf(SVN_ERR_FS_CORRUPT, NULL,
                               _("Final line in revision file longer than 64 "
                                 "characters"));
    }

  i++;

  if (root_offset)
    *root_offset = apr_atoi64(&buf[i]);

  /* find the next space */
  for ( ; i < (num_bytes - 2) ; i++)
    if (buf[i] == ' ') break;

  if (i == (num_bytes - 2))
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Final line in revision file missing space"));

  i++;

  /* note that apr_atoi64() will stop reading as soon as it encounters
     the final newline. */
  if (changes_offset)
    *changes_offset = apr_atoi64(&buf[i]);

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__rev_get_root(svn_fs_id_t **root_id_p,
                        svn_fs_t *fs,
                        svn_revnum_t rev,
                        apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  apr_file_t *revision_file;
  apr_off_t root_offset;
  svn_fs_id_t *root_id;
  svn_boolean_t is_cached;

  SVN_ERR(ensure_revision_exists(fs, rev, pool));

  SVN_ERR(svn_cache__get((void **) root_id_p, &is_cached,
                         ffd->rev_root_id_cache, &rev, pool));
  if (is_cached)
    return SVN_NO_ERROR;

  SVN_ERR(open_pack_or_rev_file(&revision_file, fs, rev, pool));
  SVN_ERR(get_root_changes_offset(&root_offset, NULL, revision_file, fs, rev,
                                  pool));

  SVN_ERR(get_fs_id_at_offset(&root_id, revision_file, fs, rev,
                              root_offset, pool));

  SVN_ERR(svn_io_file_close(revision_file, pool));

  SVN_ERR(svn_cache__set(ffd->rev_root_id_cache, &rev, root_id, pool));

  *root_id_p = root_id;

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__set_revision_proplist(svn_fs_t *fs,
                                 svn_revnum_t rev,
                                 apr_hash_t *proplist,
                                 apr_pool_t *pool)
{
  const char *final_path = path_revprops(fs, rev, pool);
  const char *tmp_path;
  svn_stream_t *stream;

  SVN_ERR(ensure_revision_exists(fs, rev, pool));

  /* ### do we have a directory sitting around already? we really shouldn't
     ### have to get the dirname here. */
  SVN_ERR(svn_stream_open_unique(&stream, &tmp_path,
                                 svn_path_dirname(final_path, pool  ),
                                 svn_io_file_del_none, pool, pool));
  SVN_ERR(svn_hash_write2(proplist, stream, SVN_HASH_TERMINATOR, pool));
  SVN_ERR(svn_stream_close(stream));

  /* We use the rev file of this revision as the perms reference,
     because when setting revprops for the first time, the revprop
     file won't exist and therefore can't serve as its own reference.
     (Whereas the rev file should already exist at this point.) */
  return svn_fs_fs__move_into_place(tmp_path, final_path,
                                    svn_fs_fs__path_rev(fs, rev, pool),
                                    pool);
}

svn_error_t *
svn_fs_fs__revision_proplist(apr_hash_t **proplist_p,
                             svn_fs_t *fs,
                             svn_revnum_t rev,
                             apr_pool_t *pool)
{
  apr_file_t *revprop_file = NULL;
  apr_hash_t *proplist;
  svn_error_t *err = SVN_NO_ERROR;
  int i;
  apr_pool_t *iterpool;

  SVN_ERR(ensure_revision_exists(fs, rev, pool));

  proplist = apr_hash_make(pool);
  iterpool = svn_pool_create(pool);
  for (i = 0; i < RECOVERABLE_RETRY_COUNT; i++)
    {
      svn_pool_clear(iterpool);

      /* Clear err here rather than after finding a recoverable error so
       * we can return that error on the last iteration of the loop. */
      svn_error_clear(err);
      err = svn_io_file_open(&revprop_file, path_revprops(fs, rev, iterpool),
                             APR_READ | APR_BUFFERED, APR_OS_DEFAULT,
                             iterpool);
      if (err)
        {
          if (APR_STATUS_IS_ENOENT(err->apr_err))
            {
              svn_error_clear(err);
              return svn_error_createf(SVN_ERR_FS_NO_SUCH_REVISION, NULL,
                                       _("No such revision %ld"), rev);
            }
#ifdef ESTALE
          else if (APR_TO_OS_ERROR(err->apr_err) == ESTALE
                   || APR_TO_OS_ERROR(err->apr_err) == EIO
                   || APR_TO_OS_ERROR(err->apr_err) == ENOENT)
            continue;
#endif
          return err;
        }

      SVN_ERR(svn_hash__clear(proplist));
      RETRY_RECOVERABLE(err, revprop_file,
                        svn_hash_read2(proplist,
                                       svn_stream_from_aprfile2(revprop_file,
                                                                TRUE,
                                                                iterpool),
                                       SVN_HASH_TERMINATOR, pool));

      IGNORE_RECOVERABLE(err, svn_io_file_close(revprop_file, iterpool));

      break;
    }
  if (err)
    return err;
  svn_pool_destroy(iterpool);

  *proplist_p = proplist;

  return SVN_NO_ERROR;
}

/* Represents where in the current svndiff data block each
   representation is. */
struct rep_state
{
  apr_file_t *file;
  apr_off_t start;  /* The starting offset for the raw
                       svndiff/plaintext data minus header. */
  apr_off_t off;    /* The current offset into the file. */
  apr_off_t end;    /* The end offset of the raw data. */
  int ver;          /* If a delta, what svndiff version? */
  int chunk_index;
};

/* See create_rep_state, which wraps this and adds another error. */
static svn_error_t *
create_rep_state_body(struct rep_state **rep_state,
                      struct rep_args **rep_args,
                      representation_t *rep,
                      svn_fs_t *fs,
                      apr_pool_t *pool)
{
  struct rep_state *rs = apr_pcalloc(pool, sizeof(*rs));
  struct rep_args *ra;
  unsigned char buf[4];

  SVN_ERR(open_and_seek_representation(&rs->file, fs, rep, pool));
  SVN_ERR(read_rep_line(&ra, rs->file, pool));
  SVN_ERR(get_file_offset(&rs->start, rs->file, pool));
  rs->off = rs->start;
  rs->end = rs->start + rep->size;
  *rep_state = rs;
  *rep_args = ra;

  if (ra->is_delta == FALSE)
    /* This is a plaintext, so just return the current rep_state. */
    return SVN_NO_ERROR;

  /* We are dealing with a delta, find out what version. */
  SVN_ERR(svn_io_file_read_full(rs->file, buf, sizeof(buf), NULL, pool));
  if (! ((buf[0] == 'S') && (buf[1] == 'V') && (buf[2] == 'N')))
    return svn_error_create
      (SVN_ERR_FS_CORRUPT, NULL,
       _("Malformed svndiff data in representation"));
  rs->ver = buf[3];
  rs->chunk_index = 0;
  rs->off += 4;

  return SVN_NO_ERROR;
}

/* Read the rep args for REP in filesystem FS and create a rep_state
   for reading the representation.  Return the rep_state in *REP_STATE
   and the rep args in *REP_ARGS, both allocated in POOL. */
static svn_error_t *
create_rep_state(struct rep_state **rep_state,
                 struct rep_args **rep_args,
                 representation_t *rep,
                 svn_fs_t *fs,
                 apr_pool_t *pool)
{
  svn_error_t *err = create_rep_state_body(rep_state, rep_args, rep, fs, pool);
  if (err && err->apr_err == SVN_ERR_FS_CORRUPT)
    {
      fs_fs_data_t *ffd = fs->fsap_data;

      /* ### This always returns "-1" for transaction reps, because
         ### this particular bit of code doesn't know if the rep is
         ### stored in the protorev or in the mutable area (for props
         ### or dir contents).  It is pretty rare for FSFS to *read*
         ### from the protorev file, though, so this is probably OK.
         ### And anyone going to debug corruption errors is probably
         ### going to jump straight to this comment anyway! */
      return svn_error_createf(SVN_ERR_FS_CORRUPT, err,
                               "Corrupt representation '%s'",
                               representation_string(rep, ffd->format, TRUE,
                                                     pool));
    }
  return err;
}

/* Build an array of rep_state structures in *LIST giving the delta
   reps from first_rep to a plain-text or self-compressed rep.  Set
   *SRC_STATE to the plain-text rep we find at the end of the chain,
   or to NULL if the final delta representation is self-compressed.
   The representation to start from is designated by filesystem FS, id
   ID, and representation REP. */
static svn_error_t *
build_rep_list(apr_array_header_t **list,
               struct rep_state **src_state,
               svn_fs_t *fs,
               representation_t *first_rep,
               apr_pool_t *pool)
{
  representation_t rep;
  struct rep_state *rs;
  struct rep_args *rep_args;

  *list = apr_array_make(pool, 1, sizeof(struct rep_state *));
  rep = *first_rep;

  while (1)
    {
      SVN_ERR(create_rep_state(&rs, &rep_args, &rep, fs, pool));
      if (rep_args->is_delta == FALSE)
        {
          /* This is a plaintext, so just return the current rep_state. */
          *src_state = rs;
          return SVN_NO_ERROR;
        }

      /* Push this rep onto the list.  If it's self-compressed, we're done. */
      APR_ARRAY_PUSH(*list, struct rep_state *) = rs;
      if (rep_args->is_delta_vs_empty)
        {
          *src_state = NULL;
          return SVN_NO_ERROR;
        }

      rep.revision = rep_args->base_revision;
      rep.offset = rep_args->base_offset;
      rep.size = rep_args->base_length;
      rep.txn_id = NULL;
    }
}


struct rep_read_baton
{
  /* The FS from which we're reading. */
  svn_fs_t *fs;

  /* The state of all prior delta representations. */
  apr_array_header_t *rs_list;

  /* The plaintext state, if there is a plaintext. */
  struct rep_state *src_state;

  /* The index of the current delta chunk, if we are reading a delta. */
  int chunk_index;

  /* The buffer where we store undeltified data. */
  char *buf;
  apr_size_t buf_pos;
  apr_size_t buf_len;

  /* A checksum context for summing the data read in order to verify it.
     Note: we don't need to use the sha1 checksum because we're only doing
     data verification, for which md5 is perfectly safe.  */
  svn_checksum_ctx_t *md5_checksum_ctx;

  svn_boolean_t checksum_finalized;

  /* The stored checksum of the representation we are reading, its
     length, and the amount we've read so far.  Some of this
     information is redundant with rs_list and src_state, but it's
     convenient for the checksumming code to have it here. */
  svn_checksum_t *md5_checksum;

  svn_filesize_t len;
  svn_filesize_t off;

  /* The key for the fulltext cache for this rep, if there is a
     fulltext cache. */
  const char *fulltext_cache_key;
  /* The text we've been reading, if we're going to cache it. */
  svn_stringbuf_t *current_fulltext;

  /* Used for temporary allocations during the read. */
  apr_pool_t *pool;

  /* Pool used to store file handles and other data that is persistant
     for the entire stream read. */
  apr_pool_t *filehandle_pool;
};

/* Create a rep_read_baton structure for node revision NODEREV in
   filesystem FS and store it in *RB_P.  If FULLTEXT_CACHE_KEY is not
   NULL, it is the rep's key in the fulltext cache, and a stringbuf
   must be allocated to store the text.  Perform all allocations in
   POOL.  If rep is mutable, it must be for file contents. */
static svn_error_t *
rep_read_get_baton(struct rep_read_baton **rb_p,
                   svn_fs_t *fs,
                   representation_t *rep,
                   const char *fulltext_cache_key,
                   apr_pool_t *pool)
{
  struct rep_read_baton *b;

  b = apr_pcalloc(pool, sizeof(*b));
  b->fs = fs;
  b->chunk_index = 0;
  b->buf = NULL;
  b->md5_checksum_ctx = svn_checksum_ctx_create(svn_checksum_md5, pool);
  b->checksum_finalized = FALSE;
  b->md5_checksum = svn_checksum_dup(rep->md5_checksum, pool);
  b->len = rep->expanded_size;
  b->off = 0;
  b->fulltext_cache_key = fulltext_cache_key;
  b->pool = svn_pool_create(pool);
  b->filehandle_pool = svn_pool_create(pool);

  if (fulltext_cache_key)
    b->current_fulltext = svn_stringbuf_create("", b->filehandle_pool);
  else
    b->current_fulltext = NULL;

  SVN_ERR(build_rep_list(&b->rs_list, &b->src_state, fs, rep,
                         b->filehandle_pool));

  /* Save our output baton. */
  *rb_p = b;

  return SVN_NO_ERROR;
}

/* Skip forwards to THIS_CHUNK in REP_STATE and then read the next delta
   window into *NWIN. */
static svn_error_t *
read_window(svn_txdelta_window_t **nwin, int this_chunk, struct rep_state *rs,
            apr_pool_t *pool)
{
  svn_stream_t *stream;

  SVN_ERR_ASSERT(rs->chunk_index <= this_chunk);

  /* Skip windows to reach the current chunk if we aren't there yet. */
  while (rs->chunk_index < this_chunk)
    {
      SVN_ERR(svn_txdelta_skip_svndiff_window(rs->file, rs->ver, pool));
      rs->chunk_index++;
      SVN_ERR(get_file_offset(&rs->off, rs->file, pool));
      if (rs->off >= rs->end)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Reading one svndiff window read "
                                  "beyond the end of the "
                                  "representation"));
    }

  /* Read the next window. */
  stream = svn_stream_from_aprfile2(rs->file, TRUE, pool);
  SVN_ERR(svn_txdelta_read_svndiff_window(nwin, stream, rs->ver, pool));
  rs->chunk_index++;
  SVN_ERR(get_file_offset(&rs->off, rs->file, pool));

  if (rs->off > rs->end)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Reading one svndiff window read beyond "
                              "the end of the representation"));

  return SVN_NO_ERROR;
}

/* Get one delta window that is a result of combining all but the last deltas
   from the current desired representation identified in *RB, to its
   final base representation.  Store the window in *RESULT. */
static svn_error_t *
get_combined_window(svn_txdelta_window_t **result,
                    struct rep_read_baton *rb)
{
  apr_pool_t *pool, *new_pool;
  int i;
  svn_txdelta_window_t *window, *nwin;
  struct rep_state *rs;

  SVN_ERR_ASSERT(rb->rs_list->nelts >= 2);

  pool = svn_pool_create(rb->pool);

  /* Read the next window from the original rep. */
  rs = APR_ARRAY_IDX(rb->rs_list, 0, struct rep_state *);
  SVN_ERR(read_window(&window, rb->chunk_index, rs, pool));

  /* Combine in the windows from the other delta reps, if needed. */
  for (i = 1; i < rb->rs_list->nelts - 1; i++)
    {
      if (window->src_ops == 0)
        break;

      rs = APR_ARRAY_IDX(rb->rs_list, i, struct rep_state *);

      SVN_ERR(read_window(&nwin, rb->chunk_index, rs, pool));

      /* Combine this window with the current one.  Cycle pools so that we
         only need to hold three windows at a time. */
      new_pool = svn_pool_create(rb->pool);
      window = svn_txdelta_compose_windows(nwin, window, new_pool);
      svn_pool_destroy(pool);
      pool = new_pool;
    }

  *result = window;
  return SVN_NO_ERROR;
}

static svn_error_t *
rep_read_contents_close(void *baton)
{
  struct rep_read_baton *rb = baton;

  svn_pool_destroy(rb->pool);
  svn_pool_destroy(rb->filehandle_pool);

  return SVN_NO_ERROR;
}

/* Return the next *LEN bytes of the rep and store them in *BUF. */
static svn_error_t *
get_contents(struct rep_read_baton *rb,
             char *buf,
             apr_size_t *len)
{
  apr_size_t copy_len, remaining = *len, tlen;
  char *sbuf, *tbuf, *cur = buf;
  struct rep_state *rs;
  svn_txdelta_window_t *cwindow, *lwindow;

  /* Special case for when there are no delta reps, only a plain
     text. */
  if (rb->rs_list->nelts == 0)
    {
      copy_len = remaining;
      rs = rb->src_state;
      if (((apr_off_t) copy_len) > rs->end - rs->off)
        copy_len = (apr_size_t) (rs->end - rs->off);
      SVN_ERR(svn_io_file_read_full(rs->file, cur, copy_len, NULL,
                                    rb->pool));
      rs->off += copy_len;
      *len = copy_len;
      return SVN_NO_ERROR;
    }

  while (remaining > 0)
    {
      /* If we have buffered data from a previous chunk, use that. */
      if (rb->buf)
        {
          /* Determine how much to copy from the buffer. */
          copy_len = rb->buf_len - rb->buf_pos;
          if (copy_len > remaining)
            copy_len = remaining;

          /* Actually copy the data. */
          memcpy(cur, rb->buf + rb->buf_pos, copy_len);
          rb->buf_pos += copy_len;
          cur += copy_len;
          remaining -= copy_len;

          /* If the buffer is all used up, clear it and empty the
             local pool. */
          if (rb->buf_pos == rb->buf_len)
            {
              svn_pool_clear(rb->pool);
              rb->buf = NULL;
            }
        }
      else
        {

          rs = APR_ARRAY_IDX(rb->rs_list, 0, struct rep_state *);
          if (rs->off == rs->end)
            break;

          /* Get more buffered data by evaluating a chunk. */
          if (rb->rs_list->nelts > 1)
            SVN_ERR(get_combined_window(&cwindow, rb));
          else
            cwindow = NULL;
          if (!cwindow || cwindow->src_ops > 0)
            {
              rs = APR_ARRAY_IDX(rb->rs_list, rb->rs_list->nelts - 1,
                                 struct rep_state *);
              /* Read window from last representation in list. */
              /* We apply this window directly instead of combining it
                 with the others.  We do this because vdelta used to
                 be used for deltas against the empty stream, which
                 will trigger quadratic behaviour in the delta
                 combiner.  It's still likely that we'll find such
                 deltas in an old repository; it may be worth
                 considering whether or not this special case is still
                 needed in the future, though. */
              SVN_ERR(read_window(&lwindow, rb->chunk_index, rs, rb->pool));

              if (lwindow->src_ops > 0)
                {
                  if (! rb->src_state)
                    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                            _("svndiff data requested "
                                              "non-existent source"));
                  rs = rb->src_state;
                  sbuf = apr_palloc(rb->pool, lwindow->sview_len);
                  if (! ((rs->start + lwindow->sview_offset) < rs->end))
                    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                            _("svndiff requested position "
                                              "beyond end of stream"));
                  if ((rs->start + lwindow->sview_offset) != rs->off)
                    {
                      rs->off = rs->start + lwindow->sview_offset;
                      SVN_ERR(svn_io_file_seek(rs->file, APR_SET, &rs->off,
                                               rb->pool));
                    }
                  SVN_ERR(svn_io_file_read_full(rs->file, sbuf,
                                                lwindow->sview_len,
                                                NULL, rb->pool));
                  rs->off += lwindow->sview_len;
                }
              else
                sbuf = NULL;

              /* Apply lwindow to source. */
              tlen = lwindow->tview_len;
              tbuf = apr_palloc(rb->pool, tlen);
              svn_txdelta_apply_instructions(lwindow, sbuf, tbuf,
                                             &tlen);
              if (tlen != lwindow->tview_len)
                return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                        _("svndiff window length is "
                                          "corrupt"));
              sbuf = tbuf;
            }
          else
            sbuf = NULL;

          rb->chunk_index++;

          if (cwindow)
            {
              rb->buf_len = cwindow->tview_len;
              rb->buf = apr_palloc(rb->pool, rb->buf_len);
              svn_txdelta_apply_instructions(cwindow, sbuf, rb->buf,
                                             &rb->buf_len);
              if (rb->buf_len != cwindow->tview_len)
                return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                        _("svndiff window length is "
                                          "corrupt"));
            }
          else
            {
              rb->buf_len = lwindow->tview_len;
              rb->buf = sbuf;
            }

          rb->buf_pos = 0;
        }
    }

  *len = cur - buf;

  return SVN_NO_ERROR;
}

/* BATON is of type `rep_read_baton'; read the next *LEN bytes of the
   representation and store them in *BUF.  Sum as we read and verify
   the MD5 sum at the end. */
static svn_error_t *
rep_read_contents(void *baton,
                  char *buf,
                  apr_size_t *len)
{
  struct rep_read_baton *rb = baton;

  /* Get the next block of data. */
  SVN_ERR(get_contents(rb, buf, len));

  if (rb->current_fulltext)
    svn_stringbuf_appendbytes(rb->current_fulltext, buf, *len);

  /* Perform checksumming.  We want to check the checksum as soon as
     the last byte of data is read, in case the caller never performs
     a short read, but we don't want to finalize the MD5 context
     twice. */
  if (!rb->checksum_finalized)
    {
      SVN_ERR(svn_checksum_update(rb->md5_checksum_ctx, buf, *len));
      rb->off += *len;
      if (rb->off == rb->len)
        {
          svn_checksum_t *md5_checksum;

          rb->checksum_finalized = TRUE;
          svn_checksum_final(&md5_checksum, rb->md5_checksum_ctx, rb->pool);
          if (!svn_checksum_match(md5_checksum, rb->md5_checksum))
            return svn_error_createf
              (SVN_ERR_FS_CORRUPT, NULL,
               _("Checksum mismatch while reading representation:\n"
                 "   expected:  %s\n"
                 "     actual:  %s\n"),
               svn_checksum_to_cstring_display(rb->md5_checksum, rb->pool),
               svn_checksum_to_cstring_display(md5_checksum, rb->pool));
        }
    }

  if (rb->off == rb->len && rb->current_fulltext)
    {
      fs_fs_data_t *ffd = rb->fs->fsap_data;
      SVN_ERR(svn_cache__set(ffd->fulltext_cache, rb->fulltext_cache_key,
                             rb->current_fulltext, rb->pool));
      rb->current_fulltext = NULL;
    }

  return SVN_NO_ERROR;
}


/* Returns whether or not the expanded fulltext of the file is
 * cachable based on its size SIZE.  Specifically, if it will fit
 * into a memcached value.  The memcached cutoff seems to be a bit
 * (header length?) under a megabyte; we round down a little to be
 * safe.
 */
static svn_boolean_t
fulltext_size_is_cachable(svn_filesize_t size)
{
  return size < 1000000;
}

/* Return a stream in *CONTENTS_P that will read the contents of a
   representation stored at the location given by REP.  Appropriate
   for any kind of immutable representation, but only for file
   contents (not props or directory contents) in mutable
   representations.

   If REP is NULL, the representation is assumed to be empty, and the
   empty stream is returned.
*/
static svn_error_t *
read_representation(svn_stream_t **contents_p,
                    svn_fs_t *fs,
                    representation_t *rep,
                    apr_pool_t *pool)
{
  if (! rep)
    {
      *contents_p = svn_stream_empty(pool);
    }
  else
    {
      fs_fs_data_t *ffd = fs->fsap_data;
      const char *fulltext_key = NULL;
      struct rep_read_baton *rb;

      if (ffd->fulltext_cache && SVN_IS_VALID_REVNUM(rep->revision)
          && fulltext_size_is_cachable(rep->expanded_size))
        {
          svn_string_t *fulltext;
          svn_boolean_t is_cached;
          fulltext_key = apr_psprintf(pool, "%ld/%" APR_OFF_T_FMT,
                                      rep->revision, rep->offset);
          SVN_ERR(svn_cache__get((void **) &fulltext, &is_cached,
                                 ffd->fulltext_cache, fulltext_key, pool));
          if (is_cached)
            {
              *contents_p = svn_stream_from_string(fulltext, pool);
              return SVN_NO_ERROR;
            }
        }

      SVN_ERR(rep_read_get_baton(&rb, fs, rep, fulltext_key, pool));

      *contents_p = svn_stream_create(rb, pool);
      svn_stream_set_read(*contents_p, rep_read_contents);
      svn_stream_set_close(*contents_p, rep_read_contents_close);
    }

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__get_contents(svn_stream_t **contents_p,
                        svn_fs_t *fs,
                        node_revision_t *noderev,
                        apr_pool_t *pool)
{
  return read_representation(contents_p, fs, noderev->data_rep, pool);
}

/* Baton used when reading delta windows. */
struct delta_read_baton
{
  struct rep_state *rs;
  svn_checksum_t *checksum;
};

/* This implements the svn_txdelta_next_window_fn_t interface. */
static svn_error_t *
delta_read_next_window(svn_txdelta_window_t **window, void *baton,
                       apr_pool_t *pool)
{
  struct delta_read_baton *drb = baton;

  if (drb->rs->off == drb->rs->end)
    {
      *window = NULL;
      return SVN_NO_ERROR;
    }

  return read_window(window, drb->rs->chunk_index, drb->rs, pool);
}

/* This implements the svn_txdelta_md5_digest_fn_t interface. */
static const unsigned char *
delta_read_md5_digest(void *baton)
{
  struct delta_read_baton *drb = baton;

  if (drb->checksum->kind == svn_checksum_md5)
    return drb->checksum->digest;
  else
    return NULL;
}

svn_error_t *
svn_fs_fs__get_file_delta_stream(svn_txdelta_stream_t **stream_p,
                                 svn_fs_t *fs,
                                 node_revision_t *source,
                                 node_revision_t *target,
                                 apr_pool_t *pool)
{
  svn_stream_t *source_stream, *target_stream;

  /* Try a shortcut: if the target is stored as a delta against the source,
     then just use that delta. */
  if (source && source->data_rep && target->data_rep)
    {
      struct rep_state *rep_state;
      struct rep_args *rep_args;

      /* Read target's base rep if any. */
      SVN_ERR(create_rep_state(&rep_state, &rep_args, target->data_rep,
                               fs, pool));
      /* If that matches source, then use this delta as is. */
      if (rep_args->is_delta
          && (rep_args->is_delta_vs_empty
              || (rep_args->base_revision == source->data_rep->revision
                  && rep_args->base_offset == source->data_rep->offset)))
        {
          /* Create the delta read baton. */
          struct delta_read_baton *drb = apr_pcalloc(pool, sizeof(*drb));
          drb->rs = rep_state;
          drb->checksum = svn_checksum_dup(target->data_rep->md5_checksum,
                                           pool);
          *stream_p = svn_txdelta_stream_create(drb, delta_read_next_window,
                                                delta_read_md5_digest, pool);
          return SVN_NO_ERROR;
        }
      else
        SVN_ERR(svn_io_file_close(rep_state->file, pool));
    }

  /* Read both fulltexts and construct a delta. */
  if (source)
    SVN_ERR(read_representation(&source_stream, fs, source->data_rep, pool));
  else
    source_stream = svn_stream_empty(pool);
  SVN_ERR(read_representation(&target_stream, fs, target->data_rep, pool));
  svn_txdelta(stream_p, source_stream, target_stream, pool);

  return SVN_NO_ERROR;
}


/* Fetch the contents of a directory into ENTRIES.  Values are stored
   as filename to string mappings; further conversion is necessary to
   convert them into svn_fs_dirent_t values. */
static svn_error_t *
get_dir_contents(apr_hash_t *entries,
                 svn_fs_t *fs,
                 node_revision_t *noderev,
                 apr_pool_t *pool)
{
  svn_stream_t *contents;

  if (noderev->data_rep && noderev->data_rep->txn_id)
    {
      const char *filename = path_txn_node_children(fs, noderev->id, pool);

      /* The representation is mutable.  Read the old directory
         contents from the mutable children file, followed by the
         changes we've made in this transaction. */
      SVN_ERR(svn_stream_open_readonly(&contents, filename, pool, pool));
      SVN_ERR(svn_hash_read2(entries, contents, SVN_HASH_TERMINATOR, pool));
      SVN_ERR(svn_hash_read_incremental(entries, contents, NULL, pool));
      SVN_ERR(svn_stream_close(contents));
    }
  else if (noderev->data_rep)
    {
      /* The representation is immutable.  Read it normally. */
      SVN_ERR(read_representation(&contents, fs, noderev->data_rep, pool));
      SVN_ERR(svn_hash_read2(entries, contents, SVN_HASH_TERMINATOR, pool));
      SVN_ERR(svn_stream_close(contents));
    }

  return SVN_NO_ERROR;
}


static const char *
unparse_dir_entry(svn_node_kind_t kind, const svn_fs_id_t *id,
                  apr_pool_t *pool)
{
  return apr_psprintf(pool, "%s %s",
                      (kind == svn_node_file) ? KIND_FILE : KIND_DIR,
                      svn_fs_fs__id_unparse(id, pool)->data);
}

/* Given a hash ENTRIES of dirent structions, return a hash in
   *STR_ENTRIES_P, that has svn_string_t as the values in the format
   specified by the fs_fs directory contents file.  Perform
   allocations in POOL. */
static svn_error_t *
unparse_dir_entries(apr_hash_t **str_entries_p,
                    apr_hash_t *entries,
                    apr_pool_t *pool)
{
  apr_hash_index_t *hi;

  *str_entries_p = apr_hash_make(pool);

  for (hi = apr_hash_first(pool, entries); hi; hi = apr_hash_next(hi))
    {
      const void *key;
      apr_ssize_t klen;
      void *val;
      svn_fs_dirent_t *dirent;
      const char *new_val;

      apr_hash_this(hi, &key, &klen, &val);
      dirent = val;
      new_val = unparse_dir_entry(dirent->kind, dirent->id, pool);
      apr_hash_set(*str_entries_p, key, klen,
                   svn_string_create(new_val, pool));
    }

  return SVN_NO_ERROR;
}


svn_error_t *
svn_fs_fs__dir_entries_serialize(char **data,
                                 apr_size_t *data_len,
                                 void *in,
                                 apr_pool_t *pool)
{
  apr_hash_t *entries = in;
  svn_stringbuf_t *buf = svn_stringbuf_create("", pool);
  svn_stream_t *stream = svn_stream_from_stringbuf(buf, pool);

  SVN_ERR(unparse_dir_entries(&entries, entries, pool));
  SVN_ERR(svn_hash_write2(entries, stream, SVN_HASH_TERMINATOR, pool));

  *data = buf->data;
  *data_len = buf->len;

  return SVN_NO_ERROR;
}


/* Given a hash STR_ENTRIES with values as svn_string_t as specified
   in an FSFS directory contents listing, return a hash of dirents in
   *ENTRIES_P.  Perform allocations in POOL. */
static svn_error_t *
parse_dir_entries(apr_hash_t **entries_p,
                  apr_hash_t *str_entries,
                  apr_pool_t *pool)
{
  apr_hash_index_t *hi;

  *entries_p = apr_hash_make(pool);

  /* Translate the string dir entries into real entries. */
  for (hi = apr_hash_first(pool, str_entries); hi; hi = apr_hash_next(hi))
    {
      const void *key;
      void *val;
      svn_string_t *str_val;
      char *str, *last_str;
      svn_fs_dirent_t *dirent = apr_pcalloc(pool, sizeof(*dirent));

      apr_hash_this(hi, &key, NULL, &val);
      str_val = val;
      str = apr_pstrdup(pool, str_val->data);
      dirent->name = apr_pstrdup(pool, key);

      str = apr_strtok(str, " ", &last_str);
      if (str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Directory entry corrupt"));

      if (strcmp(str, KIND_FILE) == 0)
        {
          dirent->kind = svn_node_file;
        }
      else if (strcmp(str, KIND_DIR) == 0)
        {
          dirent->kind = svn_node_dir;
        }
      else
        {
          return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                  _("Directory entry corrupt"));
        }

      str = apr_strtok(NULL, " ", &last_str);
      if (str == NULL)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Directory entry corrupt"));

      dirent->id = svn_fs_fs__id_parse(str, strlen(str), pool);

      apr_hash_set(*entries_p, dirent->name, APR_HASH_KEY_STRING, dirent);
    }

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__dir_entries_deserialize(void **out,
                                   const char *data,
                                   apr_size_t data_len,
                                   apr_pool_t *pool)
{
  apr_hash_t *entries = apr_hash_make(pool);
  svn_stringbuf_t *buf = svn_stringbuf_ncreate(data, data_len, pool);
  svn_stream_t *stream = svn_stream_from_stringbuf(buf, pool);

  SVN_ERR(svn_hash_read2(entries, stream, SVN_HASH_TERMINATOR, pool));
  SVN_ERR(parse_dir_entries(&entries, entries, pool));

  *out = entries;
  return SVN_NO_ERROR;
}


svn_error_t *
svn_fs_fs__rep_contents_dir(apr_hash_t **entries_p,
                            svn_fs_t *fs,
                            node_revision_t *noderev,
                            apr_pool_t *pool)
{
  fs_fs_data_t *ffd = fs->fsap_data;
  const char *unparsed_id;
  apr_hash_t *unparsed_entries, *parsed_entries;

  /* Are we looking for an immutable directory?  We could try the
   * cache. */
  if (! svn_fs_fs__id_txn_id(noderev->id))
    {
      svn_boolean_t found;

      unparsed_id = svn_fs_fs__id_unparse(noderev->id, pool)->data;
      SVN_ERR(svn_cache__get((void **) entries_p, &found, ffd->dir_cache,
                             unparsed_id, pool));
      if (found)
        return SVN_NO_ERROR;
    }

  /* Read in the directory hash. */
  unparsed_entries = apr_hash_make(pool);
  SVN_ERR(get_dir_contents(unparsed_entries, fs, noderev, pool));
  SVN_ERR(parse_dir_entries(&parsed_entries, unparsed_entries, pool));

  /* If this is an immutable directory, let's cache the contents. */
  if (! svn_fs_fs__id_txn_id(noderev->id))
    SVN_ERR(svn_cache__set(ffd->dir_cache, unparsed_id, parsed_entries, pool));

  *entries_p = parsed_entries;
  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__get_proplist(apr_hash_t **proplist_p,
                        svn_fs_t *fs,
                        node_revision_t *noderev,
                        apr_pool_t *pool)
{
  apr_hash_t *proplist;
  svn_stream_t *stream;

  proplist = apr_hash_make(pool);

  if (noderev->prop_rep && noderev->prop_rep->txn_id)
    {
      const char *filename = path_txn_node_props(fs, noderev->id, pool);

      SVN_ERR(svn_stream_open_readonly(&stream, filename, pool, pool));
      SVN_ERR(svn_hash_read2(proplist, stream, SVN_HASH_TERMINATOR, pool));
      SVN_ERR(svn_stream_close(stream));
    }
  else if (noderev->prop_rep)
    {
      SVN_ERR(read_representation(&stream, fs, noderev->prop_rep, pool));
      SVN_ERR(svn_hash_read2(proplist, stream, SVN_HASH_TERMINATOR, pool));
      SVN_ERR(svn_stream_close(stream));
    }

  *proplist_p = proplist;

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__file_length(svn_filesize_t *length,
                       node_revision_t *noderev,
                       apr_pool_t *pool)
{
  if (noderev->data_rep)
    *length = noderev->data_rep->expanded_size;
  else
    *length = 0;

  return SVN_NO_ERROR;
}

svn_boolean_t
svn_fs_fs__noderev_same_rep_key(representation_t *a,
                                representation_t *b)
{
  if (a == b)
    return TRUE;

  if (a && (! b))
    return FALSE;

  if (b && (! a))
    return FALSE;

  if (a->offset != b->offset)
    return FALSE;

  if (a->revision != b->revision)
    return FALSE;

  if (a->reuse_count != b->reuse_count)
    return FALSE;

  return TRUE;
}

svn_error_t *
svn_fs_fs__file_checksum(svn_checksum_t **checksum,
                         node_revision_t *noderev,
                         svn_checksum_kind_t kind,
                         apr_pool_t *pool)
{
  if (noderev->data_rep)
    {
      switch(kind)
        {
          case svn_checksum_md5:
            *checksum = svn_checksum_dup(noderev->data_rep->md5_checksum,
                                         pool);
            break;
          case svn_checksum_sha1:
            *checksum = svn_checksum_dup(noderev->data_rep->sha1_checksum,
                                         pool);
            break;
          default:
            *checksum = NULL;
        }
    }
  else
    *checksum = NULL;

  return SVN_NO_ERROR;
}

representation_t *
svn_fs_fs__rep_copy(representation_t *rep,
                    apr_pool_t *pool)
{
  representation_t *rep_new;

  if (rep == NULL)
    return NULL;

  rep_new = apr_pcalloc(pool, sizeof(*rep_new));

  memcpy(rep_new, rep, sizeof(*rep_new));
  rep_new->md5_checksum = svn_checksum_dup(rep->md5_checksum, pool);
  rep_new->sha1_checksum = svn_checksum_dup(rep->sha1_checksum, pool);

  return rep_new;
}

/* Merge the internal-use-only CHANGE into a hash of public-FS
   svn_fs_path_change_t CHANGES, collapsing multiple changes into a
   single summarical (is that real word?) change per path.  Also keep
   the COPYFROM_HASH up to date with new adds and replaces.  */
static svn_error_t *
fold_change(apr_hash_t *changes,
            const change_t *change,
            apr_hash_t *copyfrom_hash)
{
  apr_pool_t *pool = apr_hash_pool_get(changes);
  apr_pool_t *copyfrom_pool = apr_hash_pool_get(copyfrom_hash);
  svn_fs_path_change_t *old_change, *new_change;
  const char *path, *copyfrom_string, *copyfrom_path = NULL;

  if ((old_change = apr_hash_get(changes, change->path, APR_HASH_KEY_STRING)))
    {
      /* This path already exists in the hash, so we have to merge
         this change into the already existing one. */

      /* Get the existing copyfrom entry for this path. */
      copyfrom_string = apr_hash_get(copyfrom_hash, change->path,
                                     APR_HASH_KEY_STRING);

      /* If this entry existed in the copyfrom hash, we don't need to
         copy it. */
      if (copyfrom_string)
        copyfrom_path = change->path;

      /* Since the path already exists in the hash, we don't have to
         dup the allocation for the path itself. */
      path = change->path;
      /* Sanity check:  only allow NULL node revision ID in the
         `reset' case. */
      if ((! change->noderev_id) && (change->kind != svn_fs_path_change_reset))
        return svn_error_create
          (SVN_ERR_FS_CORRUPT, NULL,
           _("Missing required node revision ID"));

      /* Sanity check: we should be talking about the same node
         revision ID as our last change except where the last change
         was a deletion. */
      if (change->noderev_id
          && (! svn_fs_fs__id_eq(old_change->node_rev_id, change->noderev_id))
          && (old_change->change_kind != svn_fs_path_change_delete))
        return svn_error_create
          (SVN_ERR_FS_CORRUPT, NULL,
           _("Invalid change ordering: new node revision ID "
             "without delete"));

      /* Sanity check: an add, replacement, or reset must be the first
         thing to follow a deletion. */
      if ((old_change->change_kind == svn_fs_path_change_delete)
          && (! ((change->kind == svn_fs_path_change_replace)
                 || (change->kind == svn_fs_path_change_reset)
                 || (change->kind == svn_fs_path_change_add))))
        return svn_error_create
          (SVN_ERR_FS_CORRUPT, NULL,
           _("Invalid change ordering: non-add change on deleted path"));

      /* Now, merge that change in. */
      switch (change->kind)
        {
        case svn_fs_path_change_reset:
          /* A reset here will simply remove the path change from the
             hash. */
          old_change = NULL;
          copyfrom_string = NULL;
          break;

        case svn_fs_path_change_delete:
          if (old_change->change_kind == svn_fs_path_change_add)
            {
              /* If the path was introduced in this transaction via an
                 add, and we are deleting it, just remove the path
                 altogether. */
              old_change = NULL;
            }
          else
            {
              /* A deletion overrules all previous changes. */
              old_change->change_kind = svn_fs_path_change_delete;
              old_change->text_mod = change->text_mod;
              old_change->prop_mod = change->prop_mod;
            }
          copyfrom_string = NULL;
          break;

        case svn_fs_path_change_add:
        case svn_fs_path_change_replace:
          /* An add at this point must be following a previous delete,
             so treat it just like a replace. */
          old_change->change_kind = svn_fs_path_change_replace;
          old_change->node_rev_id = svn_fs_fs__id_copy(change->noderev_id,
                                                       pool);
          old_change->text_mod = change->text_mod;
          old_change->prop_mod = change->prop_mod;
          if (change->copyfrom_rev == SVN_INVALID_REVNUM)
            copyfrom_string = apr_pstrdup(copyfrom_pool, "");
          else
            {
              copyfrom_string = apr_psprintf(copyfrom_pool,
                                             "%ld %s",
                                             change->copyfrom_rev,
                                             change->copyfrom_path);
            }
          break;

        case svn_fs_path_change_modify:
        default:
          if (change->text_mod)
            old_change->text_mod = TRUE;
          if (change->prop_mod)
            old_change->prop_mod = TRUE;
          break;
        }

      /* Point our new_change to our (possibly modified) old_change. */
      new_change = old_change;
    }
  else
    {
      /* This change is new to the hash, so make a new public change
         structure from the internal one (in the hash's pool), and dup
         the path into the hash's pool, too. */
      new_change = apr_pcalloc(pool, sizeof(*new_change));
      new_change->node_rev_id = svn_fs_fs__id_copy(change->noderev_id, pool);
      new_change->change_kind = change->kind;
      new_change->text_mod = change->text_mod;
      new_change->prop_mod = change->prop_mod;
      if (change->copyfrom_rev != SVN_INVALID_REVNUM)
        {
          copyfrom_string = apr_psprintf(copyfrom_pool, "%ld %s",
                                         change->copyfrom_rev,
                                         change->copyfrom_path);
        }
      else
        copyfrom_string = apr_pstrdup(copyfrom_pool, "");
      path = apr_pstrdup(pool, change->path);
    }

  /* Add (or update) this path. */
  apr_hash_set(changes, path, APR_HASH_KEY_STRING, new_change);

  /* If copyfrom_path is non-NULL, the key is already present in the
     hash, so we don't need to duplicate it in the copyfrom pool. */
  if (! copyfrom_path)
    {
      /* If copyfrom_string is NULL, the hash entry will be deleted,
         so we don't need to duplicate the key in the copyfrom
         pool. */
      copyfrom_path = copyfrom_string ? apr_pstrdup(copyfrom_pool, path)
        : path;
    }

  apr_hash_set(copyfrom_hash, copyfrom_path, APR_HASH_KEY_STRING,
               copyfrom_string);

  return SVN_NO_ERROR;
}

/* The 256 is an arbitrary size large enough to hold the node id and the
 * various flags. */
#define MAX_CHANGE_LINE_LEN FSFS_MAX_PATH_LEN + 256

/* Read the next entry in the changes record from file FILE and store
   the resulting change in *CHANGE_P.  If there is no next record,
   store NULL there.  Perform all allocations from POOL. */
static svn_error_t *
read_change(change_t **change_p,
            apr_file_t *file,
            apr_pool_t *pool)
{
  char buf[MAX_CHANGE_LINE_LEN];
  apr_size_t len = sizeof(buf);
  change_t *change;
  char *str, *last_str;
  svn_error_t *err;

  /* Default return value. */
  *change_p = NULL;

  err = svn_io_read_length_line(file, buf, &len, pool);

  /* Check for a blank line. */
  if (err || (len == 0))
    {
      if (err && APR_STATUS_IS_EOF(err->apr_err))
        {
          svn_error_clear(err);
          return SVN_NO_ERROR;
        }
      if ((len == 0) && (! err))
        return SVN_NO_ERROR;
      return err;
    }

  change = apr_pcalloc(pool, sizeof(*change));

  /* Get the node-id of the change. */
  str = apr_strtok(buf, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Invalid changes line in rev-file"));

  change->noderev_id = svn_fs_fs__id_parse(str, strlen(str), pool);
  if (change->noderev_id == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Invalid changes line in rev-file"));

  /* Get the change type. */
  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Invalid changes line in rev-file"));

  if (strcmp(str, ACTION_MODIFY) == 0)
    {
      change->kind = svn_fs_path_change_modify;
    }
  else if (strcmp(str, ACTION_ADD) == 0)
    {
      change->kind = svn_fs_path_change_add;
    }
  else if (strcmp(str, ACTION_DELETE) == 0)
    {
      change->kind = svn_fs_path_change_delete;
    }
  else if (strcmp(str, ACTION_REPLACE) == 0)
    {
      change->kind = svn_fs_path_change_replace;
    }
  else if (strcmp(str, ACTION_RESET) == 0)
    {
      change->kind = svn_fs_path_change_reset;
    }
  else
    {
      return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                              _("Invalid change kind in rev file"));
    }

  /* Get the text-mod flag. */
  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Invalid changes line in rev-file"));

  if (strcmp(str, FLAG_TRUE) == 0)
    {
      change->text_mod = TRUE;
    }
  else if (strcmp(str, FLAG_FALSE) == 0)
    {
      change->text_mod = FALSE;
    }
  else
    {
      return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                              _("Invalid text-mod flag in rev-file"));
    }

  /* Get the prop-mod flag. */
  str = apr_strtok(NULL, " ", &last_str);
  if (str == NULL)
    return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                            _("Invalid changes line in rev-file"));

  if (strcmp(str, FLAG_TRUE) == 0)
    {
      change->prop_mod = TRUE;
    }
  else if (strcmp(str, FLAG_FALSE) == 0)
    {
      change->prop_mod = FALSE;
    }
  else
    {
      return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                              _("Invalid prop-mod flag in rev-file"));
    }

  /* Get the changed path. */
  change->path = apr_pstrdup(pool, last_str);


  /* Read the next line, the copyfrom line. */
  len = sizeof(buf);
  SVN_ERR(svn_io_read_length_line(file, buf, &len, pool));

  if (len == 0)
    {
      change->copyfrom_rev = SVN_INVALID_REVNUM;
      change->copyfrom_path = NULL;
    }
  else
    {
      str = apr_strtok(buf, " ", &last_str);
      if (! str)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Invalid changes line in rev-file"));
      change->copyfrom_rev = atol(str);

      if (! last_str)
        return svn_error_create(SVN_ERR_FS_CORRUPT, NULL,
                                _("Invalid changes line in rev-file"));

      change->copyfrom_path = apr_pstrdup(pool, last_str);
    }

  *change_p = change;

  return SVN_NO_ERROR;
}

/* Fetch all the changed path entries from FILE and store then in
   *CHANGED_PATHS.  Folding is done to remove redundant or unnecessary
   *data.  Store a hash of paths to copyfrom revisions/paths in
   COPYFROM_HASH if it is non-NULL.  If PREFOLDED is true, assume that
   the changed-path entries have already been folded (by
   write_final_changed_path_info) and may be out of order, so we shouldn't
   remove children of replaced or deleted directories.  Do all
   allocations in POOL. */
static svn_error_t *
fetch_all_changes(apr_hash_t *changed_paths,
                  apr_hash_t *copyfrom_hash,
                  apr_file_t *file,
                  svn_boolean_t prefolded,
                  apr_pool_t *pool)
{
  change_t *change;
  apr_pool_t *iterpool = svn_pool_create(pool);
  apr_hash_t *my_hash;

  /* If we are passed a NULL copyfrom hash, manufacture one for the
     duration of this call. */
  my_hash = copyfrom_hash ? copyfrom_hash : apr_hash_make(pool);

  /* Read in the changes one by one, folding them into our local hash
     as necessary. */

  SVN_ERR(read_change(&change, file, iterpool));

  while (change)
    {
      SVN_ERR(fold_change(changed_paths, change, my_hash));

      /* Now, if our change was a deletion or replacement, we have to
         blow away any changes thus far on paths that are (or, were)
         children of this path.
         ### i won't bother with another iteration pool here -- at
         most we talking about a few extra dups of paths into what
         is already a temporary subpool.
      */

      if (((change->kind == svn_fs_path_change_delete)
           || (change->kind == svn_fs_path_change_replace))
          && ! prefolded)
        {
          apr_hash_index_t *hi;

          for (hi = apr_hash_first(iterpool, changed_paths);
               hi;
               hi = apr_hash_next(hi))
            {
              /* KEY is the path. */
              const void *hashkey;
              apr_ssize_t klen;
              apr_hash_this(hi, &hashkey, &klen, NULL);

              /* If we come across our own path, ignore it. */
              if (strcmp(change->path, hashkey) == 0)
                continue;

              /* If we come across a child of our path, remove it. */
              if (svn_path_is_child(change->path, hashkey, iterpool))
                apr_hash_set(changed_paths, hashkey, klen, NULL);
            }
        }

      /* Clear the per-iteration subpool. */
      svn_pool_clear(iterpool);

      SVN_ERR(read_change(&change, file, iterpool));
    }

  /* Destroy the per-iteration subpool. */
  svn_pool_destroy(iterpool);

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__txn_changes_fetch(apr_hash_t **changed_paths_p,
                             svn_fs_t *fs,
                             const char *txn_id,
                             apr_hash_t *copyfrom_cache,
                             apr_pool_t *pool)
{
  apr_file_t *file;
  apr_hash_t *changed_paths = apr_hash_make(pool);

  SVN_ERR(svn_io_file_open(&file, path_txn_changes(fs, txn_id, pool),
                           APR_READ | APR_BUFFERED, APR_OS_DEFAULT, pool));

  SVN_ERR(fetch_all_changes(changed_paths, copyfrom_cache, file, FALSE,
                            pool));

  SVN_ERR(svn_io_file_close(file, pool));

  *changed_paths_p = changed_paths;

  return SVN_NO_ERROR;
}

svn_error_t *
svn_fs_fs__paths_changed(apr_hash_t **changed_paths_p,
                         svn_fs_t *fs,
                         svn_revnum_t rev,
                         apr_hash_t *copyfrom_cache,
                         apr_pool_t *pool)
{
  apr_off_t changes_offset;
  apr_hash_t *changed_paths;
  apr_file_t *revision_file;

  SVN_ERR(ensure_revision_exists(fs, rev, pool));

  SVN_ERR(open_pack_or_rev_file(&revision_file, fs, rev, pool));

  SVN_ERR(get_root_changes_offset(NULL, &changes_offset, revision_file, fs,
                                  rev, pool));

  SVN_ERR(svn_io_file_seek(revision_file, APR_SET, &changes_offset, pool));

  changed_pat