atscan.net
High-performance implementation of plcbundle written in Rust
refactor(rate-limiter): improve rate limiting with better logging and refill calculation
Add detailed logging for rate limiter state before requests
Move rate limiter wait logging to be consistent across endpoints
Calculate refill rate using floating point division for precision
Add available_permits method for monitoring
tree.fail
+25
-17
+25
-17
src/plc_client.rs
+25
-17
src/plc_client.rs
···
82
82
let mut last_err = None;
83
83
84
84
for attempt in 1..=max_retries {
85
-
// Wait for rate limiter token
85
+
let export_url = format!(
86
+
"{}/export?after={}&count={}",
87
+
self.base_url, after, count
88
+
);
89
+
90
+
let permits = self.rate_limiter.available_permits();
91
+
let requests_in_period = self.count_requests_in_period();
92
+
log::debug!(
93
+
"[PLCClient] Preparing /export request: {} | permits={} | window={:?} | requests_in_window={}",
94
+
export_url,
95
+
permits,
96
+
self.rate_limit_period,
97
+
requests_in_period
98
+
);
99
+
100
+
let wait_start = Instant::now();
86
101
self.rate_limiter.wait().await;
102
+
let wait_elapsed = wait_start.elapsed();
103
+
if wait_elapsed.as_nanos() > 0 {
104
+
log::debug!("[PLCClient] Rate limiter wait: {:?}", wait_elapsed);
105
+
}
87
106
88
107
// Clear previous retry_after
89
108
*self.last_retry_after.lock().await = None;
···
195
214
// PLC directory exposes DID documents at /{did} (same as plcbundle instances)
196
215
let url = format!("{}/{}", self.base_url.trim_end_matches('/'), did);
197
216
198
-
log::debug!("Waiting for rate limiter token...");
199
-
let wait_start = Instant::now();
200
-
self.rate_limiter.wait().await;
201
-
let wait_duration = wait_start.elapsed();
202
-
if wait_duration.as_millis() > 0 {
203
-
log::debug!("Rate limiter wait: {:?}", wait_duration);
204
-
}
205
-
206
-
// Record this request attempt
207
-
self.record_request();
208
-
let request_count = self.count_requests_in_period();
209
-
log::debug!("Request count in last period: {}", request_count);
210
-
211
217
log::debug!("Fetching DID document from: {}", url);
212
218
let request_start = Instant::now();
213
219
···
307
313
let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(0));
308
314
let sem_clone = semaphore.clone();
309
315
310
-
// Calculate refill rate: period / requests_per_period
311
-
// For 72 req/min: 60 seconds / 72 = 0.833 seconds per request
312
-
let refill_rate = period / requests_per_period as u32;
316
+
let refill_rate = Duration::from_secs_f64(period.as_secs_f64() / requests_per_period as f64);
313
317
314
318
// Spawn background task to refill permits at steady rate
315
319
// CRITICAL: Add first permit immediately, then refill at steady rate
···
334
338
// Wait for a permit to become available
335
339
// This will block until a permit is available (from refill task)
336
340
let _ = self.semaphore.acquire().await;
341
+
}
342
+
343
+
fn available_permits(&self) -> usize {
344
+
self.semaphore.available_permits()
337
345
}
338
346
}
339
347