Merge pull request #89 from deso-protocol/z/caching-and-txindex

Add caching and txindex updates

Author: superzordon

entries/bls_pkid_pair.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/deso-protocol/core/lib"
 	"github.com/deso-protocol/state-consumer/consumer"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/pkg/errors"
 	"github.com/uptrace/bun"
 )
@@ -77,15 +78,15 @@ func BLSPublicKeyPKIDPairSnapshotEncoderToPGStruct(
 
 // BLSPublicKeyPKIDPairBatchOperation is the entry point for processing a batch of BLSPublicKeyPKIDPair entries.
 // It determines the appropriate handler based on the operation type and executes it.
-func BLSPublicKeyPKIDPairBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams) error {
+func BLSPublicKeyPKIDPairBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// We check before we call this function that there is at least one operation type.
 	// We also ensure before this that all entries have the same operation type.
 	operationType := entries[0].OperationType
 	var err error
 	if operationType == lib.DbOperationTypeDelete {
-		err = bulkDeleteBLSPkidPairEntry(entries, db, operationType)
+		err = bulkDeleteBLSPkidPairEntry(entries, db, operationType, cachedEntries)
 	} else {
-		err = bulkInsertBLSPkidPairEntry(entries, db, operationType, params)
+		err = bulkInsertBLSPkidPairEntry(entries, db, operationType, params, cachedEntries)
 	}
 	if err != nil {
 		return errors.Wrapf(err, "entries.StakeBatchOperation: Problem with operation type %v", operationType)
@@ -95,10 +96,13 @@ func BLSPublicKeyPKIDPairBatchOperation(entries []*lib.StateChangeEntry, db bun.
 
 // bulkInsertBLSPkidPairEntry inserts a batch of stake entries into the database.
 func bulkInsertBLSPkidPairEntry(
-	entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams,
-) error {
+	entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
+
+	// Filter out any entries that are already tracked in the cache.
+	uniqueEntries = consumer.FilterCachedEntries(uniqueEntries, cachedEntries)
+
 	uniqueBLSPkidPairEntries := consumer.FilterEntriesByPrefix(
 		uniqueEntries, lib.Prefixes.PrefixValidatorBLSPublicKeyPKIDPairEntry)
 	uniqueBLSPkidPairSnapshotEntries := consumer.FilterEntriesByPrefix(
@@ -145,11 +149,16 @@ func bulkInsertBLSPkidPairEntry(
 		}
 	}
 
+	// Update the cache with the new entries.
+	for _, entry := range uniqueEntries {
+		cachedEntries.Add(string(entry.KeyBytes), entry.EncoderBytes)
+	}
+
 	return nil
 }
 
 // bulkDeleteBLSPkidPairEntry deletes a batch of stake entries from the database.
-func bulkDeleteBLSPkidPairEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType) error {
+func bulkDeleteBLSPkidPairEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
 
@@ -182,5 +191,10 @@ func bulkDeleteBLSPkidPairEntry(entries []*lib.StateChangeEntry, db bun.IDB, ope
 		}
 	}
 
+	// Remove the deleted entries from the cache.
+	for _, key := range keysToDelete {
+		cachedEntries.Remove(string(key))
+	}
+
 	return nil
 }

entries/jailed_history.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/deso-protocol/core/lib"
 	"github.com/deso-protocol/state-consumer/consumer"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/pkg/errors"
 	"github.com/uptrace/bun"
 )
@@ -39,15 +40,15 @@ func UnjailValidatorStateChangeMetadataEncoderToPGStruct(
 
 // ValidatorBatchOperation is the entry point for processing a batch of Validator entries.
 // It determines the appropriate handler based on the operation type and executes it.
-func JailedHistoryEventBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams) error {
+func JailedHistoryEventBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// We check before we call this function that there is at least one operation type.
 	// We also ensure before this that all entries have the same operation type.
 	operationType := entries[0].OperationType
 	var err error
 	if operationType == lib.DbOperationTypeDelete {
-		err = bulkDeleteValidatorEntry(entries, db, operationType)
+		err = bulkDeleteValidatorEntry(entries, db, operationType, cachedEntries)
 	} else {
-		err = bulkInsertValidatorEntry(entries, db, operationType, params)
+		err = bulkInsertValidatorEntry(entries, db, operationType, params, cachedEntries)
 	}
 	if err != nil {
 		return errors.Wrapf(err, "entries.ValidatorBatchOperation: Problem with operation type %v", operationType)

entries/pkid.go

@@ -5,6 +5,7 @@ import (
 	"github.com/deso-protocol/core/lib"
 	"github.com/deso-protocol/state-consumer/consumer"
 	"github.com/golang/glog"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/pkg/errors"
 	"github.com/uptrace/bun"
 )
@@ -127,15 +128,15 @@ func bulkDeletePkidEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationT
 	return nil
 }
 
-func PkidBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams) error {
+func PkidBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// We check before we call this function that there is at least one operation type.
 	// We also ensure before this that all entries have the same operation type.
 	operationType := entries[0].OperationType
 	var err error
 	if operationType == lib.DbOperationTypeDelete {
-		err = bulkDeletePkid(entries, db, operationType)
+		err = bulkDeletePkid(entries, db, operationType, cachedEntries)
 	} else {
-		err = bulkInsertPkid(entries, db, operationType, params)
+		err = bulkInsertPkid(entries, db, operationType, params, cachedEntries)
 	}
 	if err != nil {
 		return errors.Wrapf(err, "entries.PostBatchOperation: Problem with operation type %v", operationType)
@@ -144,12 +145,16 @@ func PkidBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib
 }
 
 // bulkInsertPkid inserts a batch of PKIDs into the database.
-func bulkInsertPkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams) error {
+func bulkInsertPkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
 
 	uniqueLeaderScheduleEntries := consumer.FilterEntriesByPrefix(
 		uniqueEntries, lib.Prefixes.PrefixSnapshotLeaderSchedule)
+
+	// Filter out any entries that are already in the cache.
+	uniqueLeaderScheduleEntries = consumer.FilterCachedEntries(uniqueLeaderScheduleEntries, cachedEntries)
+
 	// NOTE: if we need to support parsing other indexes for PKIDs beyond LeaderSchedule,
 	// we will need to filter the uniqueEntries by the appropriate prefix and then convert
 	// the entries to the appropriate PG struct.
@@ -178,11 +183,16 @@ func bulkInsertPkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType l
 		}
 	}
 
+	// Update the cached entries with the new entries.
+	for _, entry := range uniqueLeaderScheduleEntries {
+		cachedEntries.Add(string(entry.KeyBytes), entry.EncoderBytes)
+	}
+
 	return nil
 }
 
 // bulkDeletePKID deletes a batch of PKIDs from the database.
-func bulkDeletePkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType) error {
+func bulkDeletePkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
 
@@ -201,5 +211,10 @@ func bulkDeletePkid(entries []*lib.StateChangeEntry, db bun.IDB, operationType l
 		}
 	}
 
+	// Remove the entries from the cache.
+	for _, entry := range uniqueEntries {
+		cachedEntries.Remove(string(entry.KeyBytes))
+	}
+
 	return nil
 }

entries/transaction.go

@@ -21,9 +21,9 @@ type TransactionEntry struct {
 	FeeNanos                     uint64
 	NonceExpirationBlockHeight   uint64
 	NoncePartialId               uint64
-	TxnMeta                      lib.DeSoTxnMetadata `bun:"type:jsonb"`
-	TxIndexMetadata              lib.DeSoEncoder     `bun:"type:jsonb"`
-	TxIndexBasicTransferMetadata lib.DeSoEncoder     `bun:"type:jsonb"`
+	TxnMeta                      lib.DeSoTxnMetadata              `bun:"type:jsonb"`
+	TxIndexMetadata              consumer.ConsumerTxIndexMetadata `bun:"type:jsonb"`
+	TxIndexBasicTransferMetadata lib.DeSoEncoder                  `bun:"type:jsonb"`
 	TxnMetaBytes                 []byte
 	TxnBytes                     []byte
 	TxnType                      uint16

entries/utxo_operation.go

@@ -430,7 +430,7 @@ func parseUtxoOperationBundle(
 						entry.BlockHeight,
 					)
 			}
-			txIndexMetadata, err := consumer.ComputeTransactionMetadata(transaction, blockHashHex, params, transaction.TxnFeeNanos, uint64(jj), utxoOps)
+			txIndexMetadata, txnExtraMetadata, err := consumer.ComputeTransactionMetadata(transaction, blockHashHex, params, transaction.TxnFeeNanos, uint64(jj), utxoOps)
 			if err != nil {
 				// If we fail to compute txindex metadata, log the error and continue to the next transaction.
 				// We still append this txn to the transactionUpdates slice so that we can have it in the db.
@@ -453,7 +453,10 @@ func parseUtxoOperationBundle(
 					innerTxnMetadata.BasicTransferTxindexMetadata.UtxoOps = nil
 				}
 			}
-			transactions[jj].TxIndexMetadata = metadata
+			transactions[jj].TxIndexMetadata = consumer.ConsumerTxIndexMetadata{
+				DeSoEncoder:              metadata,
+				TransactionExtraMetadata: txnExtraMetadata,
+			}
 
 			transactions[jj].TxIndexBasicTransferMetadata = txIndexMetadata.GetEncoderForTxType(lib.TxnTypeBasicTransfer)
 

entries/validator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"github.com/deso-protocol/core/lib"
 	"github.com/deso-protocol/state-consumer/consumer"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/pkg/errors"
 	"github.com/uptrace/bun"
 	"github.com/uptrace/bun/extra/bunbig"
@@ -98,15 +99,15 @@ func ValidatorEncoderToPGStruct(validatorEntry *lib.ValidatorEntry, keyBytes []b
 
 // ValidatorBatchOperation is the entry point for processing a batch of Validator entries.
 // It determines the appropriate handler based on the operation type and executes it.
-func ValidatorBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams) error {
+func ValidatorBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// We check before we call this function that there is at least one operation type.
 	// We also ensure before this that all entries have the same operation type.
 	operationType := entries[0].OperationType
 	var err error
 	if operationType == lib.DbOperationTypeDelete {
-		err = bulkDeleteValidatorEntry(entries, db, operationType)
+		err = bulkDeleteValidatorEntry(entries, db, operationType, cachedEntries)
 	} else {
-		err = bulkInsertValidatorEntry(entries, db, operationType, params)
+		err = bulkInsertValidatorEntry(entries, db, operationType, params, cachedEntries)
 	}
 	if err != nil {
 		return errors.Wrapf(err, "entries.ValidatorBatchOperation: Problem with operation type %v", operationType)
@@ -115,9 +116,13 @@ func ValidatorBatchOperation(entries []*lib.StateChangeEntry, db bun.IDB, params
 }
 
 // bulkInsertValidatorEntry inserts a batch of validator entries into the database.
-func bulkInsertValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams) error {
+func bulkInsertValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
+
+	// Filter out any entries that are already tracked in the cache.
+	uniqueEntries = consumer.FilterCachedEntries(uniqueEntries, cachedEntries)
+
 	uniqueValidatorEntries := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixValidatorByPKID)
 	uniqueSnapshotValidatorEntries := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixSnapshotValidatorSetByPKID)
 	// Create a new array to hold the bun struct.
@@ -156,11 +161,17 @@ func bulkInsertValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, opera
 			return errors.Wrapf(err, "entries.bulkInsertValidatorEntry: Error inserting snapshot validator entries")
 		}
 	}
+
+	// Add any new entries to the cache.
+	for _, entry := range uniqueEntries {
+		cachedEntries.Add(string(entry.KeyBytes), entry.EncoderBytes)
+	}
+
 	return nil
 }
 
 // bulkDeleteValidatorEntry deletes a batch of validator entries from the database.
-func bulkDeleteValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType) error {
+func bulkDeleteValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, operationType lib.StateSyncerOperationType, cachedEntries *lru.Cache[string, []byte]) error {
 	// Track the unique entries we've inserted so we don't insert the same entry twice.
 	uniqueEntries := consumer.UniqueEntries(entries)
 	uniqueKeys := consumer.KeysToDelete(uniqueEntries)
@@ -195,6 +206,12 @@ func bulkDeleteValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, opera
 		}
 	}
 
+	// Delete cached validator entries.
+	for _, key := range validatorKeysToDelete {
+		keyStr := string(key)
+		cachedEntries.Remove(keyStr)
+	}
+
 	return nil
 }
 

handler/data_handler.go

@@ -11,6 +11,7 @@ import (
 	"github.com/deso-protocol/postgres-data-handler/migrations/post_sync_migrations"
 	"github.com/deso-protocol/state-consumer/consumer"
 	"github.com/golang/glog"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/pkg/errors"
 	"github.com/uptrace/bun"
 )
@@ -25,6 +26,9 @@ type PostgresDataHandler struct {
 	// Params is a struct containing the current blockchain parameters.
 	// It is used to determine which prefix to use for public keys.
 	Params *lib.DeSoParams
+
+	// LRU containing cached entries, to reduce duplicative database operations
+	CachedEntries *lru.Cache[string, []byte]
 }
 
 // HandleEntryBatch performs a bulk operation for a batch of entries, based on the encoder type.
@@ -92,7 +96,7 @@ func (postgresDataHandler *PostgresDataHandler) HandleEntryBatch(batchedEntries
 	case lib.EncoderTypeStakeEntry:
 		err = entries.StakeBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
 	case lib.EncoderTypeValidatorEntry:
-		err = entries.ValidatorBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
+		err = entries.ValidatorBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params, postgresDataHandler.CachedEntries)
 	case lib.EncoderTypeLockedStakeEntry:
 		err = entries.LockedStakeBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
 	case lib.EncoderTypeLockedBalanceEntry:
@@ -102,11 +106,11 @@ func (postgresDataHandler *PostgresDataHandler) HandleEntryBatch(batchedEntries
 	case lib.EncoderTypeEpochEntry:
 		err = entries.EpochEntryBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
 	case lib.EncoderTypePKID:
-		err = entries.PkidBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
+		err = entries.PkidBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params, postgresDataHandler.CachedEntries)
 	case lib.EncoderTypeGlobalParamsEntry:
 		err = entries.GlobalParamsBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
 	case lib.EncoderTypeBLSPublicKeyPKIDPairEntry:
-		err = entries.BLSPublicKeyPKIDPairBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
+		err = entries.BLSPublicKeyPKIDPairBatchOperation(batchedEntries, dbHandle, postgresDataHandler.Params, postgresDataHandler.CachedEntries)
 	case lib.EncoderTypeBlockNode:
 		err = entries.BlockNodeOperation(batchedEntries, dbHandle, postgresDataHandler.Params)
 	}

handler/db_utils.go

@@ -17,6 +17,10 @@ const (
 	MigrationTypePostHypersync MigrationType = 1
 )
 
+const (
+	EntryCacheSize uint = 1000000 // 1M entries
+)
+
 // TODO: Make this a method on the PostgresDataHandler struct.
 func RunMigrations(db *bun.DB, reset bool, migrationType MigrationType) error {
 	ctx := context.Background()

main.go

@@ -10,6 +10,7 @@ import (
 	"github.com/deso-protocol/postgres-data-handler/migrations/post_sync_migrations"
 	"github.com/deso-protocol/state-consumer/consumer"
 	"github.com/golang/glog"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/spf13/viper"
 	"github.com/uptrace/bun"
 	"github.com/uptrace/bun/dialect/pgdialect"
@@ -70,6 +71,11 @@ func main() {
 	}
 	lib.GlobalDeSoParams = *params
 
+	cachedEntries, err := lru.New[string, []byte](int(handler.EntryCacheSize))
+	if err != nil {
+		glog.Fatalf("Error creating LRU cache: %v", err)
+	}
+
 	// Initialize and run a state syncer consumer.
 	stateSyncerConsumer := &consumer.StateSyncerConsumer{}
 	err = stateSyncerConsumer.InitializeAndRun(
@@ -79,8 +85,9 @@ func main() {
 		threadLimit,
 		syncMempool,
 		&handler.PostgresDataHandler{
-			DB:     db,
-			Params: params,
+			DB:            db,
+			Params:        params,
+			CachedEntries: cachedEntries,
 		},
 	)
 	if err != nil {