daemon/processing/block.go
package processing
import (
"fmt"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/lbryio/chainquery/lbrycrd"
"github.com/lbryio/chainquery/metrics"
"github.com/lbryio/chainquery/model"
"github.com/lbryio/chainquery/sockety"
"github.com/lbryio/chainquery/twilio"
"github.com/lbryio/chainquery/util"
"github.com/lbryio/lbry.go/v2/extras/errors"
"github.com/lbryio/lbry.go/v2/extras/stop"
"github.com/lbryio/sockety/socketyapi"
"github.com/volatiletech/null/v8"
"github.com/sirupsen/logrus"
"github.com/volatiletech/sqlboiler/v4/boil"
"github.com/volatiletech/sqlboiler/v4/queries/qm"
)
// BlockLock is used to lock block processing to a single parent thread.
var BlockLock = sync.Mutex{}
// ManualShutDownError Error with special handling. Used to stop the concurrency pipeline for processing blocks midway.
var ManualShutDownError = errors.Err("Daemon stopped manually!")
// RunBlockProcessing runs the processing of a block at a specific height. While any height can be passed in it is
// important to note that if the previous block is not processed it will panic to prevent corruption because blocks
// must be processed in order.
func RunBlockProcessing(stopper *stop.Group, height uint64) uint64 {
defer metrics.Processing(time.Now(), "block")
defer util.TimeTrack(time.Now(), "runBlockProcessing", "daemonprofile")
if height == 0 {
err := processGenesisBlock()
if err != nil {
logrus.Fatal("Error processing Genesis block!", err)
}
return height
}
jsonBlock, err := getBlockToProcess(&height)
if err != nil {
logrus.Error("Get Block Error: ", err)
//ToDo - Should just return error...that is for another day
return height - 1
}
reorgHeight, err := checkHandleReorg(height, jsonBlock.PreviousBlockHash)
if err != nil {
logrus.Error("Reorg Handling Error: ", err)
//ToDo - Should just return error...that is for another day
return height - 1
}
if reorgHeight != height {
return reorgHeight
}
//This is an important lock to make sure we don't concurrently save transaction inputs/outputs accidentally via the
// mempool sync.
BlockLock.Lock()
defer BlockLock.Unlock()
block, err := ProcessBlock(height, stopper, jsonBlock)
if err != nil {
metrics.ProcessingFailures.WithLabelValues("block").Inc()
rollBackHeight := height - 1
blockRemovalError := block.DeleteG()
if blockRemovalError != nil {
logrus.Panicf("Could not delete block with bad data. Data corruption imminent at height %d. The block must be remove manually to continue.", height)
}
if err.Error() == ManualShutDownError.Error() {
return rollBackHeight
}
logrus.Error("Block Processing Error: ", errors.FullTrace(err))
logrus.Warning("Ran into transaction sync error at height", height, ". Rolling block back to height", height-1)
//ToDo - Should just return error...that is for another day
return rollBackHeight
}
return height
}
// ProcessBlock processing a specific block and returns an error. Use this to process a block having a custom handling
// of the error.
func ProcessBlock(height uint64, stopper *stop.Group, jsonBlock *lbrycrd.GetBlockResponse) (*model.Block, error) {
block := parseBlockInfo(height, jsonBlock)
err := setPreviousBlockInfo(height, jsonBlock.Hash)
if err != nil {
logrus.Errorf("failed to set previous block next hash: %s", err.Error())
}
txs := jsonBlock.Tx
go sockety.SendNotification(socketyapi.SendNotificationArgs{
Service: socketyapi.BlockChain,
Type: "new_block",
IDs: []string{"blocks", strconv.Itoa(int(height))},
Data: map[string]interface{}{"block": jsonBlock},
})
return block, syncTransactionsOfBlock(stopper, txs, block.BlockTime, block.Height)
}
//setPreviousBlockInfo sets the NextBlockHash field from the previous block
func setPreviousBlockInfo(currentHeight uint64, currentBLockHash string) error {
if currentHeight < 1 {
return nil
}
prevBlock, err := model.Blocks(qm.Where(model.BlockColumns.Height+"=?", currentHeight-1)).OneG()
if err != nil {
return errors.Err(err)
}
prevBlock.NextBlockHash.SetValid(currentBLockHash)
err = prevBlock.UpdateG(boil.Infer())
return errors.Err(err)
}
func parseBlockInfo(blockHeight uint64, jsonBlock *lbrycrd.GetBlockResponse) (block *model.Block) {
block = &model.Block{}
foundBlock, _ := model.Blocks(qm.Where(model.BlockColumns.Hash+"=?", jsonBlock.Hash)).OneG()
if foundBlock != nil {
block = foundBlock
}
block.Bits = jsonBlock.Bits
block.Chainwork = jsonBlock.ChainWork
block.Confirmations = uint(jsonBlock.Confirmations)
block.Difficulty = jsonBlock.Difficulty
block.Hash = jsonBlock.Hash
block.Height = blockHeight
block.MerkleRoot = jsonBlock.MerkleRoot
block.NameClaimRoot = jsonBlock.NameClaimRoot
block.Nonce = jsonBlock.Nonce
block.PreviousBlockHash.SetValid(jsonBlock.PreviousBlockHash)
block.NextBlockHash = null.NewString(jsonBlock.NextBlockHash, jsonBlock.NextBlockHash != "")
block.BlockSize = uint64(jsonBlock.Size)
block.BlockTime = uint64(jsonBlock.Time)
block.Version = uint64(jsonBlock.Version)
block.VersionHex = jsonBlock.VersionHex
block.TXCount = int(jsonBlock.NTx)
//block.TransactionHashes.SetValid(strings.Join(jsonBlock.Tx, ",")) //we don't need this, it's extremely redundant and heavy
var err error
if foundBlock != nil {
err = block.UpdateG(boil.Infer())
} else {
err = block.InsertG(boil.Infer())
}
if err != nil {
logrus.Panic(err)
}
return block
}
func processGenesisBlock() error {
genesisVerbose, genesis, err := lbrycrd.GetGenesisBlock()
if err != nil {
return errors.Err(err)
}
//This is an important lock to make sure we don't concurrently save transaction inputs/outputs accidentally via the
// mempool sync.
BlockLock.Lock()
defer BlockLock.Unlock()
block := parseBlockInfo(0, genesis)
for _, tx := range genesisVerbose.Tx {
tx.BlockHash = genesis.Hash
err := ProcessTx(&tx, block.BlockTime, 0)
if err != nil {
return errors.Err(err)
}
}
return nil
}
type txSyncManager struct {
daemonStopper *stop.Group
queueStopper *stop.Group
syncStopper *stop.Group
workerStopper *stop.Group
resultsCh chan txProcessResult
redoJobsCh chan txToProcess
jobsCh chan txToProcess
errorsCh chan error
}
func syncTransactionsOfBlock(stopper *stop.Group, txs []string, blockTime uint64, blockHeight uint64) error {
q("SYNC: started - " + strconv.Itoa(int(blockHeight)))
// Initialization
const maxErrorsPerBlockSync = 2
// Error handling logic can be tested by decreasing the number of times a transaction can fail to 0.
manager := txSyncManager{
daemonStopper: stop.New(stopper),
queueStopper: stop.New(nil),
syncStopper: stop.New(nil),
workerStopper: stop.New(nil),
errorsCh: make(chan error, maxErrorsPerBlockSync),
resultsCh: make(chan txProcessResult),
redoJobsCh: make(chan txToProcess, len(txs)),
jobsCh: make(chan txToProcess),
}
workers := util.Min(len(txs), MaxParallelTxProcessing)
initTxWorkers(manager.workerStopper, workers, manager.jobsCh, manager.resultsCh)
// Queue up n threads of transactions
manager.queueStopper.Add(1)
go queueTx(txs, blockTime, blockHeight, &manager)
q("SYNC: launched queueing")
//Setup reprocessing queue
manager.syncStopper.Add(1)
go reprocessQueue(&manager)
// Handle the results
manager.syncStopper.Add(1)
go handleTxResults(len(txs), &manager)
q("SYNC: launched handling")
// Check for queueing errors ( ie. lbrycrd fetch)
err := <-manager.errorsCh
if err != nil {
return errors.Err(err)
}
q("SYNC: received 1st on errorCh")
// Wait for first handling error/nil first error is sent when tx fails x times or daemon is shutdown
err = <-manager.errorsCh
q("SYNC: received 2nd on errorCh")
q("SYNC: stop workers...")
manager.workerStopper.Stop()
q("SYNC: stop sync...")
manager.syncStopper.Stop()
q("SYNC: wait for workers... - " + strconv.Itoa(int(blockHeight)))
manager.workerStopper.StopAndWait()
q("SYNC: wait for sync - " + strconv.Itoa(int(blockHeight)))
manager.syncStopper.StopAndWait()
q("SYNC: stopped - " + strconv.Itoa(int(blockHeight)))
q("SYNC: closing redo channel")
close(manager.redoJobsCh)
q("SYNC: closing result channel")
close(manager.resultsCh)
q("SYNC: closing error channel")
close(manager.errorsCh)
q("SYNC: closing jobs channel")
close(manager.jobsCh)
q("SYNC: finished - " + strconv.Itoa(int(blockHeight)))
return err
}
//q enables extensive logging on the concurrency of Chainquery. If there is every a deadlock and it's reproducible
// you can use this to debug it. Don't get stuck on the 'q' name either. It was literally a rare single letter that
// I chose, thats' it.
func q(a string) {
if false {
println(a)
}
}
// MaxFailures tells Chainquery how many failures a transaction can have before we rollback the block and try to process it
// it again. This is to stop an indefinite loop. Since transactions can be dependant on one another they can fail if not
// processed in the right order. We allow parallel processing by putting transactions into a queue, and if they fail to
// process, for example if its dependant transaction has not been processed yet, then we allow to go back into the queue
// x times ( MaxFailures ).
var MaxFailures int
func handleTxResults(nrToHandle int, manager *txSyncManager) {
defer manager.syncStopper.Done()
q("HANDLE: start handling")
leftToProcess := nrToHandle
for {
q("HANDLE: waiting for next result...")
select {
case <-manager.daemonStopper.Ch():
logrus.Info("stopping tx sync...")
handleFailure(errors.Err("Daemon stopped manually!"), manager)
return
case <-manager.syncStopper.Ch():
q("HANDLE: stopping handling...")
return
case txResult := <-manager.resultsCh:
q("HANDLE: start handling new result.." + txResult.tx.Txid)
leftToProcess--
if txResult.failcount > MaxFailures {
err := errors.Prefix("transaction "+txResult.tx.Txid+" failed more than "+strconv.Itoa(MaxFailures)+" times", txResult.err)
handleFailure(err, manager)
continue
}
if txResult.err != nil { // Try again if fails this time.
leftToProcess++
q("HANDLE: start sending to worker..." + txResult.tx.Txid)
manager.redoJobsCh <- txToProcess{tx: txResult.tx, blockTime: txResult.blockTime, failcount: txResult.failcount, blockHeight: txResult.blockHeight}
q("HANDLE: end sending to worker..." + txResult.tx.Txid)
q("HANDLE: finish handling new result.." + txResult.tx.Txid)
//continue
}
if leftToProcess == 0 {
q("HANDLE: start passing done..")
manager.errorsCh <- nil
q("HANDLE: end passing done..")
q("HANDLE: end handling..")
return
}
q("HANDLE: go to next loop..")
continue
}
}
}
func queueTx(txs []string, blockTime uint64, blockHeight uint64, manager *txSyncManager) {
defer manager.queueStopper.Done()
q("QUEUE: start of queuing")
txRawMap := make(map[string]*lbrycrd.TxRawResult)
depthMap := make(map[string]int, len(txs))
txSlice := make([]*lbrycrd.TxRawResult, len(txs))
for i := range txs {
select {
case <-manager.queueStopper.Ch():
q("QUEUE: stopping lbrycrd getting...")
manager.errorsCh <- nil
return
default:
q("QUEUE: start getting lbrycrd transaction..." + txs[i])
jsonTx, err := lbrycrd.GetRawTransactionResponse(txs[i])
if err != nil {
manager.errorsCh <- errors.Prefix("GetRawTxError"+txs[i], err)
return
}
txRawMap[jsonTx.Txid] = jsonTx
txSlice[i] = jsonTx
depthMap[jsonTx.Txid] = 1
q("QUEUE: end getting lbrycrd transaction..." + txs[i])
}
}
txSet, ok := optimizeOrderToProcess(txRawMap, depthMap)
q("QUEUE: start interaction of " + strconv.Itoa(len(txSet)) + " transactions")
if !ok {
txSet = txSlice
}
for _, jsonTx := range txSet {
select {
case <-manager.queueStopper.Ch():
q("QUEUE: stopping processing " + jsonTx.Txid)
manager.errorsCh <- nil
return
default:
q("QUEUE: start processing..." + jsonTx.Txid)
manager.jobsCh <- txToProcess{tx: jsonTx, blockTime: blockTime, blockHeight: blockHeight}
q("QUEUE: end processing..." + jsonTx.Txid)
}
}
q("QUEUE: end of queuing")
manager.errorsCh <- nil
q("QUEUE: end of queuing...passed nil to errorCh")
}
//flush is a helper function for handling the results.
func flush(channel <-chan txProcessResult) {
for range channel {
}
}
func handleFailure(err error, manager *txSyncManager) {
q("HANDLE: start passing error...")
manager.queueStopper.Stop()
q("HANDLE: flushing channel...")
//Clears queue if any additional finished jobs come in at this point.
// It also stops blocking during stopping.
go flush(manager.resultsCh)
q("HANDLE: stopping queue...")
manager.queueStopper.StopAndWait()
q("HANDLE: stopped queue...")
q("HANDLE: stopping workers...")
manager.workerStopper.Stop()
q("HANDLE: stopped workers...")
manager.errorsCh <- err
q("HANDLE: finish passing error...")
}
func getBlockToProcess(height *uint64) (*lbrycrd.GetBlockResponse, error) {
hash, err := lbrycrd.GetBlockHash(*height)
if err != nil {
return nil, errors.Prefix(fmt.Sprintf("GetBlockHash Error(%d)", *height), err)
}
jsonBlock, err := lbrycrd.GetBlock(*hash)
if err != nil {
return nil, errors.Prefix("GetBlock Error("+*hash+")", err)
}
return jsonBlock, nil
}
func checkHandleReorg(height uint64, chainPrevHash string) (uint64, error) {
prevHeight := height - 1
depth := 0
if height > 0 {
prevBlock, err := model.Blocks(qm.Where(model.BlockColumns.Height+"=?", prevHeight), qm.Load("BlockHashTransactions")).OneG()
if err != nil {
return height, errors.Prefix("error getting block@"+strconv.Itoa(int(prevHeight)), err)
}
//Recursively delete blocks until they match or a reorg of depth 100 == failure of logic.
for prevBlock.Hash != chainPrevHash && depth < 100 && prevHeight > 0 {
hashes := make([]string, len(prevBlock.R.BlockHashTransactions))
for i, th := range prevBlock.R.BlockHashTransactions {
hashes[i] = th.Hash
}
fmt.Printf("block %s at height %d to be removed due to reorg. TX-> %s", prevBlock.Hash, prevBlock.Height, strings.Join(hashes, ","))
logrus.Printf("block %s at height %d to be removed due to reorg. TX-> %s", prevBlock.Hash, prevBlock.Height, strings.Join(hashes, ","))
// Delete because it needs to be reprocessed due to reorg
err = prevBlock.DeleteG()
if err != nil {
return height, errors.Prefix("error deleting block@"+strconv.Itoa(int(prevHeight)), err)
}
depth++
// Set chainPrevHash to new previous blocks prevhash to check next depth
jsonBlock, err := getBlockToProcess(&prevHeight)
if err != nil {
return height, errors.Prefix("error getting block@"+strconv.Itoa(int(prevHeight))+" from lbrycrd", err)
}
chainPrevHash = jsonBlock.PreviousBlockHash
// Decrement height and set prevBlock to the new previous
prevHeight--
prevBlock, err = model.Blocks(qm.Where(model.BlockColumns.Height+"=?", prevHeight)).OneG()
if err != nil {
return height, errors.Prefix("error getting previous block@"+strconv.Itoa(int(prevHeight)), err)
}
}
if depth > 0 {
message := fmt.Sprintf("Reorg detected of depth %d at height %d,(last matching height %d) handling reorg processing!", depth, height, prevHeight)
logrus.Warning(message)
if depth > 2 {
twilio.SendSMS(message)
}
return prevHeight, nil
}
}
return height, nil
}
func reprocessQueue(manager *txSyncManager) {
defer manager.syncStopper.Done()
for {
select {
case <-manager.syncStopper.Ch():
q("REDO: stopping redo jobs")
return
case redoJob := <-manager.redoJobsCh:
q("REDO: start send new redo job - " + redoJob.tx.Txid)
manager.jobsCh <- redoJob
q("REDO: end send new redo job - " + redoJob.tx.Txid)
}
}
}
func checkDepth(tx *lbrycrd.TxRawResult, txMap map[string]*lbrycrd.TxRawResult, depthMap map[string]int, start time.Time) bool {
if time.Since(start) > 5*time.Second {
return false
}
for _, vin := range tx.Vin {
if txchild, ok := txMap[vin.TxID]; ok {
depthMap[vin.TxID]++
return checkDepth(txchild, txMap, depthMap, start)
}
}
return true
}
func optimizeOrderToProcess(txMap map[string]*lbrycrd.TxRawResult, depthMap map[string]int) ([]*lbrycrd.TxRawResult, bool) {
start := time.Now()
successful := false
for _, tx := range txMap {
successful = checkDepth(tx, txMap, depthMap, start)
}
type depthPair struct {
TxID string
Count int
}
var list []depthPair
for k, v := range depthMap {
list = append(list, depthPair{k, v})
}
sort.Slice(list, func(i, j int) bool {
return list[i].Count > list[j].Count
})
orderedTx := make([]*lbrycrd.TxRawResult, len(list))
for i := range list {
orderedTx[i] = txMap[list[i].TxID]
//Additional debugging to output the order in which transactions are processed.
q("Tx " + list[i].TxID + " Count " + strconv.Itoa(list[i].Count))
}
return orderedTx, successful
}