diff --git a/README.md b/README.md
index c8d5b0c..1438175 100644
--- a/README.md
+++ b/README.md
@@ -188,6 +188,7 @@ Transform and compose observable streams:
 - `ThrottleFirst` - Emit the first value of each time window and drop the rest
 - `ThrottleLast` - Emit only the latest value of each time window, when the window expires
 - `ThrottleFirstLast` - Emit the first value of each time window immediately, then the latest value when the window expires
+- `Timeout` - Fail with a `TimeoutException` if no value is observed within a time window (measured from subscription and reset on each value)
 
 #### Concurrency & Scheduling
 - `ObserveOn` - Control execution context for downstream operators
diff --git a/src/R3Async.Tests/Operators/TimeoutTest.cs b/src/R3Async.Tests/Operators/TimeoutTest.cs
new file mode 100644
index 0000000..dea1898
--- /dev/null
+++ b/src/R3Async.Tests/Operators/TimeoutTest.cs
@@ -0,0 +1,179 @@
+using Microsoft.Extensions.Time.Testing;
+using R3Async.Subjects;
+using Shouldly;
+#pragma warning disable CS1998
+
+namespace R3Async.Tests.Operators;
+
+public class TimeoutTest
+{
+    static readonly TimeSpan DueTime = TimeSpan.FromMilliseconds(100);
+
+    [Fact]
+    public async Task Timeout_FailsWhenNoValueArrivesInTime()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        timeProvider.Advance(DueTime);
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBeOfType<TimeoutException>();
+    }
+
+    [Fact]
+    public async Task Timeout_ValuesResetTheWindow()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        // Each value arrives before the window expires, so no timeout occurs.
+        timeProvider.Advance(TimeSpan.FromMilliseconds(50));
+        await subject.OnNextAsync(1, CancellationToken.None);
+        timeProvider.Advance(TimeSpan.FromMilliseconds(50));
+        await subject.OnNextAsync(2, CancellationToken.None);
+        timeProvider.Advance(TimeSpan.FromMilliseconds(50));
+        await subject.OnNextAsync(3, CancellationToken.None);
+
+        completedTcs.Task.IsCompleted.ShouldBeFalse();
+        lock (results) results.ShouldBe(new[] { 1, 2, 3 });
+
+        // Now let the window expire with no value.
+        timeProvider.Advance(DueTime);
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBeOfType<TimeoutException>();
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsCompletionBeforeTimeout()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subject.OnCompletedAsync(Result.Success);
+
+        var result = await completedTcs.Task;
+        result.IsSuccess.ShouldBeTrue();
+
+        // A later timer expiry must not produce anything further.
+        timeProvider.Advance(DueTime);
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsSourceFailure()
+    {
+        var expected = new InvalidOperationException("boom");
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subject.OnCompletedAsync(Result.Failure(expected));
+
+        var result = await completedTcs.Task;
+        result.IsFailure.ShouldBeTrue();
+        result.Exception.ShouldBe(expected);
+    }
+
+    [Fact]
+    public async Task Timeout_ForwardsOnErrorResumeWithoutTerminating()
+    {
+        var expected = new InvalidOperationException("boom");
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var errors = new List<Exception>();
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { lock (errors) errors.Add(ex); },
+            async result => { },
+            CancellationToken.None);
+
+        await subject.OnErrorResumeAsync(expected, CancellationToken.None);
+        await subject.OnNextAsync(1, CancellationToken.None);
+
+        lock (errors) errors.ShouldBe(new[] { expected });
+        lock (results) results.ShouldBe(new[] { 1 });
+    }
+
+    [Fact]
+    public async Task Timeout_DisposeCancelsTimer()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        await subscription.DisposeAsync();
+
+        // Firing the timer after disposal must not deliver a timeout.
+        timeProvider.Advance(DueTime);
+
+        completedTcs.Task.IsCompleted.ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task Timeout_TimeoutDisposesSourceSubscription()
+    {
+        var timeProvider = new FakeTimeProvider();
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var completedTcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        await using var subscription = await subject.Values.Timeout(DueTime, timeProvider).SubscribeAsync(
+            async (x, token) => { lock (results) results.Add(x); },
+            async (ex, token) => { },
+            async result => completedTcs.TrySetResult(result),
+            CancellationToken.None);
+
+        timeProvider.Advance(DueTime);
+        await completedTcs.Task;
+
+        // Values pushed after the timeout must not be delivered.
+        await subject.OnNextAsync(1, CancellationToken.None);
+        lock (results) results.ShouldBeEmpty();
+    }
+}
diff --git a/src/R3Async.Tests/SubscribeExtensionsTest.cs b/src/R3Async.Tests/SubscribeExtensionsTest.cs
new file mode 100644
index 0000000..66c3602
--- /dev/null
+++ b/src/R3Async.Tests/SubscribeExtensionsTest.cs
@@ -0,0 +1,52 @@
+using R3Async.Subjects;
+using Shouldly;
+#pragma warning disable CS1998
+
+namespace R3Async.Tests;
+
+public class SubscribeExtensionsTest
+{
+    [Fact]
+    public async Task SubscribeAsync_AsyncOnNextWithSyncCallbacks_ForwardsEverything()
+    {
+        var subject = Subject.Create<int>();
+
+        var results = new List<int>();
+        var errors = new List<Exception>();
+        Result? completed = null;
+
+        await using var subscription = await subject.Values.SubscribeAsync(
+            async (x, token) => results.Add(x),
+            onErrorResume: ex => errors.Add(ex),
+            onCompleted: result => completed = result,
+            CancellationToken.None);
+
+        var expectedError = new InvalidOperationException("boom");
+        await subject.OnNextAsync(1, CancellationToken.None);
+        await subject.OnErrorResumeAsync(expectedError, CancellationToken.None);
+        await subject.OnNextAsync(2, CancellationToken.None);
+        await subject.OnCompletedAsync(Result.Success);
+
+        results.ShouldBe(new[] { 1, 2 });
+        errors.ShouldBe(new[] { expectedError });
+        completed.ShouldNotBeNull();
+        completed.Value.IsSuccess.ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task SubscribeAsync_AsyncOnNextWithSyncCallbacks_NullCallbacksAreAllowed()
+    {
+        var subject = Subject.Create<int>();
+        var results = new List<int>();
+
+        await using var subscription = await subject.Values.SubscribeAsync(
+            async (x, token) => results.Add(x),
+            onErrorResume: null,
+            onCompleted: null,
+            CancellationToken.None);
+
+        await subject.OnNextAsync(42, CancellationToken.None);
+
+        results.ShouldBe(new[] { 42 });
+    }
+}
diff --git a/src/R3Async/AsyncObservableSubscribeExtensions.cs b/src/R3Async/AsyncObservableSubscribeExtensions.cs
index 2f32aa8..5fd7b8f 100644
--- a/src/R3Async/AsyncObservableSubscribeExtensions.cs
+++ b/src/R3Async/AsyncObservableSubscribeExtensions.cs
@@ -62,6 +62,29 @@ public ValueTask<IAsyncDisposable> SubscribeAsync(Action<T> onNext,
             return source.SubscribeAsync(observer, cancellationToken);
         }
 
+        public ValueTask<IAsyncDisposable> SubscribeAsync(Func<T, CancellationToken, ValueTask> onNextAsync,
+                                                          Action<Exception>? onErrorResume,
+                                                          Action<Result>? onCompleted = null,
+                                                          CancellationToken cancellationToken = default)
+        {
+            if (onNextAsync is null)
+                throw new ArgumentNullException(nameof(onNextAsync));
+            if (source is null)
+                throw new ArgumentNullException(nameof(source));
+
+            var observer = new AnonymousAsyncObserver<T>(onNextAsync, onErrorResume is null ? null : (e, _) =>
+            {
+                onErrorResume(e);
+                return default;
+            }, onCompleted is null ? null : x =>
+            {
+                onCompleted(x);
+                return default;
+            });
+
+            return source.SubscribeAsync(observer, cancellationToken);
+        }
+
         public ValueTask<IAsyncDisposable> SubscribeAsync()
         {
             return source.SubscribeAsync(static (_, _)  => default, CancellationToken.None);
diff --git a/src/R3Async/Factories/Interval.cs b/src/R3Async/Factories/Interval.cs
index 9e54d50..55222c6 100644
--- a/src/R3Async/Factories/Interval.cs
+++ b/src/R3Async/Factories/Interval.cs
@@ -22,7 +22,7 @@ public static AsyncObservable<long> Interval(TimeSpan period, TimeProvider? time
                 else
                 {
                     var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-                    await using var _ = timeProvider.CreateTimer(x => ((TaskCompletionSource<bool>)x!).TrySetResult(true), tcs, period, Timeout.InfiniteTimeSpan);
+                    await using var _ = timeProvider.CreateTimer(x => ((TaskCompletionSource<bool>)x!).TrySetResult(true), tcs, period, System.Threading.Timeout.InfiniteTimeSpan);
                     using var __ = cancellationToken.Register(x => ((TaskCompletionSource<bool>)x!).TrySetCanceled(cancellationToken), tcs);
                     await tcs.Task;
                 }
diff --git a/src/R3Async/Factories/ToAsyncObservable.cs b/src/R3Async/Factories/ToAsyncObservable.cs
index 0557e94..54d2679 100644
--- a/src/R3Async/Factories/ToAsyncObservable.cs
+++ b/src/R3Async/Factories/ToAsyncObservable.cs
@@ -12,7 +12,7 @@ public static AsyncObservable<T> ToAsyncObservable<T>(this Task<T> @this)
     {
         return CreateAsBackgroundJob<T>(async (obs, cancellationToken) =>
         {
-            var result = await @this.WaitAsync(Timeout.InfiniteTimeSpan, cancellationToken);
+            var result = await @this.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, cancellationToken);
             await obs.OnNextAsync(result, cancellationToken);
             await obs.OnCompletedAsync(Result.Success);
         }, true);
@@ -22,7 +22,7 @@ public static AsyncObservable<Unit> ToAsyncObservable(this Task @this)
     {
         return CreateAsBackgroundJob<Unit>(async (obs, cancellationToken) =>
         {
-            await @this.WaitAsync(Timeout.InfiniteTimeSpan, cancellationToken);
+            await @this.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, cancellationToken);
             await obs.OnNextAsync(Unit.Default, cancellationToken);
             await obs.OnCompletedAsync(Result.Success);
         }, true);
diff --git a/src/R3Async/Operators/TakeUntil.cs b/src/R3Async/Operators/TakeUntil.cs
index 4142c4a..659fb92 100644
--- a/src/R3Async/Operators/TakeUntil.cs
+++ b/src/R3Async/Operators/TakeUntil.cs
@@ -304,7 +304,7 @@ void Stop(Result result)
 
                     try
                     {
-                        await tcs.Task.WaitAsync(Timeout.InfiniteTimeSpan, _disposeCancellationToken);
+                        await tcs.Task.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, _disposeCancellationToken);
                         try
                         {
                             await disposable.DisposeAsync();
@@ -448,7 +448,7 @@ async void WaitAndComplete(Task task)
                 {
                     try
                     {
-                        await task.WaitAsync(Timeout.InfiniteTimeSpan, _disposeCancellationToken);
+                        await task.WaitAsync(System.Threading.Timeout.InfiniteTimeSpan, _disposeCancellationToken);
                         await ForwardOnCompletedAsync(Result.Success);
                     }
                     catch (Exception e)
diff --git a/src/R3Async/Operators/Timeout.cs b/src/R3Async/Operators/Timeout.cs
new file mode 100644
index 0000000..0513451
--- /dev/null
+++ b/src/R3Async/Operators/Timeout.cs
@@ -0,0 +1,165 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using R3Async.Internals;
+
+namespace R3Async;
+
+public static partial class AsyncObservable
+{
+    extension<T>(AsyncObservable<T> @this)
+    {
+        public AsyncObservable<T> Timeout(TimeSpan dueTime, TimeProvider? timeProvider = null)
+            => new TimeoutObservable<T>(@this, dueTime, timeProvider ?? TimeProvider.System);
+    }
+}
+
+internal sealed class TimeoutObservable<T>(AsyncObservable<T> source, TimeSpan dueTime, TimeProvider timeProvider) : AsyncObservable<T>
+{
+    protected override async ValueTask<IAsyncDisposable> SubscribeAsyncCore(AsyncObserver<T> observer, CancellationToken cancellationToken)
+    {
+        var subscription = new TimeoutSubscription(observer, dueTime, timeProvider);
+        try
+        {
+            await subscription.SubscribeAsync(source, cancellationToken);
+        }
+        catch
+        {
+            await subscription.DisposeAsync();
+            throw;
+        }
+
+        return subscription;
+    }
+
+    sealed class TimeoutSubscription : IAsyncDisposable
+    {
+        readonly AsyncObserver<T> _observer;
+        readonly TimeSpan _dueTime;
+        readonly SingleAssignmentAsyncDisposable _sourceDisposable = new();
+        readonly CancellationTokenSource _disposeCts = new();
+        readonly CancellationToken _disposeCancellationToken;
+        readonly AsyncGate _gate = new();
+        readonly TimeProvider _timeProvider;
+        readonly SerialAsyncDisposable _timerDisposable = new();
+        long _version;
+        bool _terminated;
+
+        public TimeoutSubscription(AsyncObserver<T> observer, TimeSpan dueTime, TimeProvider timeProvider)
+        {
+            _observer = observer;
+            _dueTime = dueTime;
+            _timeProvider = timeProvider;
+            _disposeCancellationToken = _disposeCts.Token;
+        }
+
+        public async ValueTask SubscribeAsync(AsyncObservable<T> source, CancellationToken subscriptionToken)
+        {
+            // The first window opens before subscribing, so a source that takes longer than
+            // dueTime to produce its first value times out even if subscription itself is slow.
+            long version;
+            using (await _gate.LockAsync())
+            {
+                version = _version;
+            }
+
+            await ScheduleTimerAsync(version);
+            var subscription = await source.SubscribeAsync(new TimeoutObserver(this), subscriptionToken);
+            await _sourceDisposable.SetDisposableAsync(subscription);
+        }
+
+        async ValueTask OnNextAsync(T value, CancellationToken cancellationToken)
+        {
+            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_disposeCancellationToken, cancellationToken);
+            long version;
+            using (await _gate.LockAsync())
+            {
+                if (_terminated) return;
+                version = unchecked(++_version);
+                await _observer.OnNextAsync(value, linkedCts.Token);
+            }
+
+            await ScheduleTimerAsync(version);
+        }
+
+        async ValueTask ScheduleTimerAsync(long version)
+        {
+            var timerSubscription = _timeProvider.CreateTimer(static state =>
+            {
+                var (self, scheduledVersion) = ((TimeoutSubscription, long))state!;
+                self.OnTimerFired(scheduledVersion);
+            }, (this, version), _dueTime, System.Threading.Timeout.InfiniteTimeSpan);
+            await _timerDisposable.SetDisposableAsync(timerSubscription);
+        }
+
+        async void OnTimerFired(long version)
+        {
+            try
+            {
+                using (await _gate.LockAsync())
+                {
+                    if (_terminated || version != _version || _disposeCancellationToken.IsCancellationRequested) return;
+                    _terminated = true;
+                }
+
+                _disposeCts.Cancel();
+                await _timerDisposable.DisposeAsync();
+                using (await _gate.LockAsync())
+                {
+                    await _observer.OnCompletedAsync(Result.Failure(new TimeoutException($"No value was observed within {_dueTime}.")));
+                }
+
+                await _sourceDisposable.DisposeAsync();
+                _disposeCts.Dispose();
+            }
+            catch (Exception e)
+            {
+                UnhandledExceptionHandler.OnUnhandledException(e);
+            }
+        }
+
+        async ValueTask OnErrorResumeAsync(Exception error, CancellationToken cancellationToken)
+        {
+            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_disposeCancellationToken, cancellationToken);
+            using (await _gate.LockAsync())
+            {
+                if (_terminated) return;
+                await _observer.OnErrorResumeAsync(error, linkedCts.Token);
+            }
+        }
+
+        async ValueTask CompleteAsync(Result? result)
+        {
+            using (await _gate.LockAsync())
+            {
+                if (_terminated) return;
+                _terminated = true;
+            }
+
+            _disposeCts.Cancel();
+            await _timerDisposable.DisposeAsync();
+            if (result is not null)
+            {
+                using (await _gate.LockAsync())
+                {
+                    await _observer.OnCompletedAsync(result.Value);
+                }
+            }
+
+            await _sourceDisposable.DisposeAsync();
+            _disposeCts.Dispose();
+        }
+
+        public ValueTask DisposeAsync() => CompleteAsync(null);
+
+        sealed class TimeoutObserver(TimeoutSubscription subscription) : AsyncObserver<T>
+        {
+            protected override ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken)
+                => subscription.OnNextAsync(value, cancellationToken);
+            protected override ValueTask OnErrorResumeAsyncCore(Exception error, CancellationToken cancellationToken)
+                => subscription.OnErrorResumeAsync(error, cancellationToken);
+            protected override ValueTask OnCompletedAsyncCore(Result result)
+                => subscription.CompleteAsync(result);
+        }
+    }
+}