fix: resolve bug in method to fetch the pool owner from the latest update id

Author: fabianbormann

src/main/java/org/cardanofoundation/rewards/calculation/domain/PoolHistory.java

@@ -3,6 +3,7 @@
 import lombok.*;
 
 import java.math.BigInteger;
+import java.util.HashSet;
 import java.util.List;
 
 @Getter
@@ -21,7 +22,7 @@ public class PoolHistory {
     private Double margin;
     private BigInteger fixedCost;
     private BigInteger pledge;
-    private List<Delegator> delegators;
+    private HashSet<Delegator> delegators;
     private int blockCount;
     private int epoch;
 }

src/main/java/org/cardanofoundation/rewards/validation/data/provider/DbSyncDataProvider.java

@@ -20,6 +20,8 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+import static org.cardanofoundation.rewards.calculation.constants.RewardConstants.MAINNET_SHELLEY_START_EPOCH;
+
 @Service
 @Slf4j
 @Profile("db-sync")
@@ -68,6 +70,10 @@ public AdaPots getAdaPotsForEpoch(int epoch) {
 
     @Override
     public Epoch getEpochInfo(int epoch) {
+        if (epoch < MAINNET_SHELLEY_START_EPOCH) {
+            return null;
+        }
+
         DbSyncEpoch dbSyncEpoch = dbSyncEpochRepository.findByNumber(epoch);
         Epoch epochInfo = EpochMapper.fromDbSyncEpoch(dbSyncEpoch);
 
@@ -103,7 +109,7 @@ public List<PoolHistory> fetchPoolHistoryInBatches(Integer epoch, int batchSize,
                 .distinct()
                 .toList();
 
-        List<LatestPoolUpdate> latestUpdates = dbSyncPoolUpdateRepository.findLatestActiveUpdatesInEpoch(epoch, poolIds);
+        HashSet<LatestPoolUpdate> latestUpdates = dbSyncPoolUpdateRepository.findLatestActiveUpdatesInEpoch(epoch, poolIds);
 
         List<Long> updateIds = latestUpdates.stream()
                 .map(LatestPoolUpdate::getId)
@@ -122,15 +128,15 @@ public List<PoolHistory> fetchPoolHistoryInBatches(Integer epoch, int batchSize,
         int batches = poolIdBatches.size();
         for (List<String> poolIdBatch : poolIdBatches) {
             log.info("fetching pool history batch " + i + " / " + batches + " for epoch " + epoch + " with " + poolIdBatch.size() + " pools");
-            List<PoolEpochStake> poolEpochStakes = dbSyncEpochStakeRepository.getAllPoolsActiveStakesInEpoch(epoch, poolIdBatch);
+            HashSet<PoolEpochStake> poolEpochStakes = dbSyncEpochStakeRepository.getAllPoolsActiveStakesInEpoch(epoch, poolIdBatch);
 
             for (String poolId : poolIdBatch) {
                 PoolHistory poolHistory = new PoolHistory();
 
-                List<Delegator> delegators = poolEpochStakes.stream()
+                HashSet<Delegator> delegators = poolEpochStakes.stream()
                         .filter(epochStake -> epochStake.getPoolId().equals(poolId))
                         .map(DelegatorMapper::fromPoolEpochStake)
-                        .toList();
+                        .collect(Collectors.toCollection(HashSet::new));
 
                 if (!delegators.isEmpty()) {
                     BigInteger activeStake = delegators.stream()
@@ -198,7 +204,7 @@ public PoolHistory getPoolHistory(String poolId, int epoch) {
         PoolHistory poolHistory = new PoolHistory();
         List<PoolEpochStake> poolEpochStakes = dbSyncEpochStakeRepository.getPoolActiveStakeInEpoch(poolId, epoch);
         BigInteger activeStake = BigInteger.ZERO;
-        List<Delegator> delegators = new ArrayList<>();
+        HashSet<Delegator> delegators = new HashSet<>();
         for (PoolEpochStake poolEpochStake : poolEpochStakes) {
             activeStake = activeStake.add(poolEpochStake.getAmount());
             Delegator delegator = Delegator.builder()

src/main/java/org/cardanofoundation/rewards/validation/data/provider/KoiosDataProvider.java

@@ -119,7 +119,7 @@ public PoolHistory getPoolHistory(String poolId, int epoch) {
 
         if (history == null) return null;
 
-        List<Delegator> poolMemberInEpoch = getPoolMemberInEpoch(poolId, epoch);
+        HashSet<Delegator> poolMemberInEpoch = getPoolMemberInEpoch(poolId, epoch);
         history.setDelegators(poolMemberInEpoch);
         return history;
     }
@@ -316,8 +316,8 @@ public HashSet<String> getRegisteredAccountsUntilNow(Integer epoch, HashSet<Stri
         return null;
     }
 
-    private List<Delegator> getPoolMemberInEpoch(String poolId, int epoch) {
-        List<Delegator> delegators = new ArrayList<>();
+    private HashSet<Delegator> getPoolMemberInEpoch(String poolId, int epoch) {
+        HashSet<Delegator> delegators = new HashSet<>();
         try {
             List<PoolDelegatorHistory> poolDelegatorsHistory = koiosBackendService
                     .getPoolService().getPoolDelegatorsHistory(poolId, epoch, Options.EMPTY).getValue();

src/main/java/org/cardanofoundation/rewards/validation/repository/DbSyncEpochStakeRepository.java

@@ -6,6 +6,7 @@
 import org.springframework.data.jpa.repository.Query;
 
 import java.math.BigInteger;
+import java.util.HashSet;
 import java.util.List;
 
 public interface DbSyncEpochStakeRepository extends ReadOnlyRepository<DbSyncEpochStake, Long> {
@@ -22,7 +23,7 @@ public interface DbSyncEpochStakeRepository extends ReadOnlyRepository<DbSyncEpo
             	JOIN pool_hash ON pool_hash.id=epoch_stake.pool_id
             	JOIN stake_address ON stake_address.id = epoch_stake.addr_id
             WHERE epoch_no=:epoch AND pool_hash.view IN :poolIds AND amount > 0""")
-    List<PoolEpochStake> getAllPoolsActiveStakesInEpoch(Integer epoch, List<String> poolIds);
+    HashSet<PoolEpochStake> getAllPoolsActiveStakesInEpoch(Integer epoch, List<String> poolIds);
 
     @Query("SELECT SUM(amount) FROM DbSyncEpochStake WHERE epoch=:epoch")
     BigInteger getEpochStakeByEpoch(Integer epoch);

src/main/java/org/cardanofoundation/rewards/validation/repository/DbSyncPoolUpdateRepository.java

@@ -22,14 +22,17 @@ public interface DbSyncPoolUpdateRepository extends ReadOnlyRepository<DbSyncPoo
 
     @Query(nativeQuery = true, value = """
             SELECT pool_update.id, pool_hash.view AS poolId, pledge, margin,
-                   fixed_cost AS fixedCost, stake_address.view AS rewardAddress
+            	   fixed_cost AS fixedCost, stake_address.view AS rewardAddress
             FROM pool_update
-                JOIN stake_address ON stake_address.id=pool_update.reward_addr_id
-                JOIN pool_hash ON pool_hash.id=hash_id
-            WHERE pool_hash.view IN :poolIds AND pool_update.registered_tx_id IN (
-                SELECT MAX(registered_tx_id) FROM pool_update WHERE active_epoch_no <= :epoch GROUP BY hash_id
-            );""")
-    List<LatestPoolUpdate> findLatestActiveUpdatesInEpoch(Integer epoch, List<String> poolIds);
+            	JOIN stake_address ON stake_address.id=pool_update.reward_addr_id
+            	JOIN pool_hash ON pool_hash.id=pool_update.hash_id AND pool_hash.view IN :poolIds
+            	JOIN (SELECT MAX(registered_tx_id) AS registered_tx_id, hash_id
+            			FROM pool_update WHERE active_epoch_no <= :epoch
+            			GROUP BY hash_id
+            	) AS latest_update ON latest_update.hash_id=pool_update.hash_id AND latest_update.registered_tx_id=pool_update.registered_tx_id
+            WHERE pool_update.active_epoch_no <= :epoch
+            """)
+    HashSet<LatestPoolUpdate> findLatestActiveUpdatesInEpoch(Integer epoch, List<String> poolIds);
 
     @Query(nativeQuery = true, value = """
             SELECT COUNT(*) FROM (

src/test/java/org/cardanofoundation/rewards/validation/EpochValidationTest.java

@@ -43,7 +43,7 @@ public void testCalculateEpochPots(final int epoch, DataProvider dataProvider, b
     }
 
     static Stream<Integer> dataProviderEpochRange() {
-        return IntStream.range(236, 460).boxed();
+        return IntStream.range(208, 460).boxed();
     }
 
     @ParameterizedTest

src/test/java/org/cardanofoundation/rewards/validation/PoolRewardValidationTest.java

@@ -111,10 +111,11 @@ void calculateAutoStakeIIPoolRewardInEpoch215() {
     }
 
     @Test
+    @EnabledIf(expression = "#{environment.acceptsProfiles('db-sync')}", loadContext = true, reason = "DB Sync data provider must be available for this test")
     void calculateIOG1PoolRewardInEpoch241() {
         String poolId = "pool1mxqjlrfskhd5kql9kak06fpdh8xjwc76gec76p3taqy2qmfzs5z";
         int epoch = 241;
-        Test_calculatePoolReward(poolId, epoch, DataProviderType.JSON);
+        Test_calculatePoolReward(poolId, epoch, DataProviderType.DB_SYNC);
     }
     @Test
     void calculateXYZPoolRewardInEpoch213() {