Vdb 269- fetch logs by hash (#122)

* Upgrade geth from 1.8.15 to 1.8.18

* Update vat_tune to use shared repository methods

* Query blockchain by block hash instead of block number range

* Remove hash validation from repositories

* Fix vow flog integration test

* Update README Travis build sticker

* Update constants formatting per go fmt

* Update EthPublicKeyParser.ParsePublicKey to use discv5.PubkeyID method

* Address PR comments

pkg/transformers/bite/repository.go

@@ -42,12 +42,6 @@ func (repository BiteRepository) Create(headerID int64, models []interface{}) er
 			return fmt.Errorf("model of type %T, not %T", model, BiteModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, biteModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.bite (header_id, ilk, urn, ink, art, iart, tab, nflip, log_idx, tx_idx, raw_log)
         VALUES($1, $2, $3, $4::NUMERIC, $5::NUMERIC, $6::NUMERIC, $7::NUMERIC, $8::NUMERIC, $9, $10, $11)`,

pkg/transformers/cat_file/chop_lump/repository.go

@@ -39,12 +39,6 @@ func (repository CatFileChopLumpRepository) Create(headerID int64, models []inte
 			return fmt.Errorf("model of type %T, not %T", model, CatFileChopLumpModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, chopLump.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.cat_file_chop_lump (header_id, ilk, what, data, tx_idx, log_idx, raw_log)
 			VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/cat_file/flip/repository.go

@@ -38,12 +38,6 @@ func (repository CatFileFlipRepository) Create(headerID int64, models []interfac
 			return fmt.Errorf("model of type %T, not %T", model, CatFileFlipModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, flip.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = repository.db.Exec(
 			`INSERT into maker.cat_file_flip (header_id, ilk, what, flip, tx_idx, log_idx, raw_log)
 			VALUES($1, $2, $3, $4, $5, $6, $7)`,

pkg/transformers/cat_file/pit_vow/repository.go

@@ -38,12 +38,6 @@ func (repository CatFilePitVowRepository) Create(headerID int64, models []interf
 			return fmt.Errorf("model of type %T, not %T", model, CatFilePitVowModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vow.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = repository.db.Exec(
 			`INSERT into maker.cat_file_pit_vow (header_id, what, data, tx_idx, log_idx, raw_log)
 			VALUES($1, $2, $3, $4, $5, $6)`,

pkg/transformers/deal/repository.go

@@ -39,12 +39,6 @@ func (repository DealRepository) Create(headerID int64, models []interface{}) er
 			return fmt.Errorf("model of type %T, not %T", model, DealModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, dealModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.deal (header_id, bid_id, contract_address, log_idx, tx_idx, raw_log)
 					 VALUES($1, $2, $3, $4, $5, $6)`,

pkg/transformers/dent/repository.go

@@ -44,12 +44,6 @@ func (repository DentRepository) Create(headerID int64, models []interface{}) er
 			return fmt.Errorf("model of type %T, not %T", model, DentModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, dent.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.dent (header_id, bid_id, lot, bid, guy, tic, log_idx, tx_idx, raw_log)
 			VALUES($1, $2, $3, $4, $5, $6, $7, $8, $9)`,

pkg/transformers/drip_drip/repository.go

@@ -38,12 +38,6 @@ func (repository DripDripRepository) Create(headerID int64, models []interface{}
 			return fmt.Errorf("model of type %T, not %T", model, DripDripModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, dripDrip.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.drip_drip (header_id, ilk, log_idx, tx_idx, raw_log)
         VALUES($1, $2, $3, $4, $5)`,

pkg/transformers/drip_file/ilk/repository.go

@@ -39,12 +39,6 @@ func (repository DripFileIlkRepository) Create(headerID int64, models []interfac
 			return fmt.Errorf("model of type %T, not %T", model, DripFileIlkModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, ilk.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_ilk (header_id, ilk, vow, tax, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/drip_file/repo/repository.go

@@ -39,12 +39,6 @@ func (repository DripFileRepoRepository) Create(headerID int64, models []interfa
 			return fmt.Errorf("model of type %T, not %T", model, DripFileRepoModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, repo.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_repo (header_id, what, data, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3::NUMERIC, $4, $5, $6)`,

pkg/transformers/drip_file/vow/repository.go

@@ -39,12 +39,6 @@ func (repository DripFileVowRepository) Create(headerID int64, models []interfac
 			return fmt.Errorf("model of type %T, not %T", model, DripFileVowModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vow.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.drip_file_vow (header_id, what, data, log_idx, tx_idx, raw_log)
         	VALUES($1, $2, $3, $4, $5, $6)`,

pkg/transformers/factories/log_note_transformer.go

@@ -52,7 +52,7 @@ func (transformer LogNoteTransformer) Execute() error {
 	log.Printf("Fetching %v event logs for %d headers \n", transformerName, len(missingHeaders))
 	for _, header := range missingHeaders {
 		// Fetch the missing logs for a given header
-		matchingLogs, err := transformer.Fetcher.FetchLogs(transformer.Config.ContractAddresses, topic, header.BlockNumber)
+		matchingLogs, err := transformer.Fetcher.FetchLogs(transformer.Config.ContractAddresses, topic, header)
 		if err != nil {
 			log.Printf("Error fetching matching logs in %v transformer: %v", transformerName, err)
 			return err

pkg/transformers/factories/transformer.go

@@ -48,7 +48,7 @@ func (transformer Transformer) Execute() error {
 	}
 	log.Printf("Fetching %v event logs for %d headers \n", transformerName, len(missingHeaders))
 	for _, header := range missingHeaders {
-		logs, err := transformer.Fetcher.FetchLogs(config.ContractAddresses, topics, header.BlockNumber)
+		logs, err := transformer.Fetcher.FetchLogs(config.ContractAddresses, topics, header)
 		if err != nil {
 			log.Printf("Error fetching matching logs in %v transformer: %v", transformerName, err)
 			return err

pkg/transformers/flap_kick/repository.go

@@ -37,12 +37,6 @@ func (repository *FlapKickRepository) Create(headerID int64, models []interface{
 			return fmt.Errorf("model of type %T, not %T", model, FlapKickModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, flapKickModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.flap_kick (header_id, bid_id, lot, bid, gal, "end", tx_idx, log_idx, raw_log)
         VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8, $9)`,

pkg/transformers/flip_kick/repository.go

@@ -38,12 +38,6 @@ func (repository FlipKickRepository) Create(headerID int64, models []interface{}
 			return fmt.Errorf("model of type %T, not %T", model, FlipKickModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, flipKickModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.flip_kick (header_id, bid_id, lot, bid, gal, "end", urn, tab, tx_idx, log_idx, raw_log)
         VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8::NUMERIC, $9, $10, $11)`,

pkg/transformers/flop_kick/repository.go

@@ -38,12 +38,6 @@ func (repository FlopKickRepository) Create(headerID int64, models []interface{}
 			return fmt.Errorf("model of type %T, not %T", flopKick, Model{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, flopKickModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.flop_kick (header_id, bid_id, lot, bid, gal, "end", tx_idx, log_idx, raw_log)
         VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8, $9)`,

pkg/transformers/frob/repository.go

@@ -39,12 +39,6 @@ func (repository FrobRepository) Create(headerID int64, models []interface{}) er
 			return fmt.Errorf("model of type %T, not %T", model, FrobModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, frobModel.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(`INSERT INTO maker.frob (header_id, art, dart, dink, iart, ilk, ink, urn, raw_log, log_idx, tx_idx)
 		VALUES($1, $2::NUMERIC, $3::NUMERIC, $4::NUMERIC, $5::NUMERIC, $6, $7::NUMERIC, $8, $9, $10, $11)`,
 			headerID, frobModel.Art, frobModel.Dart, frobModel.Dink, frobModel.IArt, frobModel.Ilk, frobModel.Ink, frobModel.Urn, frobModel.Raw, frobModel.LogIndex, frobModel.TransactionIndex)

pkg/transformers/integration_tests/vow_flog.go

@@ -39,7 +39,7 @@ var _ = Describe("VowFlog LogNoteTransformer", func() {
 		db := test_config.NewTestDB(blockchain.Node())
 		test_config.CleanTestDB(db)
 
-		err = persistHeader(db, blockNumber)
+		err = persistHeader(db, blockNumber, blockchain)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(1).To(Equal(1))
 

pkg/transformers/pit_file/debt_ceiling/repository.go

@@ -39,12 +39,6 @@ func (repository PitFileDebtCeilingRepository) Create(headerID int64, models []i
 			return fmt.Errorf("model of type %T, not %T", model, PitFileDebtCeilingModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, pitFileDC.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.pit_file_debt_ceiling (header_id, what, data, log_idx, tx_idx, raw_log)
         VALUES($1, $2, $3::NUMERIC, $4, $5, $6)`,

pkg/transformers/pit_file/ilk/repository.go

@@ -39,12 +39,6 @@ func (repository PitFileIlkRepository) Create(headerID int64, models []interface
 			return fmt.Errorf("model of type %T, not %T", model, PitFileIlkModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, pitFileIlk.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.pit_file_ilk (header_id, ilk, what, data, log_idx, tx_idx, raw_log)
         VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/price_feeds/repository.go

@@ -38,12 +38,6 @@ func (repository PriceFeedRepository) Create(headerID int64, models []interface{
 			return fmt.Errorf("model of type %T, not %T", model, PriceFeedModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, priceUpdate.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(`INSERT INTO maker.price_feeds (block_number, header_id, medianizer_address, usd_value, log_idx, tx_idx, raw_log)
 		VALUES ($1, $2, $3, $4::NUMERIC, $5, $6, $7)`, priceUpdate.BlockNumber, headerID, priceUpdate.MedianizerAddress, priceUpdate.UsdValue, priceUpdate.LogIndex, priceUpdate.TransactionIndex, priceUpdate.Raw)
 		if err != nil {

pkg/transformers/shared/constants/label.go

@@ -1,34 +1,34 @@
 package constants
 
 var (
-	BiteLabel                = "bite"
-	CatFileChopLumpLabel     = "catFileChopLump"
-	CatFileFlipLabel         = "catFileFlip"
-	CatFilePitVowLabel       = "catFilePitVow"
-	DealLabel                = "deal"
-	DentLabel                = "dent"
-	DripDripLabel            = "dripDrip"
-	DripFileIlkLabel         = "dripFileIlk"
-	DripFileRepoLabel        = "dripFileRepo"
-	DripFileVowLabel         = "dripFileVow"
-	FlapKickLabel            = "flapKick"
-	FlipKickLabel            = "flipKick"
-	FlopKickLabel            = "flopKick"
-	FrobLabel                = "frob"
-	PitFileDebtCeilingLabel  = "pitFileDebtCeiling"
-	PitFileIlkLabel          = "pitFileIlk"
-	PriceFeedLabel           = "priceFeed"
-	TendLabel                = "tend"
-	VatFluxLabel             = "vatFlux"
-	VatFoldLabel             = "vatFold"
-	VatGrabLabel             = "vatGrab"
-	VatHealLabel             = "vatHeal"
-	VatInitLabel             = "vatInit"
-	VatMoveLabel             = "vatMove"
-	VatSlipLabel             = "vatSlip"
-	VatTollLabel             = "vatToll"
-	VatTuneLabel             = "vatTune"
-	VowFlogLabel             = "vowFlog"
+	BiteLabel               = "bite"
+	CatFileChopLumpLabel    = "catFileChopLump"
+	CatFileFlipLabel        = "catFileFlip"
+	CatFilePitVowLabel      = "catFilePitVow"
+	DealLabel               = "deal"
+	DentLabel               = "dent"
+	DripDripLabel           = "dripDrip"
+	DripFileIlkLabel        = "dripFileIlk"
+	DripFileRepoLabel       = "dripFileRepo"
+	DripFileVowLabel        = "dripFileVow"
+	FlapKickLabel           = "flapKick"
+	FlipKickLabel           = "flipKick"
+	FlopKickLabel           = "flopKick"
+	FrobLabel               = "frob"
+	PitFileDebtCeilingLabel = "pitFileDebtCeiling"
+	PitFileIlkLabel         = "pitFileIlk"
+	PriceFeedLabel          = "priceFeed"
+	TendLabel               = "tend"
+	VatFluxLabel            = "vatFlux"
+	VatFoldLabel            = "vatFold"
+	VatGrabLabel            = "vatGrab"
+	VatHealLabel            = "vatHeal"
+	VatInitLabel            = "vatInit"
+	VatMoveLabel            = "vatMove"
+	VatSlipLabel            = "vatSlip"
+	VatTollLabel            = "vatToll"
+	VatTuneLabel            = "vatTune"
+	VowFlogLabel            = "vowFlog"
 )
 
 func AllTransformerLabels() []string {

pkg/transformers/shared/log_fetcher.go

@@ -15,8 +15,6 @@
 package shared
 
 import (
-	"math/big"
-
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -25,7 +23,7 @@ import (
 )
 
 type LogFetcher interface {
-	FetchLogs(contractAddresses []string, topics [][]common.Hash, blockNumber int64) ([]types.Log, error)
+	FetchLogs(contractAddresses []string, topics [][]common.Hash, header core.Header) ([]types.Log, error)
 }
 
 type SettableLogFetcher interface {
@@ -47,12 +45,11 @@ func NewFetcher(blockchain core.BlockChain) Fetcher {
 	}
 }
 
-func (fetcher Fetcher) FetchLogs(contractAddresses []string, topics [][]common.Hash, blockNumber int64) ([]types.Log, error) {
-	block := big.NewInt(blockNumber)
+func (fetcher Fetcher) FetchLogs(contractAddresses []string, topics [][]common.Hash, header core.Header) ([]types.Log, error) {
 	addresses := hexStringsToAddresses(contractAddresses)
+	blockHash := common.HexToHash(header.Hash)
 	query := ethereum.FilterQuery{
-		FromBlock: block,
-		ToBlock:   block,
+		BlockHash: &blockHash,
 		Addresses: addresses,
 		Topics:    topics,
 	}

pkg/transformers/shared/log_fetcher_test.go

@@ -15,13 +15,12 @@
 package shared_test
 
 import (
-	"math/big"
-
 	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
+	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/fakes"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 )
@@ -31,18 +30,20 @@ var _ = Describe("Fetcher", func() {
 		It("fetches logs based on the given query", func() {
 			blockChain := fakes.NewMockBlockChain()
 			fetcher := shared.NewFetcher(blockChain)
-			blockNumber := int64(123)
+			header := fakes.FakeHeader
+
 			addresses := []string{"0xfakeAddress", "0xanotherFakeAddress"}
 			topicZeros := [][]common.Hash{{common.BytesToHash([]byte{1, 2, 3, 4, 5})}}
 
-			_, err := fetcher.FetchLogs(addresses, topicZeros, blockNumber)
+			_, err := fetcher.FetchLogs(addresses, topicZeros, header)
 
 			address1 := common.HexToAddress("0xfakeAddress")
 			address2 := common.HexToAddress("0xanotherFakeAddress")
 			Expect(err).NotTo(HaveOccurred())
+
+			blockHash := common.HexToHash(header.Hash)
 			expectedQuery := ethereum.FilterQuery{
-				FromBlock: big.NewInt(blockNumber),
-				ToBlock:   big.NewInt(blockNumber),
+				BlockHash: &blockHash,
 				Addresses: []common.Address{address1, address2},
 				Topics:    topicZeros,
 			}
@@ -54,7 +55,7 @@ var _ = Describe("Fetcher", func() {
 			blockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
 			fetcher := shared.NewFetcher(blockChain)
 
-			_, err := fetcher.FetchLogs([]string{}, [][]common.Hash{}, int64(1))
+			_, err := fetcher.FetchLogs([]string{}, [][]common.Hash{}, core.Header{})
 
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(fakes.FakeError))

pkg/transformers/shared/repository.go

@@ -2,41 +2,10 @@ package shared
 
 import (
 	"database/sql"
-	"encoding/json"
-	"errors"
-	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 )
 
-var ErrHeaderMismatch = errors.New("log's block hash does not match associated header's block hash")
-
-func ValidateHeaderConsistency(headerID int64, rawLog []byte, db *postgres.DB) error {
-	var log types.Log
-	err := json.Unmarshal(rawLog, &log)
-	if err != nil {
-		return err
-	}
-	var hash string
-	err = db.Get(&hash, `SELECT hash FROM public.headers WHERE id = $1`, headerID)
-	if err != nil {
-		return err
-	}
-	if hash != log.BlockHash.String() {
-		err = deleteHeader(headerID, db)
-		if err != nil {
-			return err
-		}
-		return ErrHeaderMismatch
-	}
-	return nil
-}
-
-func deleteHeader(headerID int64, db *postgres.DB) error {
-	_, err := db.Exec(`DELETE FROM public.headers WHERE id = $1`, headerID)
-	return err
-}
-
 func MarkHeaderChecked(headerID int64, db *postgres.DB, checkedHeadersColumn string) error {
 	_, err := db.Exec(`INSERT INTO public.checked_headers (header_id, `+checkedHeadersColumn+`)
 		VALUES ($1, $2) 
@@ -59,14 +28,14 @@ func MissingHeaders(startingBlockNumber, endingBlockNumber int64, db *postgres.D
 	var err error
 
 	if endingBlockNumber == -1 {
-		query = `SELECT headers.id, headers.block_number FROM headers
+		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
 				LEFT JOIN checked_headers on headers.id = header_id
 				WHERE (header_id ISNULL OR ` + checkedHeadersColumn + ` IS FALSE)
 				AND headers.block_number >= $1
 				AND headers.eth_node_fingerprint = $2`
 		err = db.Select(&result, query, startingBlockNumber, db.Node.ID)
 	} else {
-		query = `SELECT headers.id, headers.block_number FROM headers
+		query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
 				LEFT JOIN checked_headers on headers.id = header_id
 				WHERE (header_id ISNULL OR ` + checkedHeadersColumn + ` IS FALSE)
 				AND headers.block_number >= $1

pkg/transformers/tend/repository.go

@@ -44,12 +44,6 @@ func (repository TendRepository) Create(headerID int64, models []interface{}) er
 			return fmt.Errorf("model of type %T, not %T", model, TendModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, tend.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.tend (header_id, bid_id, lot, bid, guy, tic, log_idx, tx_idx, raw_log)
 			VALUES($1, $2, $3::NUMERIC, $4::NUMERIC, $5, $6, $7, $8, $9)`,

pkg/transformers/test_data/mocks/log_fetcher.go

@@ -29,10 +29,10 @@ type MockLogFetcher struct {
 	SetBcCalled              bool
 }
 
-func (mlf *MockLogFetcher) FetchLogs(contractAddresses []string, topics [][]common.Hash, blockNumber int64) ([]types.Log, error) {
+func (mlf *MockLogFetcher) FetchLogs(contractAddresses []string, topics [][]common.Hash, header core.Header) ([]types.Log, error) {
 	mlf.FetchedContractAddresses = append(mlf.FetchedContractAddresses, contractAddresses)
 	mlf.FetchedTopics = topics
-	mlf.FetchedBlocks = append(mlf.FetchedBlocks, blockNumber)
+	mlf.FetchedBlocks = append(mlf.FetchedBlocks, header.BlockNumber)
 
 	return mlf.FetchedLogs, mlf.fetcherError
 }

pkg/transformers/test_data/shared_behaviors/repository_behaviors.go

@@ -15,7 +15,6 @@
 package shared_behaviors
 
 import (
-	"github.com/ethereum/go-ethereum/common"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	"github.com/vulcanize/vulcanizedb/pkg/core"
@@ -24,7 +23,6 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
 	"github.com/vulcanize/vulcanizedb/pkg/fakes"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
-	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
 	"github.com/vulcanize/vulcanizedb/test_config"
 	"math/rand"
@@ -103,35 +101,6 @@ func SharedRepositoryCreateBehaviors(inputs *CreateBehaviorInputs) {
 			Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
 		})
 
-		Describe("when log's block hash does not match associated header's block hash", func() {
-			var (
-				differentHeaderID int64
-				err               error
-			)
-
-			BeforeEach(func() {
-				differentHeader := fakes.FakeHeader
-				differentHeader.BlockNumber = fakes.FakeHeader.BlockNumber + 1
-				differentHeader.Hash = common.BytesToHash(append(fakes.FakeHash.Bytes(), 1)).String()
-				differentHeaderID, err = headerRepository.CreateOrUpdateHeader(differentHeader)
-				Expect(err).NotTo(HaveOccurred())
-
-				err = repository.Create(differentHeaderID, []interface{}{logEventModel})
-			})
-
-			It("returns error", func() {
-				Expect(err).To(HaveOccurred())
-				Expect(err).To(MatchError(shared.ErrHeaderMismatch))
-			})
-
-			It("removes header", func() {
-				var headerIDs []int64
-				err = db.Select(&headerIDs, `SELECT id FROM public.headers`)
-				Expect(err).NotTo(HaveOccurred())
-				Expect(headerIDs).NotTo(ContainElement(differentHeaderID))
-			})
-		})
-
 		It("allows for multiple log events of the same type in one transaction if they have different log indexes", func() {
 			err = repository.Create(headerID, []interface{}{logEventModel})
 			Expect(err).NotTo(HaveOccurred())

pkg/transformers/vat_flux/repository.go

@@ -39,12 +39,6 @@ func (repository VatFluxRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatFluxModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatFlux.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(`INSERT INTO maker.vat_flux (header_id, ilk, dst, src, rad, tx_idx, log_idx, raw_log)
 		VALUES($1, $2, $3, $4, $5::NUMERIC, $6, $7, $8)`,
 			headerID, vatFlux.Ilk, vatFlux.Dst, vatFlux.Src, vatFlux.Rad, vatFlux.TransactionIndex, vatFlux.LogIndex, vatFlux.Raw)

pkg/transformers/vat_fold/repository.go

@@ -39,12 +39,6 @@ func (repository VatFoldRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatFoldModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatFold.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vat_fold (header_id, ilk, urn, rate, log_idx, tx_idx, raw_log)
 				VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/vat_grab/repository.go

@@ -25,12 +25,6 @@ func (repository VatGrabRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatGrabModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatGrab.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vat_grab (header_id, ilk, urn, v, w, dink, dart, log_idx, tx_idx, raw_log)
 	   VALUES($1, $2, $3, $4, $5, $6::NUMERIC, $7::NUMERIC, $8, $9, $10)`,

pkg/transformers/vat_heal/repository.go

@@ -44,12 +44,6 @@ func (repository VatHealRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatHealModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatHeal.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err := tx.Exec(`INSERT INTO maker.vat_heal (header_id, urn, v, rad, log_idx, tx_idx, raw_log)
 		VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,
 			headerID, vatHeal.Urn, vatHeal.V, vatHeal.Rad, vatHeal.LogIndex, vatHeal.TransactionIndex, vatHeal.Raw)

pkg/transformers/vat_init/repository.go

@@ -39,12 +39,6 @@ func (repository VatInitRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatInitModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatInit.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT INTO maker.vat_init (header_id, ilk, log_idx, tx_idx, raw_log)
 			VALUES($1, $2, $3, $4, $5)`,

pkg/transformers/vat_move/repository.go

@@ -39,12 +39,6 @@ func (repository VatMoveRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatMoveModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatMove.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT INTO maker.vat_move (header_id, src, dst, rad, log_idx, tx_idx, raw_log)
 				VALUES ($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/vat_slip/repository.go

@@ -24,12 +24,6 @@ func (repository VatSlipRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatSlipModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatSlip.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vat_slip (header_id, ilk, guy, rad, tx_idx, log_idx, raw_log)
 			VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/vat_toll/repository.go

@@ -24,12 +24,6 @@ func (repository VatTollRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatTollModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatToll.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vat_toll (header_id, ilk, urn, take, tx_idx, log_idx, raw_log)
 			VALUES($1, $2, $3, $4::NUMERIC, $5, $6, $7)`,

pkg/transformers/vat_tune/repository.go

@@ -5,6 +5,7 @@ import (
 	"github.com/vulcanize/vulcanizedb/pkg/core"
 	"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
 	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
+	"github.com/vulcanize/vulcanizedb/pkg/transformers/shared/constants"
 )
 
 type VatTuneRepository struct {
@@ -23,12 +24,6 @@ func (repository VatTuneRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VatTuneModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, vatTune.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vat_tune (header_id, ilk, urn, v, w, dink, dart, tx_idx, log_idx, raw_log)
 	   VALUES($1, $2, $3, $4, $5, $6::NUMERIC, $7::NUMERIC, $8, $9, $10)`,
@@ -40,10 +35,7 @@ func (repository VatTuneRepository) Create(headerID int64, models []interface{})
 		}
 	}
 
-	_, err = tx.Exec(`INSERT INTO public.checked_headers (header_id, vat_tune_checked)
-		VALUES ($1, $2)
-	ON CONFLICT (header_id) DO
-		UPDATE SET vat_tune_checked = $2`, headerID, true)
+	err = shared.MarkHeaderCheckedInTransaction(headerID, tx, constants.VatTuneChecked)
 	if err != nil {
 		tx.Rollback()
 		return err
@@ -52,28 +44,11 @@ func (repository VatTuneRepository) Create(headerID int64, models []interface{})
 }
 
 func (repository VatTuneRepository) MarkHeaderChecked(headerID int64) error {
-	_, err := repository.db.Exec(`INSERT INTO public.checked_headers (header_id, vat_tune_checked)
-		VALUES ($1, $2)
-	ON CONFLICT (header_id) DO
-		UPDATE SET vat_tune_checked = $2`, headerID, true)
-	return err
+	return shared.MarkHeaderChecked(headerID, repository.db, constants.VatTuneChecked)
 }
 
 func (repository VatTuneRepository) MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error) {
-	var result []core.Header
-	err := repository.db.Select(
-		&result,
-		`SELECT headers.id, headers.block_number FROM headers
-	          LEFT JOIN checked_headers on headers.id = header_id
-	          WHERE (header_id ISNULL OR vat_tune_checked IS FALSE)
-	          AND headers.block_number >= $1
-	          AND headers.block_number <= $2
-	          AND headers.eth_node_fingerprint = $3`,
-		startingBlockNumber,
-		endingBlockNumber,
-		repository.db.Node.ID,
-	)
-	return result, err
+	return shared.MissingHeaders(startingBlockNumber, endingBlockNumber, repository.db, constants.VatTuneChecked)
 }
 
 func (repository *VatTuneRepository) SetDB(db *postgres.DB) {

pkg/transformers/vow_flog/repository.go

@@ -39,12 +39,6 @@ func (repository VowFlogRepository) Create(headerID int64, models []interface{})
 			return fmt.Errorf("model of type %T, not %T", model, VowFlogModel{})
 		}
 
-		err = shared.ValidateHeaderConsistency(headerID, flog.Raw, repository.db)
-		if err != nil {
-			tx.Rollback()
-			return err
-		}
-
 		_, err = tx.Exec(
 			`INSERT into maker.vow_flog (header_id, era, log_idx, tx_idx, raw_log)
 			VALUES($1, $2::NUMERIC, $3, $4, $5)`,