feat: store and forward consumer by joyhe208 · Pull Request #10 · nominal-io/nominal-streaming

joyhe208 · 2025-09-08T21:04:11Z

TL;DR

Added store-and-forward capability to the Nominal streaming client, enabling resilient data streaming with automatic recovery from network failures.

What changed?

Added StoreAndForwardNominalCoreConsumer that streams data to Nominal Core and falls back to local file storage when streaming fails
Implemented automatic file re-upload functionality when the connection is restored
Added StreamHealthMonitor to track stream health and detect when to attempt re-uploads
Replaced StreamingClient with NominalApiClients that includes streaming, upload, and ingest clients
Added support for multipart uploads of large files
Added reqwest and async-channel dependencies for HTTP requests and async communication
Reorganized client structure to support both streaming and batch uploads

How to test?

Create a stream with store-and-forward capability:

let consumer = StoreAndForwardNominalCoreConsumer::new(
    core_consumer,
    fallback_consumer,
    ReuploadOpts::default()
);

Test resilience by simulating network failures (the consumer will automatically fall back to file storage)
Verify automatic re-upload by checking logs for "File path queued for reupload" messages after connectivity is restored

Why make this change?

This change significantly improves the reliability of the Nominal streaming client by:

Ensuring no data is lost during network outages or service disruptions
Automatically recovering and re-uploading data when connectivity is restored
Providing a more robust solution for production environments where network reliability cannot be guaranteed

The store-and-forward pattern is essential for mission-critical data pipelines that cannot afford to lose data during temporary outages.

alxhill

one main q, haven't taken a deep dive in the re-upload code itself but the broad monitoring/checking approach looks good

src/client.rs

alxhill · 2025-09-10T18:54:30Z

src/stream.rs

+        let consumer = Arc::new(consumer);
+
+        Self::create_internal(
+            opts,
+            |running, unflushed_points, request_rx, dispatcher_id| {
+                thread::Builder::new()
+                    .name(format!("nmstream_dispatch_{dispatcher_id}"))
+                    .spawn({
+                        let consumer = Arc::clone(&consumer);
+                        debug!("starting request dispatcher from factory");
+                        move || {
+                            request_dispatcher(running, unflushed_points, request_rx, consumer);
+                        }
+                    })
+                    .unwrap();
+            },
+        )
+    }
+
+    pub fn new_with_consumer_factory<C: WriteRequestConsumerFactory + 'static>(
+        consumer_factory: C,
+        opts: NominalStreamOpts,
+    ) -> Self {
+        Self::create_internal(
+            opts,
+            |running, unflushed_points, request_rx, dispatcher_id| {
+                let consumer = Arc::new(
+                    consumer_factory
+                        .create_consumer(dispatcher_id)
+                        .expect("Failed to create consumer"),
+                );
+
+                thread::Builder::new()
+                    .name(format!("nmstream_dispatch_{dispatcher_id}"))
+                    .spawn({
+                        debug!("starting request dispatcher from factory");
+                        move || {
+                            request_dispatcher(running, unflushed_points, request_rx, consumer);
+                        }
+                    })
+                    .unwrap();
+            },
+        )
+    }


In main, I added a builder so users don't have to construct the consumers manually - good to integrate this so the "defaults" are sane/easy to use

alxhill · 2025-09-10T19:48:59Z

src/consumer.rs

+pub trait WriteRequestConsumerFactory: Send + Sync {
+    type Consumer: WriteRequestConsumer;
+    fn create_consumer(&self, id: usize) -> Result<Self::Consumer, Box<dyn Error + Send + Sync>>;
+}


So I don't love that we need a whole separate "factory" trait here, feels a little unnecessary just for creating one consumer for each thread?

Two ideas on my mind are:

have the consumer use thread::current().id() to identify itself. works, but makes it hard to do a separate "init" stage for e.g file creation

add WriteRequestConsumer: Clone + ... as a bound. then, consumers that don't need a factory are implemented on Arc<T> instead of T (i.e impl WriteRequestConsumer for Arc<NominalCoreConsumer>), while the AvroFileConsumer can implement clone() manually, copy over the opts and increment an ID.

thoughts? particularly on 2?

#2 makes sense mostly. But i'm not sure how flexible the Clone solution will be in the long run because clone() can't return a result which might be awkward if the creation of the new consumer actually fails. And it can't take any arguments either (e.g. an id) - not that we would really need an id, i guess we could just generate something randomly.
I do agree the trait is a bit too heavy weight, though. We could potentially do a lighter weight approach where we have new_with_consumer_fn or something like that that takes in a function (with params thread id, etc) that defines how to create the new consumer.

Seems vaguely reasonable that cloning a consumer is infallible - is there a case you can think of where it could fail in a recoverable way?

For the ID, was thinking that the clone() implementation would store a refcount of some kind, so each clone gets the unique next value. Internally this would be a refcount: Arc<AtomicUsize> + id: usize in the Consumer. If this feels weird, your idea of a consumer generator is probably fine too - so long as we can abstract it away from end users via the builder

Yea the refcount thing makes sense. Wondering if it'd be confusing for the AvroFileConsumer's clone to not be writing to the same underlying file? Since i think that's what the derived implementation would do.

I guess people could just use arc if they needed access to the same file

joyhe208 · 2025-09-16T16:47:25Z

feat: store and forward consumer #10 👈 (View in Graphite)
main

This stack of pull requests is managed by Graphite. Learn more about stacking.

joyhe208 · 2025-09-16T16:52:56Z

src/consumer.rs

+    }
+}
+
+impl WriteRequestConsumer for Arc<AvroFileConsumer> {


For when we don't want to create a new file for new threads

joyhe208 · 2025-09-16T16:53:38Z

src/consumer.rs

-    fallback: F,
+pub struct StoreAndForwardNominalCoreConsumer<A: AuthProvider> {
+    core_consumer: NominalCoreConsumer<A>,
+    fallback_consumer: Arc<Mutex<AvroFileConsumer>>,


needs to be wrapped in a mutex to do the file rotation :/

Could alternatively have a separate rotate_file method on AvroFileConsumer that switches out the file instead of cloning the consumer so we can get rid of this lock

joyhe208 · 2025-09-26T21:18:38Z

src/consumer.rs

 }

+impl Clone for AvroFileConsumer {
+    fn clone(&self) -> Self {


joyhe208 added 2 commits September 8, 2025 17:03

feat: automatic reupload

bbce110

rename

3266bfd

joyhe208 changed the title ~~feat: automatic reupload~~ feat: store and forward consumer Sep 9, 2025

alxhill self-requested a review September 10, 2025 01:03

Merge branch 'main' into joy/automatic-reupload

39d95d6

alxhill reviewed Sep 10, 2025

View reviewed changes

joyhe208 and others added 7 commits September 15, 2025 11:03

fix rebase

e9a2224

get rid of streaming client

48dc44b

Merge branch 'main' into joy/automatic-reupload

851626e

Fixes

2c4b8b3

update

a3ec2dc

update

83a285d

builder update

f6d8da3

joyhe208 commented Sep 16, 2025

View reviewed changes

joyhe208 added 4 commits September 16, 2025 14:04

workspace rid

34d5d26

test script

58caecb

fmt

3bddb9c

remove test

b46b0a3

joyhe208 requested a review from alxhill September 23, 2025 19:02

joyhe208 and others added 2 commits September 23, 2025 15:03

Merge branch 'main' into joy/automatic-reupload

660f305

lock

c667b9d

joyhe208 commented Sep 26, 2025

View reviewed changes

src/consumer.rs

}

impl Clone for AvroFileConsumer {

fn clone(&self) -> Self {

Copy link

Contributor Author

joyhe208 Sep 26, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alxhill

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: store and forward consumer#10

feat: store and forward consumer#10
joyhe208 wants to merge 16 commits intomainfrom
joy/automatic-reupload

joyhe208 commented Sep 8, 2025 •

edited

Loading

Uh oh!

alxhill left a comment

Uh oh!

Uh oh!

alxhill Sep 10, 2025

Uh oh!

alxhill Sep 10, 2025

Uh oh!

joyhe208 Sep 15, 2025

Uh oh!

alxhill Sep 15, 2025

Uh oh!

joyhe208 Sep 15, 2025

Uh oh!

joyhe208 Sep 15, 2025

Uh oh!

joyhe208 commented Sep 16, 2025 •

edited

Loading

Uh oh!

joyhe208 Sep 16, 2025

Uh oh!

joyhe208 Sep 16, 2025

Uh oh!

joyhe208 Sep 16, 2025

Uh oh!

joyhe208 Sep 26, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

joyhe208 commented Sep 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

TL;DR

What changed?

How to test?

Why make this change?

Uh oh!

alxhill left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

joyhe208 commented Sep 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

joyhe208 commented Sep 8, 2025 •

edited

Loading

joyhe208 commented Sep 16, 2025 •

edited

Loading