fix(uploads): retry failed files and report full storage
ci / db / db (push) Successful in 1m51s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m40s
ci / module boundaries / boundaries (push) Successful in 7s
ci / api / api (push) Successful in 33s
ci / ts / ts (packages + web) (push) Failing after 5m16s
ci / db / db (push) Successful in 1m51s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m40s
ci / module boundaries / boundaries (push) Successful in 7s
ci / api / api (push) Successful in 33s
ci / ts / ts (packages + web) (push) Failing after 5m16s
This commit is contained in:
1 parent
b39e4dc28f
commit
5e2d1e902b
10 files changed
+205
-18
No files matched your search
@@ -0,0 +1,84 @@
|
||||
import { expect, test } from '@playwright/test';
|
||||
|
||||
test('locks the comment and retries one failed upload while another is running', async ({ page }, info) => {
|
||||
await page.addInitScript(() => localStorage.setItem('pipu.locale', 'ru'));
|
||||
await page.goto('/login');
|
||||
await page.getByLabel('Логин', { exact: true }).fill('mobile-admin');
|
||||
await page.getByRole('textbox', { name: 'Пароль', exact: true }).fill('mobile-admin-password');
|
||||
await page.getByRole('button', { name: 'Войти', exact: true }).click();
|
||||
await expect(page.getByRole('button', { name: 'Меню профиля' })).toBeVisible();
|
||||
await page.evaluate(() => {
|
||||
const original = window.fetch;
|
||||
window.fetch = async (input, init) => {
|
||||
const req = new Request(input, init);
|
||||
const path = new URL(req.url).pathname;
|
||||
const json = (data: unknown, status = 200) => new Response(JSON.stringify(data), { status, headers: { 'Content-Type': 'application/json' } });
|
||||
if (path === '/api/v1/uploads/quota') return json({ limit_bytes: 1e9, used_bytes: 0, per_user_limit_bytes: null, user_used_bytes: 0, max_file_bytes: 1e9 });
|
||||
if (path === '/api/v1/uploads/precheck') {
|
||||
const body = await req.json();
|
||||
return json({ files: body.files.map((f: { client_id: string }) => ({ client_id: f.client_id, result: 'accepted' })) });
|
||||
}
|
||||
if (path === '/api/v1/uploads' && req.method === 'POST') {
|
||||
const body = await req.json();
|
||||
document.documentElement.dataset.uploadComment = body.comment;
|
||||
document.documentElement.dataset.uploadReservations = String(Number(document.documentElement.dataset.uploadReservations ?? 0) + 1);
|
||||
return json({ batch_id: 'test-batch', files: body.files.map((f: { client_id: string; name: string }) => ({ client_id: f.client_id, upload_file_id: f.name, tus_url: `/api/v1/uploads/files/${f.name}` })) }, 201);
|
||||
}
|
||||
return original(input, init);
|
||||
};
|
||||
history.pushState({}, '', '/uploads/new');
|
||||
window.dispatchEvent(new PopStateEvent('popstate'));
|
||||
});
|
||||
|
||||
let retry = false;
|
||||
let failures = 0;
|
||||
let resumedOffset: string | undefined;
|
||||
let releaseSecond!: () => void;
|
||||
const secondPending = new Promise<void>((resolve) => { releaseSecond = resolve; });
|
||||
await page.route('**/api/v1/uploads/files/*', async (route) => {
|
||||
const req = route.request();
|
||||
const first = req.url().endsWith('/first.flac');
|
||||
if (req.method() === 'HEAD') {
|
||||
await route.fulfill({ status: 200, headers: { 'Upload-Length': '1024', 'Upload-Offset': first && failures > 0 ? '256' : '0', 'Tus-Resumable': '1.0.0' } });
|
||||
return;
|
||||
}
|
||||
if (first && !retry) {
|
||||
failures++;
|
||||
await route.fulfill({ status: 507, contentType: 'application/problem+json', body: JSON.stringify({ status: 507, code: 'uploads.storage_full', detail: 'upload storage is full' }) });
|
||||
return;
|
||||
}
|
||||
if (first) resumedOffset = req.headers()['upload-offset'];
|
||||
else await secondPending;
|
||||
await route.fulfill({ status: 204, headers: { 'Upload-Offset': '1024', 'Tus-Resumable': '1.0.0' } });
|
||||
});
|
||||
|
||||
const inputs = page.locator('input[type=file]');
|
||||
await inputs.first().setInputFiles(['first.flac', 'second.flac'].map((name) => ({ name, mimeType: 'audio/flac', buffer: Buffer.alloc(1024) })));
|
||||
const comment = page.getByRole('textbox', { name: 'Комментарий для администратора' });
|
||||
await comment.fill('Lossless album');
|
||||
await expect(comment).toBeEnabled();
|
||||
const send = page.getByRole('button', { name: /^Отправить/ });
|
||||
await expect(send).toBeEnabled();
|
||||
await send.click();
|
||||
await expect(comment).toBeDisabled();
|
||||
await expect(page.getByRole('button', { name: 'Очистить список' })).toBeDisabled();
|
||||
await expect(page.getByText(/На сервере закончилось место для загрузок/)).toBeVisible();
|
||||
const retryButton = page.getByRole('button', { name: 'Повторить загрузку first.flac', exact: true });
|
||||
await expect(retryButton).toBeEnabled();
|
||||
await page.setViewportSize({ width: 390, height: 844 });
|
||||
await expect.poll(() => page.evaluate(() => document.documentElement.scrollWidth - innerWidth)).toBeLessThanOrEqual(1);
|
||||
await page.screenshot({ path: info.outputPath('upload-retry.png'), fullPage: true });
|
||||
expect(failures).toBe(1);
|
||||
retry = true;
|
||||
await retryButton.click();
|
||||
await expect.poll(() => resumedOffset).toBe('256');
|
||||
await expect(retryButton).toHaveCount(0);
|
||||
await expect(comment).toBeDisabled();
|
||||
await expect(page.locator('html')).toHaveAttribute('data-upload-reservations', '1');
|
||||
await expect(page.locator('html')).toHaveAttribute('data-upload-comment', 'Lossless album');
|
||||
releaseSecond();
|
||||
await expect(page.getByText(/Готово: 2 файла/)).toBeVisible();
|
||||
await page.getByRole('button', { name: 'Загрузить ещё', exact: true }).click();
|
||||
await inputs.first().setInputFiles({ name: 'third.flac', mimeType: 'audio/flac', buffer: Buffer.alloc(1024) });
|
||||
await expect(comment).toBeEnabled();
|
||||
});
|
||||
@@ -68,6 +68,9 @@ export class Uploader {
|
||||
private hashQueue: string[] = [];
|
||||
private hashing = false;
|
||||
private uploads = new Map<string, tus.Upload>();
|
||||
private transferQueue: { id: string; url: string }[] | null = null;
|
||||
private transferWorkers = 0;
|
||||
private transferDone: (() => void) | null = null;
|
||||
private refreshing: Promise<unknown> | null = null;
|
||||
|
||||
constructor(private readonly session: AuthSession) {
|
||||
@@ -241,7 +244,7 @@ export class Uploader {
|
||||
}
|
||||
}
|
||||
this.persist();
|
||||
await Promise.all(Array.from({ length: Math.min(PARALLEL, queue.length) }, () => this.drain(queue)));
|
||||
await this.transfer(queue);
|
||||
} catch (err) {
|
||||
this.set({ error: err });
|
||||
} finally {
|
||||
@@ -249,24 +252,57 @@ export class Uploader {
|
||||
}
|
||||
}
|
||||
|
||||
get canRetryFailed(): boolean {
|
||||
return !this.state.busy || this.transferQueue !== null;
|
||||
}
|
||||
|
||||
/** Повторить оборвавшиеся загрузки: место уже зарезервировано, tus продолжит с принятого смещения. */
|
||||
async retryFailed() {
|
||||
const failed = this.state.items.filter((it) => it.stage === 'error' && it.tusUrl);
|
||||
if (this.state.busy || !failed.length) return;
|
||||
async retryFailed(id?: string) {
|
||||
const failed = this.state.items.filter((it) => it.stage === 'error' && it.tusUrl && it.file && (!id || it.id === id));
|
||||
if (!this.canRetryFailed || !failed.length) return;
|
||||
const queue = failed.map((it) => ({ id: it.id, url: it.tusUrl! }));
|
||||
for (const it of failed) this.patch(it.id, { stage: 'uploading', message: undefined, code: undefined });
|
||||
// При продолжающейся пачке ставим повтор в её очередь, сохраняя предел параллельности.
|
||||
if (this.transferQueue) {
|
||||
this.transferQueue.push(...queue);
|
||||
this.pumpTransfers();
|
||||
return;
|
||||
}
|
||||
this.set({ busy: true, error: null });
|
||||
try {
|
||||
const queue = failed.map((it) => {
|
||||
this.patch(it.id, { stage: 'uploading', message: undefined });
|
||||
return { id: it.id, url: it.tusUrl! };
|
||||
});
|
||||
await Promise.all(Array.from({ length: Math.min(PARALLEL, queue.length) }, () => this.drain(queue)));
|
||||
await this.transfer(queue);
|
||||
} finally {
|
||||
this.set({ busy: false });
|
||||
}
|
||||
}
|
||||
|
||||
private async drain(queue: { id: string; url: string }[]) {
|
||||
for (let next = queue.shift(); next; next = queue.shift()) await this.upload(next.id, next.url);
|
||||
private transfer(queue: { id: string; url: string }[]): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
this.transferQueue = queue;
|
||||
this.transferDone = resolve;
|
||||
this.pumpTransfers();
|
||||
});
|
||||
}
|
||||
|
||||
private pumpTransfers() {
|
||||
const queue = this.transferQueue;
|
||||
if (!queue) return;
|
||||
while (this.transferWorkers < PARALLEL && queue.length) {
|
||||
const next = queue.shift()!;
|
||||
this.transferWorkers++;
|
||||
void this.upload(next.id, next.url).catch((err: unknown) => {
|
||||
this.patch(next.id, { stage: 'error', message: err instanceof Error ? err.message : String(err) });
|
||||
}).finally(() => {
|
||||
this.transferWorkers--;
|
||||
this.pumpTransfers();
|
||||
});
|
||||
}
|
||||
if (this.transferWorkers === 0) {
|
||||
this.transferQueue = null;
|
||||
const done = this.transferDone;
|
||||
this.transferDone = null;
|
||||
done?.();
|
||||
}
|
||||
}
|
||||
|
||||
private upload(id: string, url: string): Promise<void> {
|
||||
@@ -291,7 +327,7 @@ export class Uploader {
|
||||
return true;
|
||||
}
|
||||
// Сеть, перегрузка, рассинхрон смещения — повторяем (tus-js-client спросит HEAD и продолжит).
|
||||
return status === undefined || status === 0 || status >= 500 || status === 409 || status === 423;
|
||||
return status === undefined || status === 0 || (status >= 500 && status !== 507) || status === 409 || status === 423;
|
||||
},
|
||||
onProgress: (sent, total) => this.patch(id, { progress: sent / Math.max(total, 1) }),
|
||||
onSuccess: () => {
|
||||
@@ -302,7 +338,8 @@ export class Uploader {
|
||||
},
|
||||
onError: (err) => {
|
||||
this.uploads.delete(id);
|
||||
this.patch(id, { stage: 'error', message: err.message });
|
||||
const status = (err as tus.DetailedError).originalResponse?.getStatus();
|
||||
this.patch(id, { stage: 'error', message: err.message, code: status === 507 ? 'uploads.storage_full' : undefined });
|
||||
resolve();
|
||||
},
|
||||
});
|
||||
@@ -398,7 +435,7 @@ export class Uploader {
|
||||
if (!queue.length) return;
|
||||
this.set({ busy: true });
|
||||
try {
|
||||
await Promise.all(Array.from({ length: Math.min(PARALLEL, queue.length) }, () => this.drain(queue)));
|
||||
await this.transfer(queue);
|
||||
} finally {
|
||||
this.set({ busy: false });
|
||||
this.persist();
|
||||
|
||||
@@ -129,7 +129,7 @@ export function SuggestPage() {
|
||||
{st.items.length > 0 && (
|
||||
<div className={u.files} aria-label={t('uploads.files')}>
|
||||
{st.items.map((it) => (
|
||||
<FileRow key={it.id} item={it} locked={st.busy || finished} onRemove={() => uploader.remove(it.id)} />
|
||||
<FileRow key={it.id} item={it} locked={st.busy || finished} retryDisabled={!uploader.canRetryFailed} onRetry={() => void uploader.retryFailed(it.id)} onRemove={() => uploader.remove(it.id)} />
|
||||
))}
|
||||
</div>
|
||||
)}
|
||||
@@ -141,6 +141,7 @@ export function SuggestPage() {
|
||||
<textarea
|
||||
className={u.textarea}
|
||||
value={comment}
|
||||
disabled={st.busy}
|
||||
maxLength={1000}
|
||||
placeholder={t('uploads.comment_hint')}
|
||||
onChange={(e) => setComment(e.target.value)}
|
||||
@@ -188,7 +189,13 @@ export function SuggestPage() {
|
||||
);
|
||||
}
|
||||
|
||||
function FileRow({ item, locked, onRemove }: { item: UploadItem; locked: boolean; onRemove: () => void }) {
|
||||
function FileRow({ item, locked, retryDisabled, onRetry, onRemove }: {
|
||||
item: UploadItem;
|
||||
locked: boolean;
|
||||
retryDisabled: boolean;
|
||||
onRetry: () => void;
|
||||
onRemove: () => void;
|
||||
}) {
|
||||
const { t, i18n } = useTranslation(['admin', 'errors']);
|
||||
const tr = t as unknown as (k: string) => string;
|
||||
const showProgress = item.stage === 'hashing' || item.stage === 'uploading';
|
||||
@@ -204,7 +211,7 @@ function FileRow({ item, locked, onRemove }: { item: UploadItem; locked: boolean
|
||||
note = { text: tr(`errors:${item.code ?? 'uploads.duplicate'}`), cls: u.bad };
|
||||
break;
|
||||
case 'error':
|
||||
note = { text: t('uploads.file_error', { message: item.message ?? '' }), cls: u.bad };
|
||||
note = { text: item.code === 'uploads.storage_full' ? tr('errors:uploads.storage_full') : t('uploads.file_error', { message: item.message ?? '' }), cls: u.bad };
|
||||
break;
|
||||
case 'ready':
|
||||
if (item.possibleDuplicate) note = { text: t('uploads.possible_dup'), cls: u.warn };
|
||||
@@ -229,7 +236,11 @@ function FileRow({ item, locked, onRemove }: { item: UploadItem; locked: boolean
|
||||
{note && <span className={note.cls}>{note.text}</span>}
|
||||
</div>
|
||||
<span className={u.size}>{formatBytes(item.size, i18n.language)}</span>
|
||||
{locked ? (
|
||||
{item.stage === 'error' && item.tusUrl && item.file ? (
|
||||
<Button variant="secondary" disabled={retryDisabled} aria-label={t('uploads.retry_file', { name: item.relPath })} onClick={onRetry}>
|
||||
{t('uploads.retry')}
|
||||
</Button>
|
||||
) : locked ? (
|
||||
<span />
|
||||
) : (
|
||||
<Button variant="ghost" aria-label={t('uploads.remove', { name: item.relPath })} onClick={onRemove}>
|
||||
|
||||
@@ -293,6 +293,11 @@
|
||||
resize: vertical;
|
||||
}
|
||||
|
||||
.textarea:disabled {
|
||||
opacity: 0.6;
|
||||
cursor: not-allowed;
|
||||
}
|
||||
|
||||
@media (max-width: 720px) {
|
||||
.queueRow {
|
||||
grid-template-columns: auto minmax(0, 1fr);
|
||||
|
||||
@@ -270,6 +270,7 @@
|
||||
"failed_some_many": "{{count}} files did not upload — their space is reserved, you can continue.",
|
||||
"failed_some_other": "{{count}} files did not upload — their space is reserved, you can continue.",
|
||||
"retry_failed": "Continue upload",
|
||||
"retry_file": "Retry upload of {{name}}",
|
||||
"uploading_now": "Uploading…",
|
||||
"go_mine": "My uploads →",
|
||||
"more": "Upload more",
|
||||
|
||||
@@ -80,6 +80,7 @@
|
||||
"update_unavailable": "Update is unavailable"
|
||||
},
|
||||
"uploads": {
|
||||
"storage_full": "The server has no space left for uploads. Ask an administrator to free up space, then retry the upload.",
|
||||
"queue_full": "The upload queue is full",
|
||||
"user_quota_full": "Your upload queue limit is exceeded",
|
||||
"too_large": "The file is larger than allowed",
|
||||
|
||||
@@ -270,6 +270,7 @@
|
||||
"failed_some_many": "Не загрузились {{count}} файлов — место за ними зарезервировано, можно продолжить.",
|
||||
"failed_some_other": "Не загрузились {{count}} файла — место за ними зарезервировано, можно продолжить.",
|
||||
"retry_failed": "Продолжить загрузку",
|
||||
"retry_file": "Повторить загрузку {{name}}",
|
||||
"uploading_now": "Загружаю…",
|
||||
"go_mine": "Мои предложения →",
|
||||
"more": "Загрузить ещё",
|
||||
|
||||
@@ -80,6 +80,7 @@
|
||||
"update_unavailable": "Обновление недоступно"
|
||||
},
|
||||
"uploads": {
|
||||
"storage_full": "На сервере закончилось место для загрузок. Попросите администратора освободить место, затем повторите загрузку.",
|
||||
"queue_full": "В очереди на добавление нет места",
|
||||
"user_quota_full": "Превышен ваш лимит в очереди на добавление",
|
||||
"too_large": "Файл больше допустимого размера",
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"sync"
|
||||
"syscall"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/google/uuid"
|
||||
@@ -162,6 +163,10 @@ func (t *TUS) patch(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
// Обрыв посреди куска: принятое сохранено, клиент спросит HEAD и продолжит.
|
||||
t.log.WarnContext(r.Context(), "tus: write interrupted", "upload_id", f.ID, "offset", now, "err", werr)
|
||||
if errors.Is(werr, syscall.ENOSPC) || errors.Is(werr, syscall.EDQUOT) {
|
||||
httpx.WriteProblem(w, http.StatusInsufficientStorage, "uploads.storage_full", "upload storage is full")
|
||||
return
|
||||
}
|
||||
httpx.WriteProblem(w, http.StatusInternalServerError, "common.internal", "write interrupted")
|
||||
return
|
||||
}
|
||||
|
||||
@@ -3,12 +3,14 @@ package uploads
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"testing"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
@@ -19,6 +21,45 @@ import (
|
||||
"pipu/api/internal/auth"
|
||||
)
|
||||
|
||||
type failingReader struct{ err error }
|
||||
|
||||
func (r failingReader) Read([]byte) (int, error) { return 0, r.err }
|
||||
|
||||
func TestTusInterruptedWriteCanResume(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
err error
|
||||
status int
|
||||
code string
|
||||
}{
|
||||
{"disk full", &os.PathError{Op: "write", Path: "/uploads/file", Err: syscall.ENOSPC}, http.StatusInsufficientStorage, "uploads.storage_full"},
|
||||
{"disk quota", syscall.EDQUOT, http.StatusInsufficientStorage, "uploads.storage_full"},
|
||||
{"connection interrupted", io.ErrUnexpectedEOF, http.StatusInternalServerError, "common.internal"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
e := newTusEnv(t, 10)
|
||||
req := httptest.NewRequest(http.MethodPatch, TusURL(e.files.f.ID), io.MultiReader(bytes.NewBufferString("hello"), failingReader{tc.err}))
|
||||
req.Header.Set("Tus-Resumable", tusVersion)
|
||||
req.Header.Set("Content-Type", "application/offset+octet-stream")
|
||||
req.Header.Set("Upload-Offset", "0")
|
||||
req = req.WithContext(auth.WithPrincipal(req.Context(), auth.Principal{UserID: e.author, Role: auth.RoleModerator}))
|
||||
rec := httptest.NewRecorder()
|
||||
e.h.ServeHTTP(rec, req)
|
||||
require.Equal(t, tc.status, rec.Code, rec.Body.String())
|
||||
assert.Contains(t, rec.Body.String(), tc.code)
|
||||
assert.NotContains(t, rec.Body.String(), "/uploads/file")
|
||||
assert.EqualValues(t, 5, e.files.f.Received)
|
||||
assert.Equal(t, 0, e.complete)
|
||||
assert.Equal(t, "5", e.do(e.author, http.MethodHead, 0, "").Header().Get("Upload-Offset"))
|
||||
require.Equal(t, http.StatusNoContent, e.do(e.author, http.MethodPatch, 5, "world").Code)
|
||||
assert.Equal(t, 1, e.complete)
|
||||
got, err := os.ReadFile(filepath.Join(e.dir, e.files.f.StagingPath))
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, "helloworld", string(got))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type fakeFiles struct{ f File }
|
||||
|
||||
func (m *fakeFiles) File(_ context.Context, id uuid.UUID) (File, error) {
|
||||
|
||||
Reference in new issue
Block a user