feat: Serving as a builtin source with Monovertex #2382

BulkBeing · 2025-02-06T07:31:25Z

APIs:

POST /v1/process/sync
POST /v1/process/async
GET /v1/process/fetch?id=req_id
GET /v1/process/status?id=req_id

The responsibilty of generating a unique request id is moved to the Store implementation if the user doesn't provide one with X-Numaflow-Id header. If one is provided and if it already exist in the Store, we return 409 conflict status.

Since request id is generated by Store (or validated if user provides it), we use different tid for tracing (logs).

A request (/sync and /async) is registered in the store as the first step. With Redis store implementation, this adds an entry:

request:${request_id}:status  = 'pending'

When the processing is completed and all callbacks are received, the removal of request id from the tracker implementation causes the Redis status key to change value to 'completed'.

The callbacks will be stored in Redis in a list with the key request:${request}:callbacks and the result will be saved using the key request:${request}:results.

Example: The entries in Redis follows below structure:

request:01951c1a-418f-74f3-a2b2-ee480d877f80:status  = "completed"
request:01951c1a-418f-74f3-a2b2-ee480d877f80:results = {"hello":"world"}

request:01951c1a-418f-74f3-a2b2-ee480d877f80:callbacks = [
    {"id":"01951c1a-418f-74f3-a2b2-ee480d877f80","vertex":"serve-sink","cb_time":1739933238792,"from_vertex":"serving-in","responses":[{"tags":null}]}

    {"id":"01951c1a-418f-74f3-a2b2-ee480d877f80","vertex":"serving-in","cb_time":1739933237779,"from_vertex":"serving-in","responses":[{"tags":null}]}
]

Monovertex still needs to make a callback (localhost) when Sink processing is successful. I attempted to remove this by storing a handle to the callback state within the callback handler implementation. However it resulted in generics everywhere (start_source_forwarder, create_components, etc.) since callback state is generic over store.

Currently the callbacks are stored when serving pod receives a callback from each vertex for each message. Doing this only when pod is shutting down or when all callback is received can reduce the Redis API calls. This can be implemented in the future. Currently I'm prioritizing things that will make the main branch in a working/usable state.

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

codecov · 2025-02-06T07:38:03Z

Codecov Report

Attention: Patch coverage is 82.24852% with 90 lines in your changes missing coverage. Please review.

Project coverage is 70.99%. Comparing base (d5d4408) to head (c14531b).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/serving/src/app/jetstream_proxy.rs	72.04%	45 Missing ⚠️
rust/serving/src/app/callback/store/redisstore.rs	79.13%	24 Missing ⚠️
rust/serving/src/config.rs	83.33%	17 Missing ⚠️
rust/serving/src/app/callback/store/memstore.rs	93.10%	2 Missing ⚠️
rust/serving/src/app/response.rs	0.00%	1 Missing ⚠️
rust/serving/src/app/tracker.rs	97.05%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2382      +/-   ##
==========================================
+ Coverage   70.98%   70.99%   +0.01%     
==========================================
  Files         367      368       +1     
  Lines       53042    53321     +279     
==========================================
+ Hits        37651    37856     +205     
- Misses      14311    14387      +76     
+ Partials     1080     1078       -2

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

…t in in-memory store Signed-off-by: Sreekanth <[email protected]>

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

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

yhl25

Partial review

yhl25 · 2025-02-19T07:58:59Z

rust/serving/src/app.rs

+                    // Generate a tid with good enough randomness and not too long
+                    // Example of a UUID v7: 01951b72-d0f4-711e-baba-4efe03d9cb76
+                    // We use the characters representing timestamp in milliseconds (without '-'), and last 5 characters for randomness.
+                    let uuid = Uuid::now_v7().to_string();


We only use uuid if the user has not passed id right?

This uuid is not used to represent the data stored in Store. Added details in the PR description.

Since request id is generated by the Store (or validated if user provides it), we use different tid for tracing (logs).

what is the point of this TID? just to log?

Yes, all log lines corresponding to a request will have this tid. Ideally the same should be used by the callback requests for that request too, but currently it's not propagated.

yhl25 · 2025-02-19T08:00:53Z

rust/serving/src/app/callback/store.rs

@@ -17,13 +19,51 @@ pub(crate) enum PayloadToSave {
    },
 }

+#[derive(Debug, PartialEq)]
+pub(crate) enum PipelineResult {


Consider a generic name that works for mvtx as well?

yhl25 · 2025-02-19T08:06:41Z

rust/serving/src/app/callback/store/memstore.rs

-    async fn retrieve_datum(&mut self, id: &str) -> Result<Vec<Vec<u8>>, Error> {
-        let id = format!("{}_{}", id, SAVED);
+    async fn retrieve_datum(&mut self, id: &str) -> StoreResult<PipelineResult> {
+        let id = format!("{}_{}", id, "saved");


Keep constant ?

yhl25 · 2025-02-19T08:22:25Z

rust/serving/src/app/callback/store/redisstore.rs

+            }
+        }
+    }
+    async fn done(&mut self, id: String) -> StoreResult<()> {


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

vigith · 2025-02-20T03:10:49Z

just returning a status is not enough, we will have to return body. e.g., 409 status code is not definite enough for clients.

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

vigith

let's do this in follow up PR

vigith · 2025-02-20T03:16:28Z

rust/serving/src/app/callback/state.rs

-    pub(crate) fn register(&mut self, id: String) -> oneshot::Receiver<Result<String, Error>> {
+    pub(crate) async fn register(
+        &mut self,
+        id: Option<String>,


let's make sure that the ID is no longer optional.

vigith · 2025-02-20T03:18:19Z

rust/serving/src/app/callback/store.rs

+    /// Register a request id in the store. If user provides a request id, the same should be returned
+    /// if it doesn't already exist in the store. An error should be returned if the user-specified request id
+    /// already exists in the store. If the `id` is `None`, the store should generate a new unique request id.


let's make sure ID is created at the top (when the request is received)

vigith · 2025-02-20T03:19:55Z

rust/serving/src/app/callback/store/redisstore.rs

+                }
+                Ok(id)
+            }
+            None => {


if ID is generated from above, we do not have to do this. This will be an over complication once we have other stores (IPS, DDB, etc.)

Pipeline graph with empty edges to represent Monovertex

62fc3b0

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

BulkBeing added 18 commits February 6, 2025 15:46

Callback for async requests should not fail for request id not presen…

1d45879

…t in in-memory store Signed-off-by: Sreekanth <[email protected]>

Merge branch 'main' of https://github.com/numaproj/numaflow into mvtx…

251a2b6

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

Register request id to Redis

08b49c0

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

Update processing completed status in Redis

0490bc6

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

Return after registering in Redis for monovertex async endpoint

61d86a4

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

/serve API endpoint to retrive results

6bcf7b1

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

Fix redis key name

75b1041

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

Generate new request id if user doesn't specify one

07e6c27

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

Set expiry to redis commands

966a132

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

Return 409 if the request id already exists

a7c36df

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

Return proper message graph for Monovertex

a405704

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

Remove v4 feature of uuid crate

97b1f37

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

Fix unit tests

b79e82b

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

Fix unit tests

9b7dfe2

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

Rename store trait method

51c5b4f

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

Unit tests for /serve API

184a4e1

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

Unit tests monovertex config parsing

0fbe28a

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

Minor fixes

82ff763

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

BulkBeing marked this pull request as ready for review February 19, 2025 06:36

BulkBeing requested review from whynowy and vigith as code owners February 19, 2025 06:36

Rename /message endpoint to /status

826417c

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

yhl25 reviewed Feb 19, 2025

View reviewed changes

BulkBeing and others added 2 commits February 19, 2025 15:31

Refactor PipelineResult to ProcessingStatus in processing state manag…

3b87861

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

Merge branch 'main' into mvtx-serving

b84a0dd

vigith added 2 commits February 19, 2025 19:29

chore: code review

152dd02

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

chore: fmt

dd9aebb

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

fix: test

c14531b

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

yhl25 approved these changes Feb 20, 2025

View reviewed changes

vigith approved these changes Feb 20, 2025

View reviewed changes

vigith merged commit 01cc33d into main Feb 20, 2025
28 checks passed

vigith deleted the mvtx-serving branch February 20, 2025 04:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Serving as a builtin source with Monovertex #2382

feat: Serving as a builtin source with Monovertex #2382

BulkBeing commented Feb 6, 2025 •

edited

Loading

codecov bot commented Feb 6, 2025 •

edited

Loading

yhl25 left a comment

yhl25 Feb 19, 2025

BulkBeing Feb 19, 2025

vigith Feb 20, 2025

BulkBeing Feb 20, 2025

yhl25 Feb 19, 2025

yhl25 Feb 19, 2025

yhl25 Feb 19, 2025

yhl25 Feb 19, 2025

vigith commented Feb 20, 2025

vigith left a comment

vigith Feb 20, 2025

vigith Feb 20, 2025

vigith Feb 20, 2025

feat: Serving as a builtin source with Monovertex #2382

feat: Serving as a builtin source with Monovertex #2382

Conversation

BulkBeing commented Feb 6, 2025 • edited Loading

APIs:

codecov bot commented Feb 6, 2025 • edited Loading

Codecov Report

yhl25 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

vigith commented Feb 20, 2025

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

BulkBeing commented Feb 6, 2025 •

edited

Loading

codecov bot commented Feb 6, 2025 •

edited

Loading