Fix FK constraint error when running host on sqlite (#2141)

* Check for blocks before adding to avoid FK/ UK constraint

go/host/storage/hostdb/batch.go

@@ -30,7 +30,7 @@ func AddBatch(dbtx *dbTransaction, statements *SQLStatements, batch *common.ExtB
 		return fmt.Errorf("could not encode L2 transactions: %w", err)
 	}
 
-	_, err = dbtx.tx.Exec(statements.InsertBatch,
+	_, err = dbtx.Tx.Exec(statements.InsertBatch,
 		batch.SeqNo().Uint64(),       // sequence
 		batch.Hash(),                 // full hash
 		batch.Header.Number.Uint64(), // height
@@ -51,20 +51,20 @@ func AddBatch(dbtx *dbTransaction, statements *SQLStatements, batch *common.ExtB
 			args = append(args, txHash.Bytes(), batch.SeqNo().Uint64())
 		}
 		insert = strings.TrimRight(insert, ",")
-		_, err = dbtx.tx.Exec(insert, args...)
+		_, err = dbtx.Tx.Exec(insert, args...)
 		if err != nil {
 			return fmt.Errorf("failed to insert transactions. cause: %w", err)
 		}
 	}
 
 	var currentTotal int
-	err = dbtx.tx.QueryRow(selectTxCount).Scan(&currentTotal)
+	err = dbtx.Tx.QueryRow(selectTxCount).Scan(&currentTotal)
 	if err != nil {
 		return fmt.Errorf("failed to query transaction count: %w", err)
 	}
 
 	newTotal := currentTotal + len(batch.TxHashes)
-	_, err = dbtx.tx.Exec(statements.UpdateTxCount, newTotal)
+	_, err = dbtx.Tx.Exec(statements.UpdateTxCount, newTotal)
 	if err != nil {
 		return fmt.Errorf("failed to update transaction count: %w", err)
 	}

go/host/storage/hostdb/block.go

@@ -1,25 +1,29 @@
 package hostdb
 
 import (
+	"database/sql"
 	"fmt"
 
+	gethcommon "github.com/ethereum/go-ethereum/common"
+
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ten-protocol/go-ten/go/common"
 )
 
 const (
 	selectBlocks = "SELECT b.id, b.hash, b.header, r.hash FROM block_host b join rollup_host r on r.compression_block=b.id ORDER BY b.id DESC "
+	selectBlock  = "SELECT id FROM block_host WHERE hash = "
 )
 
 // AddBlock stores a block header with the given rollupHash it contains in the host DB
-func AddBlock(dbtx *dbTransaction, statements *SQLStatements, b *types.Header) error {
+func AddBlock(dbtx *sql.Tx, statements *SQLStatements, b *types.Header) error {
 	header, err := rlp.EncodeToBytes(b)
 	if err != nil {
 		return fmt.Errorf("could not encode block header. Cause: %w", err)
 	}
 
-	_, err = dbtx.tx.Exec(statements.InsertBlock,
+	_, err = dbtx.Exec(statements.InsertBlock,
 		b.Hash().Bytes(), // hash
 		header,           // l1 block header
 	)
@@ -30,6 +34,18 @@ func AddBlock(dbtx *dbTransaction, statements *SQLStatements, b *types.Header) e
 	return nil
 }
 
+// GetBlockId returns the block ID given the hash.
+func GetBlockId(db *sql.Tx, statements *SQLStatements, hash gethcommon.Hash) (*int64, error) {
+	query := selectBlock + statements.Placeholder
+	var blockId int64
+	err := db.QueryRow(query, hash.Bytes()).Scan(&blockId)
+	if err != nil {
+		return nil, fmt.Errorf("query execution for select block failed: %w", err)
+	}
+
+	return &blockId, nil
+}
+
 // GetBlockListing returns a paginated list of blocks in descending order against the order they were added
 func GetBlockListing(db HostDB, pagination *common.QueryPagination) (*common.BlockListingResponse, error) {
 	query := selectBlocks + db.GetSQLStatement().Pagination

go/host/storage/hostdb/block_test.go

@@ -0,0 +1,107 @@
+package hostdb
+
+import (
+	"database/sql"
+	"errors"
+	"math/big"
+	"strings"
+	"testing"
+	"time"
+
+	gethcommon "github.com/ethereum/go-ethereum/common"
+
+	"github.com/ethereum/go-ethereum/core/types"
+)
+
+func TestCanStoreAndRetrieveBlock(t *testing.T) {
+	db, _ := createSQLiteDB(t)
+	block1 := createBlock(batchNumber)
+	block2 := createBlock(batchNumber + 1)
+
+	// verify we get ErrNoRows for a non-existent block
+	randomHash := gethcommon.Hash{}
+	randomHash.SetBytes(make([]byte, 32)) // 32 bytes for appropriate length
+	dbtx, _ := db.NewDBTransaction()
+	statements := db.GetSQLStatement()
+	_, err := GetBlockId(dbtx.Tx, statements, randomHash)
+	if !errors.Is(err, sql.ErrNoRows) {
+		t.Errorf("expected sql.ErrNoRows for non-existent block, got: %v", err)
+	}
+	dbtx.Rollback()
+
+	dbtx, _ = db.NewDBTransaction()
+	err = AddBlock(dbtx.Tx, statements, &block1)
+	if err != nil {
+		t.Errorf("could not store block1: %s", err)
+	}
+	err = dbtx.Write()
+	if err != nil {
+		t.Errorf("could not commit block1: %s", err)
+	}
+
+	dbtx, _ = db.NewDBTransaction()
+	err = AddBlock(dbtx.Tx, statements, &block2)
+	if err != nil {
+		t.Errorf("could not store block2: %s", err)
+	}
+	err = dbtx.Write()
+	if err != nil {
+		t.Errorf("could not commit block2: %s", err)
+	}
+
+	dbtx, _ = db.NewDBTransaction()
+	blockId, err := GetBlockId(dbtx.Tx, statements, block2.Hash())
+	if err != nil {
+		t.Errorf("stored block but could not retrieve block ID: %s", err)
+	}
+	if *blockId != 2 {
+		t.Errorf("expected block ID 2, got %d", *blockId)
+	}
+	dbtx.Rollback()
+}
+
+func TestAddBlockWithForeignKeyConstraint(t *testing.T) {
+	db, _ := createSQLiteDB(t)
+	dbtx, _ := db.NewDBTransaction()
+	statements := db.GetSQLStatement()
+	metadata := createRollupMetadata(batchNumber - 10)
+	rollup := createRollup(batchNumber)
+	block := types.NewBlock(&types.Header{}, nil, nil, nil)
+
+	// add block
+	err := AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
+	if err != nil {
+		t.Errorf("could not store block. Cause: %s", err)
+	}
+	dbtx.Write()
+	dbtx, _ = db.NewDBTransaction()
+
+	// add rollup referencing block
+	err = AddRollup(dbtx, db.GetSQLStatement(), &rollup, &metadata, block)
+	if err != nil {
+		t.Errorf("could not store rollup. Cause: %s", err)
+	}
+	dbtx.Write()
+
+	// this should fail due to the UNIQUE constraint on block_host.hash
+	dbtx, _ = db.NewDBTransaction()
+	err = AddBlock(dbtx.Tx, statements, block.Header())
+	if !strings.Contains(err.Error(), "UNIQUE constraint failed: block_host.hash") {
+		t.Fatalf("expected UNIQUE constraint error, got: %v", err)
+	}
+
+	// verify the block still exists
+	_, err = GetBlockId(dbtx.Tx, statements, block.Hash())
+	if err != nil {
+		t.Fatalf("failed to get block id after duplicate insert: %v", err)
+	}
+
+	dbtx.Rollback()
+}
+
+func createBlock(blockNum int64) types.Header {
+	return types.Header{
+		Number: big.NewInt(blockNum),
+		Time:   uint64(time.Now().Unix()),
+	}
+}

go/host/storage/hostdb/hostdb.go

@@ -38,7 +38,7 @@ func (db *hostDB) NewDBTransaction() (*dbTransaction, error) {
 	}
 
 	return &dbTransaction{
-		tx: tx,
+		Tx: tx,
 	}, nil
 }
 
@@ -50,18 +50,18 @@ func (db *hostDB) Close() error {
 }
 
 type dbTransaction struct {
-	tx *sql.Tx
+	Tx *sql.Tx
 }
 
 func (b *dbTransaction) Write() error {
-	if err := b.tx.Commit(); err != nil {
+	if err := b.Tx.Commit(); err != nil {
 		return fmt.Errorf("failed to commit host db transaction. Cause: %w", err)
 	}
 	return nil
 }
 
 func (b *dbTransaction) Rollback() error {
-	if err := b.tx.Rollback(); err != nil {
+	if err := b.Tx.Rollback(); err != nil {
 		return fmt.Errorf("failed to rollback host transaction. Cause: %w", err)
 	}
 	return nil

go/host/storage/hostdb/rollup.go

@@ -29,12 +29,12 @@ func AddRollup(dbtx *dbTransaction, statements *SQLStatements, rollup *common.Ex
 	}
 
 	var blockId int
-	err = dbtx.tx.QueryRow("select id from block_host where hash="+statements.Placeholder, block.Hash().Bytes()).Scan(&blockId)
+	err = dbtx.Tx.QueryRow("select id from block_host where hash="+statements.Placeholder, block.Hash().Bytes()).Scan(&blockId)
 	if err != nil {
 		return fmt.Errorf("could not read block id: %w", err)
 	}
 
-	_, err = dbtx.tx.Exec(statements.InsertRollup,
+	_, err = dbtx.Tx.Exec(statements.InsertRollup,
 		rollup.Header.Hash().Bytes(),         //  hash
 		metadata.FirstBatchSequence.Uint64(), // first batch sequence
 		rollup.Header.LastBatchSeqNo,         // last batch sequence

go/host/storage/hostdb/rollup_test.go

@@ -22,7 +22,7 @@ func TestCanStoreAndRetrieveRollup(t *testing.T) {
 	rollup := createRollup(batchNumber)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -62,7 +62,7 @@ func TestGetRollupByBlockHash(t *testing.T) {
 	rollup := createRollup(batchNumber)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -94,7 +94,7 @@ func TestGetLatestRollup(t *testing.T) {
 	rollup1 := createRollup(rollup1LastSeq)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -139,7 +139,7 @@ func TestGetRollupBySeqNo(t *testing.T) {
 	rollup1 := createRollup(rollup1LastSeq)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -194,7 +194,7 @@ func TestGetRollupListing(t *testing.T) {
 	rollup1 := createRollup(rollup1LastSeq)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -297,7 +297,7 @@ func TestGetRollupByHash(t *testing.T) {
 	rollup1 := createRollup(rollup1LastSeq)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err = AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err = AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}
@@ -338,7 +338,7 @@ func TestGetRollupBatches(t *testing.T) {
 	batchOne := createBatch(batchNumber, txHashesOne)
 	block := types.NewBlock(&types.Header{}, nil, nil, nil)
 	dbtx, _ := db.NewDBTransaction()
-	err := AddBlock(dbtx, db.GetSQLStatement(), block.Header())
+	err := AddBlock(dbtx.Tx, db.GetSQLStatement(), block.Header())
 	if err != nil {
 		t.Errorf("could not store block. Cause: %s", err)
 	}

go/host/storage/hostdb/sql_statements.go

@@ -26,7 +26,7 @@ func SQLiteSQLStatements() *SQLStatements {
 		InsertTransactions: "INSERT INTO transaction_host (hash, b_sequence) VALUES ",
 		UpdateTxCount:      "UPDATE transaction_count SET total=? WHERE id=1",
 		InsertRollup:       "INSERT INTO rollup_host (hash, start_seq, end_seq, time_stamp, ext_rollup, compression_block) values (?,?,?,?,?,?)",
-		InsertBlock:        "INSERT OR REPLACE INTO block_host (hash, header) values (?,?)",
+		InsertBlock:        "INSERT INTO block_host (hash, header) values (?,?)",
 		Pagination:         "LIMIT ? OFFSET ?",
 		Placeholder:        "?",
 	}

go/host/storage/storage.go

@@ -1,10 +1,12 @@
 package storage
 
 import (
+	"database/sql"
 	"errors"
 	"fmt"
 	"io"
 	"math/big"
+	"strings"
 
 	gethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -78,14 +80,28 @@ func (s *storageImpl) AddRollup(rollup *common.ExtRollup, metadata *common.Publi
 func (s *storageImpl) AddBlock(b *types.Header) error {
 	dbtx, err := s.db.NewDBTransaction()
 	if err != nil {
-		return err
+		return fmt.Errorf("could not create DB transaction - %w", err)
+	}
+	defer dbtx.Rollback()
+
+	_, err = hostdb.GetBlockId(dbtx.Tx, s.db.GetSQLStatement(), b.Hash())
+	switch {
+	case err == nil:
+		// Block already exists
+		s.logger.Debug("Block already exists", "hash", b.Hash().Hex())
+		return nil
+	case !errors.Is(err, sql.ErrNoRows):
+		return fmt.Errorf("error checking block existence: %w", err)
 	}
 
-	if err := hostdb.AddBlock(dbtx, s.db.GetSQLStatement(), b); err != nil {
-		if err := dbtx.Rollback(); err != nil {
-			return err
+	if err := hostdb.AddBlock(dbtx.Tx, s.db.GetSQLStatement(), b); err != nil {
+		if IsConstraintError(err) {
+			s.logger.Debug("Block already exists",
+				"hash", b.Hash().Hex(),
+				"error", err)
+			return nil
 		}
-		return fmt.Errorf("could not add block to host. Cause: %w", err)
+		return fmt.Errorf("could not add block to host: %w", err)
 	}
 
 	if err := dbtx.Write(); err != nil {
@@ -204,3 +220,19 @@ func NewStorage(backingDB hostdb.HostDB, logger gethlog.Logger) Storage {
 		logger: logger,
 	}
 }
+
+// SQLite constraint error messages
+const (
+	ErrUniqueBlockHash = "UNIQUE constraint failed: block_host.hash"
+	ErrForeignKey      = "FOREIGN KEY constraint failed"
+)
+
+// IsConstraintError returns true if the error is a known constraint error
+func IsConstraintError(err error) bool {
+	if err == nil {
+		return false
+	}
+	errMsg := err.Error()
+	return strings.Contains(errMsg, ErrUniqueBlockHash) ||
+		strings.Contains(errMsg, ErrForeignKey)
+}