mariadb/sql/sql_backup.cc

/* Copyright (c) 2026, MariaDB plc

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335  USA */

#include "my_global.h"
#include "mdl.h"
#include "mysys_err.h"
#include "sql_class.h"
#include "sql_base.h"
#include "sql_backup.h"
#include "sql_backup_interface.h"
#include "sql_parse.h"
#include "my_atomic_wrapper.h"
#include "tpool.h"
#include "aligned.h"

static bool backup_execute(THD *thd, const char *target, const char *command,
                           int threads);

/**
   @brief BACKUP SERVER command

   BACKUP SERVER copies the storage of a running server optionally
   using multiple CONCURRENT threads, each one writing to the target
   directory or its own tar stream.

   BACKUP SERVER TO '/path/to/directory' [ 1 CONCURRENT ];
   BACKUP SERVER WITH [ 1 CONCURRENT ] 'command';

   The operation is divided into multiple stages; @see backup_stages

   Each stage is divided into "start" and "end", which are invoked
   by this thread, and a number of "step" in between that may be
   invoked by multiple CONCURRENT threads.
   @see handlerton::backup_start
   @see handlerton::backup_step
   @see handlerton::backup_end

   Most stages are tied with BACKUP STAGE locks; the actual first work stage:
   @see BACKUP_START
   @see MDL_BACKUP_START

   The snapshot for transactional storage engines is determined in:
   @see BACKUP_LOCK_COMMIT
   @see MDL_BACKUP_WAIT_COMMIT

   BACKUP_PREPARE, BACKUP_ABORTED, BACKUP_FINISHED are executed while
   not holding any MDL.

   @param thd   client connection
   @return whether the operation failed
*/
bool Sql_cmd_backup::execute(THD *thd)
{
  return backup_execute(thd, target, command, threads);
}

#ifdef _WIN32
using tpool::pread;
using tpool::pwrite;
#else
# include <sys/types.h>
# include <dirent.h>
# include <sys/mman.h>

#endif

/**
   Copy a file using positioned reads.
   @tparam stream true=write to a stream, false=pwrite to a file
   @param in_fd   source file
   @param out_fd  destination
   @param o       start offset
   @param end     last offset (exclusive)
   @return error code (non-positive)
   @retval 0  on success
*/
template<bool stream>
static ssize_t pread_write(backup::handle in_fd, backup_fd out_fd,
                           uint64_t o, uint64_t end) noexcept
{
  assert(end >= o);
  uint64_t count{end - o};
#if SIZEOF_SIZE_T < 8
  size_t size{65536};
#else
  constexpr size_t max_size{1 << 20};
  /* An integer multiple of 4096 bytes, up to max_size */
  size_t size{count >= max_size ? max_size : (size_t(count) + 4095) & ~4095};
#endif
  char *b= static_cast<char*>(aligned_malloc(size, 4096));
  if (!b)
    return -1;
  ssize_t ret;
  for (;;)
  {
    if (count < size)
      size= count;
    ret= pread(in_fd, b, ssize_t(size), o);
    count-= uint64_t(ret);
    const uint64_t n{o + ret};
    if (ret > 0)
    {
#ifdef POSIX_FADV_DONTNEED
      std::ignore= posix_fadvise(in_fd, o, ret, POSIX_FADV_DONTNEED);
#endif
#ifdef POSIX_FADV_WILLNEED
      if (count)
        std::ignore= posix_fadvise(in_fd, n, size, POSIX_FADV_WILLNEED);
#endif
      if (!stream)
        ret= pwrite(out_fd, b, size_t(ret), o);
      else if (backup_stream_write(out_fd, b, size_t(ret)))
      {
        ret= -1;
        break;
      }
    }
    if (ret < 0)
      break;
    if (!count)
    {
      ret= 0;
      break;
    }
    if (!ret)
    {
      ret= -1;
      break;
    }
    o= n;
  }
  aligned_free(b);
  return ret;
}

#ifdef __APPLE__
/* The inline copy_entire_file() invokes fcopyfile() */
#elif defined _WIN32
/* CopyFileEx() should be used */
#else
/** Copy a file (whole content).
@param src  source file descriptor
@param dst  target to append src to
@return error code (non-positive)
@retval 0   on success */
int copy_entire_file(int src, int dst) noexcept
{
  uint64_t end(lseek(src, 0, SEEK_END));
#ifdef POSIX_FADV_NOREUSE
  std::ignore= posix_fadvise(dst, 0, 0, POSIX_FADV_NOREUSE);
#endif
  int ret;
# ifdef copy_file_shortcut
  ret= int(copy_file_shortcut(src, dst, 0, end));
  if (ret == 1)
# endif
    ret= backup::copy(src, dst, 0, end);
  return ret;
}
#endif

#if defined __linux__ || defined __FreeBSD__
using copying_step= ssize_t(int,int,size_t,off_t*);
template<copying_step step,bool nonblocking>
static ssize_t stepwise(int in_fd, int out_fd, off_t offset, off_t end)
  noexcept
{
  for (;;)
  {
    const size_t c{size_t(std::min<off_t>(end - offset, INT_MAX >> 20 << 20))};
    ssize_t ret= step(in_fd, out_fd, c, &offset);
    if (ret < 0)
    {
      if (nonblocking && errno == EAGAIN)
        continue;
      return ret;
    }
    if (offset == end)
      return 0;
    if (!ret)
      return -1;
  }
}

/* Copy between files in a single (type of) file system */
static inline ssize_t
copy_step(int in_fd, int out_fd, size_t count, off_t *offset) noexcept
{
  return copy_file_range(in_fd, offset, out_fd, offset, count, 0);
}
# define cfr(src,dst,start,end) stepwise<copy_step,false>(src, dst, start, end)
#endif

namespace backup {
/** Copy a portion of a file.
@param src   source file descriptor
@param dst   target to append src to
@param start first offset to copy
@param end   last offset to copy (exclusive)
@return error code (non-positive)
@retval 0   on success */
int copy(handle src, backup_fd dst, uint64_t start, uint64_t end) noexcept
{
  assert(end >= start);
#ifdef __FreeBSD__
  /* On FreeBSD, copy_file_range() without flags just works */
  return int(cfr(src, dst, off_t(start), off_t(end)));
#else
  ssize_t ret{pread_write<false>(src, dst, start, end)};
  assert(ret <= 0);
  return int(ret);
#endif
}

#ifndef _WIN32
/**
   Copy or stream a file.
   @param target     BACKUP SERVER target (possibly, a directory)
   @param sink       per-thread context (possibly, a stream to write to)
   @param src        source file descriptor from open(), to be closed here
   @param path       file name
   @return error code (also errno will be set)
   @retval 0 on success
*/
int copy_or_stream(const backup_target &target, const backup_sink &sink,
                   int src, const char *path) noexcept
{
  int ret= -1, dst= sink.stream;
  if (src < 0)
  {
    my_error(ER_CANT_OPEN_FILE, MYF(0), path, errno);
    return ret;
  }
  if (dst < 0)
  {
    dst= openat(target.fd, path, O_CREAT | O_EXCL | O_WRONLY, 0666);
    if (dst < 0)
      my_error(ER_CANT_CREATE_FILE, MYF(0), path, errno);
    else
    {
      ret= copy_entire_file(src, dst) | close(dst);
      if (ret)
      write_error:
        my_error(ER_ERROR_ON_WRITE, MYF(0), path, errno);
    }
  }
  else
  {
    uint64_t end= (uint64_t) lseek(src, 0, SEEK_END);
    ret= backup_stream_start(dst, path, 0644, end, NULL, 0) ||
      backup::append(src, dst, 0, end) ||
      backup_stream_zeropad(dst, (size_t) end);
    if (ret)
      goto write_error;
  }
  close(src);
  return ret;
}
#endif

/**
   Copy or stream a file.
   @param target     BACKUP SERVER target (possibly, a directory)
   @param sink       per-thread context (possibly, a stream to write to)
   @param path       file name
   @param dir_prefix length of the path prefix to omit from the backup
   @return error code (also errno will be set)
   @retval 0 on success (errno might not be touched)
*/
int copy_or_stream(const backup_target &target, const backup_sink &sink,
                   const char *path, size_t dir_prefix) noexcept
{
#ifndef _WIN32
  return copy_or_stream(target, sink,
                        open(path, O_RDONLY), path + dir_prefix);
#else
  int ret= -1;
  if (sink.stream == INVALID_HANDLE_VALUE)
  {
    int len= snprintf(NULL, 0, "%s/%s", target.path, path + dir_prefix) + 1;
    char *dstpath= static_cast<char*>(malloc(len));
    if (!dstpath)
      my_error(ER_TOO_LONG_IDENT, MYF(0), path + dir_prefix);
    else
    {
      snprintf(dstpath, len, "%s/%s", target.path, path + dir_prefix);
      while (!CopyFileEx(path, dstpath, NULL, NULL, NULL,
                         COPY_FILE_NO_BUFFERING))
        switch (GetLastError()) {
        default:
          my_osmaperr(GetLastError());
          my_error(ER_CANT_CREATE_FILE, MYF(0), dstpath, errno);
          goto done;
        case ERROR_SHARING_VIOLATION:
        case ERROR_LOCK_VIOLATION:
          Sleep(10);
        }
      ret= 0;
    done:
      free(dstpath);
    }
  }
  else
  {
    LARGE_INTEGER li;
    HANDLE src, dst= sink.stream;
    for (;;)
    {
      src= CreateFile(path, GENERIC_READ,
                      FILE_SHARE_READ|FILE_SHARE_WRITE|FILE_SHARE_DELETE,
                      my_win_file_secattr(), OPEN_EXISTING,
                      FILE_ATTRIBUTE_NORMAL, NULL);
      if (src != INVALID_HANDLE_VALUE)
        break;
      switch (GetLastError()) {
      case ERROR_SHARING_VIOLATION:
      case ERROR_LOCK_VIOLATION:
        Sleep(10);
        continue;
      }

      my_osmaperr(GetLastError());
      my_error(ER_FILE_NOT_FOUND, MYF(ME_ERROR_LOG), path, errno);
      return ret;
    }

    ret= !GetFileSizeEx(src, &li) ||
      backup_stream_start(dst, path + dir_prefix, 0644, li.QuadPart,
                          NULL, 0) ||
      backup::append(src, dst, 0, li.QuadPart) ||
      backup_stream_zeropad(dst, (size_t) li.QuadPart);
    (void) CloseHandle(src);

    if (ret)
    {
      my_osmaperr(GetLastError());
      my_error(ER_ERROR_ON_WRITE, MYF(0), path + dir_prefix, errno);
    }
  }
  return ret;
#endif
}

/**
   Append a file snippet to stream,
   after a corresponding call to backup_stream_start().

   Note that tar uses 512-byte blocks. If end-start is not a multiple of
   512 bytes, backup_stream_zeropad() must be invoked.
   @param src      source file
   @param stream   backup stream
   @param start first offset to copy
   @param end   last offset to copy (exclusive)
   @return error code (non-positive)
   @retval 0   on success
*/
int append(handle src, backup_fd stream, uint64_t start, uint64_t end) noexcept
{
  assert(stream != backup_sink::NO_STREAM);
  assert(end >= start);
#if 0 /* On Linux, this turned out to be slower in our tests! */
  /*
    It is unsafe to zero-copy send from src unless we can guarantee
    that the receiving end of the pipe has consumed everything before
    our caller re-enables writes to this src region.

    For Linux mmap(2), even for MAP_PRIVATE, the following is documented:
    It is unspecified whether changes made to the file after the
    mmap(2) call are visible in the mapped region.

    Similarly, for Linux sendfile(2): If out_fd refers to a socket or
    pipe with zero-copy support, callers must ensure the transferred
    portions of the file referred to by in_fd remain unmodified until
    the reader on the other end of out_fd has consumed the
    transferred data.
  */
# ifdef __linux__
  const int pipe_size{fcntl(stream, F_GETPIPE_SZ)};
# elif defined __FreeBSD__ || defined __APPLE__
  // https://unix.stackexchange.com/questions/11946/how-big-is-the-pipe-buffer
  constexpr int pipe_size{65536};
# else
  constexpr int pipe_size{0};
# endif
  if (pipe_size > 0)
  {
    /*
      It should be safe to invoke the zero-copy path if we can
      guarantee that the recipient will drain the buffer before the
      final nonzero-copy path completes. We assume that this is
      guaranteed by writing at least pipe_size bytes via the
      nonzero-copy path after this shortcut.
    */
    const uint64_t fast_end{start + uint64_t(pipe_size)};
    if (fast_end < end)
    {
      if (int ret= backup_stream_append_async(src, stream, start, fast_end))
        return ret;
      start= fast_end;
    }
  }
#endif
  return int(pread_write<true>(src, stream, start, end));
}

}

/**
   Zero-pad a the stream to a multiple of 512 bytes.

   @param stream  backup stream
   @param written least significant bits of the number of payload appended
   @return error code (non-positive)
   @retval 0   on success
*/
int backup_stream_zeropad(backup_fd stream, size_t written) noexcept
{
  static constexpr const char zerobuf[511]{'\0'};
  written&= 511;
  return written ? backup_stream_write(stream, zerobuf, 512 - written) : 0;
}

#ifdef __linux__
/**
   Try to copy a portion of a file via copy_file_range(2).
   @param src   source file descriptor
   @param dst   target to append src to
   @param start first offset to copy
   @param end   last offset to copy (exclusive)
   @return error code (non-positive)
   @retval 0   on success
   @retval 1   if a fallback to backup::copy() is needed
*/
int copy_file_range_try(int src, int dst, uint64_t start, uint64_t end)
  noexcept
{
  assert(end >= start);
  ssize_t ret{cfr(src, dst, off_t(start), off_t(end))};
  assert(ret <= 0);
  if (ret && (errno == EOPNOTSUPP || errno == EXDEV))
    return 1;
#ifdef POSIX_FADV_DONTNEED
  std::ignore= posix_fadvise(src, start, end - start, POSIX_FADV_DONTNEED);
#endif
  return int(ret);
}
#endif

/** Append to the configuration file.
@param target   backup target directory
@param config   the configuration file snippet to append
@param size     length of the snippet
@return error code (non-positive)
@retval 0   on success */
int backup_config_append(IF_WIN(const char*, int) target,
                         const char *config, size_t size) noexcept
{
  /* FIXME: append to a pre-created configuration file */
#ifdef _WIN32
  HANDLE dst;
  {
    std::string path{target};
    path.append("/backup.cnf");
    dst= CreateFile(path.c_str(), GENERIC_WRITE, 0,
                    my_win_file_secattr(), CREATE_NEW,
                    FILE_ATTRIBUTE_NORMAL, nullptr);
    if (dst != INVALID_HANDLE_VALUE)
    {
      BOOL ok;
      for (;;)
      {
        DWORD written;
        ok= WriteFile(dst, config, DWORD(size), &written, nullptr);
        if (ok || GetLastError() != ERROR_IO_PENDING)
          break;
        assert(written < DWORD(size));
        config+= written;
        size-= size_t(written);
      }
      if (CloseHandle(dst) & ok)
        return 0;
    }
  }
#else
  int dst= openat(target, "backup.cnf",
                  O_CREAT | O_EXCL | O_TRUNC | O_WRONLY, 0666);
  if (dst >= 0)
  {
    ssize_t ret;
    for (; (ret= write(dst, config, size)) >= 0; config+= ret, size -= ret)
    {
      assert(size_t(ret) <= size);
      if (!(size-= size_t(ret)))
      {
        ret= 0;
        break;
      }
    }
    if (!(close(dst) | ret))
      return 0;
  }
#endif
  my_error(ER_CANT_CREATE_FILE, MYF(ME_ERROR_LOG), "backup.cnf", errno);
  return -1;
}

/**
   Invoke handlerton::backup_file() on a storage engine in a thread
   that may or may not be associated with a BACKUP SERVER connection,
   between handlerton::backup_start() and handlerton::backup_end()
   of the same backup stage.
   @param thd     the BACKUP SERVER session
   @param plugin  storage engine
   @param arg     the name to check
   @return whether the name should be excluded
*/
static my_bool backup_name_filtered(THD *thd, plugin_ref plugin, void *arg)
  noexcept
{
  const handlerton *hton= plugin_hton(plugin);
  LEX_CSTRING &name{*static_cast<LEX_CSTRING*>(arg)};
  return hton->backup_file && !hton->backup_file(BACKUP_LOCK_COMMIT, name);
}

/** process-wide backup context */
struct backup_context
{
  /** Engine-specific context, shared by all threads.  Only
  modified by the calling thread while no backup_step() is pending;
  backup_step() must use find(), never operator[], because
  operator[] is not safe for concurrent use. */
  std::unordered_map<const handlerton*,void*> ha_data;
#ifndef _WIN32
  /** directory stream */
  DIR *dir{};
  /** the readdir(dir) result for which subdir was opened */
  const struct dirent *d{};
  /** subdirectory stream, or NULL if iterating to next entry in dir */
  DIR *subdir{};
#else
  /** directory iterator */
  HANDLE dir{INVALID_HANDLE_VALUE};
    /** subdirectory iterator, or INVALID_HANDLE_VALUE */
  HANDLE subdir{INVALID_HANDLE_VALUE};
#endif
  /** whether the operation failed */
  bool fail{};
  /** mutex protecting d, subdir, fail */
  std::mutex mutex;
  /** Number of background tasks executing backup_step_callback */
  Atomic_counter<int> pending{0};
#ifdef _WIN32
  /** FindFirstFileA()/FindNextFile() buffer for dir */
  WIN32_FIND_DATAA d{};
  /** FindFirstFileA()/FindNextFile() buffer for subdir */
  WIN32_FIND_DATAA sd{};
#endif

#ifndef _WIN32
  backup_context()
  {
    int dfd= open(mysql_data_home, O_DIRECTORY);
    if (dfd >= 0)
    {
      if (!(dir= fdopendir(dfd)))
        close(dfd);
    }
  }
  bool invalid() const noexcept { return !dir; }
  ~backup_context()
  {
    assert(!pending);
    if (dir)
      closedir(dir);
    if (subdir)
      closedir(subdir);
  }
#else
  backup_context() : dir(FindFirstFileA("*.*", &d)) {}
  bool invalid() const noexcept { return dir == INVALID_HANDLE_VALUE; }
  ~backup_context()
  {
    assert(!pending);
    if (dir != INVALID_HANDLE_VALUE)
      FindClose(dir);
    if (subdir != INVALID_HANDLE_VALUE)
      FindClose(subdir);
  }
#endif
  /**
     Rewind the directory.
     @return whether the operation failed
  */
  bool rewinddir() noexcept // FIXME: not used yet
  {
    assert(!invalid());
#ifndef _WIN32
    assert(!d);
    assert(!subdir);
    ::rewinddir(dir);
#else
    assert(subdir == INVALID_HANDLE_VALUE);
    FindClose(dir);
    dir= FindFirstFileA("*.*", &d);
    if (invalid())
    {
      dir_error(mysql_data_home);
      return true;
    }
#endif
    return false;
  }

  /**
     Create a subdirectory unless we are streaming or it pre-exists.
     @param target    BACKUP SERVER TO target directory
     @param name      base name of subdirectory to create
     @return error code (also errno will be set)
     @retval 0 on success (errno might not be touched)
  */
  static int mkdir(const backup_target &target, const char *name) noexcept
  {
#ifdef _WIN32
    if (!target.path)
      return 0;
    int ret= snprintf(NULL, 0, "%s/%s", target.path, name) + 1;
    char *path= static_cast<char*>(malloc(ret));
    if (!path)
    {
      my_error(ER_TOO_LONG_IDENT, MYF(0), name);
      return 1;
    }
    snprintf(path, ret, "%s/%s", target.path, name);
    if (CreateDirectory(path, NULL))
      ret= 0;
    else
    {
      const DWORD err= GetLastError();
      ret= err != ERROR_ALREADY_EXISTS;
      if (ret)
      {
        my_osmaperr(err);
        my_error(ER_CANT_CREATE_FILE, MYF(0), path, errno);
      }
    }
    free(path);
    return ret;
#else
    if (target.fd == -1 || likely(!mkdirat(target.fd, name, 0777)) ||
        errno == EEXIST)
      return 0;
    my_error(ER_CANT_CREATE_FILE, MYF(0), name, errno);
    return 1;
#endif
  }

  /**
     Determine if a built-in file may be backed up.
     @param file_name   candidate file name
     @param len         strlen(file_name)
     @return whether the file may be included
  */
  static bool is_db_file(const char *file_name, size_t len) noexcept
  {
    uint32_t suffix;
    assert(len >= 4);
    memcpy(&suffix, file_name + len - 4, 4);
    switch (suffix) {
#ifdef WORDS_BIGENDIAN
    case 0x2e41524d: /* .ARM ENGINE=ARCHIVE metadata */
    case 0x2e41525a: /* .ARZ ENGINE=ARCHIVE compressed data */
    case 0x2e43534d: /* .CSM ENGINE=CSV metadata */
    case 0x2e435356: /* .CSV ENGINE=CSV data ("comma separated values") */
    case 0x2e4d5247: /* .MRG ENGINE=MRG_MyISAM */
    case 0x2e4d5944: /* .MYD ENGINE=MyISAM data heap */
    case 0x2e4d5949: /* .MYI ENGINE=MyISAM indexes */
    case 0x2e545247: /* .TRG trigger definition */
    case 0x2e54524e: /* .TRN trigger name */
    case 0x2e66726d: /* .frm form (SHOW CREATE TABLE) */
    case 0x2e706172: /* .par PARTITION metadata */
    case 0x2e69736c: /* .isl ENGINE=InnoDB DATA DIRECTORY "symbolic link" */
#else
    case 0x4d52412e: /* .ARM ENGINE=ARCHIVE metadata */
    case 0x5a52412e: /* .ARZ ENGINE=ARCHIVE compressed data */
    case 0x4d53432e: /* .CSM ENGINE=CSV metadata */
    case 0x5653432e: /* .CSV ENGINE=CSV data ("comma separated values") */
    case 0x47524d2e: /* .MRG ENGINE=MRG_MyISAM */
    case 0x44594d2e: /* .MYD ENGINE=MyISAM data heap */
    case 0x49594d2e: /* .MYI ENGINE=MyISAM indexes */
    case 0x4752542e: /* .TRG trigger definition */
    case 0x4e52542e: /* .TRN trigger name */
    case 0x6d72662e: /* .frm form (SHOW CREATE TABLE) */
    case 0x7261702e: /* .par PARTITION metadata */
    case 0x6c73692e: /* .isl ENGINE=InnoDB DATA DIRECTORY "symbolic link" */
#endif
      return true;
    }
    return len == 6 && !memcmp(file_name, C_STRING_WITH_LEN("db.opt"));
  }

  /**
     Report that a directory cannot be read.
     @param name   directory name
  */
  ATTRIBUTE_COLD ATTRIBUTE_NOINLINE static void dir_error(const char *name)
    noexcept
  {
#ifdef _WIN32
    my_osmaperr(GetLastError());
#endif
    my_error(ER_CANT_READ_DIR, MYF(0), name, errno);
  }

  /**
     Consume a data file name and copy the file if needed.
     @param target     BACKUP SERVER target
     @param sink       per-thread context (possibly, a stream to write to)
     @retval 1 on success if some work remains
     @retval 0 on successful completion
     @retval -1 on error
  */
  int step(const backup_target &target, const backup_sink &sink) noexcept
  {
    const char *filename{};
    char path[FN_REFLEN + 2];
    int left{};
#ifndef _WIN32
    struct stat sb;
    assert(dir);
#else
    assert(dir != INVALID_HANDLE_VALUE);
#endif
    {
      std::lock_guard<std::mutex> _{mutex};

      if (fail)
      {
      err_exit:
        fail= true;
        my_error(ER_UNKNOWN_ERROR, MYF(0));
        return -1;
      }
#ifndef _WIN32
      else if (!subdir)
      {
        while (struct dirent *de= readdir(dir))
        {
          switch (de->d_type) {
          default:
            continue;
          case DT_LNK:
          case DT_UNKNOWN:
            if (fstatat(dirfd(dir), de->d_name, &sb, 0) ||
                (sb.st_mode & S_IFMT) != S_IFDIR)
              continue;
            /* fall through */
          case DT_DIR:
            if (de->d_name[0] == '.' &&
                (de->d_name[1] == '\0' ||
                 (de->d_name[1] == '.' && de->d_name[2] == '\0')))
              continue;
            if (is_in_ignore_db_dirs_list(de->d_name))
              continue;
          }
          if (!mkdir(target, de->d_name))
          {
            int dfd= openat(dirfd(dir), de->d_name, O_DIRECTORY);
            d= de;
            if (dfd >= 0)
            {
              if ((subdir= fdopendir(dfd)))
                goto consume_subdir;
              close(dfd);
            }
            dir_error(de->d_name);
          }
          goto err_exit;
        }
      }
      else
      {
      consume_subdir:
        assert(d);
        assert(d->d_type == DT_DIR || d->d_type == DT_LNK ||
               d->d_type == DT_UNKNOWN);
        struct dirent *sd;
        while ((sd= readdir(subdir)))
        {
          const char *const name= sd->d_name;
          switch (sd->d_type) {
          default:
            continue;
          case DT_REG:
            break;
          case DT_LNK:
          case DT_UNKNOWN:
            if (fstatat(dirfd(subdir), name, &sb, 0) ||
                (sb.st_mode & S_IFMT) != S_IFREG)
              continue;
          }
          /* Consume a file name */
          if ((int) sizeof path <=
              snprintf(path, sizeof path, "%s/%s", d->d_name, name))
          {
            path[(sizeof path) - 1]= '\0';
            my_error(ER_TOO_LONG_IDENT, MYF(0), path);
            goto err_exit;
          }
          filename= path;
          break;
        }

        left= 1;
        if (!sd)
        {
          closedir(subdir);
          d= nullptr;
          subdir= nullptr;
        }
      }
#else
      else if (subdir == INVALID_HANDLE_VALUE)
      {
        do
        {
          if (!(d.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY))
            continue;
          if (d.cFileName[0] == '.' &&
              (d.cFileName[1] == '\0' ||
               (d.cFileName[1] == '.' && d.cFileName[2] == '\0')))
            continue;
          if (is_in_ignore_db_dirs_list(d.cFileName))
            continue;
          if ((int) sizeof path <=
              snprintf(path, sizeof path, "%s/*.*", d.cFileName))
          name_too_long:
            my_error(ER_TOO_LONG_IDENT, MYF(0), path);
          else if (mkdir(target, d.cFileName));
          else if ((subdir= FindFirstFileA(path, &sd)) !=
                   INVALID_HANDLE_VALUE)
            goto consume_subdir;
          else
            dir_error(path);
          goto err_exit;
        }
        while (FindNextFile(dir, &d));
      }
      else
      {
      consume_subdir:
        do
        {
          const char *const name= sd.cFileName;
          if (sd.dwFileAttributes & FILE_ATTRIBUTE_DIRECTORY)
            continue;
          /* Consume a file name */
          if ((int) sizeof path <=
              snprintf(path, sizeof path, "%s/%s", d.cFileName, name))
            goto name_too_long;
          filename= path;
        }
        while ((left= FindNextFile(subdir, &sd)) && !filename);

        if (!left)
        {
          FindClose(subdir);
          subdir= INVALID_HANDLE_VALUE;
          left= FindNextFile(dir, &d);
        }
      }
#endif
    }

    if (!filename)
      return left;

    assert(left >= 0);

    do
    {
      const char *name{strrchr(filename, '/')};
      assert(name); /* we constructed this with %s/%s */
      size_t len= strlen(++name);
      static_assert(tmp_file_prefix_length == 4);
      if (len < 4 ||
          /*
            As noted in MDEV-25854, file names that start with #sql
            must be excluded from the backup. For example, a call to
            MDL_context::upgrade_shared_lock() in
            mysql_inplace_alter_table() could time out, resulting in
            cleanup_table_after_inplace_alter() deleting a
            #sql-alter*.frm file before we get a chance to copy it.
          */
          !memcmp(name, tmp_file_prefix, tmp_file_prefix_length))
        continue;

      if (!is_db_file(name, len))
      {
        LEX_CSTRING n{name, len};
        if (plugin_foreach_with_mask(nullptr, backup_name_filtered,
                                     MYSQL_STORAGE_ENGINE_PLUGIN,
                                     PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                                     &n))
          continue;
      }

      if (backup::copy_or_stream(target, sink,
#ifndef _WIN32
                                 openat(dirfd(dir), filename, O_RDONLY),
#endif
                                 filename))
        return -1;
    }
    while (false);

    return left;
  }

#ifndef _WIN32
#else

#endif
};

/** per-thread backup context */
struct backup_target_stage
{
  /** target directory or stream */
  backup_target target;
  /** current backup stage */
  backup_stages stage;
  /** backup worker state (output stream) */
  backup_sink sink;
  /** stream object for streaming backup */
  FILE *stream;
  /** handlerton::backup_step return value in multi-threaded operation */
  int ret;
  /** process-wide backup context */
  backup_context &context;

  /**
     Scan the data directory and copy files if needed.
     @return whether an error occurred
  */
  int step() const noexcept
  {
    if (stage == BACKUP_LOCK_COMMIT)
      return context.step(target, sink);
    return 0;
  }
};

/**
   Inform a storage engine of an upcoming backup by invoking
   handlerton::backup_start(BACKUP_PREPARE) before
   acquiring any locks.
   @param thd     current session
   @param plugin  storage engine
   @return whether the operation failed
*/
static my_bool backup_preparation(THD *thd, plugin_ref plugin, void*) noexcept
{
  const auto bs= plugin_hton(plugin)->backup_start;
  return bs && bs(thd, nullptr, BACKUP_PREPARE, nullptr);
}

/**
   Invoke handlerton::backup_start() on a storage engine,
   when there are no pending handlerton::backup_step() in any thread.
   @param thd     current session
   @param plugin  storage engine
   @param arg     the backup_target_stage context
   @return whether the operation failed
*/
static my_bool backup_start(THD *thd, plugin_ref plugin, void *arg) noexcept
{
  const handlerton *hton= plugin_hton(plugin);
  backup_target_stage &t{*static_cast<backup_target_stage*>(arg)};
  assert(int{t.stage} >= 0);
  if (hton->backup_start)
  {
    t.sink.ha_data= t.context.ha_data[hton];
    void *data= hton->backup_start(thd, &t.target, t.stage, &t.sink);
    if (data == reinterpret_cast<void*>(-1))
      return true;
    assert(!t.context.ha_data[hton] || t.context.ha_data[hton] == data);
    t.context.ha_data[hton]= data;
  }
  return false;
}

/**
   Invoke handlerton::backup_end() on a storage engine,
   when there are no pending handlerton::backup_step() in any thread.
   @param thd     current session
   @param plugin  storage engine
   @param arg     the backup_target_stage context
   @return whether the operation failed
*/
static my_bool backup_end(THD *thd, plugin_ref plugin, void *arg) noexcept
{
  const handlerton *hton= plugin_hton(plugin);
  backup_target_stage &t{*static_cast<backup_target_stage*>(arg)};
  if (hton->backup_end)
  {
    t.sink.ha_data= t.context.ha_data[hton];
    return hton->backup_end(thd, &t.target, t.stage, &t.sink);
  }
  return false;
}

/**
   Invoke handlerton::backup_step() on a storage engine in a thread
   that may or may not be associated with a BACKUP SERVER connection,
   between handlerton::backup_start() and handlerton::backup_end()
   of the same stage.
   @param thd     the BACKUP SERVER session
   @param plugin  storage engine
   @param arg     the backup_target_phase context
   @return whether the operation failed
*/
static my_bool backup_step(THD *thd, plugin_ref plugin, void *arg) noexcept
{
  const handlerton *hton= plugin_hton(plugin);
  backup_target_stage &t{*static_cast<backup_target_stage*>(arg)};
  assert(int{t.stage} >= 0);
  if (hton->backup_step)
  {
    const auto it= t.context.ha_data.find(hton);
    t.sink.ha_data= it == t.context.ha_data.end() ? nullptr : it->second;
    while (int res= hton->backup_step(thd, &t.target, t.stage, &t.sink))
      if (res < 0)
        return true;
  }
  return false;
}


/**
   Execute all handlerton::backup_step() until completion or failure.
   @param thd           current connection
   @param target_stage  backup target and stage
   @return whether the operation failed
*/
static bool backup_step_one(THD *thd, backup_target_stage *target_stage)
{
  while (int ret= target_stage->step())
    if (ret < 0)
      return ret;
  return plugin_foreach_with_mask(thd, backup_step,
                                  MYSQL_STORAGE_ENGINE_PLUGIN,
                                  PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                                  target_stage);
}


/**
   Initialize a thread that was created by the thread pool.

   Any thread that uses mysys functions must call my_thread_init()
   before it does so; my_errno and other mysys variables are stored in
   a thread local variable that my_thread_init() allocates.
*/
static void backup_thread_init() noexcept
{
  my_thread_init();
  my_thread_set_name("backup_worker");
  /* Make the thread visible in performance_schema.threads */
#ifdef HAVE_PSI_THREAD_INTERFACE
  PSI_thread *psi{PSI_CALL_new_thread(key_thread_backup_worker, NULL, 0)};
  PSI_CALL_set_thread_os_id(psi);
  PSI_CALL_set_thread(psi);
#endif
}


/** Clean up a thread that was initialized by backup_thread_init() */
static void backup_thread_end() noexcept
{
  PSI_CALL_delete_current_thread();
  my_thread_end();
}


/** Invoke backup_step() in a background task */
static void backup_step_callback(void *arg) noexcept
{
  backup_target_stage &t{*static_cast<backup_target_stage*>(arg)};
  assert(!t.ret);
  while (int ret= t.step())
    if (ret < 0)
    {
      t.ret= true;
      goto done;
    }
  t.ret= plugin_foreach_with_mask(nullptr, backup_step,
                                  MYSQL_STORAGE_ENGINE_PLUGIN,
                                  PLUGIN_IS_DELETED|PLUGIN_IS_READY, &t);
 done:
#ifndef NDEBUG
  auto was_pending=
#endif
    t.context.pending--;
  assert(was_pending);
}

/**
   Execute all handlerton::backup_step() until completion or failure.
   @param thd           current connection
   @param target_stage  backup target and stage
   @param threads       number of execution threads
   @param tp            thread pool
*/
static bool backup_steps(THD *thd, backup_target_stage *target_stage,
                         int threads, tpool::thread_pool *tp)
{
  auto &pending{target_stage->context.pending};
  assert(!pending);
  if (unlikely(thd->killed != NOT_KILLED))
    return true;
  if (threads == 1)
    return backup_step_one(thd, target_stage);
  tpool::task *const tasks=
    static_cast<tpool::task*>(alloca(threads * sizeof *tasks));
  pending= threads - 1;
  assert(target_stage[0].sink.id == threads - 1);
  for (int n{threads}; --n; )
  {
    assert(target_stage[n].sink.id == (threads - 1) - n);
    target_stage[n].stage= target_stage->stage;
    tp->submit_task(new (&tasks[n]) tpool::task{backup_step_callback,
                                                &target_stage[n]});
  }
  bool fail= backup_step_one(thd, target_stage);
  while (pending)
    std::this_thread::sleep_for(std::chrono::milliseconds(100));

  if (fail)
    return fail;

  for (int n{threads}; --n; )
    if (target_stage[n].ret)
    {
      my_error(ER_UNKNOWN_ERROR, MYF(0));
      return true;
    }

  return false;
}


/**
   BACKUP SERVER driver

   @param thd     client connection
   @param target  BACKUP SERVER TO directory to be created
   @param command BACKUP SERVER WITH command
   @param threads number of CONCURRENT threads to employ
   @return whether the operation failed
*/
static bool backup_execute(THD *thd, const char *target, const char *command,
                           int threads)
{
  assert(!!target == !command);
  assert(threads > 0);

  if (!target)
  {
    /*
      Disallow path separators and the quote character.
    */
    if (false ||
#ifndef _WIN32
        strchr(command, '\'') /* quote */ ||
#else
        strchr(command, '"') /* quote */ ||
        strchr(command, '\\') /* special inside quotes */ ||
#endif
        strchr(command, '/') /* disallow path separator */)
    {
      my_error(ER_WRONG_ARGUMENTS, MYF(0), "BACKUP SERVER WITH");
      return true;
    }
  }
  else if (error_if_data_home_dir(target, "BACKUP SERVER TO"))
    return true;

  bool fail= plugin_foreach_with_mask(thd, backup_preparation,
                                      MYSQL_STORAGE_ENGINE_PLUGIN,
                                      PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                                      nullptr);
  if (fail)
  {
    my_error(ER_OUT_OF_RESOURCES, MYF(0));
    return true;
  }

  if (run_backup_stage(thd, BACKUP_START))
    return true;

  tpool::thread_pool *tp= nullptr;
  backup_context context{};
  if (context.invalid())
  {
    context.dir_error(mysql_data_home);
  release_and_exit:
    run_backup_stage(thd, BACKUP_END);
    return true;
  }
#ifndef NDEBUG
# define calloca(n) memset(alloca(n), 0, n)
#else
# define calloca(n) alloca(n)
#endif
  backup_target_stage *target_stage= static_cast<backup_target_stage*>
    (calloca(threads * sizeof *target_stage));
#undef calloca

  if (threads <= 1);
  else if (!(tp= tpool::create_thread_pool_generic()))
  {
  oor:
    my_error(ER_OUT_OF_RESOURCES, MYF(0));
  err_exit:
    delete tp;
#ifndef NDEBUG
    for (int t{threads}; t--; )
      assert(!target_stage[t].stream);
#endif
    goto release_and_exit;
  }
  else
    tp->set_thread_callbacks(backup_thread_init, backup_thread_end);

  if (command)
  {
    char cmd[1024];
    for (int t{threads}; t; t--)
    {
#ifndef _WIN32
      /*
        popen(3) will prepend "/bin/sh -c" which preserves the literal
        meaning of all characters enclosed in single quotes.
      */
      constexpr const char cmd_fmt[]= "'mariadb-backup-%s' %d";
#else
      /*
        my_win_popen() will prepend "cmd.exe /c ";
        add double quotes around its first argument
      */
      constexpr const char cmd_fmt[]= "\"mariadb-backup-%s\" %d";
#endif
      if (size_t(snprintf(cmd, sizeof cmd, cmd_fmt, command, t)) >= sizeof cmd)
        goto oor;
      FILE *f= my_popen(cmd, "w");
      if (!f)
      {
        while (t++ < threads)
        {
          my_pclose(target_stage[threads - t].stream);
#ifndef NDEBUG
          target_stage[threads - t].stream= nullptr;
#endif
        }
        goto oor;
      }
#ifdef _WIN32
      HANDLE sink= (HANDLE) _get_osfhandle(_fileno(f));
#else
      int sink= fileno(f);
# ifdef __linux__
      /* Request a larger pipe buffer; the default appears to be 64 KiB. */
      fcntl(sink, F_SETPIPE_SZ, 1 << 20/* 1 MiB */);
# endif
#endif
      new (&target_stage[threads - t])
        backup_target_stage{backup_target{IF_WIN(nullptr, -1)},
          BACKUP_START, backup_sink{t - 1, sink, nullptr}, f, 0, context};
    }
  }
  else if (my_mkdir(target, 0755, MYF(MY_WME)))
    goto err_exit;
  else
  {
#ifndef _WIN32
    const int dir{open(target, O_DIRECTORY)};
    if (dir < 0)
    {
      my_error(EE_CANT_MKDIR, MYF(ME_BELL), target, errno);
      goto err_exit;
    }
#endif
    for (int t{threads}; t--; )
    {
      new (&target_stage[threads - 1 - t])
        backup_target_stage{backup_target{IF_WIN(target, dir)},
          BACKUP_START, backup_sink{t, backup_sink::NO_STREAM, nullptr},
          nullptr, 0, context};
    }
  }

  int stage{BACKUP_START};
  goto backup_start;

  for (; stage <= int{BACKUP_LOCK_COMMIT}; stage++)
  {
    fail= run_backup_stage(thd, backup_stages(stage));

    if (fail)
      break;

  backup_start:
    target_stage->stage= backup_stages(stage);
    fail= plugin_foreach_with_mask(thd, backup_start,
                                   MYSQL_STORAGE_ENGINE_PLUGIN,
                                   PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                                   target_stage);
    if (fail)
      break;
    fail= backup_steps(thd, target_stage, threads, tp) ||
      plugin_foreach_with_mask(thd, backup_end, MYSQL_STORAGE_ENGINE_PLUGIN,
                               PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                               target_stage);
    if (fail)
      break;
  }

#ifdef FIXME_ARIA_LOG_LSN
  /*
    TODO: Rotate the Aria log when determining the commit LSN so that
    we can safely copy the last complete log file.
  */
  fail= run_backup_stage(thd, BACKUP_END) || fail;
#endif

  if (!fail)
  {
    target_stage->stage= BACKUP_FINISHED;
    fail= plugin_foreach_with_mask(thd, backup_start,
                                   MYSQL_STORAGE_ENGINE_PLUGIN,
                                   PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                                   target_stage) ||
      backup_steps(thd, target_stage, threads, tp);
  }
  else
  {
    target_stage->stage= BACKUP_ABORTED;
    plugin_foreach_with_mask(thd, backup_end, MYSQL_STORAGE_ENGINE_PLUGIN,
                             PLUGIN_IS_DELETED|PLUGIN_IS_READY, target_stage);
    target_stage->stage= BACKUP_FINISHED;
  }

  fail=
    plugin_foreach_with_mask(thd, backup_end, MYSQL_STORAGE_ENGINE_PLUGIN,
                             PLUGIN_IS_DELETED|PLUGIN_IS_READY,
                             target_stage) || fail;

#ifndef FIXME_ARIA_LOG_LSN
  /*
    We cannot call BACKUP_END earlier as the Aria logs must be copied under
    BLOCK_COMMIT.  If we relase the commit lock, the redo log will be filled
    with data that we do not want.
  */
  fail= run_backup_stage(thd, BACKUP_END) || fail;
#endif
  delete tp;

  if (command)
  {
    if (fail)
      /* Ensure that each stream is an invalid tar stream. */
      for (int t= threads; t--; )
        std::ignore= putc('\0', target_stage[t].stream);
    int pfail{0};
    for (int t= threads; t--; )
      pfail|= my_pclose(target_stage[t].stream);
    if (!fail && (fail= !!pfail))
      my_error(ER_NET_ERROR_ON_WRITE, MYF(0));
  }
#ifndef _WIN32
  else
    std::ignore= close(target_stage->target.fd);
#endif

  if (!fail)
    my_ok(thd);
  return fail;
}

/**
   Encode an octal string.
   @param start   first byte of buffer
   @param end     first byte after buffer
   @param n       number to encode
*/
static void ustar_write_octal(char *start, char *end, uint64_t n) noexcept
{
  for (*--end= '\0'; --end >= start; n>>= 3)
    *end= char('0' + (n & 7));
}

/**
   Encode a quantity in 12 bytes.
   @param start   first byte of the buffer
   @param n       number to encode
*/
static void ustar_write_dozen(char *start, uint64_t n) noexcept
{
  if (n < 1ULL << 33)
    ustar_write_octal(start, start + 12, n);
  else
  {
    const uint32_t head{my_htobe32(1U << 31)};
    n= my_htobe64(n);
    memcpy(start, &head, 4);
    memcpy(start + 4, &n, 8);
  }
}

/** Initialize a ustar block
@param buf  GNU tape archiver --format=oldgnu header block
@param name name of the block
@param mode file access mode
@param size physical size of the following block */
static void ustar_block_init(char *buf, const char *name, mode_t mode,
                             uint64_t size) noexcept
{
  strncpy(buf, name, 99);
  buf[99]= '\0';
  ustar_write_octal(buf + 100, buf + 108, uint64_t(mode));
  ustar_write_octal(buf + 108, buf + 116, 0/* POSIX uid */);
  ustar_write_octal(buf + 116, buf + 124, 0/* POSIX gid */);
  ustar_write_dozen(buf + 124, size);
  /* last modification time */
  ustar_write_octal(buf + 136, buf + 148, 0);
  memset(buf + 148, ' ', 9); /* initial block checksum and dummy type */
  memset(buf + 157, '\0', 100); /* name of linked file (unused) */
  memcpy(buf + 257, "ustar  ", 8);
  strncpy(buf + 265, "root" /* POSIX owner name */, 32);
  strncpy(buf + 297, "root" /* POSIX group name */, 512 - 297);
  /* The caller will fill in the rest and invoke ustar_block_checksum() */
}

/**
   Compute and write the POSIX tar block checksum.
   @param buf    POSIX tar block
*/
static void ustar_block_checksum(char *buf) noexcept
{
  uint16_t sum{0};
  for (int i{0}; i < 512; i++)
    sum+= uint16_t{uint8_t(buf[i])};
  ustar_write_octal(buf + 148, buf + 155, sum);
}

/**
   Write data to a stream.
   @param stream  backup stream
   @param buf     source buffer
   @param size    length of the buffer (usually an integer multiple of 512)
   @return error code (non-positive)
   @retval 0 on success
*/
int backup_stream_write(backup_fd stream, const void *buf, size_t size)
  noexcept
{
#ifdef _WIN32
  for (DWORD sz= DWORD(size);;)
  {
    DWORD wrote;
    if (WriteFile(stream, buf, sz, &wrote, nullptr))
    {
      assert(wrote == sz);
      return 0;
    }
    else if (GetLastError() != ERROR_IO_PENDING)
      return -1;
    buf= static_cast<const char*>(buf) + wrote;
    sz-= wrote;
  }
#else
  do
  {
    ssize_t wrote= write(stream, buf, size);
    assert(wrote <= ssize_t(size));
    if (wrote < 0)
    {
      if (errno == EAGAIN)
        continue;
      return -1;
    }
    buf= static_cast<const char*>(buf) + wrote;
    size-= size_t(wrote);
  }
  while (size);
#endif
  return 0;
}

/**
   Copy a prefix of a NUL terminated string to a buffer, NUL-padded.
   @param b     output buffer
   @param s     NUL terminated string
   @param size  size of buf, in bytes
*/
static inline char *ustar_zeropad(char *b, const char *s, size_t size) noexcept
{
#if defined __GNUC__ && __GNUC__ >= 8
# pragma GCC diagnostic push
# pragma GCC diagnostic ignored "-Wstringop-truncation"
#endif
  return strncpy(b, s, size);
#if defined __GNUC__ && __GNUC__ >= 8
# pragma GCC diagnostic pop
#endif
}

/** Start streaming a file.
@param stream   backup stream
@param name     file name
@param mode     file access mode
@param size     physical length of the file, in bytes
@param chunks   payload chunks of a sparse file, or nullptr
@param n_chunks number of chunks; 0 unless sparse file
@return error code (non-positive)
@retval 0   on success */
int backup_stream_start(backup_fd stream,
                        const char *name, mode_t mode, uint64_t size,
                        const backup_chunk *chunks, size_t n_chunks) noexcept
{
#ifndef NDEBUG
  if (size_t n{n_chunks})
  {
    /* The physical size must be a sum of the chunk sizes. */
    uint64_t s{size};
    do s-= chunks[--n].length; while (n);
    assert(!s);
  }
#endif
  assert(stream != backup_sink::NO_STREAM);
  char buf[512];
  size_t s= strlen(name);
  if (s > 99)
  {
    /* Write a block that contains the full name length,
    followed by blocks that contain the full name, in
    tar --format=oldgnu */
    ustar_block_init(buf, "././@LongLink", 0644, s + 1);
    buf[156]= 'L';
    ustar_block_checksum(buf);
    if (int err= backup_stream_write(stream, buf, sizeof buf))
      return err;
    const size_t whole{s & ~((sizeof buf) - 1)};
    if (whole)
      if (int err= backup_stream_write(stream, name, whole))
        return err;
    /* After 512*n bytes of file name there will be a zerofilled block. */
    ustar_zeropad(buf, name + whole, sizeof buf);
    if (int err= backup_stream_write(stream, buf, sizeof buf))
      return err;
  }

  ustar_block_init(buf, name, mode, size);
  if (!n_chunks)
    buf[156]= '0';
  else
  {
    buf[156]= 'S';
    char *h= &buf[386];
    if (n_chunks > 4)
      return -1; // FIXME; support more chunks
    for (size_t i= 0; i < n_chunks; i++, h+= 24)
    {
      ustar_write_dozen(h, chunks[i].offset);
      ustar_write_dozen(h + 12, chunks[i].length);
    }
    ustar_write_dozen(&buf[0x1e3], chunks[n_chunks - 1].offset +
                      chunks[n_chunks - 1].length);
  }
  ustar_block_checksum(buf);
  return backup_stream_write(stream, buf, sizeof buf);
}

/** Append to the configuration file.
@param target   backup stream
@param config   the configuration file snippet to append
@param size     length of the snippet
@return error code (non-positive)
@retval 0   on success */
int backup_stream_config(backup_fd stream,
                         const char *config, size_t size) noexcept
{
  /* FIXME: append to a pre-created configuration file */
  if (int ret=
      backup_stream_start(stream, "backup.cnf", 0644, size, nullptr, 0))
    return ret;
  char buf[512];
  const size_t whole{size & ~((sizeof buf) - 1)};
  if (whole)
    if (int err= backup_stream_write(stream, config, whole))
      return err;
  if (size == whole)
    return 0;
  ustar_zeropad(buf, config + whole, sizeof buf);
  return backup_stream_write(stream, buf, sizeof buf);
}

#ifndef _WIN32
# ifdef __linux__
#  include <sys/sendfile.h>
/** Copy a file to a stream or to a regular file. */
static inline ssize_t
send_step(int in_fd, int out_fd, size_t count, off_t *offset) noexcept
{
  return sendfile(out_fd, in_fd, offset, count);
}
# endif

/**
   Zero-copy append an immutable file snippet to a stream.

   The caller guarantees that this section of the file will remain
   intact until the stream is closed. This guarantee is needed
   because the receiving end of the stream pipe might delay consuming
   the data, and the operating system might point the pipe buffer
   to the block cache for a long time.

   Note that tar uses 512-byte blocks. If end-start is not a multiple of
   512 bytes, backup_stream_zeropad() must be invoked.
   @param src    source file
   @param stream backup stream
   @param start  first offset to copy
   @param end    last offset to copy (exclusive)
   @return error code (non-positive)
   @retval 0   on success
*/
int backup_stream_append_async(int src, int stream,
                               uint64_t start, uint64_t end) noexcept
{
  assert(stream != backup_sink::NO_STREAM);
# ifdef __linux__
  return int(stepwise<send_step,true>(src, stream, off_t(start), off_t(end)));
# else
  return int(pread_write<true>(src, stream, start, end));
# endif
}
#endif