You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: source.rs
    path of file: ./repos/holochain/crates/kitsune_p2p/fetch/src
    
    the code of the file until where you have to start completion: use indexmap::IndexSet;
use kitsune_p2p_types::KAgent;
use std::{default, ops::Deref};

use crate::{backoff::FetchBackoff, FetchConfig};

/// The number of times to probe a source between backoff attempts. This needs to be enough to reasonably allow
/// a source which might be slow to respond to one or two requests to respond to at least one of the probes but
/// not so high that it wastes time for this node trying to talk to a source that is not responding.
const NUM_PROBE_ATTEMPTS: u32 = 10;

/// A source to fetch from: either a node, or an agent on a node
#[derive(Debug, Clone, Par
    
    async fn source_state_single_backoff_then_recover() {
        let mut source_state: SourceState = Default::default();
        let config = Arc::new(TestFetchConfig(1, 1));

        assert!(source_state.should_use());

        // Exhaust the retries
        for _ in 0..=config.source_unavailable_timeout_threshold() {
            // The source should continue being used even with timeouts. It's only when we hit the limit that it shouldn't.
            assert!(source_state.should_use());

            // The check should keep passing
            source_state.is_valid(config.clone());

            // Record another timeout
            source_state.record_timeout();
        }

        // The source is still ready because it hasn't been checked
        assert!(source_state.should_use());

        // Now it goes into a backoff state
        source_state.is_valid(config.clone());
        assert!(!source_state.should_use());

        tokio::time::advance(Duration::from_secs(2)).await;

        // Now the backoff should go into a ready state and permit a number of probes
        for _ in 0..NUM_PROBE_ATTEMPTS {
            assert!(source_state.should_use());
        }

        // The probes have all been used and the backoff should be waiting again
        assert!(!source_state.should_use());

        // Now get a single successful response
        source_state.record_response();

        // Go back to a ready state
        assert!(source_state.should_use());
    }