Process fetch responses #64

jost-s · 2025-01-03T02:16:15Z

Add a queue to process incoming responses to fetch requests. This completes the cycle from sending fetch request -> sending fetch response -> receiving fetch response.

There's a rename stacked onto this PR. Proto definition changes Request to FetchRequest and Response to FetchResponse. That makes the PR look bigger than it is.

resolves #62

crates/api/src/op_store.rs

crates/core/src/factories/core_fetch.rs

neonphog

Looks good to me 👍

ThetaSinner

Looking great. I have a few comments to add but all of them are minor :)

ThetaSinner · 2025-01-17T11:18:37Z

crates/api/src/op_store.rs

@@ -3,6 +3,7 @@
 use crate::{
    builder, config, BoxFut, DhtArc, K2Result, OpId, SpaceId, Timestamp,
 };
+use bytes::Bytes;


Most places we've used bytes::Bytes rather than importing. I'm not super fussy which we use but I'd slightly prefer to be consistent.

"Your OCD is your own problem" is a valid response to this minor comment :)

Oh and that is actually the only change to this file, is to make sure this type is imported. Do you have a preference for importing everywhere then?

I've seen this and adopted it too, until I wondered why we're doing this. In several places there are these half-qualified types and it makes the code generally more wordy and harder to read. Coming to think of it, I only find it useful when there are identical or similar types in different crates. Bytes is unique and so are probably all of our own types like the stores and the factories.
In that light I prefer importing everywhere.

ThetaSinner · 2025-01-17T11:26:22Z

crates/core/src/factories/core_fetch.rs

@@ -466,6 +476,33 @@ impl CoreFetch {
        }
    }

+    async fn spawn_incoming_response_task(
+        incoming_response_rx: Arc<


Does this need to be in an arc+mutex? We're doing a single receive task here and not multiple like with the other side?

It doesn't 👍

ThetaSinner · 2025-01-17T11:27:43Z

crates/core/src/factories/core_fetch.rs

@@ -303,6 +304,15 @@ impl CoreFetch {
            ));
        tasks.push(incoming_request_task);

+        // Spawn incoming response task.
+        let incoming_response_task =
+            tokio::task::spawn(CoreFetch::spawn_incoming_response_task(


Minor thing - it's slightly odd to have the function called spawn_xxx_task but have it actually return a future task that needs to be spawned.

Either the tokio::spawn could go inside the function and the function just returns the join handle or could the function just be incoming_response_task?

Did that. I like it better too.

crates/core/src/factories/core_fetch.rs

ThetaSinner · 2025-01-17T11:37:05Z

crates/core/src/factories/core_fetch/test/incoming_response_queue.rs

+    time::Duration,
+};
+
+const SPACE_ID: SpaceId = SpaceId(id::Id(bytes::Bytes::from_static(b"space1")));


Id love some test data somewhere that has a few things like this. Not something that needs to be done on this PR but if anybody spots an opportunity to put something like that in place, it'd be very welcome. I'll look for an opportunity too :)

I've added the test space id to the test utils.

ThetaSinner · 2025-01-17T11:38:10Z

crates/core/src/factories/core_fetch/test/incoming_response_queue.rs

+
+#[tokio::test(flavor = "multi_thread")]
+async fn incoming_ops_are_written_to_op_store() {
+    let builder =


Can the setup be extracted from the tests? It looks like there's some duplicate code here to wire up the peer store, transport and fetch module

And also done for the other test files.

ThetaSinner · 2025-01-17T11:40:18Z

crates/core/src/factories/core_fetch/test/incoming_response_queue.rs

+        )
+        .unwrap();
+
+    tokio::time::sleep(Duration::from_millis(10)).await;


Please can this be a tokio::time::timeout with a higher timeout that checks for the condition until success?

Not exactly, because it's giving fetch the chance to process incoming ops to later on check if something has not happened. I don't see how I could check for the ops having been processed and nothing having been written to the ops store to know when to assert this.

…onse

jost-s changed the base branch from main to feat/fetch-response-queue January 3, 2025 02:16

jost-s force-pushed the feat/process-fetch-responses branch from 59c92a7 to 153b705 Compare January 3, 2025 02:21

jost-s commented Jan 3, 2025

View reviewed changes

crates/api/src/op_store.rs Outdated Show resolved Hide resolved

jost-s force-pushed the feat/process-fetch-responses branch from 153b705 to 85f839e Compare January 3, 2025 02:28

jost-s requested a review from a team January 3, 2025 02:28

jost-s force-pushed the feat/fetch-response-queue branch from 4e00657 to 57b3547 Compare January 3, 2025 02:32

jost-s force-pushed the feat/process-fetch-responses branch from 85f839e to f338b3a Compare January 3, 2025 02:34

jost-s commented Jan 3, 2025

View reviewed changes

crates/core/src/factories/core_fetch.rs Outdated Show resolved Hide resolved

jost-s force-pushed the feat/fetch-response-queue branch 9 times, most recently from b975539 to 5c335a0 Compare January 8, 2025 22:56

Base automatically changed from feat/fetch-response-queue to main January 9, 2025 00:32

jost-s force-pushed the feat/process-fetch-responses branch 4 times, most recently from fa3efea to a4a1e86 Compare January 16, 2025 19:54

neonphog previously approved these changes Jan 16, 2025

View reviewed changes

ThetaSinner reviewed Jan 17, 2025

View reviewed changes

jost-s added 2 commits January 17, 2025 12:36

refactor(api): rename Request -> FetchRequest & Response -> FetchResp…

a0996cc

…onse

feat(core): process incoming responses to fetch requests

205ce3a

jost-s dismissed neonphog’s stale review via 205ce3a January 17, 2025 21:21

jost-s force-pushed the feat/process-fetch-responses branch from a4a1e86 to 205ce3a Compare January 17, 2025 21:21

feat(test-utils): add test space id

94e20dc

jost-s requested a review from ThetaSinner January 18, 2025 02:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Process fetch responses #64

Process fetch responses #64

jost-s commented Jan 3, 2025 •

edited

Loading

neonphog left a comment

ThetaSinner left a comment

ThetaSinner Jan 17, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 17, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 17, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 18, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 18, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 18, 2025

jost-s Jan 18, 2025

ThetaSinner Jan 17, 2025

jost-s Jan 17, 2025

Process fetch responses #64

Are you sure you want to change the base?

Process fetch responses #64

Conversation

jost-s commented Jan 3, 2025 • edited Loading

neonphog left a comment

Choose a reason for hiding this comment

ThetaSinner 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

jost-s commented Jan 3, 2025 •

edited

Loading