   (when-not (or (= 1 (:onyx/max-peers task-map))
                 (= 1 (:onyx/n-peers task-map)))
     (throw (ex-info "Read log tasks must set :onyx/max-peers 1" task-map)))
+  (s/validate DatomicReadLogTaskMap task-map)
+
   (let [start-tx (:datomic/log-start-tx task-map)
         max-tx (:datomic/log-end-tx task-map)
         {:keys [read-ch shutdown-ch commit-ch]} pipeline

;; output plugins

(defn inject-write-tx-resources
  [{:keys [onyx.core/pipeline]} lifecycle]
  {:datomic/conn (:conn pipeline)})

(defn inject-write-bulk-tx-resources

            [onyx.static.default-vals :refer [defaults]]
            [onyx.static.uuid :refer [random-uuid]]
            [onyx.extensions :as extensions]
            [schema.core :as s]
            [onyx.schema :as os]
            [taoensso.timbre :refer [info debug fatal]]))

;;;;;;;;;;;;;;
;;;;;;;;;;;;;;
;; task schemas

(def UserTaskMapKey
  (os/build-allowed-key-ns :datomic))

(def DatomicReadLogTaskMap
  (s/->Both [os/TaskMap 
             {:datomic/uri s/Str
              (s/optional-key :datomic/log-start-tx) s/Int
              (s/optional-key :datomic/log-end-tx) s/Int
              :checkpoint/force-reset? s/Bool
              (s/optional-key :onyx/max-peers) (s/enum 1)
              (s/optional-key :onyx/n-peers) (s/enum 1)
              UserTaskMapKey s/Any}]))

(def DatomicReadDatomsTaskMap
  (s/->Both [os/TaskMap 
             {:datomic/uri s/Str
              :datomic/t s/Int
              :datomic/datoms-index s/Keyword
              :datomic/datoms-components [s/Any]
              :datomic/datoms-per-segment s/Int
              (s/optional-key :onyx/max-peers) (s/enum 1)
              (s/optional-key :onyx/n-peers) (s/enum 1)
              UserTaskMapKey s/Any}]))


(def DatomicReadIndexRangeTaskMap
  (s/->Both [os/TaskMap 
             {:datomic/uri s/Str
              :datomic/t s/Int
              :datomic/index-attribute s/Any
              :datomic/index-range-start s/Any
              :datomic/index-range-end s/Any
              :datomic/datoms-per-segment s/Int
              (s/optional-key :onyx/max-peers) (s/enum 1)
              (s/optional-key :onyx/n-peers) (s/enum 1)
              UserTaskMapKey s/Any}]))

(def DatomicWriteDatomsTaskMap
  (s/->Both [os/TaskMap 
             {:datomic/uri db-uri
              (s/optional-key :datomic/partition) (s/either s/Int s/Keyword)
              (s/optional-key :onyx/max-peers) (s/enum 1)
              (s/optional-key :onyx/n-peers) (s/enum 1)
              UserTaskMapKey s/Any}]))

;;; Helpers

(defn safe-connect [task-map]
  (if-let [uri (:datomic/uri task-map)]
