Rerun conditional migrations

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/migration/InitializeEntityBalanceMigration.java

@@ -18,9 +18,11 @@
 
 import com.google.common.base.Stopwatch;
 import com.hedera.mirror.importer.MirrorProperties;
+import com.hedera.mirror.importer.parser.balance.InitializeBalanceEvent;
 import jakarta.inject.Named;
 import org.flywaydb.core.api.MigrationVersion;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.context.event.EventListener;
 import org.springframework.jdbc.core.JdbcOperations;
 
 @Named
@@ -87,4 +89,12 @@ protected void doMigrate() {
         int count = jdbcOperations.update(INITIALIZE_ENTITY_BALANCE_SQL);
         log.info("Initialized {} entities balance in {}", count, stopwatch);
     }
+
+    @EventListener
+    public void reRunMigration(InitializeBalanceEvent event) {
+        log.info(
+                "Re-running the InitializeEntityBalanceMigration on InitializeBalanceEvent created at {}",
+                event.getTimestamp());
+        doMigrate();
+    }
 }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/migration/SyntheticCryptoTransferApprovalMigration.java

@@ -17,8 +17,12 @@
 package com.hedera.mirror.importer.migration;
 
 import com.hedera.mirror.common.domain.entity.EntityId;
+import com.hedera.mirror.common.domain.transaction.RecordFile;
 import com.hedera.mirror.importer.MirrorProperties;
 import com.hedera.mirror.importer.db.DBProperties;
+import com.hedera.mirror.importer.exception.ImporterException;
+import com.hedera.mirror.importer.parser.record.RecordStreamFileListener;
+import com.hedera.mirror.importer.repository.RecordFileRepository;
 import com.hederahashgraph.api.proto.java.Key;
 import jakarta.inject.Named;
 import java.time.Duration;
@@ -30,13 +34,16 @@
 import lombok.Data;
 import org.flywaydb.core.api.MigrationVersion;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.data.util.Version;
 import org.springframework.jdbc.core.DataClassRowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.transaction.support.TransactionOperations;
 
 @Named
-public class SyntheticCryptoTransferApprovalMigration extends AsyncJavaMigration<Long> {
+public class SyntheticCryptoTransferApprovalMigration extends AsyncJavaMigration<Long>
+        implements RecordStreamFileListener {
 
+    public static final Version HAPI_VERSION_0_38_10 = new Version(0, 38, 10);
     // The contract id of the first synthetic transfer that could have exhibited this problem
     private static final long GRANDFATHERED_ID = 2119900L;
     // The created timestamp of the grandfathered id contract
@@ -140,6 +147,32 @@ with contractresults as (
             update token_transfer set is_approval = true where consensus_timestamp = :consensus_timestamp and token_id = :token_id and account_id = :sender
             """;
 
+    private final RecordFileRepository recordFileRepository;
+
+    @Override
+    public void onStart() throws ImporterException {
+        // Not Applicable
+    }
+
+    @Override
+    public void onEnd(RecordFile streamFile) throws ImporterException {
+        if (streamFile == null) {
+            return;
+        }
+
+        if ((streamFile.getHapiVersion()).isGreaterThanOrEqualTo(HAPI_VERSION_0_38_10)) {
+            var latestFile = recordFileRepository.findLatestWithOffset(1).orElse(null);
+            if (latestFile != null && latestFile.getHapiVersion().isLessThan(HAPI_VERSION_0_38_10)) {
+                migrateAsync();
+            }
+        }
+    }
+
+    @Override
+    public void onError() {
+        // Not Applicable
+    }
+
     private enum TRANSFER_TYPE {
         CRYPTO_TRANSFER,
         NFT_TRANSFER,
@@ -157,10 +190,12 @@ public SyntheticCryptoTransferApprovalMigration(
             DBProperties dbProperties,
             NamedParameterJdbcTemplate transferJdbcTemplate,
             MirrorProperties mirrorProperties,
-            TransactionOperations transactionOperations) {
+            TransactionOperations transactionOperations,
+            RecordFileRepository recordFileRepository) {
         super(transferJdbcTemplate, dbProperties.getSchema(), transactionOperations);
         this.transferJdbcTemplate = transferJdbcTemplate;
         this.mirrorProperties = mirrorProperties;
+        this.recordFileRepository = recordFileRepository;
     }
 
     @Override

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/migration/SyntheticNftAllowanceOwnerMigration.java

@@ -17,15 +17,23 @@
 package com.hedera.mirror.importer.migration;
 
 import com.google.common.base.Stopwatch;
+import com.hedera.mirror.common.domain.transaction.RecordFile;
 import com.hedera.mirror.importer.MirrorProperties;
 import com.hedera.mirror.importer.config.Owner;
+import com.hedera.mirror.importer.exception.ImporterException;
+import com.hedera.mirror.importer.parser.record.RecordStreamFileListener;
+import com.hedera.mirror.importer.repository.RecordFileRepository;
 import jakarta.inject.Named;
 import org.flywaydb.core.api.MigrationVersion;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.data.util.Version;
 import org.springframework.jdbc.core.JdbcTemplate;
 
 @Named
-public class SyntheticNftAllowanceOwnerMigration extends RepeatableMigration {
+public class SyntheticNftAllowanceOwnerMigration extends RepeatableMigration implements RecordStreamFileListener {
+    // 779
+    public static final Version HAPI_VERSION_0_37_0 = new Version(0, 37, 0);
+    private final RecordFileRepository recordFileRepository;
 
     private static final String UPDATE_NFT_ALLOWANCE_OWNER_SQL =
             """
@@ -107,9 +115,13 @@ insert into nft_allowance (approved_for_all, owner, payer_account_id, spender, t
     private final JdbcTemplate jdbcTemplate;
 
     @Lazy
-    public SyntheticNftAllowanceOwnerMigration(@Owner JdbcTemplate jdbcTemplate, MirrorProperties mirrorProperties) {
+    public SyntheticNftAllowanceOwnerMigration(
+            @Owner JdbcTemplate jdbcTemplate,
+            MirrorProperties mirrorProperties,
+            RecordFileRepository recordFileRepository) {
         super(mirrorProperties.getMigration());
         this.jdbcTemplate = jdbcTemplate;
+        this.recordFileRepository = recordFileRepository;
     }
 
     @Override
@@ -129,4 +141,28 @@ protected void doMigrate() {
         jdbcTemplate.execute(UPDATE_NFT_ALLOWANCE_OWNER_SQL);
         log.info("Updated nft allowance owners in {}", stopwatch);
     }
+
+    @Override
+    public void onStart() throws ImporterException {
+        // Not Applicable
+    }
+
+    @Override
+    public void onEnd(RecordFile streamFile) throws ImporterException {
+        if (streamFile == null) {
+            return;
+        }
+
+        if ((streamFile.getHapiVersion()).isGreaterThanOrEqualTo(HAPI_VERSION_0_37_0)) {
+            var latestFile = recordFileRepository.findLatestWithOffset(1).orElse(null);
+            if (latestFile != null && latestFile.getHapiVersion().isLessThan(HAPI_VERSION_0_37_0)) {
+                doMigrate();
+            }
+        }
+    }
+
+    @Override
+    public void onError() {
+        // Not Applicable
+    }
 }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/migration/SyntheticTokenAllowanceOwnerMigration.java

@@ -17,15 +17,23 @@
 package com.hedera.mirror.importer.migration;
 
 import com.google.common.base.Stopwatch;
+import com.hedera.mirror.common.domain.transaction.RecordFile;
 import com.hedera.mirror.importer.MirrorProperties;
 import com.hedera.mirror.importer.config.Owner;
+import com.hedera.mirror.importer.exception.ImporterException;
+import com.hedera.mirror.importer.parser.record.RecordStreamFileListener;
+import com.hedera.mirror.importer.repository.RecordFileRepository;
 import jakarta.inject.Named;
 import org.flywaydb.core.api.MigrationVersion;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.data.util.Version;
 import org.springframework.jdbc.core.JdbcTemplate;
 
 @Named
-public class SyntheticTokenAllowanceOwnerMigration extends RepeatableMigration {
+public class SyntheticTokenAllowanceOwnerMigration extends RepeatableMigration implements RecordStreamFileListener {
+    // 10158
+    public static final Version HAPI_VERSION_0_37_0 = new Version(0, 37, 0);
+    private final RecordFileRepository recordFileRepository;
 
     private static final String UPDATE_TOKEN_ALLOWANCE_OWNER_SQL =
             """
@@ -110,9 +118,13 @@ insert into token_allowance (amount, amount_granted, owner, payer_account_id, sp
     private final JdbcTemplate jdbcTemplate;
 
     @Lazy
-    public SyntheticTokenAllowanceOwnerMigration(@Owner JdbcTemplate jdbcTemplate, MirrorProperties mirrorProperties) {
+    public SyntheticTokenAllowanceOwnerMigration(
+            @Owner JdbcTemplate jdbcTemplate,
+            MirrorProperties mirrorProperties,
+            RecordFileRepository recordFileRepository) {
         super(mirrorProperties.getMigration());
         this.jdbcTemplate = jdbcTemplate;
+        this.recordFileRepository = recordFileRepository;
     }
 
     @Override
@@ -132,4 +144,27 @@ protected void doMigrate() {
         jdbcTemplate.execute(UPDATE_TOKEN_ALLOWANCE_OWNER_SQL);
         log.info("Updated token allowance owners in {}", stopwatch);
     }
+
+    @Override
+    public void onStart() throws ImporterException {
+        // Not Applicable
+    }
+
+    @Override
+    public void onEnd(RecordFile streamFile) throws ImporterException {
+        if (streamFile == null) {
+            return;
+        }
+        if ((streamFile.getHapiVersion()).isGreaterThanOrEqualTo(HAPI_VERSION_0_37_0)) {
+            var latestFile = recordFileRepository.findLatestWithOffset(1).orElse(null);
+            if (latestFile != null && latestFile.getHapiVersion().isLessThan(HAPI_VERSION_0_37_0)) {
+                doMigrate();
+            }
+        }
+    }
+
+    @Override
+    public void onError() {
+        // Not Applicable
+    }
 }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/migration/TokenAccountBalanceMigration.java

@@ -18,9 +18,11 @@
 
 import com.google.common.base.Stopwatch;
 import com.hedera.mirror.importer.MirrorProperties;
+import com.hedera.mirror.importer.parser.balance.InitializeBalanceEvent;
 import jakarta.inject.Named;
 import org.flywaydb.core.api.MigrationVersion;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.context.event.EventListener;
 import org.springframework.jdbc.core.JdbcOperations;
 
 @Named
@@ -89,4 +91,10 @@ protected void doMigrate() {
         int count = jdbcOperations.update(UPDATE_TOKEN_ACCOUNT_SQL);
         log.info("Migrated {} token account balances in {}", count, stopwatch);
     }
+
+    @EventListener
+    public void reRunMigration(InitializeBalanceEvent event) {
+        log.info("Running this on InitializeBalanceEvent created at {}", event.getTimestamp());
+        doMigrate();
+    }
 }

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/AbstractStreamFileParser.java

@@ -24,13 +24,15 @@
 import io.micrometer.core.instrument.Timer;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.concurrent.atomic.AtomicReference;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 public abstract class AbstractStreamFileParser<T extends StreamFile<?>> implements StreamFileParser<T> {
 
     public static final String STREAM_PARSE_DURATION_METRIC_NAME = "hedera.mirror.parse.duration";
 
+    protected final AtomicReference<T> last;
     protected final Logger log = LogManager.getLogger(getClass());
     protected final MeterRegistry meterRegistry;
     protected final ParserProperties parserProperties;
@@ -44,6 +46,8 @@ protected AbstractStreamFileParser(
             MeterRegistry meterRegistry,
             ParserProperties parserProperties,
             StreamFileRepository<T, Long> streamFileRepository) {
+
+        this.last = new AtomicReference<>();
         this.meterRegistry = meterRegistry;
         this.parserProperties = parserProperties;
         this.streamFileRepository = streamFileRepository;
@@ -86,7 +90,7 @@ public void parse(T streamFile) {
                 success = true;
                 Instant consensusInstant = Instant.ofEpochSecond(0L, streamFile.getConsensusEnd());
                 parseLatencyMetric.record(Duration.between(consensusInstant, Instant.now()));
-            } catch (Throwable e) {
+            } catch (Exception e) {
                 log.error("Error parsing file {} after {}", streamFile.getName(), stopwatch, e);
                 throw e;
             } finally {
@@ -102,14 +106,18 @@ private boolean shouldParse(T streamFile) {
         if (!parserProperties.isEnabled()) {
             return false;
         }
+        var lastStreamFileFromDB = streamFileRepository.findLatest();
 
-        var last = streamFileRepository.findLatest();
-
-        if (last.isEmpty()) {
+        if (lastStreamFileFromDB.isEmpty()) {
             return true;
         }
 
-        var lastStreamFile = last.get();
+        if (last.get() == null) {
+            last.set(lastStreamFileFromDB.get());
+            last.get().setItems(null);
+        }
+
+        var lastStreamFile = lastStreamFileFromDB.get();
         var name = streamFile.getName();
 
         if (lastStreamFile.getConsensusEnd() >= streamFile.getConsensusStart()) {

hedera-mirror-importer/src/main/java/com/hedera/mirror/importer/parser/balance/AccountBalanceFileParser.java

@@ -34,6 +34,8 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.jetbrains.annotations.Nullable;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.retry.annotation.Backoff;
 import org.springframework.retry.annotation.Retryable;
 import org.springframework.transaction.annotation.Transactional;
@@ -44,6 +46,7 @@
 @Named
 public class AccountBalanceFileParser extends AbstractStreamFileParser<AccountBalanceFile> {
 
+    private final ApplicationEventPublisher applicationEventPublisher;
     private final BatchPersister batchPersister;
     private final MirrorDateRangePropertiesProcessor mirrorDateRangePropertiesProcessor;
     private final BalanceStreamFileListener streamFileListener;
@@ -53,10 +56,12 @@ public AccountBalanceFileParser(
             MeterRegistry meterRegistry,
             BalanceParserProperties parserProperties,
             StreamFileRepository<AccountBalanceFile, Long> accountBalanceFileRepository,
+            ApplicationEventPublisher applicationEventPublisher,
             MirrorDateRangePropertiesProcessor mirrorDateRangePropertiesProcessor,
             BalanceStreamFileListener streamFileListener) {
         super(meterRegistry, parserProperties, accountBalanceFileRepository);
         this.batchPersister = batchPersister;
+        this.applicationEventPublisher = applicationEventPublisher;
         this.mirrorDateRangePropertiesProcessor = mirrorDateRangePropertiesProcessor;
         this.streamFileListener = streamFileListener;
     }
@@ -84,6 +89,7 @@ protected void doParse(AccountBalanceFile accountBalanceFile) {
         DateRangeFilter filter = mirrorDateRangePropertiesProcessor.getDateRangeFilter(StreamType.BALANCE);
         int batchSize = ((BalanceParserProperties) parserProperties).getBatchSize();
         long count = 0L;
+        AccountBalanceFile lastBalanceFile = publishInitializeAccountBalanceEvent();
 
         if (filter.filter(accountBalanceFile.getConsensusTimestamp())) {
             List<AccountBalance> accountBalances = new ArrayList<>(batchSize);
@@ -121,5 +127,17 @@ protected void doParse(AccountBalanceFile accountBalanceFile) {
         accountBalanceFile.setLoadEnd(loadEnd.getEpochSecond());
         streamFileListener.onEnd(accountBalanceFile);
         streamFileRepository.save(accountBalanceFile);
+
+        last.compareAndSet(lastBalanceFile, accountBalanceFile);
+    }
+
+    @Nullable
+    private AccountBalanceFile publishInitializeAccountBalanceEvent() {
+        var lastBalanceFile = last.get();
+        // If no account balance file has been processed since last startup.
+        if (lastBalanceFile == null) {
+            applicationEventPublisher.publishEvent(new InitializeBalanceEvent(this));
+        }
+        return lastBalanceFile;
     }
 }