nango/packages/tasks · na/nango

文件	最后提交记录	最后更新时间
lib	perf(orch): don't wait for locked schedule when scheduling immediate task (#6926) Mass triggering of the same sync is causing `schedules.search()` queries to pile up and wait because of the `for update` condition used to ensure only one task is active per schedule. - An unique index is now in place, coupled with the `ON CONFLICT` statement added in this commit to make postgres take care of the unique active task requirement. - `immediate()` doesn't wait for schedule row to be available if already locked and fail fast instead. - Ensures conflicting tasks and locked schedules aren't retried <!-- Describe the problem and your solution --> <!-- Issue ticket number and link (if applicable) --> <!-- Testing instructions (skip if just adding/editing providers) --> <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/NangoHQ/nango/pull/6926?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a> <!-- End of auto-generated description by cubic. -->	1 个月前
package.json	fix: upgrade dd-trace (#6905) <!-- Describe the problem and your solution --> <!-- Issue ticket number and link (if applicable) --> <!-- Testing instructions (skip if just adding/editing providers) --> <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/NangoHQ/nango/pull/6905?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a> <!-- End of auto-generated description by cubic. -->	1 个月前
tsconfig.json	feat: add generic task queue package wired into the server (#6312) Generic `task-queue` package: an abstraction layer over `@nangohq/scheduler` for defining task types and enqueuing background work, wired into the server. The goal is good DevX — declare a task type with a Zod-validated payload + handler, then `enqueue` it with full type-checking. ## DevX Instantiating and starting processor: ```ts const definitions = [reminderTask, syncTask] as const; // add new task types here export const taskQueue = new TaskQueue({ definitions, dbUrl, // postgres connection string dbSchema: 'nango_tasks', // isolated schema dbPoolMax: 10, // processorMaxConcurrency, processorPollIntervalMs, schedulerConfig, logger — all optional }); await taskQueue.migrate(); // run scheduler migrations into the schema taskQueue.start(); // start daemons + processor loop (every replica) // on shutdown: await taskQueue.stop(); ``` Defining a task payload and handler: ```ts // Simple export const reminderTask = defineTask({ type: 'reminder', schema: z.object({ userId: z.string(), message: z.string() }), handle: async (payload, ctx) => { // payload: { userId: string; message: string } // ctx: { taskId, attempt, logger } ctx.logger.info(`reminder for ${payload.userId} (attempt ${ctx.attempt})`); return Ok(undefined); // return Err(...) to fail → auto-retried up to retryMax } }); // Override options export const syncTask = defineTask({ type: 'sync', schema: z.object({ connectionId: z.string(), model: z.string() }), groupKey: (p) => `sync:${p.connectionId}`, // 1 sync per connection, connections run in parallel groupMaxConcurrency: 1, retryMax: 5, startedToCompletedTimeoutSecs: 1800, handle: async (payload) => { await runSync(payload.connectionId, payload.model); return Ok(undefined); } }); ``` Enqueue: ```ts // run as soon as a worker picks it up await taskQueue.enqueue('reminder', { userId: 'u_1', message: 'hi' }); // run later — "in 30 days" await taskQueue.enqueue('reminder', { userId: 'u_1', message: 'trial ending' }, { startsAfter: addDays(new Date(), 30) }); // batch const res = await taskQueue.enqueueBatch([ { type: 'sync', payload: { connectionId: 'c_1', model: 'Contact' } }, { type: 'sync', payload: { connectionId: 'c_2', model: 'Contact' } }, { type: 'reminder', payload: { userId: 'u_9', message: 'hi' }, groupKey: 'reminders:vip' } ]); ``` ## Operational notes - Isolation — runs in its own Postgres schema (default `nango_tasks`) on the main Nango DB. The schema name is validated as a Postgres identifier and escaped (`??`) in migrations. - Migrations — `migrate()` runs the scheduler migrations into the schema on a dedicated connection with no statement timeout. It's also part of the manual `migrate.ts` path, so replicas started with `NANGO_MIGRATE_AT_START=false` aren't left with a broken queue. - Scheduling model — tasks are one-shot: `enqueue(...)` runs as soon as a worker is free; `{ startsAfter }` defers it. The created→started timeout is measured from `startsAfter`, so long-deferred tasks don't expire while waiting. No recurring schedules. - Concurrency — `processorMaxConcurrency` caps in-flight tasks per replica; the processor only claims as many tasks as it has free worker slots and waits on slot availability rather than polling blindly. `groupKey` + `groupMaxConcurrency` bound concurrency within a group (e.g. one sync per connection). - Retries & timeouts — a handler that returns `Err` or throws marks the task FAILED and it's retried up to `retryMax`. Tasks that exceed their timeouts are expired (and retried) by the scheduler's expiring daemon. - Resilience — the scheduler daemons self-heal: a transient error (e.g. a DB blip) is reported and the loop keeps ticking instead of dying, so a single failure can't silently leave the queue half-running. - Multi-replica — every replica runs the processor and daemons; Postgres advisory locks prevent duplicate scheduling/expiring/cleaning across replicas.	3 个月前