From 12f223f328904803b249f6ff6497dab34d2af80f Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 30 Jul 2026 14:23:38 +0400 Subject: [PATCH 1/6] fix(console): skip-and-log malformed entities in admin exports instead of failing mid-stream One bad row (e.g. a destination with null config reading clickhouseSettings) crashed the streamed /api/admin/export/* response mid-stream, producing a truncated HTTP 200 that every consumer cached or choked on (JITSU-139). - build each entity's JSON before writing, wrap per-entity work in try/catch: malformed entities are skipped and logged with the 'System error:' marker - guard clickhouseSettings parsing with a typeof check - if an export still fails mid-stream, destroy the socket instead of ending the response, so consumers see an aborted transfer, not valid-looking JSON Co-Authored-By: Claude Fable 5 --- .../pages/api/admin/export/[name]/index.ts | 537 ++++++++++-------- 1 file changed, 304 insertions(+), 233 deletions(-) diff --git a/webapps/console/pages/api/admin/export/[name]/index.ts b/webapps/console/pages/api/admin/export/[name]/index.ts index fc083f4a5..aee380cc4 100644 --- a/webapps/console/pages/api/admin/export/[name]/index.ts +++ b/webapps/console/pages/api/admin/export/[name]/index.ts @@ -52,6 +52,17 @@ function dateMax(...dates: (Date | undefined)[]): Date | undefined { return dates.reduce((acc, d) => (d && (!acc || d.getTime() > acc.getTime()) ? d : acc), undefined); } +// One malformed entity must not poison the whole export: exports are streamed, +// so an uncaught error mid-stream truncates the payload for every consumer. +// "System error:" is the unified marker for log-based alerting — keep in sync +// with logging.SystemErrorf in bulker/jitsubase +function logExportEntityError(exportName: string, entityId: string, e: unknown) { + getLog() + .atError() + .withCause(e) + .log(`System error: Failed to export entity '${entityId}' of '${exportName}': ${getErrorMessage(e)}. Skipping`); +} + // Extract functionsClasses from workspace featuresEnabled array // Looks for feature like "functionsClass=dedicated" or "functionsClass=premium,dedicated" function extractFunctionsClasses(featuresEnabled: string[]): string[] { @@ -161,39 +172,36 @@ async function exportBulkerConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { data: data_, from, id, to, updatedAt, workspace } of objects) { - const data = data_ || {}; - if (data?.disabled) { - continue; // skip disabled connections - } - const destinationType = to.config.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - if (needComma) { - writer.write(","); + try { + const data = data_ || {}; + if (data?.disabled) { + continue; // skip disabled connections } - const credentials = omit(to.config, "destinationType", "type", "name"); - if (destinationType === "clickhouse") { - if ((data as any).clickhouseSettings) { - const extraParams = Object.fromEntries( - ((data as any).clickhouseSettings as string) - .split("\n") - .filter(s => s.includes("=")) - .map(s => s.split("=")) - .map(([k, v]) => [k.trim(), v.trim()]) - ); - credentials.parameters = { ...(credentials.parameters || {}), ...extraParams }; - } - if (!credentials.provisioned) { - credentials.loadAsJson = false; + const destinationType = to.config.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { + const credentials = omit(to.config, "destinationType", "type", "name"); + if (destinationType === "clickhouse") { + if (typeof (data as any).clickhouseSettings === "string") { + const extraParams = Object.fromEntries( + ((data as any).clickhouseSettings as string) + .split("\n") + .filter(s => s.includes("=")) + .map(s => s.split("=")) + .map(([k, v]) => [k.trim(), v.trim()]) + ); + credentials.parameters = { ...(credentials.parameters || {}), ...extraParams }; + } + if (!credentials.provisioned) { + credentials.loadAsJson = false; + } } - } - // if (data.timestampColumn) { - // // use timestampColumn field as discriminator field when doing local deduplication - // // inside batch of two rows having the same messageId(pk) will be chosen the one with the highest timestampColumn value - // data.discriminatorField = [data.timestampColumn]; - // } - writer.write( - JSON.stringify({ + // if (data.timestampColumn) { + // // use timestampColumn field as discriminator field when doing local deduplication + // // inside batch of two rows having the same messageId(pk) will be chosen the one with the highest timestampColumn value + // data.discriminatorField = [data.timestampColumn]; + // } + const payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -202,9 +210,15 @@ async function exportBulkerConnections(writer: Writer) { options: omit(data as any, "clickhouseSettings"), updatedAt: dateMax(updatedAt, to.updatedAt), credentials: credentials, - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } + } catch (e) { + logExportEntityError("bulker-connections", id, e); } } if (objects.length < batchSize) { @@ -226,14 +240,11 @@ async function exportBulkerConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} destinations objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { id, workspace, config, updatedAt } of objects) { - const destinationType = config.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + try { + const destinationType = config.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { + const payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -246,9 +257,15 @@ async function exportBulkerConnections(writer: Writer) { }, updatedAt: updatedAt, credentials: omit(config, "destinationType", "type", "name"), - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } + } catch (e) { + logExportEntityError("bulker-connections", id, e); } } if (objects.length < batchSize) { @@ -276,7 +293,10 @@ async function exportBulkerConnections(writer: Writer) { needComma = true; } } catch (e) { - console.error("Error getting backup connections", e); + getLog() + .atError() + .withCause(e) + .log(`System error: Failed to export backup connections for 'bulker-connections': ${getErrorMessage(e)}`); } } @@ -327,20 +347,17 @@ async function exportRotorConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { data: data_, from, id, to, updatedAt, workspace } of objects) { - const data = data_ || {}; - if (data?.disabled) { - continue; // skip disabled connections - } - const destinationType = to.config.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (!coreDestinationType) { - getLog().atError().log(`Unknown destination type: ${destinationType} for connection ${id}`); - } - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + try { + const data = data_ || {}; + if (data?.disabled) { + continue; // skip disabled connections + } + const destinationType = to.config.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (!coreDestinationType) { + getLog().atError().log(`Unknown destination type: ${destinationType} for connection ${id}`); + } + const payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -365,9 +382,15 @@ async function exportRotorConnections(writer: Writer) { updatedAt: dateMax(updatedAt, to.updatedAt), credentials: omit(to.config, "destinationType", "type", "name"), credentialsHash: hash(omit(to.config, "destinationType", "type", "name")), - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("rotor-connections", id, e); + } } if (objects.length < batchSize) { break; @@ -388,14 +411,11 @@ async function exportRotorConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} destinations objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { id, workspace, config, updatedAt } of objects) { - const destinationType = config?.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + try { + const destinationType = config?.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { + const payload = JSON.stringify({ id: id, type: destinationType, workspaceId: workspace.id, @@ -406,9 +426,15 @@ async function exportRotorConnections(writer: Writer) { updatedAt: updatedAt, credentials: omit(config, "destinationType", "type", "name"), credentialsHash: hash(omit(config, "destinationType", "type", "name")), - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } + } catch (e) { + logExportEntityError("rotor-connections", id, e); } } if (objects.length < batchSize) { @@ -416,34 +442,31 @@ async function exportRotorConnections(writer: Writer) { } } for (const pb of profileBuilders) { - if (needComma) { - writer.write(","); - } - const cred = { - ...(pb.intermediateStorageCredentials ?? ({} as any)), - profileWindowDays: (pb.connectionOptions ?? ({} as any)).profileWindow, - profileBuilderId: pb.id, - eventsCollectionName: `profiles-raw-${pb.workspace.id}-${pb.id}`, - traitsCollectionName: `profiles-traits-${pb.workspace.id}-${pb.id}`, - }; - const opts = { - functionsEnv: (pb.connectionOptions ?? ({} as any)).variables, - functions: [ - { - functionId: "builtin.transformation.user-recognition", - }, - ...((pb.connectionOptions ?? ({} as any)).functions || []), - ], - functionsServer: selectFunctionsServer( - functionsServers, - pb.workspace.id, - pb.id, - functionsClassFunc(pb.workspace) - ), - workspaceUpdatedAt: pb.workspace.updatedAt, - }; - writer.write( - JSON.stringify({ + try { + const cred = { + ...(pb.intermediateStorageCredentials ?? ({} as any)), + profileWindowDays: (pb.connectionOptions ?? ({} as any)).profileWindow, + profileBuilderId: pb.id, + eventsCollectionName: `profiles-raw-${pb.workspace.id}-${pb.id}`, + traitsCollectionName: `profiles-traits-${pb.workspace.id}-${pb.id}`, + }; + const opts = { + functionsEnv: (pb.connectionOptions ?? ({} as any)).variables, + functions: [ + { + functionId: "builtin.transformation.user-recognition", + }, + ...((pb.connectionOptions ?? ({} as any)).functions || []), + ], + functionsServer: selectFunctionsServer( + functionsServers, + pb.workspace.id, + pb.id, + functionsClassFunc(pb.workspace) + ), + workspaceUpdatedAt: pb.workspace.updatedAt, + }; + const payload = JSON.stringify({ __debug: { workspace: { id: pb.workspaceId }, }, @@ -459,9 +482,15 @@ async function exportRotorConnections(writer: Writer) { updatedAt: pb.updatedAt, credentials: cred, credentialsHash: hash(cred), - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("rotor-connections", pb.id, e); + } } writer.write("]"); } @@ -488,17 +517,20 @@ async function exportFunctions(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + try { + const payload = JSON.stringify({ ...omit(row, "deleted", "config"), ...row.config, codeHash: hash(row.config?.code || row.config?.draft || ""), - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("functions", row.id, e); + } } if (objects.length < batchSize) { break; @@ -584,17 +616,14 @@ async function exportStreamsWithDestinations(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for streams-with-destinations export`); lastId = objects[objects.length - 1].id; for (const obj of objects) { - if (needComma) { - writer.write(","); - } - const throttlePercent = - workspacesWithClasses.get(obj.workspace.id)?.status !== "active" - ? getNumericOption("throttle", obj.workspace) - : undefined; - const shardNumber = obj.config.shard || getNumericOption("shard", obj.workspace); - const classicKeys = classicKeysMap[obj.id] || ({} as ClassicKeys); - writer.write( - JSON.stringify({ + try { + const throttlePercent = + workspacesWithClasses.get(obj.workspace.id)?.status !== "active" + ? getNumericOption("throttle", obj.workspace) + : undefined; + const shardNumber = obj.config.shard || getNumericOption("shard", obj.workspace); + const classicKeys = classicKeysMap[obj.id] || ({} as ClassicKeys); + const payload = JSON.stringify({ __debug: { workspace: { id: obj.workspace.id, name: obj.workspace.slug }, }, @@ -652,9 +681,15 @@ async function exportStreamsWithDestinations(writer: Writer) { }, })), ], - }) - ); - needComma = true; + }); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("streams-with-destinations", obj.id, e); + } } if (objects.length < batchSize) { break; @@ -732,12 +767,17 @@ async function exportWorkspaces(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { - if (needComma) { - writer.write(","); + try { + row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); + const payload = JSON.stringify(row); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("workspaces", row.id, e); } - row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); - writer.write(JSON.stringify(row)); - needComma = true; } if (objects.length < batchSize) { break; @@ -802,30 +842,35 @@ async function exportWorkspacesWithProfiles(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { - if (needComma) { - writer.write(","); - } - row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); - row.profileBuilders = row.profileBuilders - .filter(pb => pb.version > 0) - .map(pb => { - pb.functions = pb.functions.map(f => { - return { - ...omit(f.function, "config"), - ...f.function.config, - }; + try { + row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); + row.profileBuilders = row.profileBuilders + .filter(pb => pb.version > 0) + .map(pb => { + pb.functions = pb.functions.map(f => { + return { + ...omit(f.function, "config"), + ...f.function.config, + }; + }); + // Add functionsServer routing info for profile builder + (pb as any).functionsServer = selectProfileBuilderFunctionsServer( + functionsServers, + row.id, + pb.id, + functionsClassFunc(row.id) + ); + return pb; }); - // Add functionsServer routing info for profile builder - (pb as any).functionsServer = selectProfileBuilderFunctionsServer( - functionsServers, - row.id, - pb.id, - functionsClassFunc(row.id) - ); - return pb; - }); - writer.write(JSON.stringify(row)); - needComma = true; + const payload = JSON.stringify(row); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("workspaces-with-profiles", row.id, e); + } } if (objects.length < batchSize) { break; @@ -884,89 +929,28 @@ async function exportSyncs(writer: Writer) { lastId = objects[objects.length - 1].id; const enriched = objects.flatMap(({ data, from, id, to, updatedAt, workspace }) => { - // Every sync is scheduled by syncctl CronJobs — emit all of them. - // (Destination-type filters below still skip syncs whose pod template - // can't run them, e.g. non-bulker mixpanel-with-syncs.) - let destinationConfig: any = { ...(to.config as any) }; - const destinationType = destinationConfig.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (!coreDestinationType) { - getLog() - .atError() - .log(`Unknown destination type: ${destinationType} for sync ${id} - skipping export of this sync`); - return []; - } - if (!coreDestinationType.usesBulker && coreDestinationType.id !== "webhook") { - // Non-bulker destinations (e.g. mixpanel-with-syncs) used to run - // synchronously inside the console process via scheduleSync's - // runSynchronously branch — they were never scheduled by GCS, and - // the autonomous CronJob path doesn't support them either. Skip - // them out of the export so syncctl doesn't try to reconcile a - // CronJob whose Pod template can't actually run them. - getLog() - .atError() - .log( - `Sync ${id} has destination type ${destinationType} which does not use bulker - skipping export of this sync` - ); + try { + // Every sync is scheduled by syncctl CronJobs — emit all of them. + // (Destination-type filters below still skip syncs whose pod template + // can't run them, e.g. non-bulker mixpanel-with-syncs.) + return exportSyncEntity({ data, from, id, to, updatedAt, workspace }); + } catch (e) { + logExportEntityError("syncs", id, e); return []; } - const syncData = (data ?? {}) as Record; - let serviceConfig: any = { ...(from.config as any) }; - - // versionHash MUST be derived from the raw persisted credentials — - // matches the formula used by scheduleSync and sources/discover when - // they store catalog rows in source_catalog. Hashing post-mutation or - // post-OAuth-refresh creds would make sidecar's catalog lookup miss - // the rows that scheduleSync wrote. - const versionHash = `${workspace.id}_${from.id}_${juavaHash("md5", stableHash(serviceConfig.credentials))}`; - - // scheduleSync applies this default for these packages — apply it to - // a separate `credentials` value used only in the runtime source.config - // (so it doesn't leak into versionHash above). - if ( - serviceConfig.package === "airbyte/source-postgres" || - serviceConfig.package === "airbyte/source-mssql" || - serviceConfig.package === "airbyte/source-singlestore" - ) { - serviceConfig = { - ...serviceConfig, - credentials: { ...serviceConfig.credentials, sync_checkpoint_records: 200000 }, - }; - } - - // ClickHouse-without-provisioning override (mirrors scheduleSync). - if (destinationType === "clickhouse" && !destinationConfig.provisioned) { - destinationConfig = { ...destinationConfig, loadAsJson: false }; - } - - return [ - { - id, - workspaceId: workspace.id, - workspaceSlug: workspace.slug, - fromId: from.id, - toId: to.id, - source: serviceConfig, - destination: destinationConfig, - schedule: syncData.schedule, - timezone: syncData.timezone ?? "Etc/UTC", - // Everything from sync.data minus the fields already promoted to - // top-level (schedule, timezone), plus the computed versionHash. - options: { - ...omit(syncData, "schedule", "timezone"), - versionHash, - }, - updatedAt: dateMax(updatedAt, from.updatedAt, to.updatedAt), - }, - ]; }); for (const item of enriched) { - if (needComma) { - writer.write(","); + try { + const payload = JSON.stringify(item); + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; + } catch (e) { + logExportEntityError("syncs", item.id, e); } - writer.write(JSON.stringify(item)); - needComma = true; } if (objects.length < batchSize) { @@ -976,6 +960,81 @@ async function exportSyncs(writer: Writer) { writer.write("]"); } +function exportSyncEntity({ data, from, id, to, updatedAt, workspace }: any) { + let destinationConfig: any = { ...(to.config as any) }; + const destinationType = destinationConfig.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (!coreDestinationType) { + getLog() + .atError() + .log(`Unknown destination type: ${destinationType} for sync ${id} - skipping export of this sync`); + return []; + } + if (!coreDestinationType.usesBulker && coreDestinationType.id !== "webhook") { + // Non-bulker destinations (e.g. mixpanel-with-syncs) used to run + // synchronously inside the console process via scheduleSync's + // runSynchronously branch — they were never scheduled by GCS, and + // the autonomous CronJob path doesn't support them either. Skip + // them out of the export so syncctl doesn't try to reconcile a + // CronJob whose Pod template can't actually run them. + getLog() + .atError() + .log( + `Sync ${id} has destination type ${destinationType} which does not use bulker - skipping export of this sync` + ); + return []; + } + const syncData = (data ?? {}) as Record; + let serviceConfig: any = { ...(from.config as any) }; + + // versionHash MUST be derived from the raw persisted credentials — + // matches the formula used by scheduleSync and sources/discover when + // they store catalog rows in source_catalog. Hashing post-mutation or + // post-OAuth-refresh creds would make sidecar's catalog lookup miss + // the rows that scheduleSync wrote. + const versionHash = `${workspace.id}_${from.id}_${juavaHash("md5", stableHash(serviceConfig.credentials))}`; + + // scheduleSync applies this default for these packages — apply it to + // a separate `credentials` value used only in the runtime source.config + // (so it doesn't leak into versionHash above). + if ( + serviceConfig.package === "airbyte/source-postgres" || + serviceConfig.package === "airbyte/source-mssql" || + serviceConfig.package === "airbyte/source-singlestore" + ) { + serviceConfig = { + ...serviceConfig, + credentials: { ...serviceConfig.credentials, sync_checkpoint_records: 200000 }, + }; + } + + // ClickHouse-without-provisioning override (mirrors scheduleSync). + if (destinationType === "clickhouse" && !destinationConfig.provisioned) { + destinationConfig = { ...destinationConfig, loadAsJson: false }; + } + + return [ + { + id, + workspaceId: workspace.id, + workspaceSlug: workspace.slug, + fromId: from.id, + toId: to.id, + source: serviceConfig, + destination: destinationConfig, + schedule: syncData.schedule, + timezone: syncData.timezone ?? "Etc/UTC", + // Everything from sync.data minus the fields already promoted to + // top-level (schedule, timezone), plus the computed versionHash. + options: { + ...omit(syncData, "schedule", "timezone"), + versionHash, + }, + updatedAt: dateMax(updatedAt, from.updatedAt, to.updatedAt), + }, + ]; +} + const exports: Export[] = [ { name: "bulker-connections", @@ -1158,7 +1217,19 @@ export default createRoute() if (query.dateOnly) { res.write(JSON.stringify({ lastModified: lastModified.toISOString() })); } else { - await exp.data(res); + try { + await exp.data(res); + } catch (e) { + // Headers are already sent, so this can't become an HTTP 500. Destroy the + // socket so consumers see an aborted response instead of a seemingly + // complete but truncated JSON document served with status 200. + getLog() + .atError() + .withCause(e) + .log(`System error: Export '${query.name}' failed mid-stream: ${getErrorMessage(e)}`); + res.destroy(e instanceof Error ? e : new Error(getErrorMessage(e))); + return; + } } res.end(); }) From 2be5b91af96b71cad933208fa4379ec381954fde Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 30 Jul 2026 14:23:50 +0400 Subject: [PATCH 2/6] fix(bulker): mark startup failures and repository refresh errors as system errors Failure-to-start (Fatalf, e.g. 'Cannot serve without repository') and repository refresh failures now carry the 'System error:' log marker used for unified log-based alerting across Go and Node components (JITSU-139). Co-Authored-By: Claude Fable 5 --- bulker/bulkerapp/app/postgres_configuration_source.go | 2 +- bulker/jitsubase/appbase/abstract_repository.go | 2 +- bulker/jitsubase/logging/global_logger.go | 8 ++++++-- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/bulker/bulkerapp/app/postgres_configuration_source.go b/bulker/bulkerapp/app/postgres_configuration_source.go index 262002220..7e5696033 100644 --- a/bulker/bulkerapp/app/postgres_configuration_source.go +++ b/bulker/bulkerapp/app/postgres_configuration_source.go @@ -125,7 +125,7 @@ func (r *PostgresConfigurationSource) refresh(notify bool) { var err error defer func() { if err != nil { - r.Errorf("Error refreshing repository: %v", err) + r.SystemErrorf("Error refreshing repository: %v", err) metrics.ConfigurationSourceError("error").Inc() if !r.inited.Load() { if r.cacheDir != "" { diff --git a/bulker/jitsubase/appbase/abstract_repository.go b/bulker/jitsubase/appbase/abstract_repository.go index 1caf28b97..ed6a4ace1 100644 --- a/bulker/jitsubase/appbase/abstract_repository.go +++ b/bulker/jitsubase/appbase/abstract_repository.go @@ -118,7 +118,7 @@ func (r *AbstractRepository[T]) refresh(notify bool) { var err error defer func() { if err != nil { - r.Errorf("Error refreshing repository: %v", err) + r.SystemErrorf("Error refreshing repository: %v", err) if !r.inited.Load() { if r.cacheDir != "" { r.loadCached() diff --git a/bulker/jitsubase/logging/global_logger.go b/bulker/jitsubase/logging/global_logger.go index d7e44da6a..db5c19423 100644 --- a/bulker/jitsubase/logging/global_logger.go +++ b/bulker/jitsubase/logging/global_logger.go @@ -121,10 +121,14 @@ func Warn(v ...any) { log.Warnln(v...) } +// Fatal-level failures abort the process (typically failure to start), so they +// carry the "System error:" marker used by log-based alerting. func Fatal(v ...any) { - log.Fatal(v...) + msg := []any{"System error:"} + msg = append(msg, v...) + log.Fatal(msg...) } func Fatalf(format string, v ...any) { - log.Fatalf(format, v...) + log.Fatalf("System error: "+format, v...) } From eae73fef6b5e5edbbf92fcfd5dfce0c83d659ce4 Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 30 Jul 2026 14:23:50 +0400 Subject: [PATCH 3/6] fix(cfgkpr): validate JSON before caching repository payloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit config-keeper cached whatever bytes the console export returned. During JITSU-139 the export failed mid-stream and produced truncated JSON with HTTP 200; cfgkpr cached it and served it to every consumer for hours. Reject payloads that are not complete valid JSON — the repository keeps serving the previous good data and the refresh failure is logged as a system error. p.js (JavaScript, not JSON) is exempt. Co-Authored-By: Claude Fable 5 --- bulker/config-keeper/app.go | 12 ++++++++++-- bulker/config-keeper/router.go | 2 +- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/bulker/config-keeper/app.go b/bulker/config-keeper/app.go index 7bd84c29c..8097af79e 100644 --- a/bulker/config-keeper/app.go +++ b/bulker/config-keeper/app.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "fmt" "github.com/jitsucom/bulker/jitsubase/appbase" "io" @@ -19,7 +20,11 @@ type Context struct { } type RawRepositoryData struct { - data atomic.Pointer[[]byte] + // validateJSON rejects payloads that are not complete, valid JSON. A repository + // source that fails mid-stream can produce a truncated body with HTTP 200 — + // without this check such payload gets cached and served to every consumer. + validateJSON bool + data atomic.Pointer[[]byte] } func (r *RawRepositoryData) Init(reader io.Reader, tag any) error { @@ -27,6 +32,9 @@ func (r *RawRepositoryData) Init(reader io.Reader, tag any) error { if err != nil { return err } + if r.validateJSON && !json.Valid(data) { + return fmt.Errorf("payload is not valid JSON (%d bytes) - keeping previous data", len(data)) + } r.data.Store(&data) return nil } @@ -59,7 +67,7 @@ func (a *Context) InitContext(settings *appbase.AppSettings) error { "p.js": a.pScript, } for _, rep := range strings.Split(reps, ",") { - a.repositories[rep] = appbase.NewHTTPRepository[[]byte](rep, baseUrl+"/"+rep, token, appbase.HTTPTagLastModified, &RawRepositoryData{}, 2, refreshPeriodSec, cacheDir) + a.repositories[rep] = appbase.NewHTTPRepository[[]byte](rep, baseUrl+"/"+rep, token, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, refreshPeriodSec, cacheDir) } router := NewRouter(a) diff --git a/bulker/config-keeper/router.go b/bulker/config-keeper/router.go index 5cc862785..e2c4ca65f 100644 --- a/bulker/config-keeper/router.go +++ b/bulker/config-keeper/router.go @@ -77,7 +77,7 @@ func (r *Router) RepositoryHandler(c *gin.Context) { repository, ok := r.appContext.repositories[repName] if !ok { r.Infof("Repository %s not found, initializing", repName) - repository = appbase.NewHTTPRepository[[]byte](repName, r.appContext.config.RepositoryBaseURL+"/"+repName, r.appContext.config.RepositoryAuthToken, appbase.HTTPTagLastModified, &RawRepositoryData{}, 2, r.appContext.config.RepositoryRefreshPeriodSec, r.appContext.config.CacheDir) + repository = appbase.NewHTTPRepository[[]byte](repName, r.appContext.config.RepositoryBaseURL+"/"+repName, r.appContext.config.RepositoryAuthToken, appbase.HTTPTagLastModified, &RawRepositoryData{validateJSON: true}, 2, r.appContext.config.RepositoryRefreshPeriodSec, r.appContext.config.CacheDir) initTimeout := time.After(time.Second * 60) ticker := time.NewTicker(time.Second) defer ticker.Stop() From be5a30d43fc0e934d310e14d8473d31cb2493eae Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 30 Jul 2026 14:23:51 +0400 Subject: [PATCH 4/6] fix(rotor): unify repository refresh error logging with system-error marker inmem-store refresh failures now log 'System error: Error refreshing repository ' at error level, matching the Go-side message from jitsubase appbase, so one Datadog monitor covers both stacks (JITSU-139). Initialize-with-no-cache failure carries the marker too. Co-Authored-By: Claude Fable 5 --- libs/core-functions-lib/src/lib/inmem-store.ts | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/libs/core-functions-lib/src/lib/inmem-store.ts b/libs/core-functions-lib/src/lib/inmem-store.ts index d795393b4..5ec71f8ef 100644 --- a/libs/core-functions-lib/src/lib/inmem-store.ts +++ b/libs/core-functions-lib/src/lib/inmem-store.ts @@ -75,7 +75,12 @@ export const createInMemoryStore = (definition: StoreDefinition): InMemory status = "ok"; lastRefresh = new Date(); } catch (e) { - log.atWarn().withCause(e).log(`Failed to refresh store ${definition.name}. Using an old value`); + // "System error:" is the unified marker for log-based alerting — keep in sync + // with logging.SystemErrorf in bulker/jitsubase + log + .atError() + .withCause(e) + .log(`System error: Error refreshing repository ${definition.name}. Using an old value`); status = "outdated"; } }; @@ -109,6 +114,13 @@ export const createInMemoryStore = (definition: StoreDefinition): InMemory const cachedInstance = loadFromCache(definition); if (!cachedInstance) { status = "failed"; + log + .atError() + .log( + `System error: Failed to initialize store ${definition.name}. Initial load failed with ${getErrorMessage( + e + )} and no local cache found` + ); reject( new Error( `Failed to initialize store ${definition.name}. Initial load failed with ${getErrorMessage( From bb712b4a8466baadf8a9d9ad44238ecfa64dce22 Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Thu, 30 Jul 2026 14:39:32 +0400 Subject: [PATCH 5/6] fix(bulker): repository refresh errors are not system errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refresh failures are recoverable — consumers keep serving last-known-good config. Only failure to init/start an app carries the 'System error:' marker (Fatalf and store-init-without-cache keep it). Refresh errors stay at error level with unified 'Error refreshing repository' wording across Go and Node for the dedicated Datadog monitor. Co-Authored-By: Claude Fable 5 --- bulker/bulkerapp/app/postgres_configuration_source.go | 2 +- bulker/jitsubase/appbase/abstract_repository.go | 2 +- libs/core-functions-lib/src/lib/inmem-store.ts | 10 ++++------ 3 files changed, 6 insertions(+), 8 deletions(-) diff --git a/bulker/bulkerapp/app/postgres_configuration_source.go b/bulker/bulkerapp/app/postgres_configuration_source.go index 7e5696033..262002220 100644 --- a/bulker/bulkerapp/app/postgres_configuration_source.go +++ b/bulker/bulkerapp/app/postgres_configuration_source.go @@ -125,7 +125,7 @@ func (r *PostgresConfigurationSource) refresh(notify bool) { var err error defer func() { if err != nil { - r.SystemErrorf("Error refreshing repository: %v", err) + r.Errorf("Error refreshing repository: %v", err) metrics.ConfigurationSourceError("error").Inc() if !r.inited.Load() { if r.cacheDir != "" { diff --git a/bulker/jitsubase/appbase/abstract_repository.go b/bulker/jitsubase/appbase/abstract_repository.go index ed6a4ace1..1caf28b97 100644 --- a/bulker/jitsubase/appbase/abstract_repository.go +++ b/bulker/jitsubase/appbase/abstract_repository.go @@ -118,7 +118,7 @@ func (r *AbstractRepository[T]) refresh(notify bool) { var err error defer func() { if err != nil { - r.SystemErrorf("Error refreshing repository: %v", err) + r.Errorf("Error refreshing repository: %v", err) if !r.inited.Load() { if r.cacheDir != "" { r.loadCached() diff --git a/libs/core-functions-lib/src/lib/inmem-store.ts b/libs/core-functions-lib/src/lib/inmem-store.ts index 5ec71f8ef..0a209d594 100644 --- a/libs/core-functions-lib/src/lib/inmem-store.ts +++ b/libs/core-functions-lib/src/lib/inmem-store.ts @@ -75,12 +75,10 @@ export const createInMemoryStore = (definition: StoreDefinition): InMemory status = "ok"; lastRefresh = new Date(); } catch (e) { - // "System error:" is the unified marker for log-based alerting — keep in sync - // with logging.SystemErrorf in bulker/jitsubase - log - .atError() - .withCause(e) - .log(`System error: Error refreshing repository ${definition.name}. Using an old value`); + // Not a system error (the store keeps serving the old value) — but the message + // wording matches the Go-side repository refresh error in bulker/jitsubase so + // one log query covers both stacks + log.atError().withCause(e).log(`Error refreshing repository ${definition.name}. Using an old value`); status = "outdated"; } }; From e307cd57864f53933dab2dbd39f7b68fccb04be9 Mon Sep 17 00:00:00 2001 From: Ildar Nurislamov Date: Mon, 3 Aug 2026 19:06:34 +0400 Subject: [PATCH 6/6] fix(console): write outside per-entity catches; syncs export fails closed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on the skip-and-log hardening: Writes no longer live inside the per-entity try/catch. A failing writer.write means the response is gone, and swallowing it kept the export scanning every remaining DB row for a stream nobody would ever read — while the outer error handling never learned about it. Each site now materializes and serializes inside the guard, and writes outside it, so stream failures abort the export immediately. One malformed row still skips only that row. The syncs export now fails closed instead of omitting entities. syncctl reconciles desired state from this export and deletes CronJobs that are absent from it, so skipping a sync whose row failed to materialize (or serialize) would tear down a healthy sync. Any per-entity failure fails the whole export; consumers keep their last known good snapshot — stale is safe, wrong is not. Also rebased onto current newjitsu: the branch was based on 079fd5da9, whose export changes were since reverted and re-landed as 3d5ef9046. Co-Authored-By: Claude Fable 5 --- .../pages/api/admin/export/[name]/index.ts | 204 ++++++++++++------ 1 file changed, 140 insertions(+), 64 deletions(-) diff --git a/webapps/console/pages/api/admin/export/[name]/index.ts b/webapps/console/pages/api/admin/export/[name]/index.ts index aee380cc4..5a016a62a 100644 --- a/webapps/console/pages/api/admin/export/[name]/index.ts +++ b/webapps/console/pages/api/admin/export/[name]/index.ts @@ -172,6 +172,7 @@ async function exportBulkerConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { data: data_, from, id, to, updatedAt, workspace } of objects) { + let payload: string | undefined; try { const data = data_ || {}; if (data?.disabled) { @@ -201,7 +202,7 @@ async function exportBulkerConnections(writer: Writer) { // // inside batch of two rows having the same messageId(pk) will be chosen the one with the highest timestampColumn value // data.discriminatorField = [data.timestampColumn]; // } - const payload = JSON.stringify({ + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -211,15 +212,22 @@ async function exportBulkerConnections(writer: Writer) { updatedAt: dateMax(updatedAt, to.updatedAt), credentials: credentials, }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("bulker-connections", id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -240,11 +248,12 @@ async function exportBulkerConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} destinations objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { id, workspace, config, updatedAt } of objects) { + let payload: string | undefined; try { const destinationType = config.destinationType; const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - const payload = JSON.stringify({ + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -258,15 +267,22 @@ async function exportBulkerConnections(writer: Writer) { updatedAt: updatedAt, credentials: omit(config, "destinationType", "type", "name"), }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("bulker-connections", id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -347,6 +363,7 @@ async function exportRotorConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { data: data_, from, id, to, updatedAt, workspace } of objects) { + let payload: string | undefined; try { const data = data_ || {}; if (data?.disabled) { @@ -357,7 +374,7 @@ async function exportRotorConnections(writer: Writer) { if (!coreDestinationType) { getLog().atError().log(`Unknown destination type: ${destinationType} for connection ${id}`); } - const payload = JSON.stringify({ + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -383,14 +400,21 @@ async function exportRotorConnections(writer: Writer) { credentials: omit(to.config, "destinationType", "type", "name"), credentialsHash: hash(omit(to.config, "destinationType", "type", "name")), }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("rotor-connections", id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -411,11 +435,12 @@ async function exportRotorConnections(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} destinations objects for bulker export`); lastId = objects[objects.length - 1].id; for (const { id, workspace, config, updatedAt } of objects) { + let payload: string | undefined; try { const destinationType = config?.destinationType; const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - const payload = JSON.stringify({ + payload = JSON.stringify({ id: id, type: destinationType, workspaceId: workspace.id, @@ -427,21 +452,29 @@ async function exportRotorConnections(writer: Writer) { credentials: omit(config, "destinationType", "type", "name"), credentialsHash: hash(omit(config, "destinationType", "type", "name")), }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("rotor-connections", id, e); + continue; } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; } } for (const pb of profileBuilders) { + let payload: string | undefined; try { const cred = { ...(pb.intermediateStorageCredentials ?? ({} as any)), @@ -466,7 +499,7 @@ async function exportRotorConnections(writer: Writer) { ), workspaceUpdatedAt: pb.workspace.updatedAt, }; - const payload = JSON.stringify({ + payload = JSON.stringify({ __debug: { workspace: { id: pb.workspaceId }, }, @@ -483,14 +516,21 @@ async function exportRotorConnections(writer: Writer) { credentials: cred, credentialsHash: hash(cred), }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("rotor-connections", pb.id, e); + continue; } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; } writer.write("]"); } @@ -517,20 +557,28 @@ async function exportFunctions(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { + let payload: string | undefined; try { - const payload = JSON.stringify({ + payload = JSON.stringify({ ...omit(row, "deleted", "config"), ...row.config, codeHash: hash(row.config?.code || row.config?.draft || ""), }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("functions", row.id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -616,6 +664,7 @@ async function exportStreamsWithDestinations(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for streams-with-destinations export`); lastId = objects[objects.length - 1].id; for (const obj of objects) { + let payload: string | undefined; try { const throttlePercent = workspacesWithClasses.get(obj.workspace.id)?.status !== "active" @@ -623,7 +672,7 @@ async function exportStreamsWithDestinations(writer: Writer) { : undefined; const shardNumber = obj.config.shard || getNumericOption("shard", obj.workspace); const classicKeys = classicKeysMap[obj.id] || ({} as ClassicKeys); - const payload = JSON.stringify({ + payload = JSON.stringify({ __debug: { workspace: { id: obj.workspace.id, name: obj.workspace.slug }, }, @@ -682,14 +731,21 @@ async function exportStreamsWithDestinations(writer: Writer) { })), ], }); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("streams-with-destinations", obj.id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -767,17 +823,25 @@ async function exportWorkspaces(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { + let payload: string | undefined; try { row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); - const payload = JSON.stringify(row); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; + payload = JSON.stringify(row); } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("workspaces", row.id, e); + continue; + } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -842,6 +906,7 @@ async function exportWorkspacesWithProfiles(writer: Writer) { getLog().atDebug().log(`Got batch of ${objects.length} objects for bulker export`); lastId = objects[objects.length - 1].id; for (const row of objects) { + let payload: string | undefined; try { row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); row.profileBuilders = row.profileBuilders @@ -862,15 +927,22 @@ async function exportWorkspacesWithProfiles(writer: Writer) { ); return pb; }); - const payload = JSON.stringify(row); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; + payload = JSON.stringify(row); } catch (e) { + // Only entity materialization/serialization is guarded: one malformed row + // must not take down the whole export. Writes happen OUTSIDE the try so a + // failing stream aborts the export instead of silently scanning on. logExportEntityError("workspaces-with-profiles", row.id, e); + continue; } + if (payload === undefined) { + continue; + } + if (needComma) { + writer.write(","); + } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) { break; @@ -935,22 +1007,26 @@ async function exportSyncs(writer: Writer) { // can't run them, e.g. non-bulker mixpanel-with-syncs.) return exportSyncEntity({ data, from, id, to, updatedAt, workspace }); } catch (e) { + // Do NOT skip-and-continue here: syncctl reconciles desired state from + // this export and deletes CronJobs that are absent from it, so omitting + // a sync because its row failed to materialize would tear down a healthy + // sync. Fail the whole export instead — consumers keep their last known + // good snapshot (stale is safe, wrong is not). logExportEntityError("syncs", id, e); - return []; + throw e; } }); for (const item of enriched) { - try { - const payload = JSON.stringify(item); - if (needComma) { - writer.write(","); - } - writer.write(payload); - needComma = true; - } catch (e) { - logExportEntityError("syncs", item.id, e); + // No per-entity skip anywhere in syncs: omission reads as deletion to + // syncctl (see catch above), so a serialization failure also fails the + // whole export rather than dropping the entity. + const payload = JSON.stringify(item); + if (needComma) { + writer.write(","); } + writer.write(payload); + needComma = true; } if (objects.length < batchSize) {