Merge pull request #320 from mohamedawnallah/failGracefullyOnDBHeaderWrites

chainimport[1/3]: import block and filter headers on startup before falling back to P2P synchronization

Author: Roasbeef

blockmanager.go

@@ -84,7 +84,7 @@ type blockManagerCfg struct {
 
 	// RegFilterHeaders is the store where filter headers for the regular
 	// compact filters are persistently stored.
-	RegFilterHeaders *headerfs.FilterHeaderStore
+	RegFilterHeaders headerfs.FilterHeaderStore
 
 	// TimeSource is used to access a time estimate based on the clocks of
 	// the connected peers.
@@ -702,7 +702,7 @@ waitForHeaders:
 // network, if it can, and resolves any conflicts between them. It then writes
 // any verified headers to the store.
 func (b *blockManager) getUncheckpointedCFHeaders(
-	store *headerfs.FilterHeaderStore, fType wire.FilterType) error {
+	store headerfs.FilterHeaderStore, fType wire.FilterType) error {
 
 	// Get the filter header store's chain tip.
 	filterTip, filtHeight, err := store.ChainTip()
@@ -928,7 +928,7 @@ func (c *checkpointedCFHeadersQuery) handleResponse(req, resp wire.Message,
 // checkpoints we got from the network. It assumes that the filter header store
 // matches the checkpoints up to the tip of the store.
 func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
-	store *headerfs.FilterHeaderStore, fType wire.FilterType) {
+	store headerfs.FilterHeaderStore, fType wire.FilterType) {
 
 	// We keep going until we've caught up the filter header store with the
 	// latest known checkpoint.
@@ -1174,7 +1174,7 @@ func (b *blockManager) getCheckpointedCFHeaders(checkpoints []*chainhash.Hash,
 // filter header field in the next message range before writing to disk, and
 // the current height after writing the headers.
 func (b *blockManager) writeCFHeadersMsg(msg *wire.MsgCFHeaders,
-	store *headerfs.FilterHeaderStore) (*chainhash.Hash, uint32, error) {
+	store headerfs.FilterHeaderStore) (*chainhash.Hash, uint32, error) {
 
 	// Check that the PrevFilterHeader is the same as the last stored so we
 	// can prevent misalignment.
@@ -1364,7 +1364,7 @@ func verifyCheckpoint(prevCheckpoint, nextCheckpoint *chainhash.Hash,
 // information.
 func (b *blockManager) resolveConflict(
 	checkpoints map[string][]*chainhash.Hash,
-	store *headerfs.FilterHeaderStore, fType wire.FilterType) (
+	store headerfs.FilterHeaderStore, fType wire.FilterType) (
 	[]*chainhash.Hash, error) {
 
 	// First check the served checkpoints against the hardcoded ones.
@@ -1913,7 +1913,7 @@ func (b *blockManager) getCheckpts(lastHash *chainhash.Hash,
 // existing store up to the tip of the store. If all of the peers match but
 // the store doesn't, the height at which the mismatch occurs is returned.
 func checkCFCheckptSanity(cp map[string][]*chainhash.Hash,
-	headerStore *headerfs.FilterHeaderStore) (int, error) {
+	headerStore headerfs.FilterHeaderStore) (int, error) {
 
 	// Get the known best header to compare against checkpoints.
 	_, storeTip, err := headerStore.ChainTip()

blockmanager_test.go

@@ -55,7 +55,7 @@ func (m *mockDispatcher) Query(requests []*query.Request,
 
 // setupBlockManager initialises a blockManager to be used in tests.
 func setupBlockManager(t *testing.T) (*blockManager, headerfs.BlockHeaderStore,
-	*headerfs.FilterHeaderStore, error) {
+	headerfs.FilterHeaderStore, error) {
 
 	// Set up the block and filter header stores.
 	tempDir := t.TempDir()
@@ -798,8 +798,11 @@ func TestBlockManagerDetectBadPeers(t *testing.T) {
 	for _, test := range testCases {
 		// Create a mock block header store. We only need to be able to
 		// serve a header for the target index.
-		blockHeaders := newMockBlockHeaderStore()
-		blockHeaders.heights[targetIndex] = blockHeader
+		mBlockHeaderStore := &headerfs.MockBlockHeaderStore{}
+
+		mBlockHeaderStore.On("FetchHeaderByHeight", targetIndex).Return(
+			&blockHeader, nil,
+		)
 
 		// We set up the mock queryAllPeers to only respond according to
 		// the active testcase.
@@ -851,7 +854,7 @@ func TestBlockManagerDetectBadPeers(t *testing.T) {
 
 		bm := &blockManager{
 			cfg: &blockManagerCfg{
-				BlockHeaders:  blockHeaders,
+				BlockHeaders:  mBlockHeaderStore,
 				queryAllPeers: queryAllPeers,
 			},
 		}

headerfs/file.go

@@ -50,19 +50,11 @@ func (h *headerStore) appendRaw(header []byte) error {
 // amount of bytes read past the seek distance is determined by the specified
 // header type.
 func (h *headerStore) readRaw(seekDist uint64) ([]byte, error) {
-	var headerSize uint32
-
-	// Based on the defined header type, we'll determine the number of
-	// bytes that we need to read past the sync point.
-	switch h.indexType {
-	case Block:
-		headerSize = 80
-
-	case RegularFilter:
-		headerSize = 32
-
-	default:
-		return nil, fmt.Errorf("unknown index type: %v", h.indexType)
+	// Based on the defined header type, we'll determine the number of bytes
+	// that we need to read past the sync point.
+	headerSize, err := h.indexType.Size()
+	if err != nil {
+		return nil, err
 	}
 
 	// TODO(roasbeef): add buffer pool
@@ -140,7 +132,7 @@ func (h *blockHeaderStore) readHeader(height uint32) (wire.BlockHeader, error) {
 
 // readHeader reads a single filter header at the specified height from the
 // flat files on disk.
-func (f *FilterHeaderStore) readHeader(height uint32) (*chainhash.Hash, error) {
+func (f *filterHeaderStore) readHeader(height uint32) (*chainhash.Hash, error) {
 	seekDistance := uint64(height) * 32
 
 	rawHeader, err := f.readRaw(seekDistance)
@@ -158,7 +150,7 @@ func (f *FilterHeaderStore) readHeader(height uint32) (*chainhash.Hash, error) {
 //
 // NOTE: The end height is _inclusive_ so we'll fetch all headers from the
 // startHeight up to the end height, including the final header.
-func (f *FilterHeaderStore) readHeaderRange(startHeight uint32,
+func (f *filterHeaderStore) readHeaderRange(startHeight uint32,
 	endHeight uint32) ([]chainhash.Hash, error) {
 
 	// Based on the defined header type, we'll determine the number of

headerfs/file_test.go

@@ -158,9 +158,11 @@ func TestAppendRow(t *testing.T) {
 				truncFn: test.truncFn,
 			}
 
+			hdrFile := &headerFile{file: mockFile}
+
 			// Create a header store with our mock file.
 			h := &headerStore{
-				file:        mockFile,
+				headerFile:  hdrFile,
 				headerIndex: &headerIndex{indexType: Block},
 			}
 
@@ -207,8 +209,10 @@ func BenchmarkHeaderStoreAppendRaw(b *testing.B) {
 	tmpFile, cleanup := createFile(b, "header_benchmark")
 	defer cleanup()
 
+	hdrFile := &headerFile{file: tmpFile}
+
 	store := &headerStore{
-		file:        tmpFile,
+		headerFile:  hdrFile,
 		headerIndex: &headerIndex{indexType: Block},
 	}