diff --git a/src/main/java/lion/Domain/GalleryTask.java b/src/main/java/lion/Domain/GalleryTask.java index 0c96818..79eaae5 100644 --- a/src/main/java/lion/Domain/GalleryTask.java +++ b/src/main/java/lion/Domain/GalleryTask.java @@ -12,16 +12,16 @@ public class GalleryTask { public static final byte COMPRESS_COMPLETE = 4; @JsonInclude(JsonInclude.Include.NON_NULL) - private String name; + private volatile String name; private int gid; - private byte status; + private volatile byte status; - private int proceeding; + private volatile int proceeding; @JsonIgnore - private String path; + private volatile String path; @JsonIgnore public boolean is_download_complete(){ diff --git a/src/main/java/lion/Service/DownloadCheckService.java b/src/main/java/lion/Service/DownloadCheckService.java index 26c579b..0c77d88 100644 --- a/src/main/java/lion/Service/DownloadCheckService.java +++ b/src/main/java/lion/Service/DownloadCheckService.java @@ -10,6 +10,11 @@ import java.io.*; import java.util.*; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.nio.file.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.zip.CRC32; +import java.util.zip.ZipEntry; +import java.util.zip.ZipFile; @Slf4j public class DownloadCheckService { @@ -23,11 +28,20 @@ public class DownloadCheckService { final ArrayList compress_queue; + final Map retryAfter = new ConcurrentHashMap<>(); + public DownloadCheckService(Map queue){ + this(queue, true); + } + + // Tests use temporary directories and invoke scans explicitly. + DownloadCheckService(Map queue, boolean startScheduler){ this.queue = queue; compress_queue = new ArrayList<>(0); - convert_thread = new ScheduledThreadPoolExecutor(1); - convert_thread.scheduleAtFixedRate(this::compress, 0, 5, TimeUnit.SECONDS); + if (startScheduler) { + convert_thread = new ScheduledThreadPoolExecutor(1); + convert_thread.scheduleWithFixedDelay(this::compress, 0, 5, TimeUnit.SECONDS); + } } public boolean downloadCheck(){ @@ -55,7 +69,7 @@ public class DownloadCheckService { while(fileIterator.hasNext()){ File file = fileIterator.next(); - if(!file.getName().contains(String.valueOf(galleryTask.getGid()))) + if(!file.isDirectory() || !matchesGid(file.getName(), galleryTask.getGid())) continue; galleryTask.setStatus(GalleryTask.DOWNLOADING); @@ -81,7 +95,8 @@ public class DownloadCheckService { //压缩队列 for(GalleryTask galleryTask: queue.values()) - if (galleryTask.is_download_complete()) { + if (galleryTask.is_download_complete() + && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) { galleryTask.setStatus(GalleryTask.COMPRESSING); synchronized (compress_queue) { compress_queue.add(galleryTask); @@ -95,36 +110,87 @@ public class DownloadCheckService { * 压缩线程：将压缩队列的任务复制一份，进行转换 */ public void compress() { - if(compress_queue.isEmpty()) - return; ArrayList galleryTasks; synchronized (compress_queue) { + if (compress_queue.isEmpty()) + return; galleryTasks = new ArrayList<>(compress_queue); compress_queue.clear(); } for (GalleryTask galleryTask : galleryTasks) { + Path temporary = null; try { log.info("开始压缩:{}", galleryTask.getName()); - File file = new File(storagePath + galleryTask.getName()); - if (file.isDirectory() || file.mkdirs()) { - log.info("{}文件夹创建成功", galleryTask.getName()); - } else { - log.error("{}文件夹创建失败", galleryTask.getName()); - continue; + Path directory = Paths.get(storagePath, galleryTask.getName()); + Files.createDirectories(directory); + Path archive = directory.resolve(galleryTask.getName() + ".zip"); + temporary = Files.createTempFile(directory, ".compress-", ".zip.part"); + ZipUtil.zip(galleryTask.getPath(), temporary.toString()); + if (!isValidArchive(temporary.toFile())) + throw new IOException("压缩包校验失败"); + // Publish only a closed, verified archive. A crash leaves a .part file. + try { + Files.move(temporary, archive, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); + } catch (AtomicMoveNotSupportedException e) { + Files.move(temporary, archive, StandardCopyOption.REPLACE_EXISTING); } - - ZipUtil.zip(galleryTask.getPath(), storagePath + galleryTask.getName() + "/" + galleryTask.getName() + ".zip"); - log.info("{}压缩完成", galleryTask.getName()); - - FileUtil.del(galleryTask.getPath()); + temporary = null; galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE); - } catch (Exception e){ - log.error("{}压缩失败:{}", galleryTask, e.getMessage()); + retryAfter.remove(galleryTask.getGid()); + if (!FileUtil.del(galleryTask.getPath())) + log.warn("压缩已完成，但源目录清理失败: {}", galleryTask.getPath()); + log.info("{}压缩完成", galleryTask.getName()); + } catch (Exception e) { + // Keep the source and restore an existing, retryable protocol state. + if (!galleryTask.is_compress_complete()) { + retryAfter.put(galleryTask.getGid(), System.currentTimeMillis() + 30_000); + galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE); + } + log.error("{}压缩或清理失败，源文件保留，稍后可重试", galleryTask.getName(), e); + } finally { + if (temporary != null) { + try { Files.deleteIfExists(temporary); } + catch (IOException e) { log.warn("清理压缩临时文件失败: {}", temporary, e); } + } } } } + static boolean isValidArchive(File file) { + if (!file.isFile()) + return false; + try (ZipFile zip = new ZipFile(file)) { + if (zip.size() == 0) + return false; + byte[] buffer = new byte[8192]; + Enumeration entries = zip.entries(); + while (entries.hasMoreElements()) { + ZipEntry entry = entries.nextElement(); + if (entry.isDirectory()) + continue; + CRC32 crc = new CRC32(); + long size = 0; + try (InputStream input = zip.getInputStream(entry)) { + int count; + while ((count = input.read(buffer)) != -1) { + crc.update(buffer, 0, count); + size += count; + } + } + if (size != entry.getSize() || crc.getValue() != entry.getCrc()) + return false; + } + return true; + } catch (IOException e) { + return false; + } + } + public GalleryTask addToQueue(GalleryTask galleryTask){ + GalleryTask active = queue.get(galleryTask.getGid()); + if (active != null && active.is_compressing()) + return active; + retryAfter.remove(galleryTask.getGid()); // A reconnect can resend a task whose name is stale (for example, the // downloader appended a resolution suffix). Resolve completed archives // by gid first, because the gid is stable while the directory name is not. @@ -176,7 +242,7 @@ public class DownloadCheckService { for(File directory : directories){ if(matchesGid(directory.getName(), gid) - && new File(directory, directory.getName() + ".zip").isFile()) + && isValidArchive(new File(directory, directory.getName() + ".zip"))) return directory; } return null; diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index ffbd9b7..86eb467 100644 --- a/src/main/java/lion/storageNode.java +++ b/src/main/java/lion/storageNode.java @@ -52,7 +52,7 @@ public class storageNode { thread.setDaemon(true); return thread; }); - queue = new HashMap<>(); + queue = new ConcurrentHashMap<>(); tempQueue = new HashMap<>(); lock = new ReentrantLock(); diff --git a/src/test/java/lion/Service/DownloadCheckServiceTest.java b/src/test/java/lion/Service/DownloadCheckServiceTest.java new file mode 100644 index 0000000..6b5c3f3 --- /dev/null +++ b/src/test/java/lion/Service/DownloadCheckServiceTest.java @@ -0,0 +1,51 @@ +package lion.Service; + +import lion.Domain.GalleryTask; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import java.nio.file.*; +import java.util.concurrent.ConcurrentHashMap; +import static org.junit.jupiter.api.Assertions.*; + +class DownloadCheckServiceTest { + @Test + void failedCompressionRetainsSourceAndCanRetry(@TempDir Path root) throws Exception { + var queue = new ConcurrentHashMap(); + var service = new DownloadCheckService(queue, false); + Path downloads = Files.createDirectory(root.resolve("downloads")); + Path source = Files.createDirectory(downloads.resolve("sample [123]")); + Files.writeString(source.resolve("galleryinfo.txt"), "metadata"); + Files.writeString(source.resolve("1.jpg"), "image bytes"); + Path storage = root.resolve("storage"); + Files.writeString(storage, "block mkdir"); + service.downloadPath = downloads.toString(); + service.storagePath = storage.toString(); + GalleryTask task = new GalleryTask(); + task.setGid(123); + service.addToQueue(task); + service.downloadCheck(); + service.compress(); + assertEquals(GalleryTask.DOWNLOAD_COMPLETE, task.getStatus()); + assertTrue(Files.exists(source.resolve("1.jpg"))); + Files.delete(storage); + Files.createDirectory(storage); + service.addToQueue(task); // manual retry removes the backoff + service.downloadCheck(); + service.compress(); + assertEquals(GalleryTask.COMPRESS_COMPLETE, task.getStatus()); + assertTrue(DownloadCheckService.isValidArchive(storage.resolve("sample [123]/sample [123].zip").toFile())); + assertFalse(Files.exists(source)); + } + + @Test + void corruptArchiveIsNotTreatedAsCompleted(@TempDir Path root) throws Exception { + Path stored = Files.createDirectories(root.resolve("stored/sample [123]")); + Files.writeString(stored.resolve("sample [123].zip"), "partial zip"); + var service = new DownloadCheckService(new ConcurrentHashMap<>(), false); + service.storagePath = root.resolve("stored").toString(); + service.downloadPath = root.resolve("downloads").toString(); + GalleryTask task = new GalleryTask(); + task.setGid(123); + assertFalse(service.addToQueue(task).is_compress_complete()); + } +}