Function valve 

pub fn valve(
    core: &Core,
    binding: &Arc<dyn ProducerBinding>,
    source: NodeId,
    control: NodeId,
    gate_fn_id: FnId,
    cancel: Option<CancellationToken>,
) -> NodeId

Boolean-gated passthrough. Subscribes to both source and control.

• Control DATA: evaluates binding.predicate_each(gate_fn_id, &[handle]) to determine gate state. true = open, false = closed. On transition from open to closed, if cancel is Some, invokes cancel.cancel().
• Source DATA: if gate is open, retains + emits. If closed, drops the handle (source DATA while gate is closed is silently discarded).
• Source COMPLETE/ERROR: forwarded unchanged.
• Control COMPLETE: does NOT auto-complete the valve (per TS completeWhenDepsComplete: false equivalent).
• Control ERROR: terminates the valve with ERROR.