services/retl-runner/src/progress.ts
Rama newjitsu · réplica de lectura del repositorio original.
import type { ArtifactHead, BatchHead } from "./artifacts/state";
import type { ReverseBatchCounts, ReverseDeliveryStats, ReverseRecordCounts } from "@jitsu/protocols/reverse-etl-stats";
export function batchStatistics(head: ArtifactHead): ReverseDeliveryStats {
const empty = (): ReverseBatchCounts => ({
total: 0,
prepared: 0,
unconfirmed: 0,
pending: 0,
accepted: 0,
rejected: 0,
partial: 0,
cancelled: 0,
});
const emptyRecords = (): ReverseRecordCounts => ({
total: 0,
prepared: 0,
unconfirmed: 0,
pending: 0,
accepted: 0,
rejected: 0,
cancelled: 0,
});
const recordCounts = { upsert: emptyRecords(), remove: emptyRecords() };
const stats: ReverseDeliveryStats = {
version: 1,
runId: head.runId,
observedAt: new Date().toISOString(),
upsert: empty(),
remove: empty(),
records: { accepted: 0, pending: 0, rejected: 0 },
recordCounts,
...(head.snapshot?.sealed && head.snapshot.summary
? {
snapshot: {
sourceRows: head.snapshot.keys,
uniqueMembers: head.snapshot.summary.uniqueMembers,
...(head.snapshot.summary.projectedMembers === undefined
? {}
: { projectedMembers: head.snapshot.summary.projectedMembers }),
...(head.snapshot.summary.excludedRows === undefined
? {}
: { excludedRows: head.snapshot.summary.excludedRows }),
},
}
: {}),
...(head.snapshot?.strategy === "native-replace"
? { replacement: head.snapshot.replacementStatus ?? "not_started" }
: {}),
};
for (const batch of head.batches) {
const counts = stats[batch.action];
const size = batch.last - batch.first + 1;
const status =
batch.status === "prepared"
? "prepared"
: batch.status === "unknown"
? "unconfirmed"
: batch.status === "cancelled"
? "cancelled"
: batch.staged > 0
? "pending"
: batch.accepted === size
? "accepted"
: batch.rejected === size
? "rejected"
: "partial";
counts.total++;
counts[status]++;
const records = recordCounts[batch.action];
records.total += size;
records.accepted += batch.accepted;
records.rejected += batch.rejected;
// Keep known outcomes even when the rest of a batch is cancelled or unconfirmed.
const remaining = size - batch.accepted - batch.rejected;
if (batch.status === "prepared") records.prepared += remaining;
else if (batch.status === "unknown") records.unconfirmed += remaining;
else if (batch.status === "cancelled") records.cancelled += remaining;
else records.pending += batch.staged;
stats.records.accepted += batch.accepted;
stats.records.pending += batch.staged;
stats.records.rejected += batch.rejected;
}
return stats;
}
export function deliveryTotals(head: ArtifactHead, action: "upsert" | "remove") {
const totals = {
batches: 0,
submittedBatches: 0,
records: 0,
submitted: 0,
accepted: 0,
pending: 0,
rejected: 0,
unconfirmed: 0,
cancelled: 0,
};
for (const batch of head.batches) {
if (batch.action !== action) continue;
const records = batch.last - batch.first + 1;
const submitted = batch.submittedRecords ?? batch.accepted + batch.staged;
totals.batches++;
totals.records += records;
totals.submitted += submitted;
if (submitted) totals.submittedBatches++;
totals.accepted += batch.accepted;
totals.pending += batch.staged;
totals.rejected += batch.rejected;
if (batch.status === "prepared" || batch.status === "unknown") totals.unconfirmed += records;
if (batch.status === "cancelled") totals.cancelled += records - batch.accepted - batch.rejected;
}
return totals;
}
/** Only aggregate core-owned counts. No payloads, identifiers, cursors or provider messages. */
export class RunProgress {
/** Changes to delivery evidence, not log messages or refreshed observation timestamps. */
deliveryChanged = false;
private deliveryBaseline?: string;
private lastStats = "";
private lastPlan = "";
private summarized = false;
private lastReplacementStatus?: string;
private readonly batches = new Map<
string,
Pick<BatchHead, "status" | "accepted" | "staged" | "rejected"> & { submitted: number; restored: boolean }
>();
constructor(
private readonly write: (message: string) => Promise<void>,
private readonly writeStats?: (stats: ReverseDeliveryStats) => Promise<void>
) {}
private async statistics(head: ArtifactHead) {
if (!this.writeStats) return;
const stats = batchStatistics(head);
const { observedAt, ...value } = stats;
const key = JSON.stringify(value);
if (key === this.lastStats) return;
try {
await this.writeStats(stats);
this.lastStats = key;
} catch {
process.stderr.write('{"event":"reverse_etl_statistics_unavailable"}\n');
}
}
async log(message: string) {
try {
await this.write(message);
return true;
} catch {
// Observability failure must never change delivery/recovery decisions.
process.stderr.write('{"event":"reverse_etl_progress_unavailable"}\n');
return false;
}
}
/** Restored batches seed progress without presenting earlier submissions as new uploads. */
async observe(head: ArtifactHead, restored = false) {
const { observedAt, snapshot: _snapshot, ...delivery } = batchStatistics(head);
const fingerprint = JSON.stringify(delivery);
if (!restored && this.deliveryBaseline !== undefined && fingerprint !== this.deliveryBaseline)
this.deliveryChanged = true;
this.deliveryBaseline = fingerprint;
await this.statistics(head);
const snapshot = head.snapshot;
if (snapshot?.sealed) {
const p = snapshot.summary;
const plan =
p && snapshot.strategy === "native-replace"
? `Full audience replacement: ${snapshot.keys} source rows, ${p.uniqueMembers} unique members${
p.projectedMembers === undefined
? ""
: `, ${p.projectedMembers} projected members, ${
p.projectedMembers - p.uniqueMembers
} duplicates collapsed, ${p.excludedRows ?? 0} source rows excluded by projection`
}. All ${
p.uniqueMembers
} members will be uploaded, including unchanged members. Older audience membership is cleaned up only after every upload is accepted.`
: p
? `Mirror comparison (entire logical run): ${p.baselineMembers} baseline members; ${p.newMembers} new, ${
p.changedMembers
} changed, ${
p.unconfirmedRefreshMembers === undefined
? `${p.refreshMembers} unchanged requiring refresh (reason unavailable in this saved snapshot)`
: `${p.refreshMembers - p.unconfirmedRefreshMembers} unchanged due for expiry refresh, ${
p.unconfirmedRefreshMembers
} unconfirmed requiring refresh`
}, ${p.unchangedMembers} unchanged skipped, ${p.removals} to remove. Snapshot: ${
snapshot.keys
} source rows, ${p.uniqueMembers} unique members` +
(p.projectedMembers === undefined
? "; original duplicate count unavailable"
: `, ${p.projectedMembers} projected members, ${
p.projectedMembers - p.uniqueMembers
} duplicates collapsed, ${p.excludedRows ?? 0} source rows excluded by projection`) +
"."
: "Original mirror comparison counts are unavailable for this older run; saved delivery totals remain available.";
if (plan !== this.lastPlan && (await this.log(plan))) this.lastPlan = plan;
}
const replacementStatus = snapshot?.replacementStatus;
if (restored) this.lastReplacementStatus = replacementStatus;
else if (replacementStatus && replacementStatus !== this.lastReplacementStatus) {
const message = {
prepared: "Submitting full-audience cleanup; all snapshot uploads are accepted.",
pending: "Google is processing full-audience cleanup; the next status check will poll the saved request.",
accepted: "Full-audience cleanup accepted. Google does not report the number of members removed.",
}[replacementStatus];
if (await this.log(message)) this.lastReplacementStatus = replacementStatus;
}
for (const batch of head.batches) {
const previous = this.batches.get(batch.id);
const current = {
status: batch.status,
accepted: batch.accepted,
staged: batch.staged,
rejected: batch.rejected,
submitted: batch.submittedRecords ?? batch.accepted + batch.staged,
restored: restored || previous?.restored === true,
};
if (restored) {
this.batches.set(batch.id, current);
continue;
}
if (JSON.stringify(previous) === JSON.stringify(current)) continue;
const records = batch.last - batch.first + 1;
const action = batch.action === "upsert" ? "additions/upserts" : "removals";
let message: string;
if (batch.status === "prepared") message = `Preparing ${records} ${action}.`;
else if (batch.status === "unknown")
message = `Confirmation missing for ${records} ${action}; check destination status before retrying.`;
else if (batch.status === "cancelled")
message = `Cancelled ${records - batch.accepted - batch.rejected} ${action}; ${
batch.accepted
} previously accepted, ${batch.rejected} rejected in this batch.`;
else {
const submitted = current.submitted - (previous?.submitted ?? 0);
message =
(submitted > 0 && !current.restored
? `Submitted ${submitted} ${action}; `
: `Status updated for ${records} ${action}: `) +
`${batch.accepted} accepted, ${batch.staged} pending processing, ${batch.rejected} rejected in this batch.`;
}
if (await this.log(message)) this.batches.set(batch.id, current);
}
}
/** One cumulative summary at the attempt boundary, including earlier attempts' delivery. */
async summarize(head: ArtifactHead) {
await this.statistics(head);
if (this.summarized) return;
const snapshot = head.snapshot;
const replacement = snapshot?.strategy === "native-replace";
const additions = deliveryTotals(head, "upsert"),
removals = deliveryTotals(head, "remove");
const format = (value: typeof additions) =>
`${value.submitted} confirmed submitted in ${value.submittedBatches} batches; ${value.accepted} accepted, ${value.pending} pending processing, ${value.rejected} rejected, ${value.unconfirmed} prepared/unconfirmed, ${value.cancelled} cancelled`;
let message = `Delivery totals (entire logical run, including earlier attempts): ${
replacement ? "full-snapshot uploads" : "additions/upserts"
}: ${format(additions)}.`;
if (replacement) {
const status = snapshot.replacementStatus;
message += ` Full-audience cleanup: ${
status === "prepared" ? "prepared/unconfirmed" : status ?? "not started"
}; removed-member count is not provided by Google.`;
if (status === "prepared")
message += " Cleanup may have reached Google; do not replay without checking its outcome.";
if (!status) message += " Cleanup waits for every snapshot upload to be accepted.";
} else message += ` Removals: ${format(removals)}.`;
if (additions.unconfirmed || removals.unconfirmed)
message += " Unconfirmed batches may have reached the destination; they are not safe to replay blindly.";
if (head.batches.some(batch => batch.status === "acknowledged" && batch.submittedRecords === undefined))
message +=
" Older receipts may not retain the exact submitted count; confirmed submission totals are a lower bound.";
const p = snapshot?.summary;
if (p && replacement) {
message += ` Not yet prepared: ${Math.max(0, p.uniqueMembers - additions.records)} snapshot members.`;
} else if (p) {
const plannedAdditions = p.newMembers + p.changedMembers + p.refreshMembers;
const remainingAdditions = Math.max(0, plannedAdditions - additions.records);
const remainingRemovals = Math.max(0, p.removals - removals.records);
message += ` Not yet prepared: ${remainingAdditions} additions/updates/refreshes, ${remainingRemovals} removals.`;
if (remainingRemovals && additions.accepted < plannedAdditions)
message += " Removals are blocked until all additions/updates/refreshes are accepted.";
if (!plannedAdditions && !p.removals) message += " No audience changes or expiry refreshes needed.";
}
this.summarized = await this.log(message);
}
}