607 lines
24 KiB
Rust
607 lines
24 KiB
Rust
//! `GET /api/agents/{name}/term/stream` — an SSE relay of one agent's
|
|
//! terminal rows, sourced straight from the swarm queue.
|
|
//!
|
|
//! **No `hive` in the path, on purpose.** An agent is not pinned to a
|
|
//! hive forever — it can move — so a URL naming one would go stale the
|
|
//! moment it did. `crate::agent_status`'s bucket is read only to answer
|
|
//! 404 for an agent that has never reported.
|
|
//!
|
|
//! **Replay, then the live tail, through one consumer.** `hive-agent`
|
|
//! publishes each already-classified `TermMsg` row to the core subject
|
|
//! `$SWARM.term.{agent}`, and the stream `term-main-{agent}`
|
|
//! (`swarm_queue_client::agent_term`) keeps them. The handler creates an
|
|
//! ephemeral consumer on that stream starting at the newest
|
|
//! `agent_term::REPLAY_ROWS` rows, so the reader gets those and then every
|
|
//! new row, in stream order, with no gap and no duplicate between the two.
|
|
//! Each event's `id` is the row's stream sequence; a browser that
|
|
//! reconnects sends it back as `Last-Event-ID` and resumes right after it.
|
|
//!
|
|
//! **No stream, no replay.** While the stream does not exist yet (up to a
|
|
//! minute after the agent is declared) or cannot be read, the handler
|
|
//! falls back to a core subscription: rows published while the reader is
|
|
//! attached, and nothing before. `crate::subagent_term` shares both: [`replay_or_live`].
|
|
//!
|
|
//! **The payload is passed through opaquely**, the same way
|
|
//! `crate::status::HiveStatus::snapshot` stores a hive's status
|
|
//! snapshot: no reason to depend on `hive-agent`'s `TermMsg` type just
|
|
//! to re-serialise a row this daemon never inspects. The bytes on the
|
|
//! wire are already what `hive-agent::swarm_term::publish` sent.
|
|
|
|
use std::convert::Infallible;
|
|
use std::time::SystemTime;
|
|
|
|
use axum::extract::{Path, State};
|
|
use axum::http::HeaderMap;
|
|
use axum::response::sse::{Event, KeepAlive, Sse};
|
|
use futures_util::stream::BoxStream;
|
|
use futures_util::{Stream, StreamExt as _};
|
|
use swarm_queue_client::agent_term;
|
|
|
|
use crate::routes::AppState;
|
|
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/agents/{name}/term/stream",
|
|
params(("name" = String, Path, description = "agent whose terminal to stream")),
|
|
responses(
|
|
(status = 200, description = "server-sent event stream; each event's `data` is \
|
|
one already-classified TermMsg row, JSON as hive-agent published it \
|
|
(opaque to this daemon); the newest retained rows first, then live rows. \
|
|
Each event's `id` is the row's stream sequence, and a `Last-Event-ID` \
|
|
request header resumes right after that row",
|
|
body = String, content_type = "text/event-stream"),
|
|
(status = 400, description = "the agent name is not shaped like an identifier \
|
|
(problem+json)", body = String),
|
|
(status = 404, description = "the agent has never reported to the swarm \
|
|
(problem+json)", body = String),
|
|
(status = 503, description = "no swarm queue is configured on this host, no \
|
|
agent-status reader is wired up, or the queue could not be reached \
|
|
(problem+json)", body = String),
|
|
(status = 500, description = "reading the agent's status failed, or the \
|
|
subscribe itself failed (problem+json)", body = String),
|
|
),
|
|
tag = "agents"
|
|
)]
|
|
pub(crate) async fn stream_agent_term(
|
|
State(state): State<AppState>,
|
|
Path(agent): Path<String>,
|
|
headers: HeaderMap,
|
|
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, problem_details::ProblemDetails> {
|
|
let Some(status) = state.status.as_ref() else {
|
|
return Err(crate::routes::error_problem(
|
|
axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
|
"no swarm queue is configured on this host",
|
|
));
|
|
};
|
|
let Some(agent_status) = state.agent_status.as_ref() else {
|
|
return Err(crate::routes::error_problem(
|
|
axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
|
"no agent-status reader is configured on this host",
|
|
));
|
|
};
|
|
let agent = hive_types::Ident::parse(&agent)
|
|
.map_err(|reason| {
|
|
crate::routes::error_problem(axum::http::StatusCode::BAD_REQUEST, reason)
|
|
})?
|
|
.into_string();
|
|
|
|
// One-agent "roster": reuses `AgentStatusReader::view`'s already-tested
|
|
// row logic rather than a second, narrower lookup that could disagree
|
|
// with it about what "never reported" means.
|
|
agent_status
|
|
.view(std::slice::from_ref(&agent), SystemTime::now())
|
|
.await
|
|
.map_err(|e| {
|
|
let detail = format!("{e:#}");
|
|
tracing::warn!(%agent, error = %detail, "term stream: reading agent status failed");
|
|
crate::routes::error_problem(axum::http::StatusCode::INTERNAL_SERVER_ERROR, &detail)
|
|
})?
|
|
.into_iter()
|
|
.find_map(|row| row.hive)
|
|
.ok_or_else(|| {
|
|
crate::routes::error_problem(
|
|
axum::http::StatusCode::NOT_FOUND,
|
|
&format!("{agent:?} has never reported to the swarm"),
|
|
)
|
|
})?;
|
|
|
|
let client = status.queue_client();
|
|
// Before subscribing: an unconnected client does not fail a
|
|
// subscribe request outright, but there is no point opening a
|
|
// subscription against a queue that is not there — the honest
|
|
// answer is 503, the same shape every other queue-backed route
|
|
// here already uses.
|
|
swarm_queue_client::ensure_connected(&client).map_err(|e| {
|
|
crate::routes::error_problem(
|
|
axum::http::StatusCode::SERVICE_UNAVAILABLE,
|
|
&swarm_queue_client::chain(&e),
|
|
)
|
|
})?;
|
|
|
|
let stream = replay_or_live(
|
|
&client,
|
|
&agent_term::stream_name(&agent),
|
|
swarm_queue_client::term_subject(&agent),
|
|
last_event_id(&headers),
|
|
"term",
|
|
)
|
|
.await?;
|
|
Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
|
|
}
|
|
|
|
/// The SSE events of a terminal relay: the rows [`replay`] reads from
|
|
/// `stream_name`, or, when the stream cannot be read, a core subscription
|
|
/// to `subject` that sends live rows only. `what` names the relay in logs
|
|
/// and errors.
|
|
pub(crate) async fn replay_or_live(
|
|
client: &async_nats::Client,
|
|
stream_name: &str,
|
|
subject: String,
|
|
resume_after: Option<u64>,
|
|
what: &str,
|
|
) -> Result<BoxStream<'static, Result<Event, Infallible>>, problem_details::ProblemDetails> {
|
|
match replay(client, stream_name, subject.clone(), resume_after).await {
|
|
Ok(rows) => Ok(rows.map(|row| Ok(row_event(&row))).boxed()),
|
|
Err(error) => {
|
|
tracing::warn!(%subject, %error, "{what} stream: no replay, live rows only");
|
|
let subscriber = subscribe(client, subject, what).await?;
|
|
Ok(subscriber
|
|
.map(|msg| Ok(Event::default().data(String::from_utf8_lossy(&msg.payload))))
|
|
.boxed())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// One row read from a terminal stream.
|
|
pub(crate) struct Row {
|
|
/// Its stream sequence, when the delivery names one.
|
|
pub(crate) seq: Option<u64>,
|
|
pub(crate) payload: bytes::Bytes,
|
|
}
|
|
|
|
/// The rows on `subject` in the stream `stream_name`: every row after
|
|
/// `resume_after`, or for a fresh reader the newest
|
|
/// [`agent_term::REPLAY_ROWS`] of them, then every new row.
|
|
///
|
|
/// When `subject` is the stream's only subject, its sequences are the
|
|
/// subject's rows and [`agent_term::replay_start`] alone picks the start.
|
|
/// In a stream that holds several subjects, as a `term-sub-<agent>` holds
|
|
/// every subagent of the agent, the newest sequences may hold none of this
|
|
/// subject's rows, so a fresh reader's consumer starts at the oldest row and
|
|
/// the reader drops all but the newest [`agent_term::REPLAY_ROWS`] it counts
|
|
/// as pending when the consumer is created.
|
|
pub(crate) async fn replay(
|
|
client: &async_nats::Client,
|
|
stream_name: &str,
|
|
subject: String,
|
|
resume_after: Option<u64>,
|
|
) -> Result<BoxStream<'static, Row>, String> {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
let stream = js
|
|
.get_stream(stream_name)
|
|
.await
|
|
.map_err(|e| format!("opening {stream_name}: {}", swarm_queue_client::chain(&e)))?;
|
|
let info = stream.cached_info();
|
|
let last_sequence = info.state.last_sequence;
|
|
let shared = info.config.subjects != [subject.as_str()];
|
|
// The case `replay_start` resumes in: a sequence the stream has held.
|
|
let resumed = resume_after.is_some_and(|seen| seen <= last_sequence);
|
|
let trim = shared && !resumed;
|
|
let start = if trim {
|
|
1
|
|
} else {
|
|
agent_term::replay_start(last_sequence, resume_after)
|
|
};
|
|
let deliver_subject = client.new_inbox();
|
|
// The inbox's last token is a fresh nuid, which is also a valid name.
|
|
let consumer_name = format!(
|
|
"term-replay-{}",
|
|
deliver_subject.rsplit('.').next().unwrap_or_default()
|
|
);
|
|
let config =
|
|
agent_term::replay_consumer_config(subject.clone(), consumer_name, deliver_subject, start);
|
|
let consumer = stream.create_consumer(config).await.map_err(|e| {
|
|
format!(
|
|
"creating a consumer on {stream_name}: {}",
|
|
swarm_queue_client::chain(&e)
|
|
)
|
|
})?;
|
|
let skip = if trim {
|
|
consumer
|
|
.cached_info()
|
|
.num_pending
|
|
.saturating_sub(agent_term::REPLAY_ROWS)
|
|
} else {
|
|
0
|
|
};
|
|
let messages = consumer.messages().await.map_err(|e| {
|
|
format!(
|
|
"subscribing to {stream_name}: {}",
|
|
swarm_queue_client::chain(&e)
|
|
)
|
|
})?;
|
|
tracing::info!(stream = %stream_name, %subject, start, skip, "term stream: client attached");
|
|
Ok(rows_until_error(messages, subject)
|
|
.skip(usize::try_from(skip).unwrap_or(usize::MAX))
|
|
.boxed())
|
|
}
|
|
|
|
/// The rows of a consumer's deliveries, ending at the first error. A missed
|
|
/// heartbeat is such an error, and ending there closes the SSE response, so
|
|
/// the browser reconnects with `Last-Event-ID` and gets a fresh consumer.
|
|
fn rows_until_error<E: std::fmt::Display>(
|
|
messages: impl Stream<Item = Result<async_nats::jetstream::Message, E>> + Send + 'static,
|
|
subject: String,
|
|
) -> BoxStream<'static, Row> {
|
|
messages
|
|
.scan((), move |(), msg| {
|
|
let row = match msg {
|
|
Ok(msg) => Some(Row {
|
|
seq: msg.info().ok().map(|info| info.stream_sequence),
|
|
payload: msg.message.payload,
|
|
}),
|
|
Err(e) => {
|
|
tracing::warn!(%subject, error = %e, "term stream: consumer failed, ending the stream");
|
|
None
|
|
}
|
|
};
|
|
std::future::ready(row)
|
|
})
|
|
.boxed()
|
|
}
|
|
|
|
/// One row as an SSE event, carrying its stream sequence as the event `id`.
|
|
fn row_event(row: &Row) -> Event {
|
|
let event = Event::default().data(String::from_utf8_lossy(&row.payload));
|
|
match row.seq {
|
|
Some(seq) => event.id(seq.to_string()),
|
|
None => event,
|
|
}
|
|
}
|
|
|
|
/// The stream sequence a reconnecting browser last received, from its
|
|
/// `Last-Event-ID` header. Absent or unparseable means a fresh reader.
|
|
pub(crate) fn last_event_id(headers: &HeaderMap) -> Option<u64> {
|
|
headers
|
|
.get("last-event-id")?
|
|
.to_str()
|
|
.ok()?
|
|
.trim()
|
|
.parse()
|
|
.ok()
|
|
}
|
|
|
|
/// Subscribe to `subject`. `what` names the relay in logs and errors.
|
|
pub(crate) async fn subscribe(
|
|
client: &async_nats::Client,
|
|
subject: String,
|
|
what: &str,
|
|
) -> Result<async_nats::Subscriber, problem_details::ProblemDetails> {
|
|
let subscriber = client.subscribe(subject.clone()).await.map_err(|e| {
|
|
tracing::warn!(%subject, error = %e, "{what} stream: subscribe failed");
|
|
crate::routes::error_problem(
|
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
|
&format!("subscribing to {subject} failed: {e}"),
|
|
)
|
|
})?;
|
|
tracing::info!(%subject, "{what} stream: client attached");
|
|
Ok(subscriber)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
use std::time::Duration;
|
|
|
|
use crate::test_nats::TestNats;
|
|
|
|
const AGENT: &str = "iris";
|
|
|
|
async fn publish_rows(client: &async_nats::Client, rows: std::ops::RangeInclusive<u64>) {
|
|
for i in rows {
|
|
client
|
|
.publish(
|
|
swarm_queue_client::term_subject(AGENT),
|
|
format!("row{i}").into(),
|
|
)
|
|
.await
|
|
.expect("publish");
|
|
}
|
|
client.flush().await.expect("flush");
|
|
}
|
|
|
|
/// Wait until the agent's stream holds row `last`, so the next step sees it.
|
|
async fn stored(client: &async_nats::Client, last: u64) {
|
|
stored_in(client, &agent_term::stream_name(AGENT), last).await;
|
|
}
|
|
|
|
async fn stored_in(client: &async_nats::Client, stream_name: &str, last: u64) {
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
for _ in 0..100 {
|
|
let mut stream = js.get_stream(stream_name).await.expect("the stream");
|
|
if stream.info().await.expect("info").state.last_sequence >= last {
|
|
return;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
panic!("{stream_name} never stored row {last}");
|
|
}
|
|
|
|
async fn replay_agent(
|
|
client: &async_nats::Client,
|
|
resume_after: Option<u64>,
|
|
) -> Result<BoxStream<'static, Row>, String> {
|
|
replay(
|
|
client,
|
|
&agent_term::stream_name(AGENT),
|
|
swarm_queue_client::term_subject(AGENT),
|
|
resume_after,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Row `i` of subagents `scout` and `builder` in turn, so `scout`'s row
|
|
/// `i` is stream sequence `2i - 1` and the newest sequences hold only
|
|
/// half as many of its rows.
|
|
async fn publish_subagent_rows(
|
|
client: &async_nats::Client,
|
|
rows: std::ops::RangeInclusive<u64>,
|
|
) {
|
|
for i in rows {
|
|
for subagent in ["scout", "builder"] {
|
|
client
|
|
.publish(
|
|
swarm_queue_client::subagent_term::subject(AGENT, subagent),
|
|
format!("{subagent}{i}").into(),
|
|
)
|
|
.await
|
|
.expect("publish");
|
|
}
|
|
}
|
|
client.flush().await.expect("flush");
|
|
}
|
|
|
|
async fn replay_scout(
|
|
client: &async_nats::Client,
|
|
resume_after: Option<u64>,
|
|
) -> Result<BoxStream<'static, Row>, String> {
|
|
replay(
|
|
client,
|
|
&swarm_queue_client::subagent_term::stream_name(AGENT),
|
|
swarm_queue_client::subagent_term::subject(AGENT, "scout"),
|
|
resume_after,
|
|
)
|
|
.await
|
|
}
|
|
|
|
fn scout_rows(range: std::ops::RangeInclusive<u64>) -> Vec<(u64, String)> {
|
|
range.map(|i| (2 * i - 1, format!("scout{i}"))).collect()
|
|
}
|
|
|
|
async fn take(rows: &mut BoxStream<'static, Row>, n: usize) -> Vec<(u64, String)> {
|
|
let mut out = Vec::with_capacity(n);
|
|
for _ in 0..n {
|
|
let row = tokio::time::timeout(Duration::from_secs(5), rows.next())
|
|
.await
|
|
.unwrap_or_else(|_| panic!("row {} of {n} never came; got {out:?}", out.len() + 1))
|
|
.expect("the row stream ended");
|
|
out.push((
|
|
row.seq.expect("a delivered row names its sequence"),
|
|
String::from_utf8_lossy(&row.payload).into_owned(),
|
|
));
|
|
}
|
|
out
|
|
}
|
|
|
|
fn rows(range: std::ops::RangeInclusive<u64>) -> Vec<(u64, String)> {
|
|
range.map(|i| (i, format!("row{i}"))).collect()
|
|
}
|
|
|
|
async fn nothing_more(rows: &mut BoxStream<'static, Row>) {
|
|
if let Ok(Some(row)) = tokio::time::timeout(Duration::from_millis(500), rows.next()).await {
|
|
panic!("an extra row: seq {:?}", row.seq);
|
|
}
|
|
}
|
|
|
|
/// A consumer the server dropped, as on a restart, ends the row stream
|
|
/// once its heartbeats stop, so the SSE response closes and the browser
|
|
/// reconnects instead of keeping a tail that never delivers.
|
|
#[tokio::test]
|
|
async fn a_dropped_consumer_ends_the_row_stream() {
|
|
use futures_util::TryStreamExt as _;
|
|
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let stream = agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the stream");
|
|
publish_rows(&client, 1..=3).await;
|
|
stored(&client, 3).await;
|
|
let mut replayed = replay_agent(&client, None).await.expect("a replay");
|
|
assert_eq!(take(&mut replayed, 3).await, rows(1..=3));
|
|
let consumers: Vec<String> = stream
|
|
.consumer_names()
|
|
.try_collect()
|
|
.await
|
|
.expect("consumer names");
|
|
let [consumer] = consumers.as_slice() else {
|
|
panic!("expected the one replay consumer, got {consumers:?}");
|
|
};
|
|
stream
|
|
.delete_consumer(consumer)
|
|
.await
|
|
.expect("deleting the consumer");
|
|
let end = tokio::time::timeout(agent_term::IDLE_HEARTBEAT * 4, replayed.next())
|
|
.await
|
|
.expect("the row stream ends once heartbeats stop");
|
|
assert!(
|
|
end.is_none(),
|
|
"a row arrived after the consumer was deleted"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn ensuring_the_terminal_stream_again_leaves_it_as_it_is() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let created = agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the first ensure creates the stream");
|
|
publish_rows(&client, 1..=3).await;
|
|
stored(&client, 3).await;
|
|
agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("a second ensure succeeds");
|
|
// Read back from the server, not from what the ensure returned.
|
|
let after = async_nats::jetstream::new(client.clone())
|
|
.get_stream(agent_term::stream_name(AGENT))
|
|
.await
|
|
.expect("the stream")
|
|
.info()
|
|
.await
|
|
.expect("its info")
|
|
.clone();
|
|
assert_eq!(after.config, created.cached_info().config);
|
|
assert_eq!(after.state.messages, 3);
|
|
}
|
|
|
|
/// An existing stream is opened, not created again: one made when the
|
|
/// queue announced a different `max_payload` keeps its own config and
|
|
/// rows, where a second create would be refused for the mismatch.
|
|
#[tokio::test]
|
|
async fn an_existing_terminal_stream_with_other_limits_is_opened_as_it_is() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let js = async_nats::jetstream::new(client.clone());
|
|
js.create_stream(async_nats::jetstream::stream::Config {
|
|
name: agent_term::stream_name(AGENT),
|
|
subjects: vec![agent_term::stream_subject(AGENT)],
|
|
max_message_size: 4096,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.expect("an earlier stream");
|
|
publish_rows(&client, 1..=2).await;
|
|
stored(&client, 2).await;
|
|
let opened = agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("an existing stream opens");
|
|
assert_eq!(opened.cached_info().config.max_message_size, 4096);
|
|
assert_eq!(opened.cached_info().state.messages, 2);
|
|
}
|
|
|
|
/// Live rows published after the reader attached, before it read any
|
|
/// replayed row, still arrive after the whole replay.
|
|
#[tokio::test]
|
|
async fn the_replay_reaches_the_reader_before_the_live_tail() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the stream");
|
|
publish_rows(&client, 1..=150).await;
|
|
stored(&client, 150).await;
|
|
let mut replayed = replay_agent(&client, None).await.expect("a replay");
|
|
publish_rows(&client, 151..=153).await;
|
|
assert_eq!(take(&mut replayed, 103).await, rows(51..=153));
|
|
nothing_more(&mut replayed).await;
|
|
}
|
|
|
|
/// A reader that reconnects with the last id it was sent gets every row
|
|
/// after it exactly once, including those published while it was away.
|
|
#[tokio::test]
|
|
async fn a_reconnect_gets_every_later_row_once() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
agent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the stream");
|
|
publish_rows(&client, 1..=150).await;
|
|
stored(&client, 150).await;
|
|
let mut first = replay_agent(&client, None).await.expect("a replay");
|
|
let seen = take(&mut first, 100).await;
|
|
drop(first);
|
|
publish_rows(&client, 151..=160).await;
|
|
stored(&client, 160).await;
|
|
let last_id = seen.last().expect("rows were seen").0.to_string();
|
|
let resume_after = last_event_id(&with_last_event_id(&last_id));
|
|
let mut resumed = replay_agent(&client, resume_after)
|
|
.await
|
|
.expect("a resumed replay");
|
|
publish_rows(&client, 161..=163).await;
|
|
assert_eq!(take(&mut resumed, 13).await, rows(151..=163));
|
|
nothing_more(&mut resumed).await;
|
|
}
|
|
|
|
/// A subagent's fresh reader gets its own newest rows, in order, then its
|
|
/// live rows, though its sibling's rows fill half the stream's newest
|
|
/// sequences.
|
|
#[tokio::test]
|
|
async fn a_subagent_replays_its_own_newest_rows_before_the_live_tail() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT);
|
|
swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the stream");
|
|
publish_subagent_rows(&client, 1..=150).await;
|
|
stored_in(&client, &stream_name, 300).await;
|
|
let mut replayed = replay_scout(&client, None).await.expect("a replay");
|
|
publish_subagent_rows(&client, 151..=153).await;
|
|
assert_eq!(take(&mut replayed, 103).await, scout_rows(51..=153));
|
|
nothing_more(&mut replayed).await;
|
|
}
|
|
|
|
/// A subagent's reader that reconnects with the last id it was sent gets
|
|
/// every later row of that subagent once.
|
|
#[tokio::test]
|
|
async fn a_subagent_reconnect_gets_every_later_row_once() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let stream_name = swarm_queue_client::subagent_term::stream_name(AGENT);
|
|
swarm_queue_client::subagent_term::open_or_create(&client, AGENT)
|
|
.await
|
|
.expect("the stream");
|
|
publish_subagent_rows(&client, 1..=150).await;
|
|
stored_in(&client, &stream_name, 300).await;
|
|
let mut first = replay_scout(&client, None).await.expect("a replay");
|
|
let seen = take(&mut first, 100).await;
|
|
drop(first);
|
|
publish_subagent_rows(&client, 151..=160).await;
|
|
stored_in(&client, &stream_name, 320).await;
|
|
let last_id = seen.last().expect("rows were seen").0.to_string();
|
|
let resume_after = last_event_id(&with_last_event_id(&last_id));
|
|
let mut resumed = replay_scout(&client, resume_after)
|
|
.await
|
|
.expect("a resumed replay");
|
|
publish_subagent_rows(&client, 161..=163).await;
|
|
assert_eq!(take(&mut resumed, 13).await, scout_rows(151..=163));
|
|
nothing_more(&mut resumed).await;
|
|
}
|
|
|
|
fn with_last_event_id(value: &str) -> HeaderMap {
|
|
let mut headers = HeaderMap::new();
|
|
headers.insert(
|
|
"last-event-id",
|
|
value.parse().expect("a valid header value"),
|
|
);
|
|
headers
|
|
}
|
|
|
|
#[test]
|
|
fn a_reconnect_resumes_after_the_sequence_it_sends() {
|
|
assert_eq!(last_event_id(&with_last_event_id("917")), Some(917));
|
|
assert_eq!(
|
|
agent_term::replay_start(1000, last_event_id(&with_last_event_id("917"))),
|
|
918
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn no_or_a_foreign_last_event_id_is_a_fresh_reader() {
|
|
assert_eq!(last_event_id(&HeaderMap::new()), None);
|
|
assert_eq!(last_event_id(&with_last_event_id("r12")), None);
|
|
assert_eq!(last_event_id(&with_last_event_id("")), None);
|
|
}
|
|
}
|