foreign/node/src/wire/message/message-batch.test.ts

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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
//
//   http://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 { describe, it } from "node:test";
import assert from "node:assert/strict";
import { u128ToBuf } from "../number.utils.js";
import {
  BATCH_HEADER_SIZE,
  FRAME_HEADER_SIZE,
  serializeBatchHeader,
  deserializeBatchHeader,
} from "./iggy-header.utils.js";
import {
  encodeMessagesBatch,
  serializeSendMessages,
} from "./message.utils.js";
import {
  HeaderValue,
  HeaderKeyFactory,
  serializeHeaders,
} from "./header.utils.js";
import {
  deserializeBatchMessages,
  deserializePollMessages,
} from "./poll.utils.js";

// Golden vectors generated by the Rust implementation for stream id 1,
// topic id 2, balanced partitioning, and two messages:
// {id=7, originTimestamp=1000, payload="first-payload"} and
// {id=8, originTimestamp=1050, payload="second-payload",
// userHeaders="user-header-bytes"}.
const PRODUCE_BATCH_ONLY =
  "000000000000000000000000000000000000000000000000e803000000000000" +
  "8c01000000000000a91f38c86307267c02000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "bfd2b9205a759675070000000000000000000000000000000000000000000000" +
  "000000000d000000000000000000000066697273742d7061796c6f6164d66b7e" +
  "1c758eb7c0080000000000000000000000000000000100000032000000110000" +
  "000e00000000000000000000007365636f6e642d7061796c6f6164757365722d" +
  "6865616465722d6279746573";

const POLL_BODY =
  "0300000065000000000000000200000003000000000000006400000000000000" +
  "8813000000000000e8030000000000008c01000000000000c96826b38a8feed2" +
  "0200000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "0000000000000000000000000000000000000000000000000000000000000000" +
  "00000000000000000000000000000000bfd2b9205a7596750700000000000000" +
  "00000000000000000000000000000000000000000d0000000000000000000000" +
  "66697273742d7061796c6f6164d66b7e1c758eb7c00800000000000000000000" +
  "00000000000100000032000000110000000e0000000000000000000000736563" +
  "6f6e642d7061796c6f6164757365722d6865616465722d6279746573";

const goldenMessages = () => [
  {
    id: u128ToBuf(7n),
    payload: Buffer.from("first-payload"),
    userHeaders: Buffer.alloc(0),
    originTimestamp: 1000n,
  },
  {
    id: u128ToBuf(8n),
    payload: Buffer.from("second-payload"),
    userHeaders: Buffer.from("user-header-bytes"),
    originTimestamp: 1050n,
  },
];

describe("message batch golden vectors", () => {
  it("encodes the golden produce batch byte-for-byte", () => {
    const batch = encodeMessagesBatch(goldenMessages());
    assert.equal(batch.toString("hex"), PRODUCE_BATCH_ONLY);
  });

  it("prefixes the batch with the golden send metadata", () => {
    const body = serializeSendMessages(1, 2, [
      { id: 7n, payload: "first-payload" },
      { id: 8n, payload: "second-payload" },
    ]);
    const metadata = Buffer.from(
      "120000000104010000000104020000000100" + "02000000",
      "hex",
    );
    assert.deepEqual(body.subarray(0, metadata.length), metadata);
    assert.equal(
      body.length,
      metadata.length + BATCH_HEADER_SIZE + FRAME_HEADER_SIZE * 2 +
        "first-payload".length + "second-payload".length,
    );
  });

  it("decodes the golden poll body", () => {
    const body = Buffer.from(POLL_BODY, "hex");
    assert.equal(body.readUInt32LE(0), 3);
    assert.equal(body.readBigUInt64LE(4), 101n);
    assert.equal(body.readUInt32LE(12), 2);

    const messages = deserializeBatchMessages(body, 16);
    assert.equal(messages.length, 2);

    assert.equal(messages[0].id, 7n);
    assert.equal(messages[0].offset, 100n);
    assert.equal(messages[0].timestamp, 5000n);
    assert.equal(messages[0].originTimestamp, 1000n);
    assert.equal(messages[0].checksum, 0x7596755a20b9d2bfn);
    assert.equal(messages[0].payload.toString(), "first-payload");
    assert.equal(messages[0].userHeaders.length, 0);

    assert.equal(messages[1].id, 8n);
    assert.equal(messages[1].offset, 101n);
    assert.equal(messages[1].timestamp, 5000n);
    assert.equal(messages[1].originTimestamp, 1050n);
    assert.equal(messages[1].checksum, 0xc0b78e751c7e6bd6n);
    assert.equal(messages[1].payload.toString(), "second-payload");
    assert.equal(messages[1].userHeaders.toString(), "user-header-bytes");
  });

  it("rejects a non-zero reserved frame field", () => {
    const body = Buffer.from(POLL_BODY, "hex");
    body.writeUInt8(1, 16 + BATCH_HEADER_SIZE + 40);
    assert.throws(
      () => deserializeBatchMessages(body, 16),
      /reserved/,
    );
  });

  it("rejects an empty batch", () => {
    assert.throws(() => encodeMessagesBatch([]));
    assert.throws(() => serializeSendMessages(1, 2, []));
  });

  it("rejects a timestamp delta beyond u32", () => {
    const messages = goldenMessages();
    messages[1].originTimestamp =
      messages[0].originTimestamp + 0x1_0000_0000n;
    assert.throws(() => encodeMessagesBatch(messages), /u32/);
  });

  it("round-trips messages through the public poll decoder", () => {
    const userHeaders = [
      {
        key: HeaderKeyFactory.String("trace"),
        value: HeaderValue.String("golden"),
      },
    ];
    const batch = encodeMessagesBatch([
      {
        id: u128ToBuf(7n),
        payload: Buffer.from("first-payload"),
        userHeaders: Buffer.alloc(0),
        originTimestamp: 3_000_000n,
      },
      {
        id: u128ToBuf(8n),
        payload: Buffer.from("second-payload"),
        userHeaders: serializeHeaders(userHeaders),
        originTimestamp: 3_000_500n,
      },
    ]);
    const stamped = deserializeBatchHeader(batch);
    stamped.partitionId = 3n;
    stamped.baseOffset = 100n;
    stamped.baseTimestamp = 5_000_000n;
    serializeBatchHeader(stamped).copy(batch);

    const envelope = Buffer.alloc(16);
    envelope.writeUInt32LE(3, 0);
    envelope.writeBigUInt64LE(101n, 4);
    envelope.writeUInt32LE(2, 12);
    const response = deserializePollMessages(
      Buffer.concat([envelope, batch]),
    );

    assert.equal(response.partitionId, 3);
    assert.equal(response.currentOffset, 101n);
    assert.equal(response.count, 2);
    assert.equal(response.messages.length, 2);

    const [first, second] = response.messages;
    assert.equal(first.headers.id, 7n);
    assert.equal(first.headers.offset, 100n);
    assert.deepEqual(first.headers.timestamp, new Date(5_000));
    assert.deepEqual(first.headers.originTimestamp, new Date(3_000));
    assert.equal(first.payload.toString(), "first-payload");
    assert.deepEqual(first.userHeaders, []);

    assert.equal(second.headers.id, 8n);
    assert.equal(second.headers.offset, 101n);
    assert.deepEqual(second.headers.originTimestamp, new Date(3_000));
    assert.equal(second.payload.toString(), "second-payload");
    assert.equal(second.userHeaders.length, 1);
    assert.equal(second.userHeaders[0].key.value, "trace");
    assert.equal(second.userHeaders[0].value.value, "golden");
  });
});