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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
//! Browser Pool - Manages multiple browser instances for concurrent auditing
//!
//! Provides a pool of browser pages that can be checked out for use,
//! enabling concurrent URL processing while managing resource usage.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use chromiumoxide::Page;
use tokio::sync::{Mutex, Semaphore};
use tracing::{debug, info, warn};
use super::manager::{BrowserManager, BrowserOptions};
use crate::error::{AuditError, Result};
/// Configuration for the browser pool
#[derive(Debug, Clone)]
pub struct PoolConfig {
/// Maximum number of concurrent browser pages
pub max_pages: usize,
/// Browser options for all instances
pub browser_options: BrowserOptions,
/// Timeout for acquiring a page from the pool
pub acquire_timeout_secs: u64,
}
impl Default for PoolConfig {
fn default() -> Self {
Self {
max_pages: 4,
browser_options: BrowserOptions::default(),
acquire_timeout_secs: 60,
}
}
}
/// A pooled page that automatically returns to the pool when dropped
pub struct PooledPage {
/// The underlying page
page: Option<Page>,
/// Reference to the pool for returning the page
pool: Arc<BrowserPoolInner>,
}
impl PooledPage {
/// Get a reference to the underlying page
pub fn page(&self) -> Result<&Page> {
self.page.as_ref().ok_or_else(|| {
crate::error::AuditError::ConfigError("Page already returned to pool".to_string())
})
}
}
impl Drop for PooledPage {
fn drop(&mut self) {
if let Some(page) = self.page.take() {
// `tokio::spawn` panics if no runtime is active for this thread —
// normally guaranteed for the program's whole lifetime, but not
// during process shutdown/panic-unwind after the runtime itself
// has already been torn down (#QA-041). Skip the spawn rather
// than panic in that case; the page (and its Chrome tab) is
// simply leaked, same as if the process were killed outright.
if tokio::runtime::Handle::try_current().is_ok() {
let pool = Arc::clone(&self.pool);
tokio::spawn(async move {
pool.return_page(page).await;
});
} else {
warn!("No active tokio runtime; dropping page without returning it to the pool");
}
}
}
}
/// Inner pool state
struct BrowserPoolInner {
/// The browser manager
browser: BrowserManager,
/// Available pages
pages: Mutex<Vec<Page>>,
/// Semaphore for limiting concurrent pages
semaphore: Semaphore,
/// Total pages created
pages_created: AtomicUsize,
/// Maximum pages allowed
max_pages: usize,
/// Acquire timeout
acquire_timeout: Duration,
}
impl BrowserPoolInner {
/// Return a page to the pool
async fn return_page(&self, page: Page) {
// Reset the page for reuse. Guard with a short timeout — some pages become
// unresponsive after navigating to heavy sites and the goto call would hang
// indefinitely, starving the semaphore.
let reset = tokio::time::timeout(Duration::from_secs(10), page.goto("about:blank")).await;
match reset {
Ok(Ok(_)) => {
let mut pages = self.pages.lock().await;
pages.push(page);
self.semaphore.add_permits(1);
debug!("Page returned to pool ({} available)", pages.len());
}
Ok(Err(e)) => {
warn!("Failed to reset page: {}", e);
self.discard_page(page).await;
}
Err(_) => {
warn!("Page reset timed out after 10s, discarding page");
self.discard_page(page).await;
}
}
}
/// Discard a page that couldn't be reset for reuse: close its Chrome tab
/// and give back its slot in `pages_created`, not just its semaphore
/// permit (#QA-040). Without the `pages_created` decrement, every page
/// lost this way permanently shrinks the pool's effective capacity —
/// `acquire()`'s "shouldn't happen" exhaustion branch was actually
/// reachable once enough pages were discarded on a long batch run,
/// collapsing concurrency to zero with free semaphore permits sitting
/// unused. `page.close()` is itself timeout-guarded since a page
/// unresponsive enough to fail the reset above may also fail to close;
/// the bookkeeping happens either way so a hung close can't reintroduce
/// the same exhaustion bug.
async fn discard_page(&self, page: Page) {
match tokio::time::timeout(Duration::from_secs(5), page.close()).await {
Ok(Ok(())) => {}
Ok(Err(e)) => warn!("Failed to close discarded page: {}", e),
Err(_) => warn!("Closing discarded page timed out after 5s"),
}
self.pages_created.fetch_sub(1, Ordering::SeqCst);
self.semaphore.add_permits(1);
}
}
/// Browser Pool - manages multiple browser pages for concurrent processing
pub struct BrowserPool {
inner: Arc<BrowserPoolInner>,
}
impl BrowserPool {
/// Create a new browser pool with the given configuration
///
/// # Arguments
/// * `config` - Pool configuration
///
/// # Returns
/// * `Ok(BrowserPool)` - Pool created successfully
/// * `Err(AuditError)` - Failed to create pool
pub async fn new(config: PoolConfig) -> Result<Self> {
info!("Creating browser pool with max {} pages", config.max_pages);
// Launch the browser
let browser = BrowserManager::with_options(config.browser_options).await?;
let inner = Arc::new(BrowserPoolInner {
browser,
pages: Mutex::new(Vec::with_capacity(config.max_pages)),
semaphore: Semaphore::new(config.max_pages),
pages_created: AtomicUsize::new(0),
max_pages: config.max_pages,
acquire_timeout: Duration::from_secs(config.acquire_timeout_secs),
});
Ok(Self { inner })
}
/// Create a pool with default configuration
pub async fn with_concurrency(concurrency: usize) -> Result<Self> {
Self::new(PoolConfig {
max_pages: concurrency,
..Default::default()
})
.await
}
/// Acquire a page from the pool
///
/// This will block until a page is available or timeout occurs.
///
/// # Returns
/// * `Ok(PooledPage)` - A page from the pool
/// * `Err(AuditError)` - Failed to acquire page
pub async fn acquire(&self) -> Result<PooledPage> {
debug!("Acquiring page from pool...");
// Wait for a permit with timeout
let permit =
tokio::time::timeout(self.inner.acquire_timeout, self.inner.semaphore.acquire())
.await
.map_err(|_| AuditError::PoolTimeout {
timeout_secs: self.inner.acquire_timeout.as_secs(),
})?
.map_err(|_| AuditError::PoolClosed)?;
// Forget the permit since we manage page count manually
permit.forget();
// Try to get an existing page from the pool
let mut pages = self.inner.pages.lock().await;
if let Some(page) = pages.pop() {
debug!("Reusing existing page from pool");
return Ok(PooledPage {
page: Some(page),
pool: Arc::clone(&self.inner),
});
}
drop(pages); // Release the lock before creating a new page
// Create a new page if we haven't hit the limit
let current = self.inner.pages_created.fetch_add(1, Ordering::SeqCst);
if current >= self.inner.max_pages {
// This shouldn't happen due to semaphore, but handle it anyway
self.inner.pages_created.fetch_sub(1, Ordering::SeqCst);
return Err(AuditError::PoolExhausted);
}
debug!(
"Creating new page ({}/{})",
current + 1,
self.inner.max_pages
);
let page = self.inner.browser.new_page().await?;
Ok(PooledPage {
page: Some(page),
pool: Arc::clone(&self.inner),
})
}
/// Get the browser manager for navigation
pub fn browser(&self) -> &BrowserManager {
&self.inner.browser
}
/// Get the number of available pages in the pool
pub async fn available_pages(&self) -> usize {
self.inner.pages.lock().await.len()
}
/// Get pool statistics
pub fn stats(&self) -> PoolStats {
PoolStats {
max_pages: self.inner.max_pages,
pages_created: self.inner.pages_created.load(Ordering::SeqCst),
permits_available: self.inner.semaphore.available_permits(),
}
}
/// Close the pool and release all resources
pub async fn close(self) -> Result<()> {
info!("Closing browser pool...");
// Close all pooled pages
let mut pages = self.inner.pages.lock().await;
for page in pages.drain(..) {
if let Err(e) = page.close().await {
warn!("Failed to close pooled page: {}", e);
}
}
// The browser will be closed when the Arc is dropped
info!("Browser pool closed");
Ok(())
}
}
/// Pool statistics
#[derive(Debug, Clone)]
pub struct PoolStats {
/// Maximum number of pages allowed
pub max_pages: usize,
/// Total pages created
pub pages_created: usize,
/// Currently available permits
pub permits_available: usize,
}
impl std::fmt::Display for PoolStats {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"Pool: {}/{} pages, {} available",
self.pages_created, self.max_pages, self.permits_available
)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_pool_config_default() {
let config = PoolConfig::default();
assert_eq!(config.max_pages, 4);
assert_eq!(config.acquire_timeout_secs, 60);
}
#[test]
fn test_pool_stats_display() {
let stats = PoolStats {
max_pages: 4,
pages_created: 2,
permits_available: 2,
};
let display = format!("{}", stats);
assert!(display.contains("2/4"));
assert!(display.contains("2 available"));
}
}