Add LeaderScheduleEntry

Author: lazynina

entries/epoch.go

@@ -14,9 +14,8 @@ type EpochEntry struct {
 	InitialBlockHeight              uint64
 	InitialView                     uint64
 	FinalBlockHeight                uint64
-	CreatedAtBlockTimestampNanoSecs uint64
-
-	BadgerKey []byte `pg:",pk,use_zero"`
+	CreatedAtBlockTimestampNanoSecs int64
+	SnapshotAtEpochNumber           uint64
 }
 
 type PGEpochEntry struct {
@@ -33,12 +32,19 @@ type PGEpochUtxoOps struct {
 
 // Convert the EpochEntry DeSo encoder to the PGEpochEntry struct used by bun.
 func EpochEntryEncoderToPGStruct(epochEntry *lib.EpochEntry, keyBytes []byte, params *lib.DeSoParams) EpochEntry {
+
+	var snapshotAtEpochNumber uint64
+	// Epochs use data snapshotted from two epochs ago. Epochs 0 and 1 use data from epoch 0.
+	if epochEntry.EpochNumber >= 2 {
+		snapshotAtEpochNumber = epochEntry.EpochNumber - 2
+	}
 	return EpochEntry{
-		EpochNumber:        epochEntry.EpochNumber,
-		InitialBlockHeight: epochEntry.InitialBlockHeight,
-		InitialView:        epochEntry.InitialView,
-		FinalBlockHeight:   epochEntry.FinalBlockHeight,
-		BadgerKey:          keyBytes,
+		EpochNumber:                     epochEntry.EpochNumber,
+		InitialBlockHeight:              epochEntry.InitialBlockHeight,
+		InitialView:                     epochEntry.InitialView,
+		FinalBlockHeight:                epochEntry.FinalBlockHeight,
+		CreatedAtBlockTimestampNanoSecs: epochEntry.CreatedAtBlockTimestampNanoSecs,
+		SnapshotAtEpochNumber:           snapshotAtEpochNumber,
 	}
 }
 
@@ -49,8 +55,11 @@ func EpochEntryBatchOperation(entries []*lib.StateChangeEntry, db *bun.DB, param
 	// We also ensure before this that all entries have the same operation type.
 	operationType := entries[0].OperationType
 	var err error
+	// Core only tracks the current epoch entry and never deletes them.
+	// In order to track all historical epoch entries, we don't use the badger
+	// key to uniquely identify them, but rather the epoch number.
 	if operationType == lib.DbOperationTypeDelete {
-		err = bulkDeleteEpochEntry(entries, db, operationType)
+		return errors.Wrapf(err, "entries.EpochEntryBatchOperation: Delete operation type not supported")
 	} else {
 		err = bulkInsertEpochEntry(entries, db, operationType, params)
 	}
@@ -76,31 +85,11 @@ func bulkInsertEpochEntry(entries []*lib.StateChangeEntry, db *bun.DB, operation
 	query := db.NewInsert().Model(&pgEntrySlice)
 
 	if operationType == lib.DbOperationTypeUpsert {
-		query = query.On("CONFLICT (badger_key) DO UPDATE")
+		query = query.On("CONFLICT (epoch_number) DO UPDATE")
 	}
 
 	if _, err := query.Returning("").Exec(context.Background()); err != nil {
 		return errors.Wrapf(err, "entries.bulkInsertEpochEntry: Error inserting entries")
 	}
 	return nil
 }
-
-// bulkDeleteEpochEntry deletes a batch of locked stake entries from the database.
-func bulkDeleteEpochEntry(entries []*lib.StateChangeEntry, db *bun.DB, operationType lib.StateSyncerOperationType) error {
-	// Track the unique entries we've inserted so we don't insert the same entry twice.
-	uniqueEntries := consumer.UniqueEntries(entries)
-
-	// Transform the entries into a list of keys to delete.
-	keysToDelete := consumer.KeysToDelete(uniqueEntries)
-
-	// Execute the delete query.
-	if _, err := db.NewDelete().
-		Model(&PGEpochEntry{}).
-		Where("badger_key IN (?)", bun.In(keysToDelete)).
-		Returning("").
-		Exec(context.Background()); err != nil {
-		return errors.Wrapf(err, "entries.bulkDeleteEpochEntry: Error deleting entries")
-	}
-
-	return nil
-}

entries/pkid.go

@@ -1,7 +1,6 @@
 package entries
 
 import (
-	"bytes"
 	"context"
 	"github.com/deso-protocol/core/lib"
 	"github.com/deso-protocol/state-consumer/consumer"
@@ -131,7 +130,6 @@ func bulkDeletePkidEntry(entries []*lib.StateChangeEntry, db *bun.DB, operationT
 func PkidBatchOperation(entries []*lib.StateChangeEntry, db *bun.DB, params *lib.DeSoParams) 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.
-	glog.Infof("PkidBatchOperation: Putting %d entries into the database", len(entries))
 	operationType := entries[0].OperationType
 	var err error
 	if operationType == lib.DbOperationTypeDelete {
@@ -149,21 +147,23 @@ func PkidBatchOperation(entries []*lib.StateChangeEntry, db *bun.DB, params *lib
 func bulkInsertPkid(entries []*lib.StateChangeEntry, db *bun.DB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams) 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)
 	// NOTE: if we need to support parsing other indexes for PKIDs beyond LeaderSchedule,
-	// we will need to create an
+	// we will need to filter the uniqueEntries by the appropriate prefix and then convert
+	// the entries to the appropriate PG struct.
 	// Create a new array to hold the bun struct.
-	pgEntrySlice := make([]*PGLeaderScheduleEntry, len(uniqueEntries))
+	pgEntrySlice := make([]*PGLeaderScheduleEntry, len(uniqueLeaderScheduleEntries))
 
 	// Loop through the entries and convert them to PGPostEntry.
-	for ii, entry := range uniqueEntries {
-		if bytes.Equal(entry.KeyBytes[0:1], lib.Prefixes.PrefixSnapshotLeaderSchedule) {
-			leaderScheduleEntry := LeaderScheduleEncoderToPGStruct(entry.Encoder.(*lib.PKID), entry.KeyBytes, params)
-			if leaderScheduleEntry == nil {
-				glog.Errorf("bulkInsertPkid: Error converting LeaderScheduleEntry to PG struct")
-				continue
-			}
-			pgEntrySlice[ii] = &PGLeaderScheduleEntry{LeaderScheduleEntry: *leaderScheduleEntry}
+	for ii, entry := range uniqueLeaderScheduleEntries {
+		leaderScheduleEntry := LeaderScheduleEncoderToPGStruct(entry.Encoder.(*lib.PKID), entry.KeyBytes, params)
+		if leaderScheduleEntry == nil {
+			glog.Errorf("bulkInsertPkid: Error converting LeaderScheduleEntry to PG struct")
+			continue
 		}
+		pgEntrySlice[ii] = &PGLeaderScheduleEntry{LeaderScheduleEntry: *leaderScheduleEntry}
 	}
 
 	query := db.NewInsert().Model(&pgEntrySlice)
@@ -185,12 +185,12 @@ func bulkDeletePkid(entries []*lib.StateChangeEntry, db *bun.DB, operationType l
 
 	// Transform the entries into a list of keys to delete.
 	keysToDelete := consumer.KeysToDelete(uniqueEntries)
-	keysToDelete = consumer.FilterKeysByPrefix(keysToDelete, lib.Prefixes.PrefixSnapshotLeaderSchedule)
+	leaderSchedKeysToDelete := consumer.FilterKeysByPrefix(keysToDelete, lib.Prefixes.PrefixSnapshotLeaderSchedule)
 
 	// Execute the delete query.
 	if _, err := db.NewDelete().
 		Model(&LeaderScheduleEntry{}).
-		Where("badger_key IN (?)", bun.In(keysToDelete)).
+		Where("badger_key IN (?)", bun.In(leaderSchedKeysToDelete)).
 		Returning("").
 		Exec(context.Background()); err != nil {
 		return errors.Wrapf(err, "entries.bulkDeletePkid: Error deleting entries")

migrations/initial_migrations/20240129000003_create_epoch_entry_table.go

@@ -12,14 +12,18 @@ import (
 func createEpochEntryTable(db *bun.DB, tableName string) error {
 	_, err := db.Exec(strings.Replace(`
 			CREATE TABLE {tableName} (
-				epoch_number BIGINT NOT NULL,
+				epoch_number BIGINT PRIMARY KEY NOT NULL,
 				initial_block_height BIGINT NOT NULL,
 				initial_view BIGINT NOT NULL,
 				final_block_height BIGINT NOT NULL,
 				created_at_block_timestamp_nano_secs BIGINT NOT NULL,
-
-				badger_key BYTEA PRIMARY KEY 
+				snapshot_at_epoch_number BIGINT NOT NULL			
 			);
+
+			CREATE INDEX {tableName}_epoch_number_idx ON {tableName} (epoch_number);
+			CREATE INDEX {tableName}_initial_block_height_idx ON {tableName} (initial_block_height);
+			CREATE INDEX {tableName}_final_block_height_idx ON {tableName} (final_block_height);
+			CREATE INDEX {tableName}_snapshot_at_epoch_number_idx ON {tableName} (snapshot_at_epoch_number);
 		`, "{tableName}", tableName, -1))
 	// TODO: What other fields do we need indexed?
 	return err

migrations/initial_migrations/20240215000001_create_leader_schedule.go

@@ -19,7 +19,6 @@ func createLeaderScheduleTable(db *bun.DB, tableName string) error {
 			CREATE INDEX {tableName}_snapshot_at_epoch_number_idx ON {tableName} (snapshot_at_epoch_number);
 			CREATE INDEX {tableName}_snapshot_at_epoch_number_leader_index_idx ON {tableName} (snapshot_at_epoch_number, leader_index);
 		`, "{tableName}", tableName, -1))
-	// TODO: What other fields do we need indexed?
 	return err
 }
 

migrations/post_sync_migrations/20240213000002_create_pos_fk_comments.go

@@ -22,7 +22,6 @@ func init() {
                 comment on column locked_stake_entry.badger_key is E'@omit';
                 comment on column yield_curve_point.badger_key is E'@omit';
                 comment on column locked_balance_entry.badger_key is E'@omit';
-                comment on column epoch_entry.badger_key is E'@omit';
                 comment on table transaction_partition_34 is E'@omit';
                 comment on table transaction_partition_35 is E'@omit';
                 comment on table transaction_partition_36 is E'@omit';

migrations/post_sync_migrations/20240215000002_create_leader_schedule_fk_comments.go

@@ -6,11 +6,10 @@ import (
 	"github.com/uptrace/bun"
 )
 
-// TODO: revisit access group relationships when we refactor the messaging app to use the graphql API.
 func init() {
 	Migrations.MustRegister(func(ctx context.Context, db *bun.DB) error {
 		_, err := db.Exec(`
-                comment on table leader_schedule_entry is E'@foreignKey (validator_pkid) references account (pkid)|@foreignFieldName leaderScheduleEntries|@fieldName leaderAccount\n@foreignKey (validator_pkid) references validator_entry (validator_pkid)|@foreignFieldName leaderScheduleEntries|@fieldName validatorEntry';
+                comment on table leader_schedule_entry is E'@foreignKey (validator_pkid) references account (pkid)|@foreignFieldName leaderScheduleEntries|@fieldName leaderAccount\n@foreignKey (validator_pkid) references validator_entry (validator_pkid)|@foreignFieldName leaderScheduleEntries|@fieldName validatorEntry\n@foreignKey (snapshot_at_epoch_number) references epoch_entry (snapshot_at_epoch_number)|@foreignFieldName leaderScheduleEntries|@fieldName epochEntryBySnapshot';
                 comment on column leader_schedule_entry.badger_key is E'@omit';
 		`)
 		if err != nil {