clicktype-batch 0.2.0

Async batching system for ClickType with backpressure and metrics
Documentation 
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

//! Batch configuration

use std::time::Duration;

/// Batch configuration
#[derive(Debug, Clone, PartialEq)]
pub struct BatchConfig {
    /// Maximum number of rows before automatic flush
    pub max_rows: usize,

    /// Maximum wait time before flush
    pub max_wait: Duration,

    /// Initial buffer size in bytes
    pub initial_buffer_size: usize,

    /// Maximum buffer size before backpressure
    pub max_buffer_size: usize,

    /// Maximum buffer capacity to retain after flush (prevents memory leaks)
    /// If capacity exceeds this value, the buffer will be reallocated.
    /// Defaults to 2 * initial_buffer_size.
    pub buffer_shrink_threshold: usize,

    /// Maximum number of retries on insert failure
    pub max_retries: u32,

    /// Base delay between retries (exponential backoff)
    pub retry_base_delay: Duration,

    /// MPSC channel capacity
    pub channel_capacity: usize,
}

impl Default for BatchConfig {
    fn default() -> Self {
        Self {
            max_rows: 50_000,
            max_wait: Duration::from_secs(5),
            initial_buffer_size: 1024 * 1024,        // 1 MB
            max_buffer_size: 64 * 1024 * 1024,       // 64 MB
            buffer_shrink_threshold: 2 * 1024 * 1024, // 2 MB
            max_retries: 3,
            retry_base_delay: Duration::from_millis(100),
            channel_capacity: 10_000,
        }
    }
}

impl BatchConfig {
    /// High throughput preset
    pub fn high_throughput() -> Self {
        Self {
            max_rows: 100_000,
            max_wait: Duration::from_secs(10),
            initial_buffer_size: 4 * 1024 * 1024,    // 4 MB
            max_buffer_size: 256 * 1024 * 1024,      // 256 MB
            buffer_shrink_threshold: 8 * 1024 * 1024, // 8 MB
            channel_capacity: 100_000,
            ..Default::default()
        }
    }

    /// Low latency preset
    pub fn low_latency() -> Self {
        Self {
            max_rows: 1_000,
            max_wait: Duration::from_millis(100),
            initial_buffer_size: 64 * 1024,          // 64 KB
            max_buffer_size: 16 * 1024 * 1024,       // 16 MB
            buffer_shrink_threshold: 128 * 1024,     // 128 KB
            channel_capacity: 1_000,
            ..Default::default()
        }
    }
}

/// Statistics from a flush operation
#[derive(Debug, Clone)]
pub struct FlushStats {
    /// Number of rows flushed
    pub rows_flushed: usize,

    /// Bytes sent
    pub bytes_sent: usize,

    /// Time taken for flush
    pub duration: Duration,

    /// Age of the batch (time from first row)
    pub batch_age: Duration,
}