Using sidekiq to process transactions and event logs.

.env

@@ -4,12 +4,19 @@
 # "db" for docker inside database, "host.docker.internal" for host database(only work for mac and windows for docker 18.03+)
 DB_HOST="db"
 DB_PORT=5432
+DB_POOL=64
 DB_USERNAME="postgres"
 DB_PASSWORD="postgres"
 DB_NAME_DEV="re-birth_development"
 DB_NAME_TEST="re-birth_test"
 DB_NAME_PRO="re-birth_production"
 
+# Redis config
+REDIS_URL="redis://localhost:6379/8"
+# set password if have
+# REDIS_PASSWORD=""
+REDIS_NAMESPACE="rebirth"
+
 # secret
 # run `rails secret` to generate one and rewrite in .env.local
 SECRET_KEY_BASE="dee66a8ef3281bc6bd74a5637ede4d846c455df93ca0d6c098da1d52b65dda8f87933c421a077643b903d8d3c5328178e39f729dda0a06e96d268c5c81530df5"

Gemfile

@@ -56,10 +56,20 @@ gem 'health_check', '~> 3.0'
 # appchain sdk
 gem "appchain.rb", github: "cryptape/appchain.rb"
 
+# Redis
+gem 'hiredis', '~> 0.6.1'
+gem 'redis', '~> 4.0', '>= 4.0.3'
+gem 'redis-namespace', '~> 1.6'
+gem 'redis-objects', '~> 1.4', '>= 1.4.3'
+
+# Sidekiq
+gem 'sidekiq', '~> 5.2', '>= 5.2.3'
+
 # Deployment
 gem 'mina', require: false
 gem 'mina-puma', require: false
 gem 'mina-multistage', require: false
+gem 'mina-sidekiq', '~> 1.0', '>= 1.0.3', require: false
 
 group :development, :test do
   # Call 'byebug' anywhere in the code to stop execution and get a debugger console

Gemfile.lock

@@ -82,6 +82,7 @@ GEM
       url
     coderay (1.1.2)
     concurrent-ruby (1.1.1)
+    connection_pool (2.2.2)
     crack (0.4.3)
       safe_yaml (~> 1.0.0)
     crass (1.0.4)
@@ -112,6 +113,7 @@ GEM
     hashdiff (0.3.7)
     health_check (3.0.0)
       railties (>= 5.0)
+    hiredis (0.6.1)
     i18n (1.1.1)
       concurrent-ruby (~> 1.0)
     jaro_winkler (1.5.1)
@@ -150,6 +152,8 @@ GEM
     mina-puma (1.1.0)
       mina (~> 1.2.0)
       puma (>= 2.13)
+    mina-sidekiq (1.0.3)
+      mina (>= 1.0.2)
     mini_mime (1.0.1)
     mini_portile2 (2.3.0)
     minitest (5.11.3)
@@ -174,6 +178,8 @@ GEM
     puma (3.12.0)
     rack (2.0.6)
     rack-cors (1.0.2)
+    rack-protection (2.0.4)
+      rack
     rack-test (1.1.0)
       rack (>= 1.0, < 3)
     rails (5.2.1)
@@ -210,6 +216,11 @@ GEM
     rb-fsevent (0.10.3)
     rb-inotify (0.9.10)
       ffi (>= 0.5.0, < 2)
+    redis (4.0.3)
+    redis-namespace (1.6.0)
+      redis (>= 3.0.4)
+    redis-objects (1.4.3)
+      redis (~> 4.0)
     rlp (0.7.3)
     rspec-core (3.8.0)
       rspec-support (~> 3.8.0)
@@ -239,6 +250,10 @@ GEM
     ruby-progressbar (1.10.0)
     ruby_dep (1.5.0)
     safe_yaml (1.0.4)
+    sidekiq (5.2.3)
+      connection_pool (~> 2.2, >= 2.2.2)
+      rack-protection (>= 1.5.0)
+      redis (>= 3.3.5, < 5)
     simplecov (0.16.1)
       docile (~> 1.1)
       json (>= 1.8, < 3)
@@ -295,11 +310,13 @@ DEPENDENCIES
   faraday (~> 0.15.2)
   google-protobuf (~> 3.6)
   health_check (~> 3.0)
+  hiredis (~> 0.6.1)
   kaminari (~> 1.1, >= 1.1.1)
   listen (>= 3.0.5, < 3.2)
   mina
   mina-multistage
   mina-puma
+  mina-sidekiq (~> 1.0, >= 1.0.3)
   oj (~> 3.6, >= 3.6.2)
   pg (>= 0.18, < 2.0)
   pry (= 0.11.3)
@@ -308,8 +325,12 @@ DEPENDENCIES
   rack-cors
   rails (~> 5.2.0)
   ransack (~> 2.0, >= 2.0.1)
+  redis (~> 4.0, >= 4.0.3)
+  redis-namespace (~> 1.6)
+  redis-objects (~> 1.4, >= 1.4.3)
   rspec-rails (~> 3.7)
   rubocop (~> 0.59)
+  sidekiq (~> 5.2, >= 5.2.3)
   simplecov
   spring
   spring-watcher-listen (~> 2.0.0)

app/controllers/concerns/split_requests_concern.rb

@@ -6,8 +6,6 @@ module SplitRequestsConcern
 
   # real time sync methods
   SYNC_METHODS = %w(
-    getBlockByNumber
-    getBlockByHash
     getTransaction
   ).freeze
 

app/models/block.rb

@@ -6,7 +6,7 @@ class Block < ApplicationRecord
   has_many :erc20_transfers
 
   # store_accessor :header, :number
-  store_accessor :body, :transactions
+  # store_accessor :body, :transactions
 
   validates :cita_hash, presence: true, uniqueness: true
   validates :block_number, presence: true, uniqueness: true

app/models/cita_sync/persist.rb

@@ -15,27 +15,33 @@ def save_blocks?
       # @param hex_num_str [String] block number in hex_num_str
       # @return [Block, SyncError] return SyncError if rpc return an error
       def save_block(hex_num_str)
-        data = CitaSync::Api.get_block_by_number(hex_num_str, true)
+        data = CitaSync::Api.get_block_by_number(hex_num_str, false)
         result = data["result"]
         error = data["error"]
 
         # handle error
-        return handle_error("getBlockByNumber", [hex_num_str, true], error) unless error.nil?
+        return handle_error("getBlockByNumber", [hex_num_str, false], error) unless error.nil?
 
         # handle for result.nil
         return nil if result.nil?
 
         block_number_hex_str = result.dig("header", "number")
         block_number = HexUtils.to_decimal(block_number_hex_str)
+        transaction_hashes = result.dig("body", "transactions")
         block = Block.new(
           version: result["version"],
           cita_hash: result["hash"],
           header: result["header"],
-          body: result["body"],
+          # body: result["body"],
           block_number: block_number,
-          transaction_count: result.dig("body", "transactions").count
+          transaction_count: transaction_hashes.count
         )
         block.save if save_blocks?
+
+        transaction_hashes.each do |hash|
+          SaveTransactionWorker.perform_async(hash)
+        end
+
         block
       end
 
@@ -53,9 +59,9 @@ def save_transaction(hash)
         return handle_error("getTransaction", [hash], error) unless error.nil?
         return nil if result.nil?
 
-        block = if save_blocks?
-                  Block.find_by(block_number: HexUtils.to_decimal(result["blockNumber"]))
-                end
+        block = nil
+        block = Block.find_by(block_number: HexUtils.to_decimal(result["blockNumber"])) if save_blocks?
+
         content = result["content"]
         message = Message.new(content)
         transaction = Transaction.new(
@@ -80,7 +86,7 @@ def save_transaction(hash)
           transaction.error_message = receipt_result["errorMessage"]
         end
         transaction.save
-        save_event_logs(receipt_result["logs"]) unless receipt_result.nil?
+        SaveEventLogsWorker.perform_async(receipt_result["logs"]) unless receipt_result.nil?
         transaction
       end
 
@@ -100,9 +106,7 @@ def save_event_logs(logs)
         event_logs = EventLog.create(attrs)
         # if event log is a registered ERC20 contract address, process it
         event_logs.each do |el|
-          if Erc20Transfer.exists?(address: el.address&.downcase) && Erc20Transfer.transfer?(el.topics)
-            Erc20Transfer.save_from_event_log(el)
-          end
+          SaveErc20TransferWorker.perform_async(el.id)
         end
       end
 
@@ -156,23 +160,6 @@ def save_abi(addr, block_number)
         [abi, data]
       end
 
-      # save one block with it's transactions and meta data
-      #
-      # @param block_number_hex_str [String] hex string with "0x" prefix
-      # @return [void]
-      def save_block_with_infos(block_number_hex_str)
-        # merge to one commit, can be faster
-        ApplicationRecord.transaction do
-          block = save_block(block_number_hex_str)
-          return if block.nil?
-
-          hashes = block.transactions.map { |t| t&.with_indifferent_access&.dig :hash }
-          hashes.each do |hash|
-            save_transaction(hash)
-          end
-        end
-      end
-
       # save blocks and there's transactions and meta data, from next db block to last block in chain
       #
       # @return [void]
@@ -185,7 +172,7 @@ def save_blocks_with_infos
         ((last_block_number + 1)..block_number).each do |num|
           hex_str = HexUtils.to_hex(num)
           ApplicationRecord.transaction do
-            save_block_with_infos(hex_str)
+            save_block(hex_str)
             SyncInfo.current_block_number = num
           end
         end
@@ -196,10 +183,7 @@ def save_blocks_with_infos
       # @return [void]
       def realtime_sync
         loop do
-          begin
-            save_blocks_with_infos
-          rescue
-          end
+          save_blocks_with_infos
         end
       end
 

app/workers/save_erc20_transfer_worker.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class SaveErc20TransferWorker
+  include Sidekiq::Worker
+
+  def perform(event_log_id)
+    event_log = EventLog.find_by(id: event_log_id)
+    return if event_log.nil?
+
+    if Erc20Transfer.exists?(address: event_log.address&.downcase) && Erc20Transfer.transfer?(event_log.topics)
+      Erc20Transfer.save_from_event_log(event_log)
+    end
+  end
+end

app/workers/save_event_logs_worker.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class SaveEventLogsWorker
+  include Sidekiq::Worker
+
+  def perform(logs)
+    CitaSync::Persist.save_event_logs(logs)
+  end
+end

app/workers/save_transaction_worker.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+class SaveTransactionWorker
+  include Sidekiq::Worker
+
+  def perform(hash)
+    CitaSync::Persist.save_transaction(hash)
+  end
+end

config/database.yml

@@ -21,7 +21,7 @@ default: &default
   # http://guides.rubyonrails.org/configuring.html#database-pooling
   host: <%= ENV.fetch("DB_HOST") { "localhost" } %>
   port: <%= ENV.fetch("DB_PORT") { "5432" } %>
-  pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
+  pool: <%= ENV.fetch("DB_POOL") { 64 } %>
   username: <%= ENV.fetch("DB_USERNAME") { "postgres" } %>
   password: <%= ENV.fetch("DB_PASSWORD") { "postgres" } %>
 

config/deploy.rb

@@ -5,6 +5,7 @@
 # require 'mina/rbenv'  # for rbenv support. (https://rbenv.org)
 require 'mina/rvm'    # for rvm support. (https://rvm.io)
 require 'mina/puma'
+require 'mina_sidekiq/tasks'
 
 # Basic settings:
 #   domain       - The hostname to SSH to.
@@ -42,7 +43,6 @@
 
 set :shared_files, fetch(:shared_files, []).push(
   'config/puma.rb',
-  'config/master.key',
   '.env.local'
 )
 
@@ -83,7 +83,9 @@
 
     on :launch do
       in_path(fetch(:current_path)) do
+        invoke :'sidekiq:quiet'
         invoke :'puma:phased_restart'
+        invoke :'sidekiq:restart'
       end
     end
   end

config/initializers/redis.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+require "redis"
+require "redis/objects"
+
+redis_config = Rails.application.config_for(:redis)
+
+$redis = Redis.new(url: redis_config["url"], driver: :hiredis, password: redis_config["password"])
+Redis::Objects.redis = $redis

config/initializers/sidekiq.rb

@@ -0,0 +1,12 @@
+redis_config = Rails.application.config_for(:redis)
+sidekiq_url = redis_config["url"]
+redis_password = redis_config["password"]
+
+namespace = ENV.fetch("REDIS_NAMESPACE") { "rebirth" }
+
+Sidekiq.configure_server do |config|
+  config.redis = { url: sidekiq_url, driver: :hiredis, password: redis_password, namespace: namespace }
+end
+Sidekiq.configure_client do |config|
+  config.redis = { url: sidekiq_url, driver: :hiredis, password: redis_password, namespace: namespace }
+end

config/redis.yml

@@ -0,0 +1,12 @@
+defaults: &defaults
+  url: <%= ENV.fetch("REDIS_URL") || "redis://localhost:6379/8" %>
+  password: <%= ENV["REDIS_PASSWORD"] %>
+
+development:
+  <<: *defaults
+
+test:
+  <<: *defaults
+
+production:
+  <<: *defaults

config/routes.rb

@@ -1,6 +1,9 @@
 Rails.application.routes.draw do
   # For details on the DSL available within this file, see http://guides.rubyonrails.org/routing.html
 
+  require "sidekiq/web"
+  mount Sidekiq::Web => "/sidekiq"
+
   # resources :cita, only: [:index]
   root to: "application#homepage"
   post "/", to: "cita#index"

config/sidekiq.yml

@@ -0,0 +1,6 @@
+---
+:concurrency: <%= ENV["SIDEKIQ_THREADS"]&.to_i || 25 %>
+:pidfile: tmp/pids/sidekiq.pid
+:logfile: log/sidekiq.log
+:queues:
+  - [default, 1]