Enable out-of-order transaction processing for asset table

README.md

@@ -38,6 +38,10 @@ Then with a local `DATABASE_URL` var exported like this `export DATABASE_URL=pos
 
 If you need to install `sea-orm-cli` run `cargo install sea-orm-cli`.
 
+Note: The current SeaORM types were generated using version 0.9.3 so unless you want to upgrade you can install using `cargo install sea-orm-cli --version 0.9.3`.
+
+Also note: The migration `m20230224_093722_performance_improvements` needs to be commented out of the migration lib.rs in order for the Sea ORM `Relations` to generate correctly.
+
 #### Developing Locally
  *Prerequisites*
  * A Postgres Server running with the database setup according to ./init.sql

digital_asset_types/src/dao/generated/asset.rs

@@ -43,6 +43,7 @@ pub struct Model {
     pub slot_updated: i64,
     pub data_hash: Option<String>,
     pub creator_hash: Option<String>,
+    pub compressed_seq: Option<i64>,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveColumn)]
@@ -72,6 +73,7 @@ pub enum Column {
     SlotUpdated,
     DataHash,
     CreatorHash,
+    CompressedSeq,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DerivePrimaryKey)]
@@ -124,6 +126,7 @@ impl ColumnTrait for Column {
             Self::SlotUpdated => ColumnType::BigInteger.def(),
             Self::DataHash => ColumnType::Char(Some(50u32)).def().null(),
             Self::CreatorHash => ColumnType::Char(Some(50u32)).def().null(),
+            Self::CompressedSeq => ColumnType::BigInteger.def().null(),
         }
     }
 }

digital_asset_types/src/dao/generated/sea_orm_active_enums.rs

@@ -7,17 +7,29 @@ use serde::{Deserialize, Serialize};
 #[sea_orm(
     rs_type = "String",
     db_type = "Enum",
-    enum_name = "specification_versions"
+    enum_name = "v1_account_attachments"
 )]
-pub enum SpecificationVersions {
+pub enum V1AccountAttachments {
+    #[sea_orm(string_value = "edition")]
+    Edition,
+    #[sea_orm(string_value = "edition_marker")]
+    EditionMarker,
+    #[sea_orm(string_value = "master_edition_v1")]
+    MasterEditionV1,
+    #[sea_orm(string_value = "master_edition_v2")]
+    MasterEditionV2,
+    #[sea_orm(string_value = "unknown")]
+    Unknown,
+}
+#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "owner_type")]
+pub enum OwnerType {
+    #[sea_orm(string_value = "single")]
+    Single,
+    #[sea_orm(string_value = "token")]
+    Token,
     #[sea_orm(string_value = "unknown")]
     Unknown,
-    #[sea_orm(string_value = "v0")]
-    V0,
-    #[sea_orm(string_value = "v1")]
-    V1,
-    #[sea_orm(string_value = "v2")]
-    V2,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
@@ -36,18 +48,6 @@ pub enum RoyaltyTargetType {
     Unknown,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
-#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "task_status")]
-pub enum TaskStatus {
-    #[sea_orm(string_value = "failed")]
-    Failed,
-    #[sea_orm(string_value = "pending")]
-    Pending,
-    #[sea_orm(string_value = "running")]
-    Running,
-    #[sea_orm(string_value = "success")]
-    Success,
-}
-#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "chain_mutability")]
 pub enum ChainMutability {
     #[sea_orm(string_value = "immutable")]
@@ -58,42 +58,32 @@ pub enum ChainMutability {
     Unknown,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
-#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "owner_type")]
-pub enum OwnerType {
-    #[sea_orm(string_value = "single")]
-    Single,
-    #[sea_orm(string_value = "token")]
-    Token,
-    #[sea_orm(string_value = "unknown")]
-    Unknown,
-}
-#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
-#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "mutability")]
-pub enum Mutability {
-    #[sea_orm(string_value = "immutable")]
-    Immutable,
-    #[sea_orm(string_value = "mutable")]
-    Mutable,
-    #[sea_orm(string_value = "unknown")]
-    Unknown,
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "task_status")]
+pub enum TaskStatus {
+    #[sea_orm(string_value = "failed")]
+    Failed,
+    #[sea_orm(string_value = "pending")]
+    Pending,
+    #[sea_orm(string_value = "running")]
+    Running,
+    #[sea_orm(string_value = "success")]
+    Success,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
     rs_type = "String",
     db_type = "Enum",
-    enum_name = "v1_account_attachments"
+    enum_name = "specification_versions"
 )]
-pub enum V1AccountAttachments {
-    #[sea_orm(string_value = "edition")]
-    Edition,
-    #[sea_orm(string_value = "edition_marker")]
-    EditionMarker,
-    #[sea_orm(string_value = "master_edition_v1")]
-    MasterEditionV1,
-    #[sea_orm(string_value = "master_edition_v2")]
-    MasterEditionV2,
+pub enum SpecificationVersions {
     #[sea_orm(string_value = "unknown")]
     Unknown,
+    #[sea_orm(string_value = "v0")]
+    V0,
+    #[sea_orm(string_value = "v1")]
+    V1,
+    #[sea_orm(string_value = "v2")]
+    V2,
 }
 #[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
 #[sea_orm(
@@ -116,10 +106,20 @@ pub enum SpecificationAssetClass {
     Print,
     #[sea_orm(string_value = "PRINTABLE_NFT")]
     PrintableNft,
+    #[sea_orm(string_value = "PROGRAMMABLE_NFT")]
+    ProgrammableNft,
     #[sea_orm(string_value = "TRANSFER_RESTRICTED_NFT")]
     TransferRestrictedNft,
     #[sea_orm(string_value = "unknown")]
     Unknown,
-    #[sea_orm(string_value = "PNFT")]
-    ProgrammableNft,
+}
+#[derive(Debug, Clone, PartialEq, EnumIter, DeriveActiveEnum, Serialize, Deserialize)]
+#[sea_orm(rs_type = "String", db_type = "Enum", enum_name = "mutability")]
+pub enum Mutability {
+    #[sea_orm(string_value = "immutable")]
+    Immutable,
+    #[sea_orm(string_value = "mutable")]
+    Mutable,
+    #[sea_orm(string_value = "unknown")]
+    Unknown,
 }

digital_asset_types/tests/common.rs

@@ -142,14 +142,15 @@ pub fn create_asset(
             royalty_target_type,
             royalty_target,
             royalty_amount,
-            asset_data: Some(id.clone()),
+            asset_data: Some(id),
             burnt: false,
             created_at: None,
             specification_asset_class: SpecificationAssetClass::Nft,
             slot_updated: 0,
             data_hash: None,
             alt_id: None,
             creator_hash: None,
+            compressed_seq: Some(0),
         },
     )
 }

docker-compose.yaml

@@ -95,7 +95,7 @@ services:
     - ./ledger:/config:rw
     - ./solana-test-validator-geyser-config:/plugin-config:rw
     environment:
-      RUST_LOG: info
+      RUST_LOG: error
       PLUGIN_MESSENGER_CONFIG.messenger_type: "Redis"
       PLUGIN_MESSENGER_CONFIG.connection_config: '{redis_connection_str="redis://redis"}'
     ports:

migration/src/lib.rs

@@ -16,6 +16,8 @@ mod m20230203_205959_improve_upsert_perf;
 mod m20230224_093722_performance_improvements;
 mod m20230310_162227_add_indexes_to_bg;
 mod m20230317_121944_remove_indexes_for_perf;
+mod m20230526_120101_add_compressed_seq;
+mod m20230601_120101_add_pnft_enum_val;
 
 pub struct Migrator;
 
@@ -39,6 +41,8 @@ impl MigratorTrait for Migrator {
             Box::new(m20230224_093722_performance_improvements::Migration),
             Box::new(m20230310_162227_add_indexes_to_bg::Migration),
             Box::new(m20230317_121944_remove_indexes_for_perf::Migration),
+            Box::new(m20230526_120101_add_compressed_seq::Migration),
+            Box::new(m20230601_120101_add_pnft_enum_val::Migration),
         ]
     }
 }

migration/src/m20230224_093722_performance_improvements.rs

@@ -135,7 +135,7 @@ impl MigrationTrait for Migration {
         Ok(())
     }
 
-    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
         Ok(())
     }
 }

migration/src/m20230526_120101_add_compressed_seq.rs

@@ -0,0 +1,32 @@
+use digital_asset_types::dao::asset;
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(asset::Entity)
+                    .add_column(ColumnDef::new(Alias::new("compressed_seq")).big_integer())
+                    .to_owned(),
+            )
+            .await
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(asset::Entity)
+                    .drop_column(Alias::new("compressed_seq"))
+                    .to_owned(),
+            )
+            .await
+    }
+}

migration/src/m20230601_120101_add_pnft_enum_val.rs

@@ -0,0 +1,29 @@
+use sea_orm_migration::{
+    prelude::*,
+    sea_orm::{ConnectionTrait, DatabaseBackend, Statement},
+};
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        manager
+            .get_connection()
+            .execute(Statement::from_string(
+                DatabaseBackend::Postgres,
+                "ALTER TYPE specification_asset_class ADD VALUE IF NOT EXISTS 'PROGRAMMABLE_NFT';"
+                    .to_string(),
+            ))
+            .await?;
+
+        Ok(())
+    }
+
+    async fn down(&self, _manager: &SchemaManager) -> Result<(), DbErr> {
+        // Replace the sample below with your own migration scripts
+        Ok(())
+    }
+}

nft_ingester/src/program_transformers/bubblegum/burn.rs

@@ -1,13 +1,14 @@
-use crate::{
-    error::IngesterError,
-};
-use super::{update_asset, save_changelog_event};
+use super::{save_changelog_event, upsert_asset_with_leaf_schema};
+use crate::error::IngesterError;
 use blockbuster::{
     instruction::InstructionBundle,
     programs::bubblegum::{BubblegumInstruction, LeafSchema},
 };
 use digital_asset_types::dao::asset;
-use sea_orm::{entity::*, ConnectionTrait, TransactionTrait};
+use sea_orm::{
+    entity::*, query::*, sea_query::OnConflict, ConnectionTrait, DbBackend, EntityTrait,
+    TransactionTrait,
+};
 
 pub async fn burn<'c, T>(
     parsing_result: &BubblegumInstruction,
@@ -19,18 +20,51 @@ where
 {
     if let (Some(le), Some(cl)) = (&parsing_result.leaf_update, &parsing_result.tree_update) {
         let seq = save_changelog_event(cl, bundle.slot, txn).await?;
+        #[allow(unreachable_patterns)]
         return match le.schema {
-            LeafSchema::V1 { id, .. } => {
+            LeafSchema::V1 {
+                id,
+                delegate,
+                owner,
+                ..
+            } => {
                 let id_bytes = id.to_bytes().to_vec();
-                let asset_to_update = asset::ActiveModel {
-                    id: Unchanged(id_bytes.clone()),
+
+                let asset_model = asset::ActiveModel {
+                    id: Set(id_bytes.to_vec()),
                     burnt: Set(true),
-                    seq: Set(seq as i64), // gummyroll seq
                     ..Default::default()
                 };
-                // Don't send sequence number with this update, because we will always
-                // run this update even if it's from a backfill/replay.
-                update_asset(txn, id_bytes, None, asset_to_update).await
+
+                // Upsert asset table `burnt` column.
+                let query = asset::Entity::insert(asset_model)
+                    .on_conflict(
+                        OnConflict::columns([asset::Column::Id])
+                            .update_columns([
+                                asset::Column::Burnt,
+                                //TODO maybe handle slot updated.
+                            ])
+                            .to_owned(),
+                    )
+                    .build(DbBackend::Postgres);
+                txn.execute(query).await?;
+
+                // Partial update of asset table with just leaf schema elements.
+                let delegate = if owner == delegate {
+                    None
+                } else {
+                    Some(delegate.to_bytes().to_vec())
+                };
+                let owner_bytes = owner.to_bytes().to_vec();
+                upsert_asset_with_leaf_schema(
+                    txn,
+                    id_bytes.clone(),
+                    le.leaf_hash.to_vec(),
+                    delegate,
+                    owner_bytes,
+                    seq as i64,
+                )
+                .await
             }
             _ => Err(IngesterError::NotImplemented),
         };