fix(prune): correct checkpoint when RocksDB tx lookup deletes nothing (#21842)

Co-authored-by: Georgios Konstantopoulos <me@gakonst.com>
Co-authored-by: Amp <amp@ampcode.com>

crates/prune/prune/src/segments/user/transaction_lookup.rs

@@ -271,7 +271,8 @@ impl TransactionLookup {
         let done = deleted == hashes.len() && tx_range_end == end;
         trace!(target: "pruner", %deleted, %done, "Pruned transaction lookup (RocksDB)");
 
-        let last_pruned_transaction = if deleted > 0 { start + deleted as u64 - 1 } else { start };
+        let last_pruned_transaction =
+            if deleted > 0 { start + deleted as u64 - 1 } else { tx_range_end };
 
         let last_pruned_block = provider
             .block_by_transaction_id(last_pruned_transaction)?
@@ -532,4 +533,98 @@ mod tests {
             );
         }
     }
+
+    /// Tests that when `RocksDB` prune deletes nothing (limit exhausted), checkpoint doesn't
+    /// advance.
+    ///
+    /// This test simulates a scenario where:
+    /// 1. Some transactions have already been pruned (checkpoint at tx 5)
+    /// 2. The deleted entries limit is exhausted before any new deletions
+    /// 3. The checkpoint should NOT advance to the next start position
+    #[cfg(all(unix, feature = "rocksdb"))]
+    #[test]
+    fn prune_rocksdb_zero_deleted_checkpoint() {
+        use reth_db_api::models::StorageSettings;
+        use reth_provider::RocksDBProviderFactory;
+        use reth_storage_api::StorageSettingsCache;
+
+        let db = TestStageDB::default();
+        let mut rng = generators::rng();
+
+        let blocks = random_block_range(
+            &mut rng,
+            1..=10,
+            BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
+        );
+        db.insert_blocks(blocks.iter(), StorageKind::Static).expect("insert blocks");
+
+        // Collect transaction hashes and their tx numbers
+        let mut tx_hash_numbers = Vec::new();
+        for block in &blocks {
+            tx_hash_numbers.reserve_exact(block.transaction_count());
+            for transaction in &block.body().transactions {
+                tx_hash_numbers.push((*transaction.tx_hash(), tx_hash_numbers.len() as u64));
+            }
+        }
+
+        // Insert into RocksDB
+        {
+            let rocksdb = db.factory.rocksdb_provider();
+            let mut batch = rocksdb.batch();
+            for (hash, tx_num) in &tx_hash_numbers {
+                batch.put::<tables::TransactionHashNumbers>(*hash, tx_num).unwrap();
+            }
+            batch.commit().expect("commit rocksdb batch");
+        }
+
+        // Enable RocksDB storage for transaction hash numbers
+        db.factory.set_storage_settings_cache(
+            StorageSettings::legacy().with_transaction_hash_numbers_in_rocksdb(true),
+        );
+
+        let to_block: BlockNumber = 6;
+        let prune_mode = PruneMode::Before(to_block);
+
+        // Simulate that we've already pruned up to tx 5, so start will be tx 6
+        let previous_checkpoint =
+            Some(PruneCheckpoint { block_number: Some(2), tx_number: Some(5), prune_mode });
+
+        // Create a limiter with limit of 1, but exhaust it before pruning
+        // This means deleted_entries_limit_left() = Some(0)
+        let mut limiter = PruneLimiter::default().set_deleted_entries_limit(1);
+        limiter.increment_deleted_entries_count(); // Exhaust the limit
+
+        let input = PruneInput { previous_checkpoint, to_block, limiter };
+        let segment = TransactionLookup::new(prune_mode);
+
+        let provider = db.factory.database_provider_rw().unwrap();
+        let result = segment.prune(&provider, input).unwrap();
+        provider.commit().expect("commit");
+
+        // With an exhausted limit, nothing should be deleted
+        assert_eq!(result.pruned, 0, "Nothing should be pruned with exhausted limit");
+
+        // The checkpoint tx_number should NOT advance to 6 (start)
+        // With the bug: checkpoint.tx_number = start = 6 (WRONG - claims tx 6 was pruned)
+        // With the fix: checkpoint.tx_number = tx_range_end = 5 (correct - no advancement)
+        if let Some(checkpoint) = &result.checkpoint {
+            assert_eq!(
+                checkpoint.tx_number,
+                Some(5),
+                "Checkpoint should stay at 5 (previous), not advance to 6 (start)"
+            );
+        }
+
+        // All RocksDB entries should still exist (nothing was actually deleted)
+        {
+            let rocksdb = db.factory.rocksdb_provider();
+            let remaining: Vec<_> =
+                rocksdb.iter::<tables::TransactionHashNumbers>().unwrap().collect();
+            assert_eq!(
+                remaining.len(),
+                tx_hash_numbers.len(),
+                "All RocksDB entries should still exist"
+            );
+        }
+    }
 }