Merge pull request #52 from vulcanize/account_transformer_staging

Account transformer staging
This commit is contained in:
Ian Norden 2019-04-08 10:35:38 -05:00 committed by GitHub
commit 6805d3ea29
30 changed files with 535 additions and 215 deletions

View File

@ -1,5 +1,5 @@
-- +goose Up -- +goose Up
CREATE TABLE receipts CREATE TABLE full_sync_receipts
( (
id SERIAL PRIMARY KEY, id SERIAL PRIMARY KEY,
transaction_id INTEGER NOT NULL REFERENCES full_sync_transactions (id) ON DELETE CASCADE, transaction_id INTEGER NOT NULL REFERENCES full_sync_transactions (id) ON DELETE CASCADE,
@ -13,4 +13,4 @@ CREATE TABLE receipts
-- +goose Down -- +goose Down
DROP TABLE receipts; DROP TABLE full_sync_receipts;

View File

@ -1,5 +1,5 @@
-- +goose Up -- +goose Up
CREATE INDEX transaction_id_index ON receipts (transaction_id); CREATE INDEX transaction_id_index ON full_sync_receipts (transaction_id);
-- +goose Down -- +goose Down
DROP INDEX transaction_id_index; DROP INDEX transaction_id_index;

View File

@ -8,7 +8,7 @@ ALTER TABLE logs
ALTER TABLE logs ALTER TABLE logs
ADD CONSTRAINT receipts_fk ADD CONSTRAINT receipts_fk
FOREIGN KEY (receipt_id) FOREIGN KEY (receipt_id)
REFERENCES receipts (id) REFERENCES full_sync_receipts (id)
ON DELETE CASCADE; ON DELETE CASCADE;

View File

@ -1,44 +1,44 @@
-- +goose Up -- +goose Up
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ADD COLUMN block_id INT; ADD COLUMN block_id INT;
UPDATE receipts UPDATE full_sync_receipts
SET block_id = ( SET block_id = (
SELECT block_id FROM full_sync_transactions WHERE full_sync_transactions.id = receipts.transaction_id SELECT block_id FROM full_sync_transactions WHERE full_sync_transactions.id = full_sync_receipts.transaction_id
); );
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ALTER COLUMN block_id SET NOT NULL; ALTER COLUMN block_id SET NOT NULL;
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ADD CONSTRAINT blocks_fk ADD CONSTRAINT blocks_fk
FOREIGN KEY (block_id) FOREIGN KEY (block_id)
REFERENCES blocks (id) REFERENCES blocks (id)
ON DELETE CASCADE; ON DELETE CASCADE;
ALTER TABLE receipts ALTER TABLE full_sync_receipts
DROP COLUMN transaction_id; DROP COLUMN transaction_id;
-- +goose Down -- +goose Down
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ADD COLUMN transaction_id INT; ADD COLUMN transaction_id INT;
CREATE INDEX transaction_id_index ON receipts (transaction_id); CREATE INDEX transaction_id_index ON full_sync_receipts (transaction_id);
UPDATE receipts UPDATE full_sync_receipts
SET transaction_id = ( SET transaction_id = (
SELECT id FROM full_sync_transactions WHERE full_sync_transactions.hash = receipts.tx_hash SELECT id FROM full_sync_transactions WHERE full_sync_transactions.hash = full_sync_receipts.tx_hash
); );
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ALTER COLUMN transaction_id SET NOT NULL; ALTER COLUMN transaction_id SET NOT NULL;
ALTER TABLE receipts ALTER TABLE full_sync_receipts
ADD CONSTRAINT transaction_fk ADD CONSTRAINT transaction_fk
FOREIGN KEY (transaction_id) FOREIGN KEY (transaction_id)
REFERENCES full_sync_transactions (id) REFERENCES full_sync_transactions (id)
ON DELETE CASCADE; ON DELETE CASCADE;
ALTER TABLE receipts ALTER TABLE full_sync_receipts
DROP COLUMN block_id; DROP COLUMN block_id;

View File

@ -2,15 +2,15 @@
CREATE TABLE light_sync_transactions ( CREATE TABLE light_sync_transactions (
id SERIAL PRIMARY KEY, id SERIAL PRIMARY KEY,
header_id INTEGER NOT NULL REFERENCES headers(id) ON DELETE CASCADE, header_id INTEGER NOT NULL REFERENCES headers(id) ON DELETE CASCADE,
hash TEXT, hash VARCHAR(66),
gas_limit NUMERIC, gas_limit NUMERIC,
gas_price NUMERIC, gas_price NUMERIC,
input_data BYTEA, input_data BYTEA,
nonce NUMERIC, nonce NUMERIC,
raw BYTEA, raw BYTEA,
tx_from TEXT, tx_from VARCHAR(44),
tx_index INTEGER, tx_index INTEGER,
tx_to TEXT, tx_to VARCHAR(44),
"value" NUMERIC, "value" NUMERIC,
UNIQUE (header_id, hash) UNIQUE (header_id, hash)
); );

View File

@ -0,0 +1,18 @@
-- +goose Up
CREATE TABLE light_sync_receipts(
id SERIAL PRIMARY KEY,
transaction_id INTEGER NOT NULL REFERENCES light_sync_transactions(id) ON DELETE CASCADE,
header_id INTEGER NOT NULL REFERENCES headers(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),
rlp BYTEA,
UNIQUE(header_id, transaction_id)
);
-- +goose Down
DROP TABLE light_sync_receipts;

View File

@ -2,8 +2,8 @@
-- PostgreSQL database dump -- PostgreSQL database dump
-- --
-- Dumped from database version 10.6 -- Dumped from database version 10.5
-- Dumped by pg_dump version 10.6 -- Dumped by pg_dump version 10.4
SET statement_timeout = 0; SET statement_timeout = 0;
SET lock_timeout = 0; SET lock_timeout = 0;
@ -151,6 +151,42 @@ CREATE TABLE public.eth_nodes (
); );
--
-- Name: full_sync_receipts; Type: TABLE; Schema: public; Owner: -
--
CREATE TABLE public.full_sync_receipts (
id integer NOT NULL,
contract_address character varying(42),
cumulative_gas_used numeric,
gas_used numeric,
state_root character varying(66),
status integer,
tx_hash character varying(66),
block_id integer NOT NULL
);
--
-- Name: full_sync_receipts_id_seq; Type: SEQUENCE; Schema: public; Owner: -
--
CREATE SEQUENCE public.full_sync_receipts_id_seq
AS integer
START WITH 1
INCREMENT BY 1
NO MINVALUE
NO MAXVALUE
CACHE 1;
--
-- Name: full_sync_receipts_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: -
--
ALTER SEQUENCE public.full_sync_receipts_id_seq OWNED BY public.full_sync_receipts.id;
-- --
-- Name: full_sync_transactions; Type: TABLE; Schema: public; Owner: - -- Name: full_sync_transactions; Type: TABLE; Schema: public; Owner: -
-- --
@ -258,6 +294,44 @@ CREATE SEQUENCE public.headers_id_seq
ALTER SEQUENCE public.headers_id_seq OWNED BY public.headers.id; ALTER SEQUENCE public.headers_id_seq OWNED BY public.headers.id;
--
-- Name: light_sync_receipts; Type: TABLE; Schema: public; Owner: -
--
CREATE TABLE public.light_sync_receipts (
id integer NOT NULL,
transaction_id integer NOT NULL,
header_id integer NOT NULL,
contract_address character varying(42),
cumulative_gas_used numeric,
gas_used numeric,
state_root character varying(66),
status integer,
tx_hash character varying(66),
rlp bytea
);
--
-- Name: light_sync_receipts_id_seq; Type: SEQUENCE; Schema: public; Owner: -
--
CREATE SEQUENCE public.light_sync_receipts_id_seq
AS integer
START WITH 1
INCREMENT BY 1
NO MINVALUE
NO MAXVALUE
CACHE 1;
--
-- Name: light_sync_receipts_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: -
--
ALTER SEQUENCE public.light_sync_receipts_id_seq OWNED BY public.light_sync_receipts.id;
-- --
-- Name: light_sync_transactions; Type: TABLE; Schema: public; Owner: - -- Name: light_sync_transactions; Type: TABLE; Schema: public; Owner: -
-- --
@ -265,15 +339,15 @@ ALTER SEQUENCE public.headers_id_seq OWNED BY public.headers.id;
CREATE TABLE public.light_sync_transactions ( CREATE TABLE public.light_sync_transactions (
id integer NOT NULL, id integer NOT NULL,
header_id integer NOT NULL, header_id integer NOT NULL,
hash text, hash character varying(66),
gas_limit numeric, gas_limit numeric,
gas_price numeric, gas_price numeric,
input_data bytea, input_data bytea,
nonce numeric, nonce numeric,
raw bytea, raw bytea,
tx_from text, tx_from character varying(44),
tx_index integer, tx_index integer,
tx_to text, tx_to character varying(44),
value numeric value numeric
); );
@ -413,26 +487,27 @@ ALTER SEQUENCE public.queued_storage_id_seq OWNED BY public.queued_storage.id;
-- --
-- Name: receipts; Type: TABLE; Schema: public; Owner: - -- Name: uncles; Type: TABLE; Schema: public; Owner: -
-- --
CREATE TABLE public.receipts ( CREATE TABLE public.uncles (
id integer NOT NULL, id integer NOT NULL,
contract_address character varying(42), hash character varying(66) NOT NULL,
cumulative_gas_used numeric, block_id integer NOT NULL,
gas_used numeric, reward numeric NOT NULL,
state_root character varying(66), miner character varying(42) NOT NULL,
status integer, raw jsonb,
tx_hash character varying(66), block_timestamp numeric,
block_id integer NOT NULL eth_node_id integer NOT NULL,
eth_node_fingerprint character varying(128)
); );
-- --
-- Name: receipts_id_seq; Type: SEQUENCE; Schema: public; Owner: - -- Name: uncles_id_seq; Type: SEQUENCE; Schema: public; Owner: -
-- --
CREATE SEQUENCE public.receipts_id_seq CREATE SEQUENCE public.uncles_id_seq
AS integer AS integer
START WITH 1 START WITH 1
INCREMENT BY 1 INCREMENT BY 1
@ -442,10 +517,10 @@ CREATE SEQUENCE public.receipts_id_seq
-- --
-- Name: receipts_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: - -- Name: uncles_id_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: -
-- --
ALTER SEQUENCE public.receipts_id_seq OWNED BY public.receipts.id; ALTER SEQUENCE public.uncles_id_seq OWNED BY public.uncles.id;
-- --
@ -560,6 +635,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); ALTER TABLE ONLY public.eth_nodes ALTER COLUMN id SET DEFAULT nextval('public.nodes_id_seq'::regclass);
--
-- Name: full_sync_receipts id; Type: DEFAULT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.full_sync_receipts ALTER COLUMN id SET DEFAULT nextval('public.full_sync_receipts_id_seq'::regclass);
-- --
-- Name: full_sync_transactions id; Type: DEFAULT; Schema: public; Owner: - -- Name: full_sync_transactions id; Type: DEFAULT; Schema: public; Owner: -
-- --
@ -581,6 +663,13 @@ ALTER TABLE ONLY public.goose_db_version ALTER COLUMN id SET DEFAULT nextval('pu
ALTER TABLE ONLY public.headers ALTER COLUMN id SET DEFAULT nextval('public.headers_id_seq'::regclass); ALTER TABLE ONLY public.headers ALTER COLUMN id SET DEFAULT nextval('public.headers_id_seq'::regclass);
--
-- Name: light_sync_receipts id; Type: DEFAULT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.light_sync_receipts ALTER COLUMN id SET DEFAULT nextval('public.light_sync_receipts_id_seq'::regclass);
-- --
-- Name: light_sync_transactions id; Type: DEFAULT; Schema: public; Owner: - -- Name: light_sync_transactions id; Type: DEFAULT; Schema: public; Owner: -
-- --
@ -610,10 +699,10 @@ ALTER TABLE ONLY public.queued_storage ALTER COLUMN id SET DEFAULT nextval('publ
-- --
-- Name: receipts id; Type: DEFAULT; Schema: public; Owner: - -- Name: uncles id; Type: DEFAULT; Schema: public; Owner: -
-- --
ALTER TABLE ONLY public.receipts ALTER COLUMN id SET DEFAULT nextval('public.receipts_id_seq'::regclass); ALTER TABLE ONLY public.uncles ALTER COLUMN id SET DEFAULT nextval('public.uncles_id_seq'::regclass);
-- --
@ -670,6 +759,14 @@ ALTER TABLE ONLY public.eth_nodes
ADD CONSTRAINT eth_node_uc UNIQUE (genesis_block, network_id, eth_node_id); ADD CONSTRAINT eth_node_uc UNIQUE (genesis_block, network_id, eth_node_id);
--
-- Name: full_sync_receipts full_sync_receipts_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.full_sync_receipts
ADD CONSTRAINT full_sync_receipts_pkey PRIMARY KEY (id);
-- --
-- Name: full_sync_transactions full_sync_transactions_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- Name: full_sync_transactions full_sync_transactions_pkey; Type: CONSTRAINT; Schema: public; Owner: -
-- --
@ -694,6 +791,22 @@ ALTER TABLE ONLY public.headers
ADD CONSTRAINT headers_pkey PRIMARY KEY (id); ADD CONSTRAINT headers_pkey PRIMARY KEY (id);
--
-- Name: light_sync_receipts light_sync_receipts_header_id_transaction_id_key; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.light_sync_receipts
ADD CONSTRAINT light_sync_receipts_header_id_transaction_id_key UNIQUE (header_id, transaction_id);
--
-- Name: light_sync_receipts light_sync_receipts_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.light_sync_receipts
ADD CONSTRAINT light_sync_receipts_pkey PRIMARY KEY (id);
-- --
-- Name: light_sync_transactions light_sync_transactions_header_id_hash_key; Type: CONSTRAINT; Schema: public; Owner: - -- Name: light_sync_transactions light_sync_transactions_header_id_hash_key; Type: CONSTRAINT; Schema: public; Owner: -
-- --
@ -743,11 +856,19 @@ ALTER TABLE ONLY public.queued_storage
-- --
-- Name: receipts receipts_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- Name: uncles uncles_block_id_hash_key; Type: CONSTRAINT; Schema: public; Owner: -
-- --
ALTER TABLE ONLY public.receipts ALTER TABLE ONLY public.uncles
ADD CONSTRAINT receipts_pkey PRIMARY KEY (id); ADD CONSTRAINT uncles_block_id_hash_key UNIQUE (block_id, hash);
--
-- Name: uncles uncles_pkey; Type: CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.uncles
ADD CONSTRAINT uncles_pkey PRIMARY KEY (id);
-- --
@ -825,10 +946,10 @@ CREATE INDEX tx_to_index ON public.full_sync_transactions USING btree (tx_to);
-- --
-- Name: receipts blocks_fk; Type: FK CONSTRAINT; Schema: public; Owner: - -- Name: full_sync_receipts blocks_fk; Type: FK CONSTRAINT; Schema: public; Owner: -
-- --
ALTER TABLE ONLY public.receipts ALTER TABLE ONLY public.full_sync_receipts
ADD CONSTRAINT blocks_fk FOREIGN KEY (block_id) REFERENCES public.blocks(id) ON DELETE CASCADE; ADD CONSTRAINT blocks_fk FOREIGN KEY (block_id) REFERENCES public.blocks(id) ON DELETE CASCADE;
@ -856,6 +977,22 @@ ALTER TABLE ONLY public.headers
ADD CONSTRAINT headers_eth_node_id_fkey FOREIGN KEY (eth_node_id) REFERENCES public.eth_nodes(id) ON DELETE CASCADE; ADD CONSTRAINT headers_eth_node_id_fkey FOREIGN KEY (eth_node_id) REFERENCES public.eth_nodes(id) ON DELETE CASCADE;
--
-- Name: light_sync_receipts light_sync_receipts_header_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.light_sync_receipts
ADD CONSTRAINT light_sync_receipts_header_id_fkey FOREIGN KEY (header_id) REFERENCES public.headers(id) ON DELETE CASCADE;
--
-- Name: light_sync_receipts light_sync_receipts_transaction_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.light_sync_receipts
ADD CONSTRAINT light_sync_receipts_transaction_id_fkey FOREIGN KEY (transaction_id) REFERENCES public.light_sync_transactions(id) ON DELETE CASCADE;
-- --
-- Name: light_sync_transactions light_sync_transactions_header_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: - -- Name: light_sync_transactions light_sync_transactions_header_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
-- --
@ -877,7 +1014,23 @@ ALTER TABLE ONLY public.blocks
-- --
ALTER TABLE ONLY public.logs ALTER TABLE ONLY public.logs
ADD CONSTRAINT receipts_fk FOREIGN KEY (receipt_id) REFERENCES public.receipts(id) ON DELETE CASCADE; ADD CONSTRAINT receipts_fk FOREIGN KEY (receipt_id) REFERENCES public.full_sync_receipts(id) ON DELETE CASCADE;
--
-- Name: uncles uncles_block_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.uncles
ADD CONSTRAINT uncles_block_id_fkey FOREIGN KEY (block_id) REFERENCES public.blocks(id) ON DELETE CASCADE;
--
-- Name: uncles uncles_eth_node_id_fkey; Type: FK CONSTRAINT; Schema: public; Owner: -
--
ALTER TABLE ONLY public.uncles
ADD CONSTRAINT uncles_eth_node_id_fkey FOREIGN KEY (eth_node_id) REFERENCES public.eth_nodes(id) ON DELETE CASCADE;
-- --

View File

@ -0,0 +1,38 @@
[database]
name = "vulcanize_public"
hostname = "localhost"
port = 5432
[client]
ipcPath = ""
[exporter]
home = "github.com/vulcanize/vulcanizedb"
name = "accountTransformerExporter"
save = false
transformerNames = [
"account"
]
[exporter.account]
path = "transformers/account/light/initializer"
type = "eth_contract"
repository = "github.com/vulcanize/account_transformers"
migrations = "db/migrations"
rank = "0"
[token]
addresses = [
"0x58b6A8A3302369DAEc383334672404Ee733aB239",
"0x862Da0A691bb0b74038377295f8fF523D0493eB4",
]
[token.equivalents]
0x0000000000085d4780B73119b644AE5ecd22b376 = [
"0x8dd5fbCe2F6a956C3022bA3663759011Dd51e73E"
]
0x58b6A8A3302369DAEc383334672404Ee733aB239 = [
"0x8e306b005773bee6bA6A6e8972Bc79D766cC15c8"
]
[account]
start = 0
addresses = []

View File

@ -53,7 +53,7 @@ func (r *blockRetriever) retrieveFirstBlockFromReceipts(contractAddr string) (in
err := r.db.Get( err := r.db.Get(
&firstBlock, &firstBlock,
`SELECT number FROM blocks `SELECT number FROM blocks
WHERE id = (SELECT block_id FROM receipts WHERE id = (SELECT block_id FROM full_sync_receipts
WHERE lower(contract_address) = $1 WHERE lower(contract_address) = $1
ORDER BY block_id ASC ORDER BY block_id ASC
LIMIT 1)`, LIMIT 1)`,

View File

@ -18,9 +18,9 @@ package repository
import ( import (
"fmt" "fmt"
"github.com/jmoiron/sqlx"
"github.com/hashicorp/golang-lru" "github.com/hashicorp/golang-lru"
"github.com/jmoiron/sqlx"
"github.com/vulcanize/vulcanizedb/pkg/core" "github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres" "github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
@ -104,7 +104,6 @@ func (r *headerRepository) MarkHeaderChecked(headerID int64, id string) error {
VALUES ($1, $2) VALUES ($1, $2)
ON CONFLICT (header_id) DO ON CONFLICT (header_id) DO
UPDATE SET `+id+` = checked_headers.`+id+` + 1`, headerID, 1) UPDATE SET `+id+` = checked_headers.`+id+` + 1`, headerID, 1)
return err return err
} }
@ -124,7 +123,6 @@ func (r *headerRepository) MarkHeaderCheckedForAll(headerID int64, ids []string)
} }
pgStr = pgStr[:len(pgStr)-2] pgStr = pgStr[:len(pgStr)-2]
_, err := r.db.Exec(pgStr, headerID) _, err := r.db.Exec(pgStr, headerID)
return err return err
} }
@ -134,7 +132,6 @@ func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids [
if err != nil { if err != nil {
return err return err
} }
for _, header := range headers { for _, header := range headers {
pgStr := "INSERT INTO public.checked_headers (header_id, " pgStr := "INSERT INTO public.checked_headers (header_id, "
for _, id := range ids { for _, id := range ids {
@ -155,8 +152,8 @@ func (r *headerRepository) MarkHeadersCheckedForAll(headers []core.Header, ids [
return err return err
} }
} }
err = tx.Commit()
return tx.Commit() return err
} }
// Returns missing headers for the provided checked_headers column id // Returns missing headers for the provided checked_headers column id
@ -164,14 +161,13 @@ func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber
var result []core.Header var result []core.Header
var query string var query string
var err error var err error
if endingBlockNumber == -1 { if endingBlockNumber == -1 {
query = `SELECT headers.id, headers.block_number, headers.hash FROM headers query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL OR checked_headers.` + id + `=0) WHERE (header_id ISNULL OR checked_headers.` + id + `=0)
AND headers.block_number >= $1 AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2 AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else { } else {
query = `SELECT headers.id, headers.block_number, headers.hash FROM headers query = `SELECT headers.id, headers.block_number, headers.hash FROM headers
@ -180,10 +176,9 @@ func (r *headerRepository) MissingHeaders(startingBlockNumber, endingBlockNumber
AND headers.block_number >= $1 AND headers.block_number >= $1
AND headers.block_number <= $2 AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3 AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
} }
return contiguousHeaders(result, startingBlockNumber), err return contiguousHeaders(result, startingBlockNumber), err
} }
@ -192,34 +187,30 @@ func (r *headerRepository) MissingHeadersForAll(startingBlockNumber, endingBlock
var result []core.Header var result []core.Header
var query string var query string
var err error var err error
baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id ISNULL` WHERE (header_id ISNULL`
for _, id := range ids { for _, id := range ids {
baseQuery += ` OR checked_headers.` + id + `= 0` baseQuery += ` OR checked_headers.` + id + `= 0`
} }
if endingBlockNumber == -1 { if endingBlockNumber == -1 {
endStr := `) AND headers.block_number >= $1 endStr := `) AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2 AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
query = baseQuery + endStr query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else { } else {
endStr := `) AND headers.block_number >= $1 endStr := `) AND headers.block_number >= $1
AND headers.block_number <= $2 AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3 AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
query = baseQuery + endStr query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
} }
return contiguousHeaders(result, startingBlockNumber), err return contiguousHeaders(result, startingBlockNumber), err
} }
// Takes in an ordered sequence of headers and returns only the first contiguous segment // Returns a continuous set of headers that is contiguous with the provided startingBlockNumber
// Enforce continuity with previous segment with the appropriate startingBlockNumber
func contiguousHeaders(headers []core.Header, startingBlockNumber int64) []core.Header { func contiguousHeaders(headers []core.Header, startingBlockNumber int64) []core.Header {
if len(headers) < 1 { if len(headers) < 1 {
return headers return headers
@ -243,7 +234,6 @@ func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlock
var result []core.Header var result []core.Header
var query string var query string
var err error var err error
baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers baseQuery := `SELECT headers.id, headers.block_number, headers.hash FROM headers
LEFT JOIN checked_headers on headers.id = header_id LEFT JOIN checked_headers on headers.id = header_id
WHERE (header_id IS NOT NULL` WHERE (header_id IS NOT NULL`
@ -255,23 +245,37 @@ func (r *headerRepository) MissingMethodsCheckedEventsIntersection(startingBlock
baseQuery += id + ` =0 AND ` baseQuery += id + ` =0 AND `
} }
baseQuery = baseQuery[:len(baseQuery)-5] + `) ` baseQuery = baseQuery[:len(baseQuery)-5] + `) `
if endingBlockNumber == -1 { if endingBlockNumber == -1 {
endStr := `AND headers.block_number >= $1 endStr := `AND headers.block_number >= $1
AND headers.eth_node_fingerprint = $2 AND headers.eth_node_fingerprint = $2
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
query = baseQuery + endStr query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, r.db.Node.ID)
} else { } else {
endStr := `AND headers.block_number >= $1 endStr := `AND headers.block_number >= $1
AND headers.block_number <= $2 AND headers.block_number <= $2
AND headers.eth_node_fingerprint = $3 AND headers.eth_node_fingerprint = $3
ORDER BY headers.block_number LIMIT 100` ORDER BY headers.block_number`
query = baseQuery + endStr query = baseQuery + endStr
err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID) err = r.db.Select(&result, query, startingBlockNumber, endingBlockNumber, r.db.Node.ID)
} }
return continuousHeaders(result), err
}
return result, err // Returns a continuous set of headers
func continuousHeaders(headers []core.Header) []core.Header {
if len(headers) < 1 {
return headers
}
previousHeader := headers[0].BlockNumber
for i := 1; i < len(headers); i++ {
previousHeader++
if headers[i].BlockNumber != previousHeader {
return headers[:i]
}
}
return headers
} }
// Check the repositories column id cache for a value // Check the repositories column id cache for a value
@ -280,7 +284,7 @@ func (r *headerRepository) CheckCache(key string) (interface{}, bool) {
} }
// Used to mark a header checked as part of some external transaction so as to group into one commit // Used to mark a header checked as part of some external transaction so as to group into one commit
func MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error { func (r *headerRepository) MarkHeaderCheckedInTransaction(headerID int64, tx *sqlx.Tx, eventID string) error {
_, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`) _, err := tx.Exec(`INSERT INTO public.checked_headers (header_id, `+eventID+`)
VALUES ($1, $2) VALUES ($1, $2)
ON CONFLICT (header_id) DO ON CONFLICT (header_id) DO

View File

@ -203,18 +203,6 @@ var _ = Describe("Repository", func() {
Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633))) Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
}) })
It("Returns at most 100 headers", func() {
add102Headers(coreHeaderRepo)
err := contractHeaderRepo.AddCheckColumns(eventIDs)
Expect(err).ToNot(HaveOccurred())
missingHeaders, err := contractHeaderRepo.MissingHeadersForAll(6194632, 6194733, eventIDs)
Expect(err).ToNot(HaveOccurred())
Expect(len(missingHeaders)).To(Equal(100))
Expect(missingHeaders[0].BlockNumber).To(Equal(int64(6194632)))
Expect(missingHeaders[1].BlockNumber).To(Equal(int64(6194633)))
})
It("Fails if one of the eventIDs does not yet exist in check_headers table", func() { It("Fails if one of the eventIDs does not yet exist in check_headers table", func() {
addHeaders(coreHeaderRepo) addHeaders(coreHeaderRepo)
err := contractHeaderRepo.AddCheckColumns(eventIDs) err := contractHeaderRepo.AddCheckColumns(eventIDs)
@ -357,12 +345,3 @@ func addDiscontinuousHeaders(coreHeaderRepo repositories.HeaderRepository) {
coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2) coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader2)
coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader4) coreHeaderRepo.CreateOrUpdateHeader(mocks.MockHeader4)
} }
func add102Headers(coreHeaderRepo repositories.HeaderRepository) {
baseHeader := mocks.MockHeader1
for i := 6194632; i < 6194733; i++ {
_, err := coreHeaderRepo.CreateOrUpdateHeader(baseHeader)
Expect(err).ToNot(HaveOccurred())
baseHeader.BlockNumber++
}
}

View File

@ -243,7 +243,10 @@ func TearDown(db *postgres.DB) {
_, err = tx.Exec("DELETE FROM light_sync_transactions") _, err = tx.Exec("DELETE FROM light_sync_transactions")
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM receipts`) _, err = tx.Exec(`DELETE FROM full_sync_receipts`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM light_sync_receipts`)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DROP TABLE checked_headers`) _, err = tx.Exec(`DROP TABLE checked_headers`)

View File

@ -23,7 +23,7 @@ import (
"github.com/hashicorp/golang-lru" "github.com/hashicorp/golang-lru"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/light/repository" "github.com/vulcanize/vulcanizedb/libraries/shared/repository"
"github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types" "github.com/vulcanize/vulcanizedb/pkg/contract_watcher/shared/types"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres" "github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
) )

View File

@ -18,11 +18,12 @@ package core
type Receipt struct { type Receipt struct {
Bloom string Bloom string
ContractAddress string ContractAddress string `db:"contract_address"`
CumulativeGasUsed uint64 CumulativeGasUsed uint64 `db:"cumulative_gas_used"`
GasUsed uint64 GasUsed uint64 `db:"gas_used"`
Logs []Log Logs []Log
StateRoot string StateRoot string `db:"state_root"`
Status int Status int
TxHash string TxHash string `db:"tx_hash"`
Rlp []byte `db:"rlp"`
} }

View File

@ -23,7 +23,7 @@ type TransactionModel struct {
GasPrice int64 `db:"gas_price"` GasPrice int64 `db:"gas_price"`
Hash string Hash string
Nonce uint64 Nonce uint64
Raw []byte Raw []byte `db:"raw"`
Receipt Receipt
To string `db:"tx_to"` To string `db:"tx_to"`
TxIndex int64 `db:"tx_index"` TxIndex int64 `db:"tx_index"`

View File

@ -19,9 +19,9 @@ package repositories
import ( import (
"database/sql" "database/sql"
"errors" "errors"
"github.com/jmoiron/sqlx" "github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/libraries/shared/utilities"
"github.com/vulcanize/vulcanizedb/pkg/core" "github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore" "github.com/vulcanize/vulcanizedb/pkg/datastore"
@ -144,8 +144,8 @@ func (blockRepository BlockRepository) insertBlock(block core.Block) (int64, err
block.IsFinal, block.IsFinal,
block.Miner, block.Miner,
block.ExtraData, block.ExtraData,
utilities.NullToZero(block.Reward), nullStringToZero(block.Reward),
utilities.NullToZero(block.UnclesReward), nullStringToZero(block.UnclesReward),
blockRepository.database.Node.ID). blockRepository.database.Node.ID).
Scan(&blockId) Scan(&blockId)
if insertBlockErr != nil { if insertBlockErr != nil {
@ -199,7 +199,7 @@ func (blockRepository BlockRepository) createUncle(tx *sqlx.Tx, blockId int64, u
(hash, block_id, reward, miner, raw, block_timestamp, eth_node_id, eth_node_fingerprint) (hash, block_id, reward, miner, raw, block_timestamp, eth_node_id, eth_node_fingerprint)
VALUES ($1, $2, $3, $4, $5, $6, $7::NUMERIC, $8) VALUES ($1, $2, $3, $4, $5, $6, $7::NUMERIC, $8)
RETURNING id`, RETURNING id`,
uncle.Hash, blockId, utilities.NullToZero(uncle.Reward), uncle.Miner, uncle.Raw, uncle.Timestamp, blockRepository.database.NodeID, blockRepository.database.Node.ID) uncle.Hash, blockId, nullStringToZero(uncle.Reward), uncle.Miner, uncle.Raw, uncle.Timestamp, blockRepository.database.NodeID, blockRepository.database.Node.ID)
return err return err
} }
@ -260,7 +260,7 @@ func (blockRepository BlockRepository) createReceipt(tx *sqlx.Tx, blockId int64,
//Not currently persisting log bloom filters //Not currently persisting log bloom filters
var receiptId int var receiptId int
err := tx.QueryRow( err := tx.QueryRow(
`INSERT INTO receipts `INSERT INTO full_sync_receipts
(contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id) (contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id)
VALUES ($1, $2, $3, $4, $5, $6, $7) VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id`, RETURNING id`,

View File

@ -329,11 +329,11 @@ var _ = Describe("Saving blocks", func() {
GasPrice: gasPrice, GasPrice: gasPrice,
Hash: "x1234", Hash: "x1234",
Nonce: nonce, Nonce: nonce,
Raw: raw.Bytes(),
Receipt: core.Receipt{}, Receipt: core.Receipt{},
To: to, To: to,
TxIndex: 2, TxIndex: 2,
Value: value.String(), Value: value.String(),
Raw: []byte{},
} }
block := core.Block{ block := core.Block{
Number: 123, Number: 123,

View File

@ -19,7 +19,10 @@ package repositories
import ( import (
"database/sql" "database/sql"
"errors" "errors"
"github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/core" "github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres" "github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
) )
@ -64,6 +67,40 @@ func (repository HeaderRepository) CreateTransactions(headerID int64, transactio
return nil return nil
} }
func (repository HeaderRepository) CreateTransactionInTx(tx *sqlx.Tx, headerID int64, transaction core.TransactionModel) (int64, error) {
var txId int64
err := tx.QueryRowx(`INSERT INTO public.light_sync_transactions
(header_id, hash, gas_limit, gas_price, input_data, nonce, raw, tx_from, tx_index, tx_to, "value")
VALUES ($1, $2, $3::NUMERIC, $4::NUMERIC, $5, $6::NUMERIC, $7, $8, $9::NUMERIC, $10, $11::NUMERIC)
ON CONFLICT (header_id, hash) DO UPDATE
SET (gas_limit, gas_price, input_data, nonce, raw, tx_from, tx_index, tx_to, "value") = ($3::NUMERIC, $4::NUMERIC, $5, $6::NUMERIC, $7, $8, $9::NUMERIC, $10, $11::NUMERIC)
RETURNING id`,
headerID, transaction.Hash, transaction.GasLimit, transaction.GasPrice,
transaction.Data, transaction.Nonce, transaction.Raw, transaction.From,
transaction.TxIndex, transaction.To, transaction.Value).Scan(&txId)
if err != nil {
log.Error("header_repository: error inserting transaction: ", err)
return txId, err
}
return txId, err
}
func (repository HeaderRepository) CreateReceiptInTx(tx *sqlx.Tx, headerID, transactionID int64, receipt core.Receipt) (int64, error) {
var receiptId int64
err := tx.QueryRowx(`INSERT INTO public.light_sync_receipts
(header_id, transaction_id, contract_address, cumulative_gas_used, gas_used, state_root, status, tx_hash, rlp)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, transaction_id) DO UPDATE
SET (contract_address, cumulative_gas_used, gas_used, state_root, status, tx_hash, rlp) = ($3, $4::NUMERIC, $5::NUMERIC, $6, $7, $8, $9)
RETURNING id`,
headerID, transactionID, receipt.ContractAddress, receipt.CumulativeGasUsed, receipt.GasUsed, receipt.StateRoot, receipt.Status, receipt.TxHash, receipt.Rlp).Scan(&receiptId)
if err != nil {
log.Error("header_repository: error inserting receipt: ", err)
return receiptId, err
}
return receiptId, err
}
func (repository HeaderRepository) GetHeader(blockNumber int64) (core.Header, error) { func (repository HeaderRepository) GetHeader(blockNumber int64) (core.Header, error) {
var header core.Header var header core.Header
err := repository.database.Get(&header, `SELECT id, block_number, hash, raw, block_timestamp FROM headers WHERE block_number = $1 AND eth_node_fingerprint = $2`, err := repository.database.Get(&header, `SELECT id, block_number, hash, raw, block_timestamp FROM headers WHERE block_number = $1 AND eth_node_fingerprint = $2`,

View File

@ -180,6 +180,66 @@ var _ = Describe("Block header repository", func() {
}) })
}) })
Describe("creating a receipt", func() {
It("adds a receipt in a tx", func() {
headerID, err := repo.CreateOrUpdateHeader(header)
Expect(err).NotTo(HaveOccurred())
fromAddress := common.HexToAddress("0x1234")
toAddress := common.HexToAddress("0x5678")
txHash := common.HexToHash("0x9876")
txIndex := big.NewInt(123)
transaction := core.TransactionModel{
Data: []byte{},
From: fromAddress.Hex(),
GasLimit: 0,
GasPrice: 0,
Hash: txHash.Hex(),
Nonce: 0,
Raw: []byte{},
To: toAddress.Hex(),
TxIndex: txIndex.Int64(),
Value: "0",
}
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
txId, txErr := repo.CreateTransactionInTx(tx, headerID, transaction)
Expect(txErr).ToNot(HaveOccurred())
contractAddr := common.HexToAddress("0x1234")
stateRoot := common.HexToHash("0x5678")
receipt := core.Receipt{
ContractAddress: contractAddr.Hex(),
TxHash: txHash.Hex(),
GasUsed: 10,
CumulativeGasUsed: 100,
StateRoot: stateRoot.Hex(),
Rlp: []byte{1, 2, 3},
}
_, receiptErr := repo.CreateReceiptInTx(tx, headerID, txId, receipt)
commitErr := tx.Commit()
Expect(commitErr).ToNot(HaveOccurred())
Expect(receiptErr).ToNot(HaveOccurred())
type idModel struct {
TransactionId int64 `db:"transaction_id"`
core.Receipt
}
var dbReceipt idModel
err = db.Get(&dbReceipt,
`SELECT transaction_id, contract_address, cumulative_gas_used, gas_used, state_root, status, tx_hash, rlp
FROM public.light_sync_receipts WHERE header_id = $1`, headerID)
Expect(err).NotTo(HaveOccurred())
Expect(dbReceipt.TransactionId).To(Equal(txId))
Expect(dbReceipt.TxHash).To(Equal(txHash.Hex()))
Expect(dbReceipt.ContractAddress).To(Equal(contractAddr.Hex()))
Expect(dbReceipt.CumulativeGasUsed).To(Equal(uint64(100)))
Expect(dbReceipt.GasUsed).To(Equal(uint64(10)))
Expect(dbReceipt.StateRoot).To(Equal(stateRoot.Hex()))
Expect(dbReceipt.Rlp).To(Equal([]byte{1, 2, 3}))
})
})
Describe("creating a transaction", func() { Describe("creating a transaction", func() {
var ( var (
headerID int64 headerID int64
@ -206,6 +266,7 @@ var _ = Describe("Block header repository", func() {
To: toAddress.Hex(), To: toAddress.Hex(),
TxIndex: txIndex.Int64(), TxIndex: txIndex.Int64(),
Value: "0", Value: "0",
Receipt: core.Receipt{},
}, { }, {
Data: []byte{}, Data: []byte{},
From: fromAddress.Hex(), From: fromAddress.Hex(),
@ -245,6 +306,87 @@ var _ = Describe("Block header repository", func() {
}) })
}) })
Describe("creating a transaction in a sqlx tx", func() {
It("adds a transaction", func() {
headerID, err := repo.CreateOrUpdateHeader(header)
Expect(err).NotTo(HaveOccurred())
fromAddress := common.HexToAddress("0x1234")
toAddress := common.HexToAddress("0x5678")
txHash := common.HexToHash("0x9876")
txIndex := big.NewInt(123)
transaction := core.TransactionModel{
Data: []byte{},
From: fromAddress.Hex(),
GasLimit: 0,
GasPrice: 0,
Hash: txHash.Hex(),
Nonce: 0,
Raw: []byte{1, 2, 3},
To: toAddress.Hex(),
TxIndex: txIndex.Int64(),
Value: "0",
}
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
_, insertErr := repo.CreateTransactionInTx(tx, headerID, transaction)
commitErr := tx.Commit()
Expect(commitErr).ToNot(HaveOccurred())
Expect(insertErr).NotTo(HaveOccurred())
var dbTransaction core.TransactionModel
err = db.Get(&dbTransaction,
`SELECT hash, gas_limit, gas_price, input_data, nonce, raw, tx_from, tx_index, tx_to, "value"
FROM public.light_sync_transactions WHERE header_id = $1`, headerID)
Expect(err).NotTo(HaveOccurred())
Expect(dbTransaction).To(Equal(transaction))
})
It("silently upserts", func() {
headerID, err := repo.CreateOrUpdateHeader(header)
Expect(err).NotTo(HaveOccurred())
fromAddress := common.HexToAddress("0x1234")
toAddress := common.HexToAddress("0x5678")
txHash := common.HexToHash("0x9876")
txIndex := big.NewInt(123)
transaction := core.TransactionModel{
Data: []byte{},
From: fromAddress.Hex(),
GasLimit: 0,
GasPrice: 0,
Hash: txHash.Hex(),
Nonce: 0,
Raw: []byte{},
Receipt: core.Receipt{},
To: toAddress.Hex(),
TxIndex: txIndex.Int64(),
Value: "0",
}
tx1, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
txId1, insertErr := repo.CreateTransactionInTx(tx1, headerID, transaction)
commit1Err := tx1.Commit()
Expect(commit1Err).ToNot(HaveOccurred())
Expect(insertErr).NotTo(HaveOccurred())
tx2, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
txId2, insertErr := repo.CreateTransactionInTx(tx2, headerID, transaction)
commit2Err := tx2.Commit()
Expect(commit2Err).ToNot(HaveOccurred())
Expect(insertErr).NotTo(HaveOccurred())
Expect(txId1).To(Equal(txId2))
var dbTransactions []core.TransactionModel
err = db.Select(&dbTransactions,
`SELECT hash, gas_limit, gas_price, input_data, nonce, raw, tx_from, tx_index, tx_to, "value"
FROM public.light_sync_transactions WHERE header_id = $1`, headerID)
Expect(err).NotTo(HaveOccurred())
Expect(len(dbTransactions)).To(Equal(1))
})
})
Describe("Getting a header", func() { Describe("Getting a header", func() {
It("returns header if it exists", func() { It("returns header if it exists", func() {
_, err = repo.CreateOrUpdateHeader(header) _, err = repo.CreateOrUpdateHeader(header)

View File

@ -56,7 +56,7 @@ func (receiptRepository ReceiptRepository) CreateReceiptsAndLogs(blockId int64,
func createReceipt(receipt core.Receipt, blockId int64, tx *sqlx.Tx) (int64, error) { func createReceipt(receipt core.Receipt, blockId int64, tx *sqlx.Tx) (int64, error) {
var receiptId int64 var receiptId int64
err := tx.QueryRow( err := tx.QueryRow(
`INSERT INTO receipts `INSERT INTO full_sync_receipts
(contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id) (contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id)
VALUES ($1, $2, $3, $4, $5, $6, $7) VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id`, RETURNING id`,
@ -87,7 +87,7 @@ func (receiptRepository ReceiptRepository) CreateReceipt(blockId int64, receipt
tx, _ := receiptRepository.DB.Beginx() tx, _ := receiptRepository.DB.Beginx()
var receiptId int64 var receiptId int64
err := tx.QueryRow( err := tx.QueryRow(
`INSERT INTO receipts `INSERT INTO full_sync_receipts
(contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id) (contract_address, tx_hash, cumulative_gas_used, gas_used, state_root, status, block_id)
VALUES ($1, $2, $3, $4, $5, $6, $7) VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING id`, RETURNING id`,
@ -109,7 +109,7 @@ func (receiptRepository ReceiptRepository) GetReceipt(txHash string) (core.Recei
gas_used, gas_used,
state_root, state_root,
status status
FROM receipts FROM full_sync_receipts
WHERE tx_hash = $1`, txHash) WHERE tx_hash = $1`, txHash)
receipt, err := loadReceipt(row) receipt, err := loadReceipt(row)
if err != nil { if err != nil {

View File

@ -67,6 +67,9 @@ func (ci *ColdImporter) createBlocksAndTransactions(hash []byte, i int64) (int64
func (ci *ColdImporter) createReceiptsAndLogs(hash []byte, number int64, blockId int64) error { func (ci *ColdImporter) createReceiptsAndLogs(hash []byte, number int64, blockId int64) error {
receipts := ci.ethDB.GetBlockReceipts(hash, number) receipts := ci.ethDB.GetBlockReceipts(hash, number)
coreReceipts := common.ToCoreReceipts(receipts) coreReceipts, err := common.ToCoreReceipts(receipts)
if err != nil {
return err
}
return ci.receiptRepository.CreateReceiptsAndLogs(blockId, coreReceipts) return ci.receiptRepository.CreateReceiptsAndLogs(blockId, coreReceipts)
} }

View File

@ -126,7 +126,8 @@ var _ = Describe("Geth cold importer", func() {
err := importer.Execute(blockNumber, blockNumber, "node_id") err := importer.Execute(blockNumber, blockNumber, "node_id")
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
expectedReceipts := vulcCommon.ToCoreReceipts(fakeReceipts) expectedReceipts, err := vulcCommon.ToCoreReceipts(fakeReceipts)
Expect(err).ToNot(HaveOccurred())
mockReceiptRepository.AssertCreateReceiptsAndLogsCalledWith(blockId, expectedReceipts) mockReceiptRepository.AssertCreateReceiptsAndLogsCalledWith(blockId, expectedReceipts)
}) })

View File

@ -290,7 +290,8 @@ var _ = Describe("Conversion of GethBlock to core.Block", func() {
Expect(coreTransaction.Nonce).To(Equal(gethTransaction.Nonce())) Expect(coreTransaction.Nonce).To(Equal(gethTransaction.Nonce()))
coreReceipt := coreTransaction.Receipt coreReceipt := coreTransaction.Receipt
expectedReceipt := vulcCommon.ToCoreReceipt(gethReceipt) expectedReceipt, err := vulcCommon.ToCoreReceipt(gethReceipt)
Expect(err).ToNot(HaveOccurred())
Expect(coreReceipt).To(Equal(expectedReceipt)) Expect(coreReceipt).To(Equal(expectedReceipt))
}) })
@ -329,7 +330,8 @@ var _ = Describe("Conversion of GethBlock to core.Block", func() {
Expect(coreTransaction.To).To(Equal("")) Expect(coreTransaction.To).To(Equal(""))
coreReceipt := coreTransaction.Receipt coreReceipt := coreTransaction.Receipt
expectedReceipt := vulcCommon.ToCoreReceipt(gethReceipt) expectedReceipt, err := vulcCommon.ToCoreReceipt(gethReceipt)
Expect(err).ToNot(HaveOccurred())
Expect(coreReceipt).To(Equal(expectedReceipt)) Expect(coreReceipt).To(Equal(expectedReceipt))
}) })
}) })

View File

@ -26,16 +26,19 @@ import (
"github.com/vulcanize/vulcanizedb/pkg/core" "github.com/vulcanize/vulcanizedb/pkg/core"
) )
func ToCoreReceipts(gethReceipts types.Receipts) []core.Receipt { func ToCoreReceipts(gethReceipts types.Receipts) ([]core.Receipt, error) {
var coreReceipts []core.Receipt var coreReceipts []core.Receipt
for _, receipt := range gethReceipts { for _, receipt := range gethReceipts {
coreReceipt := ToCoreReceipt(receipt) coreReceipt, err := ToCoreReceipt(receipt)
if err != nil {
return nil, err
}
coreReceipts = append(coreReceipts, coreReceipt) coreReceipts = append(coreReceipts, coreReceipt)
} }
return coreReceipts return coreReceipts, nil
} }
func ToCoreReceipt(gethReceipt *types.Receipt) core.Receipt { func ToCoreReceipt(gethReceipt *types.Receipt) (core.Receipt, error) {
bloom := hexutil.Encode(gethReceipt.Bloom.Bytes()) bloom := hexutil.Encode(gethReceipt.Bloom.Bytes())
var postState string var postState string
var status int var status int
@ -43,6 +46,12 @@ func ToCoreReceipt(gethReceipt *types.Receipt) core.Receipt {
logs := dereferenceLogs(gethReceipt) logs := dereferenceLogs(gethReceipt)
contractAddress := setContractAddress(gethReceipt) contractAddress := setContractAddress(gethReceipt)
rlpBuff := new(bytes.Buffer)
receiptForStorage := types.ReceiptForStorage(*gethReceipt)
err := receiptForStorage.EncodeRLP(rlpBuff)
if err != nil {
return core.Receipt{}, err
}
return core.Receipt{ return core.Receipt{
Bloom: bloom, Bloom: bloom,
ContractAddress: contractAddress, ContractAddress: contractAddress,
@ -52,7 +61,8 @@ func ToCoreReceipt(gethReceipt *types.Receipt) core.Receipt {
StateRoot: postState, StateRoot: postState,
TxHash: gethReceipt.TxHash.Hex(), TxHash: gethReceipt.TxHash.Hex(),
Status: status, Status: status,
} Rlp: rlpBuff.Bytes(),
}, nil
} }
func setContractAddress(gethReceipt *types.Receipt) string { func setContractAddress(gethReceipt *types.Receipt) string {

View File

@ -17,6 +17,7 @@
package common_test package common_test
import ( import (
"bytes"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
@ -40,6 +41,11 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
TxHash: common.HexToHash("0x97d99bc7729211111a21b12c933c949d4f31684f1d6954ff477d0477538ff017"), TxHash: common.HexToHash("0x97d99bc7729211111a21b12c933c949d4f31684f1d6954ff477d0477538ff017"),
} }
rlpBuff := new(bytes.Buffer)
receiptForStorage := types.ReceiptForStorage(receipt)
err := receiptForStorage.EncodeRLP(rlpBuff)
Expect(err).ToNot(HaveOccurred())
expected := core.Receipt{ expected := core.Receipt{
Bloom: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", Bloom: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
ContractAddress: "", ContractAddress: "",
@ -49,9 +55,11 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
StateRoot: "0x88abf7e73128227370aa7baa3dd4e18d0af70e92ef1f9ef426942fbe2dddb733", StateRoot: "0x88abf7e73128227370aa7baa3dd4e18d0af70e92ef1f9ef426942fbe2dddb733",
Status: -99, Status: -99,
TxHash: receipt.TxHash.Hex(), TxHash: receipt.TxHash.Hex(),
Rlp: rlpBuff.Bytes(),
} }
coreReceipt := vulcCommon.ToCoreReceipt(&receipt) coreReceipt, err := vulcCommon.ToCoreReceipt(&receipt)
Expect(err).ToNot(HaveOccurred())
Expect(coreReceipt.Bloom).To(Equal(expected.Bloom)) Expect(coreReceipt.Bloom).To(Equal(expected.Bloom))
Expect(coreReceipt.ContractAddress).To(Equal(expected.ContractAddress)) Expect(coreReceipt.ContractAddress).To(Equal(expected.ContractAddress))
Expect(coreReceipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed)) Expect(coreReceipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed))
@ -60,7 +68,7 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
Expect(coreReceipt.StateRoot).To(Equal(expected.StateRoot)) Expect(coreReceipt.StateRoot).To(Equal(expected.StateRoot))
Expect(coreReceipt.Status).To(Equal(expected.Status)) Expect(coreReceipt.Status).To(Equal(expected.Status))
Expect(coreReceipt.TxHash).To(Equal(expected.TxHash)) Expect(coreReceipt.TxHash).To(Equal(expected.TxHash))
Expect(bytes.Compare(coreReceipt.Rlp, expected.Rlp)).To(Equal(0))
}) })
It("converts geth receipt to internal receipt format (post Byzantium has status", func() { It("converts geth receipt to internal receipt format (post Byzantium has status", func() {
@ -74,6 +82,11 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
TxHash: common.HexToHash("0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223"), TxHash: common.HexToHash("0xe340558980f89d5f86045ac11e5cc34e4bcec20f9f1e2a427aa39d87114e8223"),
} }
rlpBuff := new(bytes.Buffer)
receiptForStorage := types.ReceiptForStorage(receipt)
err := receiptForStorage.EncodeRLP(rlpBuff)
Expect(err).ToNot(HaveOccurred())
expected := core.Receipt{ expected := core.Receipt{
Bloom: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000", Bloom: "0x00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000",
ContractAddress: receipt.ContractAddress.Hex(), ContractAddress: receipt.ContractAddress.Hex(),
@ -83,9 +96,11 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
StateRoot: "", StateRoot: "",
Status: 1, Status: 1,
TxHash: receipt.TxHash.Hex(), TxHash: receipt.TxHash.Hex(),
Rlp: rlpBuff.Bytes(),
} }
coreReceipt := vulcCommon.ToCoreReceipt(&receipt) coreReceipt, err := vulcCommon.ToCoreReceipt(&receipt)
Expect(err).ToNot(HaveOccurred())
Expect(coreReceipt.Bloom).To(Equal(expected.Bloom)) Expect(coreReceipt.Bloom).To(Equal(expected.Bloom))
Expect(coreReceipt.ContractAddress).To(Equal("")) Expect(coreReceipt.ContractAddress).To(Equal(""))
Expect(coreReceipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed)) Expect(coreReceipt.CumulativeGasUsed).To(Equal(expected.CumulativeGasUsed))
@ -94,5 +109,6 @@ var _ = Describe("Conversion of GethReceipt to core.Receipt", func() {
Expect(coreReceipt.StateRoot).To(Equal(expected.StateRoot)) Expect(coreReceipt.StateRoot).To(Equal(expected.StateRoot))
Expect(coreReceipt.Status).To(Equal(expected.Status)) Expect(coreReceipt.Status).To(Equal(expected.Status))
Expect(coreReceipt.TxHash).To(Equal(expected.TxHash)) Expect(coreReceipt.TxHash).To(Equal(expected.TxHash))
Expect(bytes.Compare(coreReceipt.Rlp, expected.Rlp)).To(Equal(0))
}) })
}) })

View File

@ -127,7 +127,10 @@ func (rtc *RpcTransactionConverter) appendReceiptToTransaction(transaction core.
if err != nil { if err != nil {
return transaction, err return transaction, err
} }
receipt := vulcCommon.ToCoreReceipt(gethReceipt) receipt, err := vulcCommon.ToCoreReceipt(gethReceipt)
if err != nil {
return transaction, err
}
transaction.Receipt = receipt transaction.Receipt = receipt
return transaction, nil return transaction, nil
} }

View File

@ -1,88 +0,0 @@
// 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 test_helpers
import (
"fmt"
"github.com/ethereum/go-ethereum/ethclient"
"github.com/ethereum/go-ethereum/rpc"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/geth"
"github.com/vulcanize/vulcanizedb/pkg/geth/client"
rpc2 "github.com/vulcanize/vulcanizedb/pkg/geth/converters/rpc"
"github.com/vulcanize/vulcanizedb/pkg/geth/node"
)
// TODO: consider whether this should be moved to libraries/shared
func SetupDBandBC() (*postgres.DB, core.BlockChain) {
infuraIPC := "http://kovan0.vulcanize.io:8545"
rawRpcClient, err := rpc.Dial(infuraIPC)
Expect(err).NotTo(HaveOccurred())
rpcClient := client.NewRpcClient(rawRpcClient, infuraIPC)
ethClient := ethclient.NewClient(rawRpcClient)
blockChainClient := client.NewEthClient(ethClient)
blockChainNode := node.MakeNode(rpcClient)
transactionConverter := rpc2.NewRpcTransactionConverter(ethClient)
blockChain := geth.NewBlockChain(blockChainClient, rpcClient, blockChainNode, transactionConverter)
db, err := postgres.NewDB(config.Database{
Hostname: "localhost",
Name: "vulcanize_private",
Port: 5432,
}, blockChain.Node())
Expect(err).NotTo(HaveOccurred())
return db, blockChain
}
func TearDown(db *postgres.DB) {
tx, err := db.Beginx()
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM headers`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM logs`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM log_filters`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM full_sync_transactions`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec("DELETE FROM light_sync_transactions")
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM receipts`)
Expect(err).NotTo(HaveOccurred())
_, err = tx.Exec(`DELETE FROM checked_headers`)
Expect(err).NotTo(HaveOccurred())
err = tx.Commit()
Expect(err).NotTo(HaveOccurred())
}
func DropTestSchema(db *postgres.DB, schema string) {
_, err := db.Exec(fmt.Sprintf(`DROP SCHEMA IF EXISTS %s CASCADE`, schema))
Expect(err).NotTo(HaveOccurred())
}

View File

@ -1,3 +0,0 @@
### Test
This empty directory is for holding the temporary .so and .go files generated during the generator_tests

View File

@ -115,7 +115,8 @@ func CleanTestDB(db *postgres.DB) {
db.MustExec("DELETE FROM log_filters") db.MustExec("DELETE FROM log_filters")
db.MustExec("DELETE FROM logs") db.MustExec("DELETE FROM logs")
db.MustExec("DELETE FROM queued_storage") db.MustExec("DELETE FROM queued_storage")
db.MustExec("DELETE FROM receipts") db.MustExec("DELETE FROM full_sync_receipts")
db.MustExec("DELETE FROM light_sync_receipts")
db.MustExec("DELETE FROM watched_contracts") db.MustExec("DELETE FROM watched_contracts")
} }