levels: Expired keys and delete markers are never purged

db_test.go

@@ -1259,6 +1259,17 @@ func TestExpiry(t *testing.T) {
 			return nil
 		})
 		require.NoError(t, err)
+
+		// Reopen the database to force the memory table to disk.
+		err = db.Close()
+		require.NoError(t, err)
+		db, err = Open(db.opt)
+		require.NoError(t, err)
+
+		// Only one table and two keys remain, the `!badger!head` key and `answer1`.
+		tables := db.Tables(true)
+		require.Equal(t, 1, len(tables))
+		require.Equal(t, uint64(2), tables[0].KeyCount)
 	})
 }
 

levels.go

@@ -585,13 +585,15 @@ func (s *levelsController) compactBuildTables(
 				lastValidVersion := vs.Meta&bitDiscardEarlierVersions > 0 ||
 					numVersions == s.kv.opt.NumVersionsToKeep
 
-				if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) || lastValidVersion {
+				isExpired := isDeletedOrExpired(vs.Meta, vs.ExpiresAt)
+
+				if isExpired || lastValidVersion {
 					// If this version of the key is deleted or expired, skip all the rest of the
 					// versions. Ensure that we're only removing versions below readTs.
 					skipKey = y.SafeCopy(skipKey, it.Key())
 
 					switch {
-					case lastValidVersion:
+					case !isExpired && lastValidVersion:
 						// Add this key. We have set skipKey, so the following key versions
 						// would be skipped.
 					case hasOverlap: