 
 
 
+; publish all out channels before wiring
 (defn wire-stage [peer {:keys [chans] :as stage}]
   (go (if chans stage
-          (let [[in out] [(chan) (chan)]]
-            (<! (wire peer [in out]))
-            (assoc stage :chans [in out])))))
-
-
+          (let [log (atom {})
+                in (debug/chan log ["STAGE" :in])
+                out (debug/chan log ["STAGE" :out])]
+            (<! (wire peer [in (async/pub out :type)]))
+            (assoc stage :chans [(async/pub in :type) out])))))
+
+;; push perspective
 (defn sync! [{:keys [type author chans new-values meta] :as stage}]
-  (go (let [[in out] chans
-            p (async/pub in :type)
+  (go (let [[p out] chans
             fch (chan)
+            pch (chan)
             repo (:id meta)]
 
         (async/sub p :fetch fch)
+        (async/sub p :meta-pubed pch)
         (case type
           :meta-pub  (>! out {:type :meta-pub :user author :meta meta})
           :meta-sub (do
                       (>! out {:type :meta-sub :metas {author #{repo}}})
                       (>! out {:type :meta-pub :user author :meta meta})))
 
-        (let [to-fetch (select-keys new-values (:ids (<! fch)))]
-          (>! out {:type :fetched
-                   :user author :repo repo
-                   :values to-fetch}))
+        (go (let [to-fetch (select-keys new-values (:ids (<! fch)))]
+              (>! out {:type :fetched
+                       :user author :repo repo
+                       :values to-fetch})))
+
+        (let [m (alt! pch (timeout 10000))]
+          (when-not m
+            (throw (IllegalStateException. (str "No meta-pubed ack received for" meta)))))
+        (async/unsub p :fetch fch)
         (async/unsub p :fetch fch)
 
         (dissoc stage :type :new-values))))

  ([peer author repo schema]
     (load-stage! peer author repo schema (chan) (chan)))
  ([peer author repo schema [in out]]
     (go (<! (wire peer [in out]))
         {:author author
          :schema schema
          :meta (<! (-get-in (:store (:volatile peer)) [author repo]))

  (def peer (client-peer "CLIENT" (new-store)))
  (require '[clojure.core.incubator :refer [dissoc-in]])
  (dissoc-in @peer [:volatile :log])
  (clojure.pprint/pprint (get-in @peer [:volatile :log]))

  (go (let [printfn (partial println "STAGE:")
            stage (repo/new-repository "me@mail.com"
                                       {:type "s" :version 1}
                                       "Testing."
                                       false
                                       {:some 42})
            stage (<! (wire-stage peer stage))
            out (second (:chans stage))
            _ (>! out {:type :meta-sub
                       :metas {"me@mail.com" #{(:id (:meta stage))}}})
            _ (<! (timeout 1000))
            _ (>! out {:type :connect
                       :ip4 "127.0.0.1"
                       :port 9090})
            _ (<! (timeout 1000))
            new-stage (<! (sync! stage))]
        (-> new-stage
