src/ledger/bin/app/merging/merge_resolver.cc - fuchsia - Git at Google

 // Copyright 2016 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/ledger/bin/app/merging/merge_resolver.h"

 #include <lib/callback/scoped_callback.h>
 #include <lib/callback/trace_callback.h>
 #include <lib/callback/waiter.h>
 #include <lib/fit/defer.h>
 #include <lib/fit/function.h>

 #include <algorithm>
 #include <memory>
 #include <queue>
 #include <set>
 #include <utility>

 #include "src/ledger/bin/app/merging/common_ancestor.h"
 #include "src/ledger/bin/app/merging/ledger_merge_manager.h"
 #include "src/ledger/bin/app/merging/merge_strategy.h"
 #include "src/ledger/bin/app/page_manager.h"
 #include "src/ledger/bin/app/page_utils.h"
 #include "src/ledger/bin/cobalt/cobalt.h"
 #include "src/ledger/lib/coroutine/coroutine.h"
 #include "src/ledger/lib/coroutine/coroutine_waiter.h"
 #include "src/lib/fxl/memory/ref_ptr.h"
 #include "src/lib/fxl/memory/weak_ptr.h"

 namespace ledger {

 // Enumerates merge candidates' indexes among current head commits.
 class MergeResolver::MergeCandidates {
  public:
   MergeCandidates();

   // Resets the MergeCandidates and sets the total number of head commits to
   // |head_count|.
   void ResetCandidates(size_t head_count);

   // Returns whether MergeCandidates should be reset. A reset is necessary
   // when the head commits have changed, i.e. when there is a successful merge
   // or on a new commit.
   bool NeedsReset() { return needs_reset_; }

   // Returns the current pair on indexes of head commits to be merged.
   std::pair<size_t, size_t> GetCurrentPair();

   // Returns whether there is a merge candidate pair available.
   bool HasCandidate();

   // Returns true if there was a network error in one of the previous merge
   // attempts. This does not include merges before |ResetCandidates| was
   // called.
   bool HadNetworkErrors() { return had_network_errors_; }

   // Should be called after a successful merge.
   void OnMergeSuccess();

   // Should be called after an unsuccessful merge.
   void OnMergeError(storage::Status status);

   // Should be called when new commits are available.
   void OnNewCommits();

   // Returns the number of head commits.
   size_t head_count() { return head_count_; }

  private:
   // Advances to the next available pair of merge candidates.
   void PrepareNext();

   size_t head_count_;
   std::pair<size_t, size_t> current_pair_;
   bool needs_reset_ = true;
   bool had_network_errors_ = false;
 };

 MergeResolver::MergeCandidates::MergeCandidates() {}

 void MergeResolver::MergeCandidates::ResetCandidates(size_t head_count) {
   head_count_ = head_count;
   current_pair_ = {0, 1};
   needs_reset_ = false;
   had_network_errors_ = false;
 }

 bool MergeResolver::MergeCandidates::HasCandidate() {
   return current_pair_.first != head_count_ - 1;
 }

 std::pair<size_t, size_t> MergeResolver::MergeCandidates::GetCurrentPair() {
   return current_pair_;
 }

 void MergeResolver::MergeCandidates::OnMergeSuccess() { needs_reset_ = true; }

 void MergeResolver::MergeCandidates::OnMergeError(storage::Status status) {
   if (status == storage::Status::NETWORK_ERROR) {
     // The contents of the common ancestor are unavailable locally and it wasn't
     // possible to retrieve them through the network: Ignore this pair of heads
     // for now.
     had_network_errors_ = true;
     PrepareNext();
   } else {
     FXL_LOG(WARNING) << "Merging failed. Will try again later.";
   }
 }

 void MergeResolver::MergeCandidates::OnNewCommits() { needs_reset_ = true; }

 void MergeResolver::MergeCandidates::PrepareNext() {
   ++current_pair_.second;
   if (current_pair_.second == head_count_) {
     ++current_pair_.first;
     current_pair_.second = current_pair_.first + 1;
   }
 }

 MergeResolver::MergeResolver(fit::closure on_destroyed,
                              Environment* environment,
                              storage::PageStorage* storage,
                              std::unique_ptr<backoff::Backoff> backoff)
     : coroutine_service_(environment->coroutine_service()),
       storage_(storage),
       backoff_(std::move(backoff)),
       merge_candidates_(std::make_unique<MergeCandidates>()),
       on_destroyed_(std::move(on_destroyed)),
       task_runner_(environment->dispatcher()) {
   storage_->AddCommitWatcher(this);
   PostCheckConflicts(DelayedStatus::DONT_DELAY);
 }

 MergeResolver::~MergeResolver() {
   storage_->RemoveCommitWatcher(this);
   on_destroyed_();
 }

 void MergeResolver::set_on_empty(fit::closure on_empty_callback) {
   on_empty_callback_ = std::move(on_empty_callback);
 }

 bool MergeResolver::IsEmpty() { return !merge_in_progress_; }

 bool MergeResolver::HasUnfinishedMerges() {
   return merge_in_progress_ || check_conflicts_in_progress_ ||
          check_conflicts_task_count_ != 0 || in_delay_ ||
          merge_candidates_->HadNetworkErrors();
 }

 void MergeResolver::SetMergeStrategy(std::unique_ptr<MergeStrategy> strategy) {
   if (merge_in_progress_) {
     FXL_DCHECK(strategy_);
     // The new strategy can be the empty strategy (nullptr), so we need a
     // separate boolean to know if we have a pending strategy change to make.
     has_next_strategy_ = true;
     next_strategy_ = std::move(strategy);
     strategy_->Cancel();
     return;
   }
   strategy_.swap(strategy);
   if (strategy_) {
     PostCheckConflicts(DelayedStatus::DONT_DELAY);
   }
 }

 void MergeResolver::SetPageManager(PageManager* page_manager) {
   FXL_DCHECK(page_manager_ == nullptr);
   page_manager_ = page_manager;
 }

 void MergeResolver::RegisterNoConflictCallback(
     fit::function<void(ConflictResolutionWaitStatus)> callback) {
   no_conflict_callbacks_.push_back(std::move(callback));
 }

 void MergeResolver::OnNewCommits(
     const std::vector<std::unique_ptr<const storage::Commit>>& /*commits*/,
     storage::ChangeSource source) {
   merge_candidates_->OnNewCommits();
   PostCheckConflicts(source == storage::ChangeSource::LOCAL
                          ? DelayedStatus::DONT_DELAY
                          // We delay remote commits.
                          : DelayedStatus::MAY_DELAY);
 }

 void MergeResolver::PostCheckConflicts(DelayedStatus delayed_status) {
   check_conflicts_task_count_++;
   task_runner_.PostTask([this, delayed_status] {
     check_conflicts_task_count_--;
     CheckConflicts(delayed_status);
   });
 }

 void MergeResolver::CheckConflicts(DelayedStatus delayed_status) {
   if (!strategy_ || merge_in_progress_ || check_conflicts_in_progress_ ||
       in_delay_) {
     // No strategy is set, or a merge is already in progress, or we are already
     // checking for conflicts, or we are delaying merges. Let's bail out early.
     return;
   }
   check_conflicts_in_progress_ = true;
   std::vector<std::unique_ptr<const storage::Commit>> heads;
   storage::Status s = storage_->GetHeadCommits(&heads);
   check_conflicts_in_progress_ = false;

   if (merge_candidates_->NeedsReset()) {
     merge_candidates_->ResetCandidates(heads.size());
   }
   FXL_DCHECK(merge_candidates_->head_count() == heads.size())
       << merge_candidates_->head_count() << " != " << heads.size();

   if (s != storage::Status::OK || heads.size() == 1 ||
       !(merge_candidates_->HasCandidate())) {
     // An error occurred, or there is no conflict we can resolve. In
     // either case, return early.
     if (s != storage::Status::OK) {
       FXL_LOG(ERROR) << "Failed to get head commits with status " << s;
     } else if (heads.size() == 1) {
       for (auto& callback : no_conflict_callbacks_) {
         callback(has_merged_ ? ConflictResolutionWaitStatus::CONFLICTS_RESOLVED
                              : ConflictResolutionWaitStatus::NO_CONFLICTS);
       }
       no_conflict_callbacks_.clear();
       has_merged_ = false;
     }
     if (on_empty_callback_) {
       on_empty_callback_();
     }
     return;
   }
   if (!strategy_) {
     if (on_empty_callback_) {
       on_empty_callback_();
     }
     return;
   }
   merge_in_progress_ = true;
   std::pair<size_t, size_t> head_indexes = merge_candidates_->GetCurrentPair();
   ResolveConflicts(delayed_status, std::move(heads[head_indexes.first]),
                    std::move(heads[head_indexes.second]));
 }

 void MergeResolver::ResolveConflicts(
     DelayedStatus delayed_status, std::unique_ptr<const storage::Commit> head1,
     std::unique_ptr<const storage::Commit> head2) {
   auto cleanup = fit::defer(task_runner_.MakeScoped([this, delayed_status] {
     // |merge_in_progress_| must be reset before calling
     // |on_empty_callback_|.
     merge_in_progress_ = false;

     if (has_next_strategy_) {
       strategy_ = std::move(next_strategy_);
       next_strategy_.reset();
       has_next_strategy_ = false;
     }
     PostCheckConflicts(delayed_status);
     // Call on_empty_callback_ at the very end as it might delete the
     // resolver.
     if (on_empty_callback_) {
       on_empty_callback_();
     }
   }));
   uint64_t id = TRACE_NONCE();
   TRACE_ASYNC_BEGIN("ledger", "merge", id);
   auto tracing = fit::defer([id] { TRACE_ASYNC_END("ledger", "merge", id); });

   FXL_DCHECK(storage::Commit::TimestampOrdered(head1, head2));

   if (head1->GetParentIds().size() == 2 && head2->GetParentIds().size() == 2) {
     if (delayed_status == DelayedStatus::MAY_DELAY) {
       // If trying to merge 2 merge commits, add some delay with
       // exponential backoff.
       auto delay_callback = [this] {
         in_delay_ = false;
         CheckConflicts(DelayedStatus::DONT_DELAY);
       };
       in_delay_ = true;
       task_runner_.PostDelayedTask(
           TRACE_CALLBACK(std::move(delay_callback), "ledger", "merge_delay"),
           backoff_->GetNext());
       cleanup.cancel();
       merge_in_progress_ = false;
       // We don't want to continue merging if nobody is interested
       // (all clients disconnected).
       if (on_empty_callback_) {
         on_empty_callback_();
       }
       return;
     }
     // If delayed_status is not initial, report the merge.
     ReportEvent(CobaltEvent::MERGED_COMMITS_MERGED);
   } else {
     // No longer merging 2 merge commits, reinitialize the exponential
     // backoff.
     backoff_->Reset();
   }

   // Merge the first two commits using the most recent one as the
   // base.
   RecursiveMergeOneStep(
       std::move(head1), std::move(head2),
       [cleanup = std::move(cleanup), tracing = std::move(tracing)] {
         ReportEvent(CobaltEvent::COMMITS_MERGED);
       });
 }

 void MergeResolver::RecursiveMergeOneStep(
     std::unique_ptr<const storage::Commit> left,
     std::unique_ptr<const storage::Commit> right,
     fit::closure on_successful_merge) {
   coroutine_service_->StartCoroutine(
       [this, left = std::move(left), right = std::move(right),
        on_successful_merge = std::move(on_successful_merge)](
           coroutine::CoroutineHandler* handler) mutable {
         TRACE_DURATION("ledger", "recursive_merge");
         storage::Status status =
             RecursiveMergeSync(handler, std::move(left), std::move(right));
         if (status == storage::Status::INTERRUPTED) {
           return;
         }
         if (status != storage::Status::OK) {
           FXL_LOG(ERROR) << "Recursive merge failed";
           return;
         }
         on_successful_merge();
       });
 }

 storage::Status MergeResolver::MergeCommitsToContentOfLeftSync(
     coroutine::CoroutineHandler* handler,
     std::unique_ptr<const storage::Commit> left,
     std::unique_ptr<const storage::Commit> right) {
   std::unique_ptr<storage::Journal> journal =
       storage_->StartMergeCommit(std::move(left), std::move(right));
   has_merged_ = true;

   storage::Status status;
   std::unique_ptr<const storage::Commit> commit;
   auto sync_call_status = coroutine::SyncCall(
       handler,
       [this, journal = std::move(journal)](
           fit::function<void(storage::Status status,
                              std::unique_ptr<const storage::Commit>)>
               callback) mutable {
         storage_->CommitJournal(std::move(journal), std::move(callback));
       },
       &status, &commit);
   if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
     return storage::Status::INTERRUPTED;
   }
   return status;
 }

 // Synchronously get the commit with id |commit_id|. Try |candidate| if it has
 // the right id, otherwise fetch it from storage.
 storage::Status MergeResolver::GetCommitSync(
     coroutine::CoroutineHandler* handler, storage::CommitIdView commit_id,
     std::unique_ptr<const storage::Commit> candidate,
     std::unique_ptr<const storage::Commit>* result) {
   // Exit early if we already have the commit.
   if (candidate->GetId() == commit_id) {
     *result = std::move(candidate);
     return storage::Status::OK;
   }

   storage::Status status;
   auto sync_call_status = coroutine::SyncCall(
       handler,
       [this,
        commit_id](fit::function<void(storage::Status status,
                                      std::unique_ptr<const storage::Commit>)>
                       callback) {
         storage_->GetCommit(commit_id, std::move(callback));
       },
       &status, result);
   if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
     return storage::Status::INTERRUPTED;
   }
   // If the strategy has been changed, bail early.
   if (has_next_strategy_) {
     return storage::Status::INTERRUPTED;
   }
   return status;
 }

 // Requests the merges of |right_commit| and any element of |left_commits|, and
 // return them in |merges|.
 storage::Status MergeResolver::FindMergesSync(
     coroutine::CoroutineHandler* handler,
     const std::vector<storage::CommitId>& left_commits,
     storage::CommitId right_commit, std::vector<storage::CommitId>* merges) {
   auto waiter = fxl::MakeRefCounted<
       callback::Waiter<storage::Status, std::vector<storage::CommitId>>>(
       storage::Status::OK);
   for (const auto& left_commit : left_commits) {
     storage_->GetMergeCommitIds(left_commit, right_commit,
                                 waiter->NewCallback());
   }
   storage::Status status;
   std::vector<std::vector<storage::CommitId>> merge_lists;
   if (coroutine::Wait(handler, std::move(waiter), &status, &merge_lists) ==
       coroutine::ContinuationStatus::INTERRUPTED) {
     return storage::Status::INTERRUPTED;
   }
   // If the strategy has been changed, bail early.
   if (has_next_strategy_) {
     return storage::Status::INTERRUPTED;
   }
   if (status != storage::Status::OK) {
     return status;
   }
   for (auto& merge_list : merge_lists) {
     merges->insert(merges->end(), std::make_move_iterator(merge_list.begin()),
                    std::make_move_iterator(merge_list.end()));
   }
   return storage::Status::OK;
 }

 // Try to build a merge of all commits in |ancestors|. Either the merge already
 // exists and is returned in |final_merge| or one intermediate merge is
 // constructed.
 storage::Status MergeResolver::MergeSetSync(
     coroutine::CoroutineHandler* handler,
     std::vector<std::unique_ptr<const storage::Commit>> ancestors,
     std::unique_ptr<const storage::Commit>* final_merge) {
   FXL_DCHECK(!ancestors.empty());

   // Sort ancestors by timestamp. This guarantees that, when we call the merge
   // strategy, the right-hand side commit is always the most recent, and also
   // matches (as much as possible) the order in which heads would be merged.
   std::sort(ancestors.begin(), ancestors.end(),
             storage::Commit::TimestampOrdered);

   // Build a merge of the first N ancestors. This holds the list of available
   // merges of all the ancestors examined until now. Since merges have the
   // maximum timestamp of their parents as timestamps, all commits in this list
   // are older than the Nth ancestor, but they may have lower or higher commit
   // ids.
   std::vector<storage::CommitId> merges;
   // The first ancestor is a merge of itself.
   merges.push_back(ancestors[0]->GetId());

   for (auto it = ancestors.begin() + 1, end = ancestors.end(); it < end; it++) {
     // Request the merges of the ancestor |*it| and any element of |merges|.
     std::vector<storage::CommitId> next_merges;
     auto& next_ancestor = *it;

     storage::Status status =
         FindMergesSync(handler, merges, next_ancestor->GetId(), &next_merges);
     if (status != storage::Status::OK) {
       return status;
     }
     // If |next_merges| is empty, the merges we need are not present yet. We
     // call RecursiveMergeOneStep recursively.
     if (next_merges.empty()) {
       // Try to create the merge in a deterministic way: order merges by id.
       std::sort(merges.begin(), merges.end());

       // Get |merge[0]| from storage, or from |ancestors[0]| if they are the
       // same commit.
       std::unique_ptr<const storage::Commit> last_merge;
       storage::Status status = GetCommitSync(
           handler, merges[0], std::move(ancestors[0]), &last_merge);
       if (status != storage::Status::OK) {
         return status;
       }
       // We know that |last_merge->GetTimestamp() <=
       // next_ancestor->GetTimestamp()| but the commit id of |last_merge| may be
       // higher. In case of equality we need to reorder the calls.
       if (!storage::Commit::TimestampOrdered(last_merge, next_ancestor)) {
         FXL_DCHECK(last_merge->GetTimestamp() == next_ancestor->GetTimestamp());
         return RecursiveMergeSync(handler, std::move(next_ancestor),
                                   std::move(last_merge));
       }
       return RecursiveMergeSync(handler, std::move(last_merge),
                                 std::move(next_ancestor));
     }
     merges = std::move(next_merges);
   }

   FXL_DCHECK(!merges.empty());

   // Try to create the merge in a deterministic way: order by id.
   std::sort(merges.begin(), merges.end());
   return GetCommitSync(handler, merges[0], std::move(ancestors[0]),
                        final_merge);
 }

 // Does one step of recursive merging: tries to merge |left| and |right| and
 // either produces a merge commit, or calls itself recursively to merge some
 // common ancestors. Assumes that |left| is older than |right| according to
 // |storage::Commit::TimestampOrdered|.
 storage::Status MergeResolver::RecursiveMergeSync(
     coroutine::CoroutineHandler* handler,
     std::unique_ptr<const storage::Commit> left,
     std::unique_ptr<const storage::Commit> right) {
   FXL_DCHECK(storage::Commit::TimestampOrdered(left, right));

   CommitComparison comparison;
   std::vector<std::unique_ptr<const storage::Commit>> common_ancestors;
   storage::Status status;
   {
     TRACE_DURATION("ledger", "merge_common_ancestor");
     status =
         FindCommonAncestors(handler, storage_, left->Clone(), right->Clone(),
                             &comparison, &common_ancestors);
   }
   if (status != storage::Status::OK) {
     return status;
   }
   // If the strategy has been changed, bail early.
   if (has_next_strategy_) {
     return storage::Status::INTERRUPTED;
   }

   if (comparison == CommitComparison::LEFT_SUBSET_OF_RIGHT) {
     return MergeCommitsToContentOfLeftSync(handler, std::move(right),
                                            std::move(left));
   } else if (comparison == CommitComparison::RIGHT_SUBSET_OF_LEFT) {
     return MergeCommitsToContentOfLeftSync(handler, std::move(left),
                                            std::move(right));
   } else if (comparison == CommitComparison::EQUIVALENT) {
     // The commits are equivalent so we can merge to the content
     // of either of them.
     return MergeCommitsToContentOfLeftSync(handler, std::move(left),
                                            std::move(right));
   }

   FXL_DCHECK(!common_ancestors.empty());

   // MergeSetSync has 3 possible results:
   //  - a non-OK storage::Status
   //  - a commit returned in merge_base
   //  - OK with an empty merge_base
   std::unique_ptr<const storage::Commit> merge_base;
   status = MergeSetSync(handler, std::move(common_ancestors), &merge_base);
   if (status != storage::Status::OK) {
     return status;
   }
   if (!merge_base) {
     // A commit was made, resume when notified of it.
     return storage::Status::OK;
   }

   has_merged_ = true;

   storage::Status merge_status;
   auto sync_call_status = coroutine::SyncCall(
       handler,
       [this, left = std::move(left), right = std::move(right),
        merge_base = std::move(merge_base)](
           fit::function<void(storage::Status)> callback) mutable {
         strategy_->Merge(storage_, page_manager_, std::move(left),
                          std::move(right), std::move(merge_base),
                          TRACE_CALLBACK(std::move(callback), "ledger",
                                         "merge_strategy_merge"));
       },
       &merge_status);
   if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
     return storage::Status::INTERRUPTED;
   }
   if (merge_status != storage::Status::OK) {
     merge_candidates_->OnMergeError(merge_status);
     return storage::Status::ILLEGAL_STATE;
   }
   merge_candidates_->OnMergeSuccess();
   return storage::Status::OK;
 }

 }  // namespace ledger
	// Copyright 2016 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/ledger/bin/app/merging/merge_resolver.h"

	#include <lib/callback/scoped_callback.h>
	#include <lib/callback/trace_callback.h>
	#include <lib/callback/waiter.h>
	#include <lib/fit/defer.h>
	#include <lib/fit/function.h>

	#include <algorithm>
	#include <memory>
	#include <queue>
	#include <set>
	#include <utility>

	#include "src/ledger/bin/app/merging/common_ancestor.h"
	#include "src/ledger/bin/app/merging/ledger_merge_manager.h"
	#include "src/ledger/bin/app/merging/merge_strategy.h"
	#include "src/ledger/bin/app/page_manager.h"
	#include "src/ledger/bin/app/page_utils.h"
	#include "src/ledger/bin/cobalt/cobalt.h"
	#include "src/ledger/lib/coroutine/coroutine.h"
	#include "src/ledger/lib/coroutine/coroutine_waiter.h"
	#include "src/lib/fxl/memory/ref_ptr.h"
	#include "src/lib/fxl/memory/weak_ptr.h"

	namespace ledger {

	// Enumerates merge candidates' indexes among current head commits.
	class MergeResolver::MergeCandidates {
	public:
	MergeCandidates();

	// Resets the MergeCandidates and sets the total number of head commits to
	// \|head_count\|.
	void ResetCandidates(size_t head_count);

	// Returns whether MergeCandidates should be reset. A reset is necessary
	// when the head commits have changed, i.e. when there is a successful merge
	// or on a new commit.
	bool NeedsReset() { return needs_reset_; }

	// Returns the current pair on indexes of head commits to be merged.
	std::pair<size_t, size_t> GetCurrentPair();

	// Returns whether there is a merge candidate pair available.
	bool HasCandidate();

	// Returns true if there was a network error in one of the previous merge
	// attempts. This does not include merges before \|ResetCandidates\| was
	// called.
	bool HadNetworkErrors() { return had_network_errors_; }

	// Should be called after a successful merge.
	void OnMergeSuccess();

	// Should be called after an unsuccessful merge.
	void OnMergeError(storage::Status status);

	// Should be called when new commits are available.
	void OnNewCommits();

	// Returns the number of head commits.
	size_t head_count() { return head_count_; }

	private:
	// Advances to the next available pair of merge candidates.
	void PrepareNext();

	size_t head_count_;
	std::pair<size_t, size_t> current_pair_;
	bool needs_reset_ = true;
	bool had_network_errors_ = false;
	};

	MergeResolver::MergeCandidates::MergeCandidates() {}

	void MergeResolver::MergeCandidates::ResetCandidates(size_t head_count) {
	head_count_ = head_count;
	current_pair_ = {0, 1};
	needs_reset_ = false;
	had_network_errors_ = false;
	}

	bool MergeResolver::MergeCandidates::HasCandidate() {
	return current_pair_.first != head_count_ - 1;
	}

	std::pair<size_t, size_t> MergeResolver::MergeCandidates::GetCurrentPair() {
	return current_pair_;
	}

	void MergeResolver::MergeCandidates::OnMergeSuccess() { needs_reset_ = true; }

	void MergeResolver::MergeCandidates::OnMergeError(storage::Status status) {
	if (status == storage::Status::NETWORK_ERROR) {
	// The contents of the common ancestor are unavailable locally and it wasn't
	// possible to retrieve them through the network: Ignore this pair of heads
	// for now.
	had_network_errors_ = true;
	PrepareNext();
	} else {
	FXL_LOG(WARNING) << "Merging failed. Will try again later.";
	}
	}

	void MergeResolver::MergeCandidates::OnNewCommits() { needs_reset_ = true; }

	void MergeResolver::MergeCandidates::PrepareNext() {
	++current_pair_.second;
	if (current_pair_.second == head_count_) {
	++current_pair_.first;
	current_pair_.second = current_pair_.first + 1;
	}
	}

	MergeResolver::MergeResolver(fit::closure on_destroyed,
	Environment* environment,
	storage::PageStorage* storage,
	std::unique_ptr<backoff::Backoff> backoff)
	: coroutine_service_(environment->coroutine_service()),
	storage_(storage),
	backoff_(std::move(backoff)),
	merge_candidates_(std::make_unique<MergeCandidates>()),
	on_destroyed_(std::move(on_destroyed)),
	task_runner_(environment->dispatcher()) {
	storage_->AddCommitWatcher(this);
	PostCheckConflicts(DelayedStatus::DONT_DELAY);
	}

	MergeResolver::~MergeResolver() {
	storage_->RemoveCommitWatcher(this);
	on_destroyed_();
	}

	void MergeResolver::set_on_empty(fit::closure on_empty_callback) {
	on_empty_callback_ = std::move(on_empty_callback);
	}

	bool MergeResolver::IsEmpty() { return !merge_in_progress_; }

	bool MergeResolver::HasUnfinishedMerges() {
	return merge_in_progress_ \|\| check_conflicts_in_progress_ \|\|
	check_conflicts_task_count_ != 0 \|\| in_delay_ \|\|
	merge_candidates_->HadNetworkErrors();
	}

	void MergeResolver::SetMergeStrategy(std::unique_ptr<MergeStrategy> strategy) {
	if (merge_in_progress_) {
	FXL_DCHECK(strategy_);
	// The new strategy can be the empty strategy (nullptr), so we need a
	// separate boolean to know if we have a pending strategy change to make.
	has_next_strategy_ = true;
	next_strategy_ = std::move(strategy);
	strategy_->Cancel();
	return;
	}
	strategy_.swap(strategy);
	if (strategy_) {
	PostCheckConflicts(DelayedStatus::DONT_DELAY);
	}
	}

	void MergeResolver::SetPageManager(PageManager* page_manager) {
	FXL_DCHECK(page_manager_ == nullptr);
	page_manager_ = page_manager;
	}

	void MergeResolver::RegisterNoConflictCallback(
	fit::function<void(ConflictResolutionWaitStatus)> callback) {
	no_conflict_callbacks_.push_back(std::move(callback));
	}

	void MergeResolver::OnNewCommits(
	const std::vector<std::unique_ptr<const storage::Commit>>& /commits/,
	storage::ChangeSource source) {
	merge_candidates_->OnNewCommits();
	PostCheckConflicts(source == storage::ChangeSource::LOCAL
	? DelayedStatus::DONT_DELAY
	// We delay remote commits.
	: DelayedStatus::MAY_DELAY);
	}

	void MergeResolver::PostCheckConflicts(DelayedStatus delayed_status) {
	check_conflicts_task_count_++;
	task_runner_.PostTask([this, delayed_status] {
	check_conflicts_task_count_--;
	CheckConflicts(delayed_status);
	});
	}

	void MergeResolver::CheckConflicts(DelayedStatus delayed_status) {
	if (!strategy_ \|\| merge_in_progress_ \|\| check_conflicts_in_progress_ \|\|
	in_delay_) {
	// No strategy is set, or a merge is already in progress, or we are already
	// checking for conflicts, or we are delaying merges. Let's bail out early.
	return;
	}
	check_conflicts_in_progress_ = true;
	std::vector<std::unique_ptr<const storage::Commit>> heads;
	storage::Status s = storage_->GetHeadCommits(&heads);
	check_conflicts_in_progress_ = false;

	if (merge_candidates_->NeedsReset()) {
	merge_candidates_->ResetCandidates(heads.size());
	}
	FXL_DCHECK(merge_candidates_->head_count() == heads.size())
	<< merge_candidates_->head_count() << " != " << heads.size();

	if (s != storage::Status::OK \|\| heads.size() == 1 \|\|
	!(merge_candidates_->HasCandidate())) {
	// An error occurred, or there is no conflict we can resolve. In
	// either case, return early.
	if (s != storage::Status::OK) {
	FXL_LOG(ERROR) << "Failed to get head commits with status " << s;
	} else if (heads.size() == 1) {
	for (auto& callback : no_conflict_callbacks_) {
	callback(has_merged_ ? ConflictResolutionWaitStatus::CONFLICTS_RESOLVED
	: ConflictResolutionWaitStatus::NO_CONFLICTS);
	}
	no_conflict_callbacks_.clear();
	has_merged_ = false;
	}
	if (on_empty_callback_) {
	on_empty_callback_();
	}
	return;
	}
	if (!strategy_) {
	if (on_empty_callback_) {
	on_empty_callback_();
	}
	return;
	}
	merge_in_progress_ = true;
	std::pair<size_t, size_t> head_indexes = merge_candidates_->GetCurrentPair();
	ResolveConflicts(delayed_status, std::move(heads[head_indexes.first]),
	std::move(heads[head_indexes.second]));
	}

	void MergeResolver::ResolveConflicts(
	DelayedStatus delayed_status, std::unique_ptr<const storage::Commit> head1,
	std::unique_ptr<const storage::Commit> head2) {
	auto cleanup = fit::defer(task_runner_.MakeScoped([this, delayed_status] {
	// \|merge_in_progress_\| must be reset before calling
	// \|on_empty_callback_\|.
	merge_in_progress_ = false;

	if (has_next_strategy_) {
	strategy_ = std::move(next_strategy_);
	next_strategy_.reset();
	has_next_strategy_ = false;
	}
	PostCheckConflicts(delayed_status);
	// Call on_empty_callback_ at the very end as it might delete the
	// resolver.
	if (on_empty_callback_) {
	on_empty_callback_();
	}
	}));
	uint64_t id = TRACE_NONCE();
	TRACE_ASYNC_BEGIN("ledger", "merge", id);
	auto tracing = fit::defer([id] { TRACE_ASYNC_END("ledger", "merge", id); });

	FXL_DCHECK(storage::Commit::TimestampOrdered(head1, head2));

	if (head1->GetParentIds().size() == 2 && head2->GetParentIds().size() == 2) {
	if (delayed_status == DelayedStatus::MAY_DELAY) {
	// If trying to merge 2 merge commits, add some delay with
	// exponential backoff.
	auto delay_callback = [this] {
	in_delay_ = false;
	CheckConflicts(DelayedStatus::DONT_DELAY);
	};
	in_delay_ = true;
	task_runner_.PostDelayedTask(
	TRACE_CALLBACK(std::move(delay_callback), "ledger", "merge_delay"),
	backoff_->GetNext());
	cleanup.cancel();
	merge_in_progress_ = false;
	// We don't want to continue merging if nobody is interested
	// (all clients disconnected).
	if (on_empty_callback_) {
	on_empty_callback_();
	}
	return;
	}
	// If delayed_status is not initial, report the merge.
	ReportEvent(CobaltEvent::MERGED_COMMITS_MERGED);
	} else {
	// No longer merging 2 merge commits, reinitialize the exponential
	// backoff.
	backoff_->Reset();
	}

	// Merge the first two commits using the most recent one as the
	// base.
	RecursiveMergeOneStep(
	std::move(head1), std::move(head2),
	[cleanup = std::move(cleanup), tracing = std::move(tracing)] {
	ReportEvent(CobaltEvent::COMMITS_MERGED);
	});
	}

	void MergeResolver::RecursiveMergeOneStep(
	std::unique_ptr<const storage::Commit> left,
	std::unique_ptr<const storage::Commit> right,
	fit::closure on_successful_merge) {
	coroutine_service_->StartCoroutine(
	[this, left = std::move(left), right = std::move(right),
	on_successful_merge = std::move(on_successful_merge)](
	coroutine::CoroutineHandler* handler) mutable {
	TRACE_DURATION("ledger", "recursive_merge");
	storage::Status status =
	RecursiveMergeSync(handler, std::move(left), std::move(right));
	if (status == storage::Status::INTERRUPTED) {
	return;
	}
	if (status != storage::Status::OK) {
	FXL_LOG(ERROR) << "Recursive merge failed";
	return;
	}
	on_successful_merge();
	});
	}

	storage::Status MergeResolver::MergeCommitsToContentOfLeftSync(
	coroutine::CoroutineHandler* handler,
	std::unique_ptr<const storage::Commit> left,
	std::unique_ptr<const storage::Commit> right) {
	std::unique_ptr<storage::Journal> journal =
	storage_->StartMergeCommit(std::move(left), std::move(right));
	has_merged_ = true;

	storage::Status status;
	std::unique_ptr<const storage::Commit> commit;
	auto sync_call_status = coroutine::SyncCall(
	handler,
	[this, journal = std::move(journal)](
	fit::function<void(storage::Status status,
	std::unique_ptr<const storage::Commit>)>
	callback) mutable {
	storage_->CommitJournal(std::move(journal), std::move(callback));
	},
	&status, &commit);
	if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
	return storage::Status::INTERRUPTED;
	}
	return status;
	}

	// Synchronously get the commit with id \|commit_id\|. Try \|candidate\| if it has
	// the right id, otherwise fetch it from storage.
	storage::Status MergeResolver::GetCommitSync(
	coroutine::CoroutineHandler* handler, storage::CommitIdView commit_id,
	std::unique_ptr<const storage::Commit> candidate,
	std::unique_ptr<const storage::Commit>* result) {
	// Exit early if we already have the commit.
	if (candidate->GetId() == commit_id) {
	*result = std::move(candidate);
	return storage::Status::OK;
	}

	storage::Status status;
	auto sync_call_status = coroutine::SyncCall(
	handler,
	[this,
	commit_id](fit::function<void(storage::Status status,
	std::unique_ptr<const storage::Commit>)>
	callback) {
	storage_->GetCommit(commit_id, std::move(callback));
	},
	&status, result);
	if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
	return storage::Status::INTERRUPTED;
	}
	// If the strategy has been changed, bail early.
	if (has_next_strategy_) {
	return storage::Status::INTERRUPTED;
	}
	return status;
	}

	// Requests the merges of \|right_commit\| and any element of \|left_commits\|, and
	// return them in \|merges\|.
	storage::Status MergeResolver::FindMergesSync(
	coroutine::CoroutineHandler* handler,
	const std::vector<storage::CommitId>& left_commits,
	storage::CommitId right_commit, std::vector<storage::CommitId>* merges) {
	auto waiter = fxl::MakeRefCounted<
	callback::Waiter<storage::Status, std::vector<storage::CommitId>>>(
	storage::Status::OK);
	for (const auto& left_commit : left_commits) {
	storage_->GetMergeCommitIds(left_commit, right_commit,
	waiter->NewCallback());
	}
	storage::Status status;
	std::vector<std::vector<storage::CommitId>> merge_lists;
	if (coroutine::Wait(handler, std::move(waiter), &status, &merge_lists) ==
	coroutine::ContinuationStatus::INTERRUPTED) {
	return storage::Status::INTERRUPTED;
	}
	// If the strategy has been changed, bail early.
	if (has_next_strategy_) {
	return storage::Status::INTERRUPTED;
	}
	if (status != storage::Status::OK) {
	return status;
	}
	for (auto& merge_list : merge_lists) {
	merges->insert(merges->end(), std::make_move_iterator(merge_list.begin()),
	std::make_move_iterator(merge_list.end()));
	}
	return storage::Status::OK;
	}

	// Try to build a merge of all commits in \|ancestors\|. Either the merge already
	// exists and is returned in \|final_merge\| or one intermediate merge is
	// constructed.
	storage::Status MergeResolver::MergeSetSync(
	coroutine::CoroutineHandler* handler,
	std::vector<std::unique_ptr<const storage::Commit>> ancestors,
	std::unique_ptr<const storage::Commit>* final_merge) {
	FXL_DCHECK(!ancestors.empty());

	// Sort ancestors by timestamp. This guarantees that, when we call the merge
	// strategy, the right-hand side commit is always the most recent, and also
	// matches (as much as possible) the order in which heads would be merged.
	std::sort(ancestors.begin(), ancestors.end(),
	storage::Commit::TimestampOrdered);

	// Build a merge of the first N ancestors. This holds the list of available
	// merges of all the ancestors examined until now. Since merges have the
	// maximum timestamp of their parents as timestamps, all commits in this list
	// are older than the Nth ancestor, but they may have lower or higher commit
	// ids.
	std::vector<storage::CommitId> merges;
	// The first ancestor is a merge of itself.
	merges.push_back(ancestors[0]->GetId());

	for (auto it = ancestors.begin() + 1, end = ancestors.end(); it < end; it++) {
	// Request the merges of the ancestor \|*it\| and any element of \|merges\|.
	std::vector<storage::CommitId> next_merges;
	auto& next_ancestor = *it;

	storage::Status status =
	FindMergesSync(handler, merges, next_ancestor->GetId(), &next_merges);
	if (status != storage::Status::OK) {
	return status;
	}
	// If \|next_merges\| is empty, the merges we need are not present yet. We
	// call RecursiveMergeOneStep recursively.
	if (next_merges.empty()) {
	// Try to create the merge in a deterministic way: order merges by id.
	std::sort(merges.begin(), merges.end());

	// Get \|merge[0]\| from storage, or from \|ancestors[0]\| if they are the
	// same commit.
	std::unique_ptr<const storage::Commit> last_merge;
	storage::Status status = GetCommitSync(
	handler, merges[0], std::move(ancestors[0]), &last_merge);
	if (status != storage::Status::OK) {
	return status;
	}
	// We know that \|last_merge->GetTimestamp() <=
	// next_ancestor->GetTimestamp()\| but the commit id of \|last_merge\| may be
	// higher. In case of equality we need to reorder the calls.
	if (!storage::Commit::TimestampOrdered(last_merge, next_ancestor)) {
	FXL_DCHECK(last_merge->GetTimestamp() == next_ancestor->GetTimestamp());
	return RecursiveMergeSync(handler, std::move(next_ancestor),
	std::move(last_merge));
	}
	return RecursiveMergeSync(handler, std::move(last_merge),
	std::move(next_ancestor));
	}
	merges = std::move(next_merges);
	}

	FXL_DCHECK(!merges.empty());

	// Try to create the merge in a deterministic way: order by id.
	std::sort(merges.begin(), merges.end());
	return GetCommitSync(handler, merges[0], std::move(ancestors[0]),
	final_merge);
	}

	// Does one step of recursive merging: tries to merge \|left\| and \|right\| and
	// either produces a merge commit, or calls itself recursively to merge some
	// common ancestors. Assumes that \|left\| is older than \|right\| according to
	// \|storage::Commit::TimestampOrdered\|.
	storage::Status MergeResolver::RecursiveMergeSync(
	coroutine::CoroutineHandler* handler,
	std::unique_ptr<const storage::Commit> left,
	std::unique_ptr<const storage::Commit> right) {
	FXL_DCHECK(storage::Commit::TimestampOrdered(left, right));

	CommitComparison comparison;
	std::vector<std::unique_ptr<const storage::Commit>> common_ancestors;
	storage::Status status;
	{
	TRACE_DURATION("ledger", "merge_common_ancestor");
	status =
	FindCommonAncestors(handler, storage_, left->Clone(), right->Clone(),
	&comparison, &common_ancestors);
	}
	if (status != storage::Status::OK) {
	return status;
	}
	// If the strategy has been changed, bail early.
	if (has_next_strategy_) {
	return storage::Status::INTERRUPTED;
	}

	if (comparison == CommitComparison::LEFT_SUBSET_OF_RIGHT) {
	return MergeCommitsToContentOfLeftSync(handler, std::move(right),
	std::move(left));
	} else if (comparison == CommitComparison::RIGHT_SUBSET_OF_LEFT) {
	return MergeCommitsToContentOfLeftSync(handler, std::move(left),
	std::move(right));
	} else if (comparison == CommitComparison::EQUIVALENT) {
	// The commits are equivalent so we can merge to the content
	// of either of them.
	return MergeCommitsToContentOfLeftSync(handler, std::move(left),
	std::move(right));
	}

	FXL_DCHECK(!common_ancestors.empty());

	// MergeSetSync has 3 possible results:
	// - a non-OK storage::Status
	// - a commit returned in merge_base
	// - OK with an empty merge_base
	std::unique_ptr<const storage::Commit> merge_base;
	status = MergeSetSync(handler, std::move(common_ancestors), &merge_base);
	if (status != storage::Status::OK) {
	return status;
	}
	if (!merge_base) {
	// A commit was made, resume when notified of it.
	return storage::Status::OK;
	}

	has_merged_ = true;

	storage::Status merge_status;
	auto sync_call_status = coroutine::SyncCall(
	handler,
	[this, left = std::move(left), right = std::move(right),
	merge_base = std::move(merge_base)](
	fit::function<void(storage::Status)> callback) mutable {
	strategy_->Merge(storage_, page_manager_, std::move(left),
	std::move(right), std::move(merge_base),
	TRACE_CALLBACK(std::move(callback), "ledger",
	"merge_strategy_merge"));
	},
	&merge_status);
	if (sync_call_status == coroutine::ContinuationStatus::INTERRUPTED) {
	return storage::Status::INTERRUPTED;
	}
	if (merge_status != storage::Status::OK) {
	merge_candidates_->OnMergeError(merge_status);
	return storage::Status::ILLEGAL_STATE;
	}
	merge_candidates_->OnMergeSuccess();
	return storage::Status::OK;
	}

	} // namespace ledger