Fix unacknowledged msg after cancelling match

Author: yuxuanleong

Backend/MatchingService/rabbitmq/setup.js

@@ -13,7 +13,7 @@ const queueNames = [
     'medium.cplusplus',
     'hard.python',
     'hard.java',
-    'hard.cplusplus'
+    'hard.cplusplus',
 ];
 
 async function setupRabbitMQ() {
@@ -27,17 +27,17 @@ async function setupRabbitMQ() {
         const channel = await connection.createChannel();
 
         // Declare the matching exchange (topic)
-        await channel.assertExchange(matching_exchange_name, "topic", { durable: false });
+        await channel.assertExchange(matching_exchange_name, "topic", { durable: true });
 
         // Declare the dead-letter exchange (fanout)
-        await channel.assertExchange(dead_letter_exchange_name, "fanout", { durable: false });
+        await channel.assertExchange(dead_letter_exchange_name, "fanout", { durable: true });
 
         // Declare and bind all main queues with TTL and DLQ bindings
         for (let queueName of queueNames) {
             await channel.deleteQueue(queueName);  // Ensure we start fresh for each setup
 
             await channel.assertQueue(queueName, {
-                durable: false, 
+                durable: true, 
                 arguments: {
                     'x-message-ttl': 10000,  // 60 seconds TTL
                     'x-dead-letter-exchange': dead_letter_exchange_name  // Bind to dead-letter exchange
@@ -51,12 +51,12 @@ async function setupRabbitMQ() {
         await channel.deleteQueue(dead_letter_queue_name);
 
         // Declare the dead-letter queue and bind it to the dead-letter exchange
-        await channel.assertQueue(dead_letter_queue_name, { durable: false });
+        await channel.assertQueue(dead_letter_queue_name, { durable: true });
         await channel.bindQueue(dead_letter_queue_name, dead_letter_exchange_name, ''); // Bind with no routing key
 
         // Declare and bind the cancel queue
         await channel.deleteQueue(cancel_queue_name);  // Delete any existing cancel queue
-        await channel.assertQueue(cancel_queue_name, { durable: false }); // Declare the cancel queue
+        await channel.assertQueue(cancel_queue_name, { durable: true }); // Declare the cancel queue
         await channel.bindQueue(cancel_queue_name, matching_exchange_name, 'cancel'); // Bind with the "cancel" routing key
 
         console.log("RabbitMQ setup complete with queues, DLQ, and bindings.");

Backend/MatchingService/rabbitmq/subscriber.js

@@ -23,20 +23,20 @@ function matchUsers(channel, msg, userId, language, difficulty) {
         waitingUsers[criteriaKey] = [];
     }
 
-    waitingUsers[criteriaKey].push({ userId, msg }); // Store both userId and the message
+    // Store both the userId, message, and the channel in waitingUsers
+    waitingUsers[criteriaKey].push({ userId, msg, channel });
     console.log(`User ${userId} added to ${criteriaKey}. Waiting list: ${waitingUsers[criteriaKey].length}`);
 
     // Check if there are 2 or more users waiting for this criteria
     if (waitingUsers[criteriaKey].length >= 2) {
         const matchedUsers = waitingUsers[criteriaKey].splice(0, 2); // Match the first two users
         console.log(`Matched users: ${matchedUsers.map(user => user.userId)}`);
 
-        // Send match success (this could trigger WebSocket communication)
-        // notifyUsers(matchedUsers.map(user => user.userId));
+        // Notify users of the match
         notifyUsers(matchedUsers.map(user => user.userId), 'Match found!', 'match');
 
         // Acknowledge the messages for both matched users
-        matchedUsers.forEach(({ msg }) => {
+        matchedUsers.forEach(({ msg, channel }) => {
             acknowledgeMessage(channel, msg);
         });
 
@@ -46,6 +46,7 @@ function matchUsers(channel, msg, userId, language, difficulty) {
     return false;
 }
 
+
 async function acknowledgeMessage(channel, msg) {
     return new Promise((resolve, reject) => {
         try {
@@ -104,7 +105,6 @@ async function rejectMessage(channel, msg, userId) {
 
 async function consumeQueue() {
     try {
-        // Connect
         const connection = await amqp.connect(process.env.RABBITMQ_URL);
         const channel = await connection.createChannel();
 
@@ -123,26 +123,23 @@ async function consumeQueue() {
                     // Call matchUsers with channel, message, and user details
                     const matched = matchUsers(channel, msg, userId, language, difficulty);
 
-                    // If not matched, set a timeout for rejection
                     if (!matched) {
                         console.log(`No match for ${userId}, waiting for rejection timeout.`);
 
-                        // Set a timeout for rejection after 10 seconds
                         const timeoutId = setTimeout(async () => {
                             await rejectMessage(channel, msg, userId);
                         }, 10000); // 10 seconds delay
 
-                        // Store the timeout ID
                         timeoutMap[userId] = timeoutId;
                     }
                 }
-            });
+            }, { noAck: false }); // Ensure manual acknowledgment
         }
 
         console.log("Listening to matchmaking queues");
 
         await consumeCancelQueue();
-        console.log("Listening to Cancel Queue")
+        console.log("Listening to Cancel Queue");
     } catch (error) {
         console.error('Error consuming RabbitMQ queue:', error);
     }
@@ -184,45 +181,62 @@ async function consumeCancelQueue() {
         await channel.consume('cancel_queue', async (msg) => {
             if (msg !== null) {
                 const { userId } = JSON.parse(msg.content.toString());
-
                 console.log(`Received cancel request for user: ${userId}`);
 
                 // Process the cancel request
                 await cancelMatching(channel, msg, userId);
             }
-        });
+        }, { noAck: false }); // Ensure manual acknowledgment
 
         console.log("Listening for cancel requests");
     } catch (error) {
         console.error('Error consuming cancel queue:', error);
     }
 }
 
-async function cancelMatching(channel, msg, userId) {
+async function cancelMatching(cancelChannel, cancelMsg, userId) {
     try {
-        // Loop through waitingUsers to find the user
+        let foundOriginalMsg = false;
+
+        // Loop through waitingUsers to find the original message for the user
         Object.keys(waitingUsers).forEach(criteriaKey => {
             const userIndex = waitingUsers[criteriaKey].findIndex(user => user.userId === userId);
 
             if (userIndex !== -1) {
+                const { msg, channel } = waitingUsers[criteriaKey][userIndex]; // Get original msg and its channel
+
+                // Acknowledge the original matchmaking message from the queue (e.g., easy.python)
+                if (msg && channel) {
+                    console.log(`Acknowledging original message for user ${userId} in queue ${criteriaKey}`);
+                    channel.ack(msg); // Use the same channel that consumed the message to acknowledge it
+                    foundOriginalMsg = true;
+                }
+
+                // Remove the user from the waiting list
                 waitingUsers[criteriaKey].splice(userIndex, 1);
                 console.log(`User ${userId} removed from waiting list for ${criteriaKey}`);
             }
         });
 
-        // Clean up the timeout
+        // If original message not found, log a warning
+        if (!foundOriginalMsg) {
+            console.warn(`Original message for user ${userId} not found in matchmaking queues.`);
+        }
+
+        // Clear any timeouts for the user
         if (timeoutMap[userId]) {
             clearTimeout(timeoutMap[userId]);
             delete timeoutMap[userId];
         }
 
-        // Acknowledge the cancel message
-        channel.ack(msg);
-
+        // Acknowledge the cancel message from the cancel queue
+        cancelChannel.ack(cancelMsg);
         console.log(`Cancel processed for user ${userId}`);
+
     } catch (error) {
         console.error(`Failed to process cancel for user ${userId}:`, error);
     }
 }
 
+
 module.exports = { consumeQueue, consumeDLQ };