Compare commits
No commits in common. "4637d1e0b7323b332591a54bfbc43eba4b4a315b" and "1cb0da05e25777979eb304d6f7d0a94c182f04c0" have entirely different histories.
4637d1e0b7
...
1cb0da05e2
3 changed files with 6 additions and 93 deletions
|
|
@ -750,31 +750,8 @@
|
||||||
chunkPool.tick();
|
chunkPool.tick();
|
||||||
};
|
};
|
||||||
|
|
||||||
// Reprise après un échec : la session serveur est conservée dans `job.sessionId`,
|
|
||||||
// on récupère les morceaux déjà reçus pour ne renvoyer que les manquants.
|
|
||||||
let meta;
|
let meta;
|
||||||
let alreadyReceived = [];
|
for (;;) {
|
||||||
if (job.sessionId) {
|
|
||||||
try {
|
|
||||||
const status = await fetch('/upload/' + job.sessionId);
|
|
||||||
if (aborted) return;
|
|
||||||
if (status.ok) {
|
|
||||||
meta = await status.json();
|
|
||||||
alreadyReceived = Array.isArray(meta.received) ? meta.received : [];
|
|
||||||
} else {
|
|
||||||
job.sessionId = null; // session expirée côté serveur : on repart de zéro
|
|
||||||
}
|
|
||||||
} catch {
|
|
||||||
if (aborted) return;
|
|
||||||
entry.stopped = true;
|
|
||||||
chunkPool.remove(entry);
|
|
||||||
finish(job);
|
|
||||||
fail(job, 'Erreur réseau (reprise)');
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
const resumed = Boolean(meta);
|
|
||||||
while (!resumed) {
|
|
||||||
let res;
|
let res;
|
||||||
try {
|
try {
|
||||||
res = await fetch('/upload/init' + uploadQuery(job), {
|
res = await fetch('/upload/init' + uploadQuery(job), {
|
||||||
|
|
@ -817,18 +794,12 @@
|
||||||
}
|
}
|
||||||
if (job.state === 'running') setStatus(ui, 'Envoi…', 'muted');
|
if (job.state === 'running') setStatus(ui, 'Envoi…', 'muted');
|
||||||
sessionId = meta.id;
|
sessionId = meta.id;
|
||||||
job.sessionId = sessionId;
|
|
||||||
const chunkSize = meta.chunkSize;
|
const chunkSize = meta.chunkSize;
|
||||||
const chunkCount = meta.chunkCount;
|
const chunkCount = meta.chunkCount;
|
||||||
const loadedPer = new Array(chunkCount).fill(0);
|
const loadedPer = new Array(chunkCount).fill(0);
|
||||||
const attempts = new Array(chunkCount).fill(0);
|
const attempts = new Array(chunkCount).fill(0);
|
||||||
const emit = () => report(loadedPer.reduce((a, b) => a + b, 0));
|
const emit = () => report(loadedPer.reduce((a, b) => a + b, 0));
|
||||||
const skip = new Set(alreadyReceived);
|
let doneCount = 0;
|
||||||
for (const i of skip) {
|
|
||||||
loadedPer[i] = Math.min(chunkSize, file.size - i * chunkSize);
|
|
||||||
}
|
|
||||||
let doneCount = skip.size;
|
|
||||||
if (doneCount > 0) emit();
|
|
||||||
|
|
||||||
function retry(index, status, releasePermit) {
|
function retry(index, status, releasePermit) {
|
||||||
if (aborted || entry.stopped) {
|
if (aborted || entry.stopped) {
|
||||||
|
|
@ -839,15 +810,10 @@
|
||||||
if (attempts[index] > CHUNK_RETRIES) {
|
if (attempts[index] > CHUNK_RETRIES) {
|
||||||
entry.stopped = true;
|
entry.stopped = true;
|
||||||
entry.pending.length = 0;
|
entry.pending.length = 0;
|
||||||
if (status === 404) job.sessionId = null; // plus rien à reprendre côté serveur
|
|
||||||
failureMessage =
|
failureMessage =
|
||||||
status === 404
|
status === 404
|
||||||
? 'Session expirée côté serveur, réessayez'
|
? 'Session expirée côté serveur, réessayez'
|
||||||
: 'Erreur morceau ' +
|
: 'Erreur morceau ' + index + ' (' + (status || 'réseau') + ')';
|
||||||
index +
|
|
||||||
' (' +
|
|
||||||
(status || 'réseau') +
|
|
||||||
') — « Réessayer » reprend ici';
|
|
||||||
finalize('fail');
|
finalize('fail');
|
||||||
releasePermit();
|
releasePermit();
|
||||||
return;
|
return;
|
||||||
|
|
@ -861,7 +827,7 @@
|
||||||
releasePermit(); // libère le permis ; le morceau est re-planifié après le backoff
|
releasePermit(); // libère le permis ; le morceau est re-planifié après le backoff
|
||||||
}
|
}
|
||||||
|
|
||||||
entry.pending = Array.from({ length: chunkCount }, (_, i) => i).filter((i) => !skip.has(i));
|
entry.pending = Array.from({ length: chunkCount }, (_, i) => i);
|
||||||
entry.priming = false;
|
entry.priming = false;
|
||||||
entry.send = (index) =>
|
entry.send = (index) =>
|
||||||
new Promise((releasePermit) => {
|
new Promise((releasePermit) => {
|
||||||
|
|
@ -912,16 +878,15 @@
|
||||||
xhr.send(file.slice(start, end));
|
xhr.send(file.slice(start, end));
|
||||||
});
|
});
|
||||||
|
|
||||||
if (doneCount === chunkCount) finalize('ok'); // reprise : tout était déjà reçu
|
|
||||||
chunkPool.tick();
|
chunkPool.tick();
|
||||||
const outcome = await finished;
|
const outcome = await finished;
|
||||||
chunkPool.remove(entry);
|
chunkPool.remove(entry);
|
||||||
|
|
||||||
if (outcome === 'abort') return;
|
if (outcome === 'abort') return;
|
||||||
if (outcome === 'fail') {
|
if (outcome === 'fail') {
|
||||||
// La session serveur est conservée : « Réessayer » reprend aux morceaux manquants.
|
|
||||||
for (const x of inflight) x.abort();
|
for (const x of inflight) x.abort();
|
||||||
finish(job);
|
finish(job);
|
||||||
|
dropSession(sessionId);
|
||||||
if (failureMessage) fail(job, failureMessage);
|
if (failureMessage) fail(job, failureMessage);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
@ -930,10 +895,10 @@
|
||||||
const res = await fetch('/upload/' + sessionId + '/complete', { method: 'POST' });
|
const res = await fetch('/upload/' + sessionId + '/complete', { method: 'POST' });
|
||||||
const body = await res.json().catch(() => ({}));
|
const body = await res.json().catch(() => ({}));
|
||||||
finish(job);
|
finish(job);
|
||||||
if (res.ok || res.status === 404) job.sessionId = null;
|
|
||||||
applyResult(job, res.status, body);
|
applyResult(job, res.status, body);
|
||||||
} catch {
|
} catch {
|
||||||
finish(job);
|
finish(job);
|
||||||
|
dropSession(sessionId);
|
||||||
fail(job, 'Erreur réseau (finalisation)');
|
fail(job, 'Erreur réseau (finalisation)');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -42,11 +42,6 @@ interface InitResponse {
|
||||||
chunkCount: number;
|
chunkCount: number;
|
||||||
}
|
}
|
||||||
|
|
||||||
interface ResumeState extends InitResponse {
|
|
||||||
/** Index des morceaux déjà reçus par le serveur. */
|
|
||||||
received: number[];
|
|
||||||
}
|
|
||||||
|
|
||||||
interface ChunkAck {
|
interface ChunkAck {
|
||||||
received: number;
|
received: number;
|
||||||
total: number;
|
total: number;
|
||||||
|
|
@ -80,7 +75,6 @@ function clampChunkSize(value: unknown): number {
|
||||||
/**
|
/**
|
||||||
* Upload fragmenté et reprenable :
|
* Upload fragmenté et reprenable :
|
||||||
* - `POST /upload/init` crée une session, renvoie `{ id, chunkSize, chunkCount }`
|
* - `POST /upload/init` crée une session, renvoie `{ id, chunkSize, chunkCount }`
|
||||||
* - `GET /upload/:id` état de la session (morceaux reçus) pour reprendre
|
|
||||||
* - `PUT /upload/:id/:index` envoie un morceau (corps brut octet-stream)
|
* - `PUT /upload/:id/:index` envoie un morceau (corps brut octet-stream)
|
||||||
* - `POST /upload/:id/complete` assemble et finalise, renvoie 201 + métadonnées
|
* - `POST /upload/:id/complete` assemble et finalise, renvoie 201 + métadonnées
|
||||||
* - `DELETE /upload/:id` abandonne (supprime le fichier partiel)
|
* - `DELETE /upload/:id` abandonne (supprime le fichier partiel)
|
||||||
|
|
@ -146,18 +140,6 @@ export const chunkedUploadRoutes: FastifyPluginCallback = (app, _options, done)
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
||||||
// Permet au client de reprendre un upload en échec là où il s'est arrêté.
|
|
||||||
app.get('/upload/:id', { preHandler: requireSession }, (request): ResumeState => {
|
|
||||||
const { id } = request.params as { id: string };
|
|
||||||
const session = app.uploadSessions.use(id);
|
|
||||||
return {
|
|
||||||
id: session.id,
|
|
||||||
chunkSize: session.chunkSize,
|
|
||||||
chunkCount: session.chunkCount,
|
|
||||||
received: [...session.received].sort((a, b) => a - b),
|
|
||||||
};
|
|
||||||
});
|
|
||||||
|
|
||||||
app.put(
|
app.put(
|
||||||
'/upload/:id/:index',
|
'/upload/:id/:index',
|
||||||
{ preHandler: requireSession },
|
{ preHandler: requireSession },
|
||||||
|
|
|
||||||
|
|
@ -207,40 +207,6 @@ describe('upload fragmenté', () => {
|
||||||
expect(files[0]).toMatch(/\.mediadrop-part$/);
|
expect(files[0]).toMatch(/\.mediadrop-part$/);
|
||||||
});
|
});
|
||||||
|
|
||||||
it('GET /upload/:id liste les morceaux reçus et permet de reprendre', async () => {
|
|
||||||
const data = randomBytes(MIB + 10);
|
|
||||||
const { body } = await init({
|
|
||||||
size: data.length,
|
|
||||||
chunkSize: MIB,
|
|
||||||
target: 'tree',
|
|
||||||
path: 'a/clip.mp4',
|
|
||||||
});
|
|
||||||
expect(body.chunkCount).toBe(2);
|
|
||||||
expect(await putChunk(body.id!, 1, data.subarray(MIB))).toBe(200);
|
|
||||||
|
|
||||||
const state = await fetch(`${baseUrl}/upload/${body.id!}`, { headers: { cookie } });
|
|
||||||
expect(state.status).toBe(200);
|
|
||||||
expect(await state.json()).toEqual({
|
|
||||||
id: body.id,
|
|
||||||
chunkSize: MIB,
|
|
||||||
chunkCount: 2,
|
|
||||||
received: [1],
|
|
||||||
});
|
|
||||||
|
|
||||||
expect(await putChunk(body.id!, 0, data.subarray(0, MIB))).toBe(200);
|
|
||||||
const done = await complete(body.id!);
|
|
||||||
expect(done.status).toBe(201);
|
|
||||||
expect(await readFile(join(dir, 'a', 'clip.mp4'))).toEqual(data);
|
|
||||||
}, 20_000);
|
|
||||||
|
|
||||||
it('GET /upload/:id : 404 si inconnue, 401 sans session', async () => {
|
|
||||||
const unknown = await fetch(`${baseUrl}/upload/inexistante`, { headers: { cookie } });
|
|
||||||
expect(unknown.status).toBe(404);
|
|
||||||
const { body } = await init({ size: MIB, chunkSize: MIB });
|
|
||||||
const anonymous = await fetch(`${baseUrl}/upload/${body.id!}`);
|
|
||||||
expect(anonymous.status).toBe(401);
|
|
||||||
});
|
|
||||||
|
|
||||||
it('404 sur une session inconnue', async () => {
|
it('404 sur une session inconnue', async () => {
|
||||||
expect(await putChunk('inexistante', 0, Buffer.alloc(10))).toBe(404);
|
expect(await putChunk('inexistante', 0, Buffer.alloc(10))).toBe(404);
|
||||||
expect((await complete('inexistante')).status).toBe(404);
|
expect((await complete('inexistante')).status).toBe(404);
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue