From f5edd3b6f918f4814a8d6b6d22bb70bfdd4390d4 Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 04:05:54 +0900 Subject: [PATCH 1/2] perf(ollama): count fragmented NDJSON bytes incrementally --- scripts/test-layout/layout.json | 2 +- src/adapters/ollama-native.ts | 16 ++-- structure/transports/byte-accounting.md | 8 ++ tests/fixtures/test-layout-expected.json | 1 + .../ollama/ollama-native-scaling.test.ts | 91 +++++++++++++++++++ 5 files changed, 111 insertions(+), 7 deletions(-) create mode 100644 tests/providers/ollama/ollama-native-scaling.test.ts diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 2442a618c9c..cf35d39ba8b 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1331,7 +1331,7 @@ "ocx-launcher-source.test.ts": "cli", "ocx-run.test.ts": "cli", "ollama-native-parser.test.ts": "providers/ollama", "ollama-native-reasoning-wire.test.ts": "providers/ollama", "ollama-native-structured-output.test.ts": "providers/ollama", "ollama-native-tool-continuations.test.ts": "providers/ollama", - "ollama-native-v4.test.ts": "providers/ollama", "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama", "ollama-show-enrichment.test.ts": "providers/ollama", + "ollama-native-v4.test.ts": "providers/ollama", "ollama-native-scaling.test.ts": "providers/ollama", "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama", "ollama-show-enrichment.test.ts": "providers/ollama", "ollama-show-ignore-abort.test.ts": "providers/ollama", "opaque-blob-wrapped-rejection.test.ts": "responses", "openai-api-virtual-models.test.ts": "adapters/openai", "openai-chat-bounded-tool-names.test.ts": "adapters/openai", "openai-chat-dangling-toolcalls.test.ts": "adapters/openai", "openai-chat-developer-position.test.ts": "adapters/openai", "openai-chat-eof.test.ts": "adapters/openai", "openai-chat-hardening.test.ts": "adapters/openai", "openai-empty-assistant-history.test.ts": "adapters/openai", "openai-chat-image-normalization.test.ts": "adapters/openai", diff --git a/src/adapters/ollama-native.ts b/src/adapters/ollama-native.ts index 61ceeb6d827..e612011347d 100644 --- a/src/adapters/ollama-native.ts +++ b/src/adapters/ollama-native.ts @@ -1051,7 +1051,6 @@ async function* parseOllamaNativeStream( } const reader = response.body.getReader(); const decoder = new TextDecoder("utf-8", { fatal: true }); - const encoder = new TextEncoder(); const state = streamState(allowParallelToolCalls); const issuedIds = issuedToolCallIds ?? new Set(); let buffer = ""; @@ -1060,15 +1059,20 @@ async function* parseOllamaNativeStream( const ingest = function* (text: string): Generator { if (!text) return; + const hasNewline = text.includes("\n"); buffer += text; - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; + // After draining, the carry contains no newline. Only new text can complete a record. + const lines = hasNewline ? buffer.split("\n") : []; + if (hasNewline) buffer = lines.pop() ?? ""; // The safety bound applies to the genuinely retained residual — the incomplete NDJSON record // still being assembled — and to each complete record below. It must never depend on the // transport read size (one read may carry many valid records) nor transiently charge // old + replacement together. On a ceiling violation the old reservation is left in place so // the generator's finally releases exactly what is held. - const residualBytes = encoder.encode(buffer).byteLength; + // TextDecoder emits complete code points; fragment byte counts add without a surrogate fixup. + const residualBytes = hasNewline + ? Buffer.byteLength(buffer, "utf8") + : bufferBytes + Buffer.byteLength(text, "utf8"); if (residualBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) { throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES); } @@ -1077,7 +1081,7 @@ async function* parseOllamaNativeStream( for (const rawLine of lines) { const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; - const lineBytes = encoder.encode(line).byteLength; + const lineBytes = Buffer.byteLength(line, "utf8"); if (lineBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) { throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES); } @@ -1112,7 +1116,7 @@ async function* parseOllamaNativeStream( yield* ingest(decoder.decode()); if (!state.terminal && buffer.length > 0) { const rawLine = buffer.endsWith("\r") ? buffer.slice(0, -1) : buffer; - const lineBytes = encoder.encode(rawLine).byteLength; + const lineBytes = Buffer.byteLength(rawLine, "utf8"); if (lineBytes > TRANSLATOR_MAX_SSE_EVENT_BYTES) throw new TranslatorBudgetExceededError("live_transient", TRANSLATOR_MAX_SSE_EVENT_BYTES); try { // The EOF record keeps its residual charge while it is parsed and consumed — releasing it diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index ff55305299f..3eefc7360a8 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -140,6 +140,14 @@ envelopes, and real HTTP streams; `tests/server/inbound-body-budget.test.ts` cov ## Stream-buffer accounting +Native Ollama NDJSON in `src/adapters/ollama-native.ts` measures new decoded fragments without +allocating UTF-8 arrays. A fragment without a newline extends the cached residual byte count; +only a fragment containing a newline triggers splitting of the accumulated record. Complete +records and the remaining residual retain their independent size checks and existing reservation +order, including CRLF and EOF handling. The fatal streaming decoder emits complete code points, +so fragment counts remain exact across split UTF-8 reads. `tests/providers/ollama/ollama-native-scaling.test.ts` +checks linear measurement and splitting work, framing equivalence, invalid UTF-8 and cancellation. + `src/lib/pinned-http.ts` pauses its socket response when the receiving Web Stream queue fills and resumes when the reader asks for more. For gzip and deflate, the final decoded queue also gates subsequent socket input even when the decoder would accept more coded bytes. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 7939d5f9243..b0113b507e3 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1071,6 +1071,7 @@ "ocx-launcher-desktop-handoff.test.ts": "cli", "cli-path-diagnostics.test.ts": "cli", "ocx-run.test.ts": "cli", "ollama-native-parser.test.ts": "providers/ollama", "ollama-native-reasoning-wire.test.ts": "providers/ollama", + "ollama-native-scaling.test.ts": "providers/ollama", "ollama-native-structured-output.test.ts": "providers/ollama", "ollama-native-v4.test.ts": "providers/ollama", "ollama-native.test.ts": "providers/ollama", "ollama-show-enrichment-v7.test.ts": "providers/ollama", "ollama-show-enrichment.test.ts": "providers/ollama", "ollama-show-ignore-abort.test.ts": "providers/ollama", diff --git a/tests/providers/ollama/ollama-native-scaling.test.ts b/tests/providers/ollama/ollama-native-scaling.test.ts new file mode 100644 index 00000000000..ee60e0240f1 --- /dev/null +++ b/tests/providers/ollama/ollama-native-scaling.test.ts @@ -0,0 +1,91 @@ +import { expect, spyOn, test } from "bun:test"; +import { createOllamaNativeAdapter } from "../../../src/adapters/ollama-native"; +import type { AdapterEvent } from "../../../src/types"; +import { createTestTranslatorBudget } from "../../helpers/translator-budget"; + +const encoder = new TextEncoder(); +const adapter = () => createOllamaNativeAdapter({ adapter: "ollama-native", baseUrl: "http://127.0.0.1:11434", models: ["fixture"] }); +const record = (content: string, done = true) => JSON.stringify({ model: "fixture", message: { role: "assistant", content }, done, ...(done ? { done_reason: "stop", eval_count: 7 } : {}) }); + +function response(bytes: Uint8Array, chunkSize: number, emptyChunks = false, keepOpen = false) { + let offset = 0; + let empty = emptyChunks; + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { + if (empty) { empty = false; controller.enqueue(new Uint8Array(0)); return; } + if (offset === bytes.length) { if (!keepOpen) controller.close(); return; } + const end = Math.min(bytes.length, offset + chunkSize); + controller.enqueue(bytes.subarray(offset, end)); + offset = end; + empty = emptyChunks; + }, + cancel() { cancelled = true; }, + }); + return { response: new Response(body), cancelled: () => cancelled }; +} + +async function collect(input: Response) { + const budget = createTestTranslatorBudget(); + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(input, budget)) events.push(event); + expect(budget.snapshot().currentBytes).toBe(0); + return events; +} + +for (const metric of ["measurement", "splitting"] as const) { + test(`fragmented Ollama records have linear ${metric} work`, async () => { + const content = "x".repeat(256 * 1024); + const bytes = encoder.encode(record(content) + "\n"); + const input = response(bytes, 256, false, true); + const encode = TextEncoder.prototype.encode; + const byteLength = Buffer.byteLength; + const split = String.prototype.split; + let measuredChars = 0; + let splitChars = 0; + const encoding = spyOn(TextEncoder.prototype, "encode").mockImplementation(function (value = "") { + measuredChars += value.length; + return Reflect.apply(encode, this, [value]); + }); + const measuring = spyOn(Buffer, "byteLength").mockImplementation((value, encoding) => { + if (typeof value === "string") measuredChars += value.length; + return byteLength(value, encoding); + }); + const splitting = spyOn(String.prototype, "split").mockImplementation(function (...args: Parameters) { + if (args[0] === "\n") splitChars += this.length; + return Reflect.apply(split, this, args); + }); + let events: AdapterEvent[]; + try { events = await collect(input.response); } + finally { encoding.mockRestore(); measuring.mockRestore(); splitting.mockRestore(); } + expect(events.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe(content); + expect(events.at(-1)).toMatchObject({ type: "done", usage: { outputTokens: 7 } }); + expect(input.cancelled()).toBe(true); + // Bound actual work, independent of CPU speed. Both UTF-8 counting APIs are observed. + if (metric === "measurement") expect(measuredChars).toBeLessThanOrEqual(bytes.length * 8); + else expect(splitChars).toBeLessThanOrEqual(bytes.length * 2); + }); +} + +for (const ending of ["\n", "\r\n", ""] as const) { + test(`Ollama fragmented UTF-8 preserves BOM, blank lines and ${JSON.stringify(ending)} termination`, async () => { + const bytes = encoder.encode("\ufeff\n" + record("한글 😀", false) + "\r\n\n" + record("끝") + ending); + const whole = await collect(response(bytes, bytes.length).response); + const fragmented = await collect(response(bytes, 1, true).response); + expect(fragmented).toEqual(whole); + expect(fragmented.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe("한글 😀끝"); + expect(fragmented.at(-1)?.type).toBe("done"); + }); +} + +test("invalid UTF-8 after a retained partial record releases its budget and cancels", async () => { + const prefix = encoder.encode('{"message":{"content":"partial '); + const bytes = new Uint8Array(prefix.length + 1); + bytes.set(prefix); + bytes[prefix.length] = 0xff; + const input = response(bytes, prefix.length, false, true); + const events = await collect(input.response); + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ type: "error" }); + expect(input.cancelled()).toBe(true); +}); From be7b78186f58900932ad428a590ba91e7b36ff9e Mon Sep 17 00:00:00 2001 From: 0disoft Date: Sat, 31 Oct 2026 17:00:13 +0900 Subject: [PATCH 2/2] test(ollama): assert fragmented UTF-8 budget boundaries --- .../ollama/ollama-native-scaling.test.ts | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/providers/ollama/ollama-native-scaling.test.ts b/tests/providers/ollama/ollama-native-scaling.test.ts index ee60e0240f1..a9052bbfdbd 100644 --- a/tests/providers/ollama/ollama-native-scaling.test.ts +++ b/tests/providers/ollama/ollama-native-scaling.test.ts @@ -78,6 +78,46 @@ for (const ending of ["\n", "\r\n", ""] as const) { }); } +for (const deficit of [0, 1]) { + test(`fragmented Unicode residual ${deficit ? "rejects one byte below" : "admits at"} its UTF-8 byte limit`, async () => { + const content = "é한😀"; + const text = record(content); + const bytes = encoder.encode(text); + expect(bytes.length).toBeGreaterThan(text.length); + const budget = createTestTranslatorBudget({ maxTurnBytes: bytes.length - deficit }); + let offset = 0; + let cancelled = false; + let beforeEof: ReturnType | undefined; + const body = new ReadableStream({ + pull(controller) { + if (offset < bytes.length) { + controller.enqueue(bytes.subarray(offset, ++offset)); + return; + } + // With no read-ahead, every fragment has been ingested; no delimiter or EOF yet. + beforeEof = budget.snapshot(); + controller.close(); + }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 }); + const events: AdapterEvent[] = []; + for await (const event of adapter().parseStream(new Response(body), budget)) events.push(event); + if (deficit === 0) { + expect(beforeEof).toMatchObject({ currentBytes: bytes.length, highWaterBytes: bytes.length, overflows: 0 }); + expect(events.filter(event => event.type === "text_delta").map(event => event.text).join("")).toBe(content); + expect(events.at(-1)?.type).toBe("done"); + expect(budget.snapshot().highWaterBytes).toBe(bytes.length); + } else { + expect(beforeEof).toBeUndefined(); // Refusal must happen while the record is still buffered. + expect(events).toHaveLength(1); + expect(events[0]).toMatchObject({ type: "error", code: "translation_buffer_limit" }); + expect(budget.snapshot().overflows).toBe(1); + expect(cancelled).toBe(true); + } + expect(budget.snapshot().currentBytes).toBe(0); + }); +} + test("invalid UTF-8 after a retained partial record releases its budget and cancels", async () => { const prefix = encoder.encode('{"message":{"content":"partial '); const bytes = new Uint8Array(prefix.length + 1);