deep-stack/ipld-eth-server
6
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

Rename transactions to full_sync_transactions

Browse Source 
- Table has foreign key to blocks
- Add transaction RLP and transaction index to table
- Enables other tables with standalone transactions or transactions
  associated with other data structures (e.g. headers)
This commit is contained in:
Rob Mulholand 2019-03-19 14:11:26 -05:00
parent 48f6114791
commit d93006321b
30 changed files with 498 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
db/migrations/00002_create_transactions_table.sql → db/migrations/00002_create_full_sync_transactions_table.sql
View File

@ -1,16 +1,18 @@
-- +goose Up
CREATE TABLE transactions (
CREATE TABLE full_sync_transactions (
id SERIAL PRIMARY KEY,
block_id INTEGER NOT NULL REFERENCES blocks(id) ON DELETE CASCADE,
input_data VARCHAR,
tx_from VARCHAR(66),
gaslimit NUMERIC,
gasprice NUMERIC,
hash VARCHAR(66),
input_data VARCHAR,
nonce NUMERIC,
raw BYTEA,
tx_from VARCHAR(66),
tx_index INTEGER,
tx_to VARCHAR(66),
"value" NUMERIC
);
-- +goose Down
DROP TABLE transactions;
DROP TABLE full_sync_transactions;

5
db/migrations/00009_add_block_id_index_to_full_sync_transactions.sql Normal file
View File

@ -0,0 +1,5 @@
-- +goose Up
CREATE INDEX block_id_index ON full_sync_transactions (block_id);
-- +goose Down
DROP INDEX block_id_index;

5
db/migrations/00009_add_block_id_index_to_transactions.sql
View File

@ -1,5 +0,0 @@
-- +goose Up
CREATE INDEX block_id_index ON transactions (block_id);
-- +goose Down
DROP INDEX block_id_index;

5
db/migrations/00011_add_tx_to_index_to_full_sync_transactions.sql Normal file
View File

@ -0,0 +1,5 @@
-- +goose Up
CREATE INDEX tx_to_index ON full_sync_transactions(tx_to);
-- +goose Down
DROP INDEX tx_to_index;

5
db/migrations/00011_add_tx_to_index_to_transactions.sql
View File

@ -1,5 +0,0 @@
-- +goose Up
CREATE INDEX tx_to_index ON transactions(tx_to);
-- +goose Down
DROP INDEX tx_to_index;

5
db/migrations/00012_add_tx_from_index_to_full_sync_transactions.sql Normal file
View File

@ -0,0 +1,5 @@
-- +goose Up
CREATE INDEX tx_from_index ON full_sync_transactions(tx_from);
-- +goose Down
DROP INDEX tx_from_index;

5
db/migrations/00012_add_tx_from_index_to_transactions.sql
View File

@ -1,5 +0,0 @@
-- +goose Up
CREATE INDEX tx_from_index ON transactions(tx_from);
-- +goose Down
DROP INDEX tx_from_index;

7
db/migrations/00013_create_receipts_table.sql
View File

@ -2,16 +2,13 @@
CREATE TABLE receipts
(
id SERIAL PRIMARY KEY,
transaction_id INTEGER NOT NULL,
transaction_id INTEGER NOT NULL REFERENCES full_sync_transactions (id) ON DELETE CASCADE,
contract_address VARCHAR(42),
cumulative_gas_used NUMERIC,
gas_used NUMERIC,
state_root VARCHAR(66),
status INTEGER,
tx_hash VARCHAR(66),
CONSTRAINT transaction_fk FOREIGN KEY (transaction_id)
REFERENCES transactions (id)
ON DELETE CASCADE
tx_hash VARCHAR(66)
);

6
db/migrations/00020_associate_receipts_with_blocks.sql
View File

@ -4,7 +4,7 @@ ALTER TABLE receipts
UPDATE receipts
SET block_id = (
SELECT block_id FROM transactions WHERE transactions.id = receipts.transaction_id
SELECT block_id FROM full_sync_transactions WHERE full_sync_transactions.id = receipts.transaction_id
);
ALTER TABLE receipts
@ -28,7 +28,7 @@ CREATE INDEX transaction_id_index ON receipts (transaction_id);
UPDATE receipts
SET transaction_id = (
SELECT id FROM transactions WHERE transactions.hash = receipts.tx_hash
SELECT id FROM full_sync_transactions WHERE full_sync_transactions.hash = receipts.tx_hash
);
ALTER TABLE receipts
@ -37,7 +37,7 @@ ALTER TABLE receipts
ALTER TABLE receipts
ADD CONSTRAINT transaction_fk
FOREIGN KEY (transaction_id)
REFERENCES transactions (id)
REFERENCES full_sync_transactions (id)
ON DELETE CASCADE;
ALTER TABLE receipts

130
db/schema.sql
View File

@ -151,6 +151,46 @@ CREATE TABLE public.eth_nodes (
);
--
-- Name: full_sync_transactions; Type: TABLE; Schema: public; Owner: -
--
CREATE TABLE public.full_sync_transactions (
id integer NOT NULL,
block_id integer NOT NULL,
gaslimit numeric,
gasprice numeric,
hash character varying(66),
input_data character varying,
nonce numeric,
raw bytea,
tx_from character varying(66),
tx_index integer,
tx_to character varying(66),
value numeric
);
--
-- Name: full_sync_transactions_id_seq; Type: SEQUENCE; Schema: public; Owner: -
--
CREATE SEQUENCE public.full_sync_transactions_id_seq
AS integer
START WITH 1
INCREMENT BY 1
NO MINVALUE
NO MAXVALUE
CACHE 1;
--
-- Name: full_sync_transactions_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: -
--
ALTER SEQUENCE public.full_sync_transactions_id_seq OWNED BY public.full_sync_transactions.id;
--
-- Name: goose_db_version; Type: TABLE; Schema: public; Owner: -
--
@ -368,44 +408,6 @@ CREATE SEQUENCE public.receipts_id_seq
ALTER SEQUENCE public.receipts_id_seq OWNED BY public.receipts.id;
--
-- Name: transactions; Type: TABLE; Schema: public; Owner: -
--
CREATE TABLE public.transactions (
id integer NOT NULL,
block_id integer NOT NULL,
input_data character varying,
tx_from character varying(66),
gaslimit numeric,
gasprice numeric,
hash character varying(66),
nonce numeric,
tx_to character varying(66),
value numeric
);
--
-- Name: transactions_id_seq; Type: SEQUENCE; Schema: public; Owner: -
--
CREATE SEQUENCE public.transactions_id_seq
AS integer
START WITH 1
INCREMENT BY 1
NO MINVALUE
NO MAXVALUE
CACHE 1;
--
-- Name: transactions_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: -
--
ALTER SEQUENCE public.transactions_id_seq OWNED BY public.transactions.id;
--
-- Name: watched_contracts; Type: TABLE; Schema: public; Owner: -
--
@ -481,6 +483,13 @@ ALTER TABLE ONLY public.checked_headers ALTER COLUMN id SET DEFAULT nextval('pub
ALTER TABLE ONLY public.eth_nodes ALTER COLUMN id SET DEFAULT nextval('public.nodes_id_seq'::regclass);
--
-- Name: full_sync_transactions id; Type: DEFAULT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.full_sync_transactions ALTER COLUMN id SET DEFAULT nextval('public.full_sync_transactions_id_seq'::regclass);
--
-- Name: goose_db_version id; Type: DEFAULT; Schema: public; Owner: -
--
@ -523,13 +532,6 @@ ALTER TABLE ONLY public.queued_storage ALTER COLUMN id SET DEFAULT nextval('publ
ALTER TABLE ONLY public.receipts ALTER COLUMN id SET DEFAULT nextval('public.receipts_id_seq'::regclass);
--
-- Name: transactions id; Type: DEFAULT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.transactions ALTER COLUMN id SET DEFAULT nextval('public.transactions_id_seq'::regclass);
--
-- Name: watched_contracts contract_id; Type: DEFAULT; Schema: public; Owner: -
--
@ -577,6 +579,14 @@ ALTER TABLE ONLY public.eth_nodes
ADD CONSTRAINT eth_node_uc UNIQUE (genesis_block, network_id, eth_node_id);
--
-- Name: full_sync_transactions full_sync_transactions_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.full_sync_transactions
ADD CONSTRAINT full_sync_transactions_pkey PRIMARY KEY (id);
--
-- Name: goose_db_version goose_db_version_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
@ -633,14 +643,6 @@ ALTER TABLE ONLY public.receipts
ADD CONSTRAINT receipts_pkey PRIMARY KEY (id);
--
-- Name: transactions transactions_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.transactions
ADD CONSTRAINT transactions_pkey PRIMARY KEY (id);
--
-- Name: watched_contracts watched_contracts_contract_hash_key; Type: CONSTRAINT; Schema: public; Owner: -
--
@ -661,7 +663,7 @@ ALTER TABLE ONLY public.watched_contracts
-- Name: block_id_index; Type: INDEX; Schema: public; Owner: -
--
CREATE INDEX block_id_index ON public.transactions USING btree (block_id);
CREATE INDEX block_id_index ON public.full_sync_transactions USING btree (block_id);
--
@ -689,14 +691,14 @@ CREATE INDEX number_index ON public.blocks USING btree (number);
-- Name: tx_from_index; Type: INDEX; Schema: public; Owner: -
--
CREATE INDEX tx_from_index ON public.transactions USING btree (tx_from);
CREATE INDEX tx_from_index ON public.full_sync_transactions USING btree (tx_from);
--
-- Name: tx_to_index; Type: INDEX; Schema: public; Owner: -
--
CREATE INDEX tx_to_index ON public.transactions USING btree (tx_to);
CREATE INDEX tx_to_index ON public.full_sync_transactions USING btree (tx_to);
--
@ -723,6 +725,14 @@ ALTER TABLE ONLY public.headers
ADD CONSTRAINT eth_nodes_fk FOREIGN KEY (eth_node_id) REFERENCES public.eth_nodes(id) ON DELETE CASCADE;
--
-- Name: full_sync_transactions full_sync_transactions_block_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.full_sync_transactions
ADD CONSTRAINT full_sync_transactions_block_id_fkey FOREIGN KEY (block_id) REFERENCES public.blocks(id) ON DELETE CASCADE;
--
-- Name: blocks node_fk; Type: FK CONSTRAINT; Schema: public; Owner: -
--
@ -739,14 +749,6 @@ ALTER TABLE ONLY public.logs
ADD CONSTRAINT receipts_fk FOREIGN KEY (receipt_id) REFERENCES public.receipts(id) ON DELETE CASCADE;
--
-- Name: transactions transactions_block_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.transactions
ADD CONSTRAINT transactions_block_id_fkey FOREIGN KEY (block_id) REFERENCES public.blocks(id) ON DELETE CASCADE;
--
-- PostgreSQL database dump complete
--

24
integration_test/contract_watcher_full_transformer_test.go
View File

@ -40,8 +40,10 @@ var _ = Describe("contractWatcher full transformer", func() {
Describe("Init", func() {
It("Initializes transformer's contract objects", func() {
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
_, insertErr := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
Expect(insertErrTwo).NotTo(HaveOccurred())
t := transformer.NewTransformer(test_helpers.TusdConfig, blockChain, db)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
@ -63,8 +65,10 @@ var _ = Describe("contractWatcher full transformer", func() {
})
It("Does nothing if watched events are unset", func() {
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
_, insertErr := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
Expect(insertErrTwo).NotTo(HaveOccurred())
var testConf config.ContractConfig
testConf = test_helpers.TusdConfig
testConf.Events = nil
@ -80,8 +84,10 @@ var _ = Describe("contractWatcher full transformer", func() {
Describe("Execute", func() {
BeforeEach(func() {
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
_, insertErr := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(mocks.TransferBlock2)
Expect(insertErrTwo).NotTo(HaveOccurred())
})
It("Transforms watched contract data into custom repositories", func() {
@ -181,8 +187,10 @@ var _ = Describe("contractWatcher full transformer", func() {
Describe("Execute- against ENS registry contract", func() {
BeforeEach(func() {
blockRepository.CreateOrUpdateBlock(mocks.NewOwnerBlock1)
blockRepository.CreateOrUpdateBlock(mocks.NewOwnerBlock2)
_, insertErr := blockRepository.CreateOrUpdateBlock(mocks.NewOwnerBlock1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(mocks.NewOwnerBlock2)
Expect(insertErrTwo).NotTo(HaveOccurred())
})
It("Transforms watched contract data into custom repositories", func() {

12
integration_test/contract_watcher_light_transformer_test.go
View File

@ -38,8 +38,10 @@ var _ = Describe("contractWatcher light transformer", func() {
Describe("Init", func() {
It("Initializes transformer's contract objects", func() {
headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
_, insertErr := headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
Expect(insertErrTwo).NotTo(HaveOccurred())
t := transformer.NewTransformer(test_helpers.TusdConfig, blockChain, db)
err = t.Init()
Expect(err).ToNot(HaveOccurred())
@ -61,8 +63,10 @@ var _ = Describe("contractWatcher light transformer", func() {
})
It("Does nothing if nothing if no addresses are configured", func() {
headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
_, insertErr := headerRepository.CreateOrUpdateHeader(mocks.MockHeader1)
Expect(insertErr).NotTo(HaveOccurred())
_, insertErrTwo := headerRepository.CreateOrUpdateHeader(mocks.MockHeader3)
Expect(insertErrTwo).NotTo(HaveOccurred())
var testConf config.ContractConfig
testConf = test_helpers.TusdConfig
testConf.Addresses = nil

76
pkg/contract_watcher/full/retriever/block_retriever_test.go
View File

@ -17,6 +17,7 @@
package retriever_test
import (
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"strings"
@ -32,6 +33,7 @@ import (
var _ = Describe("Block Retriever", func() {
var db *postgres.DB
var r retriever.BlockRetriever
var rawTransaction []byte
var blockRepository repositories.BlockRepository
// Contains no contract address
@ -45,6 +47,10 @@ var _ = Describe("Block Retriever", func() {
db, _ = test_helpers.SetupDBandBC()
blockRepository = *repositories.NewBlockRepository(db)
r = retriever.NewBlockRetriever(db)
gethTransaction := types.Transaction{}
var err error
rawTransaction, err = gethTransaction.MarshalJSON()
Expect(err).NotTo(HaveOccurred())
})
AfterEach(func() {
@ -53,18 +59,23 @@ var _ = Describe("Block Retriever", func() {
Describe("RetrieveFirstBlock", func() {
It("Retrieves block number where contract first appears in receipt, if available", func() {
// Contains the address in the receipt
block2 := core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
Number: 2,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
GasPrice: 0,
GasLimit: 0,
Nonce: 0,
Raw: rawTransaction,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
ContractAddress: constants.TusdContractAddress,
Logs: []core.Log{},
},
TxIndex: 0,
Value: "0",
}},
}
@ -73,7 +84,11 @@ var _ = Describe("Block Retriever", func() {
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad456yui",
Number: 3,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
GasPrice: 0,
GasLimit: 0,
Nonce: 0,
Raw: rawTransaction,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
ContractAddress: constants.TusdContractAddress,
@ -91,12 +106,17 @@ var _ = Describe("Block Retriever", func() {
Data: "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
}},
},
TxIndex: 0,
Value: "0",
}},
}
blockRepository.CreateOrUpdateBlock(block1)
blockRepository.CreateOrUpdateBlock(block2)
blockRepository.CreateOrUpdateBlock(block3)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(block1)
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(block2)
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := blockRepository.CreateOrUpdateBlock(block3)
Expect(insertErrThree).NotTo(HaveOccurred())
i, err := r.RetrieveFirstBlock(strings.ToLower(constants.TusdContractAddress))
Expect(err).NotTo(HaveOccurred())
@ -104,12 +124,15 @@ var _ = Describe("Block Retriever", func() {
})
It("Retrieves block number where contract first appears in event logs if it cannot find the address in a receipt", func() {
block2 := core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
Number: 2,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
GasPrice: 0,
GasLimit: 0,
Nonce: 0,
Raw: rawTransaction,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad6546ae",
ContractAddress: "",
@ -127,6 +150,8 @@ var _ = Describe("Block Retriever", func() {
Data: "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
}},
},
TxIndex: 0,
Value: "0",
}},
}
@ -134,7 +159,11 @@ var _ = Describe("Block Retriever", func() {
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad456yui",
Number: 3,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
GasPrice: 0,
GasLimit: 0,
Nonce: 0,
Raw: rawTransaction,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad234hfs",
ContractAddress: "",
@ -152,12 +181,17 @@ var _ = Describe("Block Retriever", func() {
Data: "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
}},
},
TxIndex: 0,
Value: "0",
}},
}
blockRepository.CreateOrUpdateBlock(block1)
blockRepository.CreateOrUpdateBlock(block2)
blockRepository.CreateOrUpdateBlock(block3)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(block1)
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(block2)
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := blockRepository.CreateOrUpdateBlock(block3)
Expect(insertErrThree).NotTo(HaveOccurred())
i, err := r.RetrieveFirstBlock(constants.DaiContractAddress)
Expect(err).NotTo(HaveOccurred())
@ -177,9 +211,12 @@ var _ = Describe("Block Retriever", func() {
Transactions: []core.Transaction{},
}
blockRepository.CreateOrUpdateBlock(block1)
blockRepository.CreateOrUpdateBlock(block2)
blockRepository.CreateOrUpdateBlock(block3)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(block1)
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(block2)
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := blockRepository.CreateOrUpdateBlock(block3)
Expect(insertErrThree).NotTo(HaveOccurred())
_, err := r.RetrieveFirstBlock(constants.DaiContractAddress)
Expect(err).To(HaveOccurred())
@ -200,9 +237,12 @@ var _ = Describe("Block Retriever", func() {
Transactions: []core.Transaction{},
}
blockRepository.CreateOrUpdateBlock(block1)
blockRepository.CreateOrUpdateBlock(block2)
blockRepository.CreateOrUpdateBlock(block3)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(block1)
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(block2)
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := blockRepository.CreateOrUpdateBlock(block3)
Expect(insertErrThree).NotTo(HaveOccurred())
i, err := r.RetrieveMostRecentBlock()
Expect(err).ToNot(HaveOccurred())

2
pkg/contract_watcher/shared/helpers/test_helpers/database.go
View File

@ -237,7 +237,7 @@ func TearDown(db *postgres.DB) {
_, err = tx.Exec(`DELETE FROM log_filters`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM transactions`)
_, err = tx.Exec(`DELETE FROM full_sync_transactions`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM receipts`)

28
pkg/contract_watcher/shared/helpers/test_helpers/mocks/entities.go
View File

@ -34,7 +34,10 @@ var TransferBlock1 = core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ert",
Number: 6194633,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
GasLimit: 0,
GasPrice: 0,
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
Nonce: 0,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654aaa",
ContractAddress: "",
@ -52,6 +55,8 @@ var TransferBlock1 = core.Block{
Data: "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
}},
},
TxIndex: 0,
Value: "0",
}},
}
@ -59,7 +64,10 @@ var TransferBlock2 = core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ooo",
Number: 6194634,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
GasLimit: 0,
GasPrice: 0,
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
Nonce: 0,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654eee",
ContractAddress: "",
@ -77,6 +85,8 @@ var TransferBlock2 = core.Block{
Data: "0x000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc200000000000000000000000089d24a6b4ccb1b6faa2625fe562bdd9a23260359000000000000000000000000000000000000000000000000392d2e2bda9c00000000000000000000000000000000000000000000000000927f41fa0a4a418000000000000000000000000000000000000000000000000000000000005adcfebe",
}},
},
TxIndex: 0,
Value: "0",
}},
}
@ -84,7 +94,10 @@ var NewOwnerBlock1 = core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ppp",
Number: 6194635,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654bbb",
GasLimit: 0,
GasPrice: 0,
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654bbb",
Nonce: 0,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654bbb",
ContractAddress: "",
@ -102,6 +115,8 @@ var NewOwnerBlock1 = core.Block{
Data: "0x000000000000000000000000000000000000000000000000000000000000af21",
}},
},
TxIndex: 0,
Value: "0",
}},
}
@ -109,7 +124,10 @@ var NewOwnerBlock2 = core.Block{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad123ggg",
Number: 6194636,
Transactions: []core.Transaction{{
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654lll",
GasLimit: 0,
GasPrice: 0,
Hash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654lll",
Nonce: 0,
Receipt: core.Receipt{
TxHash: "0x135391a0962a63944e5908e6fedfff90fb4be3e3290a21017861099bad654lll",
ContractAddress: "",
@ -127,6 +145,8 @@ var NewOwnerBlock2 = core.Block{
Data: "0x000000000000000000000000000000000000000000000000000000000000af21",
}},
},
TxIndex: 0,
Value: "0",
}},
}

14
pkg/core/transaction.go
View File

@ -17,13 +17,15 @@
package core
type Transaction struct {
Hash string `db:"hash"`
Data string `db:"input_data"`
Nonce uint64 `db:"nonce"`
To string `db:"tx_to"`
From string `db:"tx_from"`
GasLimit uint64 `db:"gaslimit"`
GasPrice int64 `db:"gasprice"`
GasLimit uint64
GasPrice int64
Hash string
Nonce uint64
Raw []byte
Receipt
Value string `db:"value"`
To string `db:"tx_to"`
TxIndex int64 `db:"tx_index"`
Value string
}

38
pkg/datastore/postgres/errors.go Normal file
View File

@ -0,0 +1,38 @@
package postgres
import (
"errors"
"fmt"
)
const (
BeginTransactionFailedMsg = "failed to begin transaction"
DbConnectionFailedMsg = "db connection failed"
DeleteQueryFailedMsg = "delete query failed"
InsertQueryFailedMsg = "insert query failed"
SettingNodeFailedMsg = "unable to set db node"
)
func ErrBeginTransactionFailed(beginErr error) error {
return formatError(BeginTransactionFailedMsg, beginErr.Error())
}
func ErrDBConnectionFailed(connectErr error) error {
return formatError(DbConnectionFailedMsg, connectErr.Error())
}
func ErrDBDeleteFailed(deleteErr error) error {
return formatError(DeleteQueryFailedMsg, deleteErr.Error())
}
func ErrDBInsertFailed(insertErr error) error {
return formatError(InsertQueryFailedMsg, insertErr.Error())
}
func ErrUnableToSetNode(setErr error) error {
return formatError(SettingNodeFailedMsg, setErr.Error())
}
func formatError(msg, err string) error {
return errors.New(fmt.Sprintf("%s: %s", msg, err))
}

23
pkg/datastore/postgres/postgres.go
View File

@ -17,8 +17,6 @@
package postgres
import (
"errors"
"github.com/jmoiron/sqlx"
_ "github.com/lib/pq" //postgres driver
"github.com/vulcanize/vulcanizedb/pkg/config"
@ -31,23 +29,18 @@ type DB struct {
NodeID int64
}
var (
ErrDBInsertFailed = errors.New("postgres: insert failed")
ErrDBDeleteFailed = errors.New("postgres: delete failed")
ErrDBConnectionFailed = errors.New("postgres: db connection failed")
ErrUnableToSetNode = errors.New("postgres: unable to set node")
)
var ()
func NewDB(databaseConfig config.Database, node core.Node) (*DB, error) {
connectString := config.DbConnectionString(databaseConfig)
db, err := sqlx.Connect("postgres", connectString)
if err != nil {
return &DB{}, ErrDBConnectionFailed
db, connectErr := sqlx.Connect("postgres", connectString)
if connectErr != nil {
return &DB{}, ErrDBConnectionFailed(connectErr)
}
pg := DB{DB: db, Node: node}
err = pg.CreateNode(&node)
if err != nil {
return &DB{}, ErrUnableToSetNode
nodeErr := pg.CreateNode(&node)
if nodeErr != nil {
return &DB{}, ErrUnableToSetNode(nodeErr)
}
return &pg, nil
}
@ -66,7 +59,7 @@ func (db *DB) CreateNode(node *core.Node) error {
RETURNING id`,
node.GenesisBlock, node.NetworkID, node.ID, node.ClientName).Scan(&nodeId)
if err != nil {
return ErrUnableToSetNode
return ErrUnableToSetNode(err)
}
db.NodeID = nodeId
return nil

8
pkg/datastore/postgres/postgres_test.go
View File

@ -109,14 +109,18 @@ var _ = Describe("Postgres DB", func() {
_, err := postgres.NewDB(invalidDatabase, node)
Expect(err).To(Equal(postgres.ErrDBConnectionFailed))
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring(postgres.DbConnectionFailedMsg))
})
It("throws error when can't create node", func() {
badHash := fmt.Sprintf("x %s", strings.Repeat("1", 100))
node := core.Node{GenesisBlock: badHash, NetworkID: 1, ID: "x123", ClientName: "geth"}
_, err := postgres.NewDB(test_config.DBConfig, node)
Expect(err).To(Equal(postgres.ErrUnableToSetNode))
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring(postgres.SettingNodeFailedMsg))
})
It("does not commit log if log is invalid", func() {

79
pkg/datastore/postgres/repositories/block_repository.go
View File

@ -121,26 +121,42 @@ func (blockRepository BlockRepository) GetBlock(blockNumber int64) (core.Block,
func (blockRepository BlockRepository) insertBlock(block core.Block) (int64, error) {
var blockId int64
tx, _ := blockRepository.database.Beginx()
err := tx.QueryRow(
tx, beginErr := blockRepository.database.Beginx()
if beginErr != nil {
return 0, postgres.ErrBeginTransactionFailed(beginErr)
}
insertBlockErr := tx.QueryRow(
`INSERT INTO blocks
(eth_node_id, number, gaslimit, gasused, time, difficulty, hash, nonce, parenthash, size, uncle_hash, is_final, miner, extra_data, reward, uncles_reward, eth_node_fingerprint)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
RETURNING id `,
blockRepository.database.NodeID, block.Number, block.GasLimit, block.GasUsed, block.Time, block.Difficulty, block.Hash, block.Nonce, block.ParentHash, block.Size, block.UncleHash, block.IsFinal, block.Miner, block.ExtraData, block.Reward, block.UnclesReward, blockRepository.database.Node.ID).
Scan(&blockId)
if err != nil {
tx.Rollback()
return 0, err
if insertBlockErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
log.Error("failed to rollback transaction: ", rollbackErr)
}
return 0, postgres.ErrDBInsertFailed(insertBlockErr)
}
if len(block.Transactions) > 0 {
err = blockRepository.createTransactions(tx, blockId, block.Transactions)
if err != nil {
tx.Rollback()
return 0, postgres.ErrDBInsertFailed
insertTxErr := blockRepository.createTransactions(tx, blockId, block.Transactions)
if insertTxErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
log.Warn("failed to rollback transaction: ", rollbackErr)
}
return 0, postgres.ErrDBInsertFailed(insertTxErr)
}
}
tx.Commit()
commitErr := tx.Commit()
if commitErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
log.Warn("failed to rollback transaction: ", rollbackErr)
}
return 0, commitErr
}
return blockId, nil
}
@ -166,11 +182,11 @@ func nullStringToZero(s string) string {
func (blockRepository BlockRepository) createTransaction(tx *sqlx.Tx, blockId int64, transaction core.Transaction) error {
_, err := tx.Exec(
`INSERT INTO transactions
(block_id, hash, nonce, tx_to, tx_from, gaslimit, gasprice, value, input_data)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8::NUMERIC, $9)
RETURNING id`,
blockId, transaction.Hash, transaction.Nonce, transaction.To, transaction.From, transaction.GasLimit, transaction.GasPrice, nullStringToZero(transaction.Value), transaction.Data)
`INSERT INTO full_sync_transactions
(block_id, gaslimit, gasprice, hash, input_data, nonce, raw, tx_from, tx_index, tx_to, "value")
VALUES ($1, $2::NUMERIC, $3::NUMERIC, $4, $5, $6::NUMERIC, $7, $8, $9::NUMERIC, $10, $11::NUMERIC)
RETURNING id`, blockId, transaction.GasLimit, transaction.GasPrice, transaction.Hash, transaction.Data,
transaction.Nonce, transaction.Raw, transaction.From, transaction.TxIndex, transaction.To, transaction.Value)
if err != nil {
return err
}
@ -215,6 +231,7 @@ func (blockRepository BlockRepository) createReceipt(tx *sqlx.Tx, blockId int64,
func (blockRepository BlockRepository) getBlockHash(block core.Block) (string, bool) {
var retrievedBlockHash string
// TODO: handle possible error
blockRepository.database.Get(&retrievedBlockHash,
`SELECT hash
FROM blocks
@ -232,7 +249,7 @@ func (blockRepository BlockRepository) createLogs(tx *sqlx.Tx, logs []core.Log,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data, receiptId,
)
if err != nil {
return postgres.ErrDBInsertFailed
return postgres.ErrDBInsertFailed(err)
}
}
return nil
@ -244,12 +261,10 @@ func blockExists(retrievedBlockHash string) bool {
func (blockRepository BlockRepository) removeBlock(blockNumber int64) error {
_, err := blockRepository.database.Exec(
`DELETE FROM
blocks
WHERE number=$1 AND eth_node_id=$2`,
`DELETE FROM blocks WHERE number=$1 AND eth_node_id=$2`,
blockNumber, blockRepository.database.NodeID)
if err != nil {
return postgres.ErrDBDeleteFailed
return postgres.ErrDBDeleteFailed(err)
}
return nil
}
@ -266,17 +281,19 @@ func (blockRepository BlockRepository) loadBlock(blockRows *sqlx.Row) (core.Bloc
return core.Block{}, err
}
transactionRows, err := blockRepository.database.Queryx(`
SELECT hash,
nonce,
tx_to,
tx_from,
gaslimit,
gasprice,
value,
input_data
FROM transactions
WHERE block_id = $1
ORDER BY hash`, block.ID)
SELECT hash,
gaslimit,
gasprice,
input_data,
nonce,
raw,
tx_from,
tx_index,
tx_to,
value
FROM full_sync_transactions
WHERE block_id = $1
ORDER BY hash`, block.ID)
if err != nil {
log.Error("loadBlock: error fetting transactions: ", err)
return core.Block{}, err

172
pkg/datastore/postgres/repositories/block_repository_test.go
View File

@ -17,6 +17,10 @@
package repositories_test
import (
"bytes"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"math/big"
"strconv"
@ -51,7 +55,8 @@ var _ = Describe("Saving blocks", func() {
block := core.Block{
Number: 123,
}
blockRepository.CreateOrUpdateBlock(block)
_, insertErr := blockRepository.CreateOrUpdateBlock(block)
Expect(insertErr).NotTo(HaveOccurred())
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkID: 1,
@ -98,8 +103,9 @@ var _ = Describe("Saving blocks", func() {
UnclesReward: unclesReward,
}
blockRepository.CreateOrUpdateBlock(block)
_, insertErr := blockRepository.CreateOrUpdateBlock(block)
Expect(insertErr).NotTo(HaveOccurred())
savedBlock, err := blockRepository.GetBlock(blockNumber)
Expect(err).NotTo(HaveOccurred())
Expect(savedBlock.Reward).To(Equal(blockReward))
@ -127,42 +133,54 @@ var _ = Describe("Saving blocks", func() {
It("saves one transaction associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}},
Transactions: []core.Transaction{fakes.FakeTransaction},
}
blockRepository.CreateOrUpdateBlock(block)
_, insertErr := blockRepository.CreateOrUpdateBlock(block)
savedBlock, _ := blockRepository.GetBlock(123)
Expect(insertErr).NotTo(HaveOccurred())
savedBlock, getErr := blockRepository.GetBlock(123)
Expect(getErr).NotTo(HaveOccurred())
Expect(len(savedBlock.Transactions)).To(Equal(1))
})
It("saves two transactions associated to the block", func() {
block := core.Block{
Number: 123,
Transactions: []core.Transaction{{}, {}},
Transactions: []core.Transaction{fakes.FakeTransaction, fakes.FakeTransaction},
}
blockRepository.CreateOrUpdateBlock(block)
_, insertErr := blockRepository.CreateOrUpdateBlock(block)
savedBlock, _ := blockRepository.GetBlock(123)
Expect(insertErr).NotTo(HaveOccurred())
savedBlock, getErr := blockRepository.GetBlock(123)
Expect(getErr).NotTo(HaveOccurred())
Expect(len(savedBlock.Transactions)).To(Equal(2))
})
It(`replaces blocks and transactions associated to the block
when a more new block is in conflict (same block number + nodeid)`, func() {
blockOne := core.Block{
Number: 123,
Hash: "xabc",
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
Number: 123,
Hash: "xabc",
Transactions: []core.Transaction{
fakes.GetFakeTransaction("x123", core.Receipt{}),
fakes.GetFakeTransaction("x345", core.Receipt{}),
},
}
blockTwo := core.Block{
Number: 123,
Hash: "xdef",
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
Number: 123,
Hash: "xdef",
Transactions: []core.Transaction{
fakes.GetFakeTransaction("x678", core.Receipt{}),
fakes.GetFakeTransaction("x9ab", core.Receipt{}),
},
}
blockRepository.CreateOrUpdateBlock(blockOne)
blockRepository.CreateOrUpdateBlock(blockTwo)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(blockOne)
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(blockTwo)
Expect(insertErrTwo).NotTo(HaveOccurred())
savedBlock, _ := blockRepository.GetBlock(123)
Expect(len(savedBlock.Transactions)).To(Equal(2))
@ -173,14 +191,21 @@ var _ = Describe("Saving blocks", func() {
It(`does not replace blocks when block number is not unique
but block number + node id is`, func() {
blockOne := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x123"}, {Hash: "x345"}},
Number: 123,
Transactions: []core.Transaction{
fakes.GetFakeTransaction("x123", core.Receipt{}),
fakes.GetFakeTransaction("x345", core.Receipt{}),
},
}
blockTwo := core.Block{
Number: 123,
Transactions: []core.Transaction{{Hash: "x678"}, {Hash: "x9ab"}},
Number: 123,
Transactions: []core.Transaction{
fakes.GetFakeTransaction("x678", core.Receipt{}),
fakes.GetFakeTransaction("x9ab", core.Receipt{}),
},
}
blockRepository.CreateOrUpdateBlock(blockOne)
_, insertErrOne := blockRepository.CreateOrUpdateBlock(blockOne)
Expect(insertErrOne).NotTo(HaveOccurred())
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkID: 1,
@ -189,10 +214,14 @@ var _ = Describe("Saving blocks", func() {
test_config.CleanTestDB(dbTwo)
repositoryTwo := repositories.NewBlockRepository(dbTwo)
blockRepository.CreateOrUpdateBlock(blockOne)
repositoryTwo.CreateOrUpdateBlock(blockTwo)
retrievedBlockOne, _ := blockRepository.GetBlock(123)
retrievedBlockTwo, _ := repositoryTwo.GetBlock(123)
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(blockOne)
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := repositoryTwo.CreateOrUpdateBlock(blockTwo)
Expect(insertErrThree).NotTo(HaveOccurred())
retrievedBlockOne, getErrOne := blockRepository.GetBlock(123)
Expect(getErrOne).NotTo(HaveOccurred())
retrievedBlockTwo, getErrTwo := repositoryTwo.GetBlock(123)
Expect(getErrTwo).NotTo(HaveOccurred())
Expect(retrievedBlockOne.Transactions[0].Hash).To(Equal("x123"))
Expect(retrievedBlockTwo.Transactions[0].Hash).To(Equal("x678"))
@ -223,46 +252,51 @@ var _ = Describe("Saving blocks", func() {
var value = new(big.Int)
value.SetString("34940183920000000000", 10)
inputData := "0xf7d8c8830000000000000000000000000000000000000000000000000000000000037788000000000000000000000000000000000000000000000000000000000003bd14"
gethTransaction := types.NewTransaction(nonce, common.HexToAddress(to), value, gasLimit, big.NewInt(gasPrice), common.FromHex(inputData))
var raw bytes.Buffer
rlpErr := gethTransaction.EncodeRLP(&raw)
Expect(rlpErr).NotTo(HaveOccurred())
transaction := core.Transaction{
Hash: "x1234",
GasPrice: gasPrice,
GasLimit: gasLimit,
Nonce: nonce,
To: to,
From: from,
Value: value.String(),
Data: inputData,
From: from,
GasLimit: gasLimit,
GasPrice: gasPrice,
Hash: "x1234",
Nonce: nonce,
Raw: raw.Bytes(),
Receipt: core.Receipt{},
To: to,
TxIndex: 2,
Value: value.String(),
}
block := core.Block{
Number: 123,
Transactions: []core.Transaction{transaction},
}
blockRepository.CreateOrUpdateBlock(block)
_, insertErr := blockRepository.CreateOrUpdateBlock(block)
Expect(insertErr).NotTo(HaveOccurred())
savedBlock, _ := blockRepository.GetBlock(123)
savedBlock, err := blockRepository.GetBlock(123)
Expect(err).NotTo(HaveOccurred())
Expect(len(savedBlock.Transactions)).To(Equal(1))
savedTransaction := savedBlock.Transactions[0]
Expect(savedTransaction.Data).To(Equal(transaction.Data))
Expect(savedTransaction.Hash).To(Equal(transaction.Hash))
Expect(savedTransaction.To).To(Equal(to))
Expect(savedTransaction.From).To(Equal(from))
Expect(savedTransaction.Nonce).To(Equal(nonce))
Expect(savedTransaction.GasLimit).To(Equal(gasLimit))
Expect(savedTransaction.GasPrice).To(Equal(gasPrice))
Expect(savedTransaction.Value).To(Equal(value.String()))
Expect(savedTransaction).To(Equal(transaction))
})
Describe("The missing block numbers", func() {
It("is empty the starting block number is the highest known block number", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 1})
_, insertErr := blockRepository.CreateOrUpdateBlock(core.Block{Number: 1})
Expect(insertErr).NotTo(HaveOccurred())
Expect(len(blockRepository.MissingBlockNumbers(1, 1, node.ID))).To(Equal(0))
})
It("is empty if copies of block exist from both current node and another", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 0})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 1})
_, insertErrOne := blockRepository.CreateOrUpdateBlock(core.Block{Number: 0})
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(core.Block{Number: 1})
Expect(insertErrTwo).NotTo(HaveOccurred())
nodeTwo := core.Node{
GenesisBlock: "0x456",
NetworkID: 1,
@ -270,7 +304,8 @@ var _ = Describe("Saving blocks", func() {
dbTwo, err := postgres.NewDB(test_config.DBConfig, nodeTwo)
Expect(err).NotTo(HaveOccurred())
repositoryTwo := repositories.NewBlockRepository(dbTwo)
repositoryTwo.CreateOrUpdateBlock(core.Block{Number: 0})
_, insertErrThree := repositoryTwo.CreateOrUpdateBlock(core.Block{Number: 0})
Expect(insertErrThree).NotTo(HaveOccurred())
missing := blockRepository.MissingBlockNumbers(0, 1, node.ID)
@ -278,42 +313,53 @@ var _ = Describe("Saving blocks", func() {
})
It("is the only missing block number", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 2})
_, insertErr := blockRepository.CreateOrUpdateBlock(core.Block{Number: 2})
Expect(insertErr).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(1, 2, node.ID)).To(Equal([]int64{1}))
})
It("is both missing block numbers", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
_, insertErr := blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
Expect(insertErr).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(1, 3, node.ID)).To(Equal([]int64{1, 2}))
})
It("goes back to the starting block number", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
_, insertErr := blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(insertErr).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(4, 6, node.ID)).To(Equal([]int64{4, 5}))
})
It("only includes missing block numbers", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
_, insertErrOne := blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(insertErrTwo).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(4, 6, node.ID)).To(Equal([]int64{5}))
})
It("includes blocks created by a different node", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
_, insertErrOne := blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(core.Block{Number: 6})
Expect(insertErrTwo).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(4, 6, "Different node id")).To(Equal([]int64{4, 5, 6}))
})
It("is a list with multiple gaps", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 5})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 10})
_, insertErrOne := blockRepository.CreateOrUpdateBlock(core.Block{Number: 4})
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(core.Block{Number: 5})
Expect(insertErrTwo).NotTo(HaveOccurred())
_, insertErrThree := blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
Expect(insertErrThree).NotTo(HaveOccurred())
_, insertErrFour := blockRepository.CreateOrUpdateBlock(core.Block{Number: 10})
Expect(insertErrFour).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(3, 10, node.ID)).To(Equal([]int64{3, 6, 7, 9}))
})
@ -323,8 +369,10 @@ var _ = Describe("Saving blocks", func() {
})
It("only returns requested range even when other gaps exist", func() {
blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
_, insertErrOne := blockRepository.CreateOrUpdateBlock(core.Block{Number: 3})
Expect(insertErrOne).NotTo(HaveOccurred())
_, insertErrTwo := blockRepository.CreateOrUpdateBlock(core.Block{Number: 8})
Expect(insertErrTwo).NotTo(HaveOccurred())
Expect(blockRepository.MissingBlockNumbers(1, 5, node.ID)).To(Equal([]int64{1, 2, 4, 5}))
})
@ -334,11 +382,13 @@ var _ = Describe("Saving blocks", func() {
It("sets the status of blocks within n-20 of chain HEAD as final", func() {
blockNumberOfChainHead := 25
for i := 0; i < blockNumberOfChainHead; i++ {
blockRepository.CreateOrUpdateBlock(core.Block{Number: int64(i), Hash: strconv.Itoa(i)})
_, err := blockRepository.CreateOrUpdateBlock(core.Block{Number: int64(i), Hash: strconv.Itoa(i)})
Expect(err).NotTo(HaveOccurred())
}
blockRepository.SetBlocksStatus(int64(blockNumberOfChainHead))
setErr := blockRepository.SetBlocksStatus(int64(blockNumberOfChainHead))
Expect(setErr).NotTo(HaveOccurred())
blockOne, err := blockRepository.GetBlock(1)
Expect(err).ToNot(HaveOccurred())
Expect(blockOne.IsFinal).To(Equal(true))

4
pkg/datastore/postgres/repositories/contract_repository.go
View File

@ -42,7 +42,7 @@ func (contractRepository ContractRepository) CreateContract(contract core.Contra
SET contract_hash = $1, contract_abi = $2
`, contract.Hash, abiToInsert)
if err != nil {
return postgres.ErrDBInsertFailed
return postgres.ErrDBInsertFailed(err)
}
return nil
}
@ -86,7 +86,7 @@ func (contractRepository ContractRepository) addTransactions(contract core.Contr
gasprice,
value,
input_data
FROM transactions
FROM full_sync_transactions
WHERE tx_to = $1
ORDER BY block_id DESC`, contract.Hash)
if err != nil {

14
pkg/datastore/postgres/repositories/logs_repository.go
View File

@ -32,18 +32,18 @@ type LogRepository struct {
func (logRepository LogRepository) CreateLogs(lgs []core.Log, receiptId int64) error {
tx, _ := logRepository.DB.Beginx()
for _, tlog := range lgs {
_, err := tx.Exec(
_, insertLogErr := tx.Exec(
`INSERT INTO logs (block_number, address, tx_hash, index, topic0, topic1, topic2, topic3, data, receipt_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
`,
tlog.BlockNumber, tlog.Address, tlog.TxHash, tlog.Index, tlog.Topics[0], tlog.Topics[1], tlog.Topics[2], tlog.Topics[3], tlog.Data, receiptId,
)
if err != nil {
err = tx.Rollback()
if err != nil {
logrus.Error("CreateLogs: could not perform rollback: ", err)
if insertLogErr != nil {
rollbackErr := tx.Rollback()
if rollbackErr != nil {
logrus.Error("CreateLogs: could not perform rollback: ", rollbackErr)
}
return postgres.ErrDBInsertFailed
return postgres.ErrDBInsertFailed(insertLogErr)
}
}
err := tx.Commit()
@ -52,7 +52,7 @@ func (logRepository LogRepository) CreateLogs(lgs []core.Log, receiptId int64) e
if err != nil {
logrus.Error("CreateLogs: could not perform rollback: ", err)
}
return postgres.ErrDBInsertFailed
return postgres.ErrDBInsertFailed(err)
}
return nil
}

7
pkg/datastore/postgres/repositories/logs_repository_test.go
View File

@ -17,6 +17,7 @@
package repositories_test
import (
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"sort"
. "github.com/onsi/ginkgo"
@ -201,11 +202,7 @@ var _ = Describe("Logs Repository", func() {
Status: 1,
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction :=
core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
transaction := fakes.GetFakeTransaction(receipt.TxHash, receipt)
block := core.Block{Transactions: []core.Transaction{transaction}}
_, err := blockRepository.CreateOrUpdateBlock(block)

11
pkg/datastore/postgres/repositories/receipts_repository_test.go
View File

@ -23,6 +23,7 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/datastore"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/test_config"
)
@ -117,10 +118,7 @@ var _ = Describe("Receipts Repository", func() {
TxHash: "0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223",
}
transaction := core.Transaction{
Hash: expected.TxHash,
Receipt: expected,
}
transaction := fakes.GetFakeTransaction(expected.TxHash, expected)
block := core.Block{Transactions: []core.Transaction{transaction}}
_, err := blockRepository.CreateOrUpdateBlock(block)
@ -147,10 +145,7 @@ var _ = Describe("Receipts Repository", func() {
receipt := core.Receipt{
TxHash: "0x002c4799161d809b23f67884eb6598c9df5894929fe1a9ead97ca175d360f547",
}
transaction := core.Transaction{
Hash: receipt.TxHash,
Receipt: receipt,
}
transaction := fakes.GetFakeTransaction(receipt.TxHash, receipt)
block := core.Block{
Transactions: []core.Transaction{transaction},

40
pkg/fakes/data.go
View File

@ -17,6 +17,7 @@
package fakes
import (
"bytes"
"encoding/json"
"errors"
"strconv"
@ -48,3 +49,42 @@ func GetFakeHeader(blockNumber int64) core.Header {
Timestamp: strconv.FormatInt(fakeTimestamp, 10),
}
}
var fakeTransaction types.Transaction
var rawTransaction bytes.Buffer
var _ = fakeTransaction.EncodeRLP(&rawTransaction)
var FakeTransaction = core.Transaction{
Data: "",
From: "",
GasLimit: 0,
GasPrice: 0,
Hash: "",
Nonce: 0,
Raw: rawTransaction.Bytes(),
Receipt: core.Receipt{},
To: "",
TxIndex: 0,
Value: "0",
}
func GetFakeTransaction(hash string, receipt core.Receipt) core.Transaction {
gethTransaction := types.Transaction{}
var raw bytes.Buffer
err := gethTransaction.EncodeRLP(&raw)
if err != nil {
panic("failed to marshal transaction creating test fake")
}
return core.Transaction{
Data: "",
From: "",
GasLimit: 0,
GasPrice: 0,
Hash: hash,
Nonce: 0,
Raw: raw.Bytes(),
Receipt: receipt,
To: "",
TxIndex: 0,
Value: "0",
}
}

6
pkg/geth/converters/common/block_converter_test.go
View File

@ -17,6 +17,7 @@
package common_test
import (
"bytes"
"io/ioutil"
"log"
"math/big"
@ -227,6 +228,9 @@ var _ = Describe("Conversion of GethBlock to core.Block", func() {
big.NewInt(3),
hexutil.MustDecode("0xf7d8c8830000000000000000000000000000000000000000000000000000000000037788000000000000000000000000000000000000000000000000000000000003bd14"),
)
var rawTransaction bytes.Buffer
encodeErr := gethTransaction.EncodeRLP(&rawTransaction)
Expect(encodeErr).NotTo(HaveOccurred())
gethReceipt := &types.Receipt{
Bloom: types.BytesToBloom(hexutil.MustDecode("0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000")),
@ -261,6 +265,8 @@ var _ = Describe("Conversion of GethBlock to core.Block", func() {
Expect(coreTransaction.From).To(Equal("0x0000000000000000000000000000000000000123"))
Expect(coreTransaction.GasLimit).To(Equal(gethTransaction.Gas()))
Expect(coreTransaction.GasPrice).To(Equal(gethTransaction.GasPrice().Int64()))
Expect(coreTransaction.Raw).To(Equal(rawTransaction.Bytes()))
Expect(coreTransaction.TxIndex).To(Equal(int64(0)))
Expect(coreTransaction.Value).To(Equal(gethTransaction.Value().String()))
Expect(coreTransaction.Nonce).To(Equal(gethTransaction.Nonce()))

25
pkg/geth/converters/rpc/transaction_converter.go
View File

@ -17,6 +17,7 @@
package rpc
import (
"bytes"
"context"
"log"
"strings"
@ -52,7 +53,10 @@ func (rtc *RpcTransactionConverter) ConvertTransactionsToCore(gethBlock *types.B
log.Println("transaction sender: ", err)
return err
}
coreTransaction := transToCoreTrans(transaction, &from)
coreTransaction, convertErr := transToCoreTrans(transaction, &from, int64(gethTransactionIndex))
if convertErr != nil {
return convertErr
}
coreTransaction, err = rtc.appendReceiptToTransaction(coreTransaction)
if err != nil {
log.Println("receipt: ", err)
@ -79,18 +83,25 @@ func (rtc *RpcTransactionConverter) appendReceiptToTransaction(transaction core.
return transaction, nil
}
func transToCoreTrans(transaction *types.Transaction, from *common.Address) core.Transaction {
func transToCoreTrans(transaction *types.Transaction, from *common.Address, transactionIndex int64) (core.Transaction, error) {
data := hexutil.Encode(transaction.Data())
var raw bytes.Buffer
encodeErr := transaction.EncodeRLP(&raw)
if encodeErr != nil {
return core.Transaction{}, encodeErr
}
return core.Transaction{
Hash: transaction.Hash().Hex(),
Nonce: transaction.Nonce(),
To: strings.ToLower(addressToHex(transaction.To())),
Data: data,
From: strings.ToLower(addressToHex(from)),
GasLimit: transaction.Gas(),
GasPrice: transaction.GasPrice().Int64(),
Hash: transaction.Hash().Hex(),
Nonce: transaction.Nonce(),
Raw: raw.Bytes(),
To: strings.ToLower(addressToHex(transaction.To())),
TxIndex: transactionIndex,
Value: transaction.Value().String(),
Data: data,
}
}, nil
}
func addressToHex(to *common.Address) string {

2
pkg/plugin/test_helpers/database.go
View File

@ -66,7 +66,7 @@ func TearDown(db *postgres.DB) {
_, err = tx.Exec(`DELETE FROM log_filters`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM transactions`)
_, err = tx.Exec(`DELETE FROM full_sync_transactions`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM receipts`)

2
test_config/test_config.go
View File

@ -108,13 +108,13 @@ func CleanTestDB(db *postgres.DB) {
db.MustExec("DELETE FROM blocks")
db.MustExec("DELETE FROM checked_headers")
// can't delete from eth_nodes since this function is called after the required eth_node is persisted
db.MustExec("DELETE FROM full_sync_transactions")
db.MustExec("DELETE FROM goose_db_version")
db.MustExec("DELETE FROM headers")
db.MustExec("DELETE FROM log_filters")
db.MustExec("DELETE FROM logs")
db.MustExec("DELETE FROM queued_storage")
db.MustExec("DELETE FROM receipts")
db.MustExec("DELETE FROM transactions")
db.MustExec("DELETE FROM watched_contracts")
}