services/retl-runner/src/execute.ts
Rama newjitsu · réplica de lectura del repositorio original.
import { randomUUID } from "node:crypto";
import type { JsonObject, ReverseEtlContext } from "@jitsu/protocols/reverse-etl";
import type { WarehouseReader, CompositeCursor } from "@jitsu/warehouse-query";
import { ReverseRunConfig } from "@jitsu/warehouse-query/src/runtime";
import { createBufferedSyncStore, recordKey } from "@jitsu/destination-functions/src/reverse-etl/identity";
import { runReverseEtl } from "@jitsu/destination-functions/src/reverse-etl/run";
import { requiresManualReconciliation } from "@jitsu/destination-functions/src/reverse-etl/failure";
import { rejectionDetailsMessage } from "@jitsu/destination-functions/src/reverse-etl/rejection-details";
import { Database, openPersistence } from "./persistence";
import type { ControlRow } from "./persistence/rows";
import { ensure, PersistenceResetRequiredError } from "./persistence/types";
import { openMirrorPersistence, runSnapshotMirror, resumeSnapshotMirror, MirrorRunError } from "./mirror";
import type { AdapterRegistry } from "./adapters";
import type { RunLease } from "./lease";
import { Tasks, type TaskResult } from "./tasks";
import { recoverRun, type RecoveryReport } from "./recovery";
import { failureMessage, reportFailure, type FailureStage } from "./diagnostics";
import { RunProgress } from "./progress";
import { prepareSource } from "./source-preflight";
import { reverseDestinationMetadata } from "@jitsu/destination-functions/src/reverse-etl/catalog";
import { objectMappingErrors } from "@jitsu/destination-functions/src/reverse-etl/editor";
export interface ExecuteOptions {
config: ReverseRunConfig;
taskId: string;
trigger: "manual" | "scheduled" | "recovery";
recoveryOf?: string;
workerId?: string;
refreshAttempt?: number;
db: Database;
lease: RunLease;
adapters: AdapterRegistry;
controller: AbortController;
/** Fresh console admission. No cached config fallback on errors. */
admit(): Promise<ReverseRunConfig>;
reader(config: Record<string, unknown>): WarehouseReader;
heartbeatMs?: number;
}
export async function execute(input: ExecuteOptions): Promise<TaskResult> {
const { db, lease, controller } = input;
const tasks = new Tasks(
db,
input.config.id,
input.recoveryOf ?? input.taskId,
input.config.workspaceId,
input.workerId ?? input.taskId
);
const progress = new RunProgress(
message => tasks.progress(message),
stats => tasks.statistics(stats)
);
let reportProgress: (() => Promise<void>) | undefined;
let rejectedRows: (() => Promise<boolean>) | undefined;
let refreshRunId: string | undefined;
let deliveryFinished = false;
const recoveryReport: RecoveryReport = {};
let reader: WarehouseReader | undefined;
let preparedSource: Awaited<ReturnType<typeof prepareSource>> | undefined;
let started = false;
let held = false;
let timer: ReturnType<typeof setTimeout> | undefined;
let renewing: Promise<void> | undefined;
let leaseRenewal: Promise<void> | undefined;
let stopped = false;
let ownershipLost = false;
let stage: FailureStage = "lease_acquire";
const signal = controller.signal;
// Provisioning and foreground admission share renewal with the heartbeat.
// Parallel GET/PUT renewals would conflict on Kubernetes resourceVersion.
const renewLease = () =>
(leaseRenewal ??= lease.renew().finally(() => {
leaseRenewal = undefined;
}));
const tick = async () => {
try {
await renewLease();
await tasks.heartbeat();
} catch {
ownershipLost = true;
controller.abort();
}
if (!stopped && !signal.aborted)
timer = setTimeout(() => {
renewing = tick();
}, input.heartbeatMs ?? 10_000);
};
try {
signal.throwIfAborted();
await lease.acquire();
held = true;
stage = "task_start";
refreshRunId = await tasks.start(
input.trigger,
input.recoveryOf,
input.config.configRevision,
input.refreshAttempt
);
started = true;
stage = "admission";
const config = ReverseRunConfig.parse(await input.admit());
if (Object.keys(config.options.objectMappings ?? {}).length) {
const stream = reverseDestinationMetadata
.get(String(config.destination.destinationType))
?.streams.find(s => s.id === config.options.stream);
ensure(stream, "Object composition is not supported for this stream");
const errors = objectMappingErrors(stream!, config.options);
ensure(!errors.length, errors.join(" "));
}
// Automatic checks are only queued with enabled snapshots. A pause after
// queuing must still stop them; explicit paused refresh gets a paused snapshot.
ensure(
!config.options.disabled || (input.config.options.disabled && input.trigger === "recovery" && refreshRunId),
"Paused sync only permits saved delivery status refresh"
);
stage = "execution";
ensure(
config.id === input.config.id &&
config.workspaceId === input.config.workspaceId &&
config.configRevision === input.config.configRevision,
"Reverse configuration changed before admission"
);
// The object storage was chosen from the startup configuration, and retention is deliberately outside the revision
// hash, so a retention bucket enabled, disabled or changed since then would be accepted here yet written with the old
// storage choice. Stop before anything is constructed or sent; the next run starts with the new setting.
ensure(
config.retention?.bucket === input.config.retention?.bucket,
"Reverse ETL retention configuration changed before admission"
);
// A mirror run seals its whole snapshot, with the projected provider data, in the main store before its first batch,
// so the per-batch guard on projections would come too late. Retention is for upsert streams only: refuse it up front.
ensure(!(config.retention && config.options.mode === "mirror"), "Retention is not supported for mirror syncs");
signal.throwIfAborted();
const bind = input.adapters.get(String(config.destination.destinationType));
ensure(bind, "Reverse destination is not enabled in this runner");
// Audience provisioning can outlive a lease period; supervise it just like delivery.
timer = setTimeout(() => {
renewing = tick();
}, input.heartbeatMs ?? 10_000);
const adapter = await bind(config, {
db,
signal,
log: message => progress.log(message),
validateSource: async validation => {
ensure(!preparedSource && input.trigger !== "recovery", "Cannot extract source during status refresh");
await progress.log(
"Validating warehouse snapshot before creating an audience; no audience changes submitted yet."
);
preparedSource = await prepareSource(config, validation, input.reader(config.warehouse), signal, db.limits);
await progress.log(`Validated ${preparedSource.count} source rows; using this same snapshot for delivery.`);
},
});
ensure(adapter.stream.name === config.options.stream, "Reverse stream does not match adapter");
if (config.options.mode === "mirror")
ensure(
adapter.mirror && adapter.mirror.stream === adapter.stream && adapter.verifyMirrorBaseline,
"Mirror adapter not verified"
);
const logicalRunId =
refreshRunId ??
(await db.transaction(async client => {
const result = await client.query<Pick<ControlRow, "run_id" | "phase">>(
"SELECT run_id,phase FROM reverse_sync_control WHERE workspace_id=$1 AND sync_id=$2 AND NOT detached AND phase NOT IN ('complete','aborted') ORDER BY run_order DESC LIMIT 1",
[config.workspaceId, config.id]
);
const previous = result.rows[0];
return previous && !["complete", "aborted"].includes(previous.phase) ? previous.run_id : randomUUID();
}));
// Recheck the Kubernetes lease before opening persistence. Durable state still
// rejects changed revision/target/mode, but does not authorize worker ownership.
await renewLease();
signal.throwIfAborted();
const runInput = {
workspaceId: config.workspaceId,
syncId: config.id,
taskId: tasks.taskId,
logicalRunId,
targetIdentity: adapter.targetIdentity,
configRevision: config.configRevision,
mode: config.options.mode,
extraction: config.model.cursor ? ("cursor" as const) : ("full" as const),
insertOnly: adapter.insertOnly,
};
const mirror = config.options.mode === "mirror";
// Restoring large object-backed baselines must not outlive worker ownership.
const run = mirror
? await openMirrorPersistence(db, runInput)
: await openPersistence(db, runInput, adapter.project);
const scope = run.scope;
run.core.onPublish = head => progress.observe(head);
reportProgress = () => progress.summarize(run.core.head);
rejectedRows = () => run.core.hasRejected();
ensure(input.trigger !== "recovery" || run.recovery, "Recovery must not start a fresh extraction");
await tasks.heartbeat();
const saved = await run.core.state();
await tasks.progress(
run.recovery
? "Refreshing submitted Reverse ETL delivery status"
: mirror
? "Extracting warehouse snapshot; no audience changes submitted yet"
: "Starting Reverse ETL extraction and delivery"
);
await progress.log(
`${
run.recovery ? "Continuing saved delivery; warehouse SQL is not re-read. " : ""
}Delivery counts describe records the destination's API accepted, not downstream results such as audience matching or attribution. Submission is not acceptance; the destination may report its own metrics separately.`
);
await progress.observe(run.core.head, true);
const context: ReverseEtlContext<JsonObject, JsonObject> = {
...run.scope,
mode: config.options.mode,
fullRefresh: !config.model.cursor,
credentials: adapter.credentials,
options: adapter.options ?? (config.options.streamOptions as JsonObject),
signal,
store: createBufferedSyncStore(saved.store),
delivery: run.delivery,
// Provider messages can contain identifiers/tokens. Persist core-owned
// lifecycle messages only until a structured redacted logging contract exists.
log: { info() {}, warn() {}, debug() {}, error() {} },
fetch: (url, opts) => fetch(url, { ...opts, signal: AbortSignal.any([signal, AbortSignal.timeout(30_000)]) }),
};
const source = async function* (after: CompositeCursor | undefined, sourceSignal: AbortSignal) {
sourceSignal.throwIfAborted();
if (preparedSource) {
ensure(!after, "Validated mirror source cannot resume from a cursor");
yield* preparedSource.rows(sourceSignal);
return;
}
reader = input.reader(config.warehouse);
let rows = 0,
deleted = 0,
previouslySubmitted = 0,
complete = false;
try {
for await (const row of reader.stream(config.model, after, sourceSignal)) {
rows++;
if (row.deleted) deleted++;
const key = recordKey(config.model.primaryKey.map(column => row.row[column] as string | number | boolean));
if (adapter.insertOnly && run.core.hasEventKey(key)) {
previouslySubmitted++;
continue;
}
yield {
...row,
key,
};
}
complete = true;
} finally {
if (adapter.insertOnly)
await progress.log(
`Event deduplication: ${rows} source rows, ${previouslySubmitted} previously submitted events omitted, ${
rows - previouslySubmitted
} new events.`
);
if (!mirror)
await progress.log(
`Source extraction ${complete ? "complete" : "interrupted"}: ${rows} rows read in this attempt, ${
rows - deleted
} upsert rows, ${deleted} explicit removal rows. Source rows are not confirmed deliveries.`
);
}
};
const hooks = adapter.recovery?.(saved.providerState);
let result: string;
if (mirror) {
const targetBaseline = await adapter.verifyMirrorBaseline!(signal);
const options = {
persistence: run as Awaited<ReturnType<typeof openMirrorPersistence>>,
adapter: adapter.mirror!,
context,
targetBaseline,
};
const phase = (await run.core.recoveryStatus()).phase;
const rejected = run.recovery && (await run.core.hasRejected());
if (run.recovery && phase !== "abort_prepared" && !rejected && (await run.snapshots.status())?.sealed) {
// Local finish recovery needs no provider hooks; attachment is fail-closed.
result = (
await resumeSnapshotMirror(
options,
hooks ?? {
attachWriter: async () => {
throw new Error("Provider reconciliation required");
},
}
)
).delivery;
} else if (run.recovery)
result = await recoverRun(
run,
context,
hooks,
adapter.mirror!.batchDelivery === "asynchronous",
recoveryReport
);
else
result = (
await runSnapshotMirror({
...options,
mapping: config.options.mapping,
objectMappings: config.options.objectMappings,
source: sig => source(undefined, sig),
onSnapshotProgress: progress =>
tasks.progress(
progress.sealed
? `Snapshot complete: ${progress.sourceRows} source rows, ${progress.projectedMembers} projected audience members, ${progress.uniqueMembers} unique audience members, ${progress.duplicatesCollapsed} duplicates collapsed. Comparing audience membership and submitting changes.`
: `Extracted ${progress.sourceRows} source rows into snapshot; no audience changes submitted yet`
),
})
).delivery;
} else if (run.recovery)
result = await recoverRun(run, context, hooks, adapter.stream.batchDelivery === "asynchronous", recoveryReport);
else
result = (
await runReverseEtl({
stream: { ...adapter.stream, batchSize: Math.min(adapter.stream.batchSize, db.limits.batchRecords) },
context,
mapping: config.options.mapping,
objectMappings: config.options.objectMappings,
sourceKeyField: adapter.insertOnly ? "__sourceKey" : undefined,
source,
checkpointEvery: config.options.checkpointEvery,
})
).delivery;
// A committed delivery cannot be polled again; subsequent bookkeeping errors
// must not restore PENDING against a terminal persistence control.
deliveryFinished = result !== "pending";
// Stop renewal before terminal task status; never let a late heartbeat read
// our own SUCCESS as cancellation. Still hold the Kubernetes lease until finally.
stopped = true;
clearTimeout(timer);
await renewing;
signal.throwIfAborted();
await reportProgress();
if (result === "pending") {
// Interrupted legacy runs remain recoverable but cannot admit overlapping extraction.
await run.core.detach();
return await tasks.wait(logicalRunId, config.configRevision);
}
const success = result === "accepted";
const changed = await tasks.finish(
success ? "COMPLETE" : "FAILED",
success
? "Reverse ETL delivery committed"
: `Recovery completed cleanup; next run will restart extraction${
recoveryReport.rejectionCode
? `. The destination rejected a row. Reason code: ${recoveryReport.rejectionCode}.`
: ""
}${rejectionDetailsMessage(recoveryReport.rejectionDetails)}`
);
return changed && success ? "COMPLETE" : "FAILED";
} catch (error) {
reportFailure(stage, error);
if (!started && stage === "lease_acquire")
await tasks
.admissionFailed(
input.trigger,
`Worker could not acquire the sync lease. ${failureMessage(error, tasks.taskId)}`
)
.catch(() => undefined);
stopped = true;
clearTimeout(timer);
await renewing;
const status = signal.aborted && !ownershipLost ? "CANCELLED" : "FAILED";
await reportProgress?.();
// A final provider result is delivery evidence, not a failed status lookup.
// Retain its binding for explicit continuation, but stop automatic polling.
if (started && requiresManualReconciliation(error)) {
await tasks.finish(status, failureMessage(error, tasks.taskId)).catch(() => undefined);
return status;
}
if (started && refreshRunId && !deliveryFinished && !progress.deliveryChanged) {
return await tasks.refreshFailed(
`Status refresh failed without updating delivery outcomes. ${failureMessage(error, tasks.taskId)}`,
error instanceof PersistenceResetRequiredError
);
}
if (
started &&
refreshRunId &&
!deliveryFinished &&
!signal.aborted &&
!ownershipLost &&
!(error instanceof PersistenceResetRequiredError) &&
!(await rejectedRows?.().catch(() => true))
) {
// Polling errors do not discard a submitted run or create a new task. A
// permanent row rejection still fails immediately; the polling deadline remains bounded.
const result = await tasks.wait(refreshRunId, input.config.configRevision);
await tasks.logError(`Status refresh could not complete. ${failureMessage(error, tasks.taskId)}`);
return result;
}
if (started)
await tasks
.finish(
status,
ownershipLost
? "The worker lost ownership or could not report progress. Contact support or your Jitsu administrator; do not reset sync state."
: signal.aborted
? "Reverse ETL cancelled; unresolved delivery retained"
: error instanceof MirrorRunError || error instanceof PersistenceResetRequiredError
? error.message
: failureMessage(error, input.taskId)
)
.catch(() => undefined);
return status;
} finally {
stopped = true;
clearTimeout(timer);
await renewing;
await reader?.close().catch(() => undefined);
await preparedSource?.close().catch(() => undefined);
if (held) await lease.release().catch(() => undefined);
}
}