Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/unclaimed-view-pool-leak.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@workflow/core": patch
---

Serialize only the viewed bytes of a `Float16Array` (no longer the whole Node `Buffer` pool behind it), and support `Float16Array` in the QuickJS engine with the same wire format as the node:vm engine.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed for v5, fixed in b61f19b: Float16Array is now listed next to Float32Array/Float64Array in docs/content/docs/v5/foundations/serialization.mdx, and the PR description has a Docs Preview section.

I left v4 alone on purpose. Nothing in this PR applies to 4.x: stable has no QuickJS engine and no hardened viewInfo (it calls devalue's stringify with the default operations). Whether to list Float16Array in the v4 docs belongs with a fix on stable, not here.

2 changes: 1 addition & 1 deletion docs/content/docs/v5/foundations/serialization.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ The following types can be serialized and passed through workflow functions:
- `BigInt64Array`, `BigUint64Array`
- `DataView`
- `Date`
- `Float32Array`, `Float64Array`
- `Float16Array`, `Float32Array`, `Float64Array`
- `Int8Array`, `Int16Array`, `Int32Array`
- `Map<Serializable, Serializable>`
- `RegExp`
Expand Down
99 changes: 99 additions & 0 deletions packages/core/src/runtime/quickjs-serde.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import { createRequire } from 'node:module';
import { StreamError } from '@workflow/errors';
import { QuickJS } from 'quickjs-wasi';
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
import { serialize as nodeEngineSerialize } from '../serialization/workflow.js';
import {
deserialize as referenceDeserialize,
serialize as referenceSerialize,
Expand Down Expand Up @@ -424,6 +425,104 @@ describe('full round trip through the host serde only', () => {
});
});

describe.skipIf(typeof Float16Array !== 'function')(
'Float16Array (no reducer: devalue built-in branch)',
() => {
// `Float16Array` has no workflow reducer on any codec, so the reference
// codec, the node:vm engine and this one must all emit devalue's
// built-in `[tag, buffer]` form, and a subview must carry only its
// viewed bytes.
const expectParity = (bytes: Uint8Array, hostValue: unknown) => {
expect(text(bytes)).toBe(text(referenceSerialize(hostValue)));
expect(text(bytes)).toBe(text(nodeEngineSerialize(hostValue)));
};

it('matches the reference codec for a whole-buffer Float16Array', () => {
const bytes = serializeGuest('new Float16Array([0.5, -1, 65504])');
expectParity(bytes, new Float16Array([0.5, -1, 65504]));
expect(
checkInGuest(
bytes,
`function (v) {
return { isF16: v instanceof Float16Array, values: Array.from(v) };
}`
)
).toEqual({ isF16: true, values: [0.5, -1, 65504] });
});

it('carries a Float16Array subview without the rest of its buffer', () => {
const bytes = serializeGuest(
'new Float16Array(new Uint8Array([1,2,0,60,0,64,7,8]).buffer, 2, 2)'
);
// Base64 of [0, 60, 0, 64] (1.0, 2.0) and nothing else.
expect(text(bytes)).toBe(
'devl[["Float16Array",1],["ArrayBuffer",2],"ADwAQA=="]'
);
expectParity(
bytes,
new Float16Array(
new Uint8Array([1, 2, 0, 60, 0, 64, 7, 8]).buffer,
2,
2
)
);
expect(
checkInGuest(
bytes,
`function (v) {
return {
isF16: v instanceof Float16Array,
byteOffset: v.byteOffset,
bufferByteLength: v.buffer.byteLength,
values: Array.from(v),
};
}`
)
).toEqual({
isF16: true,
byteOffset: 0,
bufferByteLength: 4,
values: [1, 2],
});
});

it('serializes an empty Float16Array subview', () => {
const bytes = serializeGuest(
'new Float16Array(new ArrayBuffer(8), 4, 0)'
);
expectParity(bytes, new Float16Array(new ArrayBuffer(8), 4, 0));
expect(
checkInGuest(
bytes,
`function (v) {
return { isF16: v instanceof Float16Array, length: v.length };
}`
)
).toEqual({ isF16: true, length: 0 });
});

it('revives a payload carrying subview bounds', () => {
// What devalue's built-in encoding wrote before the hardened
// `viewInfo` copied subviews: the whole buffer, bounds beside it.
const legacy = new TextEncoder().encode(
'devl[["Float16Array",1,2,2],["ArrayBuffer",2],"AQIAPABABwg="]'
);
expect(
checkInGuest(
legacy,
`function (v) {
return {
isF16: v instanceof Float16Array,
byteOffset: v.byteOffset,
values: Array.from(v),
};
}`
)
).toEqual({ isF16: true, byteOffset: 2, values: [1, 2] });
});
}
);

describe('NUL (U+0000) safety across the WASM boundary', () => {
// `JS_ToCString` is NUL-terminated: naive extraction truncates guest
// strings at the first U+0000 and mangles NUL-bearing property keys
Expand Down
30 changes: 29 additions & 1 deletion packages/core/src/runtime/quickjs-serde.ts
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,9 @@ const BRANDED_SAMPLES = `({
Float64Array: new Float64Array(0),
BigInt64Array: new BigInt64Array(0),
BigUint64Array: new BigUint64Array(0),
...(typeof Float16Array === 'function'
? { Float16Array: new Float16Array(0) }
: {}),
})`;

/**
Expand Down Expand Up @@ -203,6 +206,7 @@ const CAPTURE_INTRINSICS = `(() => {
Int8Array, Uint8Array, Uint8ClampedArray, Int16Array, Uint16Array,
Int32Array, Uint32Array, Float32Array, Float64Array,
BigInt64Array, BigUint64Array,
Float16Array: g.Float16Array,
Error,
AggregateError: g.AggregateError,
EvalError, RangeError, ReferenceError, SyntaxError, TypeError, URIError,
Expand Down Expand Up @@ -486,6 +490,14 @@ export function createQuickJSSerde(
]) {
typedArrayConstructors.set(name, at(name));
}
// `Float16Array` has no workflow reducer: it takes devalue's built-in
// typed-array branch, matching the node:vm engine's wire format. Optional
// because a capture root adopted from a snapshot taken before this capture
// existed does not carry it.
{
const float16Array = optional('Float16Array');
if (float16Array) typedArrayConstructors.set('Float16Array', float16Array);
}

const symbols = new Map<SymbolName, JSValueHandle>();
{
Expand Down Expand Up @@ -828,7 +840,23 @@ export function createQuickJSSerde(
if (!isDataView) {
info.length = call(i.viewLength, value).consume((h) => h.toNumber());
}
return info;
if (info.byteLength === info.bufferByteLength) return info;
// Reached only for views no reducer claims (`Float16Array`). devalue's
// built-in encoding emits the whole backing buffer plus offset/length,
// which would put bytes outside the view on the wire. Hand it a fresh
// guest buffer holding only the viewed bytes instead, so it encodes the
// compact `[tag, buffer]` form, byte-identical to the node:vm engine
// (serialization/hardened.ts). The guest buffer is claimed by the
// `ArrayBuffer` reducer like any other and swept with the pass's
// handle scope.
const viewed = viewBytes(value).slice();
return {
buffer: vm.newArrayBuffer(viewed.buffer as ArrayBuffer),
byteOffset: 0,
byteLength: info.byteLength,
bufferByteLength: info.byteLength,
length: info.length,
};
},
toArrayBuffer: (value) =>
isHandle(value) ? value.toArrayBuffer() : d.toArrayBuffer(value),
Expand Down
82 changes: 82 additions & 0 deletions packages/core/src/serialization.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2916,6 +2916,88 @@ describe('DataView serialization', () => {
});
});

describe.skipIf(typeof Float16Array !== 'function')(
'unclaimed typed array serialization',
() => {
/**
* No reducer claims `Float16Array`, so it reaches devalue's built-in
* typed-array encoding, which emits the view's whole backing buffer plus
* offset/length. The hardened `viewInfo` operation must hand devalue only
* the viewed bytes, or a view onto Node's shared `Buffer` pool leaks
* unrelated allocations into the event log. See the `DataView` tests
* above for the same hazard on a claimed type.
*/
it('writes only the viewed bytes of a pooled Float16Array to a step return', async () => {
const neighbour = Buffer.allocUnsafe(256);
neighbour.fill('SECRET-POOL-RESIDUE-');

const viewed = Buffer.allocUnsafe(4);
viewed.set([0x00, 0x3c, 0x00, 0x40]); // 1.0, 2.0 as little-endian f16
const f16 = new Float16Array(viewed.buffer, viewed.byteOffset, 2);

// Precondition: really a pooled view.
expect(f16.buffer.byteLength).toBeGreaterThan(f16.byteLength);

const serialized = (await dehydrateStepReturnValue(
f16,
mockRunId,
noEncryptionKey,
[]
)) as Uint8Array;

const wire = new TextDecoder().decode(serialized);
expect(wire).not.toContain(Buffer.from('SECRET').toString('base64'));
expect(wire).toBe(
`devl[["Float16Array",1],["ArrayBuffer",2],"${Buffer.from([0x00, 0x3c, 0x00, 0x40]).toString('base64')}"]`
);

const hydrated = (await hydrateStepReturnValue(
serialized,
mockRunId,
noEncryptionKey
)) as Float16Array;

expect(hydrated).toBeInstanceOf(Float16Array);
expect(hydrated.byteOffset).toBe(0);
expect(hydrated.buffer.byteLength).toBe(4);
expect([...hydrated]).toEqual([1, 2]);
});

it('round-trips a Float16Array that spans its whole buffer', async () => {
const f16 = new Float16Array([0.5, -1, 65504]);
const serialized = await dehydrateStepReturnValue(
f16,
mockRunId,
noEncryptionKey,
[]
);
const hydrated = (await hydrateStepReturnValue(
serialized,
mockRunId,
noEncryptionKey
)) as Float16Array;
expect([...hydrated]).toEqual([0.5, -1, 65504]);
});

it('round-trips a zero-length Float16Array subview', async () => {
const serialized = await dehydrateStepReturnValue(
new Float16Array(new ArrayBuffer(8), 4, 0),
mockRunId,
noEncryptionKey,
[]
);
const hydrated = (await hydrateStepReturnValue(
serialized,
mockRunId,
noEncryptionKey
)) as Float16Array;
expect(hydrated).toBeInstanceOf(Float16Array);
expect(hydrated.length).toBe(0);
expect(hydrated.buffer.byteLength).toBe(0);
});
}
);

describe('step function serialization', () => {
const { globalThis: vmGlobalThis } = createContext({
seed: 'test',
Expand Down
37 changes: 36 additions & 1 deletion packages/core/src/serialization/codec-devalue-vm.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,40 @@ import { type Reducers, type Revivers, SerializationFormat } from './types.js';
const encoder = new TextEncoder();
const decoder = new TextDecoder();

/**
* devalue's built-in typed-array encoding (reached only by views no reducer
* claims, i.e. `Float16Array`) emits the view's whole backing buffer plus
* offset/length, putting bytes outside the view on the wire. Copy a
* subview's viewed bytes into a fresh buffer instead, so devalue emits the
* compact `[tag, buffer]` form, as both engines do (serialization/hardened.ts,
* runtime/quickjs-serde.ts).
*/
const stringifyOptions = {
operations: {
viewInfo: (view: ArrayBufferView & { length?: number }) => {
const length = view instanceof DataView ? 0 : (view.length ?? 0);
if (view.byteLength === view.buffer.byteLength) {
return {
buffer: view.buffer,
byteOffset: view.byteOffset,
byteLength: view.byteLength,
length,
bufferByteLength: view.buffer.byteLength,
};
}
const copy = new Uint8Array(view.byteLength);
copy.set(new Uint8Array(view.buffer, view.byteOffset, view.byteLength));
return {
buffer: copy.buffer,
byteOffset: 0,
byteLength: view.byteLength,
length,
bufferByteLength: view.byteLength,
};
},
},
};

// ---- AbortController / AbortSignal (workflow VM context) ----
// Mirrors the node:vm engine's workflow-context abort reducers/revivers in
// serialization.ts: reduce by reading the stream/hook symbols stamped at
Expand Down Expand Up @@ -162,7 +196,8 @@ export const devalueVmCodec: Codec = {
const reducers = getReducersForMode(mode);
const str = stringify(
value,
reducers as Record<string, (value: any) => any>
reducers as Record<string, (value: any) => any>,
stringifyOptions
);
return encoder.encode(str);
},
Expand Down
43 changes: 36 additions & 7 deletions packages/core/src/serialization/hardened.ts
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,16 @@ const typedArrayByteLength = protoGetter<number>(
'byteLength'
);
const typedArrayLength = protoGetter<number>(TypedArrayPrototype, 'length');
// `%TypedArray%.prototype.set` copies by internal slots and, unlike `slice`,
// never consults a species constructor.
const typedArraySet = uncurryThis(
(TypedArrayPrototype as Uint8Array).set as (
this: Uint8Array,
source: ArrayLike<number>
) => void
);
const HostArrayBuffer = ArrayBuffer;
const HostUint8Array = Uint8Array;
const dataViewBuffer = protoGetter<ArrayBufferLike>(
DataView.prototype,
'buffer'
Expand Down Expand Up @@ -652,14 +662,33 @@ export const hardenedStringifyOperations: Partial<StringifyOperations> = {

viewInfo: (view: ArrayBufferView) => {
const info = viewInfo(view);
// devalue only reads `length` for typed-array subviews; DataView
// subviews are encoded from `byteLength`.
const length = types.isDataView(view) ? 0 : typedArrayLength(view);
const bufferByteLength = types.isSharedArrayBuffer(info.buffer)
? sharedArrayBufferByteLength(info.buffer)
: arrayBufferByteLength(info.buffer);
if (info.byteLength === bufferByteLength) {
return { ...info, length, bufferByteLength };
}
// Reached only for views the reducers did not claim (e.g. `Float16Array`,
// which no reducer set handles). devalue's built-in encoding emits the
// view's *whole* backing buffer plus offset/length, and a view onto
// Node's shared `Buffer` pool (`Buffer.allocUnsafe`, small `Buffer.from`)
// would put 8 KiB of unrelated, possibly sensitive allocations into the
// event log. Copy just the viewed bytes into a fresh buffer, so devalue
// encodes the compact `[tag, buffer]` form, which every reader accepts.
const copy = new HostArrayBuffer(info.byteLength);
typedArraySet(
new HostUint8Array(copy),
new HostUint8Array(info.buffer, info.byteOffset, info.byteLength)
);
return {
...info,
// devalue only reads `length` for typed-array subviews; DataView
// subviews are encoded from `byteLength`.
length: types.isDataView(view) ? 0 : typedArrayLength(view),
bufferByteLength: types.isSharedArrayBuffer(info.buffer)
? sharedArrayBufferByteLength(info.buffer)
: arrayBufferByteLength(info.buffer),
buffer: copy,
byteOffset: 0,
byteLength: info.byteLength,
length,
bufferByteLength: info.byteLength,
};
},

Expand Down
Loading