khepri_cluster: Acquire lock before doing a read/modify/write on ?PT_STORE_IDS

[Why]
The code was reading the persistent term, adding/removing a key and
writing the persistent term again.

When several stores are started and/or stopped in parallel, these
unprotected concurent read/modify/write operations could cause some
store properties to be lost in the process.

[How]
The code now acquires a local lock before proceeding with this
read/modify/write.

Author: dumbbell

src/khepri_cluster.erl

@@ -1468,11 +1468,17 @@ complete_ra_server_config(#{cluster_name := StoreId,
 
 remember_store(RaSystem, #{cluster_name := StoreId} = RaServerConfig) ->
     ?assert(maps:is_key(id, RaServerConfig)),
-    StoreIds = persistent_term:get(?PT_STORE_IDS, #{}),
-    Props = #{ra_system => RaSystem,
-              ra_server_config => RaServerConfig},
-    StoreIds1 = StoreIds#{StoreId => Props},
-    persistent_term:put(?PT_STORE_IDS, StoreIds1),
+    Lock = store_ids_lock(),
+    global:set_lock(Lock, [node()]),
+    try
+        StoreIds = persistent_term:get(?PT_STORE_IDS, #{}),
+        Props = #{ra_system => RaSystem,
+                  ra_server_config => RaServerConfig},
+        StoreIds1 = StoreIds#{StoreId => Props},
+        persistent_term:put(?PT_STORE_IDS, StoreIds1)
+    after
+            global:del_lock(Lock, [node()])
+    end,
     ok.
 
 -spec get_store_prop(StoreId, PropName) -> Ret when
@@ -1502,14 +1508,23 @@ get_store_prop(StoreId, PropName) ->
 
 forget_store(StoreId) ->
     ok = khepri_machine:clear_cache(StoreId),
-    StoreIds = persistent_term:get(?PT_STORE_IDS, #{}),
-    StoreIds1 = maps:remove(StoreId, StoreIds),
-    case maps:size(StoreIds1) of
-        0 -> _ = persistent_term:erase(?PT_STORE_IDS);
-        _ -> ok = persistent_term:put(?PT_STORE_IDS, StoreIds1)
+    Lock = store_ids_lock(),
+    global:set_lock(Lock, [node()]),
+    try
+        StoreIds = persistent_term:get(?PT_STORE_IDS, #{}),
+        StoreIds1 = maps:remove(StoreId, StoreIds),
+        case maps:size(StoreIds1) of
+            0 -> _ = persistent_term:erase(?PT_STORE_IDS);
+            _ -> ok = persistent_term:put(?PT_STORE_IDS, StoreIds1)
+        end
+    after
+            global:del_lock(Lock, [node()])
     end,
     ok.
 
+store_ids_lock() ->
+    {?PT_STORE_IDS, self()}.
+
 -spec get_store_ids() -> [StoreId] when
       StoreId :: khepri:store_id().
 %% @doc Returns the list of running stores.