durable-actors 0.7.14

Standalone regional durable-actors control plane, host, and durability runtime
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

use crate::request_traces::metrics::{
    ClassMetrics, OverviewMetrics, QueueWaitQuery, QueueWaitRow, SocketSession, TimeRange,
};
use anyhow::Result;
use rusqlite::{Connection, params};

pub(super) fn overview(
    connection: &mut Connection,
    project_id: &str,
    range: &TimeRange,
) -> Result<OverviewMetrics> {
    let mut statement = connection.prepare(include_str!("overview.sql"))?;
    let rows = statement.query_map(
        params![
            range.from_ms.map(|ms| ms as i64),
            range.to_ms.map(|ms| ms as i64),
            project_id
        ],
        |row| {
            let attempts: i64 = row.get(2)?;
            let completed: i64 = row.get(3)?;
            Ok(ClassMetrics {
                actor_name: row.get(0)?,
                count: row.get::<_, i64>(1)? as u64,
                success: (attempts > 0).then(|| 100.0 * completed as f64 / attempts as f64),
                p95: row.get(4)?,
                queue_p95: row.get(5)?,
            })
        },
    )?;
    let mut metrics = OverviewMetrics::default();
    for row in rows {
        let row = row?;
        if row.actor_name.is_empty() {
            metrics.total = row;
        } else {
            metrics.classes.push(row);
        }
    }
    Ok(metrics)
}

pub(super) fn queue_waits(
    connection: &mut Connection,
    project_id: &str,
    query: &QueueWaitQuery,
) -> Result<Vec<QueueWaitRow>> {
    let mut statement = connection.prepare(include_str!("queue_waits.sql"))?;
    let rows = statement.query_map(
        params![
            query.from_ms.map(|ms| ms as i64),
            query.to_ms.map(|ms| ms as i64),
            query.actor_name,
            project_id
        ],
        |row| {
            Ok(QueueWaitRow {
                actor_name: row.get(0)?,
                actor_id: row.get(1)?,
                admitted: row.get::<_, i64>(2)? as u64,
                average_ms: row.get(3)?,
                max_ms: row.get(4)?,
            })
        },
    )?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}

pub(super) fn websockets(
    connection: &mut Connection,
    project_id: &str,
    range: &TimeRange,
) -> Result<Vec<SocketSession>> {
    let mut statement = connection.prepare(include_str!("websockets.sql"))?;
    let rows = statement.query_map(
        params![
            range.from_ms.map(|ms| ms as i64),
            range.to_ms.map(|ms| ms as i64),
            project_id
        ],
        |row| {
            let connect_event: Option<String> = row.get(7)?;
            let metadata = connect_event
                .map(|event| serde_json::from_str::<serde_json::Value>(&event))
                .transpose()
                .map_err(|error| {
                    rusqlite::Error::FromSqlConversionFailure(
                        7,
                        rusqlite::types::Type::Text,
                        Box::new(error),
                    )
                })?
                .and_then(|mut event| event.as_object_mut()?.remove("metadata"));
            Ok(SocketSession {
                connection_id: row.get(0)?,
                actor_name: row.get(1)?,
                actor_id: row.get(2)?,
                host_id: row.get(3)?,
                opened_at_ms: row.get::<_, Option<i64>>(4)?.map(|ms| ms as u64),
                closed_at_ms: row.get::<_, Option<i64>>(5)?.map(|ms| ms as u64),
                last_seen_ms: row.get::<_, Option<i64>>(6)?.map(|ms| ms as u64),
                messages: row.get::<_, i64>(8)? as u64,
                failures: row.get::<_, i64>(9)? as u64,
                metadata,
            })
        },
    )?;
    Ok(rows.collect::<rusqlite::Result<_>>()?)
}