From f4d3e5cdcffa2b1f55604586f52a157fbbbf3304 Mon Sep 17 00:00:00 2001
From: Tomasz Grabiec <tgrabiec@scylladb.com>
Date: Tue, 18 Jul 2017 12:53:36 +0200
Subject: [PATCH] Merge "Drop mutations that raced with truncate" from Duarte

Instead of retrying, just drop mutations that raced with a truncate.

* git@github.com:duarten/scylla.git truncate-reorder/v1:
  database: Rename replay_position_reordered_exception
  database: Drop mutations that raced with truncate

(cherry picked from commit 63caa58b70875f5ff5f8463f938e973789f519f4)
---
 database.cc | 41 ++++++++++++++---------------------------
 database.hh |  2 +-
 2 files changed, 15 insertions(+), 28 deletions(-)

diff --git a/database.cc b/database.cc
index 85947a2050..fee58a16a1 100644
--- a/database.cc
+++ b/database.cc
@@ -3055,7 +3055,7 @@ void column_family::apply_streaming_big_mutation(schema_ptr m_schema, utils::UUI
 void
 column_family::check_valid_rp(const db::replay_position& rp) const {
     if (rp != db::replay_position() && rp < _lowest_allowed_rp) {
-        throw replay_position_reordered_exception();
+        throw mutation_reordered_with_truncate_exception();
     }
 }
 
@@ -3201,25 +3201,24 @@ future<mutation> database::apply_counter_update(schema_ptr s, const frozen_mutat
     }
 }
 
+static future<> maybe_handle_reorder(std::exception_ptr exp) {
+    try {
+        std::rethrow_exception(exp);
+        return make_exception_future(exp);
+    } catch (mutation_reordered_with_truncate_exception&) {
+        // This mutation raced with a truncate, so we can just drop it.
+        dblog.debug("replay_position reordering detected");
+        return make_ready_future<>();
+    }
+}
+
 future<> database::apply_with_commitlog(column_family& cf, const mutation& m, timeout_clock::time_point timeout) {
     if (cf.commitlog() != nullptr) {
         return do_with(freeze(m), [this, &m, &cf, timeout] (frozen_mutation& fm) {
             commitlog_entry_writer cew(m.schema(), fm);
             return cf.commitlog()->add_entry(m.schema()->id(), cew, timeout);
         }).then([this, &m, &cf, timeout] (db::rp_handle h) {
-            return apply_in_memory(m, cf, std::move(h), timeout).handle_exception([this, &cf, &m, timeout] (auto ep) {
-                try {
-                    std::rethrow_exception(ep);
-                } catch (replay_position_reordered_exception&) {
-                    // expensive, but we're assuming this is super rare.
-                    // if we failed to apply the mutation due to future re-ordering
-                    // (which should be the ever only reason for rp mismatch in CF)
-                    // let's just try again, add the mutation to the CL once more,
-                    // and assume success in inevitable eventually.
-                    dblog.debug("replay_position reordering detected");
-                    return this->apply_with_commitlog(cf, m, timeout);
-                }
-            });
+            return apply_in_memory(m, cf, std::move(h), timeout).handle_exception(maybe_handle_reorder);
         });
     }
     return apply_in_memory(m, cf, {}, timeout);
@@ -3230,19 +3229,7 @@ future<> database::apply_with_commitlog(schema_ptr s, column_family& cf, utils::
     if (cl != nullptr) {
         commitlog_entry_writer cew(s, m);
         return cf.commitlog()->add_entry(uuid, cew, timeout).then([&m, this, s, timeout, cl](db::rp_handle h) {
-            return this->apply_in_memory(m, s, std::move(h), timeout).handle_exception([this, s, &m, timeout] (auto ep) {
-                try {
-                    std::rethrow_exception(ep);
-                } catch (replay_position_reordered_exception&) {
-                    // expensive, but we're assuming this is super rare.
-                    // if we failed to apply the mutation due to future re-ordering
-                    // (which should be the ever only reason for rp mismatch in CF)
-                    // let's just try again, add the mutation to the CL once more,
-                    // and assume success in inevitable eventually.
-                    dblog.debug("replay_position reordering detected");
-                    return this->apply(s, m, timeout);
-                }
-            });
+            return this->apply_in_memory(m, s, std::move(h), timeout).handle_exception(maybe_handle_reorder);
         });
     }
     return apply_in_memory(m, std::move(s), {}, timeout);
diff --git a/database.hh b/database.hh
index ed437bbaf9..bfc1f79a20 100644
--- a/database.hh
+++ b/database.hh
@@ -116,7 +116,7 @@ void make(database& db, bool durable, bool volatile_testing_only);
 }
 }
 
-class replay_position_reordered_exception : public std::exception {};
+class mutation_reordered_with_truncate_exception : public std::exception {};
 
 using shared_memtable = lw_shared_ptr<memtable>;
 class memtable_list;