anytype_rpc/
deadline.rs

1//! Logical gRPC deadline configuration, propagation, and local enforcement.
2
3use std::{
4    error::Error as StdError,
5    ffi::OsString,
6    fmt,
7    future::Future,
8    marker::PhantomData,
9    pin::Pin,
10    sync::{
11        Arc, OnceLock,
12        atomic::{AtomicU64, Ordering},
13    },
14    task::{Context, Poll},
15    time::Duration,
16};
17
18use http_body::{Body as HttpBody, Frame, SizeHint};
19use prost::bytes::Buf;
20use tonic::{
21    Code, GrpcMethod, Request, Status, TimeoutExpired,
22    body::Body,
23    codegen::{Service, http},
24    metadata::{Ascii, MetadataMap, MetadataValue},
25};
26
27tokio::task_local! {
28    static ENCLOSING_DEADLINE: GrpcEnclosingDeadline;
29}
30
31/// Process environment variable that overrides inherited gRPC deadlines.
32pub const ANYTYPE_GRPC_TIMEOUT_SECS: &str = "ANYTYPE_GRPC_TIMEOUT_SECS";
33/// Largest credential, ordinary, setup, idle, or lifetime deadline.
34pub const MAX_GRPC_TIMEOUT: Duration = Duration::from_secs(3_600);
35/// Largest long-operation deadline.
36pub const MAX_LONG_GRPC_TIMEOUT: Duration = Duration::from_secs(7_200);
37/// Largest cleanup deadline.
38pub const MAX_CLEANUP_GRPC_TIMEOUT: Duration = Duration::from_secs(30);
39/// Default credential-session setup deadline.
40pub const DEFAULT_CREDENTIAL_GRPC_TIMEOUT: Duration = Duration::from_secs(120);
41/// Default ordinary unary RPC deadline.
42pub const DEFAULT_ORDINARY_GRPC_TIMEOUT: Duration = Duration::from_secs(120);
43/// Default long unary RPC deadline.
44pub const DEFAULT_LONG_GRPC_TIMEOUT: Duration = Duration::from_secs(1_800);
45/// Default stream response-header deadline.
46pub const DEFAULT_STREAM_SETUP_GRPC_TIMEOUT: Duration = Duration::from_secs(120);
47/// Default cleanup RPC deadline.
48pub const DEFAULT_CLEANUP_GRPC_TIMEOUT: Duration = Duration::from_secs(5);
49
50const DEADLINE_SOURCE_METADATA: &str = "x-anytype-deadline-source";
51const DEADLINE_CLASS_METADATA: &str = "x-anytype-deadline-class";
52const DEADLINE_OUTCOME_METADATA: &str = "x-anytype-deadline-outcome";
53const GRPC_TIMEOUT_HEADER: &str = "grpc-timeout";
54
55/// Logical gRPC deadlines applied by an Anytype gRPC client.
56///
57/// `None` disables a boundary. Credential, ordinary, setup, idle, and stream
58/// lifetime values must be between one and 3,600 seconds. Long unary values
59/// may be as large as 7,200 seconds, while cleanup values may be at most 30
60/// seconds.
61#[derive(Clone, Copy, Debug, PartialEq, Eq)]
62pub struct GrpcTimeoutPolicy {
63    /// Deadline for account-key or app-key session creation.
64    pub credential_setup: Option<Duration>,
65    /// Deadline for ordinary unary reads and mutations.
66    pub ordinary_unary: Option<Duration>,
67    /// Deadline for export, import, and equivalent long unary operations.
68    pub long_unary: Option<Duration>,
69    /// Deadline through successful streaming response headers.
70    pub stream_setup: Option<Duration>,
71    /// Optional no-progress deadline for an established stream.
72    pub stream_idle: Option<Duration>,
73    /// Optional total lifetime for an established stream.
74    pub stream_total_lifetime: Option<Duration>,
75    /// Deadline for a cleanup RPC.
76    pub cleanup: Option<Duration>,
77}
78
79impl Default for GrpcTimeoutPolicy {
80    fn default() -> Self {
81        Self {
82            credential_setup: Some(DEFAULT_CREDENTIAL_GRPC_TIMEOUT),
83            ordinary_unary: Some(DEFAULT_ORDINARY_GRPC_TIMEOUT),
84            long_unary: Some(DEFAULT_LONG_GRPC_TIMEOUT),
85            stream_setup: Some(DEFAULT_STREAM_SETUP_GRPC_TIMEOUT),
86            stream_idle: None,
87            stream_total_lifetime: None,
88            cleanup: Some(DEFAULT_CLEANUP_GRPC_TIMEOUT),
89        }
90    }
91}
92
93impl GrpcTimeoutPolicy {
94    /// Resolves an explicit policy or the process environment and defaults.
95    ///
96    /// An explicit policy ignores [`ANYTYPE_GRPC_TIMEOUT_SECS`].
97    pub fn resolve(explicit: Option<Self>) -> Result<Self, GrpcTimeoutConfigError> {
98        if let Some(policy) = explicit {
99            return policy.validate();
100        }
101        Self::from_environment(std::env::var_os(ANYTYPE_GRPC_TIMEOUT_SECS))
102    }
103
104    /// Validates every finite boundary and returns the unchanged policy.
105    pub fn validate(self) -> Result<Self, GrpcTimeoutConfigError> {
106        for (field, value, maximum) in [
107            ("credential_setup", self.credential_setup, MAX_GRPC_TIMEOUT),
108            ("ordinary_unary", self.ordinary_unary, MAX_GRPC_TIMEOUT),
109            ("long_unary", self.long_unary, MAX_LONG_GRPC_TIMEOUT),
110            ("stream_setup", self.stream_setup, MAX_GRPC_TIMEOUT),
111            ("stream_idle", self.stream_idle, MAX_GRPC_TIMEOUT),
112            (
113                "stream_total_lifetime",
114                self.stream_total_lifetime,
115                MAX_GRPC_TIMEOUT,
116            ),
117            ("cleanup", self.cleanup, MAX_CLEANUP_GRPC_TIMEOUT),
118        ] {
119            if let Some(duration) = value
120                && !(Duration::from_secs(1)..=maximum).contains(&duration)
121            {
122                return Err(GrpcTimeoutConfigError::InvalidField {
123                    field,
124                    maximum_seconds: maximum.as_secs(),
125                });
126            }
127        }
128        Ok(self)
129    }
130
131    /// Returns the configured duration for one deadline class.
132    #[must_use]
133    pub const fn duration(self, class: GrpcTimeoutClass) -> Option<Duration> {
134        match class {
135            GrpcTimeoutClass::CredentialSetup => self.credential_setup,
136            GrpcTimeoutClass::OrdinaryUnary => self.ordinary_unary,
137            GrpcTimeoutClass::LongUnary => self.long_unary,
138            GrpcTimeoutClass::StreamSetup => self.stream_setup,
139            GrpcTimeoutClass::StreamIdle => self.stream_idle,
140            GrpcTimeoutClass::StreamLifetime => self.stream_total_lifetime,
141            GrpcTimeoutClass::Cleanup => self.cleanup,
142        }
143    }
144
145    fn from_environment(value: Option<OsString>) -> Result<Self, GrpcTimeoutConfigError> {
146        let Some(value) = value else {
147            return Ok(Self::default());
148        };
149        let value = value
150            .into_string()
151            .map_err(|_| GrpcTimeoutConfigError::InvalidEnvironment)?;
152        if value.is_empty()
153            || !value.bytes().all(|byte| byte.is_ascii_digit())
154            || (value.len() > 1 && value.starts_with('0'))
155        {
156            return Err(GrpcTimeoutConfigError::InvalidEnvironment);
157        }
158        let seconds = value
159            .parse::<u64>()
160            .map_err(|_| GrpcTimeoutConfigError::InvalidEnvironment)?;
161        if seconds > MAX_GRPC_TIMEOUT.as_secs() {
162            return Err(GrpcTimeoutConfigError::EnvironmentOutOfRange);
163        }
164        let inherited = (seconds != 0).then(|| Duration::from_secs(seconds));
165        Ok(Self {
166            credential_setup: inherited,
167            ordinary_unary: inherited,
168            long_unary: inherited,
169            stream_setup: inherited,
170            stream_idle: None,
171            stream_total_lifetime: None,
172            cleanup: Some(DEFAULT_CLEANUP_GRPC_TIMEOUT),
173        })
174    }
175}
176
177/// Invalid logical gRPC deadline configuration.
178#[derive(Clone, Copy, Debug, PartialEq, Eq)]
179pub enum GrpcTimeoutConfigError {
180    /// A programmatic field was zero, subsecond, or above its finite maximum.
181    InvalidField {
182        /// Stable field name.
183        field: &'static str,
184        /// Largest supported whole-second value.
185        maximum_seconds: u64,
186    },
187    /// The process override was not an exact supported ASCII decimal.
188    InvalidEnvironment,
189    /// The process override exceeded 3,600 seconds.
190    EnvironmentOutOfRange,
191    /// Absolute deadline arithmetic was not representable.
192    UnrepresentableDeadline,
193}
194
195impl fmt::Display for GrpcTimeoutConfigError {
196    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
197        match self {
198            Self::InvalidField {
199                field,
200                maximum_seconds,
201            } => write!(
202                formatter,
203                "grpc_timeouts.{field} must be disabled or between 1 and {maximum_seconds} seconds"
204            ),
205            Self::InvalidEnvironment => write!(
206                formatter,
207                "{ANYTYPE_GRPC_TIMEOUT_SECS} must be an ASCII decimal from 0 through 3600"
208            ),
209            Self::EnvironmentOutOfRange => write!(
210                formatter,
211                "{ANYTYPE_GRPC_TIMEOUT_SECS} must not exceed 3600 seconds"
212            ),
213            Self::UnrepresentableDeadline => {
214                formatter.write_str("gRPC absolute deadline is not representable")
215            }
216        }
217    }
218}
219
220impl StdError for GrpcTimeoutConfigError {}
221
222/// Closed logical gRPC deadline taxonomy.
223#[derive(Clone, Copy, Debug, PartialEq, Eq)]
224pub enum GrpcTimeoutClass {
225    /// Credential-session setup.
226    CredentialSetup,
227    /// Ordinary unary RPC.
228    OrdinaryUnary,
229    /// Long unary RPC.
230    LongUnary,
231    /// Streaming response setup.
232    StreamSetup,
233    /// Established stream no-progress boundary.
234    StreamIdle,
235    /// Established stream total lifetime.
236    StreamLifetime,
237    /// Cleanup RPC.
238    Cleanup,
239}
240
241impl GrpcTimeoutClass {
242    const fn as_str(self) -> &'static str {
243        match self {
244            Self::CredentialSetup => "credential_setup",
245            Self::OrdinaryUnary => "ordinary_unary",
246            Self::LongUnary => "long_unary",
247            Self::StreamSetup => "stream_setup",
248            Self::StreamIdle => "stream_idle",
249            Self::StreamLifetime => "stream_lifetime",
250            Self::Cleanup => "cleanup",
251        }
252    }
253
254    fn from_str(value: &str) -> Option<Self> {
255        match value {
256            "credential_setup" => Some(Self::CredentialSetup),
257            "ordinary_unary" => Some(Self::OrdinaryUnary),
258            "long_unary" => Some(Self::LongUnary),
259            "stream_setup" => Some(Self::StreamSetup),
260            "stream_idle" => Some(Self::StreamIdle),
261            "stream_lifetime" => Some(Self::StreamLifetime),
262            "cleanup" => Some(Self::Cleanup),
263            _ => None,
264        }
265    }
266}
267
268impl fmt::Display for GrpcTimeoutClass {
269    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
270        formatter.write_str(self.as_str())
271    }
272}
273
274/// Effect of a gRPC timeout on an operation.
275#[derive(Clone, Copy, Debug, PartialEq, Eq)]
276pub enum GrpcTimeoutOutcome {
277    /// A read did not complete, or dispatch had not begun when time expired.
278    ReadAborted,
279    /// A mutation may have reached the server.
280    MutationIndeterminate,
281    /// An established stream was terminated.
282    StreamTerminated,
283}
284
285impl GrpcTimeoutOutcome {
286    const fn as_str(self) -> &'static str {
287        match self {
288            Self::ReadAborted => "read_aborted",
289            Self::MutationIndeterminate => "mutation_indeterminate",
290            Self::StreamTerminated => "stream_terminated",
291        }
292    }
293
294    fn from_str(value: &str) -> Option<Self> {
295        match value {
296            "read_aborted" => Some(Self::ReadAborted),
297            "mutation_indeterminate" => Some(Self::MutationIndeterminate),
298            "stream_terminated" => Some(Self::StreamTerminated),
299            _ => None,
300        }
301    }
302}
303
304impl fmt::Display for GrpcTimeoutOutcome {
305    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
306        formatter.write_str(self.as_str())
307    }
308}
309
310/// Origin of an observed gRPC deadline expiration.
311#[derive(Clone, Copy, Debug, PartialEq, Eq)]
312pub enum GrpcTimeoutSource {
313    /// The local Tokio absolute deadline expired.
314    Local,
315    /// The peer returned `DeadlineExceeded`.
316    Server,
317}
318
319impl fmt::Display for GrpcTimeoutSource {
320    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
321        formatter.write_str(match self {
322            Self::Local => "local",
323            Self::Server => "server",
324        })
325    }
326}
327
328/// Stable, payload-free gRPC deadline classification.
329#[derive(Clone, Copy, Debug, PartialEq, Eq)]
330pub struct GrpcDeadlineError {
331    /// Expired deadline class.
332    pub class: GrpcTimeoutClass,
333    /// Safety outcome of the interrupted operation.
334    pub outcome: GrpcTimeoutOutcome,
335    /// Whether expiration was enforced locally or reported by the peer.
336    pub source: GrpcTimeoutSource,
337    /// Elapsed local operation time when known.
338    pub elapsed: Duration,
339}
340
341impl GrpcDeadlineError {
342    /// Classifies a deadline status without retaining its message or metadata.
343    #[must_use]
344    pub fn from_status(
345        status: &Status,
346        fallback_class: GrpcTimeoutClass,
347        fallback_outcome: GrpcTimeoutOutcome,
348        elapsed: Duration,
349    ) -> Option<Self> {
350        if status.code() != Code::DeadlineExceeded {
351            return None;
352        }
353        let class = metadata_text(status.metadata(), DEADLINE_CLASS_METADATA)
354            .and_then(GrpcTimeoutClass::from_str)
355            .unwrap_or(fallback_class);
356        let outcome = metadata_text(status.metadata(), DEADLINE_OUTCOME_METADATA)
357            .and_then(GrpcTimeoutOutcome::from_str)
358            .unwrap_or(fallback_outcome);
359        let source = if metadata_text(status.metadata(), DEADLINE_SOURCE_METADATA) == Some("local")
360        {
361            GrpcTimeoutSource::Local
362        } else {
363            GrpcTimeoutSource::Server
364        };
365        Some(Self {
366            class,
367            outcome,
368            source,
369            elapsed,
370        })
371    }
372}
373
374impl fmt::Display for GrpcDeadlineError {
375    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
376        write!(
377            formatter,
378            "gRPC deadline expired class={} outcome={} source={} elapsed_ms={}",
379            self.class,
380            self.outcome,
381            self.source,
382            self.elapsed.as_millis()
383        )
384    }
385}
386
387impl StdError for GrpcDeadlineError {}
388
389/// An optional absolute budget supplied by an enclosing workflow.
390#[derive(Clone, Copy, Debug, PartialEq, Eq)]
391pub struct GrpcEnclosingDeadline(tokio::time::Instant);
392
393impl GrpcEnclosingDeadline {
394    /// Captures an absolute deadline relative to the current Tokio clock.
395    pub fn from_now(duration: Duration) -> Result<Self, GrpcTimeoutConfigError> {
396        tokio::time::Instant::now()
397            .checked_add(duration)
398            .map(Self)
399            .ok_or(GrpcTimeoutConfigError::UnrepresentableDeadline)
400    }
401
402    /// Wraps an already captured Tokio instant.
403    #[must_use]
404    pub const fn from_instant(deadline: tokio::time::Instant) -> Self {
405        Self(deadline)
406    }
407
408    /// Returns the captured absolute Tokio instant.
409    #[must_use]
410    pub const fn instant(self) -> tokio::time::Instant {
411        self.0
412    }
413}
414
415/// Runs gRPC work under one caller-owned absolute deadline.
416///
417/// The deadline is applied by the transport layer to every generated call,
418/// including credential setup and calls whose request options are inferred.
419pub async fn scope_grpc_enclosing_deadline<F, T>(deadline: GrpcEnclosingDeadline, operation: F) -> T
420where
421    F: Future<Output = T>,
422{
423    ENCLOSING_DEADLINE.scope(deadline, operation).await
424}
425
426/// Per-request profile, outcome, and optional enclosing budget.
427#[derive(Clone, Copy, Debug, PartialEq, Eq)]
428pub struct GrpcCallOptions {
429    /// Selected logical profile.
430    pub class: GrpcTimeoutClass,
431    /// Safety outcome if the call expires after possible dispatch.
432    pub outcome: GrpcTimeoutOutcome,
433    /// Optional smaller enclosing absolute deadline.
434    pub enclosing: Option<GrpcEnclosingDeadline>,
435}
436
437impl GrpcCallOptions {
438    /// Creates request options for a deadline class and safety outcome.
439    #[must_use]
440    pub const fn new(class: GrpcTimeoutClass, outcome: GrpcTimeoutOutcome) -> Self {
441        Self {
442            class,
443            outcome,
444            enclosing: None,
445        }
446    }
447
448    /// Adds an enclosing absolute deadline.
449    #[must_use]
450    pub const fn enclosing(mut self, deadline: GrpcEnclosingDeadline) -> Self {
451        self.enclosing = Some(deadline);
452        self
453    }
454
455    /// Ordinary read options.
456    #[must_use]
457    pub const fn ordinary_read() -> Self {
458        Self::new(
459            GrpcTimeoutClass::OrdinaryUnary,
460            GrpcTimeoutOutcome::ReadAborted,
461        )
462    }
463
464    /// Ordinary mutation options.
465    #[must_use]
466    pub const fn ordinary_mutation() -> Self {
467        Self::new(
468            GrpcTimeoutClass::OrdinaryUnary,
469            GrpcTimeoutOutcome::MutationIndeterminate,
470        )
471    }
472
473    /// Long read options.
474    #[must_use]
475    pub const fn long_read() -> Self {
476        Self::new(GrpcTimeoutClass::LongUnary, GrpcTimeoutOutcome::ReadAborted)
477    }
478
479    /// Stream setup options.
480    #[must_use]
481    pub const fn stream_setup() -> Self {
482        Self::new(
483            GrpcTimeoutClass::StreamSetup,
484            GrpcTimeoutOutcome::StreamTerminated,
485        )
486    }
487
488    /// Cleanup mutation options.
489    #[must_use]
490    pub const fn cleanup() -> Self {
491        Self::new(
492            GrpcTimeoutClass::Cleanup,
493            GrpcTimeoutOutcome::MutationIndeterminate,
494        )
495    }
496}
497
498impl Default for GrpcCallOptions {
499    fn default() -> Self {
500        // Unknown RPCs are classified conservatively until their owner marks
501        // read semantics explicitly.
502        Self::ordinary_mutation()
503    }
504}
505
506/// Adds explicit deadline semantics to a tonic request.
507pub fn with_grpc_call_options<T>(mut request: Request<T>, options: GrpcCallOptions) -> Request<T> {
508    request.extensions_mut().insert(options);
509    request
510}
511
512/// Raw response-body progress shared with an established stream controller.
513///
514/// Tonic preserves this value in the streaming response extensions. Each data
515/// frame advances its generation before tonic attempts to decode a message.
516#[derive(Clone)]
517pub struct GrpcTransportProgress {
518    inner: Arc<GrpcTransportProgressInner>,
519}
520
521struct GrpcTransportProgressInner {
522    generation: AtomicU64,
523    notify: tokio::sync::Notify,
524}
525
526impl GrpcTransportProgress {
527    fn new() -> Self {
528        Self {
529            inner: Arc::new(GrpcTransportProgressInner {
530                generation: AtomicU64::new(0),
531                notify: tokio::sync::Notify::new(),
532            }),
533        }
534    }
535
536    fn record(&self) {
537        self.inner.generation.fetch_add(1, Ordering::Relaxed);
538        self.inner.notify.notify_waiters();
539    }
540
541    fn generation(&self) -> u64 {
542        self.inner.generation.load(Ordering::Relaxed)
543    }
544
545    async fn changed_after(&self, generation: u64) -> u64 {
546        loop {
547            let notified = self.inner.notify.notified();
548            let current = self.generation();
549            if current != generation {
550                return current;
551            }
552            notified.await;
553        }
554    }
555}
556
557impl fmt::Debug for GrpcTransportProgress {
558    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
559        formatter
560            .debug_struct("GrpcTransportProgress")
561            .field("generation", &self.generation())
562            .finish()
563    }
564}
565
566/// A service wrapper that propagates and locally enforces logical deadlines.
567///
568/// Generated tonic clients can use this wrapper exactly as they use a channel.
569/// Existing shorter `grpc-timeout` metadata is preserved as the winning bound.
570#[derive(Clone)]
571pub struct GrpcDeadlineService<S> {
572    inner: S,
573    policy: GrpcTimeoutPolicy,
574}
575
576impl<S> fmt::Debug for GrpcDeadlineService<S> {
577    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
578        formatter
579            .debug_struct("GrpcDeadlineService")
580            .field("policy", &self.policy)
581            .field("inner", &"redacted")
582            .finish()
583    }
584}
585
586impl<S> GrpcDeadlineService<S> {
587    /// Wraps a tonic service after validating its timeout policy.
588    pub fn try_new(inner: S, policy: GrpcTimeoutPolicy) -> Result<Self, GrpcTimeoutConfigError> {
589        Ok(Self {
590            inner,
591            policy: policy.validate()?,
592        })
593    }
594
595    pub(crate) const fn new_resolved(inner: S, policy: GrpcTimeoutPolicy) -> Self {
596        Self { inner, policy }
597    }
598
599    /// Returns the resolved policy used by this service.
600    #[must_use]
601    pub const fn policy(&self) -> GrpcTimeoutPolicy {
602        self.policy
603    }
604
605    /// Returns a reference to the wrapped service.
606    #[must_use]
607    pub const fn inner(&self) -> &S {
608        &self.inner
609    }
610}
611
612/// Service error used by [`GrpcDeadlineService`].
613pub enum GrpcDeadlineServiceError<E> {
614    /// The wrapped transport failed.
615    Transport {
616        /// Closed gRPC status classification captured before redaction.
617        code: Code,
618        /// Retains the transport error type without retaining its payload.
619        error_type: PhantomData<fn() -> E>,
620    },
621    /// A local deadline expired or was exhausted before dispatch.
622    Deadline(Status),
623    /// Existing `grpc-timeout` metadata was malformed.
624    InvalidTimeout(Status),
625}
626
627impl<E> fmt::Debug for GrpcDeadlineServiceError<E> {
628    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
629        match self {
630            Self::Transport { code, .. } => formatter
631                .debug_struct("GrpcDeadlineServiceError::Transport")
632                .field("code", code)
633                .field("source", &"redacted")
634                .finish(),
635            Self::Deadline(_) => {
636                formatter.write_str("GrpcDeadlineServiceError::Deadline(redacted)")
637            }
638            Self::InvalidTimeout(_) => {
639                formatter.write_str("GrpcDeadlineServiceError::InvalidTimeout(redacted)")
640            }
641        }
642    }
643}
644
645impl<E> fmt::Display for GrpcDeadlineServiceError<E> {
646    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
647        formatter.write_str(match self {
648            Self::Transport { .. } => "gRPC transport failed",
649            Self::Deadline(_) => "gRPC local deadline expired",
650            Self::InvalidTimeout(_) => "invalid gRPC timeout metadata",
651        })
652    }
653}
654
655impl<E> StdError for GrpcDeadlineServiceError<E> {
656    fn source(&self) -> Option<&(dyn StdError + 'static)> {
657        match self {
658            Self::Transport { code, .. } => Some(redacted_transport_status_source(*code)),
659            Self::Deadline(status) => Some(redacted_deadline_status_source(status)),
660            Self::InvalidTimeout(_) => Some(redacted_invalid_timeout_source()),
661        }
662    }
663}
664
665impl<E> GrpcDeadlineServiceError<E>
666where
667    E: StdError + Send + Sync + 'static,
668{
669    fn transport(source: E) -> Self {
670        let code = closed_grpc_error_code(source);
671        Self::Transport {
672            code,
673            error_type: PhantomData,
674        }
675    }
676}
677
678fn closed_grpc_error_code<E>(error: E) -> Code
679where
680    E: StdError + Send + Sync + 'static,
681{
682    Status::from_error(Box::new(error)).code()
683}
684
685fn redacted_transport_status_source(code: Code) -> &'static Status {
686    static SOURCES: [OnceLock<Status>; 17] = [const { OnceLock::new() }; 17];
687    let index = match code {
688        Code::Ok => 0,
689        Code::Cancelled => 1,
690        Code::Unknown => 2,
691        Code::InvalidArgument => 3,
692        Code::DeadlineExceeded => 4,
693        Code::NotFound => 5,
694        Code::AlreadyExists => 6,
695        Code::PermissionDenied => 7,
696        Code::ResourceExhausted => 8,
697        Code::FailedPrecondition => 9,
698        Code::Aborted => 10,
699        Code::OutOfRange => 11,
700        Code::Unimplemented => 12,
701        Code::Internal => 13,
702        Code::Unavailable => 14,
703        Code::DataLoss => 15,
704        Code::Unauthenticated => 16,
705    };
706    SOURCES[index].get_or_init(|| Status::new(code, "gRPC transport failed (details redacted)"))
707}
708
709fn redacted_deadline_status_source(status: &Status) -> &'static Status {
710    static SOURCES: [OnceLock<Status>; 21] = [const { OnceLock::new() }; 21];
711    let class = metadata_text(status.metadata(), DEADLINE_CLASS_METADATA)
712        .and_then(GrpcTimeoutClass::from_str)
713        .unwrap_or(GrpcTimeoutClass::OrdinaryUnary);
714    let outcome = metadata_text(status.metadata(), DEADLINE_OUTCOME_METADATA)
715        .and_then(GrpcTimeoutOutcome::from_str)
716        .unwrap_or(GrpcTimeoutOutcome::MutationIndeterminate);
717    let class_index = match class {
718        GrpcTimeoutClass::CredentialSetup => 0,
719        GrpcTimeoutClass::OrdinaryUnary => 1,
720        GrpcTimeoutClass::LongUnary => 2,
721        GrpcTimeoutClass::StreamSetup => 3,
722        GrpcTimeoutClass::StreamIdle => 4,
723        GrpcTimeoutClass::StreamLifetime => 5,
724        GrpcTimeoutClass::Cleanup => 6,
725    };
726    let outcome_index = match outcome {
727        GrpcTimeoutOutcome::ReadAborted => 0,
728        GrpcTimeoutOutcome::MutationIndeterminate => 1,
729        GrpcTimeoutOutcome::StreamTerminated => 2,
730    };
731    SOURCES[class_index * 3 + outcome_index]
732        .get_or_init(|| local_deadline_status(GrpcCallOptions::new(class, outcome), Duration::ZERO))
733}
734
735fn redacted_invalid_timeout_source() -> &'static Status {
736    static SOURCE: OnceLock<Status> = OnceLock::new();
737    SOURCE.get_or_init(|| Status::invalid_argument("invalid gRPC timeout metadata"))
738}
739
740type BoxServiceFuture<T, E> = Pin<
741    Box<
742        dyn Future<
743                Output = Result<http::Response<GrpcDeadlineBody<T>>, GrpcDeadlineServiceError<E>>,
744            > + Send
745            + 'static,
746    >,
747>;
748
749/// Response body that retains a unary RPC's absolute deadline through trailers.
750pub struct GrpcDeadlineBody<B> {
751    inner: Pin<Box<B>>,
752    sleep: Option<Pin<Box<tokio::time::Sleep>>>,
753    progress: Option<GrpcTransportProgress>,
754    options: GrpcCallOptions,
755    started: tokio::time::Instant,
756    finished: bool,
757}
758
759impl<B> fmt::Debug for GrpcDeadlineBody<B> {
760    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
761        formatter
762            .debug_struct("GrpcDeadlineBody")
763            .field("class", &self.options.class)
764            .field("outcome", &self.options.outcome)
765            .field("deadline_enabled", &self.sleep.is_some())
766            .field("progress_enabled", &self.progress.is_some())
767            .field("finished", &self.finished)
768            .field("inner", &"redacted")
769            .finish()
770    }
771}
772
773impl<B> GrpcDeadlineBody<B> {
774    fn new(
775        inner: B,
776        deadline: Option<tokio::time::Instant>,
777        progress: Option<GrpcTransportProgress>,
778        options: GrpcCallOptions,
779        started: tokio::time::Instant,
780    ) -> Self {
781        Self {
782            inner: Box::pin(inner),
783            sleep: deadline.map(|deadline| Box::pin(tokio::time::sleep_until(deadline))),
784            progress,
785            options,
786            started,
787            finished: false,
788        }
789    }
790}
791
792impl<B> HttpBody for GrpcDeadlineBody<B>
793where
794    B: HttpBody,
795    B::Error: StdError + Send + Sync + 'static,
796{
797    type Data = B::Data;
798    type Error = GrpcDeadlineServiceError<B::Error>;
799
800    fn poll_frame(
801        mut self: Pin<&mut Self>,
802        context: &mut Context<'_>,
803    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
804        let this = self.as_mut().get_mut();
805        if this.finished {
806            return Poll::Ready(None);
807        }
808
809        if let Some(sleep) = this.sleep.as_mut()
810            && sleep.as_mut().poll(context).is_ready()
811        {
812            this.finished = true;
813            return Poll::Ready(Some(Err(GrpcDeadlineServiceError::Deadline(
814                local_deadline_status(this.options, this.started.elapsed()),
815            ))));
816        }
817
818        match this.inner.as_mut().poll_frame(context) {
819            Poll::Ready(Some(Ok(mut frame))) => {
820                if frame.data_ref().is_some_and(|data| data.remaining() > 0)
821                    && let Some(progress) = this.progress.as_ref()
822                {
823                    progress.record();
824                }
825                if let Some(trailers) = frame.trailers_mut() {
826                    annotate_server_deadline_headers(trailers, this.options);
827                }
828                return Poll::Ready(Some(Ok(frame)));
829            }
830            Poll::Ready(Some(Err(source))) => {
831                this.finished = true;
832                return Poll::Ready(Some(Err(GrpcDeadlineServiceError::transport(source))));
833            }
834            Poll::Ready(None) => {
835                this.finished = true;
836                return Poll::Ready(None);
837            }
838            Poll::Pending => {}
839        }
840        Poll::Pending
841    }
842
843    fn is_end_stream(&self) -> bool {
844        self.finished || self.inner.is_end_stream()
845    }
846
847    fn size_hint(&self) -> SizeHint {
848        self.inner.size_hint()
849    }
850}
851
852impl<S, ResponseBody> Service<http::Request<Body>> for GrpcDeadlineService<S>
853where
854    S: Service<http::Request<Body>, Response = http::Response<ResponseBody>>
855        + Clone
856        + Send
857        + 'static,
858    S::Error: StdError + Send + Sync + 'static,
859    S::Future: Send + 'static,
860    ResponseBody: Send + 'static,
861{
862    type Response = http::Response<GrpcDeadlineBody<ResponseBody>>;
863    type Error = GrpcDeadlineServiceError<S::Error>;
864    type Future = BoxServiceFuture<ResponseBody, S::Error>;
865
866    fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
867        let _ = context;
868        Poll::Ready(Ok(()))
869    }
870
871    fn call(&mut self, mut request: http::Request<Body>) -> Self::Future {
872        let mut options = request
873            .extensions()
874            .get::<GrpcCallOptions>()
875            .copied()
876            .unwrap_or_else(|| inferred_call_options(request.extensions().get::<GrpcMethod>()));
877        if let Ok(scoped) = ENCLOSING_DEADLINE.try_with(|deadline| *deadline) {
878            options.enclosing = Some(match options.enclosing {
879                Some(explicit) => {
880                    GrpcEnclosingDeadline::from_instant(explicit.instant().min(scoped.instant()))
881                }
882                None => scoped,
883            });
884        }
885        let started = tokio::time::Instant::now();
886        let caller_deadline = match request.headers().get(GRPC_TIMEOUT_HEADER) {
887            Some(value) => match parse_grpc_timeout(value) {
888                Ok(Some(duration)) => match started.checked_add(duration) {
889                    Some(deadline) => Some(deadline),
890                    None => {
891                        let status =
892                            Status::invalid_argument("unrepresentable grpc-timeout metadata");
893                        return Box::pin(async move {
894                            Err(GrpcDeadlineServiceError::InvalidTimeout(status))
895                        });
896                    }
897                },
898                Ok(None) => None,
899                Err(_) => {
900                    let status = Status::invalid_argument("invalid grpc-timeout metadata");
901                    return Box::pin(async move {
902                        Err(GrpcDeadlineServiceError::InvalidTimeout(status))
903                    });
904                }
905            },
906            None => None,
907        };
908        let selected = select_deadline(
909            self.policy.duration(options.class),
910            options.enclosing,
911            request.headers().get(GRPC_TIMEOUT_HEADER),
912            started,
913        );
914        let duration = match selected {
915            Ok(duration) => duration,
916            Err(SelectDeadlineError::Expired) => {
917                let status = local_deadline_status(
918                    GrpcCallOptions {
919                        outcome: GrpcTimeoutOutcome::ReadAborted,
920                        ..options
921                    },
922                    Duration::ZERO,
923                );
924                return Box::pin(async move { Err(GrpcDeadlineServiceError::Deadline(status)) });
925            }
926            Err(SelectDeadlineError::InvalidHeader) => {
927                let status = Status::invalid_argument("invalid grpc-timeout metadata");
928                return Box::pin(
929                    async move { Err(GrpcDeadlineServiceError::InvalidTimeout(status)) },
930                );
931            }
932        };
933
934        let absolute_deadline = match duration {
935            Some(duration) => match started.checked_add(duration) {
936                Some(deadline) => Some(deadline),
937                None => {
938                    let status = Status::invalid_argument("unrepresentable gRPC deadline");
939                    return Box::pin(async move {
940                        Err(GrpcDeadlineServiceError::InvalidTimeout(status))
941                    });
942                }
943            },
944            None => None,
945        };
946        let mut inner = self.inner.clone();
947        Box::pin(async move {
948            let ready = async { std::future::poll_fn(|context| inner.poll_ready(context)).await };
949            let ready_result = if let Some(deadline) = absolute_deadline {
950                tokio::pin!(ready);
951                tokio::select! {
952                    biased;
953                    result = &mut ready => Some(result),
954                    () = tokio::time::sleep_until(deadline) => None,
955                }
956            } else {
957                Some(ready.await)
958            };
959            match ready_result {
960                Some(Ok(())) => {}
961                Some(Err(source)) if error_chain_contains_timeout(&source) => {
962                    let status = local_deadline_status(
963                        GrpcCallOptions {
964                            outcome: GrpcTimeoutOutcome::ReadAborted,
965                            ..options
966                        },
967                        started.elapsed(),
968                    );
969                    return Err(GrpcDeadlineServiceError::Deadline(status));
970                }
971                Some(Err(source)) => return Err(GrpcDeadlineServiceError::transport(source)),
972                None => {
973                    let status = local_deadline_status(
974                        GrpcCallOptions {
975                            outcome: GrpcTimeoutOutcome::ReadAborted,
976                            ..options
977                        },
978                        started.elapsed(),
979                    );
980                    return Err(GrpcDeadlineServiceError::Deadline(status));
981                }
982            }
983
984            if let Some(deadline) = absolute_deadline {
985                let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
986                if remaining.is_zero() {
987                    let status = local_deadline_status(
988                        GrpcCallOptions {
989                            outcome: GrpcTimeoutOutcome::ReadAborted,
990                            ..options
991                        },
992                        started.elapsed(),
993                    );
994                    return Err(GrpcDeadlineServiceError::Deadline(status));
995                }
996                let propagated = if options.class == GrpcTimeoutClass::StreamSetup {
997                    caller_deadline
998                        .map(|deadline| {
999                            deadline.saturating_duration_since(tokio::time::Instant::now())
1000                        })
1001                        .filter(|duration| !duration.is_zero())
1002                } else {
1003                    Some(remaining)
1004                };
1005                if let Some(propagated) = propagated {
1006                    let value = grpc_timeout_value(propagated).ok_or_else(|| {
1007                        GrpcDeadlineServiceError::InvalidTimeout(Status::invalid_argument(
1008                            "unrepresentable grpc-timeout metadata",
1009                        ))
1010                    })?;
1011                    let header = http::HeaderValue::from_str(&value).map_err(|_| {
1012                        GrpcDeadlineServiceError::InvalidTimeout(Status::invalid_argument(
1013                            "invalid grpc-timeout metadata",
1014                        ))
1015                    })?;
1016                    request.headers_mut().insert(GRPC_TIMEOUT_HEADER, header);
1017                }
1018            }
1019
1020            let future = inner.call(request);
1021            let result = if let Some(deadline) = absolute_deadline {
1022                tokio::pin!(future);
1023                tokio::select! {
1024                    biased;
1025                    result = &mut future => Some(result),
1026                    () = tokio::time::sleep_until(deadline) => None,
1027                }
1028            } else {
1029                Some(future.await)
1030            };
1031
1032            match result {
1033                Some(Ok(mut response)) => {
1034                    sanitize_response_deadline_headers(response.headers_mut(), options);
1035                    let body_deadline = (options.class != GrpcTimeoutClass::StreamSetup)
1036                        .then_some(absolute_deadline)
1037                        .flatten();
1038                    let progress = (options.class == GrpcTimeoutClass::StreamSetup)
1039                        .then(GrpcTransportProgress::new);
1040                    if let Some(progress) = progress.as_ref() {
1041                        response.extensions_mut().insert(progress.clone());
1042                    }
1043                    Ok(response.map(|body| {
1044                        GrpcDeadlineBody::new(body, body_deadline, progress, options, started)
1045                    }))
1046                }
1047                Some(Err(source)) if error_chain_contains_timeout(&source) => {
1048                    Err(GrpcDeadlineServiceError::Deadline(local_deadline_status(
1049                        options,
1050                        started.elapsed(),
1051                    )))
1052                }
1053                Some(Err(source)) => Err(GrpcDeadlineServiceError::transport(source)),
1054                None => Err(GrpcDeadlineServiceError::Deadline(local_deadline_status(
1055                    options,
1056                    started.elapsed(),
1057                ))),
1058            }
1059        })
1060    }
1061}
1062
1063/// Established-stream idle and lifetime controller.
1064#[derive(Debug)]
1065pub struct GrpcStreamDeadline {
1066    idle: Option<Duration>,
1067    idle_deadline: Option<tokio::time::Instant>,
1068    lifetime_deadline: Option<tokio::time::Instant>,
1069    enclosing: Option<GrpcEnclosingDeadline>,
1070    transport_progress: Option<GrpcTransportProgress>,
1071    observed_transport_generation: u64,
1072    started: tokio::time::Instant,
1073}
1074
1075/// Error returned while waiting for established-stream progress.
1076pub enum GrpcStreamError {
1077    /// An idle, lifetime, enclosing, or peer deadline expired.
1078    Deadline(GrpcDeadlineError),
1079    /// The stream failed for a reason other than deadline expiration.
1080    Status(Status),
1081}
1082
1083impl fmt::Debug for GrpcStreamError {
1084    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1085        match self {
1086            Self::Deadline(source) => formatter
1087                .debug_tuple("GrpcStreamError::Deadline")
1088                .field(source)
1089                .finish(),
1090            Self::Status(_) => formatter.write_str("GrpcStreamError::Status(redacted)"),
1091        }
1092    }
1093}
1094
1095impl fmt::Display for GrpcStreamError {
1096    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1097        match self {
1098            Self::Deadline(source) => source.fmt(formatter),
1099            Self::Status(_) => formatter.write_str("gRPC stream failed (details redacted)"),
1100        }
1101    }
1102}
1103
1104impl StdError for GrpcStreamError {
1105    fn source(&self) -> Option<&(dyn StdError + 'static)> {
1106        match self {
1107            Self::Deadline(source) => Some(source),
1108            Self::Status(_) => None,
1109        }
1110    }
1111}
1112
1113impl GrpcStreamDeadline {
1114    /// Starts established-stream boundaries after successful response headers.
1115    pub fn new(
1116        policy: GrpcTimeoutPolicy,
1117        enclosing: Option<GrpcEnclosingDeadline>,
1118    ) -> Result<Self, GrpcTimeoutConfigError> {
1119        let started = tokio::time::Instant::now();
1120        let idle = policy.stream_idle;
1121        let idle_deadline = checked_deadline(started, idle)?;
1122        let lifetime_deadline = checked_deadline(started, policy.stream_total_lifetime)?;
1123        Ok(Self {
1124            idle,
1125            idle_deadline,
1126            lifetime_deadline,
1127            enclosing,
1128            transport_progress: None,
1129            observed_transport_generation: 0,
1130            started,
1131        })
1132    }
1133
1134    /// Connects raw body progress supplied in a tonic streaming response.
1135    #[must_use]
1136    pub fn with_transport_progress(mut self, progress: Option<GrpcTransportProgress>) -> Self {
1137        self.set_transport_progress(progress);
1138        self
1139    }
1140
1141    /// Replaces the raw body progress source after a successful stream reopen.
1142    pub fn set_transport_progress(&mut self, progress: Option<GrpcTransportProgress>) {
1143        self.observed_transport_generation = progress
1144            .as_ref()
1145            .map_or(0, GrpcTransportProgress::generation);
1146        self.transport_progress = progress;
1147    }
1148
1149    /// Restarts only the idle boundary after a successful stream reopen.
1150    ///
1151    /// The total lifetime and enclosing absolute deadline remain unchanged.
1152    pub fn reset_idle(&mut self) -> Result<(), GrpcTimeoutConfigError> {
1153        self.idle_deadline = checked_deadline(tokio::time::Instant::now(), self.idle)?;
1154        if let Some(progress) = self.transport_progress.as_ref() {
1155            self.observed_transport_generation = progress.generation();
1156        }
1157        Ok(())
1158    }
1159
1160    /// Records delivery of a decoded message when raw body progress is unavailable.
1161    ///
1162    /// When transport progress is attached, its DATA frame already advanced the
1163    /// idle boundary and decoding must not grant a second idle window.
1164    pub fn observe_decoded_message(&mut self) -> Result<(), GrpcTimeoutConfigError> {
1165        if self.transport_progress.is_some() {
1166            Ok(())
1167        } else {
1168            self.reset_idle()
1169        }
1170    }
1171
1172    /// Returns the earliest absolute lifetime or enclosing workflow boundary.
1173    #[must_use]
1174    pub fn workflow_deadline(&self) -> Option<GrpcEnclosingDeadline> {
1175        [
1176            self.lifetime_deadline,
1177            self.enclosing.map(GrpcEnclosingDeadline::instant),
1178        ]
1179        .into_iter()
1180        .flatten()
1181        .min()
1182        .map(GrpcEnclosingDeadline::from_instant)
1183    }
1184
1185    /// Runs reconnect or replay work under the retained lifetime/workflow bound.
1186    ///
1187    /// This boundary does not reset the idle timer because reconnect work is not
1188    /// established-stream transport progress.
1189    pub async fn phase<T, F>(&self, future: F) -> Result<T, GrpcStreamError>
1190    where
1191        F: Future<Output = T>,
1192    {
1193        let Some(deadline) = self.workflow_deadline().map(GrpcEnclosingDeadline::instant) else {
1194            return Ok(future.await);
1195        };
1196        if deadline <= tokio::time::Instant::now() {
1197            return Err(GrpcStreamError::Deadline(self.stream_error(
1198                GrpcTimeoutClass::StreamLifetime,
1199                GrpcTimeoutSource::Local,
1200            )));
1201        }
1202        tokio::pin!(future);
1203        tokio::select! {
1204            biased;
1205            () = tokio::time::sleep_until(deadline) => Err(GrpcStreamError::Deadline(
1206                self.stream_error(GrpcTimeoutClass::StreamLifetime, GrpcTimeoutSource::Local),
1207            )),
1208            value = &mut future => Ok(value),
1209        }
1210    }
1211
1212    /// Runs established-stream work under idle, lifetime, and enclosing bounds.
1213    ///
1214    /// Raw nonempty DATA progress resets idle while the work is pending. Work
1215    /// completion itself does not count as stream progress.
1216    pub async fn established_phase<T, F>(&mut self, future: F) -> Result<T, GrpcStreamError>
1217    where
1218        F: Future<Output = T>,
1219    {
1220        tokio::pin!(future);
1221        loop {
1222            self.observe_transport_progress()?;
1223            let now = tokio::time::Instant::now();
1224            let boundary = earliest_stream_boundary(
1225                self.idle_deadline,
1226                self.lifetime_deadline,
1227                self.enclosing.map(GrpcEnclosingDeadline::instant),
1228            );
1229            if let Some((deadline, class)) = boundary
1230                && deadline <= now
1231            {
1232                return Err(GrpcStreamError::Deadline(
1233                    self.stream_error(class, GrpcTimeoutSource::Local),
1234                ));
1235            }
1236
1237            let progress = self.transport_progress.clone();
1238            let observed = self.observed_transport_generation;
1239            let progress_wait = async move {
1240                match progress {
1241                    Some(progress) => progress.changed_after(observed).await,
1242                    None => std::future::pending().await,
1243                }
1244            };
1245            let result = if let Some((deadline, class)) = boundary {
1246                tokio::select! {
1247                    biased;
1248                    () = tokio::time::sleep_until(deadline) => {
1249                        return Err(GrpcStreamError::Deadline(
1250                            self.stream_error(class, GrpcTimeoutSource::Local),
1251                        ));
1252                    }
1253                    value = &mut future => return Ok(value),
1254                    generation = progress_wait => generation,
1255                }
1256            } else {
1257                tokio::select! {
1258                    biased;
1259                    value = &mut future => return Ok(value),
1260                    generation = progress_wait => generation,
1261                }
1262            };
1263            self.observed_transport_generation = result;
1264            self.idle_deadline =
1265                checked_deadline(tokio::time::Instant::now(), self.idle).map_err(|_| {
1266                    GrpcStreamError::Deadline(
1267                        self.stream_error(GrpcTimeoutClass::StreamIdle, GrpcTimeoutSource::Local),
1268                    )
1269                })?;
1270        }
1271    }
1272
1273    /// Waits for the next transport-progress future and resets idle on success.
1274    pub async fn next<T, F>(&mut self, future: F) -> Result<T, GrpcStreamError>
1275    where
1276        F: Future<Output = Result<T, Status>>,
1277    {
1278        tokio::pin!(future);
1279        loop {
1280            self.observe_transport_progress()?;
1281            let now = tokio::time::Instant::now();
1282            let boundary = earliest_stream_boundary(
1283                self.idle_deadline,
1284                self.lifetime_deadline,
1285                self.enclosing.map(GrpcEnclosingDeadline::instant),
1286            );
1287            if let Some((deadline, class)) = boundary
1288                && deadline <= now
1289            {
1290                return Err(GrpcStreamError::Deadline(
1291                    self.stream_error(class, GrpcTimeoutSource::Local),
1292                ));
1293            }
1294
1295            let progress = self.transport_progress.clone();
1296            let observed = self.observed_transport_generation;
1297            let progress_wait = async move {
1298                match progress {
1299                    Some(progress) => progress.changed_after(observed).await,
1300                    None => std::future::pending().await,
1301                }
1302            };
1303            let result = if let Some((deadline, class)) = boundary {
1304                tokio::select! {
1305                    biased;
1306                    () = tokio::time::sleep_until(deadline) => None,
1307                    result = &mut future => Some(StreamWait::Result(result, class)),
1308                    generation = progress_wait => Some(StreamWait::Progress(generation)),
1309                }
1310            } else {
1311                tokio::select! {
1312                    biased;
1313                    result = &mut future => Some(StreamWait::Result(
1314                        result,
1315                        GrpcTimeoutClass::StreamLifetime,
1316                    )),
1317                    generation = progress_wait => Some(StreamWait::Progress(generation)),
1318                }
1319            };
1320
1321            match result {
1322                Some(StreamWait::Progress(generation)) => {
1323                    self.observed_transport_generation = generation;
1324                    self.idle_deadline =
1325                        checked_deadline(tokio::time::Instant::now(), self.idle).map_err(|_| {
1326                            GrpcStreamError::Deadline(self.stream_error(
1327                                GrpcTimeoutClass::StreamIdle,
1328                                GrpcTimeoutSource::Local,
1329                            ))
1330                        })?;
1331                }
1332                Some(StreamWait::Result(Ok(value), _)) => {
1333                    self.idle_deadline =
1334                        checked_deadline(tokio::time::Instant::now(), self.idle).map_err(|_| {
1335                            GrpcStreamError::Deadline(self.stream_error(
1336                                GrpcTimeoutClass::StreamIdle,
1337                                GrpcTimeoutSource::Local,
1338                            ))
1339                        })?;
1340                    return Ok(value);
1341                }
1342                Some(StreamWait::Result(Err(status), class)) => {
1343                    return if let Some(error) = GrpcDeadlineError::from_status(
1344                        &status,
1345                        class,
1346                        GrpcTimeoutOutcome::StreamTerminated,
1347                        self.started.elapsed(),
1348                    ) {
1349                        Err(GrpcStreamError::Deadline(error))
1350                    } else {
1351                        Err(GrpcStreamError::Status(status))
1352                    };
1353                }
1354                None => {
1355                    let (_, class) = boundary.unwrap_or((now, GrpcTimeoutClass::StreamLifetime));
1356                    return Err(GrpcStreamError::Deadline(
1357                        self.stream_error(class, GrpcTimeoutSource::Local),
1358                    ));
1359                }
1360            }
1361        }
1362    }
1363
1364    fn observe_transport_progress(&mut self) -> Result<(), GrpcStreamError> {
1365        let Some(progress) = self.transport_progress.as_ref() else {
1366            return Ok(());
1367        };
1368        let generation = progress.generation();
1369        if generation == self.observed_transport_generation {
1370            return Ok(());
1371        }
1372        self.observed_transport_generation = generation;
1373        self.idle_deadline =
1374            checked_deadline(tokio::time::Instant::now(), self.idle).map_err(|_| {
1375                GrpcStreamError::Deadline(
1376                    self.stream_error(GrpcTimeoutClass::StreamIdle, GrpcTimeoutSource::Local),
1377                )
1378            })?;
1379        Ok(())
1380    }
1381
1382    fn stream_error(
1383        &self,
1384        class: GrpcTimeoutClass,
1385        source: GrpcTimeoutSource,
1386    ) -> GrpcDeadlineError {
1387        GrpcDeadlineError {
1388            class,
1389            outcome: GrpcTimeoutOutcome::StreamTerminated,
1390            source,
1391            elapsed: self.started.elapsed(),
1392        }
1393    }
1394}
1395
1396enum StreamWait<T> {
1397    Result(Result<T, Status>, GrpcTimeoutClass),
1398    Progress(u64),
1399}
1400
1401#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1402enum SelectDeadlineError {
1403    Expired,
1404    InvalidHeader,
1405}
1406
1407fn select_deadline(
1408    profile: Option<Duration>,
1409    enclosing: Option<GrpcEnclosingDeadline>,
1410    existing: Option<&http::HeaderValue>,
1411    now: tokio::time::Instant,
1412) -> Result<Option<Duration>, SelectDeadlineError> {
1413    let existing = existing.map(parse_grpc_timeout).transpose()?.flatten();
1414    let enclosing = enclosing
1415        .map(GrpcEnclosingDeadline::instant)
1416        .map(|deadline| {
1417            (deadline > now)
1418                .then(|| deadline.saturating_duration_since(now))
1419                .ok_or(SelectDeadlineError::Expired)
1420        })
1421        .transpose()?;
1422    let selected = [profile, enclosing, existing].into_iter().flatten().min();
1423    if selected == Some(Duration::ZERO) {
1424        return Err(SelectDeadlineError::Expired);
1425    }
1426    Ok(selected)
1427}
1428
1429fn parse_grpc_timeout(value: &http::HeaderValue) -> Result<Option<Duration>, SelectDeadlineError> {
1430    let value = value
1431        .to_str()
1432        .map_err(|_| SelectDeadlineError::InvalidHeader)?;
1433    if value.is_empty() || value.len() > 9 {
1434        return Err(SelectDeadlineError::InvalidHeader);
1435    }
1436    let (digits, unit) = value.split_at(value.len() - 1);
1437    if digits.is_empty() || digits.len() > 8 || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
1438        return Err(SelectDeadlineError::InvalidHeader);
1439    }
1440    let amount = digits
1441        .parse::<u64>()
1442        .map_err(|_| SelectDeadlineError::InvalidHeader)?;
1443    let duration = match unit {
1444        "H" => Duration::from_secs(amount.saturating_mul(3_600)),
1445        "M" => Duration::from_secs(amount.saturating_mul(60)),
1446        "S" => Duration::from_secs(amount),
1447        "m" => Duration::from_millis(amount),
1448        "u" => Duration::from_micros(amount),
1449        "n" => Duration::from_nanos(amount),
1450        _ => return Err(SelectDeadlineError::InvalidHeader),
1451    };
1452    Ok(Some(duration))
1453}
1454
1455fn grpc_timeout_value(duration: Duration) -> Option<String> {
1456    fn format_unit(value: u128, unit: char) -> Option<String> {
1457        (value <= 99_999_999).then(|| format!("{value}{unit}"))
1458    }
1459
1460    format_unit(duration.as_nanos(), 'n')
1461        .or_else(|| format_unit(duration.as_micros(), 'u'))
1462        .or_else(|| format_unit(duration.as_millis(), 'm'))
1463        .or_else(|| format_unit(duration.as_secs().into(), 'S'))
1464        .or_else(|| format_unit((duration.as_secs() / 60).into(), 'M'))
1465        .or_else(|| format_unit((duration.as_secs() / 3_600).into(), 'H'))
1466}
1467
1468fn local_deadline_status(options: GrpcCallOptions, _elapsed: Duration) -> Status {
1469    let mut metadata = MetadataMap::new();
1470    metadata.insert(
1471        DEADLINE_SOURCE_METADATA,
1472        MetadataValue::<Ascii>::from_static("local"),
1473    );
1474    if let Ok(value) = options.class.as_str().parse() {
1475        metadata.insert(DEADLINE_CLASS_METADATA, value);
1476    }
1477    if let Ok(value) = options.outcome.as_str().parse() {
1478        metadata.insert(DEADLINE_OUTCOME_METADATA, value);
1479    }
1480    Status::with_metadata(Code::DeadlineExceeded, "gRPC deadline exceeded", metadata)
1481}
1482
1483fn annotate_server_deadline_headers(headers: &mut http::HeaderMap, options: GrpcCallOptions) {
1484    headers.insert(
1485        DEADLINE_SOURCE_METADATA,
1486        http::HeaderValue::from_static("server"),
1487    );
1488    if let Ok(value) = http::HeaderValue::from_str(options.class.as_str()) {
1489        headers.insert(DEADLINE_CLASS_METADATA, value);
1490    }
1491    if let Ok(value) = http::HeaderValue::from_str(options.outcome.as_str()) {
1492        headers.insert(DEADLINE_OUTCOME_METADATA, value);
1493    }
1494}
1495
1496fn sanitize_response_deadline_headers(headers: &mut http::HeaderMap, options: GrpcCallOptions) {
1497    let has_terminal_status = headers.contains_key("grpc-status");
1498    headers.remove(DEADLINE_SOURCE_METADATA);
1499    headers.remove(DEADLINE_CLASS_METADATA);
1500    headers.remove(DEADLINE_OUTCOME_METADATA);
1501    if has_terminal_status {
1502        annotate_server_deadline_headers(headers, options);
1503    }
1504}
1505
1506macro_rules! generated_method_profiles {
1507    (
1508        credential: [$($credential:literal),* $(,)?],
1509        stream: [$($stream:literal),* $(,)?],
1510        cleanup: [$($cleanup:literal),* $(,)?],
1511        long_read: [$($long_read:literal),* $(,)?],
1512        long_mutation: [$($long_mutation:literal),* $(,)?],
1513        ordinary_read: [$($ordinary_read:literal),* $(,)?],
1514        ordinary_mutation: [$($ordinary_mutation:literal),* $(,)?],
1515    ) => {
1516        fn reviewed_generated_call_options(method: &str) -> Option<GrpcCallOptions> {
1517            Some(match method {
1518                $($credential => GrpcCallOptions::new(
1519                    GrpcTimeoutClass::CredentialSetup,
1520                    GrpcTimeoutOutcome::MutationIndeterminate,
1521                ),)*
1522                $($stream => GrpcCallOptions::stream_setup(),)*
1523                $($cleanup => GrpcCallOptions::cleanup(),)*
1524                $($long_read => GrpcCallOptions::long_read(),)*
1525                $($long_mutation => GrpcCallOptions::new(
1526                    GrpcTimeoutClass::LongUnary,
1527                    GrpcTimeoutOutcome::MutationIndeterminate,
1528                ),)*
1529                $($ordinary_read => GrpcCallOptions::ordinary_read(),)*
1530                $($ordinary_mutation => GrpcCallOptions::ordinary_mutation(),)*
1531                _ => return None,
1532            })
1533        }
1534
1535        #[cfg(test)]
1536        const REVIEWED_GENERATED_METHODS: &[&str] = &[
1537            $($credential,)*
1538            $($stream,)*
1539            $($cleanup,)*
1540            $($long_read,)*
1541            $($long_mutation,)*
1542            $($ordinary_read,)*
1543            $($ordinary_mutation,)*
1544        ];
1545    };
1546}
1547
1548generated_method_profiles! {
1549    credential: [
1550        "AccountLocalLinkNewChallenge",
1551        "AccountLocalLinkSolveChallenge",
1552        "WalletCreateSession",
1553    ],
1554    stream: [
1555        "ListenSessionEvents",
1556    ],
1557    cleanup: [
1558        "WalletCloseSession",
1559        "AccountMigrateCancel",
1560        "SpaceJoinCancel",
1561        "ObjectClose",
1562        "ObjectCrossSpaceSearchUnsubscribe",
1563        "ObjectSearchUnsubscribe",
1564        "FileDiscardPreload",
1565        "FileCacheCancelDownload",
1566        "ProcessCancel",
1567        "ProcessUnsubscribe",
1568        "ChatUnsubscribe",
1569        "ChatUnsubscribeFromMessagePreviews",
1570    ],
1571    long_read: [
1572        "WorkspaceExport",
1573        "ObjectListExport",
1574        "ObjectExport",
1575        "TemplateExportAll",
1576        "BlockExport",
1577        "DebugExportLocalstore",
1578        "DebugExportReport",
1579        "FileDownload",
1580    ],
1581    long_mutation: [
1582        "AccountRecoverFromLegacyExport",
1583        "ObjectImport",
1584        "ObjectImportUseCase",
1585        "ObjectImportExperience",
1586        "BlockUpload",
1587        "FileUpload",
1588    ],
1589    ordinary_read: [
1590        "AppGetVersion",
1591        "AccountLocalLinkListApps",
1592        "WorkspaceGetCurrent",
1593        "WorkspaceGetAll",
1594        "SpaceInviteGetCurrent",
1595        "SpaceInviteGetGuest",
1596        "SpaceInviteView",
1597        "PublishingList",
1598        "PublishingResolveUri",
1599        "PublishingGetStatus",
1600        "ObjectShow",
1601        "ObjectGraph",
1602        "ObjectSearch",
1603        "ObjectSearchWithMeta",
1604        "ObjectCleanupSuggestions",
1605        "ObjectImportList",
1606        "ObjectImportNotionValidateToken",
1607        "ObjectDateByTimestamp",
1608        "RelationOptions",
1609        "RelationListWithValue",
1610        "ObjectRelationListAvailable",
1611        "ObjectTypeListConflictingRelations",
1612        "HistoryShowVersion",
1613        "HistoryGetVersions",
1614        "HistoryDiffVersions",
1615        "FileSpaceUsage",
1616        "FileNodeUsage",
1617        "NavigationListObjects",
1618        "NavigationGetObjectInfoWithLinks",
1619        "TemplateGetPlaceholders",
1620        "LinkPreview",
1621        "UnsplashSearch",
1622        "UnsplashDownload",
1623        "GalleryDownloadManifest",
1624        "GalleryDownloadIndex",
1625        "BlockPreview",
1626        "DebugStat",
1627        "DebugTree",
1628        "DebugTreeHeads",
1629        "DebugSpaceSummary",
1630        "DebugStackGoroutines",
1631        "DebugPing",
1632        "DebugSubscriptions",
1633        "DebugOpenedObjects",
1634        "DebugAccountSelectTrace",
1635        "DebugAnystoreObjectChanges",
1636        "DebugNetCheck",
1637        "NotificationList",
1638        "MembershipGetStatus",
1639        "MembershipIsNameValid",
1640        "MembershipGetPortalLinkUrl",
1641        "MembershipGetVerificationEmailStatus",
1642        "MembershipGetTiers",
1643        "MembershipCodeGetInfo",
1644        "MembershipV2GetProducts",
1645        "MembershipV2GetStatus",
1646        "MembershipV2GetPortalLink",
1647        "MembershipV2AnyNameIsValid",
1648        "MembershipV2CartGet",
1649        "NameServiceUserAccountGet",
1650        "NameServiceResolveName",
1651        "NameServiceResolveAnyId",
1652        "DeviceList",
1653        "ChatGetMessages",
1654        "ChatGetMessagesByIds",
1655        "ChatUnreadMessages",
1656        "ChatReadReactions",
1657        "ChatSearch",
1658        "ChatGetPinnedMessages",
1659        "AIWritingTools",
1660        "AIAutofill",
1661        "AIListSummary",
1662    ],
1663    ordinary_mutation: [
1664        "AppSetDeviceState",
1665        "AppShutdown",
1666        "WalletCreate",
1667        "WalletRecover",
1668        "WalletConvert",
1669        "AccountLocalLinkCreateApp",
1670        "AccountLocalLinkRevokeApp",
1671        "WorkspaceCreate",
1672        "WorkspaceOpen",
1673        "WorkspaceObjectAdd",
1674        "WorkspaceObjectListAdd",
1675        "WorkspaceObjectListRemove",
1676        "WorkspaceSelect",
1677        "WorkspaceSetInfo",
1678        "WorkspaceSetHomepage",
1679        "AccountRecover",
1680        "AccountMigrate",
1681        "AccountCreate",
1682        "AccountDelete",
1683        "AccountPreloadRemainingSpaces",
1684        "AccountRevertDeletion",
1685        "AccountSelect",
1686        "AccountEnableLocalNetworkSync",
1687        "AccountChangeJsonApiAddr",
1688        "AccountStop",
1689        "AccountMove",
1690        "AccountConfigUpdate",
1691        "AccountChangeNetworkConfigAndRestart",
1692        "SpaceDelete",
1693        "SpaceInviteGenerate",
1694        "SpaceInviteChange",
1695        "SpaceInviteRevoke",
1696        "SpaceJoin",
1697        "SpaceStopSharing",
1698        "SpaceRequestApprove",
1699        "SpaceRequestDecline",
1700        "SpaceLeaveApprove",
1701        "SpaceMakeShareable",
1702        "SpaceParticipantRemove",
1703        "SpaceParticipantPermissionsChange",
1704        "SpaceSetOrder",
1705        "SpaceUnsetOrder",
1706        "SpaceChangeOwnership",
1707        "SpaceDeleteCorruptedBackup",
1708        "SpaceParticipantsAddList",
1709        "PublishingCreate",
1710        "PublishingRemove",
1711        "ObjectOpen",
1712        "ObjectRefresh",
1713        "ObjectCreate",
1714        "ObjectCreateBookmark",
1715        "ObjectCreateFromUrl",
1716        "ObjectCreateSet",
1717        "ObjectSearchSubscribe",
1718        "ObjectCrossSpaceSearchSubscribe",
1719        "ObjectSubscribeIds",
1720        "ObjectGroupsSubscribe",
1721        "ObjectSetDetails",
1722        "ObjectDuplicate",
1723        "ObjectSetObjectType",
1724        "ObjectSetLayout",
1725        "ObjectSetInternalFlags",
1726        "ObjectSetIsFavorite",
1727        "ObjectSetIsArchived",
1728        "ObjectSetSource",
1729        "ObjectListDuplicate",
1730        "ObjectListDelete",
1731        "ObjectListSetIsArchived",
1732        "ObjectCleanupSuggestionIgnore",
1733        "ObjectListSetIsFavorite",
1734        "ObjectListSetObjectType",
1735        "ObjectListSetDetails",
1736        "ObjectListModifyDetailValues",
1737        "ObjectApplyTemplate",
1738        "ObjectToSet",
1739        "ObjectToCollection",
1740        "ObjectShareByLink",
1741        "ObjectUndo",
1742        "ObjectRedo",
1743        "ObjectBookmarkFetch",
1744        "ObjectCollectionAdd",
1745        "ObjectCollectionRemove",
1746        "ObjectCollectionSort",
1747        "ObjectCreateRelation",
1748        "ObjectCreateRelationOption",
1749        "RelationListRemoveOption",
1750        "RelationOptionSetOrder",
1751        "ObjectRelationAdd",
1752        "ObjectRelationDelete",
1753        "ObjectRelationAddFeatured",
1754        "ObjectRelationRemoveFeatured",
1755        "ObjectCreateObjectType",
1756        "ObjectTypeRelationAdd",
1757        "ObjectTypeRelationRemove",
1758        "ObjectTypeRecommendedRelationsSet",
1759        "ObjectTypeRecommendedFeaturedRelationsSet",
1760        "ObjectTypeResolveLayoutConflicts",
1761        "ObjectTypeSetOrder",
1762        "HistorySetVersion",
1763        "FileSpaceOffload",
1764        "FileReconcile",
1765        "FileListOffload",
1766        "FileDrop",
1767        "FileSetAutoDownload",
1768        "FileCacheDownload",
1769        "FileAutoDownloadSetLimit",
1770        "TemplateCreateFromObject",
1771        "TemplateClone",
1772        "TemplateSetPlaceholders",
1773        "TemplateDeletePlaceholders",
1774        "BlockReplace",
1775        "BlockCreate",
1776        "BlockSplit",
1777        "BlockMerge",
1778        "BlockCopy",
1779        "BlockPaste",
1780        "BlockCut",
1781        "BlockSetFields",
1782        "BlockSetCarriage",
1783        "BlockListDelete",
1784        "BlockListMoveToExistingObject",
1785        "BlockListMoveToNewObject",
1786        "BlockListConvertToObjects",
1787        "BlockListSetFields",
1788        "BlockListDuplicate",
1789        "BlockListSetBackgroundColor",
1790        "BlockListSetAlign",
1791        "BlockListSetVerticalAlign",
1792        "BlockListTurnInto",
1793        "BlockTextSetText",
1794        "BlockTextSetColor",
1795        "BlockTextSetStyle",
1796        "BlockTextSetChecked",
1797        "BlockTextSetIcon",
1798        "BlockTextListSetColor",
1799        "BlockTextListSetMark",
1800        "BlockTextListSetStyle",
1801        "BlockTextListClearStyle",
1802        "BlockTextListClearContent",
1803        "BlockFileSetName",
1804        "BlockFileSetTargetObjectId",
1805        "BlockImageSetName",
1806        "BlockVideoSetName",
1807        "BlockFileCreateAndUpload",
1808        "BlockFileListSetStyle",
1809        "BlockDataviewViewCreate",
1810        "BlockDataviewViewDelete",
1811        "BlockDataviewViewUpdate",
1812        "BlockDataviewViewSetActive",
1813        "BlockDataviewViewSetPosition",
1814        "BlockDataviewSetSource",
1815        "BlockDataviewRelationSet",
1816        "BlockDataviewRelationAdd",
1817        "BlockDataviewRelationDelete",
1818        "BlockDataviewGroupOrderUpdate",
1819        "BlockDataviewObjectOrderUpdate",
1820        "BlockDataviewObjectOrderMove",
1821        "BlockDataviewCreateFromExistingObject",
1822        "BlockDataviewFilterAdd",
1823        "BlockDataviewFilterRemove",
1824        "BlockDataviewFilterReplace",
1825        "BlockDataviewFilterSort",
1826        "BlockDataviewSortAdd",
1827        "BlockDataviewSortRemove",
1828        "BlockDataviewSortReplace",
1829        "BlockDataviewSortSort",
1830        "BlockDataviewViewRelationAdd",
1831        "BlockDataviewViewRelationRemove",
1832        "BlockDataviewViewRelationReplace",
1833        "BlockDataviewViewRelationSort",
1834        "BlockTableCreate",
1835        "BlockTableExpand",
1836        "BlockTableRowCreate",
1837        "BlockTableRowDelete",
1838        "BlockTableRowDuplicate",
1839        "BlockTableRowSetHeader",
1840        "BlockTableColumnCreate",
1841        "BlockTableColumnMove",
1842        "BlockTableColumnDelete",
1843        "BlockTableColumnDuplicate",
1844        "BlockTableRowListFill",
1845        "BlockTableRowListClean",
1846        "BlockTableColumnListFill",
1847        "BlockTableSort",
1848        "BlockCreateWidget",
1849        "BlockWidgetSetTargetId",
1850        "BlockWidgetSetLayout",
1851        "BlockWidgetSetLimit",
1852        "BlockWidgetSetViewId",
1853        "BlockLinkCreateWithObject",
1854        "BlockLinkListSetAppearance",
1855        "BlockBookmarkFetch",
1856        "BlockBookmarkCreateAndFetch",
1857        "BlockRelationSetKey",
1858        "BlockRelationAdd",
1859        "BlockDivListSetStyle",
1860        "BlockLatexSetText",
1861        "ProcessSubscribe",
1862        "LogSend",
1863        "DebugRunProfiler",
1864        "DebugCleanupReport",
1865        "InitialSetParameters",
1866        "NotificationReply",
1867        "NotificationTest",
1868        "MembershipRegisterPaymentRequest",
1869        "MembershipGetVerificationEmail",
1870        "MembershipVerifyEmailCode",
1871        "MembershipFinalize",
1872        "MembershipVerifyAppStoreReceipt",
1873        "MembershipCodeRedeem",
1874        "MembershipV2AnyNameAllocate",
1875        "MembershipV2CartUpdate",
1876        "MembershipV2SubscribeToUpdates",
1877        "BroadcastPayloadEvent",
1878        "DeviceSetName",
1879        "DeviceNetworkStateSet",
1880        "ChatAddMessage",
1881        "ChatEditMessageContent",
1882        "ChatToggleMessageReaction",
1883        "ChatDeleteMessage",
1884        "ChatSubscribeLastMessages",
1885        "ChatReadMessages",
1886        "ChatSubscribeToMessagePreviews",
1887        "ObjectChatAdd",
1888        "ObjectAddDiscussion",
1889        "ChatReadAll",
1890        "ChatSetPinnedMessages",
1891        "ChatAddNotificationSubscriber",
1892        "ChatRemoveNotificationSubscriber",
1893        "AIObjectCreateFromUrl",
1894        "PushNotificationRegisterToken",
1895        "PushNotificationSetSpaceMode",
1896        "PushNotificationSetForceModeIds",
1897        "PushNotificationResetIds",
1898    ],
1899}
1900
1901fn inferred_call_options(method: Option<&GrpcMethod<'_>>) -> GrpcCallOptions {
1902    let Some(method) = method else {
1903        return GrpcCallOptions::default();
1904    };
1905    if method.service() != "anytype.ClientCommands" {
1906        return GrpcCallOptions::default();
1907    }
1908    reviewed_generated_call_options(method.method()).unwrap_or_default()
1909}
1910
1911fn metadata_text<'a>(metadata: &'a MetadataMap, key: &'static str) -> Option<&'a str> {
1912    metadata.get(key).and_then(|value| value.to_str().ok())
1913}
1914
1915fn error_chain_contains_timeout(error: &(dyn StdError + 'static)) -> bool {
1916    let mut source = Some(error);
1917    while let Some(current) = source {
1918        if current.downcast_ref::<TimeoutExpired>().is_some() {
1919            return true;
1920        }
1921        source = current.source();
1922    }
1923    false
1924}
1925
1926fn checked_deadline(
1927    start: tokio::time::Instant,
1928    duration: Option<Duration>,
1929) -> Result<Option<tokio::time::Instant>, GrpcTimeoutConfigError> {
1930    duration
1931        .map(|duration| {
1932            start
1933                .checked_add(duration)
1934                .ok_or(GrpcTimeoutConfigError::UnrepresentableDeadline)
1935        })
1936        .transpose()
1937}
1938
1939fn earliest_stream_boundary(
1940    idle: Option<tokio::time::Instant>,
1941    lifetime: Option<tokio::time::Instant>,
1942    enclosing: Option<tokio::time::Instant>,
1943) -> Option<(tokio::time::Instant, GrpcTimeoutClass)> {
1944    [
1945        idle.map(|deadline| (deadline, GrpcTimeoutClass::StreamIdle)),
1946        lifetime.map(|deadline| (deadline, GrpcTimeoutClass::StreamLifetime)),
1947        enclosing.map(|deadline| (deadline, GrpcTimeoutClass::StreamLifetime)),
1948    ]
1949    .into_iter()
1950    .flatten()
1951    .min_by_key(|(deadline, _)| *deadline)
1952}
1953
1954#[cfg(test)]
1955mod tests {
1956    use std::{
1957        convert::Infallible,
1958        sync::{
1959            Arc, Mutex,
1960            atomic::{AtomicUsize, Ordering},
1961        },
1962    };
1963
1964    use tokio::sync::oneshot;
1965
1966    use super::*;
1967
1968    #[derive(Clone)]
1969    struct ScriptService {
1970        calls: Arc<AtomicUsize>,
1971        header: Arc<Mutex<Option<oneshot::Sender<Option<String>>>>>,
1972    }
1973
1974    #[derive(Debug)]
1975    struct PendingBody;
1976
1977    impl HttpBody for PendingBody {
1978        type Data = tonic::codegen::Bytes;
1979        type Error = Infallible;
1980
1981        fn poll_frame(
1982            self: Pin<&mut Self>,
1983            _context: &mut Context<'_>,
1984        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
1985            Poll::Pending
1986        }
1987
1988        fn is_end_stream(&self) -> bool {
1989            false
1990        }
1991
1992        fn size_hint(&self) -> SizeHint {
1993            SizeHint::default()
1994        }
1995    }
1996
1997    struct StatusErrorBody {
1998        status: Option<Status>,
1999    }
2000
2001    #[derive(Debug)]
2002    struct HostileTransportError(&'static str);
2003
2004    impl fmt::Display for HostileTransportError {
2005        fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2006            formatter.write_str(self.0)
2007        }
2008    }
2009
2010    impl StdError for HostileTransportError {}
2011
2012    impl HttpBody for StatusErrorBody {
2013        type Data = tonic::codegen::Bytes;
2014        type Error = Status;
2015
2016        fn poll_frame(
2017            mut self: Pin<&mut Self>,
2018            _context: &mut Context<'_>,
2019        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
2020            Poll::Ready(self.status.take().map(Err))
2021        }
2022    }
2023
2024    #[derive(Clone)]
2025    struct StatusReadyFailureService;
2026
2027    impl Service<http::Request<Body>> for StatusReadyFailureService {
2028        type Response = http::Response<Body>;
2029        type Error = Status;
2030        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2031
2032        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2033            Poll::Ready(Err(Status::unavailable("HOSTILE_READINESS_SECRET")))
2034        }
2035
2036        fn call(&mut self, _request: http::Request<Body>) -> Self::Future {
2037            std::future::ready(Ok(http::Response::new(Body::empty())))
2038        }
2039    }
2040
2041    #[derive(Debug)]
2042    struct TrailerBody {
2043        frame: Option<Frame<tonic::codegen::Bytes>>,
2044    }
2045
2046    impl HttpBody for TrailerBody {
2047        type Data = tonic::codegen::Bytes;
2048        type Error = Infallible;
2049
2050        fn poll_frame(
2051            mut self: Pin<&mut Self>,
2052            _context: &mut Context<'_>,
2053        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
2054            Poll::Ready(self.frame.take().map(Ok))
2055        }
2056
2057        fn is_end_stream(&self) -> bool {
2058            self.frame.is_none()
2059        }
2060
2061        fn size_hint(&self) -> SizeHint {
2062            SizeHint::default()
2063        }
2064    }
2065
2066    #[derive(Debug)]
2067    struct SignaledDataBody {
2068        ready: oneshot::Receiver<()>,
2069        data: tonic::codegen::Bytes,
2070        emitted: bool,
2071    }
2072
2073    impl HttpBody for SignaledDataBody {
2074        type Data = tonic::codegen::Bytes;
2075        type Error = Infallible;
2076
2077        fn poll_frame(
2078            mut self: Pin<&mut Self>,
2079            context: &mut Context<'_>,
2080        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
2081            if self.emitted {
2082                return Poll::Pending;
2083            }
2084            match Pin::new(&mut self.ready).poll(context) {
2085                Poll::Ready(_) => {
2086                    self.emitted = true;
2087                    let data = std::mem::take(&mut self.data);
2088                    Poll::Ready(Some(Ok(Frame::data(data))))
2089                }
2090                Poll::Pending => Poll::Pending,
2091            }
2092        }
2093
2094        fn is_end_stream(&self) -> bool {
2095            false
2096        }
2097
2098        fn size_hint(&self) -> SizeHint {
2099            SizeHint::default()
2100        }
2101    }
2102
2103    #[derive(Clone)]
2104    struct PendingReadyService {
2105        calls: Arc<AtomicUsize>,
2106    }
2107
2108    impl Service<http::Request<Body>> for PendingReadyService {
2109        type Response = http::Response<Body>;
2110        type Error = Infallible;
2111        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2112
2113        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2114            Poll::Pending
2115        }
2116
2117        fn call(&mut self, _request: http::Request<Body>) -> Self::Future {
2118            self.calls.fetch_add(1, Ordering::Relaxed);
2119            std::future::ready(Ok(http::Response::new(Body::empty())))
2120        }
2121    }
2122
2123    #[derive(Clone)]
2124    struct ReadyAtDeadlineService {
2125        calls: Arc<AtomicUsize>,
2126        ready: Arc<Mutex<Pin<Box<tokio::time::Sleep>>>>,
2127        header: Arc<Mutex<Option<oneshot::Sender<Option<String>>>>>,
2128    }
2129
2130    impl Service<http::Request<Body>> for ReadyAtDeadlineService {
2131        type Response = http::Response<Body>;
2132        type Error = Infallible;
2133        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2134
2135        fn poll_ready(&mut self, context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2136            let Ok(mut ready) = self.ready.lock() else {
2137                return Poll::Pending;
2138            };
2139            ready.as_mut().poll(context).map(|()| Ok(()))
2140        }
2141
2142        fn call(&mut self, request: http::Request<Body>) -> Self::Future {
2143            self.calls.fetch_add(1, Ordering::Relaxed);
2144            if let Ok(mut slot) = self.header.lock()
2145                && let Some(sender) = slot.take()
2146            {
2147                let value = request
2148                    .headers()
2149                    .get(GRPC_TIMEOUT_HEADER)
2150                    .and_then(|value| value.to_str().ok())
2151                    .map(str::to_owned);
2152                let _ = sender.send(value);
2153            }
2154            std::future::ready(Ok(http::Response::new(Body::empty())))
2155        }
2156    }
2157
2158    struct DelayedGrpcBody {
2159        delay: Pin<Box<tokio::time::Sleep>>,
2160        state: u8,
2161    }
2162
2163    impl HttpBody for DelayedGrpcBody {
2164        type Data = tonic::codegen::Bytes;
2165        type Error = Infallible;
2166
2167        fn poll_frame(
2168            mut self: Pin<&mut Self>,
2169            context: &mut Context<'_>,
2170        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
2171            match self.state {
2172                0 => {
2173                    if self.delay.as_mut().poll(context).is_pending() {
2174                        return Poll::Pending;
2175                    }
2176                    self.state = 1;
2177                    Poll::Ready(Some(Ok(Frame::data(tonic::codegen::Bytes::from_static(
2178                        &[0, 0, 0, 0, 0],
2179                    )))))
2180                }
2181                1 => {
2182                    self.state = 2;
2183                    let mut trailers = http::HeaderMap::new();
2184                    trailers.insert("grpc-status", http::HeaderValue::from_static("0"));
2185                    Poll::Ready(Some(Ok(Frame::trailers(trailers))))
2186                }
2187                _ => Poll::Ready(None),
2188            }
2189        }
2190
2191        fn is_end_stream(&self) -> bool {
2192            self.state >= 2
2193        }
2194
2195        fn size_hint(&self) -> SizeHint {
2196            SizeHint::default()
2197        }
2198    }
2199
2200    #[derive(Clone)]
2201    struct DelayedStreamService {
2202        header: Arc<Mutex<Option<oneshot::Sender<Option<String>>>>>,
2203    }
2204
2205    impl tonic::server::NamedService for DelayedStreamService {
2206        const NAME: &'static str = "anytype.ClientCommands";
2207    }
2208
2209    impl Service<http::Request<Body>> for DelayedStreamService {
2210        type Response = http::Response<Body>;
2211        type Error = Infallible;
2212        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2213
2214        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2215            Poll::Ready(Ok(()))
2216        }
2217
2218        fn call(&mut self, request: http::Request<Body>) -> Self::Future {
2219            if let Ok(mut slot) = self.header.lock()
2220                && let Some(sender) = slot.take()
2221            {
2222                let header = request
2223                    .headers()
2224                    .get(GRPC_TIMEOUT_HEADER)
2225                    .and_then(|value| value.to_str().ok())
2226                    .map(str::to_owned);
2227                let _ = sender.send(header);
2228            }
2229            let body = DelayedGrpcBody {
2230                delay: Box::pin(tokio::time::sleep(Duration::from_millis(1_100))),
2231                state: 0,
2232            };
2233            let response = http::Response::builder()
2234                .status(200)
2235                .header("content-type", "application/grpc")
2236                .body(Body::new(body))
2237                .unwrap_or_else(|_| http::Response::new(Body::empty()));
2238            std::future::ready(Ok(response))
2239        }
2240    }
2241
2242    struct ListenerIncoming {
2243        listener: tokio::net::TcpListener,
2244    }
2245
2246    impl tonic::codegen::tokio_stream::Stream for ListenerIncoming {
2247        type Item = std::io::Result<tokio::net::TcpStream>;
2248
2249        fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
2250            self.listener
2251                .poll_accept(context)
2252                .map(|result| Some(result.map(|(stream, _)| stream)))
2253        }
2254    }
2255
2256    #[derive(Clone)]
2257    struct ServerDeadlineService;
2258
2259    impl Service<http::Request<Body>> for ServerDeadlineService {
2260        type Response = http::Response<Body>;
2261        type Error = Infallible;
2262        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2263
2264        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2265            Poll::Ready(Ok(()))
2266        }
2267
2268        fn call(&mut self, _request: http::Request<Body>) -> Self::Future {
2269            let mut trailers = http::HeaderMap::new();
2270            trailers.insert("grpc-status", http::HeaderValue::from_static("4"));
2271            trailers.insert(
2272                "grpc-message",
2273                http::HeaderValue::from_static("UNTRUSTED_SERVER_SECRET"),
2274            );
2275            trailers.insert(
2276                DEADLINE_SOURCE_METADATA,
2277                http::HeaderValue::from_static("local"),
2278            );
2279            trailers.insert(
2280                DEADLINE_OUTCOME_METADATA,
2281                http::HeaderValue::from_static("mutation_indeterminate"),
2282            );
2283            let body = Body::new(TrailerBody {
2284                frame: Some(Frame::trailers(trailers)),
2285            });
2286            let response = http::Response::builder()
2287                .status(200)
2288                .header("content-type", "application/grpc")
2289                .body(body)
2290                .unwrap_or_else(|_| http::Response::new(Body::empty()));
2291            std::future::ready(Ok(response))
2292        }
2293    }
2294
2295    #[derive(Clone)]
2296    struct HeaderThenStallService {
2297        header: Arc<Mutex<Option<oneshot::Sender<Option<String>>>>>,
2298    }
2299
2300    impl Service<http::Request<Body>> for HeaderThenStallService {
2301        type Response = http::Response<Body>;
2302        type Error = Infallible;
2303        type Future = std::future::Ready<Result<Self::Response, Self::Error>>;
2304
2305        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2306            Poll::Ready(Ok(()))
2307        }
2308
2309        fn call(&mut self, request: http::Request<Body>) -> Self::Future {
2310            if let Ok(mut slot) = self.header.lock()
2311                && let Some(sender) = slot.take()
2312            {
2313                let value = request
2314                    .headers()
2315                    .get(GRPC_TIMEOUT_HEADER)
2316                    .and_then(|value| value.to_str().ok())
2317                    .map(str::to_owned);
2318                let _ = sender.send(value);
2319            }
2320            let response = http::Response::builder()
2321                .status(200)
2322                .header("content-type", "application/grpc")
2323                .body(Body::new(PendingBody))
2324                .unwrap_or_else(|_| http::Response::new(Body::empty()));
2325            std::future::ready(Ok(response))
2326        }
2327    }
2328
2329    fn header_then_stall_service() -> (HeaderThenStallService, oneshot::Receiver<Option<String>>) {
2330        let (sender, receiver) = oneshot::channel();
2331        (
2332            HeaderThenStallService {
2333                header: Arc::new(Mutex::new(Some(sender))),
2334            },
2335            receiver,
2336        )
2337    }
2338
2339    impl Service<http::Request<Body>> for ScriptService {
2340        type Response = http::Response<Body>;
2341        type Error = Infallible;
2342        type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
2343
2344        fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
2345            Poll::Ready(Ok(()))
2346        }
2347
2348        fn call(&mut self, request: http::Request<Body>) -> Self::Future {
2349            self.calls.fetch_add(1, Ordering::Relaxed);
2350            if let Ok(mut slot) = self.header.lock()
2351                && let Some(sender) = slot.take()
2352            {
2353                let value = request
2354                    .headers()
2355                    .get(GRPC_TIMEOUT_HEADER)
2356                    .and_then(|value| value.to_str().ok())
2357                    .map(str::to_owned);
2358                let _ = sender.send(value);
2359            }
2360            Box::pin(std::future::pending())
2361        }
2362    }
2363
2364    fn scripted_service() -> (
2365        ScriptService,
2366        Arc<AtomicUsize>,
2367        oneshot::Receiver<Option<String>>,
2368    ) {
2369        let calls = Arc::new(AtomicUsize::new(0));
2370        let (sender, receiver) = oneshot::channel();
2371        (
2372            ScriptService {
2373                calls: calls.clone(),
2374                header: Arc::new(Mutex::new(Some(sender))),
2375            },
2376            calls,
2377            receiver,
2378        )
2379    }
2380
2381    #[test]
2382    fn defaults_and_finite_ranges_match_policy() {
2383        let policy = GrpcTimeoutPolicy::default();
2384        assert_eq!(policy.credential_setup, Some(Duration::from_secs(120)));
2385        assert_eq!(policy.ordinary_unary, Some(Duration::from_secs(120)));
2386        assert_eq!(policy.long_unary, Some(Duration::from_secs(1_800)));
2387        assert_eq!(policy.stream_setup, Some(Duration::from_secs(120)));
2388        assert_eq!(policy.stream_idle, None);
2389        assert_eq!(policy.stream_total_lifetime, None);
2390        assert_eq!(policy.cleanup, Some(Duration::from_secs(5)));
2391
2392        for duration in [Duration::ZERO, Duration::from_millis(999)] {
2393            assert!(
2394                GrpcTimeoutPolicy {
2395                    ordinary_unary: Some(duration),
2396                    ..policy
2397                }
2398                .validate()
2399                .is_err()
2400            );
2401        }
2402        assert!(
2403            GrpcTimeoutPolicy {
2404                long_unary: Some(Duration::from_secs(7_201)),
2405                ..policy
2406            }
2407            .validate()
2408            .is_err()
2409        );
2410        assert!(
2411            GrpcTimeoutPolicy {
2412                cleanup: Some(Duration::from_secs(31)),
2413                ..policy
2414            }
2415            .validate()
2416            .is_err()
2417        );
2418    }
2419
2420    #[test]
2421    fn environment_override_has_exact_grammar_and_limited_scope() {
2422        let disabled = GrpcTimeoutPolicy::from_environment(Some(OsString::from("0")))
2423            .expect("zero disables inherited generic profiles");
2424        assert_eq!(disabled.credential_setup, None);
2425        assert_eq!(disabled.ordinary_unary, None);
2426        assert_eq!(disabled.long_unary, None);
2427        assert_eq!(disabled.stream_setup, None);
2428        assert_eq!(disabled.cleanup, Some(Duration::from_secs(5)));
2429
2430        let finite = GrpcTimeoutPolicy::from_environment(Some(OsString::from("17")))
2431            .expect("finite override");
2432        assert_eq!(finite.credential_setup, Some(Duration::from_secs(17)));
2433        assert_eq!(finite.long_unary, Some(Duration::from_secs(17)));
2434        assert_eq!(finite.stream_idle, None);
2435
2436        for malformed in [
2437            "",
2438            "00",
2439            "01",
2440            " 1",
2441            "+1",
2442            "-1",
2443            "1.0",
2444            "3601",
2445            "18446744073709551616",
2446        ] {
2447            assert!(
2448                GrpcTimeoutPolicy::from_environment(Some(OsString::from(malformed))).is_err(),
2449                "accepted {malformed:?}"
2450            );
2451        }
2452    }
2453
2454    #[test]
2455    fn explicit_policy_resolution_returns_the_validated_policy() {
2456        let explicit = GrpcTimeoutPolicy {
2457            credential_setup: None,
2458            ordinary_unary: Some(Duration::from_secs(11)),
2459            long_unary: Some(Duration::from_secs(7_200)),
2460            stream_setup: None,
2461            stream_idle: Some(Duration::from_secs(12)),
2462            stream_total_lifetime: None,
2463            cleanup: Some(Duration::from_secs(30)),
2464        };
2465        assert_eq!(
2466            GrpcTimeoutPolicy::resolve(Some(explicit)).expect("explicit policy"),
2467            explicit
2468        );
2469        assert!(
2470            GrpcDeadlineService::try_new(
2471                (),
2472                GrpcTimeoutPolicy {
2473                    cleanup: Some(Duration::from_secs(31)),
2474                    ..explicit
2475                }
2476            )
2477            .is_err()
2478        );
2479    }
2480
2481    #[test]
2482    fn generated_method_profiles_cover_the_checked_in_client_inventory() {
2483        let generated_source = include_str!("gen/anytype.rs");
2484        let mut generated = generated_source
2485            .split("GrpcMethod::new(")
2486            .skip(1)
2487            .filter_map(|tail| {
2488                let mut quoted = tail.split('"');
2489                let _ = quoted.next()?;
2490                let service = quoted.next()?;
2491                let _ = quoted.next()?;
2492                let method = quoted.next()?;
2493                (service == "anytype.ClientCommands").then_some(method)
2494            })
2495            .collect::<Vec<_>>();
2496        generated.sort_unstable();
2497        generated.dedup();
2498
2499        let reviewed_count = REVIEWED_GENERATED_METHODS.len();
2500        let mut reviewed = REVIEWED_GENERATED_METHODS.to_vec();
2501        reviewed.sort_unstable();
2502        reviewed.dedup();
2503        assert_eq!(
2504            reviewed.len(),
2505            reviewed_count,
2506            "profile authority has duplicates"
2507        );
2508        let generated_without_profile = generated
2509            .iter()
2510            .copied()
2511            .filter(|method| reviewed.binary_search(method).is_err())
2512            .collect::<Vec<_>>();
2513        assert!(
2514            generated_without_profile.is_empty(),
2515            "generated RPCs missing reviewed profiles: {generated_without_profile:?}"
2516        );
2517        let reviewed_without_generated = reviewed
2518            .iter()
2519            .copied()
2520            .filter(|method| generated.binary_search(method).is_err())
2521            .collect::<Vec<_>>();
2522        // A reviewed forward profile may land before its generated snapshot;
2523        // only explicitly named transition methods may be absent locally.
2524        let mut expected_forward_profiles =
2525            ["ObjectCleanupSuggestions", "ObjectCleanupSuggestionIgnore"]
2526                .into_iter()
2527                .filter(|method| generated.binary_search(method).is_err())
2528                .collect::<Vec<_>>();
2529        expected_forward_profiles.sort_unstable();
2530        assert_eq!(
2531            reviewed_without_generated, expected_forward_profiles,
2532            "profile authority contains an unexpected method absent from generated RPCs"
2533        );
2534    }
2535
2536    #[test]
2537    fn generated_method_defaults_are_closed_and_conservative() {
2538        let read = GrpcMethod::new("anytype.ClientCommands", "ObjectShow");
2539        assert_eq!(
2540            inferred_call_options(Some(&read)).outcome,
2541            GrpcTimeoutOutcome::ReadAborted
2542        );
2543        let stream = GrpcMethod::new("anytype.ClientCommands", "ListenSessionEvents");
2544        assert_eq!(
2545            inferred_call_options(Some(&stream)).class,
2546            GrpcTimeoutClass::StreamSetup
2547        );
2548        let import = GrpcMethod::new("anytype.ClientCommands", "ObjectImport");
2549        assert_eq!(
2550            inferred_call_options(Some(&import)),
2551            GrpcCallOptions::new(
2552                GrpcTimeoutClass::LongUnary,
2553                GrpcTimeoutOutcome::MutationIndeterminate,
2554            )
2555        );
2556        let close = GrpcMethod::new("anytype.ClientCommands", "ObjectClose");
2557        assert_eq!(
2558            inferred_call_options(Some(&close)).class,
2559            GrpcTimeoutClass::Cleanup
2560        );
2561        let unknown = GrpcMethod::new("anytype.ClientCommands", "FutureMutation");
2562        assert_eq!(
2563            inferred_call_options(Some(&unknown)).outcome,
2564            GrpcTimeoutOutcome::MutationIndeterminate
2565        );
2566    }
2567
2568    #[test]
2569    fn cleanup_suggestion_methods_have_explicit_read_and_mutation_outcomes() {
2570        let suggestions = GrpcMethod::new("anytype.ClientCommands", "ObjectCleanupSuggestions");
2571        assert_eq!(
2572            inferred_call_options(Some(&suggestions)),
2573            GrpcCallOptions::ordinary_read()
2574        );
2575        let ignore = GrpcMethod::new("anytype.ClientCommands", "ObjectCleanupSuggestionIgnore");
2576        assert_eq!(
2577            inferred_call_options(Some(&ignore)),
2578            GrpcCallOptions::ordinary_mutation()
2579        );
2580    }
2581
2582    #[cfg(unix)]
2583    #[test]
2584    fn environment_rejects_non_unicode() {
2585        use std::os::unix::ffi::OsStringExt;
2586
2587        assert!(GrpcTimeoutPolicy::from_environment(Some(OsString::from_vec(vec![0xff]))).is_err());
2588    }
2589
2590    #[tokio::test(start_paused = true)]
2591    async fn request_header_and_local_deadline_use_the_same_budget() {
2592        let (inner, calls, header) = scripted_service();
2593        let mut service = GrpcDeadlineService::try_new(
2594            inner,
2595            GrpcTimeoutPolicy {
2596                ordinary_unary: Some(Duration::from_secs(10)),
2597                ..GrpcTimeoutPolicy::default()
2598            },
2599        )
2600        .expect("valid policy");
2601        let request = http::Request::new(Body::empty());
2602        let call = tokio::spawn(async move { service.call(request).await });
2603
2604        assert_eq!(
2605            header.await.expect("captured header").as_deref(),
2606            Some("10000000u")
2607        );
2608        assert_eq!(calls.load(Ordering::Relaxed), 1);
2609        tokio::time::advance(Duration::from_secs(10)).await;
2610        let error = call
2611            .await
2612            .expect("service task")
2613            .expect_err("local timeout");
2614        assert!(matches!(
2615            error,
2616            GrpcDeadlineServiceError::Deadline(ref status)
2617                if status.code() == Code::DeadlineExceeded
2618        ));
2619        let source = error.source().expect("payload-free deadline source");
2620        assert!(!format!("{source:?}").contains("SECRET"));
2621    }
2622
2623    #[tokio::test(start_paused = true)]
2624    async fn scoped_enclosing_deadline_caps_generated_header_and_local_wait() {
2625        let (inner, calls, header) = scripted_service();
2626        let mut service = GrpcDeadlineService::try_new(
2627            inner,
2628            GrpcTimeoutPolicy {
2629                ordinary_unary: Some(Duration::from_secs(120)),
2630                ..GrpcTimeoutPolicy::default()
2631            },
2632        )
2633        .expect("valid policy");
2634        let enclosing = GrpcEnclosingDeadline::from_now(Duration::from_secs(3))
2635            .expect("representable enclosing deadline");
2636        let call = tokio::spawn(async move {
2637            scope_grpc_enclosing_deadline(enclosing, async move {
2638                service.call(http::Request::new(Body::empty())).await
2639            })
2640            .await
2641        });
2642
2643        assert_eq!(
2644            header.await.expect("captured header").as_deref(),
2645            Some("3000000u")
2646        );
2647        assert_eq!(calls.load(Ordering::Relaxed), 1);
2648        tokio::time::advance(Duration::from_secs(3)).await;
2649        assert!(matches!(
2650            call.await.expect("service task").expect_err("local timeout"),
2651            GrpcDeadlineServiceError::Deadline(ref status)
2652                if status.code() == Code::DeadlineExceeded
2653        ));
2654    }
2655
2656    #[tokio::test(start_paused = true)]
2657    async fn generated_tonic_client_preserves_local_timeout_classification() {
2658        use crate::{
2659            anytype::{ClientCommandsClient, rpc::account::local_link::list_apps},
2660            deadline::with_grpc_call_options,
2661        };
2662
2663        let (inner, calls, header) = scripted_service();
2664        let service = GrpcDeadlineService::try_new(
2665            inner,
2666            GrpcTimeoutPolicy {
2667                ordinary_unary: Some(Duration::from_secs(4)),
2668                ..GrpcTimeoutPolicy::default()
2669            },
2670        )
2671        .expect("valid policy");
2672        let mut client = ClientCommandsClient::new(service);
2673        let request = with_grpc_call_options(
2674            Request::new(list_apps::Request {}),
2675            GrpcCallOptions::ordinary_read(),
2676        );
2677        let call = tokio::spawn(async move { client.account_local_link_list_apps(request).await });
2678        assert_eq!(
2679            header.await.expect("captured header").as_deref(),
2680            Some("4000000u")
2681        );
2682        assert_eq!(calls.load(Ordering::Relaxed), 1);
2683        tokio::time::advance(Duration::from_secs(4)).await;
2684        let status = call
2685            .await
2686            .expect("generated client task")
2687            .expect_err("local timeout");
2688        let error = GrpcDeadlineError::from_status(
2689            &status,
2690            GrpcTimeoutClass::LongUnary,
2691            GrpcTimeoutOutcome::MutationIndeterminate,
2692            Duration::from_secs(4),
2693        )
2694        .expect("classified deadline");
2695        assert_eq!(error.class, GrpcTimeoutClass::OrdinaryUnary);
2696        assert_eq!(error.outcome, GrpcTimeoutOutcome::ReadAborted);
2697        assert_eq!(error.source, GrpcTimeoutSource::Local);
2698    }
2699
2700    #[tokio::test(start_paused = true)]
2701    async fn unary_deadline_remains_active_after_response_headers() {
2702        use crate::anytype::{ClientCommandsClient, rpc::account::local_link::list_apps};
2703
2704        let (inner, header) = header_then_stall_service();
2705        let service = GrpcDeadlineService::try_new(
2706            inner,
2707            GrpcTimeoutPolicy {
2708                ordinary_unary: Some(Duration::from_secs(6)),
2709                ..GrpcTimeoutPolicy::default()
2710            },
2711        )
2712        .expect("valid policy");
2713        let mut client = ClientCommandsClient::new(service);
2714        let request = with_grpc_call_options(
2715            Request::new(list_apps::Request {}),
2716            GrpcCallOptions::ordinary_read(),
2717        );
2718        let call = tokio::spawn(async move { client.account_local_link_list_apps(request).await });
2719        assert_eq!(
2720            header.await.expect("captured header").as_deref(),
2721            Some("6000000u")
2722        );
2723        tokio::time::advance(Duration::from_secs(6)).await;
2724        let status = call
2725            .await
2726            .expect("generated client task")
2727            .expect_err("stalled unary body must expire");
2728        let error = GrpcDeadlineError::from_status(
2729            &status,
2730            GrpcTimeoutClass::OrdinaryUnary,
2731            GrpcTimeoutOutcome::ReadAborted,
2732            Duration::from_secs(6),
2733        )
2734        .expect("classified deadline");
2735        assert_eq!(error.source, GrpcTimeoutSource::Local);
2736    }
2737
2738    #[tokio::test(start_paused = true)]
2739    async fn successful_stream_headers_disarm_the_setup_deadline() {
2740        use crate::anytype::{ClientCommandsClient, StreamRequest};
2741
2742        let (inner, header) = header_then_stall_service();
2743        let service = GrpcDeadlineService::try_new(
2744            inner,
2745            GrpcTimeoutPolicy {
2746                stream_setup: Some(Duration::from_secs(5)),
2747                ..GrpcTimeoutPolicy::default()
2748            },
2749        )
2750        .expect("valid policy");
2751        let mut client = ClientCommandsClient::new(service);
2752        let request = Request::new(StreamRequest {
2753            token: String::new(),
2754        });
2755        let request = with_grpc_call_options(request, GrpcCallOptions::stream_setup());
2756        let response = client
2757            .listen_session_events(request)
2758            .await
2759            .expect("successful response headers");
2760        assert_eq!(header.await.expect("captured header").as_deref(), None);
2761        let mut stream = response.into_inner();
2762        tokio::time::advance(Duration::from_secs(6)).await;
2763        assert!(
2764            tokio::time::timeout(Duration::from_millis(1), stream.message())
2765                .await
2766                .is_err(),
2767            "setup deadline must not terminate an established stream body"
2768        );
2769    }
2770
2771    #[tokio::test(start_paused = true)]
2772    async fn stream_setup_preserves_a_caller_supplied_whole_call_timeout() {
2773        use crate::anytype::{ClientCommandsClient, StreamRequest};
2774
2775        let (inner, header) = header_then_stall_service();
2776        let service = GrpcDeadlineService::try_new(
2777            inner,
2778            GrpcTimeoutPolicy {
2779                stream_setup: Some(Duration::from_secs(5)),
2780                ..GrpcTimeoutPolicy::default()
2781            },
2782        )
2783        .expect("valid policy");
2784        let mut client = ClientCommandsClient::new(service);
2785        let mut request = Request::new(StreamRequest {
2786            token: String::new(),
2787        });
2788        request.set_timeout(Duration::from_secs(2));
2789        let request = with_grpc_call_options(request, GrpcCallOptions::stream_setup());
2790        let _response = client
2791            .listen_session_events(request)
2792            .await
2793            .expect("successful response headers");
2794        assert_eq!(
2795            header.await.expect("captured caller timeout").as_deref(),
2796            Some("2000000u")
2797        );
2798    }
2799
2800    #[tokio::test]
2801    async fn tonic_server_stream_remains_established_beyond_setup_budget() {
2802        use crate::anytype::{ClientCommandsClient, StreamRequest};
2803
2804        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
2805            .await
2806            .expect("bind tonic test server");
2807        let address = listener.local_addr().expect("tonic test address");
2808        let (header_sender, header_receiver) = oneshot::channel();
2809        let service = DelayedStreamService {
2810            header: Arc::new(Mutex::new(Some(header_sender))),
2811        };
2812        let (shutdown_sender, shutdown_receiver) = oneshot::channel();
2813        let server = tokio::spawn(async move {
2814            tonic::transport::Server::builder()
2815                .add_service(service)
2816                .serve_with_incoming_shutdown(ListenerIncoming { listener }, async move {
2817                    let _ = shutdown_receiver.await;
2818                })
2819                .await
2820        });
2821        let channel = tonic::transport::Endpoint::from_shared(format!("http://{address}"))
2822            .expect("valid tonic test endpoint")
2823            .connect()
2824            .await
2825            .expect("connect tonic test client");
2826        let deadline_service = GrpcDeadlineService::try_new(
2827            channel,
2828            GrpcTimeoutPolicy {
2829                stream_setup: Some(Duration::from_secs(1)),
2830                ..GrpcTimeoutPolicy::default()
2831            },
2832        )
2833        .expect("valid stream policy");
2834        let mut client = ClientCommandsClient::new(deadline_service);
2835        let request = with_grpc_call_options(
2836            Request::new(StreamRequest {
2837                token: String::new(),
2838            }),
2839            GrpcCallOptions::stream_setup(),
2840        );
2841        let response = client
2842            .listen_session_events(request)
2843            .await
2844            .expect("stream response headers");
2845        assert_eq!(
2846            header_receiver.await.expect("captured server header"),
2847            None,
2848            "stream setup must remain a local-only header boundary"
2849        );
2850        assert!(
2851            response
2852                .extensions()
2853                .get::<GrpcTransportProgress>()
2854                .is_some(),
2855            "tonic response extensions must carry raw transport progress"
2856        );
2857
2858        let mut stream = response.into_inner();
2859        let message = tokio::time::timeout(Duration::from_secs(2), stream.message())
2860            .await
2861            .expect("established stream wait")
2862            .expect("stream transport")
2863            .expect("delayed event");
2864        assert!(message.context_id.is_empty());
2865
2866        let _ = shutdown_sender.send(());
2867        server
2868            .await
2869            .expect("tonic server task")
2870            .expect("tonic server shutdown");
2871    }
2872
2873    #[tokio::test]
2874    async fn server_deadline_uses_request_classification_and_cannot_spoof_local_source() {
2875        use crate::anytype::{ClientCommandsClient, rpc::account::local_link::list_apps};
2876
2877        let service =
2878            GrpcDeadlineService::try_new(ServerDeadlineService, GrpcTimeoutPolicy::default())
2879                .expect("valid policy");
2880        let mut client = ClientCommandsClient::new(service);
2881        let request = with_grpc_call_options(
2882            Request::new(list_apps::Request {}),
2883            GrpcCallOptions::ordinary_read(),
2884        );
2885        let status = client
2886            .account_local_link_list_apps(request)
2887            .await
2888            .expect_err("server deadline");
2889        let error = GrpcDeadlineError::from_status(
2890            &status,
2891            GrpcTimeoutClass::LongUnary,
2892            GrpcTimeoutOutcome::MutationIndeterminate,
2893            Duration::from_millis(1),
2894        )
2895        .expect("classified deadline");
2896        assert_eq!(error.class, GrpcTimeoutClass::OrdinaryUnary);
2897        assert_eq!(error.outcome, GrpcTimeoutOutcome::ReadAborted);
2898        assert_eq!(error.source, GrpcTimeoutSource::Server);
2899        assert!(!error.to_string().contains("UNTRUSTED_SERVER_SECRET"));
2900    }
2901
2902    #[tokio::test(start_paused = true)]
2903    async fn enclosing_deadline_and_existing_tighter_timeout_win() {
2904        let now = tokio::time::Instant::now();
2905        let enclosing = GrpcEnclosingDeadline::from_instant(now + Duration::from_secs(7));
2906        let selected = select_deadline(Some(Duration::from_secs(120)), Some(enclosing), None, now)
2907            .expect("valid deadline");
2908        assert_eq!(selected, Some(Duration::from_secs(7)));
2909
2910        let header = http::HeaderValue::from_static("3000000u");
2911        let selected = select_deadline(
2912            Some(Duration::from_secs(120)),
2913            Some(enclosing),
2914            Some(&header),
2915            now,
2916        )
2917        .expect("valid deadline");
2918        assert_eq!(selected, Some(Duration::from_secs(3)));
2919
2920        let exhausted = http::HeaderValue::from_static("0n");
2921        assert_eq!(
2922            select_deadline(None, None, Some(&exhausted), now),
2923            Err(SelectDeadlineError::Expired)
2924        );
2925    }
2926
2927    #[tokio::test(start_paused = true)]
2928    async fn expired_enclosing_deadline_prevents_dispatch() {
2929        let (inner, calls, _header) = scripted_service();
2930        let mut service = GrpcDeadlineService::try_new(inner, GrpcTimeoutPolicy::default())
2931            .expect("valid policy");
2932        let mut request = http::Request::new(Body::empty());
2933        request
2934            .extensions_mut()
2935            .insert(GrpcCallOptions::ordinary_mutation().enclosing(
2936                GrpcEnclosingDeadline::from_instant(tokio::time::Instant::now()),
2937            ));
2938        let error = service.call(request).await.expect_err("expired deadline");
2939        assert_eq!(calls.load(Ordering::Relaxed), 0);
2940        let status = match &error {
2941            GrpcDeadlineServiceError::Deadline(status) => status,
2942            other => panic!("unexpected service error: {other:?}"),
2943        };
2944        assert_eq!(status.code(), Code::DeadlineExceeded);
2945        let classified = GrpcDeadlineError::from_status(
2946            status,
2947            GrpcTimeoutClass::OrdinaryUnary,
2948            GrpcTimeoutOutcome::MutationIndeterminate,
2949            Duration::ZERO,
2950        );
2951        assert_eq!(
2952            classified.map(|error| error.outcome),
2953            Some(GrpcTimeoutOutcome::ReadAborted)
2954        );
2955        assert!(error.source().is_some());
2956    }
2957
2958    #[tokio::test(start_paused = true)]
2959    async fn readiness_wait_consumes_cleanup_and_enclosing_budget() {
2960        let calls = Arc::new(AtomicUsize::new(0));
2961        let inner = PendingReadyService {
2962            calls: calls.clone(),
2963        };
2964        let mut service = GrpcDeadlineService::try_new(inner, GrpcTimeoutPolicy::default())
2965            .expect("valid policy");
2966        let enclosing = GrpcEnclosingDeadline::from_now(Duration::from_secs(2))
2967            .expect("valid enclosing deadline");
2968        let options = GrpcCallOptions::cleanup().enclosing(enclosing);
2969        let mut request = http::Request::new(Body::empty());
2970        request.extensions_mut().insert(options);
2971        let call = tokio::spawn(async move { service.call(request).await });
2972
2973        tokio::time::advance(Duration::from_secs(2)).await;
2974        let error = call
2975            .await
2976            .expect("service task")
2977            .expect_err("pending readiness must expire");
2978        let status = match error {
2979            GrpcDeadlineServiceError::Deadline(status) => status,
2980            other => panic!("unexpected service error: {other:?}"),
2981        };
2982        let classified = GrpcDeadlineError::from_status(
2983            &status,
2984            GrpcTimeoutClass::Cleanup,
2985            GrpcTimeoutOutcome::MutationIndeterminate,
2986            Duration::from_secs(2),
2987        )
2988        .expect("classified readiness deadline");
2989        assert_eq!(classified.class, GrpcTimeoutClass::Cleanup);
2990        assert_eq!(classified.outcome, GrpcTimeoutOutcome::ReadAborted);
2991        assert_eq!(calls.load(Ordering::Relaxed), 0);
2992    }
2993
2994    #[tokio::test(start_paused = true)]
2995    async fn stream_setup_ready_at_exact_deadline_does_not_dispatch() {
2996        let calls = Arc::new(AtomicUsize::new(0));
2997        let ready_at = tokio::time::Instant::now() + Duration::from_secs(1);
2998        let inner = ReadyAtDeadlineService {
2999            calls: calls.clone(),
3000            ready: Arc::new(Mutex::new(Box::pin(tokio::time::sleep_until(ready_at)))),
3001            header: Arc::new(Mutex::new(None)),
3002        };
3003        let mut service = GrpcDeadlineService::try_new(
3004            inner,
3005            GrpcTimeoutPolicy {
3006                stream_setup: Some(Duration::from_secs(1)),
3007                ..GrpcTimeoutPolicy::default()
3008            },
3009        )
3010        .expect("valid policy");
3011        let mut request = http::Request::new(Body::empty());
3012        request
3013            .extensions_mut()
3014            .insert(GrpcCallOptions::stream_setup());
3015        let call = tokio::spawn(async move { service.call(request).await });
3016
3017        tokio::task::yield_now().await;
3018        tokio::time::advance(Duration::from_secs(1)).await;
3019        let error = call
3020            .await
3021            .expect("service task")
3022            .expect_err("exhausted setup deadline");
3023        assert!(matches!(error, GrpcDeadlineServiceError::Deadline(_)));
3024        assert_eq!(calls.load(Ordering::Relaxed), 0);
3025    }
3026
3027    #[tokio::test(start_paused = true)]
3028    async fn caller_stream_timeout_propagates_only_its_remaining_absolute_budget() {
3029        let calls = Arc::new(AtomicUsize::new(0));
3030        let ready_at = tokio::time::Instant::now() + Duration::from_millis(1_900);
3031        let (header_sender, header_receiver) = oneshot::channel();
3032        let inner = ReadyAtDeadlineService {
3033            calls: calls.clone(),
3034            ready: Arc::new(Mutex::new(Box::pin(tokio::time::sleep_until(ready_at)))),
3035            header: Arc::new(Mutex::new(Some(header_sender))),
3036        };
3037        let mut service = GrpcDeadlineService::try_new(
3038            inner,
3039            GrpcTimeoutPolicy {
3040                stream_setup: Some(Duration::from_secs(5)),
3041                ..GrpcTimeoutPolicy::default()
3042            },
3043        )
3044        .expect("valid policy");
3045        let mut request = http::Request::new(Body::empty());
3046        request
3047            .headers_mut()
3048            .insert(GRPC_TIMEOUT_HEADER, http::HeaderValue::from_static("2S"));
3049        request
3050            .extensions_mut()
3051            .insert(GrpcCallOptions::stream_setup());
3052        let call = tokio::spawn(async move { service.call(request).await });
3053
3054        tokio::task::yield_now().await;
3055        tokio::time::advance(Duration::from_millis(1_900)).await;
3056        call.await
3057            .expect("service task")
3058            .expect("remaining caller timeout dispatches");
3059        let header = header_receiver
3060            .await
3061            .expect("captured remaining caller timeout")
3062            .expect("caller timeout header preserved");
3063        let remaining = parse_grpc_timeout(
3064            &http::HeaderValue::from_str(&header).expect("valid propagated header"),
3065        )
3066        .expect("parse propagated header")
3067        .expect("finite propagated header");
3068        assert!((Duration::from_millis(99)..=Duration::from_millis(100)).contains(&remaining));
3069        assert_eq!(calls.load(Ordering::Relaxed), 1);
3070    }
3071
3072    #[tokio::test(start_paused = true)]
3073    async fn expired_body_deadline_precedes_queued_data_and_trailers() {
3074        let options = GrpcCallOptions::ordinary_read();
3075        for frame in [
3076            Frame::data(tonic::codegen::Bytes::from_static(b"queued")),
3077            Frame::trailers(http::HeaderMap::new()),
3078        ] {
3079            let deadline = tokio::time::Instant::now() + Duration::from_secs(1);
3080            let mut body = GrpcDeadlineBody::new(
3081                TrailerBody { frame: Some(frame) },
3082                Some(deadline),
3083                None,
3084                options,
3085                tokio::time::Instant::now(),
3086            );
3087            tokio::time::advance(Duration::from_secs(1)).await;
3088            let frame = std::future::poll_fn(|context| Pin::new(&mut body).poll_frame(context))
3089                .await
3090                .expect("deadline frame");
3091            assert!(matches!(frame, Err(GrpcDeadlineServiceError::Deadline(_))));
3092        }
3093    }
3094
3095    #[tokio::test(start_paused = true)]
3096    async fn raw_body_progress_resets_idle_before_a_message_decodes() {
3097        let progress = GrpcTransportProgress::new();
3098        let (ready, receiver) = oneshot::channel();
3099        let body = SignaledDataBody {
3100            ready: receiver,
3101            data: tonic::codegen::Bytes::from_static(b"partial"),
3102            emitted: false,
3103        };
3104        let mut body = GrpcDeadlineBody::new(
3105            body,
3106            None,
3107            Some(progress.clone()),
3108            GrpcCallOptions::stream_setup(),
3109            tokio::time::Instant::now(),
3110        );
3111        let policy = GrpcTimeoutPolicy {
3112            stream_idle: Some(Duration::from_secs(5)),
3113            ..GrpcTimeoutPolicy::default()
3114        };
3115        let mut deadlines = GrpcStreamDeadline::new(policy, None)
3116            .expect("stream deadlines")
3117            .with_transport_progress(Some(progress));
3118        let waiting = tokio::spawn(async move {
3119            let undecoded = std::future::poll_fn(move |context| {
3120                let _ = Pin::new(&mut body).poll_frame(context);
3121                Poll::<Result<(), Status>>::Pending
3122            });
3123            deadlines.next(undecoded).await
3124        });
3125
3126        tokio::time::advance(Duration::from_secs(4)).await;
3127        let _ = ready.send(());
3128        tokio::task::yield_now().await;
3129        tokio::time::advance(Duration::from_secs(4)).await;
3130        assert!(
3131            !waiting.is_finished(),
3132            "raw progress must reset stream idle"
3133        );
3134        tokio::time::advance(Duration::from_secs(1)).await;
3135        let error = waiting
3136            .await
3137            .expect("stream task")
3138            .expect_err("idle expires after progress window");
3139        assert!(matches!(
3140            error,
3141            GrpcStreamError::Deadline(GrpcDeadlineError {
3142                class: GrpcTimeoutClass::StreamIdle,
3143                ..
3144            })
3145        ));
3146    }
3147
3148    #[tokio::test(start_paused = true)]
3149    async fn decoded_delivery_does_not_grant_a_second_idle_window_after_raw_progress() {
3150        let progress = GrpcTransportProgress::new();
3151        let mut deadlines = GrpcStreamDeadline::new(
3152            GrpcTimeoutPolicy {
3153                stream_idle: Some(Duration::from_secs(5)),
3154                ..GrpcTimeoutPolicy::default()
3155            },
3156            None,
3157        )
3158        .expect("stream deadlines")
3159        .with_transport_progress(Some(progress.clone()));
3160        let control = tokio::spawn(async move {
3161            let result = deadlines
3162                .established_phase(tokio::time::sleep(Duration::from_secs(8)))
3163                .await;
3164            (deadlines, result)
3165        });
3166
3167        tokio::task::yield_now().await;
3168        tokio::time::advance(Duration::from_secs(4)).await;
3169        progress.record();
3170        tokio::task::yield_now().await;
3171        tokio::time::advance(Duration::from_secs(4)).await;
3172        let (mut deadlines, result) = control.await.expect("control phase task");
3173        result.expect("raw progress keeps long control work alive");
3174        deadlines
3175            .observe_decoded_message()
3176            .expect("decoded delivery observation");
3177
3178        let waiting = tokio::spawn(async move {
3179            deadlines
3180                .established_phase(std::future::pending::<()>())
3181                .await
3182        });
3183        tokio::time::advance(Duration::from_millis(999)).await;
3184        assert!(!waiting.is_finished());
3185        tokio::time::advance(Duration::from_millis(1)).await;
3186        let error = waiting
3187            .await
3188            .expect("idle task")
3189            .expect_err("idle remains anchored to raw DATA arrival");
3190        assert!(matches!(
3191            error,
3192            GrpcStreamError::Deadline(GrpcDeadlineError {
3193                class: GrpcTimeoutClass::StreamIdle,
3194                ..
3195            })
3196        ));
3197    }
3198
3199    #[tokio::test(start_paused = true)]
3200    async fn empty_raw_data_does_not_reset_stream_idle() {
3201        let progress = GrpcTransportProgress::new();
3202        let (ready, receiver) = oneshot::channel();
3203        let body = SignaledDataBody {
3204            ready: receiver,
3205            data: tonic::codegen::Bytes::new(),
3206            emitted: false,
3207        };
3208        let mut body = GrpcDeadlineBody::new(
3209            body,
3210            None,
3211            Some(progress.clone()),
3212            GrpcCallOptions::stream_setup(),
3213            tokio::time::Instant::now(),
3214        );
3215        let policy = GrpcTimeoutPolicy {
3216            stream_idle: Some(Duration::from_secs(5)),
3217            ..GrpcTimeoutPolicy::default()
3218        };
3219        let mut deadlines = GrpcStreamDeadline::new(policy, None)
3220            .expect("stream deadlines")
3221            .with_transport_progress(Some(progress));
3222        let waiting = tokio::spawn(async move {
3223            let undecoded = std::future::poll_fn(move |context| {
3224                let _ = Pin::new(&mut body).poll_frame(context);
3225                Poll::<Result<(), Status>>::Pending
3226            });
3227            deadlines.next(undecoded).await
3228        });
3229
3230        tokio::time::advance(Duration::from_secs(4)).await;
3231        let _ = ready.send(());
3232        tokio::task::yield_now().await;
3233        tokio::time::advance(Duration::from_secs(1)).await;
3234        let error = waiting
3235            .await
3236            .expect("stream task")
3237            .expect_err("empty DATA must not reset idle");
3238        assert!(matches!(
3239            error,
3240            GrpcStreamError::Deadline(GrpcDeadlineError {
3241                class: GrpcTimeoutClass::StreamIdle,
3242                ..
3243            })
3244        ));
3245    }
3246
3247    async fn assert_dedicated_reader_progresses_during_stalled_work() {
3248        let progress = GrpcTransportProgress::new();
3249        let (ready, receiver) = oneshot::channel();
3250        let mut body = GrpcDeadlineBody::new(
3251            SignaledDataBody {
3252                ready: receiver,
3253                data: tonic::codegen::Bytes::from_static(b"partial"),
3254                emitted: false,
3255            },
3256            None,
3257            Some(progress.clone()),
3258            GrpcCallOptions::stream_setup(),
3259            tokio::time::Instant::now(),
3260        );
3261        let reader = tokio::spawn(async move {
3262            std::future::poll_fn(move |context| {
3263                let _ = Pin::new(&mut body).poll_frame(context);
3264                Poll::<()>::Pending
3265            })
3266            .await;
3267        });
3268        let mut deadlines = GrpcStreamDeadline::new(
3269            GrpcTimeoutPolicy {
3270                stream_idle: Some(Duration::from_secs(5)),
3271                ..GrpcTimeoutPolicy::default()
3272            },
3273            None,
3274        )
3275        .expect("stream deadlines")
3276        .with_transport_progress(Some(progress));
3277        let work = tokio::spawn(async move {
3278            deadlines
3279                .established_phase(std::future::pending::<()>())
3280                .await
3281        });
3282
3283        tokio::task::yield_now().await;
3284        tokio::time::advance(Duration::from_secs(4)).await;
3285        let _ = ready.send(());
3286        tokio::task::yield_now().await;
3287        tokio::time::advance(Duration::from_secs(4)).await;
3288        assert!(!work.is_finished(), "raw progress must reset stream idle");
3289        tokio::time::advance(Duration::from_secs(1)).await;
3290        let error = work
3291            .await
3292            .expect("established work task")
3293            .expect_err("idle expires after the raw-progress window");
3294        assert!(matches!(
3295            error,
3296            GrpcStreamError::Deadline(GrpcDeadlineError {
3297                class: GrpcTimeoutClass::StreamIdle,
3298                ..
3299            })
3300        ));
3301        reader.abort();
3302    }
3303
3304    #[tokio::test(start_paused = true)]
3305    async fn raw_progress_remains_observable_during_stalled_control_work() {
3306        assert_dedicated_reader_progresses_during_stalled_work().await;
3307    }
3308
3309    #[tokio::test(start_paused = true)]
3310    async fn raw_progress_remains_observable_during_stalled_output_work() {
3311        assert_dedicated_reader_progresses_during_stalled_work().await;
3312    }
3313
3314    #[tokio::test(start_paused = true)]
3315    async fn raw_progress_remains_observable_during_stalled_resubscribe_work() {
3316        assert_dedicated_reader_progresses_during_stalled_work().await;
3317    }
3318
3319    #[test]
3320    fn debug_output_redacts_inner_errors_statuses_and_bodies() {
3321        let transport =
3322            GrpcDeadlineServiceError::<Status>::transport(Status::internal("TRANSPORT_SECRET"));
3323        assert!(!format!("{transport:?}").contains("TRANSPORT_SECRET"));
3324        let source = transport.source().expect("payload-free transport source");
3325        assert!(!source.to_string().contains("TRANSPORT_SECRET"));
3326        assert!(!format!("{source:?}").contains("TRANSPORT_SECRET"));
3327        assert_eq!(
3328            Status::from_error(Box::new(transport)).code(),
3329            Code::Internal
3330        );
3331        let deadline = GrpcDeadlineServiceError::<Status>::Deadline(Status::deadline_exceeded(
3332            "DEADLINE_SECRET",
3333        ));
3334        assert!(!format!("{deadline:?}").contains("DEADLINE_SECRET"));
3335        let source = deadline.source().expect("payload-free deadline source");
3336        assert!(!source.to_string().contains("DEADLINE_SECRET"));
3337        assert!(!format!("{source:?}").contains("DEADLINE_SECRET"));
3338        let invalid = GrpcDeadlineServiceError::<Status>::InvalidTimeout(Status::invalid_argument(
3339            "INVALID_SECRET",
3340        ));
3341        assert!(!format!("{invalid:?}").contains("INVALID_SECRET"));
3342        let source = invalid.source().expect("payload-free invalid source");
3343        assert!(!source.to_string().contains("INVALID_SECRET"));
3344        assert!(!format!("{source:?}").contains("INVALID_SECRET"));
3345        let stream = GrpcStreamError::Status(Status::internal("SERVER_SECRET"));
3346        assert!(!format!("{stream:?}").contains("SERVER_SECRET"));
3347        let body = GrpcDeadlineBody::new(
3348            PendingBody,
3349            None,
3350            None,
3351            GrpcCallOptions::ordinary_read(),
3352            tokio::time::Instant::now(),
3353        );
3354        assert!(!format!("{body:?}").contains("PendingBody"));
3355        let service =
3356            GrpcDeadlineService::new_resolved("INNER_SECRET", GrpcTimeoutPolicy::default());
3357        assert!(!format!("{service:?}").contains("INNER_SECRET"));
3358    }
3359
3360    #[tokio::test]
3361    async fn redacted_transport_errors_preserve_readiness_and_body_status_codes() {
3362        let mut service = GrpcDeadlineService::new_resolved(
3363            StatusReadyFailureService,
3364            GrpcTimeoutPolicy::default(),
3365        );
3366        let readiness = service
3367            .call(http::Request::new(Body::empty()))
3368            .await
3369            .expect_err("readiness failure");
3370        let readiness_text = format!("{readiness:?} {readiness}");
3371        assert!(!readiness_text.contains("HOSTILE_READINESS_SECRET"));
3372        let readiness_status = Status::from_error(Box::new(readiness));
3373        assert_eq!(readiness_status.code(), Code::Unavailable);
3374        assert!(!format!("{readiness_status:?}").contains("HOSTILE_READINESS_SECRET"));
3375
3376        let mut body = GrpcDeadlineBody::new(
3377            StatusErrorBody {
3378                status: Some(Status::cancelled("HOSTILE_BODY_SECRET")),
3379            },
3380            None,
3381            None,
3382            GrpcCallOptions::ordinary_read(),
3383            tokio::time::Instant::now(),
3384        );
3385        let body_error = std::future::poll_fn(|context| Pin::new(&mut body).poll_frame(context))
3386            .await
3387            .expect("body failure frame")
3388            .expect_err("body status failure");
3389        let body_text = format!("{body_error:?} {body_error}");
3390        assert!(!body_text.contains("HOSTILE_BODY_SECRET"));
3391        let body_status = Status::from_error(Box::new(body_error));
3392        assert_eq!(body_status.code(), Code::Cancelled);
3393        assert!(!format!("{body_status:?}").contains("HOSTILE_BODY_SECRET"));
3394    }
3395
3396    #[test]
3397    fn redacted_transport_errors_preserve_non_status_tonic_classification() {
3398        let timeout = GrpcDeadlineServiceError::<TimeoutExpired>::transport(TimeoutExpired(()));
3399        assert_eq!(
3400            Status::from_error(Box::new(timeout)).code(),
3401            Code::Cancelled
3402        );
3403
3404        let connect =
3405            GrpcDeadlineServiceError::<tonic::ConnectError>::transport(tonic::ConnectError(
3406                Box::new(HostileTransportError("HOSTILE_CONNECT_SECRET\ncontrol")),
3407            ));
3408        let rendered = format!("{connect:?} {connect}");
3409        assert!(!rendered.contains("HOSTILE_CONNECT_SECRET"));
3410        let status = Status::from_error(Box::new(connect));
3411        assert_eq!(status.code(), Code::Unavailable);
3412        assert!(!format!("{status:?}").contains("HOSTILE_CONNECT_SECRET"));
3413    }
3414
3415    #[tokio::test(start_paused = true)]
3416    async fn stream_idle_resets_but_lifetime_does_not() {
3417        let policy = GrpcTimeoutPolicy {
3418            stream_idle: Some(Duration::from_secs(3)),
3419            stream_total_lifetime: Some(Duration::from_secs(8)),
3420            ..GrpcTimeoutPolicy::default()
3421        };
3422        let mut deadlines = GrpcStreamDeadline::new(policy, None).expect("stream deadlines");
3423        tokio::time::advance(Duration::from_secs(2)).await;
3424        assert!(matches!(
3425            deadlines.next(async { Ok::<_, Status>(1) }).await,
3426            Ok(1)
3427        ));
3428        tokio::time::advance(Duration::from_secs(2)).await;
3429        assert!(matches!(
3430            deadlines.next(async { Ok::<_, Status>(2) }).await,
3431            Ok(2)
3432        ));
3433        tokio::time::advance(Duration::from_secs(3)).await;
3434        let error = deadlines
3435            .next(std::future::pending::<Result<(), Status>>())
3436            .await
3437            .expect_err("idle timeout");
3438        let GrpcStreamError::Deadline(error) = error else {
3439            panic!("expected deadline error");
3440        };
3441        assert_eq!(error.class, GrpcTimeoutClass::StreamIdle);
3442        assert_eq!(error.outcome, GrpcTimeoutOutcome::StreamTerminated);
3443    }
3444
3445    #[tokio::test(start_paused = true)]
3446    async fn stream_progress_and_reopen_do_not_reset_total_lifetime() {
3447        let policy = GrpcTimeoutPolicy {
3448            stream_idle: Some(Duration::from_secs(4)),
3449            stream_total_lifetime: Some(Duration::from_secs(6)),
3450            ..GrpcTimeoutPolicy::default()
3451        };
3452        let mut deadlines = GrpcStreamDeadline::new(policy, None).expect("stream deadlines");
3453        tokio::time::advance(Duration::from_secs(3)).await;
3454        assert!(matches!(
3455            deadlines.next(async { Ok::<_, Status>(()) }).await,
3456            Ok(())
3457        ));
3458        tokio::time::advance(Duration::from_secs(2)).await;
3459        deadlines.reset_idle().expect("reopen resets idle only");
3460        tokio::time::advance(Duration::from_secs(1)).await;
3461        let error = deadlines
3462            .next(std::future::pending::<Result<(), Status>>())
3463            .await
3464            .expect_err("lifetime timeout");
3465        let GrpcStreamError::Deadline(error) = error else {
3466            panic!("expected deadline error");
3467        };
3468        assert_eq!(error.class, GrpcTimeoutClass::StreamLifetime);
3469    }
3470
3471    #[test]
3472    fn server_and_local_statuses_share_classification_but_retain_source() {
3473        let options = GrpcCallOptions::ordinary_mutation();
3474        let local = local_deadline_status(options, Duration::from_secs(1));
3475        let local = GrpcDeadlineError::from_status(
3476            &local,
3477            GrpcTimeoutClass::LongUnary,
3478            GrpcTimeoutOutcome::ReadAborted,
3479            Duration::from_secs(1),
3480        )
3481        .expect("local deadline");
3482        assert_eq!(local.class, GrpcTimeoutClass::OrdinaryUnary);
3483        assert_eq!(local.outcome, GrpcTimeoutOutcome::MutationIndeterminate);
3484        assert_eq!(local.source, GrpcTimeoutSource::Local);
3485
3486        let server = Status::deadline_exceeded("untrusted upstream detail");
3487        let server = GrpcDeadlineError::from_status(
3488            &server,
3489            GrpcTimeoutClass::LongUnary,
3490            GrpcTimeoutOutcome::ReadAborted,
3491            Duration::from_secs(1),
3492        )
3493        .expect("server deadline");
3494        assert_eq!(server.class, GrpcTimeoutClass::LongUnary);
3495        assert_eq!(server.source, GrpcTimeoutSource::Server);
3496        assert!(!server.to_string().contains("untrusted upstream detail"));
3497    }
3498}
anytype_rpc/deadline.rs

anytype_rpc/
deadline.rs
