liminal-server 0.25.4

Standalone server for the liminal messaging bus
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

//! Connection and revocation owners retire the same set of keyed links.

use super::{OpenLink, OpenRefusal};
use crate::link::{CloseLinks, LinkEnd};
use crate::server::connection::supervisor::runtime::ConnectionRuntime;
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex, MutexGuard};

#[derive(Debug, Default)]
pub(super) struct Contents {
    pub(super) links: BTreeMap<u32, OpenLink>,
    closed: Option<LinkEnd>,
}

#[derive(Debug, Default)]
struct LinkSet(Mutex<Contents>);

impl CloseLinks for LinkSet {
    fn end_links(&self, reason: LinkEnd) -> Result<(), String> {
        let mut contents = self
            .0
            .lock()
            .map_err(|_| "connection link set lock poisoned")?;
        if contents.closed.is_some() {
            return Ok(());
        }
        contents.closed = Some(reason.clone());
        let mut failure = None;
        for open in std::mem::take(&mut contents.links).into_values() {
            if let Err(error) = open.life.end(reason.clone()) {
                failure = Some(error);
            }
        }
        drop(contents);
        failure.map_or(Ok(()), Err)
    }
}

impl Drop for LinkSet {
    fn drop(&mut self) {
        if let Err(error) = self.end_links(LinkEnd::PeerGone) {
            tracing::error!(%error, "connection link set drop failed");
        }
    }
}

#[derive(Debug, Default)]
pub(in crate::server::connection) struct ConnectionLinks(Arc<LinkSet>);

impl ConnectionLinks {
    pub(super) fn lock(&self) -> Result<MutexGuard<'_, Contents>, String> {
        self.0
            .0
            .lock()
            .map_err(|_| "connection link set lock poisoned".to_owned())
    }

    pub(in crate::server::connection) fn contains_key(&self, stream: u32) -> Result<bool, String> {
        Ok(self.lock()?.links.contains_key(&stream))
    }

    pub(super) fn bind(
        &self,
        pid: u64,
        runtime: &ConnectionRuntime,
        grants: Option<&crate::auth_revocation::ConnectionGrants>,
    ) -> Result<(), OpenRefusal> {
        let links: Arc<dyn CloseLinks> = self.0.clone();
        runtime
            .link_life
            .register(pid, &links)
            .map_err(OpenRefusal::Refused)?;
        if let Some(authority) = runtime.revocation() {
            authority
                .link_life
                .register(pid, &links)
                .map_err(OpenRefusal::Refused)?;
        }
        // Register before checking the fence so a racing revoke sees this set,
        // or its already-denied binding prevents the later insertion.
        if grants.is_some_and(|grants| grants.close_pending().is_some()) {
            self.0
                .end_links(LinkEnd::Revoked)
                .map_err(OpenRefusal::Refused)?;
        }
        Ok(())
    }

    pub(super) fn insert(
        &self,
        runtime: &ConnectionRuntime,
        stream: u32,
        mut open: OpenLink,
    ) -> Result<(), OpenRefusal> {
        let mut contents = self.lock().map_err(OpenRefusal::Refused)?;
        if let Some(reason) = &contents.closed {
            return Err(OpenRefusal::Refused(format!(
                "connection links ended: {reason}"
            )));
        }
        open.life = runtime
            .link_life
            .opened(*open.transport.peer())
            .map_err(OpenRefusal::Refused)?;
        if let Some(signals) = runtime.signals.get() {
            signals
                .observe_link(&open.life)
                .map_err(OpenRefusal::Refused)?;
        }
        if let Some(previous) = contents.links.insert(stream, open) {
            previous
                .life
                .end(LinkEnd::Replaced)
                .map_err(OpenRefusal::Refused)?;
        }
        drop(contents);
        Ok(())
    }
}