feat(runtime): expose resident driver boundaries
Expose exact Actor instance identity and manifest mailbox capacity so Host supervisors can distinguish idempotent starts from fault recovery and preserve revision-specific limits. Expose resident stream chunk and endpoint-policy validation to drivers, and mark terminal streams closing before dispatching their final callback. Cover Actor identity reuse and replacement in the Runtime integration suite.
This commit is contained in:
@@ -515,6 +515,15 @@ impl ResidentSession {
|
||||
self.actor.key()
|
||||
}
|
||||
|
||||
/// Returns the largest stream chunk that a driver may deliver.
|
||||
///
|
||||
/// A stream driver should size its read buffer to this value or a smaller
|
||||
/// protocol-specific limit. The Runtime applies the same limit to both
|
||||
/// incoming `stream-data` events and Component `write-stream` effects.
|
||||
pub fn max_stream_chunk_bytes(&self) -> usize {
|
||||
self.actor.resident_stream_chunk_limit()
|
||||
}
|
||||
|
||||
/// Registers a configured listener or datagram endpoint after policy checks.
|
||||
///
|
||||
/// Registration does not bind an operating-system socket. The driver binds
|
||||
@@ -664,6 +673,7 @@ impl ResidentSession {
|
||||
reason: StreamCloseReason,
|
||||
) -> Result<Vec<ResidentOperation>, ResidentHostError> {
|
||||
self.ensure_kind(stream_id, ResidentResourceKind::Stream, "stream")?;
|
||||
*self.stream_state_mut(stream_id)? = StreamState::Closing;
|
||||
let result = self.dispatch(ResidentEvent::StreamClosed { stream_id, reason });
|
||||
self.resources.remove(&stream_id);
|
||||
result
|
||||
@@ -1056,7 +1066,13 @@ impl ResidentSession {
|
||||
}
|
||||
|
||||
impl ResidentPolicy {
|
||||
fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> {
|
||||
/// Validates one endpoint without registering or binding it.
|
||||
///
|
||||
/// Management layers can call this while loading static configuration so
|
||||
/// an address outside policy fails before any Actor is started. Drivers
|
||||
/// must still register the endpoint immediately before bind because policy
|
||||
/// is also a runtime ownership boundary.
|
||||
pub fn validate_endpoint(&self, endpoint: &ResidentEndpoint) -> Result<(), ResidentHostError> {
|
||||
match endpoint {
|
||||
ResidentEndpoint::Tcp { bind, .. } if !self.tcp_listen.allows(*bind) => {
|
||||
Err(ResidentHostError::EndpointDenied(format!(
|
||||
|
||||
@@ -615,10 +615,33 @@ impl ActorHandle {
|
||||
self.execution
|
||||
}
|
||||
|
||||
/// Returns whether two handles address the exact same Actor instance.
|
||||
///
|
||||
/// Service keys are revision identities and therefore survive an Actor
|
||||
/// fault/restart. Host resource supervisors use this stronger check to
|
||||
/// retain themselves for an idempotent start but replace stale resources
|
||||
/// when the Runtime created a fresh Store for the same revision.
|
||||
pub fn is_same_instance(&self, other: &Self) -> bool {
|
||||
Arc::ptr_eq(&self.status, &other.status)
|
||||
}
|
||||
|
||||
/// Returns the immutable mailbox capacity configured for this Actor.
|
||||
///
|
||||
/// Host supervisors use the same bound for their external event queue so a
|
||||
/// restored revision retains the limits stored in its own manifest instead
|
||||
/// of inheriting newer process defaults.
|
||||
pub fn mailbox_capacity(&self) -> usize {
|
||||
self.limits.mailbox_capacity
|
||||
}
|
||||
|
||||
pub(crate) fn resident_resource_limit(&self) -> usize {
|
||||
self.limits.resident.max_resources
|
||||
}
|
||||
|
||||
pub(crate) fn resident_stream_chunk_limit(&self) -> usize {
|
||||
self.limits.resident.max_stream_chunk_bytes
|
||||
}
|
||||
|
||||
/// Sends an invocation and waits up to the remaining service deadline.
|
||||
///
|
||||
/// Cloned handles compete for the same bounded mailbox. A Wasm trap is
|
||||
|
||||
@@ -96,12 +96,16 @@ fn faulted_actor_can_be_started_again() {
|
||||
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||
let (key, actor) = start_component(&runtime, "fault", "fault_component.wasm");
|
||||
|
||||
let already_running = runtime.start(&key, Vec::new()).unwrap();
|
||||
assert!(actor.is_same_instance(&already_running));
|
||||
|
||||
assert!(matches!(
|
||||
actor.invoke(b"fault".to_vec()),
|
||||
Err(RuntimeError::ActorFault { .. })
|
||||
));
|
||||
|
||||
let restarted = runtime.start(&key, Vec::new()).unwrap();
|
||||
assert!(!actor.is_same_instance(&restarted));
|
||||
assert_eq!(restarted.invoke(b"ready".to_vec()).unwrap(), b"ready");
|
||||
|
||||
runtime.stop(&key).unwrap();
|
||||
|
||||
Reference in New Issue
Block a user