Welcome to the official Cloudflare Developers server. Here you can ask for help and stay updated with the latest news
83,498Members
View on DiscordResources
Similar Threads
Was this page helpful?
HH
JH
A
delaySeconds no longer work?AA
A
A
SS
S
U
.retry(12 hours), and I can't figure out whether I've mis-configured something or am hitting an edge case.after: now.toISO()), and some upon delay (message has a field after: remindOn.toISO(), typically 3-5 days from now). The are 0-3 batches of these messages per day sent out worldwide, as in these are not high volume..retry({delaySeconds: X}) of messages which should not be processed yet: if the after.retry()-ed messages. This doesn't seem to happen globally either, but rather per region, as in that "fresh new unrelated message" will only trigger the re-processing of geographically proximate messages. E.g. I had somebody in Germany visit my website on Sept 11 and fill out the booking form for Sept 16 lesson; they received the "immediate" messages, but the "reminders" got stuck in limbo until Sept 11 until a completely different person somewhere in Russian filled out the form which triggered new messages to be added to the queue.UUUU
B
BBU
BBBA
AAABAA
I
SSSS
O
MMOO
CO
TC
A
body property wrapping them and when do they not? The docs are not clear on thisAbody when they are sent in a batch, but not wrapped when they are sent individually?O
B
delaySeconds.retry(12 hours)after: now.toISO()after: remindOn.toISO().retry({delaySeconds: X}).retry() await platform.env.bookings.sendBatch([
{ body: { action: 'notify', target: 'student', via: 'email', after: now.toISO(), message: 'requested_lesson', ...booking }},
...2 messages omitted...
message: 'requested_lesson', ...booking }},
{ body: { action: 'notify', target: 'student', via: 'email', after: remindOn.toISO(), message: 'confirm_lesson', ...booking }},
...2 messages omitted...
]); async queue(batch: MessageBatch<ActionMessage>, env: Env): Promise<void> {
const now = DateTime.now();
for (let message: Message<ActionMessage> of batch.messages) {
const sendAfter = DateTime.fromISO(message.body.after);
if (sendAfter > now) {
const delaySeconds = Math.round(Math.min(
Duration.fromObject({ hours: 12}).as('seconds'),
sendAfter.diff(now).as('seconds')));
message.retry({
delaySeconds: delaySeconds
});
console.info(`message ${message.id} DELAYED: ${JSON.stringify(message.body)} by ${delaySeconds}`);
} else ...[[queues.consumers]]
queue = "litcondit-bookings"
max_batch_size = 10
max_batch_timeout = 5
max_retries = 100
dead_letter_queue = "litcondit-bookings-dlq"bodybodyinterface TypedMessageBatch<Name extends string, Body> {
readonly messages: readonly Message<Body>[];
readonly queue: Name;
retryAll(options?: QueueRetryOptions): void;
ackAll(): void;
}
type MessageBatch =
| TypedMessageBatch<"queue-1", { foo: number }>
| TypedMessageBatch<"queue-2", { bar: number }>;
export default class extends WorkerEntrypoint<Env> {
async queue(_batch: MessageBatch) {
switch (batch.queue) {
case "queue-1": {
// message.body is typed
break;
}
}
}
}const Body = z.union(
z.object({ type: "A", data: {...} }),
z.object({ type: "B", data: {...} })
);
try {
msg = Body.parse(await req.json()); // msg is now guaranteed to match the `Body` schema
} catch (err) {
return new Response("Invalid body", { status: 400 });
}
await env.QUEUE.send(msg);