From a13f77e5c787fd3a1310f907cf1483bb6f152157 Mon Sep 17 00:00:00 2001 From: Lynn Mecham Date: Thu, 15 Oct 2026 12:29:25 -0600 Subject: [PATCH 1/3] fix(stream): resend the whole combined Firehose record on partial failure The Firehose path sends one combined record, so RequestResponses has a single entry and the index-based retry filter dropped every chunk after the first. Leave records unchanged and back off so the whole record is resent. Bump to 7.1.17. Task: TASK-002 Co-Authored-By: Claude Sonnet 5.5 --- lib/stream/leo-stream.js | 10 +--------- package-lock.json | 4 ++-- package.json | 2 +- test/lib.stream.leo-stream.write.utest.ts | 8 ++++++-- 4 files changed, 10 insertions(+), 14 deletions(-) diff --git a/lib/stream/leo-stream.js b/lib/stream/leo-stream.js index d06e02d..dc8bfee 100644 --- a/lib/stream/leo-stream.js +++ b/lib/stream/leo-stream.js @@ -884,15 +884,7 @@ module.exports = function(configure) { logger.error("toLeo.firehose.putRecordBatch", err); retry.backoff(); } else if (data.FailedPutCount && data.FailedPutCount > 0) { - var left = []; - for (var i = 0; i < data.RequestResponses.length; i++) { - var row = data.RequestResponses[i]; - if (row.ErrorCode) { - left.push(records[i]); - } - } - logger.info(`Count failed firehose put record batch (failed, retry):(${data.FailedPutCount},${left.length})`); - records = left; + logger.info(`Count failed firehose put record batch (failed, retry):(${data.FailedPutCount},${records.length})`); retry.backoff(); } else { logger.timeEnd("firehose request" + loggerTimeId); diff --git a/package-lock.json b/package-lock.json index 4bb0253..1c656f5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "license": "MIT", "dependencies": { "async": "2.6.4", diff --git a/package.json b/package.json index 461a9bd..4dd464a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "description": "Load data onto the LEO Platform", "homepage": "https://leoplatform.io", "main": "index.js", diff --git a/test/lib.stream.leo-stream.write.utest.ts b/test/lib.stream.leo-stream.write.utest.ts index cd8da77..44e4c10 100644 --- a/test/lib.stream.leo-stream.write.utest.ts +++ b/test/lib.stream.leo-stream.write.utest.ts @@ -191,10 +191,14 @@ describe("lib/stream/leo-stream.js write", function () { assert.deepEqual(events.map((e) => e.payload), [{ n: 1 }, { n: 2 }]); }); - it("firehose retries only the failed records of a partial failure", async function () { + it("firehose resends the whole combined record when it is rejected", async function () { putRecordBatch.onFirstCall().callsFake((p: any, cb: any) => cb(null, { FailedPutCount: 1, RequestResponses: [{ ErrorCode: "ServiceUnavailableException" }] })); - await write([evt(1)], ls.toLeo("bot1", { ...fh, backoff: { initialDelay: 1, maxDelay: 2 } })); + await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, records: 1, backoff: { initialDelay: 1, maxDelay: 2 } })); assert.equal(putRecordBatch.callCount, 2); + const first = putRecordBatch.getCall(0).args[0].Records[0].Data; + const second = putRecordBatch.getCall(1).args[0].Records[0].Data; + assert.equal(first.toString().trim().split("\n").length, 3); + assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); }); it("firehose retries after an error", async function () { From 6a1982296e4968e2b517d5241c2d5d48e112949a Mon Sep 17 00:00:00 2001 From: Lynn Mecham Date: Thu, 15 Oct 2026 12:30:24 -0600 Subject: [PATCH 2/3] test(stream): make the Firehose resend test put multiple chunks in one record Use one-event chunks with gzipFirehose so the first putRecordBatch carries several chunk lines; the retry must resend them byte for byte. Fails against the old index-based retry. Task: TASK-002 Co-Authored-By: Claude Sonnet 5.5 --- test/lib.stream.leo-stream.write.utest.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/lib.stream.leo-stream.write.utest.ts b/test/lib.stream.leo-stream.write.utest.ts index 44e4c10..2647e42 100644 --- a/test/lib.stream.leo-stream.write.utest.ts +++ b/test/lib.stream.leo-stream.write.utest.ts @@ -193,11 +193,11 @@ describe("lib/stream/leo-stream.js write", function () { it("firehose resends the whole combined record when it is rejected", async function () { putRecordBatch.onFirstCall().callsFake((p: any, cb: any) => cb(null, { FailedPutCount: 1, RequestResponses: [{ ErrorCode: "ServiceUnavailableException" }] })); - await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, records: 1, backoff: { initialDelay: 1, maxDelay: 2 } })); + await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, gzipFirehose: true, records: 10, chunk: { records: 1 }, backoff: { initialDelay: 1, maxDelay: 2 } })); assert.equal(putRecordBatch.callCount, 2); const first = putRecordBatch.getCall(0).args[0].Records[0].Data; const second = putRecordBatch.getCall(1).args[0].Records[0].Data; - assert.equal(first.toString().trim().split("\n").length, 3); + assert.isAtLeast(first.toString().trim().split("\n").length, 2); assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); }); From 4e8d6d0ca2fb050877a5387b5f5a47c315b32d03 Mon Sep 17 00:00:00 2001 From: Lynn Mecham Date: Mon, 19 Oct 2026 13:03:53 -0600 Subject: [PATCH 3/3] fix(stream): log Firehose error code and check resent records The partial-failure log now labels its two numbers (failed Firehose records, buffered chunks) and includes the first response's ErrorCode and ErrorMessage. The gzip resend test decodes the resent record, and a new test covers the raw (gzip off) resend path. Task: TASK-002 Co-Authored-By: Claude Opus 5.5 --- lib/stream/leo-stream.js | 3 ++- test/lib.stream.leo-stream.write.utest.ts | 13 +++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/lib/stream/leo-stream.js b/lib/stream/leo-stream.js index dc8bfee..69b6c3b 100644 --- a/lib/stream/leo-stream.js +++ b/lib/stream/leo-stream.js @@ -884,7 +884,8 @@ module.exports = function(configure) { logger.error("toLeo.firehose.putRecordBatch", err); retry.backoff(); } else if (data.FailedPutCount && data.FailedPutCount > 0) { - logger.info(`Count failed firehose put record batch (failed, retry):(${data.FailedPutCount},${records.length})`); + let response = (data.RequestResponses && data.RequestResponses[0]) || {}; + logger.info(`Count failed firehose put record batch (failed records, buffered chunks):(${data.FailedPutCount},${records.length})`, response.ErrorCode, response.ErrorMessage); retry.backoff(); } else { logger.timeEnd("firehose request" + loggerTimeId); diff --git a/test/lib.stream.leo-stream.write.utest.ts b/test/lib.stream.leo-stream.write.utest.ts index 2647e42..875b8d2 100644 --- a/test/lib.stream.leo-stream.write.utest.ts +++ b/test/lib.stream.leo-stream.write.utest.ts @@ -199,6 +199,19 @@ describe("lib/stream/leo-stream.js write", function () { const second = putRecordBatch.getCall(1).args[0].Records[0].Data; assert.isAtLeast(first.toString().trim().split("\n").length, 2); assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); + const resent = second.toString().trim().split("\n").flatMap((l: string) => decodeGzipRecord(Buffer.from(l, "base64"))); + assert.deepEqual(resent.map((e) => e.payload), [{ n: 1 }, { n: 2 }, { n: 3 }]); + }); + + it("firehose resends the whole raw record when it is rejected", async function () { + putRecordBatch.onFirstCall().callsFake((p: any, cb: any) => cb(null, { FailedPutCount: 1, RequestResponses: [{ ErrorCode: "ServiceUnavailableException" }] })); + await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, records: 10, chunk: { records: 1 }, backoff: { initialDelay: 1, maxDelay: 2 } })); + assert.equal(putRecordBatch.callCount, 2); + const first = putRecordBatch.getCall(0).args[0].Records[0].Data; + const second = putRecordBatch.getCall(1).args[0].Records[0].Data; + assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); + const resent = second.toString().trim().split("\n").map((l: string) => JSON.parse(l)); + assert.deepEqual(resent.map((e: any) => e.payload), [{ n: 1 }, { n: 2 }, { n: 3 }]); }); it("firehose retries after an error", async function () {