Merge pull request #458 from district0x/syncer-fixes

Keep event count to avoid duplicates and syncer fixes

Author: ginesdt

server/src/district/server/web3_events.cljs

@@ -166,11 +166,10 @@
           (fn [err checkpoint]
             (web3-eth/get-block-number
               @web3
-              (fn [_err-block last-block-number]
+              (fn [err-block last-block-number]
+                (when (or err-block (nil? last-block-number))
+                  (throw (js/Error. "Failed to get current block number")))
                 (let [{:keys [last-processed-block processed-log-indexes]} checkpoint
-                      ; Current backtrack parameter is not considered (and therefore next-block-to-process neither)
-                      ; If want to use it, it needs to be made sure that event handlers are idempotent (i.e. can be run twice without the DB changing)
-                      ; This is not the case for FundsIn/FundsOut and some other messages in Ethlance for example
                       next-block-to-process (max 0 (- last-processed-block backtrack))]
                   (if skip-past-events-replay?
                     (do
@@ -179,7 +178,7 @@
                     (smart-contracts/replay-past-events-in-order
                       events
                       dispatch
-                      {:from-block (max last-processed-block from-block 0)
+                      {:from-block (max next-block-to-process from-block 0)
                        :crash-on-event-fail? crash-on-event-fail?
                        :skip-log-indexes processed-log-indexes
                        :to-block last-block-number

server/src/ethlance/server/db.cljs

@@ -645,6 +645,27 @@
                       :token-detail/decimals (:decimals token-details)}))))
 
 
+(defn get-last-event
+  [conn contract-key event-name]
+  (safe-go
+    (<? (db/get conn {:select [:event/last-log-index :event/last-block-number :event/count]
+                      :from [:Event]
+                      :where [:and
+                              [:= :event/contract-key contract-key]
+                              [:= :event/event-name event-name]]}))))
+
+
+(defn upsert-event!
+  [conn event]
+  (safe-go
+    (<?
+      (db/run! conn {:insert-into :Event
+                :values [(select-keys event (get-table-column-names :Event))]
+                :upsert (array-map
+                          :on-conflict [:event/event-name :event/contract-key]
+                          :do-update-set (keys event))}))))
+
+
 (def get-checkpoint-query
   {:select [:*]
    :from [:ContractEventCheckpoint]

server/src/ethlance/server/db/schema.cljs

@@ -414,4 +414,14 @@
     [[:id :serial]
      [:checkpoint :json]
      [:created-at :timestamp]
-     [(sql/call :primary-key :id)]]}])
+     [(sql/call :primary-key :id)]]}
+
+   {:table-name :Event
+    :table-columns
+    [[:event/contract-key :varchar not-nil]
+     [:event/event-name :varchar not-nil]
+     [:event/last-log-index :integer not-nil]
+     [:event/last-block-number :integer not-nil]
+     [:event/count :integer not-nil]
+     ;; PK
+     [(sql/call :primary-key :event/contract-key :event/event-name)]]}])

server/src/ethlance/server/syncer.cljs

@@ -9,9 +9,11 @@
     [district.server.async-db :as db]
     [district.server.smart-contracts :as smart-contracts]
     [district.time :as time]
+    [district.server.config :refer [config]]
     [district.server.web3 :refer [ping-start ping-stop web3]]
     [district.server.web3-events :as web3-events]
     [district.shared.async-helpers :refer [<? safe-go]]
+    [ethlance.server.db :as ethlance-db]
     [ethlance.server.event-replay-queue :as replay-queue]
     [ethlance.server.tracing.api :as t-api]
     [ethlance.server.syncer.handlers :as handlers]
@@ -24,8 +26,9 @@
 
 
 (defstate ^{:on-reload :noop} syncer
-  :start (start {})
-  :stop (stop syncer))
+  :start (start (merge (:syncer @config)
+                       (:syncer (mount/args))))
+  :stop (stop))
 
 
 ;;
@@ -60,6 +63,8 @@
       (async/go
         (let [contract-key (-> event :contract :contract-key)
               event-key (-> event :event)
+              event-name (name event-key)
+              log-index (-> event :log-index)
               handler (get contract-ev->handler [contract-key event-key])
               span (t-api/start-span (str (name (or contract-key "UnnamedContract")) "." (name (or event-key "UnnamedEvent"))))
               conn (<? (db/get-connection))]
@@ -72,21 +77,35 @@
                                                                 (if timestamp
                                                                   (bn/number timestamp)
                                                                   block-timestamp))))
-                  _ (db/begin-tx conn)
-                  res (t-api/with-span-context span #(handler conn err event))
-                  _ (db/commit-tx conn)]
-              (t-api/set-span-ok! span)
-              ;; Calling a handler can throw or return a go block (when using safe-go)
-              ;; in the case of async ones, the go block will return the js/Error.
-              ;; In either cases push the event to the queue, so it can be replayed later
-              (when (satisfies? ReadPort res)
-                (let [r (<! res)]
-                  (when (instance? js/Error r)
-                    (throw r))
+                  {:keys [:event/last-block-number :event/last-log-index :event/count]
+                   :or {last-block-number -1
+                        last-log-index -1
+                        count 0}} (<? (ethlance-db/get-last-event conn (name contract-key) event-name))]
+              (log/debug "Handling event..." event)
+              (if (or (> block-number last-block-number)
+                      (and (= block-number last-block-number) (> log-index last-log-index)))
+                (let [_ (db/begin-tx conn)
+                      res (t-api/with-span-context span #(handler conn err event))]
                   (t-api/set-span-ok! span)
-                  (t-api/end-span! span)
-                  (log/info "Syncer: OK" r)
-                  r)))
+                  ;; Calling a handler can throw or return a go block (when using safe-go)
+                  ;; in the case of async ones, the go block will return the js/Error.
+                  ;; In either cases push the event to the queue, so it can be replayed later
+                  (when (satisfies? ReadPort res)
+                    (let [r (<! res)]
+                      (when (instance? js/Error r)
+                        (throw r))
+                      (<? (ethlance-db/upsert-event! conn {:event/last-log-index log-index
+                                                           :event/last-block-number block-number
+                                                           :event/count (inc count)
+                                                           :event/event-name event-name
+                                                           :event/contract-key (name contract-key)}))
+                      (db/commit-tx conn)
+                      (log/info "Handled new event" event)
+                      (t-api/set-span-ok! span)
+                      (t-api/end-span! span)
+                      (log/info "Syncer: OK" r)
+                      r)))
+                (log/info "Skipping handling of a persisted event" event)))
             (catch js/Error error
               (log/error "Syncer: ERROR" error)
               (replay-queue/push-event conn event)
@@ -105,7 +124,7 @@
         (let [connected? (true? (<! (web3-eth/is-listening? @web3)))]
           (when connected?
             (do
-              (log/debug (str "disconnecting from provider to force reload. Last block: " @last-block-number))
+              (log/debug (str "disconnecting from provider to force reload"))
               (web3-core/disconnect @web3))))))
     interval))