Experimente / superlog /apps /proxy /src /ingest-queue.stream.test.ts

Initial public upload from Projekte/Experimente

95b671e verified about 1 month ago

4.83 kB

import { strict as assert } from "node:assert";
import { test } from "node:test";
import { PayloadTooLargeError, type SpillSink } from "./body-capture.js";
import { IngestQueue, type IngestQueueConfig, type SpillSinkFactory } from "./ingest-queue.js";

class FakeSqs {
  sent: string[] = [];
  // biome-ignore lint/suspicious/noExplicitAny: test double for the AWS client surface
  async send(cmd: any): Promise<unknown> {
    if (cmd.constructor.name !== "SendMessageCommand") {
      throw new Error(`unexpected SQS command: ${cmd.constructor.name}`);
    }
    this.sent.push(cmd.input.MessageBody);
    return { MessageId: "m-1" };
  }
}

class RecordingSink implements SpillSink {
  written: Buffer[] = [];
  finished = false;
  aborted = false;
  async write(chunk: Uint8Array): Promise<void> {
    this.written.push(Buffer.from(chunk));
  }
  async finish(): Promise<void> {
    this.finished = true;
  }
  async abort(): Promise<void> {
    this.aborted = true;
  }
  get bytes(): number {
    return this.written.reduce((n, b) => n + b.length, 0);
  }
}

const noopLogger = { info: () => {}, warn: () => {}, error: () => {} };

function buildQueue(overrides: Partial<IngestQueueConfig> = {}): {
  queue: IngestQueue;
  sqs: FakeSqs;
  sinks: RecordingSink[];
} {
  const config: IngestQueueConfig = {
    queueUrl: "http://localhost/queue",
    region: "us-west-2",
    oversizeBucket: "test-bucket",
    oversizePrefix: "otlp-oversize",
    maxMessageBytes: 4000, // inline raw threshold ≈ floor((4000-1024)*3/4) = 2232 bytes
    maxBodyBytes: 10_000,
    consumerEnabled: false,
    waitTimeSeconds: 20,
    visibilityTimeoutSeconds: 120,
    batchSize: 10,
    consumerConcurrency: 4,
    ...overrides,
  };
  const sinks: RecordingSink[] = [];
  const spillSinkFactory: SpillSinkFactory = () => {
    const sink = new RecordingSink();
    sinks.push(sink);
    return sink;
  };
  const queue = new IngestQueue(config, noopLogger, spillSinkFactory);
  const sqs = new FakeSqs();
  (queue as unknown as { sqs: FakeSqs }).sqs = sqs;
  return { queue, sqs, sinks };
}

async function* streamOf(...chunks: string[]): AsyncIterable<Uint8Array> {
  for (const chunk of chunks) yield Buffer.from(chunk);
}

test("small body is buffered and enqueued inline, never opening a spill sink", async () => {
  const { queue, sqs, sinks } = buildQueue();
  const result = await queue.enqueueStream({
    path: "/v1/traces",
    projectId: "proj-1",
    contentType: "application/x-protobuf",
    body: streamOf("hello world"),
  });

  assert.equal(result.storage, "inline");
  assert.equal(result.bytes, 11);
  assert.equal(sinks.length, 0, "small body must not spill");
  assert.equal(sqs.sent.length, 1);
  const sent = sqs.sent[0];
  assert.ok(sent);
  const msg = JSON.parse(sent);
  assert.equal(msg.body.storage, "inline");
  assert.equal(Buffer.from(msg.body.base64, "base64").toString(), "hello world");
});

test("large body streams to the spill sink and enqueues an s3 pointer with the byte count", async () => {
  const { queue, sqs, sinks } = buildQueue();
  // 3000 raw bytes > the ~2232 inline threshold → spills.
  const chunk = "X".repeat(1000);
  const result = await queue.enqueueStream({
    path: "/v1/traces",
    projectId: "proj-1",
    contentType: "application/x-protobuf",
    body: streamOf(chunk, chunk, chunk),
  });

  assert.equal(result.storage, "s3");
  assert.equal(result.bytes, 3000);
  assert.equal(sinks.length, 1, "exactly one spill sink opened");
  const sink = sinks[0];
  assert.ok(sink);
  assert.equal(sink.bytes, 3000, "every byte streamed through the sink intact");
  assert.equal(sink.finished, true);
  assert.equal(sink.aborted, false);

  assert.equal(sqs.sent.length, 1);
  const sent = sqs.sent[0];
  assert.ok(sent);
  const msg = JSON.parse(sent);
  assert.equal(msg.body.storage, "s3");
  assert.equal(msg.body.bucket, "test-bucket");
  assert.equal(msg.body.sizeBytes, 3000);
  assert.match(msg.body.key, /^otlp-oversize\/\d{4}\/\d{2}\/\d{2}\/.+\.otlp$/);
});

test("a body over maxBodyBytes is rejected with PayloadTooLargeError and enqueues nothing", async () => {
  // maxBodyBytes (2500) is below the inline threshold (~2232 from maxMessageBytes
  // 4000), so the cap trips while still buffering. The spill-then-abort path is
  // covered separately by the body-capture unit test; here we assert the invariant
  // that an over-cap body never reaches the queue.
  const { queue, sqs } = buildQueue({ maxBodyBytes: 2500 });
  await assert.rejects(
    queue.enqueueStream({
      path: "/v1/traces",
      projectId: "proj-1",
      contentType: "application/x-protobuf",
      body: streamOf("Y".repeat(1500), "Y".repeat(1500)), // 3000 > 2500
    }),
    PayloadTooLargeError,
  );
  assert.equal(sqs.sent.length, 0, "over-cap body must never reach the queue");
});

	import { strict as assert } from "node:assert";
	import { test } from "node:test";
	import { PayloadTooLargeError, type SpillSink } from "./body-capture.js";
	import { IngestQueue, type IngestQueueConfig, type SpillSinkFactory } from "./ingest-queue.js";

	class FakeSqs {
	sent: string[] = [];
	// biome-ignore lint/suspicious/noExplicitAny: test double for the AWS client surface
	async send(cmd: any): Promise<unknown> {
	if (cmd.constructor.name !== "SendMessageCommand") {
	throw new Error(`unexpected SQS command: ${cmd.constructor.name}`);
	}
	this.sent.push(cmd.input.MessageBody);
	return { MessageId: "m-1" };
	}
	}

	class RecordingSink implements SpillSink {
	written: Buffer[] = [];
	finished = false;
	aborted = false;
	async write(chunk: Uint8Array): Promise<void> {
	this.written.push(Buffer.from(chunk));
	}
	async finish(): Promise<void> {
	this.finished = true;
	}
	async abort(): Promise<void> {
	this.aborted = true;
	}
	get bytes(): number {
	return this.written.reduce((n, b) => n + b.length, 0);
	}
	}

	const noopLogger = { info: () => {}, warn: () => {}, error: () => {} };

	function buildQueue(overrides: Partial<IngestQueueConfig> = {}): {
	queue: IngestQueue;
	sqs: FakeSqs;
	sinks: RecordingSink[];
	} {
	const config: IngestQueueConfig = {
	queueUrl: "http://localhost/queue",
	region: "us-west-2",
	oversizeBucket: "test-bucket",
	oversizePrefix: "otlp-oversize",
	maxMessageBytes: 4000, // inline raw threshold ≈ floor((4000-1024)*3/4) = 2232 bytes
	maxBodyBytes: 10_000,
	consumerEnabled: false,
	waitTimeSeconds: 20,
	visibilityTimeoutSeconds: 120,
	batchSize: 10,
	consumerConcurrency: 4,
	...overrides,
	};
	const sinks: RecordingSink[] = [];
	const spillSinkFactory: SpillSinkFactory = () => {
	const sink = new RecordingSink();
	sinks.push(sink);
	return sink;
	};
	const queue = new IngestQueue(config, noopLogger, spillSinkFactory);
	const sqs = new FakeSqs();
	(queue as unknown as { sqs: FakeSqs }).sqs = sqs;
	return { queue, sqs, sinks };
	}

	async function* streamOf(...chunks: string[]): AsyncIterable<Uint8Array> {
	for (const chunk of chunks) yield Buffer.from(chunk);
	}

	test("small body is buffered and enqueued inline, never opening a spill sink", async () => {
	const { queue, sqs, sinks } = buildQueue();
	const result = await queue.enqueueStream({
	path: "/v1/traces",
	projectId: "proj-1",
	contentType: "application/x-protobuf",
	body: streamOf("hello world"),
	});

	assert.equal(result.storage, "inline");
	assert.equal(result.bytes, 11);
	assert.equal(sinks.length, 0, "small body must not spill");
	assert.equal(sqs.sent.length, 1);
	const sent = sqs.sent[0];
	assert.ok(sent);
	const msg = JSON.parse(sent);
	assert.equal(msg.body.storage, "inline");
	assert.equal(Buffer.from(msg.body.base64, "base64").toString(), "hello world");
	});

	test("large body streams to the spill sink and enqueues an s3 pointer with the byte count", async () => {
	const { queue, sqs, sinks } = buildQueue();
	// 3000 raw bytes > the ~2232 inline threshold → spills.
	const chunk = "X".repeat(1000);
	const result = await queue.enqueueStream({
	path: "/v1/traces",
	projectId: "proj-1",
	contentType: "application/x-protobuf",
	body: streamOf(chunk, chunk, chunk),
	});

	assert.equal(result.storage, "s3");
	assert.equal(result.bytes, 3000);
	assert.equal(sinks.length, 1, "exactly one spill sink opened");
	const sink = sinks[0];
	assert.ok(sink);
	assert.equal(sink.bytes, 3000, "every byte streamed through the sink intact");
	assert.equal(sink.finished, true);
	assert.equal(sink.aborted, false);

	assert.equal(sqs.sent.length, 1);
	const sent = sqs.sent[0];
	assert.ok(sent);
	const msg = JSON.parse(sent);
	assert.equal(msg.body.storage, "s3");
	assert.equal(msg.body.bucket, "test-bucket");
	assert.equal(msg.body.sizeBytes, 3000);
	assert.match(msg.body.key, /^otlp-oversize\/\d{4}\/\d{2}\/\d{2}\/.+\.otlp$/);
	});

	test("a body over maxBodyBytes is rejected with PayloadTooLargeError and enqueues nothing", async () => {
	// maxBodyBytes (2500) is below the inline threshold (~2232 from maxMessageBytes
	// 4000), so the cap trips while still buffering. The spill-then-abort path is
	// covered separately by the body-capture unit test; here we assert the invariant
	// that an over-cap body never reaches the queue.
	const { queue, sqs } = buildQueue({ maxBodyBytes: 2500 });
	await assert.rejects(
	queue.enqueueStream({
	path: "/v1/traces",
	projectId: "proj-1",
	contentType: "application/x-protobuf",
	body: streamOf("Y".repeat(1500), "Y".repeat(1500)), // 3000 > 2500
	}),
	PayloadTooLargeError,
	);
	assert.equal(sqs.sent.length, 0, "over-cap body must never reach the queue");
	});