feat: ask pattern for typed request-response (Cycle 15)
Add Runtime::ask() and Ask<R> wrapper for convenient request-response. Creates a temporary inbox, sends the request (with reply address via closure), and provides recv_ticking() for automatic tick-until-response. Purely sugar over the existing inbox pattern — no implicit auto-reply. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
4d18874909
commit
902471b1f4
3 changed files with 166 additions and 1 deletions
|
|
@ -2,7 +2,7 @@
|
||||||
|
|
||||||
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
## Current Stage: Phase 1 — Research + First Improvement Cycle
|
||||||
|
|
||||||
### Status: Cycle 14 COMPLETE
|
### Status: Cycle 15 COMPLETE
|
||||||
|
|
||||||
## Plan Overview
|
## Plan Overview
|
||||||
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
|
1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅
|
||||||
|
|
@ -124,6 +124,28 @@
|
||||||
- `stop_nonexistent_actor_returns_error` — stop on bad address returns Err
|
- `stop_nonexistent_actor_returns_error` — stop on bad address returns Err
|
||||||
- **Result**: 82 tests pass, all workspace compiles
|
- **Result**: 82 tests pass, all workspace compiles
|
||||||
|
|
||||||
|
### Cycle 15: Ask Pattern (Request-Response)
|
||||||
|
- **Research**: Studied ask/call/request-response patterns across Erlang gen_server:call (From + reply),
|
||||||
|
Akka ask (temporary actor + Future), Ractor call (RpcReplyPort), Kameo ask (async + Reply trait),
|
||||||
|
xactor Handler (return value auto-routing)
|
||||||
|
- Key finding: swactor's synchronous tick model requires explicit reply_to, not implicit routing
|
||||||
|
- Decision: convenience wrapper over existing inbox pattern, not implicit auto-reply
|
||||||
|
- **Implementation**: `Ask<R>` struct + `Runtime::ask()` method
|
||||||
|
- `Ask<R>`: wraps `Inbox<R>` with `try_recv()` and `recv_ticking(rt, max_ticks)`
|
||||||
|
- `rt.ask(addr, |reply_to| Msg { reply_to })` — creates inbox, builds message, sends, returns Ask<R>
|
||||||
|
- `ask.recv_ticking(&rt, max_ticks)` — ticks until response or timeout (single-threaded only)
|
||||||
|
- `ask.try_recv()` — poll without ticking (works in both modes)
|
||||||
|
- `ask.reply_addr()` — access inbox address for manual use
|
||||||
|
- Purely sugar over `new_inbox → send_to → tick → try_recv` pattern
|
||||||
|
- Zero changes to ContextInner or ActorInterface — no implicit auto-reply magic
|
||||||
|
- **Tests**: 5 new behavioral tests
|
||||||
|
- `ask_recv_ticking_returns_response` — basic PingPong ask roundtrip
|
||||||
|
- `ask_multiple_times_tracks_state` — 3 sequential asks to CounterActor
|
||||||
|
- `ask_timeout_when_no_response` — ask dead actor → timeout error
|
||||||
|
- `ask_try_recv_returns_none_before_tick` — poll before tick → None, after tick → Some
|
||||||
|
- `ask_reply_addr_is_accessible` — reply address is valid
|
||||||
|
- **Result**: 127 tests pass (120 behavioral + 7 proptest), all workspace compiles, zero warnings
|
||||||
|
|
||||||
### Cycle 14: Actor Groups (Pub-Sub)
|
### Cycle 14: Actor Groups (Pub-Sub)
|
||||||
- **Research**: Studied group/pub-sub patterns across Erlang pg (scopes, join/leave/get_members),
|
- **Research**: Studied group/pub-sub patterns across Erlang pg (scopes, join/leave/get_members),
|
||||||
Akka DistributedPubSub (mediator, topics), Ractor pg (join/leave/broadcast), Bastion (Dispatcher),
|
Akka DistributedPubSub (mediator, topics), Ractor pg (join/leave/broadcast), Bastion (Dispatcher),
|
||||||
|
|
|
||||||
|
|
@ -32,6 +32,39 @@ impl<M: Message> Inbox<M> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Pending ask response — wraps an inbox with convenience recv methods.
|
||||||
|
///
|
||||||
|
/// Created by [`Runtime::ask`]. Provides `try_recv()` for polling and
|
||||||
|
/// `recv_ticking()` for automatic tick-until-response.
|
||||||
|
pub struct Ask<R: Message> {
|
||||||
|
inbox: Inbox<R>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<R: Message> Ask<R> {
|
||||||
|
/// Try to receive the response without ticking.
|
||||||
|
pub fn try_recv(&self) -> Option<R> {
|
||||||
|
self.inbox.try_recv()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Tick the runtime until a response arrives or `max_ticks` is exhausted.
|
||||||
|
///
|
||||||
|
/// Only valid for single-threaded runtimes (panics if `num_threads >= 2`).
|
||||||
|
pub fn recv_ticking(&self, rt: &Runtime, max_ticks: usize) -> Result<R, Error> {
|
||||||
|
for _ in 0..max_ticks {
|
||||||
|
rt.tick();
|
||||||
|
if let Some(resp) = self.inbox.try_recv() {
|
||||||
|
return Ok(resp);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(Error::from("ask timeout: no response within max_ticks"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the reply address (for manual message construction).
|
||||||
|
pub fn reply_addr(&self) -> &ActorAddress {
|
||||||
|
self.inbox.addr()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Handle for dealing with a runtime that has started via the `Runtime::run()` method.
|
/// Handle for dealing with a runtime that has started via the `Runtime::run()` method.
|
||||||
pub struct RuntimeHandle {
|
pub struct RuntimeHandle {
|
||||||
pub runtime: Arc<Runtime>,
|
pub runtime: Arc<Runtime>,
|
||||||
|
|
@ -301,6 +334,27 @@ impl Runtime {
|
||||||
self.group_registry.group_names()
|
self.group_registry.group_names()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send a request and get a handle for the response.
|
||||||
|
///
|
||||||
|
/// Creates a temporary inbox, calls `msg_builder` with the inbox's address
|
||||||
|
/// (so you can embed it as `reply_to`), sends the message, and returns an
|
||||||
|
/// [`Ask`] handle for receiving the response.
|
||||||
|
///
|
||||||
|
/// ```ignore
|
||||||
|
/// let ask = rt.ask(actor, |reply_to| GetValue { reply_to })?;
|
||||||
|
/// let value = ask.recv_ticking(&rt, 10)?;
|
||||||
|
/// ```
|
||||||
|
pub fn ask<Req: Message, Resp: Message>(
|
||||||
|
&self,
|
||||||
|
addr: ActorAddress,
|
||||||
|
msg_builder: impl FnOnce(ActorAddress) -> Req,
|
||||||
|
) -> Result<Ask<Resp>, Error> {
|
||||||
|
let inbox = self.new_inbox::<Resp>()?;
|
||||||
|
let msg = msg_builder(*inbox.addr());
|
||||||
|
self.send_to(addr, msg)?;
|
||||||
|
Ok(Ask { inbox })
|
||||||
|
}
|
||||||
|
|
||||||
/// Send a message to an actor address
|
/// Send a message to an actor address
|
||||||
pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
|
pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
|
||||||
let result = self.send_any(addr, Box::new(msg));
|
let result = self.send_any(addr, Box::new(msg));
|
||||||
|
|
|
||||||
|
|
@ -3548,3 +3548,92 @@ fn ctx_publish_broadcasts_from_handler() {
|
||||||
}
|
}
|
||||||
assert!(pong_count >= 2, "at least 2 PingPong members should reply, got {pong_count}");
|
assert!(pong_count >= 2, "at least 2 PingPong members should reply, got {pong_count}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── Ask Pattern ─────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
/// Given a PingPong actor,
|
||||||
|
/// when I ask with recv_ticking,
|
||||||
|
/// then I get the Pong response.
|
||||||
|
#[test]
|
||||||
|
fn ask_recv_ticking_returns_response() {
|
||||||
|
let rt = Runtime::new(RuntimeConfig::default());
|
||||||
|
let actor = rt.spawn(PingPongActor).unwrap();
|
||||||
|
rt.tick(); // on_start
|
||||||
|
|
||||||
|
let pong: Pong = rt.ask(actor, |reply_to| Ping { reply_to })
|
||||||
|
.unwrap()
|
||||||
|
.recv_ticking(&rt, 10)
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(pong, Pong);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Given a CounterActor,
|
||||||
|
/// when I ask multiple times,
|
||||||
|
/// then each response reflects the updated state.
|
||||||
|
#[test]
|
||||||
|
fn ask_multiple_times_tracks_state() {
|
||||||
|
let rt = Runtime::new(RuntimeConfig::default());
|
||||||
|
let actor = rt.spawn(CounterActor { count: 0 }).unwrap();
|
||||||
|
rt.tick(); // on_start
|
||||||
|
|
||||||
|
let c1: Count = rt.ask(actor, |reply_to| Increment { reply_to })
|
||||||
|
.unwrap().recv_ticking(&rt, 10).unwrap();
|
||||||
|
let c2: Count = rt.ask(actor, |reply_to| Increment { reply_to })
|
||||||
|
.unwrap().recv_ticking(&rt, 10).unwrap();
|
||||||
|
let c3: Count = rt.ask(actor, |reply_to| Increment { reply_to })
|
||||||
|
.unwrap().recv_ticking(&rt, 10).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(c1, Count(1));
|
||||||
|
assert_eq!(c2, Count(2));
|
||||||
|
assert_eq!(c3, Count(3));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Given a dead actor,
|
||||||
|
/// when I ask and tick,
|
||||||
|
/// then recv_ticking returns a timeout error.
|
||||||
|
#[test]
|
||||||
|
fn ask_timeout_when_no_response() {
|
||||||
|
let rt = Runtime::new(RuntimeConfig::default());
|
||||||
|
let actor = rt.spawn(PingPongActor).unwrap();
|
||||||
|
rt.tick();
|
||||||
|
rt.stop_actor(actor).unwrap();
|
||||||
|
rt.tick(); // actor dies
|
||||||
|
|
||||||
|
// Ask the dead actor — message is undeliverable, no response
|
||||||
|
let result = rt.ask::<Ping, Pong>(actor, |reply_to| Ping { reply_to });
|
||||||
|
// send_to may succeed (message goes to transfer queue) or fail (addr removed)
|
||||||
|
// Either way, no response will come
|
||||||
|
if let Ok(ask) = result {
|
||||||
|
let err = ask.recv_ticking(&rt, 5);
|
||||||
|
assert!(err.is_err(), "should timeout with no response");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Given an ask handle,
|
||||||
|
/// when I use try_recv before ticking,
|
||||||
|
/// then it returns None (response hasn't arrived yet).
|
||||||
|
#[test]
|
||||||
|
fn ask_try_recv_returns_none_before_tick() {
|
||||||
|
let rt = Runtime::new(RuntimeConfig::default());
|
||||||
|
let actor = rt.spawn(PingPongActor).unwrap();
|
||||||
|
rt.tick(); // on_start
|
||||||
|
|
||||||
|
let ask = rt.ask::<Ping, Pong>(actor, |reply_to| Ping { reply_to }).unwrap();
|
||||||
|
assert!(ask.try_recv().is_none(), "no response before ticking");
|
||||||
|
|
||||||
|
rt.tick(); // process message
|
||||||
|
assert_eq!(ask.try_recv(), Some(Pong));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Given an ask, the reply_addr() returns the inbox address for manual use.
|
||||||
|
#[test]
|
||||||
|
fn ask_reply_addr_is_accessible() {
|
||||||
|
let rt = Runtime::new(RuntimeConfig::default());
|
||||||
|
let actor = rt.spawn(PingPongActor).unwrap();
|
||||||
|
rt.tick();
|
||||||
|
|
||||||
|
let ask = rt.ask::<Ping, Pong>(actor, |reply_to| Ping { reply_to }).unwrap();
|
||||||
|
let addr = *ask.reply_addr();
|
||||||
|
// The address should be valid (non-zero)
|
||||||
|
assert_ne!(addr, ActorAddress::default());
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue