Skip to content

Commit 7ae0a61

Browse files
lazyninalazynina
committed
Snapshot validator entries and block signers + associated views
1 parent c69997a commit 7ae0a61

File tree

6 files changed

+281
-20
lines changed

6 files changed

+281
-20
lines changed

‎entries/block.go‎

Lines changed: 72 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package entries
33
import (
44
"context"
55
"encoding/hex"
6+
"reflect"
67
"time"
78

89
"github.com/deso-protocol/core/lib"
@@ -34,12 +35,43 @@ type PGBlockEntry struct {
3435
BlockEntry
3536
}
3637

38+
type BlockSigner struct {
39+
BlockHash string
40+
SignerIndex uint64
41+
}
42+
43+
type PGBlockSigner struct {
44+
bun.BaseModel `bun:"table:block_signer"`
45+
BlockSigner
46+
}
47+
3748
// Convert the UserAssociation DeSo encoder to the PG struct used by bun.
38-
func BlockEncoderToPGStruct(block *lib.MsgDeSoBlock, keyBytes []byte, params *lib.DeSoParams) *PGBlockEntry {
49+
func BlockEncoderToPGStruct(block *lib.MsgDeSoBlock, keyBytes []byte, params *lib.DeSoParams) (*PGBlockEntry, []*PGBlockSigner) {
3950
blockHash, _ := block.Hash()
51+
blockHashHex := hex.EncodeToString(blockHash[:])
52+
qc := block.Header.GetQC()
53+
blockSigners := []*PGBlockSigner{}
54+
if !isInterfaceNil(qc) {
55+
aggSig := qc.GetAggregatedSignature()
56+
if !isInterfaceNil(aggSig) {
57+
signersList := aggSig.GetSignersList()
58+
for ii := 0; ii < signersList.Size(); ii++ {
59+
// Skip signers that didn't sign.
60+
if !signersList.Get(ii) {
61+
continue
62+
}
63+
blockSigners = append(blockSigners, &PGBlockSigner{
64+
BlockSigner: BlockSigner{
65+
BlockHash: blockHashHex,
66+
SignerIndex: uint64(ii),
67+
},
68+
})
69+
}
70+
}
71+
}
4072
return &PGBlockEntry{
4173
BlockEntry: BlockEntry{
42-
BlockHash: hex.EncodeToString(blockHash[:]),
74+
BlockHash: blockHashHex,
4375
PrevBlockHash: hex.EncodeToString(block.Header.PrevBlockHash[:]),
4476
TxnMerkleRoot: hex.EncodeToString(block.Header.TransactionMerkleRoot[:]),
4577
Timestamp: consumer.UnixNanoToTime(uint64(block.Header.TstampNanoSecs)),
@@ -53,7 +85,7 @@ func BlockEncoderToPGStruct(block *lib.MsgDeSoBlock, keyBytes []byte, params *li
5385
ProposerVotePartialSignature: block.Header.ProposerVotePartialSignature.ToString(),
5486
BadgerKey: keyBytes,
5587
},
56-
}
88+
}, blockSigners
5789
}
5890

5991
// PostBatchOperation is the entry point for processing a batch of post entries. It determines the appropriate handler
@@ -120,11 +152,13 @@ func bulkInsertBlockEntry(entries []*lib.StateChangeEntry, db *bun.DB, operation
120152
// Create a new array to hold the bun struct.
121153
pgBlockEntrySlice := make([]*PGBlockEntry, 0)
122154
pgTransactionEntrySlice := make([]*PGTransactionEntry, 0)
155+
pgBlockSignersEntrySlice := make([]*PGBlockSigner, 0)
123156

124157
for _, entry := range uniqueBlocks {
125158
block := entry.Encoder.(*lib.MsgDeSoBlock)
126-
blockEntry := BlockEncoderToPGStruct(block, entry.KeyBytes, params)
159+
blockEntry, blockSigners := BlockEncoderToPGStruct(block, entry.KeyBytes, params)
127160
pgBlockEntrySlice = append(pgBlockEntrySlice, blockEntry)
161+
pgBlockSignersEntrySlice = append(pgBlockSignersEntrySlice, blockSigners...)
128162
for jj, transaction := range block.Txns {
129163
indexInBlock := uint64(jj)
130164
pgTransactionEntry, err := TransactionEncoderToPGStruct(
@@ -166,6 +200,19 @@ func bulkInsertBlockEntry(entries []*lib.StateChangeEntry, db *bun.DB, operation
166200
return errors.Wrapf(err, "entries.bulkInsertBlock: Error inserting transaction entries")
167201
}
168202

203+
if len(pgBlockSignersEntrySlice) > 0 {
204+
// Execute the insert query.
205+
query := db.NewInsert().Model(&pgBlockSignersEntrySlice)
206+
207+
if operationType == lib.DbOperationTypeUpsert {
208+
query = query.On("CONFLICT (block_hash, signer_index) DO UPDATE")
209+
}
210+
211+
if _, err := query.Returning("").Exec(context.Background()); err != nil {
212+
return errors.Wrapf(err, "entries.bulkInsertBlockEntry: Error inserting block signers")
213+
}
214+
}
215+
169216
return nil
170217
}
171218

@@ -214,5 +261,26 @@ func bulkDeleteBlockEntriesFromKeysToDelete(db *bun.DB, keysToDelete [][]byte) e
214261
Exec(context.Background()); err != nil {
215262
return errors.Wrapf(err, "entries.bulkDeleteBlockEntry: Error deleting utxo operation entries")
216263
}
264+
265+
// Delete any signers associated with the block.
266+
if _, err := db.NewDelete().
267+
Model(&PGBlockSigner{}).
268+
Where("block_hash IN (?)", bun.In(blockHashHexesToDelete)).
269+
Returning("").
270+
Exec(context.Background()); err != nil {
271+
return errors.Wrapf(err, "entries.bulkDeleteBlockEntry: Error deleting block signers")
272+
}
217273
return nil
218274
}
275+
276+
// golang interface types are stored as a tuple of (type, value). A single i==nil check is not enough to
277+
// determine if a pointer that implements an interface is nil. This function checks if the interface is nil
278+
// by checking if the pointer itself is nil.
279+
func isInterfaceNil(i interface{}) bool {
280+
if i == nil {
281+
return true
282+
}
283+
284+
value := reflect.ValueOf(i)
285+
return value.Kind() == reflect.Ptr && value.IsNil()
286+
}

‎entries/utxo_operation.go‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -93,13 +93,14 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
9393
transactionUpdates := make([]*PGTransactionEntry, 0)
9494
affectedPublicKeys := make([]*PGAffectedPublicKeyEntry, 0)
9595
blockEntries := make([]*PGBlockEntry, 0)
96+
pgBlockSigners := make([]*PGBlockSigner, 0)
9697
stakeRewardEntries := make([]*PGStakeReward, 0)
9798
jailedHistoryEntries := make([]*PGJailedHistoryEvent, 0)
9899

99100
// Start timer to track how long it takes to insert the entries.
100101
start := time.Now()
101102

102-
fmt.Printf("entries.bulkInsertUtxoOperationsEntry: Inserting %v entries\n", len(uniqueEntries))
103+
glog.V(2).Infof("entries.bulkInsertUtxoOperationsEntry: Inserting %v entries\n", len(uniqueEntries))
103104
transactionCount := 0
104105

105106
// Whether we are inserting transactions for the first time, or just updating them.
@@ -126,8 +127,9 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
126127
if entry.Block != nil {
127128
insertTransactions = true
128129
block := entry.Block
129-
blockEntry := BlockEncoderToPGStruct(block, entry.KeyBytes, params)
130+
blockEntry, blockSigners := BlockEncoderToPGStruct(block, entry.KeyBytes, params)
130131
blockEntries = append(blockEntries, blockEntry)
132+
pgBlockSigners = append(pgBlockSigners, blockSigners...)
131133
for ii, txn := range block.Txns {
132134
indexInBlock := uint64(ii)
133135
pgTxn, err := TransactionEncoderToPGStruct(
@@ -277,7 +279,7 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
277279
transactionCount += len(innerTransactionsUtxoOperations)
278280
// Print how long it took to insert the entries.
279281
}
280-
fmt.Printf("entries.bulkInsertUtxoOperationsEntry: Processed %v txns in %v s\n", transactionCount, time.Since(start))
282+
glog.V(2).Infof("entries.bulkInsertUtxoOperationsEntry: Processed %v txns in %v s\n", transactionCount, time.Since(start))
281283

282284
start = time.Now()
283285

@@ -299,6 +301,16 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
299301
return errors.Wrapf(err, "entries.bulkInsertBlock: Error inserting entries")
300302
}
301303

304+
blockSignerQuery := db.NewInsert().Model(&pgBlockSigners)
305+
306+
if operationType == lib.DbOperationTypeUpsert {
307+
blockSignerQuery = blockSignerQuery.On("CONFLICT (block_hash, signer_index) DO UPDATE")
308+
}
309+
310+
if _, err := blockSignerQuery.Exec(context.Background()); err != nil {
311+
return errors.Wrapf(err, "entries.bulkInsertBlockSigners: Error inserting block signer entries")
312+
}
313+
302314
} else {
303315
values := db.NewValues(&transactionUpdates)
304316
_, err := db.NewUpdate().
@@ -317,7 +329,7 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
317329
}
318330
}
319331

320-
fmt.Printf("entries.bulkInsertUtxoOperationsEntry: Updated %v txns in %v s\n", len(transactionUpdates), time.Since(start))
332+
glog.V(2).Infof("entries.bulkInsertUtxoOperationsEntry: Updated %v txns in %v s\n", len(transactionUpdates), time.Since(start))
321333

322334
start = time.Now()
323335

@@ -329,7 +341,7 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
329341
}
330342
}
331343

332-
fmt.Printf("entries.bulkInsertUtxoOperationsEntry: Inserted %v affected public keys in %v s\n", len(affectedPublicKeys), time.Since(start))
344+
glog.V(2).Infof("entries.bulkInsertUtxoOperationsEntry: Inserted %v affected public keys in %v s\n", len(affectedPublicKeys), time.Since(start))
333345

334346
start = time.Now()
335347

@@ -340,7 +352,7 @@ func bulkInsertUtxoOperationsEntry(entries []*lib.StateChangeEntry, db *bun.DB,
340352
return errors.Wrapf(err, "InsertStakeRewards: Problem inserting stake rewards")
341353
}
342354
}
343-
fmt.Printf("entries.bulkInsertUtxoOperationsEntry: Inserted %v stake rewards in %v s\n", len(stakeRewardEntries), time.Since(start))
355+
glog.V(2).Infof("entries.bulkInsertUtxoOperationsEntry: Inserted %v stake rewards in %v s\n", len(stakeRewardEntries), time.Since(start))
344356

345357
if len(jailedHistoryEntries) > 0 {
346358
_, err := db.NewInsert().Model(&jailedHistoryEntries).On("CONFLICT (validator_pkid, jailed_at_epoch_number, unjailed_at_epoch_number) DO NOTHING").Exec(context.Background())
@@ -410,6 +422,9 @@ func parseUtxoOperationBundle(
410422
}
411423
txIndexMetadata, err := consumer.ComputeTransactionMetadata(transaction, blockHashHex, params, transaction.TxnFeeNanos, uint64(jj), utxoOps)
412424
if err != nil {
425+
glog.Errorf("parseUtxoOperationBundle: Problem computing transaction metadata for "+
426+
"entry %+v at block height %v: %v", entry, entry.BlockHeight, err)
427+
// TODO: swallow error and continue.
413428
return nil,
414429
nil,
415430
nil,

‎entries/validator.go‎

Lines changed: 100 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,28 @@ type PGValidatorEntryUtxoOps struct {
3838
UtxoOperation
3939
}
4040

41+
type SnapshotValidatorEntry struct {
42+
ValidatorPKID string `bun:",nullzero"`
43+
Domains []string `bun:",array"`
44+
DisableDelegatedStake bool
45+
DelegatedStakeCommissionBasisPoints uint64
46+
VotingPublicKey string `bun:",nullzero"`
47+
VotingAuthorization string `bun:",nullzero"`
48+
// Use bunbig.Int to store the balance as a numeric in the pg database.
49+
TotalStakeAmountNanos *bunbig.Int `pg:",use_zero"`
50+
LastActiveAtEpochNumber uint64
51+
JailedAtEpochNumber uint64
52+
SnapshotAtEpochNumber uint64 `pg:",use_zero"`
53+
54+
ExtraData map[string]string `bun:"type:jsonb"`
55+
BadgerKey []byte `pg:",pk,use_zero"`
56+
}
57+
58+
type PGSnapshotValidatorEntry struct {
59+
bun.BaseModel `bun:"table:snapshot_validator_entry"`
60+
SnapshotValidatorEntry
61+
}
62+
4163
// Convert the ValidatorEntry DeSo encoder to the PGValidatorEntry struct used by bun.
4264
func ValidatorEncoderToPGStruct(validatorEntry *lib.ValidatorEntry, keyBytes []byte, params *lib.DeSoParams) ValidatorEntry {
4365
pgValidatorEntry := ValidatorEntry{
@@ -96,23 +118,43 @@ func ValidatorBatchOperation(entries []*lib.StateChangeEntry, db *bun.DB, params
96118
func bulkInsertValidatorEntry(entries []*lib.StateChangeEntry, db *bun.DB, operationType lib.StateSyncerOperationType, params *lib.DeSoParams) error {
97119
// Track the unique entries we've inserted so we don't insert the same entry twice.
98120
uniqueEntries := consumer.UniqueEntries(entries)
121+
uniqueValidatorEntries := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixValidatorByPKID)
122+
uniqueSnapshotValidatorEntries := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixSnapshotValidatorSetByPKID)
99123
// Create a new array to hold the bun struct.
100-
pgEntrySlice := make([]*PGValidatorEntry, len(uniqueEntries))
124+
pgEntrySlice := make([]*PGValidatorEntry, len(uniqueValidatorEntries))
125+
pgSnapshotEntrySlice := make([]*PGSnapshotValidatorEntry, len(uniqueSnapshotValidatorEntries))
101126

102127
// Loop through the entries and convert them to PGEntry.
103-
for ii, entry := range uniqueEntries {
128+
for ii, entry := range uniqueValidatorEntries {
104129
pgEntrySlice[ii] = &PGValidatorEntry{ValidatorEntry: ValidatorEncoderToPGStruct(entry.Encoder.(*lib.ValidatorEntry), entry.KeyBytes, params)}
105130
}
131+
for ii, entry := range uniqueSnapshotValidatorEntries {
132+
pgSnapshotEntrySlice[ii] = &PGSnapshotValidatorEntry{SnapshotValidatorEntry: SnapshotValidatorEncoderToPGStruct(entry.Encoder.(*lib.ValidatorEntry), entry.KeyBytes, params)}
133+
}
106134

107135
// Execute the insert query.
108-
query := db.NewInsert().Model(&pgEntrySlice)
136+
if len(pgEntrySlice) > 0 {
137+
query := db.NewInsert().Model(&pgEntrySlice)
138+
139+
if operationType == lib.DbOperationTypeUpsert {
140+
query = query.On("CONFLICT (badger_key) DO UPDATE")
141+
}
109142

110-
if operationType == lib.DbOperationTypeUpsert {
111-
query = query.On("CONFLICT (badger_key) DO UPDATE")
143+
if _, err := query.Returning("").Exec(context.Background()); err != nil {
144+
return errors.Wrapf(err, "entries.bulkInsertValidatorEntry: Error inserting validator entries")
145+
}
112146
}
113147

114-
if _, err := query.Returning("").Exec(context.Background()); err != nil {
115-
return errors.Wrapf(err, "entries.bulkInsertValidatorEntry: Error inserting entries")
148+
if len(pgSnapshotEntrySlice) > 0 {
149+
query := db.NewInsert().Model(&pgSnapshotEntrySlice)
150+
151+
if operationType == lib.DbOperationTypeUpsert {
152+
query = query.On("CONFLICT (badger_key) DO UPDATE")
153+
}
154+
155+
if _, err := query.Returning("").Exec(context.Background()); err != nil {
156+
return errors.Wrapf(err, "entries.bulkInsertValidatorEntry: Error inserting snapshot validator entries")
157+
}
116158
}
117159
return nil
118160
}
@@ -123,16 +165,64 @@ func bulkDeleteValidatorEntry(entries []*lib.StateChangeEntry, db *bun.DB, opera
123165
uniqueEntries := consumer.UniqueEntries(entries)
124166

125167
// Transform the entries into a list of keys to delete.
126-
keysToDelete := consumer.KeysToDelete(uniqueEntries)
168+
validatorEntriesToDelete := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixValidatorByPKID)
127169

128-
// Execute the delete query.
170+
snapshotValidatorEntriesToDelete := consumer.FilterEntriesByPrefix(uniqueEntries, lib.Prefixes.PrefixSnapshotValidatorSetByPKID)
171+
172+
// Execute the delete query for validator entries.
129173
if _, err := db.NewDelete().
130174
Model(&PGValidatorEntry{}).
131-
Where("badger_key IN (?)", bun.In(keysToDelete)).
175+
Where("badger_key IN (?)", bun.In(validatorEntriesToDelete)).
132176
Returning("").
133177
Exec(context.Background()); err != nil {
134178
return errors.Wrapf(err, "entries.bulkDeleteValidatorEntry: Error deleting entries")
135179
}
136180

181+
// Execute the delete query.
182+
if _, err := db.NewDelete().
183+
Model(&PGSnapshotValidatorEntry{}).
184+
Where("badger_key IN (?)", bun.In(snapshotValidatorEntriesToDelete)).
185+
Returning("").
186+
Exec(context.Background()); err != nil {
187+
return errors.Wrapf(err, "entries.bulkDeleteSnapshotValidatorEntry: Error deleting entries")
188+
}
189+
137190
return nil
138191
}
192+
193+
// Convert the SnapshotValidatorEntry DeSo encoder to the PGSnapshotValidatorEntry struct used by bun.
194+
func SnapshotValidatorEncoderToPGStruct(validatorEntry *lib.ValidatorEntry, keyBytes []byte, params *lib.DeSoParams) SnapshotValidatorEntry {
195+
pgValidatorEntry := SnapshotValidatorEntry{
196+
ExtraData: consumer.ExtraDataBytesToString(validatorEntry.ExtraData),
197+
BadgerKey: keyBytes,
198+
}
199+
200+
if validatorEntry.ValidatorPKID != nil {
201+
pgValidatorEntry.ValidatorPKID = consumer.PublicKeyBytesToBase58Check((*validatorEntry.ValidatorPKID)[:], params)
202+
}
203+
204+
if validatorEntry.Domains != nil {
205+
pgValidatorEntry.Domains = make([]string, len(validatorEntry.Domains))
206+
for ii, domain := range validatorEntry.Domains {
207+
pgValidatorEntry.Domains[ii] = string(domain)
208+
}
209+
}
210+
211+
pgValidatorEntry.DisableDelegatedStake = validatorEntry.DisableDelegatedStake
212+
pgValidatorEntry.DelegatedStakeCommissionBasisPoints = validatorEntry.DelegatedStakeCommissionBasisPoints
213+
214+
if validatorEntry.VotingPublicKey != nil {
215+
pgValidatorEntry.VotingPublicKey = validatorEntry.VotingPublicKey.ToString()
216+
}
217+
218+
if validatorEntry.VotingAuthorization != nil {
219+
pgValidatorEntry.VotingAuthorization = validatorEntry.VotingAuthorization.ToString()
220+
}
221+
222+
pgValidatorEntry.TotalStakeAmountNanos = bunbig.FromMathBig(validatorEntry.TotalStakeAmountNanos.ToBig())
223+
pgValidatorEntry.LastActiveAtEpochNumber = validatorEntry.LastActiveAtEpochNumber
224+
pgValidatorEntry.JailedAtEpochNumber = validatorEntry.JailedAtEpochNumber
225+
keyBytesWithoutPrefix := keyBytes[1:]
226+
pgValidatorEntry.SnapshotAtEpochNumber = lib.DecodeUint64(keyBytesWithoutPrefix[:8])
227+
return pgValidatorEntry
228+
}

0 commit comments

Comments
 (0)