Use expire in case of deregistration event miss

Author: Croway

core/core-service-discovery/src/main/java/ai/wanaku/core/service/discovery/ValkeyRegistry.java

@@ -2,7 +2,6 @@
 
 import ai.wanaku.api.types.management.State;
 import io.quarkus.arc.lookup.LookupIfProperty;
-import io.quarkus.arc.properties.IfBuildProperty;
 import io.valkey.StreamEntryID;
 import io.valkey.params.XAddParams;
 import io.valkey.resps.StreamEntry;
@@ -26,6 +25,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.jboss.logging.Logger;
 
 /**
@@ -48,6 +49,15 @@ public class ValkeyRegistry implements ServiceRegistry {
     @Inject
     JedisPool jedisPool;
 
+    /**
+     * In case of a deregister event is not triggered (for example, kill -9),
+     * The entry in Valkey will be removed after expireTime.
+     *
+     * NB {wanaku.service.expire-time} > {wanaku.service.provider.registration.interval}
+     */
+    @ConfigProperty(name = "wanaku.service.expire-time", defaultValue = "60")
+    private int expireTime;
+
     /**
      * Registers a new service with the given configurations.
      *
@@ -67,13 +77,17 @@ public void register(ServiceTarget serviceTarget, Map<String, String> configurat
             LOG.infof("Service %s with target %s registered", serviceTarget.getService(), serviceTarget.toAddress());
 
             for (var entry : configurations.entrySet()) {
-                if (jedis.hget(serviceTarget.getService(), entry.getKey()) == null) {
+                if (!jedis.hexists(serviceTarget.getService(), entry.getKey())) {
                     LOG.infof("Registering configuration %s for service %s", entry.getKey(), serviceTarget.getService());
                     Configuration configuration = new Configuration();
                     configuration.setDescription(entry.getValue());
                     jedis.hset(serviceTarget.getService(), entry.getKey(), configuration.toJson());
                 }
             }
+
+            jedis.expire(serviceTarget.getService(), expireTime);
+            // Need to wait for https://github.com/valkey-io/valkey/issues/640
+            // jedis.expireMember(serviceKey, serviceTarget.getService(), EXPIRE_TIME);
         } catch (Exception e) {
             LOG.errorf(e, "Failed to register service %s: %s", serviceTarget.getService(), e.getMessage());
         }
@@ -176,7 +190,9 @@ public Map<String, Service> getEntries(ServiceType serviceType) {
             for (String key : services) {
                 Service service = newService(jedis, key);
 
-                entries.put(key, service);
+                if (service != null) {
+                    entries.put(key, service);
+                }
             }
 
         } catch (Exception e) {
@@ -223,8 +239,15 @@ private static Service newService(Jedis jedis, String key) {
      * @return A Service object representing the created service.
      */
     private static Service toService(Jedis jedis, String key, Set<String> configs) {
+        String address = jedis.hget(key, ReservedKeys.WANAKU_TARGET_ADDRESS);
+        if (address == null) {
+            return null;
+        }
+
         Service service = new Service();
 
+        service.setTarget(address);
+
         Map<String, Configuration> configurationMap = new HashMap<>();
 
         for (String config : configs) {
@@ -238,9 +261,6 @@ private static Service toService(Jedis jedis, String key, Set<String> configs) {
         configurations.setConfigurations(configurationMap);
         service.setConfigurations(configurations);
 
-        String address = jedis.hget(key, ReservedKeys.WANAKU_TARGET_ADDRESS);
-        service.setTarget(address);
-
         return service;
     }