HHH-9868 Infinispan 2LC can store stale data

* invalidation blocks putFromLoads until the transaction with invalidation
  is committed
* rewritten the naked puts support: timestamp is stored in the pendingPutMap
  and removal of the record relies on pending puts' idle expiration or
  piggy-backs on release from putFromLoad

Author: rvansa

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/access/PutFromLoadValidator.java

@@ -9,7 +9,6 @@
 import org.hibernate.cache.CacheException;
 import org.hibernate.cache.infinispan.impl.BaseRegion;
 import org.hibernate.cache.infinispan.util.Caches;
-
 import org.infinispan.AdvancedCache;
 import org.infinispan.util.logging.Log;
 import org.infinispan.util.logging.LogFactory;
@@ -111,7 +110,8 @@ public boolean putFromLoad(Object key, Object value, long txTimestamp, Object ve
 			return false;
 		}
 
-		if ( !putValidator.acquirePutFromLoadLock( key ) ) {
+		PutFromLoadValidator.Lock lock = putValidator.acquirePutFromLoadLock(key);
+		if ( lock == null) {
 			if ( TRACE_ENABLED ) {
 				log.tracef( "Put from load lock not acquired for key %s", key );
 			}
@@ -131,7 +131,7 @@ public boolean putFromLoad(Object key, Object value, long txTimestamp, Object ve
 			}
 		}
 		finally {
-			putValidator.releasePutFromLoadLock( key );
+			putValidator.releasePutFromLoadLock( key, lock);
 		}
 
 		return true;
@@ -185,7 +185,7 @@ public boolean update(Object key, Object value, Object currentVersion, Object pr
     * @throws CacheException if removing the cached item fails
     */
 	public void remove(Object key) throws CacheException {
-		if ( !putValidator.invalidateKey( key ) ) {
+		if ( !putValidator.beginInvalidatingKey(key)) {
 			throw new CacheException(
 					"Failed to invalidate pending putFromLoad calls for key " + key + " from region " + region.getName()
 			);
@@ -216,7 +216,7 @@ public void removeAll() throws CacheException {
     * @throws CacheException if evicting the item fails
     */
 	public void evict(Object key) throws CacheException {
-		if ( !putValidator.invalidateKey( key ) ) {
+		if ( !putValidator.beginInvalidatingKey(key)) {
 			throw new CacheException(
 					"Failed to invalidate pending putFromLoad calls for key " + key + " from region " + region.getName()
 			);
@@ -240,4 +240,19 @@ public void evictAll() throws CacheException {
 		Caches.broadcastEvictAll( cache );
 	}
 
+	/**
+	 * Called when we have finished the attempted update/delete (which may or
+	 * may not have been successful), after transaction completion.  This method
+	 * is used by "asynchronous" concurrency strategies.
+	 *
+	 * @param key The item key
+	 * @throws org.hibernate.cache.CacheException Propogated from underlying {@link org.hibernate.cache.spi.Region}
+	 */
+	public void unlockItem(Object key) throws CacheException {
+		if ( !putValidator.endInvalidatingKey(key) ) {
+			// TODO: localization
+			log.warn("Failed to end invalidating pending putFromLoad calls for key " + key + " from region "
+					+ region.getName() + "; the key won't be cached in the future.");
+		}
+	}
 }

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/access/TransactionalAccessDelegate.java

@@ -75,6 +75,7 @@ public SoftLock lockRegion() throws CacheException {
 	}
 
 	public void unlockItem(SessionImplementor session, Object key, SoftLock lock) throws CacheException {
+		delegate.unlockItem(key);
 	}
 
 	public void unlockRegion(SoftLock lock) throws CacheException {

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/collection/TransactionalAccess.java

@@ -84,6 +84,7 @@ public SoftLock lockRegion() throws CacheException {
 	}
 
 	public void unlockItem(SessionImplementor session, Object key, SoftLock lock) throws CacheException {
+		delegate.unlockItem(key);
 	}
 
 	public void unlockRegion(SoftLock lock) throws CacheException {

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/entity/TransactionalAccess.java

@@ -89,6 +89,7 @@ public SoftLock lockRegion() throws CacheException {
 
 	@Override
 	public void unlockItem(SessionImplementor session, Object key, SoftLock lock) throws CacheException {
+		delegate.unlockItem(key);
 	}
 
 	@Override