Add notification_blocking_timeout for time-bounded notification waits#89
Open
Add notification_blocking_timeout for time-bounded notification waits#89
Conversation
Adds a `notification-blocking-timeout` WIT function and plumbs it through the runtime as `wait_with_timeout(Duration) -> Option<Notification>`. When the timeout expires without a notification, returns None instead of suspending the task. This allows workflows to implement retry/fallback logic when notifications aren't reliably delivered. Bumps WIT package version to 2.7.0. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
…eout The existing test only verifies timeout expiry and pre-queued notifications. This new test sends a notification while the task is blocked in wait_with_timeout(120s) and asserts the task completes within 10s, confirming the broadcast channel wake-up path works correctly. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
The previous implementation held the worker slot for the entire user timeout without ever suspending. Now the select loop tracks both the user-provided deadline and the config suspend_timeout. If the suspend timeout fires first, the task suspends and frees the worker (matching notification_blocking behavior). If the user timeout fires first, the function returns None. A notification arriving wakes the task promptly in either case. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
…eout Adds notify_wait_timeout_after_suspend which forces suspension via suspend_timeout(Duration::ZERO), confirms the task enters the suspended state, then sends a notification and asserts the task completes promptly (< 15s) rather than sleeping until the 120s user timeout expires. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
Three DST tests using DstScheduler, DstClock, DstEntropy, and DstEventSource: - dst_notify_timeout_suspend_then_wake: forces immediate suspension via zero suspend_timeout, confirms suspended state, injects notification via DB + event source, asserts timely completion. - dst_notify_timeout_wake_before_suspend: uses long suspend_timeout so suspension never occurs, injects notification event directly through DstEventSource while task is actively waiting, asserts prompt wake-up. - dst_notify_timeout_records_scheduler_events: verifies the DstScheduler records the expected TaskClaimed (>= 2 for initial + resume) and TaskCompleted (exactly 1) events through a full suspend/wake cycle. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
Two new deterministic simulation tests that exercise the RecvError::Lagged path in notification_blocking_timeout: - dst_notify_timeout_recovers_from_lag: inserts the real notification into the DB first, then floods the broadcast channel with 200 events to cause lag. After recovery the task re-polls the DB, finds its notification, and completes promptly. - dst_notify_timeout_recovers_from_lag_then_notified: floods the channel to cause lag *without* a pending notification. After the task recovers and re-polls (finding nothing), it loops back into the select. The real notification is then delivered normally and the task completes promptly. https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| let notif = result.expect("expected to receive a notification before timeout"); | ||
| assert_eq!(notif.event, "wakeup"); | ||
|
|
||
| print!("ok\n"); |
Check warning
Code scanning / clippy
using print!() with a format string that ends in a single newline Warning
| let data: String = notif.json().expect("invalid json data"); | ||
| assert_eq!(data, "hello timeout"); | ||
|
|
||
| print!("ok\n"); |
Check warning
Code scanning / clippy
using print!() with a format string that ends in a single newline Warning
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds support for time-bounded notification waits in the Durable SDK. Tasks can now call
wait_with_timeout(duration)to block until a notification arrives or a timeout expires, returningOption<Notification>.Key Changes
notification-blocking-timeoutfunction todurable:core/notifyinterface (version bumped to 2.7.0)RecvError::Lagged)wait_with_timeout()to bothdurable-coreanddurablecratesImplementation Details
The timeout implementation uses a
tokio::select!loop that races three conditions:When the user timeout expires, the implementation performs one final database poll before returning
None, ensuring notifications that arrived concurrently are not missed. If the suspend timeout fires first, the task is suspended to free the worker slot, and the notification is delivered via the normal resume path.The broadcast channel lag handling ensures robustness: if the channel's buffer overflows, the task recovers by re-polling the database for pending notifications.
https://claude.ai/code/session_01CxLxPCCymXVh9ob6wL5X16