Lattice bimorphism emits every item twice when new data arrives in the same tick.

[rohitkulshreshtha]

Lattice bimorphism emits every item twice - once when it matches from the LHS -> RHS and then again when it matches from RHS -> LHS.

The lattice_reduce in the lattice dimorphism tests is probably masking this issue. The following test recreates it:

 #[test]
    pub fn test_cartesian_product() {

        tracing_subscriber::fmt::init();

        let (out_send, out_recv) = hydroflow::util::unbounded_channel::<_>();

        let mut df = hydroflow_syntax! {
        lhs = source_iter_delta(0..1)
            -> map(SetUnionSingletonSet::new_from)
            -> state::<'static, SetUnionHashSet<u32>>();
        rhs = source_iter_delta(1..2)
            -> map(SetUnionSingletonSet::new_from)
            -> state::<'static, SetUnionHashSet<u32>>();

        lhs -> [0]my_join;
        rhs -> [1]my_join;

        my_join = lattice_bimorphism(CartesianProductBimorphism::<HashSet<_>>::default(), #lhs, #rhs)
            //-> lattice_reduce()
            -> for_each(|x| out_send.send(x).unwrap());
    };

        df.run_available();

        assert_eq!(
            &[SetUnionHashSet::new(HashSet::from_iter([
                (0, 1),
            ]))],
            &*collect_ready::<Vec<_>, _>(out_recv)
        );
    }

The test fails because out_recv contains SetUnion twice - once when it is pulled from LHS and then again when it is pulled from RHS.