tap-node 0.7.0

Transaction Authorization Protocol (TAP) node implementation for routing and processing messages
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

//! Stress testing for TAP Node
//!
//! Run with: cargo bench --bench stress_test

use std::str::FromStr;
use std::sync::Arc;
use std::time::Instant;
use tap_agent::TapAgent;
use tap_msg::message::TapMessageBody;
use tap_msg::message::{Party, Transfer};
use tap_msg::PlainMessage;
use tap_node::message::processor_pool::ProcessorPoolConfig;
use tap_node::{NodeConfig, TapNode};

use criterion::{criterion_group, criterion_main, Criterion};
use std::collections::HashMap;
use tokio::time::Duration;

/// Create a test message
async fn create_test_message(
    from_did: &str,
    to_did: &str,
    index: usize,
) -> (PlainMessage, Transfer) {
    // Create a simple transfer message
    let body = Transfer {
        transaction_id: Some(uuid::Uuid::new_v4().to_string()),
        asset: tap_caip::AssetId::from_str(
            "eip155:1/erc20:0x6b175474e89094c44da98b954eedeac495271d0f",
        )
        .unwrap(),
        originator: Some(Party::new(from_did)),
        beneficiary: Some(Party::new(to_did)),
        amount: format!("{}.00", index),
        agents: vec![],
        settlement_id: None,
        expiry: None,
        transaction_value: None,
        memo: Some(format!("Test message {}", index)),
        metadata: HashMap::new(),
        connection_id: None,
    };

    // Convert to DIDComm message
    let message = body.to_didcomm(from_did).unwrap();
    (message, body)
}

fn stress_test(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();

    // Setup a node with processing pool
    let pool_config = ProcessorPoolConfig {
        workers: 16,
        channel_capacity: 1000,
        worker_timeout: Duration::from_secs(30),
    };

    let node_config = NodeConfig {
        debug: false,
        max_agents: None,
        enable_message_logging: false,
        log_message_content: false,
        processor_pool: Some(pool_config),
        event_logger: None,
        #[cfg(feature = "storage")]
        storage_path: None,
        #[cfg(feature = "storage")]
        agent_did: None,
        #[cfg(feature = "storage")]
        tap_root: None,
        decision_mode: Default::default(),
    };

    // For testing, we'll create some DIDs that don't rely on external resolvers
    let mut group = c.benchmark_group("tap_node_stress");
    group.sample_size(10); // Reduce sample size for stress tests

    // Test with different message batch sizes
    for &batch_size in &[10, 100, 1000] {
        group.bench_function(format!("process_{}_messages", batch_size), |b| {
            b.iter(|| {
                rt.block_on(async {
                    // Create a new node for each benchmark iteration
                    let node = TapNode::new(node_config.clone());

                    // Create two ephemeral agents for the test
                    let (agent1, agent1_did) = TapAgent::from_ephemeral_key().await.unwrap();
                    let (agent2, agent2_did) = TapAgent::from_ephemeral_key().await.unwrap();

                    let agent1 = Arc::new(agent1);
                    let agent2 = Arc::new(agent2);

                    // Register the agents with the node
                    node.register_agent(agent1).await.unwrap();
                    node.register_agent(agent2).await.unwrap();

                    // Submit messages
                    let start = Instant::now();
                    let mut futures = Vec::with_capacity(batch_size);

                    for i in 0..batch_size {
                        let (message, _) = create_test_message(&agent1_did, &agent2_did, i).await;
                        let message_value = serde_json::to_value(message).unwrap();
                        futures.push(node.receive_message(message_value));
                    }

                    // Wait for all messages to be processed
                    for future in futures {
                        let _ = future.await;
                    }

                    let duration = start.elapsed();
                    println!(
                        "Processed {} messages in {:?} ({:.2} msg/s)",
                        batch_size,
                        duration,
                        batch_size as f64 / duration.as_secs_f64()
                    );
                });
            });
        });
    }

    group.finish();
}

criterion_group!(benches, stress_test);
criterion_main!(benches);