1use std::{
2 collections::HashMap,
3 net::SocketAddr,
4 sync::{
5 Arc, Mutex,
6 atomic::{AtomicU64, Ordering},
7 },
8 time::Duration,
9};
10
11use datum_net::quic::quinn;
12use prost::Message as ProstMessage;
13use tokio::{
14 io::{AsyncRead, AsyncWrite, AsyncWriteExt},
15 net::TcpStream,
16 sync::{mpsc, oneshot},
17 task::JoinHandle,
18};
19
20use crate::dcp::{
21 DcpError, DcpResult,
22 frame::{read_frame, write_frame},
23 proto::{
24 ClientKind, ClusterJobList, ClusterNodeInfo, ClusterNodeList, CompleteShardingAsk,
25 ConfigValue, DcpFrame, Event, ForwardShardEnvelopes, GetConfig, GetShardAllocations, Hello,
26 JobStatus as WireJobStatus, JobStatusRequest, ListClusterJobs, ListJobs, MetricFrame,
27 MetricSample, OpenShardPipe, PlacementSpec, PutConfig, RememberClusterAssignment,
28 RememberShardAllocations, Request, Response, ResponseStatus, ShardAllocation,
29 ShardAllocationRequest, ShardAllocationTable, ShardEnvelopeBatchResult, ShardPipeFrame,
30 StartJob, StopJob, SubmitClusterJob, SubscribeEvents, SubscribeMetrics, dcp_frame, request,
31 },
32 server::connect_quic_stream,
33};
34
35type PendingMap = HashMap<u64, oneshot::Sender<DcpResult<Response>>>;
36
37#[derive(Clone)]
38pub struct DcpClient {
39 inner: Arc<ClientInner>,
40 _reader_task: Arc<JoinHandle<()>>,
41 _writer_task: Arc<JoinHandle<()>>,
42 _quic_endpoint: Option<quinn::Endpoint>,
43 _quic_connection: Option<quinn::Connection>,
44}
45
46struct ClientInner {
47 outbound: mpsc::Sender<DcpFrame>,
48 pending: Mutex<PendingMap>,
49 event_subscriptions: Mutex<HashMap<u64, mpsc::Sender<Event>>>,
50 metric_subscriptions: Mutex<HashMap<u64, mpsc::Sender<MetricSample>>>,
51 next_request_id: AtomicU64,
52}
53
54pub struct EventSubscription {
55 receiver: mpsc::Receiver<Event>,
56}
57
58impl EventSubscription {
59 pub async fn recv(&mut self) -> Option<Event> {
60 self.receiver.recv().await
61 }
62}
63
64pub struct MetricSubscription {
65 receiver: mpsc::Receiver<MetricSample>,
66}
67
68impl MetricSubscription {
69 pub async fn recv(&mut self) -> Option<MetricSample> {
70 self.receiver.recv().await
71 }
72}
73
74#[derive(Clone)]
75pub struct ShardPipeClient {
76 outbound: mpsc::Sender<ShardPipeFrame>,
77 _reader_task: Arc<JoinHandle<()>>,
78 _writer_task: Arc<JoinHandle<()>>,
79 _quic_endpoint: Option<quinn::Endpoint>,
80 _quic_connection: Option<quinn::Connection>,
81}
82
83impl ShardPipeClient {
84 pub async fn connect_tcp(addr: SocketAddr, hello: Hello) -> DcpResult<Self> {
85 let mut stream = TcpStream::connect(addr).await?;
86 stream.set_nodelay(true)?;
87 {
88 let (mut reader, mut writer) = stream.split();
89 open_shard_pipe(&mut reader, &mut writer, hello).await?;
90 }
91 let (reader, writer) = stream.into_split();
92 Ok(Self::from_parts(reader, writer, None, None))
93 }
94
95 pub async fn connect_quic(
96 addr: SocketAddr,
97 server_name: &str,
98 client_config: quinn::ClientConfig,
99 hello: Hello,
100 ) -> DcpResult<Self> {
101 let (endpoint, connection, mut reader, mut writer) =
102 connect_quic_stream(addr, server_name, client_config).await?;
103 open_shard_pipe(&mut reader, &mut writer, hello).await?;
104 Ok(Self::from_parts(
105 reader,
106 writer,
107 Some(endpoint),
108 Some(connection),
109 ))
110 }
111
112 fn from_parts<R, W>(
113 reader: R,
114 writer: W,
115 quic_endpoint: Option<quinn::Endpoint>,
116 quic_connection: Option<quinn::Connection>,
117 ) -> Self
118 where
119 R: AsyncRead + Unpin + Send + 'static,
120 W: AsyncWrite + Unpin + Send + 'static,
121 {
122 let (outbound, outbound_receiver) = mpsc::channel(256);
123 let reader_task = tokio::spawn(async move {
124 read_shard_pipe_loop(reader).await;
125 });
126 let writer_task = tokio::spawn(async move {
127 let _ = write_shard_pipe_loop(writer, outbound_receiver).await;
128 });
129 Self {
130 outbound,
131 _reader_task: Arc::new(reader_task),
132 _writer_task: Arc::new(writer_task),
133 _quic_endpoint: quic_endpoint,
134 _quic_connection: quic_connection,
135 }
136 }
137
138 pub async fn send(&self, frame: ShardPipeFrame) -> DcpResult<()> {
139 self.outbound
140 .send(frame)
141 .await
142 .map_err(|_| DcpError::Closed)
143 }
144}
145
146impl DcpClient {
147 pub async fn connect_tcp(addr: SocketAddr, hello: Hello) -> DcpResult<Self> {
148 let stream = TcpStream::connect(addr).await?;
149 stream.set_nodelay(true)?;
150 let (reader, writer) = stream.into_split();
151 let client = Self::from_parts(reader, writer, None, None);
152 client.hello(hello).await?;
153 Ok(client)
154 }
155
156 pub async fn connect_quic(
157 addr: SocketAddr,
158 server_name: &str,
159 client_config: quinn::ClientConfig,
160 hello: Hello,
161 ) -> DcpResult<Self> {
162 let (endpoint, connection, reader, writer) =
163 connect_quic_stream(addr, server_name, client_config).await?;
164 let client = Self::from_parts(reader, writer, Some(endpoint), Some(connection));
165 client.hello(hello).await?;
166 Ok(client)
167 }
168
169 fn from_parts<R, W>(
170 reader: R,
171 writer: W,
172 quic_endpoint: Option<quinn::Endpoint>,
173 quic_connection: Option<quinn::Connection>,
174 ) -> Self
175 where
176 R: AsyncRead + Unpin + Send + 'static,
177 W: AsyncWrite + Unpin + Send + 'static,
178 {
179 let (outbound, outbound_receiver) = mpsc::channel(256);
180 let inner = Arc::new(ClientInner {
181 outbound,
182 pending: Mutex::new(HashMap::new()),
183 event_subscriptions: Mutex::new(HashMap::new()),
184 metric_subscriptions: Mutex::new(HashMap::new()),
185 next_request_id: AtomicU64::new(1),
186 });
187 let reader_task = {
188 let inner = Arc::clone(&inner);
189 tokio::spawn(async move {
190 read_loop(reader, inner).await;
191 })
192 };
193 let writer_task = tokio::spawn(async move {
194 let _ = write_loop(writer, outbound_receiver).await;
195 });
196 Self {
197 inner,
198 _reader_task: Arc::new(reader_task),
199 _writer_task: Arc::new(writer_task),
200 _quic_endpoint: quic_endpoint,
201 _quic_connection: quic_connection,
202 }
203 }
204
205 pub async fn hello(&self, hello: Hello) -> DcpResult<()> {
206 let response = self
207 .send_frame_with_response(0, DcpFrame::hello(hello))
208 .await?;
209 ensure_ok(response).map(|_| ())
210 }
211
212 pub async fn request(&self, request: Request) -> DcpResult<Response> {
213 let response = self
214 .send_frame_with_response(request.request_id, DcpFrame::request(request))
215 .await?;
216 ensure_ok(response)
217 }
218
219 pub async fn list_jobs(&self) -> DcpResult<Vec<WireJobStatus>> {
220 let response = self
221 .send_command(request::Command::ListJobs(ListJobs {}), 0)
222 .await?;
223 let list = crate::dcp::proto::JobList::decode(response.payload.as_slice())?;
224 Ok(list.jobs)
225 }
226
227 pub async fn list_cluster_jobs(&self, timeout_ms: u64) -> DcpResult<ClusterJobList> {
228 let request = self.send_command(
229 request::Command::ListClusterJobs(ListClusterJobs { timeout_ms }),
230 cluster_request_deadline(timeout_ms),
231 );
232 let response = timeout_cluster_request(timeout_ms, request).await?;
233 Ok(ClusterJobList::decode(response.payload.as_slice())?)
234 }
235
236 pub async fn cluster_node_info(&self, timeout_ms: u64) -> DcpResult<ClusterNodeList> {
237 let request = self.send_command(
238 request::Command::ClusterNodeInfo(ClusterNodeInfo { timeout_ms }),
239 cluster_request_deadline(timeout_ms),
240 );
241 let response = timeout_cluster_request(timeout_ms, request).await?;
242 Ok(ClusterNodeList::decode(response.payload.as_slice())?)
243 }
244
245 pub async fn start_job(
246 &self,
247 factory_name: impl Into<String>,
248 instance_name: impl Into<String>,
249 params: HashMap<String, String>,
250 ) -> DcpResult<WireJobStatus> {
251 self.job_status_response(request::Command::StartJob(StartJob {
252 factory_name: factory_name.into(),
253 instance_name: instance_name.into(),
254 params,
255 cluster: None,
256 }))
257 .await
258 }
259
260 pub(crate) async fn start_cluster_job_on_node(
261 &self,
262 factory_name: impl Into<String>,
263 instance_name: impl Into<String>,
264 params: HashMap<String, String>,
265 cluster: crate::dcp::proto::ClusterJobStart,
266 ) -> DcpResult<WireJobStatus> {
267 self.job_status_response(request::Command::StartJob(StartJob {
268 factory_name: factory_name.into(),
269 instance_name: instance_name.into(),
270 params,
271 cluster: Some(cluster),
272 }))
273 .await
274 }
275
276 pub async fn submit_cluster_job(
277 &self,
278 factory_name: impl Into<String>,
279 instance_name: impl Into<String>,
280 params: HashMap<String, String>,
281 placement: PlacementSpec,
282 timeout_ms: u64,
283 ) -> DcpResult<WireJobStatus> {
284 let request = self.send_command(
285 request::Command::SubmitClusterJob(SubmitClusterJob {
286 factory_name: factory_name.into(),
287 instance_name: instance_name.into(),
288 params,
289 placement: Some(placement),
290 timeout_ms,
291 }),
292 cluster_request_deadline(timeout_ms),
293 );
294 let response = timeout_cluster_request(timeout_ms, request).await?;
295 Ok(WireJobStatus::decode(response.payload.as_slice())?)
296 }
297
298 pub async fn drain_job(&self, name: impl Into<String>) -> DcpResult<WireJobStatus> {
299 self.job_status_response(request::Command::DrainJob(crate::dcp::proto::DrainJob {
300 name: name.into(),
301 cluster: false,
302 }))
303 .await
304 }
305
306 pub async fn drain_cluster_job(
307 &self,
308 name: impl Into<String>,
309 timeout_ms: u64,
310 ) -> DcpResult<WireJobStatus> {
311 let request = self.send_command(
312 request::Command::DrainJob(crate::dcp::proto::DrainJob {
313 name: name.into(),
314 cluster: true,
315 }),
316 cluster_request_deadline(timeout_ms),
317 );
318 let response = timeout_cluster_request(timeout_ms, request).await?;
319 Ok(WireJobStatus::decode(response.payload.as_slice())?)
320 }
321
322 pub async fn stop_job(&self, name: impl Into<String>) -> DcpResult<WireJobStatus> {
323 self.job_status_response(request::Command::StopJob(StopJob {
324 name: name.into(),
325 cluster: false,
326 }))
327 .await
328 }
329
330 pub async fn stop_cluster_job(
331 &self,
332 name: impl Into<String>,
333 timeout_ms: u64,
334 ) -> DcpResult<WireJobStatus> {
335 let request = self.send_command(
336 request::Command::StopJob(StopJob {
337 name: name.into(),
338 cluster: true,
339 }),
340 cluster_request_deadline(timeout_ms),
341 );
342 let response = timeout_cluster_request(timeout_ms, request).await?;
343 Ok(WireJobStatus::decode(response.payload.as_slice())?)
344 }
345
346 pub async fn restart_job(&self, name: impl Into<String>) -> DcpResult<WireJobStatus> {
347 self.job_status_response(request::Command::RestartJob(
348 crate::dcp::proto::RestartJob {
349 name: name.into(),
350 cluster: false,
351 },
352 ))
353 .await
354 }
355
356 pub async fn job_status(&self, name: impl Into<String>) -> DcpResult<WireJobStatus> {
357 self.job_status_response(request::Command::JobStatus(JobStatusRequest {
358 name: name.into(),
359 cluster: false,
360 }))
361 .await
362 }
363
364 pub async fn cluster_job_status(
365 &self,
366 name: impl Into<String>,
367 timeout_ms: u64,
368 ) -> DcpResult<WireJobStatus> {
369 let request = self.send_command(
370 request::Command::JobStatus(JobStatusRequest {
371 name: name.into(),
372 cluster: true,
373 }),
374 cluster_request_deadline(timeout_ms),
375 );
376 let response = timeout_cluster_request(timeout_ms, request).await?;
377 Ok(WireJobStatus::decode(response.payload.as_slice())?)
378 }
379
380 pub(crate) async fn remember_cluster_assignment(
381 &self,
382 instance_name: impl Into<String>,
383 assignment: crate::dcp::proto::ClusterJobStart,
384 ) -> DcpResult<()> {
385 let response = self
386 .send_command(
387 request::Command::RememberClusterAssignment(RememberClusterAssignment {
388 instance_name: instance_name.into(),
389 assignment: Some(assignment),
390 }),
391 0,
392 )
393 .await?;
394 let _ = response;
395 Ok(())
396 }
397
398 pub async fn allocate_shard(
399 &self,
400 type_name: impl Into<String>,
401 shard_id: impl Into<String>,
402 timeout_ms: u64,
403 ) -> DcpResult<ShardAllocation> {
404 let request = self.send_command(
405 request::Command::AllocateShard(ShardAllocationRequest {
406 type_name: type_name.into(),
407 shard_id: shard_id.into(),
408 timeout_ms,
409 }),
410 cluster_request_deadline(timeout_ms),
411 );
412 let response = timeout_cluster_request(timeout_ms, request).await?;
413 Ok(ShardAllocation::decode(response.payload.as_slice())?)
414 }
415
416 pub async fn remember_shard_allocations(&self, table: ShardAllocationTable) -> DcpResult<()> {
417 let response = self
418 .send_command(
419 request::Command::RememberShardAllocations(RememberShardAllocations {
420 table: Some(table),
421 }),
422 0,
423 )
424 .await?;
425 let _ = response;
426 Ok(())
427 }
428
429 pub async fn get_shard_allocations(
430 &self,
431 type_name: impl Into<String>,
432 timeout_ms: u64,
433 ) -> DcpResult<ShardAllocationTable> {
434 let request = self.send_command(
435 request::Command::GetShardAllocations(GetShardAllocations {
436 type_name: type_name.into(),
437 }),
438 cluster_request_deadline(timeout_ms),
439 );
440 let response = timeout_cluster_request(timeout_ms, request).await?;
441 Ok(ShardAllocationTable::decode(response.payload.as_slice())?)
442 }
443
444 pub async fn forward_shard_envelopes(
445 &self,
446 batch: ForwardShardEnvelopes,
447 timeout_ms: u64,
448 ) -> DcpResult<ShardEnvelopeBatchResult> {
449 let request = self.send_command(
450 request::Command::ForwardShardEnvelopes(batch),
451 cluster_request_deadline(timeout_ms),
452 );
453 let response = timeout_cluster_request(timeout_ms, request).await?;
454 Ok(ShardEnvelopeBatchResult::decode(
455 response.payload.as_slice(),
456 )?)
457 }
458
459 pub async fn complete_sharding_ask(&self, reply: CompleteShardingAsk) -> DcpResult<()> {
460 let response = self
461 .send_command(request::Command::CompleteShardingAsk(reply), 0)
462 .await?;
463 let _ = response;
464 Ok(())
465 }
466
467 pub async fn subscribe_events(&self) -> DcpResult<EventSubscription> {
468 let request_id = self.next_request_id();
469 let (sender, receiver) = mpsc::channel(256);
470 self.inner
471 .event_subscriptions
472 .lock()
473 .expect("DCP event subscriptions poisoned")
474 .insert(request_id, sender);
475 let request = Request {
476 request_id,
477 deadline_ms: 0,
478 command: Some(request::Command::SubscribeEvents(SubscribeEvents {
479 buffer: 0,
480 })),
481 };
482 if let Err(error) = self.request(request).await {
483 self.inner
484 .event_subscriptions
485 .lock()
486 .expect("DCP event subscriptions poisoned")
487 .remove(&request_id);
488 return Err(error);
489 }
490 Ok(EventSubscription { receiver })
491 }
492
493 pub async fn subscribe_metrics(&self, interval_ms: u64) -> DcpResult<MetricSubscription> {
494 let request_id = self.next_request_id();
495 let (sender, receiver) = mpsc::channel(16);
496 self.inner
497 .metric_subscriptions
498 .lock()
499 .expect("DCP metric subscriptions poisoned")
500 .insert(request_id, sender);
501 let request = Request {
502 request_id,
503 deadline_ms: 0,
504 command: Some(request::Command::SubscribeMetrics(SubscribeMetrics {
505 interval_ms,
506 })),
507 };
508 if let Err(error) = self.request(request).await {
509 self.inner
510 .metric_subscriptions
511 .lock()
512 .expect("DCP metric subscriptions poisoned")
513 .remove(&request_id);
514 return Err(error);
515 }
516 Ok(MetricSubscription { receiver })
517 }
518
519 pub async fn get_config(&self, key: impl Into<String>) -> DcpResult<ConfigValue> {
520 let response = self
521 .send_command(
522 request::Command::GetConfig(GetConfig { key: key.into() }),
523 0,
524 )
525 .await?;
526 Ok(ConfigValue::decode(response.payload.as_slice())?)
527 }
528
529 pub async fn put_config(
530 &self,
531 key: impl Into<String>,
532 value: impl Into<String>,
533 ) -> DcpResult<ConfigValue> {
534 let response = self
535 .send_command(
536 request::Command::PutConfig(PutConfig {
537 key: key.into(),
538 value: value.into(),
539 }),
540 0,
541 )
542 .await?;
543 Ok(ConfigValue::decode(response.payload.as_slice())?)
544 }
545
546 async fn job_status_response(&self, command: request::Command) -> DcpResult<WireJobStatus> {
547 let response = self.send_command(command, 0).await?;
548 Ok(WireJobStatus::decode(response.payload.as_slice())?)
549 }
550
551 async fn send_command(
552 &self,
553 command: request::Command,
554 deadline_ms: u64,
555 ) -> DcpResult<Response> {
556 let request_id = self.next_request_id();
557 let request = Request {
558 request_id,
559 deadline_ms,
560 command: Some(command),
561 };
562 self.request(request).await
563 }
564
565 async fn send_frame_with_response(
566 &self,
567 request_id: u64,
568 frame: DcpFrame,
569 ) -> DcpResult<Response> {
570 let (sender, receiver) = oneshot::channel();
571 self.inner
572 .pending
573 .lock()
574 .expect("DCP pending map poisoned")
575 .insert(request_id, sender);
576 if self.inner.outbound.send(frame).await.is_err() {
577 self.inner
578 .pending
579 .lock()
580 .expect("DCP pending map poisoned")
581 .remove(&request_id);
582 return Err(DcpError::Closed);
583 }
584 receiver.await.map_err(|_| DcpError::Closed)?
585 }
586
587 fn next_request_id(&self) -> u64 {
588 self.inner.next_request_id.fetch_add(1, Ordering::Relaxed)
589 }
590}
591
592async fn read_loop<R>(mut reader: R, inner: Arc<ClientInner>)
593where
594 R: AsyncRead + Unpin,
595{
596 loop {
597 match read_frame(&mut reader).await {
598 Ok(Some(frame)) => handle_inbound_frame(frame, &inner).await,
599 Ok(None) => {
600 fail_pending(&inner, DcpError::Closed);
601 break;
602 }
603 Err(error) => {
604 fail_pending(&inner, error);
605 break;
606 }
607 }
608 }
609}
610
611async fn handle_inbound_frame(frame: DcpFrame, inner: &Arc<ClientInner>) {
612 match frame.frame {
613 Some(dcp_frame::Frame::Response(response)) => {
614 if let Some(sender) = inner
615 .pending
616 .lock()
617 .expect("DCP pending map poisoned")
618 .remove(&response.request_id)
619 {
620 let _ = sender.send(Ok(response));
621 }
622 }
623 Some(dcp_frame::Frame::Event(event_frame)) => {
624 if let Some(event) = event_frame.event {
625 let sender = inner
626 .event_subscriptions
627 .lock()
628 .expect("DCP event subscriptions poisoned")
629 .get(&event_frame.subscription_id)
630 .cloned();
631 if let Some(sender) = sender {
632 let _ = sender.send(event).await;
633 }
634 }
635 }
636 Some(dcp_frame::Frame::Metric(MetricFrame {
637 subscription_id,
638 sample: Some(sample),
639 })) => {
640 let sender = inner
641 .metric_subscriptions
642 .lock()
643 .expect("DCP metric subscriptions poisoned")
644 .get(&subscription_id)
645 .cloned();
646 if let Some(sender) = sender {
647 let _ = sender.send(sample).await;
648 }
649 }
650 _ => {}
651 }
652}
653
654fn fail_pending(inner: &ClientInner, error: DcpError) {
655 let pending = std::mem::take(&mut *inner.pending.lock().expect("DCP pending map poisoned"));
656 let message = error.to_string();
657 for (_, sender) in pending {
658 let _ = sender.send(Err(DcpError::Protocol(message.clone())));
659 }
660}
661
662async fn write_loop<W>(mut writer: W, mut outbound: mpsc::Receiver<DcpFrame>) -> DcpResult<()>
663where
664 W: AsyncWrite + Unpin,
665{
666 while let Some(frame) = outbound.recv().await {
667 write_frame(&mut writer, &frame).await?;
668 }
669 let _ = writer.shutdown().await;
670 Ok(())
671}
672
673async fn open_shard_pipe<R, W>(reader: &mut R, writer: &mut W, hello: Hello) -> DcpResult<()>
674where
675 R: AsyncRead + Unpin,
676 W: AsyncWrite + Unpin,
677{
678 write_frame(writer, &DcpFrame::hello(hello)).await?;
679 let Some(response) = read_frame(reader).await? else {
680 return Err(DcpError::Closed);
681 };
682 match response.frame {
683 Some(dcp_frame::Frame::Response(response)) => {
684 ensure_ok(response)?;
685 }
686 _ => {
687 return Err(DcpError::Protocol(
688 "DCP shard pipe hello did not receive a response".to_owned(),
689 ));
690 }
691 }
692
693 let request = Request {
694 request_id: 1,
695 deadline_ms: 0,
696 command: Some(request::Command::OpenShardPipe(OpenShardPipe {})),
697 };
698 write_frame(writer, &DcpFrame::request(request)).await?;
699 let Some(response) = read_frame(reader).await? else {
700 return Err(DcpError::Closed);
701 };
702 match response.frame {
703 Some(dcp_frame::Frame::Response(response)) => ensure_ok(response).map(|_| ()),
704 _ => Err(DcpError::Protocol(
705 "DCP shard pipe open did not receive a response".to_owned(),
706 )),
707 }
708}
709
710async fn read_shard_pipe_loop<R>(mut reader: R)
711where
712 R: AsyncRead + Unpin,
713{
714 while let Ok(Some(_frame)) = read_frame(&mut reader).await {}
715}
716
717async fn write_shard_pipe_loop<W>(
718 mut writer: W,
719 mut outbound: mpsc::Receiver<ShardPipeFrame>,
720) -> DcpResult<()>
721where
722 W: AsyncWrite + Unpin,
723{
724 while let Some(frame) = outbound.recv().await {
725 write_frame(&mut writer, &DcpFrame::shard_pipe(frame)).await?;
726 }
727 let _ = writer.shutdown().await;
728 Ok(())
729}
730
731fn ensure_ok(response: Response) -> DcpResult<Response> {
732 let status = response.response_status();
733 if status == ResponseStatus::Ok {
734 Ok(response)
735 } else {
736 Err(DcpError::response(status, response.message))
737 }
738}
739
740async fn timeout_cluster_request<F>(timeout_ms: u64, request: F) -> DcpResult<Response>
741where
742 F: std::future::Future<Output = DcpResult<Response>>,
743{
744 if timeout_ms == 0 {
745 return request.await;
746 }
747 tokio::time::timeout(
748 Duration::from_millis(timeout_ms).saturating_add(Duration::from_millis(250)),
749 request,
750 )
751 .await
752 .map_err(|_| {
753 DcpError::response(
754 ResponseStatus::DeadlineExceeded,
755 "cluster request timed out",
756 )
757 })?
758}
759
760fn cluster_request_deadline(timeout_ms: u64) -> u64 {
761 if timeout_ms == 0 {
762 0
763 } else {
764 timeout_ms.saturating_add(250)
765 }
766}
767
768impl Default for DcpClient {
769 fn default() -> Self {
770 panic!("DcpClient cannot be default-constructed; use connect_tcp or connect_quic")
771 }
772}
773
774#[must_use]
775pub fn default_hello(node_id: impl Into<String>, client_kind: ClientKind) -> Hello {
776 Hello::new(node_id, client_kind)
777}