Auto-release acquired stories on Disconnect (#642)

Author: iameneko

client/cpp/include/chronolog_client.h

@@ -111,10 +111,9 @@ class ChronologClientImpl;
 
 // Top-level Chronolog Client. Implementation details in ChronologClientImpl.
 //
-// NOTE: ReleaseStory() must be called for all acquired stories before Disconnect();
-// otherwise Disconnect() returns CL_ERR_ACQUIRED (-4) and the client record is not removed.
-// TODO: Visor should auto-release acquired stories on Disconnect() — removing this requirement.
-// The destructor calls Disconnect() as best-effort; the Client is always safe to delete.
+// Disconnect() auto-releases any stories still acquired by the client so the
+// caller never has to walk its own acquisitions to clean up. The destructor
+// calls Disconnect() as best-effort; the Client is always safe to delete.
 class Client
 {
 public:

src/chrono-visor/include/ChronicleMetaDirectory.h

@@ -43,7 +43,14 @@ class ChronicleMetaDirectory
     int release_story(chronolog::ClientId const& client_id,
                       const std::string& chronicle_name,
                       const std::string& story_name,
-                      StoryId&);
+                      StoryId&,
+                      bool& was_last_acquirer);
+
+    // Release every story currently acquired by client_id. Returns the StoryIds
+    // of stories whose last acquirer was this client, so the caller can notify
+    // the recording groups to stop only those.
+    int release_all_acquired_stories(chronolog::ClientId const& client_id,
+                                     std::vector<StoryId>& released_with_no_acquirers_left);
 
     int get_chronicle_attr(std::string const& name, const std::string& key, std::string& value);
 

src/chrono-visor/include/ClientRegistryManager.h

@@ -31,6 +31,9 @@ class ClientRegistryManager
 
     ClientInfo* get_client_info(chronolog::ClientId const& client_id);
 
+    int get_acquired_stories_snapshot(chronolog::ClientId const& client_id,
+                                      std::vector<std::pair<uint64_t, Story*>>& snapshot);
+
     int add_story_acquisition(chronolog::ClientId const& client_id, uint64_t& sid, Story* pStory);
 
     int remove_story_acquisition(chronolog::ClientId const& client_id, uint64_t& sid);

src/chrono-visor/src/ChronicleMetaDirectory.cpp

@@ -262,12 +262,13 @@ int ChronicleMetaDirectory::acquire_story(chl::ClientId const& client_id,
  *         chronolog::CL_ERR_NOT_EXIST if the Chronicle does not exist \n
  *         chronolog::CL_ERR_UNKNOWN otherwise
  */
-//TO_DO return acquisition_count after the story has been released
 int ChronicleMetaDirectory::release_story(chl::ClientId const& client_id,
                                           const std::string& chronicle_name,
                                           const std::string& story_name,
-                                          StoryId& story_id)
+                                          StoryId& story_id,
+                                          bool& was_last_acquirer)
 {
+    was_last_acquirer = false;
     LOG_DEBUG("[ChronicleMetaDirectory] ClientID={} releasing StoryName={} in ChronicleName={}",
               client_id,
               story_name.c_str(),
@@ -294,8 +295,11 @@ int ChronicleMetaDirectory::release_story(chl::ClientId const& client_id,
         if(acquirerMapRecord != acquirerMap.end())
         {
             /* All checks passed and entry found, manipulate metadata */
-            /* Decrement AcquisitionCount */
-            pStory->decrementAcquisitionCount();
+            /* Decrement AcquisitionCount; the post-decrement value tells the caller
+             * whether other clients still hold the story. The recording group must
+             * only be told to stop when the last acquirer releases. */
+            uint64_t remaining_acquirers = pStory->decrementAcquisitionCount();
+            was_last_acquirer = (remaining_acquirers == 0);
             story_id = pStory->getSid();
             /* Remove this client from acquirerClientList of the Story */
             pStory->removeAcquirerClient(client_id);
@@ -317,6 +321,76 @@ int ChronicleMetaDirectory::release_story(chl::ClientId const& client_id,
     return ret;
 }
 
+int ChronicleMetaDirectory::release_all_acquired_stories(chl::ClientId const& client_id,
+                                                         std::vector<StoryId>& released_with_no_acquirers_left)
+{
+    released_with_no_acquirers_left.clear();
+    if(clientRegistryManager_ == nullptr)
+    {
+        return chronolog::CL_ERR_UNKNOWN;
+    }
+
+    // Snapshot under the ClientRegistryManager's mutex; release_story below
+    // mutates acquiredStoryList_ via the manager and an unsynchronized walk
+    // here would race with concurrent Acquire/Release on this client.
+    std::vector<std::pair<uint64_t, Story*>> snapshot;
+    int snapshot_ret = clientRegistryManager_->get_acquired_stories_snapshot(client_id, snapshot);
+    if(snapshot_ret == chronolog::CL_ERR_NOT_EXIST)
+    {
+        // Nothing to release if the client has no record (e.g. already disconnected).
+        return chronolog::CL_SUCCESS;
+    }
+    if(snapshot_ret != chronolog::CL_SUCCESS)
+    {
+        return snapshot_ret;
+    }
+    released_with_no_acquirers_left.reserve(snapshot.size());
+
+    for(auto& [sid, pStory]: snapshot)
+    {
+        if(pStory == nullptr)
+        {
+            continue;
+        }
+
+        uint64_t cid = pStory->getCid();
+        std::string const story_name = pStory->getName();
+
+        std::string chronicle_name;
+        {
+            std::lock_guard<std::mutex> chronicleMapLock(g_chronicleMetaDirectoryMutex_);
+            auto chronicleRecord = chronicleMap_->find(cid);
+            if(chronicleRecord == chronicleMap_->end())
+            {
+                continue;
+            }
+            chronicle_name = chronicleRecord->second->getName();
+        }
+
+        StoryId released_id{0};
+        bool was_last_acquirer = false;
+        int ret = release_story(client_id, chronicle_name, story_name, released_id, was_last_acquirer);
+        if(ret == chronolog::CL_SUCCESS)
+        {
+            // Only report ids whose last acquirer was this client. If other
+            // clients still hold the story, the recording group must keep going.
+            if(was_last_acquirer)
+            {
+                released_with_no_acquirers_left.push_back(released_id);
+            }
+        }
+        else
+        {
+            LOG_WARNING("[ChronicleMetaDirectory] Failed to auto-release StoryName={} for ClientID={}: rc={}",
+                        story_name.c_str(),
+                        client_id,
+                        ret);
+        }
+    }
+
+    return chronolog::CL_SUCCESS;
+}
+
 int ChronicleMetaDirectory::get_chronicle_attr(std::string const& name, const std::string& key, std::string& value)
 {
     LOG_DEBUG("[ChronicleMetaDirectory] Getting attributes Key={} from ChronicleName={}", key.c_str(), name.c_str());

src/chrono-visor/src/ClientRegistryManager.cpp

@@ -47,6 +47,23 @@ ClientInfo* ClientRegistryManager::get_client_info(chl::ClientId const& client_i
 }
 //////////////////////
 
+int ClientRegistryManager::get_acquired_stories_snapshot(chl::ClientId const& client_id,
+                                                         std::vector<std::pair<uint64_t, Story*>>& snapshot)
+{
+    snapshot.clear();
+    std::lock_guard<std::mutex> clientRegistryLock(g_clientRegistryMutex_);
+    auto clientRegistryRecord = clientRegistry_->find(client_id);
+    if(clientRegistryRecord == clientRegistry_->end())
+    {
+        return chronolog::CL_ERR_NOT_EXIST;
+    }
+    auto const& acquiredStoryList = clientRegistryRecord->second.acquiredStoryList_;
+    snapshot.reserve(acquiredStoryList.size());
+    snapshot.assign(acquiredStoryList.begin(), acquiredStoryList.end());
+    return chronolog::CL_SUCCESS;
+}
+//////////////////////
+
 int ClientRegistryManager::add_story_acquisition(chl::ClientId const& client_id, uint64_t& sid, Story* pStory)
 {
     std::lock_guard<std::mutex> clientRegistryLock(g_clientRegistryMutex_);

src/chrono-visor/src/VisorClientPortal.cpp

@@ -137,6 +137,25 @@ int chronolog::VisorClientPortal::ClientConnect(uint32_t client_euid,
 int chronolog::VisorClientPortal::ClientDisconnect(chronolog::ClientId const& client_id)
 {
     LOG_INFO("Client Disconnected. ClientID={}", client_id);
+
+    // Auto-release any stories the client is still holding so the client
+    // record can always be cleaned up. Without this, remove_client_record
+    // refuses to remove a client that has acquired stories and the client
+    // would have to call ReleaseStory for every acquisition before Disconnect.
+    //
+    // release_all_acquired_stories reports back only the stories whose last
+    // acquirer was this client; stories still held by other clients keep
+    // recording.
+    std::vector<StoryId> released_with_no_acquirers_left;
+    chronicleMetaDirectory.release_all_acquired_stories(client_id, released_with_no_acquirers_left);
+    if(theKeeperRegistry != nullptr)
+    {
+        for(StoryId const& released_id: released_with_no_acquirers_left)
+        {
+            theKeeperRegistry->notifyRecordingGroupOfStoryRecordingStop(released_id);
+        }
+    }
+
     return clientManager.remove_client_record(client_id);
 }
 
@@ -271,7 +290,8 @@ chl::AcquireStoryResponseMsg chronolog::VisorClientPortal::AcquireStory(chl::Cli
                                                                                              player))
     {
         // RPC notification to the keepers might have failed, release the newly acquired story
-        chronicleMetaDirectory.release_story(client_id, chronicle_name, story_name, story_id);
+        bool was_last_acquirer = false;
+        chronicleMetaDirectory.release_story(client_id, chronicle_name, story_name, story_id, was_last_acquirer);
         //we do know that there's no need notify keepers of the story ending in this case as it hasn't started...
         return chronolog::AcquireStoryResponseMsg(chronolog::CL_ERR_NO_KEEPERS, story_id, empty_keeper_service_ids);
     }
@@ -304,13 +324,19 @@ int chronolog::VisorClientPortal::ReleaseStory(chl::ClientId const& client_id,
     }
 
     StoryId story_id(0);
-    auto return_code = chronicleMetaDirectory.release_story(client_id, chronicle_name, story_name, story_id);
+    bool was_last_acquirer = false;
+    auto return_code =
+            chronicleMetaDirectory.release_story(client_id, chronicle_name, story_name, story_id, was_last_acquirer);
     if(chronolog::CL_SUCCESS != return_code)
     {
         return return_code;
     }
 
-    theKeeperRegistry->notifyRecordingGroupOfStoryRecordingStop(story_id);
+    // Only stop the recording group if no other client still holds the story.
+    if(was_last_acquirer)
+    {
+        theKeeperRegistry->notifyRecordingGroupOfStoryRecordingStop(story_id);
+    }
 
     return chronolog::CL_SUCCESS;
 }

tests/integration/client/client_metadata_rpc_test.cpp

@@ -130,7 +130,7 @@ int main(int argc, char** argv)
         }
 
         ret = client.Disconnect();
-        assert(ret == chronolog::CL_ERR_NO_KEEPERS || ret == chronolog::CL_ERR_ACQUIRED);
+        assert(ret == chronolog::CL_SUCCESS || ret == chronolog::CL_ERR_NO_KEEPERS);
 
         t1 = std::chrono::steady_clock::now();
         std::vector<std::string> stories_names_retrieved;