[2.0.x] fix(stores): stop re-processed blocks from duplicating metadata, assets, script and MIR rows (backport of #1266)

components/common/src/main/java/com/bloxbean/cardano/yaci/store/common/util/UUIDUtil.java

@@ -0,0 +1,70 @@
+package com.bloxbean.cardano.yaci.store.common.util;
+
+import org.jooq.DSLContext;
+import org.jooq.Field;
+import org.jooq.SQLDialect;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.UUID;
+
+public class UUIDUtil {
+    private static final UUID NAMESPACE = UUID.fromString("1cd62603-57e9-4254-a1a4-b1ae636cf00e");
+
+    /**
+     * Name based UUID (RFC 4122 version 5) of the given parts, so a row derived from the same chain data
+     * always gets the same id. Each part is length prefixed and null is encoded as length -1.
+     */
+    public static UUID nameUUID(String... parts) {
+        MessageDigest sha1;
+        try {
+            sha1 = MessageDigest.getInstance("SHA-1");
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException(e);
+        }
+
+        sha1.update(toBytes(NAMESPACE));
+        for (String part : parts) {
+            if (part == null) {
+                sha1.update(ByteBuffer.allocate(4).putInt(-1).array());
+            } else {
+                byte[] bytes = part.getBytes(StandardCharsets.UTF_8);
+                sha1.update(ByteBuffer.allocate(4).putInt(bytes.length).array());
+                sha1.update(bytes);
+            }
+        }
+
+        byte[] hash = sha1.digest();
+        hash[6] = (byte) ((hash[6] & 0x0f) | 0x50); //version 5
+        hash[8] = (byte) ((hash[8] & 0x3f) | 0x80); //IETF variant
+        ByteBuffer buffer = ByteBuffer.wrap(hash, 0, 16);
+        return new UUID(buffer.getLong(), buffer.getLong());
+    }
+
+    /**
+     * jOOQ binds UUID as a string on MySQL, where the id column is binary(16) as written by Hibernate.
+     */
+    public static Field<UUID> idField(DSLContext dsl, Field<UUID> idField) {
+        if (dsl.dialect().family() != SQLDialect.MYSQL)
+            return idField;
+
+        return DSL.field(idField.getUnqualifiedName(),
+                SQLDataType.BINARY(16).asConvertedDataType(UUID.class, UUIDUtil::fromBytes, UUIDUtil::toBytes));
+    }
+
+    private static byte[] toBytes(UUID uuid) {
+        return ByteBuffer.allocate(16)
+                .putLong(uuid.getMostSignificantBits())
+                .putLong(uuid.getLeastSignificantBits())
+                .array();
+    }
+
+    private static UUID fromBytes(byte[] bytes) {
+        ByteBuffer buffer = ByteBuffer.wrap(bytes);
+        return new UUID(buffer.getLong(), buffer.getLong());
+    }
+}

components/common/src/test/java/com/bloxbean/cardano/yaci/store/common/util/UUIDUtilTest.java

@@ -0,0 +1,35 @@
+package com.bloxbean.cardano.yaci.store.common.util;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class UUIDUtilTest {
+
+    @Test
+    void nameUUID_sameInput_sameUUID() {
+        String txHash = "9163a0795ca1dee7d51f0601f65ec2d90408a653ed89cd45e36c73e04ccdaa8e";
+
+        assertThat(UUIDUtil.nameUUID(txHash, "721")).isEqualTo(UUIDUtil.nameUUID(txHash, "721"));
+        assertThat(UUIDUtil.nameUUID(txHash, "721")).isNotEqualTo(UUIDUtil.nameUUID(txHash, "674"));
+    }
+
+    @Test
+    void nameUUID_isVersion5WithIetfVariant() {
+        UUID uuid = UUIDUtil.nameUUID("tx", "1");
+
+        assertThat(uuid.version()).isEqualTo(5);
+        assertThat(uuid.variant()).isEqualTo(2);
+    }
+
+    @Test
+    void nameUUID_differentSplits_differentUUIDs() {
+        assertThat(UUIDUtil.nameUUID("ab", "c")).isNotEqualTo(UUIDUtil.nameUUID("a", "bc"));
+        assertThat(UUIDUtil.nameUUID("abc")).isNotEqualTo(UUIDUtil.nameUUID("abc", ""));
+        assertThat(UUIDUtil.nameUUID("tx", "0", null)).isNotEqualTo(UUIDUtil.nameUUID("tx", "0", "null"));
+        assertThat(UUIDUtil.nameUUID("tx", "0", null)).isNotEqualTo(UUIDUtil.nameUUID("tx", "0", ""));
+        assertThat(UUIDUtil.nameUUID("tx", "0", null)).isNotEqualTo(UUIDUtil.nameUUID("tx", "0"));
+    }
+}

components/core/src/main/java/com/bloxbean/cardano/yaci/store/core/service/StartService.java

@@ -122,8 +122,14 @@ public void start() {
         log.info("From >> " + from);
         log.info("TO >> " + to);
 
-        //Send a rollback event to rollback data at and after this slot.
-        //This is because, during start up the from block will be processed again (For BlockFetch).
+        log.debug("Block diff configuration to start sync >>> " + storeProperties.getBlockDiffToStartSyncProtocol());
+        long diff = tip.getPoint().getSlot() - from.getSlot();
+        //BlockFetch requests the range from the from block inclusive, so the from block is processed again.
+        //ChainSync starts after the intersection, so it is not.
+        boolean fetchRange = !storeProperties.isPrimaryInstance() || diff > storeProperties.getBlockDiffToStartSyncProtocol();
+
+        //Send a rollback event to rollback data after this slot, and at this slot for stores
+        //that cannot overwrite the from block's rows when it is processed again (For BlockFetch).
         if (from.getSlot() > 0) {
             //Publish pre-rollback event to clean data
             PreRollbackEvent preRollbackEvent = PreRollbackEvent.builder()
@@ -139,6 +145,7 @@ public void start() {
                     .rollbackTo(new Point(from.getSlot(), from.getHash()))
                     .currentBlock(tip.getBlock())
                     .currentPoint(new Point(tip.getPoint().getSlot(), tip.getPoint().getHash()))
+                    .rollbackPointReprocessed(fetchRange)
                     .build();
             log.info("Publishing rollback event to clean data after restart >>> " + rollbackEvent);
             publisher.publishEvent(rollbackEvent);
@@ -147,18 +154,13 @@ public void start() {
         if (blockNumber != null && blockNumber > 0L)
             publisher.publishEvent(new PreSyncEvent(blockNumber));
 
-        log.debug("Block diff configuration to start sync >>> " + storeProperties.getBlockDiffToStartSyncProtocol());
-        long diff = tip.getPoint().getSlot() - from.getSlot();
-        if (storeProperties.isPrimaryInstance()) {
-            if (diff > storeProperties.getBlockDiffToStartSyncProtocol()) {
+        if (fetchRange) {
+            if (storeProperties.isPrimaryInstance())
                 log.info("Start BlockRangeSync");
-                blockFetchService.startFetch(from, to);
-            } else {
-                log.info("Start BlockSync");
-                blockFetchService.startSync(from);
-            }
-        } else {
             blockFetchService.startFetch(from, to);
+        } else {
+            log.info("Start BlockSync");
+            blockFetchService.startSync(from);
         }
     }
 

components/core/src/test/java/com/bloxbean/cardano/yaci/store/core/service/StartServiceTest.java

@@ -0,0 +1,108 @@
+package com.bloxbean.cardano.yaci.store.core.service;
+
+import com.bloxbean.cardano.yaci.core.model.Era;
+import com.bloxbean.cardano.yaci.core.protocol.chainsync.messages.Point;
+import com.bloxbean.cardano.yaci.core.protocol.chainsync.messages.Tip;
+import com.bloxbean.cardano.yaci.helper.GenesisBlockFinder;
+import com.bloxbean.cardano.yaci.store.common.config.StoreProperties;
+import com.bloxbean.cardano.yaci.store.common.domain.Cursor;
+import com.bloxbean.cardano.yaci.store.common.service.CursorService;
+import com.bloxbean.cardano.yaci.store.core.configuration.GenesisConfig;
+import com.bloxbean.cardano.yaci.store.events.RollbackEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.context.ApplicationEventPublisher;
+import reactor.core.publisher.Mono;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class StartServiceTest {
+    private static final long CURSOR_SLOT = 115239162;
+
+    @Mock
+    private ApplicationEventPublisher publisher;
+    @Mock
+    private BlockFetchService blockFetchService;
+    @Mock
+    private TipFinderService tipFinderService;
+    @Mock
+    private GenesisBlockFinder genesisBlockFinder;
+    @Mock
+    private CursorService cursorService;
+    @Mock
+    private InitService initService;
+    @Mock
+    private GenesisConfig genesisConfig;
+
+    private StoreProperties storeProperties;
+    private StartService startService;
+
+    @BeforeEach
+    void setUp() {
+        storeProperties = new StoreProperties();
+        startService = new StartService(publisher, blockFetchService, tipFinderService, genesisBlockFinder,
+                cursorService, initService, storeProperties, genesisConfig);
+
+        when(cursorService.getStartCursor()).thenReturn(Optional.of(
+                new Cursor(CURSOR_SLOT, "cursor_hash", 10_000_000L, "prev_hash", Era.Babbage)));
+    }
+
+    @Test
+    void start_farFromTip_rangeFetchReprocessesCursorBlock() {
+        tipAt(CURSOR_SLOT + storeProperties.getBlockDiffToStartSyncProtocol() + 1);
+
+        startService.start();
+
+        assertThat(rollbackEvent().isRollbackPointReprocessed()).isTrue();
+        assertThat(rollbackEvent().generatedKeyRollbackSlot()).isEqualTo(CURSOR_SLOT - 1);
+        verify(blockFetchService).startFetch(eq(new Point(CURSOR_SLOT, "cursor_hash")), any());
+        verify(blockFetchService, never()).startSync(any());
+    }
+
+    @Test
+    void start_nearTip_chainSyncDoesNotReprocessCursorBlock() {
+        tipAt(CURSOR_SLOT + 10);
+
+        startService.start();
+
+        assertThat(rollbackEvent().isRollbackPointReprocessed()).isFalse();
+        assertThat(rollbackEvent().generatedKeyRollbackSlot()).isEqualTo(CURSOR_SLOT);
+        verify(blockFetchService).startSync(new Point(CURSOR_SLOT, "cursor_hash"));
+        verify(blockFetchService, never()).startFetch(any(), any());
+    }
+
+    @Test
+    void start_withStopSlot_rangeFetchReprocessesCursorBlock() {
+        storeProperties.setSyncStopSlot(CURSOR_SLOT + 10);
+        storeProperties.setSyncStopBlockhash("stop_hash");
+        tipAt(CURSOR_SLOT + 10);
+
+        startService.start();
+
+        assertThat(rollbackEvent().isRollbackPointReprocessed()).isTrue();
+        verify(blockFetchService).startFetch(new Point(CURSOR_SLOT, "cursor_hash"), new Point(CURSOR_SLOT + 10, "stop_hash"));
+    }
+
+    private void tipAt(long slot) {
+        when(tipFinderService.getTip()).thenReturn(Mono.just(new Tip(new Point(slot, "tip_hash"), 10_000_100L)));
+    }
+
+    private RollbackEvent rollbackEvent() {
+        ArgumentCaptor<Object> captor = ArgumentCaptor.forClass(Object.class);
+        verify(publisher, atLeastOnce()).publishEvent(captor.capture());
+        return captor.getAllValues().stream()
+                .filter(RollbackEvent.class::isInstance)
+                .map(RollbackEvent.class::cast)
+                .findFirst()
+                .orElseThrow();
+    }
+}

components/events/src/main/java/com/bloxbean/cardano/yaci/store/events/RollbackEvent.java

@@ -14,4 +14,20 @@ public class RollbackEvent {
     private long currentBlock;
 
     private boolean remotePublish; //Is published by a remote publisher
+
+    /**
+     * True when the block at {@link #rollbackTo} is going to be processed again, as on a restart that
+     * refetches the cursor block. Stores that rewrite a block's rows idempotently can ignore it. Stores
+     * whose rows have generated keys must then delete the rows at {@code rollbackTo} as well, or that
+     * block's rows are written twice.
+     */
+    private boolean rollbackPointReprocessed;
+
+    /**
+     * The slot after which a store whose rows have generated keys (and so are not overwritten when a
+     * block is processed again) must delete them.
+     */
+    public long generatedKeyRollbackSlot() {
+        return rollbackPointReprocessed ? rollbackTo.getSlot() - 1 : rollbackTo.getSlot();
+    }
 }

publish-common.gradle

@@ -44,7 +44,7 @@ if (!name.equalsIgnoreCase("stores") && !name.equalsIgnoreCase("components") &&
 
 //JOOQ generator
 //Add store modules to this array
-def jooq_modules = ['utxo', 'transaction', 'blocks', 'script', 'account', 'governance', 'governance-aggr', 'staking', 'adapot', 'epoch']
+def jooq_modules = ['utxo', 'transaction', 'blocks', 'script', 'account', 'governance', 'governance-aggr', 'staking', 'adapot', 'epoch', 'assets', 'metadata', 'mir']
 if (jooq_modules.contains(name)) {
     configurations {
         flywayMigration

stores/assets/src/main/java/com/bloxbean/cardano/yaci/store/assets/AssetsStoreConfiguration.java

@@ -6,6 +6,7 @@
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.AssetStorageReaderImpl;
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.mapper.AssetMapper;
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.repository.TxAssetRepository;
+import org.jooq.DSLContext;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.autoconfigure.domain.EntityScan;
@@ -31,8 +32,8 @@ public class AssetsStoreConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
-    public AssetStorage assetStorage(TxAssetRepository txAssetRepository, AssetMapper assetMapper) {
-        return new AssetStorageImpl(txAssetRepository, assetMapper);
+    public AssetStorage assetStorage(TxAssetRepository txAssetRepository, AssetMapper assetMapper, DSLContext dslContext) {
+        return new AssetStorageImpl(txAssetRepository, assetMapper, dslContext);
     }
 
     @Bean

stores/assets/src/main/java/com/bloxbean/cardano/yaci/store/assets/processor/AssetRollbackProcessor.java

@@ -22,9 +22,9 @@ public class AssetRollbackProcessor {
 
     @EventListener
     @Transactional
-    //TODO -- tests
     public void handleRollbackEvent(RollbackEvent rollbackEvent) {
-        int count = assetStorage.deleteBySlotGreaterThan(rollbackEvent.getRollbackTo().getSlot());
+        //Rows have random ids, so a block processed again must not find its rows still there
+        int count = assetStorage.deleteBySlotGreaterThan(rollbackEvent.generatedKeyRollbackSlot());
 
         log.info("Rollback -- {} transaction_assets records", count);
     }

stores/assets/src/main/java/com/bloxbean/cardano/yaci/store/assets/storage/impl/AssetStorageImpl.java

@@ -5,23 +5,61 @@
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.mapper.AssetMapper;
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.model.TxAssetEntity;
 import com.bloxbean.cardano.yaci.store.assets.storage.impl.repository.TxAssetRepository;
+import com.bloxbean.cardano.yaci.store.common.util.UUIDUtil;
 import com.bloxbean.cardano.yaci.store.plugin.aspect.Plugin;
 import lombok.RequiredArgsConstructor;
+import org.jooq.DSLContext;
 
+import java.time.LocalDateTime;
 import java.util.List;
 
+import static com.bloxbean.cardano.yaci.store.assets.jooq.Tables.ASSETS;
+
 @RequiredArgsConstructor
 public class AssetStorageImpl implements AssetStorage {
     private final static String PLUGIN_ASSET_SAVE = "asset.save";
 
     private final TxAssetRepository txAssetRepository;
     private final AssetMapper assetMapper;
+    private final DSLContext dsl;
 
     @Override
     @Plugin(key = PLUGIN_ASSET_SAVE)
     public void saveAll(List<TxAsset> txAssetList) {
         List<TxAssetEntity> txAssetEntities = txAssetList.stream().map(assetMapper::toTxAssetEntity).toList();
-        txAssetRepository.saveAll(txAssetEntities);
+
+        //Id from (tx_hash, unit), so a block processed again overwrites its rows. Unit is policy + hex asset name,
+        //unique in a tx's mint, while asset_name is a decoded name which can repeat
+        var idField = UUIDUtil.idField(dsl, ASSETS.ID);
+        LocalDateTime localDateTime = LocalDateTime.now();
+        var inserts = txAssetEntities.stream()
+                .map(entity -> dsl.insertInto(ASSETS)
+                        .set(idField, UUIDUtil.nameUUID(entity.getTxHash(), entity.getUnit()))
+                        .set(ASSETS.SLOT, entity.getSlot())
+                        .set(ASSETS.TX_HASH, entity.getTxHash())
+                        .set(ASSETS.POLICY, entity.getPolicy())
+                        .set(ASSETS.ASSET_NAME, entity.getAssetName())
+                        .set(ASSETS.UNIT, entity.getUnit())
+                        .set(ASSETS.FINGERPRINT, entity.getFingerprint())
+                        .set(ASSETS.QUANTITY, entity.getQuantity() != null ? entity.getQuantity().longValue() : null)
+                        .set(ASSETS.MINT_TYPE, entity.getMintType() != null ? entity.getMintType().name() : null)
+                        .set(ASSETS.BLOCK, entity.getBlockNumber())
+                        .set(ASSETS.BLOCK_TIME, entity.getBlockTime())
+                        .set(ASSETS.UPDATE_DATETIME, localDateTime)
+                        .onDuplicateKeyUpdate()
+                        .set(ASSETS.SLOT, entity.getSlot())
+                        .set(ASSETS.TX_HASH, entity.getTxHash())
+                        .set(ASSETS.POLICY, entity.getPolicy())
+                        .set(ASSETS.ASSET_NAME, entity.getAssetName())
+                        .set(ASSETS.UNIT, entity.getUnit())
+                        .set(ASSETS.FINGERPRINT, entity.getFingerprint())
+                        .set(ASSETS.QUANTITY, entity.getQuantity() != null ? entity.getQuantity().longValue() : null)
+                        .set(ASSETS.MINT_TYPE, entity.getMintType() != null ? entity.getMintType().name() : null)
+                        .set(ASSETS.BLOCK, entity.getBlockNumber())
+                        .set(ASSETS.BLOCK_TIME, entity.getBlockTime())
+                        .set(ASSETS.UPDATE_DATETIME, localDateTime)).toList();
+
+        dsl.batch(inserts).execute();
     }
 
     @Override