1#![doc(
4 html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
5 html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
6 issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
7)]
8#![cfg_attr(not(test), warn(unused_crate_dependencies))]
9#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
10
11use crate::metrics::PayloadBuilderMetrics;
12use alloy_eips::merge::SLOT_DURATION;
13use alloy_primitives::{B256, U256};
14use futures_core::ready;
15use futures_util::FutureExt;
16use reth_payload_builder::{KeepPayloadJobAlive, PayloadId, PayloadJob, PayloadJobGenerator};
17use reth_payload_builder_primitives::PayloadBuilderError;
18use reth_payload_primitives::{BuiltPayload, PayloadBuilderAttributes, PayloadKind};
19use reth_primitives::{NodePrimitives, SealedHeader};
20use reth_primitives_traits::HeaderTy;
21use reth_provider::{BlockReaderIdExt, CanonStateNotification, StateProviderFactory};
22use reth_revm::{cached::CachedReads, cancelled::CancelOnDrop};
23use reth_tasks::TaskSpawner;
24use std::{
25 fmt,
26 future::Future,
27 ops::Deref,
28 pin::Pin,
29 sync::Arc,
30 task::{Context, Poll},
31 time::{Duration, SystemTime, UNIX_EPOCH},
32};
33use tokio::{
34 sync::{oneshot, Semaphore},
35 time::{Interval, Sleep},
36};
37use tracing::{debug, trace, warn};
38
39mod better_payload_emitter;
40mod metrics;
41mod stack;
42
43pub use better_payload_emitter::BetterPayloadEmitter;
44pub use stack::PayloadBuilderStack;
45
46pub type HeaderForPayload<P> = <<P as BuiltPayload>::Primitives as NodePrimitives>::BlockHeader;
48
49#[derive(Debug)]
51pub struct BasicPayloadJobGenerator<Client, Tasks, Builder> {
52 client: Client,
54 executor: Tasks,
56 config: BasicPayloadJobGeneratorConfig,
58 payload_task_guard: PayloadTaskGuard,
60 builder: Builder,
64 pre_cached: Option<PrecachedState>,
66}
67
68impl<Client, Tasks, Builder> BasicPayloadJobGenerator<Client, Tasks, Builder> {
71 pub fn with_builder(
74 client: Client,
75 executor: Tasks,
76 config: BasicPayloadJobGeneratorConfig,
77 builder: Builder,
78 ) -> Self {
79 Self {
80 client,
81 executor,
82 payload_task_guard: PayloadTaskGuard::new(config.max_payload_tasks),
83 config,
84 builder,
85 pre_cached: None,
86 }
87 }
88
89 #[inline]
98 fn max_job_duration(&self, unix_timestamp: u64) -> Duration {
99 let duration_until_timestamp = duration_until(unix_timestamp);
100
101 let duration_until_timestamp = duration_until_timestamp.min(self.config.deadline * 3);
103
104 self.config.deadline + duration_until_timestamp
105 }
106
107 #[inline]
110 fn job_deadline(&self, unix_timestamp: u64) -> tokio::time::Instant {
111 tokio::time::Instant::now() + self.max_job_duration(unix_timestamp)
112 }
113
114 pub const fn tasks(&self) -> &Tasks {
116 &self.executor
117 }
118
119 fn maybe_pre_cached(&self, parent: B256) -> Option<CachedReads> {
122 self.pre_cached.as_ref().filter(|pc| pc.block == parent).map(|pc| pc.cached.clone())
123 }
124}
125
126impl<Client, Tasks, Builder> PayloadJobGenerator
129 for BasicPayloadJobGenerator<Client, Tasks, Builder>
130where
131 Client: StateProviderFactory
132 + BlockReaderIdExt<Header = HeaderForPayload<Builder::BuiltPayload>>
133 + Clone
134 + Unpin
135 + 'static,
136 Tasks: TaskSpawner + Clone + Unpin + 'static,
137 Builder: PayloadBuilder + Unpin + 'static,
138 Builder::Attributes: Unpin + Clone,
139 Builder::BuiltPayload: Unpin + Clone,
140{
141 type Job = BasicPayloadJob<Tasks, Builder>;
142
143 fn new_payload_job(
144 &self,
145 attributes: <Self::Job as PayloadJob>::PayloadAttributes,
146 ) -> Result<Self::Job, PayloadBuilderError> {
147 let parent_header = if attributes.parent().is_zero() {
148 self.client
150 .latest_header()
151 .map_err(PayloadBuilderError::from)?
152 .ok_or_else(|| PayloadBuilderError::MissingParentHeader(B256::ZERO))?
153 } else {
154 self.client
156 .sealed_header_by_hash(attributes.parent())
157 .map_err(PayloadBuilderError::from)?
158 .ok_or_else(|| PayloadBuilderError::MissingParentHeader(attributes.parent()))?
159 };
160
161 let config = PayloadConfig::new(Arc::new(parent_header.clone()), attributes);
162
163 let until = self.job_deadline(config.attributes.timestamp());
164 let deadline = Box::pin(tokio::time::sleep_until(until));
165
166 let cached_reads = self.maybe_pre_cached(parent_header.hash());
167
168 let mut job = BasicPayloadJob {
169 config,
170 executor: self.executor.clone(),
171 deadline,
172 interval: tokio::time::interval(self.config.interval),
174 best_payload: PayloadState::Missing,
175 pending_block: None,
176 cached_reads,
177 payload_task_guard: self.payload_task_guard.clone(),
178 metrics: Default::default(),
179 builder: self.builder.clone(),
180 };
181
182 job.spawn_build_job();
184
185 Ok(job)
186 }
187
188 fn on_new_state<N: NodePrimitives>(&mut self, new_state: CanonStateNotification<N>) {
189 let mut cached = CachedReads::default();
190
191 let committed = new_state.committed();
193 let new_execution_outcome = committed.execution_outcome();
194 for (addr, acc) in new_execution_outcome.bundle_accounts_iter() {
195 if let Some(info) = acc.info.clone() {
196 let storage =
199 acc.storage.iter().map(|(key, slot)| (*key, slot.present_value)).collect();
200 cached.insert_account(addr, info, storage);
201 }
202 }
203
204 self.pre_cached = Some(PrecachedState { block: committed.tip().hash(), cached });
205 }
206}
207
208#[derive(Debug, Clone)]
212pub struct PrecachedState {
213 pub block: B256,
215 pub cached: CachedReads,
217}
218
219#[derive(Debug, Clone)]
221pub struct PayloadTaskGuard(Arc<Semaphore>);
222
223impl Deref for PayloadTaskGuard {
224 type Target = Semaphore;
225
226 fn deref(&self) -> &Self::Target {
227 &self.0
228 }
229}
230
231impl PayloadTaskGuard {
234 pub fn new(max_payload_tasks: usize) -> Self {
236 Self(Arc::new(Semaphore::new(max_payload_tasks)))
237 }
238}
239
240#[derive(Debug, Clone)]
242pub struct BasicPayloadJobGeneratorConfig {
243 interval: Duration,
245 deadline: Duration,
249 max_payload_tasks: usize,
251}
252
253impl BasicPayloadJobGeneratorConfig {
256 pub const fn interval(mut self, interval: Duration) -> Self {
258 self.interval = interval;
259 self
260 }
261
262 pub const fn deadline(mut self, deadline: Duration) -> Self {
264 self.deadline = deadline;
265 self
266 }
267
268 pub fn max_payload_tasks(mut self, max_payload_tasks: usize) -> Self {
274 assert!(max_payload_tasks > 0, "max_payload_tasks must be greater than 0");
275 self.max_payload_tasks = max_payload_tasks;
276 self
277 }
278}
279
280impl Default for BasicPayloadJobGeneratorConfig {
281 fn default() -> Self {
282 Self {
283 interval: Duration::from_secs(1),
284 deadline: SLOT_DURATION,
286 max_payload_tasks: 3,
287 }
288 }
289}
290
291#[derive(Debug)]
302pub struct BasicPayloadJob<Tasks, Builder>
303where
304 Builder: PayloadBuilder,
305{
306 config: PayloadConfig<Builder::Attributes, HeaderForPayload<Builder::BuiltPayload>>,
308 executor: Tasks,
310 deadline: Pin<Box<Sleep>>,
312 interval: Interval,
314 best_payload: PayloadState<Builder::BuiltPayload>,
316 pending_block: Option<PendingPayload<Builder::BuiltPayload>>,
318 payload_task_guard: PayloadTaskGuard,
320 cached_reads: Option<CachedReads>,
325 metrics: PayloadBuilderMetrics,
327 builder: Builder,
331}
332
333impl<Tasks, Builder> BasicPayloadJob<Tasks, Builder>
334where
335 Tasks: TaskSpawner + Clone + 'static,
336 Builder: PayloadBuilder + Unpin + 'static,
337 Builder::Attributes: Unpin + Clone,
338 Builder::BuiltPayload: Unpin + Clone,
339{
340 fn spawn_build_job(&mut self) {
342 trace!(target: "payload_builder", id = %self.config.payload_id(), "spawn new payload build task");
343 let (tx, rx) = oneshot::channel();
344 let cancel = CancelOnDrop::default();
345 let _cancel = cancel.clone();
346 let guard = self.payload_task_guard.clone();
347 let payload_config = self.config.clone();
348 let best_payload = self.best_payload.payload().cloned();
349 self.metrics.inc_initiated_payload_builds();
350 let cached_reads = self.cached_reads.take().unwrap_or_default();
351 let builder = self.builder.clone();
352 self.executor.spawn_blocking(Box::pin(async move {
353 let _permit = guard.acquire().await;
355 let args =
356 BuildArguments { cached_reads, config: payload_config, cancel, best_payload };
357 let result = builder.try_build(args);
358 let _ = tx.send(result);
359 }));
360
361 self.pending_block = Some(PendingPayload { _cancel, payload: rx });
362 }
363}
364
365impl<Tasks, Builder> Future for BasicPayloadJob<Tasks, Builder>
366where
367 Tasks: TaskSpawner + Clone + 'static,
368 Builder: PayloadBuilder + Unpin + 'static,
369 Builder::Attributes: Unpin + Clone,
370 Builder::BuiltPayload: Unpin + Clone,
371{
372 type Output = Result<(), PayloadBuilderError>;
373
374 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
375 let this = self.get_mut();
376
377 if this.deadline.as_mut().poll(cx).is_ready() {
379 trace!(target: "payload_builder", "payload building deadline reached");
380 return Poll::Ready(Ok(()))
381 }
382
383 while this.interval.poll_tick(cx).is_ready() {
385 if this.pending_block.is_none() && !this.best_payload.is_frozen() {
388 this.spawn_build_job();
389 }
390 }
391
392 if let Some(mut fut) = this.pending_block.take() {
394 match fut.poll_unpin(cx) {
395 Poll::Ready(Ok(outcome)) => match outcome {
396 BuildOutcome::Better { payload, cached_reads } => {
397 this.cached_reads = Some(cached_reads);
398 debug!(target: "payload_builder", value = %payload.fees(), "built better payload");
399 this.best_payload = PayloadState::Best(payload);
400 }
401 BuildOutcome::Freeze(payload) => {
402 debug!(target: "payload_builder", "payload frozen, no further building will occur");
403 this.best_payload = PayloadState::Frozen(payload);
404 }
405 BuildOutcome::Aborted { fees, cached_reads } => {
406 this.cached_reads = Some(cached_reads);
407 trace!(target: "payload_builder", worse_fees = %fees, "skipped payload build of worse block");
408 }
409 BuildOutcome::Cancelled => {
410 unreachable!("the cancel signal never fired")
411 }
412 },
413 Poll::Ready(Err(error)) => {
414 debug!(target: "payload_builder", %error, "payload build attempt failed");
416 this.metrics.inc_failed_payload_builds();
417 }
418 Poll::Pending => {
419 this.pending_block = Some(fut);
420 }
421 }
422 }
423
424 Poll::Pending
425 }
426}
427
428impl<Tasks, Builder> PayloadJob for BasicPayloadJob<Tasks, Builder>
429where
430 Tasks: TaskSpawner + Clone + 'static,
431 Builder: PayloadBuilder + Unpin + 'static,
432 Builder::Attributes: Unpin + Clone,
433 Builder::BuiltPayload: Unpin + Clone,
434{
435 type PayloadAttributes = Builder::Attributes;
436 type ResolvePayloadFuture = ResolveBestPayload<Self::BuiltPayload>;
437 type BuiltPayload = Builder::BuiltPayload;
438
439 fn best_payload(&self) -> Result<Self::BuiltPayload, PayloadBuilderError> {
440 if let Some(payload) = self.best_payload.payload() {
441 Ok(payload.clone())
442 } else {
443 self.metrics.inc_requested_empty_payload();
450 self.builder.build_empty_payload(self.config.clone())
451 }
452 }
453
454 fn payload_attributes(&self) -> Result<Self::PayloadAttributes, PayloadBuilderError> {
455 Ok(self.config.attributes.clone())
456 }
457
458 fn resolve_kind(
459 &mut self,
460 kind: PayloadKind,
461 ) -> (Self::ResolvePayloadFuture, KeepPayloadJobAlive) {
462 let best_payload = self.best_payload.payload().cloned();
463 if best_payload.is_none() && self.pending_block.is_none() {
464 self.spawn_build_job();
466 }
467
468 let maybe_better = self.pending_block.take();
469 let mut empty_payload = None;
470
471 if best_payload.is_none() {
472 debug!(target: "payload_builder", id=%self.config.payload_id(), "no best payload yet to resolve, building empty payload");
473
474 let args = BuildArguments {
475 cached_reads: self.cached_reads.take().unwrap_or_default(),
476 config: self.config.clone(),
477 cancel: CancelOnDrop::default(),
478 best_payload: None,
479 };
480
481 match self.builder.on_missing_payload(args) {
482 MissingPayloadBehaviour::AwaitInProgress => {
483 debug!(target: "payload_builder", id=%self.config.payload_id(), "awaiting in progress payload build job");
484 }
485 MissingPayloadBehaviour::RaceEmptyPayload => {
486 debug!(target: "payload_builder", id=%self.config.payload_id(), "racing empty payload");
487
488 self.metrics.inc_requested_empty_payload();
490 let (tx, rx) = oneshot::channel();
492 let config = self.config.clone();
493 let builder = self.builder.clone();
494 self.executor.spawn_blocking(Box::pin(async move {
495 let res = builder.build_empty_payload(config);
496 let _ = tx.send(res);
497 }));
498
499 empty_payload = Some(rx);
500 }
501 MissingPayloadBehaviour::RacePayload(job) => {
502 debug!(target: "payload_builder", id=%self.config.payload_id(), "racing fallback payload");
503 let (tx, rx) = oneshot::channel();
505 self.executor.spawn_blocking(Box::pin(async move {
506 let _ = tx.send(job());
507 }));
508 empty_payload = Some(rx);
509 }
510 };
511 }
512
513 let fut = ResolveBestPayload {
514 best_payload,
515 maybe_better,
516 empty_payload: empty_payload.filter(|_| kind != PayloadKind::WaitForPending),
517 };
518
519 (fut, KeepPayloadJobAlive::No)
520 }
521}
522
523#[derive(Debug, Clone)]
525pub enum PayloadState<P> {
526 Missing,
528 Best(P),
530 Frozen(P),
534}
535
536impl<P> PayloadState<P> {
537 pub const fn is_frozen(&self) -> bool {
539 matches!(self, Self::Frozen(_))
540 }
541
542 pub const fn payload(&self) -> Option<&P> {
544 match self {
545 Self::Missing => None,
546 Self::Best(p) | Self::Frozen(p) => Some(p),
547 }
548 }
549}
550
551#[derive(Debug)]
561pub struct ResolveBestPayload<Payload> {
562 pub best_payload: Option<Payload>,
564 pub maybe_better: Option<PendingPayload<Payload>>,
566 pub empty_payload: Option<oneshot::Receiver<Result<Payload, PayloadBuilderError>>>,
568}
569
570impl<Payload> ResolveBestPayload<Payload> {
571 const fn is_empty(&self) -> bool {
572 self.best_payload.is_none() && self.maybe_better.is_none() && self.empty_payload.is_none()
573 }
574}
575
576impl<Payload> Future for ResolveBestPayload<Payload>
577where
578 Payload: Unpin,
579{
580 type Output = Result<Payload, PayloadBuilderError>;
581
582 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
583 let this = self.get_mut();
584
585 if let Some(fut) = Pin::new(&mut this.maybe_better).as_pin_mut() {
587 if let Poll::Ready(res) = fut.poll(cx) {
588 this.maybe_better = None;
589 if let Ok(Some(payload)) = res.map(|out| out.into_payload())
590 .inspect_err(|err| warn!(target: "payload_builder", %err, "failed to resolve pending payload"))
591 {
592 debug!(target: "payload_builder", "resolving better payload");
593 return Poll::Ready(Ok(payload))
594 }
595 }
596 }
597
598 if let Some(best) = this.best_payload.take() {
599 debug!(target: "payload_builder", "resolving best payload");
600 return Poll::Ready(Ok(best))
601 }
602
603 if let Some(fut) = Pin::new(&mut this.empty_payload).as_pin_mut() {
604 if let Poll::Ready(res) = fut.poll(cx) {
605 this.empty_payload = None;
606 return match res {
607 Ok(res) => {
608 if let Err(err) = &res {
609 warn!(target: "payload_builder", %err, "failed to resolve empty payload");
610 } else {
611 debug!(target: "payload_builder", "resolving empty payload");
612 }
613 Poll::Ready(res)
614 }
615 Err(err) => Poll::Ready(Err(err.into())),
616 }
617 }
618 }
619
620 if this.is_empty() {
621 return Poll::Ready(Err(PayloadBuilderError::MissingPayload))
622 }
623
624 Poll::Pending
625 }
626}
627
628#[derive(Debug)]
630pub struct PendingPayload<P> {
631 _cancel: CancelOnDrop,
633 payload: oneshot::Receiver<Result<BuildOutcome<P>, PayloadBuilderError>>,
635}
636
637impl<P> PendingPayload<P> {
638 pub const fn new(
640 cancel: CancelOnDrop,
641 payload: oneshot::Receiver<Result<BuildOutcome<P>, PayloadBuilderError>>,
642 ) -> Self {
643 Self { _cancel: cancel, payload }
644 }
645}
646
647impl<P> Future for PendingPayload<P> {
648 type Output = Result<BuildOutcome<P>, PayloadBuilderError>;
649
650 fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
651 let res = ready!(self.payload.poll_unpin(cx));
652 Poll::Ready(res.map_err(Into::into).and_then(|res| res))
653 }
654}
655
656#[derive(Clone, Debug)]
658pub struct PayloadConfig<Attributes, Header = alloy_consensus::Header> {
659 pub parent_header: Arc<SealedHeader<Header>>,
661 pub attributes: Attributes,
663}
664
665impl<Attributes, Header> PayloadConfig<Attributes, Header>
666where
667 Attributes: PayloadBuilderAttributes,
668{
669 pub const fn new(parent_header: Arc<SealedHeader<Header>>, attributes: Attributes) -> Self {
671 Self { parent_header, attributes }
672 }
673
674 pub fn payload_id(&self) -> PayloadId {
676 self.attributes.payload_id()
677 }
678}
679
680#[derive(Debug)]
682pub enum BuildOutcome<Payload> {
683 Better {
685 payload: Payload,
687 cached_reads: CachedReads,
689 },
690 Aborted {
692 fees: U256,
694 cached_reads: CachedReads,
696 },
697 Cancelled,
699
700 Freeze(Payload),
702}
703
704impl<Payload> BuildOutcome<Payload> {
705 pub fn into_payload(self) -> Option<Payload> {
707 match self {
708 Self::Better { payload, .. } | Self::Freeze(payload) => Some(payload),
709 _ => None,
710 }
711 }
712
713 pub const fn is_better(&self) -> bool {
715 matches!(self, Self::Better { .. })
716 }
717
718 pub const fn is_aborted(&self) -> bool {
720 matches!(self, Self::Aborted { .. })
721 }
722
723 pub const fn is_cancelled(&self) -> bool {
725 matches!(self, Self::Cancelled)
726 }
727
728 pub(crate) fn map_payload<F, P>(self, f: F) -> BuildOutcome<P>
730 where
731 F: FnOnce(Payload) -> P,
732 {
733 match self {
734 Self::Better { payload, cached_reads } => {
735 BuildOutcome::Better { payload: f(payload), cached_reads }
736 }
737 Self::Aborted { fees, cached_reads } => BuildOutcome::Aborted { fees, cached_reads },
738 Self::Cancelled => BuildOutcome::Cancelled,
739 Self::Freeze(payload) => BuildOutcome::Freeze(f(payload)),
740 }
741 }
742}
743
744#[derive(Debug)]
746pub enum BuildOutcomeKind<Payload> {
747 Better {
749 payload: Payload,
751 },
752 Aborted {
754 fees: U256,
756 },
757 Cancelled,
759 Freeze(Payload),
761}
762
763impl<Payload> BuildOutcomeKind<Payload> {
764 pub fn with_cached_reads(self, cached_reads: CachedReads) -> BuildOutcome<Payload> {
766 match self {
767 Self::Better { payload } => BuildOutcome::Better { payload, cached_reads },
768 Self::Aborted { fees } => BuildOutcome::Aborted { fees, cached_reads },
769 Self::Cancelled => BuildOutcome::Cancelled,
770 Self::Freeze(payload) => BuildOutcome::Freeze(payload),
771 }
772 }
773}
774
775#[derive(Debug)]
781pub struct BuildArguments<Attributes, Payload: BuiltPayload> {
782 pub cached_reads: CachedReads,
784 pub config: PayloadConfig<Attributes, HeaderTy<Payload::Primitives>>,
786 pub cancel: CancelOnDrop,
788 pub best_payload: Option<Payload>,
790}
791
792impl<Attributes, Payload: BuiltPayload> BuildArguments<Attributes, Payload> {
793 pub const fn new(
795 cached_reads: CachedReads,
796 config: PayloadConfig<Attributes, HeaderTy<Payload::Primitives>>,
797 cancel: CancelOnDrop,
798 best_payload: Option<Payload>,
799 ) -> Self {
800 Self { cached_reads, config, cancel, best_payload }
801 }
802}
803
804pub trait PayloadBuilder: Send + Sync + Clone {
813 type Attributes: PayloadBuilderAttributes;
815 type BuiltPayload: BuiltPayload;
817
818 fn try_build(
831 &self,
832 args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
833 ) -> Result<BuildOutcome<Self::BuiltPayload>, PayloadBuilderError>;
834
835 fn on_missing_payload(
839 &self,
840 _args: BuildArguments<Self::Attributes, Self::BuiltPayload>,
841 ) -> MissingPayloadBehaviour<Self::BuiltPayload> {
842 MissingPayloadBehaviour::RaceEmptyPayload
843 }
844
845 fn build_empty_payload(
847 &self,
848 config: PayloadConfig<Self::Attributes, HeaderForPayload<Self::BuiltPayload>>,
849 ) -> Result<Self::BuiltPayload, PayloadBuilderError>;
850}
851
852pub enum MissingPayloadBehaviour<Payload> {
856 AwaitInProgress,
858 RaceEmptyPayload,
860 RacePayload(Box<dyn FnOnce() -> Result<Payload, PayloadBuilderError> + Send>),
862}
863
864impl<Payload> fmt::Debug for MissingPayloadBehaviour<Payload> {
865 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
866 match self {
867 Self::AwaitInProgress => write!(f, "AwaitInProgress"),
868 Self::RaceEmptyPayload => {
869 write!(f, "RaceEmptyPayload")
870 }
871 Self::RacePayload(_) => write!(f, "RacePayload"),
872 }
873 }
874}
875
876impl<Payload> Default for MissingPayloadBehaviour<Payload> {
877 fn default() -> Self {
878 Self::RaceEmptyPayload
879 }
880}
881
882#[inline(always)]
886pub fn is_better_payload<T: BuiltPayload>(best_payload: Option<&T>, new_fees: U256) -> bool {
887 if let Some(best_payload) = best_payload {
888 new_fees > best_payload.fees()
889 } else {
890 true
891 }
892}
893
894fn duration_until(unix_timestamp_secs: u64) -> Duration {
898 let unix_now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default();
899 let timestamp = Duration::from_secs(unix_timestamp_secs);
900 timestamp.saturating_sub(unix_now)
901}