   {:pre [(#{:active :inactive} topology-initial-status)]}                
   (let [storm-cluster-state (:storm-cluster-state nimbus)
         conf (:conf nimbus)
-        storm-conf (read-storm-conf conf storm-id)
+        storage (:storage nimbus)
+        storm-conf (read-storm-conf storage conf storm-id)
         topology (system-topology! storm-conf (read-storm-topology conf storm-id))
         num-executors (->> (all-components topology) (map-val num-start-executors))]
     (log-message "Activating " storm-name ": " storm-id)

  (fn [kill-time]
    (let [delay (if kill-time
                  kill-time
                  (get (read-storm-conf (:conf nimbus) storm-id)
                       TOPOLOGY-MESSAGE-TIMEOUT-SECS))]
      (delay-event nimbus
                   storm-id

(defn inbox [nimbus]
  (master-inbox (:conf nimbus)))

(defn- read-storm-conf [storage conf storm-id]
  (let [stormroot (nimbus-storage-stormdist-root storm-id)]
    (merge conf
           (Utils/deserialize
            (IOUtils/toByteArray
             (.open storage (master-stormconf-path stormroot)))))))

(defn set-topology-status! [nimbus storm-id status]
  (let [storm-cluster-state (:storm-cluster-state nimbus)]
