Skip to content

problem: no notifications for new targets (JSON, Pulsar) #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
splix wants to merge 1 commit into
base: master
Choose a base branch
from
Open

problem: no notifications for new targets (JSON, Pulsar) #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 79 additions & 5 deletions README.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -373,27 +373,101 @@ NOTE: Applicable only for Ethereum-compatible blockchains.
[source, json]
----
{
"version":"https://schema.emrld.io/dshackle-archive/notify",
"version":"https://schema.emrld.io/dshackle-archive/notify/v2",
"ts":"2022-05-20T23:14:24.481327Z",
"blockchain":"ETH",
"type":"transactions",
"run":"stream",
"maturity": "finalized",
"heightStart":14813875,
"heightEnd":14813875,
"location":"gs://my-bucket/blockchain-archive/eth/014000000/014813000/014813875.txes.avro"
"location": {
"type": "file",
"url": "s3://my-bucket/blockchain-archive/eth/014000000/014813000/014813875.txes.avro"
}
}
----

.Where
- `version` id of the current JSON format
- `ts` timestamp of the archive event
- `blockchain` blockchain
- `type` type of file (`transactions`, `blocks`, or `traces`)
- `type` type of data (`transactions`, `blocks`, or `traces`)
- `run` mode in which the Dshackle Archive is run (`archive`, `stream`, `copy` or `compact`)
- `maturity` block maturity level (`latest` or `finalized`); `finalized` is applicable to Ethereum PoS chains only
- `heightStart` and `heightEnd` range of blocks in the archived files
- `location` a URL to the archived file
- `heightStart` and `heightEnd` range of blocks covered by the notification
- `location` where the data landed; an object distinguished by its own `type` field, see <<notification-location>>

[[notification-location]]
==== Location types

The shape of `location` depends on the target the archive writes to.
The `location.type` defines the structure of the location object.

===== `file`

One row-batched file (the Avro layout):

[source, json]
----
{
"type": "file",
"url": "s3://my-bucket/eth/014000000/range-014813000_014813999.txes.avro"
}
----

===== `files`

Per-field files of a single height (ex. for the `--format=json`).

One notification is sent per kind per height.
Each entry of `files` describes one block or one transaction, pointing to its individual files.

[source, json]
----
{
"type": "files",
"files": [
{
"txId": "0x40846886cf7b8...",
"tx": "s3://my-bucket/eth/014000000/014813000/014813875/tx-0x40846886cf7b8....json",
"raw": "s3://my-bucket/eth/014000000/014813000/014813875/raw-0x40846886cf7b8....hex",
"receipt": "s3://my-bucket/eth/014000000/014813000/014813875/receipt-0x40846886cf7b8....json"
}
]
}
----

.Per-entry fields (only the produced ones are present)
- `txId` - transaction id, on per-transaction entries
- `block` - URL of the block JSON
- `uncles` - URLs of the uncle JSONs, in uncle-index order
- `tx` - URL of the transaction JSON
- `raw` - URL of the raw transaction (hex)
- `receipt` - URL of the transaction receipt JSON
- `calls` - URL of the `callTracer` trace JSON
- `stateDiff` - URL of the `prestateTracer` trace JSON

===== `pulsar`

Messages published to per-field topics of an Apache Pulsar broker (For the `--stream.url` target).

One notification is sent per kind per height; each message names its actual topic and is identified by its broker message id (`ledgerId:entryId:partition[:batchIndex]`):

[source, json]
----
{
"type": "pulsar",
"messages": [
{
"topic": "persistent://public/default/archive-eth-tx-json",
"field": "tx-json",
"txId": "0x40846886cf7b8...",
"messageId": "125:4:-1"
}
]
}
----

== Community

Expand Down
95 changes: 38 additions & 57 deletions src/archiver/archiver.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,10 @@ use chrono::Utc;
use tokio::sync::mpsc::Sender;
use tokio_util::sync::CancellationToken;
use crate::blockchain::{BlockchainData, BlockchainTypes};
use crate::archiver::datakind::{DataKind, DataOptions};
use crate::archiver::datakind::DataOptions;
use crate::archiver::ProcessOutcome;
use crate::notify::empty::EmptyNotifier;
use crate::notify::{Maturity, Notification, Notifier, RunMode};
use crate::notify::{Maturity, Notification, NotificationBuilder, Notifier, RunMode};
use crate::archiver::range::{Height, Range};
use crate::global;
use crate::storage::WriteTarget;
Expand Down Expand Up @@ -89,26 +89,17 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Height> for Archiver<B, TS>
) -> anyhow::Result<()> {
let start_time = Utc::now();

let notification = Notification {
// common fields
version: Notification::version(),
ts: Utc::now(),
let template = NotificationBuilder {
blockchain: self.data_provider.blockchain_id(),
run: mode,
height_start: what.height,
height_end: what.height,
maturity,

// specific fields, should be overridden later
file_type: DataKind::Blocks,
location: "".to_string(),
};

let (blocks, blocks_notif) = match self
.process_blocks(Range::Single(what.clone()), notification.clone(), options, cancel)
let (blocks, blocks_notifs) = match self
.process_blocks(Range::Single(what.clone()), template.clone(), options, cancel)
.await?
{
ProcessOutcome::Completed { value, notification } => (value, notification),
ProcessOutcome::Completed { value, notifications } => (value, notifications),
ProcessOutcome::Cancelled => {
tracing::info!(
"Block {} cancelled (re-org) — skipping tx/trace fetch",
Expand All @@ -129,28 +120,28 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Height> for Archiver<B, TS>
let (tx_side, trace_side) = tokio::join! {
async {
if options.include_tx() {
match self.process_txes(range.clone(), notification.clone(), &blocks, options, cancel).await {
match self.process_txes(range.clone(), template.clone(), &blocks, options, cancel).await {
Ok(outcome) => Some(outcome),
Err(e) => {
tracing::warn!("Failed to archive txes for block {}: {}", what, e);
None
}
}
} else {
Some(ProcessOutcome::Completed { value: (), notification: None })
Some(ProcessOutcome::Completed { value: (), notifications: vec![] })
}
},
async {
if options.include_trace() {
match self.process_traces(range.clone(), notification.clone(), &blocks, options, cancel).await {
match self.process_traces(range.clone(), template.clone(), &blocks, options, cancel).await {
Ok(outcome) => Some(outcome),
Err(e) => {
tracing::warn!("Failed to archive traces for block {}: {}", what, e);
None
}
}
} else {
Some(ProcessOutcome::Completed { value: (), notification: None })
Some(ProcessOutcome::Completed { value: (), notifications: vec![] })
}
}
};
Expand All @@ -170,11 +161,12 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Height> for Archiver<B, TS>
// artifacts have already been cleaned up via writer Drop in the
// process_* functions; the doomed block is invisible to consumers.
if !cancelled && !global::get_shutdown().is_signalled() {
self.publish_notifications([
blocks_notif,
extract_notification(tx_side.as_ref()),
extract_notification(trace_side.as_ref()),
])
self.publish_notifications(
blocks_notifs
.into_iter()
.chain(extract_notifications(tx_side.as_ref()))
.chain(extract_notifications(trace_side.as_ref())),
)
.await;
}

Expand Down Expand Up @@ -214,26 +206,17 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Range> for Archiver<B, TS>
let start_time = Utc::now();
tracing::debug!("Archiving range: {}", what);

let notification = Notification {
// common fields
version: Notification::version(),
ts: Utc::now(),
let template = NotificationBuilder {
blockchain: self.data_provider.blockchain_id(),
run: mode,
height_start: what.start(),
height_end: what.end(),
maturity,

// specific fields, should be overridden later
file_type: DataKind::Blocks,
location: "".to_string(),
};

let (blocks, blocks_notif) = match self
.process_blocks(what.clone(), notification.clone(), options, cancel)
let (blocks, blocks_notifs) = match self
.process_blocks(what.clone(), template.clone(), options, cancel)
.await?
{
ProcessOutcome::Completed { value, notification } => (value, notification),
ProcessOutcome::Completed { value, notifications } => (value, notifications),
ProcessOutcome::Cancelled => {
tracing::info!(range = %what, "Range archive cancelled before tx/trace fetch");
return Ok(());
Expand All @@ -244,17 +227,17 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Range> for Archiver<B, TS>
async {
if options.include_tx() {
tracing::debug!(range = %what, "Process txes");
self.process_txes(what.clone(), notification.clone(), &blocks, options, cancel).await
self.process_txes(what.clone(), template.clone(), &blocks, options, cancel).await
} else {
Ok(ProcessOutcome::Completed { value: (), notification: None })
Ok(ProcessOutcome::Completed { value: (), notifications: vec![] })
}
},
async {
if options.include_trace() {
tracing::debug!(range = %what, "Process traces");
self.process_traces(what.clone(), notification.clone(), &blocks, options, cancel).await
self.process_traces(what.clone(), template.clone(), &blocks, options, cancel).await
} else {
Ok(ProcessOutcome::Completed { value: (), notification: None })
Ok(ProcessOutcome::Completed { value: (), notifications: vec![] })
}
}
);
Expand All @@ -278,11 +261,12 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Range> for Archiver<B, TS>

let cancelled = tx_outcome.is_cancelled() || trace_outcome.is_cancelled();
if !cancelled {
self.publish_notifications([
blocks_notif,
extract_notification(Some(&tx_outcome)),
extract_notification(Some(&trace_outcome)),
])
self.publish_notifications(
blocks_notifs
.into_iter()
.chain(extract_notifications(Some(&tx_outcome)))
.chain(extract_notifications(Some(&trace_outcome))),
)
.await;
}

Expand All @@ -303,20 +287,17 @@ impl<B: BlockchainTypes, TS: WriteTarget> ArchiveAll<Range> for Archiver<B, TS>
}
}

/// Extract the deferred [`Notification`] from a [`ProcessOutcome`], if any.
/// Extract the deferred [`Notification`]s from a [`ProcessOutcome`].
///
/// Returns `None` when the side errored (outer Option is None), was
/// cancelled, or completed without producing a notification (target skipped
/// Returns nothing when the side errored (outer Option is None), was
/// cancelled, or completed without producing notifications (target skipped
/// an existing file). The archiver coordinator collects these across all
/// three process_* calls and publishes them as a batch — only when the
/// whole run is known to be uncancelled.
fn extract_notification<T>(side: Option<&ProcessOutcome<T>>) -> Option<Notification> {
fn extract_notifications<T>(side: Option<&ProcessOutcome<T>>) -> Vec<Notification> {
match side {
Some(ProcessOutcome::Completed {
notification: Some(n),
..
}) => Some(n.clone()),
_ => None,
Some(ProcessOutcome::Completed { notifications, .. }) => notifications.clone(),
_ => vec![],
}
}

Expand All @@ -325,8 +306,8 @@ impl<B: BlockchainTypes, TS: WriteTarget> Archiver<B, TS> {
/// the notification channel are logged but not propagated — the data
/// itself has already landed in the archive; a failed notify shouldn't
/// fail the whole run.
async fn publish_notifications(&self, notifs: impl IntoIterator<Item = Option<Notification>>) {
for notif in notifs.into_iter().flatten() {
async fn publish_notifications(&self, notifs: impl IntoIterator<Item = Notification>) {
for notif in notifs {
if let Err(e) = self.notifications.send(notif).await {
tracing::warn!("Failed to publish notification: {}", e);
}
Expand Down
Loading
Loading