Validation Receipts Scenario (#83)

* initial working validation receipts scenario

* fix early loop break

* address code review comments

.github/workflows/test.yaml

@@ -207,6 +207,23 @@ jobs:
           # Stop local network services
           pkill hc-run-local
 
+      - name: Smoke test - validation_receipts
+        run: |
+          set -x
+
+          # Start local network services
+          nix develop .#ci -c bash -c "hc-run-local-services --bootstrap-port 4422 --signal-port 4423 &"
+          # Start a TryCP instance
+          nix develop .#ci -c bash -c "source ./scripts/trycp.sh && start_trycp &"
+
+          # Run the scenario for 10 seconds
+          RUST_LOG=warn CONDUCTOR_CONFIG="CI" MIN_PEERS=2 nix run .#validation_receipts -- --targets targets-ci.yaml --instances-per-target 2 --duration 10 --no-progress
+
+          # Stop the TryCP instance
+          nix develop .#ci -c bash -c "source ./scripts/trycp.sh && stop_trycp"
+          # Stop local network services
+          pkill hc-run-local
+
       - name: Build scenario bundles
         if: runner.os == 'Linux'
         run: |

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "scenarios/two_party_countersigning",
     "scenarios/write_validated",
     "scenarios/trycp_write_validated",
+    "scenarios/validation_receipts",
 
     "zomes/return_single_value/coordinator",
     "zomes/crud/coordinator",

influx/templates/dashboards/validation_receipts.json

@@ -0,0 +1 @@
+[{"apiVersion":"influxdata.com/v2alpha1","kind":"Dashboard","metadata":{"name":"flamboyant-turing-9d2001"},"spec":{"charts":[{"axes":[{"base":"10","name":"x","scale":"linear"},{"base":"10","name":"y","scale":"linear","suffix":"s"}],"colorizeRows":true,"colors":[{"id":"YbLgrZIWds0VUbHbU7eW0","name":"Color Blind Friendly - Light","type":"scale","hex":"#FFFFFF"},{"id":"cmiZoIh8t_6i85GQDbQUT","name":"Color Blind Friendly - Light","type":"scale","hex":"#E69F00"},{"id":"NwsgnzwbcJ-LDzHg5uCXw","name":"Color Blind Friendly - Light","type":"scale","hex":"#56B4E9"},{"id":"IL2nM3qfc2633nv72-XP3","name":"Color Blind Friendly - Light","type":"scale","hex":"#009E73"},{"id":"3v3QgnLTCUXG8Um3UK_q8","name":"Color Blind Friendly - Light","type":"scale","hex":"#F0E442"},{"id":"za7MKZmBZiDcEcDry8PWo","name":"Color Blind Friendly - Light","type":"scale","hex":"#0072B2"},{"id":"PWbWJNYzZqeVB-Kbughen","name":"Color Blind Friendly - Light","type":"scale","hex":"#D55E00"},{"id":"aU8r-mDCUZGdDm3G3v7Bd","name":"Color Blind Friendly - Light","type":"scale","hex":"#CC79A7"}],"geom":"line","height":4,"hoverDimension":"auto","kind":"Xy","legendColorizeRows":true,"legendOpacity":1,"legendOrientationThreshold":100000000,"name":"Validation Receipts Complete","opacity":1,"orientationThreshold":100000000,"position":"overlaid","queries":[{"query":"from(bucket: \"windtunnel\")\n  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)\n  |> filter(fn: (r) => r[\"_measurement\"] == \"wt.custom.validation_receipts_complete_time\")\n  |> filter(fn: (r) => r[\"scenario_name\"] == \"validation_receipts\")\n  |> filter(fn: (r) => r[\"run_id\"] == v.RunId)\n  |> aggregateWindow(every: v.windowPeriod, fn: mean, createEmpty: false)\n  |> yield(name: \"mean\")"}],"staticLegend":{"colorizeRows":true,"opacity":1,"orientationThreshold":100000000,"widthRatio":1},"width":12,"widthRatio":1,"xCol":"_time","yCol":"_value"}],"name":"Validation Receipts"}}]

scenarios/validation_receipts/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "validation_receipts"
+version = "0.1.0"
+edition = "2021"
+build = "../scenario_build.rs"
+
+[dependencies]
+anyhow = { workspace = true }
+tokio = { workspace = true }
+rand = { workspace = true }
+log = { workspace = true }
+
+holochain_types = { workspace = true }
+trycp_wind_tunnel_runner = { workspace = true }
+
+[build-dependencies]
+happ_builder = { workspace = true }
+
+[lints]
+workspace = true
+
+[package.metadata.required-dna]
+name = "crud"
+zomes = ["crud"]
+
+[package.metadata.required-happ]
+name = "crud"
+dnas = ["crud"]

scenarios/validation_receipts/README.md

@@ -0,0 +1,46 @@
+## validation_receipts
+
+### Description
+
+Creates an entry, wait for required validation receipts, then repeat.
+
+**warning** This is a TryCP-based scenario and needs to be run differently to other scenarios.
+
+### Waiting for peer discovery
+
+This scenario reads the environment variable `MIN_PEERS` and waits for at least that many peers to be available before
+starting the agent behaviour. It will wait up to two minutes then proceed regardless.
+
+The scenario is not able to check that you have configured more peers than the minimum you have set, so you should
+ensure that you have configured enough peers to meet the minimum.
+
+Note that the number of peers seen by each node includes itself. So having two nodes means that each node will 
+immediately see one peer after app installation.
+
+You need around at least 10 peers, or the nodes will never get the required number of validation receipts.
+
+### NO_VALIDATION_COMPLETE
+
+By default this scenario will wait for a complete set of validation receipts before moving on to commit the next record. If you want to publish new records on every round, building up an ever-growing list of action hashes to check for validation complete, run with the `NO_VALIDATION_COMPLETE=1` environment variable.
+
+Example:
+
+```bash
+NO_VALIDATION_COMPLETE=1 RUST_LOG=info CONDUCTOR_CONFIG="CI" TRYCP_RUST_LOG="info" MIN_PEERS=10 cargo run --package validation_receipts -- --targets targets-ci.yaml --instances-per-target 10 --duration 300
+```
+
+### Suggested command
+
+You can run the scenario locally with the following command:
+
+```bash
+RUST_LOG=info CONDUCTOR_CONFIG="CI" TRYCP_RUST_LOG="info" MIN_PEERS=10 cargo run --package validation_receipts -- --targets targets-ci.yaml --instances-per-target 10 --duration 300
+```
+
+This assumes that `trycp_server` is running. See the script `scripts/trycp.sh` and run with `start_trycp`.
+
+To run the scenario against the current target list, you can run:
+
+```bash
+RUST_LOG=info MIN_PEERS=40 cargo run --package validation_receipts -- --targets targets.yaml --duration 500
+```

scenarios/validation_receipts/src/main.rs

@@ -0,0 +1,198 @@
+use holochain_types::prelude::*;
+use std::collections::HashMap;
+use std::time::{Duration, Instant};
+use trycp_wind_tunnel_runner::embed_conductor_config;
+use trycp_wind_tunnel_runner::prelude::*;
+
+embed_conductor_config!();
+
+type OpType = String;
+type ReceiptsComplete = bool;
+
+#[derive(Debug, Default)]
+pub struct ScenarioValues {
+    /// Action hash to a map of validation receipt types:
+    /// - if the sub map is empty, we haven't received any receipts yet,
+    ///   so we're still pending
+    /// - if any of the receipts_complete are false, we are still pending
+    /// - if all of the receipts_complete are true, we are complete
+    ///   so the action should be removed from the map
+    pending_actions: HashMap<ActionHash, HashMap<OpType, ReceiptsComplete>>,
+}
+
+impl ScenarioValues {
+    fn mut_op_complete(&mut self, action_hash: &ActionHash, op_type: String) -> &mut bool {
+        self.pending_actions
+            .get_mut(action_hash)
+            .unwrap()
+            .entry(op_type)
+            .or_default()
+    }
+
+    fn mut_any_pending(&mut self) -> bool {
+        self.pending_actions.retain(|_, m| {
+            if m.is_empty() {
+                return true;
+            }
+            let mut all_complete = true;
+            for c in m.values() {
+                if !c {
+                    all_complete = false;
+                    break;
+                }
+            }
+            !all_complete
+        });
+
+        !self.pending_actions.is_empty()
+    }
+}
+
+impl UserValuesConstraint for ScenarioValues {}
+
+fn agent_setup(
+    ctx: &mut AgentContext<TryCPRunnerContext, TryCPAgentContext<ScenarioValues>>,
+) -> HookResult {
+    connect_trycp_client(ctx)?;
+    reset_trycp_remote(ctx)?;
+
+    let client = ctx.get().trycp_client();
+    let agent_name = ctx.agent_name().to_string();
+
+    ctx.runner_context()
+        .executor()
+        .execute_in_place(async move {
+            client
+                .configure_player(agent_name.clone(), conductor_config().to_string(), None)
+                .await?;
+
+            client.startup(agent_name.clone(), None).await?;
+
+            Ok(())
+        })?;
+
+    install_app(ctx, scenario_happ_path!("crud"), &"crud".to_string())?;
+    try_wait_for_min_peers(ctx, Duration::from_secs(120))?;
+
+    Ok(())
+}
+
+fn agent_behaviour(
+    ctx: &mut AgentContext<TryCPRunnerContext, TryCPAgentContext<ScenarioValues>>,
+) -> HookResult {
+    let reporter = ctx.runner_context().reporter();
+
+    let action_hash: ActionHash = call_zome(
+        ctx,
+        "crud",
+        "create_sample_entry",
+        "this is a test entry value",
+        Some(Duration::from_secs(80)),
+    )?;
+
+    ctx.get_mut()
+        .scenario_values
+        .pending_actions
+        .insert(action_hash, HashMap::new());
+
+    let start = Instant::now();
+
+    'outer: loop {
+        // sleep a bit, we don't want to busy loop
+        ctx.runner_context()
+            .executor()
+            .execute_in_place(async move {
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                Ok(())
+            })?;
+
+        // get our list of pending actions
+        let action_hash_list = ctx
+            .get()
+            .scenario_values
+            .pending_actions
+            .keys()
+            .cloned()
+            .collect::<Vec<_>>();
+
+        for action_hash in action_hash_list {
+            // for each action, get the validation receipts
+            let response: Vec<ValidationReceiptSet> = call_zome(
+                ctx,
+                "crud",
+                "get_sample_entry_validation_receipts",
+                action_hash.clone(),
+                Some(Duration::from_secs(80)),
+            )?;
+
+            for set in response.iter() {
+                let cur = *ctx
+                    .get_mut()
+                    .scenario_values
+                    .mut_op_complete(&action_hash, set.op_type.clone());
+
+                if set.receipts_complete && !cur {
+                    // if the action wasn't already complete report the time
+                    // and mark it complete
+                    reporter.add_custom(
+                        ReportMetric::new("validation_receipts_complete_time")
+                            .with_tag("op_type", set.op_type.clone())
+                            .with_field("value", start.elapsed().as_secs_f64()),
+                    );
+                    *ctx.get_mut()
+                        .scenario_values
+                        .mut_op_complete(&action_hash, set.op_type.clone()) = true;
+                }
+            }
+        }
+
+        // if there are no remaining pending actions, break out of the loop
+        if !ctx.get_mut().scenario_values.mut_any_pending() {
+            break 'outer;
+        }
+
+        // if we were instructed to not wait for validation complete,
+        // don't wait for validation complete
+        if std::env::var_os("NO_VALIDATION_COMPLETE").is_some() {
+            break 'outer;
+        }
+    }
+
+    Ok(())
+}
+
+fn agent_teardown(
+    ctx: &mut AgentContext<TryCPRunnerContext, TryCPAgentContext<ScenarioValues>>,
+) -> HookResult {
+    if let Err(e) = dump_logs(ctx) {
+        log::warn!("Failed to dump logs: {:?}", e);
+    }
+
+    // Best effort to remove data and cleanup.
+    // You should comment out this line if you want to examine the result of the scenario run!
+    let _ = reset_trycp_remote(ctx);
+
+    // Alternatively, you can just shut down the remote conductor instead of shutting it down and removing data.
+    // shutdown_remote(ctx)?;
+
+    disconnect_trycp_client(ctx)?;
+
+    Ok(())
+}
+
+fn main() -> WindTunnelResult<()> {
+    let builder = TryCPScenarioDefinitionBuilder::<
+        TryCPRunnerContext,
+        TryCPAgentContext<ScenarioValues>,
+    >::new_with_init(env!("CARGO_PKG_NAME"))?
+    .into_std()
+    .use_agent_setup(agent_setup)
+    .use_agent_behaviour(agent_behaviour)
+    .use_agent_teardown(agent_teardown);
+
+    let agents_at_completion = run(builder)?;
+
+    println!("Finished with {} agents", agents_at_completion);
+
+    Ok(())
+}

zomes/crud/coordinator/src/lib.rs

@@ -25,6 +25,13 @@ fn get_sample_entry(hash: ActionHash) -> ExternResult<Option<Record>> {
     get(hash, GetOptions::local())
 }
 
+#[hdk_extern]
+fn get_sample_entry_validation_receipts(
+    hash: ActionHash,
+) -> ExternResult<Vec<ValidationReceiptSet>> {
+    get_validation_receipts(GetValidationReceiptsInput::new(hash))
+}
+
 #[hdk_extern]
 fn chain_query_count_len() -> ExternResult<u32> {
     let q = ChainQueryFilter::new()