address code review comments

conductor-config-ci.yaml

@@ -1,6 +1,3 @@
-dpki:
-  device_seed_lair_tag: ci
-  no_dpki: true
 network:
   network_type: quic_bootstrap
   transport_pool:

scenarios/validation_receipts/README.md

@@ -19,6 +19,16 @@ 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:

scenarios/validation_receipts/src/main.rs

@@ -1,12 +1,52 @@
 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 {}
+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 {}
 
@@ -50,49 +90,73 @@ fn agent_behaviour(
         Some(Duration::from_secs(80)),
     )?;
 
-    let response: Option<Record> = call_zome(
-        ctx,
-        "crud",
-        "get_sample_entry",
-        action_hash.clone(),
-        Some(Duration::from_secs(80)),
-    )?;
-
-    assert!(response.is_some(), "Expected record to be found");
+    ctx.get_mut()
+        .scenario_values
+        .pending_actions
+        .insert(action_hash, HashMap::new());
 
     let start = Instant::now();
 
     'outer: loop {
-        let response: Vec<ValidationReceiptSet> = call_zome(
-            ctx,
-            "crud",
-            "get_sample_entry_validation_receipts",
-            action_hash.clone(),
-            Some(Duration::from_secs(80)),
-        )?;
-
-        // only check for complete if we actually get data back
-        // before any receipts are received, this is an empty vec
-        if !response.is_empty() {
-            let mut all_complete = true;
-
-            'inner: for set in response.iter() {
-                if !set.receipts_complete {
-                    all_complete = false;
-                    break 'inner;
+        // 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 all_complete {
-                break 'outer;
-            }
+        // if there are no remaining pending actions, break out of the loop
+        if !ctx.get_mut().scenario_values.mut_any_pending() {
+            break 'outer;
         }
-    }
 
-    reporter.add_custom(
-        ReportMetric::new("validation_receipts_complete_time")
-            .with_field("value", start.elapsed().as_secs_f64()),
-    );
+        // 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(())
 }