lion/storageNode
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复压缩失败恢复并校验归档完整性

Browse Source
This commit is contained in:
root
2026-09-08 09:18:21 +08:00
parent 942aecf27c
commit cbfd634f0d
4 changed files with 142 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/lion/Domain/GalleryTask.java
+4 -4
View File
@@ -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(){
src/main/java/lion/Service/DownloadCheckService.java
+86 -20
View File
@@ -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<GalleryTask> compress_queue;
final Map<Integer, Long> retryAfter = new ConcurrentHashMap<>();
public DownloadCheckService(Map<Integer, GalleryTask> queue){
this(queue, true);
}
// Tests use temporary directories and invoke scans explicitly.
DownloadCheckService(Map<Integer, GalleryTask> 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<GalleryTask> 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<? extends ZipEntry> 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;
src/main/java/lion/storageNode.java
+1 -1
View File
@@ -52,7 +52,7 @@ public class storageNode {
thread.setDaemon(true);
return thread;
});
queue = new HashMap<>();
queue = new ConcurrentHashMap<>();
tempQueue = new HashMap<>();
lock = new ReentrantLock();
src/test/java/lion/Service/DownloadCheckServiceTest.java
+51
View File
@@ -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<Integer, GalleryTask>();
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());
}
}