/* File Name: Multi.go Copyright: 2021 Peernet s.r.o. Author: Peter Kleissner Multi-blockchain store implementation. Keys used in the key-value store: 1. Key: Public key compressed, Value: Header 2. Key: Public key compressed + version + block number, Value: Block */ package blockchain import ( "encoding/binary" "errors" "time" "github.com/PeernetOfficial/core/btcec" "github.com/PeernetOfficial/core/store" ) const ( MultiStatusOK = 0 // Success. MultiStatusErrorReadHeader = 1 // Error reading header for blockchain. MultiStatusHeaderNA = 2 // Header not available. This indicates nothing is stored for the blockchain. MultiStatusInvalidRemote = 3 // Invalid remote reported blockchain version and height. (local cache is "newer" which should not happen) MultiStatusEqual = 4 // Remote blockchain and local cache are equal. MultiStatusNewVersion = 5 // Local cache is out of date. A new version of the blockchain is available. MultiStatusNewBlocks = 6 // Local cache is out of date. Additional blocks are available. ) // MultiStore stores multiple blockchains. type MultiStore struct { path string // Path of the blockchain on disk. Depends on key-value store whether a filename or folder. Database store.Store // The database storing the blockchain. // callbacks FilterStatisticUpdate func(multi *MultiStore, header *MultiBlockchainHeader, statsOld BlockchainStats) FilterBlockchainDelete func(multi *MultiStore, header *MultiBlockchainHeader) } func InitMultiStore(path string) (multi *MultiStore, err error) { multi = &MultiStore{path: path} // open existing blockchain file or create new one if multi.Database, err = store.NewPogrebStore(path); err != nil { return nil, err } return multi, nil } /* Header for blockchains: Offset Size Info 0 8 Version of the blockchain 8 8 Height of the blockchain 16 8 Count of blocks 24 8 Date first block added 32 8 Date last block added 40 8 Stats: Count of file records (from available blocks) 48 8 Stats: Size of all files combined (from available blocks) 56 8 * n List of block numbers that are stored Note: The statistics fields only count available stored blocks. */ const multiBlockchainHeaderSize = 56 // This is a header for a single blockchain stored in a multi store. type MultiBlockchainHeader struct { PublicKey *btcec.PublicKey // Public Key of the blockchain Height uint64 // Height is exchanged as uint32 in the protocol, but stored as uint64. Version uint64 // Version is always uint64. DateFirstBlockAdded time.Time // Date the first block was added DateLastBlockAdded time.Time // Date the last block was added ListBlocks []uint64 // List of block numbers that are stored Stats BlockchainStats // Statistics about the blockchain (only about stored blocks) } // Keep statistics about blocks stored for a blockchain type BlockchainStats struct { CountFileRecords uint64 // Count of file records in stored blocks SizeAllFiles uint64 // Size of all files combined in stored blocks } func decodeBlockchainHeader(publicKey *btcec.PublicKey, buffer []byte) (header *MultiBlockchainHeader, err error) { if len(buffer) < multiBlockchainHeaderSize { return nil, errors.New("header length too small") } header = &MultiBlockchainHeader{PublicKey: publicKey} header.Version = binary.LittleEndian.Uint64(buffer[0:8]) header.Height = binary.LittleEndian.Uint64(buffer[8:16]) countBlocks := binary.LittleEndian.Uint64(buffer[16:24]) header.DateFirstBlockAdded = time.Unix(int64(binary.LittleEndian.Uint64(buffer[24:32])), 0) header.DateLastBlockAdded = time.Unix(int64(binary.LittleEndian.Uint64(buffer[32:40])), 0) header.Stats.CountFileRecords = binary.LittleEndian.Uint64(buffer[40:48]) header.Stats.SizeAllFiles = binary.LittleEndian.Uint64(buffer[48:56]) if uint64(len(buffer)) < multiBlockchainHeaderSize+8*countBlocks { return nil, errors.New("header length too small") } index := multiBlockchainHeaderSize for n := uint64(0); n < countBlocks; n++ { blockN := binary.LittleEndian.Uint64(buffer[index : index+8]) header.ListBlocks = append(header.ListBlocks, blockN) index += 8 } return header, nil } // Reads a blockchains header if available. func (multi *MultiStore) ReadBlockchainHeader(publicKey *btcec.PublicKey) (header *MultiBlockchainHeader, found bool, err error) { buffer, found := multi.Database.Get(publicKey.SerializeCompressed()) if !found { return nil, false, nil } header, err = decodeBlockchainHeader(publicKey, buffer) return header, err == nil, err } // WriteBlockchainHeader writes a blockchain header. If one exists, it will be overwritten. func (multi *MultiStore) WriteBlockchainHeader(header *MultiBlockchainHeader) (err error) { raw := make([]byte, multiBlockchainHeaderSize+8*len(header.ListBlocks)) binary.LittleEndian.PutUint64(raw[0:8], header.Version) binary.LittleEndian.PutUint64(raw[8:16], header.Height) binary.LittleEndian.PutUint64(raw[16:24], uint64(len(header.ListBlocks))) binary.LittleEndian.PutUint64(raw[24:32], uint64(header.DateFirstBlockAdded.UTC().Unix())) binary.LittleEndian.PutUint64(raw[32:40], uint64(header.DateLastBlockAdded.UTC().Unix())) binary.LittleEndian.PutUint64(raw[40:48], header.Stats.CountFileRecords) binary.LittleEndian.PutUint64(raw[48:56], header.Stats.SizeAllFiles) index := multiBlockchainHeaderSize for _, blockN := range header.ListBlocks { binary.LittleEndian.PutUint64(raw[index:index+8], blockN) index += 8 } return multi.Database.Set(header.PublicKey.SerializeCompressed(), raw) } func lookupKeyForBlock(publicKey *btcec.PublicKey, version, blockNumber uint64) (key []byte) { var buffer [16]byte binary.LittleEndian.PutUint64(buffer[0:8], version) binary.LittleEndian.PutUint64(buffer[8:16], blockNumber) key = append(key, publicKey.SerializeCompressed()...) key = append(key, buffer[:]...) return key } // ReadBlock reads a raw block func (multi *MultiStore) ReadBlock(publicKey *btcec.PublicKey, version, blockNumber uint64) (raw []byte, found bool) { return multi.Database.Get(lookupKeyForBlock(publicKey, version, blockNumber)) } // WriteBlock writes a raw block. It does not update the blockchain header. func (multi *MultiStore) WriteBlock(publicKey *btcec.PublicKey, version, blockNumber uint64, raw []byte) (err error) { return multi.Database.Set(lookupKeyForBlock(publicKey, version, blockNumber), raw) } // AssessBlockchainHeader reads the blockchain header, if available, and assesses the status. func (multi *MultiStore) AssessBlockchainHeader(publicKey *btcec.PublicKey, version, height uint64) (header *MultiBlockchainHeader, status int, err error) { // check if there is an existing header for the blockchain header, found, err := multi.ReadBlockchainHeader(publicKey) if err != nil { return nil, MultiStatusErrorReadHeader, err } else if !found { return nil, MultiStatusHeaderNA, nil } if header.Version == version && header.Height == height { return header, MultiStatusEqual, nil } // Check if existing version is newer than reported - indicates illegal behavior by the remote peer. // Improper refactoring could happen if the local blockchain folder is deleted and automatically recreated. if header.Version > version || (header.Version == version && header.Height > height) { return header, MultiStatusInvalidRemote, nil } if version > header.Version { return header, MultiStatusNewVersion, nil } return header, MultiStatusNewBlocks, nil } // Deletes an entire blockchain from the store. It will delete each block individually and then the header. func (multi *MultiStore) DeleteBlockchain(header *MultiBlockchainHeader) { // first delete all blocks for _, blockN := range header.ListBlocks { multi.Database.Delete(lookupKeyForBlock(header.PublicKey, header.Version, blockN)) } // delete the header multi.Database.Delete(header.PublicKey.SerializeCompressed()) if multi.FilterBlockchainDelete != nil { multi.FilterBlockchainDelete(multi, header) } } func (multi *MultiStore) NewBlockchainHeader(publicKey *btcec.PublicKey, version, height uint64) (header *MultiBlockchainHeader, err error) { timeN := time.Now().UTC() header = &MultiBlockchainHeader{ PublicKey: publicKey, Height: height, Version: version, DateFirstBlockAdded: timeN, DateLastBlockAdded: timeN, } return header, multi.WriteBlockchainHeader(header) } // Updates the statistics fields of the blockchain header based on the new decoded block records. // It does not write the blockchain header; multi.WriteBlockchainHeader must called to store the changes. // The caller must make sure not to call this function on records already processed. func (multi *MultiStore) UpdateBlockchainStatistics(header *MultiBlockchainHeader, recordsDecoded []interface{}) { updatedStats := false statsOld := header.Stats for _, decodedR := range recordsDecoded { if file, ok := decodedR.(BlockRecordFile); ok { header.Stats.SizeAllFiles += file.Size header.Stats.CountFileRecords++ updatedStats = true } } if updatedStats && multi.FilterStatisticUpdate != nil { multi.FilterStatisticUpdate(multi, header, statsOld) } } // Iterates over all blockchains stored in the cache func (multi *MultiStore) IterateBlockchains(callback func(header *MultiBlockchainHeader)) { multi.Database.Iterate(func(key, value []byte) { if len(key) == 33 { // Length 33 indicates key = Public key compressed, value = Header if blockchainPublicKey, err := btcec.ParsePubKey(key, btcec.S256()); err == nil { if header, err := decodeBlockchainHeader(blockchainPublicKey, value); err == nil { callback(header) } } } }) } // IngestBlock ingests a new block into the store. It fails if a block is already stored for the given blockchain and block number. // It will update the blockchain header including the statistics. func (multi *MultiStore) IngestBlock(header *MultiBlockchainHeader, blockNumber uint64, raw []byte, failIfInvalid bool) (decoded *BlockDecoded, err error) { // check if already exists if _, found := multi.ReadBlock(header.PublicKey, header.Version, blockNumber); found { return nil, errors.New("already exists") } // decode it decoded, status, err := DecodeBlockRaw(raw) if failIfInvalid && err != nil { return nil, err } // store the transferred block in the cache multi.WriteBlock(header.PublicKey, header.Version, blockNumber, raw) header.ListBlocks = append(header.ListBlocks, blockNumber) // update blockchain header stats if records were decoded if status == StatusOK { multi.UpdateBlockchainStatistics(header, decoded.RecordsDecoded) } // update the blockchain header multi.WriteBlockchainHeader(header) return decoded, nil }