ossa_core/protocol/
v0.rs

1use std::collections::BTreeSet;
2use std::marker::Send;
3
4// use async_session_types::{Eps, Recv, Send};
5use bytes::{BufMut, Bytes, BytesMut};
6use futures::{SinkExt, StreamExt};
7use serde::{Deserialize, Serialize};
8use std::fmt::Debug;
9use tokio::{
10    io::{AsyncReadExt, AsyncWriteExt},
11    net::TcpStream,
12    runtime::Runtime,
13    sync::{
14        mpsc::{self, Receiver, Sender, UnboundedSender},
15        watch,
16    },
17};
18use tokio_stream::wrappers::ReceiverStream;
19use tokio_util::sync::{PollSendError, PollSender};
20
21use crate::protocol::heartbeat::v0::Heartbeat;
22use crate::protocol::manager::v0::Manager;
23use crate::protocol::MiniProtocolArgs;
24use crate::store;
25use crate::util;
26use crate::{
27    auth::DeviceId,
28    core::{OssaType, StoreStatuses},
29    network::{
30        multiplexer::{run_miniprotocol_async, Multiplexer, MultiplexerCommand, Party, StreamId},
31        protocol::MiniProtocol,
32    },
33    store::ecg::ECGHeader,
34};
35
36// Required since Rust can't handle proper existentials.
37pub(crate) enum MiniProtocols<StoreId, Hash, HeaderId, Header> {
38    Heartbeat(Heartbeat),
39    Manager(Manager<StoreId, Hash, HeaderId, Header>),
40}
41
42impl<
43        StoreId: Send + Sync + Copy + AsRef<[u8]> + Ord + Debug + Serialize + for<'a> Deserialize<'a>,
44        Hash: Send,
45        HeaderId: Send,
46        Header: Send,
47    > MiniProtocols<StoreId, Hash, HeaderId, Header>
48{
49    pub(crate) async fn run_async<O: OssaType>(
50        self,
51        is_client: bool,
52        stream_id: StreamId,
53        sender: Sender<(StreamId, Bytes)>,
54        receiver: Receiver<BytesMut>,
55    ) {
56        match self {
57            MiniProtocols::Heartbeat(p) => {
58                run_miniprotocol_async::<_, O>(p, is_client, stream_id, sender, receiver).await
59            }
60            MiniProtocols::Manager(p) => {
61                run_miniprotocol_async::<_, O>(p, is_client, stream_id, sender, receiver).await
62            }
63        }
64    }
65}
66
67// Multiplexer:
68//  StreamId (u32)
69//  DataLength (u32?)
70//
71// Miniprotocols:
72// 0 - Heartbeat (Server)
73// 1 - StreamManagement Client (AdvertiseStores, CloseConnection, TerminateConnection, CreateStream, CloseStream? (probably not))
74// 2 - StreamManagement Server
75// ...
76// N - (N is odd for client, even for server):
77//     - StoreSync i
78/// Miniprotocols initially run when connected for V0.
79fn initial_miniprotocols<StoreId, Hash, HeaderId, Header>(
80    party: Party,
81    args: MiniProtocolArgs<StoreId, Hash, HeaderId, Header>,
82    multiplexer_cmd_send: UnboundedSender<MultiplexerCommand>,
83) -> Vec<MiniProtocols<StoreId, Hash, HeaderId, Header>> {
84    let (client_chan, server_chan) = if let Party::Client = party {
85        (Some(args.manager_channel), None)
86    } else {
87        (None, Some(args.manager_channel))
88    };
89
90    // Order impacts stream id in multiplexer!
91    vec![
92        MiniProtocols::Heartbeat(Heartbeat {}),
93        MiniProtocols::Manager(Manager::new(
94            Party::Client,
95            args.peer_id,
96            args.active_stores.clone(),
97            client_chan,
98            1,
99            multiplexer_cmd_send.clone(),
100        )),
101        MiniProtocols::Manager(Manager::new(
102            Party::Server,
103            args.peer_id,
104            args.active_stores,
105            server_chan,
106            2,
107            multiplexer_cmd_send,
108        )),
109    ]
110}
111
112pub(crate) async fn run_miniprotocols_server<O: OssaType>(
113    stream: TcpStream,
114    args: MiniProtocolArgs<
115        O::StoreId,
116        O::Hash,
117        <O::ECGHeader as ECGHeader>::HeaderId,
118        O::ECGHeader,
119    >,
120) {
121    run_miniprotocols::<O>(stream, args, Party::Server).await
122}
123
124pub(crate) async fn run_miniprotocols_client<O: OssaType>(
125    stream: TcpStream,
126    args: MiniProtocolArgs<
127        O::StoreId,
128        O::Hash,
129        <O::ECGHeader as ECGHeader>::HeaderId,
130        O::ECGHeader,
131    >,
132) {
133    run_miniprotocols::<O>(stream, args, Party::Client).await
134}
135
136async fn run_miniprotocols<O: OssaType>(
137    stream: TcpStream,
138    args: MiniProtocolArgs<
139        O::StoreId,
140        O::Hash,
141        <O::ECGHeader as ECGHeader>::HeaderId,
142        O::ECGHeader,
143    >,
144    party: Party,
145) {
146    // Start multiplexer.
147    let (mux_cmd_send, mux_cmd_recv) = mpsc::unbounded_channel();
148    let multiplexer = Multiplexer::new(party, mux_cmd_recv);
149
150    multiplexer
151        .run_with_miniprotocols::<O>(stream, initial_miniprotocols(party, args, mux_cmd_send))
152        .await;
153}
154
155// # Protocols run between peers.
156
157// request_store_metadata_header :: RequestStoreMetadataHeaderV0 -> Either<ProtocolError, ResponseStoreMetadataHeaderV0>
158// pub type StoreMetadataHeader<StoreId> = Send<
159//     StoreMetadataHeaderRequest<StoreId>,
160//     Recv<ProtocolResult<StoreMetadataHeaderResponse<StoreId>>, Eps>,
161// >;
162// pub type StoreMetadataBody =
163//     Send<StoreMetadataBodyRequest, Recv<ProtocolResult<StoreMetadataBodyResponse>, Eps>>;
164
165//
166// TODO: Do we need to do a handshake to establish a MAC key (is an encryption needed)?.
167//
168
169// # Messages sent by protocols.
170#[derive(Debug, Serialize, Deserialize)]
171pub enum MsgStoreMetadataHeader<StoreId> {
172    Request(StoreMetadataHeaderRequest<StoreId>),
173    Response(StoreMetadataHeaderResponse<StoreId>),
174}
175
176impl<StoreId> Into<MsgStoreMetadataHeader<StoreId>> for StoreMetadataHeaderRequest<StoreId> {
177    fn into(self) -> MsgStoreMetadataHeader<StoreId> {
178        MsgStoreMetadataHeader::Request(self)
179    }
180}
181impl<StoreId> Into<MsgStoreMetadataHeader<StoreId>> for StoreMetadataHeaderResponse<StoreId> {
182    fn into(self) -> MsgStoreMetadataHeader<StoreId> {
183        MsgStoreMetadataHeader::Response(self)
184    }
185}
186impl<StoreId> TryInto<StoreMetadataHeaderRequest<StoreId>> for MsgStoreMetadataHeader<StoreId> {
187    type Error = ();
188    fn try_into(self) -> Result<StoreMetadataHeaderRequest<StoreId>, ()> {
189        match self {
190            MsgStoreMetadataHeader::Request(r) => Ok(r),
191            MsgStoreMetadataHeader::Response(_) => Err(()),
192        }
193    }
194}
195impl<StoreId> TryInto<StoreMetadataHeaderResponse<StoreId>> for MsgStoreMetadataHeader<StoreId> {
196    type Error = ();
197    fn try_into(self) -> Result<StoreMetadataHeaderResponse<StoreId>, ()> {
198        match self {
199            MsgStoreMetadataHeader::Response(r) => Ok(r),
200            MsgStoreMetadataHeader::Request(_) => Err(()),
201        }
202    }
203}
204
205#[derive(Debug, Deserialize, Serialize)]
206pub struct StoreMetadataHeaderRequest<StoreId> {
207    pub store_id: StoreId,
208    pub body_request: Option<StoreMetadataBodyRequest>,
209}
210
211#[derive(Debug, Deserialize, Serialize)]
212pub struct StoreMetadataHeaderResponse<StoreId> {
213    pub header: store::v0::MetadataHeader<StoreId>,
214    // pub body: Option<StoreMetadataBodyResponse<StoreId>>,
215}
216
217pub type StoreMetadataBodyRequest = (); // TODO: Eventually request certain chunks.
218pub type StoreMetadataBodyResponse<StoreId> = store::v0::MetadataBody<StoreId>; // TODO: Eventually request certain chunks.
219
220pub type ProtocolResult<T> = Result<T, ProtocolError>;
221pub type ProtocolError = String; // TODO: Eventually more informative error type.
ossa_core/protocol/v0.rs

ossa_core/protocol/
v0.rs
