Verify session existence before update in ReactiveRedisOperationsSessionRepository

Currently, ReactiveRedisOperationsSessionRepository#save does not ensure session's existence before executing update. This can result in an invalid session record in Redis, since write use only delta, and in turn to error while retrieving the invalid session record.

This commit adds check for session existence if session is being updated.

Closes gh-1111

spring-session-data-redis/src/integration-test/java/org/springframework/session/data/redis/ReactiveRedisOperationsSessionRepositoryITests.java

@@ -16,6 +16,8 @@
 
 package org.springframework.session.data.redis;
 
+import java.time.Instant;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -191,6 +193,28 @@ public class ReactiveRedisOperationsSessionRepositoryITests extends AbstractRedi
 		assertThat(this.repository.findById(originalId).block()).isNull();
 	}
 
+	// gh-1111
+	@Test
+	public void changeSessionSaveOldSessionInstance() {
+		ReactiveRedisOperationsSessionRepository.RedisSession toSave = this.repository
+				.createSession().block();
+		String sessionId = toSave.getId();
+
+		this.repository.save(toSave).block();
+
+		ReactiveRedisOperationsSessionRepository.RedisSession session = this.repository
+				.findById(sessionId).block();
+		session.changeSessionId();
+		session.setLastAccessedTime(Instant.now());
+		this.repository.save(session).block();
+
+		toSave.setLastAccessedTime(Instant.now());
+		this.repository.save(toSave).block();
+
+		assertThat(this.repository.findById(sessionId).block()).isNull();
+		assertThat(this.repository.findById(session.getId()).block()).isNotNull();
+	}
+
 	@Configuration
 	@EnableRedisWebSession
 	static class Config extends BaseConfig {

spring-session-data-redis/src/main/java/org/springframework/session/data/redis/ReactiveRedisOperationsSessionRepository.java

@@ -143,24 +143,39 @@ public class ReactiveRedisOperationsSessionRepository implements
 
 	@Override
 	public Mono<Void> save(RedisSession session) {
-		return session.saveDelta().and((s) -> {
-			if (session.isNew) {
-				session.setNew(false);
-			}
-
-			s.onComplete();
-		});
+		Mono<Void> result = session.saveChangeSessionId().and(session.saveDelta())
+				.and((s) -> {
+					session.isNew = false;
+					s.onComplete();
+				});
+		if (session.isNew) {
+			return result;
+		}
+		else if (session.hasChangedSessionId()) {
+			String sessionKey = getSessionKey(session.originalSessionId);
+			return this.sessionRedisOperations.hasKey(sessionKey)
+					.flatMap((exists) -> exists ? result : Mono.empty());
+		}
+		else {
+			String sessionKey = getSessionKey(session.getId());
+			return this.sessionRedisOperations.hasKey(sessionKey)
+					.flatMap((exists) -> exists ? result : Mono.empty());
+		}
 	}
 
 	@Override
 	public Mono<RedisSession> findById(String id) {
 		String sessionKey = getSessionKey(id);
 
+		// @formatter:off
 		return this.sessionRedisOperations.opsForHash().entries(sessionKey)
 				.collectMap((e) -> e.getKey().toString(), Map.Entry::getValue)
-				.filter((map) -> !map.isEmpty()).map(new SessionMapper(id))
-				.filter((session) -> !session.isExpired()).map(RedisSession::new)
+				.filter((map) -> !map.isEmpty())
+				.map(new SessionMapper(id))
+				.filter((session) -> !session.isExpired())
+				.map(RedisSession::new)
 				.switchIfEmpty(Mono.defer(() -> deleteById(id).then(Mono.empty())));
+		// @formatter:on
 	}
 
 	@Override
@@ -285,12 +300,8 @@ public class ReactiveRedisOperationsSessionRepository implements
 			return this.cached.isExpired();
 		}
 
-		public void setNew(boolean isNew) {
-			this.isNew = isNew;
-		}
-
-		public boolean isNew() {
-			return this.isNew;
+		private boolean hasChangedSessionId() {
+			return !getId().equals(this.originalSessionId);
 		}
 
 		private void flushImmediateIfNecessary() {
@@ -305,38 +316,35 @@ public class ReactiveRedisOperationsSessionRepository implements
 		}
 
 		private Mono<Void> saveDelta() {
-			String sessionId = getId();
-			Mono<Void> changeSessionId = saveChangeSessionId(sessionId);
-
 			if (this.delta.isEmpty()) {
-				return changeSessionId.and(Mono.empty());
+				return Mono.empty();
 			}
 
-			String sessionKey = getSessionKey(sessionId);
-
+			String sessionKey = getSessionKey(getId());
 			Mono<Boolean> update = ReactiveRedisOperationsSessionRepository.this.sessionRedisOperations
 					.opsForHash().putAll(sessionKey, this.delta);
-
 			Mono<Boolean> setTtl = ReactiveRedisOperationsSessionRepository.this.sessionRedisOperations
 					.expire(sessionKey, getMaxInactiveInterval());
 
-			return changeSessionId.and(update).and(setTtl).and((s) -> {
+			return update.and(setTtl).and((s) -> {
 				this.delta.clear();
 				s.onComplete();
 			}).then();
 		}
 
-		private Mono<Void> saveChangeSessionId(String sessionId) {
-			if (sessionId.equals(this.originalSessionId)) {
+		private Mono<Void> saveChangeSessionId() {
+			if (!hasChangedSessionId()) {
 				return Mono.empty();
 			}
 
+			String sessionId = getId();
+
 			Publisher<Void> replaceSessionId = (s) -> {
 				this.originalSessionId = sessionId;
 				s.onComplete();
 			};
 
-			if (isNew()) {
+			if (this.isNew) {
 				return Mono.from(replaceSessionId);
 			}
 			else {

spring-session-data-redis/src/test/java/org/springframework/session/data/redis/ReactiveRedisOperationsSessionRepositoryTests.java

@@ -183,6 +183,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 
 	@Test
 	public void saveSessionNothingChanged() {
+		given(this.redisOperations.hasKey(anyString())).willReturn(Mono.just(true));
 		given(this.redisOperations.expire(anyString(), any()))
 				.willReturn(Mono.just(true));
 
@@ -191,12 +192,14 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 
 		StepVerifier.create(this.repository.save(session)).verifyComplete();
 
+		verify(this.redisOperations).hasKey(anyString());
 		verifyZeroInteractions(this.redisOperations);
 		verifyZeroInteractions(this.hashOperations);
 	}
 
 	@Test
 	public void saveLastAccessChanged() {
+		given(this.redisOperations.hasKey(anyString())).willReturn(Mono.just(true));
 		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
 		given(this.hashOperations.putAll(anyString(), any())).willReturn(Mono.just(true));
 		given(this.redisOperations.expire(anyString(), any()))
@@ -206,6 +209,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 		session.setLastAccessedTime(Instant.ofEpochMilli(12345678L));
 		Mono.just(session).subscribe(this.repository::save);
 
+		verify(this.redisOperations).hasKey(anyString());
 		verify(this.redisOperations).opsForHash();
 		verify(this.hashOperations).putAll(anyString(), this.delta.capture());
 		verify(this.redisOperations).expire(anyString(), any());
@@ -219,6 +223,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 
 	@Test
 	public void saveSetAttribute() {
+		given(this.redisOperations.hasKey(anyString())).willReturn(Mono.just(true));
 		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
 		given(this.hashOperations.putAll(anyString(), any())).willReturn(Mono.just(true));
 		given(this.redisOperations.expire(anyString(), any()))
@@ -229,6 +234,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 		session.setAttribute(attrName, "attrValue");
 		Mono.just(session).subscribe(this.repository::save);
 
+		verify(this.redisOperations).hasKey(anyString());
 		verify(this.redisOperations).opsForHash();
 		verify(this.hashOperations).putAll(anyString(), this.delta.capture());
 		verify(this.redisOperations).expire(anyString(), any());
@@ -242,6 +248,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 
 	@Test
 	public void saveRemoveAttribute() {
+		given(this.redisOperations.hasKey(anyString())).willReturn(Mono.just(true));
 		given(this.redisOperations.opsForHash()).willReturn(this.hashOperations);
 		given(this.hashOperations.putAll(anyString(), any())).willReturn(Mono.just(true));
 		given(this.redisOperations.expire(anyString(), any()))
@@ -252,6 +259,7 @@ public class ReactiveRedisOperationsSessionRepositoryTests {
 		session.removeAttribute(attrName);
 		Mono.just(session).subscribe(this.repository::save);
 
+		verify(this.redisOperations).hasKey(anyString());
 		verify(this.redisOperations).opsForHash();
 		verify(this.hashOperations).putAll(anyString(), this.delta.capture());
 		verify(this.redisOperations).expire(anyString(), any());