IGNITE-24793 Fix expired entries leak

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEntryEx.java

@@ -114,6 +114,11 @@ public interface GridCacheEntryEx {
      */
     public boolean hasValue();
 
+    /**
+     * @return {@code True} if has value or value bytes and the value is not expired yet.
+     */
+    public boolean hasNonExpiredValue();
+
     /**
      * @param val New value.
      * @param ttl Time to live.

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java

@@ -993,6 +993,31 @@ protected void recordNodeId(UUID nodeId, AffinityTopologyVersion topVer) {
         @Nullable GridCacheVersion dhtVer,
         @Nullable Long updateCntr
     ) throws IgniteCheckedException, GridCacheEntryRemovedException {
+        // Explicit expire time is already reached, the value is expired: remove the entry instead of storing
+        // an already expired row. A row with expire time in the past must never be written to the row store,
+        // otherwise the (expireTime, link) pair, already processed by the TTL cleanup worker, can be recreated
+        // and the pending entries tree becomes inconsistent (see IGNITE-25194).
+        if (CU.isExpired(drExpireTime)) {
+            return innerRemove(
+                tx,
+                evtNodeId,
+                affNodeId,
+                retval,
+                evt,
+                metrics,
+                keepBinary,
+                keepBinaryInInterceptor,
+                oldValPresent,
+                oldVal,
+                topVer,
+                drType,
+                explicitVer,
+                taskName,
+                dhtVer,
+                updateCntr
+            );
+        }
+
         CacheObject old;
 
         final boolean valid = valid(tx != null ? tx.topologyVersion() : topVer);
@@ -1077,6 +1102,12 @@ else if (interceptorVal != val0)
                 if (ttl == -1L) {
                     ttl = ttlExtras();
                     expireTime = expireTimeExtras();
+
+                    // The current value is already expired (but not cleaned up yet), so the update is actually
+                    // a creation: calculate a new expire time instead of inheriting the expired one. A row with
+                    // expire time in the past must never be written to the row store (see IGNITE-25194).
+                    if (CU.isExpired(expireTime))
+                        expireTime = CU.toExpireTime(ttl);
                 }
                 else
                     expireTime = CU.toExpireTime(ttl);
@@ -2441,6 +2472,18 @@ private boolean checkExpired() throws IgniteCheckedException {
         }
     }
 
+    /** {@inheritDoc} */
+    @Override public final boolean hasNonExpiredValue() {
+        lockEntry();
+
+        try {
+            return hasValueUnlocked() && !CU.isExpired(expireTimeExtras());
+        }
+        finally {
+            unlockEntry();
+        }
+    }
+
     /**
      * @return {@code True} if this entry has value.
      */
@@ -2526,6 +2569,18 @@ private boolean skipInterceptor(@Nullable GridCacheVersion explicitVer) {
 
             long expTime = expireTime < 0 ? CU.toExpireTime(ttl) : expireTime;
 
+            CacheDataRow expiredRow = null;
+
+            // The value is already expired: store it as removed instead of storing an already expired row.
+            // A row with expire time in the past must never be written to the row store (see IGNITE-25194).
+            if (val != null && CU.isExpired(expTime)) {
+                val = null;
+
+                // Pre-created row is already inserted to the row store and must be removed.
+                expiredRow = row;
+                row = null;
+            }
+
             val = cctx.kernalContext().cacheObjects().prepareForCache(val, cctx);
 
             final boolean unswapped = ((flags & IS_UNSWAPPED_MASK) != 0);
@@ -2583,6 +2638,12 @@ else if (val == null)
             else
                 update = storeValue(val, expTime, ver, p, row);
 
+            // If update is not applied, the pre-created row is removed by the caller (see CacheDataStore#insertRows).
+            if (expiredRow != null && update) {
+                cctx.offheap().dataStore(localPartition()).rowStore()
+                    .removeRow(expiredRow.link(), cctx.group().statisticsHolderData());
+            }
+
             if (update) {
                 update(val, expTime, ttl, ver, true);
 
@@ -4354,9 +4415,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
         /** */
         private CacheDataRow oldRow;
 
-        /** */
-        private boolean oldRowExpiredFlag;
-
         /** */
         private IgniteTree.OperationType treeOp = IgniteTree.OperationType.PUT;
 
@@ -4379,14 +4437,14 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
 
         /** {@inheritDoc} */
         @Override public void call(@Nullable CacheDataRow oldRow) throws IgniteCheckedException {
+            this.oldRow = oldRow;
+
             if (oldRow != null) {
                 oldRow.key(entry.key);
 
                 oldRow = checkRowExpired(oldRow);
             }
 
-            this.oldRow = oldRow;
-
             if (predicate != null && !predicate.apply(oldRow)) {
                 treeOp = IgniteTree.OperationType.NOOP;
 
@@ -4426,11 +4484,6 @@ private static class UpdateClosure implements IgniteCacheOffheapManager.OffheapI
             return oldRow;
         }
 
-        /** {@inheritDoc} */
-        @Override public boolean oldRowExpiredFlag() {
-            return oldRowExpiredFlag;
-        }
-
         /**
          * Checks row for expiration and fire expire events if needed.
          *
@@ -4476,8 +4529,6 @@ private CacheDataRow checkRowExpired(CacheDataRow row) throws IgniteCheckedExcep
 
             entry.updatePlatformCache(null, null);
 
-            oldRowExpiredFlag = true;
-
             return null;
         }
     }
@@ -4637,11 +4688,6 @@ private static class AtomicCacheUpdateClosure implements IgniteCacheOffheapManag
             return oldRow;
         }
 
-        /** {@inheritDoc} */
-        @Override public boolean oldRowExpiredFlag() {
-            return oldRowExpiredFlag;
-        }
-
         /** {@inheritDoc} */
         @Override public CacheDataRow newRow() {
             return newRow;
@@ -4994,7 +5040,7 @@ else if (newSysTtl == CU.TTL_ZERO) {
                 newSysExpireTime = newExpireTime = conflictCtx.expireTime();
             }
 
-            if (newExpireTime > 0 && newExpireTime < U.currentTimeMillis()) {
+            if (CU.isExpired(newExpireTime)) {
                 op = DELETE;
 
                 writeObj = null;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java

@@ -1067,6 +1067,14 @@ public static String cacheOrGroupName(CacheConfiguration<?, ?> ccfg) {
         return ccfg.getGroupName() == null ? ccfg.getName() : ccfg.getGroupName();
     }
 
+    /**
+     * @param expireTime Expire time.
+     * @return {@code True} if the given expire time is set and already reached.
+     */
+    public static boolean isExpired(long expireTime) {
+        return expireTime > 0 && expireTime <= U.currentTimeMillis();
+    }
+
     /**
      * Convert TTL to expire time.
      *

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManager.java

@@ -418,12 +418,6 @@ interface OffheapInvokeClosure extends IgniteTree.InvokeClosure<CacheDataRow> {
          * @return Old row.
          */
         @Nullable public CacheDataRow oldRow();
-
-        /**
-         * Flag that indicates if oldRow was expired during invoke.
-         * @return {@code true} if old row was expired, {@code false} otherwise.
-         */
-        public boolean oldRowExpiredFlag();
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheOffheapManagerImpl.java

@@ -997,7 +997,8 @@ private long allocateForTree() throws IgniteCheckedException {
 
             try {
                 batch.add(new DataRowCacheAware(info.key(),
-                    info.value(),
+                    // Already expired value is stored as removed, don't insert it to the row store (see IGNITE-25194).
+                    CU.isExpired(info.expireTime()) ? null : info.value(),
                     info.version(),
                     part.id(),
                     info.expireTime(), info.cacheId(), grp.storeCacheIdInDataPage()));
@@ -1507,9 +1508,7 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo
                 case PUT: {
                     assert c.newRow() != null : c;
 
-                    CacheDataRow oldRow = c.oldRow();
-
-                    finishUpdate(cctx, c.newRow(), oldRow, c.oldRowExpiredFlag());
+                    finishUpdate(cctx, c.newRow(), c.oldRow());
 
                     break;
                 }
@@ -1666,21 +1665,12 @@ private void invoke0(GridCacheContext cctx, CacheSearchRow row, OffheapInvokeClo
          * @param oldRow Old row if available.
          * @throws IgniteCheckedException If failed.
          */
-        private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow)
-            throws IgniteCheckedException {
-            finishUpdate(cctx, newRow, oldRow, false);
-        }
-
-        /**
-         * @param cctx Cache context.
-         * @param newRow New row.
-         * @param oldRow Old row if available.
-         * @param oldRowExpired Old row expiration flag
-         * @throws IgniteCheckedException If failed.
-         */
-        private void finishUpdate(GridCacheContext cctx, CacheDataRow newRow, @Nullable CacheDataRow oldRow, boolean oldRowExpired)
-            throws IgniteCheckedException {
-            if (oldRow == null && !oldRowExpired)
+        private void finishUpdate(
+            GridCacheContext<?, ?> cctx,
+            CacheDataRow newRow,
+            @Nullable CacheDataRow oldRow
+        ) throws IgniteCheckedException {
+            if (oldRow == null)
                 incrementSize(cctx.cacheId());
 
             GridCacheQueryManager qryMgr = cctx.queries();

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java

@@ -584,6 +584,11 @@ else if (conflictCtx.isMerge()) {
                                         // Nullify explicit version so that innerSet/innerRemove will work as usual.
                                         explicitVer = null;
 
+                                    // Explicit expire time is already reached: remove instead of storing an already
+                                    // expired value (the same way as for TTL_ZERO, see IGNITE-25194).
+                                    if ((op == CREATE || op == UPDATE) && CU.isExpired(txEntry.conflictExpireTime()))
+                                        op = DELETE;
+
                                     GridCacheVersion dhtVer = cached.isNear() ? writeVersion() : null;
 
                                     if (!near() && cacheCtx.group().logDataRecords() &&

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java

@@ -604,7 +604,7 @@ protected GridCacheEntryEx entryEx(GridCacheContext cacheCtx, IgniteTxKey key, A
                                         if (expiry != null) {
                                             txEntry.cached().unswap(false);
 
-                                            Duration duration = cached.hasValue() ?
+                                            Duration duration = cached.hasNonExpiredValue() ?
                                                 expiry.getExpiryForUpdate() : expiry.getExpiryForCreation();
 
                                             txEntry.ttl(CU.toTtl(duration));
@@ -624,7 +624,8 @@ protected GridCacheEntryEx entryEx(GridCacheContext cacheCtx, IgniteTxKey key, A
                                     ExpiryPolicy expiry = cacheCtx.expiryForTxEntry(txEntry);
 
                                     if (expiry != null) {
-                                        Duration duration = cached.hasValue() ?
+                                        // Expired (but not cleaned up yet) value is treated as absent.
+                                        Duration duration = cached.hasNonExpiredValue() ?
                                             expiry.getExpiryForUpdate() : expiry.getExpiryForCreation();
 
                                         long ttl = CU.toTtl(duration);
@@ -676,6 +677,11 @@ else if (conflictCtx.isUseNew()) {
                                     txEntry.conflictVersion(explicitVer);
                                 }
 
+                                // Explicit expire time is already reached: remove instead of storing an already
+                                // expired value (the same way as for TTL_ZERO, see IGNITE-25194).
+                                if ((op == CREATE || op == UPDATE) && CU.isExpired(txEntry.conflictExpireTime()))
+                                    op = DELETE;
+
                                 if (dhtVer == null)
                                     dhtVer = explicitVer != null ? explicitVer : writeVersion();
 

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheTestEntryEx.java

@@ -298,6 +298,11 @@ void recheckLock() {
         return val != null;
     }
 
+    /** @inheritDoc */
+    @Override public boolean hasNonExpiredValue() {
+        return hasValue();
+    }
+
     /** @inheritDoc */
     @Override public CacheObject rawPut(CacheObject val, long ttl) {
         CacheObject old = this.val;

modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/ttl/CacheSizeTtlTest.java

@@ -22,23 +22,29 @@
 import java.util.stream.IntStream;
 import javax.cache.expiry.Duration;
 import javax.cache.expiry.ModifiedExpiryPolicy;
+import javax.cache.expiry.TouchedExpiryPolicy;
 import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.IgniteSystemProperties;
 import org.apache.ignite.Ignition;
 import org.apache.ignite.cache.CachePeekMode;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
 import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.jetbrains.annotations.NotNull;
 import org.junit.Test;
 
 import static java.util.Collections.singleton;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ;
 
 /**
  * Tests for cache.size() with ttl enabled.
@@ -57,6 +63,26 @@ public class CacheSizeTtlTest extends GridCommonAbstractTest {
         stopAllGrids();
     }
 
+    /** */
+    @Test
+    @WithSystemProperty(key = IgniteSystemProperties.IGNITE_UNWIND_THROTTLING_TIMEOUT, value = "10000")
+    public void testEntriesLeak() throws Exception {
+        IgniteEx srv = startGrid(getConfiguration().setIncludeEventTypes(EVT_CACHE_OBJECT_READ));
+
+        srv.events().localListen(evt -> {
+            doSleep(2000L);
+            return true;
+        }, EVT_CACHE_OBJECT_READ);
+
+        IgniteCache<Object, Object> cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME)
+            .withExpiryPolicy(new TouchedExpiryPolicy(new Duration(SECONDS, 5)));
+
+        cache.put(1, 1);
+        cache.get(1);
+
+        assertTrue(GridTestUtils.waitForCondition(() -> cache.size(CachePeekMode.PRIMARY) == 0, 15_000L));
+    }
+
     /**
      * Tests that cache.size() works correctly for massive amount of puts and ttl.
      */