mirror of
https://github.com/requarks/wiki.git
synced 2026-10-12 14:01:58 -05:00
fix: postgres claim for job scheduler + notification fan-out
This commit is contained in:
4 files changed
+23
-8
No files matched your search
@@ -24,6 +24,13 @@ RUN apt-get update && apt-get install -qy \
|
||||
unzip \
|
||||
wget
|
||||
|
||||
# GitHub CLI, from GitHub's own apt repository: Debian's package trails upstream by many releases
|
||||
RUN mkdir -p -m 755 /etc/apt/keyrings && \
|
||||
wget -nv -O /etc/apt/keyrings/githubcli-archive-keyring.gpg https://cli.github.com/packages/githubcli-archive-keyring.gpg && \
|
||||
chmod go+r /etc/apt/keyrings/githubcli-archive-keyring.gpg && \
|
||||
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main" > /etc/apt/sources.list.d/github-cli.list && \
|
||||
apt-get update && apt-get install -qy gh
|
||||
|
||||
# The Puppeteer extension installs its own browser, in app-init.sh: the one Chrome release it is built
|
||||
# against, where a distro Chromium moves on with every update and eventually stops being one Puppeteer
|
||||
# can drive. Only the headless shell, which is all the renderer uses.
|
||||
|
||||
@@ -14,7 +14,7 @@ import {
|
||||
jobSchedule as jobScheduleTable,
|
||||
jobHistory as jobHistoryTable
|
||||
} from '../db/schema.ts'
|
||||
import { and, eq, inArray, lt, sql } from 'drizzle-orm'
|
||||
import { and, eq, lt, sql } from 'drizzle-orm'
|
||||
import type { PoolClient } from 'pg'
|
||||
|
||||
/** What the scheduler hands an in-process task besides its payload. */
|
||||
@@ -384,13 +384,13 @@ export default {
|
||||
let jobs: any[] = []
|
||||
try {
|
||||
jobs = await WIKI.db.transaction(async (trx: any) => {
|
||||
// -> `= ANY(ARRAY(...))` rather than `IN (...)`: an IN subquery may be planned as a semi join
|
||||
// that re-runs it per outer row, each run skipping what was already deleted, so the LIMIT
|
||||
// stops limiting and every due job is claimed at once. ARRAY() is evaluated exactly once.
|
||||
const claimed = await trx
|
||||
.delete(jobsTable)
|
||||
.where(
|
||||
inArray(
|
||||
jobsTable.id,
|
||||
sql`(SELECT id FROM jobs WHERE ("waitUntil" IS NULL OR "waitUntil" <= NOW()) ORDER BY id FOR UPDATE SKIP LOCKED LIMIT ${availableWorkers})`
|
||||
)
|
||||
sql`${jobsTable.id} = ANY(ARRAY(SELECT id FROM jobs WHERE ("waitUntil" IS NULL OR "waitUntil" <= NOW()) ORDER BY id FOR UPDATE SKIP LOCKED LIMIT ${availableWorkers}))`
|
||||
)
|
||||
.returning()
|
||||
for (const job of claimed) {
|
||||
|
||||
@@ -1124,10 +1124,14 @@ class Rendering {
|
||||
*/
|
||||
renderer ??= await this.createRenderer()
|
||||
|
||||
// -> `= (subquery)`, never `IN (subquery)`: postgres may plan an IN as a semi join that re-runs
|
||||
// the subquery per outer row, and each re-run skips the rows already deleted and hands back
|
||||
// the next one — so a LIMIT 1 claim empties the whole queue. A scalar subquery is an
|
||||
// InitPlan, evaluated exactly once.
|
||||
const claimed = await WIKI.db
|
||||
.delete(renderQueueTable)
|
||||
.where(
|
||||
inArray(
|
||||
eq(
|
||||
renderQueueTable.id,
|
||||
sql`(SELECT id FROM "pageRenderQueue" ORDER BY "createdAt" FOR UPDATE SKIP LOCKED LIMIT 1)`
|
||||
)
|
||||
|
||||
@@ -133,20 +133,24 @@ class RunContext {
|
||||
* `SKIP LOCKED` so that several instances can drain the outbox at once without two of them taking
|
||||
* the same event, and a claim older than the task timeout counts as abandoned: whatever held it has
|
||||
* stopped without saying so, and the cursor it left is where the next claim carries on.
|
||||
*
|
||||
* `= ANY(ARRAY(...))` and not `IN (...)`: postgres may plan an IN subquery as a semi join that re-runs
|
||||
* it per outer row, each run skipping the rows already claimed, so the LIMIT stops limiting and the
|
||||
* whole outbox is claimed at once. ARRAY() is evaluated exactly once.
|
||||
*/
|
||||
async function claimEvents(): Promise<ClaimedEvent[]> {
|
||||
const staleSeconds = (WIKI.config.scheduler?.taskTimeout ?? 300) + 60
|
||||
const result = await WIKI.db.execute(sql`
|
||||
UPDATE ${eventsTable}
|
||||
SET "claimedAt" = now(), "claimedBy" = ${WIKI.INSTANCE_ID}
|
||||
WHERE id IN (
|
||||
WHERE id = ANY(ARRAY(
|
||||
SELECT id FROM ${eventsTable}
|
||||
WHERE "processedAt" IS NULL
|
||||
AND ("claimedAt" IS NULL OR "claimedAt" < now() - make_interval(secs => ${staleSeconds}))
|
||||
ORDER BY "createdAt"
|
||||
LIMIT ${EVENT_BATCH_SIZE}
|
||||
FOR UPDATE SKIP LOCKED
|
||||
)
|
||||
))
|
||||
RETURNING id, kind, origin, "siteId", "actorId", data, recipients, cursor, "createdAt"
|
||||
`)
|
||||
return (result.rows as any[])
|
||||
|
||||
Reference in new issue
Block a user