Struct holochain::core::workflow::app_validation_workflow::AppValidationWorkspace

pub struct AppValidationWorkspace { /* private fields */ }

Implementations

impl AppValidationWorkspace

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

Examples found in repository

src/core/queue_consumer.rs (lines 171-178)

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 validation_workspace(
    &self
) -> Result<HostFnWorkspaceRead, AppValidationError>

Examples found in repository

src/core/workflow/app_validation_workflow.rs (line 437)

async fn validate_op_outer(
    dna_hash: Arc<DnaHash>,
    op: &Op,
    conductor_handle: &ConductorHandle,
    workspace: &AppValidationWorkspace,
    network: &HolochainP2pDna,
) -> AppValidationOutcome<Outcome> {
    // Get the workspace for the validation calls
    let host_fn_workspace = workspace.validation_workspace().await?;

    // Get the ribosome
    let ribosome = conductor_handle
        .get_ribosome(dna_hash.as_ref())
        .map_err(|_| AppValidationError::DnaMissing((*dna_hash).clone()))?;

    validate_op(op, host_fn_workspace, network, &ribosome).await
}

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

Examples found in repository

src/core/workflow/app_validation_workflow.rs (line 122)

async fn app_validation_workflow_inner(
    dna_hash: Arc<DnaHash>,
    workspace: Arc<AppValidationWorkspace>,
    conductor: ConductorHandle,
    network: &HolochainP2pDna,
    dht_query_cache: DhtDbQueryCache,
) -> WorkflowResult<WorkComplete> {
    let db = workspace.dht_db.clone().into();
    let sorted_ops = validation_query::get_ops_to_app_validate(&db).await?;
    let start_len = sorted_ops.len();
    tracing::debug!("validating {} ops", start_len);
    let start = (start_len >= NUM_CONCURRENT_OPS).then(std::time::Instant::now);
    let saturated = start.is_some();

    // Validate all the ops
    let iter = sorted_ops.into_iter().map({
        let network = network.clone();
        let workspace = workspace.clone();
        move |so| {
            let network = network.clone();
            let conductor = conductor.clone();
            let workspace = workspace.clone();
            let dna_hash = dna_hash.clone();
            async move {
                let (op, op_hash) = so.into_inner();
                let op_type = op.get_type();
                let action = op.action();
                let dependency = get_dependency(op_type, &action);
                let op_light = op.to_light();

                // If this is agent activity, track it for the cache.
                let activity = matches!(op_type, DhtOpType::RegisterAgentActivity).then(|| {
                    (
                        action.author().clone(),
                        action.action_seq(),
                        matches!(dependency, Dependency::Null),
                    )
                });

                // Validate this op
                let mut cascade = workspace.full_cascade(network.clone());
                let r = match dhtop_to_op(op, &mut cascade).await {
                    Ok(op) => {
                        validate_op_outer(dna_hash, &op, &conductor, &(*workspace), &network).await
                    }
                    Err(e) => Err(e),
                };
                (op_hash, dependency, op_light, r, activity)
            }
        }
    });

    // Create a stream of concurrent validation futures.
    // This will run NUM_CONCURRENT_OPS validation futures concurrently and
    // return up to NUM_CONCURRENT_OPS * 100 results.
    let mut iter = futures::stream::iter(iter)
        .buffer_unordered(NUM_CONCURRENT_OPS)
        .ready_chunks(NUM_CONCURRENT_OPS * 100);

    // Spawn a task to actually drive the stream.
    // This allows the stream to make progress in the background while
    // we are committing previous results to the database.
    let (tx, rx) = tokio::sync::mpsc::channel(NUM_CONCURRENT_OPS * 100);
    let jh = tokio::spawn(async move {
        while let Some(op) = iter.next().await {
            // Send the result to task that will commit to the database.
            if tx.send(op).await.is_err() {
                tracing::warn!("app validation task has failed to send ops. This is not a problem if the conductor is shutting down");
                break;
            }
        }
    });

    // Create a stream that will chunk up to NUM_CONCURRENT_OPS * 100 ready results.
    let mut iter =
        tokio_stream::wrappers::ReceiverStream::new(rx).ready_chunks(NUM_CONCURRENT_OPS * 100);

    let mut total = 0;
    let mut round_time = start.is_some().then(std::time::Instant::now);
    // Pull in a chunk of results.
    while let Some(chunk) = iter.next().await {
        tracing::debug!(
            "Committing {} ops",
            chunk.iter().map(|c| c.len()).sum::<usize>()
        );
        let (t, a, r, activity) = workspace
            .dht_db
            .async_commit(move |txn| {
                let mut total = 0;
                let mut awaiting = 0;
                let mut rejected = 0;
                let mut agent_activity = Vec::new();
                for outcome in chunk.into_iter().flatten() {
                    let (op_hash, dependency, op_light, outcome, activity) = outcome;
                    // Get the outcome or return the error
                    let outcome = outcome.or_else(|outcome_or_err| outcome_or_err.try_into())?;

                    // Collect all agent activity.
                    if let Some(activity) = activity {
                        // If the activity is accepted or rejected then it's ready to integrate.
                        if matches!(&outcome, Outcome::Accepted | Outcome::Rejected(_)) {
                            agent_activity.push(activity);
                        }
                    }


                    if let Outcome::AwaitingDeps(_) | Outcome::Rejected(_) = &outcome {
                        warn!(
                            msg = "DhtOp has failed app validation",
                            outcome = ?outcome,
                        );
                    }
                    match outcome {
                        Outcome::Accepted => {
                            total += 1;
                            if let Dependency::Null = dependency {
                                put_integrated(txn, &op_hash, ValidationStatus::Valid)?;
                            } else {
                                put_integration_limbo(txn, &op_hash, ValidationStatus::Valid)?;
                            }
                        }
                        Outcome::AwaitingDeps(deps) => {
                            awaiting += 1;
                            let status = ValidationLimboStatus::AwaitingAppDeps(deps);
                            put_validation_limbo(txn, &op_hash, status)?;
                        }
                        Outcome::Rejected(_) => {
                            rejected += 1;
                            tracing::warn!("Received invalid op! Warrants aren't implemented yet, so we can't do anything about this right now, but be warned that somebody on the network has maliciously hacked their node.\nOp: {:?}", op_light);
                            if let Dependency::Null = dependency {
                                put_integrated(txn, &op_hash, ValidationStatus::Rejected)?;
                            } else {
                                put_integration_limbo(txn, &op_hash, ValidationStatus::Rejected)?;
                            }
                        }
                    }
                }
                WorkflowResult::Ok((total, awaiting, rejected, agent_activity))
            })
            .await?;
        // Once the database transaction is committed, add agent activity to the cache
        // that is ready for integration.
        for (author, seq, has_no_dependency) in activity {
            // Any activity with no dependency is integrated in this workflow.
            // TODO: This will no longer be true when [#1212](https://github.com/holochain/holochain/pull/1212) lands.
            if has_no_dependency {
                dht_query_cache
                    .set_activity_to_integrated(&author, seq)
                    .await?;
            } else {
                dht_query_cache
                    .set_activity_ready_to_integrate(&author, seq)
                    .await?;
            }
        }
        total += t;
        if let (Some(start), Some(round_time)) = (start, &mut round_time) {
            let round_el = round_time.elapsed();
            *round_time = std::time::Instant::now();
            let avg_ops_ps = total as f64 / start.elapsed().as_micros() as f64 * 1_000_000.0;
            let ops_ps = t as f64 / round_el.as_micros() as f64 * 1_000_000.0;
            tracing::warn!(
                "App validation is saturated. Util {:.2}%. OPS/s avg {:.2}, this round {:.2}",
                (start_len - total) as f64 / NUM_CONCURRENT_OPS as f64 * 100.0,
                avg_ops_ps,
                ops_ps
            );
        }
        tracing::debug!(
            "{} committed, {} awaiting sys dep, {} rejected. {} committed this round",
            t,
            a,
            r,
            total
        );
    }
    jh.await?;
    tracing::debug!("accepted {} ops", total);
    Ok(if saturated {
        WorkComplete::Incomplete
    } else {
        WorkComplete::Complete
    })
}