solution: can stream to a Pulsar topic

src/archiver/archiver.rs

@@ -11,14 +11,27 @@ use crate::archiver::range::{Height, Range};
 use crate::global;
 use crate::storage::WriteTarget;
 
-#[derive(Clone)]
 pub struct Archiver<B: BlockchainTypes, TS: WriteTarget> {
     b: PhantomData<B>,
     pub target: Arc<TS>,
     pub data_provider: Arc<B::DataProvider>,
     pub notifications: Sender<Notification>,
 }
 
+// Manual `Clone` impl: all fields are reference-counted, so cloning works
+// without requiring `B: Clone` or `TS: Clone` (which derive(Clone) would
+// otherwise demand).
+impl<B: BlockchainTypes, TS: WriteTarget> Clone for Archiver<B, TS> {
+    fn clone(&self) -> Self {
+        Self {
+            b: PhantomData,
+            target: self.target.clone(),
+            data_provider: self.data_provider.clone(),
+            notifications: self.notifications.clone(),
+        }
+    }
+}
+
 impl<B: BlockchainTypes, TS: WriteTarget> Archiver<B, TS> {
 
     pub fn new_simple(target: Arc<TS>, data_provider: Arc<B::DataProvider>) -> Self {

src/archiver/mod.rs

@@ -7,8 +7,10 @@ pub mod filenames;
 pub mod range_bag;
 pub mod blocks_config;
 pub mod range_group;
+pub mod resume;
 
 pub use archiver::{ArchiveAll, Archiver};
+pub use resume::{ScanResume, StreamResume};
 
 use crate::blockchain::BlockchainTypes;
 

src/archiver/resume.rs

@@ -0,0 +1,101 @@
+// 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.
+
+//! Stream-resume strategies.
+//!
+//! When the `stream` command starts with `--continue`, it asks the target to
+//! report which heights/fields are already archived and re-archives the
+//! missing tail. The mechanism is target-specific:
+//!
+//! - File-based backends ([`crate::storage::ScanTarget`]) can list existing
+//!   files and compute the missing kinds per range. See [`ScanResume`].
+//! - Streaming backends (Pulsar, future Kafka) will eventually grow a
+//!   broker-side variant that tail-reads each topic; the v1 implementation
+//!   ships without it.
+//!
+//! The trait lives in the archiver module rather than alongside the
+//! `stream` command because it's an archival operation (it calls back into
+//! [`crate::archiver::Archiver::archive`]); the command just dispatches it.
+
+use std::sync::Arc;
+
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::archiver::archiver::Archiver;
+use crate::archiver::datakind::DataOptions;
+use crate::archiver::range::{Height, Range};
+use crate::archiver::ArchiveAll;
+use crate::blockchain::BlockchainTypes;
+use crate::notify::RunMode;
+use crate::storage::ScanTarget;
+
+/// Resume strategy invoked once at stream startup, *before* the first new
+/// block is published. Implementations re-archive the tail of the chain so
+/// that gaps left by a previous run are filled in before live tailing begins.
+#[async_trait]
+pub trait StreamResume: Send + Sync {
+    /// Make the archive complete up to (and including) `up_to`'s parent
+    /// height, then return. Errors are the caller's to handle — typically
+    /// callers log and continue so a transient resume failure doesn't block
+    /// fresh blocks from being archived.
+    async fn ensure_continued(&self, up_to: Height) -> Result<()>;
+}
+
+/// [`ScanTarget`]-backed [`StreamResume`] implementation that walks the last
+/// `continue_blocks` heights of the archive and re-archives anything reported
+/// as incomplete.
+pub struct ScanResume<B: BlockchainTypes, TS: ScanTarget> {
+    archiver: Archiver<B, TS>,
+    continue_blocks: u64,
+    data_options: DataOptions,
+}
+
+impl<B: BlockchainTypes, TS: ScanTarget> ScanResume<B, TS> {
+    pub fn new(archiver: Archiver<B, TS>, continue_blocks: u64, data_options: DataOptions) -> Self {
+        Self { archiver, continue_blocks, data_options }
+    }
+
+    /// Convenience constructor that hands back the boxed trait object the
+    /// command layer wants to store. Saves callers from importing
+    /// `Arc<dyn StreamResume>` and the trait at once.
+    pub fn boxed(
+        archiver: Archiver<B, TS>,
+        continue_blocks: u64,
+        data_options: DataOptions,
+    ) -> Arc<dyn StreamResume>
+    where
+        B: 'static,
+        TS: 'static,
+    {
+        Arc::new(Self::new(archiver, continue_blocks, data_options))
+    }
+}
+
+#[async_trait]
+impl<B, TS> StreamResume for ScanResume<B, TS>
+where
+    B: BlockchainTypes + 'static,
+    TS: ScanTarget + 'static,
+{
+    async fn ensure_continued(&self, height: Height) -> Result<()> {
+        let range = Range::up_to(self.continue_blocks, &Range::Single(height));
+        let options = self.data_options.clone();
+        let missing = self
+            .archiver
+            .target
+            .find_incomplete_tables(range, &options)
+            .await?;
+        for (range, kinds) in missing {
+            let range_opts = options.clone().only_include(&kinds);
+            for height in range.iter().collect::<Vec<u64>>() {
+                self.archiver
+                    .archive(Height::from(height), RunMode::Stream, None, &range_opts)
+                    .await?;
+            }
+        }
+        Ok(())
+    }
+}

src/args.rs

@@ -60,6 +60,9 @@ pub struct Args {
     #[command(flatten)]
     pub aws: Option<Aws>,
 
+    #[command(flatten)]
+    pub stream: Option<Stream>,
+
     /// Target directory
     #[arg(long = "dir", short)]
     pub dir: Option<String>,
@@ -137,6 +140,7 @@ impl Default for Args {
             connection: Connection::default(),
             notify: None,
             aws: None,
+            stream: None,
             dir: None,
             continue_last: false,
             tail: None,
@@ -259,6 +263,53 @@ pub struct Aws {
     pub trust_tls: bool,
 }
 
+/// Streaming-target options. Picked up only by the `stream` command.
+///
+/// The broker is identified by the URL scheme on `--stream.url`: `pulsar://`
+/// selects Apache Pulsar. When `--stream.url` is set the archive runs against
+/// a topic-per-field broker target instead of a file backend, so `--dir`,
+/// `--auth.aws.*`, and `--format` are ignored.
+#[derive(Parser, Debug, Clone)]
+pub struct Stream {
+    /// Publish stream data to a broker at the given URL. The scheme selects
+    /// the backend:
+    ///
+    /// - `pulsar://HOST:PORT` — Apache Pulsar.
+    ///
+    /// Selecting a streaming target restricts the run to the `stream` command —
+    /// `archive`, `fix`, `verify`, and `compact` are rejected at startup
+    /// because topics are append-only.
+    #[arg(long = "stream.url", required = false, alias = "stream-url")]
+    pub stream_url: Option<String>,
+
+    /// Prefix used to build the per-field topic names. Each field is published
+    /// to `<prefix>-<field>` (e.g. `<prefix>-blocks`, `<prefix>-tx-json`).
+    /// For Pulsar, include the full topic path up to the prefix, e.g.
+    /// `persistent://public/default/archive-eth`.
+    #[arg(long = "stream.topics", required = false, alias = "stream-topics")]
+    pub stream_topics: Option<String>,
+}
+
+impl Default for Stream {
+    fn default() -> Self {
+        Self {
+            stream_url: None,
+            stream_topics: None,
+        }
+    }
+}
+
+impl Stream {
+    /// True when the args carry a Pulsar streaming target (URL scheme
+    /// `pulsar://`). Other schemes will route to other backends in the future.
+    pub fn is_pulsar(&self) -> bool {
+        self.stream_url
+            .as_deref()
+            .map(|u| u.starts_with("pulsar://"))
+            .unwrap_or(false)
+    }
+}
+
 impl Default for Aws {
     fn default() -> Self {
         Self {