fix(ai): meter converted audio from reported output duration (#4366)

* fix(ai): meter converted audio from reported output duration

* fix(ai): size undocumented voice changer output formats as the default

* fix(ai): skip WAV headers when metering and renew streamed charge holds

A wav_* output's RIFF header was counted as samples, rounding a 3 s clip
up to 4 s. The streamed-charge hold now renews until the charge lands, so
a slow reader can't outlast it.

* fix(ai): time Opus output by its pages and renew holds through slow opens

Opus is variable bitrate, so a mostly silent clip billed far under its
length at the nominal rate; its length now comes from the last Ogg
page's granule position. The streamed-charge hold renews from before the
upstream opens, since opening can run as long as the hold's TTL, and is
released even if recording the charge throws. Speech-to-text extra_body
can no longer override response_format or stream, which billing reads.
This commit is contained in:
Daniel Salazar authored and GitHub committed 2026-10-28 16:05:36 -07:00
1 parent 7c6c0f2d39
commit c2b38786f7
9 files changed
+729 -31

No files matched your search

@@ -47,7 +47,13 @@ import { PuterServer } from '../../server.js';
import type { MeteringService } from '../../services/metering/MeteringService.js';
import { setupTestServer } from '../../testUtil.js';
import { generateDefaultFsentries } from '../../util/userProvisioning.js';
import type { VoiceChangerDriver } from './VoiceChangerDriver.js';
import type { DriverStreamResult } from '../meta.js';
import {
OggOpusClock,
outputBytesPerSecond,
type VoiceChangerDriver,
wavDataOffset,
} from './VoiceChangerDriver.js';
import {
callerError,
readStreamResult,
@@ -330,11 +336,11 @@ describe('VoiceChangerDriver.convert success path', () => {
it('meters one usage line at the per-second rate from costs.ts', async () => {
const { actor } = await makeUser();
// 32 KB of mp3_44100_128 back = 2 seconds.
fetchSpy.mockResolvedValueOnce(
okResponse(new ArrayBuffer(8), 'audio/mpeg'),
okResponse(new ArrayBuffer(32_000), 'audio/mpeg'),
);
// 32 KB audio at 16 kbit/s = 2 seconds rounded up.
const buf = Buffer.alloc(32_000);
await readStreamResult(
await withActor(actor, () =>
@@ -344,7 +350,7 @@ describe('VoiceChangerDriver.convert success path', () => {
const usageType = 'elevenlabs:eleven_multilingual_sts_v2:second';
const perSecond = VOICE_CHANGER_COSTS[usageType];
const expectedSeconds = Math.max(1, Math.ceil(32_000 / 16000));
const expectedSeconds = 2;
const calls = incrementUsageSpy.mock.calls.filter(
([, type]) => type === usageType,
@@ -365,7 +371,7 @@ describe('VoiceChangerDriver.convert success path', () => {
.map(([, , , cost]) => cost);
const convert = async (charges: number) => {
fetchSpy.mockResolvedValueOnce(
okResponse(new ArrayBuffer(8), 'audio/mpeg'),
okResponse(new ArrayBuffer(32_000), 'audio/mpeg'),
);
await readStreamResult(
await withActor(actor, () =>
@@ -385,6 +391,230 @@ describe('VoiceChangerDriver.convert success path', () => {
expect(held).toEqual([base, base! * 2]);
});
it.each<[string, number]>([
['mp3_44100_128', 16_000],
['mp3_22050_32', 4_000],
['opus_48000_64', 8_000],
['pcm_44100', 88_200],
['ulaw_8000', 8_000],
['wav_44100', 88_200],
['mp3_44100_999999', 16_000],
['flac_44100', 16_000],
])('sizes %s at %s bytes per second', (format, bytes) => {
expect(outputBytesPerSecond(format)).toBe(bytes);
});
/** A 16-bit mono WAV header, with an optional LIST chunk before data. */
const wavHeader = (sampleRate: number, dataBytes: number, list = 0) => {
const listChunk = list
? Buffer.concat([
Buffer.from('LIST'),
Buffer.from(new Uint32Array([list]).buffer),
Buffer.alloc(list),
])
: Buffer.alloc(0);
const fmt = Buffer.alloc(24);
fmt.write('fmt ', 0, 'ascii');
fmt.writeUInt32LE(16, 4);
fmt.writeUInt16LE(1, 8);
fmt.writeUInt16LE(1, 10);
fmt.writeUInt32LE(sampleRate, 12);
fmt.writeUInt32LE(sampleRate * 2, 16);
fmt.writeUInt16LE(2, 20);
fmt.writeUInt16LE(16, 22);
const data = Buffer.alloc(8);
data.write('data', 0, 'ascii');
data.writeUInt32LE(dataBytes, 4);
const body = Buffer.concat([fmt, listChunk, data]);
const riff = Buffer.alloc(12);
riff.write('RIFF', 0, 'ascii');
riff.writeUInt32LE(4 + body.length + dataBytes, 4);
riff.write('WAVE', 8, 'ascii');
return Buffer.concat([riff, body]);
};
it("finds where a WAV stream's samples start", () => {
expect(wavDataOffset(wavHeader(16_000, 0))).toBe(44);
expect(wavDataOffset(wavHeader(16_000, 0, 26))).toBe(78);
expect(wavDataOffset(wavHeader(16_000, 0).subarray(0, 40))).toBe(
undefined,
);
expect(wavDataOffset(Buffer.alloc(64))).toBe(undefined);
});
/** One Ogg page carrying `body` as a single packet. */
const oggPage = (granule: bigint, sequence: number, body: Buffer) => {
const lacing: number[] = [];
let left = body.length;
while (left >= 255) {
lacing.push(255);
left -= 255;
}
lacing.push(left);
const header = Buffer.alloc(27);
header.write('OggS', 0, 'ascii');
header.writeBigUInt64LE(granule, 6);
header.writeUInt32LE(sequence, 18);
header[26] = lacing.length;
return Buffer.concat([header, Buffer.from(lacing), body]);
};
/** A sparse Ogg Opus stream: `seconds` of audio in very few bytes. */
const oggOpus = (seconds: number, preSkip = 312) => {
const opusHead = Buffer.alloc(19);
opusHead.write('OpusHead', 0, 'ascii');
opusHead[8] = 1;
opusHead[9] = 1;
opusHead.writeUInt16LE(preSkip, 10);
opusHead.writeUInt32LE(48_000, 12);
const pages = [
oggPage(0n, 0, opusHead),
oggPage(0n, 1, Buffer.from('OpusTags')),
];
for (let s = 1; s <= seconds; s++) {
pages.push(
oggPage(BigInt(preSkip + s * 48_000), s + 1, Buffer.alloc(300)),
);
}
return Buffer.concat(pages);
};
it('times an Ogg Opus stream by its last granule, however it is chunked', () => {
const stream = oggOpus(7);
const whole = new OggOpusClock();
whole.push(stream);
expect(whole.seconds).toBe(7);
const trickle = new OggOpusClock();
for (const byte of stream) trickle.push(Buffer.from([byte]));
expect(trickle.seconds).toBe(7);
});
it('does not time a stream that is not Ogg Opus', () => {
const clock = new OggOpusClock();
clock.push(Buffer.alloc(4096, 1));
expect(clock.seconds).toBe(undefined);
});
describe('meters the length of the audio that comes back', () => {
const usageType = 'elevenlabs:eleven_multilingual_sts_v2:second';
const billed = async () => {
await vi.waitFor(() =>
expect(
incrementUsageSpy.mock.calls.filter(
([, type]) => type === usageType,
),
).toHaveLength(1),
);
const [, , count, cost] = incrementUsageSpy.mock.calls.find(
([, type]) => type === usageType,
)!;
expect(cost).toBe(VOICE_CHANGER_COSTS[usageType]! * count);
return count;
};
const convert = async (
actor: Actor,
inputBytes: number,
args: Record<string, unknown> = {},
) =>
readStreamResult(
await withActor(actor, () =>
driver.convert({
audio: dataUrl(Buffer.alloc(inputBytes), 'audio/ogg'),
...args,
}),
),
);
it('bills 600 s for a compact 600 s clip, not its byte size', async () => {
const { actor } = await makeUser();
// 600 s of Opus at 8 kbps in; 600 s of mp3_44100_128 out.
fetchSpy.mockResolvedValueOnce(
okResponse(new ArrayBuffer(600 * 16_000)),
);
await convert(actor, 441_078);
expect(await billed()).toBe(600);
});
it('sizes the output by the requested output_format', async () => {
const { actor } = await makeUser();
// pcm_16000 is 16-bit mono: 32,000 bytes per second.
fetchSpy.mockResolvedValueOnce(
okResponse(new ArrayBuffer(3 * 32_000), 'audio/pcm'),
);
await convert(actor, 1_000_000, { output_format: 'pcm_16000' });
expect(await billed()).toBe(3);
});
it('does not bill a WAV header as audio', async () => {
const { actor } = await makeUser();
// 3 s of wav_16000 is 96,000 bytes of samples plus its header.
fetchSpy.mockResolvedValueOnce(
okResponse(
new Uint8Array(
Buffer.concat([
wavHeader(16_000, 3 * 32_000, 26),
Buffer.alloc(3 * 32_000),
]),
).buffer,
'audio/wav',
),
);
await convert(actor, 1_000_000, { output_format: 'wav_16000' });
expect(await billed()).toBe(3);
});
it('bills sparse Opus output by its duration, not its bytes', async () => {
const { actor } = await makeUser();
// 20 s of mostly silent opus_48000_64 is far under 20 s of bytes.
const body = oggOpus(20);
expect(body.length).toBeLessThan(8_000 * 2);
fetchSpy.mockResolvedValueOnce(
okResponse(new Uint8Array(body).buffer, 'audio/ogg'),
);
await convert(actor, 1_000, { output_format: 'opus_48000_64' });
expect(await billed()).toBe(20);
});
it('sizes an undocumented output_format as the default', async () => {
const { actor } = await makeUser();
// 3 s at the default mp3_44100_128.
fetchSpy.mockResolvedValueOnce(
okResponse(new ArrayBuffer(3 * 16_000)),
);
await convert(actor, 1_000_000, {
output_format: 'mp3_44100_999999',
});
expect(await billed()).toBe(3);
});
it('charges at least the input estimate when the caller hangs up', async () => {
const { actor } = await makeUser();
fetchSpy.mockResolvedValueOnce(
new Response(
new ReadableStream({
start(controller) {
controller.enqueue(new Uint8Array(16_000));
},
}),
{ status: 200, headers: { 'content-type': 'audio/mpeg' } },
),
);
const result = (await withActor(actor, () =>
driver.convert({
audio: dataUrl(Buffer.alloc(80_000), 'audio/mpeg'),
}),
)) as DriverStreamResult;
for await (const _chunk of result.stream) break;
expect(await billed()).toBe(5);
await vi.waitFor(async () =>
expect(
await server.stores.creditHold.outstanding(actor.user.uuid),
).toBe(0),
);
});
});
it('meters nothing when ElevenLabs breaks off before any audio', async () => {
const { actor } = await makeUser();
fetchSpy.mockResolvedValueOnce(
@@ -17,9 +17,11 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
import { pipeline, Readable, Transform } from 'node:stream';
import { Context } from '../../core/context.js';
import { HttpError } from '../../core/http/HttpError.js';
import type { DriverStreamResult } from '../meta.js';
import { OUTPUT_FORMATS } from '../ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.js';
import { PuterDriver } from '../types.js';
import {
type AiMeteringService,
@@ -198,10 +200,9 @@ export class VoiceChangerDriver extends PuterDriver {
legacyCode: 'bad_request',
});
// Metering: estimate duration from file size if we don't parse metadata.
// 16 kbit/s is a safe lower bound for speech audio; pre-check credits
// before we hit the ElevenLabs API. Post-usage we increment by the same
// estimate — duration parsing is deferred to v2.1 if needed.
// The hold estimates seconds from the input at 128 kbps: compact audio
// is under-held rather than refused. Usage is metered from the length
// of the audio that comes back.
const estimatedSeconds = Math.max(
1,
Math.ceil(loaded.buffer.byteLength / 16000),
@@ -263,6 +264,10 @@ export class VoiceChangerDriver extends PuterDriver {
if (search) url.search = search;
const metering = this.#aiMetering;
const output = measuredOutput(
outputBytesPerSecond(outputFormat),
outputFormat,
);
return withStreamedAiCharge(
metering,
actor,
@@ -287,16 +292,206 @@ export class VoiceChangerDriver extends PuterDriver {
dataType: 'stream',
content_type:
response.headers.get('content-type') ?? 'audio/mpeg',
stream: upstreamBodyStream(response),
stream: output.measure(upstreamBodyStream(response)),
};
},
() =>
metering.incrementUsage(
async () => {
const measured = await output.seconds;
// A stream cut short was still converted in full upstream.
const seconds =
measured.complete && measured.seconds
? measured.seconds
: Math.max(measured.seconds ?? 0, estimatedSeconds);
return metering.incrementUsage(
actor,
usageKey,
estimatedSeconds,
estimatedCost,
),
seconds,
ucentsPerSecond * seconds,
);
},
);
}
}
/**
* Bytes per second of an ElevenLabs `output_format`. A format outside the
* documented set is sized as the default, so a made-up bitrate can't shrink the
* bill.
*/
export function outputBytesPerSecond(format: string): number {
const known = OUTPUT_FORMATS.has(format) ? format : DEFAULT_OUTPUT_FORMAT;
const [codec, rate, kbps] = known.split('_');
if (codec === 'mp3' || codec === 'opus') return Number(kbps) * 125;
if (codec === 'ulaw' || codec === 'alaw') return Number(rate);
// pcm and wav: 16-bit mono.
return Number(rate) * 2;
}
// Past this, a WAV whose data chunk hasn't started is billed by raw length.
const WAV_HEADER_MAX_BYTES = 64 * 1024;
/**
* Where the samples of a RIFF/WAVE stream start, once `head` reaches its `data`
* chunk; undefined until then, or if `head` isn't a WAV.
*/
export function wavDataOffset(head: Buffer): number | undefined {
if (head.length < 12) return undefined;
if (
head.toString('ascii', 0, 4) !== 'RIFF' ||
head.toString('ascii', 8, 12) !== 'WAVE'
)
return undefined;
let offset = 12;
while (offset + 8 <= head.length) {
if (head.toString('ascii', offset, offset + 4) === 'data')
return offset + 8;
const size = head.readUInt32LE(offset + 4);
offset += 8 + size + (size & 1);
}
return undefined;
}
const OGG_HEADER_BYTES = 27;
const NO_GRANULE = 2n ** 64n - 1n;
/**
* Walks an Ogg Opus stream page by page as it arrives. Opus is variable
* bitrate, so its length comes from the last page's granule position (48 kHz
* samples, less the stream's pre-skip) rather than its byte count.
*/
export class OggOpusClock {
#header = Buffer.alloc(0);
#skip = 0;
#firstBody: Buffer | undefined = Buffer.alloc(0);
#firstBodyBytes = 0;
#granule: bigint | undefined;
#preSkip = 0;
#broken = false;
push(chunk: Buffer): void {
let offset = 0;
while (!this.#broken && offset < chunk.length) {
if (this.#skip > 0) {
const n = Math.min(this.#skip, chunk.length - offset);
if (this.#firstBody && this.#firstBody.length < 19) {
this.#firstBody = Buffer.concat([
this.#firstBody,
chunk.subarray(offset, offset + n),
]);
}
this.#skip -= n;
offset += n;
if (this.#skip === 0 && this.#firstBody) this.#readOpusHead();
continue;
}
const want =
this.#header.length < OGG_HEADER_BYTES
? OGG_HEADER_BYTES
: OGG_HEADER_BYTES + this.#header[26]!;
const n = Math.min(
want - this.#header.length,
chunk.length - offset,
);
this.#header = Buffer.concat([
this.#header,
chunk.subarray(offset, offset + n),
]);
offset += n;
if (this.#header.length < want) continue;
if (this.#header.toString('ascii', 0, 4) !== 'OggS') {
this.#broken = true;
return;
}
if (want === OGG_HEADER_BYTES && this.#header[26]! > 0) continue;
const granule = this.#header.readBigUInt64LE(6);
if (granule !== NO_GRANULE) this.#granule = granule;
let body = 0;
for (const lacing of this.#header.subarray(OGG_HEADER_BYTES)) {
body += lacing;
}
this.#header = Buffer.alloc(0);
this.#skip = body;
if (this.#firstBody) this.#firstBodyBytes = body;
}
}
/** Whole seconds so far, once the stream has proven to be Ogg Opus. */
get seconds(): number | undefined {
if (this.#broken || this.#granule === undefined) return undefined;
const samples = Number(this.#granule) - this.#preSkip;
return samples > 0
? Math.max(1, Math.ceil(samples / 48_000))
: undefined;
}
#readOpusHead(): void {
const head = this.#firstBody!;
this.#firstBody = undefined;
if (
this.#firstBodyBytes < 19 ||
head.toString('ascii', 0, 8) !== 'OpusHead'
) {
this.#broken = true;
return;
}
this.#preSkip = head.readUInt16LE(10);
}
}
/**
* Counts the bytes of a converted stream. `seconds` settles when the stream
* ends or closes: its length in whole seconds (undefined when nothing came
* back), and whether the upstream sent all of it. A WAV's header isn't audio,
* so it isn't counted; Opus is timed by its pages.
*/
function measuredOutput(bytesPerSecond: number, format: string) {
let bytes = 0;
let head: Buffer | undefined = format.startsWith('wav_')
? Buffer.alloc(0)
: undefined;
const opus = format.startsWith('opus_') ? new OggOpusClock() : undefined;
let headerBytes = 0;
let settle!: (measured: { seconds?: number; complete: boolean }) => void;
const seconds = new Promise<{ seconds?: number; complete: boolean }>(
(resolve) => {
settle = resolve;
},
);
const lengthOf = () => {
const timed = opus?.seconds;
if (timed !== undefined) return timed;
const audio = bytes - headerBytes;
return audio > 0
? Math.max(1, Math.ceil(audio / bytesPerSecond))
: undefined;
};
const measure = (source: Readable): Readable => {
const counter = new Transform({
transform(chunk: Buffer, _encoding, callback) {
bytes += chunk.length;
opus?.push(chunk);
if (head) {
head = Buffer.concat([head, chunk]);
const offset = wavDataOffset(head);
if (offset !== undefined) headerBytes = offset;
if (
offset !== undefined ||
head.length >= WAV_HEADER_MAX_BYTES
)
head = undefined;
}
callback(null, chunk);
},
flush(callback) {
settle({ seconds: lengthOf(), complete: true });
callback();
},
});
counter.once('close', () =>
settle({ seconds: lengthOf(), complete: false }),
);
pipeline(source, counter, () => {});
return counter;
};
return { measure, seconds };
}
@@ -508,6 +508,26 @@ describe('SpeechToTextDriver model selection and payload shape', () => {
expect(sent.temperature).toBe(0.2);
});
it('keeps extra_body from overriding the response shape billing reads', async () => {
const { actor } = await makeUser();
transcriptionsCreateMock.mockResolvedValueOnce({ text: 'x' });
await withActor(actor, () =>
driver.transcribe({
file: dataUrl(Buffer.from('a'), 'audio/mp3'),
model: 'gpt-transcribe',
extra_body: {
response_format: 'text',
stream: true,
chunking_strategy: 'auto',
},
}),
);
const sent = transcriptionsCreateMock.mock.calls[0]![0];
expect(sent.extra_body).toEqual({ chunking_strategy: 'auto' });
});
it('does not forward timestamp_granularities, which gpt-transcribe ignores', async () => {
const { actor } = await makeUser();
transcriptionsCreateMock.mockResolvedValueOnce({ text: 'x' });
@@ -684,6 +704,72 @@ describe('SpeechToTextDriver metering', () => {
expect(creditCall[1]).toBe(expected);
});
it('meters the duration OpenAI reports, not the byte estimate', async () => {
const { actor } = await makeUser();
const usageType = 'openai:gpt-transcribe:second';
const billed = async (bytes: number, seconds: number) => {
incrementUsageSpy.mockClear();
transcriptionsCreateMock.mockResolvedValueOnce({
text: 'x',
usage: { type: 'duration', seconds },
});
await withActor(actor, () =>
driver.transcribe({
file: dataUrl(Buffer.alloc(bytes, 1), 'audio/ogg'),
}),
);
const [, , count, cost] = incrementUsageSpy.mock.calls.find(
([, type]) => type === usageType,
)!;
expect(cost).toBe(SPEECH_TO_TEXT_COSTS[usageType]! * count);
return count;
};
// 600 s of Opus at 8 kbps, and 60 s of 44.1 kHz stereo WAV.
expect(await billed(441_078, 600)).toBe(600);
expect(await billed(10_584_078, 59.2)).toBe(60);
});
it('asks for json behind response_format=text so usage comes back', async () => {
const { actor } = await makeUser();
transcriptionsCreateMock.mockResolvedValueOnce({
text: 'plain transcript',
usage: { type: 'duration', seconds: 42 },
});
const result = await withActor(actor, () =>
driver.transcribe({
file: dataUrl(Buffer.from('a'), 'audio/mp3'),
response_format: 'text',
}),
);
expect(result).toBe('plain transcript');
expect(transcriptionsCreateMock.mock.calls[0]![0].response_format).toBe(
'json',
);
const usageType = 'openai:gpt-transcribe:second';
const [, , count] = incrementUsageSpy.mock.calls.find(
([, type]) => type === usageType,
)!;
expect(count).toBe(42);
});
it('falls back to the byte estimate when OpenAI reports token usage', async () => {
const { actor } = await makeUser();
transcriptionsCreateMock.mockResolvedValueOnce({
text: 'x',
usage: { type: 'tokens', input_tokens: 10, output_tokens: 2 },
});
await withActor(actor, () =>
driver.transcribe({
file: dataUrl(Buffer.alloc(32000, 0), 'audio/mp3'),
}),
);
const usageType = 'openai:gpt-transcribe:second';
const [, , count] = incrementUsageSpy.mock.calls.find(
([, type]) => type === usageType,
)!;
expect(count).toBe(2);
});
it('holds and meters at the AI cost factor', async () => {
const { actor } = await makeUser();
const usageType = 'openai:gpt-transcribe:second';
@@ -170,9 +170,9 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider {
);
}
// Estimate seconds from raw bytes — 16 kbps is a conservative speech-audio
// lower bound. Full metadata parsing (music-metadata) is deferred — clients
// aren't observably sensitive to billing-time delta vs real duration.
// The hold estimates seconds from bytes at 128 kbps: compact audio is
// under-held rather than refused, and usage is metered from the
// duration OpenAI reports.
const estimatedSeconds = Math.max(
1,
Math.ceil(loaded.buffer.byteLength / 16000),
@@ -198,8 +198,13 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider {
file: openaiFile,
model: selectedModel,
};
// A `text` response carries no usage, so json is asked for and
// its text returned.
if (args.response_format)
payload.response_format = args.response_format;
payload.response_format =
args.response_format === 'text'
? 'json'
: args.response_format;
if (args.language) payload.language = args.language;
if (typeof args.temperature === 'number')
payload.temperature = args.temperature;
@@ -207,7 +212,15 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider {
// The API ignores a bare `logprobs` flag; it only honors `include`.
if (args.logprobs && caps.canLogprobs)
payload.include = ['logprobs'];
if (args.extra_body) payload.extra_body = args.extra_body;
if (args.extra_body && typeof args.extra_body === 'object') {
// Billing reads usage off the response; these can strip it.
const extra = {
...(args.extra_body as Record<string, unknown>),
};
delete extra.response_format;
delete extra.stream;
payload.extra_body = extra;
}
const result = await this.#openai.audio.transcriptions.create(
payload as unknown as Parameters<
@@ -215,11 +228,12 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider {
>[0],
);
const billedSeconds = reportedSeconds(result) ?? estimatedSeconds;
this.deps.metering.incrementUsage(
actor,
usageType,
estimatedSeconds,
ucentsPerSecond * estimatedSeconds,
billedSeconds,
ucentsPerSecond * billedSeconds,
);
// Text response_format: return raw string; otherwise forward the OpenAI object.
@@ -234,3 +248,18 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider {
}
}
}
/** Whole seconds of audio OpenAI reports billing for, if it reports them. */
function reportedSeconds(result: unknown): number | undefined {
if (!result || typeof result !== 'object') return undefined;
const { usage, duration } = result as {
usage?: { type?: string; seconds?: unknown };
duration?: unknown;
};
const seconds = usage?.type === 'duration' ? usage.seconds : duration;
return typeof seconds === 'number' &&
Number.isFinite(seconds) &&
seconds > 0
? Math.max(1, Math.ceil(seconds))
: undefined;
}
@@ -48,7 +48,7 @@ const VOICE_ID_PATTERN = /^[A-Za-z0-9]{20}$/;
const VOICE_SEARCH_MAX_PAGES = 5;
// https://elevenlabs.io/docs/api-reference/text-to-speech/convert
// mp3_44100_192 needs the Creator tier; pcm/wav at 44.1 kHz need Pro.
const OUTPUT_FORMATS = new Set([
export const OUTPUT_FORMATS = new Set([
'alaw_8000',
'mp3_22050_32',
'mp3_24000_48',
@@ -651,6 +651,122 @@ describe('AI cost factor', () => {
}
});
it('renews the hold until a slow charge lands', async () => {
vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] });
const refresh = vi.spyOn(server.stores.creditHold, 'refresh');
const check = vi
.spyOn(metering, 'hasEnoughCredits')
.mockResolvedValue(true);
let landCharge = () => {};
const charge = new Promise<void>((resolve) => {
landCharge = resolve;
});
try {
const result = await withStreamedAiCharge(
scoped,
actor,
usageType,
1000,
async () => ({
dataType: 'stream',
content_type: 'audio/mpeg',
stream: Readable.from([Buffer.from('ab')]),
}),
() => charge,
);
expect(await read(result.stream)).toBeUndefined();
await vi.advanceTimersByTimeAsync(11 * 60 * 1000);
expect(refresh).toHaveBeenCalledTimes(2);
expect(await metering.getOutstandingHolds(actor)).toBe(1000);
landCharge();
await vi.waitFor(async () =>
expect(await metering.getOutstandingHolds(actor)).toBe(0),
);
await vi.advanceTimersByTimeAsync(10 * 60 * 1000);
expect(refresh).toHaveBeenCalledTimes(2);
} finally {
check.mockRestore();
refresh.mockRestore();
vi.useRealTimers();
}
});
it('renews the hold while a slow upstream is still opening', async () => {
vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] });
const refresh = vi.spyOn(server.stores.creditHold, 'refresh');
const check = vi
.spyOn(metering, 'hasEnoughCredits')
.mockResolvedValue(true);
let opened = () => {};
const opening = new Promise<void>((resolve) => {
opened = resolve;
});
try {
const pending = withStreamedAiCharge(
scoped,
actor,
usageType,
1000,
async () => {
await opening;
return {
dataType: 'stream',
content_type: 'audio/mpeg',
stream: Readable.from([Buffer.from('ab')]),
};
},
() => undefined,
);
await vi.waitFor(async () =>
expect(await metering.getOutstandingHolds(actor)).toBe(
1000,
),
);
await vi.advanceTimersByTimeAsync(9 * 60 * 1000);
expect(refresh).toHaveBeenCalledTimes(1);
opened();
expect(await read((await pending).stream)).toBeUndefined();
await vi.waitFor(async () =>
expect(await metering.getOutstandingHolds(actor)).toBe(0),
);
} finally {
check.mockRestore();
refresh.mockRestore();
vi.useRealTimers();
}
});
it('releases the hold when recording the charge throws', async () => {
const check = vi
.spyOn(metering, 'hasEnoughCredits')
.mockResolvedValue(true);
try {
const result = await withStreamedAiCharge(
scoped,
actor,
usageType,
1000,
async () => ({
dataType: 'stream',
content_type: 'audio/mpeg',
stream: Readable.from([Buffer.from('ab')]),
}),
() => {
throw new Error('metering down');
},
);
expect(await read(result.stream)).toBeUndefined();
await vi.waitFor(async () =>
expect(await metering.getOutstandingHolds(actor)).toBe(0),
);
} finally {
check.mockRestore();
}
});
it('charges nothing when the upstream fails before its first byte', async () => {
const { record, stream } = await relay(
new Readable({
+26 -6
View File
@@ -416,10 +416,14 @@ export async function withAiCreditHold<T>(
}
}
// Under the hold store's default TTL, so a renewal always lands first.
const STREAM_HOLD_RENEW_MS = 5 * 60 * 1000;
/**
* Holds `amount` while `open` starts an upstream audio stream, then charges it
* through `record` once the first byte arrives, or once the caller hangs up. An
* upstream that fails or ends before its first byte is not charged.
* upstream that fails or ends before its first byte is not charged. The hold is
* renewed until the charge lands, however long a slow reader takes.
*/
export async function withStreamedAiCharge(
metering: AiMeteringService,
@@ -431,10 +435,20 @@ export async function withStreamedAiCharge(
): Promise<DriverStreamResult> {
const hold = await metering.reserveAiCredits(actor, usageType, amount);
if (!hold) throw insufficientCreditsError();
// Started before `open`, which can itself run as long as the hold's TTL.
const renew = setInterval(async () => {
try {
await hold.extend?.();
} catch {
// The next tick retries.
}
}, STREAM_HOLD_RENEW_MS);
renew.unref?.();
let result: DriverStreamResult;
try {
result = await open();
} catch (e) {
clearInterval(renew);
await hold.release();
throw e;
}
@@ -445,11 +459,17 @@ export async function withStreamedAiCharge(
settled = true;
// The relay runs outside the request, so the hold waits on the charge
// itself. A failed release leaves the hold to its TTL.
const release = () => hold.release().catch(() => {});
void Promise.resolve(charge ? record() : undefined).then(
release,
release,
);
const release = async () => {
clearInterval(renew);
try {
await hold.release();
} catch {
// ignored
}
};
void Promise.resolve()
.then(() => (charge ? record() : undefined))
.then(release, release);
};
const source = result.stream;
let chunks: AsyncIterator<unknown> | undefined;
@@ -2264,6 +2264,28 @@ describe('MeteringService', () => {
expect(allowed.remaining).toBe(allowed.monthUsageAllowance + 5_000);
});
it('reads a spent addons record once across gate checks', async () => {
const addonsKey = `${POLICY_PREFIX}:actor:${actor.user.uuid}:addons`;
const getSpy = vi.spyOn(server.stores.kv, 'get');
try {
for (let i = 0; i < 3; i++) await target.getAllowedUsage(actor);
const addonReads = getSpy.mock.calls.filter(
([{ key }]) => key === addonsKey,
);
expect(addonReads).toHaveLength(1);
} finally {
getSpy.mockRestore();
}
});
it('sees credit granted after a gate check memoized the spent record', async () => {
const before = await target.getAllowedUsage(actor);
expect(before.addons).toEqual({});
await target.updateAddonCredit(actor.user.uuid!, 5_000);
const after = await target.getAllowedUsage(actor);
expect(after.remaining).toBe(after.monthUsageAllowance + 5_000);
});
it('clamps remaining at zero when over allowance with no credits', async () => {
const sub = await target.getActorSubscription(actor);
await target.incrementUsage(
@@ -1578,7 +1578,7 @@ export class MeteringService extends PuterService {
}> {
const [userSubscription, addons, totals] = await Promise.all([
this.getActorSubscription(actor),
this.getActorAddons(actor),
this.#addonsForSettle(actor),
this.currentMonthTotals(actor),
]);
@@ -2244,7 +2244,7 @@ export class MeteringService extends PuterService {
: null;
}
/** `getActorAddons`, through `#spentAddons`. */
/** `getActorAddons`, through `#spentAddons`. Shared by gates and settles. */
async #addonsForSettle(actor: Actor): Promise<UsageAddons> {
const uuid = actor.user?.uuid;
const cached = uuid ? this.#spentAddons.get(uuid) : undefined;