mongreldb_client/
native.rs

1//! High-level native RPC client over pooled multiplexed TLS connections.
2
3use std::collections::VecDeque;
4use std::io::Cursor;
5use std::sync::Arc;
6use std::time::{Duration, SystemTime, UNIX_EPOCH};
7
8use arrow::ipc::reader::StreamReader;
9use arrow::record_batch::RecordBatch;
10use mongreldb_protocol::native;
11use mongreldb_protocol::native_transport::{NativeRpcClientConfig, NativeRpcClientPool};
12use mongreldb_protocol::{NATIVE_API_MAJOR, NATIVE_API_MINOR};
13use mongreldb_query::QueryId;
14use prost::Message;
15
16use crate::{ClientError, ClientResult, SecretString};
17
18pub use mongreldb_protocol::native_transport::NativeRpcClientConfig as Config;
19
20#[derive(Clone)]
21pub struct NativeClient {
22    pool: Arc<NativeRpcClientPool>,
23    database_id: [u8; 16],
24    request_timeout: Duration,
25    max_retries: usize,
26}
27
28impl NativeClient {
29    pub async fn connect(
30        config: NativeRpcClientConfig,
31        connections: usize,
32        database_id: [u8; 16],
33    ) -> ClientResult<Self> {
34        let request_timeout = config.request_timeout;
35        let pool = NativeRpcClientPool::connect(config, connections)
36            .await
37            .map_err(|error| ClientError::Transport(error.to_string()))?;
38        Ok(Self {
39            pool,
40            database_id,
41            request_timeout,
42            max_retries: 2,
43        })
44    }
45
46    pub fn with_max_retries(mut self, max_retries: usize) -> Self {
47        self.max_retries = max_retries;
48        self
49    }
50
51    pub async fn authenticate_password(
52        &self,
53        username: impl Into<String>,
54        password: &SecretString,
55    ) -> ClientResult<NativeSession> {
56        let username = username.into();
57        let client_nonce = QueryId::random()
58            .map_err(|error| ClientError::Transport(error.to_string()))?
59            .to_string();
60        let mut exchange = mongreldb_core::ScramClientSession::begin(
61            &username,
62            secrecy::ExposeSecret::expose_secret(password),
63            &client_nonce,
64        )
65        .map_err(|error| ClientError::Transport(error.to_string()))?;
66        let begin = self
67            .pool
68            .client()
69            .auth()
70            .begin_scram(native::BeginScramRequest {
71                context: Some(context(self.request_timeout, None)?),
72                username,
73                client_first_bare: exchange.client_first_bare().into(),
74                client_nonce,
75            })
76            .await
77            .map_err(native_error)?
78            .into_inner();
79        let (client_final_without_proof, client_proof) = exchange
80            .respond(&begin.server_first)
81            .map_err(|error| ClientError::Transport(error.to_string()))?;
82        let finish = self
83            .pool
84            .client()
85            .auth()
86            .finish_scram(native::FinishScramRequest {
87                context: Some(context(self.request_timeout, None)?),
88                exchange_id: begin.exchange_id,
89                client_final_without_proof,
90                client_proof,
91            })
92            .await
93            .map_err(native_error)?
94            .into_inner();
95        exchange
96            .verify_server_final(&finish.server_final)
97            .map_err(|error| ClientError::Transport(error.to_string()))?;
98        self.open_session(
99            finish
100                .authentication
101                .ok_or_else(|| ClientError::Decode("SCRAM authentication missing".into()))?,
102        )
103        .await
104    }
105
106    pub async fn authenticate_anonymous(&self) -> ClientResult<NativeSession> {
107        self.authenticate(None).await
108    }
109
110    /// Authenticate one MySQL wire handshake without exposing its password to
111    /// the adapter. The native server validates the `caching_sha2_password`
112    /// proof against the catalog verifier and returns a normal live session.
113    pub async fn authenticate_mysql_caching_sha2(
114        &self,
115        username: impl Into<String>,
116        nonce: &[u8],
117        proof: &[u8],
118    ) -> ClientResult<NativeSession> {
119        self.authenticate(Some(
120            native::authenticate_request::Credential::MysqlCachingSha2(
121                native::MysqlCachingSha2Credential {
122                    username: username.into(),
123                    nonce: nonce.to_vec(),
124                    proof: proof.to_vec(),
125                },
126            ),
127        ))
128        .await
129    }
130
131    async fn authenticate(
132        &self,
133        credential: Option<native::authenticate_request::Credential>,
134    ) -> ClientResult<NativeSession> {
135        let auth = self
136            .pool
137            .client()
138            .auth()
139            .authenticate(native::AuthenticateRequest {
140                context: Some(context(self.request_timeout, None)?),
141                credential,
142            })
143            .await
144            .map_err(native_error)?
145            .into_inner();
146        self.open_session(auth).await
147    }
148
149    async fn open_session(
150        &self,
151        auth: native::AuthenticateResponse,
152    ) -> ClientResult<NativeSession> {
153        let session_id = self
154            .pool
155            .client()
156            .session()
157            .open_session(native::OpenSessionRequest {
158                context: Some(context(self.request_timeout, None)?),
159                identity: auth.identity,
160                database_id: self.database_id.to_vec(),
161                auth_token: auth.auth_token,
162            })
163            .await
164            .map_err(native_error)?
165            .into_inner()
166            .session_id;
167        Ok(NativeSession {
168            pool: Arc::clone(&self.pool),
169            session_id,
170            database_id: self.database_id,
171            request_timeout: self.request_timeout,
172            max_retries: self.max_retries,
173        })
174    }
175}
176
177#[derive(Clone)]
178pub struct NativeSession {
179    pool: Arc<NativeRpcClientPool>,
180    session_id: Vec<u8>,
181    database_id: [u8; 16],
182    request_timeout: Duration,
183    max_retries: usize,
184}
185
186impl NativeSession {
187    pub fn id(&self) -> &[u8] {
188        &self.session_id
189    }
190
191    pub async fn close(self) -> ClientResult<()> {
192        self.pool
193            .client()
194            .session()
195            .close_session(native::CloseSessionRequest {
196                context: Some(context(self.request_timeout, None)?),
197                session_id: self.session_id,
198            })
199            .await
200            .map_err(native_error)?;
201        Ok(())
202    }
203
204    pub async fn create_table(
205        &self,
206        table: impl Into<String>,
207        schema: &mongreldb_core::Schema,
208    ) -> ClientResult<native::CreateTableResponse> {
209        if schema.columns.iter().any(|column| {
210            matches!(
211                column.embedding_source.as_ref(),
212                Some(mongreldb_core::EmbeddingSource::LocalModel { .. })
213            )
214        }) {
215            return Err(ClientError::Decode(
216                "native CreateTable cannot transport node-local model paths; use ConfiguredModel or GeneratedColumnSpec"
217                    .into(),
218            ));
219        }
220        let full_schema_required = !schema.indexes.is_empty()
221            || !schema.constraints.checks.is_empty()
222            || schema
223                .columns
224                .iter()
225                .any(|column| column.default_value.is_some() || column.embedding_source.is_some());
226        let legacy_columns = schema
227            .columns
228            .iter()
229            .map(native_column)
230            .collect::<ClientResult<Vec<_>>>();
231        let use_legacy_fields = !full_schema_required && legacy_columns.is_ok();
232        let schema_json = serde_json::to_vec(schema).map_err(|error| {
233            ClientError::Decode(format!("native schema encode failed: {error}"))
234        })?;
235        let response = self
236            .pool
237            .client()
238            .catalog()
239            .create_table(native::CreateTableRequest {
240                context: Some(context(self.request_timeout, None)?),
241                session_id: self.session_id.clone(),
242                table: table.into(),
243                schema_id: schema.schema_id,
244                columns: if use_legacy_fields {
245                    legacy_columns.unwrap_or_default()
246                } else {
247                    Vec::new()
248                },
249                uniques: if use_legacy_fields {
250                    schema
251                        .constraints
252                        .uniques
253                        .iter()
254                        .map(|constraint| native::UniqueConstraint {
255                            id: u32::from(constraint.id),
256                            name: constraint.name.clone(),
257                            columns: constraint.columns.iter().map(|id| u32::from(*id)).collect(),
258                        })
259                        .collect()
260                } else {
261                    Vec::new()
262                },
263                foreign_keys: if use_legacy_fields {
264                    schema
265                        .constraints
266                        .foreign_keys
267                        .iter()
268                        .map(native_foreign_key)
269                        .collect()
270                } else {
271                    Vec::new()
272                },
273                schema_json,
274            })
275            .await
276            .map_err(native_error)?
277            .into_inner();
278        Ok(response)
279    }
280
281    pub async fn schema(&self, table: impl Into<String>) -> ClientResult<mongreldb_core::Schema> {
282        let response = self
283            .pool
284            .client()
285            .catalog()
286            .get_schema(native::GetSchemaRequest {
287                context: Some(context(self.request_timeout, None)?),
288                database_id: self.database_id.to_vec(),
289                table: table.into(),
290                session_id: self.session_id.clone(),
291            })
292            .await
293            .map_err(native_error)?
294            .into_inner();
295        if response.schema_json.is_empty() {
296            return Err(ClientError::Decode(
297                "native schema response omitted complete schema_json".into(),
298            ));
299        }
300        serde_json::from_slice(&response.schema_json)
301            .map_err(|error| ClientError::Decode(format!("native schema decode failed: {error}")))
302    }
303
304    pub async fn prepare(&self, sql: impl Into<String>) -> ClientResult<NativePrepared> {
305        let response = self
306            .pool
307            .client()
308            .query()
309            .prepare(native::PrepareRequest {
310                context: Some(context(self.request_timeout, None)?),
311                session_id: self.session_id.clone(),
312                sql: sql.into(),
313            })
314            .await
315            .map_err(native_error)?
316            .into_inner();
317        Ok(NativePrepared {
318            statement_id: response.statement_id,
319            schema_version: response.schema_version,
320        })
321    }
322
323    pub async fn execute(
324        &self,
325        sql: impl Into<String>,
326        idempotency_key: Option<&str>,
327    ) -> ClientResult<NativeExecuteResult> {
328        let sql = sql.into();
329        let retryable = idempotency_key.is_some() || read_only_sql(&sql);
330        self.execute_request(
331            native::execute_request::Command::Sql(sql),
332            Vec::new(),
333            idempotency_key,
334            retryable,
335        )
336        .await
337    }
338
339    pub async fn execute_prepared(
340        &self,
341        statement: NativePrepared,
342        parameters: &[mongreldb_protocol::request::ParameterValue],
343    ) -> ClientResult<NativeExecuteResult> {
344        let parameters = parameters
345            .iter()
346            .map(|parameter| {
347                bincode::serialize(parameter)
348                    .map_err(|error| ClientError::Decode(error.to_string()))
349            })
350            .collect::<ClientResult<Vec<_>>>()?;
351        self.execute_request(
352            native::execute_request::Command::PreparedStatementId(statement.statement_id),
353            parameters,
354            None,
355            true,
356        )
357        .await
358    }
359
360    async fn execute_request(
361        &self,
362        command: native::execute_request::Command,
363        parameters: Vec<Vec<u8>>,
364        idempotency_key: Option<&str>,
365        retryable: bool,
366    ) -> ClientResult<NativeExecuteResult> {
367        let mut attempt = 0;
368        loop {
369            let query_id = random_query_id()?;
370            let request = native::ExecuteRequest {
371                context: Some(context(self.request_timeout, idempotency_key)?),
372                session_id: self.session_id.clone(),
373                query_id: query_id.clone(),
374                command: Some(command.clone()),
375                parameters: parameters.clone(),
376            };
377            match self.pool.client().query().execute(request).await {
378                Ok(response) => return decode_execute(response.into_inner()),
379                Err(error)
380                    if retryable && attempt < self.max_retries && native_retryable(&error) =>
381                {
382                    attempt += 1;
383                }
384                Err(error) => return Err(native_error(error)),
385            }
386        }
387    }
388
389    pub async fn execute_stream(&self, sql: impl Into<String>) -> ClientResult<NativeArrowStream> {
390        let sql = sql.into();
391        let mut attempt = 0;
392        loop {
393            let query_id = random_query_id()?;
394            let request = native::ExecuteRequest {
395                context: Some(context(self.request_timeout, None)?),
396                session_id: self.session_id.clone(),
397                query_id: query_id.clone(),
398                command: Some(native::execute_request::Command::Sql(sql.clone())),
399                parameters: Vec::new(),
400            };
401            match self.pool.client().query().execute_stream(request).await {
402                Ok(response) => {
403                    return Ok(NativeArrowStream {
404                        query_id,
405                        inner: response.into_inner(),
406                        buffered: VecDeque::new(),
407                        finished: false,
408                    })
409                }
410                Err(error)
411                    if read_only_sql(&sql)
412                        && attempt < self.max_retries
413                        && native_retryable(&error) =>
414                {
415                    attempt += 1;
416                }
417                Err(error) => return Err(native_error(error)),
418            }
419        }
420    }
421
422    pub async fn cancel(&self, query_id: &[u8]) -> ClientResult<NativeCancelResult> {
423        let response = self
424            .pool
425            .client()
426            .query()
427            .cancel_query(native::CancelQueryRequest {
428                context: Some(context(self.request_timeout, None)?),
429                query_id: query_id.to_vec(),
430                session_id: self.session_id.clone(),
431            })
432            .await
433            .map(tonic::Response::into_inner)
434            .map_err(native_error)?;
435        Ok(NativeCancelResult {
436            outcome: native::CancelOutcome::try_from(response.outcome)
437                .unwrap_or(native::CancelOutcome::Unspecified),
438            durable: response.durable.unwrap_or_default(),
439        })
440    }
441
442    pub async fn query_status(&self, query_id: &[u8]) -> ClientResult<NativeQueryStatus> {
443        let response = self
444            .pool
445            .client()
446            .query()
447            .get_query_status(native::GetQueryStatusRequest {
448                context: Some(context(self.request_timeout, None)?),
449                query_id: query_id.to_vec(),
450                session_id: self.session_id.clone(),
451            })
452            .await
453            .map(tonic::Response::into_inner)
454            .map_err(native_error)?;
455        Ok(NativeQueryStatus {
456            query_id: response.query_id,
457            phase: response.phase,
458            error: response.error,
459            durable: response.durable.unwrap_or_default(),
460        })
461    }
462
463    pub async fn begin(&self, isolation: native::IsolationLevel) -> ClientResult<Vec<u8>> {
464        self.pool
465            .client()
466            .transaction()
467            .begin(native::BeginTransactionRequest {
468                context: Some(context(self.request_timeout, None)?),
469                session_id: self.session_id.clone(),
470                isolation: isolation as i32,
471            })
472            .await
473            .map(|response| response.into_inner().transaction_id)
474            .map_err(native_error)
475    }
476
477    pub async fn commit(&self) -> ClientResult<()> {
478        self.transaction("commit").await
479    }
480
481    pub async fn rollback(&self) -> ClientResult<()> {
482        self.transaction("rollback").await
483    }
484
485    async fn transaction(&self, operation: &str) -> ClientResult<()> {
486        let request = native::TransactionRequest {
487            context: Some(context(self.request_timeout, None)?),
488            session_id: self.session_id.clone(),
489        };
490        let mut client = self.pool.client().transaction();
491        match operation {
492            "commit" => client.commit(request).await,
493            _ => client.rollback(request).await,
494        }
495        .map_err(native_error)?;
496        Ok(())
497    }
498}
499
500fn native_column(column: &mongreldb_core::ColumnDef) -> ClientResult<native::CreateColumn> {
501    use mongreldb_core::TypeId;
502    let (data_type, decimal_precision, decimal_scale) = match &column.ty {
503        TypeId::Bool => (native::ColumnType::Bool, 0, 0),
504        TypeId::Int8 => (native::ColumnType::Int8, 0, 0),
505        TypeId::Int16 => (native::ColumnType::Int16, 0, 0),
506        TypeId::Int32 => (native::ColumnType::Int32, 0, 0),
507        TypeId::Int64 => (native::ColumnType::Int64, 0, 0),
508        TypeId::UInt8 => (native::ColumnType::Uint8, 0, 0),
509        TypeId::UInt16 => (native::ColumnType::Uint16, 0, 0),
510        TypeId::UInt32 => (native::ColumnType::Uint32, 0, 0),
511        TypeId::UInt64 => (native::ColumnType::Uint64, 0, 0),
512        TypeId::Float32 => (native::ColumnType::Float32, 0, 0),
513        TypeId::Float64 => (native::ColumnType::Float64, 0, 0),
514        TypeId::TimestampNanos => (native::ColumnType::TimestampNanos, 0, 0),
515        TypeId::Date32 => (native::ColumnType::Date32, 0, 0),
516        TypeId::Date64 => (native::ColumnType::Date64, 0, 0),
517        TypeId::Time64 => (native::ColumnType::Time64, 0, 0),
518        TypeId::Bytes => (native::ColumnType::Bytes, 0, 0),
519        TypeId::Json => (native::ColumnType::Json, 0, 0),
520        TypeId::Decimal128 { precision, scale } => (
521            native::ColumnType::Decimal128,
522            u32::from(*precision),
523            i32::from(*scale),
524        ),
525        _ => {
526            return Err(ClientError::Decode(format!(
527                "native CreateTable does not support column type {:?}",
528                column.ty
529            )))
530        }
531    };
532    let allowed_flags = mongreldb_core::ColumnFlags::NULLABLE
533        | mongreldb_core::ColumnFlags::PRIMARY_KEY
534        | mongreldb_core::ColumnFlags::AUTO_INCREMENT;
535    if column.flags.bits() & !allowed_flags != 0 {
536        return Err(ClientError::Decode(format!(
537            "native CreateTable does not support flags on column {:?}",
538            column.name
539        )));
540    }
541    Ok(native::CreateColumn {
542        id: u32::from(column.id),
543        name: column.name.clone(),
544        data_type: data_type as i32,
545        nullable: column.flags.contains(mongreldb_core::ColumnFlags::NULLABLE),
546        primary_key: column
547            .flags
548            .contains(mongreldb_core::ColumnFlags::PRIMARY_KEY),
549        decimal_precision,
550        decimal_scale,
551        auto_increment: column
552            .flags
553            .contains(mongreldb_core::ColumnFlags::AUTO_INCREMENT),
554    })
555}
556
557fn native_foreign_key(constraint: &mongreldb_core::constraint::ForeignKey) -> native::ForeignKey {
558    let action = |action| match action {
559        mongreldb_core::constraint::FkAction::Restrict => native::ForeignKeyAction::Restrict,
560        mongreldb_core::constraint::FkAction::Cascade => native::ForeignKeyAction::Cascade,
561        mongreldb_core::constraint::FkAction::SetNull => native::ForeignKeyAction::SetNull,
562    };
563    native::ForeignKey {
564        id: u32::from(constraint.id),
565        name: constraint.name.clone(),
566        columns: constraint.columns.iter().map(|id| u32::from(*id)).collect(),
567        referenced_table: constraint.ref_table.clone(),
568        referenced_columns: constraint
569            .ref_columns
570            .iter()
571            .map(|id| u32::from(*id))
572            .collect(),
573        on_delete: action(constraint.on_delete) as i32,
574        on_update: action(constraint.on_update) as i32,
575    }
576}
577
578#[derive(Debug, Clone, Copy, PartialEq, Eq)]
579pub struct NativePrepared {
580    pub statement_id: u64,
581    pub schema_version: u64,
582}
583
584/// Structural durable write outcome from native RPC (P0.6). No string parsing.
585pub type NativeDurableOutcome = native::DurableOutcome;
586
587#[derive(Debug, Clone)]
588pub struct NativeCancelResult {
589    pub outcome: native::CancelOutcome,
590    pub durable: NativeDurableOutcome,
591}
592
593#[derive(Debug, Clone)]
594pub struct NativeQueryStatus {
595    pub query_id: Vec<u8>,
596    pub phase: i32,
597    pub error: Option<native::ErrorDetail>,
598    pub durable: NativeDurableOutcome,
599}
600
601pub struct NativeExecuteResult {
602    pub query_id: Vec<u8>,
603    pub batches: Vec<RecordBatch>,
604    pub rows_affected: u64,
605    pub committed: bool,
606    pub commit_epoch: Option<u64>,
607    pub idempotency_replayed: bool,
608    pub original_query_id: Vec<u8>,
609    /// Prefer this over legacy `committed` / `commit_epoch` fields.
610    pub durable: NativeDurableOutcome,
611}
612
613pub struct NativeArrowStream {
614    query_id: Vec<u8>,
615    inner: tonic::Streaming<native::ArrowFrame>,
616    buffered: VecDeque<RecordBatch>,
617    finished: bool,
618}
619
620impl NativeArrowStream {
621    pub fn query_id(&self) -> &[u8] {
622        &self.query_id
623    }
624
625    pub async fn next_batch(&mut self) -> ClientResult<Option<RecordBatch>> {
626        if let Some(batch) = self.buffered.pop_front() {
627            return Ok(Some(batch));
628        }
629        while !self.finished {
630            let Some(frame) = self.inner.message().await.map_err(native_error)? else {
631                self.finished = true;
632                return Ok(None);
633            };
634            if frame.end_of_stream {
635                self.finished = true;
636            }
637            if frame.ipc.is_empty() {
638                continue;
639            }
640            self.buffered.extend(decode_ipc(&frame.ipc)?);
641            if let Some(batch) = self.buffered.pop_front() {
642                return Ok(Some(batch));
643            }
644        }
645        Ok(None)
646    }
647}
648
649fn decode_execute(response: native::ExecuteResponse) -> ClientResult<NativeExecuteResult> {
650    let mut batches = Vec::new();
651    for frame in &response.frames {
652        batches.extend(decode_ipc(&frame.ipc)?);
653    }
654    let durable = response.durable.unwrap_or_else(|| native::DurableOutcome {
655        committed: response.committed,
656        committed_statements: 0,
657        last_commit_hlc: Vec::new(),
658        first_commit_statement_index: None,
659        last_commit_statement_index: None,
660        completed_statements: 0,
661        current_statement_index: 0,
662        terminal_state: String::new(),
663        serialization_state: String::new(),
664        last_commit_epoch: (response.commit_epoch != 0).then_some(response.commit_epoch),
665    });
666    Ok(NativeExecuteResult {
667        query_id: response.query_id,
668        batches,
669        rows_affected: response.rows_affected,
670        committed: durable.committed || response.committed,
671        commit_epoch: durable
672            .last_commit_epoch
673            .or((response.commit_epoch != 0).then_some(response.commit_epoch)),
674        idempotency_replayed: response.idempotency_replayed,
675        original_query_id: response.original_query_id,
676        durable,
677    })
678}
679
680fn decode_ipc(ipc: &[u8]) -> ClientResult<Vec<RecordBatch>> {
681    if ipc.is_empty() {
682        return Ok(Vec::new());
683    }
684    StreamReader::try_new(Cursor::new(ipc), None)
685        .map_err(|error| ClientError::Decode(error.to_string()))?
686        .collect::<Result<Vec<_>, _>>()
687        .map_err(|error| ClientError::Decode(error.to_string()))
688}
689
690fn context(
691    timeout: Duration,
692    idempotency_key: Option<&str>,
693) -> ClientResult<native::RequestContext> {
694    let request_id = QueryId::random()
695        .map_err(|error| ClientError::Transport(error.to_string()))?
696        .to_string();
697    Ok(native::RequestContext {
698        version: Some(native::ApiVersion {
699            major: NATIVE_API_MAJOR,
700            minor: NATIVE_API_MINOR,
701        }),
702        request_id,
703        deadline_unix_micros: now_unix_micros()
704            .saturating_add(timeout.as_micros().min(u128::from(u64::MAX)) as u64),
705        idempotency_key: idempotency_key.unwrap_or_default().into(),
706    })
707}
708
709fn random_query_id() -> ClientResult<Vec<u8>> {
710    QueryId::random()
711        .map(|id| id.as_bytes().to_vec())
712        .map_err(|error| ClientError::Transport(error.to_string()))
713}
714
715fn now_unix_micros() -> u64 {
716    SystemTime::now()
717        .duration_since(UNIX_EPOCH)
718        .unwrap_or_default()
719        .as_micros()
720        .min(u128::from(u64::MAX)) as u64
721}
722
723fn read_only_sql(sql: &str) -> bool {
724    matches!(
725        mongreldb_query::classify_sql_idempotency(sql),
726        mongreldb_query::SqlIdempotencyClass::ReadOnly
727    )
728}
729
730fn native_retryable(status: &tonic::Status) -> bool {
731    if matches!(status.code(), tonic::Code::Unavailable) {
732        return true;
733    }
734    native_detail(status).is_some_and(|detail| detail.retryable)
735}
736
737fn native_error(status: tonic::Status) -> ClientError {
738    match native_detail(&status) {
739        Some(detail) => ClientError::Native {
740            code: format!("{:?}", status.code()),
741            category_code: Some(detail.category_code),
742            category: detail.category,
743            message: detail.message,
744            retryable: detail.retryable,
745        },
746        None => ClientError::Native {
747            code: format!("{:?}", status.code()),
748            category_code: None,
749            category: "transport".into(),
750            message: status.message().into(),
751            retryable: status.code() == tonic::Code::Unavailable,
752        },
753    }
754}
755
756fn native_detail(status: &tonic::Status) -> Option<native::ErrorDetail> {
757    (!status.details().is_empty())
758        .then(|| native::ErrorDetail::decode(status.details()).ok())
759        .flatten()
760}
mongreldb_client/native.rs

mongreldb_client/
native.rs
