Struct holochain::core::workflow::sys_validation_workflow::SysValidationWorkspace

pub struct SysValidationWorkspace { /* private fields */ }

Implementations

impl SysValidationWorkspace

pub fn new(
    authored_db: DbRead<DbKindAuthored>,
    dht_db: DbRead<DbKindDht>,
    dht_query_cache: DhtDbQueryCache,
    cache: DbWrite<DbKindCache>,
    dna_def: Arc<DnaDef>
) -> Self

Examples found in repository

src/core/queue_consumer.rs (lines 205-211)

pub async fn spawn_queue_consumer_tasks(
    cell_id: CellId,
    network: HolochainP2pDna,
    space: &Space,
    conductor_handle: ConductorHandle,
    task_sender: sync::mpsc::Sender<ManagedTaskAdd>,
    stop: sync::broadcast::Sender<()>,
) -> (QueueTriggers, InitialQueueTriggers) {
    let Space {
        authored_db,
        dht_db,
        cache_db: cache,
        dht_query_cache,
        ..
    } = space;

    let keystore = conductor_handle.keystore().clone();
    let dna_hash = Arc::new(cell_id.dna_hash().clone());
    let queue_consumer_map = conductor_handle.get_queue_consumer_workflows();

    // Publish
    let (tx_publish, handle) = spawn_publish_dht_ops_consumer(
        cell_id.agent_pubkey().clone(),
        authored_db.clone(),
        conductor_handle.clone(),
        stop.subscribe(),
        Box::new(network.clone()),
    );
    task_sender
        .send(ManagedTaskAdd::cell_critical(
            handle,
            cell_id.clone(),
            "publish_dht_ops_consumer",
        ))
        .await
        .expect("Failed to manage workflow handle");

    // Validation Receipt
    // One per space.
    let (tx_receipt, handle) =
        queue_consumer_map.spawn_once_validation_receipt(dna_hash.clone(), || {
            spawn_validation_receipt_consumer(
                dna_hash.clone(),
                dht_db.clone(),
                conductor_handle.clone(),
                stop.subscribe(),
                network.clone(),
            )
        });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "validation_receipt_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    // Integration
    // One per space.
    let (tx_integration, handle) =
        queue_consumer_map.spawn_once_integration(dna_hash.clone(), || {
            spawn_integrate_dht_ops_consumer(
                dna_hash.clone(),
                dht_db.clone(),
                dht_query_cache.clone(),
                stop.subscribe(),
                tx_receipt.clone(),
                network.clone(),
            )
        });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "integrate_dht_ops_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let dna_def = conductor_handle
        .get_dna_def(&*dna_hash)
        .expect("Dna must be in store");

    // App validation
    // One per space.
    let (tx_app, handle) = queue_consumer_map.spawn_once_app_validation(dna_hash.clone(), || {
        spawn_app_validation_consumer(
            dna_hash.clone(),
            AppValidationWorkspace::new(
                authored_db.clone().into(),
                dht_db.clone(),
                space.dht_query_cache.clone(),
                cache.clone(),
                keystore.clone(),
                Arc::new(dna_def),
            ),
            conductor_handle.clone(),
            stop.subscribe(),
            tx_integration.clone(),
            network.clone(),
            dht_query_cache.clone(),
        )
    });
    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "app_validation_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let dna_def = conductor_handle
        .get_dna_def(&*dna_hash)
        .expect("Dna must be in store");

    // Sys validation
    // One per space.
    let (tx_sys, handle) = queue_consumer_map.spawn_once_sys_validation(dna_hash.clone(), || {
        spawn_sys_validation_consumer(
            SysValidationWorkspace::new(
                authored_db.clone().into(),
                dht_db.clone().into(),
                dht_query_cache.clone(),
                cache.clone(),
                Arc::new(dna_def),
            ),
            space.clone(),
            conductor_handle.clone(),
            stop.subscribe(),
            tx_app.clone(),
            network.clone(),
        )
    });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "sys_validation_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let (tx_cs, handle) = queue_consumer_map.spawn_once_countersigning(dna_hash.clone(), || {
        spawn_countersigning_consumer(
            space.clone(),
            stop.subscribe(),
            network.clone(),
            tx_sys.clone(),
        )
    });
    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "countersigning_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    (
        QueueTriggers {
            sys_validation: tx_sys.clone(),
            publish_dht_ops: tx_publish.clone(),
            countersigning: tx_cs,
            integrate_dht_ops: tx_integration.clone(),
        },
        InitialQueueTriggers::new(tx_sys, tx_publish, tx_app, tx_integration, tx_receipt),
    )
}

pub async fn is_chain_empty(
    &self,
    author: &AgentPubKey
) -> SourceChainResult<bool>

Examples found in repository

src/core/sys_validate.rs (line 193)

pub async fn check_valid_if_dna(
    action: &Action,
    workspace: &SysValidationWorkspace,
) -> SysValidationResult<()> {
    match action {
        Action::Dna(_) => {
            if !workspace.is_chain_empty(action.author()).await? {
                Err(PrevActionError::InvalidRoot).map_err(|e| ValidationOutcome::from(e).into())
            } else if action.timestamp() < workspace.dna_def().modifiers.origin_time {
                // If the Dna timestamp is ahead of the origin time, every other action
                // will be inductively so also due to the prev_action check
                Err(PrevActionError::InvalidRootOriginTime)
                    .map_err(|e| ValidationOutcome::from(e).into())
            } else {
                Ok(())
            }
        }
        _ => Ok(()),
    }
}

pub async fn action_seq_is_empty(
    &self,
    action: &Action
) -> SourceChainResult<bool>

Examples found in repository

src/core/sys_validate.rs (line 214)

pub async fn check_chain_rollback(
    action: &Action,
    workspace: &SysValidationWorkspace,
) -> SysValidationResult<()> {
    let empty = workspace.action_seq_is_empty(action).await?;

    // Ok or log warning
    if empty {
        Ok(())
    } else {
        // TODO: implement real rollback detection once we know what that looks like
        tracing::error!(
            "Chain rollback detected at position {} for agent {:?} from action {:?}",
            action.action_seq(),
            action.author(),
            action,
        );
        Ok(())
    }
}

pub fn local_cascade(&self) -> Cascade

Create a cascade with local data only

Examples found in repository

src/core/sys_validate.rs (line 659)

async fn check_and_hold<I: Into<AnyDhtHash> + Clone>(
    hash: &I,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
) -> SysValidationResult<Source> {
    let hash: AnyDhtHash = hash.clone().into();
    // Create a workspace with just the local stores
    let mut local_cascade = workspace.local_cascade();
    if let Some(el) = local_cascade
        .retrieve(hash.clone(), Default::default())
        .await?
    {
        return Ok(Source::Local(el));
    }
    // Create a workspace with just the network
    let mut network_only_cascade = workspace.full_cascade(network);
    match network_only_cascade
        .retrieve(hash.clone(), Default::default())
        .await?
    {
        Some(el) => Ok(Source::Network(el.privatized())),
        None => Err(ValidationOutcome::NotHoldingDep(hash).into()),
    }
}

pub fn full_cascade<Network: HolochainP2pDnaT + Clone + 'static + Send>(
    &self,
    network: Network
) -> Cascade<Network>

Examples found in repository

src/core/sys_validate.rs (line 667)

async fn check_and_hold<I: Into<AnyDhtHash> + Clone>(
    hash: &I,
    workspace: &SysValidationWorkspace,
    network: HolochainP2pDna,
) -> SysValidationResult<Source> {
    let hash: AnyDhtHash = hash.clone().into();
    // Create a workspace with just the local stores
    let mut local_cascade = workspace.local_cascade();
    if let Some(el) = local_cascade
        .retrieve(hash.clone(), Default::default())
        .await?
    {
        return Ok(Source::Local(el));
    }
    // Create a workspace with just the network
    let mut network_only_cascade = workspace.full_cascade(network);
    match network_only_cascade
        .retrieve(hash.clone(), Default::default())
        .await?
    {
        Some(el) => Ok(Source::Network(el.privatized())),
        None => Err(ValidationOutcome::NotHoldingDep(hash).into()),
    }
}

src/core/workflow/sys_validation_workflow.rs (line 325)

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(())
}

pub fn dna_def(&self) -> Arc<DnaDef>

Get a reference to the sys validation workspace’s dna def.

Examples found in repository

src/core/sys_validate.rs (line 195)

pub async fn check_valid_if_dna(
    action: &Action,
    workspace: &SysValidationWorkspace,
) -> SysValidationResult<()> {
    match action {
        Action::Dna(_) => {
            if !workspace.is_chain_empty(action.author()).await? {
                Err(PrevActionError::InvalidRoot).map_err(|e| ValidationOutcome::from(e).into())
            } else if action.timestamp() < workspace.dna_def().modifiers.origin_time {
                // If the Dna timestamp is ahead of the origin time, every other action
                // will be inductively so also due to the prev_action check
                Err(PrevActionError::InvalidRootOriginTime)
                    .map_err(|e| ValidationOutcome::from(e).into())
            } else {
                Ok(())
            }
        }
        _ => Ok(()),
    }
}

Trait Implementations

impl From<&HostFnWorkspace<DbWrite<DbKindAuthored>, DbWrite<DbKindDht>>> for SysValidationWorkspace

fn from(h: &HostFnWorkspace) -> Self

Converts to this type from the input type.