forge-ops-tracker 0.8.1

// A small in-process worker thread + bounded channel, so delivery never blocks the caller that
// reported the error and never depends on the host app having any particular job backend
// configured. Ported from gems/forge_ops_tracker/lib/forge_ops_tracker/delivery_queue.rb.
//
// Unlike the Ruby gem and Python client, which start their worker lazily on first push
// specifically so a prefork server (Puma, Gunicorn) forking after the module has already loaded
// doesn't leave a dead thread in every forked child, this starts its worker eagerly in `new`:
// the same choice the .NET SDK makes, and for the same reason: Rust programs essentially never
// fork themselves at the application level, so that hazard doesn't apply here, and eager start
// keeps DeliveryQueue itself simpler (no lazy-init synchronization to get right).

use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::mpsc::{sync_channel, SyncSender, TrySendError};
use std::sync::Arc;
use std::thread;

use crate::client::Client;
use crate::event_builder::Event;

pub struct DeliveryQueue {
    sender: SyncSender<Event>,
}

impl DeliveryQueue {
    pub fn new(queue_size: usize, client: Arc<Client>) -> Self {
        let (sender, receiver) = sync_channel::<Event>(queue_size.max(1));

        thread::Builder::new()
            .name("forge-ops-tracker-delivery".to_string())
            .spawn(move || {
                for event in receiver {
                    // Per-item, not wrapping the whole loop: one bad delivery must not kill the
                    // worker for every event after it. Client::deliver already turns every
                    // failure mode of its own into a plain `false` return rather than a panic;
                    // this is a second, redundant layer of safety around it.
                    let _ = catch_unwind(AssertUnwindSafe(|| {
                        client.deliver(&event);
                    }));
                }
            })
            .expect("failed to spawn the forge-ops-tracker delivery thread");

        DeliveryQueue { sender }
    }

    /// Enqueues event for delivery, returning false (and dropping it) if the queue is already
    /// full rather than blocking the caller.
    pub fn push(&self, event: Event) -> bool {
        match self.sender.try_send(event) {
            Ok(()) => true,
            Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => false,
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::{Arc, RwLock};
    use std::time::Duration;

    use super::*;
    use crate::configuration::Configuration;

    fn test_event(n: i64) -> Event {
        let mut context = HashMap::new();
        context.insert(
            "n".to_string(),
            crate::pii_scrubber::Value::Number(n as f64),
        );
        Event {
            exception_class: "Error".to_string(),
            message: "boom".to_string(),
            backtrace: vec![],
            occurred_at: "2024-01-15T10:30:00Z".to_string(),
            environment: "production".to_string(),
            release: None,
            server_name: None,
            context,
            tags: HashMap::new(),
            sdk_name: "rust".to_string(),
            user: None,
            breadcrumbs: vec![],
        }
    }

    fn client_targeting(addr: std::net::SocketAddr) -> Arc<Client> {
        let mut config = Configuration::new();
        config.dsn = Some(format!("http://key@{addr}/events"));
        config.timeout = Duration::from_secs(5);
        Arc::new(Client::new(Arc::new(RwLock::new(config))))
    }

    #[test]
    fn push_delivers_in_background() {
        use std::io::Write;
        use std::net::TcpListener;
        use std::sync::atomic::{AtomicI32, Ordering};

        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        let received = Arc::new(AtomicI32::new(0));
        let received_clone = Arc::clone(&received);

        thread::spawn(move || {
            for stream in listener.incoming() {
                let mut stream = match stream {
                    Ok(s) => s,
                    Err(_) => break,
                };
                crate::test_support::read_full_request(&mut stream);
                received_clone.fetch_add(1, Ordering::SeqCst);
                let _ = stream.write_all(
                    b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}",
                );
            }
        });

        let client = client_targeting(addr);
        let queue = DeliveryQueue::new(10, client);

        queue.push(test_event(1));
        queue.push(test_event(2));

        let deadline = std::time::Instant::now() + Duration::from_secs(2);
        while received.load(Ordering::SeqCst) < 2 && std::time::Instant::now() < deadline {
            thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(received.load(Ordering::SeqCst), 2);
    }

    #[test]
    fn drops_when_full() {
        use std::net::TcpListener;

        // A listener that accepts a connection and then never reads/responds, so the worker
        // thread stays busy (blocked inside ureq waiting on the response) long enough for the
        // queue behind it to actually fill up.
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                std::mem::forget(stream); // hold the connection open without ever responding
            }
        });

        let mut config = Configuration::new();
        config.dsn = Some(format!("http://key@{addr}/events"));
        config.timeout = Duration::from_secs(5);
        let client = Arc::new(Client::new(Arc::new(RwLock::new(config))));
        let queue = DeliveryQueue::new(1, client);

        assert!(
            queue.push(test_event(1)),
            "first push should have succeeded"
        );
        thread::sleep(Duration::from_millis(100)); // let the worker pick it up and start blocking

        assert!(
            queue.push(test_event(2)),
            "second push should have filled the size-1 queue"
        );
        assert!(
            !queue.push(test_event(3)),
            "third push should have been dropped: queue was full"
        );
    }
}