Z/cache and pubsub

entries/block.go

@@ -124,6 +124,7 @@ func bulkInsertBlockEntry(entries []*lib.StateChangeEntry, db bun.IDB, operation
 	for _, entry := range uniqueBlocks {
 		block := entry.Encoder.(*lib.MsgDeSoBlock)
 		blockEntry, blockSigners := BlockEncoderToPGStruct(block, entry.KeyBytes, params)
+
 		pgBlockEntrySlice = append(pgBlockEntrySlice, blockEntry)
 		pgBlockSignersEntrySlice = append(pgBlockSignersEntrySlice, blockSigners...)
 		for jj, transaction := range block.Txns {

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,14 @@ 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,
+	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 +150,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 +192,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/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)
@@ -182,6 +193,12 @@ func bulkDeleteValidatorEntry(entries []*lib.StateChangeEntry, db bun.IDB, opera
 			Exec(context.Background()); err != nil {
 			return errors.Wrapf(err, "entries.bulkDeleteValidatorEntry: Error deleting entries")
 		}
+
+		// Delete cached validator entries.
+		for _, key := range validatorKeysToDelete {
+			keyStr := string(key)
+			cachedEntries.Remove(keyStr)
+		}
 	}
 
 	// Execute the delete query for snapshot validator entries.

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/deso-protocol/state-consumer v1.0.3
 	github.com/golang/glog v1.2.2
 	github.com/google/uuid v1.6.0
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/pkg/errors v0.9.1
 	github.com/spf13/viper v1.19.0
 	github.com/uptrace/bun v1.2.3
@@ -58,7 +59,7 @@ require (
 	github.com/btcsuite/btcd/btcec/v2 v2.3.4 // indirect
 	github.com/btcsuite/btcd/btcutil v1.1.6 // indirect
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0 // indirect
-	github.com/btcsuite/btclog v0.0.0-20241003133417-09c4e92e319c // indirect
+	github.com/btcsuite/btclog v0.0.0-20241017175713-3428138b75c7 // indirect
 	github.com/bwesterb/go-ristretto v1.2.3 // indirect
 	github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
@@ -79,7 +80,7 @@ require (
 	github.com/eapache/queue/v2 v2.0.0-20230407133247-75960ed334e4 // indirect
 	github.com/ebitengine/purego v0.8.0 // indirect
 	github.com/emirpasic/gods v1.18.1 // indirect
-	github.com/envoyproxy/go-control-plane v0.13.0 // indirect
+	github.com/envoyproxy/go-control-plane v0.13.1 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.1.0 // indirect
 	github.com/ethereum/go-ethereum v1.14.11 // indirect
 	github.com/fatih/color v1.17.0 // indirect
@@ -131,7 +132,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/montanaflynn/stats v0.7.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
-	github.com/nyaruka/phonenumbers v1.4.0 // indirect
+	github.com/nyaruka/phonenumbers v1.4.1 // indirect
 	github.com/oleiade/lane v1.0.1 // indirect
 	github.com/onflow/crypto v0.25.2 // indirect
 	github.com/outcaste-io/ristretto v0.2.3 // indirect
@@ -142,7 +143,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/puzpuzpuz/xsync/v3 v3.4.0 // indirect
 	github.com/richardartoul/molecule v1.0.1-0.20240531184615-7ca0df43c0b3 // indirect
-	github.com/robinjoseph08/go-pg-migrations/v3 v3.0.0 // indirect
+	github.com/robinjoseph08/go-pg-migrations/v3 v3.1.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/ryanuber/go-glob v1.0.0 // indirect
 	github.com/sagikazarmark/locafero v0.6.0 // indirect
@@ -204,7 +205,7 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20241015192408-796eee8c2d53 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20241015192408-796eee8c2d53 // indirect
 	google.golang.org/grpc v1.67.1 // indirect
-	google.golang.org/grpc/stats/opentelemetry v0.0.0-20241016173057-569c8eb0af32 // indirect
+	google.golang.org/grpc/stats/opentelemetry v0.0.0-20241017163036-56df169480cd // indirect
 	google.golang.org/protobuf v1.35.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect