feat: Implement `Sourcer` traits for serving source #2301

BulkBeing · 2024-12-23T09:48:40Z

closes: #2281

Signed-off-by: Sreekanth <[email protected]>

vigith · 2024-12-24T04:49:36Z

please resolve conflicts?

Signed-off-by: Sreekanth <[email protected]>

codecov · 2024-12-24T05:05:53Z

Codecov Report

Attention: Patch coverage is 86.10662% with 86 lines in your changes missing coverage. Please review.

Project coverage is 67.11%. Comparing base (0c86e82) to head (1dcad81).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/serving/src/source.rs	90.32%	18 Missing ⚠️
rust/numaflow-core/src/shared/create_components.rs	0.00%	17 Missing ⚠️
rust/numaflow-core/src/source/serving.rs	90.90%	13 Missing ⚠️
rust/numaflow-core/src/source.rs	0.00%	7 Missing ⚠️
rust/serving/src/config.rs	78.78%	7 Missing ⚠️
rust/numaflow-core/src/config/components.rs	0.00%	6 Missing ⚠️
rust/serving/src/pipeline.rs	0.00%	5 Missing ⚠️
rust/numaflow-core/src/message.rs	0.00%	4 Missing ⚠️
rust/serving/src/app/jetstream_proxy.rs	97.08%	4 Missing ⚠️
rust/serving/src/app.rs	95.45%	3 Missing ⚠️
... and 1 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2301      +/-   ##
==========================================
+ Coverage   67.08%   67.11%   +0.02%     
==========================================
  Files         349      351       +2     
  Lines       45055    45263     +208     
==========================================
+ Hits        30227    30378     +151     
- Misses      13756    13818      +62     
+ Partials     1072     1067       -5

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Signed-off-by: Sreekanth <[email protected]>

vigith

Why is the test coverage too low?

Signed-off-by: Sreekanth <[email protected]>

vigith · 2025-01-03T05:12:01Z

rust/Cargo.toml

@@ -40,8 +40,8 @@ verbose_file_reads = "warn"
 # This profile optimizes for runtime performance and small binary size at the expense of longer build times.
 # Compared to default release profile, this profile reduced binary size from 29MB to 21MB
 # and increased build time (with only one line change in code) from 12 seconds to 133 seconds (tested on Mac M2 Max).
-[profile.release]
-lto = "fat"
+# [profile.release]


no longer valid?

Reverted. Had commented it out for faster image builds on laptop.

vigith · 2025-01-03T05:15:59Z

rust/numaflow-core/src/config/components.rs

@@ -37,7 +38,7 @@ pub(crate) mod source {
        Generator(GeneratorConfig),
        UserDefined(UserDefinedConfig),
        Pulsar(PulsarSourceConfig),
-        Serving(serving::Settings),
+        Serving(Arc<serving::Settings>),


why Arc, is it for easy cloning? how come this is different from the rest? please document the reason?

vigith · 2025-01-03T05:16:28Z

rust/numaflow-core/src/lib.rs

@@ -55,6 +55,7 @@ mod tracker;
 mod mapper;

 pub async fn run() -> Result<()> {
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();


what is this for? please document

vigith · 2025-01-03T05:18:44Z

rust/numaflow-core/src/shared/create_components.rs

-        SourceType::Serving(_) => {
-            unimplemented!("Serving as built-in source is not yet implemented")
+        SourceType::Serving(config) => {
+            let serving = ServingSource::new(Arc::clone(config), batch_size, read_timeout).await?;


Suggested change

let serving = ServingSource::new(Arc::clone(config), batch_size, read_timeout).await?;

let serving = ServingSource::new(config, batch_size, read_timeout).await?;

why clone, can't we give the ownership?

The match is on a reference, so the config here is a &Arc<Settings>.

vigith · 2025-01-03T05:28:39Z

rust/numaflow-core/src/source/serving.rs

+        let offset = Offset::String(StringOffset::new(message.id.clone(), 0));
+
+        Ok(Message {
+            keys: Arc::from(vec![]),


Suggested change

keys: Arc::from(vec![]),

// we do not support keys from HTTP client

keys: Arc::from(vec![]),

vigith · 2025-01-03T05:31:35Z

rust/numaflow-core/src/source/serving.rs

+    }
+
+    fn partitions(&self) -> Vec<u16> {
+        vec![]


we should have atleast one partition?

vigith · 2025-01-03T05:33:15Z

rust/numaflow-core/src/source/serving.rs

+}
+
+impl super::SourceAcker for ServingSource {
+    async fn ack(&mut self, offsets: Vec<Offset>) -> Result<()> {


Suggested change

async fn ack(&mut self, offsets: Vec<Offset>) -> Result<()> {

/// HTTP response is sent only once we have confirmation that the message has been written to the ISB.

async fn ack(&mut self, offsets: Vec<Offset>) -> Result<()> {

vigith · 2025-01-03T05:38:41Z

rust/numaflow-core/src/source/serving.rs

+    }
+}
+
+impl super::SourceAcker for ServingSource {


Suggested change

impl super::SourceAcker for ServingSource {

// TODO: implement for async and sync, currently it is only for async

impl super::SourceAcker for ServingSource {

vigith · 2025-01-03T05:50:06Z

rust/serving/src/app/jetstream_proxy.rs

+        },
+    };
+
+    proxy_state.message.send(message).await.unwrap(); // FIXME:


Signed-off-by: Vigith Maurice <[email protected]>

yhl25 · 2025-01-03T06:01:50Z

rust/serving/src/source.rs

+    ) -> Result<()> {
+        let (messages_tx, messages_rx) = mpsc::channel(10000);
+        // Create a redis store to store the callbacks and the custom responses


10000 is way too much, should we restrict it to 2 * batch size?

yhl25 · 2025-01-03T06:03:12Z

rust/serving/src/source.rs

+        tokio::spawn(async move {
+            let mut serving_actor = ServingSourceActor {
+                messages: messages_rx,
+                handler_rx,
+                tracker: HashMap::new(),
+            };
+            serving_actor.run().await;
+        });
+        let app = crate::AppState {
+            message: messages_tx,
+            settings,
+            callback_state,
+        };
+        tokio::spawn(async move {
+            crate::serve(app).await.unwrap();
+        });


make sure these tokio tasks are aborted during shutdown, structured concurrency :).

This might need more work. Created an issue for it #2309

yhl25 · 2025-01-03T06:03:32Z

rust/serving/src/source.rs

+}
+
+struct ServingSourceActor {
+    messages: mpsc::Receiver<MessageWrapper>,
+    handler_rx: mpsc::Receiver<ActorMessage>,
+    tracker: HashMap<String, oneshot::Sender<()>>,
+}


yhl25 · 2025-01-03T06:05:50Z

rust/serving/src/source.rs

+
+    async fn read(&mut self, count: usize, timeout_at: Instant) -> Vec<Message> {
+        let mut messages = vec![];
+        loop {


return Result<Vec<Message>> to fix the FIXME :). Source reader trait returns a result.

yhl25 · 2025-01-03T06:06:35Z

rust/serving/src/source.rs

+            if messages.len() >= count || Instant::now() >= timeout_at {
+                break;
+            }


yhl25 · 2025-01-03T06:07:18Z

rust/serving/src/source.rs

+            let offset = offset
+                .strip_suffix("-0")
+                .expect("offset does not end with '-0'"); // FIXME: we hardcode 0 as the partition index when constructing offset
+            let confirm_save_tx = self


use vertex replica as the partition index incase of source.

yhl25 · 2025-01-03T06:07:43Z

rust/serving/src/source.rs

+        let (actor_tx, actor_rx) = mpsc::channel(1000);
+        ServingSourceActor::start(settings, actor_rx).await?;


make the channel size configurable, it can be 2 * batch size.

yhl25 · 2025-01-03T06:08:32Z

rust/serving/src/source.rs

+        timeout_at: Instant,
+        reply_to: oneshot::Sender<Vec<Message>>,
+    },


make oneshot sender return Result, to handle critical errors?

yhl25 · 2025-01-03T06:09:09Z

rust/serving/src/source.rs

+    },
+    Ack {
+        offsets: Vec<String>,
+        reply_to: oneshot::Sender<()>,


same here, make oneshot sender return Result?

Signed-off-by: Sreekanth <[email protected]>

yhl25 · 2025-01-03T14:39:41Z

rust/serving/src/source.rs

+            // Stop if the read timeout has reached or if we have collected the requested number of messages
            if messages.len() >= count || Instant::now() >= timeout_at {
                break;


Can we use ReceiverStream instead of mpsc channel to store the messages? That way you can chunk the stream using chunks_timeout()?

I can't use ReceiverStream here since chunks_timeout consumes self. This results in:

cannot move out of `self.messages` which is behind a mutable reference

ReceiverStream.chunks_timeout() would have worked if it was function that accepts the ReceiverStream and uses it in a loop{} to consume it. In this case, the Receiver end is a struct member.

Signed-off-by: Sreekanth <[email protected]>

yhl25 · 2025-01-06T05:36:46Z

rust/serving/src/source.rs

                    // The error will happen on all the subsequent read attempts too.
                    if messages.is_empty() {


Why do we need to send an error when we don't have any messages?

If the code reaches here, all the Senders are already dropped. Any subsequent read requests also will fail.
Unless there is some bug, this code path should never reach.

yhl25

LGTM!

Signed-off-by: Sreekanth <[email protected]> Signed-off-by: Vigith Maurice <[email protected]> Co-authored-by: Vigith Maurice <[email protected]>

BulkBeing added 4 commits December 23, 2024 14:08

/v1/process/async publish works

3a45e5f

Signed-off-by: Sreekanth <[email protected]>

Unit tests for configurations

198fe00

Signed-off-by: Sreekanth <[email protected]>

Remove jetstream related config options

94dcddd

Signed-off-by: Sreekanth <[email protected]>

Remove unused constants

ae42c29

Signed-off-by: Sreekanth <[email protected]>

BulkBeing added 2 commits December 24, 2024 10:19

Merge branch 'main' into serving-sourcer-traits

bdb8243

Fix compilation errors

0bc6ca9

Signed-off-by: Sreekanth <[email protected]>

BulkBeing added 4 commits December 24, 2024 12:51

Modify serving to be a built-in source

e20b022

Signed-off-by: Sreekanth <[email protected]>

Remove duplicate code, make serving a built-in source

f548609

Signed-off-by: Sreekanth <[email protected]>

Unit tests for Serving endpoints

ca8cdeb

Signed-off-by: Sreekanth <[email protected]>

Merge branch 'main' into serving-sourcer-traits

76c2371

BulkBeing force-pushed the serving-sourcer-traits branch from 824ae59 to 76c2371 Compare January 2, 2025 07:48

Merge branch 'main' into serving-sourcer-traits

57c4cb7

BulkBeing marked this pull request as ready for review January 2, 2025 07:48

BulkBeing requested review from whynowy and vigith as code owners January 2, 2025 07:48

vigith reviewed Jan 2, 2025

View reviewed changes

BulkBeing added 3 commits January 3, 2025 06:30

Unit tests for serving source

e893964

Signed-off-by: Sreekanth <[email protected]>

More unit tests for serving

a45411d

Signed-off-by: Sreekanth <[email protected]>

Unit tests for Sourcer trait implementation on Serving source

e8592bc

Signed-off-by: Sreekanth <[email protected]>

vigith reviewed Jan 3, 2025

View reviewed changes

chore: minor updates

54c2cce

Signed-off-by: Vigith Maurice <[email protected]>

yhl25 reviewed Jan 3, 2025

View reviewed changes

BulkBeing added 2 commits January 3, 2025 14:54

Fixes based on code review

8f0389b

Signed-off-by: Sreekanth <[email protected]>

Fix test

d85079a

Signed-off-by: Sreekanth <[email protected]>

yhl25 reviewed Jan 3, 2025

View reviewed changes

Avoid hot loop by using timeout on channel receive

1dcad81

Signed-off-by: Sreekanth <[email protected]>

yhl25 reviewed Jan 6, 2025

View reviewed changes

yhl25 approved these changes Jan 6, 2025

View reviewed changes

yhl25 merged commit 8d8340c into main Jan 6, 2025
28 checks passed

yhl25 deleted the serving-sourcer-traits branch January 6, 2025 05:57

BulkBeing mentioned this pull request Jan 6, 2025

serve response via ack for sync and sync-serve #2308

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Implement `Sourcer` traits for serving source #2301

feat: Implement `Sourcer` traits for serving source #2301

BulkBeing commented Dec 23, 2024

vigith commented Dec 24, 2024

codecov bot commented Dec 24, 2024 •

edited

Loading

vigith left a comment

vigith Jan 3, 2025

BulkBeing Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

BulkBeing Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

vigith Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

BulkBeing Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

yhl25 Jan 3, 2025

BulkBeing Jan 6, 2025 •

edited

Loading

yhl25 Jan 6, 2025

BulkBeing Jan 6, 2025 •

edited

Loading

yhl25 left a comment

	let serving = ServingSource::new(Arc::clone(config), batch_size, read_timeout).await?;
	let serving = ServingSource::new(config, batch_size, read_timeout).await?;

	keys: Arc::from(vec![]),
	// we do not support keys from HTTP client
	keys: Arc::from(vec![]),

	async fn ack(&mut self, offsets: Vec<Offset>) -> Result<()> {
	/// HTTP response is sent only once we have confirmation that the message has been written to the ISB.
	async fn ack(&mut self, offsets: Vec<Offset>) -> Result<()> {

	impl super::SourceAcker for ServingSource {
	// TODO: implement for async and sync, currently it is only for async
	impl super::SourceAcker for ServingSource {

		let (actor_tx, actor_rx) = mpsc::channel(1000);
		ServingSourceActor::start(settings, actor_rx).await?;

		// The error will happen on all the subsequent read attempts too.
		if messages.is_empty() {

feat: Implement Sourcer traits for serving source #2301

feat: Implement Sourcer traits for serving source #2301

Conversation

BulkBeing commented Dec 23, 2024

vigith commented Dec 24, 2024

codecov bot commented Dec 24, 2024 • edited Loading

Codecov Report

vigith left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BulkBeing Jan 6, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BulkBeing Jan 6, 2025 • edited Loading

Choose a reason for hiding this comment

yhl25 left a comment

Choose a reason for hiding this comment

feat: Implement `Sourcer` traits for serving source #2301

feat: Implement `Sourcer` traits for serving source #2301

codecov bot commented Dec 24, 2024 •

edited

Loading

BulkBeing Jan 6, 2025 •

edited

Loading

BulkBeing Jan 6, 2025 •

edited

Loading