Updated logic for payload exchange

codevalley · codevalley · commit 6ae5b83631ff · 2025-03-05T09:54:43.000+05:30
diff --git a/.gitignore b/.gitignore
@@ -21,6 +21,8 @@
 send_response.txt
 MATCH_STATS.md
 server_log.txt
+repopack-output.txt
+repomix-output.txt
 # Test coverage reports
 coverage/
 *.profraw
diff --git a/src/queue/q_matching.rs b/src/queue/q_matching.rs
@@ -108,7 +108,7 @@ impl UnifiedQueue {
     }
     
     /// Helper method to atomically match two requests
-    pub(super) fn atomic_match(&self, new_request: &QueuedRequest, matched_request_ref: &QueuedRequest) -> Result<MatchResult, BumpError> {
+    pub(super) fn atomic_match(&self, new_request: &QueuedRequest, matched_request_ref: &QueuedRequest) -> Result<(MatchResult, MatchResult), BumpError> {
         // In this new implementation:
         // - new_request is always the request being added (has full channel)
         // - matched_request_ref is just a reference object with the ID of the matched request
@@ -388,12 +388,12 @@ impl UnifiedQueue {
             log::info!("Notifying send request {} of match", send_id);
             match tx {
                 ResponseChannel::OneShot(tx) => {
-                    if let Err(e) = tx.send(send_match_result.clone()) {
+                    if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {
                         log::error!("Failed to send match result to send request {}: {:?}", send_id, e);
                     }
                 },
                 ResponseChannel::Broadcast(tx) => {
-                    if let Err(e) = tx.send(send_match_result.clone()) {
+                    if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {
                         log::error!("Failed to broadcast match result to send request {}: {:?}", send_id, e);
                     }
                 }
@@ -404,18 +404,18 @@ impl UnifiedQueue {
             log::info!("Notifying receive request {} of match", receive_id);
             match tx {
                 ResponseChannel::OneShot(tx) => {
-                    if let Err(e) = tx.send(receive_match_result.clone()) {
+                    if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {
                         log::error!("Failed to send match result to receive request {}: {:?}", receive_id, e);
                     }
                 },
                 ResponseChannel::Broadcast(tx) => {
-                    if let Err(e) = tx.send(receive_match_result.clone()) {
+                    if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {
                         log::error!("Failed to broadcast match result to receive request {}: {:?}", receive_id, e);
                     }
                 }
             }
         }
         
-        Ok(send_match_result)
+        Ok((send_match_result, receive_match_result))
     }
 }
diff --git a/src/queue/types.rs b/src/queue/types.rs
@@ -31,9 +31,9 @@ pub struct QueuedRequest {
 #[derive(Debug)]
 pub enum ResponseChannel {
     /// Traditional oneshot channel for send/receive
-    OneShot(tokio::sync::oneshot::Sender<MatchResult>),
+    OneShot(tokio::sync::oneshot::Sender<(MatchResult, MatchResult)>),
     /// Broadcast channel for bump requests
-    Broadcast(tokio::sync::broadcast::Sender<MatchResult>),
+    Broadcast(tokio::sync::broadcast::Sender<(MatchResult, MatchResult)>),
 }
 
 impl Clone for QueuedRequest {
@@ -119,7 +119,7 @@ pub enum RequestEventType {
 #[async_trait::async_trait]
 pub trait RequestQueue: Send + Sync + 'static {
     /// Add a new request to the queue
-    async fn add_request(&self, request: QueuedRequest) -> Result<Option<MatchResult>, BumpError>;
+    async fn add_request(&self, request: QueuedRequest) -> Result<Option<(MatchResult, MatchResult)>, BumpError>;
     
     /// Remove a request from the queue
     async fn remove_request(&self, request_id: &str) -> Result<(), BumpError>;
@@ -149,5 +149,5 @@ pub trait RequestQueue: Send + Sync + 'static {
     /// Applications should now create a channel when adding the request and
     /// wait on that channel directly.
     #[allow(unused)]
-    async fn wait_for_match(&self, request_id: &str, ttl_ms: u64) -> Result<Option<MatchResult>, BumpError>;
+    async fn wait_for_match(&self, request_id: &str, ttl_ms: u64) -> Result<Option<(MatchResult, MatchResult)>, BumpError>;
 }
diff --git a/src/service.rs b/src/service.rs
@@ -206,15 +206,22 @@ impl MatchingService {
         // Add the request to the queue
         log::info!("Adding send request {} to queue", request_id);
         match self.queue.add_request(queued_request).await {
-            Ok(Some(match_result)) => {
+            Ok(Some((send_result, recv_result))) => {
                 // Immediate match found
                 log::info!("Immediate match found for send request {}", request_id);
                 
+                // For send requests, we want the send side result
+                let our_match_result = if request_id == send_result.matched_with {
+                    recv_result  // We're the receive side
+                } else {
+                    send_result  // We're the send side
+                };
+                
                 return Ok(MatchResponse {
                     status: MatchStatus::Matched,
                     sender_id: Some(request_id),
-                    receiver_id: Some(match_result.matched_with),
-                    timestamp: match_result.timestamp,
+                    receiver_id: Some(our_match_result.matched_with),
+                    timestamp: our_match_result.timestamp,
                     payload: Some(request.payload),
                     message: None,
                 });
@@ -226,15 +233,22 @@ impl MatchingService {
                 // Wait for a match with timeout
                 let ttl = std::time::Duration::from_millis(request.ttl as u64);
                 match tokio::time::timeout(ttl, rx).await {
-                    Ok(Ok(match_result)) => {
+                    Ok(Ok((send_result, recv_result))) => {
                         // We got a match result from the oneshot channel
                         log::info!("Match found for send request {}", request_id);
                         
+                        // For send requests, we want the send side result
+                        let our_match_result = if request_id == send_result.matched_with {
+                            recv_result  // We're the receive side
+                        } else {
+                            send_result  // We're the send side
+                        };
+                        
                         return Ok(MatchResponse {
                             status: MatchStatus::Matched,
                             sender_id: Some(request_id),
-                            receiver_id: Some(match_result.matched_with),
-                            timestamp: match_result.timestamp,
+                            receiver_id: Some(our_match_result.matched_with),
+                            timestamp: our_match_result.timestamp,
                             payload: Some(request.payload),
                             message: None,
                         });
@@ -292,16 +306,23 @@ impl MatchingService {
         // Add the request to the unified queue
         log::info!("Adding receive request {} to queue", request_id);
         match self.queue.add_request(queued_request).await {
-            Ok(Some(match_result)) => {
+            Ok(Some((send_result, recv_result))) => {
                 // Immediate match found
                 log::info!("Immediate match found for receive request {}", request_id);
                 
+                // For receive requests, we want the receive side result
+                let our_match_result = if request_id == send_result.matched_with {
+                    recv_result  // We're the receive side
+                } else {
+                    send_result  // We're the send side
+                };
+                
                 return Ok(MatchResponse {
                     status: MatchStatus::Matched,
-                    sender_id: Some(match_result.matched_with),
+                    sender_id: Some(our_match_result.matched_with),
                     receiver_id: Some(request_id),
-                    timestamp: match_result.timestamp,
-                    payload: match_result.payload,
+                    timestamp: our_match_result.timestamp,
+                    payload: our_match_result.payload,
                     message: None,
                 });
             },
@@ -400,12 +421,21 @@ impl MatchingService {
                 // Immediate match found
                 log::info!("Immediate match found for bump request {}", request_id);
                 
+                // Choose the correct match result based on which side we are
+                let our_match_result = if request_id == send_match_result.matched_with {
+                    // We are the receive side
+                    recv_match_result
+                } else {
+                    // We are the send side
+                    send_match_result
+                };
+                
                 return Ok(MatchResponse {
                     status: MatchStatus::Matched,
                     sender_id: Some(request_id),
-                    receiver_id: Some(match_result.matched_with),
-                    timestamp: match_result.timestamp,
-                    payload: match_result.payload,
+                    receiver_id: Some(our_match_result.matched_with),
+                    timestamp: our_match_result.timestamp,
+                    payload: our_match_result.payload,
                     message: None,
                 });
             },
@@ -420,12 +450,21 @@ impl MatchingService {
                         // We got a match result from the oneshot channel
                         log::info!("Match found for bump request {}", request_id);
                         
+                        // Choose the correct match result based on which side we are
+                        let our_match_result = if request_id == send_match_result.matched_with {
+                            // We are the receive side
+                            recv_match_result
+                        } else {
+                            // We are the send side
+                            send_match_result
+                        };
+                        
                         return Ok(MatchResponse {
                             status: MatchStatus::Matched,
                             sender_id: Some(request_id),
-                            receiver_id: Some(match_result.matched_with),
-                            timestamp: match_result.timestamp,
-                            payload: match_result.payload,
+                            receiver_id: Some(our_match_result.matched_with),
+                            timestamp: our_match_result.timestamp,
+                            payload: our_match_result.payload,
                             message: None,
                         });
                     },

Original file line number	Diff line number	Diff line change
`@@ -108,7 +108,7 @@ impl UnifiedQueue {`
`108`	`108`	`}`
`109`	`109`
`110`	`110`	`/// Helper method to atomically match two requests`
`111`		`- pub(super) fn atomic_match(&self, new_request: &QueuedRequest, matched_request_ref: &QueuedRequest) -> Result<MatchResult, BumpError> {`
	`111`	`+ pub(super) fn atomic_match(&self, new_request: &QueuedRequest, matched_request_ref: &QueuedRequest) -> Result<(MatchResult, MatchResult), BumpError> {`
`112`	`112`	`// In this new implementation:`
`113`	`113`	`// - new_request is always the request being added (has full channel)`
`114`	`114`	`// - matched_request_ref is just a reference object with the ID of the matched request`
`@@ -388,12 +388,12 @@ impl UnifiedQueue {`
`388`	`388`	`log::info!("Notifying send request {} of match", send_id);`
`389`	`389`	`match tx {`
`390`	`390`	`ResponseChannel::OneShot(tx) => {`
`391`		`- if let Err(e) = tx.send(send_match_result.clone()) {`
	`391`	`+ if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {`
`392`	`392`	`log::error!("Failed to send match result to send request {}: {:?}", send_id, e);`
`393`	`393`	`}`
`394`	`394`	`},`
`395`	`395`	`ResponseChannel::Broadcast(tx) => {`
`396`		`- if let Err(e) = tx.send(send_match_result.clone()) {`
	`396`	`+ if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {`
`397`	`397`	`log::error!("Failed to broadcast match result to send request {}: {:?}", send_id, e);`
`398`	`398`	`}`
`399`	`399`	`}`
`@@ -404,18 +404,18 @@ impl UnifiedQueue {`
`404`	`404`	`log::info!("Notifying receive request {} of match", receive_id);`
`405`	`405`	`match tx {`
`406`	`406`	`ResponseChannel::OneShot(tx) => {`
`407`		`- if let Err(e) = tx.send(receive_match_result.clone()) {`
	`407`	`+ if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {`
`408`	`408`	`log::error!("Failed to send match result to receive request {}: {:?}", receive_id, e);`
`409`	`409`	`}`
`410`	`410`	`},`
`411`	`411`	`ResponseChannel::Broadcast(tx) => {`
`412`		`- if let Err(e) = tx.send(receive_match_result.clone()) {`
	`412`	`+ if let Err(e) = tx.send((send_match_result.clone(), receive_match_result.clone())) {`
`413`	`413`	`log::error!("Failed to broadcast match result to receive request {}: {:?}", receive_id, e);`
`414`	`414`	`}`
`415`	`415`	`}`
`416`	`416`	`}`
`417`	`417`	`}`
`418`	`418`
`419`		`- Ok(send_match_result)`
	`419`	`+ Ok((send_match_result, receive_match_result))`
`420`	`420`	`}`
`421`	`421`	`}`