doc_db/
dodb.rs

1use anyhow::{Result, anyhow, bail};
2use base64::Engine;
3use bytes::Bytes;
4use dodb_client::{ClientError, ClientTlsConfig, DodbClient, DodbConnection as ClientConnection};
5use dodb_core::{
6    ConditionExpectation, DocumentKey, PrimaryKey, RevisionState, SortKey, TenantId,
7    TransactionCondition, TransactionMutation, TransactionRequest,
8};
9use dodb_protocol::{ApplicationErrorKind, ProtocolLimits};
10use std::env;
11use std::io::Cursor;
12use std::net::SocketAddr;
13
14use crate::{
15    ObservedDocument, TransactCondition, TransactConflict, TransactMutation, TransactOutcome,
16    TransactRequest,
17};
18
19#[derive(Clone, Debug)]
20pub struct DodbConfig {
21    server_addr: SocketAddr,
22    server_name: String,
23    bind_addr: SocketAddr,
24    root_certificates: Vec<Vec<u8>>,
25    protocol_limits: ProtocolLimits,
26}
27
28impl DodbConfig {
29    pub fn new(
30        server_addr: SocketAddr,
31        server_name: impl Into<String>,
32        root_certificates: Vec<Vec<u8>>,
33    ) -> Self {
34        Self {
35            server_addr,
36            server_name: server_name.into(),
37            bind_addr: "0.0.0.0:0".parse().expect("valid default bind address"),
38            root_certificates,
39            protocol_limits: ProtocolLimits::default(),
40        }
41    }
42
43    pub fn from_env() -> Result<Self> {
44        let server_addr = required_env("DODB_ADDR")?.parse::<SocketAddr>()?;
45        let server_name = required_env("DODB_SERVER_NAME")?;
46        let bind_addr = env::var("DODB_BIND_ADDR")
47            .unwrap_or_else(|_| "0.0.0.0:0".to_string())
48            .parse::<SocketAddr>()?;
49        let root_certificates = root_certificates_from_env()?;
50
51        Ok(Self {
52            server_addr,
53            server_name,
54            bind_addr,
55            root_certificates,
56            protocol_limits: ProtocolLimits::default(),
57        })
58    }
59
60    pub fn with_bind_addr(mut self, bind_addr: SocketAddr) -> Self {
61        self.bind_addr = bind_addr;
62        self
63    }
64
65    pub fn with_protocol_limits(mut self, protocol_limits: ProtocolLimits) -> Self {
66        self.protocol_limits = protocol_limits;
67        self
68    }
69}
70
71#[derive(Clone)]
72pub struct DodbConnection {
73    inner: ClientConnection,
74    protocol_limits: ProtocolLimits,
75}
76
77impl DodbConnection {
78    pub async fn connect(config: &DodbConfig) -> Result<Self> {
79        let tls = client_tls(config)?;
80        let inner = ClientConnection::connect(
81            config.bind_addr,
82            config.server_addr,
83            &config.server_name,
84            tls,
85            config.protocol_limits,
86        )
87        .await
88        .map_err(client_error)?;
89        Ok(Self {
90            inner,
91            protocol_limits: config.protocol_limits,
92        })
93    }
94
95    pub fn connect_lazy(config: &DodbConfig) -> Result<Self> {
96        let tls = client_tls(config)?;
97        let inner = ClientConnection::connect_lazy(
98            config.bind_addr,
99            config.server_addr,
100            &config.server_name,
101            tls,
102            config.protocol_limits,
103        )
104        .map_err(client_error)?;
105        Ok(Self {
106            inner,
107            protocol_limits: config.protocol_limits,
108        })
109    }
110
111    pub fn close(&self) {
112        self.inner.close();
113    }
114
115    pub fn remote_addr(&self) -> SocketAddr {
116        self.inner.remote_addr()
117    }
118
119    pub(crate) fn database(&self, project_id: &str) -> Result<DodbDatabase> {
120        Ok(DodbDatabase {
121            client: self.inner.for_tenant(dodb_tenant_id(project_id)?),
122            protocol_limits: self.protocol_limits,
123        })
124    }
125}
126
127/// The tenant outside the normal eight-character base36 project namespace
128/// that is reserved for the fn0 control plane.
129pub const FN0_CONTROL_DODB_TENANT_ID: u64 = u64::MAX;
130
131/// Converts fn0's existing string project identity to its dodb tenant.
132///
133/// Normal project IDs are fixed-width lowercase base36 strings. Keeping the
134/// fixed width is important: accepting arbitrary base36 text would make
135/// aliases such as `1`, `01`, and `00000001` possible. The positional base36
136/// representation is injective, so no hash or probabilistic mapping is needed.
137pub fn dodb_tenant_id(project_id: &str) -> Result<TenantId> {
138    if project_id == "fn0-control" {
139        return Ok(TenantId::new(FN0_CONTROL_DODB_TENANT_ID));
140    }
141    if project_id == "local" {
142        bail!("local project has no dodb tenant");
143    }
144    if project_id.len() != 8
145        || !project_id
146            .bytes()
147            .all(|byte| byte.is_ascii_digit() || byte.is_ascii_lowercase())
148    {
149        bail!("invalid project ID {project_id:?}: expected eight lowercase base36 characters");
150    }
151    let value = u64::from_str_radix(project_id, 36)
152        .map_err(|error| anyhow!("invalid project ID {project_id:?}: {error}"))?;
153    Ok(TenantId::new(value))
154}
155
156/// Compatibility name for callers that describe this boundary as a project
157/// tenant lookup rather than a dodb-specific conversion.
158pub fn project_tenant_id(project_id: &str) -> Result<TenantId> {
159    dodb_tenant_id(project_id)
160}
161
162#[derive(Clone)]
163pub(crate) struct DodbDatabase {
164    client: DodbClient,
165    protocol_limits: ProtocolLimits,
166}
167
168impl DodbDatabase {
169    pub(crate) async fn get(&self, pk: &str, sk: &str) -> Result<Option<Bytes>> {
170        match self
171            .client
172            .get(document_key(pk, sk))
173            .await
174            .map_err(client_error)?
175        {
176            RevisionState::Present { value, .. } => Ok(Some(value.into())),
177            RevisionState::Missing { .. } => Ok(None),
178        }
179    }
180
181    pub(crate) async fn put(&self, pk: &str, sk: &str, data: &[u8]) -> Result<()> {
182        self.client
183            .put(document_key(pk, sk), data.to_vec())
184            .await
185            .map(|_| ())
186            .map_err(client_error)
187    }
188
189    pub(crate) async fn delete(&self, pk: &str, sk: &str) -> Result<()> {
190        self.client
191            .delete(document_key(pk, sk))
192            .await
193            .map(|_| ())
194            .map_err(client_error)
195    }
196
197    pub(crate) async fn query(
198        &self,
199        pk: &str,
200        after_sk: Option<&str>,
201        limit: usize,
202    ) -> Result<Vec<(String, Bytes)>> {
203        let mut rows = Vec::with_capacity(limit.min(self.protocol_limits.max_query_limit));
204        let mut cursor = after_sk.map(str::to_owned);
205        while rows.len() < limit {
206            let page_limit = (limit - rows.len()).min(self.protocol_limits.max_query_limit);
207            let page = self
208                .client
209                .query(
210                    PrimaryKey::new(pk.as_bytes().to_vec()),
211                    cursor
212                        .as_deref()
213                        .map(|sk| SortKey::new(sk.as_bytes().to_vec())),
214                    page_limit,
215                )
216                .await
217                .map_err(client_error)?;
218            if page.is_empty() {
219                break;
220            }
221            let page_len = page.len();
222            let mut last_sort_key = None;
223            for document in page {
224                let DocumentKey { pk: row_pk, sk } = document.key;
225                if row_pk.as_bytes() != pk.as_bytes() {
226                    bail!("dodb query returned an unexpected partition key")
227                }
228                let sort_key = string_from_key_component("sort key", sk.into_bytes())?;
229                last_sort_key = Some(sort_key.clone());
230                rows.push((sort_key, document.value.into()));
231            }
232            cursor = last_sort_key;
233            if page_len < page_limit {
234                break;
235            }
236        }
237        Ok(rows)
238    }
239
240    pub(crate) async fn scan(
241        &self,
242        after: Option<(&str, &str)>,
243        limit: usize,
244    ) -> Result<Vec<(String, String, Bytes)>> {
245        let mut rows = Vec::with_capacity(limit.min(self.protocol_limits.max_scan_limit));
246        let mut cursor = after.map(|(pk, sk)| (pk.to_owned(), sk.to_owned()));
247        while rows.len() < limit {
248            let page_limit = (limit - rows.len()).min(self.protocol_limits.max_scan_limit);
249            let page = self
250                .client
251                .scan(
252                    cursor.as_ref().map(|(pk, sk)| document_key(pk, sk)),
253                    page_limit,
254                )
255                .await
256                .map_err(client_error)?;
257            if page.is_empty() {
258                break;
259            }
260            let page_len = page.len();
261            let mut last_key = None;
262            for document in page {
263                let DocumentKey { pk, sk } = document.key;
264                let pk = string_from_key_component("partition key", pk.into_bytes())?;
265                let sk = string_from_key_component("sort key", sk.into_bytes())?;
266                last_key = Some((pk.clone(), sk.clone()));
267                rows.push((pk, sk, document.value.into()));
268            }
269            cursor = last_key;
270            if page_len < page_limit {
271                break;
272            }
273        }
274        Ok(rows)
275    }
276
277    pub(crate) async fn get_observed(&self, pk: &str, sk: &str) -> Result<ObservedDocument> {
278        let state = self
279            .client
280            .get(document_key(pk, sk))
281            .await
282            .map_err(client_error)?;
283        Ok(observed_document(state))
284    }
285
286    pub(crate) async fn transact(&self, request: &TransactRequest) -> Result<TransactOutcome> {
287        if request.conditions.is_empty() && request.mutations.is_empty() {
288            return Ok(TransactOutcome { conflict: None });
289        }
290
291        let dodb_request = TransactionRequest::new(
292            request.conditions.iter().map(dodb_condition).collect(),
293            request.mutations.iter().map(dodb_mutation).collect(),
294        );
295        match self.client.transact(dodb_request).await {
296            Ok(_) => Ok(TransactOutcome { conflict: None }),
297            Err(ClientError::Application(error))
298                if error.kind == ApplicationErrorKind::Conflict =>
299            {
300                let details = error
301                    .conflict
302                    .as_ref()
303                    .ok_or_else(|| anyhow!("dodb conflict response did not include details"))?;
304                let condition_index = request
305                    .conditions
306                    .iter()
307                    .position(|condition| condition_matches(condition, details))
308                    .ok_or_else(|| {
309                        anyhow!("dodb conflict did not match any requested transaction condition")
310                    })?;
311                Ok(TransactOutcome {
312                    conflict: Some(TransactConflict { condition_index }),
313                })
314            }
315            Err(error) => Err(client_error(error)),
316        }
317    }
318}
319
320fn required_env(name: &str) -> Result<String> {
321    env::var(name).map_err(|_| anyhow!("{name} must be set"))
322}
323
324fn client_tls(config: &DodbConfig) -> Result<ClientTlsConfig> {
325    ClientTlsConfig::from_der(config.root_certificates.clone()).map_err(client_error)
326}
327
328fn root_certificates_from_env() -> Result<Vec<Vec<u8>>> {
329    let pem = match env::var("DODB_ROOT_CERT_PEM") {
330        Ok(value) => value.into_bytes(),
331        Err(_) => {
332            let encoded = required_env("DODB_ROOT_CERT_PEM_BASE64")?;
333            base64::engine::general_purpose::STANDARD.decode(encoded)?
334        }
335    };
336    let mut reader = Cursor::new(pem);
337    let certificates = rustls_pemfile::certs(&mut reader)
338        .collect::<std::result::Result<Vec<_>, _>>()?
339        .into_iter()
340        .map(|certificate| certificate.to_vec())
341        .collect::<Vec<_>>();
342    if certificates.is_empty() {
343        bail!("DODB_ROOT_CERT_PEM must contain at least one certificate")
344    }
345    Ok(certificates)
346}
347
348fn client_error(error: ClientError) -> anyhow::Error {
349    anyhow::Error::new(error)
350}
351
352fn document_key(pk: &str, sk: &str) -> DocumentKey {
353    DocumentKey::new(pk.as_bytes().to_vec(), sk.as_bytes().to_vec())
354}
355
356fn string_from_key_component(name: &str, bytes: Vec<u8>) -> Result<String> {
357    String::from_utf8(bytes)
358        .map_err(|error| anyhow!("dodb returned an invalid UTF-8 {name}: {error}"))
359}
360
361fn observed_document(state: RevisionState) -> ObservedDocument {
362    match state {
363        RevisionState::Present { value, revision } => ObservedDocument::Present {
364            data: value.into(),
365            revision: crate::DocDbRevision::new(revision.get()),
366        },
367        RevisionState::Missing { revision } => ObservedDocument::Missing {
368            revision: Some(crate::DocDbRevision::new(revision.get())),
369        },
370    }
371}
372
373fn dodb_condition(condition: &TransactCondition) -> TransactionCondition {
374    match condition {
375        TransactCondition::RevisionEquals {
376            pk,
377            sk,
378            expected_revision,
379        } => TransactionCondition::RevisionEquals {
380            key: document_key(pk, sk),
381            expected_revision: dodb_core::Revision::new(expected_revision.value()),
382        },
383        TransactCondition::Exists { pk, sk } => TransactionCondition::Exists {
384            key: document_key(pk, sk),
385        },
386        TransactCondition::NotExists { pk, sk } => TransactionCondition::NotExists {
387            key: document_key(pk, sk),
388        },
389    }
390}
391
392fn dodb_mutation(mutation: &TransactMutation) -> TransactionMutation {
393    match mutation {
394        TransactMutation::Put { pk, sk, data } => TransactionMutation::Put {
395            key: document_key(pk, sk),
396            value: data.clone(),
397        },
398        TransactMutation::Delete { pk, sk } => TransactionMutation::Delete {
399            key: document_key(pk, sk),
400        },
401    }
402}
403
404fn condition_matches(
405    condition: &TransactCondition,
406    conflict: &dodb_protocol::ConflictDetails,
407) -> bool {
408    let (key, expectation) = match condition {
409        TransactCondition::RevisionEquals {
410            pk,
411            sk,
412            expected_revision,
413        } => (
414            document_key(pk, sk),
415            ConditionExpectation::RevisionEquals(dodb_core::Revision::new(
416                expected_revision.value(),
417            )),
418        ),
419        TransactCondition::Exists { pk, sk } => {
420            (document_key(pk, sk), ConditionExpectation::Exists)
421        }
422        TransactCondition::NotExists { pk, sk } => {
423            (document_key(pk, sk), ConditionExpectation::NotExists)
424        }
425    };
426    key == conflict.key && expectation == conflict.expected
427}
428
429#[cfg(test)]
430mod tests {
431    use super::*;
432    use crate::{
433        DocDbRevision, TransactCondition, TransactMutation, TransactRequest, dodb_with_connection,
434    };
435    use dodb_server::{
436        DodbServer, DodbServerConfig, LocalTenantService, LocalTenantServiceConfig, ServerError,
437        ServerTlsConfig,
438    };
439    use rcgen::generate_simple_self_signed;
440    use std::path::PathBuf;
441    use std::sync::Arc;
442    use tokio::task::JoinHandle;
443
444    struct TestTls {
445        certificate: Vec<u8>,
446        private_key: Vec<u8>,
447    }
448
449    fn test_tls() -> TestTls {
450        let certified = generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap();
451        TestTls {
452            certificate: certified.cert.der().to_vec(),
453            private_key: certified.signing_key.serialize_der(),
454        }
455    }
456
457    #[test]
458    fn normal_project_ids_use_exact_base36_values() {
459        let cases = [
460            ("00000000", 0),
461            ("00000001", 1),
462            ("0000000z", 35),
463            ("00000010", 36),
464            ("zzzzzzzz", 2_821_109_907_455),
465        ];
466        for (project_id, expected) in cases {
467            assert_eq!(dodb_tenant_id(project_id).unwrap(), TenantId::new(expected));
468        }
469    }
470
471    #[test]
472    fn control_uses_a_reserved_tenant_outside_the_project_namespace() {
473        assert_eq!(
474            dodb_tenant_id("fn0-control").unwrap(),
475            TenantId::new(FN0_CONTROL_DODB_TENANT_ID)
476        );
477        assert!(TenantId::new(2_821_109_907_455) < TenantId::new(u64::MAX));
478    }
479
480    #[test]
481    fn local_has_no_dodb_tenant() {
482        let error = dodb_tenant_id("local").unwrap_err().to_string();
483        assert!(error.contains("no dodb tenant"), "{error}");
484    }
485
486    #[test]
487    fn arbitrary_and_noncanonical_project_ids_are_rejected() {
488        for project_id in [
489            "1",
490            "01",
491            "000000001",
492            "ABCDEFGH",
493            "abc-defg",
494            "fn0-foo",
495            "",
496        ] {
497            assert!(
498                dodb_tenant_id(project_id).is_err(),
499                "{project_id:?} unexpectedly mapped to a dodb tenant"
500            );
501        }
502    }
503
504    #[test]
505    fn representative_canonical_ids_are_injective() {
506        let project_ids = ["00000000", "00000001", "0000000z", "00000010", "zzzzzzzz"];
507        let tenants = project_ids
508            .into_iter()
509            .map(|project_id| dodb_tenant_id(project_id).unwrap())
510            .collect::<Vec<_>>();
511        for (index, tenant) in tenants.iter().enumerate() {
512            assert!(tenants[index + 1..].iter().all(|other| other != tenant));
513        }
514    }
515
516    async fn start_server(
517        data_dir: PathBuf,
518        tls: &TestTls,
519    ) -> (
520        Arc<DodbServer<LocalTenantService>>,
521        JoinHandle<Result<(), ServerError>>,
522    ) {
523        start_server_at(data_dir, tls, "127.0.0.1:0".parse().unwrap()).await
524    }
525
526    async fn start_server_at(
527        data_dir: PathBuf,
528        tls: &TestTls,
529        listen_addr: SocketAddr,
530    ) -> (
531        Arc<DodbServer<LocalTenantService>>,
532        JoinHandle<Result<(), ServerError>>,
533    ) {
534        let service = Arc::new(
535            LocalTenantService::new(LocalTenantServiceConfig {
536                data_dir,
537                ..LocalTenantServiceConfig::default()
538            })
539            .unwrap(),
540        );
541        let server = Arc::new(
542            DodbServer::bind(
543                service,
544                DodbServerConfig {
545                    listen_addr,
546                    tls: ServerTlsConfig::from_der(
547                        vec![tls.certificate.clone()],
548                        tls.private_key.clone(),
549                    )
550                    .unwrap(),
551                    protocol_limits: ProtocolLimits::default(),
552                    max_connections: 8,
553                    max_concurrent_streams: 64,
554                    max_concurrent_requests: 64,
555                },
556            )
557            .unwrap(),
558        );
559        let task_server = Arc::clone(&server);
560        let task = tokio::spawn(async move { task_server.run().await });
561        (server, task)
562    }
563
564    async fn test_connection(
565        server: &DodbServer<LocalTenantService>,
566        tls: &TestTls,
567    ) -> DodbConnection {
568        DodbConnection::connect(
569            &DodbConfig::new(
570                server.local_addr().unwrap(),
571                "localhost",
572                vec![tls.certificate.clone()],
573            )
574            .with_protocol_limits(ProtocolLimits::default()),
575        )
576        .await
577        .unwrap()
578    }
579
580    async fn stop_server(
581        connection: &DodbConnection,
582        server: Arc<DodbServer<LocalTenantService>>,
583        task: JoinHandle<Result<(), ServerError>>,
584    ) {
585        connection.close();
586        server.shutdown().await;
587        task.await.unwrap().unwrap();
588    }
589
590    #[tokio::test]
591    async fn lazy_connection_defers_dial_and_preserves_empty_transaction_noop() {
592        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
593        let server_tls = test_tls();
594        let client_tls = test_tls();
595        let directory = tempfile::tempdir().unwrap();
596        let (server, task) = start_server(directory.path().to_owned(), &server_tls).await;
597        let connection = DodbConnection::connect_lazy(&DodbConfig::new(
598            server.local_addr().unwrap(),
599            "localhost",
600            vec![client_tls.certificate.clone()],
601        ))
602        .unwrap();
603        let database = dodb_with_connection(&connection, "00000000").unwrap();
604        assert_eq!(server.metrics().snapshot().connections_total, 0);
605
606        let empty = database
607            .transact(&TransactRequest {
608                conditions: vec![],
609                mutations: vec![],
610            })
611            .await
612            .unwrap();
613        assert!(empty.conflict.is_none());
614
615        let error = tokio::time::timeout(
616            std::time::Duration::from_secs(2),
617            database.get("lazy", "first-request"),
618        )
619        .await
620        .expect("first lazy request did not finish promptly")
621        .unwrap_err();
622        assert!(error.downcast_ref::<ClientError>().is_some(), "{error:#}");
623
624        stop_server(&connection, server, task).await;
625    }
626
627    #[tokio::test]
628    async fn dodb_backend_runs_the_shared_contract_over_real_quic() {
629        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
630        let directory = tempfile::tempdir().unwrap();
631        let tls = test_tls();
632        let (server, task) = start_server(directory.path().to_owned(), &tls).await;
633        let connection = test_connection(&server, &tls).await;
634        let database = dodb_with_connection(&connection, "00000000").unwrap();
635        let new_connection = connection.clone();
636
637        let result = crate::backend_contract::run_revisioned_missing_backend_contract(
638            database,
639            move || dodb_with_connection(&new_connection, "00000000").unwrap(),
640            "dodb",
641        )
642        .await;
643
644        stop_server(&connection, server, task).await;
645        result.unwrap();
646    }
647
648    #[tokio::test]
649    async fn dodb_backend_rejects_stale_missing_revision_after_delete() {
650        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
651        let directory = tempfile::tempdir().unwrap();
652        let tls = test_tls();
653        let (server, task) = start_server(directory.path().to_owned(), &tls).await;
654        let connection = test_connection(&server, &tls).await;
655        let database = dodb_with_connection(&connection, "00000001").unwrap();
656        let pk = "missing-revision";
657        let sk = "key";
658
659        let initial = database.get_observed(pk, sk).await.unwrap();
660        assert!(matches!(
661            initial,
662            ObservedDocument::Missing {
663                revision: Some(revision)
664            } if revision == DocDbRevision::new(0)
665        ));
666
667        database.put(pk, sk, b"present").await.unwrap();
668        database.delete(pk, sk).await.unwrap();
669        let deleted = database.get_observed(pk, sk).await.unwrap();
670        let deleted_revision = match deleted {
671            ObservedDocument::Missing {
672                revision: Some(revision),
673            } => revision,
674            _ => panic!("expected an exact missing revision"),
675        };
676        assert_ne!(deleted_revision, DocDbRevision::new(0));
677
678        let outcome = database
679            .transact(&TransactRequest {
680                conditions: vec![TransactCondition::RevisionEquals {
681                    pk: pk.to_owned(),
682                    sk: sk.to_owned(),
683                    expected_revision: DocDbRevision::new(0),
684                }],
685                mutations: vec![TransactMutation::Put {
686                    pk: pk.to_owned(),
687                    sk: sk.to_owned(),
688                    data: b"must-not-commit".to_vec(),
689                }],
690            })
691            .await
692            .unwrap();
693        assert_eq!(outcome.conflict.unwrap().condition_index, 0);
694        assert_eq!(database.get(pk, sk).await.unwrap(), None);
695
696        stop_server(&connection, server, task).await;
697    }
698
699    #[tokio::test]
700    async fn existing_database_handle_reconnects_after_same_address_restart() {
701        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
702        let directory = tempfile::tempdir().unwrap();
703        let tls = test_tls();
704        let (server, task) = start_server(directory.path().to_owned(), &tls).await;
705        let address = server.local_addr().unwrap();
706        let connection = test_connection(&server, &tls).await;
707        let database = dodb_with_connection(&connection, "00000002").unwrap();
708        let key = ("reconnect", "same-handle");
709
710        database.put(key.0, key.1, b"persisted").await.unwrap();
711
712        server.close();
713        let failure = database.get(key.0, key.1).await.unwrap_err();
714        assert!(
715            failure.downcast_ref::<ClientError>().is_some(),
716            "{failure:#}"
717        );
718        server.shutdown().await;
719        task.await.unwrap().unwrap();
720        drop(server);
721        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
722
723        let (restarted, restarted_task) =
724            start_server_at(directory.path().to_owned(), &tls, address).await;
725        let value = tokio::time::timeout(
726            std::time::Duration::from_secs(2),
727            database.get(key.0, key.1),
728        )
729        .await
730        .expect("reconnect request did not finish promptly")
731        .unwrap();
732        assert_eq!(value.as_deref(), Some(b"persisted".as_slice()));
733
734        stop_server(&connection, restarted, restarted_task).await;
735    }
736}
doc_db/dodb.rs

doc_db/
dodb.rs
