Implement method for releasing leases

src/StateMachine.cc

@@ -1142,6 +1142,23 @@ rocksdb::Status StateMachine::lease_acquire(StagingArea &stagingArea, const std:
   return operation.finalize(value.size());
 }
 
+rocksdb::Status StateMachine::lease_release(StagingArea &stagingArea, const std::string &key) {
+  WriteOperation operation(stagingArea, key, KeyType::kLease);
+  if(!operation.valid()) return wrong_type();
+
+  KeyDescriptor &descriptor = operation.descriptor();
+
+  ExpirationEventLocator event(descriptor.getEndIndex(), key);
+  THROW_ON_ERROR(stagingArea.exists(event.toSlice()));
+  stagingArea.del(event.toSlice());
+
+  LeaseLocator leaseLocator(key);
+  THROW_ON_ERROR(stagingArea.exists(leaseLocator.toSlice()));
+  stagingArea.del(leaseLocator.toSlice());
+
+  return operation.finalize(0u);
+}
+
 rocksdb::Status StateMachine::llen(StagingArea &stagingArea, const std::string &key, size_t &len) {
   len = 0;
 
@@ -1648,3 +1665,7 @@ rocksdb::Status StateMachine::lhset(const std::string &key, const std::string &f
 rocksdb::Status StateMachine::lease_acquire(const std::string &key, const std::string &value, ClockValue clockUpdate, uint64_t duration, bool &acquired, LogIndex index) {
   CHAIN(index, lease_acquire, key, value, clockUpdate, duration, acquired);
 }
+
+rocksdb::Status StateMachine::lease_release(const std::string &key, LogIndex index) {
+  CHAIN(index, lease_release, key);
+}

src/StateMachine.hh

@@ -80,6 +80,7 @@ public:
 
   void advanceClock(StagingArea &stagingArea, ClockValue newValue);
   rocksdb::Status lease_acquire(StagingArea &stagingArea, const std::string &key, const std::string &value, ClockValue clockUpdate, uint64_t duration, bool &acquired);
+  rocksdb::Status lease_release(StagingArea &stagingArea, const std::string &key);
 
   //----------------------------------------------------------------------------
   // API for transactional reads. Can be part of a mixed read-write transaction.
@@ -146,6 +147,7 @@ public:
   void getClock(ClockValue &value);
   rocksdb::Status rawGetAllVersions(const std::string &key, std::vector<rocksdb::KeyVersion> &versions);
   rocksdb::Status lease_acquire(const std::string &key, const std::string &value, ClockValue clockUpdate, uint64_t duration, bool &acquired, LogIndex index = 0);
+  rocksdb::Status lease_release(const std::string &key, LogIndex index = 0);
 
   //----------------------------------------------------------------------------
   // Internal configuration, not exposed to users through 'KEYS' and friends.

test/state-machine.cc

@@ -763,6 +763,21 @@ TEST_F(State_Machine, Leases) {
     ASSERT_FALSE(iterator.valid());
   }
 
+  ASSERT_OK(stateMachine()->lease_release("my-lease-2"));
+  int64_t count = 0;
+  std::vector<std::string> keys = { "my-lease-2" };
+  ASSERT_OK(stateMachine()->exists(keys.begin(), keys.end(), count) );
+  ASSERT_EQ(count, 0);
+
+  {
+    StagingArea stagingArea(*stateMachine());
+    ExpirationEventIterator iterator(stagingArea);
+    ASSERT_TRUE(iterator.valid());
+    ASSERT_EQ(iterator.getDeadline(), 19u);
+    ASSERT_EQ(iterator.getRedisKey(), "my-lease");
+    iterator.next();
+    ASSERT_FALSE(iterator.valid());
+  }
 }
 
 static std::string sliceToString(const rocksdb::Slice &slice) {