Improve versioning for timeseries and attributes

thingsboard · Sep 16, 2024 · 8ea2230 · 8ea2230
1 parent 87e4a23
commit 8ea2230
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 4 deletions.
diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/VersionedCaffeineTbCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/VersionedCaffeineTbCache.java
@@ -50,7 +50,7 @@ private void doPut(K key, V value, Long version) {
         lock.lock();
         try {
             TbPair<Long, V> versionValuePair = doGet(key);
-            if (versionValuePair == null || version >= versionValuePair.getFirst()) {
+            if (versionValuePair == null || version > versionValuePair.getFirst()) {
                 failAllTransactionsByKey(key);
                 cache.put(key, wrapValue(value, version));
             }

diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/VersionedRedisTbCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/VersionedRedisTbCache.java
@@ -51,15 +51,15 @@ local function setNewValue()
 
             if currentVersionBytes and #currentVersionBytes == 8 then
                 local currentVersion = struct.unpack(">I8", currentVersionBytes)
-                if newVersion >= currentVersion then
+                if newVersion > currentVersion then
                     setNewValue()
                 end
             else
                 -- If the current value is absent or the current version is not found, set the new value
                 setNewValue()
             end
             """);
-    static final byte[] SET_VERSIONED_VALUE_SHA = StringRedisSerializer.UTF_8.serialize("5e02631c0d9df032e769e9b7d0fc20b74e7e104b");
+    static final byte[] SET_VERSIONED_VALUE_SHA = StringRedisSerializer.UTF_8.serialize("0453cb1814135b706b4198b09a09f43c9f67bbfe");
 
     public VersionedRedisTbCache(String cacheName, CacheSpecsMap cacheSpecsMap, RedisConnectionFactory connectionFactory, TBRedisCacheConfiguration configuration, TbRedisSerializer<K, V> valueSerializer) {
         super(cacheName, cacheSpecsMap, connectionFactory, configuration, valueSerializer);

diff --git a/common/cache/src/main/java/org/thingsboard/server/cache/VersionedTbCache.java b/common/cache/src/main/java/org/thingsboard/server/cache/VersionedTbCache.java
@@ -51,7 +51,7 @@ default V get(K key, Supplier<V> supplier, boolean putToCache) {
     void evict(K key, Long version);
 
     default Long getVersion(V value) {
-        return 0L;
+        return null;
         /* version on edge is static to update cache correctly
         if (value == null) {
             return 0L;

diff --git a/dao/src/main/java/org/thingsboard/server/dao/attributes/AttributeCaffeineCache.java b/dao/src/main/java/org/thingsboard/server/dao/attributes/AttributeCaffeineCache.java
@@ -30,4 +30,15 @@ public AttributeCaffeineCache(CacheManager cacheManager) {
         super(cacheManager, CacheConstants.ATTRIBUTES_CACHE);
     }
 
+    @Override
+    public Long getVersion(AttributeKvEntry value) {
+        if (value == null) {
+            return 0L;
+        } else if (value.getVersion() != null) {
+            return value.getVersion();
+        } else {
+            return null;
+        }
+    }
+
 }
diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/TsLatestRedisCache.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/TsLatestRedisCache.java
@@ -52,4 +52,16 @@ public TsKvEntry deserialize(TsLatestCacheKey key, byte[] bytes) throws Serializ
             }
         });
     }
+
+    @Override
+    public Long getVersion(TsKvEntry value) {
+        if (value == null) {
+            return 0L;
+        } else if (value.getVersion() != null) {
+            return value.getVersion();
+        } else {
+            return null;
+        }
+    }
+
 }