ps_promise/methods/
with_resolvers.rs

1use std::{
2    fmt::Debug,
3    future::Future,
4    pin::Pin,
5    sync::{Arc, Mutex, PoisonError},
6    task::{Context, Poll, Waker},
7};
8
9use crate::{Promise, PromiseRejection, TaskFailure};
10
11/// Rejection payload produced when every handle returned by
12/// `Promise::with_resolvers` is dropped without settling the promise.
13#[derive(thiserror::Error, Debug, Clone, Copy, PartialEq, Eq)]
14#[error("all resolver handles were dropped without settling the promise")]
15pub struct ResolversDropped;
16
17impl<T, E> Promise<T, E>
18where
19    T: Send + 'static,
20    E: PromiseRejection,
21{
22    /// Creates a pending [`Promise`] along with handles that settle it.
23    ///
24    /// Mirrors ECMAScript's `Promise.withResolvers`: the promise stays
25    /// pending until [`Resolve::resolve`] or [`Reject::reject`] is called,
26    /// and the first settlement wins. Both handles are cloneable, and any
27    /// clone may settle the promise. If every handle, including clones, is
28    /// dropped without settling, the promise rejects with
29    /// [`ResolversDropped`] wrapped in [`TaskFailure::Error`], mapped
30    /// through [`PromiseRejection::task_failed`].
31    #[must_use = "Dropping the handles rejects the Promise!"]
32    pub fn with_resolvers() -> (Self, Resolve<T, E>, Reject<T, E>) {
33        let slot = Arc::new(Mutex::new(Slot {
34            settled: false,
35            outcome: None,
36            waker: None,
37        }));
38
39        let guard = Arc::new(HandleGuard { slot: slot.clone() });
40
41        (
42            Self::lazy(SlotFuture { slot }),
43            Resolve {
44                guard: guard.clone(),
45            },
46            Reject { guard },
47        )
48    }
49}
50
51/// Resolves the [`Promise`] created by `Promise::with_resolvers`.
52///
53/// Cloning yields another handle to the same promise; any clone may settle
54/// it, and the first settlement wins.
55pub struct Resolve<T, E>
56where
57    E: PromiseRejection,
58{
59    guard: Arc<HandleGuard<T, E>>,
60}
61
62impl<T, E> Clone for Resolve<T, E>
63where
64    E: PromiseRejection,
65{
66    fn clone(&self) -> Self {
67        Self {
68            guard: self.guard.clone(),
69        }
70    }
71}
72
73impl<T, E> Debug for Resolve<T, E>
74where
75    E: PromiseRejection,
76{
77    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        f.debug_struct("Resolve").finish_non_exhaustive()
79    }
80}
81
82impl<T, E> Resolve<T, E>
83where
84    E: PromiseRejection,
85{
86    /// Settles the associated [`Promise`] with `value`.
87    ///
88    /// Ignored if the promise is already settled.
89    pub fn resolve(self, value: T) {
90        settle(&self.guard.slot, Ok(value));
91    }
92}
93
94/// Rejects the [`Promise`] created by `Promise::with_resolvers`.
95///
96/// Cloning yields another handle to the same promise; any clone may settle
97/// it, and the first settlement wins.
98pub struct Reject<T, E>
99where
100    E: PromiseRejection,
101{
102    guard: Arc<HandleGuard<T, E>>,
103}
104
105impl<T, E> Clone for Reject<T, E>
106where
107    E: PromiseRejection,
108{
109    fn clone(&self) -> Self {
110        Self {
111            guard: self.guard.clone(),
112        }
113    }
114}
115
116impl<T, E> Debug for Reject<T, E>
117where
118    E: PromiseRejection,
119{
120    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
121        f.debug_struct("Reject").finish_non_exhaustive()
122    }
123}
124
125impl<T, E> Reject<T, E>
126where
127    E: PromiseRejection,
128{
129    /// Settles the associated [`Promise`] with `err`.
130    ///
131    /// Ignored if the promise is already settled.
132    pub fn reject(self, err: E) {
133        settle(&self.guard.slot, Err(err));
134    }
135}
136
137struct Slot<T, E> {
138    settled: bool,
139    outcome: Option<Result<T, E>>,
140    waker: Option<Waker>,
141}
142
143type SharedSlot<T, E> = Arc<Mutex<Slot<T, E>>>;
144
145/// Stores `outcome` into `slot` and wakes the consumer, unless a prior
146/// settlement already won.
147fn settle<T, E>(slot: &SharedSlot<T, E>, outcome: Result<T, E>) {
148    let mut guard = slot.lock().unwrap_or_else(PoisonError::into_inner);
149
150    if !guard.settled {
151        guard.settled = true;
152        guard.outcome = Some(outcome);
153
154        if let Some(waker) = guard.waker.take() {
155            drop(guard);
156
157            waker.wake();
158        }
159    }
160}
161
162/// Shared by every settlement handle; when the last one is dropped, rejects
163/// the promise with [`ResolversDropped`] unless it has already been settled.
164struct HandleGuard<T, E>
165where
166    E: PromiseRejection,
167{
168    slot: SharedSlot<T, E>,
169}
170
171impl<T, E> Drop for HandleGuard<T, E>
172where
173    E: PromiseRejection,
174{
175    fn drop(&mut self) {
176        settle(
177            &self.slot,
178            Err(E::task_failed(TaskFailure::Error(Arc::new(
179                ResolversDropped,
180            )))),
181        );
182    }
183}
184
185struct SlotFuture<T, E> {
186    slot: SharedSlot<T, E>,
187}
188
189impl<T, E> Future for SlotFuture<T, E>
190where
191    E: PromiseRejection,
192{
193    type Output = Result<T, E>;
194
195    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
196        let Ok(mut guard) = self.slot.lock() else {
197            return Poll::Ready(Err(E::task_failed(TaskFailure::Panic(
198                "a panic corrupted the promise's settlement state".into(),
199            ))));
200        };
201
202        if !guard.settled {
203            if let Some(waker) = &mut guard.waker {
204                waker.clone_from(cx.waker());
205            } else {
206                guard.waker = Some(cx.waker().clone());
207            }
208
209            return Poll::Pending;
210        }
211
212        Poll::Ready(
213            guard
214                .outcome
215                .take()
216                .unwrap_or_else(|| Err(E::already_consumed())),
217        )
218    }
219}
220
221#[cfg(test)]
222#[allow(clippy::expect_used)]
223mod tests {
224    use std::{
225        future::Future,
226        pin::Pin,
227        sync::{
228            atomic::{AtomicUsize, Ordering},
229            Arc, Mutex,
230        },
231        task::{Context, Poll, Wake, Waker},
232        thread,
233    };
234
235    use crate::{Promise, PromiseRejection, TaskFailure};
236
237    use super::{settle, ResolversDropped, SharedSlot, Slot, SlotFuture};
238
239    const POISONED_MSG: &str = "task panicked: a panic corrupted the promise's settlement state";
240
241    #[derive(Debug, PartialEq)]
242    enum E {
243        AlreadyConsumed,
244        Fail,
245        TaskFailed(String),
246    }
247
248    impl PromiseRejection for E {
249        fn already_consumed() -> Self {
250            Self::AlreadyConsumed
251        }
252
253        fn task_failed(failure: TaskFailure) -> Self {
254            Self::TaskFailed(failure.to_string())
255        }
256    }
257
258    /// The rejection produced when every settlement handle is dropped.
259    fn resolvers_dropped() -> E {
260        E::TaskFailed(ResolversDropped.to_string())
261    }
262
263    fn cx() -> Context<'static> {
264        Context::from_waker(Waker::noop())
265    }
266
267    fn poll_future<F: Future + Unpin>(future: &mut F) -> Poll<F::Output> {
268        Pin::new(future).poll(&mut cx())
269    }
270
271    fn empty_slot() -> SharedSlot<i32, E> {
272        Arc::new(Mutex::new(Slot {
273            settled: false,
274            outcome: None,
275            waker: None,
276        }))
277    }
278
279    /// Poisons `slot` by panicking on a helper thread while the lock is held.
280    fn poison(slot: &SharedSlot<i32, E>) {
281        let poisoner = slot.clone();
282
283        thread::spawn(move || {
284            let _guard = poisoner.lock().expect("first lock of a fresh slot");
285
286            panic!("poison the slot");
287        })
288        .join()
289        .expect_err("poisoning thread must panic");
290    }
291
292    struct CountingWaker {
293        count: AtomicUsize,
294    }
295
296    impl Wake for CountingWaker {
297        fn wake(self: Arc<Self>) {
298            self.count.fetch_add(1, Ordering::Relaxed);
299        }
300
301        fn wake_by_ref(self: &Arc<Self>) {
302            self.count.fetch_add(1, Ordering::Relaxed);
303        }
304    }
305
306    fn counting_waker() -> (Arc<CountingWaker>, Waker) {
307        let counter = Arc::new(CountingWaker {
308            count: AtomicUsize::new(0),
309        });
310
311        let waker = Waker::from(counter.clone());
312
313        (counter, waker)
314    }
315
316    #[test]
317    fn resolve_settles_promise() {
318        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
319
320        resolve.resolve(42);
321        promise.poll_settled(&mut cx());
322
323        assert_eq!(promise.consume(), Some(Ok(42)));
324    }
325
326    #[test]
327    fn reject_settles_promise() {
328        let (mut promise, _resolve, reject) = Promise::<i32, E>::with_resolvers();
329
330        reject.reject(E::Fail);
331        promise.poll_settled(&mut cx());
332
333        assert_eq!(promise.consume(), Some(Err(E::Fail)));
334    }
335
336    #[test]
337    fn pending_until_settled() {
338        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
339
340        assert!(promise.poll_pending(&mut cx()));
341
342        resolve.resolve(7);
343
344        assert!(promise.poll_settled(&mut cx()));
345        assert_eq!(promise.consume(), Some(Ok(7)));
346    }
347
348    #[test]
349    fn first_settlement_wins() {
350        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
351
352        resolve.resolve(1);
353        reject.reject(E::Fail);
354        promise.poll_settled(&mut cx());
355
356        assert_eq!(promise.consume(), Some(Ok(1)));
357    }
358
359    #[test]
360    fn rejects_when_handles_dropped() {
361        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
362
363        drop(resolve);
364        drop(reject);
365        promise.poll_settled(&mut cx());
366
367        assert_eq!(promise.consume(), Some(Err(resolvers_dropped())));
368    }
369
370    #[test]
371    fn settles_even_if_other_handle_dropped_later() {
372        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
373
374        resolve.resolve(9);
375        drop(reject);
376        promise.poll_settled(&mut cx());
377
378        assert_eq!(promise.consume(), Some(Ok(9)));
379    }
380
381    #[test]
382    fn clone_can_settle_after_originals_are_dropped() {
383        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
384
385        let clone = resolve.clone();
386
387        drop(resolve);
388        drop(reject);
389
390        clone.resolve(3);
391        promise.poll_settled(&mut cx());
392
393        assert_eq!(promise.consume(), Some(Ok(3)));
394    }
395
396    #[test]
397    fn first_settlement_wins_across_clones() {
398        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
399
400        let clone = resolve.clone();
401
402        clone.resolve(1);
403        resolve.resolve(2);
404        promise.poll_settled(&mut cx());
405
406        assert_eq!(promise.consume(), Some(Ok(1)));
407    }
408
409    #[test]
410    fn live_clone_keeps_the_promise_pending() {
411        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
412
413        let clone = reject.clone();
414
415        drop(resolve);
416        drop(reject);
417
418        assert!(promise.poll_pending(&mut cx()));
419
420        clone.reject(E::Fail);
421        promise.poll_settled(&mut cx());
422
423        assert_eq!(promise.consume(), Some(Err(E::Fail)));
424    }
425
426    #[test]
427    fn rejects_only_when_every_clone_is_dropped() {
428        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
429
430        let resolve_clone = resolve.clone();
431        let reject_clone = reject.clone();
432
433        drop(resolve);
434        drop(reject);
435
436        assert!(promise.poll_pending(&mut cx()));
437
438        drop(resolve_clone);
439        drop(reject_clone);
440        promise.poll_settled(&mut cx());
441
442        assert_eq!(promise.consume(), Some(Err(resolvers_dropped())));
443    }
444
445    #[test]
446    fn debug_output_names_the_handles() {
447        let (_promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
448
449        assert_eq!(format!("{resolve:?}"), "Resolve { .. }");
450        assert_eq!(format!("{reject:?}"), "Reject { .. }");
451    }
452
453    #[test]
454    fn resolve_wakes_pending_consumer() {
455        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
456
457        let (counter, waker) = counting_waker();
458
459        assert!(promise.poll_pending(&mut Context::from_waker(&waker)));
460        assert_eq!(counter.count.load(Ordering::SeqCst), 0);
461
462        resolve.resolve(8);
463
464        assert_eq!(counter.count.load(Ordering::SeqCst), 1);
465
466        assert!(promise.poll_settled(&mut cx()));
467        assert_eq!(promise.consume(), Some(Ok(8)));
468    }
469
470    #[test]
471    fn reject_wakes_pending_consumer() {
472        let (mut promise, _resolve, reject) = Promise::<i32, E>::with_resolvers();
473
474        let (counter, waker) = counting_waker();
475
476        assert!(promise.poll_pending(&mut Context::from_waker(&waker)));
477        assert_eq!(counter.count.load(Ordering::SeqCst), 0);
478
479        reject.reject(E::Fail);
480
481        assert_eq!(counter.count.load(Ordering::SeqCst), 1);
482
483        assert!(promise.poll_settled(&mut cx()));
484        assert_eq!(promise.consume(), Some(Err(E::Fail)));
485    }
486
487    #[test]
488    fn dropping_last_handle_wakes_pending_consumer() {
489        let (mut promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
490
491        let (counter, waker) = counting_waker();
492
493        assert!(promise.poll_pending(&mut Context::from_waker(&waker)));
494
495        drop(resolve);
496
497        assert_eq!(counter.count.load(Ordering::SeqCst), 0);
498
499        drop(reject);
500
501        assert_eq!(counter.count.load(Ordering::SeqCst), 1);
502
503        assert!(promise.poll_settled(&mut cx()));
504        assert_eq!(promise.consume(), Some(Err(resolvers_dropped())));
505    }
506
507    /// Only the waker from the most recent poll is woken; a re-poll with a
508    /// different waker replaces the prior registration, which is all the
509    /// `Future` contract requires.
510    #[test]
511    fn repoll_replaces_the_stored_waker() {
512        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
513
514        let (counter_a, waker_a) = counting_waker();
515        let (counter_b, waker_b) = counting_waker();
516
517        assert!(promise.poll_pending(&mut Context::from_waker(&waker_a)));
518        assert!(promise.poll_pending(&mut Context::from_waker(&waker_b)));
519
520        resolve.resolve(4);
521
522        assert_eq!(counter_a.count.load(Ordering::SeqCst), 0);
523        assert_eq!(counter_b.count.load(Ordering::SeqCst), 1);
524
525        assert!(promise.poll_settled(&mut cx()));
526        assert_eq!(promise.consume(), Some(Ok(4)));
527    }
528
529    #[test]
530    fn settlement_from_another_thread_wakes_and_delivers() {
531        let (mut promise, resolve, _reject) = Promise::<i32, E>::with_resolvers();
532
533        let (counter, waker) = counting_waker();
534
535        assert!(promise.poll_pending(&mut Context::from_waker(&waker)));
536
537        thread::spawn(move || resolve.resolve(21))
538            .join()
539            .expect("settler thread");
540
541        assert!(counter.count.load(Ordering::SeqCst) >= 1);
542
543        assert!(promise.poll_settled(&mut cx()));
544        assert_eq!(promise.consume(), Some(Ok(21)));
545    }
546
547    #[test]
548    fn settling_after_promise_dropped_is_ignored() {
549        let (promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
550
551        drop(promise);
552
553        resolve.resolve(1);
554        reject.reject(E::Fail);
555    }
556
557    #[test]
558    fn dropping_handles_after_promise_dropped_is_ignored() {
559        let (promise, resolve, reject) = Promise::<i32, E>::with_resolvers();
560
561        drop(promise);
562        drop(resolve);
563        drop(reject);
564    }
565
566    #[test]
567    fn settle_before_first_poll_delivers_immediately() {
568        let slot = empty_slot();
569
570        settle(&slot, Ok(11));
571
572        let mut future = SlotFuture { slot };
573
574        assert_eq!(poll_future(&mut future), Poll::Ready(Ok(11)));
575    }
576
577    /// Regression test: polling the settlement future again after it has
578    /// delivered its outcome must report `already_consumed`, not hang as
579    /// `Pending`.
580    #[test]
581    fn repoll_after_delivery_reports_already_consumed() {
582        let slot = empty_slot();
583
584        let mut future = SlotFuture { slot: slot.clone() };
585
586        assert_eq!(poll_future(&mut future), Poll::Pending);
587
588        settle(&slot, Ok(5));
589
590        assert_eq!(poll_future(&mut future), Poll::Ready(Ok(5)));
591        assert_eq!(
592            poll_future(&mut future),
593            Poll::Ready(Err(E::AlreadyConsumed))
594        );
595    }
596
597    /// A settlement arriving after the outcome was delivered must not
598    /// resurrect the future with a new outcome.
599    #[test]
600    fn settlement_after_delivery_is_ignored() {
601        let slot = empty_slot();
602
603        let mut future = SlotFuture { slot: slot.clone() };
604
605        settle(&slot, Ok(1));
606
607        assert_eq!(poll_future(&mut future), Poll::Ready(Ok(1)));
608
609        settle(&slot, Ok(2));
610
611        assert_eq!(
612            poll_future(&mut future),
613            Poll::Ready(Err(E::AlreadyConsumed))
614        );
615    }
616
617    #[test]
618    fn poisoned_slot_rejects_with_panic_failure() {
619        let slot = empty_slot();
620
621        poison(&slot);
622
623        let mut future = SlotFuture { slot };
624
625        assert_eq!(
626            poll_future(&mut future),
627            Poll::Ready(Err(E::TaskFailed(POISONED_MSG.into())))
628        );
629    }
630
631    /// Pins the chosen conservative semantics: once the slot is poisoned, the
632    /// consumer rejects even though a real outcome was stored before the
633    /// poisoning panic.
634    #[test]
635    fn poisoned_slot_rejects_even_after_settlement() {
636        let slot = empty_slot();
637
638        settle(&slot, Ok(9));
639        poison(&slot);
640
641        let mut future = SlotFuture { slot };
642
643        assert_eq!(
644            poll_future(&mut future),
645            Poll::Ready(Err(E::TaskFailed(POISONED_MSG.into())))
646        );
647    }
648
649    /// The producer side ignores poisoning, so settling a poisoned slot is a
650    /// quiet no-op rather than a second panic.
651    #[test]
652    fn settling_a_poisoned_slot_does_not_panic() {
653        let slot = empty_slot();
654
655        poison(&slot);
656
657        settle(&slot, Ok(3));
658    }
659
660    #[test]
661    fn resolvers_dropped_error_message() {
662        assert_eq!(
663            ResolversDropped.to_string(),
664            "all resolver handles were dropped without settling the promise"
665        );
666    }
667}
ps_promise/methods/with_resolvers.rs

ps_promise/methods/
with_resolvers.rs
