-async fn create(&self, _topic: &str) -> Result<()> {
-    self.create_with_retention_policy(_topic, RetentionPolicy::Limits)
-        .await
-}
-...
 async fn create_with_retention_policy_and_max_age(
     &self,
     _topic: &str,
     _retention_policy: RetentionPolicy,
     _max_age: std::time::Duration,
 ) -> Result<()> {
-    todo!()
+    Ok(())
 }
-...
+
 async fn publish(&self, _topic: &str, _value: Bytes) -> Result<()> {
-    todo!()
+    Ok(())
 }
-...
+
 async fn consume(
     &self,
     _topic: &str,
     _deliver_policy: Option<DeliverPolicy>,
 ) -> Result<Arc<mpsc::Receiver<super::Message>>> {
-    todo!()
+    let (_tx, rx) = mpsc::channel(1);
+    Ok(Arc::new(rx))
 }

 async fn consume(
     &self,
     topic: &str,
     deliver_policy: Option<queue::DeliverPolicy>,
 ) -> Result<Arc<mpsc::Receiver<super::Message>>> {
     let (tx, rx) = mpsc::channel(1024);
     let stream_name = format!("{}{}", self.prefix, format_key(topic));
-    let consumer_name = self.consumer_name.clone();
+    // Make consumer name unique per stream/topic to avoid collisions
+    let consumer_name = format!("{}_{}", stream_name, self.consumer_name);
     let is_durable = self.is_durable;
 
     {
         let client = get_nats_client().await.clone();
         let jetstream = jetstream::new(client);
         let stream = jetstream.get_stream(&stream_name).await.map_err(|e| {
             log::error!("Failed to get nats stream {}: {}", stream_name, e);
             Error::Message(format!("Failed to get nats stream {}: {}", stream_name, e))
         })?;
         let config = jetstream::consumer::pull::Config {
-            name: Some(consumer_name.to_string()),
+            name: Some(consumer_name.clone()),
             durable_name: if is_durable {
-                Some(consumer_name.to_string())
+                Some(consumer_name.clone())
             } else {
                 None
             },
             deliver_policy: get_deliver_policy(deliver_policy),
             ..Default::default()
         };
-        let consumer = stream
-            .get_or_create_consumer(&consumer_name, config)
-            .await
-            .map_err(|e| {
-                log::error!(
-                    "Failed to get_or_create_consumer for nats stream {}: {}",
-                    stream_name,
-                    e
-                );
-                Error::Message(format!(
-                    "Failed to get_or_create_consumer for nats stream {}: {}",
-                    stream_name, e
-                ))
-            })?;
+        let consumer = stream.get_or_create_consumer(&consumer_name, config).await.map_err(|e| {
+            log::error!(
+                "Failed to get_or_create_consumer for nats stream {}: {}",
+                stream_name,
+                e
+            );
+            Error::Message(format!(
+                "Failed to get_or_create_consumer for nats stream {}: {}",
+                stream_name, e
+            ))
+        })?;
         tokio::spawn(async move {
             loop {
                 let Ok(mut messages) = consumer.messages().await else {
-                    log::error!(
-                        "Failed to get nats consumer messages for stream {}",
-                        stream_name
-                    );
+                    log::error!("Failed to get nats consumer messages for stream {}", stream_name);
                     break;
                 };
                 let message = match messages.try_next().await {
                     Ok(Some(message)) => message,
                     Ok(None) => {
-                        log::warn!(
-                            "Nats consumer messages channel closed for stream {}",
-                            stream_name
-                        );
+                        log::warn!("Nats consumer messages channel closed for stream {}", stream_name);
                         break;
                     }
                     Err(e) => {
-                        log::error!(
-                            "Failed to get nats consumer messages for stream {}: {}",
-                            stream_name,
-                            e
-                        );
+                        log::error!("Failed to get nats consumer messages for stream {}: {}", stream_name, e);
                         break;
                     }
                 };
                 let message = super::Message::Nats(message);
-                tx.send(message).await.map_err(|e| {
+                if let Err(e) = tx.send(message).await {
                     log::error!("Failed to send message to channel: {}", e);
-                    Error::Message(format!("Failed to send message to channel: {}", e))
-                })?;
+                    break;
+                }
             }
-            Ok::<(), Error>(())
         });
     }
 
     Ok(Arc::new(rx))
 }

 async fn subscribe(tx: mpsc::Sender<CoordinatorEvent>) -> Result<()> {
     let queue = queue::get_queue().await;
     let mut reconnect = false;
     loop {
         if config::cluster::is_offline() {
             break;
         }
 
+        // Avoid full replay on reconnect; resume from last/new
         let deliver_policy = if reconnect {
-            queue::DeliverPolicy::All
+            queue::DeliverPolicy::Last
         } else {
             queue::DeliverPolicy::New
         };
         log::info!(
             "[COORDINATOR::EVENTS] subscribing to coordinator topic with deliver policy: {:?}",
             deliver_policy
         );
         let mut receiver: Arc<mpsc::Receiver<queue::Message>> = match queue
             .consume(COORDINATOR_STREAM, Some(deliver_policy))
             .await
         {
             Ok(receiver) => receiver,
             Err(e) => {
                 log::error!("[COORDINATOR::EVENTS] failed to subscribe to coordinator topic: {e}");
                 tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
                 reconnect = true;
                 continue;
             }
         };
+        // ... rest unchanged ...