 
 
 
+; 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))))

                                (async/into [])
                                <!
                                (zipmap ids))]
               #_(println "TO-FETCH" (:ip @peer) fetched)
               (>! out {:type :fetched
                        :user user :repo repo
                        :values fetched})
               (recur (<! fetch-ch))))))


(defn wire [peer [out in]]
  (go (let [{:keys [store chans log]} (:volatile @peer)
            ip (:ip @peer)
            [bus-in bus-out] chans
            p (async/pub in :type)
            pub-ch (debug/chan log [ip :pub])
            conn-ch (debug/chan log [ip :conn])
            sub-ch (debug/chan log [ip :sub])

                   [c-in c-out] [(debug/chan log [(str ip4 ":" port) :in])
                                 (debug/chan log [(str ip4 ":" port) :out])]
                   subs (:meta-sub @peer)
                   subed-ch (chan)]
               (client-connect! ip4 port c-in c-out)

                                        ; handshake
               (>! c-out {:type :meta-sub :metas subs})
               (put! subed-ch (<! c-in))
               (subscribe peer subed-ch bus-out c-out)

               (wire peer [c-out (async/pub c-in :type)])
               (recur (<! conn-ch))))))


