retriever unit test

pkg/ipfs/mocks/test_data.go

@@ -97,10 +97,10 @@ var (
 		Leaf:  true,
 	}}
 	emptyStorage           = make([]statediff.StorageDiff, 0)
-	address                = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476592")
+	Address                = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476592")
-	ContractLeafKey        = ipfs.AddressToKey(address)
+	ContractLeafKey        = ipfs.AddressToKey(Address)
-	anotherAddress         = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476593")
+	AnotherAddress         = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476593")
-	AnotherContractLeafKey = ipfs.AddressToKey(anotherAddress)
+	AnotherContractLeafKey = ipfs.AddressToKey(AnotherAddress)
 	testAccount            = state.Account{
 		Nonce:    NonceValue,
 		Balance:  big.NewInt(BalanceValue),
@@ -273,6 +273,34 @@ var (
 			},
 		},
 	}
+
+	MockCIDWrapper = &ipfs.CIDWrapper{
+		BlockNumber:  big.NewInt(1),
+		Headers:      []string{"mockHeaderCID"},
+		Transactions: []string{"mockTrxCID1", "mockTrxCID2"},
+		Receipts:     []string{"mockRctCID1", "mockRctCID2"},
+		Uncles:       []string{},
+		StateNodes: []ipfs.StateNodeCID{
+			{
+				CID:  "mockStateCID1",
+				Leaf: true,
+				Key:  ContractLeafKey.Hex(),
+			},
+			{
+				CID:  "mockStateCID2",
+				Leaf: true,
+				Key:  AnotherContractLeafKey.Hex(),
+			},
+		},
+		StorageNodes: []ipfs.StorageNodeCID{
+			{
+				CID:      "mockStorageCID",
+				Leaf:     true,
+				StateKey: ContractLeafKey.Hex(),
+				Key:      "0x0000000000000000000000000000000000000000000000000000000000000001",
+			},
+		},
+	}
 )
 
 // createTransactionsAndReceipts is a helper function to generate signed mock transactions and mock receipts with mock logs

pkg/seed_node/repository.go

@@ -50,24 +50,36 @@ func (repo *Repository) Index(cidPayload *ipfs.CIDPayload) error {
 	}
 	headerID, err := repo.indexHeaderCID(tx, cidPayload.HeaderCID, cidPayload.BlockNumber, cidPayload.BlockHash.Hex())
 	if err != nil {
-		log.Error(tx.Rollback())
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
 		return err
 	}
 	for uncleHash, cid := range cidPayload.UncleCIDS {
 		err = repo.indexUncleCID(tx, cid, cidPayload.BlockNumber, uncleHash.Hex())
 		if err != nil {
-			log.Error(tx.Rollback())
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			return err
 		}
 	}
 	err = repo.indexTransactionAndReceiptCIDs(tx, cidPayload, headerID)
 	if err != nil {
-		log.Error(tx.Rollback())
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
 		return err
 	}
 	err = repo.indexStateAndStorageCIDs(tx, cidPayload, headerID)
 	if err != nil {
-		log.Error(tx.Rollback())
+		rollbackErr := tx.Rollback()
+		if rollbackErr != nil {
+			log.Error(rollbackErr)
+		}
 		return err
 	}
 	return tx.Commit()

pkg/seed_node/repository_test.go

@@ -60,8 +60,8 @@ var _ = Describe("Repository", func() {
 			err = db.Select(&trxs, pgStr, 1)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(len(trxs)).To(Equal(2))
-			Expect(trxs[0]).To(Equal("mockTrxCID1"))
+			Expect(seed_node.ListContains(trxs, "mockTrxCID1")).To(BeTrue())
-			Expect(trxs[1]).To(Equal("mockTrxCID2"))
+			Expect(seed_node.ListContains(trxs, "mockTrxCID2")).To(BeTrue())
 			// check receipts were properly indexed
 			rcts := make([]string, 0)
 			pgStr = `SELECT receipt_cids.cid FROM receipt_cids, transaction_cids, header_cids
@@ -71,8 +71,8 @@ var _ = Describe("Repository", func() {
 			err = db.Select(&rcts, pgStr, 1)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(len(rcts)).To(Equal(2))
-			Expect(rcts[0]).To(Equal("mockRctCID1"))
+			Expect(seed_node.ListContains(rcts, "mockRctCID1")).To(BeTrue())
-			Expect(rcts[1]).To(Equal("mockRctCID2"))
+			Expect(seed_node.ListContains(rcts, "mockRctCID2")).To(BeTrue())
 			// check that state nodes were properly indexed
 			stateNodes := make([]ipfs.StateNodeCID, 0)
 			pgStr = `SELECT state_cids.cid, state_cids.state_key, state_cids.leaf FROM state_cids INNER JOIN header_cids ON (state_cids.header_id = header_cids.id)
@@ -80,16 +80,16 @@ var _ = Describe("Repository", func() {
 			err = db.Select(&stateNodes, pgStr, 1)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(len(stateNodes)).To(Equal(2))
-			Expect(stateNodes[0]).To(Equal(ipfs.StateNodeCID{
+			for _, stateNode := range stateNodes {
-				CID:  "mockStateCID1",
+				if stateNode.CID == "mockStateCID1" {
-				Leaf: true,
+					Expect(stateNode.Leaf).To(Equal(true))
-				Key:  mocks.ContractLeafKey.Hex(),
+					Expect(stateNode.Key).To(Equal(mocks.ContractLeafKey.Hex()))
-			}))
+				}
-			Expect(stateNodes[1]).To(Equal(ipfs.StateNodeCID{
+				if stateNode.CID == "mockStateCID2" {
-				CID:  "mockStateCID2",
+					Expect(stateNode.Leaf).To(Equal(true))
-				Leaf: true,
+					Expect(stateNode.Key).To(Equal(mocks.AnotherContractLeafKey.Hex()))
-				Key:  mocks.AnotherContractLeafKey.Hex(),
+				}
-			}))
+			}
 			// check that storage nodes were properly indexed
 			storageNodes := make([]ipfs.StorageNodeCID, 0)
 			pgStr = `SELECT storage_cids.cid, state_cids.state_key, storage_cids.storage_key, storage_cids.leaf FROM storage_cids, state_cids, header_cids

pkg/seed_node/retriever.go

@@ -77,15 +77,21 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	if !streamFilters.HeaderFilter.Off {
 		cw.Headers, err = ecr.retrieveHeaderCIDs(tx, streamFilters, blockNumber)
 		if err != nil {
-			tx.Rollback()
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			log.Error("header cid retrieval error")
 			return nil, err
 		}
 		if !streamFilters.HeaderFilter.FinalOnly {
 			cw.Uncles, err = ecr.retrieveUncleCIDs(tx, streamFilters, blockNumber)
 			if err != nil {
-				tx.Rollback()
+				rollbackErr := tx.Rollback()
-				log.Error("header cid retrieval error")
+				if rollbackErr != nil {
+					log.Error(rollbackErr)
+				}
+				log.Error("uncle cid retrieval error")
 				return nil, err
 			}
 		}
@@ -96,7 +102,10 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	if !streamFilters.TrxFilter.Off {
 		cw.Transactions, trxIds, err = ecr.retrieveTrxCIDs(tx, streamFilters, blockNumber)
 		if err != nil {
-			tx.Rollback()
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			log.Error("transaction cid retrieval error")
 			return nil, err
 		}
@@ -106,7 +115,10 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	if !streamFilters.ReceiptFilter.Off {
 		cw.Receipts, err = ecr.retrieveRctCIDs(tx, streamFilters, blockNumber, trxIds)
 		if err != nil {
-			tx.Rollback()
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			log.Error("receipt cid retrieval error")
 			return nil, err
 		}
@@ -116,7 +128,10 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	if !streamFilters.StateFilter.Off {
 		cw.StateNodes, err = ecr.retrieveStateCIDs(tx, streamFilters, blockNumber)
 		if err != nil {
-			tx.Rollback()
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			log.Error("state cid retrieval error")
 			return nil, err
 		}
@@ -126,7 +141,10 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription, bloc
 	if !streamFilters.StorageFilter.Off {
 		cw.StorageNodes, err = ecr.retrieveStorageCIDs(tx, streamFilters, blockNumber)
 		if err != nil {
-			tx.Rollback()
+			rollbackErr := tx.Rollback()
+			if rollbackErr != nil {
+				log.Error(rollbackErr)
+			}
 			log.Error("storage cid retrieval error")
 			return nil, err
 		}
@@ -187,7 +205,7 @@ func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Su
 
 func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64, trxIds []int64) ([]string, error) {
 	log.Debug("retrieving receipt cids for block ", blockNumber)
-	args := make([]interface{}, 0, 2)
+	args := make([]interface{}, 0, 4)
 	pgStr := `SELECT receipt_cids.cid FROM receipt_cids, transaction_cids, header_cids
 			WHERE receipt_cids.tx_id = transaction_cids.id 
 			AND transaction_cids.header_id = header_cids.id
@@ -196,17 +214,38 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
 	if len(streamFilters.ReceiptFilter.Topic0s) > 0 {
 		pgStr += ` AND ((receipt_cids.topic0s && $2::VARCHAR(66)[]`
 		args = append(args, pq.Array(streamFilters.ReceiptFilter.Topic0s))
-	}
+		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
-	if len(streamFilters.ReceiptFilter.Contracts) > 0 {
+			pgStr += ` AND receipt_cids.contract = ANY($3::VARCHAR(66)[]))`
-		pgStr += ` AND receipt_cids.contract = ANY($3::VARCHAR(66)[])`
+			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
+			if len(trxIds) > 0 {
+				pgStr += ` OR receipt_cids.tx_id = ANY($4::INTEGER[]))`
+				args = append(args, pq.Array(trxIds))
+			} else {
+				pgStr += `)`
+			}
+		} else {
+			pgStr += `)`
+			if len(trxIds) > 0 {
+				pgStr += ` OR receipt_cids.tx_id = ANY($3::INTEGER[]))`
+				args = append(args, pq.Array(trxIds))
+			} else {
+				pgStr += `)`
+			}
+		}
 	} else {
-		pgStr += `)`
+		if len(streamFilters.ReceiptFilter.Contracts) > 0 {
-	}
+			pgStr += ` AND (receipt_cids.contract = ANY($2::VARCHAR(66)[])`
-	if len(trxIds) > 0 {
+			args = append(args, pq.Array(streamFilters.ReceiptFilter.Contracts))
-		pgStr += ` OR receipt_cids.tx_id = ANY($4::INTEGER[]))`
+			if len(trxIds) > 0 {
-		args = append(args, pq.Array(trxIds))
+				pgStr += ` OR receipt_cids.tx_id = ANY($3::INTEGER[]))`
-	} else {
+				args = append(args, pq.Array(trxIds))
-		pgStr += `)`
+			} else {
+				pgStr += `)`
+			}
+		} else if len(trxIds) > 0 {
+			pgStr += ` AND receipt_cids.tx_id = ANY($2::INTEGER[])`
+			args = append(args, pq.Array(trxIds))
+		}
 	}
 	receiptCids := make([]string, 0)
 	err := tx.Select(&receiptCids, pgStr, args...)

pkg/seed_node/retriever_test.go

@@ -0,0 +1,305 @@
+// VulcanizeDB
+// Copyright © 2019 Vulcanize
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+package seed_node_test
+
+import (
+	"math/big"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"github.com/vulcanize/vulcanizedb/pkg/config"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs"
+	"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
+	"github.com/vulcanize/vulcanizedb/pkg/seed_node"
+)
+
+var (
+	retriever seed_node.CIDRetriever
+)
+
+var _ = Describe("Retriever", func() {
+	BeforeEach(func() {
+		db, err = seed_node.SetupDB()
+		Expect(err).ToNot(HaveOccurred())
+		repo = seed_node.NewCIDRepository(db)
+		err = repo.Index(mocks.MockCIDPayload)
+		Expect(err).ToNot(HaveOccurred())
+		retriever = seed_node.NewCIDRetriever(db)
+	})
+	AfterEach(func() {
+		seed_node.TearDownDB(db)
+	})
+	Describe("RetrieveCIDs", func() {
+		It("Retrieves the CIDs specified by the provided filtering metadata", func() {
+			openFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter:  config.HeaderFilter{},
+				TrxFilter:     config.TrxFilter{},
+				ReceiptFilter: config.ReceiptFilter{},
+				StateFilter:   config.StateFilter{},
+				StorageFilter: config.StorageFilter{},
+			}
+			cidWrapper, err := retriever.RetrieveCIDs(openFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper.Headers)).To(Equal(1))
+			Expect(cidWrapper.Headers).To(Equal(mocks.MockCIDWrapper.Headers))
+			Expect(len(cidWrapper.Transactions)).To(Equal(2))
+			Expect(seed_node.ListContains(cidWrapper.Transactions, mocks.MockCIDWrapper.Transactions[0])).To(BeTrue())
+			Expect(seed_node.ListContains(cidWrapper.Transactions, mocks.MockCIDWrapper.Transactions[1])).To(BeTrue())
+			Expect(len(cidWrapper.Receipts)).To(Equal(2))
+			Expect(seed_node.ListContains(cidWrapper.Receipts, mocks.MockCIDWrapper.Receipts[0])).To(BeTrue())
+			Expect(seed_node.ListContains(cidWrapper.Receipts, mocks.MockCIDWrapper.Receipts[1])).To(BeTrue())
+			Expect(len(cidWrapper.StateNodes)).To(Equal(2))
+			for _, stateNode := range cidWrapper.StateNodes {
+				if stateNode.CID == "mockStateCID1" {
+					Expect(stateNode.Key).To(Equal(mocks.ContractLeafKey.Hex()))
+					Expect(stateNode.Leaf).To(Equal(true))
+				}
+				if stateNode.CID == "mockStateCID2" {
+					Expect(stateNode.Key).To(Equal(mocks.AnotherContractLeafKey.Hex()))
+					Expect(stateNode.Leaf).To(Equal(true))
+				}
+			}
+			Expect(len(cidWrapper.StorageNodes)).To(Equal(1))
+			Expect(cidWrapper.StorageNodes).To(Equal(mocks.MockCIDWrapper.StorageNodes))
+		})
+	})
+
+	Describe("RetrieveCIDs", func() {
+		It("Applies filters from the provided config.Subscription", func() {
+			rctContractFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Off: true,
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Contracts: []string{"0x0000000000000000000000000000000000000001"},
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper1, err := retriever.RetrieveCIDs(rctContractFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper1.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper1.Headers)).To(Equal(0))
+			Expect(len(cidWrapper1.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper1.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper1.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper1.Receipts)).To(Equal(1))
+			Expect(cidWrapper1.Receipts[0]).To(Equal("mockRctCID2"))
+
+			rctTopicsFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Off: true,
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Topic0s: []string{"0x0000000000000000000000000000000000000000000000000000000000000004"},
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper2, err := retriever.RetrieveCIDs(rctTopicsFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper2.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper2.Headers)).To(Equal(0))
+			Expect(len(cidWrapper2.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper2.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper2.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper2.Receipts)).To(Equal(1))
+			Expect(cidWrapper2.Receipts[0]).To(Equal("mockRctCID1"))
+
+			rctTopicsAndContractFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Off: true,
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000004", "0x0000000000000000000000000000000000000000000000000000000000000005"},
+					Contracts: []string{"0x0000000000000000000000000000000000000000"},
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper3, err := retriever.RetrieveCIDs(rctTopicsAndContractFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper3.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper3.Headers)).To(Equal(0))
+			Expect(len(cidWrapper3.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper3.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper3.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper3.Receipts)).To(Equal(1))
+			Expect(cidWrapper3.Receipts[0]).To(Equal("mockRctCID1"))
+
+			rctContractsAndTopicFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Off: true,
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000005"},
+					Contracts: []string{"0x0000000000000000000000000000000000000000", "0x0000000000000000000000000000000000000001"},
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper4, err := retriever.RetrieveCIDs(rctContractsAndTopicFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper4.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper4.Headers)).To(Equal(0))
+			Expect(len(cidWrapper4.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper4.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper4.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper4.Receipts)).To(Equal(1))
+			Expect(cidWrapper4.Receipts[0]).To(Equal("mockRctCID2"))
+
+			rctsForAllCollectedTrxs := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{}, // Trx filter open so we will collect all trxs, therefore we will also collect all corresponding rcts despite rct filter
+				ReceiptFilter: config.ReceiptFilter{
+					Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
+					Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper5, err := retriever.RetrieveCIDs(rctsForAllCollectedTrxs, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper5.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper5.Headers)).To(Equal(0))
+			Expect(len(cidWrapper5.Transactions)).To(Equal(2))
+			Expect(len(cidWrapper5.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper5.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper5.Receipts)).To(Equal(2))
+			Expect(seed_node.ListContains(cidWrapper5.Receipts, "mockRctCID1")).To(BeTrue())
+			Expect(seed_node.ListContains(cidWrapper5.Receipts, "mockRctCID2")).To(BeTrue())
+
+			rctsForSelectCollectedTrxs := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Dst: []string{"0x0000000000000000000000000000000000000001"}, // We only filter for one of the trxs so we will only get the one corresponding receipt
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Topic0s:   []string{"0x0000000000000000000000000000000000000000000000000000000000000006"}, // Topic isn't one of the topics we have
+					Contracts: []string{"0x0000000000000000000000000000000000000002"},                         // Contract isn't one of the contracts we have
+				},
+				StateFilter: config.StateFilter{
+					Off: true,
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper6, err := retriever.RetrieveCIDs(rctsForSelectCollectedTrxs, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper6.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper6.Headers)).To(Equal(0))
+			Expect(len(cidWrapper6.Transactions)).To(Equal(1))
+			Expect(len(cidWrapper6.StateNodes)).To(Equal(0))
+			Expect(len(cidWrapper6.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper6.Receipts)).To(Equal(1))
+			Expect(cidWrapper6.Receipts[0]).To(Equal("mockRctCID2"))
+
+			stateFilter := config.Subscription{
+				StartingBlock: big.NewInt(0),
+				HeaderFilter: config.HeaderFilter{
+					Off: true,
+				},
+				TrxFilter: config.TrxFilter{
+					Off: true,
+				},
+				ReceiptFilter: config.ReceiptFilter{
+					Off: true,
+				},
+				StateFilter: config.StateFilter{
+					Addresses: []string{mocks.Address.Hex()},
+				},
+				StorageFilter: config.StorageFilter{
+					Off: true,
+				},
+			}
+			cidWrapper7, err := retriever.RetrieveCIDs(stateFilter, 1)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(cidWrapper7.BlockNumber).To(Equal(mocks.MockCIDWrapper.BlockNumber))
+			Expect(len(cidWrapper7.Headers)).To(Equal(0))
+			Expect(len(cidWrapper7.Transactions)).To(Equal(0))
+			Expect(len(cidWrapper7.Receipts)).To(Equal(0))
+			Expect(len(cidWrapper7.StorageNodes)).To(Equal(0))
+			Expect(len(cidWrapper7.StateNodes)).To(Equal(1))
+			Expect(cidWrapper7.StateNodes[0]).To(Equal(ipfs.StateNodeCID{
+				Leaf: true,
+				Key:  mocks.ContractLeafKey.Hex(),
+				CID:  "mockStateCID1",
+			}))
+		})
+	})
+
+	Describe("RetrieveFirstBlockNumber", func() {
+		It("Gets the number of the first block that has data in the database", func() {
+			num, err := retriever.RetrieveFirstBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(num).To(Equal(int64(1)))
+		})
+	})
+	Describe("RetrieveLastBlockNumber", func() {
+		It("Gets the number of the latest block that has data in the database", func() {
+			num, err := retriever.RetrieveLastBlockNumber()
+			Expect(err).ToNot(HaveOccurred())
+			Expect(num).To(Equal(int64(1)))
+		})
+	})
+})

pkg/seed_node/service.go

@@ -208,7 +208,7 @@ func (sap *Service) publishAndIndex(id int, publishAndIndexPayload <-chan ipfs.I
 					log.Errorf("worker %d error: %v", id, err)
 				}
 			case <-publishAndIndexQuit:
-				log.Infof("quiting publishAndIndex worked %d", id)
+				log.Infof("quiting publishAndIndex worker %d", id)
 				return
 			}
 		}

pkg/seed_node/test_helpers.go

@@ -52,3 +52,12 @@ func TearDownDB(db *postgres.DB) {
 	err = tx.Commit()
 	Expect(err).NotTo(HaveOccurred())
 }
+
+func ListContains(sss []string, s string) bool {
+	for _, str := range sss {
+		if s == str {
+			return true
+		}
+	}
+	return false
+}