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

    (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)]
      (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]))

(defn assert-not-closed [ret]
  (if (= ret (Publication/CLOSED))
    (throw (Exception. "Offered to a closed publication. Rebooting."))))

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