Merge "Fix TTL serialization breakage" from Avi

ommit 93270dd changed gc_clock to be 64-bit, to fix the Y2038
problem. While 64-bit tombstone::deletion_time is serialized in a
compatible way, TTLs (gc_clock::duration) were not.

This patchset reverts TTL serialization to the 32-bit serialization
format, and also allows opting-in to the 64-bit format in case a
cluster was installed with the broken code. Only Scylla 3.1.0 is
vulnerable.

Fixes #4855

Tests: unit (dev)
(cherry picked from commit e621db591e)

db/config.hh

@@ -757,6 +757,7 @@ public:
         " It is not enough to have ever since upgraded to newer versions of Cassandra. If you EVER used a version earlier than 2.1 in the cluster where these SSTables come from, DO NOT TURN ON THIS OPTION! You will corrupt your data. You have been warned.") \
     val(enable_shard_aware_drivers, bool, true, Used, "Enable native transport drivers to use connection-per-shard for better performance") \
     val(abort_on_internal_error, bool, false, Used, "Abort the server instead of throwing exception when internal invariants are violated.") \
+    val(enable_3_1_0_compatibility_mode, bool, false, Used, "Set to true if the cluster was initially installed from 3.1.0. If it was upgraded from an earlier version, or installed from a later version, leave this set to false. This adjusts the communication protocol to work around a bug in Scylla 3.1.0") \
     /* done! */
 
 #define _make_value_member(name, type, deflt, status, desc, ...)    \

gc_clock.hh

@@ -86,3 +86,37 @@ struct appending_hash<gc_clock::time_point> {
         }
     }
 };
+
+
+namespace ser {
+
+// Forward-declaration - defined in serializer.hh, to avoid including it here.
+
+template <typename Output>
+void serialize_gc_clock_duration_value(Output& out, int64_t value);
+
+template <typename Input>
+int64_t deserialize_gc_clock_duration_value(Input& in);
+
+template <typename T>
+struct serializer;
+
+template <>
+struct serializer<gc_clock::duration> {
+    template <typename Input>
+    static gc_clock::duration read(Input& in) {
+        return gc_clock::duration(deserialize_gc_clock_duration_value(in));
+    }
+
+    template <typename Output>
+    static void write(Output& out, gc_clock::duration d) {
+        serialize_gc_clock_duration_value(out, d.count());
+    }
+
+    template <typename Input>
+    static void skip(Input& in) {
+        read(in);
+    }
+};
+
+}

main.cc

@@ -69,6 +69,7 @@
 #include "sstables/sstables.hh"
 #include "gms/feature_service.hh"
 #include "distributed_loader.hh"
+#include "serializer.hh"
 
 namespace fs = std::filesystem;
 
@@ -408,6 +409,11 @@ int main(int ac, char** av) {
             read_config(opts, *cfg).get();
             configurable::init_all(opts, *cfg, *ext).get();
 
+            // We're writing to a non-atomic variable here. But bool writes are atomic
+            // in all supported architectures, and some broadcast or other below
+            // will apply the required memory barriers anyway.
+            ser::gc_clock_using_3_1_0_serialization = cfg->enable_3_1_0_compatibility_mode();
+
             logalloc::prime_segment_pool(memory::stats().total_memory(), memory::min_free_memory()).get();
             logging::apply_settings(cfg->logging_settings(opts));
 

serializer.hh

@@ -282,6 +282,38 @@ struct normalize<bytes_ostream> {
 template <typename T, typename U>
 struct is_equivalent : std::is_same<typename normalize<std::remove_const_t<std::remove_reference_t<T>>>::type, typename normalize<std::remove_const_t <std::remove_reference_t<U>>>::type> {
 };
+
+// gc_clock duration values were serialized as 32-bit prior to 3.1, and
+// are serialized as 64-bit in 3.1.0.
+//
+// TTL values are capped to 20 years, which fits into 32 bits, so
+// truncation is not a concern.
+
+inline bool gc_clock_using_3_1_0_serialization = false;
+
+template <typename Output>
+void
+serialize_gc_clock_duration_value(Output& out, int64_t v) {
+    if (!gc_clock_using_3_1_0_serialization) {
+        // This should have been caught by the CQL layer, so this is just
+        // for extra safety.
+        assert(int32_t(v) == v);
+        serializer<int32_t>::write(out, v);
+    } else {
+        serializer<int64_t>::write(out, v);
+    }
+}
+
+template <typename Input>
+int64_t
+deserialize_gc_clock_duration_value(Input& in) {
+    if (!gc_clock_using_3_1_0_serialization) {
+        return serializer<int32_t>::read(in);
+    } else {
+        return serializer<int64_t>::read(in);
+    }
+}
+
 }
 
 /*