mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
feat(core): Converge leader takeover through the reconciliation loop (#35454)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Danny Martini
GitHub
parent
9f2f315646
commit
1801f65d85
|
|
@ -11,6 +11,9 @@ import {
|
|||
} from '../entities/workflow-publication-outbox';
|
||||
import { isUniqueConstraintError } from '../utils/is-unique-constraint-error';
|
||||
|
||||
/** Sqlite bound-variable budget per statement; safely under every build's cap. */
|
||||
const SQLITE_ENQUEUE_CHUNK_SIZE = 999;
|
||||
|
||||
@Service()
|
||||
export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPublicationOutbox> {
|
||||
constructor(
|
||||
|
|
@ -92,75 +95,6 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a pending publication record for the current version of each active workflow
|
||||
* with at least one in-memory trigger.
|
||||
*
|
||||
* These must be re-published on startup or leader handoff to ensure that all
|
||||
* of the in-memory state is correctly initialized. For workflows with no in-memory triggers,
|
||||
* we can skip re-publishing because their triggers are already persisted.
|
||||
*
|
||||
* NOTE: we only exclude workflows where we KNOW that there are no in-memory triggers.
|
||||
* If the per-trigger data is missing (e.g. due to a crash), we will still enqueue the
|
||||
* workflow for publication, which is safe because the publication process is idempotent.
|
||||
*/
|
||||
async enqueueForLeaderHandoff(): Promise<void> {
|
||||
if (this.globalConfig.database.type === 'postgresdb') {
|
||||
await this.enqueueForLeaderHandoffWithPostgresUpsert();
|
||||
return;
|
||||
}
|
||||
|
||||
await this.enqueueForLeaderHandoffWithSqliteUpsert();
|
||||
}
|
||||
|
||||
private async enqueueForLeaderHandoffWithPostgresUpsert(): Promise<void> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
const workflowTableName = this.getTableName('workflow_entity');
|
||||
const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status');
|
||||
|
||||
await this.query(
|
||||
`INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status")
|
||||
SELECT w."id", w."activeVersionId", '${Status.Pending}'
|
||||
FROM ${workflowTableName} w
|
||||
WHERE w."activeVersionId" IS NOT NULL AND w."isArchived" = false
|
||||
AND (
|
||||
-- Enqueue workflows with at least one in-memory trigger
|
||||
EXISTS (
|
||||
SELECT 1 FROM ${triggerStatusTableName} ts
|
||||
WHERE ts."workflowId" = w."id" AND ts."triggerKind" = 'in-memory'
|
||||
)
|
||||
-- Enqueue workflows where we have no trigger status data
|
||||
OR NOT EXISTS (SELECT 1 FROM ${triggerStatusTableName} ts WHERE ts."workflowId" = w."id")
|
||||
)
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO UPDATE SET "publishedVersionId" = EXCLUDED."publishedVersionId", "updatedAt" = CURRENT_TIMESTAMP(3)`,
|
||||
);
|
||||
}
|
||||
|
||||
private async enqueueForLeaderHandoffWithSqliteUpsert(): Promise<void> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
const workflowTableName = this.getTableName('workflow_entity');
|
||||
const triggerStatusTableName = this.getTableName('workflow_publication_trigger_status');
|
||||
|
||||
await this.query(
|
||||
`INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status")
|
||||
SELECT w."id", w."activeVersionId", '${Status.Pending}'
|
||||
FROM ${workflowTableName} w
|
||||
WHERE w."activeVersionId" IS NOT NULL AND w."isArchived" = 0
|
||||
AND (
|
||||
-- Enqueue workflows with at least one in-memory trigger
|
||||
EXISTS (
|
||||
SELECT 1 FROM ${triggerStatusTableName} ts
|
||||
WHERE ts."workflowId" = w."id" AND ts."triggerKind" = 'in-memory'
|
||||
)
|
||||
-- Enqueue workflows where we have no trigger status data
|
||||
OR NOT EXISTS (SELECT 1 FROM ${triggerStatusTableName} ts WHERE ts."workflowId" = w."id")
|
||||
)
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO UPDATE SET "publishedVersionId" = excluded."publishedVersionId", "updatedAt" = STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW')`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a pending publication record for each given workflow that still
|
||||
* exists, in a single statement. Used by reconciliation, which must be able to
|
||||
|
|
@ -210,17 +144,25 @@ export class WorkflowPublicationOutboxRepository extends Repository<WorkflowPubl
|
|||
private async enqueueByWorkflowIdsWithSqliteUpsert(workflowIds: string[]): Promise<void> {
|
||||
const outboxTableName = this.getTableName('workflow_publication_outbox');
|
||||
const workflowTableName = this.getTableName('workflow_entity');
|
||||
const placeholders = workflowIds.map(() => '?').join(', ');
|
||||
|
||||
await this.query(
|
||||
`INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status")
|
||||
SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}'
|
||||
FROM ${workflowTableName} w
|
||||
WHERE w."id" IN (${placeholders})
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO NOTHING`,
|
||||
workflowIds,
|
||||
);
|
||||
// One placeholder is bound per id and sqlite caps bound variables per
|
||||
// statement, while a fresh leader can pass every active workflow at once.
|
||||
// Chunked upserts stay idempotent, so a failure mid-batch just leaves the
|
||||
// remainder for the next reconciliation pass.
|
||||
for (let i = 0; i < workflowIds.length; i += SQLITE_ENQUEUE_CHUNK_SIZE) {
|
||||
const chunk = workflowIds.slice(i, i + SQLITE_ENQUEUE_CHUNK_SIZE);
|
||||
const placeholders = chunk.map(() => '?').join(', ');
|
||||
|
||||
await this.query(
|
||||
`INSERT INTO ${outboxTableName} ("workflowId", "publishedVersionId", "status")
|
||||
SELECT w."id", COALESCE(w."activeVersionId", '${UNPUBLISH_VERSION_SENTINEL}'), '${Status.Pending}'
|
||||
FROM ${workflowTableName} w
|
||||
WHERE w."id" IN (${placeholders})
|
||||
ON CONFLICT ("workflowId", "status") WHERE "status" IN ('${Status.Pending}', '${Status.InProgress}')
|
||||
DO NOTHING`,
|
||||
chunk,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -417,9 +417,6 @@ export class Start extends BaseCommand<z.infer<typeof flagsSchema>> {
|
|||
|
||||
// Start to get active workflows and run their triggers
|
||||
if (this.globalConfig.workflows.useWorkflowPublicationService) {
|
||||
const { PublishedWorkflowEnqueuer } = await import(
|
||||
'@/workflows/publication/published-workflow-enqueuer.js'
|
||||
);
|
||||
const { WorkflowPublicationOutboxConsumer } = await import(
|
||||
'@/workflows/publication/workflow-publication-outbox-consumer.js'
|
||||
);
|
||||
|
|
@ -439,12 +436,6 @@ export class Start extends BaseCommand<z.infer<typeof flagsSchema>> {
|
|||
// listeners, so rewire to pick them up.
|
||||
Container.get(PubSubRegistry).init();
|
||||
|
||||
// Enqueue needs to happen before outbox consumer init, so it can activate
|
||||
// everything on the first drain
|
||||
if (this.instanceSettings.isLeader) {
|
||||
await Container.get(PublishedWorkflowEnqueuer).enqueueActiveWorkflows();
|
||||
}
|
||||
|
||||
// Don't await: the immediate drain activates every trigger and can take a
|
||||
// while, so let it run in the background instead of blocking startup.
|
||||
void Container.get(WorkflowPublicationOutboxConsumer)
|
||||
|
|
|
|||
|
|
@ -1,51 +0,0 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import type { WorkflowsConfig } from '@n8n/config';
|
||||
import type { WorkflowPublicationOutboxRepository } from '@n8n/db';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { PublishedWorkflowEnqueuer } from '@/workflows/publication/published-workflow-enqueuer';
|
||||
import type { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer';
|
||||
|
||||
describe('PublishedWorkflowEnqueuer', () => {
|
||||
const logger = mock<Logger>();
|
||||
logger.scoped.mockReturnValue(logger);
|
||||
|
||||
const outboxRepository = mock<WorkflowPublicationOutboxRepository>();
|
||||
const outboxConsumer = mock<WorkflowPublicationOutboxConsumer>();
|
||||
|
||||
let enqueuer: PublishedWorkflowEnqueuer;
|
||||
|
||||
function createEnqueuer(useWorkflowPublicationService = true) {
|
||||
const workflowsConfig = mock<WorkflowsConfig>({ useWorkflowPublicationService });
|
||||
return new PublishedWorkflowEnqueuer(logger, workflowsConfig, outboxRepository, outboxConsumer);
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
enqueuer = createEnqueuer();
|
||||
});
|
||||
|
||||
test('delegates to the outbox repository to enqueue active workflows requiring leader publication', async () => {
|
||||
await enqueuer.enqueueActiveWorkflows();
|
||||
|
||||
expect(outboxRepository.enqueueForLeaderHandoff).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
describe('enqueueOnLeaderTakeover', () => {
|
||||
test('enqueues active workflows and drains immediately when the publication service is enabled', async () => {
|
||||
await createEnqueuer(true).enqueueOnLeaderTakeover();
|
||||
|
||||
expect(outboxRepository.enqueueForLeaderHandoff).toHaveBeenCalledTimes(1);
|
||||
expect(outboxConsumer.startPolling).toHaveBeenCalledTimes(1);
|
||||
expect(outboxConsumer.drainPending).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
test('does nothing when the publication service is disabled', async () => {
|
||||
await createEnqueuer(false).enqueueOnLeaderTakeover();
|
||||
|
||||
expect(outboxRepository.enqueueForLeaderHandoff).not.toHaveBeenCalled();
|
||||
expect(outboxConsumer.startPolling).not.toHaveBeenCalled();
|
||||
expect(outboxConsumer.drainPending).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -318,70 +318,5 @@ describe('PublishedWorkflowTriggerDeactivator', () => {
|
|||
expect(logger.warn).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('interval lifecycle', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
const advanceOneInterval = async () =>
|
||||
await vi.advanceTimersByTimeAsync(RECONCILE_INTERVAL_SECONDS * 1000);
|
||||
|
||||
test('sweeps every reconcile interval once started', async () => {
|
||||
const deactivator = createDeactivator();
|
||||
deactivator.startGhostTriggerJanitor();
|
||||
|
||||
await advanceOneInterval();
|
||||
await advanceOneInterval();
|
||||
|
||||
expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).toHaveBeenCalledTimes(2);
|
||||
|
||||
deactivator.stopGhostTriggerJanitor();
|
||||
});
|
||||
|
||||
test('does not start when the publication service is disabled', async () => {
|
||||
createDeactivator(false).startGhostTriggerJanitor();
|
||||
|
||||
await advanceOneInterval();
|
||||
|
||||
expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('starting twice schedules only one interval', async () => {
|
||||
const deactivator = createDeactivator();
|
||||
deactivator.startGhostTriggerJanitor();
|
||||
deactivator.startGhostTriggerJanitor();
|
||||
|
||||
await advanceOneInterval();
|
||||
|
||||
expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).toHaveBeenCalledTimes(1);
|
||||
|
||||
deactivator.stopGhostTriggerJanitor();
|
||||
});
|
||||
|
||||
test('stops sweeping once stopped', async () => {
|
||||
const deactivator = createDeactivator();
|
||||
deactivator.startGhostTriggerJanitor();
|
||||
deactivator.stopGhostTriggerJanitor();
|
||||
|
||||
await advanceOneInterval();
|
||||
|
||||
expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test('does not restart after shutdown', async () => {
|
||||
const deactivator = createDeactivator();
|
||||
deactivator.shutdown();
|
||||
deactivator.startGhostTriggerJanitor();
|
||||
|
||||
await advanceOneInterval();
|
||||
|
||||
expect(activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ import type {
|
|||
import type { EventService } from '@/events/event.service';
|
||||
import type { NonWebhookTriggerRegistrar } from '@/workflows/triggers/non-webhook-trigger-registrar';
|
||||
|
||||
import type { PublishedWorkflowTriggerDeactivator } from '../published-workflow-trigger-deactivator';
|
||||
import type { WorkflowPublicationLifecycleLock } from '../workflow-publication-lifecycle-lock';
|
||||
import type { WorkflowPublicationOutboxConsumer } from '../workflow-publication-outbox-consumer';
|
||||
import { WorkflowPublicationReconciler } from '../workflow-publication-reconciler.service';
|
||||
|
|
@ -39,6 +40,7 @@ const eventService = mock<EventService>();
|
|||
const lifecycleLock = mock<WorkflowPublicationLifecycleLock>();
|
||||
const workflowRepository = mock<WorkflowRepository>();
|
||||
const activeWorkflowTriggers = mock<ActiveWorkflowTriggers>();
|
||||
const triggerDeactivator = mock<PublishedWorkflowTriggerDeactivator>();
|
||||
|
||||
let service: WorkflowPublicationReconciler;
|
||||
|
||||
|
|
@ -71,6 +73,7 @@ beforeEach(() => {
|
|||
workflowRepository.getActiveIds.mockResolvedValue([]);
|
||||
workflowRepository.findOneBy.mockResolvedValue(null);
|
||||
lifecycleLock.runExclusive.mockImplementation(async (_workflowId, fn) => await fn());
|
||||
triggerDeactivator.sweepGhostTriggers.mockResolvedValue(0);
|
||||
service = new WorkflowPublicationReconciler(
|
||||
logger,
|
||||
config,
|
||||
|
|
@ -85,6 +88,7 @@ beforeEach(() => {
|
|||
lifecycleLock,
|
||||
workflowRepository,
|
||||
activeWorkflowTriggers,
|
||||
triggerDeactivator,
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -109,32 +113,89 @@ describe('WorkflowPublicationReconciler', () => {
|
|||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('does not start when the instance is not leader', () => {
|
||||
it('schedules the loop on a follower too, ticking the ghost sweep instead of the detections', async () => {
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
|
||||
service.init();
|
||||
vi.advanceTimersByTime(10_000);
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalled();
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('startReconciler', () => {
|
||||
it('does not start when the publication service is disabled', () => {
|
||||
describe('loop lifecycle', () => {
|
||||
it('does not start when the publication service is disabled', async () => {
|
||||
Object.assign(config, { useWorkflowPublicationService: false });
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
|
||||
service.startReconciler();
|
||||
vi.advanceTimersByTime(10_000);
|
||||
service.init();
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled();
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not start if shutting down', () => {
|
||||
it('does not start after shutdown', async () => {
|
||||
service.shutdown();
|
||||
service.startReconciler();
|
||||
vi.advanceTimersByTime(10_000);
|
||||
service.init();
|
||||
await vi.advanceTimersByTimeAsync(10_000);
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled();
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('keeps ticking across a stepdown, flipping to the follower checks', async () => {
|
||||
service.init();
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled();
|
||||
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('needs no takeover kick: the next tick after promotion runs the leader detections', async () => {
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
service.init();
|
||||
|
||||
Object.assign(instanceSettings, { isLeader: true });
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('stops only at shutdown', async () => {
|
||||
service.init();
|
||||
await vi.advanceTimersByTimeAsync(5_000);
|
||||
|
||||
service.shutdown();
|
||||
triggerStatusRepository.findActivatedInMemoryTriggers.mockClear();
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('leader takeover', () => {
|
||||
it('kicks an immediate pass without waiting for the interval', async () => {
|
||||
service.startReconciler();
|
||||
|
||||
await service.reconcileOnLeaderTakeover();
|
||||
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('does nothing while the loop is not running', async () => {
|
||||
// The loop only runs when the publication service is enabled and the
|
||||
// instance is not shutting down; the takeover kick must not open a
|
||||
// side door around those gates.
|
||||
await service.reconcileOnLeaderTakeover();
|
||||
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -233,12 +294,28 @@ describe('WorkflowPublicationReconciler', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('does not run when the instance is no longer leader', async () => {
|
||||
it('runs the ghost sweep and none of the leader detections when not leader', async () => {
|
||||
Object.assign(instanceSettings, { isLeader: false });
|
||||
|
||||
await service.reconcile();
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).toHaveBeenCalledTimes(1);
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
expect(outboxRepository.findVersionSkewedWorkflowIds).not.toHaveBeenCalled();
|
||||
expect(outboxRepository.findTriggerStatusDriftedWorkflowIds).not.toHaveBeenCalled();
|
||||
expect(outboxRepository.findUnreportedPublishedWorkflowIds).not.toHaveBeenCalled();
|
||||
expect(outboxRepository.enqueueByWorkflowIds).not.toHaveBeenCalled();
|
||||
expect(eventService.emit).not.toHaveBeenCalledWith(
|
||||
'workflow-publication-reconciliation',
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not run the ghost sweep on the leader', async () => {
|
||||
await service.reconcile();
|
||||
|
||||
expect(triggerDeactivator.sweepGhostTriggers).not.toHaveBeenCalled();
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -328,15 +405,4 @@ describe('WorkflowPublicationReconciler', () => {
|
|||
expect(activeWorkflowTriggers.remove).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('stopReconciler', () => {
|
||||
it('stops the interval on leader stepdown', () => {
|
||||
service.startReconciler();
|
||||
service.stopReconciler();
|
||||
|
||||
vi.advanceTimersByTime(10_000);
|
||||
|
||||
expect(triggerStatusRepository.findActivatedInMemoryTriggers).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,48 +0,0 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { WorkflowsConfig } from '@n8n/config';
|
||||
import { WorkflowPublicationOutboxRepository } from '@n8n/db';
|
||||
import { OnLeaderTakeover } from '@n8n/decorators';
|
||||
import { Service } from '@n8n/di';
|
||||
|
||||
import { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer';
|
||||
|
||||
/**
|
||||
* Enqueues a publication outbox record for every active workflow, on leader
|
||||
* startup and on leader takeover (the two moments an instance becomes leader).
|
||||
*/
|
||||
@Service()
|
||||
export class PublishedWorkflowEnqueuer {
|
||||
constructor(
|
||||
private readonly logger: Logger,
|
||||
private readonly workflowsConfig: WorkflowsConfig,
|
||||
private readonly outboxRepository: WorkflowPublicationOutboxRepository,
|
||||
private readonly outboxConsumer: WorkflowPublicationOutboxConsumer,
|
||||
) {
|
||||
this.logger = this.logger.scoped('workflow-publication');
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue a publication outbox record for every active workflow that needs
|
||||
* leader-side publication at its current active version. Workflows with only
|
||||
* persisted (webhook) triggers are skipped: those live durably in
|
||||
* `webhook_entity` and are served by any main, so a new leader has no
|
||||
* in-memory state to rebuild for them. This is idempotent.
|
||||
*/
|
||||
async enqueueActiveWorkflows(): Promise<void> {
|
||||
await this.outboxRepository.enqueueForLeaderHandoff();
|
||||
this.logger.debug('Enqueued active workflow publication records');
|
||||
}
|
||||
|
||||
/**
|
||||
* On leader takeover, requeue every active workflow exactly like startup. Safe
|
||||
* because publication is idempotent and a no-op publish emits no pubsub, so
|
||||
* unchanged workflows stay quiet.
|
||||
*/
|
||||
@OnLeaderTakeover()
|
||||
async enqueueOnLeaderTakeover(): Promise<void> {
|
||||
if (!this.workflowsConfig.useWorkflowPublicationService) return;
|
||||
await this.enqueueActiveWorkflows();
|
||||
this.outboxConsumer.startPolling();
|
||||
await this.outboxConsumer.drainPending();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +1,6 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { WorkflowsConfig } from '@n8n/config';
|
||||
import { Time } from '@n8n/constants';
|
||||
import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
|
||||
import { OnLeaderStepdown, OnShutdown } from '@n8n/decorators';
|
||||
import { Service } from '@n8n/di';
|
||||
import { ActiveWorkflowTriggers, ErrorReporter, InstanceSettings } from 'n8n-core';
|
||||
|
||||
|
|
@ -19,7 +18,8 @@ const STUCK_LOCK_WARN_AFTER_PASSES = 3;
|
|||
|
||||
/**
|
||||
* Tears down in-memory triggers on leader stepdown and shutdown, and — as a
|
||||
* safeguard — periodically sweeps the registry while not leader: a trigger
|
||||
* safeguard — sweeps the registry while not leader (driven by the
|
||||
* reconciler's loop, which ticks the sweep on non-leader instances): a trigger
|
||||
* registration on a non-leader should never exist, so anything found is torn
|
||||
* down and reported.
|
||||
*
|
||||
|
|
@ -35,7 +35,6 @@ const STUCK_LOCK_WARN_AFTER_PASSES = 3;
|
|||
*/
|
||||
@Service()
|
||||
export class PublishedWorkflowTriggerDeactivator {
|
||||
private ghostTriggerJanitorInterval: NodeJS.Timeout | undefined;
|
||||
private isShuttingDown = false;
|
||||
private consecutiveLockSkipsByWorkflowId: Map<string, number> = new Map();
|
||||
|
||||
|
|
@ -72,7 +71,7 @@ export class PublishedWorkflowTriggerDeactivator {
|
|||
|
||||
const ghosts: string[] = [];
|
||||
|
||||
// Nothing may escape this method: the interval callback driving it has
|
||||
// Nothing may escape this method: the loop tick driving it runs with
|
||||
// nobody awaiting it, so an escaped rejection would crash the process.
|
||||
try {
|
||||
const workflowIds = this.activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds();
|
||||
|
|
@ -132,30 +131,8 @@ export class PublishedWorkflowTriggerDeactivator {
|
|||
return deactivatedWorkflows;
|
||||
}
|
||||
|
||||
@OnLeaderStepdown()
|
||||
startGhostTriggerJanitor(): void {
|
||||
if (!this.workflowsConfig.useWorkflowPublicationService) return;
|
||||
if (this.isShuttingDown) return;
|
||||
if (this.ghostTriggerJanitorInterval) return;
|
||||
this.consecutiveLockSkipsByWorkflowId = new Map();
|
||||
|
||||
this.ghostTriggerJanitorInterval = setInterval(
|
||||
async () => await this.sweepGhostTriggers(),
|
||||
this.workflowsConfig.publicationReconcileIntervalSeconds * Time.seconds.toMilliseconds,
|
||||
);
|
||||
}
|
||||
|
||||
@OnLeaderTakeover()
|
||||
stopGhostTriggerJanitor(): void {
|
||||
if (this.ghostTriggerJanitorInterval) {
|
||||
clearInterval(this.ghostTriggerJanitorInterval);
|
||||
this.ghostTriggerJanitorInterval = undefined;
|
||||
}
|
||||
}
|
||||
|
||||
@OnShutdown()
|
||||
shutdown(): void {
|
||||
this.isShuttingDown = true;
|
||||
this.stopGhostTriggerJanitor();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { WorkflowsConfig } from '@n8n/config';
|
||||
import { WorkflowPublicationOutbox, WorkflowPublicationOutboxRepository } from '@n8n/db';
|
||||
import { OnPubSubEvent, OnShutdown } from '@n8n/decorators';
|
||||
import { OnLeaderTakeover, OnPubSubEvent, OnShutdown } from '@n8n/decorators';
|
||||
import { Service } from '@n8n/di';
|
||||
import { ErrorReporter, InstanceSettings, SpanStatus, Tracing } from 'n8n-core';
|
||||
import { ensureError } from '@n8n/utils/errors/ensure-error';
|
||||
|
|
@ -93,6 +93,7 @@ export class WorkflowPublicationOutboxConsumer {
|
|||
* Also start polling - this is idempotent, so harmless if we're already polling.
|
||||
*/
|
||||
@OnPubSubEvent('workflow-publish-wake-up', { instanceType: 'main', instanceRole: 'leader' })
|
||||
@OnLeaderTakeover()
|
||||
async wakeUp(): Promise<void> {
|
||||
if (!this.workflowsConfig.useWorkflowPublicationService) return;
|
||||
// Direct callers (e.g. a short-lived CLI process waking its own local consumer)
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import {
|
|||
WorkflowPublicationTriggerStatusRepository,
|
||||
WorkflowRepository,
|
||||
} from '@n8n/db';
|
||||
import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
|
||||
import { OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
|
||||
import { Service } from '@n8n/di';
|
||||
import {
|
||||
ActiveWorkflowTriggers,
|
||||
|
|
@ -20,16 +20,23 @@ import type { WorkflowId } from 'n8n-workflow';
|
|||
import { EventService } from '@/events/event.service';
|
||||
import { NonWebhookTriggerRegistrar } from '@/workflows/triggers/non-webhook-trigger-registrar';
|
||||
|
||||
import { PublishedWorkflowTriggerDeactivator } from './published-workflow-trigger-deactivator';
|
||||
import { WorkflowPublicationLifecycleLock } from './workflow-publication-lifecycle-lock';
|
||||
import { WorkflowPublicationOutboxConsumer } from './workflow-publication-outbox-consumer';
|
||||
|
||||
/**
|
||||
* Periodically checks, on the leader, that every trigger which should be active
|
||||
* in memory really is, and re-publishes any workflow whose triggers went
|
||||
* missing. This recovers from the leader-transition edge case where a demoted
|
||||
* instance consumed a workflow's outbox record (marking it completed) and then
|
||||
* tore its triggers down, so the new leader never (re)activated them and no
|
||||
* pending record remains to reprocess.
|
||||
* The one reconciliation loop of the workflow publication service. It ticks on
|
||||
* every main instance for the whole process lifetime; what a tick does is
|
||||
* decided by the instance's role at that moment, so leadership transitions
|
||||
* never start or stop anything — only shutdown stops the loop.
|
||||
*
|
||||
* Leader ticks verify that the running state matches the published state —
|
||||
* every trigger that should be active in memory really is, and the publication
|
||||
* tables agree with each other — and heal whatever diverged. Causes are
|
||||
* deliberately not enumerated here: a known cause is a bug to fix at its
|
||||
* source; this loop exists to detect and repair what an unknown one leaves
|
||||
* behind. Non-leader ticks instead sweep the registry for ghost registrations
|
||||
* (see {@link PublishedWorkflowTriggerDeactivator}).
|
||||
*
|
||||
* Detection is scoped to non-webhook (poll/trigger) nodes, since those are the
|
||||
* ones held in the in-memory registry; webhook triggers live in the
|
||||
|
|
@ -37,16 +44,16 @@ import { WorkflowPublicationOutboxConsumer } from './workflow-publication-outbox
|
|||
* re-published. Recovery only enqueues through the outbox and wakes the consumer;
|
||||
* the applier does the actual re-registration, so this adds no activation logic.
|
||||
*
|
||||
* A third detection compares versions rather than trigger node ids: a
|
||||
* `workflow_published_version` mapping that disagrees with the workflow's
|
||||
* `activeVersionId` (a rolled-back or lost publication, or a missed unpublish)
|
||||
* is invisible to the node-id diffs when the trigger set is unchanged, so it is
|
||||
* detected in the database directly and healed through the same enqueue path.
|
||||
* Two further database detections complete the coverage: trigger-status rows
|
||||
* recorded for a version other than the active one (the mapping may already be
|
||||
* correct while the rows lag), and published workflows with no trigger-status
|
||||
* rows at all (published before the service was enabled, or a publication that
|
||||
* failed before reporting).
|
||||
* The registry checks compare the trigger-status rows against what is
|
||||
* registered in memory — and both were written by the last publication that
|
||||
* was applied. They can tell when memory drifted from what was applied, but
|
||||
* not whether what was applied is what the workflow's `activeVersionId` says
|
||||
* it should be. Three database-only detections close that gap: a
|
||||
* `workflow_published_version` mapping that disagrees with `activeVersionId`,
|
||||
* trigger-status rows recorded for a version other than the active one, and
|
||||
* published workflows with no trigger-status rows at all (also the expected
|
||||
* state for anything published before the service was enabled). All heal
|
||||
* through the same enqueue path.
|
||||
*/
|
||||
@Service()
|
||||
export class WorkflowPublicationReconciler {
|
||||
|
|
@ -70,13 +77,12 @@ export class WorkflowPublicationReconciler {
|
|||
private readonly lifecycleLock: WorkflowPublicationLifecycleLock,
|
||||
private readonly workflowRepository: WorkflowRepository,
|
||||
private readonly activeWorkflowTriggers: ActiveWorkflowTriggers,
|
||||
private readonly publishedWorkflowTriggerDeactivator: PublishedWorkflowTriggerDeactivator,
|
||||
) {
|
||||
this.logger = logger.scoped('workflow-publication');
|
||||
}
|
||||
|
||||
init() {
|
||||
if (!this.instanceSettings.isLeader) return;
|
||||
|
||||
this.startReconciler();
|
||||
|
||||
// Run an initial pass at startup rather than waiting a full interval, so a
|
||||
|
|
@ -85,12 +91,10 @@ export class WorkflowPublicationReconciler {
|
|||
}
|
||||
|
||||
/**
|
||||
* Deliberately no immediate pass on takeover: `PublishedWorkflowEnqueuer`
|
||||
* already re-enqueues every active workflow at that moment, and the
|
||||
* consumed-record race this loop recovers can only materialize after that
|
||||
* enqueue — the first interval pass catches it.
|
||||
* Starts the process-lifetime tick interval, on any role. Takeover never
|
||||
* starts or stops anything here — the loop is already ticking — but it does
|
||||
* kick an immediate pass (see {@link reconcileOnLeaderTakeover}).
|
||||
*/
|
||||
@OnLeaderTakeover()
|
||||
startReconciler() {
|
||||
if (!this.workflowsConfig.useWorkflowPublicationService) return;
|
||||
if (this.isShuttingDown || this.reconcileInterval) return;
|
||||
|
|
@ -104,26 +108,28 @@ export class WorkflowPublicationReconciler {
|
|||
this.logger.debug(`Trigger reconciliation scheduled every ${intervalSeconds}s`);
|
||||
}
|
||||
|
||||
@OnLeaderStepdown()
|
||||
stopReconciler() {
|
||||
@OnShutdown()
|
||||
shutdown() {
|
||||
this.isShuttingDown = true;
|
||||
clearInterval(this.reconcileInterval);
|
||||
this.reconcileInterval = undefined;
|
||||
}
|
||||
|
||||
@OnShutdown()
|
||||
shutdown() {
|
||||
this.isShuttingDown = true;
|
||||
this.stopReconciler();
|
||||
}
|
||||
|
||||
/**
|
||||
* One reconciliation pass: diff the triggers that should be active in memory
|
||||
* against what is registered, and re-publish any workflow with a missing
|
||||
* trigger. Errors are caught and reported so a transient failure never escapes
|
||||
* the interval; the next pass retries. Runs only on the leader.
|
||||
* One tick of the loop, gated by the instance's role at this moment. On the
|
||||
* leader: diff the triggers that should be active in memory against what is
|
||||
* registered, and re-publish any workflow with a missing trigger. On any
|
||||
* other role: sweep ghost registrations instead. Errors are caught and
|
||||
* reported so a transient failure never escapes the interval; the next pass
|
||||
* retries.
|
||||
*/
|
||||
async reconcile(): Promise<void> {
|
||||
if (!this.instanceSettings.isLeader || this.isShuttingDown) return;
|
||||
if (this.isShuttingDown) return;
|
||||
|
||||
if (!this.instanceSettings.isLeader) {
|
||||
await this.publishedWorkflowTriggerDeactivator.sweepGhostTriggers();
|
||||
return;
|
||||
}
|
||||
|
||||
await this.tracing.startSpan(
|
||||
{ name: 'Publication trigger reconciliation', op: 'publication.reconcile' },
|
||||
|
|
@ -185,6 +191,21 @@ export class WorkflowPublicationReconciler {
|
|||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* A fresh leader must converge now, not an interval later: with an empty
|
||||
* registry, this pass re-detects every workflow that needs its triggers
|
||||
* back. Pending records left by the previous leader are excluded from the
|
||||
* detections and drained by the consumer's own takeover hook instead. Gated
|
||||
* on the loop running, so a disabled service or shutdown keeps takeover
|
||||
* inert.
|
||||
*/
|
||||
@OnLeaderTakeover()
|
||||
async reconcileOnLeaderTakeover(): Promise<void> {
|
||||
if (!this.reconcileInterval) return;
|
||||
|
||||
await this.reconcile();
|
||||
}
|
||||
|
||||
private async findSurplusWorkflowIds(): Promise<WorkflowId[]> {
|
||||
const registered = this.activeWorkflowTriggers.getNonWebhookTriggerWorkflowIds();
|
||||
if (registered.length === 0) return [];
|
||||
|
|
|
|||
|
|
@ -91,7 +91,7 @@ afterAll(async () => {
|
|||
);
|
||||
Container.get(WorkflowPublicationOutboxConsumer).stopPolling();
|
||||
Container.get(WorkflowPublicationOutboxCleanupService).stopCleanup();
|
||||
Container.get(WorkflowPublicationReconciler).stopReconciler();
|
||||
Container.get(WorkflowPublicationReconciler).shutdown();
|
||||
await Container.get(ActiveWorkflowManager).removeAll();
|
||||
await testDb.terminate();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,23 +1,16 @@
|
|||
import { createActiveWorkflow, createWorkflow, testDb } from '@n8n/backend-test-utils';
|
||||
import { createActiveWorkflow, createWorkflow, newWorkflow, testDb } from '@n8n/backend-test-utils';
|
||||
import { WorkflowsConfig } from '@n8n/config';
|
||||
import { UNPUBLISH_VERSION_SENTINEL } from '@n8n/db';
|
||||
import type { WorkflowPublicationTriggerKind } from '@n8n/db';
|
||||
import {
|
||||
WorkflowPublicationOutboxRepository,
|
||||
WorkflowPublicationTriggerStatusRepository,
|
||||
WorkflowRepository,
|
||||
} from '@n8n/db';
|
||||
import { WorkflowPublicationOutboxRepository, WorkflowRepository } from '@n8n/db';
|
||||
import { Container } from '@n8n/di';
|
||||
import assert from 'node:assert';
|
||||
|
||||
describe('WorkflowPublicationOutboxRepository', () => {
|
||||
let repository: WorkflowPublicationOutboxRepository;
|
||||
let triggerStatusRepository: WorkflowPublicationTriggerStatusRepository;
|
||||
|
||||
beforeAll(async () => {
|
||||
await testDb.init();
|
||||
repository = Container.get(WorkflowPublicationOutboxRepository);
|
||||
triggerStatusRepository = Container.get(WorkflowPublicationTriggerStatusRepository);
|
||||
});
|
||||
|
||||
beforeEach(async () => {
|
||||
|
|
@ -469,134 +462,24 @@ describe('WorkflowPublicationOutboxRepository', () => {
|
|||
|
||||
expect(await repository.find({ where: { status: 'pending' } })).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('enqueueForLeaderHandoff', () => {
|
||||
beforeEach(async () => {
|
||||
await testDb.truncate([
|
||||
'WorkflowPublicationTriggerStatus',
|
||||
'WorkflowDependency',
|
||||
'WorkflowEntity',
|
||||
'WorkflowHistory',
|
||||
'WorkflowPublishHistory',
|
||||
]);
|
||||
});
|
||||
it('enqueues every workflow in a batch larger than one statement chunk', async () => {
|
||||
// A fresh leader detects every active workflow as missing at once, so
|
||||
// this call sees fleet-sized input. Sqlite binds one placeholder per id
|
||||
// and caps bound variables per statement, so large batches are split
|
||||
// into chunks; a boundary bug here would silently drop workflows.
|
||||
const workflowIds = Array.from({ length: 1100 }, (_, i) => `bulk-wf-${i}`);
|
||||
// Seed in slices: a single multi-row VALUES of this size exceeds sqlite's
|
||||
// expression-tree depth, which is not the limit under test here.
|
||||
for (let i = 0; i < workflowIds.length; i += 250) {
|
||||
await Container.get(WorkflowRepository).insert(
|
||||
workflowIds.slice(i, i + 250).map((id) => newWorkflow({ id })),
|
||||
);
|
||||
}
|
||||
|
||||
// Seed a single trigger-status row of the given kind at the workflow's active
|
||||
// version, appending to any already recorded for the workflow.
|
||||
const recordTrigger = async (
|
||||
workflow: { id: string; activeVersionId: string | null },
|
||||
nodeId: string,
|
||||
triggerKind: WorkflowPublicationTriggerKind,
|
||||
status: 'activated' | 'failed' = 'activated',
|
||||
) => {
|
||||
assert(workflow.activeVersionId);
|
||||
const existing = await triggerStatusRepository.findByWorkflowId(workflow.id);
|
||||
await triggerStatusRepository.replaceForWorkflow(workflow.id, [
|
||||
...existing.map((row) => ({
|
||||
nodeId: row.nodeId,
|
||||
versionId: row.versionId,
|
||||
status: row.status,
|
||||
triggerKind: row.triggerKind,
|
||||
errorMessage: row.errorMessage,
|
||||
})),
|
||||
{
|
||||
nodeId,
|
||||
versionId: workflow.activeVersionId,
|
||||
status,
|
||||
triggerKind,
|
||||
errorMessage: status === 'failed' ? 'boom' : null,
|
||||
},
|
||||
]);
|
||||
};
|
||||
await repository.enqueueByWorkflowIds(workflowIds);
|
||||
|
||||
const pendingWorkflowIds = async () => {
|
||||
const pending = await repository.find({ where: { status: 'pending' } });
|
||||
return pending.map((record) => record.workflowId);
|
||||
};
|
||||
|
||||
it('enqueues a workflow whose recorded triggers include an in-memory trigger', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'in-memory');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([workflow.id]);
|
||||
});
|
||||
|
||||
it('skips a workflow whose recorded triggers are all persisted', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'persisted');
|
||||
await recordTrigger(workflow, 'n2', 'persisted');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([]);
|
||||
});
|
||||
|
||||
it('enqueues a workflow that has no recorded triggers yet', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([workflow.id]);
|
||||
});
|
||||
|
||||
it('enqueues a workflow with a mix of persisted and in-memory triggers', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'persisted');
|
||||
await recordTrigger(workflow, 'n2', 'in-memory');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([workflow.id]);
|
||||
});
|
||||
|
||||
it('enqueues a workflow whose only in-memory trigger last failed to activate', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'in-memory', 'failed');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([workflow.id]);
|
||||
});
|
||||
|
||||
it('enqueues at the active version', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'in-memory');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
const pending = await repository.find({ where: { status: 'pending' } });
|
||||
expect(pending).toHaveLength(1);
|
||||
expect(pending[0].publishedVersionId).toBe(workflow.activeVersionId);
|
||||
});
|
||||
|
||||
it('skips inactive and archived workflows', async () => {
|
||||
const active = await createActiveWorkflow();
|
||||
await recordTrigger(active, 'n1', 'in-memory');
|
||||
await createWorkflow(); // inactive: no activeVersionId
|
||||
const archived = await createActiveWorkflow();
|
||||
await recordTrigger(archived, 'n1', 'in-memory');
|
||||
await Container.get(WorkflowRepository).update(archived.id, { isArchived: true });
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
expect(await pendingWorkflowIds()).toEqual([active.id]);
|
||||
});
|
||||
|
||||
it('is idempotent: re-running does not create duplicate pending records', async () => {
|
||||
const workflow = await createActiveWorkflow();
|
||||
await recordTrigger(workflow, 'n1', 'in-memory');
|
||||
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
await repository.enqueueForLeaderHandoff();
|
||||
|
||||
const pending = await repository.find({
|
||||
where: { workflowId: workflow.id, status: 'pending' },
|
||||
});
|
||||
expect(pending).toHaveLength(1);
|
||||
expect(pending[0].publishedVersionId).toBe(workflow.activeVersionId);
|
||||
expect(await repository.countBy({ status: 'pending' })).toBe(1100);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ import { ExecutionService } from '@/executions/execution.service';
|
|||
import { ExternalHooks } from '@/external-hooks';
|
||||
import { Push } from '@/push';
|
||||
import { OwnershipService } from '@/services/ownership.service';
|
||||
import { PublishedWorkflowEnqueuer } from '@/workflows/publication/published-workflow-enqueuer';
|
||||
import { PublishedWorkflowTriggerDeactivator } from '@/workflows/publication/published-workflow-trigger-deactivator';
|
||||
import { WorkflowPublicationLifecycleLock } from '@/workflows/publication/workflow-publication-lifecycle-lock';
|
||||
import { WorkflowPublicationOutboxConsumer } from '@/workflows/publication/workflow-publication-outbox-consumer';
|
||||
|
|
@ -207,28 +206,6 @@ describe('WorkflowPublicationOutboxConsumer (integration)', () => {
|
|||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
});
|
||||
|
||||
test('startup enqueue + drain registers triggers for active workflows via reconciliation', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
const trigger = scheduleNode('startup');
|
||||
const workflow = await createWorkflowWithHistory({ active: true, nodes: [trigger] }, owner);
|
||||
await setActiveVersion(workflow.id, workflow.versionId);
|
||||
await publishedVersionRepository.setPublishedVersion(workflow.id, workflow.versionId);
|
||||
|
||||
// Fresh leader startup: the workflow is active and published, but nothing is
|
||||
// registered in memory yet.
|
||||
expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined();
|
||||
|
||||
await Container.get(PublishedWorkflowEnqueuer).enqueueActiveWorkflows();
|
||||
consumer.startPolling();
|
||||
await consumer.drainPending();
|
||||
consumer.stopPolling();
|
||||
|
||||
// The reconciliation path registered the missing trigger and completed the record.
|
||||
expect(activeWorkflowTriggers.get(workflow.id)?.has(trigger.id)).toBe(true);
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
});
|
||||
|
||||
test('does no trigger work when only non-trigger content changed', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
|
|
|
|||
|
|
@ -503,7 +503,7 @@ describe('WorkflowPublicationReconciler (integration)', () => {
|
|||
);
|
||||
});
|
||||
|
||||
test('records persisted rows for a pseudo-only workflow, keeping it out of leader handoff and reconciliation', async () => {
|
||||
test('records persisted rows for a pseudo-only workflow, keeping it out of reconciliation', async () => {
|
||||
const owner = await createOwner();
|
||||
|
||||
const trigger = manualTriggerNode('pseudo');
|
||||
|
|
@ -526,14 +526,9 @@ describe('WorkflowPublicationReconciler (integration)', () => {
|
|||
triggerKind: 'persisted',
|
||||
});
|
||||
|
||||
// Leader handoff on a fresh leader: nothing is registered locally, but a
|
||||
// pseudo-only workflow has no real trigger work to republish.
|
||||
// Fresh leader: nothing is registered locally. The reconciler ignores
|
||||
// persisted rows even though the node is not registered — no re-enqueue loop.
|
||||
await activeWorkflowTriggers.remove(workflow.id);
|
||||
await outboxRepository.enqueueForLeaderHandoff();
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
|
||||
// The reconciler ignores persisted rows even though the node is not
|
||||
// registered — no re-enqueue loop.
|
||||
await reconciler.reconcile();
|
||||
expect(await outboxRepository.claimNextPendingRecord()).toBeNull();
|
||||
expect(activeWorkflowTriggers.get(workflow.id)).toBeUndefined();
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user