pact_plugin_driver/
call_chain.rs

1//! Call-chain tracking for driver <-> plugin callback cycle detection and deadline propagation.
2//!
3//! When a plugin calls back into the driver's `PluginHost` service for a capability (see
4//! [`crate::core_capabilities`] and proposal 007, Driver-plugin callback model), that call may
5//! itself be forwarded to another plugin, which could in turn call back again. This module tracks
6//! the chain of catalogue entry keys invoked under a single root call (identified by a
7//! `pact-call-chain-id` gRPC metadata value) so a cycle - the same entry key being invoked twice
8//! in the same chain - is rejected instead of deadlocking or recursing forever, and enforces an
9//! absolute deadline (`pact-deadline-ms` gRPC metadata) so a callback can never outlive the
10//! request that triggered it.
11
12use std::collections::HashMap;
13use std::sync::Mutex;
14use std::time::{Duration, SystemTime, UNIX_EPOCH};
15
16use lazy_static::lazy_static;
17use uuid::Uuid;
18
19/// gRPC metadata key carrying the call-chain ID on driver<->plugin callback requests.
20pub const CALL_CHAIN_ID_METADATA_KEY: &str = "pact-call-chain-id";
21/// gRPC metadata key carrying the absolute deadline (Unix epoch milliseconds) on driver<->plugin
22/// callback requests.
23pub const DEADLINE_METADATA_KEY: &str = "pact-deadline-ms";
24
25/// Budget given to a new call chain started by the driver, and the fallback used when a callback
26/// arrives with no deadline metadata (defensive default for a non-conforming plugin).
27pub const DEFAULT_CALL_CHAIN_TIMEOUT: Duration = Duration::from_secs(30);
28
29lazy_static! {
30  static ref CALL_CHAINS: Mutex<HashMap<String, Vec<String>>> = Mutex::new(HashMap::new());
31}
32
33/// Generate a new call-chain ID for the root of a driver -> plugin call that may trigger
34/// callbacks.
35pub fn new_call_chain_id() -> String {
36  Uuid::new_v4().to_string()
37}
38
39/// Current time as Unix epoch milliseconds.
40pub fn now_ms() -> u64 {
41  SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as u64
42}
43
44/// Absolute deadline (Unix epoch milliseconds) `timeout` from now.
45pub fn deadline_from(timeout: Duration) -> u64 {
46  now_ms() + timeout.as_millis() as u64
47}
48
49/// Absolute deadline (Unix epoch milliseconds) `DEFAULT_CALL_CHAIN_TIMEOUT` from now, for
50/// starting a new call chain.
51pub fn default_deadline_ms() -> u64 {
52  deadline_from(DEFAULT_CALL_CHAIN_TIMEOUT)
53}
54
55/// True if `deadline_ms` has already passed.
56pub fn is_expired(deadline_ms: u64) -> bool {
57  now_ms() >= deadline_ms
58}
59
60/// Remaining budget until `deadline_ms`, for use as the timeout of the next hop's own call.
61/// Zero if the deadline has already passed.
62pub fn remaining(deadline_ms: u64) -> Duration {
63  Duration::from_millis(deadline_ms.saturating_sub(now_ms()))
64}
65
66/// Holds `entry_key`'s place on `chain_id`'s call stack until dropped, at which point it is
67/// popped back off. Returned by [`push_call`].
68#[derive(Debug)]
69pub struct CallChainGuard {
70  chain_id: String,
71  entry_key: String,
72}
73
74impl Drop for CallChainGuard {
75  fn drop(&mut self) {
76    pop_call(&self.chain_id, &self.entry_key);
77  }
78}
79
80/// Push `entry_key` onto `chain_id`'s call stack, detecting cycles.
81///
82/// Call this before dispatching a callback for `entry_key` under `chain_id`; hold onto the
83/// returned guard for the duration of the dispatch so `entry_key` is popped back off when it
84/// completes, however it completes. Returns `Err` describing the current chain if `entry_key` is
85/// already on the stack - the same capability being invoked again before an earlier invocation of
86/// it has returned, i.e. a cycle - and dispatch should not proceed.
87pub fn push_call(chain_id: &str, entry_key: &str) -> Result<CallChainGuard, String> {
88  let mut chains = CALL_CHAINS.lock().expect("CALL_CHAINS mutex poisoned");
89  let stack = chains.entry(chain_id.to_string()).or_default();
90  if stack.iter().any(|key| key == entry_key) {
91    return Err(format!(
92      "Cycle detected calling '{}': already in call chain {:?}", entry_key, stack
93    ));
94  }
95  stack.push(entry_key.to_string());
96  Ok(CallChainGuard { chain_id: chain_id.to_string(), entry_key: entry_key.to_string() })
97}
98
99fn pop_call(chain_id: &str, entry_key: &str) {
100  let mut chains = CALL_CHAINS.lock().expect("CALL_CHAINS mutex poisoned");
101  if let Some(stack) = chains.get_mut(chain_id) {
102    if let Some(pos) = stack.iter().rposition(|key| key == entry_key) {
103      stack.remove(pos);
104    }
105    if stack.is_empty() {
106      chains.remove(chain_id);
107    }
108  }
109}
110
111#[cfg(test)]
112mod tests {
113  use expectest::prelude::*;
114
115  use super::*;
116
117  #[test]
118  fn push_call_succeeds_for_a_new_entry_and_pops_on_drop() {
119    let chain_id = "push_call_succeeds_for_a_new_entry_and_pops_on_drop";
120
121    {
122      let _guard = push_call(chain_id, "content-matcher/xml")
123        .expect("expected the first push for a fresh chain to succeed");
124      expect!(push_call(chain_id, "content-matcher/csv")).to(be_ok());
125    }
126
127    // both guards have dropped, so the chain should be gone and the keys reusable
128    expect!(push_call(chain_id, "content-matcher/xml")).to(be_ok());
129  }
130
131  #[test]
132  fn push_call_rejects_a_repeated_entry_key_in_the_same_chain() {
133    let chain_id = "push_call_rejects_a_repeated_entry_key_in_the_same_chain";
134    let _guard = push_call(chain_id, "content-matcher/xml").unwrap();
135
136    let result = push_call(chain_id, "content-matcher/xml");
137
138    expect!(result.is_err()).to(be_true());
139    expect!(result.unwrap_err()).to(be_equal_to("Cycle detected calling 'content-matcher/xml': already in call chain [\"content-matcher/xml\"]".to_string()));
140  }
141
142  #[test]
143  fn push_call_allows_the_same_entry_key_in_different_chains() {
144    let key = "content-matcher/xml";
145    let _guard_a = push_call("chain-a-push_call_allows_the_same_entry_key_in_different_chains", key).unwrap();
146
147    expect!(push_call("chain-b-push_call_allows_the_same_entry_key_in_different_chains", key)).to(be_ok());
148  }
149
150  #[test]
151  fn deadline_helpers_compute_expiry_and_remaining_budget() {
152    let future_deadline = deadline_from(Duration::from_secs(60));
153    expect!(is_expired(future_deadline)).to(be_false());
154    expect!(remaining(future_deadline).as_secs() > 0).to(be_true());
155
156    let past_deadline = now_ms().saturating_sub(1_000);
157    expect!(is_expired(past_deadline)).to(be_true());
158    expect!(remaining(past_deadline)).to(be_equal_to(Duration::ZERO));
159  }
160}
pact_plugin_driver/call_chain.rs

pact_plugin_driver/
call_chain.rs
