durable-actors 0.4.0

Standalone regional durable-actors control plane, host, and durability runtime
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

use crate::{
    clock::{Clock, SystemClock},
    grpc::{
        proto::{Empty, replica_service_client::ReplicaServiceClient},
        transport::{MAX_STORAGE_MESSAGE_BYTES, request},
    },
    replication::{
        ReplicaAccess, ReplicaGrant, ReplicaStream, ReplicaTarget, SessionHead, StreamHead,
    },
    state_transport::{GrpcStateTransport, StateTransport},
};
use anyhow::Result;
use async_trait::async_trait;

#[async_trait]
pub trait ReplicaPeers: Send + Sync {
    async fn initialize(&self, peer: &ReplicaTarget, session: &str) -> Result<()>;
    async fn head(&self, peer: &ReplicaTarget, stream: &ReplicaStream) -> Result<StreamHead>;
    async fn seal(&self, peer: &ReplicaTarget, session: &str) -> Result<SessionHead>;
    async fn read(&self, peer: &ReplicaTarget, object: &str) -> Result<Vec<u8>>;
}

pub struct GrpcReplicaPeers {
    access: ReplicaAccess,
    transport: GrpcStateTransport,
}

impl GrpcReplicaPeers {
    pub fn new(access: ReplicaAccess) -> Result<Self> {
        Ok(Self::with_transport(access, GrpcStateTransport::new()))
    }

    pub(crate) fn with_transport(access: ReplicaAccess, transport: GrpcStateTransport) -> Self {
        Self { access, transport }
    }

    async fn client(
        &self,
        peer: &ReplicaTarget,
        grant: ReplicaGrant,
    ) -> Result<(
        ReplicaServiceClient<tonic::transport::Channel>,
        tonic::Request<Empty>,
    )> {
        let address = self.access.url(
            &peer.url,
            &ReplicaGrant {
                host_id: peer.host_id.clone(),
                ..grant
            },
        )?;
        let (channel, token) = self.transport.capability(&address).await?;
        Ok((
            ReplicaServiceClient::new(channel)
                .max_decoding_message_size(MAX_STORAGE_MESSAGE_BYTES)
                .max_encoding_message_size(MAX_STORAGE_MESSAGE_BYTES),
            request(Empty {}, &token)?,
        ))
    }
}

#[async_trait]
impl ReplicaPeers for GrpcReplicaPeers {
    async fn initialize(&self, peer: &ReplicaTarget, session: &str) -> Result<()> {
        let (mut client, request) = self
            .client(
                peer,
                grant("INITIALIZE_SESSION", &peer.region, session, 60_000)?,
            )
            .await?;
        client.initialize(request).await?;
        Ok(())
    }

    async fn head(&self, peer: &ReplicaTarget, stream: &ReplicaStream) -> Result<StreamHead> {
        let (mut client, request) = self
            .client(
                peer,
                ReplicaGrant {
                    stream: Some(stream.clone()),
                    ..grant("HEAD", &peer.region, &stream.prefix, 60_000)?
                },
            )
            .await?;
        client.head(request).await?.into_inner().try_into()
    }

    async fn seal(&self, peer: &ReplicaTarget, session: &str) -> Result<SessionHead> {
        let (mut client, request) = self
            .client(peer, grant("SEAL_SESSION", &peer.region, session, 60_000)?)
            .await?;
        client.seal(request).await?.into_inner().try_into()
    }

    async fn read(&self, peer: &ReplicaTarget, object: &str) -> Result<Vec<u8>> {
        let address = self.access.url(
            &peer.url,
            &ReplicaGrant {
                host_id: peer.host_id.clone(),
                ..grant("GET", &peer.region, object, 60_000)?
            },
        )?;
        Ok(self.transport.read(&address).await?.to_vec())
    }
}

pub(super) fn grant(
    operation: &str,
    region: &str,
    object: &str,
    duration_ms: u64,
) -> Result<ReplicaGrant> {
    Ok(ReplicaGrant {
        operation: operation.into(),
        region: region.into(),
        object: object.into(),
        host_id: String::new(),
        expires_at_ms: SystemClock.now_ms()?.saturating_add(duration_ms),
        stream: None,
    })
}