From b798e171c3c56bc11788c821cb9fe0b42c38e706 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 10 Jun 2026 18:18:15 +0100 Subject: [PATCH] problem: no notifications for new targets (JSON, Pulsar) solution: extend the Notification struct with new types --- README.adoc | 84 ++++++++++- src/archiver/archiver.rs | 95 +++++-------- src/archiver/block.rs | 47 +++--- src/archiver/mod.rs | 21 +-- src/archiver/order.rs | 6 + src/archiver/table.rs | 61 ++++---- src/formats/json.rs | 11 ++ src/notify/location.rs | 276 ++++++++++++++++++++++++++++++++++++ src/notify/mod.rs | 42 +++++- src/notify/pulsar.rs | 4 +- src/storage/fs.rs | 12 +- src/storage/json_fs.rs | 128 ++++++++++++++++- src/storage/json_objects.rs | 22 +++ src/storage/mod.rs | 11 ++ src/storage/objects.rs | 22 ++- src/storage/pulsar.rs | 57 +++++++- 16 files changed, 761 insertions(+), 138 deletions(-) create mode 100644 src/notify/location.rs diff --git a/README.adoc b/README.adoc index 4b12a31..7408444 100644 --- a/README.adoc +++ b/README.adoc @@ -373,7 +373,7 @@ 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", @@ -381,7 +381,10 @@ NOTE: Applicable only for Ethereum-compatible blockchains. "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" + } } ---- @@ -389,11 +392,82 @@ NOTE: Applicable only for Ethereum-compatible blockchains. - `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]] +==== 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 diff --git a/src/archiver/archiver.rs b/src/archiver/archiver.rs index b9b2a5b..f611d35 100644 --- a/src/archiver/archiver.rs +++ b/src/archiver/archiver.rs @@ -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; @@ -89,26 +89,17 @@ impl ArchiveAll for Archiver ) -> 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", @@ -129,7 +120,7 @@ impl ArchiveAll for Archiver 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); @@ -137,12 +128,12 @@ impl ArchiveAll for Archiver } } } 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); @@ -150,7 +141,7 @@ impl ArchiveAll for Archiver } } } else { - Some(ProcessOutcome::Completed { value: (), notification: None }) + Some(ProcessOutcome::Completed { value: (), notifications: vec![] }) } } }; @@ -170,11 +161,12 @@ impl ArchiveAll for Archiver // 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; } @@ -214,26 +206,17 @@ impl ArchiveAll for Archiver 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(()); @@ -244,17 +227,17 @@ impl ArchiveAll for Archiver 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![] }) } } ); @@ -278,11 +261,12 @@ impl ArchiveAll for Archiver 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; } @@ -303,20 +287,17 @@ impl ArchiveAll for Archiver } } -/// 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(side: Option<&ProcessOutcome>) -> Option { +fn extract_notifications(side: Option<&ProcessOutcome>) -> Vec { match side { - Some(ProcessOutcome::Completed { - notification: Some(n), - .. - }) => Some(n.clone()), - _ => None, + Some(ProcessOutcome::Completed { notifications, .. }) => notifications.clone(), + _ => vec![], } } @@ -325,8 +306,8 @@ impl Archiver { /// 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>) { - for notif in notifs.into_iter().flatten() { + async fn publish_notifications(&self, notifs: impl IntoIterator) { + for notif in notifs { if let Err(e) = self.notifications.send(notif).await { tracing::warn!("Failed to publish notification: {}", e); } diff --git a/src/archiver/block.rs b/src/archiver/block.rs index 502d540..647aab7 100644 --- a/src/archiver/block.rs +++ b/src/archiver/block.rs @@ -1,6 +1,5 @@ use std::sync::Arc; use anyhow::anyhow; -use chrono::Utc; use tokio::sync::Semaphore; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; @@ -9,10 +8,10 @@ use crate::archiver::order::AppendSink; use crate::archiver::{BlockTransactions, ProcessOutcome}; use crate::blockchain::{BlockReference, BlockchainData, BlockchainTypes}; use crate::archiver::datakind::{DataKind, DataOptions}; -use crate::notify::Notification; +use crate::notify::NotificationBuilder; use crate::archiver::range::{Height, Range}; use crate::global; -use crate::storage::{TargetFile, TargetFileWriter, WriteTarget}; +use crate::storage::{TargetFileWriter, WriteTarget}; impl Archiver { @@ -26,7 +25,7 @@ impl Archiver { pub async fn process_blocks( &self, blocks: Range, - notification: Notification, + template: NotificationBuilder, options: &DataOptions, cancel: &CancellationToken, ) -> anyhow::Result>> { @@ -34,7 +33,7 @@ impl Archiver { if shutdown.is_signalled() { return Ok(ProcessOutcome::Completed { value: vec![], - notification: None, + notifications: vec![], }); } let dry_run = global::is_dry_run(); @@ -45,7 +44,6 @@ impl Archiver { // note even though we skip the file, we still fetch the blocks to return them tracing::debug!(range = %blocks, "Skipping existing file"); } - let file_url = file.as_ref().map(|f| f.get_url()); let file = file.map(Arc::new); // Order block appends by height when the target requires it // (streaming backends). For file backends `needs_ordering()` is false @@ -135,11 +133,16 @@ impl Archiver { .map(|(_, block, txes)| (block, txes)) .collect(); - if !dry_run { + // Build the notifications (one per location the writer reports) but + // defer the send to `archive()`. If a concurrent process_txes / + // process_traces ends up cancelled, the archiver will discard all + // queued notifications atomically so the consumer never sees a torn + // notification stream for the doomed block. + let notifications = if !dry_run { // Close the ordering layer first — it drains any buffered rows // into the underlying writer (no-op for the direct variant). Only - // then is it safe to take the sole reference to the writer and - // close it. + // then is it safe to take the sole reference to the writer, ask it + // where the data landed, and close it. if let Some(sink) = sink { let sink = Arc::into_inner(sink) .ok_or_else(|| anyhow!("AppendSink still referenced after all tasks completed"))?; @@ -148,23 +151,23 @@ impl Archiver { if let Some(file) = file { let file = Arc::into_inner(file) .ok_or_else(|| anyhow!("File writer still referenced after all tasks completed"))?; + let locations = file.locations(); let _ = file.close().await?; + locations + } else { + vec![] } - } - // Build the notification but defer the send to `archive()`. If a - // concurrent process_txes / process_traces ends up cancelled, the - // archiver will discard all queued notifications atomically so the - // consumer never sees a torn notification stream for the doomed - // block. - let notification = file_url.map(|file_url| Notification { - file_type: DataKind::Blocks, - location: file_url, - ts: Utc::now(), - ..notification - }); + } else { + // dry-run writes nothing, so there is nothing to notify about + vec![] + }; + let notifications = notifications + .into_iter() + .map(|(range, location)| template.notification(DataKind::Blocks, &range, location)) + .collect(); Ok(ProcessOutcome::Completed { value: results, - notification, + notifications, }) } } diff --git a/src/archiver/mod.rs b/src/archiver/mod.rs index cf72763..c338931 100644 --- a/src/archiver/mod.rs +++ b/src/archiver/mod.rs @@ -26,15 +26,16 @@ pub type BlockHash = String; /// upstream cancellation (e.g., a re-org invalidating the block being /// fetched). /// -/// `Completed { value, notification }` is the normal outcome — work -/// finished and produced the inner value, plus an optional notification -/// the caller should publish once the entire run is known to be -/// uncancelled. Notifications are NOT sent inside the process step itself: -/// otherwise a concurrent `process_txes` and `process_traces` could race -/// the cancellation signal and emit a torn notification stream (one side -/// publishes before observing the cancel, the other observes it and -/// stays silent), leaving downstream consumers with partial-block state -/// they can't distinguish from corruption. +/// `Completed { value, notifications }` is the normal outcome — work +/// finished and produced the inner value, plus the notifications the +/// caller should publish once the entire run is known to be uncancelled +/// (one per location the target reported; per-height targets produce one +/// per archived height). Notifications are NOT sent inside the process +/// step itself: otherwise a concurrent `process_txes` and +/// `process_traces` could race the cancellation signal and emit a torn +/// notification stream (one side publishes before observing the cancel, +/// the other observes it and stays silent), leaving downstream consumers +/// with partial-block state they can't distinguish from corruption. /// /// `Cancelled` means the run was abandoned cooperatively; any /// partially-written rows are dropped and the file is left to Drop (which @@ -46,7 +47,7 @@ pub type BlockHash = String; pub enum ProcessOutcome { Completed { value: T, - notification: Option, + notifications: Vec, }, Cancelled, } diff --git a/src/archiver/order.rs b/src/archiver/order.rs index c576a24..feaeeeb 100644 --- a/src/archiver/order.rs +++ b/src/archiver/order.rs @@ -239,6 +239,9 @@ mod tests { self.appended.lock().unwrap().push(row.height); Ok(()) } + fn locations(&self) -> Vec<(crate::archiver::range::Range, crate::notify::Location)> { + vec![] + } async fn close(self) -> Result<()> { Ok(()) } @@ -428,6 +431,9 @@ mod tests { async fn append(&self, _row: ArchiveRow) -> Result<()> { Err(anyhow!("simulated broker reject")) } + fn locations(&self) -> Vec<(crate::archiver::range::Range, crate::notify::Location)> { + vec![] + } async fn close(self) -> Result<()> { Ok(()) } diff --git a/src/archiver/table.rs b/src/archiver/table.rs index 6a0bc35..2fdc203 100644 --- a/src/archiver/table.rs +++ b/src/archiver/table.rs @@ -1,6 +1,5 @@ use std::sync::Arc; use anyhow::anyhow; -use chrono::Utc; use tokio::sync::Semaphore; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; @@ -9,17 +8,17 @@ use crate::archiver::order::AppendSink; use crate::archiver::{BlockTransactions, ProcessOutcome}; use crate::blockchain::{BlockchainData, BlockchainTypes}; use crate::archiver::datakind::{DataKind, DataOptions}; -use crate::notify::Notification; +use crate::notify::NotificationBuilder; use crate::archiver::range::Range; use crate::global; -use crate::storage::{TargetFile, TargetFileWriter, WriteTarget}; +use crate::storage::{TargetFileWriter, WriteTarget}; impl Archiver { pub async fn process_traces( &self, range: Range, - notification: Notification, + template: NotificationBuilder, blocks: &BlockTransactions, options: &DataOptions, cancel: &CancellationToken, @@ -28,7 +27,7 @@ impl Archiver { if shutdown.is_signalled() { return Ok(ProcessOutcome::Completed { value: (), - notification: None, + notifications: vec![], }); } let dry_run = global::is_dry_run(); @@ -39,13 +38,12 @@ impl Archiver { tracing::debug!(range = %range, "Skipping existing file"); return Ok(ProcessOutcome::Completed { value: (), - notification: None, + notifications: vec![], }); } let file = file.unwrap(); let options = options.trace.as_ref().unwrap(); - let file_url = file.get_url(); let file = Arc::new(file); // Order traces by a flat `(block_position, tx_index)` ordinal so block // N's traces are all published before block N+1's, and within a block @@ -114,30 +112,35 @@ impl Archiver { return Ok(ProcessOutcome::Cancelled); } - if !dry_run { + let locations = if !dry_run { + // Close the ordering layer first so any buffered rows reach the + // writer before it reports where the data landed. let sink = Arc::into_inner(sink) .ok_or_else(|| anyhow!("AppendSink still referenced after all tasks completed"))?; sink.close().await?; let file = Arc::into_inner(file) .ok_or_else(|| anyhow!("File writer still referenced after all tasks completed"))?; + let locations = file.locations(); let _ = file.close().await?; - } - let notification = Notification { - file_type: DataKind::TransactionTraces, - location: file_url, - ts: Utc::now(), - ..notification + locations + } else { + // dry-run writes nothing, so there is nothing to notify about + vec![] }; + let notifications = locations + .into_iter() + .map(|(range, location)| template.notification(DataKind::TransactionTraces, &range, location)) + .collect(); Ok(ProcessOutcome::Completed { value: (), - notification: Some(notification), + notifications, }) } pub async fn process_txes( &self, range: Range, - notification: Notification, + template: NotificationBuilder, blocks: &BlockTransactions, options: &DataOptions, cancel: &CancellationToken, @@ -146,7 +149,7 @@ impl Archiver { if shutdown.is_signalled() { return Ok(ProcessOutcome::Completed { value: (), - notification: None, + notifications: vec![], }); } let dry_run = global::is_dry_run(); @@ -157,12 +160,11 @@ impl Archiver { tracing::debug!(range = %range, "Skipping existing file"); return Ok(ProcessOutcome::Completed { value: (), - notification: None, + notifications: vec![], }); } let file = file.unwrap(); - let file_url = file.get_url(); let file = Arc::new(file); // Tx ordering: a flat `(block_position, tx_index)` ordinal across the // whole range. `blocks` is already sorted by height by @@ -232,23 +234,28 @@ impl Archiver { return Ok(ProcessOutcome::Cancelled); } - if !dry_run { + let locations = if !dry_run { + // Close the ordering layer first so any buffered rows reach the + // writer before it reports where the data landed. let sink = Arc::into_inner(sink) .ok_or_else(|| anyhow!("AppendSink still referenced after all tasks completed"))?; sink.close().await?; let file = Arc::into_inner(file) .ok_or_else(|| anyhow!("File writer still referenced after all tasks completed"))?; + let locations = file.locations(); let _ = file.close().await?; - } - let notification = Notification { - file_type: DataKind::Transactions, - location: file_url, - ts: Utc::now(), - ..notification + locations + } else { + // dry-run writes nothing, so there is nothing to notify about + vec![] }; + let notifications = locations + .into_iter() + .map(|(range, location)| template.notification(DataKind::Transactions, &range, location)) + .collect(); Ok(ProcessOutcome::Completed { value: (), - notification: Some(notification), + notifications, }) } } diff --git a/src/formats/json.rs b/src/formats/json.rs index 6bdebea..499e11d 100644 --- a/src/formats/json.rs +++ b/src/formats/json.rs @@ -33,6 +33,7 @@ use std::collections::HashMap; use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::range::Range; use crate::archiver::range_bag::RangeBag; +use crate::notify::FileSlot; use crate::record::{ArchiveRow, BlockchainType, Field}; /// A single file produced by [`encode_row`]. @@ -42,6 +43,9 @@ pub struct JsonFieldFile { pub filename: String, /// Exact bytes to write to the file. pub payload: Vec, + /// Which [`crate::notify::FileGroup`] field the file fills in the + /// notification, so writers don't re-derive it from the filename. + pub slot: FileSlot, } /// Convert an [`ArchiveRow`] into the set of per-field files it produces under @@ -64,34 +68,41 @@ fn encode_field( Field::BlockJson(bytes) => Some(JsonFieldFile { filename: "block.json".to_string(), payload: bytes.clone(), + slot: FileSlot::Block, }), Field::Uncle { index, json } => Some(JsonFieldFile { filename: format!("uncle-{}.json", index), payload: json.clone(), + slot: FileSlot::Uncle, }), Field::TxJson(bytes) => tx_id.map(|id| JsonFieldFile { filename: format!("tx-{}.json", id), payload: bytes.clone(), + slot: FileSlot::Tx, }), // Raw transaction is stored decoded to save memory; re-encode to the // node's wire format here (Ethereum prefixes with `0x`, Bitcoin does not). Field::TxRaw(bytes) => tx_id.map(|id| JsonFieldFile { filename: format!("raw-{}.hex", id), payload: encode_tx_raw(bytes, blockchain_type), + slot: FileSlot::Raw, }), Field::Receipt(bytes) => tx_id.map(|id| JsonFieldFile { filename: format!("receipt-{}.json", id), payload: bytes.clone(), + slot: FileSlot::Receipt, }), // Convenience-only fields that are already present inside the parent JSON. Field::From(_) | Field::To(_) => None, Field::Trace(bytes) => tx_id.map(|id| JsonFieldFile { filename: format!("trace-{}.json", id), payload: bytes.clone(), + slot: FileSlot::Calls, }), Field::StateDiff(bytes) => tx_id.map(|id| JsonFieldFile { filename: format!("statediff-{}.json", id), payload: bytes.clone(), + slot: FileSlot::StateDiff, }), } } diff --git a/src/notify/location.rs b/src/notify/location.rs new file mode 100644 index 0000000..bc6f8e1 --- /dev/null +++ b/src/notify/location.rs @@ -0,0 +1,276 @@ +// Copyright 2026 EmeraldPay Ltd +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. + +//! Where archived data landed, as reported inside a +//! [`Notification`](crate::notify::Notification). +//! +//! Each target type addresses its output differently — a row-batched file is +//! one URL, the JSON layout is a set of per-field files, a streaming broker is +//! a set of message ids. [`Location`] carries all of them under a single +//! `type`-tagged JSON object, so consumers can dispatch on `location.type` and +//! new target types (e.g., Kafka) extend the enum without breaking the overall +//! notification schema. + +use serde::{Deserialize, Serialize}; +use crate::archiver::range::Range; + +/// +/// Address of the archived data inside the target storage. +/// +/// Serialized with a `type` tag so a JSON consumer can distinguish the +/// location flavours, and ignore (or fail on) flavours added after it was +/// written. +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "lowercase")] +pub enum Location { + /// A single row-batched file — the Avro layout (and any future format + /// that keeps one file per (kind, range)). + File { + /// URL of the file, e.g. `s3://bucket/eth/012000000/range-012345000_012345999.txes.avro` + url: String, + }, + /// Per-field files of a single height — the JSON layout. One group per + /// block or per transaction, each pointing to the individual files. + Files { + files: Vec, + }, + /// Messages published to per-field topics of a streaming broker + /// (Apache Pulsar). Each message names its actual topic — a consumer + /// can address it directly, with no topic-name construction on its side. + Pulsar { + messages: Vec, + }, +} + +/// +/// Files produced for one logical entity — a block or a single transaction — +/// under the per-field JSON layout. Only the fields that were actually +/// produced are present in the JSON. +#[derive(Clone, Serialize, Deserialize, Debug, Default, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct FileGroup { + /// Transaction id. Present on per-transaction groups only. + #[serde(skip_serializing_if = "Option::is_none")] + pub tx_id: Option, + /// URL of the block JSON. + #[serde(skip_serializing_if = "Option::is_none")] + pub block: Option, + /// URLs of the uncle JSONs, in uncle-index order. + #[serde(skip_serializing_if = "Vec::is_empty", default)] + pub uncles: Vec, + /// URL of the transaction JSON. + #[serde(skip_serializing_if = "Option::is_none")] + pub tx: Option, + /// URL of the raw transaction (hex). + #[serde(skip_serializing_if = "Option::is_none")] + pub raw: Option, + /// URL of the transaction receipt JSON. + #[serde(skip_serializing_if = "Option::is_none")] + pub receipt: Option, + /// URL of the `callTracer` trace JSON. + #[serde(skip_serializing_if = "Option::is_none")] + pub calls: Option, + /// URL of the `prestateTracer` trace JSON. + #[serde(skip_serializing_if = "Option::is_none")] + pub state_diff: Option, +} + +/// +/// The [`FileGroup`] field a produced file belongs to. Lets the format layer +/// say _what_ a file is without the writer matching on filenames. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FileSlot { + Block, + Uncle, + Tx, + Raw, + Receipt, + Calls, + StateDiff, +} + +impl FileGroup { + pub fn set(&mut self, slot: FileSlot, url: String) { + match slot { + FileSlot::Block => self.block = Some(url), + FileSlot::Uncle => self.uncles.push(url), + FileSlot::Tx => self.tx = Some(url), + FileSlot::Raw => self.raw = Some(url), + FileSlot::Receipt => self.receipt = Some(url), + FileSlot::Calls => self.calls = Some(url), + FileSlot::StateDiff => self.state_diff = Some(url), + } + } + + /// True when the group points to no files at all (`tx_id` is metadata, + /// not a file). Such a group carries nothing to notify about. + pub fn is_empty(&self) -> bool { + self.block.is_none() + && self.uncles.is_empty() + && self.tx.is_none() + && self.raw.is_none() + && self.receipt.is_none() + && self.calls.is_none() + && self.state_diff.is_none() + } +} + +/// +/// One message published to a streaming broker. +#[derive(Clone, Serialize, Deserialize, Debug, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct MessageRef { + /// Full topic name the message went to. + pub topic: String, + /// Field label — the topic suffix (`blocks`, `tx-json`, ...), same values + /// as the `field` of the message payload itself. + pub field: String, + /// Transaction id. Present on per-transaction messages only. + #[serde(skip_serializing_if = "Option::is_none")] + pub tx_id: Option, + /// Broker message id. For Pulsar: `ledgerId:entryId:partition[:batchIndex]`. + #[serde(skip_serializing_if = "Option::is_none")] + pub message_id: Option, +} + +/// +/// The files one archived row produced, tagged with the position of the row, +/// so rows appended in fetch-completion order can be regrouped per height. +#[derive(Clone, Debug)] +pub struct RowFiles { + pub height: u64, + /// Transaction index within the block; `None` for the block row itself. + pub tx_index: Option, + pub group: FileGroup, +} + +impl Location { + /// + /// Group per-row file groups into one `files` location per height, in + /// chain-natural order (block entry first, then transactions by index). + /// + /// This is what keeps the JSON layout at no more than one notification + /// per kind per height, no matter how many files a height produced. + pub fn files_per_height(mut rows: Vec) -> Vec<(Range, Location)> { + // file targets append rows as fetches complete, not in chain order + rows.sort_by_key(|r| (r.height, r.tx_index.map(|i| i + 1).unwrap_or(0))); + + let mut by_height: std::collections::BTreeMap> = std::collections::BTreeMap::new(); + for row in rows { + by_height.entry(row.height).or_default().push(row.group); + } + by_height.into_iter() + .map(|(height, files)| (Range::Single(height.into()), Location::Files { files })) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn file_location_json() { + let location = Location::File { + url: "s3://bucket/eth/012000000/range-012345000_012345999.txes.avro".to_string(), + }; + let json = serde_json::to_string(&location).unwrap(); + assert_eq!( + json, + r#"{"type":"file","url":"s3://bucket/eth/012000000/range-012345000_012345999.txes.avro"}"# + ); + } + + #[test] + fn files_location_json() { + let mut group = FileGroup { + tx_id: Some("0xabc".to_string()), + ..Default::default() + }; + group.set(FileSlot::Tx, "s3://b/eth/.../tx-0xabc.json".to_string()); + group.set(FileSlot::Raw, "s3://b/eth/.../raw-0xabc.hex".to_string()); + group.set(FileSlot::Receipt, "s3://b/eth/.../receipt-0xabc.json".to_string()); + let location = Location::Files { files: vec![group] }; + + let json = serde_json::to_string(&location).unwrap(); + assert_eq!( + json, + r#"{"type":"files","files":[{"txId":"0xabc","tx":"s3://b/eth/.../tx-0xabc.json","raw":"s3://b/eth/.../raw-0xabc.hex","receipt":"s3://b/eth/.../receipt-0xabc.json"}]}"# + ); + } + + #[test] + fn block_files_location_json() { + let mut group = FileGroup::default(); + group.set(FileSlot::Block, "file:///archive/.../block.json".to_string()); + group.set(FileSlot::Uncle, "file:///archive/.../uncle-0.json".to_string()); + group.set(FileSlot::Uncle, "file:///archive/.../uncle-1.json".to_string()); + let location = Location::Files { files: vec![group] }; + + let json = serde_json::to_string(&location).unwrap(); + assert_eq!( + json, + r#"{"type":"files","files":[{"block":"file:///archive/.../block.json","uncles":["file:///archive/.../uncle-0.json","file:///archive/.../uncle-1.json"]}]}"# + ); + } + + #[test] + fn pulsar_location_json() { + let location = Location::Pulsar { + messages: vec![MessageRef { + topic: "persistent://public/default/archive-eth-blocks".to_string(), + field: "blocks".to_string(), + tx_id: None, + message_id: Some("125:4:-1".to_string()), + }], + }; + let json = serde_json::to_string(&location).unwrap(); + assert_eq!( + json, + r#"{"type":"pulsar","messages":[{"topic":"persistent://public/default/archive-eth-blocks","field":"blocks","messageId":"125:4:-1"}]}"# + ); + } + + #[test] + fn parses_back_by_type_tag() { + let json = r#"{"type":"file","url":"s3://bucket/file.avro"}"#; + let location: Location = serde_json::from_str(json).unwrap(); + assert_eq!(location, Location::File { url: "s3://bucket/file.avro".to_string() }); + } + + #[test] + fn groups_rows_per_height_in_chain_order() { + let tx = |height: u64, index: u64, id: &str| RowFiles { + height, + tx_index: Some(index), + group: FileGroup { + tx_id: Some(id.to_string()), + ..Default::default() + }, + }; + let block = |height: u64| RowFiles { + height, + tx_index: None, + group: FileGroup::default(), + }; + + // out of order on purpose: completion order is not chain order + let rows = vec![tx(101, 1, "0xb"), tx(100, 0, "0xa"), block(101), tx(101, 0, "0xc")]; + let locations = Location::files_per_height(rows); + + assert_eq!(locations.len(), 2); + assert_eq!(locations[0].0, Range::Single(100.into())); + assert_eq!(locations[1].0, Range::Single(101.into())); + match &locations[1].1 { + Location::Files { files } => { + assert_eq!(files.len(), 3); + assert_eq!(files[0].tx_id, None); + assert_eq!(files[1].tx_id, Some("0xc".to_string())); + assert_eq!(files[2].tx_id, Some("0xb".to_string())); + } + other => panic!("Expected files location, got {:?}", other), + } + } +} diff --git a/src/notify/mod.rs b/src/notify/mod.rs index 4e5eeb0..ee3a186 100644 --- a/src/notify/mod.rs +++ b/src/notify/mod.rs @@ -1,13 +1,17 @@ pub mod pulsar; pub mod empty; pub mod fs; +pub mod location; use serde::{Deserialize, Serialize}; use crate::archiver::datakind::DataKind; +use crate::archiver::range::Range; use tokio::sync::mpsc::{Sender}; use crate::args::Args; use anyhow::Result; +pub use location::{FileGroup, FileSlot, Location, MessageRef, RowFiles}; + /// Notification represents the metadata for an archive event. #[derive(Clone, Serialize, Deserialize, Debug)] pub struct Notification { @@ -17,7 +21,7 @@ pub struct Notification { pub ts: chrono::DateTime, /// `blockchain` blockchain pub blockchain: String, - /// `type` type of file (transactions or blocks) + /// `type` type of data (transactions, blocks, or traces) #[serde(rename = "type")] pub file_type: DataKind, /// `run` mode in which the Dshackle Archive is run (`archive`, `stream`, `copy` or `compact`) @@ -28,12 +32,40 @@ pub struct Notification { /// `heightEnd` range of blocks in the archived files #[serde(rename = "heightEnd")] pub height_end: u64, - /// `location` a URL to the archived file - pub location: String, + /// `location` where the data landed; the shape depends on the target type + pub location: Location, /// `maturity` maturity level of the block in that archive (`finalized` or `head`) pub maturity: Option, } +/// +/// Everything known about a run before the target reports where the data +/// landed. The archiver builds one [`Notification`] per location the writer +/// reports — a single file for row-batched targets, one entry per height for +/// per-height targets (JSON files, streaming brokers). +#[derive(Clone)] +pub struct NotificationBuilder { + pub blockchain: String, + pub run: RunMode, + pub maturity: Option, +} + +impl NotificationBuilder { + pub fn notification(&self, file_type: DataKind, range: &Range, location: Location) -> Notification { + Notification { + version: Notification::version(), + ts: chrono::Utc::now(), + blockchain: self.blockchain.clone(), + file_type, + run: self.run.clone(), + height_start: range.start(), + height_end: range.end(), + location, + maturity: self.maturity.clone(), + } + } +} + /// RunMode represents the mode in which the Dshackle Archive is run. #[derive(Clone, Serialize, Deserialize, Debug)] #[serde(rename_all = "lowercase")] @@ -56,8 +88,10 @@ pub enum Maturity { } impl Notification { + /// The `location` structure changed from a plain URL string to the typed + /// [`Location`] object in v2 — consumers distinguish the formats by this id. pub fn version() -> String { - "https://schema.emrld.io/dshackle-archive/notify".to_string() + "https://schema.emrld.io/dshackle-archive/notify/v2".to_string() } } diff --git a/src/notify/pulsar.rs b/src/notify/pulsar.rs index 0bef497..237016f 100644 --- a/src/notify/pulsar.rs +++ b/src/notify/pulsar.rs @@ -121,7 +121,9 @@ mod tests { height_start: 100, height_end: 120, maturity: None, - location: "file://archive/range-100_120.blocks.avro".to_string(), + location: crate::notify::Location::File { + url: "file://archive/range-100_120.blocks.avro".to_string(), + }, }).await; tokio::time::sleep(Duration::from_secs(1)).await; diff --git a/src/storage/fs.rs b/src/storage/fs.rs index 75daf82..74c0f27 100644 --- a/src/storage/fs.rs +++ b/src/storage/fs.rs @@ -9,6 +9,7 @@ use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::filenames::{Filenames, Level, LevelDouble}; use crate::archiver::range::Range; use crate::formats::avro; +use crate::notify::Location; use crate::record::ArchiveRow; use crate::storage::{ avro_reader, copy, find_incomplete_by_listing, FileReference, ReadTarget, ScanTarget, @@ -39,7 +40,7 @@ impl WriteTarget for FsStorage { if !overwrite && filename.exists() { return Ok(None); } - Ok(Some(FsFileWriter::new(filename.clone(), kind).context(format!("Path: {:?}", &filename))?)) + Ok(Some(FsFileWriter::new(filename.clone(), kind, range.clone()).context(format!("Path: {:?}", &filename))?)) } } @@ -157,16 +158,17 @@ pub struct FsFileWriter<'a> { path: PathBuf, pub writer: Option>>, kind: DataKind, + range: Range, } impl FsFileWriter<'_> { - pub fn new(path: PathBuf, kind: DataKind) -> Result { + pub fn new(path: PathBuf, kind: DataKind, range: Range) -> Result { tracing::debug!("Create file: {:?}", path); let _ = fs::create_dir_all(path.parent().unwrap())?; let file = File::create(path.clone())?; let writer = Writer::with_codec(avro::schema_for(kind), file, global::get_avro_codec()); let writer = Mutex::new(writer); - Ok(Self { path, writer: Some(writer), kind }) + Ok(Self { path, writer: Some(writer), kind, range }) } /// @@ -217,6 +219,10 @@ impl TargetFileWriter for FsFileWriter<'_> { self.append_record(data) } + fn locations(&self) -> Vec<(Range, Location)> { + vec![(self.range.clone(), Location::File { url: self.get_url() })] + } + async fn close(mut self: Self) -> Result<()> { if let Some(writer) = self.writer.take() { let mut writer = writer.lock().unwrap(); diff --git a/src/storage/json_fs.rs b/src/storage/json_fs.rs index 5f0cd23..daa8bd7 100644 --- a/src/storage/json_fs.rs +++ b/src/storage/json_fs.rs @@ -23,6 +23,7 @@ use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::filenames::Filenames; use crate::archiver::range::Range; use crate::formats::json; +use crate::notify::{FileGroup, Location, RowFiles}; use crate::record::ArchiveRow; use crate::storage::{ScanTarget, TargetFile, TargetFileWriter, WriteTarget}; @@ -80,6 +81,7 @@ impl WriteTarget for JsonFsStorage { range: range.clone(), overwrite, written_files: Mutex::new(Vec::new()), + produced: Mutex::new(Vec::new()), closed: AtomicBool::new(false), })) } @@ -123,6 +125,10 @@ pub struct JsonFsWriter { range: Range, overwrite: bool, written_files: Mutex>, + /// Per-row file groups for the notification report. Only files actually + /// written by this session — skipped pre-existing files were announced + /// when they were originally written. + produced: Mutex>, closed: AtomicBool, } @@ -153,6 +159,10 @@ impl TargetFileWriter for JsonFsWriter { .map_err(|e| anyhow!("Failed to create dir {:?}: {}", dir, e))?; let files = json::encode_row(&row); + let mut group = FileGroup { + tx_id: row.tx_id.clone(), + ..Default::default() + }; for file in files { let path = dir.join(&file.filename); if !self.overwrite && path.exists() { @@ -167,13 +177,26 @@ impl TargetFileWriter for JsonFsWriter { crate::metrics::Direction::Write, file.payload.len(), ); - self.written_files.lock().unwrap().push(path); + self.written_files.lock().unwrap().push(path.clone()); + let canonical = path.canonicalize().unwrap_or(path); + group.set(file.slot, format!("file://{}", canonical.to_str().unwrap_or("invalid"))); + } + if !group.is_empty() { + self.produced.lock().unwrap().push(RowFiles { + height: row.height, + tx_index: row.tx_index, + group, + }); } crate::progress::on_record(); crate::metrics::add_items(&self.kind, crate::metrics::Direction::Write, 1); Ok(()) } + fn locations(&self) -> Vec<(Range, Location)> { + Location::files_per_height(self.produced.lock().unwrap().clone()) + } + async fn close(self) -> Result<()> { self.closed.store(true, Ordering::Relaxed); Ok(()) @@ -400,6 +423,109 @@ mod tests { } } + #[tokio::test] + async fn reports_one_location_per_height() { + let tmp = tempdir().unwrap(); + let storage = JsonFsStorage::new(tmp.path().to_path_buf(), Filenames::with_dir("eth".to_string())); + + let writer = storage + .create(DataKind::Transactions, &Range::new(100, 101), true) + .await + .unwrap() + .unwrap(); + writer.append(tx_row(100, "0xa")).await.unwrap(); + writer.append(tx_row(101, "0xb")).await.unwrap(); + writer.append(tx_row(100, "0xc")).await.unwrap(); + + let locations = writer.locations(); + writer.close().await.unwrap(); + + assert_eq!(locations.len(), 2); + assert_eq!(locations[0].0, Range::Single(100.into())); + assert_eq!(locations[1].0, Range::Single(101.into())); + match &locations[0].1 { + crate::notify::Location::Files { files } => { + assert_eq!(files.len(), 2); + assert_eq!(files[0].tx_id, Some("0xa".to_string())); + assert!(files[0].tx.as_ref().unwrap().starts_with("file://")); + assert!(files[0].tx.as_ref().unwrap().ends_with("/000000100/tx-0xa.json")); + assert!(files[0].raw.as_ref().unwrap().ends_with("/000000100/raw-0xa.hex")); + assert!(files[0].receipt.as_ref().unwrap().ends_with("/000000100/receipt-0xa.json")); + assert_eq!(files[1].tx_id, Some("0xc".to_string())); + } + other => panic!("Expected files location, got {:?}", other), + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn archive_command_notifies_once_per_kind_per_height() { + use crate::archiver::Archiver; + use crate::args::Args; + use crate::blockchain::mock::{MockBlock, MockData, MockTx, MockType}; + use crate::command::archive::ArchiveCommand; + use crate::command::CommandExecutor; + use crate::notify::Location; + use std::sync::Arc; + + crate::testing::start_test(); + let tmp = tempdir().unwrap(); + + let data_provider: Arc = Arc::new(MockData::new("TEST")); + for h in 100..103u64 { + let txs = vec![format!("0xTX{}-A", h), format!("0xTX{}-B", h)]; + data_provider.add_block(MockBlock { + height: h, + hash: format!("0xB{}", h), + parent: format!("0xB{}", h - 1), + transactions: txs.clone(), + }); + for tx in &txs { + data_provider.add_tx(MockTx { hash: tx.clone() }); + } + } + + let storage = JsonFsStorage::new( + tmp.path().to_path_buf(), + Filenames::with_dir("test".to_string()), + ); + let (notifications_tx, mut notifications_rx) = tokio::sync::mpsc::channel(100); + let archiver: Archiver = + Archiver::new(Arc::new(storage), data_provider, notifications_tx); + + let args = Args { + range: Some("100..102".to_string()), + range_chunk: Some(10), + ..Default::default() + }; + let cmd = ArchiveCommand::new(&args, archiver).unwrap(); + cmd.execute().await.unwrap(); + + let mut received = Vec::new(); + while let Ok(n) = notifications_rx.try_recv() { + received.push(n); + } + + // one notification per kind per height: 3 heights x (blocks + txes) + assert_eq!(received.len(), 6); + for kind in [DataKind::Blocks, DataKind::Transactions] { + for h in 100..103u64 { + let matching: Vec<_> = received.iter() + .filter(|n| n.file_type == kind && n.height_start == h) + .collect(); + assert_eq!(matching.len(), 1, "expected one {} notification for height {}", kind, h); + let notification = matching[0]; + assert_eq!(notification.height_end, h); + match ¬ification.location { + Location::Files { files } => { + let expected = if kind == DataKind::Blocks { 1 } else { 2 }; + assert_eq!(files.len(), expected, "{} files at height {}", kind, h); + } + other => panic!("Expected files location, got {:?}", other), + } + } + } + } + #[tokio::test] async fn find_incomplete_tables_flags_missing_block_files() { let tmp = tempdir().unwrap(); diff --git a/src/storage/json_objects.rs b/src/storage/json_objects.rs index 0333dbb..0a80101 100644 --- a/src/storage/json_objects.rs +++ b/src/storage/json_objects.rs @@ -25,6 +25,7 @@ use crate::archiver::datakind::{DataKind, DataOptions}; use crate::archiver::filenames::Filenames; use crate::archiver::range::Range; use crate::formats::json; +use crate::notify::{FileGroup, Location, RowFiles}; use crate::record::ArchiveRow; use crate::storage::{ScanTarget, TargetFile, TargetFileWriter, WriteTarget}; @@ -77,6 +78,7 @@ impl WriteTarget for JsonObjectsStorage { kind, range: range.clone(), overwrite, + produced: std::sync::Mutex::new(Vec::new()), })) } } @@ -112,6 +114,10 @@ pub struct JsonObjectsWriter { kind: DataKind, range: Range, overwrite: bool, + /// Per-row file groups for the notification report. Only objects actually + /// written by this session — skipped pre-existing objects were announced + /// when they were originally written. + produced: std::sync::Mutex>, } impl TargetFile for JsonObjectsWriter { @@ -134,6 +140,10 @@ impl TargetFileWriter for JsonObjectsWriter { async fn append(&self, row: ArchiveRow) -> Result<()> { let dir = self.filenames.height_dir(row.height); let files = json::encode_row(&row); + let mut group = FileGroup { + tx_id: row.tx_id.clone(), + ..Default::default() + }; for file in files { let key = format!("{}/{}", dir, file.filename); let path = Path::from(key); @@ -150,12 +160,24 @@ impl TargetFileWriter for JsonObjectsWriter { .map_err(|e| anyhow!("Failed to put {}: {:?}", path, e))?; crate::progress::on_bytes(payload_len); crate::metrics::add_bytes(&self.kind, crate::metrics::Direction::Write, payload_len); + group.set(file.slot, format!("s3://{}/{}", self.bucket, path)); + } + if !group.is_empty() { + self.produced.lock().unwrap().push(RowFiles { + height: row.height, + tx_index: row.tx_index, + group, + }); } crate::progress::on_record(); crate::metrics::add_items(&self.kind, crate::metrics::Direction::Write, 1); Ok(()) } + fn locations(&self) -> Vec<(Range, Location)> { + Location::files_per_height(self.produced.lock().unwrap().clone()) + } + async fn close(self) -> Result<()> { Ok(()) } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index eea806b..4d83e1a 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -18,6 +18,7 @@ use crate::{ }, args::Args, global, + notify::Location, record::ArchiveRow, }; use anyhow::{anyhow, Result}; @@ -411,6 +412,16 @@ pub trait TargetFileWriter: TargetFile { )) } + /// + /// Where the data appended so far landed, one entry per range that should + /// get its own [`crate::notify::Notification`]. Row-batched files report a + /// single entry covering the whole session range; per-height layouts (JSON + /// files, streaming brokers) report one entry per archived height — that's + /// what keeps notifications at no more than one per kind per height. + /// + /// Call after all appends are done, before [`Self::close`] consumes the writer. + fn locations(&self) -> Vec<(Range, Location)>; + /// /// MUST BE called if everything is written ok. Otherwise, the file is deleted on Drop. async fn close(mut self: Self) -> Result<()>; diff --git a/src/storage/objects.rs b/src/storage/objects.rs index 9a5577b..5b031a8 100644 --- a/src/storage/objects.rs +++ b/src/storage/objects.rs @@ -24,6 +24,7 @@ use crate::archiver::filenames::{Filenames, Level, LevelDouble, LevelSingle}; use crate::archiver::range::Range; use crate::formats::avro; use crate::global; +use crate::notify::Location; use crate::record::ArchiveRow; use crate::storage::{ avro_reader, copy, find_incomplete_by_listing, sorted_files, FileReference, ReadTarget, @@ -55,7 +56,7 @@ impl WriteTarget for ObjectsStorage { return Ok(None); } } - Ok(Some(NewObjectsFile::new(self.os.clone(), kind, self.bucket.clone(), filename))) + Ok(Some(NewObjectsFile::new(self.os.clone(), kind, range.clone(), self.bucket.clone(), filename))) } } @@ -198,6 +199,7 @@ pub struct NewObjectsFile<'a> { bucket: String, path: Path, kind: DataKind, + range: Range, } impl TargetFile for NewObjectsFile<'_> { @@ -235,6 +237,10 @@ impl TargetFileWriter for NewObjectsFile<'_> { self.append_record(data).await } + fn locations(&self) -> Vec<(Range, Location)> { + vec![(self.range.clone(), Location::File { url: self.get_url() })] + } + async fn close(self: Self) -> anyhow::Result<()> { // Avro doesn't always write the data to the underlying writer immediately, and needs to be // flushed independently before closing the file. Otherwise, the file is correct but missing the last appended record(s). @@ -284,7 +290,7 @@ impl TargetFileReader for ExisingObjectsFile { } impl NewObjectsFile<'_> { - fn new(storage: Arc, kind: DataKind, bucket: String, path: Path) -> Self { + fn new(storage: Arc, kind: DataKind, range: Range, bucket: String, path: Path) -> Self { tracing::debug!("Create object: s3://{}/{}", bucket, path.to_string()); let buf = BufWriter::new(storage, path.clone()); let (closed_tx, closed_rx) = oneshot::channel(); @@ -297,6 +303,7 @@ impl NewObjectsFile<'_> { bucket, path, kind, + range, } } @@ -444,13 +451,20 @@ mod tests { pub async fn can_write() { testing::start_test(); let mem = Arc::new(InMemory::new()); - let file = Box::new(NewObjectsFile::new(mem.clone(), DataKind::Blocks, "test".to_string(), Path::from("test.avro"))); + let file = Box::new(NewObjectsFile::new(mem.clone(), DataKind::Blocks, Range::Single(100.into()), "test".to_string(), Path::from("test.avro"))); tokio::time::sleep(std::time::Duration::from_secs(1)).await; let added = file.append(sample_block_row(100)).await; if let Err(e) = added { panic!("Error: {:?}", e); } + assert_eq!( + file.locations(), + vec![( + Range::Single(100.into()), + Location::File { url: "s3://test/test.avro".to_string() } + )] + ); let closed = file.close().await; if let Err(e) = closed { panic!("Error: {:?}", e); @@ -622,7 +636,7 @@ mod tests { let bucket = "test".to_string(); - let file = NewObjectsFile::new(mem.clone(), DataKind::Blocks, bucket.clone(), path.clone()); + let file = NewObjectsFile::new(mem.clone(), DataKind::Blocks, Range::new(0, 9_999), bucket.clone(), path.clone()); for i in 0..10_000u64 { use chrono::TimeZone; let mut row = sample_block_row(i); diff --git a/src/storage/pulsar.rs b/src/storage/pulsar.rs index 0d001f2..2b4bb2c 100644 --- a/src/storage/pulsar.rs +++ b/src/storage/pulsar.rs @@ -38,18 +38,20 @@ //! //! [`dedup-key`]: crate::formats::stream -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use anyhow::{anyhow, Result}; use async_trait::async_trait; use pulsar::producer::{Producer, ProducerOptions}; +use pulsar::proto::MessageIdData; use pulsar::{Pulsar, TokioExecutor}; use tokio::sync::Mutex; use crate::archiver::datakind::DataKind; use crate::archiver::range::Range; use crate::formats::stream; +use crate::notify::{Location, MessageRef}; use crate::record::ArchiveRow; use crate::storage::{TargetFile, TargetFileWriter, WriteTarget}; @@ -130,6 +132,7 @@ impl WriteTarget for PulsarStorage { range: range.clone(), producers: self.producers.clone(), topic_prefix: self.topic_prefix.clone(), + published: std::sync::Mutex::new(Vec::new()), })) } @@ -139,13 +142,32 @@ impl WriteTarget for PulsarStorage { } } -/// Per-(kind, range) writer. Carries no per-session state — the producers it -/// uses are owned by [`PulsarStorage`] and shared across all writers. +/// Per-(kind, range) writer. The producers it uses are owned by +/// [`PulsarStorage`] and shared across all writers; per-session it only +/// accumulates the broker receipts for the notification report. pub struct PulsarWriter { kind: DataKind, range: Range, producers: Arc>>>>, topic_prefix: String, + published: std::sync::Mutex>, +} + +/// A broker-acknowledged message, tagged with the height it belongs to so the +/// notification report can group messages per height. +struct PublishedMessage { + height: u64, + message: MessageRef, +} + +/// Standard Pulsar string form of a message id: +/// `ledgerId:entryId:partition[:batchIndex]`. +fn format_message_id(id: &MessageIdData) -> String { + let partition = id.partition.unwrap_or(-1); + match id.batch_index { + Some(batch) => format!("{}:{}:{}:{}", id.ledger_id, id.entry_id, partition, batch), + None => format!("{}:{}:{}", id.ledger_id, id.entry_id, partition), + } } impl TargetFile for PulsarWriter { @@ -185,10 +207,20 @@ impl TargetFileWriter for PulsarWriter { .map_err(|e| anyhow!("Pulsar send failed: {:?}", e))?; // Block on the broker ack before releasing the lock — otherwise a // later message could overtake this one on the broker side. - send_future + let receipt = send_future .await .map_err(|e| anyhow!("Pulsar ack failed: {:?}", e))?; + self.published.lock().unwrap().push(PublishedMessage { + height: row.height, + message: MessageRef { + topic: format!("{}-{}", self.topic_prefix, msg.field), + field: msg.field.to_string(), + tx_id: row.tx_id.clone(), + message_id: receipt.message_id.as_ref().map(format_message_id), + }, + }); + crate::progress::on_bytes(payload_len); crate::metrics::add_bytes(&self.kind, crate::metrics::Direction::Write, payload_len); } @@ -197,6 +229,23 @@ impl TargetFileWriter for PulsarWriter { Ok(()) } + fn locations(&self) -> Vec<(Range, Location)> { + let mut by_height: BTreeMap> = BTreeMap::new(); + for published in self.published.lock().unwrap().iter() { + by_height + .entry(published.height) + .or_default() + .push(published.message.clone()); + } + by_height + .into_iter() + .map(|(height, messages)| ( + Range::Single(height.into()), + Location::Pulsar { messages }, + )) + .collect() + } + async fn close(self) -> Result<()> { // Producers are shared and outlive the writer; nothing to flush here // because every `append` already awaited its broker ack.