feat: add non-destructive GUI event history
Add a bounded cursor-based event feed with independent client cursors, integrate it into the world client, and test GUI save workflows and event-history gaps. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
parent
c697010765
commit
e446d43b91
|
|
@ -10,7 +10,7 @@ All notable changes to this project will be documented in this file.
|
|||
running image, with message, lifecycle, breakpoint, stepping, stack, and log
|
||||
views.
|
||||
- Extend the graphical client with source editing, compilation diagnostics,
|
||||
source commits, and explicit image-event collection.
|
||||
source commits, and a non-destructive, cursor-based event-history view.
|
||||
- Add committed source history browsing, revision preview, and confirmed
|
||||
rollback to the graphical source editor.
|
||||
- Add actor-local temporary behavior editing and guarded shared-source updates
|
||||
|
|
|
|||
10
README.md
10
README.md
|
|
@ -93,11 +93,11 @@ confirmation and creates a new current source version without deleting history.
|
|||
The actor inspector can also load and edit an actor-local temporary behavior,
|
||||
save it with source-identity checks, reload the behavior from the image source,
|
||||
or explicitly confirm updating the shared source from the temporary version.
|
||||
The events view has a manual collection action:
|
||||
collecting events consumes the image's shared event queue, so other clients do
|
||||
not receive events collected by the world client. Rune frame locals and
|
||||
expression evaluation are not available. The client does not create or embed a
|
||||
separate image.
|
||||
The events view polls a bounded, non-destructive event history using its own
|
||||
cursor, so reading events does not consume them for other clients. The server
|
||||
retains up to 4096 events in memory; this history is process-local and is not
|
||||
included in image snapshots. Rune frame locals and expression evaluation are
|
||||
not available. The client does not create or embed a separate image.
|
||||
|
||||
The REPL defaults to `127.0.0.1:3737`. To select another loopback endpoint,
|
||||
use `cargo run -p runa-cli -- --connect 127.0.0.1:4000 repl` and start the server with
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
//! Trusted, in-process application services for a running Runa image.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::collections::{BTreeMap, VecDeque};
|
||||
use std::num::NonZeroUsize;
|
||||
use std::sync::Arc;
|
||||
|
||||
|
|
@ -16,6 +16,18 @@ pub struct ImageApi {
|
|||
pending_paths: BTreeMap<ActorRef, (String, ImplementationId)>,
|
||||
temporary: BTreeMap<ActorRef, SourceFile>,
|
||||
events: Vec<Event>,
|
||||
event_history: VecDeque<(u64, Event)>,
|
||||
next_event_sequence: u64,
|
||||
}
|
||||
|
||||
pub const EVENT_HISTORY_LIMIT: usize = 4096;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct EventPage {
|
||||
pub entries: Vec<(u64, Event)>,
|
||||
pub next_sequence: u64,
|
||||
pub oldest_sequence: u64,
|
||||
pub has_more: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, serde::Serialize, serde::Deserialize)]
|
||||
|
|
@ -127,6 +139,8 @@ impl ImageApi {
|
|||
pending_paths: BTreeMap::new(),
|
||||
temporary: BTreeMap::new(),
|
||||
events: Vec::new(),
|
||||
event_history: VecDeque::new(),
|
||||
next_event_sequence: 1,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -518,6 +532,30 @@ impl ImageApi {
|
|||
std::mem::take(&mut self.events)
|
||||
}
|
||||
|
||||
pub fn events_after(&mut self, after: u64, limit: usize) -> EventPage {
|
||||
self.collect_runtime_events();
|
||||
let oldest_sequence = self
|
||||
.event_history
|
||||
.front()
|
||||
.map_or(self.next_event_sequence, |(sequence, _)| *sequence);
|
||||
let mut entries = self
|
||||
.event_history
|
||||
.iter()
|
||||
.filter(|(sequence, _)| *sequence > after)
|
||||
.take(limit.saturating_add(1))
|
||||
.cloned()
|
||||
.collect::<Vec<_>>();
|
||||
let has_more = entries.len() > limit;
|
||||
entries.truncate(limit);
|
||||
let next_sequence = entries.last().map_or(after, |(sequence, _)| *sequence);
|
||||
EventPage {
|
||||
entries,
|
||||
next_sequence,
|
||||
oldest_sequence,
|
||||
has_more,
|
||||
}
|
||||
}
|
||||
|
||||
fn collect_runtime_events(&mut self) {
|
||||
for event in self.runtime.drain_events() {
|
||||
if let Event::Stopped(actor) = &event {
|
||||
|
|
@ -537,7 +575,16 @@ impl ImageApi {
|
|||
.expect("pending path exists");
|
||||
self.actor_paths.insert(*actor, path);
|
||||
}
|
||||
self.events.push(event);
|
||||
self.events.push(event.clone());
|
||||
self.event_history
|
||||
.push_back((self.next_event_sequence, event));
|
||||
self.next_event_sequence = self
|
||||
.next_event_sequence
|
||||
.checked_add(1)
|
||||
.expect("image event sequence exhausted");
|
||||
if self.event_history.len() > EVENT_HISTORY_LIMIT {
|
||||
self.event_history.pop_front();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -648,6 +695,43 @@ mod tests {
|
|||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_pages_are_independent_of_the_legacy_drain_and_paginated() -> RuntimeResult<()> {
|
||||
let mut image = ImageApi::new();
|
||||
image.write_source(
|
||||
"actors/event-source",
|
||||
None,
|
||||
"pub fn handle(ctx, state, sender, message) { state }",
|
||||
)?;
|
||||
for _ in 0..130 {
|
||||
image.spawn("actors/event-source", Value::Null, capacity(2))?;
|
||||
}
|
||||
|
||||
let first = image.events_after(0, 128);
|
||||
assert_eq!(first.entries.len(), 128);
|
||||
assert_eq!(first.entries.first().unwrap().0, 1);
|
||||
assert_eq!(first.entries.last().unwrap().0, 128);
|
||||
assert_eq!(first.next_sequence, 128);
|
||||
assert_eq!(first.oldest_sequence, 1);
|
||||
assert!(first.has_more);
|
||||
|
||||
let second = image.events_after(first.next_sequence, 128);
|
||||
assert_eq!(second.entries.len(), 2);
|
||||
assert_eq!(second.entries[0].0, 129);
|
||||
assert_eq!(second.entries[1].0, 130);
|
||||
assert_eq!(second.next_sequence, 130);
|
||||
assert!(!second.has_more);
|
||||
|
||||
assert_eq!(image.events().len(), 130);
|
||||
assert!(
|
||||
image
|
||||
.events_after(second.next_sequence, 128)
|
||||
.entries
|
||||
.is_empty()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn image_api_rejects_stale_behavior_updates_without_changing_current() -> RuntimeResult<()> {
|
||||
let mut image = ImageApi::new();
|
||||
|
|
|
|||
|
|
@ -20,6 +20,13 @@ pub struct Notification {
|
|||
pub event: ImageEvent,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct EventRecordDto {
|
||||
pub sequence: u64,
|
||||
pub event: EventDto,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(tag = "event", content = "data", rename_all = "snake_case")]
|
||||
pub enum ImageEvent {
|
||||
|
|
@ -207,6 +214,7 @@ impl Request {
|
|||
Command::GetBehaviors
|
||||
| Command::ListActors
|
||||
| Command::GetEvents
|
||||
| Command::EventsAfter { .. }
|
||||
| Command::Tick { .. } => {}
|
||||
}
|
||||
Ok(())
|
||||
|
|
@ -344,6 +352,9 @@ pub enum Command {
|
|||
ticks: usize,
|
||||
instruction_budget: usize,
|
||||
},
|
||||
EventsAfter {
|
||||
after: u64,
|
||||
},
|
||||
InspectActor {
|
||||
actor_ref: String,
|
||||
},
|
||||
|
|
@ -424,6 +435,12 @@ pub enum Reply {
|
|||
Actor(ActorSnapshotDto),
|
||||
Actors(Vec<ActorSnapshotDto>),
|
||||
Events(Vec<EventDto>),
|
||||
EventPage {
|
||||
entries: Vec<EventRecordDto>,
|
||||
next_sequence: u64,
|
||||
oldest_sequence: u64,
|
||||
has_more: bool,
|
||||
},
|
||||
TickCompleted {
|
||||
ticks: usize,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -11,8 +11,8 @@ use eframe::egui;
|
|||
use runa_client::WebSocketClient;
|
||||
use runa_core::{ActorRef, StepMode, Value};
|
||||
use runa_protocol::{
|
||||
ActorSnapshotDto, ActorStatusDto, Command, EventDto, ExecutionInspectionDto, LogEntry, Reply,
|
||||
SourceCommitDto, SourceDiagnostic, SourceFileDto, SystemMessageDto,
|
||||
ActorSnapshotDto, ActorStatusDto, Command, EventRecordDto, ExecutionInspectionDto, LogEntry,
|
||||
Reply, SourceCommitDto, SourceDiagnostic, SourceFileDto, SystemMessageDto,
|
||||
};
|
||||
|
||||
#[derive(Parser)]
|
||||
|
|
@ -67,7 +67,11 @@ enum WorkerUpdate {
|
|||
revision: String,
|
||||
source: String,
|
||||
},
|
||||
Events(Vec<EventDto>),
|
||||
Events {
|
||||
entries: Vec<EventRecordDto>,
|
||||
oldest_sequence: u64,
|
||||
next_sequence: u64,
|
||||
},
|
||||
Snapshot(Box<WorkerSnapshot>),
|
||||
}
|
||||
|
||||
|
|
@ -118,7 +122,9 @@ struct WorldApp {
|
|||
selected_revision: Option<String>,
|
||||
revision_preview: Option<(String, String)>,
|
||||
rollback_confirmation: Option<(String, String)>,
|
||||
events: Vec<EventDto>,
|
||||
events: Vec<EventRecordDto>,
|
||||
event_cursor: u64,
|
||||
missed_events: u64,
|
||||
actor_behavior: Option<SourceFileDto>,
|
||||
actor_behavior_draft: String,
|
||||
actor_behavior_dirty: bool,
|
||||
|
|
@ -166,6 +172,8 @@ impl WorldApp {
|
|||
revision_preview: None,
|
||||
rollback_confirmation: None,
|
||||
events: Vec::new(),
|
||||
event_cursor: 0,
|
||||
missed_events: 0,
|
||||
actor_behavior: None,
|
||||
actor_behavior_draft: String::new(),
|
||||
actor_behavior_dirty: false,
|
||||
|
|
@ -220,6 +228,58 @@ impl WorldApp {
|
|||
});
|
||||
}
|
||||
|
||||
fn save_source(&mut self) {
|
||||
let Some(source) = &self.source else {
|
||||
return;
|
||||
};
|
||||
if !self.source_dirty {
|
||||
return;
|
||||
}
|
||||
self.submit(Command::WriteSource {
|
||||
path: source.path.clone(),
|
||||
expected_source_id: Some(source.source_id.clone()),
|
||||
source: self.source_draft.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
fn save_actor_behavior(&mut self, actor_ref: &str) {
|
||||
let Some(source) = &self.actor_behavior else {
|
||||
return;
|
||||
};
|
||||
if !self.actor_behavior_dirty {
|
||||
return;
|
||||
}
|
||||
self.submit(Command::WriteActorBehavior {
|
||||
actor_ref: actor_ref.to_owned(),
|
||||
expected_source_id: source.source_id.clone(),
|
||||
source: self.actor_behavior_draft.clone(),
|
||||
});
|
||||
}
|
||||
|
||||
fn update_shared_behavior(&mut self, actor_ref: &str) {
|
||||
let Some(temporary) = &self.actor_behavior else {
|
||||
self.notice = "Load the actor behavior before updating the shared source".to_owned();
|
||||
return;
|
||||
};
|
||||
if self.actor_behavior_dirty {
|
||||
self.notice =
|
||||
"Save or discard the actor behavior draft before updating the source".to_owned();
|
||||
return;
|
||||
}
|
||||
let Some(expected_source_id) = self.source_ids.get(&temporary.path).cloned() else {
|
||||
self.notice =
|
||||
"Could not determine the current source identity; refresh and try again".to_owned();
|
||||
return;
|
||||
};
|
||||
self.submit(Command::UpdateBehaviorSource {
|
||||
actor_ref: actor_ref.to_owned(),
|
||||
path: temporary.path.clone(),
|
||||
expected_source_id,
|
||||
temporary_source_id: temporary.source_id.clone(),
|
||||
});
|
||||
self.update_behavior_confirmation = false;
|
||||
}
|
||||
|
||||
fn read_worker_updates(&mut self) {
|
||||
loop {
|
||||
match self.updates.try_recv() {
|
||||
|
|
@ -241,6 +301,8 @@ impl WorldApp {
|
|||
self.selected_revision = None;
|
||||
self.revision_preview = None;
|
||||
self.events.clear();
|
||||
self.event_cursor = 0;
|
||||
self.missed_events = 0;
|
||||
self.actor_behavior = None;
|
||||
self.actor_behavior_draft.clear();
|
||||
self.actor_behavior_dirty = false;
|
||||
|
|
@ -313,8 +375,18 @@ impl WorldApp {
|
|||
self.revision_preview = Some((revision, source));
|
||||
}
|
||||
}
|
||||
Ok(WorkerUpdate::Events(events)) => {
|
||||
self.events.extend(events);
|
||||
Ok(WorkerUpdate::Events {
|
||||
entries,
|
||||
oldest_sequence,
|
||||
next_sequence,
|
||||
}) => {
|
||||
if oldest_sequence > self.event_cursor.saturating_add(1) {
|
||||
self.missed_events = self
|
||||
.missed_events
|
||||
.saturating_add(oldest_sequence - self.event_cursor - 1);
|
||||
}
|
||||
self.event_cursor = next_sequence.max(self.event_cursor);
|
||||
self.events.extend(entries);
|
||||
if self.events.len() > 200 {
|
||||
self.events.drain(..self.events.len() - 200);
|
||||
}
|
||||
|
|
@ -589,13 +661,8 @@ impl WorldApp {
|
|||
egui::Button::new("Save as temporary behavior"),
|
||||
)
|
||||
.clicked()
|
||||
&& let Some(source) = &self.actor_behavior
|
||||
{
|
||||
self.submit(Command::WriteActorBehavior {
|
||||
actor_ref: actor.actor_ref.clone(),
|
||||
expected_source_id: source.source_id.clone(),
|
||||
source: self.actor_behavior_draft.clone(),
|
||||
});
|
||||
self.save_actor_behavior(&actor.actor_ref);
|
||||
}
|
||||
if ui
|
||||
.add_enabled(
|
||||
|
|
@ -645,27 +712,7 @@ impl WorldApp {
|
|||
));
|
||||
ui.horizontal(|ui| {
|
||||
if ui.button("Update source").clicked() {
|
||||
let Some(temporary) = &self.actor_behavior else {
|
||||
self.update_behavior_confirmation = false;
|
||||
return;
|
||||
};
|
||||
let expected_source_id = self.source_ids.get(&temporary.path).cloned();
|
||||
match expected_source_id {
|
||||
Some(expected_source_id) => self.submit(
|
||||
Command::UpdateBehaviorSource {
|
||||
actor_ref: actor.actor_ref.clone(),
|
||||
path: temporary.path.clone(),
|
||||
expected_source_id,
|
||||
temporary_source_id: temporary.source_id.clone(),
|
||||
},
|
||||
),
|
||||
None => {
|
||||
self.notice =
|
||||
"Could not determine the current source identity; refresh and try again"
|
||||
.to_owned()
|
||||
}
|
||||
}
|
||||
self.update_behavior_confirmation = false;
|
||||
self.update_shared_behavior(&actor.actor_ref);
|
||||
}
|
||||
if ui.button("Cancel").clicked() {
|
||||
self.update_behavior_confirmation = false;
|
||||
|
|
@ -848,13 +895,8 @@ impl WorldApp {
|
|||
egui::Button::new("Save"),
|
||||
)
|
||||
.clicked()
|
||||
&& let Some(source) = &self.source
|
||||
{
|
||||
self.submit(Command::WriteSource {
|
||||
path: source.path.clone(),
|
||||
expected_source_id: Some(source.source_id.clone()),
|
||||
source: self.source_draft.clone(),
|
||||
});
|
||||
self.save_source();
|
||||
}
|
||||
if ui
|
||||
.add_enabled(
|
||||
|
|
@ -996,16 +1038,20 @@ impl WorldApp {
|
|||
|
||||
fn draw_events(&mut self, ui: &mut egui::Ui) {
|
||||
ui.heading("Image events");
|
||||
if ui.button("Collect events").clicked() {
|
||||
self.submit(Command::GetEvents);
|
||||
}
|
||||
ui.small(
|
||||
"Collecting events drains the shared image event queue, so other clients will not receive these events.",
|
||||
ui.small("Events are polled from a non-destructive shared history feed.");
|
||||
if self.missed_events > 0 {
|
||||
ui.colored_label(
|
||||
egui::Color32::YELLOW,
|
||||
format!(
|
||||
"{} older event(s) were no longer retained by the server.",
|
||||
self.missed_events
|
||||
),
|
||||
);
|
||||
}
|
||||
ui.separator();
|
||||
egui::ScrollArea::vertical().show(ui, |ui| {
|
||||
for event in self.events.iter().rev() {
|
||||
ui.monospace(format!("{event:?}"));
|
||||
ui.monospace(format!("#{} {:?}", event.sequence, event.event));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -1058,6 +1104,7 @@ fn client_worker(requests: Receiver<WorkerRequest>, updates: Sender<WorkerUpdate
|
|||
let mut selected_behavior_path = None;
|
||||
let mut log_sequence = 0;
|
||||
let mut retained_logs = Vec::new();
|
||||
let mut event_cursor = 0;
|
||||
let mut next_poll = Instant::now();
|
||||
loop {
|
||||
match requests.recv_timeout(Duration::from_millis(50)) {
|
||||
|
|
@ -1068,6 +1115,7 @@ fn client_worker(requests: Receiver<WorkerRequest>, updates: Sender<WorkerUpdate
|
|||
selected_behavior_path = None;
|
||||
log_sequence = 0;
|
||||
retained_logs.clear();
|
||||
event_cursor = 0;
|
||||
next_poll = Instant::now();
|
||||
let _ = updates.send(WorkerUpdate::Reset);
|
||||
let _ = updates.send(WorkerUpdate::Status(format!("Connected to {address}")));
|
||||
|
|
@ -1105,7 +1153,6 @@ fn client_worker(requests: Receiver<WorkerRequest>, updates: Sender<WorkerUpdate
|
|||
Command::WriteActorBehavior { .. } => "write_actor_behavior",
|
||||
Command::UpdateBehaviorSource { .. } => "update_behavior_source",
|
||||
Command::CheckSource { .. } => "check_source",
|
||||
Command::GetEvents => "get_events",
|
||||
Command::CommitSource { .. } => "commit_source",
|
||||
Command::SourceHistory { .. } => "source_history",
|
||||
Command::RevisionSource { .. } => "revision_source",
|
||||
|
|
@ -1186,9 +1233,6 @@ fn client_worker(requests: Receiver<WorkerRequest>, updates: Sender<WorkerUpdate
|
|||
_ => None,
|
||||
}
|
||||
}
|
||||
("get_events", Reply::Events(events)) => {
|
||||
Some(WorkerUpdate::Events(events))
|
||||
}
|
||||
("commit_source", Reply::SourceCommitted(commit)) => {
|
||||
Some(WorkerUpdate::Notice(format!(
|
||||
"Committed source revision {}",
|
||||
|
|
@ -1324,6 +1368,40 @@ fn client_worker(requests: Receiver<WorkerRequest>, updates: Sender<WorkerUpdate
|
|||
} else {
|
||||
(None, Vec::new())
|
||||
};
|
||||
let event_page = match request_reply(
|
||||
connected,
|
||||
Command::EventsAfter {
|
||||
after: event_cursor,
|
||||
},
|
||||
) {
|
||||
Ok(Reply::EventPage {
|
||||
entries,
|
||||
next_sequence,
|
||||
oldest_sequence,
|
||||
has_more,
|
||||
}) => {
|
||||
event_cursor = next_sequence;
|
||||
if has_more {
|
||||
next_poll = Instant::now();
|
||||
}
|
||||
(entries, oldest_sequence, next_sequence, has_more)
|
||||
}
|
||||
Ok(reply) => {
|
||||
let _ = updates.send(WorkerUpdate::Notice(format!(
|
||||
"Unexpected event-feed response: {reply:?}"
|
||||
)));
|
||||
continue;
|
||||
}
|
||||
Err(error) => {
|
||||
report_call_error(error, &updates, &mut client);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let _ = updates.send(WorkerUpdate::Events {
|
||||
entries: event_page.0,
|
||||
oldest_sequence: event_page.1,
|
||||
next_sequence: event_page.2,
|
||||
});
|
||||
let _ = updates.send(WorkerUpdate::Snapshot(Box::new(WorkerSnapshot {
|
||||
actors,
|
||||
sources,
|
||||
|
|
@ -1570,6 +1648,54 @@ fn to_json_value(value: &Value) -> serde_json::Value {
|
|||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn test_app() -> (WorldApp, Receiver<WorkerRequest>, Sender<WorkerUpdate>) {
|
||||
let (request_tx, request_rx) = mpsc::channel();
|
||||
let (update_tx, update_rx) = mpsc::channel();
|
||||
(
|
||||
WorldApp {
|
||||
requests: request_tx,
|
||||
updates: update_rx,
|
||||
endpoint: String::new(),
|
||||
status: String::new(),
|
||||
notice: String::new(),
|
||||
actors: Vec::new(),
|
||||
sources: Vec::new(),
|
||||
source_ids: BTreeMap::new(),
|
||||
selected_actor: None,
|
||||
actor: None,
|
||||
execution: None,
|
||||
logs: Vec::new(),
|
||||
breakpoints: String::new(),
|
||||
step_budget: String::new(),
|
||||
step_mode: StepMode::Into,
|
||||
spawn_behavior: String::new(),
|
||||
spawn_state: String::new(),
|
||||
message_json: String::new(),
|
||||
scheduler_ticks: String::new(),
|
||||
scheduler_budget: String::new(),
|
||||
view: WorldView::Actors,
|
||||
source: None,
|
||||
loading_source: None,
|
||||
source_draft: String::new(),
|
||||
source_diagnostics: Vec::new(),
|
||||
source_dirty: false,
|
||||
source_history: Vec::new(),
|
||||
selected_revision: None,
|
||||
revision_preview: None,
|
||||
rollback_confirmation: None,
|
||||
events: Vec::new(),
|
||||
event_cursor: 0,
|
||||
missed_events: 0,
|
||||
actor_behavior: None,
|
||||
actor_behavior_draft: String::new(),
|
||||
actor_behavior_dirty: false,
|
||||
update_behavior_confirmation: false,
|
||||
},
|
||||
request_rx,
|
||||
update_tx,
|
||||
)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_nested_json_and_actor_reference_markers() {
|
||||
let actor_ref = ActorRef::new();
|
||||
|
|
@ -1624,4 +1750,118 @@ mod tests {
|
|||
.any(|section| &source[section.byte_range.clone()] == "// note")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn source_save_uses_loaded_identity_and_does_not_overwrite_newer_edits() {
|
||||
let (mut app, requests, updates) = test_app();
|
||||
app.source = Some(SourceFileDto {
|
||||
path: "actors/counter.rn".to_owned(),
|
||||
source_id: "source-v1".to_owned(),
|
||||
source: "old".to_owned(),
|
||||
});
|
||||
app.source_draft = "submitted".to_owned();
|
||||
app.source_dirty = true;
|
||||
|
||||
app.save_source();
|
||||
|
||||
assert!(matches!(
|
||||
requests.try_recv(),
|
||||
Ok(WorkerRequest::Command(Command::WriteSource {
|
||||
path,
|
||||
expected_source_id: Some(expected_source_id),
|
||||
source,
|
||||
})) if path == "actors/counter.rn"
|
||||
&& expected_source_id == "source-v1"
|
||||
&& source == "submitted"
|
||||
));
|
||||
|
||||
app.source_draft = "newer local edit".to_owned();
|
||||
updates
|
||||
.send(WorkerUpdate::SourceSaved {
|
||||
source: SourceFileDto {
|
||||
path: "actors/counter.rn".to_owned(),
|
||||
source_id: "source-v2".to_owned(),
|
||||
source: "submitted".to_owned(),
|
||||
},
|
||||
submitted_source: "submitted".to_owned(),
|
||||
})
|
||||
.unwrap();
|
||||
app.read_worker_updates();
|
||||
|
||||
assert_eq!(app.source_draft, "newer local edit");
|
||||
assert!(app.source_dirty);
|
||||
assert_eq!(
|
||||
app.source.as_ref().map(|source| source.source_id.as_str()),
|
||||
Some("source-v2")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn actor_behavior_actions_use_compare_and_swap_identities() {
|
||||
let (mut app, requests, _) = test_app();
|
||||
let actor_ref = ActorRef::new().to_string();
|
||||
app.actor_behavior = Some(SourceFileDto {
|
||||
path: "actors/counter.rn".to_owned(),
|
||||
source_id: "temporary-v1".to_owned(),
|
||||
source: "old behavior".to_owned(),
|
||||
});
|
||||
app.actor_behavior_draft = "new behavior".to_owned();
|
||||
app.actor_behavior_dirty = true;
|
||||
|
||||
app.save_actor_behavior(&actor_ref);
|
||||
|
||||
assert!(matches!(
|
||||
requests.try_recv(),
|
||||
Ok(WorkerRequest::Command(Command::WriteActorBehavior {
|
||||
actor_ref: requested_actor,
|
||||
expected_source_id,
|
||||
source,
|
||||
})) if requested_actor == actor_ref
|
||||
&& expected_source_id == "temporary-v1"
|
||||
&& source == "new behavior"
|
||||
));
|
||||
|
||||
app.actor_behavior_dirty = false;
|
||||
app.source_ids
|
||||
.insert("actors/counter.rn".to_owned(), "shared-v3".to_owned());
|
||||
app.update_shared_behavior(&actor_ref);
|
||||
|
||||
assert!(matches!(
|
||||
requests.try_recv(),
|
||||
Ok(WorkerRequest::Command(Command::UpdateBehaviorSource {
|
||||
actor_ref: requested_actor,
|
||||
path,
|
||||
expected_source_id,
|
||||
temporary_source_id,
|
||||
})) if requested_actor == actor_ref
|
||||
&& path == "actors/counter.rn"
|
||||
&& expected_source_id == "shared-v3"
|
||||
&& temporary_source_id == "temporary-v1"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_history_reports_retention_gaps_without_double_counting() {
|
||||
let (mut app, _, updates) = test_app();
|
||||
updates
|
||||
.send(WorkerUpdate::Events {
|
||||
entries: Vec::new(),
|
||||
oldest_sequence: 7,
|
||||
next_sequence: 8,
|
||||
})
|
||||
.unwrap();
|
||||
app.read_worker_updates();
|
||||
assert_eq!(app.missed_events, 6);
|
||||
assert_eq!(app.event_cursor, 8);
|
||||
|
||||
updates
|
||||
.send(WorkerUpdate::Events {
|
||||
entries: Vec::new(),
|
||||
oldest_sequence: 7,
|
||||
next_sequence: 8,
|
||||
})
|
||||
.unwrap();
|
||||
app.read_worker_updates();
|
||||
assert_eq!(app.missed_events, 6);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -405,6 +405,22 @@ impl Server {
|
|||
Command::GetEvents => Ok(Reply::Events(
|
||||
self.image.events().iter().map(event_dto).collect(),
|
||||
)),
|
||||
Command::EventsAfter { after } => {
|
||||
let page = self.image.events_after(after, 128);
|
||||
Ok(Reply::EventPage {
|
||||
entries: page
|
||||
.entries
|
||||
.into_iter()
|
||||
.map(|(sequence, event)| runa_protocol::EventRecordDto {
|
||||
sequence,
|
||||
event: event_dto(&event),
|
||||
})
|
||||
.collect(),
|
||||
next_sequence: page.next_sequence,
|
||||
oldest_sequence: page.oldest_sequence,
|
||||
has_more: page.has_more,
|
||||
})
|
||||
}
|
||||
Command::StopActor { actor_ref } => {
|
||||
let actor_ref = parse_actor_ref(&actor_ref)?;
|
||||
self.image
|
||||
|
|
|
|||
|
|
@ -74,7 +74,7 @@ async fn serve(listener: TcpListener, mut server: Server) -> io::Result<()> {
|
|||
let mutates = !matches!(dispatch.request.command,
|
||||
Command::GetBehavior { .. } | Command::GetBehaviors |
|
||||
Command::GetBehaviorSource { .. } | Command::InspectActor { .. } |
|
||||
Command::ListActors | Command::GetEvents | Command::ListSources |
|
||||
Command::ListActors | Command::GetEvents | Command::EventsAfter { .. } | Command::ListSources |
|
||||
Command::ReadSource { .. } | Command::SourceHistory { .. } |
|
||||
Command::RevisionSource { .. } | Command::ActorBehavior { .. } | Command::CheckSource { .. } | Command::InspectExecution { .. } | Command::Logs { .. });
|
||||
let response = server.dispatch(dispatch.request);
|
||||
|
|
|
|||
|
|
@ -394,6 +394,82 @@ fn websocket_broadcasts_non_destructive_invalidations_for_mailbox_state_and_life
|
|||
assert_eq!(response.request_id, "read-fence");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn event_history_feed_is_non_destructive_and_has_independent_client_cursors() {
|
||||
let (_server, address) = start_server();
|
||||
let mut first = WebSocketClient::connect(address).unwrap();
|
||||
let mut second = WebSocketClient::connect(address).unwrap();
|
||||
let actor_ref = match first
|
||||
.request(Command::Spawn {
|
||||
behavior_name: "actors/counter".to_owned(),
|
||||
state: Value::Integer(0),
|
||||
mailbox_capacity: 4,
|
||||
})
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap()
|
||||
{
|
||||
Reply::ActorSpawned { actor_ref } => actor_ref,
|
||||
other => panic!("unexpected spawn: {other:?}"),
|
||||
};
|
||||
|
||||
let Reply::EventPage {
|
||||
entries: first_page,
|
||||
next_sequence,
|
||||
has_more: false,
|
||||
..
|
||||
} = first
|
||||
.request(Command::EventsAfter { after: 0 })
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("expected first non-destructive event page")
|
||||
};
|
||||
assert!(first_page.iter().any(|entry| {
|
||||
entry.event
|
||||
== (runa_protocol::EventDto::Spawned {
|
||||
actor_ref: actor_ref.clone(),
|
||||
})
|
||||
}));
|
||||
|
||||
let Reply::EventPage {
|
||||
entries: second_page,
|
||||
next_sequence: second_cursor,
|
||||
..
|
||||
} = second
|
||||
.request(Command::EventsAfter { after: 0 })
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("expected independent event page")
|
||||
};
|
||||
assert_eq!(second_page, first_page);
|
||||
assert_eq!(second_cursor, next_sequence);
|
||||
|
||||
let Reply::Events(legacy_events) = first.request(Command::GetEvents).unwrap().result.unwrap()
|
||||
else {
|
||||
panic!("expected legacy event drain")
|
||||
};
|
||||
assert!(
|
||||
legacy_events
|
||||
.iter()
|
||||
.any(|event| matches!(event, runa_protocol::EventDto::Spawned { actor_ref: id } if id == &actor_ref))
|
||||
);
|
||||
let Reply::EventPage { entries, .. } = second
|
||||
.request(Command::EventsAfter {
|
||||
after: second_cursor,
|
||||
})
|
||||
.unwrap()
|
||||
.result
|
||||
.unwrap()
|
||||
else {
|
||||
panic!("expected empty cursor page")
|
||||
};
|
||||
assert!(entries.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn websocket_requires_subprotocol_and_rejects_browser_origins_without_blocking_other_clients() {
|
||||
let (_server, address) = start_server();
|
||||
|
|
|
|||
|
|
@ -226,9 +226,11 @@ step, and execution-stack controls, plus a source editor with compilation
|
|||
diagnostics, explicit source commits, committed-revision previews, and
|
||||
confirmed rollback. The actor inspector can edit an actor-local temporary
|
||||
behavior and explicitly update its shared source or reload from that source.
|
||||
Image events can be collected manually; this operation drains the shared queue
|
||||
and therefore consumes events for other clients too. Network requests run on a
|
||||
worker thread, not the UI frame loop. This is a client foundation, not yet a
|
||||
The events view polls a bounded, non-destructive event history with an
|
||||
independent cursor per client; reading events does not consume them for other
|
||||
clients. The server retains up to 4096 events in memory, and this process-local
|
||||
history is not included in image snapshots. Network requests run on a worker
|
||||
thread, not the UI frame loop. This is a client foundation, not yet a
|
||||
Rune-composed graphical world; frame locals and expression evaluation remain
|
||||
unavailable.
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue