1use super::error::Result;
2#[cfg(feature = "http")]
3use super::error::{Error, ErrorKind};
4use serde_json::Value;
5#[cfg(feature = "http")]
6use serde_json::json;
7#[cfg(feature = "http")]
8use sha2::{Digest, Sha256};
9#[cfg(feature = "http")]
10use std::{
11 fs::OpenOptions,
12 io::Write,
13 path::Path,
14 sync::{Arc, Mutex},
15 time::{Duration, SystemTime, UNIX_EPOCH},
16};
17
18pub trait Transport {
19 fn call(&self, rpc: &str, method: &str, params: Value) -> Result<Value>;
20}
21
22#[cfg(feature = "http")]
23const MAX_RPC_ATTEMPTS: usize = 4;
24
25#[cfg(feature = "http")]
26#[derive(Clone)]
27pub struct HttpTransport {
28 agent: ureq::Agent,
29 trace: Option<Arc<Mutex<RpcTraceWriter>>>,
30}
31
32#[cfg(feature = "http")]
33#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
34pub enum RpcTraceMode {
35 #[default]
36 Full,
37 Summary,
38 RequestOnly,
39 ResponseMeta,
40}
41
42#[cfg(feature = "http")]
43struct RpcTraceWriter {
44 file: std::fs::File,
45 sequence: u64,
46 mode: RpcTraceMode,
47}
48
49#[cfg(feature = "http")]
50impl Default for HttpTransport {
51 fn default() -> Self {
52 Self::new()
53 }
54}
55
56#[cfg(feature = "http")]
57impl HttpTransport {
58 pub fn new() -> Self {
59 let config = ureq::Agent::config_builder()
60 .timeout_global(Some(Duration::from_secs(30)))
61 .http_status_as_error(false)
62 .build();
63 Self {
64 agent: ureq::Agent::new_with_config(config),
65 trace: None,
66 }
67 }
68
69 pub fn with_trace_jsonl(path: &Path) -> Result<Self> {
70 Self::with_trace_jsonl_mode(path, RpcTraceMode::Full)
71 }
72
73 pub fn with_trace_jsonl_mode(path: &Path, mode: RpcTraceMode) -> Result<Self> {
74 let mut transport = Self::new();
75 if let Some(parent) = path
76 .parent()
77 .filter(|parent| !parent.as_os_str().is_empty())
78 {
79 std::fs::create_dir_all(parent).map_err(|error| {
80 Error::with_kind(
81 ErrorKind::Io,
82 format!("creating RPC trace directory {}: {error}", parent.display()),
83 )
84 })?;
85 }
86 let file = OpenOptions::new()
87 .create(true)
88 .truncate(true)
89 .write(true)
90 .open(path)
91 .map_err(|error| {
92 Error::with_kind(
93 ErrorKind::Io,
94 format!("opening RPC trace {}: {error}", path.display()),
95 )
96 })?;
97 transport.trace = Some(Arc::new(Mutex::new(RpcTraceWriter {
98 file,
99 sequence: 0,
100 mode,
101 })));
102 Ok(transport)
103 }
104
105 fn trace_rpc(
106 &self,
107 rpc: &str,
108 method: &str,
109 request: &Value,
110 response: Option<&Value>,
111 http_status: Option<u16>,
112 error: Option<&str>,
113 ) {
114 let Some(trace) = &self.trace else {
115 return;
116 };
117 let Ok(mut trace) = trace.lock() else {
118 return;
119 };
120 trace.sequence += 1;
121 let mut event = json!({
122 "schema": "octra-sqlite.rpc-trace.v1",
123 "mode": trace_mode_name(trace.mode),
124 "sequence": trace.sequence,
125 "timestamp_ms": unix_timestamp_ms(),
126 "rpc": rpc,
127 "method": method,
128 "http_status": http_status,
129 "ok": error.is_none(),
130 "error": error,
131 });
132 match trace.mode {
133 RpcTraceMode::Full => {
134 event["request"] = request.clone();
135 event["response"] = response.cloned().unwrap_or(Value::Null);
136 event["request_meta"] = trace_value_meta(request);
137 event["response_meta"] = trace_optional_value_meta(response);
138 }
139 RpcTraceMode::Summary => {
140 let request_meta = trace_value_meta(request);
141 let response_meta = trace_optional_value_meta(response);
142 event["request_bytes"] = request_meta["bytes"].clone();
143 event["request_sha256"] = request_meta["sha256"].clone();
144 event["response_bytes"] = response_meta["bytes"].clone();
145 event["response_sha256"] = response_meta["sha256"].clone();
146 }
147 RpcTraceMode::RequestOnly => {
148 event["request"] = request.clone();
149 event["response_meta"] = trace_optional_value_meta(response);
150 }
151 RpcTraceMode::ResponseMeta => {
152 event["request_meta"] = trace_value_meta(request);
153 event["response_meta"] = trace_optional_value_meta(response);
154 }
155 }
156 if serde_json::to_writer(&mut trace.file, &event).is_err() {
157 return;
158 }
159 let _ = trace.file.write_all(b"\n");
160 }
161}
162
163#[cfg(feature = "http")]
164impl Transport for HttpTransport {
165 fn call(&self, rpc: &str, method: &str, params: Value) -> Result<Value> {
166 let body = json!({
167 "jsonrpc": "2.0",
168 "id": 1,
169 "method": method,
170 "params": params,
171 });
172 let retryable = method != "octra_submit";
173 for attempt in 1..=MAX_RPC_ATTEMPTS {
174 let mut response = match self.agent.post(rpc).send_json(&body) {
175 Ok(response) => response,
176 Err(error) => {
177 let message = format!("calling {method}: {error}");
178 self.trace_rpc(rpc, method, &body, None, None, Some(&message));
179 if should_retry_transport(attempt, retryable) {
180 sleep_before_retry(attempt, None);
181 continue;
182 }
183 return Err(Error::with_kind(ErrorKind::Transport, message));
184 }
185 };
186 let status = response.status();
187 let status_code = status.as_u16();
188 let retry_after = response
189 .headers()
190 .get("retry-after")
191 .and_then(|value| value.to_str().ok())
192 .and_then(parse_retry_after);
193 let text = match response.body_mut().read_to_string() {
194 Ok(text) => text,
195 Err(error) => {
196 let message = format!("reading {method} response body: {error}");
197 self.trace_rpc(rpc, method, &body, None, Some(status_code), Some(&message));
198 if should_retry_http(attempt, retryable, status_code) {
199 sleep_before_retry(attempt, retry_after);
200 continue;
201 }
202 return Err(Error::with_kind(ErrorKind::Transport, message));
203 }
204 };
205 let payload: Value = match serde_json::from_str(&text) {
206 Ok(payload) => payload,
207 Err(error) => {
208 let message = format!(
209 "decoding {method} non-JSON response from HTTP {status}: {error}; body: {}",
210 preview_text(&text, 512)
211 );
212 self.trace_rpc(rpc, method, &body, None, Some(status_code), Some(&message));
213 if should_retry_http(attempt, retryable, status_code)
214 || should_retry_non_json(attempt, retryable, &text)
215 {
216 sleep_before_retry(attempt, retry_after);
217 continue;
218 }
219 return Err(Error::with_kind(ErrorKind::Decode, message));
220 }
221 };
222 if !status.is_success() {
223 let message = format!(
224 "{method} failed with HTTP {status}: {}",
225 preview_value(&payload, 1024)
226 );
227 self.trace_rpc(
228 rpc,
229 method,
230 &body,
231 Some(&payload),
232 Some(status_code),
233 Some(&message),
234 );
235 if should_retry_http(attempt, retryable, status_code) {
236 sleep_before_retry(attempt, retry_after);
237 continue;
238 }
239 return Err(Error::with_kind(ErrorKind::Transport, message));
240 }
241 if let Some(error) = payload.get("error") {
242 let message = format!("{method} failed: {}", preview_value(error, 1024));
243 self.trace_rpc(
244 rpc,
245 method,
246 &body,
247 Some(&payload),
248 Some(status_code),
249 Some(&message),
250 );
251 if should_retry_rpc_error(attempt, retryable, error) {
252 sleep_before_retry(attempt, retry_after);
253 continue;
254 }
255 return Err(Error::with_kind(ErrorKind::Rpc, message));
256 }
257 self.trace_rpc(rpc, method, &body, Some(&payload), Some(status_code), None);
258 return Ok(payload.get("result").cloned().unwrap_or(Value::Null));
259 }
260 Err(Error::with_kind(
261 ErrorKind::Transport,
262 format!("{method} failed after retry attempts"),
263 ))
264 }
265}
266
267#[cfg(feature = "http")]
268fn should_retry_transport(attempt: usize, retryable: bool) -> bool {
269 retryable && attempt < MAX_RPC_ATTEMPTS
270}
271
272#[cfg(feature = "http")]
273fn should_retry_http(attempt: usize, retryable: bool, status_code: u16) -> bool {
274 should_retry_transport(attempt, retryable)
275 && matches!(status_code, 408 | 425 | 429 | 500 | 502 | 503 | 504)
276}
277
278#[cfg(feature = "http")]
279fn should_retry_non_json(attempt: usize, retryable: bool, text: &str) -> bool {
280 let first = text.trim_start().chars().next();
281 should_retry_transport(attempt, retryable) && matches!(first, Some('<' | '\u{feff}'))
282}
283
284#[cfg(feature = "http")]
285fn should_retry_rpc_error(attempt: usize, retryable: bool, error: &Value) -> bool {
286 if !should_retry_transport(attempt, retryable) {
287 return false;
288 }
289 let code = error.get("code").and_then(Value::as_i64);
290 let message = error
291 .get("message")
292 .and_then(Value::as_str)
293 .unwrap_or_default()
294 .to_ascii_lowercase();
295 matches!(code, Some(429) | Some(-32029))
296 || message.contains("too many requests")
297 || message.contains("rate limit")
298 || message.contains("temporarily unavailable")
299}
300
301#[cfg(feature = "http")]
302fn parse_retry_after(value: &str) -> Option<Duration> {
303 let seconds = value.trim().parse::<u64>().ok()?;
304 Some(Duration::from_secs(seconds.min(30)))
305}
306
307#[cfg(feature = "http")]
308fn sleep_before_retry(attempt: usize, retry_after: Option<Duration>) {
309 std::thread::sleep(retry_after.unwrap_or_else(|| {
310 let millis = 500_u64.saturating_mul(2_u64.saturating_pow((attempt - 1) as u32));
311 Duration::from_millis(millis.min(5_000))
312 }));
313}
314
315#[cfg(feature = "http")]
316fn preview_value(value: &Value, limit: usize) -> String {
317 preview_text(
318 &serde_json::to_string(value).unwrap_or_else(|_| value.to_string()),
319 limit,
320 )
321}
322
323#[cfg(feature = "http")]
324fn preview_text(text: &str, limit: usize) -> String {
325 let compact = text.split_whitespace().collect::<Vec<_>>().join(" ");
326 if compact.len() <= limit {
327 compact
328 } else {
329 format!(
330 "{}...<truncated {} bytes>",
331 truncate_to_char_boundary(&compact, limit),
332 compact.len()
333 )
334 }
335}
336
337#[cfg(feature = "http")]
338fn truncate_to_char_boundary(text: &str, limit: usize) -> &str {
339 if text.len() <= limit {
340 return text;
341 }
342 let mut end = 0usize;
343 for (index, _) in text.char_indices() {
344 if index > limit {
345 break;
346 }
347 end = index;
348 }
349 &text[..end]
350}
351
352#[cfg(feature = "http")]
353fn unix_timestamp_ms() -> u128 {
354 SystemTime::now()
355 .duration_since(UNIX_EPOCH)
356 .unwrap_or_default()
357 .as_millis()
358}
359
360#[cfg(feature = "http")]
361fn trace_mode_name(mode: RpcTraceMode) -> &'static str {
362 match mode {
363 RpcTraceMode::Full => "full",
364 RpcTraceMode::Summary => "summary",
365 RpcTraceMode::RequestOnly => "request_only",
366 RpcTraceMode::ResponseMeta => "response_meta",
367 }
368}
369
370#[cfg(feature = "http")]
371fn trace_optional_value_meta(value: Option<&Value>) -> Value {
372 match value {
373 Some(value) => trace_value_meta(value),
374 None => json!({
375 "bytes": null,
376 "sha256": null,
377 }),
378 }
379}
380
381#[cfg(feature = "http")]
382fn trace_value_meta(value: &Value) -> Value {
383 let bytes = serde_json::to_vec(value).unwrap_or_default();
384 json!({
385 "bytes": bytes.len(),
386 "sha256": hex::encode(Sha256::digest(&bytes)),
387 })
388}
389
390#[cfg(all(test, feature = "http"))]
391mod tests {
392 use super::*;
393 use serde_json::json;
394
395 #[test]
396 fn trace_writer_records_json_rpc_request_and_response() {
397 let path = std::env::temp_dir().join(format!(
398 "octra-sqlite-rpc-trace-{}.jsonl",
399 std::process::id()
400 ));
401 let _ = std::fs::remove_file(&path);
402 let transport = HttpTransport::with_trace_jsonl(&path).unwrap();
403 let request = json!({
404 "jsonrpc": "2.0",
405 "id": 1,
406 "method": "octra_circleViewAuth",
407 "params": ["octCircle"]
408 });
409 let response = json!({
410 "jsonrpc": "2.0",
411 "id": 1,
412 "result": "ok"
413 });
414
415 transport.trace_rpc(
416 "https://devnet.octrascan.io/rpc",
417 "octra_circleViewAuth",
418 &request,
419 Some(&response),
420 Some(200),
421 None,
422 );
423
424 let text = std::fs::read_to_string(&path).unwrap();
425 let event: Value = serde_json::from_str(text.trim()).unwrap();
426 assert_eq!(event["schema"], "octra-sqlite.rpc-trace.v1");
427 assert_eq!(event["mode"], "full");
428 assert_eq!(event["sequence"], 1);
429 assert_eq!(event["method"], "octra_circleViewAuth");
430 assert_eq!(event["request"], request);
431 assert_eq!(event["response"], response);
432 assert_eq!(
433 event["request_meta"]["bytes"],
434 serde_json::to_vec(&request).unwrap().len()
435 );
436 assert_eq!(event["request_meta"]["sha256"].as_str().unwrap().len(), 64);
437 assert!(event["error"].is_null());
438 let _ = std::fs::remove_file(path);
439 }
440
441 #[test]
442 fn trace_writer_summary_omits_rpc_bodies() {
443 let path = std::env::temp_dir().join(format!(
444 "octra-sqlite-rpc-trace-summary-{}.jsonl",
445 std::process::id()
446 ));
447 let _ = std::fs::remove_file(&path);
448 let transport = HttpTransport::with_trace_jsonl_mode(&path, RpcTraceMode::Summary).unwrap();
449 transport.trace_rpc(
450 "https://devnet.octrascan.io/rpc",
451 "octra_circleViewAuth",
452 &json!({"params": ["select * from artist;"]}),
453 Some(&json!({"result": {"rows": [[1]]}})),
454 Some(200),
455 None,
456 );
457
458 let text = std::fs::read_to_string(&path).unwrap();
459 let event: Value = serde_json::from_str(text.trim()).unwrap();
460 assert_eq!(event["mode"], "summary");
461 assert!(event.get("request").is_none());
462 assert!(event.get("response").is_none());
463 assert!(event["request_bytes"].as_u64().unwrap() > 0);
464 assert_eq!(event["request_sha256"].as_str().unwrap().len(), 64);
465 assert!(event["response_bytes"].as_u64().unwrap() > 0);
466 assert_eq!(event["response_sha256"].as_str().unwrap().len(), 64);
467 let _ = std::fs::remove_file(path);
468 }
469
470 #[test]
471 fn trace_writer_is_best_effort_after_open() {
472 let path = std::env::temp_dir().join(format!(
473 "octra-sqlite-rpc-trace-poison-{}.jsonl",
474 std::process::id()
475 ));
476 let _ = std::fs::remove_file(&path);
477 let transport = HttpTransport::with_trace_jsonl(&path).unwrap();
478 let trace = transport.trace.as_ref().unwrap().clone();
479 let _ = std::thread::spawn(move || {
480 let _guard = trace.lock().unwrap();
481 panic!("poison trace lock");
482 })
483 .join();
484
485 transport.trace_rpc(
486 "https://devnet.octrascan.io/rpc",
487 "octra_circleViewAuth",
488 &json!({"params": []}),
489 None,
490 None,
491 Some("real rpc error"),
492 );
493
494 let _ = std::fs::remove_file(path);
495 }
496
497 #[test]
498 fn retry_policy_is_mainnet_safe() {
499 assert!(should_retry_http(1, true, 429));
500 assert!(should_retry_http(1, true, 503));
501 assert!(!should_retry_http(MAX_RPC_ATTEMPTS, true, 429));
502 assert!(!should_retry_http(1, false, 429));
503 assert!(should_retry_rpc_error(
504 1,
505 true,
506 &json!({"code":429,"message":"Too Many Requests"})
507 ));
508 assert!(should_retry_rpc_error(
509 1,
510 true,
511 &json!({"message":"temporarily unavailable"})
512 ));
513 assert!(!should_retry_rpc_error(
514 1,
515 true,
516 &json!({"code":-32000,"message":"wasm export returned 1"})
517 ));
518 }
519
520 #[test]
521 fn response_previews_are_compact_and_utf8_safe() {
522 let text = "αβγδε ζηθικ λμνξο";
523 let preview = preview_text(text, 7);
524 assert!(preview.contains("<truncated"));
525 assert!(preview.is_char_boundary(preview.len()));
526 assert!(preview_text("<html> too many requests </html>", 512).contains("<html>"));
527 assert!(should_retry_non_json(1, true, " <html>429</html>"));
528 }
529}