Z/update state change entry encoder

lib/blockchain.go

@@ -2185,17 +2185,15 @@ func (bc *Blockchain) ProcessBlock(desoBlock *MsgDeSoBlock, verifySignatures boo
 				}
 				bc.timer.End("Blockchain.ProcessBlock: Transactions Db snapshot & operations")
 				if innerErr = bc.blockView.FlushToDbWithTxn(txn, blockHeight); innerErr != nil {
+					// If we're in the middle of a sync, we should notify the event manager that we failed to sync the block.
+					if bc.eventManager != nil && !bc.eventManager.isMempoolManager {
+						bc.eventManager.stateSyncerFlushed(&StateSyncerFlushedEvent{
+							FlushId:   uuid.Nil,
+							Succeeded: false,
+						})
+					}
 					return errors.Wrapf(innerErr, "ProcessBlock: Problem writing utxo view to db on simple add to tip")
 				}
-				// Immediately after the utxo view is flushed to badger, emit a state syncer flushed event, so that
-				// state syncer maintains a consistent view of the blockchain.
-				// Note: We ignore the mempool manager here, as that process handles state syncer flush events itself.
-				if bc.eventManager != nil && !bc.eventManager.isMempoolManager {
-					bc.eventManager.stateSyncerFlushed(&StateSyncerFlushedEvent{
-						FlushId:   uuid.Nil,
-						Succeeded: innerErr == nil,
-					})
-				}
 
 				bc.timer.End("Blockchain.ProcessBlock: Transactions Db utxo flush")
 				bc.timer.Start("Blockchain.ProcessBlock: Transactions Db snapshot & operations")
@@ -2548,6 +2546,16 @@ func (bc *Blockchain) ProcessBlock(desoBlock *MsgDeSoBlock, verifySignatures boo
 	// Signal the server that we've accepted this block in some way.
 	if bc.eventManager != nil {
 		bc.eventManager.blockAccepted(&BlockEvent{Block: desoBlock})
+		// Immediately after the utxo view is flushed to badger, emit a state syncer flushed event, so that
+		// state syncer maintains a consistent view of the blockchain.
+		// Note: We ignore the mempool manager here, as that process handles state syncer flush events itself.
+		if !bc.eventManager.isMempoolManager {
+			fmt.Printf("Emitting state syncer flushed event for synced block\n")
+			bc.eventManager.stateSyncerFlushed(&StateSyncerFlushedEvent{
+				FlushId:   uuid.Nil,
+				Succeeded: true,
+			})
+		}
 	}
 
 	bc.timer.Print("Blockchain.ProcessBlock: Initial")

lib/db_utils.go

@@ -460,7 +460,7 @@ type DBPrefixes struct {
 	// The Minor/Major distinction is used to deterministically map the two accessGroupIds of message's sender/recipient
 	// into a single pair based on the lexicographical ordering of the two accessGroupIds. This is done to ensure that
 	// both sides of the conversation have the same key for the same conversation, and we can store just a single message.
-	PrefixDmMessagesIndex []byte `prefix_id:"[75]" is_state:"true"`
+	PrefixDmMessagesIndex []byte `prefix_id:"[75]" is_state:"true" core_state:"true"`
 
 	// PrefixDmThreadIndex is modified by the NewMessage transaction and is used to store a DmThreadEntry
 	// for each existing dm thread. It answers the question: "Give me all the threads for a particular user."
@@ -989,6 +989,7 @@ func DBSetWithTxn(txn *badger.Txn, snap *Snapshot, key []byte, value []byte, eve
 				KeyBytes:             key,
 				EncoderBytes:         value,
 				AncestralRecordBytes: ancestralValue,
+				IsReverted:           false,
 			},
 			FlushId:      uuid.Nil,
 			IsMempoolTxn: eventManager.isMempoolManager,
@@ -1091,6 +1092,7 @@ func DBDeleteWithTxn(txn *badger.Txn, snap *Snapshot, key []byte, eventManager *
 				KeyBytes:             key,
 				EncoderBytes:         nil,
 				AncestralRecordBytes: ancestralValue,
+				IsReverted:           false,
 			},
 			FlushId:      uuid.Nil,
 			IsMempoolTxn: eventManager.isMempoolManager,

lib/event_manager.go

@@ -1,6 +1,8 @@
 package lib
 
-import "github.com/google/uuid"
+import (
+	"github.com/google/uuid"
+)
 
 type TransactionEventFunc func(event *TransactionEvent)
 type StateSyncerOperationEventFunc func(event *StateSyncerOperationEvent)

lib/server.go

@@ -381,6 +381,7 @@ func NewServer(
 
 	// Only initialize state change syncer if the directories are defined.
 	var stateChangeSyncer *StateChangeSyncer
+	fmt.Printf("State Change Dir: %v\n", _stateChangeDir)
 	if _stateChangeDir != "" {
 		// Create the state change syncer to handle syncing state changes to disk, and assign some of its methods
 		// to the event manager.
@@ -469,12 +470,14 @@ func NewServer(
 		srv.stateChangeSyncer.BlockHeight = uint64(_chain.headerTip().Height)
 	}
 
+	fmt.Printf("Before mempool\n")
 	// Create a mempool to store transactions until they're ready to be mined into
 	// blocks.
 	_mempool := NewDeSoMempool(_chain, _rateLimitFeerateNanosPerKB,
 		_minFeeRateNanosPerKB, _blockCypherAPIKey, _runReadOnlyUtxoViewUpdater, _dataDir,
 		_mempoolDumpDir, false)
 
+	fmt.Printf("After mempool\n")
 	// Initialize state syncer mempool job, if needed.
 	if srv.stateChangeSyncer != nil {
 		srv.stateChangeSyncer.StartMempoolSyncRoutine(srv)

lib/snapshot.go

@@ -1265,6 +1265,7 @@ func (snap *Snapshot) SetSnapshotChunk(mainDb *badger.DB, mainDbMutex *deadlock.
 						OperationType: DbOperationTypeInsert,
 						KeyBytes:      dbEntry.Key,
 						EncoderBytes:  dbEntry.Value,
+						IsReverted:    false,
 					},
 					FlushId: dbFlushId,
 				})