added force update of blockchain when a block is updated as temporary fix

This commit is contained in:
2023-07-20 22:09:59 +01:00
parent 9dcfcb6b55
commit 50085c6b11
2 changed files with 103 additions and 92 deletions

View File

@@ -7,138 +7,151 @@ Author: Peter Kleissner
package core
import (
"github.com/PeernetOfficial/core/blockchain"
"github.com/PeernetOfficial/core/protocol"
"github.com/enfipy/locker"
"github.com/PeernetOfficial/core/blockchain"
"github.com/PeernetOfficial/core/protocol"
"github.com/enfipy/locker"
)
// The blockchain cache stores blockchains.
type BlockchainCache struct {
BlockchainDirectory string // The directory for storing blockchains in a key-value store.
MaxBlockSize uint64 // Max block size to accept.
MaxBlockCount uint64 // Max block count to cache per peer.
LimitTotalRecords uint64 // Max count of blocks and header in total to keep across all blockchains. 0 = unlimited. Max Records * Max Block Size = Size Limit.
ReadOnly bool // Whether the cache is read only.
BlockchainDirectory string // The directory for storing blockchains in a key-value store.
MaxBlockSize uint64 // Max block size to accept.
MaxBlockCount uint64 // Max block count to cache per peer.
LimitTotalRecords uint64 // Max count of blocks and header in total to keep across all blockchains. 0 = unlimited. Max Records * Max Block Size = Size Limit.
ReadOnly bool // Whether the cache is read only.
Store *blockchain.MultiStore
peerLock *locker.Locker
Store *blockchain.MultiStore
peerLock *locker.Locker
backend *Backend
backend *Backend
}
func (backend *Backend) initBlockchainCache() {
if backend.Config.BlockchainGlobal == "" {
return
}
if backend.Config.BlockchainGlobal == "" {
return
}
backend.GlobalBlockchainCache = &BlockchainCache{
backend: backend,
BlockchainDirectory: backend.Config.BlockchainGlobal,
MaxBlockSize: backend.Config.CacheMaxBlockSize,
MaxBlockCount: backend.Config.CacheMaxBlockCount,
LimitTotalRecords: backend.Config.LimitTotalRecords,
}
backend.GlobalBlockchainCache = &BlockchainCache{
backend: backend,
BlockchainDirectory: backend.Config.BlockchainGlobal,
MaxBlockSize: backend.Config.CacheMaxBlockSize,
MaxBlockCount: backend.Config.CacheMaxBlockCount,
LimitTotalRecords: backend.Config.LimitTotalRecords,
}
var err error
backend.GlobalBlockchainCache.Store, err = blockchain.InitMultiStore(backend.Config.BlockchainGlobal)
if err != nil {
backend.LogError("initBlockchainCache", "initializing database '%s': %s", backend.Config.BlockchainGlobal, err.Error())
return
}
var err error
backend.GlobalBlockchainCache.Store, err = blockchain.InitMultiStore(backend.Config.BlockchainGlobal)
if err != nil {
backend.LogError("initBlockchainCache", "initializing database '%s': %s", backend.Config.BlockchainGlobal, err.Error())
return
}
backend.GlobalBlockchainCache.peerLock = locker.Initialize()
backend.GlobalBlockchainCache.peerLock = locker.Initialize()
// Set the blockchain cache to read-only if the record limit is reached.
if backend.Config.LimitTotalRecords > 0 && backend.GlobalBlockchainCache.Store.Database.Count() >= backend.Config.LimitTotalRecords {
backend.GlobalBlockchainCache.ReadOnly = true
}
// Set the blockchain cache to read-only if the record limit is reached.
if backend.Config.LimitTotalRecords > 0 && backend.GlobalBlockchainCache.Store.Database.Count() >= backend.Config.LimitTotalRecords {
backend.GlobalBlockchainCache.ReadOnly = true
}
backend.GlobalBlockchainCache.Store.FilterStatisticUpdate = backend.Filters.GlobalBlockchainCacheStatistic
backend.GlobalBlockchainCache.Store.FilterBlockchainDelete = backend.Filters.GlobalBlockchainCacheDelete
backend.GlobalBlockchainCache.Store.FilterStatisticUpdate = backend.Filters.GlobalBlockchainCacheStatistic
backend.GlobalBlockchainCache.Store.FilterBlockchainDelete = backend.Filters.GlobalBlockchainCacheDelete
}
// SeenBlockchainVersion shall be called with information about another peer's blockchain.
// If the reported version number is newer, all existing blocks are immediately deleted.
func (cache *BlockchainCache) SeenBlockchainVersion(peer *PeerInfo) {
cache.peerLock.Lock(string(peer.PublicKey.SerializeCompressed()))
defer cache.peerLock.Unlock(string(peer.PublicKey.SerializeCompressed()))
cache.peerLock.Lock(string(peer.PublicKey.SerializeCompressed()))
defer cache.peerLock.Unlock(string(peer.PublicKey.SerializeCompressed()))
// intermediate function to download and process blocks
downloadAndProcessBlocks := func(peer *PeerInfo, header *blockchain.MultiBlockchainHeader, offset, limit uint64) {
if limit > cache.MaxBlockCount {
limit = cache.MaxBlockCount
}
// intermediate function to download and process blocks
downloadAndProcessBlocks := func(peer *PeerInfo, header *blockchain.MultiBlockchainHeader, offset, limit uint64) {
if limit > cache.MaxBlockCount {
limit = cache.MaxBlockCount
}
peer.BlockDownload(peer.PublicKey, cache.MaxBlockCount, cache.MaxBlockSize, []protocol.BlockRange{{Offset: offset, Limit: limit}}, func(data []byte, targetBlock protocol.BlockRange, blockSize uint64, availability uint8) {
if availability != protocol.GetBlockStatusAvailable {
return
}
peer.BlockDownload(peer.PublicKey, cache.MaxBlockCount, cache.MaxBlockSize, []protocol.BlockRange{{Offset: offset, Limit: limit}}, func(data []byte, targetBlock protocol.BlockRange, blockSize uint64, availability uint8) {
if availability != protocol.GetBlockStatusAvailable {
return
}
if decoded, _ := cache.Store.IngestBlock(header, targetBlock.Offset, data, true); decoded != nil {
// index it for search
cache.backend.SearchIndex.IndexNewBlockDecoded(peer.PublicKey, peer.BlockchainVersion, targetBlock.Offset, decoded.RecordsDecoded)
}
})
}
if decoded, _ := cache.Store.IngestBlock(header, targetBlock.Offset, data, true); decoded != nil {
// index it for search
cache.backend.SearchIndex.IndexNewBlockDecoded(peer.PublicKey, peer.BlockchainVersion, targetBlock.Offset, decoded.RecordsDecoded)
}
})
}
// get the old header
header, status, err := cache.Store.AssessBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight)
if err != nil {
return
}
// get the old header
header, status, err := cache.Store.AssessBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight)
if err != nil {
return
}
switch status {
case blockchain.MultiStatusEqual:
return
switch status {
case blockchain.MultiStatusEqual:
return
case blockchain.MultiStatusInvalidRemote:
cache.Store.DeleteBlockchain(header)
case blockchain.MultiStatusInvalidRemote:
cache.Store.DeleteBlockchain(header)
cache.backend.SearchIndex.UnindexBlockchain(peer.PublicKey)
cache.backend.SearchIndex.UnindexBlockchain(peer.PublicKey)
case blockchain.MultiStatusHeaderNA:
if header, err = cache.Store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight); err != nil {
return
}
case blockchain.MultiStatusHeaderNA:
if header, err = cache.Store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight); err != nil {
return
}
downloadAndProcessBlocks(peer, header, 0, peer.BlockchainHeight)
downloadAndProcessBlocks(peer, header, 0, peer.BlockchainHeight)
case blockchain.MultiStatusNewVersion:
// delete existing data first, then create it new
cache.Store.DeleteBlockchain(header)
case blockchain.MultiStatusNewVersion:
// delete existing data first, then create it new
cache.Store.DeleteBlockchain(header)
cache.backend.SearchIndex.UnindexBlockchain(peer.PublicKey)
cache.backend.SearchIndex.UnindexBlockchain(peer.PublicKey)
if header, err = cache.Store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight); err != nil {
return
}
if header, err = cache.Store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight); err != nil {
return
}
downloadAndProcessBlocks(peer, header, 0, peer.BlockchainHeight)
downloadAndProcessBlocks(peer, header, 0, peer.BlockchainHeight)
case blockchain.MultiStatusNewBlocks:
offset := header.Height
limit := peer.BlockchainHeight - header.Height
case blockchain.MultiStatusNewBlocks:
//offset := header.Height
//limit := peer.BlockchainHeight - header.Height
//
//header.Height = peer.BlockchainHeight
//
//fmt.Println("new peer height")
//
//downloadAndProcessBlocks(peer, header, offset, limit)
// delete existing data first, then create it new
// Temporary Fix to force update the public key
cache.Store.DeleteBlockchain(header)
header.Height = peer.BlockchainHeight
cache.backend.SearchIndex.UnindexBlockchain(peer.PublicKey)
downloadAndProcessBlocks(peer, header, offset, limit)
if header, err = cache.Store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, peer.BlockchainHeight); err != nil {
return
}
}
downloadAndProcessBlocks(peer, header, 0, peer.BlockchainHeight)
if cache.LimitTotalRecords > 0 {
// Bug: This code is currently never reached if ReadOnly is true.
cache.ReadOnly = cache.Store.Database.Count() >= cache.LimitTotalRecords
}
}
if cache.LimitTotalRecords > 0 {
// Bug: This code is currently never reached if ReadOnly is true.
cache.ReadOnly = cache.Store.Database.Count() >= cache.LimitTotalRecords
}
}
// remoteBlockchainUpdate shall be called to indicate a potential update of the remotes blockchain.
// It will use the blockchain version and height to update the data lake as appropriate.
// This function is called in the Go routine of the packet worker and therefore must not stall.
func (peer *PeerInfo) remoteBlockchainUpdate() {
if peer.Backend.GlobalBlockchainCache == nil || peer.Backend.GlobalBlockchainCache.ReadOnly || peer.BlockchainVersion == 0 && peer.BlockchainHeight == 0 {
return
}
if peer.Backend.GlobalBlockchainCache == nil || peer.Backend.GlobalBlockchainCache.ReadOnly || peer.BlockchainVersion == 0 && peer.BlockchainHeight == 0 {
return
}
// TODO: This entire function should be instead a non-blocking message via a buffer channel.
go peer.Backend.GlobalBlockchainCache.SeenBlockchainVersion(peer)
// TODO: This entire function should be instead a non-blocking message via a buffer channel.
go peer.Backend.GlobalBlockchainCache.SeenBlockchainVersion(peer)
}

View File

@@ -137,8 +137,6 @@ func (api *WebapiInstance) apiProfileWrite(w http.ResponseWriter, r *http.Reques
newHeight, newVersion, status := api.Backend.UserBlockchain.ProfileWrite(fields)
api.Backend.LogError("apiProfileWrite", "Height: %v, Version %v", newHeight, newVersion)
EncodeJSON(api.Backend, w, r, apiBlockchainBlockStatus{Status: status, Height: newHeight, Version: newVersion})
}