Merge of #4173

docs/docs/04-standard-library/01-cloud/queue.md

@@ -20,6 +20,7 @@ sidebar_position: 1
 
 The `cloud.Queue` resource represents a data structure for holding a list of messages.
 Queues are typically used to decouple producers of data and the consumers of said data in distributed systems.
+Queues by default are not FIFO (first in, first out) - so the order of messages is not guaranteed.
 
 ## Usage
 

examples/tests/sdk_tests/queue/pop.main.w

@@ -9,7 +9,8 @@ test "pop" {
   let second = q.pop();
   let third = q.pop();
 
-  assert(first == "Foo");
-  assert(second == "Bar");
+  // queue is not FIFO
+  assert(first == "Foo" || first == "Bar");
+  assert(second == "Foo" || second == "Bar");
   assert(third == nil);
-}
+}

libs/wingsdk/src/cloud/queue.md

@@ -20,6 +20,7 @@ sidebar_position: 1
 
 The `cloud.Queue` resource represents a data structure for holding a list of messages.
 Queues are typically used to decouple producers of data and the consumers of said data in distributed systems.
+Queues by default are not FIFO (first in, first out) - so the order of messages is not guaranteed.
 
 ## Usage
 

libs/wingsdk/src/target-sim/queue.inflight.ts

@@ -84,7 +84,11 @@ export class Queue
     return this.context.withTrace({
       message: `Pop ().`,
       activity: async () => {
-        const message = this.messages.shift();
+        // extract a random message from the queue
+        const message = this.messages.splice(
+          Math.floor(Math.random() * this.messages.length),
+          1
+        )[0];
         return message?.payload;
       },
     });
@@ -97,11 +101,22 @@ export class Queue
       // Randomize the order of subscribers to avoid user code making
       // assumptions on the order that subscribers process messages.
       for (const subscriber of new RandomArrayIterator(this.subscribers)) {
-        const messages = this.messages.splice(0, subscriber.batchSize);
+        // Extract random messages from the queue
+        const messages = new Array<QueueMessage>();
+        for (let i = 0; i < subscriber.batchSize; i++) {
+          const message = this.messages.splice(
+            Math.floor(Math.random() * this.messages.length),
+            1
+          )[0];
+          if (message) {
+            messages.push(message);
+          }
+        }
         const messagesPayload = messages.map((m) => m.payload);
         if (messagesPayload.length === 0) {
           continue;
         }
+
         const fnClient = this.context.findInstance(
           subscriber.functionHandle!
         ) as IFunctionClient & ISimulatorResourceInstance;

libs/wingsdk/test/target-sim/__snapshots__/queue.test.ts.snap

@@ -1011,22 +1011,6 @@ async handle(message) {
 `;
 
 exports[`queue with one subscriber, default batch size of 1 1`] = `
-[
-  "wingsdk.cloud.TestRunner created.",
-  "wingsdk.cloud.Function created.",
-  "wingsdk.cloud.Queue created.",
-  "wingsdk.sim.EventMapping created.",
-  "Push (messages=A,B).",
-  "Sending messages (messages=[\\"A\\"], subscriber=sim-1).",
-  "Sending messages (messages=[\\"B\\"], subscriber=sim-1).",
-  "wingsdk.sim.EventMapping deleted.",
-  "wingsdk.cloud.Queue deleted.",
-  "wingsdk.cloud.Function deleted.",
-  "wingsdk.cloud.TestRunner deleted.",
-]
-`;
-
-exports[`queue with one subscriber, default batch size of 1 2`] = `
 {
   ".wing/my_queue-setconsumer-e645076f_c8ddc1ce.js": "exports.handler = async function(event) {
   return await (new (require(\\"[REDACTED]/wingsdk/src/target-sim/queue.setconsumer.inflight.js\\")).QueueSetConsumerHandlerClient({ handler: new ((function(){

libs/wingsdk/test/target-sim/queue.test.ts

@@ -62,7 +62,7 @@ test("queue with one subscriber, default batch size of 1", async () => {
   // THEN
   await s.stop();
 
-  expect(listMessages(s)).toMatchSnapshot();
+  expect(listMessages(s)).not.toContain("Subscriber error");
   expect(app.snapshot()).toMatchSnapshot();
 });
 
@@ -139,11 +139,14 @@ async handle() {
   // THEN
   await s.stop();
 
-  const traces = s.listTraces().map((trace) => trace.data.message);
-  expect(traces).toContain(
-    'Invoke (payload="{\\"messages\\":[\\"A\\",\\"B\\",\\"C\\",\\"D\\",\\"E\\"]}").'
-  );
-  expect(traces).toContain('Invoke (payload="{\\"messages\\":[\\"F\\"]}").');
+  const invokeMessages = s
+    .listTraces()
+    .filter(
+      (trace) =>
+        trace.sourcePath === "root/my_queue/my_queue-SetConsumer-e645076f" &&
+        trace.data.message.startsWith("Invoke")
+    );
+  expect(invokeMessages.length).toEqual(2); // queue messages are processed in two batches based on batch size
   expect(app.snapshot()).toMatchSnapshot();
 });
 
@@ -327,6 +330,7 @@ test("can pop messages from queue", async () => {
   for (let i = 0; i < messages.length; i++) {
     poppedMessages.push(await queueClient.pop());
   }
+  poppedMessages.sort();
   const poppedOnEmptyQueue = await queueClient.pop();
 
   // THEN

tools/hangar/__snapshots__/test_corpus/sdk_tests/queue/pop.main.w_compile_tf-aws.md

@@ -14,8 +14,8 @@ module.exports = function({ $q }) {
       const first = (await $q.pop());
       const second = (await $q.pop());
       const third = (await $q.pop());
-      {((cond) => {if (!cond) throw new Error("assertion failed: first == \"Foo\"")})((((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(first,"Foo")))};
-      {((cond) => {if (!cond) throw new Error("assertion failed: second == \"Bar\"")})((((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(second,"Bar")))};
+      {((cond) => {if (!cond) throw new Error("assertion failed: first == \"Foo\" || first == \"Bar\"")})(((((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(first,"Foo")) || (((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(first,"Bar"))))};
+      {((cond) => {if (!cond) throw new Error("assertion failed: second == \"Foo\" || second == \"Bar\"")})(((((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(second,"Foo")) || (((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(second,"Bar"))))};
       {((cond) => {if (!cond) throw new Error("assertion failed: third == nil")})((((a,b) => { try { return require('assert').deepStrictEqual(a,b) === undefined; } catch { return false; } })(third,undefined)))};
     }
   }