Function holochain::core::check_and_hold_store_record

pub async fn check_and_hold_store_record<F>(
    hash: &ActionHash,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
    f: F
) -> SysValidationResult<()>where
    F: FnOnce(&Record) -> SysValidationResult<()>,

If we are not holding this action then retrieve it and send it as a StoreRecord DhtOp to our incoming_dht_ops_workflow.

Apply a checks callback to the Record.

Additionally sys validation will be triggered to run again if we weren’t holding it.

Examples found in repository

src/core/workflow/sys_validation_workflow.rs (lines 359-365)

async fn validate_op_inner(
    op: &DhtOp,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    conductor_handle: &Conductor,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
) -> SysValidationResult<()> {
    match op {
        DhtOp::StoreRecord(_, action, entry) => {
            store_record(action, workspace, network.clone()).await?;
            if let Some(entry) = entry {
                // Retrieve for all other actions on countersigned entry.
                if let Entry::CounterSign(session_data, _) = &**entry {
                    let entry_hash = EntryHash::with_data_sync(&**entry);
                    let weight = action
                        .entry_rate_data()
                        .ok_or_else(|| SysValidationError::NonEntryAction(action.clone()))?;
                    for action in session_data.build_action_set(entry_hash, weight)? {
                        let hh = ActionHash::with_data_sync(&action);
                        if workspace
                            .full_cascade(network.clone())
                            .retrieve_action(hh.clone(), Default::default())
                            .await?
                            .is_none()
                        {
                            return Err(SysValidationError::ValidationOutcome(
                                ValidationOutcome::DepMissingFromDht(hh.into()),
                            ));
                        }
                    }
                }
                store_entry(
                    (action)
                        .try_into()
                        .map_err(|_| ValidationOutcome::NotNewEntry(action.clone()))?,
                    entry.as_ref(),
                    conductor_handle,
                    workspace,
                    network,
                )
                .await?;
            }
            Ok(())
        }
        DhtOp::StoreEntry(_, action, entry) => {
            // Check and hold for all other actions on countersigned entry.
            if let Entry::CounterSign(session_data, _) = &**entry {
                let dependency_check = |_original_record: &Record| Ok(());
                let entry_hash = EntryHash::with_data_sync(&**entry);
                let weight = match action {
                    NewEntryAction::Create(h) => h.weight.clone(),
                    NewEntryAction::Update(h) => h.weight.clone(),
                };
                for action in session_data.build_action_set(entry_hash, weight)? {
                    check_and_hold_store_record(
                        &ActionHash::with_data_sync(&action),
                        workspace,
                        network.clone(),
                        incoming_dht_ops_sender.clone(),
                        dependency_check,
                    )
                    .await?;
                }
            }

            store_entry(
                (action).into(),
                entry.as_ref(),
                conductor_handle,
                workspace,
                network.clone(),
            )
            .await?;

            let action = action.clone().into();
            store_record(&action, workspace, network).await?;
            Ok(())
        }
        DhtOp::RegisterAgentActivity(_, action) => {
            register_agent_activity(action, workspace, network.clone(), incoming_dht_ops_sender)
                .await?;
            store_record(action, workspace, network).await?;
            Ok(())
        }
        DhtOp::RegisterUpdatedContent(_, action, entry) => {
            register_updated_content(action, workspace, network.clone(), incoming_dht_ops_sender)
                .await?;
            if let Some(entry) = entry {
                store_entry(
                    NewEntryActionRef::Update(action),
                    entry.as_ref(),
                    conductor_handle,
                    workspace,
                    network.clone(),
                )
                .await?;
            }

            Ok(())
        }
        DhtOp::RegisterUpdatedRecord(_, action, entry) => {
            register_updated_record(action, workspace, network.clone(), incoming_dht_ops_sender)
                .await?;
            if let Some(entry) = entry {
                store_entry(
                    NewEntryActionRef::Update(action),
                    entry.as_ref(),
                    conductor_handle,
                    workspace,
                    network.clone(),
                )
                .await?;
            }

            Ok(())
        }
        DhtOp::RegisterDeletedBy(_, action) => {
            register_deleted_by(action, workspace, network, incoming_dht_ops_sender).await?;
            Ok(())
        }
        DhtOp::RegisterDeletedEntryAction(_, action) => {
            register_deleted_entry_action(action, workspace, network, incoming_dht_ops_sender)
                .await?;
            Ok(())
        }
        DhtOp::RegisterAddLink(_, action) => {
            register_add_link(action, workspace, network, incoming_dht_ops_sender).await?;
            Ok(())
        }
        DhtOp::RegisterRemoveLink(_, action) => {
            register_delete_link(action, workspace, network, incoming_dht_ops_sender).await?;
            Ok(())
        }
    }
}

// #[instrument(skip(record, call_zome_workspace, network, conductor_handle))]
/// Direct system validation call that takes
/// a Record instead of an op.
/// Does not require holding dependencies.
/// Will not await dependencies and instead returns
/// that outcome immediately.
pub async fn sys_validate_record(
    record: &Record,
    call_zome_workspace: &HostFnWorkspace,
    network: HolochainP2pDna,
    conductor_handle: &Conductor,
) -> SysValidationOutcome<()> {
    trace!(?record);
    // Create a SysValidationWorkspace with the scratches from the CallZomeWorkspace
    let workspace = SysValidationWorkspace::from(call_zome_workspace);
    let result =
        match sys_validate_record_inner(record, &workspace, network, conductor_handle).await {
            // Validation succeeded
            Ok(_) => Ok(()),
            // Validation failed so exit with that outcome
            Err(SysValidationError::ValidationOutcome(validation_outcome)) => {
                error!(msg = "Direct validation failed", ?record);
                validation_outcome.into_outcome()
            }
            // An error occurred so return it
            Err(e) => Err(OutcomeOrError::Err(e)),
        };

    result
}

async fn sys_validate_record_inner(
    record: &Record,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    conductor_handle: &Conductor,
) -> SysValidationResult<()> {
    let signature = record.signature();
    let action = record.action();
    let maybe_entry = record.entry().as_option();
    counterfeit_check(signature, action).await?;

    async fn validate(
        action: &Action,
        maybe_entry: Option<&Entry>,
        workspace: &SysValidationWorkspace,
        network: HolochainP2pDna,
        conductor_handle: &Conductor,
    ) -> SysValidationResult<()> {
        let incoming_dht_ops_sender = None;
        store_record(action, workspace, network.clone()).await?;
        if let Some((maybe_entry, EntryVisibility::Public)) =
            &maybe_entry.and_then(|e| action.entry_type().map(|et| (e, et.visibility())))
        {
            store_entry(
                (action)
                    .try_into()
                    .map_err(|_| ValidationOutcome::NotNewEntry(action.clone()))?,
                maybe_entry,
                conductor_handle,
                workspace,
                network.clone(),
            )
            .await?;
        }
        match action {
            Action::Update(action) => {
                register_updated_content(action, workspace, network, incoming_dht_ops_sender)
                    .await?;
            }
            Action::Delete(action) => {
                register_deleted_entry_action(action, workspace, network, incoming_dht_ops_sender)
                    .await?;
            }
            Action::CreateLink(action) => {
                register_add_link(action, workspace, network, incoming_dht_ops_sender).await?;
            }
            Action::DeleteLink(action) => {
                register_delete_link(action, workspace, network, incoming_dht_ops_sender).await?;
            }
            _ => {}
        }
        Ok(())
    }

    match maybe_entry {
        Some(Entry::CounterSign(session, _)) => {
            if let Some(weight) = action.entry_rate_data() {
                let entry_hash = EntryHash::with_data_sync(maybe_entry.unwrap());
                for action in session.build_action_set(entry_hash, weight)? {
                    validate(
                        &action,
                        maybe_entry,
                        workspace,
                        network.clone(),
                        conductor_handle,
                    )
                    .await?;
                }
                Ok(())
            } else {
                tracing::error!("Got countersigning entry without rate assigned. This should be impossible. But, let's see what happens.");
                validate(action, maybe_entry, workspace, network, conductor_handle).await
            }
        }
        _ => validate(action, maybe_entry, workspace, network, conductor_handle).await,
    }
}

/// Check if the op has valid signature and author.
/// Ops that fail this check should be dropped.
pub async fn counterfeit_check(signature: &Signature, action: &Action) -> SysValidationResult<()> {
    verify_action_signature(signature, action).await?;
    author_key_is_valid(action.author()).await?;
    Ok(())
}

async fn register_agent_activity(
    action: &Action,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let prev_action_hash = action.prev_action();

    // Checks
    check_prev_action(action)?;
    check_valid_if_dna(action, workspace).await?;
    if let Some(prev_action_hash) = prev_action_hash {
        check_and_hold_register_agent_activity(
            prev_action_hash,
            workspace,
            network,
            incoming_dht_ops_sender,
            |_| Ok(()),
        )
        .await?;
    }
    check_chain_rollback(action, workspace).await?;
    Ok(())
}

async fn store_record(
    action: &Action,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let prev_action_hash = action.prev_action();

    // Checks
    check_prev_action(action)?;
    if let Some(prev_action_hash) = prev_action_hash {
        let mut cascade = workspace.full_cascade(network);
        let prev_action = cascade
            .retrieve_action(prev_action_hash.clone(), Default::default())
            .await?
            .ok_or_else(|| ValidationOutcome::DepMissingFromDht(prev_action_hash.clone().into()))?;
        check_prev_timestamp(action, prev_action.action())?;
        check_prev_seq(action, prev_action.action())?;
    }
    Ok(())
}

async fn store_entry(
    action: NewEntryActionRef<'_>,
    entry: &Entry,
    conductor_handle: &Conductor,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let entry_type = action.entry_type();
    let entry_hash = action.entry_hash();

    // Checks
    check_entry_type(entry_type, entry)?;
    if let EntryType::App(app_entry_def) = entry_type {
        let entry_def =
            check_app_entry_def(workspace.dna_hash(), app_entry_def, conductor_handle).await?;
        check_not_private(&entry_def)?;
    }

    check_entry_hash(entry_hash, entry).await?;
    check_entry_size(entry)?;

    // Additional checks if this is an Update
    if let NewEntryActionRef::Update(entry_update) = action {
        let original_action_address = &entry_update.original_action_address;
        let mut cascade = workspace.full_cascade(network);
        let original_action = cascade
            .retrieve_action(original_action_address.clone(), Default::default())
            .await?
            .ok_or_else(|| {
                ValidationOutcome::DepMissingFromDht(original_action_address.clone().into())
            })?;
        update_check(entry_update, original_action.action())?;
    }

    // Additional checks if this is a countersigned entry.
    if let Entry::CounterSign(session_data, _) = entry {
        check_countersigning_session_data(EntryHash::with_data_sync(entry), session_data, action)
            .await?;
    }
    Ok(())
}

async fn register_updated_content(
    entry_update: &Update,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let original_action_address = &entry_update.original_action_address;

    let dependency_check =
        |original_record: &Record| update_check(entry_update, original_record.action());
    check_and_hold_store_entry(
        original_action_address,
        workspace,
        network,
        incoming_dht_ops_sender,
        dependency_check,
    )
    .await?;
    Ok(())
}

async fn register_updated_record(
    entry_update: &Update,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let original_action_address = &entry_update.original_action_address;

    let dependency_check =
        |original_record: &Record| update_check(entry_update, original_record.action());

    check_and_hold_store_record(
        original_action_address,
        workspace,
        network,
        incoming_dht_ops_sender,
        dependency_check,
    )
    .await?;
    Ok(())
}

async fn register_deleted_by(
    record_delete: &Delete,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
    incoming_dht_ops_sender: Option<IncomingDhtOpSender>,
) -> SysValidationResult<()> {
    // Get data ready to validate
    let removed_action_address = &record_delete.deletes_address;

    // Checks
    let dependency_check =
        |removed_action: &Record| check_new_entry_action(removed_action.action());

    check_and_hold_store_record(
        removed_action_address,
        workspace,
        network,
        incoming_dht_ops_sender,
        dependency_check,
    )
    .await?;
    Ok(())
}