diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/backupSettings.test.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/backupSettings.test.ts new file mode 100644 index 000000000..562b84f52 --- /dev/null +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/backupSettings.test.ts @@ -0,0 +1,258 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { ChangeTrackerConfig } from "../../bigquery/types"; + +const commit = jest.fn(); +const set = jest.fn(); +const settings = jest.fn(); + +const batch = jest.fn(() => ({ set, commit })); +const collection = jest.fn(() => ({ doc: (id: string) => ({ id }) })); + +jest.mock("firebase-admin", () => ({ apps: [{}] })); +jest.mock("firebase-admin/app", () => ({ initializeApp: jest.fn() })); +jest.mock("firebase-admin/firestore", () => ({ + // A fresh object per call, deliberately: the guard must key on the database id + // rather than on instance identity. + getFirestore: jest.fn(() => ({ settings, batch, collection })), +})); + +const config = { + backupTableId: "bq_failures", + firestoreInstanceId: "(default)", +} as ChangeTrackerConfig; + +const ROWS = [{ insertId: "e1", json: { event_id: "e1" } }]; + +/** Fresh module, so the module-level "already configured" set starts empty. */ +const loadHandler = () => { + let handler: any; + jest.isolateModules(() => { + handler = require("../../bigquery/handleFailedTransactions").default; + }); + return handler; +}; + +describe("handleFailedTransactions Firestore settings", () => { + beforeEach(() => { + jest.clearAllMocks(); + commit.mockResolvedValue(undefined); + settings.mockImplementation(() => undefined); + }); + + it("applies settings once across repeated failures", async () => { + const handler = loadHandler(); + + await handler(ROWS, config, new Error("insert failed")); + await handler(ROWS, config, new Error("insert failed")); + + expect(settings).toHaveBeenCalledTimes(1); + expect(commit).toHaveBeenCalledTimes(2); + }); + + it("still writes the backup when settings cannot be applied", async () => { + settings.mockImplementation(() => { + throw new Error("Firestore has already been initialized"); + }); + + const handler = loadHandler(); + + await expect( + handler(ROWS, config, new Error("insert failed")) + ).resolves.toBeUndefined(); + + expect(commit).toHaveBeenCalledTimes(1); + }); + + it("still writes the backup when the thrown value is not an Error", async () => { + // `insertData` reports whatever it caught, so a non-Error reaches here. + const handler = loadHandler(); + + await expect( + handler(ROWS, config, undefined as any) + ).resolves.toBeUndefined(); + + expect(commit).toHaveBeenCalledTimes(1); + expect(typeof set.mock.calls[0][1].error_details).toBe("string"); + }); + + it("writes one document per row, keyed by insertId", async () => { + const handler = loadHandler(); + + await handler( + [{ insertId: "a" }, { insertId: "b" }], + config, + new Error("boom") + ); + + expect(collection).toHaveBeenCalledWith("bq_failures"); + expect(set).toHaveBeenCalledTimes(2); + expect(set.mock.calls[0][0]).toMatchObject({ id: "a" }); + expect(set.mock.calls[0][1]).toMatchObject({ error_details: "boom" }); + }); +}); + +/** + * A stand-in for `PartialFailureError`: one entry per failed row nesting the + * per-field errors, and the empty message `@google-cloud/common` builds from them. + */ +const partialFailure = (groups: any[]) => + Object.assign(new Error(""), { name: "PartialFailureError", errors: groups }); + +describe("handleFailedTransactions error details", () => { + beforeEach(() => { + jest.clearAllMocks(); + commit.mockResolvedValue(undefined); + settings.mockImplementation(() => undefined); + }); + + const detailsFor = async (e: any) => { + // Cleared per call, so a test may describe more than one failure shape. + set.mockClear(); + + await loadHandler()(ROWS, config, e); + + return set.mock.calls[0][1].error_details; + }; + + it("records the nested per-field messages when the top-level message is empty", async () => { + // The shape a real rejected insert arrives in. + const details = await detailsFor( + partialFailure([ + { + errors: [ + { message: "no such field: document_id.", reason: "invalid" }, + ], + row: { insertId: "e1" }, + }, + ]) + ); + + expect(details).toBe("no such field: document_id."); + }); + + it("deduplicates messages shared across failed rows", async () => { + const details = await detailsFor( + partialFailure([ + { errors: [{ message: "no such field: document_id." }] }, + { errors: [{ message: "no such field: document_id." }] }, + { errors: [{ message: "no such field: old_data." }] }, + ]) + ); + + expect(details).toBe( + "no such field: document_id.; no such field: old_data." + ); + }); + + it("caps the number of messages and the total length", async () => { + const details = await detailsFor( + partialFailure( + Array.from({ length: 9 }, (_, i) => ({ + errors: [{ message: `${"x".repeat(400)} ${i}` }], + })) + ) + ); + + // The count must survive the truncation rather than be cut off by it. + expect(details.length).toBeLessThanOrEqual(1000); + expect(details.endsWith(" (+4 more)")).toBe(true); + expect(details).toContain("..."); + + const short = await detailsFor( + partialFailure( + Array.from({ length: 8 }, (_, i) => ({ + errors: [{ message: `field ${i}` }], + })) + ) + ); + + expect(short).toBe("field 0; field 1; field 2; field 3; field 4 (+3 more)"); + }); + + it("falls back to the reason when an entry carries no message", async () => { + // A `stopped` entry, the row BigQuery did not attempt, arrives with an empty + // message and location, so the reason is all there is. + const details = await detailsFor( + partialFailure([ + { errors: [{ message: "", location: "", reason: "stopped" }] }, + { errors: [{ message: "", location: "", reason: "stopped" }] }, + ]) + ); + + expect(details).toBe("stopped"); + }); + + it("prefers an entry's message over its reason", async () => { + const details = await detailsFor( + partialFailure([ + { + errors: [ + { message: "no such field: document_id.", reason: "invalid" }, + ], + }, + ]) + ); + + expect(details).toBe("no such field: document_id."); + }); + + it("survives an error whose message getter throws", async () => { + const hostile = { + get message(): string { + throw new Error("hostile getter"); + }, + }; + + await expect(detailsFor(hostile)).resolves.toBe("Unknown error"); + }); + + it("prefers a populated top-level message over the nested ones", async () => { + const details = await detailsFor( + Object.assign(new Error("quota exceeded"), { + errors: [{ errors: [{ message: "no such field: document_id." }] }], + }) + ); + + expect(details).toBe("quota exceeded"); + }); + + it("still writes a string for every malformed shape of `errors`", async () => { + // The handler runs inside the caller's catch block, so a throw loses the row. + const shapes: any[] = [ + partialFailure([]), + Object.assign(new Error(""), { errors: "not an array" }), + Object.assign(new Error(""), { errors: [null, undefined] }), + Object.assign(new Error(""), { errors: [{ errors: null }] }), + Object.assign(new Error(""), { errors: [{ errors: [null] }] }), + Object.assign(new Error(""), { errors: [{ errors: [{}] }] }), + Object.assign(new Error(""), { errors: [{ errors: [{ message: 42 }] }] }), + "a plain string", + 42, + null, + Object.create(null), + ]; + + for (const shape of shapes) { + jest.clearAllMocks(); + + await expect(loadHandler()(ROWS, config, shape)).resolves.toBeUndefined(); + + expect(typeof set.mock.calls[0][1].error_details).toBe("string"); + } + }); +}); diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/e2e.test.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/e2e.test.ts index c642de74c..6c1642411 100644 --- a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/e2e.test.ts +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/e2e.test.ts @@ -814,8 +814,15 @@ describe("e2e", () => { test("successfully adds old data field if it does not yet exist", async () => { const event: FirestoreDocumentChangeEvent = changeTrackerEvent({}); - /** Create a table without an old_data column */ - let schema = [{ name: "Name", type: "STRING" }]; + /** + * A valid changelog in every respect except that it predates `old_data`, + * which is the case this test is about. The base columns are never added + * to a table that already exists, so a table missing those as well would + * fail the insert outright rather than exercise the lag retry. + */ + let schema = RawChangelogSchema.fields.filter( + (field) => field.name !== "old_data" + ); let [originalRawTable] = await dataset.createTable(table_raw_changelog, { schema, diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/insertRetry.test.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/insertRetry.test.ts new file mode 100644 index 000000000..6a34d486d --- /dev/null +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/__tests__/bigquery/insertRetry.test.ts @@ -0,0 +1,887 @@ +/** + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { FirestoreBigQueryEventHistoryTracker } from "../../bigquery"; +import { ChangeTrackerConfig } from "../../bigquery/types"; +import handleFailedTransactions from "../../bigquery/handleFailedTransactions"; +import { logger } from "../../logger"; + +jest.mock("../../bigquery/handleFailedTransactions", () => ({ + __esModule: true, + default: jest.fn().mockResolvedValue(undefined), +})); + +const handleFailedTransactionsMock = handleFailedTransactions as jest.Mock; + +process.env.PROJECT_ID = "test-project"; + +const config = ( + overrides: Partial = {} +): ChangeTrackerConfig => + ({ + datasetId: "dataset", + tableId: "table", + datasetLocation: "us", + backupTableId: "backup", + transformFunction: "", + partitioning: { granularity: "NONE" }, + clustering: [], + bqProjectId: "test-project", + ...overrides, + } as ChangeTrackerConfig); + +/** + * The error shape `@google-cloud/bigquery` actually throws: `response` is the raw + * `insertAll` body, where `insertErrors` is an array, and the error's own + * `errors` is the remapped copy that drops `location`. + */ +function partialFailure( + fieldErrors: Array<{ message: string; location?: string; reason?: string }> +) { + // BigQuery always sets a reason on these entries, and classification reads it. + const entries = fieldErrors.map((fieldError) => ({ + reason: "invalid", + ...fieldError, + })); + + const e: any = new Error("insert failed"); + e.name = "PartialFailureError"; + e.errors = [ + { + row: {}, + errors: entries.map(({ message, reason }) => ({ message, reason })), + }, + ]; + e.response = { + kind: "bigquery#tableDataInsertAllResponse", + insertErrors: [{ index: 0, errors: entries }], + }; + return e; +} + +/** An error with no partial-failure body, e.g. a network or quota failure. */ +function transportFailure() { + const e: any = new Error("ECONNRESET"); + e.code = "ECONNRESET"; + return e; +} + +/** + * Carries every column the allowlist can name, so that asserting one was removed + * cannot pass because the key was never there. + */ +const ROWS = [ + { + insertId: "e1", + json: { + event_id: "e1", + data: "{}", + document_id: "d1", + old_data: null, + path_params: "{}", + created_at: "2026-01-01 00:00:00", + }, + }, +]; + +/** The row payload of the nth `insert` call, 0-indexed. */ +const payloadOf = (insert: jest.Mock, call: number) => + insert.mock.calls[call][0][0].json; + +/** A tracker whose inserts are served by `insert`, so no client is needed. */ +function trackerWith( + insert: jest.Mock, + overrides?: Partial +) { + const tracker = new FirestoreBigQueryEventHistoryTracker(config(overrides)); + + jest.spyOn(tracker as any, "bigqueryDataset").mockReturnValue({ + table: () => ({ insert }), + }); + + return tracker; +} + +/** Invokes the private insert path directly. */ +const insertData = (tracker: FirestoreBigQueryEventHistoryTracker) => + (tracker as any).insertData(ROWS); + +describe("insertData retry behaviour", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + describe("a column we just added is not streamable yet", () => { + it("retries once without the rejected column, and succeeds", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(payloadOf(insert, 0)).toHaveProperty("document_id"); + expect(payloadOf(insert, 1)).not.toHaveProperty("document_id"); + expect(payloadOf(insert, 1)).toMatchObject({ + event_id: "e1", + data: "{}", + }); + expect(insert.mock.calls[1][1]).toMatchObject({ + ignoreUnknownValues: false, + }); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("matches the inlined message form that omits location", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field: path_params." }]) + ) + .mockResolvedValueOnce(undefined); + + await expect( + insertData(trackerWith(insert, { wildcardIds: true })) + ).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(payloadOf(insert, 1)).not.toHaveProperty("path_params"); + }); + + it("does not ignore an unknown field BigQuery did not name", async () => { + // BigQuery names one unknown field per row, so a stray key alongside a + // lagging column only surfaces on the retry. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockRejectedValueOnce( + partialFailure([{ message: "no such field.", location: "injected" }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(2); + expect(payloadOf(insert, 1)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("strips one column per retry when BigQuery names them one at a time", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field.", location: "old_data" }]) + ) + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(3); + expect(payloadOf(insert, 1)).not.toHaveProperty("old_data"); + expect(payloadOf(insert, 2)).not.toHaveProperty("old_data"); + expect(payloadOf(insert, 2)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("ignores stopped rows when recognising the lag", async () => { + // With `skipInvalidRows` false BigQuery marks the rows it did not attempt + // as `stopped`, with the empty message and location a live instance sends. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "", location: "", reason: "stopped" }, + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(payloadOf(insert, 1)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("names a column once however many rows rejected it", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + const warn = jest + .spyOn(logger, "warn") + .mockImplementation(() => undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + const messages = warn.mock.calls.map(([message]) => String(message)); + warn.mockRestore(); + + expect( + messages.some((m) => m.includes("without document_id, document_id")) + ).toBe(false); + expect(messages.some((m) => m.includes("without document_id"))).toBe( + true + ); + }); + + it("gives up when a retry makes no progress", async () => { + // Bounds the recursion: the same column twice means removing it did not help. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(2); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("backs up and throws when the retry also fails", async () => { + const error = partialFailure([ + { message: "no such field.", location: "path_params" }, + ]); + const insert = jest.fn().mockRejectedValue(error); + const tracker = trackerWith(insert, { wildcardIds: true }); + + // Must start true, or asserting false below passes against an + // implementation that never clears the flag. + tracker._initialized = true; + + await expect(insertData(tracker)).rejects.toThrow("insert failed"); + + expect(insert).toHaveBeenCalledTimes(2); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + expect(tracker._initialized).toBe(false); + }); + + it("backs up the row the caller gave us, not the one the retry reduced", async () => { + // A strip followed by a terminal rejection for a different column is the + // ordinary case, and the backup is the only record of the row that is left. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field: document_id." }]) + ) + .mockRejectedValueOnce( + partialFailure([{ message: "no such field: injected_col." }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + // Without this, the assertion below passes even if nothing was stripped. + expect(payloadOf(insert, 1)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).toHaveBeenCalledWith( + ROWS, + expect.anything(), + expect.anything() + ); + }); + + it("clears initialization so a column that is really gone comes back", async () => { + // Nothing here can tell a lagging column from one that was actually + // dropped, so a warm instance must not strip it for its whole life. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field: old_data." }]) + ) + .mockResolvedValueOnce(undefined); + + const tracker = trackerWith(insert); + tracker._initialized = true; + + await expect(insertData(tracker)).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(tracker._initialized).toBe(false); + }); + + it("does not match a column that merely contains an allowlisted name", async () => { + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field: document_id_v2." }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + }); + + // `path_params` needs its own config: the column is only added, and the key + // only emitted, when wildcard ids are enabled. + const addedColumns: Array<[string, Partial]> = [ + ["document_id", {}], + ["old_data", {}], + ["path_params", { wildcardIds: true }], + ]; + + it.each(addedColumns)( + "covers %s, every column added to an existing table", + async (column, overrides) => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field.", location: column }]) + ) + .mockResolvedValueOnce(undefined); + + await expect( + insertData(trackerWith(insert, overrides)) + ).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(payloadOf(insert, 1)).not.toHaveProperty(column); + } + ); + + it("does not allowlist path_params when wildcard ids are disabled", async () => { + // The column is never created here, but a transform function can still + // inject the key, since `transformRows` uses its response verbatim. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([ + { message: "no such field.", location: "path_params" }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + }); + + describe("schema drift we did not add", () => { + it("does not retry, and does not silently drop the field", async () => { + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "user_age" }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("does not retry when only some rejected fields are ours", async () => { + const insert = jest.fn().mockRejectedValue( + partialFailure([ + { message: "no such field.", location: "document_id" }, + { message: "no such field.", location: "user_age" }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + }); + + it("does not retry a rejection that is not an unknown field", async () => { + const insert = jest.fn().mockRejectedValue( + partialFailure([ + { + message: "Cannot convert value to timestamp.", + location: "timestamp", + }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + }); + }); + + describe("the user-configured partition column", () => { + const partitioned = { + partitioning: { + granularity: "HOUR", + bigqueryColumnName: "created_at", + bigqueryColumnType: "TIMESTAMP", + firestoreFieldName: "createdAt", + }, + } as Partial; + + it("is not allowlisted, even under the strategy that adds it", async () => { + // `tableRequiresUpdate` is false for a table that is already + // time-partitioned, so on exactly that table the column is never added. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([ + { message: "no such field.", location: "created_at" }, + ]) + ); + + await expect( + insertData(trackerWith(insert, partitioned)) + ).rejects.toThrow("insert failed"); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("is not allowlisted when no partitioning is configured", async () => { + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([ + { message: "no such field.", location: "created_at" }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + }); + + it("is not allowlisted under the Firestore timestamp strategy", async () => { + // Excluded deliberately, not for want of a code path: `timestamp` keys the + // partition, so a null misfiles the row rather than costing an event. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "timestamp" }]) + ); + + await expect( + insertData( + trackerWith(insert, { + partitioning: { + granularity: "DAY", + bigqueryColumnName: "timestamp", + }, + } as Partial) + ) + ).rejects.toThrow("insert failed"); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("is not allowlisted when field partitioning names a base column", async () => { + // Keyed on the collision, not on `timestamp`, so any base-column name hits it. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "data" }]) + ); + + await expect( + insertData( + trackerWith(insert, { + partitioning: { + granularity: "DAY", + bigqueryColumnName: "data", + bigqueryColumnType: "TIMESTAMP", + firestoreFieldName: "someField", + }, + } as Partial) + ) + ).rejects.toThrow("insert failed"); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + }); + + describe("a transient blip followed by a schema lag", () => { + it("can still retry the schema lag", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce(transportFailure()) + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(3); + expect(payloadOf(insert, 1)).toHaveProperty("document_id"); + expect(payloadOf(insert, 2)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("spends the transient retry at most once", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce(transportFailure()) + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockRejectedValue(transportFailure()); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "ECONNRESET" + ); + + expect(insert).toHaveBeenCalledTimes(3); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + }); + + describe("a schema lag followed by a transient blip", () => { + it("can still retry the blip, and keeps the column stripped", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockRejectedValueOnce(transportFailure()) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(3); + expect(payloadOf(insert, 0)).toHaveProperty("document_id"); + expect(payloadOf(insert, 1)).not.toHaveProperty("document_id"); + expect(payloadOf(insert, 2)).not.toHaveProperty("document_id"); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + }); + + describe("malformed failures", () => { + it("survives a null entry in the errors array", async () => { + // Not producible by the current library, but classifying runs inside the + // catch block, where a throw loses the real error and skips the backup. + const insert = jest.fn().mockRejectedValue({ + message: "insert failed", + response: { insertErrors: [{ index: 0, errors: [null] }] }, + }); + + await expect(insertData(trackerWith(insert))).rejects.toMatchObject({ + message: "insert failed", + }); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("survives a non-object thrown value", async () => { + const insert = jest.fn().mockRejectedValue(undefined); + + await expect(insertData(trackerWith(insert))).rejects.toBeUndefined(); + + // Only shows the backup was reached, since the module is mocked here. That + // it writes a row for a non-Error is pinned in backupSettings.test.ts. + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("still reports the insert error when error logging hits a bad entry", async () => { + // `e.errors` is the remapped copy logged on the terminal path. + const error: any = new Error("insert failed"); + error.errors = [null]; + error.response = { + insertErrors: [ + { index: 0, errors: [{ message: "no such field.", location: "x" }] }, + ], + }; + + const insert = jest.fn().mockRejectedValue(error); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("still reports the insert error when errors is not an array", async () => { + const error: any = new Error("insert failed"); + error.errors = { nested: "not an array" }; + + const insert = jest.fn().mockRejectedValue(error); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + }); + }); + + describe("transient failures", () => { + it("retries a partial failure whose reasons are all retryable", async () => { + // A rate limit or backend error arrives as a partial failure, not as a + // bare transport error. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "Backend error.", reason: "backendError" }, + { message: "Row skipped.", reason: "stopped" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(insert.mock.calls[1][1]).toMatchObject({ + ignoreUnknownValues: false, + }); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("does not retry when any reason is not retryable", async () => { + const insert = jest.fn().mockRejectedValue( + partialFailure([ + { message: "Backend error.", reason: "backendError" }, + { message: "Cannot convert value.", reason: "invalid" }, + ]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("does not retry a partial failure with no reason to judge", async () => { + // Fails closed: an entry we cannot classify is not evidence of a blip. + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "", reason: undefined }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(insert).toHaveBeenCalledTimes(1); + }); + + it("retries once with options unchanged", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce(transportFailure()) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect(insert).toHaveBeenCalledTimes(2); + expect(insert.mock.calls[1][1]).toMatchObject({ + ignoreUnknownValues: false, + }); + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("backs up and throws when the retry also fails", async () => { + const insert = jest.fn().mockRejectedValue(transportFailure()); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "ECONNRESET" + ); + + expect(insert).toHaveBeenCalledTimes(2); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + }); + + describe("backup collection", () => { + it("is skipped when no backupTableId is configured", async () => { + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "user_age" }]) + ); + + await expect( + insertData(trackerWith(insert, { backupTableId: undefined })) + ).rejects.toThrow("insert failed"); + + expect(handleFailedTransactionsMock).not.toHaveBeenCalled(); + }); + + it("does not let its own failure mask the insert error", async () => { + handleFailedTransactionsMock.mockRejectedValueOnce( + new Error("firestore batch failed") + ); + + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "user_age" }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + }); + + it("is used for a terminal failure on the first attempt", async () => { + const insert = jest + .fn() + .mockRejectedValue( + partialFailure([{ message: "no such field.", location: "user_age" }]) + ); + + await expect(insertData(trackerWith(insert))).rejects.toThrow( + "insert failed" + ); + + // This failure never reaches a second attempt, so a backup condition keyed + // on "this is the second attempt" would skip it. + expect(insert).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledTimes(1); + expect(handleFailedTransactionsMock).toHaveBeenCalledWith( + ROWS, + expect.objectContaining({ backupTableId: "backup" }), + expect.any(Error) + ); + }); + }); + + describe("retry logging", () => { + let debug: jest.SpyInstance; + let warn: jest.SpyInstance; + + beforeEach(() => { + debug = jest.spyOn(logger, "debug").mockImplementation(() => undefined); + warn = jest.spyOn(logger, "warn").mockImplementation(() => undefined); + }); + + afterEach(() => { + debug.mockRestore(); + warn.mockRestore(); + }); + + it("warns rather than debugs when a retry drops columns", async () => { + // Debug is suppressed at the default log level. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + expect( + warn.mock.calls.filter(([message]) => + String(message).includes("without document_id") + ) + ).toHaveLength(1); + expect( + debug.mock.calls.filter(([message]) => + String(message).includes("without document_id") + ) + ).toHaveLength(0); + }); + + it("names the columns it dropped", async () => { + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([{ message: "no such field.", location: "old_data" }]) + ) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + const messages = warn.mock.calls.map(([message]) => String(message)); + + expect(messages.some((m) => m.includes("without old_data"))).toBe(true); + }); + + it("distinguishes the retry that drops columns from the one that does not", async () => { + // Only one of the two retries drops columns, and the logs are all an + // operator has to tell them apart. + const insert = jest + .fn() + .mockRejectedValueOnce( + partialFailure([ + { message: "no such field.", location: "document_id" }, + ]) + ) + .mockRejectedValueOnce(transportFailure()) + .mockResolvedValueOnce(undefined); + + await expect(insertData(trackerWith(insert))).resolves.toBeUndefined(); + + const messages = [...debug.mock.calls, ...warn.mock.calls].map( + ([message]) => String(message) + ); + const dropped = messages.filter((message) => + message.includes("without document_id") + ); + + expect(dropped).toHaveLength(1); + expect(dropped[0]).toContain(`${ROWS.length} row(s)`); + expect( + messages.filter((message) => message.includes("transient")) + ).toHaveLength(1); + }); + }); +}); diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/handleFailedTransactions.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/handleFailedTransactions.ts index 93e10538f..2ad56c6a9 100644 --- a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/handleFailedTransactions.ts +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/handleFailedTransactions.ts @@ -23,15 +23,126 @@ if (!admin.apps.length) { initializeApp(); } +/** `settings()` may only be called once per instance, and before it is used. */ +const settingsApplied = new Set(); + +function backupFirestore(instanceId: string) { + const db = getFirestore(instanceId); + + if (!settingsApplied.has(instanceId)) { + settingsApplied.add(instanceId); + + try { + db.settings({ ignoreUndefinedProperties: true }); + } catch (settingsError) { + // Something else reached this instance first. The backup still goes + // ahead, without `ignoreUndefinedProperties`. + } + } + + return db; +} + +/** Distinct messages recorded before the rest are counted instead. */ +const MAX_ERROR_MESSAGES = 5; + +/** Cap, so one bad insert cannot write an unbounded Firestore field. */ +const MAX_ERROR_DETAILS_LENGTH = 1000; + +function truncate( + value: string, + limit: number = MAX_ERROR_DETAILS_LENGTH +): string { + return value.length > limit ? `${value.slice(0, limit - 3)}...` : value; +} + +/** + * The per-field messages a `PartialFailureError` nests under + * `errors[].errors[].message`, deduplicated because a batch usually fails the + * same way for every row. + */ +function nestedErrorMessages(e: unknown): string { + const groups = (e as any)?.errors; + + if (!Array.isArray(groups)) return ""; + + const messages = new Set(); + + for (const group of groups) { + const inner = (group as any)?.errors; + const entries = Array.isArray(inner) ? inner : [group]; + + for (const entry of entries) { + const message = (entry as any)?.message; + + if (typeof message === "string" && message.length > 0) { + messages.add(message); + continue; + } + + // A `stopped` entry, the row BigQuery did not attempt, arrives with an + // empty `message` and `location`, so `reason` is all there is. + const reason = (entry as any)?.reason; + + if (typeof reason === "string" && reason.length > 0) { + messages.add(reason); + } + } + } + + if (messages.size === 0) return ""; + + const all = Array.from(messages); + const shown = all.slice(0, MAX_ERROR_MESSAGES); + const remaining = all.length - shown.length; + const suffix = remaining > 0 ? ` (+${remaining} more)` : ""; + + // Truncating the messages rather than the finished string, so the count is + // not the part that gets cut off. + return `${truncate( + shown.join("; "), + MAX_ERROR_DETAILS_LENGTH - suffix.length + )}${suffix}`; +} + +/** + * `insertData` reports whatever it caught, so this is not always an Error, and on + * the common failure its `message` is empty: `@google-cloud/common` builds a + * `PartialFailureError`'s message from entries that carry none, so the real + * reason sits one level down. + */ +function describeError(e: unknown): string { + // Runs inside the caller's catch block, where a throw is reported as a failed + // backup, so the whole body is guarded rather than just `String(e)`. + try { + const message = (e as any)?.message; + + if (typeof message === "string" && message.length > 0) { + return truncate(message); + } + + // Already capped, so it is not truncated a second time here. + const nested = nestedErrorMessages(e); + + if (nested.length > 0) return nested; + + return truncate(String(e)); + } catch (describeFailure) { + // A value whose `toString` or `message` throws, or an object with a null + // prototype. + return "Unknown error"; + } +} + export default async ( rows: any[], config: ChangeTrackerConfig, e: Error ): Promise => { - const db = getFirestore(config.firestoreInstanceId!); - db.settings({ - ignoreUndefinedProperties: true, - }); + const db = backupFirestore(config.firestoreInstanceId!); + + const errorDetails = describeError(e); + const batchArray = [db.batch()]; let operationCounter = 0; @@ -42,7 +153,7 @@ export default async ( batchArray[batchIndex].set(ref, { ...row, - error_details: e.message, + error_details: errorDetails, }); operationCounter++; diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/index.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/index.ts index edc7052ba..f24d9e1f4 100644 --- a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/index.ts +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/bigquery/index.ts @@ -47,6 +47,81 @@ import type { ChangeTrackerConfig } from "./types"; import { PartitioningConfig } from "./partitioning/config"; export type { ChangeTrackerConfig } from "./types"; +interface InsertAllError { + message?: string; + location?: string; + reason?: string; +} + +/** + * The `insertAll` error reasons BigQuery documents as worth retrying. `stopped` + * marks a row BigQuery did not attempt, and never appears without one of the + * others alongside it. + */ +const RETRYABLE_INSERT_REASONS = [ + "backendError", + "internalError", + "rateLimitExceeded", + "timeout", + "stopped", +]; + +/** + * Read from `response.insertErrors` rather than the error's own `errors`, which + * is a remapped copy that drops `location` and so cannot identify the column. + */ +function extractInsertErrors(e: any): InsertAllError[] { + const insertErrors = e?.response?.insertErrors; + + if (!Array.isArray(insertErrors)) return []; + + return insertErrors.flatMap((insertError) => + Array.isArray(insertError?.errors) ? insertError.errors : [] + ); +} + +/** + * Null for anything it cannot attribute, so an unrecognised entry fails the + * insert rather than dropping data. Two response forms exist: the bare + * `"no such field."` naming the column in `location`, and an inlined + * `"no such field: document_id."`. + */ +function unknownFieldColumn( + error: InsertAllError, + columns: string[] +): string | null { + // Runs inside a catch block, so a malformed entry must not throw. + const message = error?.message ?? ""; + + if (!/^no such field/i.test(message)) return null; + + if (error.location) { + return columns.includes(error.location) ? error.location : null; + } + + // Whole name, not a substring: `document_id_v2` must not match `document_id`. + const named = message.match(/^no such field:\s*(.+?)\.?$/i); + + return named && columns.includes(named[1]) ? named[1] : null; +} + +/** Rows are inserted with `raw: true`, so the payload is under `json`. */ +function withoutColumns( + rows: bigquery.RowMetadata[], + columns: string[] +): bigquery.RowMetadata[] { + if (!columns.length) return rows; + + return rows.map((row) => { + if (!row?.json) return row; + + const json = { ...row.json }; + columns.forEach((column) => delete json[column]); + + return { ...row, json }; + }); +} + /** * An FirestoreEventHistoryTracker that exports data to BigQuery. * @@ -154,39 +229,77 @@ export class FirestoreBigQueryEventHistoryTracker } /** - * Check whether a failed operation is retryable or not. - * Reasons for retrying: - * 1) We added a new column to our schema. Sometimes BQ is not ready to stream insertion records immediately - * after adding a new column to an existing table (https://issuetracker.google.com/35905247) + * The rejected columns when an insert failure is the one case it is safe to + * retry: a column added to an existing table that BigQuery is not ready to + * stream into yet (https://issuetracker.google.com/35905247). Any other unknown + * field is real drift, and dropping it would lose the user's data. + * + * Not `async`: the result is used in a guard, where a promise is always truthy. */ - private async isRetryableInsertionError(e) { - let isRetryable = true; - const expectedErrors = [ - { message: "no such field.", location: documentIdField.name }, - { message: "no such field.", location: documentPathParams.name }, - ]; - if ( - e.response && - e.response.insertErrors && - e.response.insertErrors.errors - ) { - const errors = e.response.insertErrors.errors; - errors?.forEach((error) => { - let isExpected = false; - expectedErrors?.forEach((expectedError) => { - if ( - error.message === expectedError.message && - error.location === expectedError.location - ) { - isExpected = true; - } - }); - if (!isExpected) { - isRetryable = false; - } - }); + private schemaLagColumns(e: any): string[] { + const errors = extractInsertErrors(e); + + // Without per-field detail we cannot show the retry is safe. + if (!errors.length) return []; + + const addedColumns = this.columnsAddedToExistingTables(); + const rejected: string[] = []; + + for (const error of errors) { + // A row BigQuery did not attempt because another in the request failed. + // Skipping it is what lets a multi-row batch be recognised as lag at all. + if (error?.reason === "stopped") continue; + + const column = unknownFieldColumn(error, addedColumns); + + if (!column) return []; + + rejected.push(column); } - return isRetryable; + + // One entry per row, so the same column appears once per rejected row. + return [...new Set(rejected)]; + } + + /** + * The only columns a lag retry may drop. Exact in both directions: one missing + * costs the event once the caller's retries run out, and one listed that is + * never actually added is stripped from every insert for the table's life. + * + * A null in a column the latest view groups on duplicates the document in + * `_latest` permanently. Accepted for these, whose tables already hold + * pre-upgrade rows with the same nulls; not for `timestamp`, where a null + * misfiles the row instead. + */ + private columnsAddedToExistingTables(): string[] { + const columns = [documentIdField.name, oldDataField.name]; + + // Only added, and only emitted, when wildcard ids are on. A transform + // function can inject the key regardless: `transformRows` uses its response + // verbatim. + if (this.config.wildcardIds) { + columns.push(documentPathParams.name); + } + + // The custom partition column is absent even though the Firestore field + // strategy adds it, because `tableRequiresUpdate` is false for an already + // time-partitioned table, so on exactly those tables it is never added. + return columns; + } + + /** + * Qualifies only when every entry names a reason BigQuery documents as + * retryable; anything else is a rejection of the data, which a retry cannot + * fix. A failure with no partial-failure body says nothing about the schema. + */ + private isTransientInsertionError(e: any): boolean { + const errors = extractInsertErrors(e); + + if (!errors.length) return true; + + return errors.every((error) => + RETRYABLE_INSERT_REASONS.includes(error?.reason) + ); } /** @@ -213,11 +326,18 @@ export class FirestoreBigQueryEventHistoryTracker /** * Inserts rows of data into the BigQuery raw change log table. + * + * Columns are removed at the `insert` call rather than from `rows`, so the + * backup on the terminal path still holds every column. */ private async insertData( rows: bigquery.RowMetadata[], overrideOptions: InsertRowsOptions = {}, - retry: boolean = true + // Columns a schema-lag retry has already removed from the payload. Each + // retry must remove one not removed before, which bounds the recursion. + strippedColumns: string[] = [], + // Tracked separately, so a blip cannot consume the retry a later lag needs. + allowTransientRetry: boolean = true ) { const options = { skipInvalidRows: false, @@ -230,27 +350,51 @@ export class FirestoreBigQueryEventHistoryTracker const table = dataset.table(this.rawChangeLogTableName()); logs.dataInserting(rows.length); - await table.insert(rows, options); + await table.insert(withoutColumns(rows, strippedColumns), options); logs.dataInserted(rows.length); } catch (e) { - if (retry && this.isRetryableInsertionError(e)) { - retry = false; - logs.dataInsertRetried(rows.length); + // A column we just added may not be streamable yet, so remove the ones + // BigQuery named and retry. + // + // Not `ignoreUnknownValues`: BigQuery reports one unknown field per row, so + // that would also discard fields it never mentioned. + const lagColumns = this.schemaLagColumns(e).filter( + (column) => !strippedColumns.includes(column) + ); + + if (lagColumns.length) { + logs.dataInsertRetriedWithoutColumns(rows.length, lagColumns); + + // If the column is genuinely gone rather than lagging, this makes the + // next batch re-run `initialize` and add it back. + this._initialized = false; + return this.insertData( rows, - { ...overrideOptions, ignoreUnknownValues: true }, - retry + overrideOptions, + [...strippedColumns, ...lagColumns], + allowTransientRetry ); } - // Exceeded number of retries, save in failed collection - if (!retry && this.config.backupTableId) { - await handleFailedTransactions(rows, this.config, e); + if (allowTransientRetry && this.isTransientInsertionError(e)) { + logs.dataInsertRetriedAfterTransientError(rows.length); + return this.insertData(rows, overrideOptions, strippedColumns, false); + } + + // Terminal: no further attempt will be made for these rows. + if (this.config.backupTableId) { + try { + await handleFailedTransactions(rows, this.config, e); + } catch (backupError) { + // A failed backup must not replace the insert error the caller needs. + logs.failedBackupWrite(backupError); + } } // Reinitializing in case the destintation table is modified. this._initialized = false; - logs.bigQueryTableInsertErrors(e.errors); + logs.bigQueryTableInsertErrors(e?.errors); throw e; } } diff --git a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/logs.ts b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/logs.ts index 9b1b2f658..1930bdc7d 100644 --- a/firestore-bigquery-export/firestore-bigquery-change-tracker/src/logs.ts +++ b/firestore-bigquery-export/firestore-bigquery-change-tracker/src/logs.ts @@ -137,9 +137,24 @@ export const dataInserted = (rowCount: number) => { logger.debug(`Inserted ${rowCount} row(s) of data into BigQuery`); }; -export const dataInsertRetried = (rowCount: number) => { +/** + * Warn rather than debug, and name the columns: this is the one retry path that + * leaves a column permanently null, and debug is off at the default log level. + */ +export const dataInsertRetriedWithoutColumns = ( + rowCount: number, + columns: string[] +) => { + logger.warn( + `Retrying insert of ${rowCount} row(s) of data into BigQuery without ${columns.join( + ", " + )}. BigQuery does not have those columns yet, so they will be null for these rows.` + ); +}; + +export const dataInsertRetriedAfterTransientError = (rowCount: number) => { logger.debug( - `Retried to insert ${rowCount} row(s) of data into BigQuery (ignoring unknown columns)` + `Retrying insert of ${rowCount} row(s) of data into BigQuery after a transient failure, with options unchanged` ); }; @@ -217,16 +232,28 @@ export const bigQueryTableInsertErrors = ( ) => { logger.warn(`Error when inserting data to table.`); - insertErrors?.forEach((error) => { + // Runs on the terminal path of a failed insert, so a throw here would replace + // the insert error the caller needs. + if (!Array.isArray(insertErrors)) return; + + insertErrors.forEach((error) => { logger.warn("ROW DATA JSON:"); - logger.warn(error.row); + logger.warn(error?.row); + + if (!Array.isArray(error?.errors)) return; - error.errors?.forEach((error) => - logger.warn(`ROW ERROR MESSAGE: ${error.message}`) + error.errors.forEach((error) => + logger.warn(`ROW ERROR MESSAGE: ${error?.message}`) ); }); }; +export const failedBackupWrite = (error: unknown) => { + logger.warn( + `Could not write failed rows to the backup collection. The original insert error is still thrown. Backup error: ${error}` + ); +}; + export const updatedClustering = (fields: string) => { logger.info(`Clustering updated with new settings fields: ${fields}`); };