diff --git a/gui/WarpGraphModel.cpp b/gui/WarpGraphModel.cpp
index b1bd3a5..6fcf210 100644
--- a/gui/WarpGraphModel.cpp
+++ b/gui/WarpGraphModel.cpp
@@ -90,6 +90,10 @@ bool WarpGraphModel::connectionPossible(
   if (connectionExists(connectionId)) {
     return false;
   }
+  if (m_ghostNodes.count(connectionId.outNodeId) ||
+      m_ghostNodes.count(connectionId.inNodeId)) {
+    return false;
+  }
 
   auto outIt = m_nodes.find(connectionId.outNodeId);
   auto inIt = m_nodes.find(connectionId.inNodeId);
@@ -106,6 +110,14 @@ bool WarpGraphModel::connectionPossible(
     return false;
   }
 
+  WarpNodeType outType = classifyNode(outIt->second.info);
+  WarpNodeType inType = classifyNode(inIt->second.info);
+  bool outIsVideo = (outType == WarpNodeType::kVideoSource || outType == WarpNodeType::kVideoSink);
+  bool inIsVideo = (inType == WarpNodeType::kVideoSource || inType == WarpNodeType::kVideoSink);
+  if (outIsVideo != inIsVideo) {
+    return false;
+  }
+
   return true;
 }
 
diff --git a/src/warppipe.cpp b/src/warppipe.cpp
index ab6b6e5..91ac6cc 100644
--- a/src/warppipe.cpp
+++ b/src/warppipe.cpp
@@ -1,6 +1,7 @@
 #include <algorithm>
 #include <atomic>
 #include <cerrno>
+#include <chrono>
 #include <cmath>
 #include <cstdlib>
 #include <cstring>
@@ -254,6 +255,7 @@ struct MeterStreamData {
   spa_hook listener{};
   std::atomic<float> peak_left{0.0f};
   std::atomic<float> peak_right{0.0f};
+  std::atomic<uint64_t> last_update_ms{0};
 };
 
 void NodeMeterProcess(void* data) {
@@ -261,31 +263,41 @@ void NodeMeterProcess(void* data) {
   if (!meter || !meter->stream) {
     return;
   }
-  pw_buffer* buf = pw_stream_dequeue_buffer(meter->stream);
-  if (!buf || !buf->buffer || buf->buffer->n_datas == 0) {
-    if (buf) {
-      pw_stream_queue_buffer(meter->stream, buf);
-    }
-    return;
-  }
-  spa_data* d = &buf->buffer->datas[0];
-  if (!d->data || !d->chunk) {
-    pw_stream_queue_buffer(meter->stream, buf);
-    return;
-  }
-  const float* samples = static_cast<const float*>(d->data);
-  uint32_t count = d->chunk->size / sizeof(float);
   float left = 0.0f;
   float right = 0.0f;
-  for (uint32_t i = 0; i + 1 < count; i += 2) {
-    float l = std::fabs(samples[i]);
-    float r = std::fabs(samples[i + 1]);
-    if (l > left) left = l;
-    if (r > right) right = r;
+  bool had_data = false;
+  pw_buffer* buf = nullptr;
+  while ((buf = pw_stream_dequeue_buffer(meter->stream)) != nullptr) {
+    if (!buf->buffer || buf->buffer->n_datas == 0) {
+      pw_stream_queue_buffer(meter->stream, buf);
+      continue;
+    }
+    spa_data* d = &buf->buffer->datas[0];
+    if (!d->data || !d->chunk) {
+      pw_stream_queue_buffer(meter->stream, buf);
+      continue;
+    }
+    const float* samples = static_cast<const float*>(d->data);
+    uint32_t count = d->chunk->size / sizeof(float);
+    left = 0.0f;
+    right = 0.0f;
+    for (uint32_t i = 0; i + 1 < count; i += 2) {
+      float l = std::fabs(samples[i]);
+      float r = std::fabs(samples[i + 1]);
+      if (l > left) left = l;
+      if (r > right) right = r;
+    }
+    had_data = true;
+    pw_stream_queue_buffer(meter->stream, buf);
+  }
+  if (had_data) {
+    meter->peak_left.store(left, std::memory_order_relaxed);
+    meter->peak_right.store(right, std::memory_order_relaxed);
+    auto now = std::chrono::steady_clock::now().time_since_epoch();
+    meter->last_update_ms.store(
+        static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(now).count()),
+        std::memory_order_relaxed);
   }
-  meter->peak_left.store(left, std::memory_order_relaxed);
-  meter->peak_right.store(right, std::memory_order_relaxed);
-  pw_stream_queue_buffer(meter->stream, buf);
 }
 
 static const pw_stream_events kNodeMeterEvents = {
@@ -2141,6 +2153,14 @@ Result<MeterState> Client::NodeMeterPeak(NodeId node) const {
   std::lock_guard<std::mutex> lock(impl_->cache_mutex);
   auto live_it = impl_->live_meters.find(node.value);
   if (live_it != impl_->live_meters.end() && live_it->second) {
+    auto now_ms = static_cast<uint64_t>(
+        std::chrono::duration_cast<std::chrono::milliseconds>(
+            std::chrono::steady_clock::now().time_since_epoch())
+            .count());
+    auto last = live_it->second->last_update_ms.load(std::memory_order_relaxed);
+    if (last > 0 && (now_ms - last) > 200) {
+      return {Status::Ok(), MeterState{0.0f, 0.0f}};
+    }
     MeterState state;
     state.peak_left = live_it->second->peak_left.load(std::memory_order_relaxed);
     state.peak_right = live_it->second->peak_right.load(std::memory_order_relaxed);
@@ -2156,6 +2176,14 @@ Result<MeterState> Client::NodeMeterPeak(NodeId node) const {
 Result<MeterState> Client::MeterPeak() const {
   std::lock_guard<std::mutex> lock(impl_->cache_mutex);
   if (impl_->master_meter_data) {
+    auto now_ms = static_cast<uint64_t>(
+        std::chrono::duration_cast<std::chrono::milliseconds>(
+            std::chrono::steady_clock::now().time_since_epoch())
+            .count());
+    auto last = impl_->master_meter_data->last_update_ms.load(std::memory_order_relaxed);
+    if (last > 0 && (now_ms - last) > 200) {
+      return {Status::Ok(), MeterState{0.0f, 0.0f}};
+    }
     MeterState state;
     state.peak_left = impl_->master_meter_data->peak_left.load(std::memory_order_relaxed);
     state.peak_right = impl_->master_meter_data->peak_right.load(std::memory_order_relaxed);
@@ -2367,8 +2395,12 @@ Status Client::RemoveLink(LinkId link) {
     auto it = impl_->link_proxies.find(link.value);
     if (it != impl_->link_proxies.end()) {
       if (it->second && it->second->proxy) {
+        spa_hook_remove(&it->second->listener);
         pw_proxy_destroy(it->second->proxy);
       }
+      if (impl_->registry) {
+        pw_registry_destroy(impl_->registry, link.value);
+      }
       impl_->link_proxies.erase(it);
       auto link_it2 = impl_->links.find(link.value);
       if (link_it2 != impl_->links.end()) {