Upload fragmenté : pool de connexions partagé et configurable
Remplace les 3 workers par fichier par un pool global de morceaux en vol, réparti équitablement entre les fichiers fragmentés actifs : 1 fichier utilise tout le pool, N fichiers en reçoivent une part chacun (ceil(taille / N)), la capacité libérée par un fichier terminé est redistribuée. Un fichier en cours d'init compte dans la part pour ne pas être affamé ; les morceaux déjà en vol ne sont jamais interrompus. La taille du pool est réglable via `chunkPoolSize` (config.json, défaut 30, borné à 256) et transmise au navigateur par un nouveau `GET /config`. Corps des morceaux repassé en streaming (writeChunkAt sur un flux + garde `maxBytes`) : mémoire ~constante même à pool élevé, plus de bufferisation de 16 Mo par requête. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
3a555c0bc9
commit
64a38b1e41
13 changed files with 291 additions and 79 deletions
14
README.md
14
README.md
|
|
@ -116,6 +116,7 @@ pouvant écrire les paquets de l'organisation.
|
|||
| `GET /auth/callback` | non | Fin du flux : ouvre la session, redirige vers `/` |
|
||||
| `POST /auth/logout` | session | Ferme la session (+ déconnexion RP si disponible) |
|
||||
| `GET /auth/me` | non | `{ "authenticated": bool, "name"?: string }` |
|
||||
| `GET /config` | non | Réglages client : `{ "chunkPoolSize": number }` |
|
||||
| `GET /targets` | session | Cibles : `[{ "name", "label", "allowDirectories" }]` (sans chemin) |
|
||||
| `POST /upload?target=<nom>[&path=<p>]` | session | Dépôt direct d'un fichier `multipart/form-data`, champ `file` |
|
||||
| `POST /upload/init` | session | Démarre un upload fragmenté → `{ id, chunkSize, chunkCount }` |
|
||||
|
|
@ -155,9 +156,17 @@ La page gère aussi le **glisser-déposer d'un dossier entier** et un bouton
|
|||
### Gros fichiers — upload fragmenté
|
||||
|
||||
La page bascule automatiquement en **upload fragmenté** au-delà de **32 Mo** :
|
||||
morceaux de 16 Mo, **3 en parallèle**, chaque morceau ré-essayé jusqu'à 3 fois.
|
||||
Aucune configuration ; `maxFileSizeMb` reste la limite (`10240` = 10 Go).
|
||||
morceaux de 16 Mo, chacun ré-essayé jusqu'à 3 fois. `maxFileSizeMb` reste la
|
||||
limite (`10240` = 10 Go).
|
||||
|
||||
- **Pool de connexions partagé** (`chunkPoolSize`, défaut `30`) : budget global de
|
||||
morceaux en vol, réparti équitablement entre les fichiers fragmentés en cours.
|
||||
1 fichier ⇒ il utilise tout le pool ; `N` fichiers ⇒ `⌈30 / N⌉` morceaux chacun ;
|
||||
quand un fichier se termine, sa part est redistribuée. Le navigateur lit la
|
||||
valeur via `GET /config`.
|
||||
- **HTTP/2 recommandé** : en HTTP/1.1 le navigateur plafonne à ~6 connexions par
|
||||
origine, donc un `chunkPoolSize` élevé n'a d'effet réel qu'en HTTP/2 (activer
|
||||
« HTTP/2 Support » sur l'hôte du reverse proxy).
|
||||
- Écriture en place à l'offset du morceau, mémoire constante ; assemblage sans
|
||||
seconde passe. Fichier partiel en `<nom>.mediadrop-part`, renommé à la
|
||||
finalisation.
|
||||
|
|
@ -187,6 +196,7 @@ Un seul fichier JSON (voir [`config.example.json`](config.example.json), schéma
|
|||
| `targets[].allowDirectories` | `false` | Autoriser le dépôt de dossiers (arborescence recréée sous `dir`) |
|
||||
| `maxFileSizeMb` | `512` | Taille maximale par fichier, en Mo (`10240` = 10 Go) |
|
||||
| `logLevel` | `info` | `fatal` \| `error` \| `warn` \| `info` \| `debug` \| `trace` \| `silent` |
|
||||
| `chunkPoolSize` | `30` | Morceaux en vol simultanés (upload fragmenté), répartis entre fichiers actifs ; `1`–`256`. Utile en HTTP/2 |
|
||||
| `allowedMimeTypes` | _(liste média)_ | `"all"`, ou tableau de types exacts et/ou jokers de préfixe (`"video/*"`, `"*"`) ; absent => liste par défaut |
|
||||
| `session.key` | — | **Obligatoire.** 32 octets base64 (`openssl rand -base64 32`) |
|
||||
| `session.maxAgeHours` | `12` | Durée de vie absolue de la session |
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@
|
|||
],
|
||||
"maxFileSizeMb": 512,
|
||||
"logLevel": "info",
|
||||
"chunkPoolSize": 30,
|
||||
"allowedMimeTypes": ["image/jpeg", "image/png", "image/webp", "video/mp4", "audio/mpeg"],
|
||||
"session": {
|
||||
"key": "REMPLACER_PAR_32_OCTETS_BASE64_openssl_rand_base64_32",
|
||||
|
|
|
|||
|
|
@ -43,6 +43,13 @@
|
|||
"enum": ["fatal", "error", "warn", "info", "debug", "trace", "silent"],
|
||||
"default": "info"
|
||||
},
|
||||
"chunkPoolSize": {
|
||||
"type": "integer",
|
||||
"minimum": 1,
|
||||
"maximum": 256,
|
||||
"default": 30,
|
||||
"description": "Nombre max de requêtes de morceaux en vol, réparti par le navigateur entre les fichiers fragmentés actifs (1 fichier => tout le pool, N fichiers => une part chacun). Efficace surtout en HTTP/2."
|
||||
},
|
||||
"allowedMimeTypes": {
|
||||
"description": "\"all\", ou liste de types exacts et/ou jokers de préfixe (\"video/*\", \"*/*\", \"*\"). Absent => liste média par défaut.",
|
||||
"oneOf": [
|
||||
|
|
|
|||
|
|
@ -262,9 +262,10 @@
|
|||
const queueCancel = document.getElementById('queueCancel');
|
||||
const list = document.getElementById('list');
|
||||
|
||||
const MAX_PARALLEL = 2;
|
||||
const MAX_PARALLEL = 3; // dépôts directs (petits fichiers) simultanés
|
||||
const MAX_CHUNKED_JOBS = 32; // fichiers fragmentés actifs à la fois (< 100 côté serveur)
|
||||
const CHUNK_THRESHOLD = 32 * 1024 * 1024; // au-delà : upload fragmenté et reprenable
|
||||
const CHUNK_PARALLEL = 3;
|
||||
let CHUNK_POOL = 30; // budget global de morceaux en vol ; remplacé par GET /config
|
||||
const CHUNK_RETRIES = 3;
|
||||
|
||||
let authenticated = false;
|
||||
|
|
@ -381,7 +382,6 @@
|
|||
ui.status.className = 'status ' + (kind || 'muted');
|
||||
}
|
||||
|
||||
// File d'attente : au plus MAX_PARALLEL transferts simultanés.
|
||||
const queue = [];
|
||||
const active = new Set();
|
||||
|
||||
|
|
@ -398,15 +398,30 @@
|
|||
}
|
||||
|
||||
function pump() {
|
||||
while (active.size < MAX_PARALLEL && queue.length > 0) {
|
||||
startUpload(queue.shift());
|
||||
let simple = 0;
|
||||
let chunked = 0;
|
||||
for (const j of active) {
|
||||
if (j.chunked) chunked += 1;
|
||||
else simple += 1;
|
||||
}
|
||||
const kept = [];
|
||||
for (const job of queue) {
|
||||
if (job.chunked ? chunked < MAX_CHUNKED_JOBS : simple < MAX_PARALLEL) {
|
||||
if (job.chunked) chunked += 1;
|
||||
else simple += 1;
|
||||
startUpload(job);
|
||||
} else {
|
||||
kept.push(job);
|
||||
}
|
||||
}
|
||||
queue.length = 0;
|
||||
for (const job of kept) queue.push(job);
|
||||
updateQueueRow();
|
||||
}
|
||||
|
||||
function enqueue(file, relativePath) {
|
||||
const ui = addItem(file, relativePath);
|
||||
const job = { file, relativePath, ui };
|
||||
const job = { file, relativePath, ui, chunked: file.size > CHUNK_THRESHOLD };
|
||||
queue.push(job);
|
||||
ui.cancelBtn.hidden = false;
|
||||
ui.cancelBtn.onclick = () => {
|
||||
|
|
@ -420,6 +435,64 @@
|
|||
updateQueueRow();
|
||||
}
|
||||
|
||||
// Pool global de morceaux en vol, réparti équitablement entre les fichiers
|
||||
// fragmentés actifs. Chaque fichier est plafonné à sa part équitable
|
||||
// (ceil(CHUNK_POOL / nb de fichiers actifs)) : 1 fichier => tout le pool,
|
||||
// N fichiers => une part chacun. Un fichier encore en cours d'initialisation
|
||||
// (`priming`) compte dans la part pour lui réserver un slot ; à égalité de
|
||||
// morceaux en vol, le fichier le moins servi passe en premier.
|
||||
const chunkPool = {
|
||||
free: CHUNK_POOL,
|
||||
entries: [],
|
||||
add(entry) {
|
||||
this.entries.push(entry);
|
||||
this.tick();
|
||||
},
|
||||
remove(entry) {
|
||||
const i = this.entries.indexOf(entry);
|
||||
if (i >= 0) this.entries.splice(i, 1);
|
||||
this.tick();
|
||||
},
|
||||
tick() {
|
||||
while (this.free > 0) {
|
||||
const contenders = this.entries.filter(
|
||||
(e) => !e.stopped && (e.pending.length > 0 || e.priming),
|
||||
);
|
||||
if (contenders.length === 0) return;
|
||||
const share = Math.max(1, Math.ceil(CHUNK_POOL / contenders.length));
|
||||
let best = null;
|
||||
for (const e of this.entries) {
|
||||
if (e.stopped || e.pending.length === 0 || e.inflight >= share) continue;
|
||||
if (!best || e.inflight < best.inflight) best = e;
|
||||
}
|
||||
if (!best) return;
|
||||
const target = best;
|
||||
const index = target.pending.shift();
|
||||
target.inflight += 1;
|
||||
this.free -= 1;
|
||||
target.send(index).finally(() => {
|
||||
target.inflight -= 1;
|
||||
this.free += 1;
|
||||
this.tick();
|
||||
});
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
// Taille du pool réglée côté serveur (config.json -> GET /config). Ajuste
|
||||
// `free` du même delta : sûr même si des morceaux sont déjà en vol.
|
||||
fetch('/config', { headers: { accept: 'application/json' } })
|
||||
.then((r) => (r.ok ? r.json() : null))
|
||||
.then((cfg) => {
|
||||
const n = cfg && Number(cfg.chunkPoolSize);
|
||||
if (Number.isInteger(n) && n > 0 && n !== CHUNK_POOL) {
|
||||
chunkPool.free += n - CHUNK_POOL;
|
||||
CHUNK_POOL = n;
|
||||
chunkPool.tick();
|
||||
}
|
||||
})
|
||||
.catch(() => {});
|
||||
|
||||
function makeReporter(ui, total) {
|
||||
let started = 0;
|
||||
let lastAt = 0;
|
||||
|
|
@ -527,21 +600,43 @@
|
|||
async function runChunked(job) {
|
||||
const { file, relativePath, ui } = job;
|
||||
const report = makeReporter(ui, file.size);
|
||||
const inflight = new Set();
|
||||
const inflight = new Set(); // XHR en vol (pour l'annulation)
|
||||
let aborted = false;
|
||||
let failed = false;
|
||||
let sessionId = null;
|
||||
|
||||
let resolveAll = null;
|
||||
const finished = new Promise((r) => {
|
||||
resolveAll = r;
|
||||
});
|
||||
let settled = false;
|
||||
const finalize = (outcome) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
resolveAll(outcome);
|
||||
}
|
||||
};
|
||||
|
||||
const dropSession = () => {
|
||||
if (sessionId) fetch('/upload/' + sessionId, { method: 'DELETE' }).catch(() => {});
|
||||
};
|
||||
|
||||
// Inscrit dans le pool dès maintenant, avant l'init réseau : le
|
||||
// planificateur compte ce fichier dans la part équitable (`priming`) et
|
||||
// lui réserve un slot même si son init n'a pas encore répondu.
|
||||
const entry = { inflight: 0, stopped: false, priming: true, pending: [], send: null };
|
||||
chunkPool.add(entry);
|
||||
|
||||
job.cancel = () => {
|
||||
aborted = true;
|
||||
entry.stopped = true;
|
||||
entry.pending.length = 0;
|
||||
for (const x of inflight) x.abort();
|
||||
chunkPool.remove(entry);
|
||||
dropSession();
|
||||
finish(job);
|
||||
ui.progress.textContent = '';
|
||||
setStatus(ui, 'Annulé', 'err');
|
||||
finalize('abort');
|
||||
};
|
||||
|
||||
let meta;
|
||||
|
|
@ -557,11 +652,15 @@
|
|||
});
|
||||
meta = await res.json().catch(() => ({}));
|
||||
if (!res.ok) {
|
||||
entry.stopped = true;
|
||||
chunkPool.remove(entry);
|
||||
finish(job);
|
||||
applyResult(job, res.status, meta);
|
||||
return;
|
||||
}
|
||||
} catch {
|
||||
entry.stopped = true;
|
||||
chunkPool.remove(entry);
|
||||
finish(job);
|
||||
setStatus(ui, 'Erreur réseau (init)', 'err');
|
||||
return;
|
||||
|
|
@ -571,11 +670,37 @@
|
|||
const chunkSize = meta.chunkSize;
|
||||
const chunkCount = meta.chunkCount;
|
||||
const loadedPer = new Array(chunkCount).fill(0);
|
||||
const attempts = new Array(chunkCount).fill(0);
|
||||
const emit = () => report(loadedPer.reduce((a, b) => a + b, 0));
|
||||
let nextIndex = 0;
|
||||
let doneCount = 0;
|
||||
|
||||
const sendOnce = (index) =>
|
||||
new Promise((resolve) => {
|
||||
function retry(index, status, releasePermit) {
|
||||
if (aborted || entry.stopped) {
|
||||
releasePermit();
|
||||
return;
|
||||
}
|
||||
attempts[index] += 1;
|
||||
if (attempts[index] > CHUNK_RETRIES) {
|
||||
entry.stopped = true;
|
||||
entry.pending.length = 0;
|
||||
setStatus(ui, 'Erreur morceau ' + index + ' (' + (status || 'réseau') + ')', 'err');
|
||||
finalize('fail');
|
||||
releasePermit();
|
||||
return;
|
||||
}
|
||||
setTimeout(() => {
|
||||
if (!aborted && !entry.stopped) {
|
||||
entry.pending.unshift(index);
|
||||
chunkPool.tick();
|
||||
}
|
||||
}, 400 * attempts[index]);
|
||||
releasePermit(); // libère le permis ; le morceau est re-planifié après le backoff
|
||||
}
|
||||
|
||||
entry.pending = Array.from({ length: chunkCount }, (_, i) => i);
|
||||
entry.priming = false;
|
||||
entry.send = (index) =>
|
||||
new Promise((releasePermit) => {
|
||||
const start = index * chunkSize;
|
||||
const end = Math.min(start + chunkSize, file.size);
|
||||
const xhr = new XMLHttpRequest();
|
||||
|
|
@ -588,51 +713,42 @@
|
|||
emit();
|
||||
}
|
||||
});
|
||||
const settle = (r) => {
|
||||
inflight.delete(xhr);
|
||||
resolve(r);
|
||||
};
|
||||
xhr.addEventListener('load', () => {
|
||||
inflight.delete(xhr);
|
||||
if (xhr.status >= 200 && xhr.status < 300) {
|
||||
loadedPer[index] = end - start;
|
||||
emit();
|
||||
settle({ ok: true });
|
||||
doneCount += 1;
|
||||
if (doneCount === chunkCount) finalize('ok');
|
||||
releasePermit();
|
||||
} else if (xhr.status === 401) {
|
||||
entry.stopped = true;
|
||||
entry.pending.length = 0;
|
||||
applyResult(job, 401, {});
|
||||
finalize('fail');
|
||||
releasePermit();
|
||||
} else {
|
||||
settle({ ok: false, status: xhr.status });
|
||||
retry(index, xhr.status, releasePermit);
|
||||
}
|
||||
});
|
||||
xhr.addEventListener('error', () => settle({ ok: false, status: 0 }));
|
||||
xhr.addEventListener('abort', () => settle({ ok: false, aborted: true }));
|
||||
xhr.addEventListener('error', () => {
|
||||
inflight.delete(xhr);
|
||||
retry(index, 0, releasePermit);
|
||||
});
|
||||
xhr.addEventListener('abort', () => {
|
||||
inflight.delete(xhr);
|
||||
releasePermit();
|
||||
});
|
||||
xhr.send(file.slice(start, end));
|
||||
});
|
||||
|
||||
async function worker() {
|
||||
while (!aborted && !failed && nextIndex < chunkCount) {
|
||||
const index = nextIndex;
|
||||
nextIndex += 1;
|
||||
let r = { ok: false };
|
||||
for (let attempt = 0; attempt <= CHUNK_RETRIES; attempt += 1) {
|
||||
r = await sendOnce(index);
|
||||
if (r.ok || r.aborted || r.status === 401) break;
|
||||
await new Promise((res) => setTimeout(res, 400 * (attempt + 1)));
|
||||
if (aborted) return;
|
||||
}
|
||||
if (r.ok) continue;
|
||||
if (r.aborted) return;
|
||||
failed = true;
|
||||
if (r.status === 401) applyResult(job, 401, {});
|
||||
else
|
||||
setStatus(ui, 'Erreur morceau ' + index + ' (' + (r.status || 'réseau') + ')', 'err');
|
||||
return;
|
||||
}
|
||||
}
|
||||
chunkPool.tick();
|
||||
const outcome = await finished;
|
||||
chunkPool.remove(entry);
|
||||
|
||||
const workers = [];
|
||||
for (let i = 0; i < Math.min(CHUNK_PARALLEL, chunkCount); i += 1) workers.push(worker());
|
||||
await Promise.all(workers);
|
||||
|
||||
if (aborted) return;
|
||||
if (failed) {
|
||||
if (outcome === 'abort') return;
|
||||
if (outcome === 'fail') {
|
||||
for (const x of inflight) x.abort();
|
||||
finish(job);
|
||||
dropSession();
|
||||
return;
|
||||
|
|
|
|||
14
src/app.ts
14
src/app.ts
|
|
@ -11,7 +11,7 @@ import { ensureUploadDir, removeFile } from './lib/storage.js';
|
|||
import { createUploadSessionStore } from './lib/upload-sessions.js';
|
||||
import { createOidcClient, type OidcClient } from './oidc.js';
|
||||
import { authRoutes } from './routes/auth.js';
|
||||
import { chunkedUploadRoutes, CHUNK_BODY_LIMIT } from './routes/upload-chunked.js';
|
||||
import { chunkedUploadRoutes } from './routes/upload-chunked.js';
|
||||
import { healthRoutes } from './routes/health.js';
|
||||
import { uploadRoutes } from './routes/upload.js';
|
||||
|
||||
|
|
@ -59,14 +59,10 @@ export async function buildApp(
|
|||
uploadSessions.stop();
|
||||
});
|
||||
|
||||
// Corps brut (bufferisé) pour les morceaux d'upload fragmenté.
|
||||
app.addContentTypeParser(
|
||||
'application/octet-stream',
|
||||
{ parseAs: 'buffer', bodyLimit: CHUNK_BODY_LIMIT },
|
||||
(_request, body, taskDone) => {
|
||||
taskDone(null, body);
|
||||
},
|
||||
);
|
||||
// Corps brut (flux) pour les morceaux d'upload fragmenté ; écrit en streaming.
|
||||
app.addContentTypeParser('application/octet-stream', (_request, payload, taskDone) => {
|
||||
taskDone(null, payload);
|
||||
});
|
||||
|
||||
await app.register(secureSession, {
|
||||
key: config.session.key,
|
||||
|
|
|
|||
|
|
@ -52,6 +52,12 @@ export interface AppConfig {
|
|||
maxFileSizeBytes: number;
|
||||
allowedMimeTypes: MimePolicy;
|
||||
logLevel: LogLevel;
|
||||
/**
|
||||
* Nombre maximum de requêtes de morceaux en vol simultanément, réparti par le
|
||||
* client entre les fichiers fragmentés actifs (1 fichier => tout le pool,
|
||||
* N fichiers => une part chacun). Transmis au navigateur via `GET /config`.
|
||||
*/
|
||||
chunkPoolSize: number;
|
||||
session: SessionConfig;
|
||||
oidc: OidcConfig;
|
||||
}
|
||||
|
|
@ -66,8 +72,12 @@ const DEFAULTS = {
|
|||
maxFileSizeMb: 512,
|
||||
logLevel: 'info',
|
||||
sessionMaxAgeHours: 12,
|
||||
chunkPoolSize: 30,
|
||||
} as const;
|
||||
|
||||
/** Garde-fou : au-delà, la mémoire tampon et le nombre de sockets deviennent déraisonnables. */
|
||||
const MAX_CHUNK_POOL_SIZE = 256;
|
||||
|
||||
const TARGET_NAME_PATTERN = /^[A-Za-z0-9._-]+$/;
|
||||
|
||||
function asRecord(value: unknown, label: string): Record<string, unknown> {
|
||||
|
|
@ -123,6 +133,16 @@ function optionalBoolean(value: unknown, fallback: boolean, name: string): boole
|
|||
return value;
|
||||
}
|
||||
|
||||
function parseChunkPoolSize(value: unknown): number {
|
||||
const parsed = optionalPositiveInt(value, DEFAULTS.chunkPoolSize, 'chunkPoolSize');
|
||||
if (parsed > MAX_CHUNK_POOL_SIZE) {
|
||||
throw new ConfigError(
|
||||
`chunkPoolSize ne doit pas dépasser ${String(MAX_CHUNK_POOL_SIZE)} (reçu : ${String(parsed)})`,
|
||||
);
|
||||
}
|
||||
return parsed;
|
||||
}
|
||||
|
||||
function parseLogLevel(value: unknown): LogLevel {
|
||||
if (value === undefined || value === null) {
|
||||
return DEFAULTS.logLevel;
|
||||
|
|
@ -276,6 +296,7 @@ export function parseConfig(raw: unknown): AppConfig {
|
|||
1024,
|
||||
allowedMimeTypes: parseMimePolicy(root.allowedMimeTypes),
|
||||
logLevel: parseLogLevel(root.logLevel),
|
||||
chunkPoolSize: parseChunkPoolSize(root.chunkPoolSize),
|
||||
session: parseSession(root.session),
|
||||
oidc: parseOidc(root.oidc),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -185,22 +185,42 @@ export async function createEmptyFile(filePath: string): Promise<void> {
|
|||
}
|
||||
|
||||
/**
|
||||
* Écrit `data` dans `filePath` à partir de l'octet `offset` (le fichier doit
|
||||
* exister), en une passe positionnée : sûr pour des écritures concurrentes sur
|
||||
* des zones disjointes du même fichier.
|
||||
* Écrit le flux `source` dans `filePath` à partir de l'octet `offset` (le fichier
|
||||
* doit exister). Refuse au-delà de `maxBytes`. Streaming : mémoire ~constante,
|
||||
* même à forte concurrence. Sûr pour des écritures sur des zones disjointes.
|
||||
*/
|
||||
export async function writeChunkAt(params: {
|
||||
data: Buffer;
|
||||
source: Readable;
|
||||
filePath: string;
|
||||
offset: number;
|
||||
maxBytes: number;
|
||||
}): Promise<number> {
|
||||
const handle = await open(params.filePath, 'r+');
|
||||
try {
|
||||
await handle.write(params.data, 0, params.data.length, params.offset);
|
||||
} finally {
|
||||
await handle.close();
|
||||
let written = 0;
|
||||
const limiter = new Transform({
|
||||
transform(chunk: Buffer, _encoding, callback): void {
|
||||
written += chunk.length;
|
||||
if (written > params.maxBytes) {
|
||||
callback(new PayloadTooLargeError('Morceau plus grand que la taille annoncée'));
|
||||
return;
|
||||
}
|
||||
return params.data.length;
|
||||
callback(null, chunk);
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
await pipeline(
|
||||
params.source,
|
||||
limiter,
|
||||
createWriteStream(params.filePath, {
|
||||
flags: 'r+',
|
||||
start: params.offset,
|
||||
highWaterMark: WRITE_HIGH_WATER_MARK,
|
||||
}),
|
||||
);
|
||||
} catch (error) {
|
||||
throw normalizeStreamError(error);
|
||||
}
|
||||
return written;
|
||||
}
|
||||
|
||||
/** Renomme le fichier partiel vers son chemin final (écrase si présent). */
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { stat } from 'node:fs/promises';
|
||||
import type { Readable } from 'node:stream';
|
||||
|
||||
import type { FastifyPluginCallback, FastifyReply } from 'fastify';
|
||||
|
||||
|
|
@ -19,9 +20,6 @@ const MAX_CHUNK = 32 * 1024 * 1024;
|
|||
const DEFAULT_CHUNK = 16 * 1024 * 1024;
|
||||
const PART_SUFFIX = '.mediadrop-part';
|
||||
|
||||
/** Limite de corps pour un morceau (à appliquer au parser octet-stream). */
|
||||
export const CHUNK_BODY_LIMIT = MAX_CHUNK + 4096;
|
||||
|
||||
interface InitBody {
|
||||
target?: unknown;
|
||||
path?: unknown;
|
||||
|
|
@ -124,7 +122,7 @@ export const chunkedUploadRoutes: FastifyPluginCallback = (app, _options, done)
|
|||
|
||||
app.put(
|
||||
'/upload/:id/:index',
|
||||
{ preHandler: requireSession, bodyLimit: CHUNK_BODY_LIMIT },
|
||||
{ preHandler: requireSession },
|
||||
async (request): Promise<ChunkAck> => {
|
||||
const { id, index } = request.params as { id: string; index: string };
|
||||
const session = app.uploadSessions.use(id);
|
||||
|
|
@ -134,18 +132,11 @@ export const chunkedUploadRoutes: FastifyPluginCallback = (app, _options, done)
|
|||
throw new ValidationError('Index de morceau hors limites');
|
||||
}
|
||||
|
||||
const data = request.body;
|
||||
if (!Buffer.isBuffer(data)) {
|
||||
throw new ValidationError('Corps du morceau attendu (application/octet-stream)');
|
||||
}
|
||||
if (data.length > session.chunkSize) {
|
||||
throw new PayloadTooLargeError('Morceau plus grand que la taille annoncée');
|
||||
}
|
||||
|
||||
await writeChunkAt({
|
||||
data,
|
||||
source: request.body as Readable,
|
||||
filePath: session.partPath,
|
||||
offset: chunkIndex * session.chunkSize,
|
||||
maxBytes: session.chunkSize,
|
||||
});
|
||||
session.received.add(chunkIndex);
|
||||
|
||||
|
|
|
|||
|
|
@ -23,6 +23,11 @@ interface TargetSummary {
|
|||
allowDirectories: boolean;
|
||||
}
|
||||
|
||||
interface ClientConfig {
|
||||
/** Taille du pool de connexions pour l'upload fragmenté (réglage client). */
|
||||
chunkPoolSize: number;
|
||||
}
|
||||
|
||||
interface UploadQuery {
|
||||
target?: unknown;
|
||||
path?: unknown;
|
||||
|
|
@ -39,10 +44,13 @@ function toResponse(target: DropTarget, stored: StoredFile): UploadResponse {
|
|||
}
|
||||
|
||||
/**
|
||||
* - `GET /config` — réglages client non sensibles (taille du pool de morceaux)
|
||||
* - `GET /targets` — liste des cibles (nom, libellé, dépôt de dossier autorisé)
|
||||
* - `POST /upload?target=<nom>[&path=<chemin relatif>]` — dépôt d'un fichier multipart
|
||||
*/
|
||||
export const uploadRoutes: FastifyPluginCallback = (app, _options, done) => {
|
||||
app.get('/config', (): ClientConfig => ({ chunkPoolSize: app.appConfig.chunkPoolSize }));
|
||||
|
||||
app.get('/targets', { preHandler: requireSession }, (): TargetSummary[] =>
|
||||
app.appConfig.targets.map((target) => ({
|
||||
name: target.name,
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ describe('parseConfig', () => {
|
|||
publicUrl: 'http://localhost:3000',
|
||||
maxFileSizeBytes: 512 * 1024 * 1024,
|
||||
logLevel: 'info',
|
||||
chunkPoolSize: 30,
|
||||
});
|
||||
expect(config.targets).toEqual([
|
||||
{ name: 'photos', label: 'photos', dir: './uploads/photos', allowDirectories: false },
|
||||
|
|
@ -121,6 +122,14 @@ describe('parseConfig', () => {
|
|||
expect(() => parseConfig({ ...BASE, logLevel: 'verbose' })).toThrow(ConfigError);
|
||||
});
|
||||
|
||||
it('lit et borne chunkPoolSize', () => {
|
||||
expect(parseConfig({ ...BASE, chunkPoolSize: 8 }).chunkPoolSize).toBe(8);
|
||||
expect(() => parseConfig({ ...BASE, chunkPoolSize: 0 })).toThrow(ConfigError);
|
||||
expect(() => parseConfig({ ...BASE, chunkPoolSize: -4 })).toThrow(ConfigError);
|
||||
expect(() => parseConfig({ ...BASE, chunkPoolSize: 12.5 })).toThrow(ConfigError);
|
||||
expect(() => parseConfig({ ...BASE, chunkPoolSize: 257 })).toThrow(ConfigError);
|
||||
});
|
||||
|
||||
it('interprète allowedMimeTypes', () => {
|
||||
expect(
|
||||
parseConfig({ ...BASE, allowedMimeTypes: ['image/png', 'IMAGE/WEBP'] }).allowedMimeTypes,
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ export function makeConfig(overrides: Partial<AppConfig> = {}): AppConfig {
|
|||
maxFileSizeBytes: 1024 * 1024,
|
||||
allowedMimeTypes: ['image/png', 'image/jpeg'],
|
||||
logLevel: 'silent',
|
||||
chunkPoolSize: 30,
|
||||
session: { key: Buffer.alloc(32, 7), maxAgeSeconds: 3600 },
|
||||
oidc: {
|
||||
issuer: 'https://idp.test',
|
||||
|
|
|
|||
|
|
@ -123,9 +123,10 @@ describe('writeChunkAt — réassemblage de morceaux', () => {
|
|||
for (const index of [2, 0, 1]) {
|
||||
const slice = data.subarray(index * chunkSize, (index + 1) * chunkSize);
|
||||
const written = await writeChunkAt({
|
||||
data: slice,
|
||||
source: Readable.from(slice),
|
||||
filePath: partPath,
|
||||
offset: index * chunkSize,
|
||||
maxBytes: chunkSize,
|
||||
});
|
||||
expect(written).toBe(slice.length);
|
||||
}
|
||||
|
|
@ -137,6 +138,19 @@ describe('writeChunkAt — réassemblage de morceaux', () => {
|
|||
expect(await readdir(path.join(uploadDir, 'sub'))).toEqual(['video.bin']);
|
||||
});
|
||||
|
||||
it('rejette un morceau plus grand que maxBytes', async () => {
|
||||
const partPath = path.join(uploadDir, 'p.part');
|
||||
await createEmptyFile(partPath);
|
||||
await expect(
|
||||
writeChunkAt({
|
||||
source: Readable.from(Buffer.alloc(2048)),
|
||||
filePath: partPath,
|
||||
offset: 0,
|
||||
maxBytes: 1024,
|
||||
}),
|
||||
).rejects.toBeInstanceOf(PayloadTooLargeError);
|
||||
});
|
||||
|
||||
it('removeFile ne lève pas si le fichier est absent', async () => {
|
||||
await expect(removeFile(path.join(uploadDir, 'absent'))).resolves.toBeUndefined();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -97,6 +97,24 @@ describe('GET /health', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('GET /config', () => {
|
||||
it('expose chunkPoolSize sans authentification', async () => {
|
||||
const res = await app.inject({ method: 'GET', url: '/config' });
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(res.json()).toEqual({ chunkPoolSize: 30 });
|
||||
});
|
||||
|
||||
it('reflète la valeur configurée', async () => {
|
||||
const custom = await buildApp(makeConfig({ chunkPoolSize: 12, oidc: oidcOverride(idp) }), {
|
||||
logger: false,
|
||||
});
|
||||
await custom.ready();
|
||||
const res = await custom.inject({ method: 'GET', url: '/config' });
|
||||
expect(res.json<{ chunkPoolSize: number }>().chunkPoolSize).toBe(12);
|
||||
await custom.close();
|
||||
});
|
||||
});
|
||||
|
||||
describe('GET /targets', () => {
|
||||
it('exige une session', async () => {
|
||||
const res = await app.inject({ method: 'GET', url: '/targets' });
|
||||
|
|
|
|||
Loading…
Reference in a new issue