MB-30487: [2/2] Enforce the collection max_ttl

This patch refactors the existing bucket max_ttl enforcement and embeds
it with the collection VB manifest.

Moving the bucket ttl enforcement to the manifest allows collection and
bucket TTL to be checked with one collections map lookup.

The patch updates the following operations to apply the defined limit

* add
* set
* replace
* GAT
* set-with-meta

In terms of actual enforcement, the design requires that any collection
max_ttl overrides any bucket max_ttl, this logic is in processExpiryTime
and tested in CollectionsExpiryLimitTest::operation_test

Change-Id: I15fb9f3d212b9f7027df1549d2023ff994b46dfa
Reviewed-on: http://review.couchbase.org/101230
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Ben Huddleston <ben.huddleston@couchbase.com>

daemon/mc_time.cc

@@ -99,22 +99,16 @@ rel_time_t mc_time_get_current_time(void) {
 }
 
 /// @return true if a + b would overflow rel_time_t
-static bool would_overflow(rel_time_t a, std::chrono::seconds b) {
-    return a > (std::numeric_limits<rel_time_t>::max() - b.count());
-}
-
-/**
- * @return true if a + b would overflow int64_t
- */
-static bool would_overflow(int64_t a, int64_t b) {
-    return a > (std::numeric_limits<int64_t>::max() - b);
+template <class A, class B>
+static bool would_overflow(A a, B b) {
+    return a > (std::numeric_limits<A>::max() - b);
 }
 
 // The above would_overflow(a. b) assumes rel_time_t is unsigned
 static_assert(std::is_unsigned<rel_time_t>::value,
               "would_overflow assumes rel_time_t is unsigned");
 
-rel_time_t mc_time_convert_to_real_time(rel_time_t t, cb::ExpiryLimit limit) {
+rel_time_t mc_time_convert_to_real_time(rel_time_t t) {
     rel_time_t rv = 0;
 
     int64_t epoch{memcached_epoch.load()};
@@ -123,25 +117,10 @@ rel_time_t mc_time_convert_to_real_time(rel_time_t t, cb::ExpiryLimit limit) {
     if (t > memcached_maximum_relative_time) { // t is absolute
 
         // Ensure overflow is predictable (we stay at max rel_time_t)
-        if (would_overflow(epoch, uptime)) {
-            return std::numeric_limits<rel_time_t>::max();
-        }
-        if (limit &&
-            would_overflow(gsl::narrow_cast<rel_time_t>(epoch + uptime),
-                           limit.get())) {
+        if (would_overflow<int64_t, int64_t>(epoch, uptime)) {
             return std::numeric_limits<rel_time_t>::max();
         }
 
-        if (limit && t > (epoch + uptime + limit.get().count())) {
-            if (would_overflow(gsl::narrow_cast<int64_t>(epoch + uptime),
-                               limit.get().count())) {
-                t = std::numeric_limits<rel_time_t>::max();
-            } else {
-                t = gsl::narrow<rel_time_t>((epoch + uptime) +
-                                            limit.get().count());
-            }
-        }
-
         /* if item expiration is at/before the server started, give it an
            expiration time of 1 second after the server started.
            (because 0 means don't expire).  without this, we'd
@@ -154,27 +133,34 @@ rel_time_t mc_time_convert_to_real_time(rel_time_t t, cb::ExpiryLimit limit) {
             rv = (rel_time_t)(t - epoch);
         }
     } else if (t != 0) { // t is relative
-        if (limit && t > limit.get().count()) {
-            t = gsl::narrow<rel_time_t>(limit.get().count());
-        }
-
         // Ensure overflow is predictable (we stay at max rel_time_t)
-        if (would_overflow(t, uptime)) {
+        if (would_overflow<rel_time_t, int64_t>(t, uptime)) {
             rv = std::numeric_limits<rel_time_t>::max();
         } else {
-            if (limit &&
-                would_overflow(gsl::narrow_cast<rel_time_t>(t + uptime),
-                               limit.get())) {
-                rv = std::numeric_limits<rel_time_t>::max();
-            } else {
-                rv = (rel_time_t)(t + uptime);
-            }
+            rv = (rel_time_t)(t + uptime);
         }
     }
 
     return rv;
 }
 
+time_t mc_time_limit_abstime(time_t t, std::chrono::seconds limit) {
+    auto upperbound = mc_time_convert_to_abs_time(mc_time_get_current_time());
+
+    if (would_overflow<time_t, std::chrono::seconds::rep>(upperbound,
+                                                          limit.count())) {
+        upperbound = std::numeric_limits<time_t>::max();
+    } else {
+        upperbound = upperbound + limit.count();
+    }
+
+    if (t == 0 || t > upperbound) {
+        t = upperbound;
+    }
+
+    return t;
+}
+
 /*
  * Convert the relative time to an absolute time (relative to EPOCH ;) )
  */

daemon/mc_time.h

@@ -45,11 +45,34 @@ void mc_time_clock_tick(void);
 time_t mc_time_convert_to_abs_time(const rel_time_t rel_time);
 
 /**
- * Convert a time stamp to an absolute time stamp.
+ * Convert a protocol encoded expiry time stamp to a relative time stamp
+ * (relative to the epoch time of memcached)
+ *
+ * Example 1: A relative expiry time (where t is less than 30days in seconds) of
+ * 1000s becomes epoch + 1000s
+ *
  * @param t a protocol expiry time-stamp
- * @oaram limit an optional limit, see server_api.h 'realtime' description
  */
-rel_time_t mc_time_convert_to_real_time(rel_time_t t, cb::ExpiryLimit limit);
+rel_time_t mc_time_convert_to_real_time(rel_time_t t);
+
+/**
+ * Apply a limit to an absolute timestamp (which represents an item's requested
+ * expiry time)
+ *
+ * For example if t represents 23:00 and the time we invoke this method is 22:00
+ * and the limit is 60s, then the returned value will be 22:01. The input of
+ * 23:00 exceeds 22:00 + 60s, so it is limited to 22:00 + 60s.
+ *
+ * If t == 0, then the returned value is now + limit
+ * If t < now, then the result is t, no limit needed.
+ * If t == 0 and now + limit overflows time_t, time_t::max is returned.
+ *
+ * @param t The expiry time to be limited, 0 means no expiry, 1 to time_t::max
+ *          are intepreted as the time absolute time of expiry
+ * @param limit The limit in seconds
+ * @return The expiry time after checking it against now + limit.
+ */
+time_t mc_time_limit_abstime(time_t t, std::chrono::seconds limit);
 
 #ifdef __cplusplus
 }

daemon/memcached.cc

@@ -1355,14 +1355,18 @@ struct ServerCoreApi : public ServerCoreIface {
         return mc_time_get_current_time();
     }
 
-    rel_time_t realtime(rel_time_t exptime, cb::ExpiryLimit limit) override {
-        return mc_time_convert_to_real_time(exptime, limit);
+    rel_time_t realtime(rel_time_t exptime) override {
+        return mc_time_convert_to_real_time(exptime);
     }
 
     time_t abstime(rel_time_t exptime) override {
         return mc_time_convert_to_abs_time(exptime);
     }
 
+    time_t limit_abstime(time_t t, std::chrono::seconds limit) override {
+        return mc_time_limit_abstime(t, limit);
+    }
+
     int parse_config(const char* str,
                      config_item* items,
                      FILE* error) override {

engines/default_engine/default_engine.cc

@@ -219,7 +219,7 @@ std::pair<cb::unique_item_ptr, item_info> default_engine::allocate_ex(
                     key.data(),
                     key.size(),
                     flags,
-                    server.core->realtime(exptime, cb::NoExpiryLimit),
+                    server.core->realtime(exptime),
                     (uint32_t)nbytes,
                     cookie,
                     datatype);
@@ -381,13 +381,12 @@ cb::EngineErrorItemPair default_engine::get_and_touch(
     }
 
     hash_item* it = nullptr;
-    auto ret = item_get_and_touch(
-            this,
-            cookie,
-            &it,
-            key.data(),
-            key.size(),
-            server.core->realtime(expiry_time, cb::NoExpiryLimit));
+    auto ret = item_get_and_touch(this,
+                                  cookie,
+                                  &it,
+                                  key.data(),
+                                  key.size(),
+                                  server.core->realtime(expiry_time));
 
     return cb::makeEngineErrorItemPair(
             cb::engine_errc(ret), reinterpret_cast<item*>(it), this);

engines/ep/src/collections/vbucket_manifest.cc

@@ -21,6 +21,7 @@
 #include "checkpoint_manager.h"
 #include "collections/manifest.h"
 #include "collections/vbucket_serialised_manifest_entry_generated.h"
+#include "ep_time.h"
 #include "item.h"
 #include "statwriter.h"
 #include "vbucket.h"
@@ -434,6 +435,33 @@ boost::optional<CollectionID> Manifest::shouldCompleteDeletion(
     return {};
 }
 
+void Manifest::processExpiryTime(const container::const_iterator entry,
+                                 Item& itm,
+                                 std::chrono::seconds bucketTtl) const {
+    itm.setExpTime(processExpiryTime(entry, itm.getExptime(), bucketTtl));
+}
+
+time_t Manifest::processExpiryTime(const container::const_iterator entry,
+                                   time_t t,
+                                   std::chrono::seconds bucketTtl) const {
+    std::chrono::seconds enforcedTtl{0};
+
+    if (bucketTtl.count()) {
+        enforcedTtl = bucketTtl;
+    }
+
+    // If the collection has a TTL, it gets used
+    if (entry->second.getMaxTtl()) {
+        enforcedTtl = entry->second.getMaxTtl().get();
+    }
+
+    // Note: A ttl value of 0 means no max_ttl
+    if (enforcedTtl.count()) {
+        t = ep_limit_abstime(t, enforcedTtl);
+    }
+    return t;
+}
+
 std::string Manifest::makeCollectionIdIntoString(CollectionID collection) {
     return std::string(reinterpret_cast<const char*>(&collection),
                        sizeof(CollectionID));

engines/ep/src/collections/vbucket_manifest.h

@@ -299,6 +299,34 @@ class Manifest {
             return manifest.shouldCompleteDeletion(key, bySeqno, itr);
         }
 
+        /**
+         * Check the Item's exptime against its collection config.
+         * If the collection defines a max_ttl and the Item has no expiry or
+         * an exptime which exceeds the max_ttl, set the expiry of the Item
+         * based on the collection max_ttl.
+         *
+         * @param itm The reference to the Item to check and change if needed
+         * @param bucketTtl the value of the bucket's max_ttl, 0 being none
+         */
+        void processExpiryTime(Item& itm,
+                               std::chrono::seconds bucketTtl) const {
+            manifest.processExpiryTime(itr, itm, bucketTtl);
+        }
+
+        /**
+         * t represents an absolute expiry time and this method returns t or a
+         * limited expiry time, based on the values of the bucketTtl and the
+         * collection's max_ttl.
+         *
+         * @param t an expiry time to process
+         * @param bucketTtl the value of the bucket's max_ttl, 0 being none
+         * @returns t or now + appropriate limit
+         */
+        time_t processExpiryTime(time_t t,
+                                 std::chrono::seconds bucketTtl) const {
+            return manifest.processExpiryTime(itr, t, bucketTtl);
+        }
+
         /**
          * Dump this VB::Manifest to std::cerr
          */
@@ -711,6 +739,16 @@ class Manifest {
             int64_t bySeqno,
             const container::const_iterator entry) const;
 
+    /// see comment on CachingReadHandle
+    void processExpiryTime(const container::const_iterator entry,
+                           Item& item,
+                           std::chrono::seconds bucketTtl) const;
+
+    /// see comment on CachingReadHandle
+    time_t processExpiryTime(const container::const_iterator entry,
+                             time_t t,
+                             std::chrono::seconds bucketTtl) const;
+
     /**
      * @returns true/false if $default exists
      */