// file      : bbot/agent/agent.cxx -*- C++ -*-
// license   : MIT; 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(), strto[u]ll()
#include <string.h>    // strchr()
#include <unistd.h>    // sleep(), getpid(), getuid(), fsync(), [f]stat()
#include <ifaddrs.h>   // getifaddrs(), freeifaddrs()
#include <sys/types.h> // stat, pid_t
#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 <map>
#include <atomic>
#include <chrono>
#include <random>
#include <iomanip>      // setw()
#include <iostream>
#include <system_error> // generic_category()

#include <libbutl/pager.hxx>
#include <libbutl/sha256.hxx>
#include <libbutl/openssl.hxx>
#include <libbutl/filesystem.hxx> // dir_iterator, try_rmfile(), readsymlink()
#include <libbutl/semantic-version.hxx>

#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;

// According to the standard, atomic's use in the signal handler is only safe
// if it's lock-free.
//
#if !defined(ATOMIC_INT_LOCK_FREE) || ATOMIC_INT_LOCK_FREE != 2
#error int is not lock-free on this architecture
#endif

// While we can use memory_order_relaxed in a single-threaded program, let's
// use consume/release in case this process becomes multi-threaded in the
// future.
//
static std::atomic<unsigned int> sigurs1;

using std::memory_order_consume;
using std::memory_order_release;

extern "C" void
handle_signal (int sig)
{
  switch (sig)
  {
  case SIGHUP:  exit (3); // Unimplemented feature.
  case SIGTERM: exit (0);
  case SIGUSR1: sigurs1.fetch_add (1, std::memory_order_release); break;
  default:      assert (false);
  }
}

namespace bbot
{
  agent_options ops;

  const string bs_prot ("1");

  string           tc_name;
  uint16_t         tc_num;
  path             tc_lock; // Empty if no locking.
  standard_version tc_ver;
  string           tc_id;

  uint16_t inst;     // 1-based.
  uint16_t inst_max; // 0 if priority monitoring is disabled.

  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";

            // Note: snapshot cleaned up by the caller of bootstrap_machine().
          }
          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 ())
          {
            // Note: snapshot cleaned up by the caller of bootstrap_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)))
            {
              // Note: snapshot cleaned up by the caller of bootstrap_machine().
              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;
}

// Global toolchain lock.
//
// The overall locking protocol is as follows:
//
// 1. Before enumerating the machines each agent instance acquires the global
//    toolchain lock.
//
// 2. As the agent enumerates over the machines, it tries to acquire the lock
//    for each machine.
//
// 3. If the agent encounters a machine that it needs to bootstrap, it
//    releases all the other machine locks followed by the global lock,
//    proceeds to bootstrap the machine, releases its lock, and restarts the
//    process from scratch.
//
// 4. Otherwise, upon receiving a task response for one of the machines, the
//    agent releases all the other machine locks followed by the global lock,
//    proceeds to perform the task on the selected machine, releases its lock,
//    and restarts the process from scratch.
//
// One notable implication of this protocol is that the machine locks are
// only acquired while holding the global toolchain lock but can be released
// while not holding this lock.
//
// (Note that because of this implication it can theoretically be possible
// to omit acquiring all the machine locks during the enumeration process,
// instead only acquiring the lock of the machine we need to bootstrap or
// build. However, the current approach is simpler since we still need
// to detect machines that are already locked, which entails acquiring
// the lock anyway.)
//
// Note that unlike the machine lock below, here we don't bother with removing
// the lock file.
//
class toolchain_lock
{
public:
  toolchain_lock () = default; // Empty lock.

  // Note: returns true if locking is disabled.
  //
  bool
  locked () const
  {
    return tc_lock.empty () || fl_;
  }

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

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

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

  toolchain_lock            (toolchain_lock&&) = default;
  toolchain_lock& operator= (toolchain_lock&&) = default;

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

  // Implementation details.
  //
public:
  explicit
  toolchain_lock (auto_fd&& fd)
      : fd_ (move (fd)), fl_ (true) {}

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

// Note: returns empty lock if toolchain locking is disabled.
//
static optional<toolchain_lock>
lock_toolchain (unsigned int timeout)
{
  if (tc_lock.empty ())
    return toolchain_lock ();

  auto_fd fd (fdopen (tc_lock, fdopen_mode::out | fdopen_mode::create));

  for (; flock (fd.get (), LOCK_EX | LOCK_NB) != 0; sleep (1), --timeout)
  {
    if (errno != EWOULDBLOCK)
      throw_generic_error (errno);

    if (timeout == 0)
      return nullopt;
  }

  return toolchain_lock (move (fd));
}

// Per-toolchain machine lock.
//
// 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:
  // A lock is either locked by this process or it contains information about
  // the process holding the lock.
  //
  pid_t              pid;  // Process using the machine.
  optional<uint64_t> prio; // Task priority (absent means being bootstrapped
                           // or have been suspended).

  machine_lock () = default; // Uninitialized lock.

  bool
  locked () const
  {
    return fl_;
  }

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

      if (fd_ != nullfd)
      {
        try_rmfile (fp_, ignore_errors);

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

  // Write the holding process information to the lock file.
  //
  // Must be called while holding the toolchain lock (see the lock_machine()
  // implementation for rationale).
  //
  void
  bootstrap (const toolchain_lock& tl)
  {
    assert (tl.locked () && fl_);

    if (fd_ != nullfd)
      write (nullopt);
  }

  void
  perform_task (const toolchain_lock& tl, uint64_t prio)
  {
    assert (tl.locked () && fl_);

    if (fd_ != nullfd)
      write (prio);
  }

  // Truncate the holding process information after the call to perform_task()
  // so that it doesn't contain the priority, marking the machine as being
  // suspended.
  //
  // Note that this one can be called without holding the toolchain lock.
  //
  void
  suspend_task ()
  {
    assert (fl_);

    if (fd_ != nullfd)
    {
      assert (tp_ != 0); // Must be called after perform_task().

      // While there is no direct statement to this effect in POSIX, the
      // consensus on the internet is that truncation is atomic, in a sense
      // that the reader shouldn't see a partially truncated content. Feels
      // like should be doubly so when actually truncating as opposed to
      // extending the size, which is what we do.
      //
      fdtruncate (fd_.get (), tp_);
    }
  }

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

  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:
  // If fd is nullfd, treat it as a fake lock (used for fake machines).
  //
  machine_lock (path&& fp, auto_fd&& fd)
      : fp_ (move (fp)), fd_ (move (fd)), fl_ (true) {}

  machine_lock (pid_t pi, optional<uint64_t> pr)
      : pid (pi), prio (pr), fl_ (false) {}

private:
  void
  write (optional<uint64_t> prio)
  {
    pid_t pid (getpid ());

    string l (to_string (pid));

    if (prio)
    {
      tp_ = l.size (); // Truncate position.

      l += ' ';
      l += to_string (*prio);
    }

    auto n (fdwrite (fd_.get (), l.c_str (), l.size ()));

    if (n == -1)
      throw_generic_ios_failure (errno);

    if (static_cast<size_t> (n) != l.size ())
      throw_generic_ios_failure (EFBIG);
  }

private:
  path     fp_;
  auto_fd  fd_;
  bool     fl_ = false;
  uint64_t tp_ = 0; // Truncate position.
};

// Try to lock the machine given its -<toolchain> directory. Return unlocked
// lock with pid/prio if already in use. Must be called while holding the
// toolchain lock.
//
static machine_lock
lock_machine (const toolchain_lock& tl, const dir_path& tp)
{
  assert (tl.locked ());

  path fp (tp + ".lock"); // The -<toolchain>.lock file.

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

    if (flock (fd.get (), LOCK_EX | LOCK_NB) != 0)
    {
      if (errno == EWOULDBLOCK)
      {
        // The file should contain a line in the following format:
        //
        // <pid>[ <prio>]
        //
        char buf[64]; // Sufficient for 2 64-bit numbers (20 decimals max).

        auto sn (fdread (fd.get (), buf, sizeof (buf)));

        if (sn == -1)
          throw_generic_ios_failure (errno);

        size_t n (static_cast<size_t> (sn));

        // While there would be a race between locking the file then writing
        // to it in one process and reading from it in another process, we are
        // protected by the global toolchain lock, which must be held by both
        // sides during this dance.
        //
        assert (n > 0 && n < sizeof (buf));
        buf[n] = '\0';

        // Note also that it's possible that by the time we read the pid/prio
        // the lock has already been released. But this case is no different
        // from the lock being released after we have read pid/prio but before
        // acting on this information (e.g., trying to interrupt the other
        // process), which we have to deal with anyway.
        //
        pid_t pid;
        optional<uint64_t> prio;
        {
          char* p (strchr (buf, ' '));
          char* e;

          {
            errno = 0;
            pid = strtoll (buf, &e, 10); // Note: pid_t is signed.
            assert (errno != ERANGE &&
                    e != buf        &&
                    (p != nullptr ? e == p : *e == '\0'));
          }

          if (p != nullptr)
          {
            ++p;
            errno = 0;
            prio = strtoull (p, &e, 10);
            assert (errno != ERANGE && e != p && *e == '\0');
          }
        }

        return machine_lock (pid, prio);
      }

      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));

    // Retry (note: lock is unlocked by auto_fd::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 the global toolchain lock and the list of available machines,
// (re-)bootstrapping them if necessary.
//
// Note that this function returns both machines that this process managed to
// lock as well as the machines locked by other processes (including those
// that are being bootstrapped or that have been suspended), in case the
// caller needs to interrupt one of them for a higher-priority task. In the
// latter case, the manifest is empty if the machine is bootstrapping or
// suspended and only has the machine_manifest information otherwise. (The
// bootstrapping/suspended machines have to be returned to get the correct
// count of currently active instances for the inst_max comparison.)
//
struct bootstrapped_machine
{
  dir_path                      path;
  machine_lock                  lock;
  bootstrapped_machine_manifest manifest;
};
using bootstrapped_machines = vector<bootstrapped_machine>;

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

  for (;;) // From-scratch retry loop for after bootstrap (see below).
  {
    pair<toolchain_lock, bootstrapped_machines> pr;

    {
      optional<toolchain_lock> l;
      while (!(l = lock_toolchain (60 /* seconds */)))
      {
        warn << "unable to acquire global toolchain lock " << tc_lock
             << " for 60s";
      }
      pr.first = move (*l);
    }

    toolchain_lock& tl (pr.first);
    bootstrapped_machines& r (pr.second);

    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.
          machine_lock (path (), nullfd),        // Fake lock.
          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 {}}});

      return pr;
    }

    // Compare bbot and library versions returning -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);
    };

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

    // We used to (re)-bootstrap machines as we are iterating. But with the
    // introduction of the priority monitoring functionality we need to
    // respect the --instance-max value. Which means we first need to try to
    // lock all the machines in order to determine how many of them are busy
    // then check this count against --instance-max, and only bootstrap if we
    // are not over the limit. Which means we have to store all the
    // information about a (first) machine that needs bootstrapping until
    // after we have enumerated all of them.
    //
    struct pending_bootstrap
    {
      machine_lock ml;
      dir_path tp; // -<toolchain>
      dir_path xp; // -<toolchain>-<xxx>
      machine_manifest mm;
      optional<bootstrapped_machine_manifest> bmm;
    };
    optional<pending_bootstrap> pboot;

    // The first level are machine volumes.
    //
    for (const dir_entry& ve: dir_iterator (machines, dir_iterator::no_follow))
    {
      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, dir_iterator::no_follow))
        {
          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.
            //
            machine_lock ml (lock_machine (tl, tp));

            if (!ml.locked ())
            {
              // @@ TMP: restore l4 tracing.

              machine_manifest mm;
              if (ml.prio)
              {
                // Get the machine manifest (subset of the steps performed for
                // the locked case below).
                //
                // Note that it's possible the machine we get is not what was
                // originally locked by the other process (e.g., it has been
                // upgraded since). It's also possible that if and when we
                // interrupt and lock this machine, it will be a different
                // machine (e.g., it has been upgraded since we read this
                // machine manifest). To deal with all of that we will be
                // reloading this information if/when we acquire the lock to
                // this machine.
                //
                if (sp.empty ())
                {
                  l3 ([&]{trace << "skipping " << md << ": no subvolume link";});
                  break;
                }

                l1 ([&]{trace << "keeping " << md << ": locked by " << ml.pid
                              << " with priority " << *ml.prio;});

                mm = parse_manifest<machine_manifest> (
                  sp / "manifest", "machine");
              }
              else // Bootstrapping/suspended.
              {
                l1 ([&]{trace << "keeping " << md << ": being bootstrapped "
                              << "or suspened by " << ml.pid;});
              }

              // Add the machine to the lists and bail out.
              //
              r.push_back (bootstrapped_machine {
                  move (tp),
                  move (ml),
                  bootstrapped_machine_manifest {move (mm), {}, {}}});

              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>
            //
            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.
            //
            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-bootstrap " << tp << ": new machine";});
                te = false;
              }

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

              if (int i = compare_bbot (bmm->bootstrap))
              {
                if (i < 0)
                {
                  l3 ([&]{trace << "re-bootstrap " << 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 << "bootstrap " << tp;});

            if (!te)
            {
              // Ignore any other machines that need bootstrapping.
              //
              if (!pboot)
              {
                pboot = pending_bootstrap {
                  move (ml), move (tp), move (xp), move (mm), move (bmm)};
              }
              else
                run_btrfs (trace, "subvolume", "delete", xp);

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

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

            break;
          } // Retry loop.
        } // Inner dir_iterator loop.
      }
      catch (const system_error& e)
      {
        fail << "unable to iterate over " << vd << ": " << e;
      }
    } // Outer dir_iterator loop.

    // See if there is a pending bootstrap and whether we can perform it.
    //
    // What should we do if we can't (i.e., we are in the priority minitor
    // mode)? Well, we could have found some machines that are already
    // bootstrapped (busy or not) and there may be a higher-priority task for
    // one of them, so it feels natural to return whatever we've got.
    //
    if (pboot)
    {
      dir_path& tp (pboot->tp);
      dir_path& xp (pboot->xp);

      // Determine how many machines are busy (locked by other processes) and
      // make sure it's below the --instance-max limit, if specified.
      //
      if (inst_max != 0)
      {
        size_t busy (0);
        for (const bootstrapped_machine& m: r)
          if (!m.lock.locked ())
            ++busy;

        assert (busy <= inst_max);

        if (busy == inst_max)
        {
          l1 ([&]{trace << "instance max reached attempting to bootstrap "
                        << tp;});
          run_btrfs (trace, "subvolume", "delete", xp);
          return pr;
        }
      }

      machine_lock& ml (pboot->ml);

      l3 ([&]{trace << "bootstrapping " << tp;});

      // Use the -<toolchain>-<xxx> snapshot that we have made to bootstrap
      // the new machine. Then atomically rename it to -<toolchain>.
      //
      // Also release all the machine locks that we have acquired so far as
      // well as the global toolchain lock, since the bootstrap will take a
      // while and other instances might be able to use them. Because we are
      // releasing the global lock, we have to restart the enumeration process
      // from scratch.
      //
      r.clear ();
      ml.bootstrap (tl);
      tl.unlock ();

      optional<bootstrapped_machine_manifest> bmm (
        bootstrap_machine (xp, pboot->mm, move (pboot->bmm)));

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

      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";});
        else
          warn << "bootstrapped " << tp << " bbot worker is older "
               << "than agent; assuming test setup";
      }

      continue; // Re-enumerate from scratch.
    }

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

    return pr;

  } // From-scratch retry loop.

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

// Perform the build task throwing interrupt if it has been interrupted.
//
struct interrupt {};

static result_manifest
perform_task (toolchain_lock tl, // Note: assumes ownership.
              machine_lock& ml,
              const dir_path& md,
              const bootstrapped_machine_manifest& mm,
              const task_manifest& tm)
try
{
  tracer trace ("perform_task", md.string ().c_str ());

  // Arm the interrupt handler and release the global toolchain lock.
  //
  // Note that there can be no interrupt while we are holding the global lock.
  //
  sigurs1.store (0, std::memory_order_release);
  tl.unlock ();

  result_manifest r {
    tm.name,
    tm.version,
    result_status::abort,
    operation_results {},
    nullopt /* worker_checksum */,
    nullopt /* dependency_checksum */};

  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 and
  //    interrupts.
  //
  // 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.lz4"); // Result manifest file.
  path af (pd / "upload.tar");          // Archive of build artifacts to upload.

  serialize_manifest (tm, tf, "task");

  if (ops.fake_machine_specified ())
  {
    // Note: not handling interrupts here.

    // 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");

    // If archive of build artifacts is present, then just list its content as
    // a sanity check.
    //
    bool err (!r.status);
    if (!err && file_exists (af))
    {
      try
      {
        auto_fd null (fdopen_null ());

        // Redirect stdout to stderr if the command is traced and to /dev/null
        // otherwise.
        //
        process_exit pe (
          process_run_callback (
            trace,
            null.get (), // Don't expect to read from stdin.
            verb >= 3 ? 2 : null.get (),
            2,
            "tar",
            "-tf", af));

        if (!pe)
          fail << "tar " << pe;
      }
      catch (const process_error& e)
      {
        fail << "unable execute tar: " << e;
      }
    }
  }
  else
  {
    try_rmfile (rf);
    try_rmfile (af);
    try_rmdir_r (pd / dir_path ("upload"));

    // <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 ();});

      // Note: the machine handling logic is similar to bootstrap. Except here
      // we have to cleanup the snapshot ourselves in case of suspension or
      // unexpected exit.
      //
      {
        // Start the machine.
        //
        unique_ptr<machine> m (
          start_machine (xp,
                         mm.machine,
                         mm.machine.mac,
                         ops.bridge (),
                         tftpd.port (),
                         tm.interactive.has_value ()));

        auto mg (
          make_exception_guard (
            [&m, &xp] ()
            {
              if (m != nullptr)
              {
                info << "trying to force machine " << xp << " down";
                try {m->forcedown (false);} catch (const failed&) {}
              }
            }));

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

          try
          {
            // Update the information in the machine lock to signal that the
            // machine is suspended and cannot be interrupted.
            //
            ml.suspend_task ();

            m->suspend (false);
            m->wait (false);
            m->cleanup ();
            run_btrfs (trace, "subvolume", "delete", xp);
            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;
        };

        auto check_interrupt = [&trace, &xp, &m] ()
        {
          if (sigurs1.load (std::memory_order_consume) == 0)
            return;

          // @@ l3
          l1 ([&]{trace << "machine " << xp << " interruped";});

          try {m->forcedown (false);} catch (const failed&) {}
          m->cleanup ();
          m = nullptr; // Disable exceptions guard above.
          run_btrfs (trace, "subvolume", "delete", xp);

          throw interrupt ();
        };

        // 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 and
        // checking for interrupts.
        //
        for (to = startup_to; to != 0; )
        {
          check_interrupt ();

          if (tftpd.serve (to, 2))
            break;

          if (!check_machine ())
          {
            run_btrfs (trace, "subvolume", "delete", xp);
            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 optional archive of
        // build artifacts and the result manifest afterwards. 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; )
        {
          check_interrupt ();

          if (tftpd.serve (to, 2))
            continue;

          if (!check_machine ())
          {
            if (!file_not_empty (rf))
            {
              run_btrfs (trace, "subvolume", "delete", xp);
              return r;
            }
          }

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

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

          // Parse the result manifest.
          //
          optional<result_manifest> rm;

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

          // Upload the build artifacts if the result manifest is parsed
          // successfully, the result status is not an error, and upload.tar
          // exists.
          //
          // Note that while the worker doesn't upload the build artifacts
          // archives on errors, there can be the case when the error occurred
          // while uploading the archive and so the partially uploaded file
          // may exist. Thus, we check if the result status is not an error.
          //
          // Note also that we will not bother with interrupting this process
          // assuming it will be quick (relative to the amount of work that
          // would be wasted).
          //
          bool err (!rm || !rm->status);
          if (!err && file_exists (af))
          {
            // Extract the build artifacts from the archive and upload them to
            // the controller. On error keep the result status as abort for
            // transient errors (network failure, etc) and set it to abnormal
            // otherwise (for subsequent machine suspension and
            // investigation).
            //
            optional<bool> err; // True if the error is transient.

            try
            {
              process_exit pe (
                process_run_callback (
                  trace,
                  fdopen_null (), // Don't expect to read from stdin.
                  2,              // Redirect stdout to stderr.
                  2,
                  "tar",
                  "-xf", af,
                  "-C", pd));

              if (!pe)
              {
                err = false;
                error << "tar " << pe;
              }
            }
            catch (const process_error& e)
            {
              err = false;
              error << "unable execute tar: " << e;
            }

            if (!err)
            {
              // @@ Upload the extracted artifacts.
            }

            if (err)
            {
              if (!*err)                            // Non-transient?
                r.status = result_status::abnormal; // Soft-fail below.

              rm = nullopt; // Drop the parsed manifest.
            }
          }

          if (rm)
            r = move (*rm);
        }
        else
        {
          // Suspend the machine for non-interactive builds and fall through
          // to abort for interactive (i.e., "the user went for lunch" case).
          //
          if (!tm.interactive)
            return soft_fail ("build timeout");
        }

        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 is "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;
}

static const string agent_checksum ("2"); // Logic version.

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
         << "This is free software released under the MIT license." << 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 ();

  if (ops.toolchain_lock_specified ())
  {
    const string& l (ops.toolchain_lock ());

    if (!l.empty ())
    {
      tc_lock = path (l);

      if (!tc_lock.absolute ())
        fail << "--toolchain-lock value '" << l << "' should be absolute path";
    }
  }
  else if (!(ops.fake_bootstrap ()         ||
             ops.fake_build ()             ||
             ops.fake_machine_specified () ||
             ops.fake_request_specified ()))
    tc_lock = path ("/var/lock/bbot-agent-" + tc_name + ".lock");

  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;

  inst_max = ops.instance_max ();

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

  // Controller priority to URLs map.
  //
  std::map<uint64_t, strings> controllers;

  for (int i (1); i != argc; ++i)
  {
    // [<prio>=]<url>
    //
    string a (argv[i]);

    // See if we have priority, falling back to priority 0 if absent.
    //
    uint64_t prio (0);

    // Note that we can also have `=` in <url> (e.g., parameters) so we will
    // only consider `=` as ours if prior to it we only have digits.
    //
    size_t p (a.find ('='));
    if (p != string::npos && a.find_first_not_of ("0123456789") == p)
    {
      // Require exactly four or five digits in case we later need to extend
      // the priority levels beyond the 10 possible values (e.g., DDCCBBAA).
      //
      if (p != 4 && p != 5)
        fail << "four or five-digit controller url priority expected in '"
             << a << "'";

      char* e;
      errno = 0;
      prio = strtoull (a.c_str (), &e, 10);
      assert (errno != ERANGE && e == a.c_str () + p);

      if (prio > 19999)
        fail << "out of bounds controller url priority in '" << a << "'";

      a.erase (0, p + 1);
    }

    controllers[prio].push_back (move (a));
  }

  if (controllers.empty ())
  {
    if (ops.dump_machines () || ops.fake_request_specified ())
    {
      controllers[0].push_back ("https://example.org");
    }
    else
      fail << "controller url expected" <<
        info << "run " << argv[0] << " --help for details";
  }

  // Handle SIGHUP and SIGTERM.
  //
  if (signal (SIGHUP,  &handle_signal) == SIG_ERR ||
      signal (SIGTERM, &handle_signal) == SIG_ERR ||
      signal (SIGUSR1, &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 << "interactive    " << to_string (ops.interactive()) <<
      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;

    if (inst_max != 0)
      dr << info << "instance  max  " << inst_max;

    // Note: keep last since don't restore fill/setw.
    //
    for (const pair<const uint64_t, strings>& p: controllers)
    {
      for (const string& u: p.second)
      {
        dr.os.fill ('0');
        dr << info << "controller url " << std::setw (4) << p.first << '=' << 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).
  //
  // NOTE: consider updating agent_checksum if making any logic changes.
  //
  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 */);
  }

  // Use the pkeyutl openssl command for signing the task response challenge
  // if openssl version is greater or equal to 3.0.0 and the rsautl command
  // otherwise.
  //
  // Note that openssl 3.0.0 deprecates rsautl in favor of pkeyutl.
  //
  const char* sign_cmd;

  try
  {
    optional<openssl_info> oi (openssl::info (trace, 2, ops.openssl ()));

    sign_cmd = oi                    &&
               oi->name == "OpenSSL" &&
               oi->version >= semantic_version {3, 0, 0}
               ? "pkeyutl"
               : "rsautl";
  }
  catch (const system_error& e)
  {
    fail << "unable to obtain openssl version: " << e << endf;
  }

  for (unsigned int sleep (0);; ::sleep (sleep), sleep = 0)
  {
    pair<toolchain_lock, bootstrapped_machines> er (
      enumerate_machines (ops.machines ()));

    toolchain_lock& tl (er.first);
    bootstrapped_machines& ms (er.second);

    // Determine if we should operate in the priority monitor mode and, if so,
    // the lower bound on the priorities that we should consider.
    //
    optional<uint64_t> prio_mon;
    if (inst_max != 0)
    {
      uint16_t           busy (0); // Machines locked by other processes.
      optional<uint64_t> prio;

      for (const bootstrapped_machine& m: ms)
      {
        if (!m.lock.locked ())
        {
          ++busy;
          if (m.lock.prio && (!prio || *m.lock.prio < *prio))
            prio = *m.lock.prio;
        }
      }

      assert (busy <= inst_max);

      if (busy == inst_max)
      {
        if (!prio) // All bootstrapping/suspended.
        {
          sleep = rand_sleep ();
          continue;
        }

        prio_mon = *prio;
      }
    }

    // @@ For now bail out if in the priority monitor mode.
    //
    if (prio_mon)
    {
      l1 ([&]{trace << "priority monitor, lower bound " << *prio_mon;});

      sleep = rand_sleep () / 2;
      continue;
    }

    // If we get a task, these contain all the corresponding information.
    //
    task_request_manifest tq;
    task_response_manifest tr;
    uint64_t prio;
    string url;

    // Iterate over controller priorities in reverse, that is, from highest to
    // lowest.
    //
    // @@ Note: doing it in terms of direct iterators in anticipation for
    //    lower_bound().
    //
    auto cb (controllers.begin ());
    auto ce (controllers.end ());

    for (; cb != ce; )
    {
      const pair<const uint64_t, strings>& pu (*--ce);

      prio = pu.first;
      const strings& urls (pu.second);

      // Prepare task request (it will be the same within a given priority).
      //
      tq = task_request_manifest {
        hname,
        tc_name,
        tc_ver,
        imode,
        ilogin,
        fingerprint,
        machine_header_manifests {}};

      // Note: do not assume tq.machines.size () == ms.size ().
      //
      for (const bootstrapped_machine& m: ms)
      {
        // @@ For now skip machines locked by other processes.
        //
        // @@ Note: skip machines bootstrapping/suspended.
        //
        if (m.lock.locked ())
          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 ())
      {
        // If we have no machines for this priority then we won't have any
        // for any lower priority so bail out.
        //
        break;
      }

      // 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.
      //
      // @@ TODO: need to iterate in random order somehow.
      //
      for (const string& u: urls)
      {
        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.
            string (),      // Empty result URL.
            agent_checksum,
            move (t)};

          url = u;
          break;
        }

        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 /* fail_hard */);
          }
          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 << " "
                        << "priority " << prio;});

          tr  = move (r);
          url = u;
          break;
        }
      } // url loop.

      if (!tr.session.empty ()) // Got a task.
        break;

    } // prio loop.

    if (tq.machines.empty ()) // No machines.
    {
      // Normally this means all the machines are busy so sleep a bit less.
      //
      l2 ([&]{trace << "all machines are busy, sleeping";});
      sleep = rand_sleep () / 2;
      continue;
    }

    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.
    //
    task_manifest& t (*tr.task);

    // First verify the requested machine is one of those we sent in tq.
    //
    if (find_if (tq.machines.begin (), tq.machines.end (),
                 [&t] (const machine_header_manifest& mh)
                 {
                   return mh.name == t.machine; // Yes, names, not ids.
                 }) == tq.machines.end ())
    {
      error << "task from " << url << " for unknown machine " << t.machine;

      if (ops.dump_task ())
        return 0;

      continue;
    }

    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 ();

    // Reset the worker checksum if the task's agent checksum doesn't match
    // the current one.
    //
    // Note that since the checksums are hierarchical, such reset will trigger
    // resets of the "subordinate" checksums (dependency checksum, etc).
    //
    if (!tr.agent_checksum || *tr.agent_checksum != agent_checksum)
      t.worker_checksum = nullopt;

    // Handle interrupts.
    //
    // Note that the interrupt can be triggered both by another process (the
    // interrupt exception is thrown from perform_task()) as well as by this
    // process in case we were unable to interrupt the other process (seeing
    // that we have already received a task, responding with an interrupt
    // feels like the most sensible option).
    //
    result_manifest r;
    bootstrapped_machine* pm (nullptr);
    try
    {
      // Next find the corresponding bootstrapped_machine instance in ms. Also
      // unlock all the other machines.
      //
      // @@ TODO: looks like this is also where we will interrupt the machines
      //          (thus inside the try block). Note that we have to do this
      //          while holding the toolchain lock. Would be good to
      //          unlock all the machines as well as the toolchain lock on
      //          failure.
      //
      for (bootstrapped_machine& m: ms)
      {
        if (m.manifest.machine.name == t.machine)
        {
          assert (pm == nullptr); // Sanity check.

          m.lock.perform_task (tl, prio);
          pm = &m;
        }
        else
          m.lock.unlock ();
      }
      assert (pm != nullptr);

      r = perform_task (move (tl), pm->lock, pm->path, pm->manifest, t);
    }
    catch (const interrupt&)
    {
      r = result_manifest {
        t.name,
        t.version,
        result_status::interrupt,
        operation_results {},
        nullopt /* worker_checksum */,
        nullopt /* dependency_checksum */};
    }

    if (pm != nullptr) // Let's not assume.
      pm->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 (), sign_cmd,
                  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;
    }

    result_status rs (r.status);

    // Upload the result.
    //
    result_request_manifest rq {tr.session,
                                move (challenge),
                                agent_checksum,
                                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
        {
          // Don't break lines in the manifest values not to further increase
          // the size of the result request manifest encoded representation.
          // Note that this manifest can contain quite a few lines in the
          // operation logs, potentially truncated to fit the upload limit
          // (see worker/worker.cxx for details). Breaking these lines can
          // increase the request size beyond this limit and so we can end up
          // with the request failure.
          //
          serialize_manifest (rq,
                              c.out,
                              u,
                              "result request",
                              true /* fail_hard */,
                              true /* long_lines */);
        }
        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 << ' '
                  << "status " << rs << ' '
                  << "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;
  }
}