use {
super::{
ingestor::{EventIngestionError, EventIngestor, EventPage, EventPageSource},
metrics::*,
query::EventQuery,
},
crate::sui,
futures::{stream, StreamExt as _, TryStreamExt as _},
std::time::{Duration, Instant},
tokio::sync::mpsc,
};
enum ReplayBound {
Cursor(Vec<u8>),
Checkpoint(u64),
}
enum ReplayFrame<T> {
Page(EventPage<T>),
Complete,
}
const REPLAY_PAGE_EVENTS: usize = 64;
const INDEX_PROGRESS_DELAY: Duration = Duration::from_millis(50);
const REPLAY_REQUEST_LIMIT: u32 = 1_000;
pub(super) const MIN_REPLAY_RANGE_CHECKPOINTS: u64 = 512;
const MAX_REPLAY_RANGE_CHECKPOINTS: u64 = 4_096;
const REPLAY_RANGE_WAVES: u64 = 2;
#[cfg(not(test))]
pub(super) const RECONNECT_DELAY: Duration = Duration::from_secs(2);
#[cfg(test)]
pub(super) const RECONNECT_DELAY: Duration = Duration::from_millis(10);
impl<Q: EventQuery> EventIngestor<Q> {
pub(super) async fn run(
self,
from_checkpoint: Option<u64>,
send_page: mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) {
let mut resume_checkpoint = from_checkpoint;
let mut highest_output_checkpoint = None;
loop {
match self
.run_connection(
&mut resume_checkpoint,
&mut highest_output_checkpoint,
&send_page,
)
.await
{
Ok(()) => return,
Err(error) => {
let should_reconnect = error.is_retryable();
if !self.send_error(error, &send_page).await
|| !should_reconnect
|| !self.wait(RECONNECT_DELAY, &send_page).await
{
return;
}
STREAM_RECONNECTIONS.inc();
}
}
}
}
async fn run_connection(
&self,
resume_checkpoint: &mut Option<u64>,
highest_output_checkpoint: &mut Option<u64>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> Result<(), EventIngestionError> {
let mut client = sui::grpc::client(&self.subscription_rpc_url).map_err(|error| {
EventIngestionError::Configuration(format!(
"invalid subscription gRPC URL '{}': {error}",
self.subscription_rpc_url
))
})?;
let request = sui::grpc::SubscribeEventsRequest::default()
.with_read_mask(self.read_mask.clone())
.with_filter(self.filter.clone());
let mut subscription_client = client.subscription_client();
let response = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(()),
_ = send_page.closed() => return Ok(()),
response = subscription_client.subscribe_events(request) => response,
}
.map_err(|status| EventIngestionError::rpc("subscribing to events", status))?;
let mut stream = response.into_inner();
let first = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(()),
_ = send_page.closed() => return Ok(()),
response = stream.try_next() => response,
}
.map_err(|status| EventIngestionError::rpc("establishing the event stream", status))?
.ok_or_else(|| {
EventIngestionError::rpc(
"establishing the event stream",
tonic::Status::unavailable("event stream ended before its start cursor"),
)
})?;
let live_start_cursor = Self::response_cursor(first.watermark.as_ref())?.to_vec();
if let Some(start_checkpoint) = *resume_checkpoint {
let replay_bound = match first
.watermark
.as_ref()
.and_then(|watermark| watermark.checkpoint_opt())
{
Some(checkpoint) => {
ReplayBound::Checkpoint(checkpoint.checked_add(1).ok_or_else(|| {
EventIngestionError::Protocol(
"the live event checkpoint cannot form an exclusive replay bound"
.to_owned(),
)
})?)
}
None if self.replay_rpc_url.is_some() => {
return Err(EventIngestionError::Protocol(
"the live event stream did not expose a complete checkpoint bound for \
archive replay"
.to_owned(),
));
}
None => ReplayBound::Cursor(live_start_cursor),
};
let replay_result = self
.replay_events(
start_checkpoint,
replay_bound,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await;
let replay_complete = match replay_result {
Ok(replay_complete) => replay_complete,
Err(error) if self.recover_replay_gap && error.is_replay_gap() => {
*resume_checkpoint = None;
if !self.send_error(error, send_page).await {
return Ok(());
}
true
}
Err(error) => return Err(error),
};
if !replay_complete {
return Ok(());
}
}
if !self
.process_frame(
first.event,
first.watermark.as_ref(),
EventPageSource::Live,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await?
{
return Ok(());
}
loop {
let response = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(()),
_ = send_page.closed() => return Ok(()),
response = stream.try_next() => response,
}
.map_err(|status| EventIngestionError::rpc("receiving an event stream frame", status))?
.ok_or_else(|| {
EventIngestionError::rpc(
"receiving an event stream frame",
tonic::Status::unavailable("event stream ended unexpectedly"),
)
})?;
if !self
.process_frame(
response.event,
response.watermark.as_ref(),
EventPageSource::Live,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await?
{
return Ok(());
}
}
}
async fn replay_events(
&self,
start_checkpoint: u64,
bound: ReplayBound,
resume_checkpoint: &mut Option<u64>,
highest_output_checkpoint: &mut Option<u64>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> Result<bool, EventIngestionError> {
if let ReplayBound::Checkpoint(end_checkpoint) = bound {
return self
.replay_checkpoint_ranges(
start_checkpoint,
end_checkpoint,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await;
}
self.replay_to_cursor(
start_checkpoint,
bound,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await
}
async fn replay_checkpoint_ranges(
&self,
start_checkpoint: u64,
end_checkpoint: u64,
resume_checkpoint: &mut Option<u64>,
highest_output_checkpoint: &mut Option<u64>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> Result<bool, EventIngestionError> {
if start_checkpoint >= end_checkpoint {
return Ok(true);
}
let replay_rpc_url = self
.replay_rpc_url
.as_deref()
.unwrap_or(&self.subscription_rpc_url);
let range_checkpoints = replay_range_checkpoints(
start_checkpoint,
end_checkpoint,
self.replay_concurrency.get(),
);
let ranges = std::iter::successors(Some(start_checkpoint), move |start| {
let next = start.saturating_add(range_checkpoints);
(next < end_checkpoint).then_some(next)
})
.map(|start| {
(
start,
start.saturating_add(range_checkpoints).min(end_checkpoint),
)
});
let mut waves = stream::iter(ranges).chunks(self.replay_concurrency.get());
while let Some(ranges) = waves.next().await {
let mut jobs = Vec::with_capacity(ranges.len());
let mut receivers = Vec::with_capacity(ranges.len());
for (start, end) in ranges {
let (send, receive) = mpsc::channel(1);
jobs.push((start, end, send));
receivers.push((end, receive));
}
let collect = stream::iter(jobs)
.map(|(start, end, send)| async move {
self.collect_checkpoint_range(replay_rpc_url, start, end, send_page, &send)
.await
})
.buffer_unordered(self.replay_concurrency.get())
.try_fold(true, |complete, next| async move { Ok(complete && next) });
let deliver = async {
for (end, mut receive) in receivers {
loop {
let Some(frame) = receive.recv().await else {
return Ok(false);
};
match frame {
ReplayFrame::Page(page) => {
let checkpoint = page.checkpoint;
EVENTS_PER_PAGE.observe(page.events.len() as f64);
if !self.send_page(page, send_page).await {
return Ok(false);
}
Self::advance_checkpoint(highest_output_checkpoint, checkpoint);
}
ReplayFrame::Complete => {
Self::advance_checkpoint(resume_checkpoint, end.saturating_sub(1));
break;
}
}
}
}
Ok::<_, EventIngestionError>(true)
};
let (collected, delivered) = tokio::try_join!(collect, deliver)?;
if !collected || !delivered {
return Ok(false);
}
}
Ok(true)
}
async fn collect_checkpoint_range(
&self,
replay_rpc_url: &str,
start_checkpoint: u64,
end_checkpoint: u64,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
output: &mpsc::Sender<ReplayFrame<Q::Output>>,
) -> Result<bool, EventIngestionError> {
let started = Instant::now();
let mut client = sui::grpc::client(replay_rpc_url).map_err(|error| {
EventIngestionError::Configuration(format!(
"invalid replay gRPC URL '{replay_rpc_url}': {error}"
))
})?;
let mut after_cursor: Option<Vec<u8>> = None;
let mut events = Vec::with_capacity(REPLAY_PAGE_EVENTS);
let mut event_count = 0;
loop {
let mut options = sui::grpc::QueryOptions::default()
.with_limit(REPLAY_REQUEST_LIMIT)
.with_ordering(sui::grpc::Ordering::Ascending);
if let Some(after) = &after_cursor {
options.set_after(after.clone());
}
let request = sui::grpc::ListEventsRequest::default()
.with_read_mask(self.read_mask.clone())
.with_start_checkpoint(start_checkpoint)
.with_end_checkpoint(end_checkpoint)
.with_filter(self.filter.clone())
.with_options(options);
REPLAY_REQUESTS.inc();
let mut ledger_client = client.ledger_client();
let response = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(false),
_ = send_page.closed() => return Ok(false),
response = ledger_client.list_events(request) => response,
}
.map_err(|status| {
EventIngestionError::replay_rpc(start_checkpoint, "requesting event replay", status)
})?;
let mut response = response.into_inner();
let terminal_reason = loop {
let frame = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(false),
_ = send_page.closed() => return Ok(false),
frame = response.try_next() => frame,
}
.map_err(|status| {
EventIngestionError::replay_rpc(
start_checkpoint,
"receiving an event replay frame",
status,
)
})?
.ok_or_else(|| {
EventIngestionError::Protocol(
"event replay ended without a terminal frame".to_owned(),
)
})?;
after_cursor = Some(Self::response_cursor(frame.watermark.as_ref())?.to_vec());
if let Some(event) = frame.event {
let (event, checkpoint) = self.decode_event(event).await?;
if !(start_checkpoint..end_checkpoint).contains(&checkpoint) {
return Err(EventIngestionError::Protocol(format!(
"event replay range [{start_checkpoint}, {end_checkpoint}) returned \
checkpoint {checkpoint}"
)));
}
events.extend(event);
if events.len() == REPLAY_PAGE_EVENTS {
event_count += events.len();
let page = EventPage {
events: std::mem::take(&mut events),
checkpoint,
source: EventPageSource::Replay,
};
if !self
.send_range_frame(ReplayFrame::Page(page), output, send_page)
.await
{
return Ok(false);
}
}
}
let end = frame
.end
.as_ref()
.and_then(|end| end.reason)
.and_then(|reason| sui::grpc::QueryEndReason::try_from(reason).ok());
if let Some(reason) = end {
break reason;
}
};
match terminal_reason {
sui::grpc::QueryEndReason::CheckpointBound => {
REPLAY_RANGE_DURATION.observe(started.elapsed().as_secs_f64());
REPLAY_RANGE_EVENTS.observe((event_count + events.len()) as f64);
let page = EventPage {
events,
checkpoint: end_checkpoint.saturating_sub(1),
source: EventPageSource::Replay,
};
if !self
.send_range_frame(ReplayFrame::Page(page), output, send_page)
.await
{
return Ok(false);
}
return Ok(self
.send_range_frame(ReplayFrame::Complete, output, send_page)
.await);
}
sui::grpc::QueryEndReason::ItemLimit | sui::grpc::QueryEndReason::ScanLimit => {}
sui::grpc::QueryEndReason::LedgerTip => {
if !self.wait(INDEX_PROGRESS_DELAY, send_page).await {
return Ok(false);
}
}
reason => {
return Err(EventIngestionError::Protocol(format!(
"checkpoint event replay stopped for an unsupported reason: {reason:?}"
)));
}
}
}
}
async fn send_range_frame(
&self,
frame: ReplayFrame<Q::Output>,
output: &mpsc::Sender<ReplayFrame<Q::Output>>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> bool {
tokio::select! {
_ = self.cancellation_token.cancelled() => false,
_ = send_page.closed() => false,
result = output.send(frame) => result.is_ok(),
}
}
async fn replay_to_cursor(
&self,
start_checkpoint: u64,
bound: ReplayBound,
resume_checkpoint: &mut Option<u64>,
highest_output_checkpoint: &mut Option<u64>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> Result<bool, EventIngestionError> {
let replay_rpc_url = self
.replay_rpc_url
.as_deref()
.unwrap_or(&self.subscription_rpc_url);
let mut client = sui::grpc::client(replay_rpc_url).map_err(|error| {
EventIngestionError::Configuration(format!(
"invalid replay gRPC URL '{replay_rpc_url}': {error}"
))
})?;
let mut after_cursor: Option<Vec<u8>> = None;
loop {
let mut options = sui::grpc::QueryOptions::default()
.with_limit(REPLAY_REQUEST_LIMIT)
.with_ordering(sui::grpc::Ordering::Ascending);
if let ReplayBound::Cursor(cursor) = &bound {
options.set_before(cursor.clone());
}
if let Some(after) = &after_cursor {
options.set_after(after.clone());
}
let mut request = sui::grpc::ListEventsRequest::default()
.with_read_mask(self.read_mask.clone())
.with_start_checkpoint(start_checkpoint)
.with_filter(self.filter.clone())
.with_options(options);
if let ReplayBound::Checkpoint(end_checkpoint) = &bound {
request.set_end_checkpoint(*end_checkpoint);
}
REPLAY_REQUESTS.inc();
let mut ledger_client = client.ledger_client();
let response = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(false),
_ = send_page.closed() => return Ok(false),
response = ledger_client.list_events(request) => response,
}
.map_err(|status| {
EventIngestionError::replay_rpc(start_checkpoint, "requesting event replay", status)
})?;
let mut stream = response.into_inner();
let terminal_reason = loop {
let response = tokio::select! {
_ = self.cancellation_token.cancelled() => return Ok(false),
_ = send_page.closed() => return Ok(false),
response = stream.try_next() => response,
}
.map_err(|status| {
EventIngestionError::replay_rpc(
start_checkpoint,
"receiving an event replay frame",
status,
)
})?
.ok_or_else(|| {
EventIngestionError::Protocol(
"event replay ended without a terminal frame".to_owned(),
)
})?;
after_cursor = Some(Self::response_cursor(response.watermark.as_ref())?.to_vec());
let end = response
.end
.as_ref()
.and_then(|end| end.reason)
.and_then(|reason| sui::grpc::QueryEndReason::try_from(reason).ok());
if !self
.process_frame(
response.event,
response.watermark.as_ref(),
EventPageSource::Replay,
resume_checkpoint,
highest_output_checkpoint,
send_page,
)
.await?
{
return Ok(false);
}
if let Some(reason) = end {
break reason;
}
};
match terminal_reason {
sui::grpc::QueryEndReason::CursorBound
| sui::grpc::QueryEndReason::CheckpointBound => {
return Ok(true);
}
sui::grpc::QueryEndReason::ItemLimit | sui::grpc::QueryEndReason::ScanLimit => {}
sui::grpc::QueryEndReason::LedgerTip => {
if !self.wait(INDEX_PROGRESS_DELAY, send_page).await {
return Ok(false);
}
}
reason => {
return Err(EventIngestionError::Protocol(format!(
"event replay stopped for an unsupported reason: \
{reason:?}"
)));
}
}
}
}
async fn process_frame(
&self,
event: Option<sui::grpc::Event>,
watermark: Option<&sui::grpc::Watermark>,
source: EventPageSource,
resume_checkpoint: &mut Option<u64>,
highest_output_checkpoint: &mut Option<u64>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> Result<bool, EventIngestionError> {
let watermark = watermark.ok_or_else(|| {
EventIngestionError::Protocol("event frame is missing its watermark".to_owned())
})?;
if watermark.cursor_opt().is_none() {
return Err(EventIngestionError::Protocol(
"event frame is missing its cursor".to_owned(),
));
}
let Some(event) = event else {
if let Some(checkpoint) = watermark.checkpoint_opt() {
Self::advance_checkpoint(resume_checkpoint, checkpoint);
if highest_output_checkpoint.is_none_or(|current| checkpoint > current) {
if !self
.send_page(
EventPage {
events: Vec::new(),
checkpoint,
source,
},
send_page,
)
.await
{
return Ok(false);
}
Self::advance_checkpoint(highest_output_checkpoint, checkpoint);
}
}
return Ok(true);
};
let (event, event_checkpoint) = self.decode_event(event).await?;
let events = event.into_iter().collect::<Vec<_>>();
EVENTS_PER_PAGE.observe(events.len() as f64);
let checkpoint = watermark
.checkpoint_opt()
.map_or(event_checkpoint, |checkpoint| {
checkpoint.max(event_checkpoint)
});
Self::advance_checkpoint(resume_checkpoint, checkpoint);
let should_send = !events.is_empty()
|| highest_output_checkpoint.is_none_or(|current| checkpoint > current);
if should_send {
if !self
.send_page(
EventPage {
events,
checkpoint,
source,
},
send_page,
)
.await
{
return Ok(false);
}
Self::advance_checkpoint(highest_output_checkpoint, checkpoint);
}
Ok(true)
}
async fn decode_event(
&self,
event: sui::grpc::Event,
) -> Result<(Option<Q::Output>, u64), EventIngestionError> {
FILTERED_EVENTS_RECEIVED.inc();
let event_checkpoint = event.checkpoint.ok_or_else(|| {
EventIngestionError::Protocol("event is missing its checkpoint".to_owned())
})?;
let transaction_digest = event.transaction_digest.clone().ok_or_else(|| {
EventIngestionError::Protocol("event is missing its transaction digest".to_owned())
})?;
transaction_digest
.parse::<sui::types::Digest>()
.map_err(|error| {
EventIngestionError::Protocol(format!(
"event has an invalid transaction digest: {error}"
))
})?;
let event_index = event.event_index.ok_or_else(|| {
EventIngestionError::Protocol("event is missing its event index".to_owned())
})?;
let decode_started = Instant::now();
let decoded = self.query.decode(event).await;
EVENT_PARSE_DURATION.observe(decode_started.elapsed().as_secs_f64());
let event = decoded.map_err(|source| EventIngestionError::Decode {
checkpoint: event_checkpoint,
transaction_digest,
event_index,
source: anyhow::Error::new(source),
})?;
Ok((event, event_checkpoint))
}
async fn send_error(
&self,
error: EventIngestionError,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> bool {
tokio::select! {
_ = self.cancellation_token.cancelled() => false,
_ = send_page.closed() => false,
result = send_page.send(Err(error)) => result.is_ok(),
}
}
async fn send_page(
&self,
page: EventPage<Q::Output>,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> bool {
let started = Instant::now();
let sent = tokio::select! {
_ = self.cancellation_token.cancelled() => false,
_ = send_page.closed() => false,
result = send_page.send(Ok(page)) => result.is_ok(),
};
SEND_PAGE_BACKPRESSURE_DURATION.observe(started.elapsed().as_secs_f64());
EVENT_PAGE_CHANNEL_LEN.set((send_page.max_capacity() - send_page.capacity()) as f64);
sent
}
async fn wait(
&self,
duration: Duration,
send_page: &mpsc::Sender<Result<EventPage<Q::Output>, EventIngestionError>>,
) -> bool {
tokio::select! {
_ = self.cancellation_token.cancelled() => false,
_ = send_page.closed() => false,
_ = tokio::time::sleep(duration) => true,
}
}
fn response_cursor(
watermark: Option<&sui::grpc::Watermark>,
) -> Result<&[u8], EventIngestionError> {
watermark
.and_then(sui::grpc::Watermark::cursor_opt)
.ok_or_else(|| {
EventIngestionError::Protocol("event frame is missing its cursor".to_owned())
})
}
fn advance_checkpoint(checkpoint: &mut Option<u64>, candidate: u64) {
if checkpoint.is_none_or(|current| candidate > current) {
*checkpoint = Some(candidate);
}
}
}
fn replay_range_checkpoints(start: u64, end: u64, concurrency: usize) -> u64 {
let checkpoints = end.saturating_sub(start);
let target_ranges = (concurrency as u64)
.saturating_mul(REPLAY_RANGE_WAVES)
.max(1);
checkpoints
.div_ceil(target_ranges)
.clamp(MIN_REPLAY_RANGE_CHECKPOINTS, MAX_REPLAY_RANGE_CHECKPOINTS)
}