fix(stream): resend the whole combined Firehose record on partial failure

lib/stream/leo-stream.js

@@ -884,15 +884,8 @@ 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;
+							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);

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",

test/lib.stream.leo-stream.write.utest.ts

@@ -191,10 +191,27 @@ 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, 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.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 () {