solution: can stream to a Pulsar topic

[splix]

No description provided.

[Copilot]

Pull request overview

Adds a new streaming “write-only” storage target backed by Apache Pulsar, enabling the stream command to publish per-field messages to a topic-per-field layout instead of writing files.

Changes:

• Introduce PulsarStorage (WriteTarget only) that pre-creates one producer per field topic and publishes encoded ArchiveRow data as Pulsar messages.
• Add a streaming encoder (formats::stream) that converts an ArchiveRow into per-field broker messages with metadata properties (e.g., dedup-key, height, tx-index).
• Refactor stream --continue logic into a pluggable resume strategy (archiver::resume) and wire a separate dispatch path for write-only targets.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/storage/pulsar.rs	New Pulsar write-only storage backend + end-to-end container test publishing to per-field topics.
src/storage/mod.rs	Exposes the Pulsar module and adds is_pulsar/create_pulsar helpers for CLI dispatch.
src/record.rs	Adds Field::name() as a stable external label used by streaming layout.
src/main.rs	Adds Pulsar-specific command restrictions and a write-only dispatch path; wires streaming target selection.
src/formats/stream.rs	New per-row streaming message encoding with topic labels, partition keying, and metadata properties.
src/formats/mod.rs	Exposes the new stream format module.
src/command/stream.rs	Generalizes StreamCommand over WriteTarget and plugs in optional resume strategy for --continue.
src/args.rs	Adds --stream.url / --stream.topics CLI options for selecting a streaming target.
src/archiver/resume.rs	New resume strategy abstraction and ScanResume implementation for file-based targets.
src/archiver/mod.rs	Exposes the new resume module and re-exports resume types.
src/archiver/archiver.rs	Replaces derive(Clone) with a manual Clone impl to avoid unnecessary trait bounds.

Comments suppressed due to low confidence (1)

src/main.rs:319

• StreamCommand::new returns a Result (and can reject --continue for write-only targets), but this unwrap will panic if it ever returns an error. Prefer propagating the error (e.g., make stream_write_only return Result and use ?) so unsupported combinations fail gracefully.

    async fn stream_write_only(self, args: &Args) -> StreamCommand<B, TS> {
        let notifier = self.parent.parent.notifier.unwrap();
        let notifications = notifier.start();
        let archiver = Archiver::new(
            Arc::new(self.parent.target), Arc::new(self.data), notifications
        );
        let command = StreamCommand::new(&args, archiver).await.unwrap();
        command

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[coveralls]

Coverage Report for CI Build 26834588277

Coverage decreased (-1.0%) to 48.723%

Details

• Coverage decreased (-1.0%) from the base build.
• Patch coverage: 338 uncovered changes across 18 files (219 of 557 lines covered, 39.32%).
• 17 coverage regressions across 11 files.

Uncovered Changes

Top 10 Files by Coverage Impact	Changed	Covered	%
src/archiver/archiver.rs	88	19	21.59%
src/archiver/table.rs	76	13	17.11%
src/blockchain/next_block.rs	83	33	39.76%
src/command/stream.rs	33	0	0.0%
src/main.rs	20	0	0.0%
src/archiver/order.rs	54	35	64.81%
src/storage/mod.rs	19	0	0.0%
src/archiver/resume.rs	18	0	0.0%
src/archiver/block.rs	40	24	60.0%
src/global.rs	11	3	27.27%
Total (21 files)	557	219	39.32%

Coverage Regressions

17 previously-covered lines in 11 files lost coverage.

Top 10 Files by Coverage Loss	Lines Losing Coverage	Coverage
src/archiver/archiver.rs	4	28.03%
src/metrics/mod.rs	3	10.0%
src/metrics/blockchain.rs	2	0.0%
src/archiver/datakind.rs	1	82.22%
src/archiver/range.rs	1	84.09%
src/blockchain/block_seq.rs	1	36.71%
src/command/stream.rs	1	0.0%
src/command/verify.rs	1	45.93%
src/global.rs	1	27.78%
src/notify/empty.rs	1	75.0%

---

Coverage Stats

Relevant Lines:	2623
Covered Lines:	1278
Line Coverage:	48.72%
Coverage Strength:	1.5 hits per line

---

💛 - Coveralls