scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(run): make attach stream terminal-authoritative

Browse source 
Replay persisted run events for attach requests, keep the SSE stream live
only while the run is active, and close on terminal run events instead of
returning 410 for completed runs.

The CLI now treats premature attach EOF as an error, and the affected
integration tests were stabilized around store-backed event ordering and
recovered rewind timelines.
This commit is contained in:
Bryan Helmkamp 2026-04-07 14:32:23 -04:00
parent 6c5ea35fd1
commit 3dcf2ad04b
No known key found for this signature in database
13 changed files with 478 additions and 279 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docs/api-reference/fabro-api.yaml
View file

@ -529,7 +529,7 @@ paths:
operationId: attachRunEvents
tags: [Run Internals]
summary: Attach Run Events
description: Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
description: Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
parameters:
- $ref: "#/components/parameters/RunId"
- $ref: "#/components/parameters/SinceSeq"
@ -546,12 +546,6 @@ paths:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"410":
description: Run is not live on this server
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/blobs:
post:

209
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -4,7 +4,7 @@ use std::path::Path;
#[cfg(test)]
use std::path::PathBuf;
use std::process::ExitCode;
use std::time::{Duration, Instant};
use std::time::Duration;
use anyhow::Result;
use fabro_types::{EventBody, RunEvent, RunId};
@ -17,7 +17,7 @@ use fabro_util::terminal::Styles;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::run_status::RunStatus;
use tokio::signal::ctrl_c;
use tokio::time::{sleep, timeout};
use tokio::time::sleep;
use super::run_progress;
use crate::server_client;
@ -25,10 +25,7 @@ use crate::server_client;
const INTERVIEW_UNANSWERED_MESSAGE: &str =
"Interview ended without an answer. The run is still waiting for input; reattach to answer it.";
const JSON_INTERVIEW_MESSAGE: &str = "This run is waiting for human input, but --json is non-interactive. Reattach without --json to answer it.";
#[cfg(test)]
const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_millis(250);
#[cfg(not(test))]
const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_secs(2);
const ATTACH_PREMATURE_EOF_MESSAGE: &str = "Attach stream ended before terminal run event.";
/// Attach to a running (or finished) workflow run, rendering progress live.
///
@ -73,57 +70,51 @@ pub(crate) async fn attach_run_with_client(
let replay_events = events.clone();
let next_seq = events.last().map_or(1, |event| event.seq.saturating_add(1));
let initial_exit_code = events.iter().rev().find_map(event_exit_code);
let state_exit_code = state_exit_code(&state);
if state_is_terminal(&state) || initial_exit_code.is_some() {
return replay_run_with_client(client, run_id, verbose, events, json_output).await;
return replay_run_with_client(
verbose,
events,
initial_exit_code
.or(state_exit_code)
.unwrap_or(ExitCode::from(1)),
json_output,
)
.await;
}
match client.attach_run_events(run_id, Some(next_seq)).await {
Ok(stream) => {
attach_live_run_with_client(
client,
run_id,
verbose,
events,
stream,
kill_on_detach,
styles,
json_output,
)
.await
}
Err(server_client::RunAttachStreamError::Gone) => {
replay_run_with_client(client, run_id, verbose, replay_events, json_output).await
}
Err(server_client::RunAttachStreamError::Other(err)) => Err(err),
}
let stream = client.attach_run_events(run_id, Some(next_seq)).await?;
attach_live_run_with_client(
client,
run_id,
verbose,
replay_events,
stream,
kill_on_detach,
styles,
json_output,
)
.await
}
async fn replay_run_with_client(
client: &server_client::ServerStoreClient,
run_id: &RunId,
verbose: bool,
events: Vec<EventEnvelope>,
exit_code: ExitCode,
json_output: bool,
) -> Result<ExitCode> {
let is_tty = std::io::stderr().is_terminal();
let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose);
let mut terminal_exit_code = None;
for event in events {
if let Some(exit_code) = event_exit_code(&event) {
terminal_exit_code = Some(exit_code);
}
let line = event_payload_line(&event)?;
emit_progress_line(&mut progress_ui, &line, json_output)?;
}
finish_progress(&mut progress_ui, json_output);
Ok(match terminal_exit_code {
Some(exit_code) => exit_code,
None => determine_exit_code_with_server(client, run_id).await,
})
Ok(exit_code)
}
async fn attach_live_run_with_client(
@ -141,16 +132,7 @@ async fn attach_live_run_with_client(
let ctrl_c_signal = ctrl_c();
tokio::pin!(ctrl_c_signal);
let mut next_seq = 1;
let mut terminal_exit_code = None;
let mut terminal_event_seen_at: Option<Instant> = None;
for event in existing_events {
next_seq = event.seq.saturating_add(1);
if let Some(exit_code) = event_exit_code(&event) {
terminal_exit_code = Some(exit_code);
terminal_event_seen_at = Some(Instant::now());
}
let line = event_payload_line(&event)?;
emit_progress_line(&mut progress_ui, &line, json_output)?;
}
@ -163,46 +145,28 @@ async fn attach_live_run_with_client(
}
loop {
let next_event = if let Some(seen_at) = terminal_event_seen_at {
let remaining = ATTACH_FINAL_STATUS_GRACE.saturating_sub(seen_at.elapsed());
if remaining.is_zero() {
break;
}
tokio::select! {
_ = &mut ctrl_c_signal => {
handle_detach_signal(client, run_id, kill_on_detach).await;
break;
}
result = timeout(remaining, stream.next_event()) => {
match result {
Ok(result) => result?,
Err(_) => break,
}
}
}
} else {
tokio::select! {
_ = &mut ctrl_c_signal => {
handle_detach_signal(client, run_id, kill_on_detach).await;
break;
}
result = stream.next_event() => result?,
let next_event = tokio::select! {
_ = &mut ctrl_c_signal => {
handle_detach_signal(client, run_id, kill_on_detach).await;
finish_progress(&mut progress_ui, json_output);
return Ok(ExitCode::from(1));
}
result = stream.next_event() => result?,
};
let Some(event) = next_event else {
break;
finish_progress(&mut progress_ui, json_output);
return Err(anyhow::anyhow!(ATTACH_PREMATURE_EOF_MESSAGE));
};
next_seq = event.seq.saturating_add(1);
if let Some(exit_code) = event_exit_code(&event) {
terminal_exit_code = Some(exit_code);
terminal_event_seen_at = Some(Instant::now());
}
let line = event_payload_line(&event)?;
emit_progress_line(&mut progress_ui, &line, json_output)?;
if let Some(exit_code) = event_exit_code(&event) {
finish_progress(&mut progress_ui, json_output);
return Ok(exit_code);
}
if event_starts_interview(&event) {
if let Some(exit_code) = handle_pending_server_interview(
client,
@ -217,20 +181,6 @@ async fn attach_live_run_with_client(
}
}
}
if terminal_exit_code.is_none() {
let (_, trailing_exit_code) =
emit_server_events_from(client, run_id, next_seq, &mut progress_ui, json_output)
.await?;
terminal_exit_code = trailing_exit_code;
}
finish_progress(&mut progress_ui, json_output);
Ok(match terminal_exit_code {
Some(exit_code) => exit_code,
None => determine_exit_code_with_server(client, run_id).await,
})
}
async fn handle_pending_server_interview(
@ -287,33 +237,6 @@ async fn handle_detach_signal(
}
}
async fn emit_server_events_from(
client: &server_client::ServerStoreClient,
run_id: &RunId,
next_seq: u32,
progress_ui: &mut run_progress::ProgressUI,
json_output: bool,
) -> Result<(u32, Option<ExitCode>)> {
let events = match client.list_run_events(run_id, Some(next_seq), None).await {
Ok(events) => events,
Err(err) if is_run_not_found_error(&err) => Vec::new(),
Err(err) => return Err(err),
};
let mut current_seq = next_seq;
let mut terminal_exit_code = None;
for event in events {
if let Some(exit_code) = event_exit_code(&event) {
terminal_exit_code = Some(exit_code);
}
let line = event_payload_line(&event)?;
emit_progress_line(progress_ui, &line, json_output)?;
current_seq = event.seq.saturating_add(1);
}
Ok((current_seq, terminal_exit_code))
}
fn api_question_to_question(question: &types::ApiQuestion) -> Question {
let question_type = match question.question_type {
types::QuestionType::YesNo => QuestionType::YesNo,
@ -363,11 +286,6 @@ async fn submit_server_interview_answer(
Ok(true)
}
fn is_run_not_found_error(err: &anyhow::Error) -> bool {
err.chain()
.any(|cause| cause.to_string() == "Run not found.")
}
fn state_is_terminal(state: &server_client::RunProjection) -> bool {
state.conclusion.is_some()
|| state
@ -455,38 +373,23 @@ fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool {
matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped)
}
async fn determine_exit_code_with_server(
client: &server_client::ServerStoreClient,
run_id: &RunId,
) -> ExitCode {
let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE;
loop {
if let Ok(state) = client.get_run_state(run_id).await {
if let Some(conclusion) = state.conclusion {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
};
}
fn state_exit_code(state: &server_client::RunProjection) -> Option<ExitCode> {
if let Some(conclusion) = &state.conclusion {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return Some(if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
});
}
match state.status {
Some(record) if matches!(record.status, RunStatus::Succeeded) => {
return ExitCode::from(0);
}
Some(record) if record.status.is_terminal() => return ExitCode::from(1),
Some(_) | None => {}
}
}
if Instant::now() >= deadline {
return ExitCode::from(1);
}
sleep(Duration::from_millis(100)).await;
match state.status.as_ref() {
Some(record) if record.status == RunStatus::Succeeded => Some(ExitCode::from(0)),
Some(record) if record.status.is_terminal() => Some(ExitCode::from(1)),
Some(_) | None => None,
}
}

22
lib/crates/fabro-cli/src/server_client.rs
View file

@ -34,11 +34,6 @@ pub(crate) struct RunAttachEventStream {
buffered_events: VecDeque<EventEnvelope>,
}
pub(crate) enum RunAttachStreamError {
Gone,
Other(anyhow::Error),
}
impl RunAttachEventStream {
fn new(stream: progenitor_client::ByteStream) -> Self {
Self {
@ -355,12 +350,12 @@ impl ServerStoreClient {
&self,
run_id: &RunId,
since_seq: Option<u32>,
) -> std::result::Result<RunAttachEventStream, RunAttachStreamError> {
) -> Result<RunAttachEventStream> {
let mut request = self.client.attach_run_events().id(run_id.to_string());
if let Some(seq) = since_seq.and_then(non_zero_u64_from_u32) {
request = request.since_seq(seq);
}
let response = request.send().await.map_err(map_attach_run_stream_error)?;
let response = request.send().await.map_err(map_api_error)?;
Ok(RunAttachEventStream::new(response.into_inner()))
}
@ -584,19 +579,6 @@ where
}
}
fn map_attach_run_stream_error(
err: progenitor_client::Error<types::ErrorResponse>,
) -> RunAttachStreamError {
match &err {
progenitor_client::Error::ErrorResponse(response)
if response.status() == reqwest::StatusCode::GONE =>
{
RunAttachStreamError::Gone
}
_ => RunAttachStreamError::Other(map_api_error(err)),
}
}
fn convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
where
TInput: serde::Serialize,

110
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -196,6 +196,103 @@ fn attach_uses_configured_server_target_without_server_flag() {
assert!(stdout.contains("\"event\":\"run.completed\""), "{stdout}");
}
#[test]
fn attach_errors_when_live_stream_ends_before_terminal_event() {
let context = test_context!();
let server = MockServer::start();
let run_id = unique_run_id();
server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!([
{
"run_id": run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"labels": {},
"host_repo_path": null,
"start_time": "2026-04-05T12:00:00Z",
"status": "running",
"status_reason": null,
"duration_ms": 12,
"total_cost": null
}
])
.to_string(),
);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/events"));
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"data": [{
"seq": 1,
"payload": {
"event": "run.running",
"id": "evt-run-running",
"run_id": run_id,
"ts": "2026-04-05T12:00:00Z",
"properties": {}
}
}],
"meta": { "has_more": false }
})
.to_string(),
);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/state"));
then.status(200)
.header("Content-Type", "application/json")
.body(live_run_state_response().to_string());
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/questions"))
.query_param("page[limit]", "100")
.query_param("page[offset]", "0");
then.status(200)
.header("Content-Type", "application/json")
.body(r#"{"data":[],"meta":{"has_more":false}}"#);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/attach"))
.query_param("since_seq", "2");
then.status(200)
.header("Content-Type", "text/event-stream")
.body("");
});
context.write_home(
".fabro/settings.toml",
format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()),
);
let output = context
.command()
.args(["attach", &run_id])
.output()
.expect("attach should execute");
assert!(
!output.status.success(),
"attach should fail on premature EOF"
);
let stderr = String::from_utf8(output.stderr).expect("stderr should be UTF-8");
assert!(
stderr.contains("terminal run event"),
"expected a protocol error, got:\n{stderr}"
);
}
#[test]
fn attach_replays_completed_detached_run() {
let context = test_context!();
@ -575,6 +672,19 @@ fn attach_json_errors_without_prompting_for_human_input() {
}
},
"host_repo_path": "[TEMP_DIR]",
"provenance": {
"client": {
"name": "fabro-cli",
"user_agent": "fabro-cli/0.176.2",
"version": "0.176.2"
},
"server": {
"version": "0.176.2"
},
"subject": {
"auth_method": "disabled"
}
},
"run_dir": "[RUN_DIR]",
"settings": {
"goal": "Wait for approval",

45
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -4,8 +4,8 @@ use httpmock::MockServer;
use serde_json::Value;
use super::support::{
output_stderr, resolve_run, run_state, wait_for_no_process_match, wait_for_status,
write_gated_workflow,
output_stderr, resolve_run, run_state, wait_for_event_names, wait_for_no_process_match,
wait_for_status, write_gated_workflow,
};
use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id};
@ -483,7 +483,8 @@ fn dry_run_persists_event_history_in_store() {
.assert()
.success();
context.find_run_dir(&run_id);
let run_dir = context.find_run_dir(&run_id);
wait_for_event_names(&run_dir, &["run.completed", "sandbox.cleanup.completed"]);
let output = context
.command()
.args(["logs", &run_id])
@ -516,6 +517,12 @@ fn dry_run_persists_event_history_in_store() {
.and_then(Value::as_bool),
Some(true)
);
assert!(
progress
.iter()
.any(|event| event["event"].as_str() == Some("run.completed")),
"store-backed event history should include run.completed"
);
assert_eq!(
progress.last().and_then(|event| event["event"].as_str()),
Some("sandbox.cleanup.completed")
@ -746,6 +753,19 @@ fn json_run_implies_auto_approve_for_human_gates() {
}
},
"host_repo_path": "[TEMP_DIR]",
"provenance": {
"client": {
"name": "fabro-cli",
"user_agent": "fabro-cli/0.176.2",
"version": "0.176.2"
},
"server": {
"version": "0.176.2"
},
"subject": {
"auth_method": "disabled"
}
},
"run_dir": "[RUN_DIR]",
"settings": {
"auto_approve": true,
@ -1268,25 +1288,6 @@ fn json_run_implies_auto_approve_for_human_gates() {
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.cleanup.started",
"id": "[EVENT_ID]",
"properties": {
"provider": "local"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.cleanup.completed",
"id": "[EVENT_ID]",
"properties": {
"duration_ms": "[DURATION_MS]",
"provider": "local"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
}
]
"#);

40
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -143,10 +143,15 @@ fn run_completed_dry_run(context: &TestContext, workflow: &Path) -> RunSetup {
stderr(&output)
);
}
RunSetup {
let run_setup = RunSetup {
run_dir: context.find_run_dir(&run_id),
run_id,
}
};
wait_for_event_names(
&run_setup.run_dir,
&["run.completed", "sandbox.cleanup.completed"],
);
run_setup
}
pub(crate) fn setup_created_dry_run(context: &TestContext) -> RunSetup {
@ -691,6 +696,37 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
serde_json::from_value(response["data"].clone()).expect("event list should parse")
}
pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) {
let deadline = std::time::Instant::now() + COMMAND_TIMEOUT;
loop {
let event_names = run_events(run_dir)
.into_iter()
.filter_map(|event| {
event
.payload
.as_value()
.get("event")
.and_then(serde_json::Value::as_str)
.map(ToString::to_string)
})
.collect::<Vec<_>>();
if expected
.iter()
.all(|expected_name| event_names.iter().any(|name| name == expected_name))
{
return;
}
assert!(
std::time::Instant::now() < deadline,
"timed out waiting for events {expected:?}; saw {event_names:?}"
);
std::thread::sleep(std::time::Duration::from_millis(50));
}
}
pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String {
stdout(&git_success(repo_dir, args))
}

36
lib/crates/fabro-cli/tests/it/scenario/recovery.rs
View file

@ -5,6 +5,7 @@ use fabro_checkpoint::branch::BranchStore;
use fabro_checkpoint::git::Store as GitStore;
use fabro_test::{fabro_snapshot, test_context};
use fabro_types::Checkpoint;
use fabro_workflow::operations::build_timeline;
use git2::{Repository, Signature};
use crate::support::unique_run_id;
@ -56,6 +57,17 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint {
serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap()
}
fn timeline_run_shas(repo_dir: &Path, run_id: &str) -> Vec<Option<String>> {
let repo = Repository::discover(repo_dir).unwrap();
let store = GitStore::new(repo);
build_timeline(&store, run_id)
.unwrap()
.entries
.into_iter()
.map(|entry| entry.run_commit_sha)
.collect()
}
fn init_repo_with_workflow(repo_dir: &Path) {
std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap();
std::fs::write(
@ -151,10 +163,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
exit_code: 0
----- stdout -----
----- stderr -----
@ Node Details
@1 start (no run commit)
@2 plan
@3 build
@ Node Details
@1 plan
@2 build
");
let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), &source_run_id);
@ -165,12 +176,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
None
);
assert!(rebuilt_checkpoints.len() >= 2);
let plan_sha = rebuilt_checkpoints[rebuilt_checkpoints.len() - 2]
.git_commit_sha
.clone();
let build_sha = rebuilt_checkpoints
.last()
.and_then(|checkpoint| checkpoint.git_commit_sha.clone());
let timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id);
let build_sha = timeline_shas.last().cloned().flatten();
assert!(build_sha.is_some());
let before_child = list_metadata_run_ids(repo_dir.path());
@ -204,14 +212,14 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
exit_code: 0
----- stdout -----
----- stderr -----
Rewound metadata branch to @2 (plan)
Rewound metadata branch to @2 (build)
Rewound run branch fabro/run/[ULID] to [SHA]
To resume: fabro resume [RUN_PREFIX]
");
let rewound_child = latest_metadata_checkpoint(repo_dir.path(), &source_run_id);
assert_eq!(rewound_child.git_commit_sha, plan_sha);
let rewound_timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id);
assert_eq!(rewound_timeline_shas.last().cloned().flatten(), build_sha);
let before_grandchild = list_metadata_run_ids(repo_dir.path());
context
@ -229,5 +237,5 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() {
assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run");
let grandchild_checkpoint = latest_metadata_checkpoint(repo_dir.path(), &grandchild_run_ids[0]);
assert_eq!(grandchild_checkpoint.git_commit_sha, plan_sha);
assert_eq!(grandchild_checkpoint.git_commit_sha, build_sha);
}

21
lib/crates/fabro-server/src/demo/mod.rs
View file

@ -215,7 +215,26 @@ pub(crate) async fn run_events_stub(
State(_state): State<Arc<AppState>>,
Path(_id): Path<String>,
) -> Response {
ApiError::new(StatusCode::GONE, "Event stream closed.").into_response()
let events = vec![Ok::<_, std::convert::Infallible>(
Event::default().data(
json!({
"seq": 2,
"payload": {
"id": "evt_demo_attach_completed",
"ts": "2026-04-06T15:00:02Z",
"run_id": "01JQ0000000000000000000001",
"event": "run.completed",
"properties": {
"duration_ms": 42,
"artifact_count": 0,
"status": "success"
}
}
})
.to_string(),
),
)];
Sse::new(tokio_stream::iter(events)).into_response()
}
pub(crate) async fn checkpoint_stub(

153
lib/crates/fabro-server/src/server.rs
View file

@ -30,7 +30,7 @@ use fabro_llm::types::{
};
use fabro_store::{ArtifactStore, Database, EventEnvelope, EventPayload, StageId};
use fabro_types::{
RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance,
EventBody, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance,
RunServerProvenance, RunSubjectProvenance, Settings,
};
use fabro_util::redact::redact_jsonl_line;
@ -48,11 +48,12 @@ use tokio::sync::Notify;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::task::spawn_blocking;
use tokio::time::sleep;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::BroadcastStream;
use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream};
use tower::{ServiceExt, service_fn};
use ulid::Ulid;
@ -994,6 +995,23 @@ fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
Some(Event::default().data(data))
}
fn attach_event_is_terminal(event: &EventEnvelope) -> bool {
let Ok(run_event) = RunEvent::try_from(&event.payload) else {
return false;
};
matches!(
run_event.body,
EventBody::RunCompleted(_) | EventBody::RunFailed(_)
)
}
fn run_projection_is_active(state: &fabro_store::RunProjection) -> bool {
state
.status
.as_ref()
.is_some_and(|record| record.status.is_active())
}
fn dir_size(path: &std::path::Path) -> u64 {
walkdir::WalkDir::new(path)
.into_iter()
@ -3264,20 +3282,6 @@ async fn attach_run_events(
Ok(id) => id,
Err(response) => return response,
};
{
let runs = state.runs.lock().expect("runs lock poisoned");
let Some(managed_run) = runs.get(&id) else {
return ApiError::not_found("Run not found.").into_response();
};
if !matches!(
managed_run.status,
RunStatus::Queued | RunStatus::Starting | RunStatus::Running | RunStatus::Paused
) {
return ApiError::new(StatusCode::GONE, "Run is not live on this server.")
.into_response();
}
}
let Ok(run_store) = state.store.open_run_reader(&id).await else {
return ApiError::not_found("Run not found.").into_response();
};
@ -3292,19 +3296,110 @@ async fn attach_run_events(
}
},
};
let stream = match run_store.watch_events_from(start_seq) {
Ok(stream) => stream,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
const ATTACH_REPLAY_BATCH_LIMIT: usize = 256;
let (sender, receiver) = mpsc::unbounded_channel();
tokio::spawn(async move {
let mut next_seq = start_seq;
loop {
let replay_batch = match run_store
.list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
.await
{
Ok(events) => events,
Err(_) => return,
};
let replay_has_more = replay_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
for event in replay_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
next_seq = event.seq.saturating_add(1);
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
if replay_has_more {
continue;
}
let state = match run_store.state().await {
Ok(state) => state,
Err(_) => return,
};
if run_projection_is_active(&state) {
break;
}
let tail_batch = match run_store
.list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT)
.await
{
Ok(events) => events,
Err(_) => return,
};
let tail_has_more = tail_batch.len() > ATTACH_REPLAY_BATCH_LIMIT;
for event in tail_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) {
next_seq = event.seq.saturating_add(1);
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
if tail_has_more {
continue;
}
return;
}
let mut live_stream = match run_store.watch_events_from(next_seq) {
Ok(stream) => stream,
Err(_) => return,
};
while let Some(result) = live_stream.next().await {
let Ok(event) = result else {
return;
};
let terminal = attach_event_is_terminal(&event);
if let Some(sse_event) = sse_event_from_store(&event) {
if sender
.send(Ok::<Event, std::convert::Infallible>(sse_event))
.is_err()
{
return;
}
}
if terminal {
return;
}
}
};
let stream = stream.filter_map(|result| match result {
Ok(event) => sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>),
Err(_) => None,
});
Sse::new(stream).into_response()
Sse::new(UnboundedReceiverStream::new(receiver))
.keep_alive(KeepAlive::default())
.into_response()
}
async fn get_checkpoint(
@ -6138,7 +6233,7 @@ mod tests {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cancel_before_run_transitions_to_running_closes_event_stream() {
async fn cancel_before_run_transitions_to_running_returns_empty_attach_stream() {
let state = create_app_state_with_registry_factory(|interviewer| {
std::thread::sleep(std::time::Duration::from_millis(200));
fabro_workflow::handler::default_registry(interviewer, || None)
@ -6167,7 +6262,9 @@ mod tests {
.body(Body::empty())
.unwrap();
let response = app.oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::GONE);
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert!(body.is_empty(), "expected an empty attach stream");
}
#[tokio::test]

64
lib/crates/fabro-server/tests/it/scenario/run_completion.rs
View file

@ -1,4 +1,4 @@
use axum::body::Body;
use axum::body::{Body, to_bytes};
use axum::http::{Request, StatusCode};
use tokio::time::sleep;
use tower::ServiceExt;
@ -36,22 +36,52 @@ async fn attach_run_events_returns_sse_stream() {
.unwrap();
let response = app.oneshot(req).await.unwrap();
let status = response.status();
assert_eq!(response.status(), StatusCode::OK);
let content_type = response
.headers()
.get("content-type")
.expect("content-type header should be present")
.to_str()
.unwrap();
assert!(
status == StatusCode::OK || status == StatusCode::GONE,
"unexpected status: {status}"
content_type.contains("text/event-stream"),
"expected text/event-stream, got: {content_type}"
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attach_run_events_replays_terminal_event_after_completion() {
let state = test_app_state_with_options(dry_run_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id = create_and_start_run(&app, MINIMAL_DOT).await;
let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
assert_eq!(status, "succeeded");
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/attach?since_seq=1")))
.body(Body::empty())
.unwrap();
let response = app.oneshot(req).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let body = String::from_utf8(body.to_vec()).unwrap();
let event_names = body
.lines()
.filter_map(|line| line.strip_prefix("data:"))
.filter_map(|line| serde_json::from_str::<serde_json::Value>(line.trim()).ok())
.filter_map(|event| event["payload"]["event"].as_str().map(ToString::to_string))
.collect::<Vec<_>>();
assert!(
event_names.iter().any(|event| event == "run.completed"),
"expected a replayed terminal event, got {event_names:?}"
);
assert_eq!(
event_names.last().map(String::as_str),
Some("run.completed")
);
if status == StatusCode::OK {
let content_type = response
.headers()
.get("content-type")
.expect("content-type header should be present")
.to_str()
.unwrap();
assert!(
content_type.contains("text/event-stream"),
"expected text/event-stream, got: {content_type}"
);
}
}

15
lib/crates/fabro-server/tests/it/scenario/sse.rs
View file

@ -54,15 +54,12 @@ async fn sse_stream_contains_expected_event_types() {
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
// May be 200 (stream open) or 410 (run completed before connect)
let sse_status = response.status();
assert!(
sse_status == StatusCode::OK || sse_status == StatusCode::GONE,
"expected 200 or 410, got: {sse_status}"
assert_eq!(
sse_status,
StatusCode::OK,
"expected 200, got: {sse_status}"
);
if sse_status == StatusCode::GONE {
return;
}
let content_type = response
.headers()
@ -96,8 +93,8 @@ async fn sse_stream_contains_expected_event_types() {
// Because we subscribe while the run is only guaranteed to be past
// "queued", a live stream should include at least one stage event.
// A 410 response above still covers the case where the run completed
// before we managed to attach.
// If the run completes before we attach with no unread events, an empty
// stream is still a valid 200 response.
if !event_types.is_empty() {
assert!(
event_types

26
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -240,6 +240,7 @@ impl RunDatabase {
let inner = Arc::clone(&self.inner);
let (sender, receiver) = mpsc::unbounded_channel();
tokio::spawn(async move {
let mut rx = inner.event_tx.subscribe();
let cached = {
let recent_events = inner.recent_events.lock().await;
recent_events
@ -256,8 +257,29 @@ impl RunDatabase {
}
}
let mut rx = inner.event_tx.subscribe();
while let Ok(event) = rx.recv().await {
loop {
loop {
match rx.try_recv() {
Ok(event) => {
if event.seq < next_seq {
continue;
}
next_seq = event.seq.saturating_add(1);
if sender.send(Ok(event)).is_err() {
return;
}
}
Err(broadcast::error::TryRecvError::Empty) => break,
Err(broadcast::error::TryRecvError::Lagged(_)) => continue,
Err(broadcast::error::TryRecvError::Closed) => return,
}
}
let event = match rx.recv().await {
Ok(event) => event,
Err(broadcast::error::RecvError::Lagged(_)) => continue,
Err(broadcast::error::RecvError::Closed) => return,
};
if event.seq < next_seq {
continue;
}

8
lib/packages/fabro-api-client/src/api/run-internals-api.ts
View file

@ -97,7 +97,7 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
};
},
/**
* Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
* Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
* @summary Attach Run Events
* @param {string} id Unique run identifier (ULID).
* @param {number} [sinceSeq] First event sequence number to include.
@ -732,7 +732,7 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
* Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
* @summary Attach Run Events
* @param {string} id Unique run identifier (ULID).
* @param {number} [sinceSeq] First event sequence number to include.
@ -937,7 +937,7 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
return localVarFp.appendRunEvent(id, runEvent, options).then((request) => request(axios, basePath));
},
/**
* Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
* Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
* @summary Attach Run Events
* @param {string} id Unique run identifier (ULID).
* @param {number} [sinceSeq] First event sequence number to include.
@ -1102,7 +1102,7 @@ export class RunInternalsApi extends BaseAPI {
}
/**
* Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates.
* Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active.
* @summary Attach Run Events
* @param {string} id Unique run identifier (ULID).
* @param {number} [sinceSeq] First event sequence number to include.