(VDB-699) Decouple Log extraction vs delegation to transformers

cmd/composeAndExecute.go

@@ -170,7 +170,10 @@ func composeAndExecute() {
 	var wg syn.WaitGroup
 	if len(ethEventInitializers) > 0 {
 		ew := watcher.NewEventWatcher(&db, blockChain)
-		ew.AddTransformers(ethEventInitializers)
+		err := ew.AddTransformers(ethEventInitializers)
+		if err != nil {
+			LogWithCommand.Fatalf("failed to add event transformer initializers to watcher: %s", err.Error())
+		}
 		wg.Add(1)
 		go watchEthEvents(&ew, &wg)
 	}

cmd/execute.go

@@ -114,7 +114,10 @@ func execute() {
 	var wg syn.WaitGroup
 	if len(ethEventInitializers) > 0 {
 		ew := watcher.NewEventWatcher(&db, blockChain)
-		ew.AddTransformers(ethEventInitializers)
+		err = ew.AddTransformers(ethEventInitializers)
+		if err != nil {
+			LogWithCommand.Fatalf("failed to add event transformer initializers to watcher: %s", err.Error())
+		}
 		wg.Add(1)
 		go watchEthEvents(&ew, &wg)
 	}
@@ -157,10 +160,13 @@ func watchEthEvents(w *watcher.EventWatcher, wg *syn.WaitGroup) {
 	} else {
 		recheck = constants.HeaderMissing
 	}
-	ticker := time.NewTicker(pollingInterval)
-	defer ticker.Stop()
-	for range ticker.C {
-		w.Execute(recheck)
+	errs := make(chan error)
+	go w.Execute(recheck, errs)
+	for {
+		select {
+		case err := <-errs:
+			LogWithCommand.Fatalf("error executing event watcher: %s", err.Error())
+		}
 	}
 }
 

db/migrations/00007_create_full_sync_logs_table.sql

@@ -0,0 +1,19 @@
+-- +goose Up
+CREATE TABLE full_sync_logs
+(
+    id           SERIAL PRIMARY KEY,
+    block_number BIGINT,
+    address      VARCHAR(66),
+    tx_hash      VARCHAR(66),
+    index        BIGINT,
+    topic0       VARCHAR(66),
+    topic1       VARCHAR(66),
+    topic2       VARCHAR(66),
+    topic3       VARCHAR(66),
+    data         TEXT,
+    CONSTRAINT full_sync_log_uc UNIQUE (block_number, index)
+);
+
+
+-- +goose Down
+DROP TABLE full_sync_logs;

db/migrations/00016_add_receipts_fk_to_logs.sql

@@ -1,23 +1,23 @@
 -- +goose Up
-ALTER TABLE logs
-    DROP CONSTRAINT log_uc;
+ALTER TABLE full_sync_logs
+    DROP CONSTRAINT full_sync_log_uc;
 
-ALTER TABLE logs
-  ADD COLUMN receipt_id INT;
+ALTER TABLE full_sync_logs
+    ADD COLUMN receipt_id INT;
 
-ALTER TABLE logs
-  ADD CONSTRAINT receipts_fk
-FOREIGN KEY (receipt_id)
-REFERENCES full_sync_receipts (id)
-ON DELETE CASCADE;
+ALTER TABLE full_sync_logs
+    ADD CONSTRAINT receipts_fk
+        FOREIGN KEY (receipt_id)
+            REFERENCES full_sync_receipts (id)
+            ON DELETE CASCADE;
 
 
 -- +goose Down
-ALTER TABLE logs
-  DROP CONSTRAINT receipts_fk;
+ALTER TABLE full_sync_logs
+    DROP CONSTRAINT receipts_fk;
 
-ALTER TABLE logs
-  DROP COLUMN receipt_id;
+ALTER TABLE full_sync_logs
+    DROP COLUMN receipt_id;
 
-ALTER TABLE logs
-  ADD CONSTRAINT log_uc UNIQUE (block_number, index);
+ALTER TABLE full_sync_logs
+    ADD CONSTRAINT full_sync_log_uc UNIQUE (block_number, index);

db/migrations/00018_create_watched_event_logs.sql

@@ -1,33 +1,31 @@
 -- +goose Up
 CREATE VIEW block_stats AS
-  SELECT
-    max(block_number) AS max_block,
-    min(block_number) AS min_block
-  FROM logs;
+SELECT max(block_number) AS max_block,
+       min(block_number) AS min_block
+FROM full_sync_logs;
 
 CREATE VIEW watched_event_logs AS
-  SELECT
-    log_filters.name,
-    logs.id,
-    block_number,
-    logs.address,
-    tx_hash,
-    index,
-    logs.topic0,
-    logs.topic1,
-    logs.topic2,
-    logs.topic3,
-    data,
-    receipt_id
-  FROM log_filters
-    CROSS JOIN block_stats
-    JOIN logs ON logs.address = log_filters.address
-                 AND logs.block_number >= coalesce(log_filters.from_block, block_stats.min_block)
-                 AND logs.block_number <= coalesce(log_filters.to_block, block_stats.max_block)
-  WHERE (log_filters.topic0 = logs.topic0 OR log_filters.topic0 ISNULL)
-        AND (log_filters.topic1 = logs.topic1 OR log_filters.topic1 ISNULL)
-        AND (log_filters.topic2 = logs.topic2 OR log_filters.topic2 ISNULL)
-        AND (log_filters.topic3 = logs.topic3 OR log_filters.topic3 ISNULL);
+SELECT log_filters.name,
+       full_sync_logs.id,
+       block_number,
+       full_sync_logs.address,
+       tx_hash,
+       index,
+       full_sync_logs.topic0,
+       full_sync_logs.topic1,
+       full_sync_logs.topic2,
+       full_sync_logs.topic3,
+       data,
+       receipt_id
+FROM log_filters
+         CROSS JOIN block_stats
+         JOIN full_sync_logs ON full_sync_logs.address = log_filters.address
+    AND full_sync_logs.block_number >= coalesce(log_filters.from_block, block_stats.min_block)
+    AND full_sync_logs.block_number <= coalesce(log_filters.to_block, block_stats.max_block)
+WHERE (log_filters.topic0 = full_sync_logs.topic0 OR log_filters.topic0 ISNULL)
+  AND (log_filters.topic1 = full_sync_logs.topic1 OR log_filters.topic1 ISNULL)
+  AND (log_filters.topic2 = full_sync_logs.topic2 OR log_filters.topic2 ISNULL)
+  AND (log_filters.topic3 = full_sync_logs.topic3 OR log_filters.topic3 ISNULL);
 
 -- +goose Down
 DROP VIEW watched_event_logs;

db/migrations/00023_create_headers_table.sql

@@ -1,12 +1,14 @@
 -- +goose Up
-CREATE TABLE public.headers (
-  id                    SERIAL PRIMARY KEY,
-  hash                  VARCHAR(66),
-  block_number          BIGINT,
-  raw                   JSONB,
-  block_timestamp       NUMERIC,
-  eth_node_id           INTEGER NOT NULL REFERENCES eth_nodes (id) ON DELETE CASCADE,
-  eth_node_fingerprint  VARCHAR(128)
+CREATE TABLE public.headers
+(
+    id                   SERIAL PRIMARY KEY,
+    hash                 VARCHAR(66),
+    block_number         BIGINT,
+    raw                  JSONB,
+    block_timestamp      NUMERIC,
+    check_count          INTEGER NOT NULL DEFAULT 0,
+    eth_node_id          INTEGER NOT NULL REFERENCES eth_nodes (id) ON DELETE CASCADE,
+    eth_node_fingerprint VARCHAR(128)
 );
 
 -- Index is removed when table is

db/migrations/00029_create_header_sync_logs_table.sql

@@ -0,0 +1,22 @@
+-- +goose Up
+-- SQL in this section is executed when the migration is applied.
+CREATE TABLE header_sync_logs
+(
+    id           SERIAL PRIMARY KEY,
+    header_id    INTEGER NOT NULL REFERENCES headers (id) ON DELETE CASCADE,
+    address      INTEGER NOT NULL REFERENCES addresses (id) ON DELETE CASCADE,
+    topics       BYTEA[],
+    data         BYTEA,
+    block_number BIGINT,
+    block_hash   VARCHAR(66),
+    tx_hash      VARCHAR(66),
+    tx_index     INTEGER,
+    log_index    INTEGER,
+    raw          JSONB,
+    transformed  BOOL    NOT NULL DEFAULT FALSE,
+    UNIQUE (header_id, tx_index, log_index)
+);
+
+-- +goose Down
+-- SQL in this section is executed when the migration is rolled back.
+DROP TABLE header_sync_logs;

db/migrations/00030_create_checked_logs_table.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+-- SQL in this section is executed when the migration is applied.
+CREATE TABLE public.checked_logs
+(
+    id               SERIAL PRIMARY KEY,
+    contract_address VARCHAR(42),
+    topic_zero       VARCHAR(66)
+);
+
+-- +goose Down
+-- SQL in this section is executed when the migration is rolled back.
+DROP TABLE public.checked_logs;