feat(setfamily): Add SADDEX command (dragonflydb#348)

Add SADDEX <key> <seconds> member member ....
Provides expiry semantics for set members with seconds resolution.

Important things to note:
1. The expiry is passive only, so if nobody touches that set then its members are being kept.
2. SCARD provides an upper bound estimation of set size for sets holding the expiring members.
   The reason for this is because SCARD returns a cached size and does not go over all members
   to check whether they expired. For regular sets it's exact, of course.

Fixes dragonflydb#335

src/core/dense_set.cc

@@ -492,21 +492,35 @@ uint32_t DenseSet::Scan(uint32_t cursor, const ItemCb& cb) const {
 
   uint32_t entries_idx = cursor >> (32 - capacity_log_);
 
+  auto& entries = const_cast<DenseSet*>(this)->entries_;
+
   // skip empty entries
-  while (entries_idx < entries_.size() && entries_[entries_idx].IsEmpty()) {
-    ++entries_idx;
-  }
+  do {
+    while (entries_idx < entries_.size() && entries_[entries_idx].IsEmpty()) {
+      ++entries_idx;
+    }
 
-  if (entries_idx == entries_.size()) {
-    return 0;
-  }
+    if (entries_idx == entries_.size()) {
+      return 0;
+    }
 
-  const DensePtr* curr = &entries_[entries_idx];
+    ExpireIfNeeded(nullptr, &entries[entries_idx]);
+  } while (entries_[entries_idx].IsEmpty());
+
+  DensePtr* curr = &entries[entries_idx];
 
   // when scanning add all entries in a given chain
-  while (curr != nullptr && !curr->IsEmpty()) {
+  while (true) {
     cb(curr->GetObject());
-    curr = curr->Next();
+    if (!curr->IsLink())
+      break;
+
+    DensePtr* mcurr = const_cast<DensePtr*>(curr);
+
+    if (ExpireIfNeeded(mcurr, &mcurr->AsLink()->next) && !mcurr->IsLink()) {
+      break;
+    }
+    curr = &curr->AsLink()->next;
   }
 
   // move to the next index for the next scan and check if we are done
@@ -515,6 +529,14 @@ uint32_t DenseSet::Scan(uint32_t cursor, const ItemCb& cb) const {
     return 0;
   }
 
+  // In case of displacement, we want to fully cover the bucket we traversed, therefore
+  // we check if the bucket on the right belongs to the home bucket.
+  ExpireIfNeeded(nullptr, &entries[entries_idx]);
+
+  if (entries[entries_idx].GetDisplacedDirection() == 1) { // right of the home bucket
+    cb(entries[entries_idx].GetObject());
+  }
+
   return entries_idx << (32 - capacity_log_);
 }
 

src/core/string_set_test.cc

@@ -263,9 +263,8 @@ TEST_F(StringSetTest, IntOnly) {
 
   size_t expected_seen = 0;
   auto scan_callback = [&](const sds ptr) {
-    sds s = (sds)ptr;
-    string_view str{s, sdslen(s)};
-    EXPECT_FALSE(removed.count(string(str)));
+    string str{ptr, sdslen(ptr)};
+    EXPECT_FALSE(removed.count(str));
 
     if (numbers.count(atoi(str.data()))) {
       ++expected_seen;
@@ -281,7 +280,7 @@ TEST_F(StringSetTest, IntOnly) {
     ss_->Add(to_string(val));
   } while (cursor != 0);
 
-  EXPECT_TRUE(expected_seen + removed.size() == num_ints);
+  EXPECT_GE(expected_seen + removed.size(), num_ints);
 }
 
 TEST_F(StringSetTest, XtremeScanGrow) {