corium_transactor/
authz.rs

1//! Serving the authorization database from this transactor node.
2//!
3//! The transactor already holds every database it leads, so the authz database
4//! is just another one of them: the policy source hands out a snapshot of the
5//! local `DbState` with no network hop, and the transactor's own basis watch is
6//! the change signal that triggers recompilation. This is also the surface
7//! where write admission is decided, so authorizing a transaction against the
8//! node's own authz snapshot makes admission consistent at the serialization
9//! point.
10
11use std::sync::Arc;
12
13use corium_authz::source::{PolicySource, SourceError};
14use corium_db::Db;
15
16use crate::node::TransactorNode;
17
18/// A [`PolicySource`] reading the authz database this node hosts.
19pub struct NodePolicySource {
20    node: Arc<TransactorNode>,
21    db: String,
22}
23
24impl NodePolicySource {
25    /// Reads policy from the database `db` on `node`.
26    pub fn new(node: Arc<TransactorNode>, db: impl Into<String>) -> Self {
27        Self {
28            node,
29            db: db.into(),
30        }
31    }
32}
33
34#[tonic::async_trait]
35impl PolicySource for NodePolicySource {
36    fn name(&self) -> &str {
37        &self.db
38    }
39
40    async fn snapshot(&self) -> Result<Db, SourceError> {
41        match self.node.db_state(&self.db).await {
42            Ok(state) => Ok(state.db()),
43            // Distinguish "no such database" — a misconfiguration an operator
44            // must see — from a transient read failure.
45            Err(crate::node::NodeError::UnknownDb(name)) => Err(SourceError::Unavailable(name)),
46            Err(error) => Err(SourceError::Failed {
47                db: self.db.clone(),
48                reason: error.to_string(),
49            }),
50        }
51    }
52
53    async fn changed(&self, basis_t: u64) {
54        let Ok(state) = self.node.db_state(&self.db).await else {
55            tokio::time::sleep(corium_authz::source::DEFAULT_POLL_INTERVAL).await;
56            return;
57        };
58        let mut basis = state.basis_watch();
59        // `wait_for` returns immediately when the basis has already moved past
60        // the compiled snapshot, so no policy change can be missed between the
61        // compile and the subscribe.
62        let _ = basis.wait_for(|current| *current > basis_t).await;
63    }
64}
corium_transactor/authz.rs

corium_transactor/
authz.rs
