diff --git a/src/pipewire/impl-link.c b/src/pipewire/impl-link.c
index e05c14050..b6e2ccd1b 100644
--- a/src/pipewire/impl-link.c
+++ b/src/pipewire/impl-link.c
@@ -52,6 +52,81 @@ struct impl {
 
 /** \endcond */
 
+static struct pw_node_peer *pw_node_peer_ref(struct pw_impl_node *onode, struct pw_impl_node *inode)
+{
+	struct pw_node_peer *peer;
+
+	spa_list_for_each(peer, &onode->peer_list, link) {
+		if (peer->target.node == inode) {
+			pw_log_debug("exiting peer %p from %p to %p", peer, onode, inode);
+			peer->ref++;
+			return peer;
+		}
+	}
+	peer = calloc(1, sizeof(*peer));
+	if (peer == NULL)
+		return NULL;
+
+	peer->ref = 1;
+	peer->output = onode;
+	peer->active_count = 0;
+	peer->target.node = inode;
+	peer->target.activation = inode->rt.activation;
+	peer->target.signal_func = inode->rt.target.signal_func;
+	peer->target.data = inode->rt.target.data;
+
+	spa_list_append(&onode->peer_list, &peer->link);
+	pw_log_debug("new peer %p from %p to %p", peer, onode, inode);
+	pw_impl_node_emit_peer_added(onode, inode);
+
+	return peer;
+}
+
+static void pw_node_peer_unref(struct pw_node_peer *peer)
+{
+	if (--peer->ref > 0)
+		return;
+	spa_list_remove(&peer->link);
+	pw_log_debug("remove peer %p from %p to %p", peer, peer->output, peer->target.node);
+	pw_impl_node_emit_peer_removed(peer->output, peer->target.node);
+	free(peer);
+}
+
+static void pw_node_peer_activate(struct pw_node_peer *peer)
+{
+	struct pw_node_activation_state *state;
+
+	state = &peer->target.activation->state[0];
+
+	if (peer->active_count++ == 0) {
+		spa_list_append(&peer->output->rt.target_list, &peer->target.link);
+		if (!peer->target.active && peer->output->rt.driver_target.node != NULL) {
+			state->required++;
+			peer->target.active = true;
+		}
+	}
+	pw_log_trace("%p: node:%p state:%p pending:%d/%d", peer->output,
+			peer->target.node, state, state->pending, state->required);
+}
+
+static void pw_node_peer_deactivate(struct pw_node_peer *peer)
+{
+	struct pw_node_activation_state *state;
+	state = &peer->target.activation->state[0];
+	if (--peer->active_count == 0) {
+
+		spa_list_remove(&peer->target.link);
+
+		if (peer->target.active) {
+			state->required--;
+			peer->target.active = false;
+		}
+	}
+	pw_log_trace("%p: node:%p state:%p pending:%d/%d", peer->output,
+			peer->target.node, state, state->pending, state->required);
+}
+
+
 static void info_changed(struct pw_impl_link *link)
 {
 	struct pw_resource *resource;
@@ -566,28 +641,15 @@ do_activate_link(struct spa_loop *loop,
 		 bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
 	struct pw_impl_link *this = user_data;
-	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
 
 	pw_log_trace("%p: activate", this);
 
 	spa_list_append(&this->output->rt.mix_list, &this->rt.out_mix.rt_link);
 	spa_list_append(&this->input->rt.mix_list, &this->rt.in_mix.rt_link);
 
-	if (impl->inode != impl->onode) {
-		struct pw_node_activation_state *state;
+	if (this->peer)
+		pw_node_peer_activate(this->peer);
 
-		this->rt.target.activation = impl->inode->rt.activation;
-		spa_list_append(&impl->onode->rt.target_list, &this->rt.target.link);
-
-		state = &this->rt.target.activation->state[0];
-		if (!this->rt.target.active && impl->onode->rt.driver_target.node != NULL) {
-			state->required++;
-			this->rt.target.active = true;
-		}
-
-		pw_log_trace("%p: node:%p state:%p pending:%d/%d", this, impl->inode,
-				state, state->pending, state->required);
-	}
 	return 0;
 }
 
@@ -782,26 +844,14 @@ do_deactivate_link(struct spa_loop *loop,
 		   bool async, uint32_t seq, const void *data, size_t size, void *user_data)
 {
         struct pw_impl_link *this = user_data;
-	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this);
 
 	pw_log_trace("%p: disable %p and %p", this, &this->rt.in_mix, &this->rt.out_mix);
 
 	spa_list_remove(&this->rt.out_mix.rt_link);
 	spa_list_remove(&this->rt.in_mix.rt_link);
 
-	if (impl->inode != impl->onode) {
-		struct pw_node_activation_state *state;
-
-		spa_list_remove(&this->rt.target.link);
-		state = &this->rt.target.activation->state[0];
-		if (this->rt.target.active) {
-			state->required--;
-			this->rt.target.active = false;
-		}
-
-		pw_log_trace("%p: node:%p state:%p pending:%d/%d", this, impl->inode,
-				state, state->pending, state->required);
-	}
+	if (this->peer)
+		pw_node_peer_deactivate(this->peer);
 
 	return 0;
 }
@@ -1252,9 +1302,6 @@ struct pw_impl_link *pw_context_create_link(struct pw_context *context,
 		impl->inode = input_node;
 	}
 
-	this->rt.target.signal_func = impl->inode->rt.target.signal_func;
-	this->rt.target.data = impl->inode->rt.target.data;
-
 	pw_log_debug("%p: constructed out:%p:%d.%d -> in:%p:%d.%d", impl,
 		     output_node, output->port_id, this->rt.out_mix.port.port_id,
 		     input_node, input->port_id, this->rt.in_mix.port.port_id);
@@ -1272,7 +1319,8 @@ struct pw_impl_link *pw_context_create_link(struct pw_context *context,
 	pw_impl_port_recalc_latency(output);
 	pw_impl_port_recalc_latency(input);
 
-	pw_impl_node_emit_peer_added(impl->onode, impl->inode);
+	if (impl->onode != impl->inode)
+		this->peer = pw_node_peer_ref(impl->onode, impl->inode);
 
 	return this;
 
@@ -1407,7 +1455,8 @@ void pw_impl_link_destroy(struct pw_impl_link *link)
 	if (link->registered)
 		spa_list_remove(&link->link);
 
-	pw_impl_node_emit_peer_removed(impl->onode, impl->inode);
+	if (link->peer)
+		pw_node_peer_unref(link->peer);
 
 	try_unlink_controls(impl, link->output, link->input);
 
diff --git a/src/pipewire/impl-node.c b/src/pipewire/impl-node.c
index c7144310d..b6327441e 100644
--- a/src/pipewire/impl-node.c
+++ b/src/pipewire/impl-node.c
@@ -1331,6 +1331,7 @@ struct pw_impl_node *pw_context_create_node(struct pw_context *context,
 	impl->pending_id = SPA_ID_INVALID;
 
 	spa_list_init(&this->follower_list);
+	spa_list_init(&this->peer_list);
 
 	spa_hook_list_init(&this->listener_list);
 
diff --git a/src/pipewire/private.h b/src/pipewire/private.h
index f57e1713c..48a221eb7 100644
--- a/src/pipewire/private.h
+++ b/src/pipewire/private.h
@@ -738,6 +738,8 @@ struct pw_impl_node {
 
 	struct spa_list sort_link;	/**< link used to sort nodes */
 
+	struct spa_list peer_list;	/* list of peers */
+
 	struct spa_node *node;		/**< SPA node implementation */
 	struct spa_hook listener;
 
@@ -909,6 +911,14 @@ struct pw_control_link {
 	unsigned int valid:1;
 };
 
+struct pw_node_peer {
+	int ref;
+	int active_count;
+	struct spa_list link;			/**< link in peer list */
+	struct pw_impl_node *output;		/**< the output node */
+	struct pw_node_target target;		/**< target of the input node */
+};
+
 #define pw_impl_link_emit(o,m,v,...) spa_hook_list_call(&o->listener_list, struct pw_impl_link_events, m, v, ##__VA_ARGS__)
 #define pw_impl_link_emit_destroy(l)		pw_impl_link_emit(l, destroy, 0)
 #define pw_impl_link_emit_free(l)		pw_impl_link_emit(l, free, 0)
@@ -940,10 +950,11 @@ struct pw_impl_link {
 	struct pw_control_link control;
 	struct pw_control_link notify;
 
+	struct pw_node_peer *peer;
+
 	struct {
 		struct pw_impl_port_mix out_mix;	/**< port added to the output mixer */
 		struct pw_impl_port_mix in_mix;		/**< port added to the input mixer */
-		struct pw_node_target target;		/**< target to trigger the input node */
 	} rt;
 
 	void *user_data;