heddle-thread-api 0.25.4

Native Thread clients and durable peer replication over Iroh
Documentation
use crypto::{Ed25519Signer, Signer};
use objects::object::{StateId, thread_replication::ThreadGenesis};

use super::*;

#[test]
fn partial_fetch_requires_explicit_selection_and_matching_completion() {
    let (mut open, mut ready, endpoint, _) = fixture();
    ready.full_closure_available = false;
    assert!(
        Validation::new(
            open.clone(),
            ready.clone(),
            Some(&endpoint),
            Limits::default()
        )
        .is_err(),
        "an endpoint cannot silently weaken a complete request"
    );
    open.selection.as_mut().expect("selection").allow_partial = true;
    let mut validation = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
        .expect("explicit partial request");
    validation.artifact = ready.packs.len();
    let complete = |coverage| FetchServerFrame {
        body: Some(fetch_server_frame::Body::Complete(FetchComplete {
            revision: ready.current.clone(),
            checkpoint: ready.checkpoint.clone(),
            closure: coverage,
            ..Default::default()
        })),
    };
    assert!(
        validation
            .accept(complete(Coverage::Complete as i32))
            .is_err(),
        "partial Ready cannot produce a full availability receipt"
    );
    assert!(matches!(
        validation
            .accept(complete(Coverage::Partial as i32))
            .expect("matching partial completion"),
        Item::Complete(_)
    ));
}

pub(super) fn fixture() -> (FetchOpen, TransferReady, EndpointRef, [Vec<u8>; 2]) {
    let signer = Ed25519Signer::from_seed(&[61; 32]).expect("test creator");
    let spool_id = uuid::Uuid::from_u128(0x01980000000070008000000000000001);
    let genesis = ThreadGenesis {
        owner: objects::object::thread_replication::GenesisOwner::LocalKey(
            signer.public_key().try_into().expect("public key"),
        ),
        version: 1,
        spool: spool_id.to_string(),
        parent: None,
        base: StateId::from_bytes([17; 32]),
        name: "download".into(),
        intent: "source".into(),
        creator: signer.public_key().try_into().expect("public key"),
        nonce: vec![],
    };
    let thread = ThreadRef {
        spool: Some(SpoolRef {
            id: spool_id.to_string(),
        }),
        id: Some(ThreadId {
            value: genesis.id().expect("identity").as_bytes().to_vec(),
        }),
    };
    let revision = RevisionRef {
        spool: thread.spool.clone(),
        revision: Some(revision_ref::Revision::State(
            api::heddle::api::common::StateId {
                value: [31; 32].to_vec(),
            },
        )),
    };
    let endpoint = EndpointRef {
        public_key: vec![42; 32],
        kind: EndpointKind::Weft as i32,
    };
    let artifacts = [
        b"independently verified pack bytes".to_vec(),
        b"independently verified index bytes".to_vec(),
    ];
    let packs = artifacts
        .iter()
        .enumerate()
        .map(|(i, bytes)| {
            let address = ObjectAddress {
                algorithm: "blake3".into(),
                digest: blake3::hash(bytes).as_bytes().to_vec(),
            };
            PackExtent {
                pack: Some(address.clone()),
                kind: if i == 0 {
                    pack_extent::Kind::NativePack
                } else {
                    pack_extent::Kind::NativeIndex
                } as i32,
                offset: 0,
                length: bytes.len() as u64,
                extent_digest: Some(address),
            }
        })
        .collect();
    let open = FetchOpen {
        thread: Some(thread.clone()),
        revision: Some(revision.clone()),
        selection: Some(TransferSelection {
            facets: vec![SharedFacet::Source as i32],
            ..Default::default()
        }),
        ..Default::default()
    };
    let ready = TransferReady {
        endpoint: Some(endpoint.clone()),
        thread: Some(thread),
        current: Some(revision),
        owner_genesis: Some(
            repo::sign_spool_owner_genesis(&signer, *spool_id.as_bytes()).expect("owner signature"),
        ),
        ownership: Some(OwnerState::default()),
        thread_genesis: Some(ThreadGenesisRecord {
            boundary_acceptances: Vec::new(),
            ownership_claims: vec![],
            ownership_claim_admissions: vec![],
            ownership_resolutions: vec![],
            ownership_resolution_admissions: vec![],
            genesis: Some(
                replication::opening::sign_genesis(&genesis, &signer).expect("creator signature"),
            ),
            creator_authority: vec![],
            admission: None,
        }),
        packs,
        checkpoint: Some(TransferCheckpoint {
            transfer_id: vec![9; 16],
            plan_digest: vec![5; 32],
            ..Default::default()
        }),
        budget: Some(ReadBudget {
            max_items: 128,
            max_frame_bytes: 16 * 1024,
            max_snapshot_bytes: 1024 * 1024,
        }),
        full_closure_available: true,
        ..Default::default()
    };
    (open, ready, endpoint, artifacts)
}
fn chunk(ready: &TransferReady, artifact: usize, data: Vec<u8>) -> FetchServerFrame {
    let mut extent = ready.packs[artifact].clone();
    extent.length = data.len() as u64;
    extent.extent_digest = Some(ObjectAddress {
        algorithm: "blake3".into(),
        digest: blake3::hash(&data).as_bytes().to_vec(),
    });
    FetchServerFrame {
        body: Some(fetch_server_frame::Body::Pack(PackChunk {
            extent: Some(extent),
            data,
        })),
    }
}
fn complete(ready: &TransferReady) -> FetchServerFrame {
    let mut checkpoint = ready.checkpoint.clone().expect("checkpoint");
    checkpoint.committed_bytes = ready.packs.iter().map(|p| p.length).sum();
    FetchServerFrame {
        body: Some(fetch_server_frame::Body::Complete(FetchComplete {
            revision: ready.current.clone(),
            checkpoint: Some(checkpoint),
            closure: Coverage::Complete as i32,
            missing: vec![],
        })),
    }
}
#[test]
fn native_fetch_commits_only_after_both_hashed_artifacts_and_exact_checkpoint() {
    let (open, ready, endpoint, artifacts) = fixture();
    let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
        .expect("admission");
    assert!(matches!(
        download.accept(complete(&ready)),
        Err(Error::Invalid(
            "download does not match its exact declared source coverage"
        ))
    ));
    assert!(!download.done);
    for (i, data) in artifacts.into_iter().enumerate() {
        assert!(matches!(
            download
                .accept(chunk(&ready, i, data))
                .expect("verified chunk"),
            Item::Pack(_)
        ));
    }
    assert!(matches!(
        download.accept(complete(&ready)).expect("complete"),
        Item::Complete(_)
    ));
    assert!(download.done);
    assert!(
        download.accept(complete(&ready)).is_err(),
        "terminal checkpoint cannot replay as a new commit"
    );
}
#[test]
fn native_fetch_rejects_rehashed_corruption_against_original_whole_artifact() {
    let (open, ready, endpoint, mut artifacts) = fixture();
    let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
        .expect("admission");
    artifacts[0][0] ^= 1;
    assert!(matches!(
        download.accept(chunk(&ready, 0, artifacts[0].clone())),
        Err(Error::Invalid("whole artifact hash mismatch"))
    ));
    assert!(!download.done);
}
#[test]
fn native_fetch_rejects_different_endpoint_and_truncated_index() {
    let (open, ready, endpoint, artifacts) = fixture();
    let wrong = EndpointRef {
        public_key: vec![1; 32],
        ..endpoint.clone()
    };
    assert!(matches!(
        Validation::new(open.clone(), ready.clone(), Some(&wrong), Limits::default()),
        Err(Error::Invalid(
            "admission does not match requested endpoint and revision"
        ))
    ));
    let mut download = Validation::new(open, ready.clone(), Some(&endpoint), Limits::default())
        .expect("admission");
    download
        .accept(chunk(&ready, 0, artifacts[0].clone()))
        .expect("pack");
    download
        .accept(chunk(&ready, 1, artifacts[1][..3].to_vec()))
        .expect("index prefix");
    assert!(matches!(
        download.accept(complete(&ready)),
        Err(Error::Invalid(
            "download does not match its exact declared source coverage"
        ))
    ));
}
#[test]
fn native_fetch_rejects_cross_spool_genesis_before_source() {
    let (open, mut ready, endpoint, _) = fixture();
    ready
        .owner_genesis
        .as_mut()
        .expect("owner")
        .genesis
        .as_mut()
        .expect("genesis")
        .spool_uuid = uuid::Uuid::now_v7().as_bytes().to_vec();
    assert!(matches!(
        Validation::new(open, ready, Some(&endpoint), Limits::default()),
        Err(Error::Invalid(
            "original owner genesis must bind this spool"
        ))
    ));
}