// file      : bbot/agent/agent.cxx -*- C++ -*-
// license   : TBC; see accompanying LICENSE file

#include <bbot/agent/agent.hxx>

#include <pwd.h>       // getpwuid()
#include <limits.h>    // PATH_MAX
#include <signal.h>    // signal()
#include <stdlib.h>    // rand_r()
#include <unistd.h>    // sleep(), getuid(), fsync(), [f]stat()
#include <ifaddrs.h>   // getifaddrs(), freeifaddrs()
#include <sys/types.h> // stat
#include <sys/stat.h>  // [f]stat()
#include <sys/file.h>  // flock()

#include <net/if.h>     // ifreq
#include <netinet/in.h> // sockaddr_in
#include <arpa/inet.h>  // inet_ntop()
#include <sys/ioctl.h>
#include <sys/socket.h>

#include <chrono>
#include <random>
#include <iostream>
#include <system_error> // generic_category()

#include <libbutl/pager.mxx>
#include <libbutl/sha256.mxx>
#include <libbutl/openssl.mxx>
#include <libbutl/filesystem.mxx> // dir_iterator, try_rmfile(), readsymlink()

#include <libbbot/manifest.hxx>

#include <bbot/types.hxx>
#include <bbot/utility.hxx>
#include <bbot/diagnostics.hxx>

#include <bbot/machine-manifest.hxx>
#include <bbot/bootstrap-manifest.hxx>

#include <bbot/agent/tftp.hxx>
#include <bbot/agent/machine.hxx>

using namespace butl;
using namespace bbot;

using std::cout;
using std::endl;

namespace bbot
{
  agent_options ops;

  const string bs_prot ("1");

  string           tc_name;
  uint16_t         tc_num;
  standard_version tc_ver;
  string           tc_id;

  uint16_t inst;

  uint16_t offset;

  string hname;
  string hip;
  uid_t  uid;
  string uname;
}

static void
file_sync (const path& f)
{
  auto_fd fd (fdopen (f, fdopen_mode::in));
  if (fsync (fd.get ()) != 0)
    throw_system_error (errno);
}

static bool
file_not_empty (const path& f)
{
  if (file_exists (f))
  {
    file_sync (f);
    return !file_empty (f);
  }
  return false;
}

// The btrfs tool likes to print informational messages, like "Created
// snapshot such and such". Luckily, it writes them to stdout while proper
// diagnostics goes to stderr.
//
template <typename... A>
inline void
run_btrfs (tracer& t, A&&... a)
{
  if (verb >= 4)
    run_io (t, fdopen_null (), 2, 2, "btrfs", forward<A> (a)...);
  else
    run_io (t, fdopen_null (), fdopen_null (), 2, "btrfs", forward<A> (a)...);
}

template <typename... A>
inline butl::process_exit::code_type
btrfs_exit (tracer& t, A&&... a)
{
  return verb >= 4
    ? run_io_exit (t, fdopen_null (), 2, 2, "btrfs", forward<A> (a)...)
    : run_io_exit (t,
                   fdopen_null (), fdopen_null (), 2,
                   "btrfs", forward<A> (a)...);
}

// Bootstrap the machine. Return the bootstrapped machine manifest if
// successful and nullopt otherwise (in which case the machine directory
// should be cleaned and the machine ignored for now).
//
static optional<bootstrapped_machine_manifest>
bootstrap_machine (const dir_path& md,
                   const machine_manifest& mm,
                   optional<bootstrapped_machine_manifest> obmm)
{
  tracer trace ("bootstrap_machine", md.string ().c_str ());

  bootstrapped_machine_manifest r {
    mm,
    toolchain_manifest {tc_id.empty () ? "bogus" : tc_id},
    bootstrap_manifest {
      bootstrap_manifest::versions_type {
        {"bbot",    standard_version (BBOT_VERSION_STR)},
        {"libbbot", standard_version (LIBBBOT_VERSION_STR)},
        {"libbpkg", standard_version (LIBBPKG_VERSION_STR)},
        {"libbutl", standard_version (LIBBUTL_VERSION_STR)}
      }
    }
  };

  if (ops.fake_bootstrap ())
  {
    r.machine.mac = "de:ad:be:ef:de:ad";
  }
  else
  try
  {
    // Start the TFTP server (server chroot is --tftp). Map:
    //
    // GET requests to .../toolchains/<name>/*
    // PUT requests to .../bootstrap/<name>-<instance>/*
    //
    const string in_name (tc_name + '-' + to_string (inst));
    auto_rmdir arm ((dir_path (ops.tftp ()) /= "bootstrap") /= in_name);
    try_mkdir_p (arm.path);

    // Bootstrap result manifest.
    //
    path mf (arm.path / "bootstrap.manifest");
    try_rmfile (mf);

    // @@ TMP BC: also check for the old manifest name until we migrate all
    //    the machines.
    //
    path mfo (arm.path / "manifest");
    try_rmfile (mfo);

    // Note that unlike build, here we use the same VM snapshot for retries,
    // which is not ideal.
    //
    for (size_t retry (0);; ++retry)
    {
      tftp_server tftpd ("Gr  ^/?(.+)$  /toolchains/" + tc_name + "/\\1\n" +
                         "Pr  ^/?(.+)$  /bootstrap/" + in_name + "/\\1\n",
                         ops.tftp_port () + offset);

      l3 ([&]{trace << "tftp server on port " << tftpd.port ();});

      // Start the machine.
      //
      unique_ptr<machine> m (
        start_machine (md,
                       mm,
                       obmm ? obmm->machine.mac : nullopt,
                       ops.bridge (),
                       tftpd.port (),
                       false /* pub_vnc */));

      {
        // If we are terminating with an exception then force the machine down.
        // Failed that, the machine's destructor will block waiting for its
        // completion.
        //
        auto mg (
          make_exception_guard (
            [&m, &md] ()
            {
              info << "trying to force machine " << md << " down";
              try {m->forcedown (false);} catch (const failed&) {}
            }));

        // What happens if the bootstrap process hangs? The simple thing would
        // be to force the machine down after some timeout and then fail. But
        // that won't be very helpful for investigating the cause. So instead
        // the plan is to suspend it after some timeout, issue diagnostics
        // (without failing and which Build OS monitor will relay to the
        // operator), and wait for the external intervention.
        //
        auto soft_fail = [&md, &m] (const char* msg)
        {
          {
            diag_record dr (error);
            dr << msg << " for machine " << md << ", suspending";
            m->print_info (dr);
          }

          try
          {
            m->suspend (false);
            m->wait (false);
            m->cleanup ();
            info << "resuming after machine suspension";
          }
          catch (const failed&) {}

          return nullopt;
        };

        // Check whether the machine is still running issuing diagnostics and
        // returning false if it unexpectedly terminated.
        //
        auto check_machine = [&md, &m] ()
        {
          try
          {
            size_t t (0);
            if (!m->wait (t /* seconds */, false /* fail_hard */))
              return true; // Still running.

            // Exited successfully.
          }
          catch (const failed&)
          {
            // Failed, exit code diagnostics has already been issued.
          }

          diag_record dr (error);
          dr << "machine " << md << " exited unexpectedly";
          m->print_info (dr);

          return false;
        };

        // The first request should be the toolchain download. Wait for up to
        // 5 minutes for that to arrive. In a sense we use it as an indication
        // that the machine has booted and the bootstrap process has started.
        // Why wait so long you may wonder? Well, we may be using a new MAC
        // address and operating systems like Windows may need to digest that.
        //
        size_t to;
        const size_t startup_to   (5 * 60);
        const size_t bootstrap_to (ops.bootstrap_timeout ());
        const size_t shutdown_to  (5 * 60);

        // Wait periodically making sure the machine is still alive.
        //
        for (to = startup_to; to != 0; )
        {
          if (tftpd.serve (to, 2))
            break;

          if (!check_machine ())
            return nullopt;
        }

        // This can mean two things: machine mis-configuration or what we
        // euphemistically call a "mis-boot": the VM failed to boot for some
        // unknown/random reason. Mac OS is particularly know for suffering
        // from this. So the strategy is to retry it a couple of times and
        // then suspend for investigation.
        //
        if (to == 0)
        {
          if (retry > ops.bootstrap_retries ())
            return soft_fail ("bootstrap startup timeout");

          // Note: keeping the logs behind (no cleanup).

          diag_record dr (warn);
          dr << "machine " << mm.name << " mis-booted, retrying";
          m->print_info (dr);

          try {m->forcedown (false);} catch (const failed&) {}
          continue;
        }

        l3 ([&]{trace << "completed startup in " << startup_to - to << "s";});

        // Next the bootstrap process may download additional toolchain
        // archives, build things, and then upload the result manifest. So on
        // our side we serve TFTP requests while periodically checking for the
        // manifest file. To workaround some obscure filesystem races (the
        // file's mtime/size is updated several seconds later; maybe tmpfs
        // issue?), we periodically re-check.
        //
        for (to = bootstrap_to; to != 0; )
        {
          if (tftpd.serve (to, 2))
            continue;

          if (!check_machine ())
          {
            // The exit/upload is racy so we re-check.
            //
            if (!(file_not_empty (mf) || file_not_empty (mfo)))
              return nullopt;
          }

          bool old (false);
          if (file_not_empty (mf) || (old = file_not_empty (mfo)))
          {
            if (old)
              mf = move (mfo);

            // Wait for 5 seconds of inactivity. This is our desperate attempt
            // at handling interrupted uploads.
            //
            if (!tftpd.serve (to, 5))
              break;
          }
        }

        if (to == 0)
          return soft_fail ("bootstrap timeout");

        l3 ([&]{trace << "completed bootstrap in " << bootstrap_to - to << "s";});

        // Shut the machine down cleanly.
        //
        if (!m->shutdown ((to = shutdown_to)))
          return soft_fail ("bootstrap shutdown timeout");

        l3 ([&]{trace << "completed shutdown in " << shutdown_to - to << "s";});

        m->cleanup ();
      }

      // Parse the result manifest.
      //
      r.bootstrap = parse_manifest<bootstrap_manifest> (mf, "bootstrap");

      r.machine.mac = m->mac; // Save the MAC address.

      break;
    }
  }
  catch (const system_error& e)
  {
    fail << "bootstrap error: " << e;
  }

  serialize_manifest (r, md / "manifest", "bootstrapped machine");
  return r;
}

// Machine locking.
//
// We use flock(2) which is straightforward. The tricky part is cleaning the
// file up. Here we may have a race when two processes are trying to open &
// lock the file that is being unlocked & removed by a third process. In this
// case one of these processes may still open the old file. To resolve this,
// after opening and locking the file, we verify that a new file hasn't
// appeared by stat'ing the path and file descriptor and comparing the inodes.
//
// Note that converting a lock (shared to exclusive or vice versa) is not
// guaranteed to be atomic (in case later we want to support exclusive
// bootstrap and shared build).
//
class machine_lock
{
public:
  machine_lock () = default; // Empty lock.

  ~machine_lock ()
  {
    unlock (true /* ignore_errors */);
  }

  void
  unlock (bool ignore_errors = false)
  {
    if (fl_)
    {
      fl_ = false; // We have tried.

      try_rmfile (fp_, ignore_errors);

      if (flock (fd_.get (), LOCK_UN) != 0 && !ignore_errors)
        throw_generic_error (errno);
    }
  }

  machine_lock            (machine_lock&&) = default;
  machine_lock& operator= (machine_lock&&) = default;

  machine_lock            (const machine_lock&) = delete;
  machine_lock& operator= (const machine_lock&) = delete;

  // Implementation details.
  //
public:
  machine_lock (path&& fp, auto_fd&& fd)
      : fp_ (move (fp)), fd_ (move (fd)), fl_ (true) {}

private:
  path    fp_;
  auto_fd fd_;
  bool    fl_ = false;
};

// Try to lock the machine given its -<toolchain> directory.
//
static optional<machine_lock>
lock_machine (const dir_path& tp)
{
  path fp (tp + ".lock"); // The -<toolchain>.lock file.

  for (;;)
  {
    auto_fd fd (fdopen (fp, fdopen_mode::out | fdopen_mode::create));

    if (flock (fd.get (), LOCK_EX | LOCK_NB) != 0)
    {
      if (errno == EWOULDBLOCK)
        return nullopt;

      throw_generic_error (errno);
    }

    struct stat st1, st2;

    if (fstat (fd.get (),             &st1) != 0 ||
        stat  (fp.string ().c_str (), &st2) != 0 )   // Both should succeed.
      throw_generic_error (errno);

    if (st1.st_ino == st2.st_ino)
      return machine_lock (move (fp), move (fd));

    // Note: unlocked by close().
  }
}

// Given the toolchain directory (-<toolchain>) return the snapshot path in
// the <name>-<toolchain>-<xxx> form.
//
// We include the instance number into <xxx> for debuggability.
//
static inline dir_path
snapshot_path (const dir_path& tp)
{
  return tp.directory () /=
    path::traits_type::temp_name (tp.leaf ().string () + '-' +
                                  to_string (inst));
}

// Return available machines, (re-)bootstrapping them if necessary.
//
struct bootstrapped_machine
{
  dir_path                      path;
  bootstrapped_machine_manifest manifest;
  machine_lock                  lock;
};
using bootstrapped_machines = vector<bootstrapped_machine>;

static bootstrapped_machines
enumerate_machines (const dir_path& machines)
try
{
  tracer trace ("enumerate_machines", machines.string ().c_str ());

  bootstrapped_machines r;

  if (ops.fake_machine_specified ())
  {
    auto mh (
      parse_manifest<machine_header_manifest> (
        ops.fake_machine (), "machine header"));

    r.push_back (
      bootstrapped_machine {
        dir_path (ops.machines ()) /= mh.name, // For diagnostics.
        bootstrapped_machine_manifest {
          machine_manifest {
            move (mh.id),
            move (mh.name),
            move (mh.summary),
            machine_type::kvm,
            string ("de:ad:be:ef:de:ad"),
            nullopt,
            strings ()},
          toolchain_manifest {tc_id},
          bootstrap_manifest {}},
          machine_lock ()});

    return r;
  }

  // Notice and warn if there are no machines (as opposed to all of them being
  // locked).
  //
  bool none (true);

  // The first level are machine volumes.
  //
  for (const dir_entry& ve: dir_iterator (machines,
                                          false /* ignore_dangling */))
  {
    const string vn (ve.path ().string ());

    // Ignore hidden directories.
    //
    if (ve.type () != entry_type::directory || vn[0] == '.')
      continue;

    const dir_path vd (dir_path (machines) /= vn);

    // Inside we have machines.
    //
    try
    {
      for (const dir_entry& me: dir_iterator (vd, false /* ignore_dangling */))
      {
        const string mn (me.path ().string ());

        if (me.type () != entry_type::directory || mn[0] == '.')
          continue;

        const dir_path md (dir_path (vd) /= mn);

        // Our endgoal here is to obtain a bootstrapped snapshot of this
        // machine while watching out for potential race conditions (other
        // instances as well as machines being added/upgraded/removed; see the
        // manual for details).
        //
        // So here is our overall plan:
        //
        // 1. Resolve current subvolume link for our bootstrap protocol.
        //
        // 2. Lock the machine. This excludes any other instance from trying
        //    to perform the following steps.
        //
        // 3. If there is no link, cleanup old bootstrap (if any) and ignore
        //    this machine.
        //
        // 4. Try to create a snapshot of current subvolume (this operation is
        //    atomic). If failed (e.g., someone changed the link and removed
        //    the subvolume in the meantime), retry from #1.
        //
        // 5. Compare the snapshot to the already bootstrapped version (if
        //    any) and see if we need to re-bootstrap. If so, use the snapshot
        //    as a starting point. Rename to bootstrapped at the end (atomic).
        //
        dir_path lp (dir_path (md) /= (mn + '-' + bs_prot)); // -<P>
        dir_path tp (dir_path (md) /= (mn + '-' + tc_name)); // -<toolchain>

        auto delete_bootstrapped = [&tp, &trace] () // Delete -<toolchain>.
        {
          run_btrfs (trace, "property", "set", "-ts", tp, "ro", "false");
          run_btrfs (trace, "subvolume", "delete", tp);
        };

        for (size_t retry (0);; ++retry)
        {
          if (retry != 0)
            sleep (1);

          // Resolve the link to subvolume path.
          //
          dir_path sp; // <name>-<P>.<R>

          try
          {
            sp = path_cast<dir_path> (readsymlink (lp));

            if (sp.relative ())
              sp = md / sp;
          }
          catch (const system_error& e)
          {
            // Leave the subvolume path empty if the subvolume link doesn't
            // exist and fail on any other error.
            //
            if (e.code ().category () != std::generic_category () ||
                e.code ().value () != ENOENT)
              fail << "unable to read subvolume link " << lp << ": " << e;
          }

          none = none && sp.empty ();

          // Try to lock the machine, skipping it if already locked.
          //
          optional<machine_lock> ml (lock_machine (tp));

          if (!ml)
          {
            l4 ([&]{trace << "skipping " << md << ": locked";});
            break;
          }

          bool te (dir_exists (tp));

          // If the resolution fails, then this means there is no current
          // machine subvolume (for this bootstrap protocol). In this case we
          // clean up our toolchain subvolume (-<toolchain>, if any) and
          // ignore this machine.
          //
          if (sp.empty ())
          {
            if (te)
              delete_bootstrapped ();

            l3 ([&]{trace << "skipping " << md << ": no subvolume link";});
            break;
          }

          // <name>-<toolchain>-<xxx>
          //
          const dir_path xp (snapshot_path (tp));

          if (btrfs_exit (trace, "subvolume", "snapshot", sp, xp) != 0)
          {
            if (retry >= 10)
              fail << "unable to snapshot subvolume " << sp;

            continue;
          }

          // Load the (original) machine manifest.
          //
          auto mm (
            parse_manifest<machine_manifest> (sp / "manifest", "machine"));

          // If we already have <name>-<toolchain>, see if it needs to be re-
          // bootstrapped. Things that render it obsolete:
          //
          // 1. New machine revision  (compare machine ids).
          // 2. New toolchain         (compare toolchain ids).
          // 3. New bbot/libbbot      (compare versions).
          //
          // The last case has a complication: what should we do if we have
          // bootstrapped a newer version of bbot? This would mean that we are
          // about to be stopped and upgraded (and the upgraded version will
          // probably be able to use the result). So we simply ignore this
          // machine for this run.

          // Return -1 if older, 0 if the same, and +1 if newer.
          //
          auto compare_bbot = [] (const bootstrap_manifest& m) -> int
          {
            auto cmp = [&m] (const string& n, const char* v) -> int
            {
              standard_version sv (v);
              auto i = m.versions.find (n);

              return (i == m.versions.end () || i->second < sv
                      ? -1
                      : i->second > sv ? 1 : 0);
            };

            // Start from the top assuming a new dependency cannot be added
            // without changing the dependent's version.
            //
            int r;
            return
              (r = cmp ("bbot",       BBOT_VERSION_STR)) != 0 ? r :
              (r = cmp ("libbbot", LIBBBOT_VERSION_STR)) != 0 ? r :
              (r = cmp ("libbpkg", LIBBPKG_VERSION_STR)) != 0 ? r :
              (r = cmp ("libbutl", LIBBUTL_VERSION_STR)) != 0 ? r : 0;
          };

          optional<bootstrapped_machine_manifest> bmm;
          if (te)
          {
            bmm = parse_manifest<bootstrapped_machine_manifest> (
              tp / "manifest", "bootstrapped machine");

            if (bmm->machine.id != mm.id)
            {
              l3 ([&]{trace << "re-bootstrapping " << tp << ": new machine";});
              te = false;
            }

            if (!tc_id.empty () && bmm->toolchain.id != tc_id)
            {
              l3 ([&]{trace << "re-bootstrapping " << tp << ": new toolchain";});
              te = false;
            }

            if (int i = compare_bbot (bmm->bootstrap))
            {
              if (i < 0)
              {
                l3 ([&]{trace << "re-bootstrapping " << tp << ": new bbot";});
                te = false;
              }
              else
              {
                l3 ([&]{trace << "ignoring " << tp << ": old bbot";});
                run_btrfs (trace, "subvolume", "delete", xp);
                break;
              }
            }

            if (!te)
              delete_bootstrapped ();
          }
          else
            l3 ([&]{trace << "bootstrapping " << tp;});

          if (!te)
          {
            // Use the <name>-<toolchain>-<xxx> snapshot that we have made to
            // bootstrap the new machine. Then atomically rename it to
            // <name>-<toolchain>.
            //
            // Also release all the machine locks that we have acquired so far
            // since the bootstrap will take a while and other instances might
            // be able to use them.
            //
            r.clear ();

            bmm = bootstrap_machine (xp, mm, move (bmm));

            if (!bmm)
            {
              l3 ([&]{trace << "ignoring " << tp << ": failed to bootstrap";});
              run_btrfs (trace, "subvolume", "delete", xp);
              break;
            }

            try
            {
              mvdir (xp, tp);
            }
            catch (const system_error& e)
            {
              fail << "unable to rename " << xp << " to " << tp;
            }

            l2 ([&]{trace << "bootstrapped " << bmm->machine.name;});

            // Check the bootstrapped bbot version as above and ignore this
            // machine if it's newer than us.
            //
            if (int i = compare_bbot (bmm->bootstrap))
            {
              if (i > 0)
              {
                l3 ([&]{trace << "ignoring " << tp << ": old bbot";});
                break;
              }
              else
                warn << "bootstrapped " << tp << " bbot worker is older "
                     << "than agent; assuming test setup";
            }
          }
          else
            run_btrfs (trace, "subvolume", "delete", xp);

          // Add the machine to the lists.
          //
          r.push_back (
            bootstrapped_machine {move (tp), move (*bmm), move (*ml)});

          break;
        }
      }
    }
    catch (const system_error& e)
    {
      fail << "unable to iterate over " << vd << ": " << e;
    }
  }

  if (none)
    warn << "no build machines for toolchain " << tc_name;

  return r;
}
catch (const system_error& e)
{
  fail << "unable to iterate over " << machines << ": " << e << endf;
}

static result_manifest
perform_task (const dir_path& md,
              const bootstrapped_machine_manifest& mm,
              const task_manifest& tm)
try
{
  tracer trace ("perform_task", md.string ().c_str ());

  result_manifest r {
    tm.name,
    tm.version,
    result_status::abort,
    operation_results {}};

  if (ops.fake_build ())
    return r;

  // The overall plan is as follows:
  //
  // 1. Snapshot the (bootstrapped) machine.
  //
  // 2. Save the task manifest to the TFTP directory (to be accessed by the
  //    worker).
  //
  // 3. Start the TFTP server and the machine.
  //
  // 4. Serve TFTP requests while watching out for the result manifest.
  //
  // 5. Clean up (force the machine down and delete the snapshot).
  //

  // TFTP server mapping (server chroot is --tftp):
  //
  // GET requests to .../build/<name>-<instance>/get/*
  // PUT requests to .../build/<name>-<instance>/put/*
  //
  const string in_name (tc_name + '-' + to_string (inst));
  auto_rmdir arm ((dir_path (ops.tftp ()) /= "build") /= in_name);

  dir_path gd (dir_path (arm.path) /= "get");
  dir_path pd (dir_path (arm.path) /= "put");

  try_mkdir_p (gd);
  try_mkdir_p (pd);

  path tf (gd / "task.manifest");   // Task manifest file.
  path rf (pd / "result.manifest"); // Result manifest file.

  serialize_manifest (tm, tf, "task");

  if (ops.fake_machine_specified ())
  {
    // Simply wait for the file to appear.
    //
    for (size_t i (0);; sleep (1))
    {
      if (file_not_empty (rf))
      {
        // Wait a bit to make sure we see complete manifest.
        //
        sleep (2);
        break;
      }

      if (i++ % 10 == 0)
        l3 ([&]{trace << "waiting for result manifest";});
    }

    r = parse_manifest<result_manifest> (rf, "result");
  }
  else
  {
    try_rmfile (rf);

    // <name>-<toolchain>-<xxx>
    //
    const dir_path xp (snapshot_path (md));

    for (size_t retry (0);; ++retry)
    {
      if (retry != 0)
        run_btrfs (trace, "subvolume", "delete", xp);

      run_btrfs (trace, "subvolume", "snapshot", md, xp);

      // Start the TFTP server.
      //
      tftp_server tftpd ("Gr  ^/?(.+)$  /build/" + in_name + "/get/\\1\n" +
                         "Pr  ^/?(.+)$  /build/" + in_name + "/put/\\1\n",
                         ops.tftp_port () + offset);

      l3 ([&]{trace << "tftp server on port " << tftpd.port ();});

      // Start the machine.
      //
      unique_ptr<machine> m (
        start_machine (xp,
                       mm.machine,
                       mm.machine.mac,
                       ops.bridge (),
                       tftpd.port (),
                       tm.interactive.has_value ()));

      // Note: the machine handling logic is similar to bootstrap.
      //
      {
        auto mg (
          make_exception_guard (
            [&m, &xp] ()
            {
              info << "trying to force machine " << xp << " down";
              try {m->forcedown (false);} catch (const failed&) {}
            }));

        auto soft_fail = [&xp, &m, &r] (const char* msg)
        {
          {
            diag_record dr (error);
            dr << msg << " for machine " << xp << ", suspending";
            m->print_info (dr);
          }

          try
          {
            m->suspend (false);
            m->wait (false);
            m->cleanup ();
            info << "resuming after machine suspension";
          }
          catch (const failed&) {}

          return r;
        };

        auto check_machine = [&xp, &m] ()
        {
          try
          {
            size_t t (0);
            if (!m->wait (t /* seconds */, false /* fail_hard */))
              return true;
          }
          catch (const failed&)
          {
          }

          diag_record dr (warn);
          dr << "machine " << xp << " exited unexpectedly";
          m->print_info (dr);

          return false;
        };

        // The first request should be the task manifest download. Wait for up
        // to 2 minutes for that to arrive (again, that long to deal with
        // flaky Windows networking). In a sense we use it as an indication
        // that the machine has booted and the worker process has started.
        //
        size_t to;
        const size_t startup_to (120);
        const size_t build_to   (tm.interactive
                                 ? ops.intactive_timeout ()
                                 : ops.build_timeout ());

        // Wait periodically making sure the machine is still alive.
        //
        for (to = startup_to; to != 0; )
        {
          if (tftpd.serve (to, 2))
            break;

          if (!check_machine ())
            return r;
        }

        if (to == 0)
        {
          if (retry > ops.build_retries ())
            return soft_fail ("build startup timeout");

          // Note: keeping the logs behind (no cleanup).

          diag_record dr (warn);
          dr << "machine " << mm.machine.name << " mis-booted, retrying";
          m->print_info (dr);

          try {m->forcedown (false);} catch (const failed&) {}
          continue;
        }

        l3 ([&]{trace << "completed startup in " << startup_to - to << "s";});

        // Next the worker builds things and then uploads the result manifest.
        // So on our side we serve TFTP requests while checking for the
        // manifest file. To workaround some obscure filesystem races (the
        // file's mtime/size is updated several seconds later; maybe tmpfs
        // issue?), we periodically re-check.
        //
        for (to = build_to; to != 0; )
        {
          if (tftpd.serve (to, 2))
            continue;

          if (!check_machine ())
          {
            if (!file_not_empty (rf))
              return r;
          }

          if (file_not_empty (rf))
          {
            if (!tftpd.serve (to, 5))
              break;
          }
        }

        if (to == 0)
          return soft_fail ("build timeout");

        l3 ([&]{trace << "completed build in " << build_to - to << "s";});

        // Parse the result manifest.
        //
        try
        {
          r = parse_manifest<result_manifest> (rf, "result", false);
        }
        catch (const failed&)
        {
          r.status = result_status::abnormal; // Soft-fail below.
        }

        if (r.status == result_status::abnormal)
        {
          // If the build terminated abnormally, suspend the machine for
          // investigation.
          //
          return soft_fail ("build terminated abnormally");
        }
        else
        {
          // Force the machine down (there is no need wasting time on clean
          // shutdown since the next step is to drop the snapshot). Also fail
          // softly if things go badly.
          //
          // One thing to keep in mind are DHCP leases: with this approach
          // they will not be released. However, since we reuse the same MAC
          // address since bootstrap, on the next build we should get the same
          // lease instead of a new one.
          //
          try {m->forcedown (false);} catch (const failed&) {}
          m->cleanup ();
        }
      }

      run_btrfs (trace, "subvolume", "delete", xp);
      break;
    }
  }

  // Update package name/version if the returned value as "unknown".
  //
  if (r.version == bpkg::version ("0"))
  {
    assert (r.status == result_status::abnormal);

    r.name = tm.name;
    r.version = tm.version;
  }

  return r;
}
catch (const system_error& e)
{
  fail << "build error: " << e << endf;
}

extern "C" void
handle_signal (int sig)
{
  switch (sig)
  {
  case SIGHUP:  exit (3); // Unimplemented feature.
  case SIGTERM: exit (0);
  default:      assert (false);
  }
}

int
main (int argc, char* argv[])
try
{
  cli::argv_scanner scan (argc, argv, true);
  ops.parse (scan);

  verb = ops.verbose ();

  // @@ systemd 231 added JOURNAL_STREAM environment variable which allows
  //    detecting if stderr is connected to the journal.
  //
  if (ops.systemd_daemon ())
    systemd_diagnostics (true); // With critical errors.

  tracer trace ("main");

  uid = getuid ();
  uname = getpwuid (uid)->pw_name;

  // Obtain our hostname.
  //
  {
    char buf[HOST_NAME_MAX + 1];

    if (gethostname (buf, sizeof (buf)) == -1)
      fail << "unable to obtain hostname: "
           << system_error (errno, std::generic_category ()); // Sanitize.

    hname = buf;
  }

  // Obtain our IP address as a first discovered non-loopback IPv4 address.
  //
  // Note: Linux-specific implementation.
  //
  {
    ifaddrs* i;
    if (getifaddrs (&i) == -1)
      fail << "unable to obtain IP addresses: "
           << system_error (errno, std::generic_category ()); // Sanitize.

    unique_ptr<ifaddrs, void (*)(ifaddrs*)> deleter (i, freeifaddrs);

    for (; i != nullptr; i = i->ifa_next)
    {
      sockaddr* sa (i->ifa_addr);

      if (sa != nullptr                      && // Configured.
          (i->ifa_flags & IFF_LOOPBACK) == 0 && // Not a loopback interface.
          (i->ifa_flags & IFF_UP) != 0       && // Up.
          sa->sa_family == AF_INET)             // Ignore IPv6 for now.
      {
        char buf[INET_ADDRSTRLEN]; // IPv4 address.
        if (inet_ntop (AF_INET,
                       &reinterpret_cast<sockaddr_in*> (sa)->sin_addr,
                       buf,
                       sizeof (buf)) == nullptr)
          fail << "unable to obtain IPv4 address: "
               << system_error (errno, std::generic_category ()); // Sanitize.

        hip = buf;
        break;
      }
    }

    if (hip.empty ())
      fail << "no IPv4 address configured";
  }

  // On POSIX ignore SIGPIPE which is signaled to a pipe-writing process if
  // the pipe reading end is closed. Note that by default this signal
  // terminates a process. Also note that there is no way to disable this
  // behavior on a file descriptor basis or for the write() function call.
  //
  if (signal (SIGPIPE, SIG_IGN) == SIG_ERR)
    fail << "unable to ignore broken pipe (SIGPIPE) signal: "
         << system_error (errno, std::generic_category ()); // Sanitize.

  // Version.
  //
  if (ops.version ())
  {
    cout << "bbot-agent " << BBOT_VERSION_ID << endl
         << "libbbot " << LIBBBOT_VERSION_ID << endl
         << "libbpkg " << LIBBPKG_VERSION_ID << endl
         << "libbutl " << LIBBUTL_VERSION_ID << endl
         << "Copyright (c) " << BBOT_COPYRIGHT << "." << endl
         << "TBC; All rights reserved" << endl;

    return 0;
  }

  // Help.
  //
  if (ops.help ())
  {
    pager p ("bbot-agent help", false);
    print_bbot_agent_usage (p.stream ());

    // If the pager failed, assume it has issued some diagnostics.
    //
    return p.wait () ? 0 : 1;
  }

  tc_name = ops.toolchain_name ();
  tc_num  = ops.toolchain_num ();
  tc_ver  = (ops.toolchain_ver_specified ()
             ? ops.toolchain_ver ()
             : standard_version (BBOT_VERSION_STR));
  tc_id   = ops.toolchain_id ();

  if (tc_num == 0 || tc_num > 99)
    fail << "invalid --toolchain-num value " << tc_num;

  inst = ops.instance ();

  if (inst == 0 || inst > 99)
    fail << "invalid --instance value " << inst;

  offset = (tc_num - 1) * 100 + inst;

  // Controller URLs.
  //
  if (argc < 2 &&
      !ops.dump_machines () &&
      !ops.fake_request_specified ())
  {
    fail << "controller url expected" <<
      info << "run " << argv[0] << " --help for details";
  }

  strings controllers;

  for (int i (1); i != argc; ++i)
    controllers.push_back (argv[i]);

  // Handle SIGHUP and SIGTERM.
  //
  if (signal (SIGHUP,  &handle_signal) == SIG_ERR ||
      signal (SIGTERM, &handle_signal) == SIG_ERR)
    fail << "unable to set signal handler: "
         << system_error (errno, std::generic_category ()); // Sanitize.

  optional<string> fingerprint;

  if (ops.auth_key_specified ())
  try
  {
    // Note that the process always prints to STDERR, so we redirect it to the
    // null device. We also check for the key file existence to print more
    // meaningful error message if that's not the case.
    //
    if (!file_exists (ops.auth_key ()))
      throw_generic_error (ENOENT);

    openssl os (trace,
                ops.auth_key (), path ("-"), fdopen_null (),
                ops.openssl (), "rsa",
                ops.openssl_option (), "-pubout", "-outform", "DER");

    fingerprint = sha256 (os.in).string ();
    os.in.close ();

    if (!os.wait ())
      throw_generic_error (EIO);
  }
  catch (const system_error& e)
  {
    fail << "unable to obtain authentication public key: " << e;
  }

  if (ops.systemd_daemon ())
  {
    diag_record dr;

    dr << info << "bbot agent " << BBOT_VERSION_ID;

    dr <<
      info << "cpu(s)         " << ops.cpu () <<
      info << "ram(kB)        " << ops.ram () <<
      info << "bridge         " << ops.bridge ();

    if (fingerprint)
      dr << info << "auth key fp    " << *fingerprint;

    dr <<
      info << "toolchain name " << tc_name <<
      info << "toolchain num  " << tc_num <<
      info << "toolchain ver  " << tc_ver.string () <<
      info << "toolchain id   " << tc_id <<
      info << "instance  num  " << inst;

    for (const string& u: controllers)
      dr << info << "controller url " << u;
  }

  // The work loop. The steps we go through are:
  //
  // 1. Enumerate the available machines, (re-)bootstrapping any if necessary.
  //
  // 2. Poll controller(s) for build tasks.
  //
  // 3. If no build tasks are available, go to #1 (after sleeping a bit).
  //
  // 4. If a build task is returned, do it, upload the result, and go to #1
  //    (immediately).
  //
  auto rand_sleep = [g = std::mt19937 (std::random_device {} ())] () mutable
  {
    return std::uniform_int_distribution<unsigned int> (50, 60) (g);
  };

  optional<interactive_mode> imode;
  optional<string>           ilogin;

  if (ops.interactive () != interactive_mode::false_)
  {
    imode  = ops.interactive ();
    ilogin = machine_vnc (true /* public */);
  }

  for (unsigned int sleep (0);; ::sleep (sleep), sleep = 0)
  {
    bootstrapped_machines ms (enumerate_machines (ops.machines ()));

    // Prepare task request.
    //
    task_request_manifest tq {
      hname,
      tc_name,
      tc_ver,
      imode,
      ilogin,
      fingerprint,
      machine_header_manifests {}
    };

    // Note: below we assume tq.size () == ms.size ().
    //
    for (const bootstrapped_machine& m: ms)
      tq.machines.emplace_back (m.manifest.machine.id,
                                m.manifest.machine.name,
                                m.manifest.machine.summary);

    if (ops.dump_machines ())
    {
      for (const machine_header_manifest& m: tq.machines)
        serialize_manifest (m, cout, "stdout", "machine");

      return 0;
    }

    if (tq.machines.empty ())
    {
      // Normally this means all the machines are locked so sleep a bit less.
      //
      sleep = rand_sleep () / 2;
      continue;
    }

    // Send task requests.
    //
    // Note that we have to do it while holding the lock on all the machines
    // since we don't know which machine we will need.
    //
    string url;
    task_response_manifest tr;

    if (ops.fake_request_specified ())
    {
      auto t (parse_manifest<task_manifest> (ops.fake_request (), "task"));

      tr = task_response_manifest {
        "fake-session", // Dummy session.
        nullopt,        // No challenge.
        url,            // Empty result URL.
        move (t)};

      url = "http://example.org";
    }
    else
    {
      // Note that after completing each task we always start from the
      // beginning of the list. This fact can be used to implement a poor
      // man's priority system where we will continue serving the first listed
      // controller for as long as it has tasks (and maybe in the future we
      // will implement a proper priority system).
      //
      for (const string& u: controllers)
      {
        task_response_manifest r;

        try
        {
          http_curl c (trace,
                       path ("-"),
                       path ("-"),
                       curl::post,
                       u,
                       "--header", "Content-Type: text/manifest",
                       "--retry", ops.request_retries (),
                       "--retry-max-time", ops.request_timeout (),
                       "--max-time", ops.request_timeout (),
                       "--connect-timeout", ops.connect_timeout ());

          // This is tricky/hairy: we may fail hard parsing the output before
          // seeing that curl exited with an error and failing softly.
          //
          bool f (false);

          try
          {
            serialize_manifest (tq, c.out, u, "task request", false);
          }
          catch (const failed&) {f = true;}

          c.out.close ();

          if (!f)
          try
          {
            r = parse_manifest<task_response_manifest> (
              c.in, u, "task response", false);
          }
          catch (const failed&) {f = true;}

          c.in.close ();

          if (!c.wait () || f)
            throw_generic_error (EIO);
        }
        catch (const system_error& e)
        {
          error << "unable to request task from " << u << ": " << e;
          continue;
        }

        if (r.challenge && !fingerprint) // Controller misbehaves.
        {
          error << "unexpected challenge from " << u << ": " << *r.challenge;
          continue;
        }

        if (!r.session.empty ()) // Got a task.
        {
          const task_manifest& t (*r.task);

          // For security reasons let's require the repository location to be
          // remote.
          //
          if (t.repository.local ())
          {
            error << "local repository from " << u << ": " << t.repository;
            continue;
          }

          // Make sure that the task interactivity matches the requested mode.
          //
          if (( t.interactive && !imode) ||
              (!t.interactive && imode && *imode == interactive_mode::true_))
          {
            if (t.interactive)
              error << "interactive task from " << u << ": " << *t.interactive;
            else
              error << "non-interactive task from " << u;

            continue;
          }

          l2 ([&]{trace << "task for " << t.name << '/' << t.version << " "
                        << "on " << t.machine << " "
                        << "from " << u;});

          tr  = move (r);
          url = u;
          break;
        }
      }
    }

    if (tr.session.empty ()) // No task from any of the controllers.
    {
      l2 ([&]{trace << "no tasks from any controllers, sleeping";});
      sleep = rand_sleep ();
      continue;
    }

    // We have a build task.
    //
    // First find the index of the machine we were asked to use (and verify it
    // is one of those we sent). Also unlock all the other machines.
    //
    size_t i (ms.size ());
    for (size_t j (0); j != ms.size (); ++j)
    {
      if (tq.machines[j].name == tr.task->machine)
        i = j;
      else
        ms[j].lock.unlock ();
    }

    if (i == ms.size ())
    {
      error << "task from " << url << " for unknown machine "
            << tr.task->machine;

      if (ops.dump_task ())
        return 0;

      continue;
    }

    task_manifest& t (*tr.task);

    if (ops.dump_task ())
    {
      serialize_manifest (t, cout, "stdout", "task");
      return 0;
    }

    // If we have our own repository certificate fingerprints, then use them
    // to replace what we have received from the controller.
    //
    if (!ops.trust ().empty ())
      t.trust = ops.trust ();

    const dir_path& d (); // The -<toolchain> directory.
    const bootstrapped_machine_manifest& m ();

    result_manifest r (perform_task (ms[i].path, ms[i].manifest, t));

    ms[i].lock.unlock (); // No need to hold the lock any longer.

    if (ops.dump_result ())
    {
      serialize_manifest (r, cout, "stdout", "result");
      return 0;
    }

    // Prepare the answer to the private key challenge.
    //
    optional<vector<char>> challenge;

    if (tr.challenge)
    try
    {
      assert (ops.auth_key_specified ());

      openssl os (trace,
                  fdstream_mode::text, path ("-"), 2,
                  ops.openssl (), "rsautl",
                  ops.openssl_option (), "-sign", "-inkey", ops.auth_key ());

      os.out << *tr.challenge;
      os.out.close ();

      challenge = os.in.read_binary ();
      os.in.close ();

      if (!os.wait ())
        throw_generic_error (EIO);
    }
    catch (const system_error& e)
    {
      // The task response challenge is valid (verified by manifest parser),
      // so there must be something wrong with the setup and the failure is
      // fatal.
      //
      fail << "unable to sign task response challenge: " << e;
    }

    // Upload the result.
    //
    result_request_manifest rq {tr.session, move (challenge), move (r)};
    {
      const string& u (*tr.result_url);

      try
      {
        http_curl c (trace,
                     path ("-"),
                     nullfd,     // Not expecting any data in response.
                     curl::post,
                     u,
                     "--header", "Content-Type: text/manifest",
                     "--retry", ops.request_retries (),
                     "--retry-max-time", ops.request_timeout (),
                     "--max-time", ops.request_timeout (),
                     "--connect-timeout", ops.connect_timeout ());

        // This is tricky/hairy: we may fail hard writing the input before
        // seeing that curl exited with an error and failing softly.
        //
        bool f (false);

        try
        {
          serialize_manifest (rq, c.out, u, "task request");
        }
        catch (const failed&) {f = true;}

        c.out.close ();

        if (!c.wait () || f)
          throw_generic_error (EIO);
      }
      catch (const system_error& e)
      {
        error << "unable to upload result to " << u << ": " << e;
        continue;
      }
    }

    l2 ([&]{trace << "built " << t.name << '/' << t.version << " "
                  << "on " << t.machine << " "
                  << "for " << url;});
  }
}
catch (const failed&)
{
  return 1; // Diagnostics has already been issued.
}
catch (const cli::exception& e)
{
  error << e;
  return 1;
}

namespace bbot
{
  static unsigned int rand_seed; // Seed for rand_r();

  size_t
  genrand ()
  {
    if (rand_seed == 0)
      rand_seed = static_cast<unsigned int> (
        std::chrono::system_clock::now ().time_since_epoch ().count ());

    return static_cast<size_t> (rand_r (&rand_seed));
  }

  // Note: Linux-specific implementation.
  //
  string
  iface_addr (const string& i)
  {
    if (i.size () >= IFNAMSIZ)
      throw invalid_argument ("interface name too long");

    auto_fd fd (socket (AF_INET, SOCK_DGRAM | SOCK_CLOEXEC, 0));

    if (fd.get () == -1)
      throw_system_error (errno);

    ifreq ifr;
    ifr.ifr_addr.sa_family = AF_INET;
    strcpy (ifr.ifr_name, i.c_str ());

    if (ioctl (fd.get (), SIOCGIFADDR, &ifr) == -1)
      throw_system_error (errno);

    char buf[INET_ADDRSTRLEN]; // IPv4 address.
    if (inet_ntop (AF_INET,
                   &reinterpret_cast<sockaddr_in*> (&ifr.ifr_addr)->sin_addr,
                   buf,
                   sizeof (buf)) == nullptr)
      throw_system_error (errno);

    return buf;
  }
}