Fixes #115: By introducing MQ workers
This commit is contained in:
Zef Hemel
@@ -0,0 +1,107 @@
|
||||
import { Hook, Manifest } from "../types.ts";
|
||||
import { System } from "../system.ts";
|
||||
import { DexieMQ } from "../lib/mq.dexie.ts";
|
||||
import { fullQueueName } from "../lib/mq_util.ts";
|
||||
import { Message } from "$sb/mq.ts";
|
||||
|
||||
type MQSubscription = {
|
||||
queue: string;
|
||||
batchSize?: number;
|
||||
autoAck?: boolean;
|
||||
};
|
||||
|
||||
export type MQHookT = {
|
||||
mqSubscriptions?: MQSubscription[];
|
||||
};
|
||||
|
||||
export class MQHook implements Hook<MQHookT> {
|
||||
subscriptions: (() => void)[] = [];
|
||||
|
||||
constructor(private system: System<MQHookT>, readonly mq: DexieMQ) {
|
||||
}
|
||||
|
||||
apply(system: System<MQHookT>): void {
|
||||
this.system = system;
|
||||
system.on({
|
||||
plugLoaded: () => {
|
||||
this.reloadQueues();
|
||||
},
|
||||
plugUnloaded: () => {
|
||||
this.reloadQueues();
|
||||
},
|
||||
});
|
||||
|
||||
this.reloadQueues();
|
||||
}
|
||||
|
||||
stop() {
|
||||
// console.log("Unsubscribing from all queues");
|
||||
this.subscriptions.forEach((sub) => sub());
|
||||
this.subscriptions = [];
|
||||
}
|
||||
|
||||
reloadQueues() {
|
||||
this.stop();
|
||||
for (const plug of this.system.loadedPlugs.values()) {
|
||||
if (!plug.manifest) {
|
||||
continue;
|
||||
}
|
||||
for (
|
||||
const [name, functionDef] of Object.entries(
|
||||
plug.manifest.functions,
|
||||
)
|
||||
) {
|
||||
if (!functionDef.mqSubscriptions) {
|
||||
continue;
|
||||
}
|
||||
const subscriptions = functionDef.mqSubscriptions;
|
||||
for (const subscriptionDef of subscriptions) {
|
||||
const queue = fullQueueName(plug.name!, subscriptionDef.queue);
|
||||
// console.log("Subscribing to queue", queue);
|
||||
this.subscriptions.push(
|
||||
this.mq.subscribe(
|
||||
queue,
|
||||
{
|
||||
batchSize: subscriptionDef.batchSize,
|
||||
},
|
||||
async (messages: Message[]) => {
|
||||
try {
|
||||
await plug.invoke(name, [messages]);
|
||||
if (subscriptionDef.autoAck) {
|
||||
await this.mq.batchAck(queue, messages.map((m) => m.id));
|
||||
}
|
||||
} catch (e: any) {
|
||||
console.error(
|
||||
"Execution of mqSubscription for queue",
|
||||
queue,
|
||||
"invoking",
|
||||
name,
|
||||
"with messages",
|
||||
messages,
|
||||
"failed:",
|
||||
e,
|
||||
);
|
||||
}
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
validateManifest(manifest: Manifest<MQHookT>): string[] {
|
||||
const errors: string[] = [];
|
||||
for (const functionDef of Object.values(manifest.functions)) {
|
||||
if (!functionDef.mqSubscriptions) {
|
||||
continue;
|
||||
}
|
||||
for (const subscriptionDef of functionDef.mqSubscriptions) {
|
||||
if (!subscriptionDef.queue) {
|
||||
errors.push("Missing queue name for mqSubscription");
|
||||
}
|
||||
}
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
import { IDBKeyRange, indexedDB } from "https://esm.sh/fake-indexeddb@4.0.2";
|
||||
import { DexieMQ } from "./mq.dexie.ts";
|
||||
import { assertEquals } from "../../test_deps.ts";
|
||||
import { sleep } from "../../common/async_util.ts";
|
||||
|
||||
Deno.test("Dexie MQ", async () => {
|
||||
const mq = new DexieMQ("test", indexedDB, IDBKeyRange);
|
||||
await mq.send("test", "Hello World");
|
||||
let messages = await mq.poll("test", 10);
|
||||
assertEquals(messages.length, 1);
|
||||
await mq.ack("test", messages[0].id);
|
||||
assertEquals([], await mq.poll("test", 10));
|
||||
await mq.send("test", "Hello World");
|
||||
messages = await mq.poll("test", 10);
|
||||
assertEquals(messages.length, 1);
|
||||
assertEquals([], await mq.poll("test", 10));
|
||||
await sleep(20);
|
||||
await mq.requeueTimeouts(10);
|
||||
messages = await mq.poll("test", 10);
|
||||
const stats = await mq.getAllQueueStats();
|
||||
assertEquals(stats["test"].processing, 1);
|
||||
assertEquals(messages.length, 1);
|
||||
assertEquals(messages[0].retries, 1);
|
||||
await sleep(20);
|
||||
await mq.requeueTimeouts(10, 1);
|
||||
assertEquals((await mq.fetchDLQMessages()).length, 1);
|
||||
|
||||
let receivedMessage = false;
|
||||
const unsubscribe = mq.subscribe("test123", {}, async (messages) => {
|
||||
assertEquals(messages.length, 1);
|
||||
await mq.ack("test123", messages[0].id);
|
||||
receivedMessage = true;
|
||||
});
|
||||
mq.send("test123", "Hello World");
|
||||
// Give time to process the message
|
||||
await sleep(1);
|
||||
assertEquals(receivedMessage, true);
|
||||
unsubscribe();
|
||||
|
||||
// Batch send
|
||||
await mq.batchSend("test", ["Hello", "World"]);
|
||||
const messageBatch1 = await mq.poll("test", 1);
|
||||
assertEquals(messageBatch1.length, 1);
|
||||
assertEquals(messageBatch1[0].body, "Hello");
|
||||
const messageBatch2 = await mq.poll("test", 1);
|
||||
assertEquals(messageBatch2.length, 1);
|
||||
assertEquals(messageBatch2[0].body, "World");
|
||||
|
||||
await mq.batchAck("test", [messageBatch1[0].id, messageBatch2[0].id]);
|
||||
assertEquals(await mq.fetchProcessingMessages(), []);
|
||||
// Give time to close the db
|
||||
await sleep(20);
|
||||
});
|
||||
@@ -0,0 +1,275 @@
|
||||
import Dexie, { Table } from "dexie";
|
||||
import { Message } from "$sb/mq.ts";
|
||||
|
||||
export type ProcessingMessage = Message & {
|
||||
ts: number;
|
||||
};
|
||||
|
||||
export type SubscribeOptions = {
|
||||
batchSize?: number;
|
||||
pollInterval?: number;
|
||||
};
|
||||
|
||||
export type QueueStats = {
|
||||
queued: number;
|
||||
processing: number;
|
||||
dlq: number;
|
||||
};
|
||||
|
||||
export class DexieMQ {
|
||||
db: Dexie;
|
||||
queued: Table<Message, [string, string]>;
|
||||
processing: Table<ProcessingMessage, [string, string]>;
|
||||
dlq: Table<ProcessingMessage, [string, string]>;
|
||||
|
||||
// queue -> set of run() functions
|
||||
localSubscriptions = new Map<string, Set<() => void>>();
|
||||
|
||||
constructor(
|
||||
dbName: string,
|
||||
indexedDB?: any,
|
||||
IDBKeyRange?: any,
|
||||
) {
|
||||
this.db = new Dexie(dbName, {
|
||||
indexedDB,
|
||||
IDBKeyRange,
|
||||
});
|
||||
this.db.version(1).stores({
|
||||
queued: "[queue+id], queue, id",
|
||||
processing: "[queue+id], queue, id, ts",
|
||||
dlq: "[queue+id], queue, id",
|
||||
});
|
||||
this.queued = this.db.table("queued");
|
||||
this.processing = this.db.table("processing");
|
||||
this.dlq = this.db.table("dlq");
|
||||
}
|
||||
|
||||
// Internal sequencer for messages, only really necessary when batch sending tons of messages within a millisecond
|
||||
seq = 0;
|
||||
|
||||
async batchSend(queue: string, bodies: any[]) {
|
||||
const messages = bodies.map((body) => ({
|
||||
id: `${Date.now()}-${String(++this.seq).padStart(6, "0")}`,
|
||||
queue,
|
||||
body,
|
||||
}));
|
||||
|
||||
await this.queued.bulkAdd(messages);
|
||||
|
||||
// See if we can immediately process the message with a local subscription
|
||||
const localSubscriptions = this.localSubscriptions.get(queue);
|
||||
if (localSubscriptions) {
|
||||
for (const run of localSubscriptions) {
|
||||
run();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
send(queue: string, body: any) {
|
||||
return this.batchSend(queue, [body]);
|
||||
}
|
||||
|
||||
poll(queue: string, maxItems: number): Promise<Message[]> {
|
||||
return this.db.transaction(
|
||||
"rw",
|
||||
[this.queued, this.processing],
|
||||
async (tx) => {
|
||||
const messages =
|
||||
(await tx.table<Message, [string, string]>("queued").where({ queue })
|
||||
.sortBy("id")).slice(0, maxItems);
|
||||
const ids: [string, string][] = messages.map((m) => [queue, m.id]);
|
||||
await tx.table("queued").bulkDelete(ids);
|
||||
await tx.table<ProcessingMessage, [string, string]>("processing")
|
||||
.bulkPut(
|
||||
messages.map((m) => ({
|
||||
...m,
|
||||
ts: Date.now(),
|
||||
})),
|
||||
);
|
||||
return messages;
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param queue
|
||||
* @param batchSize
|
||||
* @param callback
|
||||
* @returns a function to be called to unsubscribe
|
||||
*/
|
||||
subscribe(
|
||||
queue: string,
|
||||
options: SubscribeOptions,
|
||||
callback: (messages: Message[]) => Promise<void> | void,
|
||||
): () => void {
|
||||
let running = true;
|
||||
let timeout: number | undefined;
|
||||
const batchSize = options.batchSize || 1;
|
||||
const run = async () => {
|
||||
try {
|
||||
if (!running) {
|
||||
return;
|
||||
}
|
||||
const messages = await this.poll(queue, batchSize);
|
||||
if (messages.length > 0) {
|
||||
await callback(messages);
|
||||
}
|
||||
// If we got exactly the batch size, there might be more messages
|
||||
if (messages.length === batchSize) {
|
||||
await run();
|
||||
}
|
||||
if (timeout) {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
timeout = setTimeout(run, options.pollInterval || 5000);
|
||||
} catch (e: any) {
|
||||
console.error("Error in MQ subscription handler", e);
|
||||
}
|
||||
};
|
||||
|
||||
// Register as a local subscription handler
|
||||
const localSubscriptions = this.localSubscriptions.get(queue);
|
||||
if (!localSubscriptions) {
|
||||
this.localSubscriptions.set(queue, new Set([run]));
|
||||
} else {
|
||||
localSubscriptions.add(run);
|
||||
}
|
||||
|
||||
// Run the first time (which will schedule subsequent polling intervals)
|
||||
run();
|
||||
|
||||
// And return an unsubscribe function
|
||||
return () => {
|
||||
running = false;
|
||||
if (timeout) {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
// Remove the subscription from localSubscriptions
|
||||
const queueSubscriptions = this.localSubscriptions.get(queue);
|
||||
if (queueSubscriptions) {
|
||||
queueSubscriptions.delete(run);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
ack(queue: string, id: string) {
|
||||
return this.batchAck(queue, [id]);
|
||||
}
|
||||
|
||||
async batchAck(queue: string, ids: string[]) {
|
||||
await this.processing.bulkDelete(ids.map((id) => [queue, id]));
|
||||
}
|
||||
|
||||
async requeueTimeouts(timeout: number, maxRetries?: number) {
|
||||
const now = Date.now();
|
||||
const messages = await this.processing.where("ts").below(now - timeout)
|
||||
.toArray();
|
||||
const ids: [string, string][] = messages.map((m) => [m.queue, m.id]);
|
||||
await this.db.transaction(
|
||||
"rw",
|
||||
[this.queued, this.processing, this.dlq],
|
||||
async (tx) => {
|
||||
await tx.table("processing").bulkDelete(ids);
|
||||
const requeuedMessages: ProcessingMessage[] = [];
|
||||
const dlqMessages: ProcessingMessage[] = [];
|
||||
for (const m of messages) {
|
||||
const retries = (m.retries || 0) + 1;
|
||||
if (maxRetries && retries > maxRetries) {
|
||||
console.warn(
|
||||
"[mq]",
|
||||
"Message exceeded max retries, moving to DLQ",
|
||||
m,
|
||||
);
|
||||
dlqMessages.push({
|
||||
queue: m.queue,
|
||||
id: m.id,
|
||||
body: m.body,
|
||||
ts: Date.now(),
|
||||
retries,
|
||||
});
|
||||
} else {
|
||||
console.info("[mq]", "Message ack timed out, requeueing", m);
|
||||
requeuedMessages.push({
|
||||
...m,
|
||||
retries,
|
||||
});
|
||||
}
|
||||
}
|
||||
await tx.table("queued").bulkPut(requeuedMessages);
|
||||
await tx.table("dlq").bulkPut(dlqMessages);
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
fetchDLQMessages(): Promise<ProcessingMessage[]> {
|
||||
return this.dlq.toArray();
|
||||
}
|
||||
|
||||
fetchProcessingMessages(): Promise<ProcessingMessage[]> {
|
||||
return this.processing.toArray();
|
||||
}
|
||||
|
||||
flushDLQ(): Promise<void> {
|
||||
return this.dlq.clear();
|
||||
}
|
||||
|
||||
getQueueStats(queue: string): Promise<QueueStats> {
|
||||
return this.db.transaction(
|
||||
"r",
|
||||
[this.queued, this.processing, this.dlq],
|
||||
async (tx) => {
|
||||
const queued = await tx.table("queued").where({ queue }).count();
|
||||
const processing = await tx.table("processing").where({ queue })
|
||||
.count();
|
||||
const dlq = await tx.table("dlq").where({ queue }).count();
|
||||
return {
|
||||
queued,
|
||||
processing,
|
||||
dlq,
|
||||
};
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async getAllQueueStats(): Promise<Record<string, QueueStats>> {
|
||||
const allStatus: Record<string, QueueStats> = {};
|
||||
await this.db.transaction(
|
||||
"r",
|
||||
[this.queued, this.processing, this.dlq],
|
||||
async (tx) => {
|
||||
for (const item of await tx.table("queued").toArray()) {
|
||||
if (!allStatus[item.queue]) {
|
||||
allStatus[item.queue] = {
|
||||
queued: 0,
|
||||
processing: 0,
|
||||
dlq: 0,
|
||||
};
|
||||
}
|
||||
allStatus[item.queue].queued++;
|
||||
}
|
||||
for (const item of await tx.table("processing").toArray()) {
|
||||
if (!allStatus[item.queue]) {
|
||||
allStatus[item.queue] = {
|
||||
queued: 0,
|
||||
processing: 0,
|
||||
dlq: 0,
|
||||
};
|
||||
}
|
||||
allStatus[item.queue].processing++;
|
||||
}
|
||||
for (const item of await tx.table("dlq").toArray()) {
|
||||
if (!allStatus[item.queue]) {
|
||||
allStatus[item.queue] = {
|
||||
queued: 0,
|
||||
processing: 0,
|
||||
dlq: 0,
|
||||
};
|
||||
}
|
||||
allStatus[item.queue].dlq++;
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
return allStatus;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
// Adds a plug name to a queue name if it doesn't already have one.
|
||||
export function fullQueueName(plugName: string, queueName: string) {
|
||||
if (queueName.includes(".")) {
|
||||
return queueName;
|
||||
}
|
||||
return plugName + "." + queueName;
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
import { SysCallMapping } from "../system.ts";
|
||||
import { DexieMQ } from "../lib/mq.dexie.ts";
|
||||
import { fullQueueName } from "../lib/mq_util.ts";
|
||||
|
||||
export function mqSyscalls(
|
||||
mq: DexieMQ,
|
||||
): SysCallMapping {
|
||||
return {
|
||||
"mq.send": (ctx, queue: string, body: any) => {
|
||||
return mq.send(fullQueueName(ctx.plug.name!, queue), body);
|
||||
},
|
||||
"mq.batchSend": (ctx, queue: string, bodies: any[]) => {
|
||||
return mq.batchSend(fullQueueName(ctx.plug.name!, queue), bodies);
|
||||
},
|
||||
"mq.ack": (ctx, queue: string, id: string) => {
|
||||
return mq.ack(fullQueueName(ctx.plug.name!, queue), id);
|
||||
},
|
||||
"mq.batchAck": (ctx, queue: string, ids: string[]) => {
|
||||
return mq.batchAck(fullQueueName(ctx.plug.name!, queue), ids);
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user