Merge pull request #5396 from Sage-Bionetworks/release-562

Release 562

Author: marcomarasca

lib/lib-grid-db/src/main/java/org/sagebionetworks/grid/db/GridIndexDao.java

@@ -1,6 +1,7 @@
 package org.sagebionetworks.grid.db;
 
 import java.sql.Timestamp;
+import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
 
@@ -249,7 +250,17 @@ List<ArrayNode> getArrayNodesInOrder(String sessionIdString, Long replicaId, Log
 	 * @param setMethod
 	 * @return
 	 */
-	MessageChain createMessageChain(MessageChain setMethod);
+	MessageChain createMessageChain(MessageChain setMethod,  Duration expires);
+	
+	/**
+	 * Refresh the expiration of the provided message chain.
+	 * @param sessionId
+	 * @param replicaId
+	 * @param chainId
+	 * @param expires
+	 * @return
+	 */
+	boolean refreshMessageChain(String sessionId, Long replicaId, Integer chainId, Duration expires);
 
 	/**
 	 * Get a {@link MessageChain} if it exists.
@@ -260,6 +271,17 @@ List<ArrayNode> getArrayNodesInOrder(String sessionIdString, Long replicaId, Log
 	 * @return Optional.empty() if the chain no longer exists
 	 */
 	Optional<MessageChain> getMessageChain(String sessionId, Long replicaId, Integer chainId);
+	
+	/**
+	 * Determine if a non-expired message chain already exists for the given method name.
+	 * 
+	 * @param sessionId
+	 * @param replicaId
+	 * @param method
+	 * @return
+	 */
+	Optional<MessageChain> getNonExpiredMessageChain(String sessionId, Long replicaId, String method);
+	
 
 	/**
 	 * Delete a message chain upon completion. This will free up the ID to be

lib/lib-grid-db/src/main/java/org/sagebionetworks/grid/db/GridIndexDaoImpl.java

@@ -7,6 +7,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Timestamp;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
@@ -623,17 +624,20 @@ public Integer createNextMessageId(String sessionIdString, Long replicaId, int m
 
 	@Override
 	@GridTransaction(readOnly = false)
-	public MessageChain createMessageChain(MessageChain chain) {
+	public MessageChain createMessageChain(MessageChain chain, Duration expires) {
 		ValidateArgument.required(chain, "chain");
 		ValidateArgument.required(chain.getId(), "chain.id");
 		ValidateArgument.required(chain.getSessionId(), "chain.sessionId");
 		ValidateArgument.required(chain.getReplicaId(), "chain.replicaId");
 		ValidateArgument.required(chain.getMethod(), "chain.method");
+		ValidateArgument.required(expires, "chain.expires");
 		Long sessionId = validateReplica(chain.getSessionId(), chain.getReplicaId());
 		jdbcTemplate.update(
-				"INSERT INTO GRID_REPLICA_MESSAGE (SESSION_ID, REPLICA_ID, MESSAGE_ID, METHOD_NAME, CREATED_ON)"
-						+ " VALUES (?,?,?,?,NOW()) ON DUPLICATE KEY UPDATE METHOD_NAME = ?, CREATED_ON = NOW()",
-				sessionId, chain.getReplicaId(), chain.getId(), chain.getMethod(), chain.getMethod());
+				"INSERT INTO GRID_REPLICA_MESSAGE (SESSION_ID, REPLICA_ID, MESSAGE_ID, METHOD_NAME, CREATED_ON, EXPIRES_On)"
+						+ " VALUES (?,?,?,?,NOW(),NOW() + INTERVAL ? SECOND) ON DUPLICATE KEY"
+						+ " UPDATE METHOD_NAME = ?, CREATED_ON = NOW(), EXPIRES_ON = NOW() + INTERVAL ? SECOND",
+				sessionId, chain.getReplicaId(), chain.getId(), chain.getMethod(), expires.getSeconds(),
+				chain.getMethod(), expires.getSeconds());
 		return getMessageChain(chain.getSessionId(), chain.getReplicaId(), chain.getId()).get();
 	}
 
@@ -650,6 +654,31 @@ public Optional<MessageChain> getMessageChain(String sessionIdString, Long repli
 		}
 	}
 
+	@Override
+	@GridTransaction(readOnly = false)
+	public boolean refreshMessageChain(String sessionIdString, Long replicaId, Integer chainId, Duration expires) {
+		Long sessionId = validateReplica(sessionIdString, replicaId);
+		ValidateArgument.required(expires, "expires");
+		return jdbcTemplate.update(
+				"UPDATE GRID_REPLICA_MESSAGE SET EXPIRES_ON = NOW() + INTERVAL ? SECOND WHERE"
+						+ " SESSION_ID = ? AND REPLICA_ID = ? AND MESSAGE_ID = ?",
+				expires.getSeconds(), sessionId, replicaId, chainId) > 0;
+	}
+
+	@Override
+	public Optional<MessageChain> getNonExpiredMessageChain(String sessionIdString, Long replicaId, String method) {
+		Long sessionId = validateReplica(sessionIdString, replicaId);
+		ValidateArgument.required(method, "method");
+		try {
+			return jdbcTemplate
+					.query("SELECT * FROM GRID_REPLICA_MESSAGE WHERE SESSION_ID = ? AND REPLICA_ID = ? AND METHOD_NAME = ?"
+							+ " AND EXPIRES_ON > NOW() LIMIT 1", MESSAGE_CHAIN_MAPPER, sessionId, replicaId, method)
+					.stream().findFirst();
+		} catch (EmptyResultDataAccessException e) {
+			return Optional.empty();
+		}
+	}
+
 	@Override
 	@GridTransaction(readOnly = false)
 	public void deleteMessageChain(String sessionIdString, Long replicaId, Integer chainId) {
@@ -707,4 +736,6 @@ public Long getClockSequenceMaximum(String sessionIdString, Long replicaId) {
 				Long.class, sessionId, replicaId);
 		return max != null ? max : 1L;
 	}
+
+
 }

lib/lib-grid-db/src/main/java/org/sagebionetworks/grid/db/GridIndexManager.java

@@ -1,5 +1,6 @@
 package org.sagebionetworks.grid.db;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -60,6 +61,26 @@ public interface GridIndexManager {
 	 * @param chainId
 	 */
 	void completeMessageChain(String sessionId, Long replicaId, Integer chainId);
+	
+	/**
+	 * Refresh the expiration on the provided message chain.
+	 * 
+	 * @param sessionId
+	 * @param replicaId
+	 * @param chainId
+	 * @return
+	 */
+	boolean refreshMessageChain(String sessionId, Long replicaId, Integer chainId);
 
+	/**
+	 * Determine if a non-expired message chain already exists for the given method name.
+	 * 
+	 * @param sessionId
+	 * @param replicaId
+	 * @param method
+	 * @return
+	 */
+	Optional<MessageChain> getNonExpiredMessageChain(String sessionId, Long replicaId, String method);
+	
 	void truncateAll();
 }

lib/lib-grid-db/src/main/java/org/sagebionetworks/grid/db/GridIndexManagerImpl.java

@@ -1,5 +1,6 @@
 package org.sagebionetworks.grid.db;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -19,6 +20,7 @@
 @GridTransaction(readOnly = true)
 public class GridIndexManagerImpl implements GridIndexManager {
 
+	public static final Duration MAX_MESSAGE_DURATION = Duration.ofSeconds(60);
 	public static final int MAX_MESSAGE_ID = 65535;
 
 	private static final Logger log = LogManager.getLogger(GridIndexManagerImpl.class);
@@ -57,8 +59,8 @@ public Map<IndexType, Set<LogicalTimestamp>> applyPatch(String sessionId, Long r
 		/*
 		 * Set the replica's clock to reflect the applied patch. For bootstrap patches
 		 * (created during grid initialization), we must be careful not to increment
-		 * this replica's sequence beyond other replicas' sequences, as this could
-		 * cause outstanding bootstrap patches to be ignored during synchronization.
+		 * this replica's sequence beyond other replicas' sequences, as this could cause
+		 * outstanding bootstrap patches to be ignored during synchronization.
 		 */
 		dao.setClock(sessionId, replicaId, patchClock);
 		return changes;
@@ -98,7 +100,8 @@ public MessageChain startMessageChain(String sessionId, Long replicaId, String m
 		createReplicaIfNotExist(sessionId, replicaId);
 		Integer id = dao.createNextMessageId(sessionId, replicaId, MAX_MESSAGE_ID);
 		return dao.createMessageChain(
-				new MessageChain().setSessionId(sessionId).setReplicaId(replicaId).setMethod(method).setId(id));
+				new MessageChain().setSessionId(sessionId).setReplicaId(replicaId).setMethod(method).setId(id),
+				MAX_MESSAGE_DURATION);
 	}
 
 	@Override
@@ -117,4 +120,15 @@ public void completeMessageChain(String sessionId, Long replicaId, Integer chain
 	public void truncateAll() {
 		dao.truncateAll();
 	}
+
+	@Override
+	@GridTransaction(readOnly = false)
+	public boolean refreshMessageChain(String sessionId, Long replicaId, Integer chainId) {
+		return dao.refreshMessageChain(sessionId, replicaId, chainId, MAX_MESSAGE_DURATION);
+	}
+
+	@Override
+	public Optional<MessageChain> getNonExpiredMessageChain(String sessionId, Long replicaId, String method) {
+		return dao.getNonExpiredMessageChain(sessionId, replicaId, method);
+	}
 }

lib/lib-grid-db/src/main/resources/schema/Grid-Message-ddl.sql

@@ -4,6 +4,7 @@ CREATE TABLE IF NOT EXISTS GRID_REPLICA_MESSAGE (
 	MESSAGE_ID MEDIUMINT NOT NULL,
 	METHOD_NAME VARCHAR(256) CHARACTER SET latin1 COLLATE latin1_bin NOT NULL,
 	CREATED_ON TIMESTAMP(3) NOT NULL,
+	EXPIRES_ON TIMESTAMP(3) DEFAULT NULL,
 	PRIMARY KEY (SESSION_ID, REPLICA_ID, MESSAGE_ID),
 	CONSTRAINT `GRID_MESSAGE_FK` FOREIGN KEY (SESSION_ID, REPLICA_ID) REFERENCES GRID_REPLICA(SESSION_ID, REPLICA_ID) ON DELETE CASCADE
 )

lib/lib-grid-db/src/test/java/org/sagebionetworks/grid/db/GridIndexDaoImplTest.java

@@ -7,6 +7,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.sql.Timestamp;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -778,23 +779,30 @@ public void testCreateNextMessageId() {
 	}
 
 	@Test
-	public void testMessageChainCRUD() {
+	public void testMessageChainCRUD() throws InterruptedException {
 		gridIndexDao.createReplicaIfNotExists(sessionIdOne, replicaIdOne);
 		gridIndexDao.createReplicaIfNotExists(sessionIdTwo, replicaIdTwo);
 		int maxValues = 100;
+		Duration expires = Duration.ofSeconds(2);
 		// one
 		Integer idOne = gridIndexDao.createNextMessageId(sessionIdOne, replicaIdOne, maxValues);
 		assertEquals(Optional.empty(), gridIndexDao.getMessageChain(sessionIdOne, replicaIdOne, idOne));
 		MessageChain chainOne = new MessageChain().setSessionId(sessionIdOne).setReplicaId(replicaIdOne).setId(idOne)
 				.setMethod("methodOne");
-		MessageChain backOne = gridIndexDao.createMessageChain(chainOne);
+		MessageChain backOne = gridIndexDao.createMessageChain(chainOne, expires);
 		MessageChain expected = new MessageChain().setSessionId(sessionIdOne).setReplicaId(replicaIdOne).setId(idOne)
 				.setMethod("methodOne").setCreatedOn(backOne.getCreatedOn());
 		assertEquals(expected, backOne);
 		assertEquals(Optional.of(expected), gridIndexDao.getMessageChain(sessionIdOne, replicaIdOne, idOne));
+		assertEquals(Optional.of(expected),
+				gridIndexDao.getNonExpiredMessageChain(sessionIdOne, replicaIdOne, chainOne.getMethod()));
+		Thread.sleep(2001L);
+		assertEquals(Optional.empty(),
+				gridIndexDao.getNonExpiredMessageChain(sessionIdOne, replicaIdOne, chainOne.getMethod()));
+		
 		// update
 		chainOne.setMethod("updatedMethod");
-		backOne = gridIndexDao.createMessageChain(chainOne);
+		backOne = gridIndexDao.createMessageChain(chainOne, expires);
 		expected = new MessageChain().setSessionId(sessionIdOne).setReplicaId(replicaIdOne).setId(idOne)
 				.setMethod("updatedMethod").setCreatedOn(backOne.getCreatedOn());
 		assertEquals(expected, backOne);
@@ -803,7 +811,7 @@ public void testMessageChainCRUD() {
 		Integer idTwo = gridIndexDao.createNextMessageId(sessionIdTwo, replicaIdTwo, maxValues);
 		MessageChain chainTwo = new MessageChain().setSessionId(sessionIdTwo).setReplicaId(replicaIdTwo).setId(idTwo)
 				.setMethod("methodTwo");
-		backOne = gridIndexDao.createMessageChain(chainTwo);
+		backOne = gridIndexDao.createMessageChain(chainTwo, expires);
 		expected = new MessageChain().setSessionId(sessionIdTwo).setReplicaId(replicaIdTwo).setId(idTwo)
 				.setMethod("methodTwo").setCreatedOn(backOne.getCreatedOn());
 		assertEquals(expected, backOne);
@@ -814,6 +822,28 @@ public void testMessageChainCRUD() {
 		assertEquals(Optional.empty(), gridIndexDao.getMessageChain(sessionIdOne, replicaIdOne, idOne));
 		assertEquals(Optional.of(expected), gridIndexDao.getMessageChain(sessionIdTwo, replicaIdTwo, idTwo));
 	}
+	
+	@Test
+	public void testMessageChainExpiration() throws InterruptedException {
+		gridIndexDao.createReplicaIfNotExists(sessionIdOne, replicaIdOne);
+		int maxValues = 100;
+		Duration expires = Duration.ofSeconds(1);
+		// one
+		Integer idOne = gridIndexDao.createNextMessageId(sessionIdOne, replicaIdOne, maxValues);
+		assertEquals(Optional.empty(), gridIndexDao.getMessageChain(sessionIdOne, replicaIdOne, idOne));
+		MessageChain chainOne = new MessageChain().setSessionId(sessionIdOne).setReplicaId(replicaIdOne).setId(idOne)
+				.setMethod("methodOne");
+		// call under test
+		MessageChain backOne = gridIndexDao.createMessageChain(chainOne, expires);
+		Thread.sleep(1000L);
+		expires = Duration.ofSeconds(10);
+		// call under test
+		assertTrue(gridIndexDao.refreshMessageChain(sessionIdOne, replicaIdOne, idOne, expires));
+		Thread.sleep(1000L);
+		// all under test
+		assertEquals(Optional.of(backOne),
+				gridIndexDao.getNonExpiredMessageChain(sessionIdOne, replicaIdOne, chainOne.getMethod()));
+	}
 
 	@Test
 	public void testGetRootObject() {

lib/lib-grid-db/src/test/java/org/sagebionetworks/grid/db/GridIndexManagerImplTest.java

@@ -57,10 +57,8 @@ public class GridIndexManagerImplTest {
 	public void before() {
 		sessionId = "sessionOne";
 		replicaId = 123L;
-		newConstant = new NewConstant(
-				new LogicalTimestamp().setReplicaId(1L).setSequenceNumber(2L),
-				new ConValue(ConType.BOOLEAN, true)
-		);
+		newConstant = new NewConstant(new LogicalTimestamp().setReplicaId(1L).setSequenceNumber(2L),
+				new ConValue(ConType.BOOLEAN, true));
 		newVector = new NewVector(new LogicalTimestamp().setReplicaId(3L).setSequenceNumber(4L));
 		patch = new Patch().setPatchId(new LogicalTimestamp().setReplicaId(5L).setSequenceNumber(6L))
 				.setOperations(List.of(newConstant, newVector));
@@ -210,4 +208,25 @@ public void testIsPatchAlreadyAppliedWithNewerReplica() {
 		assertTrue(manager.isPatchAlreadyApplied(sessionId, replicaId, patch.getPatchId()));
 	}
 
+	@Test
+	public void testStartMessageChain() {
+		int nextId = 101;
+		String methodName = "some-method";
+		when(mockDao.createNextMessageId(sessionId, replicaId, GridIndexManagerImpl.MAX_MESSAGE_ID)).thenReturn(nextId);
+		MessageChain chain = new MessageChain().setId(nextId).setSessionId(sessionId).setReplicaId(replicaId)
+				.setMethod(methodName);
+		when(mockDao.createMessageChain(chain, GridIndexManagerImpl.MAX_MESSAGE_DURATION)).thenReturn(chain);
+		// call under test
+		assertEquals(chain, manager.startMessageChain(sessionId, replicaId, methodName));
+	}
+	
+	@Test
+	public void testRefreshMessageChain() {
+		int chainId = 111;
+		when(mockDao.refreshMessageChain(sessionId, replicaId, chainId, GridIndexManagerImpl.MAX_MESSAGE_DURATION))
+				.thenReturn(true);
+		// call under test
+		assertTrue(manager.refreshMessageChain(sessionId, replicaId, chainId));
+	}
+
 }