From 5268549710bf157cc10a46ade7f345a5ffbfe08d Mon Sep 17 00:00:00 2001
From: Mark Paluch <mpaluch@pivotal.io>
Date: Thu, 8 Feb 2018 14:08:50 +0100
Subject: [PATCH] DATAREDIS-673 - Prevent false positive cache hits while Redis
 keys expire.

We now prevent false positive cache hits after checking for key presence and the Redis key gets removed before fetching its value.
This can happen if we check positively for presence but the key expires (TTL) or gets removed before the GET call is able to fetch the value.

Previously, we returned a RedisCacheElement wrapping null without regard to whether the cache is able to store/return null values. We now inspect the returned value for presence to eagerly return a negative cache hit and to preserve cached null values.

Original Pull Request: #310
---
 .../data/redis/cache/RedisCache.java          | 23 +++++++++++++++----
 .../data/redis/cache/RedisCacheUnitTests.java | 19 ++++++++++++++-
 2 files changed, 36 insertions(+), 6 deletions(-)

diff --git a/src/main/java/org/springframework/data/redis/cache/RedisCache.java b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
index 0a5bce9ab..9d99f7332 100644
--- a/src/main/java/org/springframework/data/redis/cache/RedisCache.java
+++ b/src/main/java/org/springframework/data/redis/cache/RedisCache.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2017 the original author or authors.
+ * Copyright 2011-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -177,11 +177,18 @@ public class RedisCache extends AbstractValueAdaptingCache {
 			}
 		});
 
-		if (!exists.booleanValue()) {
+		if (!exists) {
 			return null;
 		}
 
-		return new RedisCacheElement(cacheKey, fromStoreValue(lookup(cacheKey)));
+		byte[] bytes = doLookup(cacheKey);
+
+		// safeguard if key gets deleted between EXISTS and GET calls.
+		if (bytes == null) {
+			return null;
+		}
+
+		return new RedisCacheElement(cacheKey, fromStoreValue(deserialize(bytes)));
 	}
 
 	/*
@@ -308,10 +315,14 @@ public class RedisCache extends AbstractValueAdaptingCache {
 	 */
 	@Override
 	protected Object lookup(Object key) {
+		return deserialize(doLookup(key));
+	}
 
+	private byte[] doLookup(Object key) {
+		
 		RedisCacheKey cacheKey = key instanceof RedisCacheKey ? (RedisCacheKey) key : getRedisCacheKey(key);
 
-		byte[] bytes = (byte[]) redisOperations.execute(new AbstractRedisCacheCallback<byte[]>(
+		return (byte[]) redisOperations.execute(new AbstractRedisCacheCallback<byte[]>(
 				new BinaryRedisCacheElement(new RedisCacheElement(cacheKey, null), cacheValueAccessor), cacheMetadata) {
 
 			@Override
@@ -319,7 +330,9 @@ public class RedisCache extends AbstractValueAdaptingCache {
 				return connection.get(element.getKeyBytes());
 			}
 		});
-
+	}
+	
+	private Object deserialize(byte[] bytes) {
 		return bytes == null ? null : cacheValueAccessor.deserializeIfNecessary(bytes);
 	}
 
diff --git a/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java b/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
index 1e18836a7..7d4b0d52c 100644
--- a/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/cache/RedisCacheUnitTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -313,6 +313,23 @@ public class RedisCacheUnitTests {
 		assertThat((String) cache.get(KEY, callableMock), equalTo(VALUE));
 		verifyZeroInteractions(callableMock);
 	}
+	
+	@Test // DATAREDIS-673
+	@SuppressWarnings("unchecked")
+	public void getWithCallableShouldReadValueFromCallableWhenKeyGetsDeletedInFlight() throws Exception {
+
+		cache = new RedisCache(CACHE_NAME, NO_PREFIX_BYTES, templateSpy, 0L);
+		Callable<Object> callableMock = mock(Callable.class);
+
+		Object result = new Object();
+		when(callableMock.call()).thenReturn(result);
+		when(connectionMock.exists(KEY_BYTES)).thenReturn(true);
+		when(connectionMock.get(KEY_BYTES)).thenReturn(null);
+
+		assertThat((String) cache.get(KEY, callableMock), equalTo(VALUE));
+		verify(callableMock).call();
+		verify(connectionMock, times(2)).get(KEY_BYTES);
+	}
 
 	@Test // DATAREDIS-468
 	public void noMultiExecForCluster() {