headerfs: fail gracefully on db header writes

Author: mohamedawnallah

headerfs/store.go

@@ -538,7 +538,35 @@ func (h *blockHeaderStore) WriteHeaders(hdrs ...BlockHeader) error {
 		headerLocs[i] = header.toIndexEntry()
 	}
 
-	return h.addHeaders(headerLocs)
+	// Attempt to add the headers to the database. If this fails, we'll need
+	// to roll back any changes to the header file to maintain consistency.
+	// The rollback process bases on the number of header serialized. If
+	// both the initial operation and the rollback fail, we return
+	// a detailed error explaining both failures to aid in debugging.
+	if err := h.addHeaders(headerLocs); err != nil {
+		// Since the probability of failing to write to the database is
+		// very low, it is mostly worth the cost of file sync operation
+		// to make sure truncate headers does it correctly.
+		if err := h.file.Sync(); err != nil {
+			return fmt.Errorf("failed to sync block headers "+
+				"file: %v", err)
+		}
+
+		headersToRollback := len(hdrs)
+		truncateErr := h.truncateHeaders(
+			uint32(headersToRollback), h.indexType,
+		)
+		if truncateErr != nil {
+			return fmt.Errorf("failed to rollback block headers "+
+				"from binary file to previous valid state: "+
+				"%v, error writing to database: %v, headers "+
+				"to rollback: %d", truncateErr, err,
+				headersToRollback)
+		}
+		return fmt.Errorf("failed to add block headers to db: %v", err)
+	}
+
+	return nil
 }
 
 // blockLocatorFromHash takes a given block hash and then creates a block
@@ -1035,7 +1063,35 @@ func (f *filterHeaderStore) WriteHeaders(hdrs ...FilterHeader) error {
 	// As the block headers should already be written, we only need to
 	// update the tip pointer for this particular header type.
 	newTip := hdrs[len(hdrs)-1].toIndexEntry().hash
-	return f.truncateIndices(&newTip, []*chainhash.Hash{}, false)
+
+	// Attempt to add the headers to the database. If this fails, we'll need
+	// to roll back any changes to the header file to maintain consistency.
+	// The rollback process bases on the number of header serialized. If
+	// both the initial operation and the rollback fail, we return
+	// a detailed error explaining both failures to aid in debugging.
+	if err := f.truncateIndices(&newTip, nil, false); err != nil {
+		// Since the probability of failing to write to the database is
+		// very low, it is mostly worth the cost of file sync operation
+		// to make sure truncate headers does it correctly.
+		if err := f.file.Sync(); err != nil {
+			return fmt.Errorf("failed to sync filter headers "+
+				"file: %v", err)
+		}
+
+		headersToRollback := len(hdrs)
+		truncateErr := f.truncateHeaders(
+			uint32(headersToRollback), f.indexType,
+		)
+		if truncateErr != nil {
+			return fmt.Errorf("failed to rollback filter headers "+
+				"binary file to previous valid state: %v, "+
+				"error writing to database: %v", truncateErr,
+				err)
+		}
+		return fmt.Errorf("failed to add filter headers to db: %v", err)
+	}
+
+	return nil
 }
 
 // ChainTip returns the latest filter header and height known to the

headerfs/store_mock.go

@@ -1,9 +1,12 @@
 package headerfs
 
 import (
+	"io"
+
 	"github.com/btcsuite/btcd/blockchain"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/btcsuite/btcwallet/walletdb"
 	"github.com/stretchr/testify/mock"
 )
 
@@ -163,3 +166,54 @@ func (m *MockFilterHeaderStore) RollbackLastBlock(
 	}
 	return args.Get(0).(*BlockStamp), args.Error(1)
 }
+
+// MockWalletDB is a mock implementation of walletdb.DB for testing.
+type MockWalletDB struct {
+	mock.Mock
+}
+
+// Update implements the walletdb.DB interface.
+func (m *MockWalletDB) Update(fn func(tx walletdb.ReadWriteTx) error) error {
+	args := m.Called(fn)
+	return args.Error(0)
+}
+
+// View implements the walletdb.DB interface.
+func (m *MockWalletDB) View(fn func(tx walletdb.ReadTx) error) error {
+	args := m.Called(fn)
+	return args.Error(0)
+}
+
+// BeginReadWriteTx implements the walletdb.DB interface.
+func (m *MockWalletDB) BeginReadWriteTx() (walletdb.ReadWriteTx, error) {
+	args := m.Called()
+
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(walletdb.ReadWriteTx), args.Error(1)
+}
+
+// BeginReadTx implements the walletdb.DB interface.
+func (m *MockWalletDB) BeginReadTx() (walletdb.ReadTx, error) {
+	args := m.Called()
+
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(walletdb.ReadTx), args.Error(1)
+}
+
+// Copy implements the walletdb.DB interface.
+func (m *MockWalletDB) Copy(w io.Writer) error {
+	args := m.Called(w)
+	return args.Error(0)
+}
+
+// Close implements the walletdb.DB interface.
+func (m *MockWalletDB) Close() error {
+	args := m.Called()
+	return args.Error(0)
+}