photon-runtime 0.1.0

Photon runtime builder and process-wide configuration
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
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

//! Compose [`AdminSnapshot`] from registries and storage checkpoint reads.

use photon_backend::{
    BackendCapabilities, HandlerDescriptor, HandlerRegistry, ShardConfig, TopicDescriptor,
    shard_storage_key,
};

use crate::Photon;

use super::types::{
    AdminBackendSummary, AdminCheckpointSummary, AdminHandlerSummary, AdminSnapshot,
    AdminTopicSummary,
};

/// Build a full admin snapshot for the given runtime (async checkpoint reads).
///
/// # Errors
///
/// Returns an error if a checkpoint load fails.
pub async fn collect_admin_snapshot(photon: &Photon) -> photon_backend::Result<AdminSnapshot> {
    let topics = collect_topics(photon);
    let handlers = collect_handlers();
    let backend = collect_backend(photon.backend_capabilities());
    let checkpoints = collect_checkpoints(photon).await?;

    Ok(AdminSnapshot {
        backend,
        topics,
        handlers,
        checkpoints,
    })
}

fn collect_topics(photon: &Photon) -> Vec<AdminTopicSummary> {
    photon
        .registry()
        .sorted_topic_names()
        .into_iter()
        .filter_map(|name| photon.registry().get(name).map(topic_from_descriptor))
        .collect()
}

fn topic_from_descriptor(descriptor: &TopicDescriptor) -> AdminTopicSummary {
    AdminTopicSummary {
        topic_name: descriptor.topic_name.to_string(),
        keyed_by: descriptor.keyed_by.map(str::to_string),
        schema_json: descriptor
            .schema_value()
            .unwrap_or_else(|| serde_json::json!({})),
    }
}

fn collect_handlers() -> Vec<AdminHandlerSummary> {
    let registry = HandlerRegistry::auto_discover();
    registry.iter().map(handler_from_descriptor).collect()
}

fn handler_from_descriptor(handler: &HandlerDescriptor) -> AdminHandlerSummary {
    let is_group = handler.is_consumer_group();
    AdminHandlerSummary {
        topic_name: handler.topic_name.to_string(),
        subscription_name: if is_group || handler.subscription_name.is_empty() {
            None
        } else {
            Some(handler.subscription_name.to_string())
        },
        consumer_group: handler.consumer_group.map(str::to_string),
        registry_key: handler.registry_key.to_string(),
        mode: if is_group {
            "consumer_group".to_string()
        } else {
            "durable".to_string()
        },
    }
}

fn collect_backend(caps: BackendCapabilities) -> AdminBackendSummary {
    AdminBackendSummary {
        telemetry_label: caps.telemetry_label.to_string(),
        supports_get_event: caps.supports_get_event,
        max_replay_window_secs: caps.max_replay_window.map(|d| d.as_secs()),
    }
}

async fn collect_checkpoints(
    photon: &Photon,
) -> photon_backend::Result<Vec<AdminCheckpointSummary>> {
    let registry = HandlerRegistry::auto_discover();
    let mut out = Vec::new();

    for handler in registry.iter() {
        if handler.is_consumer_group() {
            let group_id = handler.consumer_group.unwrap_or("");
            let shard_count = group_shard_count(photon, handler);
            for shard_id in 0..shard_count {
                let shard_key = shard_storage_key(shard_id);
                let last_seq = photon
                    .get_checkpoint_seq(group_id, handler.topic_name, Some(&shard_key))
                    .await?;
                out.push(AdminCheckpointSummary {
                    subscription_name: group_id.to_string(),
                    topic_name: handler.topic_name.to_string(),
                    topic_key: Some(shard_key),
                    last_seq,
                });
            }
        } else {
            let last_seq = photon
                .get_checkpoint_seq(handler.subscription_name, handler.topic_name, None)
                .await?;
            out.push(AdminCheckpointSummary {
                subscription_name: handler.subscription_name.to_string(),
                topic_name: handler.topic_name.to_string(),
                topic_key: None,
                last_seq,
            });
        }
    }

    Ok(out)
}

fn group_shard_count(photon: &Photon, handler: &'static HandlerDescriptor) -> u32 {
    handler
        .group_shard_count
        .or_else(|| {
            photon
                .registry()
                .get(handler.topic_name)
                .and_then(|d| d.shard_config)
                .map(|c| c.shard_count)
        })
        .unwrap_or(ShardConfig::DEFAULT_SHARD_COUNT)
}