corium_peer/
authz.rs

1//! Serving the authorization database to a peer server.
2//!
3//! A peer server hosts one application database, so the authz database reaches
4//! it the same way its own data does: over a second [`Connection`], kept in
5//! sync by the peer's subscription. Policy checks then read a local, immutable
6//! database value — no per-request round trip to the transactor — and the
7//! subscription's tx-report broadcast is the change signal that triggers
8//! recompilation.
9
10use std::sync::Arc;
11
12use corium_authz::source::{PolicySource, SourceError};
13use corium_db::Db;
14
15use crate::Connection;
16
17/// A [`PolicySource`] reading the authz database over a peer connection.
18pub struct ConnectionPolicySource {
19    connection: Arc<Connection>,
20}
21
22impl ConnectionPolicySource {
23    /// Reads policy from the database `connection` is synced to.
24    #[must_use]
25    pub fn new(connection: Arc<Connection>) -> Self {
26        Self { connection }
27    }
28}
29
30#[tonic::async_trait]
31impl PolicySource for ConnectionPolicySource {
32    fn name(&self) -> &str {
33        self.connection.db_name()
34    }
35
36    async fn snapshot(&self) -> Result<Db, SourceError> {
37        Ok(self.connection.db())
38    }
39
40    async fn changed(&self, basis_t: u64) {
41        // Subscribe before re-reading the basis: a report committed between the
42        // two is still delivered, so a policy change is never missed.
43        let mut reports = self.connection.tx_reports();
44        if self.connection.basis_t() > basis_t {
45            return;
46        }
47        loop {
48            match reports.recv().await {
49                Ok(report) if report.t > basis_t => return,
50                Ok(_) => {}
51                // Lagged: the peer applied more than the broadcast buffer held,
52                // so the basis has certainly moved.
53                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => return,
54                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
55                    tokio::time::sleep(corium_authz::source::DEFAULT_POLL_INTERVAL).await;
56                    return;
57                }
58            }
59        }
60    }
61}
corium_peer/authz.rs

corium_peer/
authz.rs
