src/sys/fuzzing/realmfuzzer/engine/runner.cc

// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/sys/fuzzing/realmfuzzer/engine/runner.h"

#include <lib/syslog/cpp/macros.h>
#include <lib/zx/clock.h>
#include <zircon/sanitizer.h>
#include <zircon/status.h>
#include <zircon/syscalls/exception.h>

#include <deque>
#include <iostream>

#include "src/lib/fxl/macros.h"
#include "src/sys/fuzzing/libfuzzer/stats.h"
#include "src/sys/fuzzing/realmfuzzer/target/process.h"

namespace fuzzing {

using ::fuchsia::fuzzer::Data;
using ::fuchsia::fuzzer::MAX_PROCESS_STATS;
using ::fuchsia::fuzzer::OutputFlags;

RunnerPtr RealmFuzzerRunner::MakePtr(ExecutorPtr executor) {
  return RunnerPtr(new RealmFuzzerRunner(std::move(executor)));
}

RealmFuzzerRunner::RealmFuzzerRunner(ExecutorPtr executor)
    : Runner(executor),
      generated_receiver_(&generated_sender_),
      leak_receiver_(&leak_sender_),
      adapter_(executor),
      provider_(executor),
      processed_receiver_(&processed_sender_),
      workflow_(this) {
  generated_receiver_.Close();
  processed_receiver_.Close();
  leak_receiver_.Close();

  seed_corpus_ = Corpus::MakePtr();
  live_corpus_ = Corpus::MakePtr();
  pool_ = std::make_shared<ModulePool>();
}

void RealmFuzzerRunner::SetAdapterHandler(TargetAdapterClient::RequestHandler handler) {
  adapter_.set_handler(std::move(handler));
}

void RealmFuzzerRunner::SetProviderHandler(CoverageDataProviderClient::RequestHandler handler) {
  provider_.Bind(std::move(handler));
}

ZxPromise<> RealmFuzzerRunner::Initialize(std::string pkg_dir, std::vector<std::string> args) {
  return adapter_.GetParameters()
      .or_else([]() -> ZxResult<std::vector<std::string>> {
        FX_LOGS(WARNING) << "Failed to get parameters from adapter";
        return fpromise::error(ZX_ERR_INTERNAL);
      })
      .and_then([this, pkg_dir = std::move(pkg_dir),
                 args = std::move(args)](const std::vector<std::string>& parameters) mutable {
        std::copy(parameters.begin(), parameters.end(), std::back_inserter(args));
        return Runner::Initialize(std::move(pkg_dir), std::move(args));
      })
      .wrap_with(workflow_.scope());
}

ZxPromise<> RealmFuzzerRunner::Configure() {
  return Runner::Configure()
      .and_then([this]() -> ZxResult<> {
        const OptionsPtr& options = this->options();
        if (bool(options->output_flags() & OutputFlags::CLOSE_STDOUT)) {
          FX_LOGS(WARNING)
              << "realmfuzzer ignored a request to close stdout: the engine does not "
                 "currently have the ability to suppress stdout from other components being fuzzed.";
        }
        if (bool(options->output_flags() & OutputFlags::CLOSE_STDERR)) {
          FX_LOGS(WARNING)
              << "realmfuzzer ignored a request to close stderr: the engine does not "
                 "currently have the ability to suppress stderr from other components being fuzzed.";
        }
        seed_corpus_->Configure(options);
        live_corpus_->Configure(options);
        mutagen_.Configure(options);
        adapter_.Configure(options);
        for (auto& [target_id, process_proxy] : process_proxies_) {
          process_proxy->Configure(options);
        }
        return fpromise::ok();
      })
      .wrap_with(workflow_);
}

zx_status_t RealmFuzzerRunner::AddToCorpus(CorpusType corpus_type, Input input) {
  switch (corpus_type) {
    case CorpusType::SEED:
      return seed_corpus_->Add(std::move(input));
    case CorpusType::LIVE:
      return live_corpus_->Add(std::move(input));
    default:
      return ZX_ERR_INVALID_ARGS;
  }
}

std::vector<Input> RealmFuzzerRunner::GetCorpus(CorpusType corpus_type) {
  CorpusPtr corpus;
  switch (corpus_type) {
    case CorpusType::SEED:
      corpus = seed_corpus_;
      break;
    case CorpusType::LIVE:
      corpus = live_corpus_;
      break;
    default:
      FX_NOTREACHED();
  }
  // The first input is always the implicit empty input; skip it.
  auto num_inputs = corpus->num_inputs();
  std::vector<Input> inputs(num_inputs - 1);
  for (size_t i = 0; i < inputs.size(); ++i) {
    corpus->At(i + 1, &inputs[i]);
  }
  return inputs;
}

zx_status_t RealmFuzzerRunner::ParseDictionary(const Input& input) {
  Dictionary dict;
  dict.Configure(options());
  if (!dict.Parse(input)) {
    return ZX_ERR_INVALID_ARGS;
  }
  mutagen_.set_dictionary(std::move(dict));
  return ZX_OK;
}

Input RealmFuzzerRunner::GetDictionaryAsInput() const { return mutagen_.dictionary().AsInput(); }

///////////////////////////////////////////////////////////////
// Asynchronous workflows.

ZxPromise<Artifact> RealmFuzzerRunner::Fuzz() {
  return FuzzInputs(/* backlog= */ options()->mutation_depth()).wrap_with(workflow_);
}

ZxPromise<Artifact> RealmFuzzerRunner::TryEach(std::vector<Input> inputs) {
  return fpromise::make_promise([this, inputs = std::move(inputs)]() mutable -> ZxResult<> {
           for (auto& input : inputs) {
             if (auto status = generated_sender_.Send(std::move(input)); status != ZX_OK) {
               FX_LOGS(ERROR) << "Input queue closed prematurely: " << zx_status_get_string(status);
               return fpromise::error(status);
             }
           }
           generated_receiver_.Close();
           return fpromise::ok();
         })
      .and_then(TestInputs(kNoPostProcessing))
      .then([](ZxResult<Artifact>& result) -> ZxResult<FuzzResult> {
        if (result.is_ok()) {
          auto artifact = result.take_value();
          return fpromise::ok(artifact.fuzz_result());
        }
        auto status = result.take_error();
        if (status == ZX_ERR_STOP) {
          return fpromise::ok(FuzzResult::NO_ERRORS);
        }
        return fpromise::error(status);
      })
      .and_then([](FuzzResult& fuzz_result) -> ZxResult<Artifact> {
        return fpromise::ok(Artifact(fuzz_result));
      })
      .wrap_with(workflow_);
}

ZxPromise<Artifact> RealmFuzzerRunner::ValidateMinimize(Input input) {
  // Check that the input can be minimized, and that minimization is bounded.
  return TestOneAsync(std::move(input), kNoPostProcessing)
      .or_else([this](const zx_status_t& status) {
        return CheckPrevious(status).and_then([]() -> ZxResult<Artifact> {
          FX_LOGS(WARNING) << "Test input did not trigger an error.";
          return fpromise::error(ZX_ERR_INVALID_ARGS);
        });
      })
      .wrap_with(workflow_, Workflow::Mode::kStart);
}

ZxPromise<Artifact> RealmFuzzerRunner::Minimize(Artifact artifact) {
  FX_DCHECK(artifact.has_input());
  auto orig_corpus = live_corpus_;
  auto orig_options = CopyOptions(options());
  return fpromise::make_promise([this, fuzz_result = artifact.fuzz_result(),
                                 input = artifact.take_input(), minimize = ZxFuture<Artifact>()](
                                    Context& context) mutable -> ZxResult<Input> {
           while (true) {
             if (!minimize) {
               // Ratchet down the input one byte.
               if (input.size() < 2) {
                 FX_LOGS(INFO) << "Input is " << input.size()
                               << " byte(s); will not minimize further.";
                 return fpromise::ok(std::move(input));
               }
               if (!options()->has_runs() && !options()->has_max_total_time()) {
                 FX_LOGS(INFO)
                     << "'max_total_time' and 'runs' are both not set. Defaulting to 10 minutes.";
                 options()->set_max_total_time(zx::min(10).get());
               }
               auto next_input = input.Duplicate();
               next_input.Truncate(input.size() - 1);
               options()->set_max_input_size(next_input.size());
               // Start each fuzzing pass using the seed corpus and the minimized input.
               live_corpus_ = Corpus::MakePtr();
               live_corpus_->Configure(options());
               auto status = live_corpus_->Add(std::move(next_input));
               if (status != ZX_OK) {
                 FX_LOGS(ERROR) << "Failed to reset corpus: " << zx_status_get_string(status);
                 return fpromise::error(status);
               }
               // Imitate libFuzzer and count from 0 so long as errors are found.
               Reset();
               run_ = 0;
               pool_->Clear();
               minimize = FuzzInputs();
             }
             if (!minimize(context)) {
               return fpromise::pending();
             }
             if (minimize.is_error()) {
               return fpromise::error(minimize.error());
             }
             auto minimized = minimize.take_value();
             if (minimized.fuzz_result() == FuzzResult::NO_ERRORS) {
               FX_LOGS(INFO) << "Did not reduce error input beyond " << input.size()
                             << " bytes; exiting.";
               return fpromise::ok(std::move(input));
             }
             // TODO(https://fxbug.dev/42166322): This needs a more rigorous way of deduplicating crashes.
             if (minimized.fuzz_result() != fuzz_result) {
               FX_LOGS(WARNING) << "Different error detected; will not minimize further.";
               return fpromise::ok(std::move(input));
             }
             input = minimized.take_input();
           }
         })
      .then(
          [this, orig_corpus, options = std::move(orig_options)](ZxResult<Input>& result) mutable {
            pool_->Clear();
            live_corpus_ = orig_corpus;
            set_options(std::move(options));
            return std::move(result);
          })
      .and_then([](Input& input) -> ZxResult<Artifact> {
        return fpromise::ok(Artifact(FuzzResult::MINIMIZED, std::move(input)));
      })
      .wrap_with(workflow_, Workflow::Mode::kFinish);
}

ZxPromise<Artifact> RealmFuzzerRunner::Cleanse(Input input) {
  // The general approach of this loop is to take tested inputs and their fuzzing results and return
  // them to |GenerateCleanInputs| as |Artifacts|.
  AsyncSender<Artifact> sender;
  auto receiver = AsyncReceiver<Artifact>::MakePtr(&sender);
  return fpromise::make_promise([this, generate = Future<>(), sender = std::move(sender),
                                 receiver = std::move(receiver), test_inputs = ZxFuture<Artifact>(),
                                 receive = Future<Input>(),
                                 result = Artifact(FuzzResult::NO_ERRORS, std::move(input)),
                                 artifacts = std::array<Artifact, 2>(),
                                 num_artifacts = 0U](Context& context) mutable -> ZxResult<Input> {
           while (true) {
             if (!generate) {
               // To set up initial conditions, simulate having just completed an "extra" attempt.
               constexpr size_t kMaxCleanseAttempts = 5;
               auto attempts_left = kMaxCleanseAttempts + 1;

               // Prepare the pipeline with some artifacts that make the attempt succeed and won't
               // be reverted. This only fails if the `sender` is closed, in which case the promise
               // below returns an error.
               for (size_t i = 0; i < 2; ++i) {
                 if (auto status =
                         sender.Send(Artifact(FuzzResult::CRASH, result.input().Duplicate()));
                     status != ZX_OK) {
                   FX_LOGS(ERROR) << "Failed to prepare fuzzing input pipeline: "
                                  << zx_status_get_string(status);
                 }
               }
               generate = GenerateCleanInputs(std::move(receiver), attempts_left);
             }
             if (!test_inputs) {
               test_inputs = TestInputs(kNoPostProcessing);
             }
             if (!receive) {
               receive = processed_receiver_.Receive();
             }
             if (generate(context) && generate.is_error()) {
               // |GenerateCleanInputs| only returns an error if its queues close unexpectedly.
               return fpromise::error(ZX_ERR_BAD_STATE);
             }
             if (test_inputs(context)) {
               if (test_inputs.is_error()) {
                 auto status = test_inputs.error();
                 if (status != ZX_ERR_STOP) {
                   return fpromise::error(status);
                 }
                 return fpromise::ok(result.take_input());
               }
               // Cleansed input triggered an error. Use it as the basis for further attempts.
               result = test_inputs.take_value();
               artifacts[0] = result.Duplicate();
               artifacts[1] = result.Duplicate();
               receive = nullptr;
               Reset();
               num_artifacts = 2;
             } else if (receive(context)) {
               if (receive.is_error()) {
                 FX_LOGS(ERROR) << "Output queue closed unexpectedly.";
                 return fpromise::error(ZX_ERR_BAD_STATE);
               }
               // Cleansed input didn't trigger an error. Save it for recycling.
               artifacts[num_artifacts++] = Artifact(FuzzResult::NO_ERRORS, receive.take_value());
             } else {
               // Still testing an input.
               return fpromise::pending();
             }
             if (num_artifacts < artifacts.size()) {
               continue;
             }
             // Recycle inputs in pairs, one for each "clean" byte.
             if (sender.Send(std::move(artifacts[0])) != ZX_OK ||
                 sender.Send(std::move(artifacts[1])) != ZX_OK) {
               // No more inputs are needed; all done.
               return fpromise::ok(result.take_input());
             }
             num_artifacts = 0;
           }
         })
      .and_then([](Input& input) -> ZxResult<Artifact> {
        return fpromise::ok(Artifact(FuzzResult::CLEANSED, std::move(input)));
      })
      .wrap_with(workflow_);
}

ZxPromise<> RealmFuzzerRunner::ValidateMerge() {
  // Accumulate the coverage from testing all the elements of the seed corpus.
  return TestOneAsync(Input(), kAccumulateCoverage)
      .or_else([this](const zx_status_t& status) {
        return CheckPrevious(status).and_then(TestCorpusAsync(seed_corpus_, kAccumulateCoverage));
      })
      .and_then([](Artifact& artifact) -> ZxResult<> {
        FX_LOGS(WARNING) << "Seed corpus contains an input that triggers an error: '"
                         << artifact.input().ToHex() << "'";
        return fpromise::error(ZX_ERR_INVALID_ARGS);
      })
      .or_else([this](zx_status_t& status) { return CheckPrevious(status); })
      .wrap_with(workflow_, Workflow::Mode::kStart);
}

ZxPromise<Artifact> RealmFuzzerRunner::Merge() {
  auto collect_errors = std::make_shared<std::vector<Input>>();
  return fpromise::make_promise([this, collect_errors] {
           // Measure what coverage each element of the live corpus provides beyond that accumulated
           // by the seed corpus. After this step the live corpus contains only valid, measured
           // inputs.
           auto unmeasured = live_corpus_;
           live_corpus_ = Corpus::MakePtr();
           live_corpus_->Configure(options());
           return TestCorpusAsync(unmeasured, kMeasureCoverageAndKeepInputs, collect_errors);
         })
      .or_else([this, collect_errors](const zx_status_t& status) {
        return CheckPrevious(status).and_then([this, collect_errors] {
          if (!collect_errors->empty()) {
            FX_LOGS(WARNING) << "Corpus contains input(s) that trigger error(s):";
            for (auto& input : *collect_errors) {
              FX_LOGS(WARNING) << "  '" << input.ToHex() << "'";
            }
          }
          // Accumulate the coverage from each element of the live corpus. The live corpus will be
          // stably sorted by size, number of features measured above, and lexicographical order.
          // Only elements that add coverage not accumulated by previous elements will be kept.
          auto measured = live_corpus_;
          live_corpus_ = Corpus::MakePtr();
          live_corpus_->Configure(options());
          return TestCorpusAsync(measured, kAccumulateCoverageAndKeepInputs);
        });
      })
      .and_then([](const Artifact& artifact) -> ZxResult<> {
        FX_LOGS(ERROR) << "Previously successful input triggered an error: '"
                       << artifact.input().ToHex() << "'";
        return fpromise::error(ZX_ERR_BAD_STATE);
      })
      .or_else([this, collect_errors](const zx_status_t& status) {
        return CheckPrevious(status).and_then([this, collect_errors]() -> ZxResult<> {
          // As a final step, keep any inputs that triggered errors.
          for (auto& input : *collect_errors) {
            if (auto status = live_corpus_->Add(std::move(input)); status != ZX_OK) {
              return fpromise::error(status);
            }
          }
          return fpromise::ok();
        });
      })
      .and_then([]() -> ZxResult<Artifact> { return fpromise::ok(Artifact(FuzzResult::MERGED)); })
      .wrap_with(workflow_, Workflow::Mode::kFinish);
}

void RealmFuzzerRunner::UpdateMonitorsWithStatus(UpdateReason reason, Status status) {
  if (bool(options()->output_flags() & OutputFlags::LIBFUZZER)) {
    std::cerr << FormatLibFuzzerStats(reason, status) << std::endl;
  }
  Runner::UpdateMonitorsWithStatus(reason, std::move(status));
}

Status RealmFuzzerRunner::CollectStatus() {
  Status status;
  status.set_running(!stopped_);
  status.set_runs(run_);

  auto elapsed = zx::clock::get_monotonic() - start_;
  status.set_elapsed(elapsed.to_nsecs());

  size_t covered_features;
  auto covered_pcs = pool_->GetCoverage(&covered_features);
  status.set_covered_pcs(covered_pcs);
  status.set_covered_features(covered_features);

  status.set_corpus_num_inputs(seed_corpus_->num_inputs() + live_corpus_->num_inputs());
  status.set_corpus_total_size(seed_corpus_->total_size() + live_corpus_->total_size());

  std::vector<ProcessStats> all_stats;
  all_stats.reserve(std::min<size_t>(process_proxies_.size(), MAX_PROCESS_STATS));
  for (auto& [target_id, process_proxy] : process_proxies_) {
    if (all_stats.size() == all_stats.capacity()) {
      break;
    }
    ProcessStats stats;
    auto status = process_proxy->GetStats(&stats);
    if (status == ZX_OK) {
      all_stats.push_back(stats);
    } else {
      FX_LOGS(WARNING) << "Failed to get stats for process: " << zx_status_get_string(status);
    }
  }
  status.set_process_stats(std::move(all_stats));
  return status;
}

ZxPromise<> RealmFuzzerRunner::Stop() {
  stopped_ = true;
  return workflow_.Stop();
}

///////////////////////////////////////////////////////////////
// Workflow-related methods.

void RealmFuzzerRunner::StartWorkflow(Scope& scope) {
  Reset();
  run_ = 0;
  pool_->Clear();
  start_ = zx::clock::get_monotonic();
  pulse_start_ = start_ + zx::sec(2);
  stopped_ = false;
  // Handle coverage data produced during the workflow.
  auto task =
      fpromise::make_promise([this,
                              fut = Future<CoverageData>()](Context& context) mutable -> Result<> {
        while (true) {
          if (!fut) {
            fut = provider_.GetCoverageData();
          }
          if (!fut(context)) {
            return fpromise::pending();
          }
          if (fut.is_error()) {
            return fpromise::error();
          }
          auto coverage = fut.take_value();
          switch (coverage.data.Which()) {
            case Data::Tag::kInstrumented:
              ConnectProcess(std::move(coverage));
              break;
            case Data::Tag::kInline8bitCounters:
              AddInline8bitCounters(std::move(coverage));
              break;
            default:
              FX_LOGS(WARNING) << "Unrecgonized coverage data type: " << coverage.data.Which();
              return fpromise::error();
          }
        }
      }).wrap_with(scope);
  executor()->schedule_task(std::move(task));
  UpdateMonitors(UpdateReason::INIT);
}

void RealmFuzzerRunner::FinishWorkflow() {
  generated_receiver_.Clear();
  processed_receiver_.Clear();
  leak_receiver_.Clear();
  stopped_ = true;
  UpdateMonitors(UpdateReason::DONE);
}

///////////////////////////////////////////////////////////////
// Methods to generate fuzzing inputs.

ZxPromise<> RealmFuzzerRunner::GenerateInputs(size_t num_inputs, size_t backlog) {
  // Set up parameters for determining what inputs to generate and for how long.
  auto max_size = options()->max_input_size();
  auto max_time = zx::duration(options()->max_total_time());
  auto deadline = max_time.get() ? zx::deadline_after(max_time) : zx::time::infinite();
  auto mutation_depth = options()->mutation_depth();
  return fpromise::make_promise([this, backlog, max_size]() -> ZxResult<> {
           // "Precycle" some inputs by making it look like they are ready for reuse.
           for (size_t i = 0; i <= backlog; i++) {
             auto status = processed_sender_.Send(Input(max_size));
             if (status != ZX_OK) {
               FX_LOGS(ERROR) << "Input queue closed prematurely while preparing to fuzz: "
                              << zx_status_get_string(status);
               return fpromise::error(status);
             }
           }
           return fpromise::ok();
         })
      .and_then([this, num_sent = 0U, num_inputs, deadline, num_mutations = mutation_depth,
                 mutation_depth,
                 recycle = Future<Input>()](Context& context) mutable -> ZxResult<> {
        while (true) {
          if (num_inputs != 0 && num_sent >= num_inputs) {
            // Run limit will be reached by inputs already queued; all done.
            return fpromise::ok();
          }
          if (zx::clock::get_monotonic() >= deadline) {
            // Time limit reached; all done.
            return fpromise::ok();
          }
          if (stopped_) {
            // Interrupted; all done.
            return fpromise::ok();
          }
          if (!recycle) {
            // Use inputs recycled from earlier runs to reduce heap allocations.
            recycle = processed_receiver_.Receive();
          }
          if (!recycle(context)) {
            return fpromise::pending();
          }
          if (recycle.is_error()) {
            // Queue was closed; all done.
            return fpromise::ok();
          }
          auto input = recycle.take_value();
          if (num_mutations >= mutation_depth) {
            // Pick an input and mutate it |mutation_depth| times in a row.
            mutagen_.reset_mutations();
            live_corpus_->Pick(mutagen_.base_input());
            live_corpus_->Pick(mutagen_.crossover());
            num_mutations = 0;
          }
          mutagen_.Mutate(&input);
          ++num_mutations;
          auto status = generated_sender_.Send(std::move(input));
          num_sent++;
          if (status != ZX_OK) {
            // Queue was closed; all done.
            return fpromise::ok();
          }
        }
      })
      .and_then([this] {
        generated_receiver_.Close();
        return fpromise::ok();
      });
}

Promise<> RealmFuzzerRunner::GenerateCleanInputs(AsyncReceiverPtr<Artifact> receiver,
                                                 size_t attempts_left) {
  // Ensure that a new attempt will be started.
  auto offset = std::numeric_limits<size_t>::max() - 1;

  // The general approach is to produce two inputs at a time, each with one byte replaced by a
  // space or 0xff. Bytes that are already a space or 0xff are skipped. Each iteration over all
  // input bytes is an attempt, and inputs are produced until an attempt doesn't produce any errors
  // or five attempts have been performed.
  return fpromise::make_promise([this, receiver = std::move(receiver), receive = Future<Artifact>(),
                                 artifacts = std::array<Artifact, 2>(), num_artifacts = 0U,
                                 attempts_left, offset, found_error = false,
                                 original = uint8_t(0)](Context& context) mutable -> Result<> {
    while (true) {
      // Recycle two artifacts.
      if (!receive) {
        receive = receiver->Receive();
      }
      if (!receive(context)) {
        return fpromise::pending();
      }
      if (receive.is_error()) {
        FX_LOGS(ERROR) << "Recycled input queue closed unexpectedly.";
        return fpromise::error();
      }
      artifacts[num_artifacts++] = receive.take_value();
      if (num_artifacts < artifacts.size()) {
        continue;
      }
      auto fuzz_result = artifacts[0].fuzz_result();
      auto input0 = artifacts[0].take_input();
      auto input1 = artifacts[1].take_input();
      auto* data0 = input0.data();
      auto* data1 = input1.data();
      if (fuzz_result == FuzzResult::NO_ERRORS) {
        // Last inputs didn't trigger any errors; restore the modified byte.
        data0[offset] = original;
        data1[offset] = original;
      } else {
        found_error = true;
      }
      // Find a "cleanable" byte, i.e. one that isn't already 0x20 or 0xff.
      do {
        if (++offset < input0.size()) {
          // Continue the current attempt.
          continue;
        }
        // Reached the end of the input. Start a new attempt.
        offset = 0;
        if (--attempts_left == 0 || !found_error) {
          // Out of attempts, or last attempt didn't trigger any error. All done.
          receiver->Close();
          generated_receiver_.Close();
          return fpromise::ok();
        }
        found_error = false;
      } while (data0[offset] == 0x20 || data0[offset] == 0xff);
      // Now actually clean the byte and send them to be tested.
      original = data0[offset];
      data0[offset] = 0x20;
      data1[offset] = 0xff;
      if (generated_sender_.Send(std::move(input0)) != ZX_OK ||
          generated_sender_.Send(std::move(input1)) != ZX_OK) {
        FX_LOGS(ERROR) << "Input queue unexpectedly closed.";
        return fpromise::error();
      }
      num_artifacts = 0;
    }
  });
}

///////////////////////////////////////////////////////////////
// Methods to perform a sequence of fuzzing runs.

ZxPromise<Artifact> RealmFuzzerRunner::FuzzInputs(size_t backlog) {
  size_t num_inputs = options()->runs();
  if (num_inputs != 0) {
    // Always perform at least enough runs to test every input in the seed and live corpora.
    // `num_fixed_inputs` cannot underflow, as the corpus always includes exactly one empty input,
    // of which the one double-counted instance is removed.
    // It also cannot overflow, as the combined `num_inputs` is less than the amount of memory being
    // used (modulo the single empty input), which by definition must fit within a `size_t`.
    size_t num_fixed_inputs = seed_corpus_->num_inputs() + live_corpus_->num_inputs() - 1;
    if (num_inputs < num_fixed_inputs) {
      num_inputs = num_fixed_inputs;
    } else {
      num_inputs -= num_fixed_inputs;
    }
  }
  return TestOneAsync(Input(), kAccumulateCoverage)
      .or_else([this](const zx_status_t& status) {
        return CheckPrevious(status).and_then(
            [this] { return TestCorpusAsync(seed_corpus_, kAccumulateCoverage); });
      })
      .or_else([this](const zx_status_t& status) {
        return CheckPrevious(status).and_then(
            [this] { return TestCorpusAsync(live_corpus_, kAccumulateCoverage); });
      })
      .or_else([this, backlog, num_inputs](const zx_status_t& status) {
        live_corpus_->Add(seed_corpus_);
        return CheckPrevious(status).and_then(
            [generate = ZxFuture<>(GenerateInputs(num_inputs, backlog)),
             test = ZxFuture<Artifact>(TestInputs(kAccumulateCoverageAndKeepInputs))](
                Context& context) mutable -> ZxResult<Artifact> {
              if (generate(context) && generate.is_error()) {
                return fpromise::error(generate.error());
              }
              if (!test(context)) {
                return fpromise::pending();
              }
              return test.take_result();
            });
      })
      .or_else([this](const zx_status_t& status) {
        return CheckPrevious(status).and_then([]() -> ZxResult<Artifact> {
          // Finished without finding an input that causes an error; return an empty artifact.
          return fpromise::ok(Artifact());
        });
      });
}

ZxPromise<Artifact> RealmFuzzerRunner::TestOneAsync(Input input, PostProcessing mode) {
  return fpromise::make_promise([this, input = std::move(input)]() mutable -> ZxResult<> {
           return AsZxResult(generated_sender_.Send(std::move(input)));
         })
      .and_then([this]() {
        generated_receiver_.Close();
        return fpromise::ok();
      })
      .and_then([this, mode] { return TestInputs(mode); });
}

ZxPromise<Artifact> RealmFuzzerRunner::TestCorpusAsync(CorpusPtr corpus, PostProcessing mode,
                                                       InputsPtr collect_errors) {
  return fpromise::make_promise([this]() -> ZxResult<> {
           // Prime the output queue.
           Reset();
           return AsZxResult(processed_sender_.Send(Input()));
         })
      .and_then([this, corpus, mode, collect_errors, test_inputs = ZxFuture<Artifact>(),
                 receive = Future<Input>(),
                 offset = 1U](Context& context) mutable -> ZxResult<Artifact> {
        while (true) {
          if (!test_inputs) {
            test_inputs = TestInputs(mode, collect_errors);
          }
          if (!receive) {
            receive = processed_receiver_.Receive();
          }
          if (test_inputs(context)) {
            // Done testing inputs.
            Reset();
            return test_inputs.take_result();
          }
          if (!receive(context)) {
            // Still testing.
            return fpromise::pending();
          }
          // Ready for the next input from the corpus.
          if (receive.is_error()) {
            FX_LOGS(ERROR) << "Output queue closed prematurely.";
            return fpromise::error(ZX_ERR_BAD_STATE);
          }
          auto input = receive.take_value();
          if (!corpus->At(offset++, &input)) {
            generated_receiver_.Close();
            continue;
          }
          if (auto status = generated_sender_.Send(std::move(input)); status != ZX_OK) {
            FX_LOGS(ERROR) << "Input queue closed prematurely: " << zx_status_get_string(status);
            return fpromise::error(status);
          }
        }
      });
}

ZxPromise<Artifact> RealmFuzzerRunner::TestInputs(PostProcessing mode, InputsPtr collect_errors) {
  constexpr size_t kMaxLeakDetectionAttempts = 1000;
  auto leak_detections = options()->detect_leaks() ? kMaxLeakDetectionAttempts : 0;
  return fpromise::make_promise(
      [this, mode, collect_errors, input = Input(), leak_detections, detect_leaks = false,
       prepare = ZxFuture<Input>(),
       run = Future<bool, FuzzResult>()](Context& context) mutable -> ZxResult<Artifact> {
        while (true) {
          // Reset process coverage and get a new input.
          if (!prepare) {
            prepare = Prepare(detect_leaks);
          }
          if (!prepare(context)) {
            return fpromise::pending();
          }
          // Make sure no errors have been received before testing an input.
          if (prepare.is_error()) {
            return fpromise::error(prepare.error());
          }
          if (!run) {
            input = std::move(prepare.value());
            run = RunOne(input);
          }
          // Now check if the run has finished and if any process reported an error.
          if (!run(context)) {
            return fpromise::pending();
          }
          auto leak_suspected = false;
          if (run.is_ok()) {
            leak_suspected = run.take_value();
          } else if (collect_errors) {
            // If collecting errors, clear errors and continue. Simulate already having attempted to
            // |detect_leaks| to true to skip analysis and leak detection.
            collect_errors->emplace_back(std::move(input));
            detect_leaks = true;
            run = nullptr;
          } else {
            return fpromise::ok(Artifact(run.error(), std::move(input)));
          }
          // Skip post-processing when repeating inputs for leak detection.
          if (!detect_leaks) {
            Analyze(input, mode);
          }
          // Iteration complete! Clear the futures so that the loop starts from the top again.
          detect_leaks = Recycle(std::move(input), leak_detections, leak_suspected, detect_leaks);
          prepare = nullptr;
        }
      });
}

///////////////////////////////////////////////////////////////
// Methods to perform individual steps of a single fuzzing run.

ZxPromise<> RealmFuzzerRunner::CheckPrevious(zx_status_t status) {
  return fpromise::make_promise([status]() -> ZxResult<> {
    if (status != ZX_ERR_STOP) {
      return fpromise::error(status);
    }
    return fpromise::ok();
  });
}

ZxPromise<Input> RealmFuzzerRunner::Prepare(bool detect_leaks) {
  return fpromise::make_promise([this, detect_leaks]() {
           // Send start signals.
           std::vector<ZxPromise<>> starts;
           for (auto& [target_id, process_proxy] : process_proxies_) {
             starts.emplace_back(process_proxy->Start(detect_leaks));
           };
           // Wait for processes to acknowledge.
           return fpromise::join_promise_vector(std::move(starts));
         })
      .then([](Result<std::vector<ZxResult<>>>& results) -> ZxResult<> {
        for (auto& result : results.value()) {
          if (result.is_error()) {
            // Ideally, processes should only return errors as a result of testing inputs.
            FX_LOGS(WARNING)
                << "Detected error between fuzzing runs. This error cannot be associated "
                   "with a specific input. The fuzzer may be non-deterministic and/or "
                   "non-hermetic, and may need to be modified to make results more "
                   "reproducible.";
            return fpromise::error(ZX_ERR_BAD_STATE);
          }
        }
        return fpromise::ok();
      })
      .and_then([this, leak = Future<Input>(),
                 generate = Future<Input>()](Context& context) mutable -> ZxResult<Input> {
        if (!leak) {
          leak = leak_receiver_.Receive();
        }
        if (!generate) {
          generate = generated_receiver_.Receive();
        }
        if (leak(context)) {
          if (leak.is_error()) {
            return fpromise::error(ZX_ERR_STOP);
          }
          return fpromise::ok(leak.take_value());
        }
        if (generate(context)) {
          if (generate.is_error()) {
            // No more inputs means the workflow is done.
            return fpromise::error(ZX_ERR_STOP);
          }
          return fpromise::ok(generate.take_value());
        }
        return fpromise::pending();
      });
}

Promise<bool, FuzzResult> RealmFuzzerRunner::RunOne(const Input& input) {
  return fpromise::make_promise([this, &input, run_limit = options()->run_limit(),
                                 timeout = Future<>(),
                                 first = true](Context& context) mutable -> Result<bool, uint64_t> {
           // Create a future for the per-run timeout. If this completes, it's an error.
           if (run_limit && !timeout) {
             timeout = executor()->MakeDelayedPromise(zx::duration(run_limit));
           }
           if (run_limit && timeout(context)) {
             return fpromise::error(kTimeoutTargetId);
           }
           if (first) {
             ++run_;
             for (auto& [target_id, process_proxy] : process_proxies_) {
               futures_.emplace_back(process_proxy->AwaitFinish());
             }
             futures_.emplace_back(adapter_.TestOneInput(input)
                                       .or_else([] { return fpromise::error(kInvalidTargetId); })
                                       .and_then([this]() -> Result<bool, uint64_t> {
                                         for (auto& [target_id, process_proxy] : process_proxies_) {
                                           if (auto status = process_proxy->Finish();
                                               status != ZX_OK) {
                                             FX_LOGS(WARNING) << "Failed to signal process: "
                                                              << zx_status_get_string(status);
                                           }
                                         }
                                         return fpromise::ok(false);
                                       }));
             first = false;
           }
           auto all_done = true;
           auto leak_suspected = false;
           for (auto& future : futures_) {
             if (!future(context)) {
               all_done = false;
               continue;
             }
             if (future.is_error()) {
               return fpromise::error(future.error());
             }
             leak_suspected |= future.value();
           }
           if (!all_done) {
             suspended_ = context.suspend_task();
             return fpromise::pending();
           }
           return fpromise::ok(leak_suspected);
         })
      .inspect([this](const Result<bool, uint64_t>& ignored) { futures_.clear(); })
      .or_else([this](const uint64_t& target_id) { return GetFuzzResult(target_id); });
}

void RealmFuzzerRunner::ConnectProcess(CoverageData coverage) {
  auto process_proxy = std::make_unique<ProcessProxy>(executor(), pool_);
  process_proxy->Configure(options());
  if (auto status = process_proxy->Connect(coverage.target_id, coverage.data.instrumented());
      status != ZX_OK) {
    FX_LOGS(WARNING) << "Failed to add process: " << zx_status_get_string(status);
    return;
  }
  auto target_id = process_proxy->target_id();
  futures_.emplace_back(process_proxy->AwaitFinish());
  process_proxies_[target_id] = std::move(process_proxy);
  // Kick |RunOne| to check the |AwaitFinish| future.
  suspended_.resume_task();
}

void RealmFuzzerRunner::AddInline8bitCounters(CoverageData coverage) {
  auto iter = process_proxies_.find(coverage.target_id);
  if (iter == process_proxies_.end()) {
    FX_LOGS(WARNING) << "Failed to add module: no such target_id: " << coverage.target_id;
    return;
  }
  auto& process_proxy = iter->second;
  if (auto status = process_proxy->AddInline8bitCounters(coverage.data.inline_8bit_counters());
      status != ZX_OK) {
    FX_LOGS(WARNING) << "Failed to add module: " << zx_status_get_string(status);
    return;
  }
}

Promise<bool, FuzzResult> RealmFuzzerRunner::GetFuzzResult(uint64_t target_id) {
  return fpromise::make_promise(
             [this, target_id, process_proxy = std::unique_ptr<ProcessProxy>(),
              result = ZxFuture<FuzzResult>()](Context& context) mutable -> ZxResult<FuzzResult> {
               if (target_id == kTimeoutTargetId) {
                 // For timeouts, dump all threads to the sanitizer log.
                 constexpr size_t kBufSize = 1ULL << 20;
                 auto buf = std::make_unique<char[]>(kBufSize);
                 for (auto& [target_id, proxy] : process_proxies_) {
                   auto len = proxy->Dump(buf.get(), kBufSize);
                   __sanitizer_log_write(buf.get(), len);
                 }
                 return fpromise::ok(FuzzResult::TIMEOUT);
               }
               if (!result) {
                 // For all other errors, wait on the result from the process exitcode.
                 auto iter = process_proxies_.find(target_id);
                 if (iter == process_proxies_.end()) {
                   FX_LOGS(ERROR) << "Cannot get error from unknown target_id: 0x" << std::hex
                                  << target_id;
                   return fpromise::error(ZX_ERR_NOT_FOUND);
                 }
                 process_proxy = std::move(iter->second);
                 process_proxies_.erase(iter);
                 result = process_proxy->GetResult();
               }
               if (!result(context)) {
                 return fpromise::pending();
               }
               return result.take_result();
             })
      .or_else([](const zx_status_t& status) -> Result<FuzzResult, FuzzResult> {
        FX_LOGS(WARNING) << "Failed to get result: " << zx_status_get_string(status);
        FX_LOGS(WARNING) << "Defaulting to error type of 'crash'.";
        return fpromise::ok(FuzzResult::CRASH);
      })
      .and_then([this](const FuzzResult& fuzz_result) -> Result<bool, FuzzResult> {
        if (fuzz_result == FuzzResult::NO_ERRORS) {
          return fpromise::ok(false);
        }
        // If it's an ignored exit(),just remove that one process_proxy and treat it like a success.
        if (fuzz_result == FuzzResult::EXIT && !options()->detect_exits()) {
          return fpromise::ok(false);
        }
        // Otherwise, it's really an error. Remove the target adapter and all proxies.
        adapter_.Disconnect();
        process_proxies_.clear();
        return fpromise::error(fuzz_result);
      });
}

void RealmFuzzerRunner::Analyze(Input& input, PostProcessing mode) {
  bool updated = false;
  switch (mode) {
    case kNoPostProcessing: {
      break;
    }
    case kAccumulateCoverage: {
      pool_->Accumulate();
      break;
    }
    case kMeasureCoverageAndKeepInputs: {
      auto num_features = pool_->Measure();
      if (num_features) {
        input.set_num_features(num_features);
        if (auto status = live_corpus_->Add(std::move(input)); status != ZX_OK) {
          FX_LOGS(WARNING) << "Failed to save input: " << zx_status_get_string(status);
        }
      }
      break;
    }
    case kAccumulateCoverageAndKeepInputs: {
      if (pool_->Accumulate()) {
        if (auto status = live_corpus_->Add(std::move(input)); status != ZX_OK) {
          FX_LOGS(WARNING) << "Failed to save input: " << zx_status_get_string(status);
        }
        UpdateMonitors(UpdateReason::NEW);
        updated = true;
      }
      break;
    }
  }
  // After a few seconds, reassure the user that the fuzzer is running by reporting each run that
  // equals a power of 2, unless it was already reported above.
  if (!updated && (run_ & (run_ - 1)) == 0 && zx::clock::get_monotonic() > pulse_start_) {
    UpdateMonitors(UpdateReason::PULSE);
  }
}

bool RealmFuzzerRunner::Recycle(Input&& input, size_t& attempts_left, bool suspected,
                                bool detecting) {
  // Determine if leak detection is needed and thereby where to send the input. Leak detection is
  // expensive, so the strategy is as follows:
  // 1. Try inputs once without leak detection.
  // 2. If leak detection is requested, check if leaks are suspected (unbalanced malloc/frees).
  // 3. If a leak if suspected, do the normal feedback analysis and then try the input again, this
  //    time with leak detection. Skip the feedback analysis on the second try.
  // 4. Keep track of how many suspected leaks don't result in an error. After
  //    |kMaxLeakDetections|, disable further leak detection.
  if (attempts_left) {
    if (detecting) {
      // Already tried detecting a leak. Decrement the number of attempts.
      --attempts_left;
      if (attempts_left == 0) {
        FX_LOGS(INFO) << "Disabling leak detection: No memory leaks have been found in any inputs "
                      << "suspected of leaking. Memory may be accumulating in some global state "
                      << "without leaking. End-of-process leak checks will still be performed.";
      }
    } else if (suspected) {
      // Leak detection is still possible, and the last run exhibited a suspected leak. Push the
      // input to the front of the queue to retry with leak detection.

      if (auto status = leak_sender_.Send(std::move(input)); status != ZX_OK) {
        FX_LOGS(WARNING) << "Failed to resend input: " << zx_status_get_string(status);
      } else {
        return true;
      }
    }
  }
  // Send input to be recycled.
  if (auto status = processed_sender_.Send(std::move(input)); status != ZX_OK) {
    FX_LOGS(WARNING) << "Failed to recycle input: " << zx_status_get_string(status);
  }
  return false;
}

///////////////////////////////////////////////////////////////
// Clean-up methods.

void RealmFuzzerRunner::Disconnect() {
  adapter_.Disconnect();
  process_proxies_.clear();
}

void RealmFuzzerRunner::Reset() {
  generated_receiver_.Reset();
  processed_receiver_.Reset();
  leak_receiver_.Reset();
}

}  // namespace fuzzing