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() 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...) } diff --git a/libs/core-functions-lib/src/lib/inmem-store.ts b/libs/core-functions-lib/src/lib/inmem-store.ts index d795393b4..0a209d594 100644 --- a/libs/core-functions-lib/src/lib/inmem-store.ts +++ b/libs/core-functions-lib/src/lib/inmem-store.ts @@ -75,7 +75,10 @@ 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`); + // 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"; } }; @@ -109,6 +112,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( diff --git a/webapps/console/pages/api/admin/export/[name]/index.ts b/webapps/console/pages/api/admin/export/[name]/index.ts index fc083f4a5..5a016a62a 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,37 @@ 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(","); + let payload: string | undefined; + 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]; + // } + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -202,10 +211,23 @@ async function exportBulkerConnections(writer: Writer) { options: omit(data as any, "clickhouseSettings"), updatedAt: dateMax(updatedAt, to.updatedAt), credentials: credentials, - }) - ); - 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; @@ -226,14 +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) { - const destinationType = config.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + let payload: string | undefined; + try { + const destinationType = config.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -246,10 +266,23 @@ async function exportBulkerConnections(writer: Writer) { }, updatedAt: updatedAt, credentials: omit(config, "destinationType", "type", "name"), - }) - ); - 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; @@ -276,7 +309,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 +363,18 @@ 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({ + let payload: string | undefined; + 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}`); + } + payload = JSON.stringify({ __debug: { workspace: { id: workspace.id, name: workspace.slug }, }, @@ -365,8 +399,21 @@ 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")), - }) - ); + }); + } 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) { @@ -388,14 +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) { - const destinationType = config?.destinationType; - const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); - if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { - if (needComma) { - writer.write(","); - } - writer.write( - JSON.stringify({ + let payload: string | undefined; + try { + const destinationType = config?.destinationType; + const coreDestinationType = getCoreDestinationTypeNonStrict(destinationType); + if (coreDestinationType?.usesBulker || coreDestinationType?.hybrid) { + payload = JSON.stringify({ id: id, type: destinationType, workspaceId: workspace.id, @@ -406,44 +451,55 @@ async function exportRotorConnections(writer: Writer) { updatedAt: updatedAt, credentials: omit(config, "destinationType", "type", "name"), credentialsHash: hash(omit(config, "destinationType", "type", "name")), - }) - ); - 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) { - 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({ + let payload: string | undefined; + 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, + }; + payload = JSON.stringify({ __debug: { workspace: { id: pb.workspaceId }, }, @@ -459,8 +515,21 @@ async function exportRotorConnections(writer: Writer) { updatedAt: pb.updatedAt, credentials: cred, credentialsHash: hash(cred), - }) - ); + }); + } 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("]"); @@ -488,16 +557,27 @@ 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({ + let payload: string | undefined; + try { + payload = JSON.stringify({ ...omit(row, "deleted", "config"), ...row.config, codeHash: hash(row.config?.code || row.config?.draft || ""), - }) - ); + }); + } 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) { @@ -584,17 +664,15 @@ 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({ + let payload: string | undefined; + 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); + payload = JSON.stringify({ __debug: { workspace: { id: obj.workspace.id, name: obj.workspace.slug }, }, @@ -652,8 +730,21 @@ async function exportStreamsWithDestinations(writer: Writer) { }, })), ], - }) - ); + }); + } 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) { @@ -732,11 +823,24 @@ 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)); + 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(","); } - row.featuresEnabled = addFunctionsClass(row.featuresEnabled ?? [], functionsClassFunc(row.id)); - writer.write(JSON.stringify(row)); + writer.write(payload); needComma = true; } if (objects.length < batchSize) { @@ -802,29 +906,42 @@ 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 + .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; + }); + 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(","); } - 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; - }); - writer.write(JSON.stringify(row)); + writer.write(payload); needComma = true; } if (objects.length < batchSize) { @@ -884,88 +1001,31 @@ 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 []; + 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) { + // 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); + throw e; } - 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), - }, - ]; }); for (const item of enriched) { + // 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(JSON.stringify(item)); + writer.write(payload); needComma = true; } @@ -976,6 +1036,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 +1293,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(); })