     (let [msg (ready replica-version src-peer-id short-id)
           len (sz/serialize control-buffer 0 msg)
           ret (.offer ^Publication publication control-buffer 0 len)]
+      (assert-not-closed ret)
       (debug "Offered ready message:" [ret msg :session-id (.sessionId publication) :site site])
       ret))
   (segment-encoder [this]

           [java.util.concurrent.atomic AtomicLong]
           [org.agrona.concurrent UnsafeBuffer]
           [org.agrona ErrorHandler]))

(deftype Publisher [peer-config src-peer-id dst-task-id slot-id site 
                    ^UnsafeBuffer buffer ^UnsafeBuffer control-buffer base-encoder segment-encoder


          (>= (endpoint-status/min-endpoint-epoch status-mon) endpoint-epoch)
          (let [ret (.offer ^Publication publication ^UnsafeBuffer buf 0 length)]
            (if (pos? ret) 
              (.addAndGet written-bytes length)
              (assert-not-closed ret))
            ret)

          :else
