                 (seq package_inventory) (update :package_inventory distinct))))))
 
 (defn exec-replace-facts
-  [{:keys [payload id received] :as command} start-time db]
+  [{:keys [payload id received] :as command} start-time db conn-status]
   (let [{:keys [certname producer_timestamp]} payload]
-    (jdbc/with-transacted-connection' db :repeatable-read
-      (scf-storage/maybe-activate-node! certname producer_timestamp)
-      (scf-storage/replace-facts! payload))
+    (jdbc/retry-with-monitored-connection
+     db conn-status :repeatable-read
+     (fn []
+       (scf-storage/maybe-activate-node! certname producer_timestamp)
+       (scf-storage/replace-facts! payload)))
     (log-command-processed-messsage id received start-time :replace-facts certname)))
 
 ;; Node deactivation

    "configure expiration" (prep-configure-expiration cmd)
    "replace catalog inputs" (prep-replace-catalog-inputs cmd)))

(def supported-command?
  (comp (kitchensink/valset command-names) :command))

(defn supported-version? [command version]
  (contains? (get supported-command-versions command #{}) version))

(defn supported-command-version? [command-name [received-command-name received-version]]
  (boolean
   (and (= command-name received-command-name)
        (supported-version? command-name received-version))))

(defn exec-command
  "Takes a command object and processes it to completion. Dispatch is
   based on the command's name and version information"
  [{command-name :command version :version delete? :delete? :as command}
   db conn-status start]
  (condp supported-command-version? [command-name version]
    "replace catalog" (exec-replace-catalog command start db)
    "replace facts" (exec-replace-facts command start db)
    "store report" (exec-store-report command start db conn-status)
    "deactivate node" (exec-deactivate-node command start db)
    "configure expiration" (exec-configure-expiration command start db)
    "replace catalog inputs" (exec-replace-catalog-inputs command start db)))

(defn process-command!
  ;; only used by testing...

(defn prep-replace-catalog-inputs [command]
  (update-in command [:payload :producer_timestamp] #(or % (now))))

(defn exec-replace-catalog-inputs [{:keys [id received payload]} start-time db conn-status]
  (let [{:keys [certname inputs catalog_uuid]
         stamp :producer_timestamp} payload]
    (when (seq inputs)
      (jdbc/retry-with-monitored-connection
       db conn-status :repeatable-read
       (fn []
         (scf-storage/maybe-activate-node! certname stamp)
         (scf-storage/replace-catalog-inputs! certname catalog_uuid inputs stamp)))
      (log-command-processed-messsage id received start-time
                                      :replace-catalog-inputs certname))))

