Mountain/Vine/Client/
SendRequest.rs

1#![allow(non_snake_case)]
2
3//! Send a request and await a response. Validates method-name length
4//! and message size, prefers the streaming multiplexer when
5//! `LAND_VINE_STREAMING=1` is on (falls through to unary on any failure
6//! except the authoritative streaming-path timeout), enforces a per-call
7//! timeout via `tokio::time::timeout`, and updates per-connection
8//! activity / failure metadata on completion.
9
10use std::time::Duration;
11
12use serde_json::{Value, from_slice, to_vec};
13use tokio::time::timeout;
14
15use crate::{
16	Vine::{
17		Client::{
18			IsShuttingDown,
19			Shared::{
20				DEFAULT_TIMEOUT_MS,
21				RecordSideCarFailure,
22				SIDECAR_CLIENTS,
23				UpdateSideCarActivity,
24				ValidateMessageSize,
25			},
26		},
27		Error::VineError,
28		Generated::GenericRequest,
29	},
30	dev_log,
31};
32
33pub async fn Fn(
34	SideCarIdentifier:&str,
35
36	Method:String,
37
38	Parameters:Value,
39
40	TimeoutMilliseconds:u64,
41) -> Result<Value, VineError> {
42	if IsShuttingDown::Fn() {
43		return Err(VineError::ClientNotConnected(SideCarIdentifier.to_string()));
44	}
45
46	if Method.is_empty() || Method.len() > 128 {
47		return Err(VineError::RPCError(
48			"Method name must be between 1 and 128 characters".to_string(),
49		));
50	}
51
52	let TimeoutDuration =
53		Duration::from_millis(if TimeoutMilliseconds > 0 { TimeoutMilliseconds } else { DEFAULT_TIMEOUT_MS });
54
55	if std::env::var("LAND_VINE_STREAMING").as_deref() == Ok("1") {
56		if let Some(Mux) = crate::Vine::Multiplexer::Multiplexer::Lookup(SideCarIdentifier) {
57			if !Mux.IsClosed() {
58				match Mux.Request(Method.clone(), Parameters.clone(), TimeoutDuration).await {
59					Ok(Result_) => {
60						UpdateSideCarActivity(SideCarIdentifier);
61
62						return Ok(Result_);
63					},
64
65					Err(VineError::RequestTimeout { .. }) => {
66						return Err(VineError::RequestTimeout {
67							SideCarIdentifier:SideCarIdentifier.to_string(),
68							MethodName:Method,
69							TimeoutMilliseconds:TimeoutDuration.as_millis() as u64,
70						});
71					},
72
73					Err(Error) => {
74						dev_log!(
75							"grpc",
76							"warn: [VineClient::SendRequest] streaming send failed for '{}::{}' ({}); falling back to \
77							 unary",
78							SideCarIdentifier,
79							Method,
80							Error
81						);
82					},
83				}
84			}
85		}
86	}
87
88	let ParameterBytes =
89		to_vec(&Parameters).map_err(|E| VineError::RPCError(format!("Failed to serialize parameters: {}", E)))?;
90
91	ValidateMessageSize(&ParameterBytes)?;
92
93	let Client = {
94		let Pool = SIDECAR_CLIENTS.lock();
95
96		Pool.get(SideCarIdentifier).cloned()
97	};
98
99	let Some(mut Client) = Client else {
100		return Err(VineError::ClientNotConnected(SideCarIdentifier.to_string()));
101	};
102
103	let RequestIdentifier = std::time::SystemTime::now()
104		.duration_since(std::time::UNIX_EPOCH)
105		.unwrap()
106		.as_nanos() as u64;
107
108	let MethodForLog = Method.clone();
109
110	let Request = GenericRequest { request_identifier:RequestIdentifier, method:Method, parameter:ParameterBytes };
111
112	let Result_ = timeout(TimeoutDuration, Client.process_mountain_request(Request)).await;
113
114	match Result_ {
115		Ok(Ok(Response)) => {
116			UpdateSideCarActivity(SideCarIdentifier);
117
118			dev_log!(
119				"grpc",
120				"[VineClient] Request sent successfully to sidecar '{}': method='{}'",
121				SideCarIdentifier,
122				MethodForLog
123			);
124
125			let InnerResponse = Response.into_inner();
126
127			let ResultBytes = InnerResponse.result;
128
129			let ResultValue:Value = from_slice(&ResultBytes)
130				.map_err(|E| VineError::RPCError(format!("Failed to deserialize response: {}", E)))?;
131
132			if let Some(ErrorData) = InnerResponse.error {
133				return Err(VineError::RPCError(format!(
134					"RPC error from sidecar: code={}, message={}",
135					ErrorData.code, ErrorData.message
136				)));
137			}
138
139			Ok(ResultValue)
140		},
141
142		Ok(Err(Status)) => {
143			RecordSideCarFailure(SideCarIdentifier);
144
145			Err(VineError::RPCError(format!("gRPC error: {}", Status)))
146		},
147
148		Err(_) => {
149			RecordSideCarFailure(SideCarIdentifier);
150
151			Err(VineError::RequestTimeout {
152				SideCarIdentifier:SideCarIdentifier.to_string(),
153				MethodName:MethodForLog,
154				TimeoutMilliseconds:TimeoutDuration.as_millis() as u64,
155			})
156		},
157	}
158}
Mountain/Vine/Client/SendRequest.rs

Mountain/Vine/Client/
SendRequest.rs
