Trying Out the Messaging Infrastructure Deno Queues in Deno

There was an announcement about Deno Queues.

Announcing Deno Queues

Deno Queues is a messaging infrastructure implemented on Deno KV.

Using Queues | Deno Docs

Deno Deploy can be used as a foundation for asynchronous processing. The announcement blog mentions the following use cases:

• Scheduled email notifications for user-initiated tasks
• Reliable Webhook processing
• Bots for Discord and Slack

Two methods, enqueue and listenQueue, are provided in the Deno.Kv namespace. Enqueue is used to send messages, and listenQueue is implemented on the receiving side.

Here is a simple sample of sending and receiving messages.

• main.ts

// Definition of notification messages
interface Notification {
  type: "email" | "sms";
  to: string;
  body: string;
}

// Message validation
function isNotification(o : unknown): o is Notification {
  return (
    ((o as Notification)?.type === "email" || (o as Notification)?.type === "sms") &&
     (o as Notification)?.to !== undefined &&
     typeof (o as Notification)?.to === "string" &&
     (o as Notification)?.body !== undefined &&
     typeof (o as Notification)?.body === "string"
     );
}

const db = await Deno.openKv();

// Receiving messages
db.listenQueue((msg) => {
  if (!isNotification(msg)) {
    console.error("invalid message", msg);
    return;
  }
  if (msg.type === "email") {
    console.log("sending email to", msg.to, msg.body);
  } else if (msg.type === "sms") {
    console.log("sending sms to", msg.to, msg.body);
  }
});

// Creating messages
const message1: Notification = { type: "email", to: "Alice", body: "Hello, Alice!" };
const message2: Notification = { type: "sms", to: "Bob", body: "Hi, Bob!" };

// Sending messages
await db.enqueue(message1, { delay: 1000 });
await db.enqueue(message2, { delay: 2000 });

In enqueue, you can specify the sending timing with the delay option.

When you run this code, messages are output at 1-second intervals.

$ deno run --unstable main.ts 
sending email to Alice Hello, Alice!
sending sms to Bob Hi, Bob!

The above sample shared the queue within the same process. Let's separate the processes for sending and receiving.

Code for the sender. The path to the local SQLite file is specified as an argument to the openKv method.

• publisher.ts

interface Notification {
  type: "email" | "sms";
  to: string;
  body: string;
}

const db = await Deno.openKv("db");

const message1: Notification = { type: "email", to: "Alice", body: "Hello, Alice!" };
const message2: Notification = { type: "sms", to: "Bob", body: "Hi, Bob!" };

await db.enqueue(message1);
await db.enqueue(message2);

console.log("messages enqueued");

Code for the receiver. Specify the same SQLite file path as the sender in the openKv method. Although redundant, the interface is defined individually.

interface Notification {
  type: "email" | "sms";
  to: string;
  body: string;
}

function isNotification(o : unknown): o is Notification {
  // Omitted
}

const db = await Deno.openKv("db");

db.listenQueue((msg) => {
  if (!isNotification(msg)) {
    console.error("invalid message", msg);
    return;
  }
  if (msg.type === "email") {
    console.log("sending email to", msg.to, msg.body);
  } else if (msg.type === "sms") {
    console.log("sending sms to", msg.to, msg.body);
  }
});

Execution of the sender. Permissions for reading and writing the database file are required.

$ deno run --unstable --allow-read --allow-write publisher.ts
messages enqueued

Execution of the receiver. The messages were successfully read.

$ deno run --unstable --allow-read --allow-write subscriber.ts
sending email to Alice Hello, Alice!
sending sms to Bob Hi, Bob!

Let's pick up from the documentation about the messaging functionality guaranteed by Deno Queues. The quoted parts below are slightly modified from Google Translate.

At least once delivery is guaranteed.

The Deno runtime guarantees "at least once delivery." This means that for most messages enqueued, the handler will be called once per listenQueue message. In some failure scenarios, the handler may be called multiple times for the same message to ensure delivery. It is important to design your application to handle duplicate messages correctly.

It seems that a design based on message duplication is required. It would be necessary to remember the processed messages on the receiving side and skip the same message if it comes again.

Regarding the guarantee of order, it seems to be best effort.

The Deno runtime makes best efforts to deliver messages in the order they were enqueued. However, strict order is not guaranteed. In some cases, messages may be delivered out of order to ensure maximum throughput.

As for the behavior on the receiving side, it seems that if the default retry count^[1] is exceeded, the message will be dropped.

The listenQueue handler is called to process messages enqueued when they are ready for delivery. If the listenQueue handler throws an exception, the runtime will automatically retry calling the handler until it succeeds or the maximum retry count is reached. Once the handler call completes successfully, the message is considered successfully processed. If the handler consistently fails to retry, the message will be dropped.

By using the atomic transactions of Deno KV, it seems possible to perform message sending/receiving and data updates atomically. If you can bundle data updates and message sending/receiving in a transaction, it could be used for purposes such as sending messages only if the data update is successful or updating data only once when receiving a message.

That's it for a brief look at Deno Queues. It's great to be able to achieve asynchronous application integration with simple code. Complex tasks can now be done even in edge environments.