Skip to main content

StorageManager

digdigdig3_station::storage::manager

Struct StorageManager 

Source 
pub struct StorageManager { /* private fields */ }
Expand description

Manages rotating daily log files for multiple concurrent streams.

§Example

use digdigdig3::core::storage::{StorageManager, StorageConfig, StreamKey};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let mgr = StorageManager::new(StorageConfig::default())?;
    let key = StreamKey {
        exchange: "binance".into(),
        account: "spot".into(),
        symbol: "BTCUSDT".into(),
        stream_kind: "ticker".into(),
    };
    mgr.append(&key, 1_700_000_000_000, b"payload").await?;
    Ok(())
}

Implementations§

Source§

impl StorageManager

Source

pub fn new(config: StorageConfig) -> Result<Self>

Create a StorageManager. Creates config.root if it doesn’t exist.

Examples found in repository?
examples/e2e_storage_read.rs (line 11)
5async fn main() -> Result<(), Box<dyn std::error::Error>> {
6    let cfg = StorageConfig {
7        root: PathBuf::from("./e2e_test_data"),
8        default_retention_days: 365,
9        orderbook_snapshot_interval_secs: 0,
10    };
11    let mgr = StorageManager::new(cfg)?;
12
13    for (exchange, account, sym, stream) in [
14        ("binance", "spot", "BTCUSDT",  "trade"),
15        ("binance", "spot", "BTCUSDT",  "ticker"),
16        ("bybit",   "spot", "BTCUSDT",  "trade"),
17        ("bybit",   "spot", "BTCUSDT",  "ticker"),
18        ("okx",     "spot", "BTC-USDT", "trade"),
19        ("okx",     "spot", "BTC-USDT", "ticker"),
20    ] {
21        let key = StreamKey {
22            exchange: exchange.to_string(),
23            account: account.to_string(),
24            symbol: sym.to_string(),
25            stream_kind: stream.to_string(),
26        };
27        // read today only — avoid iterating thousands of days from epoch
28        let now_ms = chrono::Utc::now().timestamp_millis();
29        let day_start = now_ms - (now_ms % 86_400_000);
30        let records = mgr.read_range(&key, day_start, now_ms).await?;
31        let first_ts = records.first().map(|(t, _)| *t);
32        let last_ts = records.last().map(|(t, _)| *t);
33        let span_secs = match (first_ts, last_ts) {
34            (Some(f), Some(l)) => (l - f) / 1000,
35            _ => 0,
36        };
37        println!("{exchange}:{sym}:{stream:8} records={} span={span_secs}s first_ts={first_ts:?} last_ts={last_ts:?}",
38            records.len());
39
40        if let Some((_, payload)) = records.first() {
41            match serde_json::from_slice::<serde_json::Value>(payload) {
42                Ok(v) => {
43                    let keys: Vec<_> = v.as_object()
44                        .map(|o| o.keys().map(|k| k.as_str()).collect())
45                        .unwrap_or_default();
46                    println!("  first event keys: {:?}", keys);
47                }
48                Err(e) => println!("  first event PARSE FAIL: {e}"),
49            }
50        }
51    }
52    Ok(())
53}
More examples
Hide additional examples
examples/e2e_canonical.rs (line 14)
8async fn main() -> Result<(), Box<dyn std::error::Error>> {
9    let cfg = StorageConfig {
10        root: PathBuf::from("./e2e_test_data"),
11        default_retention_days: 365,
12        orderbook_snapshot_interval_secs: 0,
13    };
14    let mgr = StorageManager::new(cfg)?;
15
16    let key = StreamKey {
17        exchange: "binance".to_string(),
18        account: "spot".to_string(),
19        symbol: "BTCUSDT".to_string(),
20        stream_kind: "trade".to_string(),
21    };
22
23    let now_ms = chrono::Utc::now().timestamp_millis();
24    let day_start = now_ms - (now_ms % 86_400_000);
25    let records = mgr.read_range(&key, day_start, now_ms).await?;
26
27    println!("Total Binance trade records: {}", records.len());
28
29    let sample: Vec<_> = records.iter().take(10).collect();
30    let mut ok = 0u32;
31    let mut failed = 0u32;
32
33    for (i, (ts_ms, payload)) in sample.iter().enumerate() {
34        match serde_json::from_slice::<StreamEvent>(payload) {
35            Ok(event) => {
36                match event.canonicalize() {
37                    Some(CanonicalEvent::Trade(t)) => {
38                        let price_ok = t.price > rust_decimal::Decimal::ZERO;
39                        let qty_ok = t.quantity > rust_decimal::Decimal::ZERO;
40                        let ts_ok = t.timestamp_ms > 1_700_000_000_000;
41                        if price_ok && qty_ok && ts_ok {
42                            ok += 1;
43                            println!("[{i}] OK  price={} qty={} ts_ms={} side={:?}",
44                                t.price, t.quantity, t.timestamp_ms, t.side);
45                        } else {
46                            failed += 1;
47                            println!("[{i}] BAD price={} qty={} ts_ms={} (record ts_ms={})",
48                                t.price, t.quantity, t.timestamp_ms, ts_ms);
49                        }
50                    }
51                    Some(other) => {
52                        failed += 1;
53                        println!("[{i}] WRONG VARIANT: {:?}", other);
54                    }
55                    None => {
56                        failed += 1;
57                        println!("[{i}] canonicalize() returned None (record ts_ms={})", ts_ms);
58                    }
59                }
60            }
61            Err(e) => {
62                failed += 1;
63                println!("[{i}] DESERIALIZE FAIL: {e}");
64            }
65        }
66    }
67
68    println!("\nResult: {ok}/10 canonical OK, {failed} failed");
69    Ok(())
70}
examples/replay_demo.rs (lines 29-33)
18async fn main() -> Result<(), Box<dyn std::error::Error>> {
19    // ── 1. Write synthetic events ─────────────────────────────────────────────
20
21    let tmp_dir = {
22        let mut d = std::env::temp_dir();
23        d.push("dig3_replay_demo");
24        d
25    };
26    std::fs::remove_dir_all(&tmp_dir).ok();
27    std::fs::create_dir_all(&tmp_dir)?;
28
29    let storage = StorageManager::new(StorageConfig {
30        root: tmp_dir.clone(),
31        default_retention_days: 365,
32        orderbook_snapshot_interval_secs: 0,
33    })?;
34
35    let key = StreamKey {
36        exchange: "binance".into(),
37        account: "spot".into(),
38        symbol: "BTCUSDT".into(),
39        stream_kind: "ticker".into(),
40    };
41
42    let base_ms = chrono::Utc::now().timestamp_millis();
43    let n_events = 200usize;
44
45    for i in 0..n_events {
46        let ts_ms = base_ms + (i as i64) * 100; // 100 ms apart → 20 s simulated
47        let price = 50_000.0 + i as f64;
48        let ev = StreamEvent::Ticker(Ticker {
49            symbol: "BTCUSDT".into(),
50            last_price: price,
51            bid_price: Some(price - 1.0),
52            ask_price: Some(price + 1.0),
53            volume_24h: Some(1000.0),
54            quote_volume_24h: None,
55            price_change_24h: None,
56            price_change_percent_24h: None,
57            high_24h: None,
58            low_24h: None,
59            timestamp: ts_ms,
60        });
61        let payload = serde_json::to_vec(&ev)?;
62        storage.append(&key, ts_ms, &payload).await?;
63    }
64    storage.flush_all().await?;
65    println!("Wrote {n_events} synthetic ticker events to {}", tmp_dir.display());
66
67    // ── 2. Replay at 10× speed ────────────────────────────────────────────────
68
69    let config = ReplayConfig {
70        storage_root: PathBuf::from(&tmp_dir),
71        rate: ReplayRate::Accelerated(10.0),
72        from_ms: Some(base_ms),
73        to_ms: Some(base_ms + (n_events as i64) * 100),
74    };
75    let hub = ReplayHub::new(config).await?;
76    hub.connect_full(ExchangeId::Binance, &[AccountType::Spot], false)
77        .await?;
78
79    let ws = hub.ws(ExchangeId::Binance, AccountType::Spot).unwrap();
80    ws.subscribe(SubscriptionRequest::ticker(Symbol::with_raw(
81        "BTC",
82        "USDT",
83        "BTCUSDT".into(),
84    )))
85    .await?;
86
87    let mut stream = ws.event_stream();
88    let mut count = 0usize;
89    let start = std::time::Instant::now();
90
91    while let Some(ev) = stream.next().await {
92        match ev {
93            Ok(StreamEvent::Ticker(_t)) => {
94                count += 1;
95                if count % 50 == 0 {
96                    println!("  {count} events received…");
97                }
98                if count >= n_events {
99                    break;
100                }
101            }
102            Ok(_other) => {}
103            Err(e) => eprintln!("stream error: {e}"),
104        }
105    }
106
107    let elapsed = start.elapsed();
108    println!(
109        "Replay done: {count} events in {:.2}s  (simulated 20s at 10×)",
110        elapsed.as_secs_f64()
111    );
112
113    // ── 3. Cleanup ────────────────────────────────────────────────────────────
114    std::fs::remove_dir_all(&tmp_dir).ok();
115
116    Ok(())
117}
Source

pub async fn append( &self, key: &StreamKey, ts_ms: i64, payload: &[u8], ) -> Result<()>

Append one record to the stream identified by key.

Rotates to a new daily file automatically at UTC midnight.

Examples found in repository?
examples/replay_demo.rs (line 62)
18async fn main() -> Result<(), Box<dyn std::error::Error>> {
19    // ── 1. Write synthetic events ─────────────────────────────────────────────
20
21    let tmp_dir = {
22        let mut d = std::env::temp_dir();
23        d.push("dig3_replay_demo");
24        d
25    };
26    std::fs::remove_dir_all(&tmp_dir).ok();
27    std::fs::create_dir_all(&tmp_dir)?;
28
29    let storage = StorageManager::new(StorageConfig {
30        root: tmp_dir.clone(),
31        default_retention_days: 365,
32        orderbook_snapshot_interval_secs: 0,
33    })?;
34
35    let key = StreamKey {
36        exchange: "binance".into(),
37        account: "spot".into(),
38        symbol: "BTCUSDT".into(),
39        stream_kind: "ticker".into(),
40    };
41
42    let base_ms = chrono::Utc::now().timestamp_millis();
43    let n_events = 200usize;
44
45    for i in 0..n_events {
46        let ts_ms = base_ms + (i as i64) * 100; // 100 ms apart → 20 s simulated
47        let price = 50_000.0 + i as f64;
48        let ev = StreamEvent::Ticker(Ticker {
49            symbol: "BTCUSDT".into(),
50            last_price: price,
51            bid_price: Some(price - 1.0),
52            ask_price: Some(price + 1.0),
53            volume_24h: Some(1000.0),
54            quote_volume_24h: None,
55            price_change_24h: None,
56            price_change_percent_24h: None,
57            high_24h: None,
58            low_24h: None,
59            timestamp: ts_ms,
60        });
61        let payload = serde_json::to_vec(&ev)?;
62        storage.append(&key, ts_ms, &payload).await?;
63    }
64    storage.flush_all().await?;
65    println!("Wrote {n_events} synthetic ticker events to {}", tmp_dir.display());
66
67    // ── 2. Replay at 10× speed ────────────────────────────────────────────────
68
69    let config = ReplayConfig {
70        storage_root: PathBuf::from(&tmp_dir),
71        rate: ReplayRate::Accelerated(10.0),
72        from_ms: Some(base_ms),
73        to_ms: Some(base_ms + (n_events as i64) * 100),
74    };
75    let hub = ReplayHub::new(config).await?;
76    hub.connect_full(ExchangeId::Binance, &[AccountType::Spot], false)
77        .await?;
78
79    let ws = hub.ws(ExchangeId::Binance, AccountType::Spot).unwrap();
80    ws.subscribe(SubscriptionRequest::ticker(Symbol::with_raw(
81        "BTC",
82        "USDT",
83        "BTCUSDT".into(),
84    )))
85    .await?;
86
87    let mut stream = ws.event_stream();
88    let mut count = 0usize;
89    let start = std::time::Instant::now();
90
91    while let Some(ev) = stream.next().await {
92        match ev {
93            Ok(StreamEvent::Ticker(_t)) => {
94                count += 1;
95                if count % 50 == 0 {
96                    println!("  {count} events received…");
97                }
98                if count >= n_events {
99                    break;
100                }
101            }
102            Ok(_other) => {}
103            Err(e) => eprintln!("stream error: {e}"),
104        }
105    }
106
107    let elapsed = start.elapsed();
108    println!(
109        "Replay done: {count} events in {:.2}s  (simulated 20s at 10×)",
110        elapsed.as_secs_f64()
111    );
112
113    // ── 3. Cleanup ────────────────────────────────────────────────────────────
114    std::fs::remove_dir_all(&tmp_dir).ok();
115
116    Ok(())
117}
Source

pub async fn flush_all(&self) -> Result<()>

Flush all open writers to the OS.

Examples found in repository?
examples/replay_demo.rs (line 64)
18async fn main() -> Result<(), Box<dyn std::error::Error>> {
19    // ── 1. Write synthetic events ─────────────────────────────────────────────
20
21    let tmp_dir = {
22        let mut d = std::env::temp_dir();
23        d.push("dig3_replay_demo");
24        d
25    };
26    std::fs::remove_dir_all(&tmp_dir).ok();
27    std::fs::create_dir_all(&tmp_dir)?;
28
29    let storage = StorageManager::new(StorageConfig {
30        root: tmp_dir.clone(),
31        default_retention_days: 365,
32        orderbook_snapshot_interval_secs: 0,
33    })?;
34
35    let key = StreamKey {
36        exchange: "binance".into(),
37        account: "spot".into(),
38        symbol: "BTCUSDT".into(),
39        stream_kind: "ticker".into(),
40    };
41
42    let base_ms = chrono::Utc::now().timestamp_millis();
43    let n_events = 200usize;
44
45    for i in 0..n_events {
46        let ts_ms = base_ms + (i as i64) * 100; // 100 ms apart → 20 s simulated
47        let price = 50_000.0 + i as f64;
48        let ev = StreamEvent::Ticker(Ticker {
49            symbol: "BTCUSDT".into(),
50            last_price: price,
51            bid_price: Some(price - 1.0),
52            ask_price: Some(price + 1.0),
53            volume_24h: Some(1000.0),
54            quote_volume_24h: None,
55            price_change_24h: None,
56            price_change_percent_24h: None,
57            high_24h: None,
58            low_24h: None,
59            timestamp: ts_ms,
60        });
61        let payload = serde_json::to_vec(&ev)?;
62        storage.append(&key, ts_ms, &payload).await?;
63    }
64    storage.flush_all().await?;
65    println!("Wrote {n_events} synthetic ticker events to {}", tmp_dir.display());
66
67    // ── 2. Replay at 10× speed ────────────────────────────────────────────────
68
69    let config = ReplayConfig {
70        storage_root: PathBuf::from(&tmp_dir),
71        rate: ReplayRate::Accelerated(10.0),
72        from_ms: Some(base_ms),
73        to_ms: Some(base_ms + (n_events as i64) * 100),
74    };
75    let hub = ReplayHub::new(config).await?;
76    hub.connect_full(ExchangeId::Binance, &[AccountType::Spot], false)
77        .await?;
78
79    let ws = hub.ws(ExchangeId::Binance, AccountType::Spot).unwrap();
80    ws.subscribe(SubscriptionRequest::ticker(Symbol::with_raw(
81        "BTC",
82        "USDT",
83        "BTCUSDT".into(),
84    )))
85    .await?;
86
87    let mut stream = ws.event_stream();
88    let mut count = 0usize;
89    let start = std::time::Instant::now();
90
91    while let Some(ev) = stream.next().await {
92        match ev {
93            Ok(StreamEvent::Ticker(_t)) => {
94                count += 1;
95                if count % 50 == 0 {
96                    println!("  {count} events received…");
97                }
98                if count >= n_events {
99                    break;
100                }
101            }
102            Ok(_other) => {}
103            Err(e) => eprintln!("stream error: {e}"),
104        }
105    }
106
107    let elapsed = start.elapsed();
108    println!(
109        "Replay done: {count} events in {:.2}s  (simulated 20s at 10×)",
110        elapsed.as_secs_f64()
111    );
112
113    // ── 3. Cleanup ────────────────────────────────────────────────────────────
114    std::fs::remove_dir_all(&tmp_dir).ok();
115
116    Ok(())
117}
Source

pub async fn read_range( &self, key: &StreamKey, from_ms: i64, to_ms: i64, ) -> Result<Vec<(i64, Vec<u8>)>>

Read records in [from_ms, to_ms] (inclusive) for key.

Spans multiple daily files automatically. Accepts any i64 (including i64::MAX sentinel for “read to end”) — values outside the representable date range are clamped to [0, MAX_SAFE_MS].

Examples found in repository?
examples/e2e_storage_read.rs (line 30)
5async fn main() -> Result<(), Box<dyn std::error::Error>> {
6    let cfg = StorageConfig {
7        root: PathBuf::from("./e2e_test_data"),
8        default_retention_days: 365,
9        orderbook_snapshot_interval_secs: 0,
10    };
11    let mgr = StorageManager::new(cfg)?;
12
13    for (exchange, account, sym, stream) in [
14        ("binance", "spot", "BTCUSDT",  "trade"),
15        ("binance", "spot", "BTCUSDT",  "ticker"),
16        ("bybit",   "spot", "BTCUSDT",  "trade"),
17        ("bybit",   "spot", "BTCUSDT",  "ticker"),
18        ("okx",     "spot", "BTC-USDT", "trade"),
19        ("okx",     "spot", "BTC-USDT", "ticker"),
20    ] {
21        let key = StreamKey {
22            exchange: exchange.to_string(),
23            account: account.to_string(),
24            symbol: sym.to_string(),
25            stream_kind: stream.to_string(),
26        };
27        // read today only — avoid iterating thousands of days from epoch
28        let now_ms = chrono::Utc::now().timestamp_millis();
29        let day_start = now_ms - (now_ms % 86_400_000);
30        let records = mgr.read_range(&key, day_start, now_ms).await?;
31        let first_ts = records.first().map(|(t, _)| *t);
32        let last_ts = records.last().map(|(t, _)| *t);
33        let span_secs = match (first_ts, last_ts) {
34            (Some(f), Some(l)) => (l - f) / 1000,
35            _ => 0,
36        };
37        println!("{exchange}:{sym}:{stream:8} records={} span={span_secs}s first_ts={first_ts:?} last_ts={last_ts:?}",
38            records.len());
39
40        if let Some((_, payload)) = records.first() {
41            match serde_json::from_slice::<serde_json::Value>(payload) {
42                Ok(v) => {
43                    let keys: Vec<_> = v.as_object()
44                        .map(|o| o.keys().map(|k| k.as_str()).collect())
45                        .unwrap_or_default();
46                    println!("  first event keys: {:?}", keys);
47                }
48                Err(e) => println!("  first event PARSE FAIL: {e}"),
49            }
50        }
51    }
52    Ok(())
53}
More examples
Hide additional examples
examples/e2e_canonical.rs (line 25)
8async fn main() -> Result<(), Box<dyn std::error::Error>> {
9    let cfg = StorageConfig {
10        root: PathBuf::from("./e2e_test_data"),
11        default_retention_days: 365,
12        orderbook_snapshot_interval_secs: 0,
13    };
14    let mgr = StorageManager::new(cfg)?;
15
16    let key = StreamKey {
17        exchange: "binance".to_string(),
18        account: "spot".to_string(),
19        symbol: "BTCUSDT".to_string(),
20        stream_kind: "trade".to_string(),
21    };
22
23    let now_ms = chrono::Utc::now().timestamp_millis();
24    let day_start = now_ms - (now_ms % 86_400_000);
25    let records = mgr.read_range(&key, day_start, now_ms).await?;
26
27    println!("Total Binance trade records: {}", records.len());
28
29    let sample: Vec<_> = records.iter().take(10).collect();
30    let mut ok = 0u32;
31    let mut failed = 0u32;
32
33    for (i, (ts_ms, payload)) in sample.iter().enumerate() {
34        match serde_json::from_slice::<StreamEvent>(payload) {
35            Ok(event) => {
36                match event.canonicalize() {
37                    Some(CanonicalEvent::Trade(t)) => {
38                        let price_ok = t.price > rust_decimal::Decimal::ZERO;
39                        let qty_ok = t.quantity > rust_decimal::Decimal::ZERO;
40                        let ts_ok = t.timestamp_ms > 1_700_000_000_000;
41                        if price_ok && qty_ok && ts_ok {
42                            ok += 1;
43                            println!("[{i}] OK  price={} qty={} ts_ms={} side={:?}",
44                                t.price, t.quantity, t.timestamp_ms, t.side);
45                        } else {
46                            failed += 1;
47                            println!("[{i}] BAD price={} qty={} ts_ms={} (record ts_ms={})",
48                                t.price, t.quantity, t.timestamp_ms, ts_ms);
49                        }
50                    }
51                    Some(other) => {
52                        failed += 1;
53                        println!("[{i}] WRONG VARIANT: {:?}", other);
54                    }
55                    None => {
56                        failed += 1;
57                        println!("[{i}] canonicalize() returned None (record ts_ms={})", ts_ms);
58                    }
59                }
60            }
61            Err(e) => {
62                failed += 1;
63                println!("[{i}] DESERIALIZE FAIL: {e}");
64            }
65        }
66    }
67
68    println!("\nResult: {ok}/10 canonical OK, {failed} failed");
69    Ok(())
70}
Source

pub fn cleanup(&self, now: DateTime<Utc>) -> Result<usize>

Run retention sweep — delete daily files older than config.default_retention_days.

Returns the count of deleted files.

Source

pub fn stream_dir(&self, key: &StreamKey) -> PathBuf

Return the directory for a stream: {root}/{exchange}/{account}/{symbol}/{stream_kind}/

Source

pub fn config(&self) -> &StorageConfig

Return reference to config.

Auto Trait Implementations§

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self>

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self>

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> PolicyExt for T
where T: ?Sized,

Source§

fn and<P, B, E>(self, other: P) -> And<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow only if self and other return Action::Follow. Read more
Source§

fn or<P, B, E>(self, other: P) -> Or<T, P>
where T: Policy<B, E>, P: Policy<B, E>,

Create a new Policy that returns Action::Follow if either self or other returns Action::Follow. Read more
Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = Infallible

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, <T as TryFrom<U>>::Error>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<V, T> VZip<V> for T
where V: MultiLane<T>,

Source§

fn vzip(self) -> V

Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self>
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self>

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more