Delegated data synchronization issues

[Voronar]

I've added two test cases that don't work for some reason. Maybe it's my fault (Willow API incorrect usage) or some bug in the Willow implementation.

The first test case is subsequent capability delegation.
And the second one is mutual capability delegation.
In both cases, entry synchronization doesn't work as I expect, they just stuck. In a successful scenario, tests should be normally terminated at the end of the corresponding test functions.

async fn delegate_path(
    path: Path,
    alfie: Peer,
    betty: Peer,
    user_alfie: UserId,
    user_betty: UserId,
    namespace_id: NamespaceId,
    session_mode: SessionMode,
) {
    let restricted_area = Area::new_path(path.clone());

    let cap_for_betty = alfie
        .delegate_caps(
            CapSelector::any(namespace_id),
            AccessMode::Read,
            DelegateTo::new(user_betty, RestrictArea::Restrict(restricted_area.clone())),
        )
        .await
        .unwrap();

    betty.import_caps(cap_for_betty).await.unwrap();

    let entry = EntryForm::new_bytes(namespace_id, path.clone(), "foo");

    alfie.insert_entry(entry, user_alfie).await.unwrap();

    let cap_selector = CapSelector::new(
        namespace_id,
        UserSelector::Any,
        AreaSelector::ContainsArea(restricted_area.clone()),
    );

    let interest = Interests::builder()
        .add_area(cap_selector.clone(), vec![restricted_area])
        .build();

    let init = SessionInit::new(interest, session_mode);

    let mut intent = betty.sync_with_peer(alfie.node_id(), init).await.unwrap();

    intent.complete().await.unwrap();

    intent.close().await;
}

#[tokio::test(flavor = "multi_thread")]
async fn subsequent_cap_delegations() {
    iroh_test::logging::setup_multithreaded();
    let mut rng = create_rng("subsequent_cap_delegations");

    let [alfie, betty] = spawn_two(&mut rng).await.unwrap();

    let user_alfie = alfie.create_user().await.unwrap();
    let user_betty = betty.create_user().await.unwrap();

    let namespace_id = alfie
        .create_namespace(NamespaceKind::Owned, user_alfie)
        .await
        .unwrap();

    let shared_path1 = Path::from_bytes(&[b"1"]).unwrap();
    let shared_path2 = Path::from_bytes(&[b"2"]).unwrap();

    tokio::spawn({
        let alfie = alfie.clone();
        let betty = betty.clone();

        async move {
            delegate_path(
                shared_path1,
                alfie,
                betty,
                user_alfie,
                user_betty,
                namespace_id,
                SessionMode::Continuous,
            )
            .await;
        }
    });

    loop {
        let entries: Vec<_> = betty
            .get_entries(namespace_id, Range3d::new_full())
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        tracing::info!("len should be 1, actual: {}", entries.len());

        if entries.len() == 1 {
            break;
        }

        sleep(Duration::from_secs(1)).await;
    }

    tokio::spawn({
        let alfie = alfie.clone();
        let betty = betty.clone();

        async move {
            delegate_path(
                shared_path2,
                alfie,
                betty,
                user_alfie,
                user_betty,
                namespace_id,
                SessionMode::Continuous,
            )
            .await;
        }
    });

    loop {
        let entries: Vec<_> = betty
            .get_entries(namespace_id, Range3d::new_full())
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        tracing::info!("len should be 2, actual: {}", entries.len());

        if entries.len() == 2 {
            break;
        }

        sleep(Duration::from_secs(1)).await;
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn mutual_cap_delegations() {
    iroh_test::logging::setup_multithreaded();
    let mut rng = create_rng("mutual_cap_delegations");

    let [alfie, betty] = spawn_two(&mut rng).await.unwrap();

    let user_alfie = alfie.create_user().await.unwrap();
    let user_betty = betty.create_user().await.unwrap();

    let namespace_id = alfie
        .create_namespace(NamespaceKind::Owned, user_alfie)
        .await
        .unwrap();

    let alfie_path = Path::from_bytes(&[b"a"]).unwrap();
    let betty_path = Path::from_bytes(&[b"b"]).unwrap();

    tokio::spawn({
        let alfie = alfie.clone();
        let betty = betty.clone();
        let alfie_path = alfie_path.clone();

        async move {
            delegate_path(
                alfie_path,
                alfie,
                betty,
                user_alfie,
                user_betty,
                namespace_id,
                SessionMode::Continuous,
            )
            .await;
        }
    });

    loop {
        let entries: Vec<_> = betty
            .get_entries(namespace_id, Range3d::new_full())
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        tracing::info!("alfie path delegated to betty: {:?}", entries);

        if entries.iter().any(|e| e.entry().path() == &alfie_path) {
            break;
        }

        sleep(Duration::from_secs(1)).await;
    }

    tokio::spawn({
        let alfie = alfie.clone();
        let betty = betty.clone();
        let betty_path = betty_path.clone();

        async move {
            delegate_path(
                betty_path,
                betty,
                alfie,
                user_betty,
                user_alfie,
                namespace_id,
                SessionMode::Continuous,
            )
            .await;
        }
    });

    loop {
        let entries: Vec<_> = alfie
            .get_entries(namespace_id, Range3d::new_full())
            .await
            .unwrap()
            .try_collect()
            .await
            .unwrap();

        tracing::info!("Betty path delegated to Alfie: {:?}", entries);

        if entries.iter().any(|e| e.entry().path() == &betty_path) {
            break;
        }

        sleep(Duration::from_secs(1)).await;
    }
}