refactor(core): improve streaming, validation, and rate limit logic
- Stream request body in blob upload fallback route to enforce size limits without buffering entire payload - Enhance admin provider panel error handling for multi-status responses - Adjust positive integer validation to require value >= 1 - Use full rate limit config for job event recording and prune by largest window - Fix PDF layout job event userId usage in docx-to-pdf upload route - Add missing windows array fallback in documents register route - Minor CI workflow and env example corrections - Update audiobooks blobstore test to use beforeAll directly These changes improve efficiency, correctness, and maintainability across API, admin, rate limiting, and test modules.
This commit is contained in:
parent
36fe5f66a3
commit
aa7700e844
9 changed files with 64 additions and 15 deletions
|
|
@ -15,7 +15,7 @@
|
||||||
API_BASE=http://localhost:8880/v1
|
API_BASE=http://localhost:8880/v1
|
||||||
API_KEY=api_key_optional
|
API_KEY=api_key_optional
|
||||||
|
|
||||||
# (Optional) TTS request/cache tuning (deaults shown below; leave unset to use defaults)
|
# (Optional) TTS request/cache tuning (defaults shown below; leave unset to use defaults)
|
||||||
# TTS_CACHE_MAX_SIZE_BYTES=268435456 # 256MB
|
# TTS_CACHE_MAX_SIZE_BYTES=268435456 # 256MB
|
||||||
# TTS_CACHE_TTL_MS=1800000 # 30 minutes
|
# TTS_CACHE_TTL_MS=1800000 # 30 minutes
|
||||||
# TTS_MAX_RETRIES=2
|
# TTS_MAX_RETRIES=2
|
||||||
|
|
|
||||||
2
.github/workflows/vitest.yml
vendored
2
.github/workflows/vitest.yml
vendored
|
|
@ -12,6 +12,8 @@ jobs:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v5
|
- uses: actions/checkout@v5
|
||||||
|
with:
|
||||||
|
persist-credentials: false
|
||||||
- uses: actions/setup-node@v5
|
- uses: actions/setup-node@v5
|
||||||
with:
|
with:
|
||||||
node-version: lts/*
|
node-version: lts/*
|
||||||
|
|
|
||||||
|
|
@ -46,14 +46,42 @@ export async function PUT(req: NextRequest) {
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
const body = Buffer.from(await req.arrayBuffer());
|
// Backstop for chunked/omitted Content-Length: stream the body so we can
|
||||||
// Backstop for chunked/omitted Content-Length: enforce on the actual bytes.
|
// bail out as soon as the running total crosses the cap instead of
|
||||||
if (body.byteLength > maxUploadBytes) {
|
// buffering the entire payload first.
|
||||||
|
const stream = req.body;
|
||||||
|
if (!stream) {
|
||||||
|
return NextResponse.json({ error: 'Missing request body' }, { status: 400 });
|
||||||
|
}
|
||||||
|
const reader = stream.getReader();
|
||||||
|
const chunks: Uint8Array[] = [];
|
||||||
|
let totalBytes = 0;
|
||||||
|
let overLimit = false;
|
||||||
|
try {
|
||||||
|
while (true) {
|
||||||
|
const { value, done } = await reader.read();
|
||||||
|
if (done) break;
|
||||||
|
if (!value) continue;
|
||||||
|
totalBytes += value.byteLength;
|
||||||
|
if (totalBytes > maxUploadBytes) {
|
||||||
|
overLimit = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
chunks.push(value);
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
if (overLimit) {
|
||||||
|
await reader.cancel().catch(() => {});
|
||||||
|
}
|
||||||
|
reader.releaseLock();
|
||||||
|
}
|
||||||
|
if (overLimit) {
|
||||||
return NextResponse.json(
|
return NextResponse.json(
|
||||||
{ error: `Upload exceeds the maximum allowed size of ${maxUploadBytes} bytes`, maxBytes: maxUploadBytes },
|
{ error: `Upload exceeds the maximum allowed size of ${maxUploadBytes} bytes`, maxBytes: maxUploadBytes },
|
||||||
{ status: 413 },
|
{ status: 413 },
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
const body = Buffer.concat(chunks, totalBytes);
|
||||||
|
|
||||||
const namespace = getOpenReaderTestNamespace(req.headers);
|
const namespace = getOpenReaderTestNamespace(req.headers);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -177,7 +177,7 @@ export async function POST(req: NextRequest) {
|
||||||
|
|
||||||
// Record upload-driven parse load (see register route for rationale).
|
// Record upload-driven parse load (see register route for rationale).
|
||||||
const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
|
const pdfRateConfig = getPdfLayoutRateConfig(await getResolvedRuntimeConfig());
|
||||||
await recordJobEvent(ctxOrRes.userId, 'pdf_layout', `docx:${randomUUID()}`, pdfRateConfig);
|
await recordJobEvent(storageUserId, 'pdf_layout', `docx:${randomUUID()}`, pdfRateConfig);
|
||||||
enqueueParsePdfJob({
|
enqueueParsePdfJob({
|
||||||
documentId: id,
|
documentId: id,
|
||||||
userId: storageUserId,
|
userId: storageUserId,
|
||||||
|
|
|
||||||
|
|
@ -196,7 +196,7 @@ export async function POST(req: NextRequest) {
|
||||||
// re-parse limiter reads. We record (not reject) here so a legitimate
|
// re-parse limiter reads. We record (not reject) here so a legitimate
|
||||||
// bulk upload always parses; the recorded load still throttles
|
// bulk upload always parses; the recorded load still throttles
|
||||||
// subsequent loopable re-parse spam via /parsed.
|
// subsequent loopable re-parse spam via /parsed.
|
||||||
await recordJobEvent(ctxOrRes.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig ?? { enabled: false });
|
await recordJobEvent(ctxOrRes.userId, 'pdf_layout', `register:${randomUUID()}`, pdfRateConfig ?? { enabled: false, windows: [] });
|
||||||
enqueueParsePdfJob({
|
enqueueParsePdfJob({
|
||||||
documentId: doc.id,
|
documentId: doc.id,
|
||||||
userId: storageUserId,
|
userId: storageUserId,
|
||||||
|
|
|
||||||
|
|
@ -72,7 +72,21 @@ async function patchDefaultProviderSlug(slug: string): Promise<void> {
|
||||||
headers: { 'Content-Type': 'application/json' },
|
headers: { 'Content-Type': 'application/json' },
|
||||||
body: JSON.stringify({ updates: { defaultTtsProvider: slug } }),
|
body: JSON.stringify({ updates: { defaultTtsProvider: slug } }),
|
||||||
});
|
});
|
||||||
if (!res.ok && res.status !== 207) throw new Error(`HTTP ${res.status}`);
|
if (res.status === 200 || res.status === 204) return;
|
||||||
|
if (res.status === 207) {
|
||||||
|
// Multi-status: the request succeeded for some keys but failed for others.
|
||||||
|
// Only treat it as success when the defaultTtsProvider field itself was
|
||||||
|
// accepted (no matching entry in the errors array).
|
||||||
|
const payload = (await res.json().catch(() => ({}))) as {
|
||||||
|
errors?: Array<{ key?: string; message?: string }>;
|
||||||
|
};
|
||||||
|
const failure = payload.errors?.find((entry) => entry?.key === 'defaultTtsProvider');
|
||||||
|
if (failure) {
|
||||||
|
throw new Error(failure.message || 'Failed to update default provider');
|
||||||
|
}
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
throw new Error(`HTTP ${res.status}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
async function patchProviderEnabled(input: { id: string; enabled: boolean }): Promise<void> {
|
async function patchProviderEnabled(input: { id: string; enabled: boolean }): Promise<void> {
|
||||||
|
|
|
||||||
|
|
@ -84,11 +84,11 @@ function positiveIntValue(defaultValue: number, envVar?: string): RuntimeConfigK
|
||||||
const trimmed = raw.trim();
|
const trimmed = raw.trim();
|
||||||
if (!trimmed) return undefined;
|
if (!trimmed) return undefined;
|
||||||
const parsed = Number(trimmed);
|
const parsed = Number(trimmed);
|
||||||
if (!Number.isFinite(parsed) || parsed <= 0) return undefined;
|
if (!Number.isFinite(parsed) || parsed < 1) return undefined;
|
||||||
return Math.floor(parsed);
|
return Math.floor(parsed);
|
||||||
},
|
},
|
||||||
validate(value) {
|
validate(value) {
|
||||||
if (typeof value !== 'number' || !Number.isFinite(value) || value <= 0) return undefined;
|
if (typeof value !== 'number' || !Number.isFinite(value) || value < 1) return undefined;
|
||||||
return Math.floor(value);
|
return Math.floor(value);
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
|
|
|
||||||
|
|
@ -131,7 +131,7 @@ export async function recordJobEvent(
|
||||||
userId: string | null | undefined,
|
userId: string | null | undefined,
|
||||||
action: JobRateAction,
|
action: JobRateAction,
|
||||||
opId: string,
|
opId: string,
|
||||||
config: Pick<JobRateConfig, 'enabled'>,
|
config: JobRateConfig,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
if (!isActive(config, userId) || !opId) return;
|
if (!isActive(config, userId) || !opId) return;
|
||||||
|
|
||||||
|
|
@ -145,15 +145,20 @@ export async function recordJobEvent(
|
||||||
// Recording is best-effort; never block op creation on ledger writes.
|
// Recording is best-effort; never block op creation on ledger writes.
|
||||||
}
|
}
|
||||||
|
|
||||||
// Opportunistic prune of rows older than a generous retention window so the
|
// Opportunistic prune of rows older than the largest configured window so
|
||||||
// ledger stays small without a separate cron.
|
// the ledger stays small without a separate cron, but never deletes events
|
||||||
|
// that could still affect an in-window count.
|
||||||
if (Math.random() < 0.05) {
|
if (Math.random() < 0.05) {
|
||||||
|
const largestWindowMs = config.windows.reduce(
|
||||||
|
(max, w) => (Number.isFinite(w.windowMs) && w.windowMs > max ? w.windowMs : max),
|
||||||
|
24 * 60 * 60 * 1000,
|
||||||
|
);
|
||||||
try {
|
try {
|
||||||
await safeDb()
|
await safeDb()
|
||||||
.delete(userJobEvents)
|
.delete(userJobEvents)
|
||||||
.where(and(
|
.where(and(
|
||||||
eq(userJobEvents.action, action),
|
eq(userJobEvents.action, action),
|
||||||
lt(userJobEvents.createdAt, now - 24 * 60 * 60 * 1000),
|
lt(userJobEvents.createdAt, now - largestWindowMs),
|
||||||
));
|
));
|
||||||
} catch {
|
} catch {
|
||||||
// ignore prune failures
|
// ignore prune failures
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
import { describe, expect, test } from 'vitest';
|
import { beforeAll, describe, expect, test } from 'vitest';
|
||||||
import {
|
import {
|
||||||
audiobookKey,
|
audiobookKey,
|
||||||
audiobookPrefix,
|
audiobookPrefix,
|
||||||
|
|
@ -15,7 +15,7 @@ function configureS3Env() {
|
||||||
}
|
}
|
||||||
|
|
||||||
describe('audiobooks-blobstore', () => {
|
describe('audiobooks-blobstore', () => {
|
||||||
test.beforeAll(() => {
|
beforeAll(() => {
|
||||||
configureS3Env();
|
configureS3Env();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue