fix: recover published archives and serialize delete with compression scheduling

This commit is contained in:
root
2026-10-07 23:58:26 +08:00
parent cbd8c3a3b9
commit 31ceb4d5e5
4 changed files with 547 additions and 114 deletions
@@ -54,6 +54,28 @@ public class DownloadCheckService {
private record VerifiedArchive(long size, long verifiedAt) {} private record VerifiedArchive(long size, long verifiedAt) {}
/**
* 状态调度使用的共享监视器就是本实例对象本身。
*
* <p>删除命令与下载扫描争用同一份任务队列与磁盘目录,两者必须真正互斥:<b>只检查
* 「是否正在压缩」的状态判定不足以构成互斥</b>。扫描线程可能在「标 COMPRESSING」与
* 「进压缩队列」之间被抢占,而删除此刻看到的队列与 active 集合都还没有该 gid,于是删除
* 先把画廊删掉,扫描随后把删除前选中的旧对象排进压缩队列,压缩线程又把归档重新写出来——
* 已删除的画廊被「复活」并上报给主站。
*
* <p>因此 {@link #downloadCheck()}、{@link #addToQueue}、{@link #removeGid}、
* {@link #isCompressing} 的状态调度,以及 {@link #compress()} 对队列的「领取」,全部在
* 本监视器下完成;{@code ReliableTaskService.executeDelete} 也用同一监视器把
* 「查忙 → 删磁盘 → 清队列」包成一个原子段。
*
* <p>锁定顺序(唯一方向,避免死锁):调用方锁 → 本实例监视器 → {@code compress_queue}。
* 真正的压缩 I/O 不在监视器内执行,删除的互斥由 {@link #activeCompressions} 标记保证。
*
* <p>共享监视器就是 {@code this}({@code DownloadCheckService} 实例)本身,因此
* {@code ReliableTaskService.executeDelete} 用 {@code synchronized (downloadCheckService)}
* 即可与本类的状态调度取到同一把锁。
*/
/** /**
* 完成通知的专用线程。 * 完成通知的专用线程。
* *
@@ -90,6 +112,7 @@ public class DownloadCheckService {
} }
public boolean downloadCheck(){ public boolean downloadCheck(){
synchronized (this) {
if(queue.isEmpty()) if(queue.isEmpty())
return false; return false;
log.info("下载检查:{}", Arrays.toString(queue.values().toArray())); log.info("下载检查:{}", Arrays.toString(queue.values().toArray()));
@@ -142,6 +165,11 @@ public class DownloadCheckService {
for(GalleryTask galleryTask: queue.values()) for(GalleryTask galleryTask: queue.values())
if (galleryTask.is_download_complete() if (galleryTask.is_download_complete()
&& System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) { && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) {
// 必须以队列里的当前对象为准:本轮遍历的引用可能已被删除流程 removeGid
// 移除(删除已把磁盘清掉),把旧对象排进压缩队列会让已删除的画廊复活。
GalleryTask current = queue.get(galleryTask.getGid());
if (current != galleryTask)
continue;
galleryTask.setStatus(GalleryTask.COMPRESSING); galleryTask.setStatus(GalleryTask.COMPRESSING);
synchronized (compress_queue) { synchronized (compress_queue) {
compress_queue.add(galleryTask); compress_queue.add(galleryTask);
@@ -150,13 +178,16 @@ public class DownloadCheckService {
return result; return result;
} }
}
/** /**
* 压缩线程:将压缩队列的任务复制一份,进行转换 * 压缩线程:将压缩队列的任务复制一份,进行转换
*/ */
public void compress() { public void compress() {
ArrayList<GalleryTask> galleryTasks; ArrayList<GalleryTask> galleryTasks;
synchronized (compress_queue) { // 领取与扫描入队共用同一监视器:原子地把队列交给压缩线程并标记 active,
// 使删除再也看不到「已标 COMPRESSING 但尚未 active」的空档。
synchronized (this) {
if (compress_queue.isEmpty()) if (compress_queue.isEmpty())
return; return;
galleryTasks = new ArrayList<>(compress_queue); galleryTasks = new ArrayList<>(compress_queue);
@@ -210,15 +241,15 @@ public class DownloadCheckService {
* <p>删除与压缩是互斥的:命中时调用方应回可重试的「忙」,而不是继续动磁盘。 * <p>删除与压缩是互斥的:命中时调用方应回可重试的「忙」,而不是继续动磁盘。
*/ */
public boolean isCompressing(int gid){ public boolean isCompressing(int gid){
synchronized (this) {
if (activeCompressions.contains(gid)) if (activeCompressions.contains(gid))
return true; return true;
synchronized (compress_queue) {
for (GalleryTask queued : compress_queue) for (GalleryTask queued : compress_queue)
if (queued.getGid() == gid) if (queued.getGid() == gid)
return true; return true;
}
return false; return false;
} }
}
/** /**
* 从压缩队列与任务队列中摘除该 gid。 * 从压缩队列与任务队列中摘除该 gid。
@@ -226,9 +257,9 @@ public class DownloadCheckService {
* <p>用于删除成功后清理本节点内部状态,避免已删除任务继续被扫描或压缩。 * <p>用于删除成功后清理本节点内部状态,避免已删除任务继续被扫描或压缩。
*/ */
public void removeGid(int gid){ public void removeGid(int gid){
synchronized (this) {
queue.remove(gid); queue.remove(gid);
retryAfter.remove(gid); retryAfter.remove(gid);
synchronized (compress_queue) {
compress_queue.removeIf(task -> task.getGid() == gid); compress_queue.removeIf(task -> task.getGid() == gid);
} }
} }
@@ -275,8 +306,31 @@ public class DownloadCheckService {
} }
} }
/**
* 受理或刷新一个任务,并按 gid 用磁盘事实重建它的状态。
*
* <p>与扫描、删除共用同一监视器:这里既要读磁盘目录,又要写任务队列,如果与删除交叉执行,
* 会出现「删除已清空磁盘,入队仍按删除前的目录刷新出下载完成」的矛盾状态。
*
* <p>重建顺序是「归档优先」:gid 稳定而目录名不稳定(下载器会追加后缀),因此先按 gid
* 找到校验有效的 ZIP 判为压缩完成,再回退到下载目录判断下载中/下载完成。只有归档有效才算
* 终态,损坏或空归档一律不得被当成完成。
*/
public GalleryTask addToQueue(GalleryTask galleryTask){ public GalleryTask addToQueue(GalleryTask galleryTask){
return addToQueue(galleryTask, true);
}
/**
* 受理或刷新一个任务的内部实现。
*
* <p>{@code notifyOnAlreadyStored=false} 供启动恢复使用:恢复时对每个任务做一次磁盘事实
* 重建,命中的是「重启前就已经存在的归档」,而不是主站刚刚新下的任务,不该为它再发一遍
* 「添加时已下载完成」通知。
*/
GalleryTask addToQueue(GalleryTask galleryTask, boolean notifyOnAlreadyStored){
synchronized (this) {
GalleryTask active = queue.get(galleryTask.getGid()); GalleryTask active = queue.get(galleryTask.getGid());
// 只有真正「正在压缩」才短路:压缩线程正持有该对象写归档,替换它会丢状态。
if (active != null && active.is_compressing()) if (active != null && active.is_compressing())
return active; return active;
retryAfter.remove(galleryTask.getGid()); retryAfter.remove(galleryTask.getGid());
@@ -285,11 +339,16 @@ public class DownloadCheckService {
// by gid first, because the gid is stable while the directory name is not. // by gid first, because the gid is stable while the directory name is not.
File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid()); File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid());
if(storedDirectory != null){ if(storedDirectory != null){
// 已有有效归档即为终态;保留队列里那个仍然活跃的对象,避免同一 gid 出现两个实例。
GalleryTask target = active == null ? galleryTask : active;
if (active != null)
queue.remove(galleryTask.getGid()); queue.remove(galleryTask.getGid());
galleryTask.setName(storedDirectory.getName()); target.setName(storedDirectory.getName());
galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE); target.setPath(storedDirectory.getPath());
notifyAlreadyStored(galleryTask.getName()); target.setStatus(GalleryTask.COMPRESS_COMPLETE);
return galleryTask; if (notifyOnAlreadyStored)
notifyAlreadyStored(target.getName());
return target;
} }
// If the task is still downloading, retain the actual directory name so // If the task is still downloading, retain the actual directory name so
@@ -299,8 +358,8 @@ public class DownloadCheckService {
GalleryTask queuedTask = queue.get(galleryTask.getGid()); GalleryTask queuedTask = queue.get(galleryTask.getGid());
if(queuedTask == null){ if(queuedTask == null){
refreshDownloadingTask(galleryTask, downloadingDirectory); refreshDownloadingTask(galleryTask, downloadingDirectory);
queue.put(galleryTask.getGid(), galleryTask); GalleryTask existing = queue.putIfAbsent(galleryTask.getGid(), galleryTask);
return galleryTask; return existing == null ? galleryTask : existing;
} }
if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete()) if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete())
refreshDownloadingTask(queuedTask, downloadingDirectory); refreshDownloadingTask(queuedTask, downloadingDirectory);
@@ -310,6 +369,7 @@ public class DownloadCheckService {
GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask); GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask);
return queuedTask == null ? galleryTask : queuedTask; return queuedTask == null ? galleryTask : queuedTask;
} }
}
private void refreshDownloadingTask(GalleryTask galleryTask, File downloadingDirectory){ private void refreshDownloadingTask(GalleryTask galleryTask, File downloadingDirectory){
galleryTask.setName(downloadingDirectory.getName()); galleryTask.setName(downloadingDirectory.getName());
@@ -67,9 +67,17 @@ public class ReliableTaskService {
/** /**
* 启动恢复。 * 启动恢复。
* *
* <p>先读回日志,再把它合并进内存队列;被中断的压缩一律退回「下载完成」,让扫描与压缩 * <p>先读回日志,再按磁盘事实重建每个任务的状态,最后并入内存队列。
* 重新评估真实文件,而不是停在一个永远等不到结果的中间态。真实文件状态由后续扫描重算, *
* 日志里的进度只作为扫描前的乐观初值。 * <p>这里必须用「有效 ZIP / 下载目录」的真实校验来重建,而不是把中断的压缩简单改成
* 「下载完成」了事:{@code compress()} 已经原子发布 ZIP 并删掉源目录之后、日志尚未落盘时
* 崩溃,重启后就是一个「日志说压缩中、磁盘上只剩归档、下载目录空无一物」的三方矛盾。
* 若只降级为下载完成,扫描会因下载目录为空而提前返回、stableop 重投又只回放回执,
* 任务就永久停在等待压缩。按 gid 命中有效归档即判为压缩完成(同时补回 name/path),
* 命中源目录则恢复下载中/下载完成与进度,两者都没有时保留任务、不捏造完成。
*
* <p>恢复只重建本地事实,绝不重新提交上游下载;operationId 与待 ACK 报告原样保留,
* 任务的 revision 过滤仍由 {@link #applyAck} 负责。
*/ */
public void restore() throws IOException { public void restore() throws IOException {
journal.load(); journal.load();
@@ -77,11 +85,19 @@ public class ReliableTaskService {
for (TaskJournal.JournalTask stored : journal.tasks().values()) { for (TaskJournal.JournalTask stored : journal.tasks().values()) {
GalleryTask task = stored.toGalleryTask(); GalleryTask task = stored.toGalleryTask();
if (task.getStatus() == GalleryTask.COMPRESSING) if (task.getStatus() == GalleryTask.COMPRESSING)
// 退回到可被扫描重排的状态;此处尚未入队,addToQueue 不会因「正在压缩」短路。
task.setStatus(GalleryTask.DOWNLOAD_COMPLETE); task.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
queue.putIfAbsent(task.getGid(), task); // 用磁盘事实重建:有效归档 → 压缩完成;源目录 → 下载中/完成(带进度);
// 都没有时保留任务原状。恢复期的命中不应再发「添加时已完成」通知。
GalleryTask restored = downloadCheckService.addToQueue(task, false);
queue.putIfAbsent(restored.getGid(), restored);
if (stored.operationId != null) if (stored.operationId != null)
operationIds.put(stored.gid, stored.operationId); operationIds.put(stored.gid, stored.operationId);
} }
// 把重建出的磁盘事实立即落盘:日志停在 COMPRESSING 时它必须被新的压缩完成状态与更高
// revision 取代。这既让后续报告不再带着过期的压缩中状态上报,也让崩溃前生成的旧快照
// ACK 在 removeTaskIfRevision 处因版本过期而被拒绝,清不掉刚恢复的新事实。
journal.syncTasks(List.copyOf(queue.values()), operationIds);
log.info("节点重启恢复: 任务={}, 待 ACK 报告={}", queue.size(), journal.pendingReports().size()); log.info("节点重启恢复: 任务={}, 待 ACK 报告={}", queue.size(), journal.pendingReports().size());
} }
} }
@@ -138,6 +154,11 @@ public class ReliableTaskService {
private byte executeDelete(TaskCommandMessage command) throws IOException { private byte executeDelete(TaskCommandMessage command) throws IOException {
int gid = command.getGid(); int gid = command.getGid();
// 删除的「查忙 → 删磁盘 → 清队列」必须与扫描的「选任务 → 标压缩中 → 排进压缩队列」
// 原子互斥。只查一次 isCompressing 会在「已标 COMPRESSING、尚未进压缩队列」的窗口里
// 判定为不忙:删除把画廊删掉,扫描随后仍把删除前选中的旧对象排进压缩队列,压缩线程
// 又把归档重新写出来,已删除的画廊被复活并上报。整段持同一监视器后该窗口不复存在。
synchronized (downloadCheckService) {
// 压缩中删除会破坏正在进行中的归档写入,先回可重试的「忙」,不写回执。 // 压缩中删除会破坏正在进行中的归档写入,先回可重试的「忙」,不写回执。
if (downloadCheckService.isCompressing(gid)) { if (downloadCheckService.isCompressing(gid)) {
log.info("任务正在压缩,删除请求延后重试: gid={}", gid); log.info("任务正在压缩,删除请求延后重试: gid={}", gid);
@@ -161,6 +182,7 @@ public class ReliableTaskService {
gid, directoryName, result)); gid, directoryName, result));
return result; return result;
} }
}
private byte deleteGallery(String name) { private byte deleteGallery(String name) {
File target = storagePath.resolve(name).toFile(); File target = storagePath.resolve(name).toFile();
@@ -0,0 +1,206 @@
package lion.Service;
import lion.Domain.GalleryTask;
import lion.Message.Main.TaskCommandMessage;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.*;
/**
* 下载扫描与删除的互斥(问题 B)。
*
* <p>缺陷:扫描线程在「把任务标成 COMPRESSING」与「排进压缩队列」之间可以被抢占;删除此刻看到
* active 集合与压缩队列里都没有该 gid,判定为「不忙」并把画廊删掉;扫描随后仍把删除前选中的旧
* 对象排进压缩队列,压缩线程又把归档写出来,已删除的画廊被复活并上报。
*
* <p>仅检查状态位({@code isCompressing})无法修复:窗口正是「状态已改、还没入队」的那段。必须让
* 「查忙 → 删磁盘 → 清队列」与「选任务 → 标状态 → 入队」落在同一把服务监视器下原子执行。
*
* <p>所有交错点用 CountDownLatch / 线程状态确定性控制,不用 sleep 概率等待。
*/
class DownloadCheckDeleteRaceTest {
/**
* 端到端交错:扫描已选中任务、正处在「标 COMPRESSING 到入队」的窗口内时,删除不能同时成功。
*
* <p>用可控任务在 {@code setStatus(COMPRESSING)} 处暂停——那正是缺陷窗口的位置,且该调用在
* 缺陷版本与修复版本中同样存在。暂停时删除线程必须阻塞在服务监视器上,直到扫描完成入队;
* 随后删除看到「正在压缩」,回可重试的忙,画廊与归档都不会被破坏。
*
* <p>缺陷版本里删除不被监视器阻挡,会在此窗口内直接完成,从而与「扫描已入队压缩」同时成功——
* 该断言正是对缺陷的判别。
*/
@Test
void deleteCannotCompleteWhileScanIsBetweenCompressingMarkAndEnqueue(@TempDir Path root) throws Exception {
Path storage = Files.createDirectories(root.resolve("gallery"));
Path downloads = Files.createDirectories(root.resolve("downloads"));
Path source = Files.createDirectories(downloads.resolve("sample [90003]"));
Files.writeString(source.resolve("1.jpg"), "a");
Files.writeString(source.resolve("galleryinfo.txt"), "meta");
// 删除目标:gid 命中的画廊目录(压缩归档还未生成)。
Path gallery = Files.createDirectories(storage.resolve("sample [90003]"));
Map<Integer, GalleryTask> queue = new ConcurrentHashMap<>();
DownloadCheckService service = new DownloadCheckService(queue, false);
service.storagePath = storage.toString();
service.downloadPath = downloads.toString();
ReliableTaskService reliable = new ReliableTaskService(
root.resolve("journal"), queue, service, storage);
PausingTask task = new PausingTask();
task.setGid(90003);
task.setName("sample [90003]");
task.setPath(source.toString());
task.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
queue.put(90003, task);
task.monitor = service;
task.arm();
Thread scan = new Thread(service::downloadCheck, "scan");
scan.start();
assertTrue(task.entered.await(5, TimeUnit.SECONDS), "扫描应到达压缩入队前的标记点");
assertTrue(task.heldMonitor, "标记点必须位于服务监视器内");
assertTrue(Files.exists(gallery), "此时画廊目录应仍然存在");
byte[] deleteResult = new byte[1];
Thread delete = new Thread(() -> {
try {
deleteResult[0] = reliable.handleCommand(
deleteCommand("op-race-e2e", 90003, "sample [90003]"));
} catch (Exception e) {
throw new RuntimeException(e);
}
}, "delete");
delete.start();
awaitBlocked(delete);
assertTrue(Files.exists(gallery), "扫描持锁期间删除不得移除画廊目录");
task.resume.countDown();
scan.join(5000);
delete.join(5000);
assertFalse(scan.isAlive());
assertFalse(delete.isAlive());
assertEquals(NodeErrorCode.BUSY_COMPRESSING, deleteResult[0],
"扫描已进入压缩调度时删除必须回可重试的忙,两者不能同时成功");
assertTrue(service.isCompressing(90003), "任务应仍处于压缩调度中");
service.compress();
assertTrue(DownloadCheckService.isValidArchive(
storage.resolve("sample [90003]/sample [90003].zip").toFile()),
"压缩应发布有效归档");
assertEquals(GalleryTask.COMPRESS_COMPLETE, task.getStatus());
}
/**
* 删除已经完成时,随后的扫描不得把已删除任务重新压缩,画廊也不得被复活。
*
* <p>覆盖「DELETE 先落地」的时序:入队按 gid 校验队列中的当前对象,避免把删除前遍历到的
* 旧对象重新排进压缩队列。
*/
@Test
void scanAfterCompletedDeleteDoesNotCompressOrResurrectGallery(@TempDir Path root) throws Exception {
Fixture fixture = fixture(root, 90004, "sample [90004]");
Files.createDirectories(root.resolve("gallery/sample [90004]"));
assertEquals(NodeErrorCode.SUCCESS,
fixture.reliable.handleCommand(deleteCommand("op-e2e-del", 90004, "sample [90004]")));
assertTrue(fixture.queue.isEmpty(), "删除成功后任务应从队列移除");
fixture.service.downloadCheck();
assertFalse(fixture.service.isCompressing(90004), "已删除任务不得再被排进压缩");
fixture.service.compress();
assertFalse(Files.exists(root.resolve("gallery/sample [90004]")),
"已删除的画廊不得被压缩复活");
}
// ---- 测试装配 ----
private record Fixture(Map<Integer, GalleryTask> queue, DownloadCheckService service,
ReliableTaskService reliable, GalleryTask task, int gid) {}
private static Fixture fixture(Path root, int gid, String name) throws Exception {
Map<Integer, GalleryTask> queue = new ConcurrentHashMap<>();
Path storage = Files.createDirectories(root.resolve("gallery"));
Path downloads = Files.createDirectories(root.resolve("downloads"));
Path source = Files.createDirectories(downloads.resolve(name));
Files.writeString(source.resolve("1.jpg"), "a");
Files.writeString(source.resolve("galleryinfo.txt"), "meta");
DownloadCheckService service = new DownloadCheckService(queue, false);
service.storagePath = storage.toString();
service.downloadPath = downloads.toString();
ReliableTaskService reliable = new ReliableTaskService(
root.resolve("journal"), queue, service, storage);
GalleryTask task = new GalleryTask();
task.setGid(gid);
task.setName(name);
task.setPath(source.toString());
task.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
queue.put(gid, task);
return new Fixture(queue, service, reliable, task, gid);
}
private static TaskCommandMessage deleteCommand(String operationId, int gid, String name) {
TaskCommandMessage command = new TaskCommandMessage();
command.setOperationId(operationId);
command.setAction(TaskCommandMessage.ACTION_DELETE);
command.setGid(gid);
command.setName(name);
return command;
}
/** 有界等待线程真正阻塞在监视器上;超时即视为「未按预期互斥」而失败。 */
private static void awaitBlocked(Thread thread) {
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
while (thread.getState() != Thread.State.BLOCKED) {
if (System.nanoTime() > deadline)
fail("线程未在预期时间内阻塞在服务监视器上,当前状态=" + thread.getState());
Thread.onSpinWait();
}
}
/**
* 可在「标记 COMPRESSING」处暂停的任务。
*
* <p>该行存在于缺陷版本与修复版本中,正是「状态已改、尚未入队」的缺陷窗口。
*/
static final class PausingTask extends GalleryTask {
final CountDownLatch entered = new CountDownLatch(1);
final CountDownLatch resume = new CountDownLatch(1);
volatile boolean armed;
volatile boolean heldMonitor;
void arm() {
armed = true;
}
@Override
public void setStatus(byte status) {
if (armed && status == COMPRESSING) {
armed = false;
heldMonitor = Thread.holdsLock(this.monitor);
entered.countDown();
try {
if (!resume.await(5, TimeUnit.SECONDS))
throw new IllegalStateException("测试未释放扫描暂停点");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
super.setStatus(status);
}
/** 服务实例;由测试设置,用于断言暂停点确实位于服务监视器内。 */
volatile Object monitor;
}
}
@@ -152,6 +152,100 @@ class ReliableTaskServiceTest {
assertNotNull(restarted.service.journal().receipt("op-r"), "命令回执必须跨重启保留"); assertNotNull(restarted.service.journal().receipt("op-r"), "命令回执必须跨重启保留");
} }
/**
* 压缩已原子发布 ZIP 并删除源目录、但 COMPRESS_COMPLETE 尚未同步进 journal 时崩溃。
*
* <p>这是问题 A 的核心回归:日志停在 COMPRESSING,磁盘上却已有有效归档、下载目录空空如也。
* 若重启只把 COMPRESSING 降级成 DOWNLOAD_COMPLETE,扫描会因空下载目录提前返回、重投又只
* 回放回执,任务永久等不到压缩。恢复必须按 gid 校验磁盘上的有效 ZIP 并重建为压缩完成。
*/
@Test
void restartRecoversPublishedArchiveAfterCrashBeforeJournalSync(@TempDir Path root) throws Exception {
Harness first = new Harness(root);
first.service.handleCommand(download("op-archive", 80060, "sample [80060]"));
// 模拟状态已推进到压缩中并落盘,随后压缩成功发布归档,但日志未再同步。
first.queue.get(80060).setStatus(GalleryTask.COMPRESSING);
first.service.refreshJournalFromQueue();
TaskJournal.ReportSnapshot snapshot = first.service.buildReport().orElseThrow();
Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80060]"));
writeValidZip(archiveDirectory.resolve("sample [80060].zip"));
// 源目录已被 compress() 删除,下载目录为空。
assertFalse(Files.exists(first.downloadsPath.resolve("sample [80060]")));
Harness restarted = new Harness(root, true);
GalleryTask restored = restarted.queue.get(80060);
assertNotNull(restored, "重启必须保留任务");
assertEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(),
"磁盘上已有有效归档时必须恢复为压缩完成,而不是停在等待压缩");
assertEquals("sample [80060]", restored.getName(), "恢复必须重建目录名");
assertEquals(archiveDirectory.toString(), restored.getPath(), "恢复必须重建归档路径");
assertTrue(restarted.service.pendingReports().stream()
.anyMatch(report -> report.reportId.equals(snapshot.reportId)),
"恢复必须保留待 ACK 报告");
assertEquals("op-archive", restarted.service.operationIds().get(80060),
"恢复必须保留 operationId,不得重新提交上游");
}
/** 没有有效归档、也没有源目录时:保留任务,但绝不捏造压缩完成。 */
@Test
void restartWithoutArchiveOrSourceKeepsTaskWithoutFabricatingCompletion(@TempDir Path root) throws Exception {
Harness first = new Harness(root);
first.service.handleCommand(download("op-empty", 80061, "sample [80061]"));
first.queue.get(80061).setStatus(GalleryTask.COMPRESSING);
first.service.refreshJournalFromQueue();
Harness restarted = new Harness(root, true);
GalleryTask restored = restarted.queue.get(80061);
assertNotNull(restored, "无归档且无源时也必须保留任务");
assertNotEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(),
"没有有效归档时不得捏造压缩完成");
assertNull(restored.getPath(), "没有真实源目录时不得编造路径");
}
/** 压缩中途源目录仍在:恢复为可继续压缩的下载完成态,并带回 path 与进度。 */
@Test
void restartWithSourcePresentRebuildsDownloadCompleteWithPathAndProgress(@TempDir Path root) throws Exception {
Harness first = new Harness(root);
first.service.handleCommand(download("op-source", 80062, "sample [80062]"));
first.queue.get(80062).setStatus(GalleryTask.COMPRESSING);
first.service.refreshJournalFromQueue();
Path source = Files.createDirectories(first.downloadsPath.resolve("sample [80062]"));
Files.writeString(source.resolve("1.jpg"), "a");
Files.writeString(source.resolve("2.jpg"), "b");
Files.writeString(source.resolve("galleryinfo.txt"), "meta");
Harness restarted = new Harness(root, true);
GalleryTask restored = restarted.queue.get(80062);
assertEquals(GalleryTask.DOWNLOAD_COMPLETE, restored.getStatus(),
"源目录仍在时应恢复为可重试的下载完成态");
assertEquals(source.toString(), restored.getPath(), "恢复必须重建源目录路径");
assertEquals(2, restored.getProceeding(), "恢复必须重建已下载页数");
}
/** 同名位置上的 ZIP 已损坏:不得被判为压缩完成,任务必须保留待重试。 */
@Test
void restartWithCorruptArchiveDoesNotComplete(@TempDir Path root) throws Exception {
Harness first = new Harness(root);
first.service.handleCommand(download("op-corrupt", 80063, "sample [80063]"));
first.queue.get(80063).setStatus(GalleryTask.COMPRESSING);
first.service.refreshJournalFromQueue();
Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80063]"));
Files.writeString(archiveDirectory.resolve("sample [80063].zip"), "partial zip");
Harness restarted = new Harness(root, true);
GalleryTask restored = restarted.queue.get(80063);
assertNotNull(restored, "损坏归档不得让任务消失");
assertNotEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(),
"损坏或空归档绝不能判为压缩完成");
}
/** 命令缺 operationId 或 action 非法时回不可重试的无效命令,且不产生回执。 */ /** 命令缺 operationId 或 action 非法时回不可重试的无效命令,且不产生回执。 */
@Test @Test
void invalidCommandsAreRejectedWithoutReceipt(@TempDir Path root) throws Exception { void invalidCommandsAreRejectedWithoutReceipt(@TempDir Path root) throws Exception {
@@ -167,12 +261,52 @@ class ReliableTaskServiceTest {
assertTrue(harness.service.journal().receipts().isEmpty(), "无效命令不得留下回执"); assertTrue(harness.service.journal().receipts().isEmpty(), "无效命令不得留下回执");
} }
/**
* 重启把任务恢复成新事实后,崩溃前生成的旧待 ACK 报告不得清掉它。
*
* <p>恢复必须把重建结果落盘({@link ReliableTaskService#restore} 末尾的 syncTasks):
* 它既让日志与磁盘一致(否则后续报告会带着过期的 COMPRESSING 上报),又用更高的 revision
* 取代旧记录,使一份内容巧合匹配的旧 ACK 被 removeTaskIfRevision 的版本过滤拒绝。
*/
@Test
void staleAckFromBeforeRestartDoesNotClearRecoveredTask(@TempDir Path root) throws Exception {
Harness first = new Harness(root);
first.service.handleCommand(download("op-stale", 80070, "sample [80070]"));
// 崩溃前:日志停在压缩中,且已生成一份覆盖该状态的待 ACK 报告。
first.queue.get(80070).setStatus(GalleryTask.COMPRESSING);
first.service.refreshJournalFromQueue();
TaskJournal.ReportSnapshot staleSnapshot = first.service.buildReport().orElseThrow();
Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80070]"));
writeValidZip(archiveDirectory.resolve("sample [80070].zip"));
Harness restarted = new Harness(root, true);
GalleryTask recovered = restarted.queue.get(80070);
assertEquals(GalleryTask.COMPRESS_COMPLETE, recovered.getStatus(), "重启应按磁盘事实恢复为压缩完成");
// 恢复结果必须已经落盘:从磁盘重新加载日志仍是新状态。
TaskJournal reloaded = new TaskJournal(root.resolve("task-journal"));
reloaded.load();
assertEquals(GalleryTask.COMPRESS_COMPLETE, reloaded.task(80070).status,
"恢复后的新事实必须同步进日志");
assertTrue(reloaded.task(80070).revision
> staleSnapshot.tasks.stream().filter(t -> t.gid == 80070).findFirst().orElseThrow().revision,
"恢复必须推进 revision,使旧快照的版本过滤失效");
restarted.service.applyAck(staleSnapshot.reportId, NodeErrorCode.SUCCESS);
assertTrue(restarted.queue.containsKey(80070),
"崩溃前的旧 ACK 不得清掉已恢复的任务");
assertTrue(restarted.service.journal().tasks().containsKey(80070),
"日志里也必须保留已恢复的任务");
}
/** 一个最小离线装配:显式目录、无调度器、无网络。 */ /** 一个最小离线装配:显式目录、无调度器、无网络。 */
static class Harness { static class Harness {
final Map<Integer, GalleryTask> queue = new ConcurrentHashMap<>(); final Map<Integer, GalleryTask> queue = new ConcurrentHashMap<>();
final DownloadCheckService downloadCheck; final DownloadCheckService downloadCheck;
final ReliableTaskService service; final ReliableTaskService service;
final Path storagePath; final Path storagePath;
final Path downloadsPath;
Harness(Path root) throws Exception { Harness(Path root) throws Exception {
this(root, false); this(root, false);
@@ -180,17 +314,28 @@ class ReliableTaskServiceTest {
Harness(Path root, boolean restore) throws Exception { Harness(Path root, boolean restore) throws Exception {
storagePath = Files.createDirectories(root.resolve("gallery")); storagePath = Files.createDirectories(root.resolve("gallery"));
Path downloads = Files.createDirectories(root.resolve("downloads")); downloadsPath = Files.createDirectories(root.resolve("downloads"));
Path journalDir = root.resolve("task-journal"); Path journalDir = root.resolve("task-journal");
downloadCheck = new DownloadCheckService(queue, false); downloadCheck = new DownloadCheckService(queue, false);
downloadCheck.storagePath = storagePath.toString(); downloadCheck.storagePath = storagePath.toString();
downloadCheck.downloadPath = downloads.toString(); downloadCheck.downloadPath = downloadsPath.toString();
service = new ReliableTaskService(journalDir, queue, downloadCheck, storagePath); service = new ReliableTaskService(journalDir, queue, downloadCheck, storagePath);
if (restore) if (restore)
service.restore(); service.restore();
} }
} }
/** 写一个真实的小 ZIP(含 CRC 校验,可被 isValidArchive 接受)。 */
static void writeValidZip(Path zip) throws Exception {
Files.createDirectories(zip.getParent());
try (java.util.zip.ZipOutputStream out =
new java.util.zip.ZipOutputStream(Files.newOutputStream(zip))) {
out.putNextEntry(new java.util.zip.ZipEntry("1.jpg"));
out.write("image bytes".getBytes(java.nio.charset.StandardCharsets.UTF_8));
out.closeEntry();
}
}
private static TaskCommandMessage download(String operationId, int gid, String name) { private static TaskCommandMessage download(String operationId, int gid, String name) {
TaskCommandMessage command = new TaskCommandMessage(); TaskCommandMessage command = new TaskCommandMessage();
command.setOperationId(operationId); command.setOperationId(operationId);