浅念
93e7046332
Parallelize the enqueueForUpload collect loop (preparer's internal bounded semaphore from the previous task keeps IO in check). Also two test hardenings from that task's code review: pin that failed and thrown prepare operations release their concurrency slot (cap=1, subsequent prepare still completes), and pin the maxConcurrentPrepares <1 -> 1 clamp (peak 1, no deadlock). Co-Authored-By: Claude Code <noreply@anthropic.com>
360 lines
13 KiB
Dart
360 lines
13 KiB
Dart
/// 上传准备器（UploadPreparer）测试（P3 计划 Task 6）。
|
||
///
|
||
/// 真实临时目录驱动（Directory.systemTemp）——复制/stat/魔数嗅探均为
|
||
/// 真实文件系统行为；任务库用内存 drift。
|
||
library;
|
||
|
||
import 'dart:io';
|
||
|
||
import 'package:flutter_test/flutter_test.dart';
|
||
import 'package:photo_live/features/uploads/preparer.dart';
|
||
import 'package:photo_live/features/uploads/task_state.dart';
|
||
import 'package:photo_live/features/uploads/task_store.dart';
|
||
|
||
/// JPEG/PNG 魔数最小字节。
|
||
final List<int> _jpgBytes = [
|
||
0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, // JFIF 头片段
|
||
0x4A, 0x46, 0x49, 0x46, 0x00,
|
||
];
|
||
final List<int> _pngBytes = [
|
||
0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, // PNG 签名
|
||
0x00, 0x00, 0x00, 0x0D,
|
||
];
|
||
|
||
void main() {
|
||
late TaskStore store;
|
||
late Directory tmpRoot;
|
||
late Directory sourceDir;
|
||
late Directory managedBase;
|
||
|
||
setUp(() async {
|
||
store = TaskStore.memory();
|
||
tmpRoot = await Directory.systemTemp.createTemp('preparer-test-');
|
||
sourceDir = Directory('${tmpRoot.path}/src')
|
||
..createSync(recursive: true);
|
||
managedBase = Directory('${tmpRoot.path}/managed')
|
||
..createSync(recursive: true);
|
||
});
|
||
|
||
tearDown(() async {
|
||
await store.close();
|
||
await tmpRoot.delete(recursive: true);
|
||
});
|
||
|
||
File sourceFile(String name, List<int> bytes) {
|
||
final f = File('${sourceDir.path}/$name');
|
||
f.writeAsBytesSync(bytes);
|
||
return f;
|
||
}
|
||
|
||
UploadPreparer preparer({
|
||
int maxBytes = UploadPreparer.kMaxUploadBytes,
|
||
int maxConcurrentPrepares = kMaxConcurrentPrepares,
|
||
}) =>
|
||
UploadPreparer(
|
||
store,
|
||
baseDirProvider: () async => managedBase,
|
||
maxBytes: maxBytes,
|
||
maxConcurrentPrepares: maxConcurrentPrepares,
|
||
);
|
||
|
||
group('prepare（快乐路径）', () {
|
||
test('jpg：受管副本存在、行 queued、魔数嗅探 contentType、bytes/filename 落库',
|
||
() async {
|
||
final src = sourceFile('IMG_0001.jpg', _jpgBytes);
|
||
|
||
final task = await preparer().prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
categoryId: 'cat-1',
|
||
sourcePath: src.path,
|
||
);
|
||
|
||
expect(task.state, UploadTaskState.queued);
|
||
expect(task.contentType, 'image/jpeg');
|
||
expect(task.bytes, _jpgBytes.length);
|
||
expect(task.filename, 'IMG_0001.jpg');
|
||
expect(task.categoryId, 'cat-1');
|
||
expect(task.lastError, isNull);
|
||
// 受管路径：{base}/uploads/{accountId}/{albumId}/{taskId}.jpg
|
||
expect(
|
||
task.sourceFilePath,
|
||
matches(RegExp(
|
||
r'[/\\]uploads[/\\]alice[/\\]alb-1[/\\][0-9a-f]{32}\.jpg$',
|
||
)),
|
||
);
|
||
expect(File(task.sourceFilePath).existsSync(), isTrue);
|
||
expect(task.id, task.sourceFilePath.split(RegExp(r'[/\\]')).last.substring(0, 32));
|
||
expect(File(task.sourceFilePath).readAsBytesSync(), _jpgBytes);
|
||
});
|
||
|
||
test('png 魔数覆盖错误扩展名（.png 伪装检查的反向：jpg 扩展实际 png）',
|
||
() async {
|
||
final src = sourceFile('photo.jpg', _pngBytes);
|
||
final task = await preparer().prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: src.path,
|
||
);
|
||
expect(task.contentType, 'image/png');
|
||
// 扩展名沿用源（副本文件名带 .jpg）——服务端以 contentType 为准。
|
||
expect(task.sourceFilePath, endsWith('.jpg'));
|
||
});
|
||
|
||
test('heic 无魔数走扩展名映射；filename 可覆盖缺省', () async {
|
||
final src = sourceFile('IMG_0002.heic', List.filled(64, 0x00));
|
||
final task = await preparer().prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: src.path,
|
||
filename: '连拍_0002.heic',
|
||
);
|
||
expect(task.contentType, 'image/heic');
|
||
expect(task.filename, '连拍_0002.heic');
|
||
});
|
||
|
||
test('入队回调触发（引擎唤醒挂接点）', () async {
|
||
final src = sourceFile('a.jpg', _jpgBytes);
|
||
var enqueued = 0;
|
||
final p = UploadPreparer(
|
||
store,
|
||
baseDirProvider: () async => managedBase,
|
||
onEnqueued: () => enqueued++,
|
||
);
|
||
await p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path);
|
||
expect(enqueued, 1);
|
||
});
|
||
});
|
||
|
||
group('prepare（预检失败 → preparationFailed）', () {
|
||
test('未知类型（.txt 且无魔数）：unsupportedType，且不产副本', () async {
|
||
final src = sourceFile('notes.txt', List.filled(16, 0x41));
|
||
final task = await preparer().prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: src.path,
|
||
);
|
||
expect(task.state, UploadTaskState.preparationFailed);
|
||
expect(task.lastError, PrepErrorCodes.unsupportedType);
|
||
expect(File(task.sourceFilePath).existsSync(), isFalse);
|
||
});
|
||
|
||
test('源文件不存在：sourceUnreadable', () async {
|
||
final task = await preparer().prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: '${sourceDir.path}/missing.jpg',
|
||
);
|
||
expect(task.state, UploadTaskState.preparationFailed);
|
||
expect(task.lastError, PrepErrorCodes.sourceUnreadable);
|
||
});
|
||
|
||
test('超本地尺寸上限：tooLarge（上限可注入小值）', () async {
|
||
final src = sourceFile('big.jpg', _jpgBytes + List.filled(200, 0x00));
|
||
final task = await preparer(maxBytes: 100).prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: src.path,
|
||
);
|
||
expect(task.state, UploadTaskState.preparationFailed);
|
||
expect(task.lastError, PrepErrorCodes.tooLarge);
|
||
});
|
||
|
||
test('复制阶段失败（源在预检后、复制前被删）：copyFailed', () async {
|
||
final src = sourceFile('race.jpg', _jpgBytes);
|
||
final p = preparer();
|
||
// 竞态模拟：stat/嗅探通过后、copy 前删源——通过先跑一次拿路径再
|
||
// 直接构造不可复制场景：用目录当源（copy 抛错）。
|
||
final dirAsSource = Directory('${sourceDir.path}/adir')
|
||
..createSync();
|
||
final task = await p.prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: dirAsSource.path,
|
||
);
|
||
// 目录 length() 在部分平台返回 stat——Windows 下非 0 会走到 copy 失败；
|
||
// Linux 下目录 stat 为 4096 同样走 copy 失败；若返回 0 则是
|
||
// sourceUnreadable——两者都是 preparationFailed，断言状态为主。
|
||
expect(task.state, UploadTaskState.preparationFailed);
|
||
expect(
|
||
task.lastError,
|
||
anyOf(PrepErrorCodes.copyFailed, PrepErrorCodes.sourceUnreadable),
|
||
);
|
||
expect(src.existsSync(), isTrue, reason: '不相干文件不受影响');
|
||
});
|
||
});
|
||
|
||
group('retry（preparationFailed → queued）', () {
|
||
test('提供新源后重试成功并清除错误', () async {
|
||
final src = sourceFile('r1.jpg', _jpgBytes);
|
||
final p = preparer();
|
||
final failed = await p.prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: '${sourceDir.path}/missing.jpg',
|
||
);
|
||
expect(failed.state, UploadTaskState.preparationFailed);
|
||
|
||
final retried = await p.retry(
|
||
accountId: 'alice',
|
||
taskId: failed.id,
|
||
sourcePath: src.path,
|
||
);
|
||
expect(retried.state, UploadTaskState.queued);
|
||
expect(retried.lastError, isNull);
|
||
expect(File(retried.sourceFilePath).existsSync(), isTrue);
|
||
});
|
||
|
||
test('非 preparationFailed 状态拒绝重试', () async {
|
||
final src = sourceFile('r2.jpg', _jpgBytes);
|
||
final p = preparer();
|
||
final queued = await p.prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: src.path,
|
||
);
|
||
await expectLater(
|
||
p.retry(accountId: 'alice', taskId: queued.id, sourcePath: src.path),
|
||
throwsA(isA<StateError>()),
|
||
);
|
||
});
|
||
});
|
||
|
||
group('有界并发执行（连拍防 IO 风暴的上限钳位）', () {
|
||
/// 构造带并发探针的准备器：baseDirProvider 在 _prepare 内被 await，
|
||
/// 其重叠度是「在途 prepare 数」的可观测下界（且受信号量上限约束）。
|
||
(UploadPreparer, int Function()) probePreparer(int limit) {
|
||
var active = 0;
|
||
var peak = 0;
|
||
final p = UploadPreparer(
|
||
store,
|
||
baseDirProvider: () async {
|
||
active++;
|
||
if (active > peak) peak = active;
|
||
await Future<void>.delayed(const Duration(milliseconds: 30));
|
||
active--;
|
||
return managedBase;
|
||
},
|
||
maxConcurrentPrepares: limit,
|
||
);
|
||
return (p, () => peak);
|
||
}
|
||
|
||
test('并发 prepare 峰值 = maxConcurrentPrepares（≤上限且真并行），9 任务全部 queued', () async {
|
||
final (p, peakOf) = probePreparer(3);
|
||
final srcs = [
|
||
for (var i = 0; i < 9; i++)
|
||
sourceFile('par-$i.jpg', _jpgBytes + [i]),
|
||
];
|
||
final tasks = await Future.wait([
|
||
for (final src in srcs)
|
||
p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path),
|
||
]);
|
||
// 全部入队、ID/幂等键互不相同、副本都在。
|
||
expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued));
|
||
expect(tasks.map((t) => t.id).toSet().length, 9);
|
||
expect(tasks.map((t) => t.clientTaskId).toSet().length, 9);
|
||
for (final t in tasks) {
|
||
expect(File(t.sourceFilePath).existsSync(), isTrue);
|
||
}
|
||
// 峰值恰为 3：≤3 证明上限钳位（防连拍 IO 风暴初衷保留）；
|
||
// 达到 3 证明未退化为串行。
|
||
expect(peakOf(), 3);
|
||
});
|
||
|
||
test('maxConcurrentPrepares=1 等价旧串行行为：峰值 1、任务仍全部成功', () async {
|
||
final (p, peakOf) = probePreparer(1);
|
||
final srcs = [
|
||
for (var i = 0; i < 4; i++)
|
||
sourceFile('ser-$i.jpg', _jpgBytes + [i]),
|
||
];
|
||
final tasks = await Future.wait([
|
||
for (final src in srcs)
|
||
p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path),
|
||
]);
|
||
expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued));
|
||
expect(peakOf(), 1, reason: '并发 1 时任意时刻在途 prepare 至多 1（串行）');
|
||
});
|
||
|
||
test('maxConcurrentPrepares<1 钳位为 1（0 不死锁、串行等价）', () async {
|
||
final (p, peakOf) = probePreparer(0);
|
||
final srcs = [
|
||
for (var i = 0; i < 3; i++)
|
||
sourceFile('clamp-$i.jpg', _jpgBytes + [i]),
|
||
];
|
||
final tasks = await Future.wait([
|
||
for (final src in srcs)
|
||
p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path),
|
||
]);
|
||
expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued));
|
||
expect(peakOf(), 1, reason: '运行时 <1 按 1 处理：峰值 1（不死锁、不并行）');
|
||
});
|
||
|
||
test('失败与抛错操作都释放并发名额（cap=1 后续 prepare 不被卡死）', () async {
|
||
final p = preparer(maxConcurrentPrepares: 1);
|
||
final src = sourceFile('after-error.jpg', _jpgBytes);
|
||
|
||
// 业务失败：源不存在 → preparationFailed（正常返回、不抛）。
|
||
final failed = await p.prepare(
|
||
accountId: 'alice',
|
||
albumId: 'alb-1',
|
||
sourcePath: '${sourceDir.path}/missing.jpg',
|
||
);
|
||
expect(failed.state, UploadTaskState.preparationFailed);
|
||
|
||
// 真错误：retry 不存在的任务 → StateError 穿过 _runBounded——
|
||
// 钉死 whenComplete 语义（若改成 then，此处名额泄漏）。
|
||
await expectLater(
|
||
p.retry(accountId: 'alice', taskId: 'no-such-task', sourcePath: src.path),
|
||
throwsA(isA<StateError>()),
|
||
);
|
||
|
||
// 名额已释放：后续 prepare 正常完成（泄漏则卡死，超时快速失败）。
|
||
final task = await p
|
||
.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path)
|
||
.timeout(const Duration(seconds: 5));
|
||
expect(task.state, UploadTaskState.queued);
|
||
});
|
||
});
|
||
|
||
group('串行回归（maxConcurrentPrepares=1 等价旧全局串行队列）', () {
|
||
test('并发=1：prepare 按提交顺序串行完成，各自独立成行', () async {
|
||
final p = preparer(maxConcurrentPrepares: 1);
|
||
final srcs = [
|
||
for (var i = 0; i < 3; i++)
|
||
sourceFile('burst-$i.jpg', _jpgBytes + [i]),
|
||
];
|
||
final tasks = await Future.wait([
|
||
for (final src in srcs)
|
||
p.prepare(accountId: 'alice', albumId: 'alb-1', sourcePath: src.path),
|
||
]);
|
||
// 全部入队、ID/幂等键互不相同、副本都在。
|
||
expect(tasks.map((t) => t.state), everyElement(UploadTaskState.queued));
|
||
expect(tasks.map((t) => t.id).toSet().length, 3);
|
||
expect(tasks.map((t) => t.clientTaskId).toSet().length, 3);
|
||
for (final t in tasks) {
|
||
expect(File(t.sourceFilePath).existsSync(), isTrue);
|
||
}
|
||
// createdAt 非降（并发=1 串行的间接证据：无交叉写库）。
|
||
final stamps = tasks.map((t) => t.createdAt).toList();
|
||
for (var i = 1; i < stamps.length; i++) {
|
||
expect(!stamps[i].isBefore(stamps[i - 1]), isTrue);
|
||
}
|
||
});
|
||
});
|
||
|
||
test('账号/相册隔离的受管目录（不同账号不同目录树）', () async {
|
||
final s1 = sourceFile('u1.jpg', _jpgBytes);
|
||
final s2 = sourceFile('u2.jpg', _jpgBytes);
|
||
final p = preparer();
|
||
final t1 = await p.prepare(
|
||
accountId: 'alice', albumId: 'alb-1', sourcePath: s1.path);
|
||
final t2 = await p.prepare(
|
||
accountId: 'bob', albumId: 'alb-2', sourcePath: s2.path);
|
||
expect(t1.sourceFilePath, contains('alice'));
|
||
expect(t1.sourceFilePath, contains('alb-1'));
|
||
expect(t2.sourceFilePath, contains('bob'));
|
||
expect(t2.sourceFilePath, contains('alb-2'));
|
||
});
|
||
}
|