liminal_server/services/
keys.rs

1//! The production [`KeySender`]: a session's keys delivered as a publish on
2//! `sessions.<id>.keys` through the server's own publish path (ACCESS-003
3//! R1).
4//!
5//! The keys are published only after Lys answered the deliberate ask
6//! outright; a held act never reaches this sender. The publish goes through
7//! [`ServicePublisher`], whose real implementation is the connection
8//! services' embedded-host publish (`publish_embedded`): the same admission,
9//! namespace refusal, durability and dedup a connection publish gets, with
10//! publisher `liminal.<host>`. The request's operation id is the
11//! idempotency key, so a request sent again is delivered at most once. The
12//! session's runner subscribes to its own `sessions.<id>.keys` subject; a
13//! subject the roster does not carry is refused by the roster, by name.
14
15use super::{KeySender, ServiceEffectError};
16use crate::server::connection::LiminalConnectionServices;
17use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};
18use std::sync::Arc;
19
20/// The schema bytes the services' messages are published under: the payload
21/// is the request's bytes, unchanged, and the schema names which service
22/// carried them.
23pub const SESSION_KEYS_SCHEMA: &[u8] = b"liminal.services.session-keys/1";
24
25/// Publishes one service message on a subject, once per operation.
26pub trait ServicePublisher: Send + Sync + std::fmt::Debug {
27    /// Publishes `envelope` on `subject` with `operation` as its idempotency
28    /// key.
29    ///
30    /// # Errors
31    /// The publish was refused or failed, by name.
32    fn publish(
33        &self,
34        subject: &str,
35        envelope: &MessageEnvelope,
36        operation: &str,
37    ) -> Result<(), ServiceEffectError>;
38}
39
40/// The connection services' embedded-host publish as a [`ServicePublisher`].
41#[derive(Debug)]
42pub struct ConnectionPublisher {
43    services: Arc<LiminalConnectionServices>,
44    host: String,
45}
46
47impl ConnectionPublisher {
48    /// Publishes through `services` as `liminal.<host>`.
49    #[must_use]
50    pub const fn new(services: Arc<LiminalConnectionServices>, host: String) -> Self {
51        Self { services, host }
52    }
53}
54
55impl ServicePublisher for ConnectionPublisher {
56    fn publish(
57        &self,
58        subject: &str,
59        envelope: &MessageEnvelope,
60        operation: &str,
61    ) -> Result<(), ServiceEffectError> {
62        self.services
63            .publish_embedded(&self.host, subject, envelope, Some(operation))
64            .map(|_outcome| ())
65            .map_err(|error| ServiceEffectError(format!("publish on {subject} refused: {error}")))
66    }
67}
68
69/// The envelope one service message travels in.
70#[must_use]
71pub fn service_envelope(schema: &[u8], payload: &[u8]) -> MessageEnvelope {
72    MessageEnvelope::new(
73        SchemaId::from_schema_bytes(schema),
74        CausalContext::independent(),
75        payload.to_vec(),
76    )
77}
78
79/// Session keys published on `sessions.<id>.keys`.
80#[derive(Debug)]
81pub struct ChannelKeySender {
82    publisher: Arc<dyn ServicePublisher>,
83}
84
85impl ChannelKeySender {
86    /// Sends through `publisher`.
87    #[must_use]
88    pub const fn new(publisher: Arc<dyn ServicePublisher>) -> Self {
89        Self { publisher }
90    }
91}
92
93/// The subject a session's keys are delivered on.
94#[must_use]
95pub fn keys_subject(session: &str) -> String {
96    format!("sessions.{session}.keys")
97}
98
99impl KeySender for ChannelKeySender {
100    fn send(&self, session: &str, keys: &[u8], operation: &str) -> Result<(), ServiceEffectError> {
101        if keys.is_empty() {
102            return Err(ServiceEffectError(format!(
103                "no keys to send to session {session}"
104            )));
105        }
106        self.publisher.publish(
107            &keys_subject(session),
108            &service_envelope(SESSION_KEYS_SCHEMA, keys),
109            operation,
110        )
111    }
112}
liminal_server/services/keys.rs

liminal_server/services/
keys.rs
