feat(web): space roles, upload queue with pipelined frames, background resume and faster sending
This commit is contained in:
justme committed
1 parent
96c4335cb5
commit
a574583943
39 files changed
+1515
-194
No files matched your search
@@ -278,8 +278,8 @@ export async function inviteChannelUser(channelId: string, input: string): Promi
|
||||
await apiFetch(`/channels/${channelId}/invitations`, { method: 'POST', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' }, body: JSON.stringify({ username, payload, domain_envelopes: domainEnvelopes }) });
|
||||
}
|
||||
export async function listChannelInvitations(channelId: string): Promise<Invitation[]> {
|
||||
const { session } = await getSession(channelId); return apiFetch(`/channels/${channelId}/invitations`, { headers: session.authHeaders('owner') }).then((r) => r.json() as Promise<Invitation[]>);
|
||||
const { session } = await getSession(channelId); return apiFetch(`/channels/${channelId}/invitations`, { headers: session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read') }).then((r) => r.json() as Promise<Invitation[]>);
|
||||
}
|
||||
export async function revokeChannelInvitation(channelId: string, input: string): Promise<void> {
|
||||
const { session } = await getSession(channelId); await apiFetch(`/channels/${channelId}/invitations/${usernameOf(input)}`, { method: 'DELETE', headers: session.authHeaders('owner') });
|
||||
const { session } = await getSession(channelId); await apiFetch(`/channels/${channelId}/invitations/${usernameOf(input)}`, { method: 'DELETE', headers: session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read') });
|
||||
}
|
||||
@@ -214,6 +214,8 @@ export async function signObjectDeletion(resource: SavedResource, session: Chann
|
||||
const { state } = await readDataKeys(resource, session);
|
||||
const object = state.objects.find(row => row.itemId === itemId);
|
||||
const signed = object?.event;
|
||||
// 末帧从未上传的对象没有签名清单:没有可签的内容,服务端允许作者和空间主直接删除
|
||||
if (object && !signed?.signature) return {};
|
||||
if (!signed?.manifestHash || !signed.frameCount || (object?.domain !== 'chat' && object?.domain !== 'files'))
|
||||
throw new Error('对象缺少已验签的密文清单，不能生成删除事件');
|
||||
const manifestHash = fromBase64Url(signed.manifestHash, 32);
|
||||
@@ -231,6 +233,8 @@ export async function verifyObjectDeletion(resource: SavedResource, session: Cha
|
||||
}
|
||||
const deleted = object?.deleteEvent;
|
||||
if (resource.mode === 'raw' && !deleted?.signature) return true;
|
||||
// 没传完的上传从未发布过有效内容,删掉它不需要事件(见 signObjectDeletion)
|
||||
if (object && !object.event?.signature && !deleted?.signature) return true;
|
||||
if (!object || !deleted?.signature || !deleted.eventId || !deleted.kind || !deleted.signerDeviceId
|
||||
|| !deleted.signerRootPublicKey || !deleted.signerDevicePublicKey || !deleted.deviceCertificateSignature
|
||||
|| !object.authorAccount || !object.event?.manifestHash || !object.event.frameCount || (object.domain !== 'chat' && object.domain !== 'files')) return false;
|
||||
|
||||
@@ -1,21 +1,12 @@
|
||||
import { removeChannelItem } from './channelItems';
|
||||
import { queueFileCatalog } from './channelThreads';
|
||||
import { enqueueFileUploads } from './uploadQueue';
|
||||
import { senderSnapshot } from './senderSnapshot';
|
||||
import { prepareObject } from './channelDataKeys';
|
||||
import { errorMessage } from '../errors';
|
||||
|
||||
import { saveResource, queueLocation } from '../database';
|
||||
|
||||
import { previewKindOf } from '@/lib/fileKind';
|
||||
import type { FileItem, FileUploadOptions, Item } from '../types';
|
||||
|
||||
import { queueLocation } from '../database';
|
||||
import type { FileUploadOptions, Item } from '../types';
|
||||
import { blobs, getSession } from './sessions';
|
||||
import { deviceIdentity } from './presentation';
|
||||
import { accountDisplay } from './accountSession';
|
||||
|
||||
|
||||
|
||||
import { flushOutbox, MAX_FRAME_DATA, MOBILE_UPLOAD_FRAME_DATA_BYTES, persistPlan } from './transfer';
|
||||
import { flushOutbox, persistPlan } from './transfer';
|
||||
import { publishWorkspace } from './presentation';
|
||||
|
||||
export async function sendText(resourceId: string, text: string, threadId?: string,
|
||||
@@ -33,93 +24,17 @@ export async function sendText(resourceId: string, text: string, threadId?: stri
|
||||
await persistPlan(resource, plan, undefined, { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch,
|
||||
eventKind: event?.kind ?? 'chat.create', ...(event?.targetItemId ? { targetItemId: event.targetItemId } : {}) });
|
||||
await queueLocation(resourceId, item.id, threadId);
|
||||
const uploaded = await flushOutbox(resource, session);
|
||||
// 只发这条消息自己的帧,不被队列里已有的大文件堵住
|
||||
const uploaded = await flushOutbox(resource, session, undefined, { only: new Set([plan.itemId]) });
|
||||
publishWorkspace(resourceId, { t: 'item', item: { ...item, state: uploaded ? { kind: 'complete' } : { kind: 'incomplete' } } });
|
||||
if (!uploaded) publishWorkspace(resourceId, { t: 'connection', state: 'offline' });
|
||||
return { itemId: plan.itemId, uploaded };
|
||||
}
|
||||
|
||||
/** 文件进入该空间的上传队列:一次传一个,其余排队。文件内部的帧并发上传,见 uploadQueue */
|
||||
export async function sendFiles(resourceId: string, files: File[], suppliedOptions?: FileUploadOptions, threadId?: string, folderId?: string, localIds?: string[]): Promise<void> {
|
||||
if (!threadId) throw new Error('请先选择频道');
|
||||
const { session, resource } = await getSession(resourceId);
|
||||
const options = suppliedOptions ?? { directPreview: true, maxDownloads: resource.maxDownloads ?? 0 };
|
||||
if (resource.kind !== 'channel' || (!session.can('owner') && !session.can('write'))) throw new Error('此链接没有上传权限');
|
||||
const device = await deviceIdentity(session.endpointSecret);
|
||||
const maxFrameData = device.kind === 'phone' ? MOBILE_UPLOAD_FRAME_DATA_BYTES : MAX_FRAME_DATA;
|
||||
const author = await senderSnapshot();
|
||||
for (const [index, file] of files.entries()) {
|
||||
const localId = localIds?.[index];
|
||||
const ts = Date.now();
|
||||
let item: FileItem | undefined;
|
||||
try {
|
||||
const prepared = await prepareObject(resource, session, threadId, 'files');
|
||||
const plan = session.encryptFile(file, {}, {
|
||||
itemId: prepared.itemIdBytes,
|
||||
...(options.directPreview ? { gated: 'auto' as const } : { gated: true, maxDownloads: options.maxDownloads }),
|
||||
ts, maxFrameData, threadId, folderId,
|
||||
});
|
||||
item = {
|
||||
id: plan.itemId, kind: 'file', ts, state: { kind: 'receiving', progress: 0, phase: 'encrypting' },
|
||||
...(author ? { author } : {}),
|
||||
...(threadId ? { threadId } : {}), ...(folderId ? { folderId } : {}), ...(localId ? { localId } : {}),
|
||||
name: file.name, mime: file.type || 'application/octet-stream', size: file.size,
|
||||
remaining: plan.gated
|
||||
? (options.directPreview ? resource.maxDownloads ?? 0 : options.maxDownloads) || null
|
||||
: null,
|
||||
preview: previewKindOf(file.type, file.name), gated: plan.gated,
|
||||
};
|
||||
resource.localItems = [...(resource.localItems ?? []).filter((entry) => entry.id !== item?.id), item];
|
||||
await saveResource(resource);
|
||||
blobs.set(`${resourceId}:${item.id}`, file);
|
||||
publishWorkspace(resourceId, { t: 'item', item });
|
||||
const updateProgress = (progress: number, phase: 'encrypting' | 'uploading', speedBytesPerSecond?: number) => {
|
||||
if (!item) return;
|
||||
item = {
|
||||
...item,
|
||||
state: {
|
||||
kind: 'receiving',
|
||||
progress: Math.max(0, Math.min(0.99, progress)),
|
||||
phase,
|
||||
...(speedBytesPerSecond === undefined ? {} : { speedBytesPerSecond }),
|
||||
},
|
||||
};
|
||||
publishWorkspace(resourceId, { t: 'item', item });
|
||||
};
|
||||
const planned = await persistPlan(resource, plan, (_frames) => {
|
||||
const processed = Math.min(file.size, _frames * MAX_FRAME_DATA);
|
||||
updateProgress(file.size === 0 ? 0.5 : Math.min(0.5, processed / file.size * 0.5), 'encrypting');
|
||||
}, { channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish' });
|
||||
const uploadedFrames = new Set<number>();
|
||||
await queueFileCatalog(resource, session, item);
|
||||
await queueLocation(resourceId, item.id, threadId, folderId);
|
||||
let uploadedBytes = 0;
|
||||
const uploadStartedAt = performance.now();
|
||||
const uploaded = await flushOutbox(resource, session, (frame, loadedBytes, totalBytes, complete) => {
|
||||
if (frame.itemId !== item?.id) return;
|
||||
if (complete && !uploadedFrames.has(frame.index)) {
|
||||
uploadedFrames.add(frame.index);
|
||||
uploadedBytes += totalBytes;
|
||||
}
|
||||
const currentBytes = uploadedBytes + (complete ? 0 : loadedBytes);
|
||||
const elapsed = Math.max(1, performance.now() - uploadStartedAt);
|
||||
updateProgress(0.5 + 0.5 * currentBytes / planned.bytes, 'uploading', currentBytes * 1000 / elapsed);
|
||||
});
|
||||
item = { ...item, state: uploaded ? { kind: 'complete' } : { kind: 'incomplete' } };
|
||||
resource.localItems = resource.localItems.map((entry) => entry.id === item?.id ? item! : entry);
|
||||
await saveResource(resource);
|
||||
publishWorkspace(resourceId, { t: 'item', item });
|
||||
if (!uploaded) publishWorkspace(resourceId, { t: 'error', message: '上传暂停了。重新打开空间并连接网络后,可以继续发送已加密内容。' });
|
||||
} catch (error) {
|
||||
if (item) {
|
||||
item = { ...item, state: { kind: 'incomplete' } };
|
||||
resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === item?.id ? item! : entry);
|
||||
await saveResource(resource);
|
||||
publishWorkspace(resourceId, { t: 'item', item });
|
||||
}
|
||||
publishWorkspace(resourceId, { t: 'error', message: `文件上传失败：${errorMessage(error)}` });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return enqueueFileUploads(resourceId, files, suppliedOptions, threadId, folderId, localIds);
|
||||
}
|
||||
|
||||
export async function updateSnapshot(resourceId: string, text: string): Promise<void> {
|
||||
|
||||
@@ -2,13 +2,13 @@ import type { ChannelSession } from '@webdrop/core';
|
||||
import type { SavedResource } from '../database';
|
||||
import type { FileItem } from '../types';
|
||||
import { persistPlan } from './transfer';
|
||||
import { changeWorkspace, createWorkspaceChannel, createWorkspaceGroup } from './workspaceStructure';
|
||||
import { changeWorkspace, createWorkspaceChannel, createWorkspaceGroup, createWorkspaceRole } from './workspaceStructure';
|
||||
import { getSession } from './sessions';
|
||||
import { initializeChannelKeys, prepareObject } from './channelDataKeys';
|
||||
import { toPublishedFile } from './fileCatalog';
|
||||
import { apiFetch } from './http';
|
||||
import { WORKSPACE_APP, type ResourceControl } from '../protocol/workspace';
|
||||
import type { WorkspaceChannelRequest, WorkspaceGroupRequest, WorkspaceNameRequest, WorkspaceFolderRequest, WorkspaceMoveRequest } from '../generated/api';
|
||||
import type { WorkspaceChannelRequest, WorkspaceGroupRequest, WorkspaceNameRequest, WorkspaceFolderRequest, WorkspaceMoveRequest, WorkspaceRoleRequest, WorkspaceRoleMembersRequest } from '../generated/api';
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
@@ -27,6 +27,14 @@ export async function renameGroup(resourceId: string, id: string, name: string):
|
||||
export async function deleteGroup(resourceId: string, id: string): Promise<void> {
|
||||
await changeWorkspace(resourceId, 'groups/' + id, 'DELETE');
|
||||
}
|
||||
export async function createRole(resourceId: string, role: WorkspaceRoleRequest): Promise<string> { return createWorkspaceRole(resourceId, role); }
|
||||
export async function updateRole(resourceId: string, id: string, role: WorkspaceRoleRequest): Promise<void> {
|
||||
await changeWorkspace(resourceId, 'roles/' + id, 'PATCH', role);
|
||||
}
|
||||
export async function deleteRole(resourceId: string, id: string): Promise<void> { await changeWorkspace(resourceId, 'roles/' + id, 'DELETE'); }
|
||||
export async function setRoleMembers(resourceId: string, id: string, members: string[]): Promise<void> {
|
||||
await changeWorkspace(resourceId, 'roles/' + id + '/members', 'PUT', { members } satisfies WorkspaceRoleMembersRequest);
|
||||
}
|
||||
export async function groupThread(resourceId: string, id: string, groupId?: string): Promise<void> {
|
||||
await changeWorkspace(resourceId, 'channels/' + id + '/group', 'PUT', { groupId: groupId ?? null } satisfies WorkspaceGroupRequest);
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import { publishWorkspace } from './presentation';
|
||||
import { channelMetadata, processChannelBytes, syncChannel } from './channelEvents';
|
||||
import { loadWorkspace } from './workspaceStructure';
|
||||
import { flushOutbox } from './transfer';
|
||||
import { resumeOutbox } from './resumeUploads';
|
||||
import { socketStream, sseStream, delay, type StreamConsumer } from './transports';
|
||||
import { errorMessage } from '../errors';
|
||||
import { saveResource } from '../database';
|
||||
@@ -43,19 +44,24 @@ export async function connectChannel(id: string): Promise<void> {
|
||||
let initialized = false;
|
||||
let transport = 0;
|
||||
let backoff = 1000;
|
||||
flushTimer = setInterval(() => {
|
||||
if (!signal.aborted) void flushOutbox(resource, session).catch((error: unknown) => {
|
||||
publishWorkspace(id, { t: 'error', message: errorMessage(error) });
|
||||
});
|
||||
}, 2500);
|
||||
// 离线队列的续传在后台跑:几个 GB 的遗留不能挡住频道加载,也不能叠出一堆并行的续传
|
||||
let resuming = false;
|
||||
const resume = () => {
|
||||
if (signal.aborted || resuming) return;
|
||||
resuming = true;
|
||||
void resumeOutbox(resource, session, signal).then(() => flushOutbox(resource, session))
|
||||
.catch((error: unknown) => { if (!signal.aborted) publishWorkspace(id, { t: 'error', message: errorMessage(error) }); })
|
||||
.finally(() => { resuming = false; });
|
||||
};
|
||||
flushTimer = setInterval(resume, 2500);
|
||||
while (!signal.aborted) {
|
||||
try {
|
||||
if (!initialized) {
|
||||
[stream.cursor] = await Promise.all([syncChannel(resource, session, signal), channelMetadata(resource, session), joining]);
|
||||
if (signal.aborted) break;
|
||||
await flushOutbox(resource, session);
|
||||
initialized = true;
|
||||
publishWorkspace(id, { t: 'ready' });
|
||||
resume();
|
||||
}
|
||||
stream.online();
|
||||
if (transport === 0) await socketStream(id, session, signal, stream);
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import type { QueuedFrame } from '../database';
|
||||
import { FrameStash } from './frameStash';
|
||||
|
||||
const frame = (index: number): QueuedFrame => ({ key: `k${index}`, resourceId: 'r', itemId: 'i', index, bytes: new Uint8Array(10), extraHeaders: {}, createdAt: index });
|
||||
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
function fakeDisk(writeMs = 5) {
|
||||
const stored = new Set<string>(); const writes: string[] = []; const removes: string[] = [];
|
||||
return {
|
||||
stored, writes, removes,
|
||||
put: async (f: QueuedFrame) => { await sleep(writeMs); stored.add(f.key); writes.push(f.key); },
|
||||
remove: async (key: string) => { stored.delete(key); removes.push(key); },
|
||||
};
|
||||
}
|
||||
|
||||
describe('FrameStash', () => {
|
||||
it('宽限期内就被确认的帧:根本不写盘', async () => {
|
||||
const disk = fakeDisk(5);
|
||||
const stash = new FrameStash(disk.put, disk.remove, 8, 100, 4);
|
||||
const frames = [0, 1, 2, 3].map(frame);
|
||||
for (const f of frames) await stash.add(f);
|
||||
await sleep(20);
|
||||
for (const f of frames) stash.accepted(f);
|
||||
await stash.drain();
|
||||
expect(disk.writes).toEqual([]);
|
||||
expect(disk.stored.size).toBe(0);
|
||||
});
|
||||
|
||||
it('超过宽限期还没确认的帧会写入,之后确认时再删除', async () => {
|
||||
const disk = fakeDisk(1);
|
||||
const stash = new FrameStash(disk.put, disk.remove, 8, 30, 4);
|
||||
const first = frame(0); await stash.add(first);
|
||||
await sleep(80);
|
||||
expect(disk.stored.has('k0')).toBe(true);
|
||||
stash.accepted(first);
|
||||
await sleep(10);
|
||||
expect(disk.stored.has('k0')).toBe(false);
|
||||
});
|
||||
|
||||
it('断网时(一直不确认)drain 之后全部帧都已落盘,续传才有东西可发', async () => {
|
||||
const disk = fakeDisk(1);
|
||||
const stash = new FrameStash(disk.put, disk.remove, 3, 5000, 4);
|
||||
for (let index = 0; index < 10; index += 1) await stash.add(frame(index));
|
||||
await stash.drain();
|
||||
expect(disk.stored.size).toBe(10);
|
||||
});
|
||||
|
||||
it('积压超过上限时 add 会等待写盘,不让内存里越堆越多', async () => {
|
||||
const disk = fakeDisk(30);
|
||||
const stash = new FrameStash(disk.put, disk.remove, 2, 5000, 1);
|
||||
let added = 0;
|
||||
const producer = (async () => { for (let index = 0; index < 6; index += 1) { await stash.add(frame(index)); added += 1; } })();
|
||||
await sleep(40);
|
||||
expect(added).toBeLessThan(6);
|
||||
await producer; await stash.drain();
|
||||
expect(disk.stored.size).toBe(6);
|
||||
});
|
||||
|
||||
it('取消后不会再写入新的帧', async () => {
|
||||
const disk = fakeDisk(5);
|
||||
const stash = new FrameStash(disk.put, disk.remove, 8, 20, 4);
|
||||
await stash.add(frame(0)); await stash.add(frame(1));
|
||||
await stash.discard();
|
||||
const before = disk.writes.length;
|
||||
await sleep(60);
|
||||
expect(disk.writes.length).toBe(before);
|
||||
});
|
||||
|
||||
it('写盘失败会让加密端停下来并抛出', async () => {
|
||||
const stash = new FrameStash(async () => { throw new Error('quota'); }, async () => {}, 1, 0, 0);
|
||||
await expect((async () => { for (let index = 0; index < 5; index += 1) await stash.add(frame(index)); await stash.drain(); })()).rejects.toThrow('quota');
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,85 @@
|
||||
import type { QueuedFrame } from '../database';
|
||||
|
||||
interface Pending { frame: QueuedFrame; at: number }
|
||||
|
||||
/**
|
||||
* 离线队列的「后写」:加密好的帧先在内存里直接上传,只有没能及时上传的才写进 IndexedDB 留作续传保险。
|
||||
* 每帧写盘要几十毫秒,同步写会把上传速度压到约 25 MB/s;网络够快时帧在宽限期内就已被确认,根本不用写。
|
||||
* 网络慢或断了,帧超过宽限期(或积压太多)就在后台写入,离线续传不受影响(protocol §5:重试发送的仍是同一份字节)。
|
||||
*/
|
||||
export class FrameStash {
|
||||
private pending: Pending[] = [];
|
||||
private readonly written = new Set<string>();
|
||||
private writing: Promise<void> | undefined;
|
||||
private error: unknown;
|
||||
private forced = false;
|
||||
private waiters: (() => void)[] = [];
|
||||
|
||||
constructor(
|
||||
private readonly put: (frame: QueuedFrame) => Promise<void>,
|
||||
private readonly remove: (key: string) => Promise<void>,
|
||||
/** 还没写入的帧超过这个数就让加密端等一等,避免断网后内存里越堆越多 */
|
||||
private readonly maxPending = 16,
|
||||
/** 帧在内存里等服务端确认的宽限时间,超过才写盘 */
|
||||
private readonly graceMs = 400,
|
||||
/** 积压超过这个数就不再等宽限期,马上写最早的。要大于上传流水线本身在途+缓冲的帧数(约 10),否则正常上传也会被逼着写盘 */
|
||||
private readonly eagerAbove = 12,
|
||||
) {}
|
||||
|
||||
/** 新加密出一帧。积压太多时等待写盘追上来 */
|
||||
async add(frame: QueuedFrame): Promise<void> {
|
||||
if (this.error) throw this.error;
|
||||
this.pending.push({ frame, at: performance.now() });
|
||||
this.kick();
|
||||
this.notify();
|
||||
while (this.pending.length > this.maxPending && !this.error) await this.wait();
|
||||
if (this.error) throw this.error;
|
||||
}
|
||||
|
||||
/** 服务端确认了这一帧:还没写盘的不用写了,已经写了的从队列里删掉 */
|
||||
accepted(frame: QueuedFrame): void {
|
||||
const at = this.pending.findIndex((entry) => entry.frame.key === frame.key);
|
||||
if (at >= 0) { this.pending.splice(at, 1); this.notify(); return; }
|
||||
if (this.written.delete(frame.key)) void this.remove(frame.key).catch(() => undefined);
|
||||
}
|
||||
|
||||
/** 等所有该写的帧都落盘(不再等宽限期)。全部上传完成时这里几乎是空的 */
|
||||
async drain(): Promise<void> {
|
||||
this.forced = true; this.kick(); this.notify();
|
||||
while ((this.pending.length || this.writing) && !this.error) await this.wait();
|
||||
if (this.error) throw this.error;
|
||||
}
|
||||
|
||||
/** 取消上传:丢掉还没写的帧,等正在写的那一帧结束。已写入的由调用方用 dequeueItem 清掉 */
|
||||
async discard(): Promise<void> {
|
||||
this.pending = [];
|
||||
this.notify();
|
||||
await this.writing;
|
||||
}
|
||||
|
||||
private kick(): void {
|
||||
if (this.writing || !this.pending.length) return;
|
||||
this.writing = (async () => {
|
||||
try {
|
||||
while (this.pending.length) {
|
||||
const head = this.pending[0]!;
|
||||
const age = performance.now() - head.at;
|
||||
if (!this.forced && age < this.graceMs && this.pending.length <= this.eagerAbove && this.pending.length < this.maxPending) {
|
||||
// 还在宽限期里:等它被确认(会被移出 pending)、或有新帧到来、或宽限期到
|
||||
await Promise.race([this.wait(), new Promise((resolve) => setTimeout(resolve, this.graceMs - age))]);
|
||||
continue;
|
||||
}
|
||||
await this.put(head.frame);
|
||||
// 写的过程中如果已经被确认,accepted() 已把它移出 pending:这次写入白做了,补一个删除
|
||||
if (this.pending[0] === head) { this.pending.shift(); this.written.add(head.frame.key); }
|
||||
else await this.remove(head.frame.key).catch(() => undefined);
|
||||
this.notify();
|
||||
}
|
||||
} catch (error) { this.error = error; }
|
||||
finally { this.writing = undefined; this.notify(); }
|
||||
})();
|
||||
}
|
||||
|
||||
private wait(): Promise<void> { return new Promise((resolve) => { this.waiters.push(resolve); }); }
|
||||
private notify(): void { const waiting = this.waiters; this.waiters = []; for (const resume of waiting) resume(); }
|
||||
}
|
||||
@@ -2,7 +2,7 @@ import { supportsApiContract } from '../contract';
|
||||
import { accountToken } from './accountSession';
|
||||
import type { ServerInfoBody } from '../generated/api';
|
||||
|
||||
export type ApiError = Error & { status?: number; code?: string };
|
||||
export type ApiError = Error & { status?: number; code?: string; /** 服务端要求的等待秒数(429 的 Retry-After) */ retryAfter?: number };
|
||||
let apiBase = '/api/v1';
|
||||
let contractCheck: Promise<ServerInfoBody> | undefined;
|
||||
|
||||
@@ -69,24 +69,37 @@ export async function apiFetch(path: string, init: RequestInit = {}): Promise<Re
|
||||
export async function apiUpload(
|
||||
path: string,
|
||||
headers: Record<string, string>,
|
||||
body: ArrayBuffer,
|
||||
body: ArrayBuffer | Uint8Array | Blob,
|
||||
onProgress: (loadedBytes: number, totalBytes: number) => void,
|
||||
signal?: AbortSignal,
|
||||
/** 这么久没有任何上传进展就放弃这次请求(弱网下连接静默卡死),由调用方重试 */
|
||||
stallTimeoutMs?: number,
|
||||
): Promise<void> {
|
||||
try {
|
||||
await getServerInfo();
|
||||
const response = await new Promise<Response>((resolve, reject) => {
|
||||
if (signal?.aborted) { reject(new DOMException('上传已取消', 'AbortError')); return; }
|
||||
const xhr = new XMLHttpRequest();
|
||||
signal?.addEventListener('abort', () => xhr.abort(), { once: true });
|
||||
let stalled = false; let stallTimer: ReturnType<typeof setTimeout> | undefined;
|
||||
const watch = () => {
|
||||
if (!stallTimeoutMs) return;
|
||||
clearTimeout(stallTimer);
|
||||
stallTimer = setTimeout(() => { stalled = true; xhr.abort(); }, stallTimeoutMs);
|
||||
};
|
||||
xhr.open('POST', apiPath(path));
|
||||
xhr.withCredentials = false;
|
||||
xhr.responseType = 'text';
|
||||
xhr.setRequestHeader('Cache-Control', 'no-store');
|
||||
// 不要设 Cache-Control 请求头:它不在服务端 CORS 允许的头里,跨域部署(API 在另一个域名)时预检会失败;POST 本来也不会被缓存
|
||||
const token = accountToken();
|
||||
if (token) xhr.setRequestHeader('X-WD-Account', token);
|
||||
for (const [name, value] of Object.entries(headers)) xhr.setRequestHeader(name, value);
|
||||
xhr.upload.onprogress = (event) => {
|
||||
watch();
|
||||
if (event.lengthComputable) onProgress(event.loaded, event.total);
|
||||
};
|
||||
xhr.onload = () => {
|
||||
clearTimeout(stallTimer);
|
||||
if (xhr.status === 0) {
|
||||
reject(new Error('network_error'));
|
||||
return;
|
||||
@@ -94,20 +107,27 @@ export async function apiUpload(
|
||||
const responseHeaders = new Headers();
|
||||
const contentType = xhr.getResponseHeader('Content-Type');
|
||||
if (contentType) responseHeaders.set('Content-Type', contentType);
|
||||
const retryAfter = xhr.getResponseHeader('Retry-After');
|
||||
if (retryAfter) responseHeaders.set('Retry-After', retryAfter);
|
||||
resolve(new Response(xhr.responseText, { status: xhr.status, statusText: xhr.statusText, headers: responseHeaders }));
|
||||
};
|
||||
xhr.onerror = () => reject(new Error('network_error'));
|
||||
xhr.onabort = () => reject(new DOMException('上传已取消', 'AbortError'));
|
||||
xhr.send(body);
|
||||
xhr.onerror = () => { clearTimeout(stallTimer); reject(new Error('network_error')); };
|
||||
xhr.onabort = () => { clearTimeout(stallTimer); reject(stalled ? new Error('network_error') : new DOMException('上传已取消', 'AbortError')); };
|
||||
watch();
|
||||
xhr.send(body as XMLHttpRequestBodyInit);
|
||||
});
|
||||
if (!response.ok) {
|
||||
const result = await response.json().catch(() => null) as { error?: unknown } | null;
|
||||
const code = typeof result?.error === 'string' && /^[a-z_]{1,40}$/.test(result.error) ? result.error : 'request_failed';
|
||||
throw fail(response.status, code);
|
||||
const error = fail(response.status, code);
|
||||
const wait = Number(response.headers.get('Retry-After'));
|
||||
if (Number.isFinite(wait) && wait > 0) error.retryAfter = wait;
|
||||
throw error;
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof Error && 'status' in error) throw error;
|
||||
if (error instanceof Error && error.message === 'api_contract_mismatch') throw error;
|
||||
if (signal?.aborted) throw error;
|
||||
throw new Error('网络连接失败');
|
||||
}
|
||||
}
|
||||
|
||||
@@ -22,7 +22,7 @@ describe('resource and rendering boundaries', () => {
|
||||
it('never synchronizes draft content or server structure snapshots inside the key vault', () => {
|
||||
const resource: SavedResource = { id: 'a', kind: 'share', mode: 'enc', role: 'owner', link: 'key', createdAt: 1, expiresAt: 2,
|
||||
endpointSecret: new Uint8Array(32), maxMembers: 0, draftItems: [], manifestEntries: [],
|
||||
workspaceMetadata: { structure: { revision: 1, channels: [], groups: [], folders: [], locations: [] }, people: [] } };
|
||||
workspaceMetadata: { structure: { revision: 1, channels: [], groups: [], roles: [], rolesSeeded: true, folders: [], locations: [] }, people: [] } };
|
||||
const exported = exportResource(resource);
|
||||
for (const field of ['draftItems', 'manifestEntries', 'workspaceMetadata', 'endpointSecret']) expect(exported).not.toHaveProperty(field);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import { dequeueItem, queuedFrame, queuedFrameKeys, saveResource, type QueuedFrame, type SavedResource } from '../database';
|
||||
import type { ChannelSession } from '@webdrop/core';
|
||||
import { publishWorkspace } from './presentation';
|
||||
import { pipelinedItems, uploadFrame } from './transfer';
|
||||
import { FrameUploader, TransferMeter } from './uploadPipeline';
|
||||
|
||||
const FRAME_CONCURRENCY = 4;
|
||||
const keyOf = (key: string) => { const [, itemId = '', index = '0'] = key.split(':'); return { itemId, index: Number(index) }; };
|
||||
/** 内容事件的签名头只在末帧上;离线队列里没有带签名头的末帧,说明上次加密被打断,这个条目无法补完 */
|
||||
const isFinalFrame = (frame: QueuedFrame): boolean => 'X-WD-Event-Signature' in frame.extraHeaders;
|
||||
|
||||
/**
|
||||
* 续传离线队列里上次没发完的内容:按条目分组,同一个条目内的帧按 FrameUploader 的规则并发上传,
|
||||
* 进度会回到聊天里的文件卡片。在后台运行,不挡住频道加载。
|
||||
* 返回 true 表示没有遗留(或都发完了)。
|
||||
*/
|
||||
export async function resumeOutbox(resource: SavedResource, session: ChannelSession, signal?: AbortSignal): Promise<boolean> {
|
||||
const items = new Map<string, number[]>();
|
||||
for (const key of await queuedFrameKeys(resource.id)) {
|
||||
const { itemId, index } = keyOf(key);
|
||||
if (pipelinedItems.has(`${resource.id}:${itemId}`)) continue;
|
||||
(items.get(itemId) ?? items.set(itemId, []).get(itemId)!).push(index);
|
||||
}
|
||||
// 上次会话里还显示「传输中」的文件,如果既没有在续传、也没有可续传的帧,就是被打断的:标成未完成,用户可以删除
|
||||
const stale = (resource.localItems ?? []).filter((item) => item.kind === 'file' && item.state.kind === 'receiving'
|
||||
&& !pipelinedItems.has(`${resource.id}:${item.id}`) && !items.has(item.id));
|
||||
for (const item of stale) await markIncomplete(resource, item.id);
|
||||
let complete = true;
|
||||
for (const [itemId, indices] of items) {
|
||||
if (signal?.aborted) return false;
|
||||
if (!await resumeItem(resource, session, itemId, indices.sort((a, b) => a - b), signal)) complete = false;
|
||||
}
|
||||
return complete;
|
||||
}
|
||||
|
||||
async function resumeItem(resource: SavedResource, session: ChannelSession, itemId: string, indices: number[], signal?: AbortSignal): Promise<boolean> {
|
||||
const keyFor = (index: number) => `${resource.id}:${itemId}:${String(index).padStart(10, '0')}`;
|
||||
const last = indices[indices.length - 1]!;
|
||||
const lastFrame = await queuedFrame(keyFor(last));
|
||||
// 没有末帧的条目补不完整:留给全局队列处理,或等用户重新发送
|
||||
if (!lastFrame || !isFinalFrame(lastFrame)) { await markIncomplete(resource, itemId); return true; }
|
||||
const queued = new Set(indices);
|
||||
const accepted: number[] = [];
|
||||
for (let index = 0; index <= last; index += 1) if (!queued.has(index)) accepted.push(index);
|
||||
const local = resource.localItems?.find((item) => item.id === itemId);
|
||||
const controller = new AbortController();
|
||||
signal?.addEventListener('abort', () => controller.abort(), { once: true });
|
||||
const meter = new TransferMeter();
|
||||
const total = last + 1;
|
||||
let lastPublished = 0;
|
||||
const publish = (sent: number, force = false) => {
|
||||
if (!local || local.kind !== 'file') return;
|
||||
const now = performance.now();
|
||||
if (!force && now - lastPublished < 150) return;
|
||||
lastPublished = now;
|
||||
const speed = meter.speed();
|
||||
publishWorkspace(resource.id, { t: 'item', item: { ...local, state: { kind: 'receiving', progress: Math.min(0.99, (accepted.length + sent) / total), phase: 'uploading', ...(speed ? { speedBytesPerSecond: speed } : {}) } } });
|
||||
};
|
||||
const uploader = new FrameUploader({
|
||||
gated: local?.kind === 'file' ? Boolean(local.gated) : false, concurrency: FRAME_CONCURRENCY, signal: controller.signal, alreadyAccepted: accepted,
|
||||
send: (frame, hooks, s) => uploadFrame(resource, session, frame, hooks, s),
|
||||
onBytes: (bytes) => { meter.add(bytes); publish(bytes / Math.max(1, lastFrame.bytes.byteLength)); },
|
||||
});
|
||||
try {
|
||||
for (const index of indices) {
|
||||
const frame = index === last ? lastFrame : await queuedFrame(keyFor(index));
|
||||
if (frame) await uploader.push(frame, index === last);
|
||||
}
|
||||
const done = await uploader.finish();
|
||||
if (local && local.kind === 'file') publishWorkspace(resource.id, { t: 'item', item: { ...local, state: done ? { kind: 'complete' } : { kind: 'incomplete' } } });
|
||||
return done;
|
||||
} catch (error) {
|
||||
if (controller.signal.aborted) return false;
|
||||
// 服务端明确拒绝(例如已被删除):这些帧再发也没用,清掉
|
||||
await dequeueItem(resource.id, itemId);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function markIncomplete(resource: SavedResource, itemId: string): Promise<void> {
|
||||
const item = resource.localItems?.find((entry) => entry.id === itemId);
|
||||
if (!item || item.state.kind === 'incomplete') return;
|
||||
const next = { ...item, state: { kind: 'incomplete' as const } };
|
||||
resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === itemId ? next : entry);
|
||||
await saveResource(resource);
|
||||
publishWorkspace(resource.id, { t: 'item', item: next });
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { fromBase64Url, IntegrityError, MAX_FRAME_DATA, type ChannelSession, type ItemDecodeResult, type ItemPlan } from '@webdrop/core';
|
||||
import { dequeueFrame, queueFrame, nextQueuedFrame, lastQueuedAt, nextQueuedLocation, dequeueLocation, type SavedResource } from '../database';
|
||||
import { dequeueFrame, queueFrame, nextQueuedFrame, lastQueuedAt, nextQueuedLocation, dequeueLocation, type QueuedFrame, type SavedResource } from '../database';
|
||||
import { registerItemLocation } from './workspaceStructure';
|
||||
import { verifyObjectEvent } from './channelDataKeys';
|
||||
import { apiFetch, apiUpload, type ApiError } from './http';
|
||||
@@ -11,21 +11,28 @@ import { storeVerifiedDownload } from './downloadStorage';
|
||||
// Frame plaintext includes a 4-byte length prefix before padding to the 256 KiB tier.
|
||||
export const MOBILE_UPLOAD_FRAME_DATA_BYTES = 256 * 1024 - 4;
|
||||
export { MAX_FRAME_DATA };
|
||||
const UPLOAD_RETRY_DELAYS_MS = [500, 1500, 4000] as const;
|
||||
|
||||
/** 正由上传流水线逐帧发送的条目(空间 id:条目 id)。全局续传 flushOutbox 要绕开它们,否则同一帧会发两遍、进度互相抢 */
|
||||
const pipelined = new Set<string>();
|
||||
export { pipelined as pipelinedItems };
|
||||
export function markPipelined(resourceId: string, itemId: string): void { pipelined.add(`${resourceId}:${itemId}`); }
|
||||
export function unmarkPipelined(resourceId: string, itemId: string): void { pipelined.delete(`${resourceId}:${itemId}`); }
|
||||
|
||||
export async function persistPlan(
|
||||
resource: SavedResource,
|
||||
plan: ItemPlan,
|
||||
onProgress?: (frames: number, bytes: number) => void,
|
||||
signedEvent?: { channelId: string; domain: 'chat' | 'files'; epoch: number; eventKind: string; targetItemId?: string },
|
||||
/** 每一帧写入离线队列之后调用;上传流水线借此边加密边上传,返回的 Promise 用来给加密端施加背压 */
|
||||
onFrame?: (frame: QueuedFrame, final: boolean) => Promise<void>,
|
||||
): Promise<{ frames: number; bytes: number }> {
|
||||
const createdAt = Math.max(Date.now(), (await lastQueuedAt(resource.id)) + 1);
|
||||
let frames = 0;
|
||||
let bytes = 0;
|
||||
const frameDigests: Uint8Array[] = [];
|
||||
let pending: { index: number; bytes: Uint8Array; extraHeaders?: Record<string, string> } | undefined;
|
||||
const persistFrame = async (frame: NonNullable<typeof pending>, extraHeaders: Record<string, string> = {}) => {
|
||||
await queueFrame({
|
||||
const persistFrame = async (frame: NonNullable<typeof pending>, extraHeaders: Record<string, string> = {}, final = false) => {
|
||||
const queued: QueuedFrame = {
|
||||
key: `${resource.id}:${plan.itemId}:${String(frame.index).padStart(10, '0')}`,
|
||||
resourceId: resource.id,
|
||||
itemId: plan.itemId,
|
||||
@@ -33,15 +40,17 @@ export async function persistPlan(
|
||||
bytes: frame.bytes,
|
||||
extraHeaders: { ...(frame.index === 0 ? plan.frame0Headers : {}), ...frame.extraHeaders, ...extraHeaders },
|
||||
createdAt: createdAt + frame.index,
|
||||
});
|
||||
};
|
||||
// 提供了 onFrame 的调用方(上传流水线)自己决定何时落盘,见 frameStash
|
||||
if (!onFrame) await queueFrame(queued);
|
||||
frames += 1;
|
||||
bytes += frame.bytes.byteLength;
|
||||
onProgress?.(frames, bytes);
|
||||
await onFrame?.(queued, final);
|
||||
};
|
||||
for await (const frame of plan.frames) {
|
||||
if (pending) await persistFrame(pending);
|
||||
const frameBuffer = frame.bytes.slice().buffer as ArrayBuffer;
|
||||
frameDigests.push(new Uint8Array(await crypto.subtle.digest('SHA-256', frameBuffer)));
|
||||
frameDigests.push(new Uint8Array(await crypto.subtle.digest('SHA-256', frame.bytes)));
|
||||
pending = frame;
|
||||
}
|
||||
if (pending) {
|
||||
@@ -50,7 +59,7 @@ export async function persistPlan(
|
||||
epoch: signedEvent.epoch, itemId: plan.itemId, eventKind: signedEvent.eventKind,
|
||||
...(signedEvent.targetItemId ? { targetItemId: signedEvent.targetItemId } : {}), frameDigests })
|
||||
: {};
|
||||
await persistFrame(pending, headers);
|
||||
await persistFrame(pending, headers, true);
|
||||
}
|
||||
return { frames, bytes };
|
||||
}
|
||||
@@ -61,25 +70,88 @@ function writeRole(session: ChannelSession): 'owner' | 'write' {
|
||||
throw new Error('此链接没有上传权限');
|
||||
}
|
||||
|
||||
/** The outbox contains ciphertext only. Every retry sends these exact bytes. */
|
||||
/** 一次请求这么久没有任何上传进展就放弃重试:弱网下连接可能静默卡死,不能一直等 */
|
||||
const FRAME_STALL_TIMEOUT_MS = 30_000;
|
||||
const MAX_UPLOAD_ATTEMPTS = 8;
|
||||
const MAX_RETRY_WAIT_MS = 60_000;
|
||||
|
||||
/** 第 n 次失败后的等待:指数退避加抖动;服务端给了 Retry-After(限流)就按它来 */
|
||||
export function retryDelayMs(attempt: number, retryAfterSeconds?: number): number {
|
||||
if (retryAfterSeconds) return Math.min(MAX_RETRY_WAIT_MS, retryAfterSeconds * 1000 + Math.random() * 250);
|
||||
return Math.min(15_000, 500 * 2 ** attempt) * (0.75 + Math.random() * 0.5);
|
||||
}
|
||||
|
||||
/**
|
||||
* 上传一帧。可重试的错误(断网、卡住、5xx、429 等)按退避重试,重试发送的永远是同一份字节(protocol §5)。
|
||||
* 返回 false 表示重试用尽(多半断网),帧仍留在离线队列;不可重试的错误会抛出;取消会抛 AbortError。
|
||||
* manageQueue 为 false 时调用方自己管理离线队列(frameStash),这里不动它。
|
||||
*/
|
||||
export async function uploadFrame(
|
||||
resource: SavedResource,
|
||||
session: ChannelSession,
|
||||
frame: QueuedFrame,
|
||||
hooks?: { onLoaded?: (loadedBytes: number, totalBytes: number) => void; onAccepted?: () => void; onRetry?: () => void; manageQueue?: boolean },
|
||||
signal?: AbortSignal,
|
||||
): Promise<boolean> {
|
||||
const role = resource.kind === 'share' ? 'owner' : writeRole(session);
|
||||
const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`;
|
||||
let maxLoadedBytes = 0;
|
||||
for (let attempt = 0; attempt < MAX_UPLOAD_ATTEMPTS; attempt += 1) {
|
||||
try {
|
||||
await apiUpload(
|
||||
path,
|
||||
{ ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders },
|
||||
// 用 Blob 发而不是字节数组:浏览器对字节数组请求体要多拷贝几次,实测上传只有 20–35 MB/s;Blob 走快速通道,单连接 100+ MB/s。
|
||||
// 每次重试都从同一份 frame.bytes 重新包装,发出的字节完全相同(protocol §5)
|
||||
new Blob([frame.bytes as BlobPart]),
|
||||
(loaded, total) => { maxLoadedBytes = Math.max(maxLoadedBytes, loaded); hooks?.onLoaded?.(maxLoadedBytes, total); },
|
||||
signal,
|
||||
FRAME_STALL_TIMEOUT_MS,
|
||||
);
|
||||
if (hooks?.manageQueue !== false) await dequeueFrame(frame.key);
|
||||
hooks?.onAccepted?.();
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (signal?.aborted || (error as Error | undefined)?.name === 'AbortError') throw error;
|
||||
const apiError = error as ApiError;
|
||||
const retryable = apiError.status === undefined || apiError.status === 408 || apiError.status === 425
|
||||
|| apiError.status === 429 || apiError.status === 499 || apiError.status >= 500;
|
||||
if (!retryable) throw error;
|
||||
if (attempt === MAX_UPLOAD_ATTEMPTS - 1) return false;
|
||||
hooks?.onRetry?.();
|
||||
maxLoadedBytes = 0;
|
||||
await new Promise((resolve) => setTimeout(resolve, retryDelayMs(attempt, apiError.retryAfter)));
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 把离线队列里的帧(和待登记的位置)发完。默认处理整个空间的队列,但跳过正由上传流水线处理的条目;
|
||||
* only 指定时只处理这些条目,发文字消息就用它,不被已入队的大文件堵住。
|
||||
*/
|
||||
export async function flushOutbox(
|
||||
resource: SavedResource,
|
||||
session: ChannelSession,
|
||||
onFrameProgress?: (frame: { itemId: string; index: number }, loadedBytes: number, totalBytes: number, complete: boolean) => void,
|
||||
scope?: { only?: ReadonlySet<string> },
|
||||
): Promise<boolean> {
|
||||
return serial(`outbox:${resource.id}`, () => flush(resource, session, onFrameProgress));
|
||||
const lane = scope?.only ? [...scope.only].sort().join(',') : 'all';
|
||||
return serial(`outbox:${resource.id}:${lane}`, () => flush(resource, session, onFrameProgress, scope));
|
||||
}
|
||||
|
||||
async function flush(
|
||||
resource: SavedResource,
|
||||
session: ChannelSession,
|
||||
onFrameProgress?: (frame: { itemId: string; index: number }, loadedBytes: number, totalBytes: number, complete: boolean) => void,
|
||||
scope?: { only?: ReadonlySet<string> },
|
||||
): Promise<boolean> {
|
||||
const accept = (itemId: string) => !pipelined.has(`${resource.id}:${itemId}`) && (!scope?.only || scope.only.has(itemId));
|
||||
for (;;) {
|
||||
const frame = await nextQueuedFrame(resource.id);
|
||||
const frame = await nextQueuedFrame(resource.id, accept);
|
||||
if (!frame) {
|
||||
if (resource.kind !== 'channel') return true;
|
||||
const location = await nextQueuedLocation(resource.id);
|
||||
const location = await nextQueuedLocation(resource.id, accept);
|
||||
if (!location) return true;
|
||||
try {
|
||||
await registerItemLocation(resource.id, { id: location.itemId, threadId: location.threadId, folderId: location.folderId });
|
||||
@@ -91,34 +163,10 @@ async function flush(
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const role = resource.kind === 'share' ? 'owner' : writeRole(session);
|
||||
const path = resource.kind === 'share' ? `/shares/${resource.id}/frames` : `/channels/${resource.id}/frames`;
|
||||
let maxLoadedBytes = 0;
|
||||
let accepted = false;
|
||||
for (let attempt = 0; attempt <= UPLOAD_RETRY_DELAYS_MS.length; attempt += 1) {
|
||||
try {
|
||||
await apiUpload(
|
||||
path,
|
||||
{ ...session.authHeaders(role), 'Content-Type': 'application/octet-stream', ...frame.extraHeaders },
|
||||
frame.bytes.slice().buffer as ArrayBuffer,
|
||||
(loaded, total) => {
|
||||
maxLoadedBytes = Math.max(maxLoadedBytes, loaded);
|
||||
onFrameProgress?.(frame, maxLoadedBytes, total, false);
|
||||
},
|
||||
);
|
||||
await dequeueFrame(frame.key);
|
||||
onFrameProgress?.(frame, frame.bytes.byteLength, frame.bytes.byteLength, true);
|
||||
accepted = true;
|
||||
break;
|
||||
} catch (error) {
|
||||
const apiError = error as ApiError;
|
||||
const retryable = apiError.status === undefined || apiError.status === 408 || apiError.status === 425
|
||||
|| apiError.status === 429 || apiError.status === 499 || apiError.status >= 500;
|
||||
if (!retryable) throw error;
|
||||
if (attempt === UPLOAD_RETRY_DELAYS_MS.length) return false;
|
||||
await new Promise((resolve) => setTimeout(resolve, UPLOAD_RETRY_DELAYS_MS[attempt]));
|
||||
}
|
||||
}
|
||||
const accepted = await uploadFrame(resource, session, frame, {
|
||||
onLoaded: (loaded, total) => onFrameProgress?.(frame, loaded, total, false),
|
||||
onAccepted: () => onFrameProgress?.(frame, frame.bytes.byteLength, frame.bytes.byteLength, true),
|
||||
});
|
||||
if (!accepted) return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import type { QueuedFrame } from '../database';
|
||||
import { estimateCipherBytes, FrameUploader, TransferMeter, type SendFrame } from './uploadPipeline';
|
||||
|
||||
const frame = (index: number, size = 100): QueuedFrame => ({ key: `k${index}`, resourceId: 'r', itemId: 'i', index, bytes: new Uint8Array(size), extraHeaders: {}, createdAt: index });
|
||||
const tick = () => new Promise((resolve) => setTimeout(resolve, 0));
|
||||
|
||||
/** 记录每一帧开始、结束的顺序和同时在途的最大数量 */
|
||||
function fakeServer(options: { delay?: (index: number) => number; reject?: (index: number) => 'pause' | 'error' | undefined } = {}) {
|
||||
const log: string[] = []; let inflight = 0; let peak = 0;
|
||||
const send: SendFrame = async (f, { onLoaded }) => {
|
||||
inflight += 1; peak = Math.max(peak, inflight); log.push(`start:${f.index}`);
|
||||
await new Promise((resolve) => setTimeout(resolve, options.delay?.(f.index) ?? 1));
|
||||
onLoaded(f.bytes.byteLength);
|
||||
inflight -= 1;
|
||||
const verdict = options.reject?.(f.index);
|
||||
if (verdict === 'error') throw new Error('boom');
|
||||
if (verdict === 'pause') return false;
|
||||
log.push(`done:${f.index}`);
|
||||
return true;
|
||||
};
|
||||
return { send, log, peak: () => peak };
|
||||
}
|
||||
|
||||
async function run(uploader: FrameUploader, count: number) {
|
||||
for (let index = 0; index < count; index += 1) await uploader.push(frame(index), index === count - 1);
|
||||
return uploader.finish();
|
||||
}
|
||||
|
||||
describe('FrameUploader', () => {
|
||||
it('受限文件:第 0 帧先传,中间帧并发,末帧等其余帧都确认后才发', async () => {
|
||||
const server = fakeServer({ delay: (index) => (index === 2 ? 20 : 2) });
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 3, send: server.send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 6)).toBe(true);
|
||||
expect(server.log.indexOf('done:0')).toBeLessThan(server.log.indexOf('start:1'));
|
||||
expect(server.peak()).toBeGreaterThan(1);
|
||||
expect(server.peak()).toBeLessThanOrEqual(3);
|
||||
const start5 = server.log.indexOf('start:5');
|
||||
for (const index of [0, 1, 2, 3, 4]) expect(server.log.indexOf(`done:${index}`)).toBeLessThan(start5);
|
||||
});
|
||||
|
||||
it('非受限文件:逐帧按顺序上传,不并发', async () => {
|
||||
const server = fakeServer();
|
||||
const uploader = new FrameUploader({ gated: false, concurrency: 4, send: server.send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 4)).toBe(true);
|
||||
expect(server.peak()).toBe(1);
|
||||
expect(server.log).toEqual(['start:0', 'done:0', 'start:1', 'done:1', 'start:2', 'done:2', 'start:3', 'done:3']);
|
||||
});
|
||||
|
||||
it('只有一帧时直接上传', async () => {
|
||||
const server = fakeServer();
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 4, send: server.send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 1)).toBe(true);
|
||||
expect(server.log).toEqual(['start:0', 'done:0']);
|
||||
});
|
||||
|
||||
it('某帧重试用尽暂停:不再发新帧,finish 返回 false', async () => {
|
||||
const server = fakeServer({ reject: (index) => (index === 1 ? 'pause' : undefined) });
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 2, send: server.send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 8)).toBe(false);
|
||||
expect(server.log).not.toContain('start:7');
|
||||
});
|
||||
|
||||
it('不可重试的错误:push 或 finish 抛出,并且不再发新帧', async () => {
|
||||
const server = fakeServer({ reject: (index) => (index === 1 ? 'error' : undefined) });
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 2, send: server.send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
await expect(run(uploader, 8)).rejects.toThrow('boom');
|
||||
expect(server.log).not.toContain('start:7');
|
||||
});
|
||||
|
||||
it('积压超过上限时 push 会等待,不让加密一口气跑完整个文件', async () => {
|
||||
const server = fakeServer({ delay: () => 15 });
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 1, send: server.send, signal: new AbortController().signal, onBytes: () => {}, buffer: 2 });
|
||||
let pushed = 0;
|
||||
const producer = (async () => { for (let index = 0; index < 6; index += 1) { await uploader.push(frame(index), index === 5); pushed += 1; } })();
|
||||
await tick(); await tick();
|
||||
expect(pushed).toBeLessThan(6);
|
||||
await producer; expect(await uploader.finish()).toBe(true);
|
||||
});
|
||||
|
||||
it('已取消的信号:push 直接中止', async () => {
|
||||
const controller = new AbortController(); controller.abort();
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 2, send: fakeServer().send, signal: controller.signal, onBytes: () => {} });
|
||||
await expect(uploader.push(frame(0), true)).rejects.toMatchObject({ name: 'AbortError' });
|
||||
});
|
||||
|
||||
it('上报的是已确认加在途的字节数', async () => {
|
||||
const seen: number[] = [];
|
||||
const uploader = new FrameUploader({ gated: false, concurrency: 1, send: fakeServer().send, signal: new AbortController().signal, onBytes: (sent) => seen.push(sent) });
|
||||
await run(uploader, 3);
|
||||
expect(Math.max(...seen)).toBe(300);
|
||||
expect(seen[seen.length - 1]).toBe(300);
|
||||
});
|
||||
});
|
||||
|
||||
describe('FrameUploader 自适应并发', () => {
|
||||
it('遇到重试就把并发减半,不再占满连接', async () => {
|
||||
let retried = false; let droppedTo = 0;
|
||||
const send: SendFrame = async (f, hooks) => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||||
if (f.index === 1 && !retried) { retried = true; hooks.onRetry(); droppedTo = uploader.currentConcurrency; }
|
||||
return true;
|
||||
};
|
||||
const uploader: FrameUploader = new FrameUploader({ gated: true, concurrency: 4, send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 20)).toBe(true);
|
||||
expect(droppedTo).toBe(2);
|
||||
});
|
||||
|
||||
it('连续顺利后并发度慢慢恢复', async () => {
|
||||
let first = true;
|
||||
const send: SendFrame = async (f, hooks) => { await new Promise((resolve) => setTimeout(resolve, 1)); if (first && f.index === 1) { first = false; hooks.onRetry(); } return true; };
|
||||
const uploader = new FrameUploader({ gated: true, concurrency: 4, send, signal: new AbortController().signal, onBytes: () => {} });
|
||||
expect(await run(uploader, 60)).toBe(true);
|
||||
expect(uploader.currentConcurrency).toBeGreaterThan(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('TransferMeter', () => {
|
||||
it('用最近一段时间估算速度和剩余时间', () => {
|
||||
const meter = new TransferMeter(3000);
|
||||
expect(meter.speed()).toBeUndefined();
|
||||
meter.add(0, 0); meter.add(1_000_000, 1000); meter.add(2_000_000, 2000);
|
||||
expect(meter.speed()).toBeCloseTo(1_000_000, -3);
|
||||
expect(meter.eta(2_000_000, 5_000_000)).toBe(3);
|
||||
});
|
||||
|
||||
it('速度只看窗口内的样本,网络变慢后很快反映出来', () => {
|
||||
const meter = new TransferMeter(3000);
|
||||
meter.add(0, 0); meter.add(10_000_000, 1000); meter.add(10_100_000, 5000); meter.add(10_200_000, 7000);
|
||||
expect(meter.speed()!).toBeLessThan(100_000);
|
||||
});
|
||||
});
|
||||
|
||||
describe('estimateCipherBytes', () => {
|
||||
it('按帧数乘第一帧密文长度估算', () => {
|
||||
expect(estimateCipherBytes(0, 1000, 1016)).toBe(1016);
|
||||
expect(estimateCipherBytes(2500, 1000, 1016)).toBe(3048);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,156 @@
|
||||
import type { QueuedFrame } from '../database';
|
||||
|
||||
/**
|
||||
* 单个文件的帧上传调度(protocol §5、服务端 AppendFrame 的约束):
|
||||
* - 第 0 帧先单独上传(它在服务端创建条目);
|
||||
* - 受限(gated)文件的中间帧彼此独立,可以并发;非受限文件的帧会按到达顺序进入日志,必须逐帧顺序上传;
|
||||
* - 末帧带整份内容的签名,服务端要求前面的帧都已入库,所以必须等其余帧全部确认后再发。
|
||||
*/
|
||||
export type SendFrame = (frame: QueuedFrame, hooks: { onLoaded: (loadedBytes: number) => void; onRetry: () => void }, signal: AbortSignal) => Promise<boolean>;
|
||||
|
||||
interface Options {
|
||||
/** 受限文件的中间帧可以并发 */
|
||||
gated: boolean;
|
||||
concurrency: number;
|
||||
send: SendFrame;
|
||||
signal: AbortSignal;
|
||||
/** 已确认 + 在途的字节数变化时回调 */
|
||||
onBytes: (sentBytes: number) => void;
|
||||
/** 生产者(加密)领先上传的最大帧数,防止整个文件堆在内存里 */
|
||||
buffer?: number;
|
||||
/** 续传:这些帧上次已经被服务端确认,不必再发 */
|
||||
alreadyAccepted?: Iterable<number>;
|
||||
}
|
||||
|
||||
export class FrameUploader {
|
||||
private ready: { frame: QueuedFrame; final: boolean }[] = [];
|
||||
private inflight = 0;
|
||||
private readonly accepted = new Set<number>();
|
||||
private readonly loaded = new Map<number, number>();
|
||||
private acceptedBytes = 0;
|
||||
private finalIndex: number | undefined;
|
||||
private paused = false;
|
||||
private error: unknown;
|
||||
private failed = false;
|
||||
private waiters: (() => void)[] = [];
|
||||
|
||||
constructor(private readonly options: Options) {
|
||||
this.dynamicLimit = options.concurrency;
|
||||
for (const index of options.alreadyAccepted ?? []) this.accepted.add(index);
|
||||
}
|
||||
|
||||
/** 已确认的帧数(含续传前已确认的) */
|
||||
get acceptedCount(): number { return this.accepted.size; }
|
||||
|
||||
/** 当前并发度:遇到重试就减半,连续顺利再慢慢加回去(弱网下少占连接,好网下跑满) */
|
||||
private dynamicLimit: number;
|
||||
private streak = 0;
|
||||
get currentConcurrency(): number { return this.limit; }
|
||||
private get limit(): number { return this.options.gated ? Math.max(1, Math.min(this.options.concurrency, this.dynamicLimit)) : 1; }
|
||||
private get bufferLimit(): number { return this.options.buffer ?? this.limit * 2 + 2; }
|
||||
|
||||
/** 一帧已加密并写入离线队列。积压太多时等待,让加密不要跑得比上传快太多 */
|
||||
async push(frame: QueuedFrame, final: boolean): Promise<void> {
|
||||
if (this.failed) throw this.error;
|
||||
if (this.options.signal.aborted) throw new DOMException('上传已取消', 'AbortError');
|
||||
if (final) this.finalIndex = frame.index;
|
||||
// 已暂停:帧留在离线队列里,等网络恢复后由 flushOutbox 续传
|
||||
if (this.paused) return;
|
||||
this.ready.push({ frame, final });
|
||||
this.pump();
|
||||
while (!final && !this.paused && !this.failed && this.ready.length + this.inflight >= this.bufferLimit) await this.wait();
|
||||
if (this.failed) throw this.error;
|
||||
}
|
||||
|
||||
/** 等所有帧确认。返回 false 表示中途暂停(重试用尽,多半断网);失败会抛出 */
|
||||
async finish(): Promise<boolean> {
|
||||
while (!this.settled()) await this.wait();
|
||||
if (this.failed) throw this.error;
|
||||
return !this.paused;
|
||||
}
|
||||
|
||||
private settled(): boolean {
|
||||
if (this.failed || this.paused) return this.inflight === 0;
|
||||
return this.finalIndex !== undefined && this.accepted.size === this.finalIndex + 1;
|
||||
}
|
||||
|
||||
private canStart(frame: QueuedFrame, final: boolean): boolean {
|
||||
if (final) return this.accepted.size === frame.index && this.inflight === 0;
|
||||
if (frame.index === 0) return this.inflight === 0;
|
||||
if (!this.options.gated) return this.accepted.size === frame.index && this.inflight === 0;
|
||||
return this.accepted.has(0);
|
||||
}
|
||||
|
||||
private pump(): void {
|
||||
while (!this.failed && !this.paused && this.inflight < this.limit) {
|
||||
const at = this.ready.findIndex((entry) => this.canStart(entry.frame, entry.final));
|
||||
if (at < 0) break;
|
||||
const [entry] = this.ready.splice(at, 1);
|
||||
this.start(entry!.frame);
|
||||
}
|
||||
this.notify();
|
||||
}
|
||||
|
||||
private start(frame: QueuedFrame): void {
|
||||
this.inflight += 1;
|
||||
let retried = false;
|
||||
this.options.send(frame, {
|
||||
onLoaded: (loaded) => { this.loaded.set(frame.index, loaded); this.report(); },
|
||||
onRetry: () => { retried = true; this.dynamicLimit = Math.max(1, Math.floor(this.dynamicLimit / 2)); this.streak = 0; },
|
||||
}, this.options.signal)
|
||||
.then((accepted) => {
|
||||
this.loaded.delete(frame.index);
|
||||
if (accepted) {
|
||||
this.accepted.add(frame.index); this.acceptedBytes += frame.bytes.byteLength;
|
||||
if (!retried && ++this.streak >= 8 && this.dynamicLimit < this.options.concurrency) { this.dynamicLimit += 1; this.streak = 0; }
|
||||
}
|
||||
else { this.paused = true; this.ready = []; }
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
this.loaded.delete(frame.index);
|
||||
if (!this.failed) { this.failed = true; this.error = error; }
|
||||
this.ready = [];
|
||||
})
|
||||
.finally(() => { this.inflight -= 1; this.report(); this.pump(); });
|
||||
}
|
||||
|
||||
private report(): void {
|
||||
let sent = this.acceptedBytes;
|
||||
for (const value of this.loaded.values()) sent += value;
|
||||
this.options.onBytes(sent);
|
||||
}
|
||||
|
||||
private wait(): Promise<void> { return new Promise((resolve) => { this.waiters.push(resolve); }); }
|
||||
private notify(): void { const waiting = this.waiters; this.waiters = []; for (const resume of waiting) resume(); }
|
||||
}
|
||||
|
||||
/** 最近几秒的滑动平均速度,网络变化后很快跟上,不会被开头的数据拖累 */
|
||||
export class TransferMeter {
|
||||
private samples: { at: number; bytes: number }[] = [];
|
||||
constructor(private readonly windowMs = 3000) {}
|
||||
|
||||
add(bytes: number, at = performance.now()): void {
|
||||
this.samples.push({ at, bytes });
|
||||
const from = at - this.windowMs;
|
||||
while (this.samples.length > 2 && this.samples[1]!.at <= from) this.samples.shift();
|
||||
}
|
||||
|
||||
/** 字节每秒;样本不足或时间太短时为 undefined */
|
||||
speed(): number | undefined {
|
||||
const first = this.samples[0]; const last = this.samples[this.samples.length - 1];
|
||||
if (!first || !last || last.at - first.at < 400) return undefined;
|
||||
const value = (last.bytes - first.bytes) * 1000 / (last.at - first.at);
|
||||
return value > 0 ? value : undefined;
|
||||
}
|
||||
|
||||
/** 剩余秒数 */
|
||||
eta(sentBytes: number, totalBytes: number): number | undefined {
|
||||
const speed = this.speed();
|
||||
return speed ? Math.max(0, Math.ceil((totalBytes - sentBytes) / speed)) : undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/** 按第一帧的密文长度估算整个文件上传的总字节数;全部加密完成后会用精确值替换 */
|
||||
export function estimateCipherBytes(fileSize: number, maxFrameData: number, firstFrameBytes: number): number {
|
||||
return Math.max(1, Math.ceil(fileSize / maxFrameData)) * firstFrameBytes;
|
||||
}
|
||||
@@ -0,0 +1,192 @@
|
||||
import { previewKindOf } from '@/lib/fileKind';
|
||||
import { dequeueFrame, dequeueItem, queueFrame, queueLocation, saveResource, type QueuedFrame, type SavedResource } from '../database';
|
||||
import { errorMessage } from '../errors';
|
||||
import type { FileItem, FileUploadOptions } from '../types';
|
||||
import { prepareObject } from './channelDataKeys';
|
||||
import { queueFileCatalog } from './channelThreads';
|
||||
import { deviceIdentity, publishWorkspace } from './presentation';
|
||||
import { senderSnapshot } from './senderSnapshot';
|
||||
import { blobs, getSession } from './sessions';
|
||||
import { flushOutbox, markPipelined, MAX_FRAME_DATA, MOBILE_UPLOAD_FRAME_DATA_BYTES, persistPlan, unmarkPipelined, uploadFrame } from './transfer';
|
||||
import { FrameStash } from './frameStash';
|
||||
import { estimateCipherBytes, FrameUploader, TransferMeter } from './uploadPipeline';
|
||||
|
||||
/** 受限文件内部同时在途的帧数。再多对单条连接的带宽提升有限,还会占满浏览器的并发连接 */
|
||||
const FRAME_CONCURRENCY = 4;
|
||||
const PROGRESS_INTERVAL_MS = 150;
|
||||
|
||||
interface Job {
|
||||
resourceId: string;
|
||||
file: File;
|
||||
options: FileUploadOptions | undefined;
|
||||
threadId: string;
|
||||
folderId: string | undefined;
|
||||
localId: string | undefined;
|
||||
controller: AbortController;
|
||||
/** 到运行时才建立的条目 id,取消时据此清理 */
|
||||
itemId?: string;
|
||||
run: () => Promise<void>;
|
||||
settle: { resolve: () => void; reject: (error: unknown) => void };
|
||||
}
|
||||
interface Lane { active?: Job; waiting: Job[] }
|
||||
|
||||
/** 每个空间一条上传队列:一次只传一个文件,其余排队;文件内部的帧可以并发(见 FrameUploader) */
|
||||
const lanes = new Map<string, Lane>();
|
||||
|
||||
const isAbort = (error: unknown): boolean => (error as Error | undefined)?.name === 'AbortError';
|
||||
|
||||
export function enqueueFileUploads(resourceId: string, files: File[], options: FileUploadOptions | undefined,
|
||||
threadId: string, folderId: string | undefined, localIds: string[] | undefined): Promise<void> {
|
||||
const lane = lanes.get(resourceId) ?? { waiting: [] };
|
||||
lanes.set(resourceId, lane);
|
||||
const done = files.map((file, index) => new Promise<void>((resolve, reject) => {
|
||||
const job: Job = { resourceId, file, options, threadId, folderId, localId: localIds?.[index], controller: new AbortController(),
|
||||
run: () => runJob(job), settle: { resolve, reject } };
|
||||
lane.waiting.push(job);
|
||||
}));
|
||||
drain(resourceId, lane);
|
||||
return Promise.allSettled(done).then((results) => {
|
||||
const failed = results.find((result): result is PromiseRejectedResult => result.status === 'rejected');
|
||||
if (failed) throw failed.reason;
|
||||
});
|
||||
}
|
||||
|
||||
/** 取消排队中或上传中的文件。返回是否找到了这个任务 */
|
||||
export function cancelFileUpload(resourceId: string, localId: string): boolean {
|
||||
const lane = lanes.get(resourceId);
|
||||
if (!lane) return false;
|
||||
const waiting = lane.waiting.findIndex((job) => job.localId === localId);
|
||||
if (waiting >= 0) { const [job] = lane.waiting.splice(waiting, 1); job!.settle.resolve(); return true; }
|
||||
if (lane.active?.localId === localId) { lane.active.controller.abort(); return true; }
|
||||
return false;
|
||||
}
|
||||
|
||||
function drain(resourceId: string, lane: Lane): void {
|
||||
if (lane.active) return;
|
||||
const job = lane.waiting.shift();
|
||||
if (!job) { lanes.delete(resourceId); return; }
|
||||
lane.active = job;
|
||||
job.run().then(job.settle.resolve, job.settle.reject).finally(() => { lane.active = undefined; drain(resourceId, lane); });
|
||||
}
|
||||
|
||||
async function runJob(job: Job): Promise<void> {
|
||||
const { resourceId, file, threadId, folderId, localId } = job;
|
||||
const signal = job.controller.signal;
|
||||
const { session, resource } = await getSession(resourceId);
|
||||
const options = job.options ?? { directPreview: true, maxDownloads: resource.maxDownloads ?? 0 };
|
||||
if (resource.kind !== 'channel' || (!session.can('owner') && !session.can('write'))) throw new Error('此链接没有上传权限');
|
||||
const device = await deviceIdentity(session.endpointSecret);
|
||||
const maxFrameData = device.kind === 'phone' ? MOBILE_UPLOAD_FRAME_DATA_BYTES : MAX_FRAME_DATA;
|
||||
const author = await senderSnapshot();
|
||||
const ts = Date.now();
|
||||
let item: FileItem | undefined;
|
||||
let stash: FrameStash | undefined;
|
||||
try {
|
||||
if (signal.aborted) return;
|
||||
const prepared = await prepareObject(resource, session, threadId, 'files');
|
||||
const plan = session.encryptFile(file, {}, {
|
||||
itemId: prepared.itemIdBytes,
|
||||
...(options.directPreview ? { gated: 'auto' as const } : { gated: true, maxDownloads: options.maxDownloads }),
|
||||
ts, maxFrameData, threadId, folderId,
|
||||
});
|
||||
item = {
|
||||
id: plan.itemId, kind: 'file', ts, state: { kind: 'receiving', progress: 0, phase: 'encrypting' },
|
||||
...(author ? { author } : {}),
|
||||
...(threadId ? { threadId } : {}), ...(folderId ? { folderId } : {}), ...(localId ? { localId } : {}),
|
||||
name: file.name, mime: file.type || 'application/octet-stream', size: file.size,
|
||||
remaining: plan.gated
|
||||
? (options.directPreview ? resource.maxDownloads ?? 0 : options.maxDownloads) || null
|
||||
: null,
|
||||
preview: previewKindOf(file.type, file.name), gated: plan.gated,
|
||||
};
|
||||
job.itemId = item.id;
|
||||
resource.localItems = [...(resource.localItems ?? []).filter((entry) => entry.id !== item?.id), item];
|
||||
await saveResource(resource);
|
||||
blobs.set(`${resourceId}:${item.id}`, file);
|
||||
publishWorkspace(resourceId, { t: 'item', item });
|
||||
markPipelined(resourceId, item.id);
|
||||
|
||||
const progress = createProgress(resourceId, () => item, (next) => { item = next; });
|
||||
// 帧先直接上传,来不及上传的才在后台写进离线队列(见 frameStash);离线队列由 stash 管理,上传成功后不用再去删
|
||||
const queued = new FrameStash(queueFrame, dequeueFrame);
|
||||
stash = queued;
|
||||
const uploader = new FrameUploader({
|
||||
gated: plan.gated, concurrency: FRAME_CONCURRENCY, signal,
|
||||
send: (frame, hooks, s) => uploadFrame(resource, session, frame, { ...hooks, manageQueue: false, onAccepted: () => queued.accepted(frame) }, s),
|
||||
onBytes: progress.sent,
|
||||
});
|
||||
const planned = await persistPlan(resource, plan, undefined,
|
||||
{ channelId: prepared.channelId, domain: prepared.domain, epoch: prepared.epoch, eventKind: 'file.publish' },
|
||||
async (frame: QueuedFrame, final) => {
|
||||
if (frame.index === 0) progress.estimate(estimateCipherBytes(file.size, maxFrameData, frame.bytes.byteLength));
|
||||
await queued.add(frame);
|
||||
await uploader.push(frame, final);
|
||||
});
|
||||
progress.exact(planned.bytes);
|
||||
await queueFileCatalog(resource, session, item);
|
||||
await queueLocation(resourceId, item.id, threadId, folderId);
|
||||
const framesUploaded = await uploader.finish();
|
||||
// 没传完的(断网暂停)帧要先写进离线队列,flushOutbox 才续传得了
|
||||
await queued.drain();
|
||||
unmarkPipelined(resourceId, item.id);
|
||||
// 帧都确认后,再发目录条目和位置登记(以及上次没发完的遗留)
|
||||
const uploaded = framesUploaded && await flushOutbox(resource, session);
|
||||
await finish(resource, item, uploaded ? { kind: 'complete' } : { kind: 'incomplete' });
|
||||
if (!uploaded) publishWorkspace(resourceId, { t: 'error', message: '上传暂停了。重新打开空间并连接网络后,可以继续发送已加密内容。' });
|
||||
} catch (error) {
|
||||
if (item) unmarkPipelined(resourceId, item.id);
|
||||
if (signal.aborted || isAbort(error)) { await stash?.discard(); await discard(resource, item); return; }
|
||||
job.controller.abort();
|
||||
// 失败时已加密还没确认的帧要保住,之后才能续传
|
||||
await stash?.drain().catch(() => undefined);
|
||||
if (item) await finish(resource, item, { kind: 'incomplete' });
|
||||
publishWorkspace(resourceId, { t: 'error', message: `文件上传失败：${errorMessage(error)}` });
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function finish(resource: SavedResource, item: FileItem, state: FileItem['state']): Promise<void> {
|
||||
const next = { ...item, state };
|
||||
resource.localItems = (resource.localItems ?? []).map((entry) => entry.id === next.id ? next : entry);
|
||||
await saveResource(resource);
|
||||
publishWorkspace(resource.id, { t: 'item', item: next });
|
||||
}
|
||||
|
||||
/** 取消:清掉离线队列里这个文件的帧和本地条目。服务端已收下的帧会随空间保留期自然过期 */
|
||||
async function discard(resource: SavedResource, item: FileItem | undefined): Promise<void> {
|
||||
if (!item) return;
|
||||
await dequeueItem(resource.id, item.id);
|
||||
resource.localItems = (resource.localItems ?? []).filter((entry) => entry.id !== item.id);
|
||||
await saveResource(resource);
|
||||
blobs.delete(`${resource.id}:${item.id}`);
|
||||
publishWorkspace(resource.id, { t: 'removed', id: item.id });
|
||||
}
|
||||
|
||||
/**
|
||||
* 把「已发送字节数」换算成一条 0–100% 的进度,并带上速度和剩余时间。
|
||||
* 总量先按第一帧估算,全部加密完成后换成精确值;最后一帧确认前封顶 99%。
|
||||
*/
|
||||
function createProgress(resourceId: string, current: () => FileItem | undefined, update: (item: FileItem) => void) {
|
||||
const meter = new TransferMeter();
|
||||
let total = 0; let lastSent = 0; let lastPublished = 0;
|
||||
const publish = (force: boolean) => {
|
||||
const item = current();
|
||||
if (!item || total <= 0) return;
|
||||
const now = performance.now();
|
||||
if (!force && now - lastPublished < PROGRESS_INTERVAL_MS) return;
|
||||
lastPublished = now;
|
||||
const speed = meter.speed(); const eta = meter.eta(lastSent, total);
|
||||
const next: FileItem = { ...item, state: {
|
||||
kind: 'receiving', progress: Math.max(0, Math.min(0.99, lastSent / total)), phase: lastSent > 0 ? 'uploading' : 'encrypting',
|
||||
sentBytes: Math.min(lastSent, total), totalBytes: total,
|
||||
...(speed === undefined ? {} : { speedBytesPerSecond: speed }), ...(eta === undefined ? {} : { etaSeconds: eta }),
|
||||
} };
|
||||
update(next);
|
||||
publishWorkspace(resourceId, { t: 'item', item: next });
|
||||
};
|
||||
return {
|
||||
estimate: (bytes: number) => { total = bytes; publish(true); },
|
||||
exact: (bytes: number) => { total = bytes; publish(true); },
|
||||
sent: (bytes: number) => { lastSent = bytes; meter.add(bytes); publish(false); },
|
||||
};
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { WorkspaceBody, WorkspaceChannelCreatedBody, WorkspaceChannelRequest, WorkspaceGroupCreatedBody, WorkspaceMoveRequest, WorkspaceNameRequest } from '../generated/api';
|
||||
import type { WorkspaceBody, WorkspaceChannelCreatedBody, WorkspaceChannelRequest, WorkspaceGroupCreatedBody, WorkspaceMoveRequest, WorkspaceNameRequest, WorkspaceRoleCreatedBody, WorkspaceRoleRequest } from '../generated/api';
|
||||
import type { FileItem } from '../types';
|
||||
import { getResource, cacheResource } from '../database';
|
||||
import { getSession } from './sessions';
|
||||
@@ -40,7 +40,7 @@ export async function loadWorkspace(id: string): Promise<WorkspaceBody> {
|
||||
export async function changeWorkspace(id: string, path: string, method: 'POST' | 'PATCH' | 'PUT' | 'DELETE', value?: unknown): Promise<WorkspaceBody> {
|
||||
const { session } = await getSession(id);
|
||||
const body = await apiFetch(`/spaces/${id}/${path}`, {
|
||||
method, headers: { ...session.authHeaders(session.can('owner') ? 'owner' : 'write'), 'Content-Type': 'application/json' }, body: JSON.stringify(value),
|
||||
method, headers: { ...session.authHeaders(session.can('owner') ? 'owner' : session.can('write') ? 'write' : 'read'), 'Content-Type': 'application/json' }, body: JSON.stringify(value),
|
||||
}).then(r => r.json() as Promise<WorkspaceBody>);
|
||||
await publish(id, body);
|
||||
return body;
|
||||
@@ -59,6 +59,13 @@ export async function createWorkspaceGroup(id: string, value: WorkspaceNameReque
|
||||
await publish(id, body.workspace);
|
||||
return body.group.id;
|
||||
}
|
||||
export async function createWorkspaceRole(id: string, value: WorkspaceRoleRequest): Promise<string> {
|
||||
const { session } = await getSession(id);
|
||||
const body = await apiFetch(`/spaces/${id}/roles`, { method: 'POST', headers: { ...session.authHeaders('owner'), 'Content-Type': 'application/json' }, body: JSON.stringify(value) })
|
||||
.then(r => r.json() as Promise<WorkspaceRoleCreatedBody>);
|
||||
await publish(id, body.workspace);
|
||||
return body.role.id;
|
||||
}
|
||||
export async function registerItemLocation(id: string, item: Pick<FileItem, 'id' | 'threadId' | 'folderId'>): Promise<void> {
|
||||
if (!item.threadId) return;
|
||||
await changeWorkspace(id, 'locations', 'PUT', { itemIds: [item.id], threadId: item.threadId, folderId: item.folderId } satisfies WorkspaceMoveRequest);
|
||||
|
||||
Reference in new issue
Block a user