Compare commits
3 Commits
1fe3368c5a
...
a89de129ea
| Author | SHA1 | Date | |
|---|---|---|---|
| a89de129ea | |||
| ddefcbed56 | |||
| 3137299bfe |
+74
-4
@@ -424,6 +424,53 @@ public class AppearancePatentTaskService {
|
|||||||
|
|
||||||
public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) {
|
public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) {
|
||||||
submitResultLocked(taskId, request);
|
submitResultLocked(taskId, request);
|
||||||
|
// 分片补传后自动恢复终态失败的组装 job —— 走与删除品牌同一套口径,
|
||||||
|
// 此前外观专利没有接该入口,分片补齐后只能人工重置 job(线上任务 28459 即如此)。
|
||||||
|
maybeRecoverTerminalFailedAssemble(taskId);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 补传恢复:此前因分片缺失导致组装 job 重试耗尽(终态失败),客户端补传缺口后
|
||||||
|
* 把失败的组装 job 重置为 PENDING 重新派发({@code resetTerminalFailedForRecovery} 自行补发 dispatch 事件)。
|
||||||
|
*
|
||||||
|
* <p>best-effort:恢复失败不得影响补传本身——分片已经落库,恢复只是让后续组装继续推进。
|
||||||
|
* 常态(无终态失败 job)下只查两次即返回,不触发分片扫描。
|
||||||
|
*/
|
||||||
|
private void maybeRecoverTerminalFailedAssemble(Long taskId) {
|
||||||
|
if (taskId == null || taskId <= 0) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
FileResultEntity result = findResultRecord(taskId);
|
||||||
|
if (result == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (taskFileJobService.hasSuccessfulAssembleJob(taskId, MODULE_TYPE, result.getId())) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (!taskFileJobService.isTerminalFailedAssembleJob(taskId, MODULE_TYPE, result.getId())) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if (!isResultSubmissionComplete(taskId)) {
|
||||||
|
log.info("[appearance-patent] 组装 job 终态失败但分片仍未补传完整,暂不恢复 taskId={} resultId={}",
|
||||||
|
taskId, result.getId());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
log.info("[appearance-patent] 分片已补传完整,恢复终态失败的组装 job taskId={} resultId={}",
|
||||||
|
taskId, result.getId());
|
||||||
|
taskFileJobService.resetTerminalFailedForRecovery(taskId, MODULE_TYPE, result.getId());
|
||||||
|
} catch (Exception ex) {
|
||||||
|
log.warn("[appearance-patent] 补传恢复检查失败(不影响本次补传)taskId={} err={}", taskId, ex.getMessage(), ex);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 按 task 取结果行(不创建);不存在返回 null。 */
|
||||||
|
private FileResultEntity findResultRecord(Long taskId) {
|
||||||
|
List<FileResultEntity> rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||||
|
.eq(FileResultEntity::getTaskId, taskId)
|
||||||
|
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||||
|
.last("limit 1"));
|
||||||
|
return rows == null || rows.isEmpty() ? null : rows.getFirst();
|
||||||
}
|
}
|
||||||
|
|
||||||
private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
|
private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
|
||||||
@@ -1223,6 +1270,7 @@ public class AppearancePatentTaskService {
|
|||||||
if (rows == null || rows.isEmpty()) {
|
if (rows == null || rows.isEmpty()) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
String conflictDetail = "未发生冲突";
|
||||||
for (int attempt = 1; attempt <= CHUNK_PAYLOAD_MERGE_RETRY_LIMIT; attempt++) {
|
for (int attempt = 1; attempt <= CHUNK_PAYLOAD_MERGE_RETRY_LIMIT; attempt++) {
|
||||||
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||||
@@ -1264,13 +1312,35 @@ public class AppearancePatentTaskService {
|
|||||||
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
|
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
|
// CAS 冲突:读回行上的当前哈希,否则只有一句 conflict 无从判断是谁改的(线上 28459 至今未能定位)
|
||||||
|
String currentHash = currentPayloadHash(chunk.getId());
|
||||||
|
conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash;
|
||||||
if (attempt < CHUNK_PAYLOAD_MERGE_RETRY_LIMIT) {
|
if (attempt < CHUNK_PAYLOAD_MERGE_RETRY_LIMIT) {
|
||||||
log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}",
|
log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}",
|
||||||
taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT);
|
taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT, oldPayloadHash, currentHash);
|
||||||
|
// 还要重试:这次写的对象会被下次重写,先删掉避免堆积
|
||||||
|
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
|
||||||
|
} else {
|
||||||
|
// 终局失败:保留这次写入的版本化对象,作为读路径「同槽位兄弟对象」兜底的恢复源。
|
||||||
|
// 行没指过去不该让该分片永久判死——删掉它才是线上 28459 丢数据的形态。
|
||||||
|
log.error("[appearance-patent] chunk payload 合并冲突重试耗尽,保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}",
|
||||||
|
taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
throw new IllegalStateException("appearance patent chunk payload update conflict");
|
throw new IllegalStateException("appearance patent chunk payload update conflict " + conflictDetail);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 读回 chunk 行当前的 payload_hash,用于 CAS 冲突定位(行已不存在/读取失败时返回可读标记)。 */
|
||||||
|
private String currentPayloadHash(Long chunkId) {
|
||||||
|
if (chunkId == null) {
|
||||||
|
return "chunkId 为空";
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
TaskChunkEntity latest = taskChunkMapper.selectById(chunkId);
|
||||||
|
return latest == null ? "行已不存在" : latest.getPayloadHash();
|
||||||
|
} catch (Exception ex) {
|
||||||
|
return "读取失败:" + ex.getMessage();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private List<AppearancePatentResultRowDto> expandRows(List<AppearancePatentResultRowDto> representatives,
|
private List<AppearancePatentResultRowDto> expandRows(List<AppearancePatentResultRowDto> representatives,
|
||||||
|
|||||||
+97
-1
@@ -5,11 +5,14 @@ import io.micrometer.core.instrument.DistributionSummary;
|
|||||||
import io.micrometer.core.instrument.MeterRegistry;
|
import io.micrometer.core.instrument.MeterRegistry;
|
||||||
import io.micrometer.core.instrument.Timer;
|
import io.micrometer.core.instrument.Timer;
|
||||||
import io.minio.GetObjectArgs;
|
import io.minio.GetObjectArgs;
|
||||||
|
import io.minio.ListObjectsArgs;
|
||||||
import io.minio.MinioClient;
|
import io.minio.MinioClient;
|
||||||
import io.minio.PutObjectArgs;
|
import io.minio.PutObjectArgs;
|
||||||
import io.minio.RemoveObjectArgs;
|
import io.minio.RemoveObjectArgs;
|
||||||
|
import io.minio.Result;
|
||||||
import io.minio.StatObjectArgs;
|
import io.minio.StatObjectArgs;
|
||||||
import io.minio.errors.ErrorResponseException;
|
import io.minio.errors.ErrorResponseException;
|
||||||
|
import io.minio.messages.Item;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import okhttp3.ConnectionPool;
|
import okhttp3.ConnectionPool;
|
||||||
import okhttp3.Dispatcher;
|
import okhttp3.Dispatcher;
|
||||||
@@ -21,8 +24,11 @@ import org.springframework.stereotype.Service;
|
|||||||
import java.io.ByteArrayInputStream;
|
import java.io.ByteArrayInputStream;
|
||||||
import java.io.ByteArrayOutputStream;
|
import java.io.ByteArrayOutputStream;
|
||||||
import java.nio.charset.StandardCharsets;
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
|
import java.util.UUID;
|
||||||
import java.util.concurrent.Semaphore;
|
import java.util.concurrent.Semaphore;
|
||||||
import java.util.concurrent.ThreadLocalRandom;
|
import java.util.concurrent.ThreadLocalRandom;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
@@ -45,6 +51,12 @@ public class RustfsObjectStorageService {
|
|||||||
"NoSuchKey", "NoSuchBucket", "NoSuchVersion",
|
"NoSuchKey", "NoSuchBucket", "NoSuchVersion",
|
||||||
"AccessDenied", "InvalidAccessKeyId", "SignatureDoesNotMatch", "InvalidBucketName");
|
"AccessDenied", "InvalidAccessKeyId", "SignatureDoesNotMatch", "InvalidBucketName");
|
||||||
|
|
||||||
|
/** 标准 UUID 字符串长度(8-4-4-4-12),用于识别版本化对象 key。 */
|
||||||
|
private static final int UUID_STRING_LENGTH = 36;
|
||||||
|
|
||||||
|
/** 兄弟对象兜底列出的上限:只为找回同槽位的版本化对象,不需要列全。 */
|
||||||
|
private static final int MAX_SIBLING_LIST_KEYS = 50;
|
||||||
|
|
||||||
private final TransientStorageProperties properties;
|
private final TransientStorageProperties properties;
|
||||||
private final ObjectProvider<MeterRegistry> meterRegistryProvider;
|
private final ObjectProvider<MeterRegistry> meterRegistryProvider;
|
||||||
private final ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider;
|
private final ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider;
|
||||||
@@ -106,7 +118,22 @@ public class RustfsObjectStorageService {
|
|||||||
return uploadBytes(objectKey, bytes, verifyAfterUpload);
|
return uploadBytes(objectKey, bytes, verifyAfterUpload);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 三参重载:是否做「上传失败补偿删除」按对象 key 形态自动判定。
|
||||||
|
*
|
||||||
|
* <p>只有版本化 key(末段以 UUID 结尾)是本次写入独占的;确定性 key 会被重传重写复用,
|
||||||
|
* 删它就可能删掉别的 DB 行仍在引用的对象(2026-09-17 线上任务 28459 的载荷对象就是这么丢的)。
|
||||||
|
*/
|
||||||
public String uploadBytes(String objectKey, byte[] content, boolean verifyAfterUpload) {
|
public String uploadBytes(String objectKey, byte[] content, boolean verifyAfterUpload) {
|
||||||
|
return uploadBytes(objectKey, content, verifyAfterUpload, isVersionedObjectKey(objectKey));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @param compensateDeleteOnFailure put 已完成、但后续可见性校验失败时,是否把该对象排进删除补偿队列。
|
||||||
|
* 仅当调用方能确认「该对象不会被其它写入复用时」才可传 true。
|
||||||
|
*/
|
||||||
|
public String uploadBytes(String objectKey, byte[] content, boolean verifyAfterUpload,
|
||||||
|
boolean compensateDeleteOnFailure) {
|
||||||
long deadlineNanos = operationDeadlineNanos();
|
long deadlineNanos = operationDeadlineNanos();
|
||||||
if (!isConfigured()) {
|
if (!isConfigured()) {
|
||||||
throw new IllegalStateException("transient storage is not configured");
|
throw new IllegalStateException("transient storage is not configured");
|
||||||
@@ -136,13 +163,45 @@ public class RustfsObjectStorageService {
|
|||||||
}
|
}
|
||||||
return uploadedObjectKey;
|
return uploadedObjectKey;
|
||||||
} catch (RuntimeException ex) {
|
} catch (RuntimeException ex) {
|
||||||
if (putCompleted.get()) {
|
if (putCompleted.get() && compensateDeleteOnFailure) {
|
||||||
enqueueDeleteRetry(objectKey, ex);
|
enqueueDeleteRetry(objectKey, ex);
|
||||||
|
} else if (putCompleted.get()) {
|
||||||
|
// 共享 key 会被重传重写:此处删除可能删掉别的行正在引用的对象,交给保留期清理兜底。
|
||||||
|
// 线上任务 28459 的 chunk-462/473/484 就是被这条无条件删除队列删掉的。
|
||||||
|
log.warn("[rustfs] 跳过上传失败补偿删除(对象非本次独占,可能被复用)objectKey={} err={}",
|
||||||
|
objectKey, ex.getMessage());
|
||||||
}
|
}
|
||||||
throw ex;
|
throw ex;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 对象 key 是否为「本次写入独占」的版本化 key:末段(去掉 {@code .json} 后缀)以 UUID 结尾。
|
||||||
|
*
|
||||||
|
* <p>只看末段——UUID 出现在中间段(如 scopeHash)不代表该对象被独占;解析失败一律按共享处理
|
||||||
|
* (保守:宁可留下孤儿对象,也不删掉可能仍被引用的对象)。
|
||||||
|
*/
|
||||||
|
static boolean isVersionedObjectKey(String objectKey) {
|
||||||
|
if (objectKey == null) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
String key = objectKey.trim();
|
||||||
|
if (key.endsWith(".json")) {
|
||||||
|
key = key.substring(0, key.length() - ".json".length());
|
||||||
|
}
|
||||||
|
int slash = key.lastIndexOf('/');
|
||||||
|
String lastSegment = slash < 0 ? key : key.substring(slash + 1);
|
||||||
|
if (lastSegment.length() < UUID_STRING_LENGTH) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
UUID.fromString(lastSegment.substring(lastSegment.length() - UUID_STRING_LENGTH));
|
||||||
|
return true;
|
||||||
|
} catch (IllegalArgumentException ex) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public String readObjectAsString(String objectKey) {
|
public String readObjectAsString(String objectKey) {
|
||||||
byte[] bytes = readObjectBytes(objectKey);
|
byte[] bytes = readObjectBytes(objectKey);
|
||||||
return new String(bytes, StandardCharsets.UTF_8);
|
return new String(bytes, StandardCharsets.UTF_8);
|
||||||
@@ -182,6 +241,43 @@ public class RustfsObjectStorageService {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 列出前缀下的对象 key,按最后修改时间倒序(最新在前)。
|
||||||
|
*
|
||||||
|
* <p>只服务于「指针指向的对象已不存在、需要找回同槽位的版本化兄弟对象」这一兜底路径,
|
||||||
|
* 因此刻意不做重试、不参与失败窗口记账:列出失败直接抛错,由调用方按原错误语义处理。
|
||||||
|
*/
|
||||||
|
public List<String> listObjectKeysNewestFirst(String prefix, int limit) {
|
||||||
|
if (!isConfigured()) {
|
||||||
|
throw new IllegalStateException("transient storage is not configured");
|
||||||
|
}
|
||||||
|
int safeLimit = Math.max(1, Math.min(limit, MAX_SIBLING_LIST_KEYS));
|
||||||
|
long deadlineNanos = operationDeadlineNanos();
|
||||||
|
List<String[]> entries = new ArrayList<>();
|
||||||
|
try {
|
||||||
|
Iterable<Result<Item>> results = buildClient(deadlineNanos).listObjects(ListObjectsArgs.builder()
|
||||||
|
.bucket(properties.getBucket())
|
||||||
|
.prefix(prefix == null ? "" : prefix)
|
||||||
|
.recursive(true)
|
||||||
|
.maxKeys(safeLimit)
|
||||||
|
.build());
|
||||||
|
for (Result<Item> result : results) {
|
||||||
|
Item item = result.get();
|
||||||
|
entries.add(new String[]{item.objectName(),
|
||||||
|
item.lastModified() == null ? "" : item.lastModified().toString()});
|
||||||
|
}
|
||||||
|
} catch (Exception ex) {
|
||||||
|
throw new IllegalStateException("transient storage list failed prefix=" + prefix
|
||||||
|
+ " err=" + ex.getMessage(), ex);
|
||||||
|
}
|
||||||
|
entries.sort((left, right) -> right[1].compareTo(left[1]));
|
||||||
|
List<String> keys = new ArrayList<>(entries.size());
|
||||||
|
for (String[] entry : entries) {
|
||||||
|
keys.add(entry[0]);
|
||||||
|
}
|
||||||
|
return keys;
|
||||||
|
}
|
||||||
|
|
||||||
public void deleteObject(String objectKey) {
|
public void deleteObject(String objectKey) {
|
||||||
deleteObject(objectKey, true, operationDeadlineNanos());
|
deleteObject(objectKey, true, operationDeadlineNanos());
|
||||||
}
|
}
|
||||||
|
|||||||
+27
-4
@@ -214,6 +214,7 @@ public class SimilarAsinPipelineSupport {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
int maxAttempts = 3;
|
int maxAttempts = 3;
|
||||||
|
String conflictDetail = "未发生冲突";
|
||||||
for (int attempt = 1; attempt <= maxAttempts; attempt++) {
|
for (int attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||||
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
|
||||||
.eq(TaskChunkEntity::getTaskId, taskId)
|
.eq(TaskChunkEntity::getTaskId, taskId)
|
||||||
@@ -275,13 +276,35 @@ public class SimilarAsinPipelineSupport {
|
|||||||
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
|
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
|
// CAS 冲突:读回行上的当前哈希,否则只有一句 conflict 无从判断是谁改的
|
||||||
|
String currentHash = currentPayloadHash(chunk.getId());
|
||||||
|
conflictDetail = "期望hash=" + oldPayloadHash + " 当前hash=" + currentHash;
|
||||||
if (attempt < maxAttempts) {
|
if (attempt < maxAttempts) {
|
||||||
log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}",
|
log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{} 期望hash={} 当前hash={}",
|
||||||
taskId, scopeHash, chunkIndex, attempt, maxAttempts);
|
taskId, scopeHash, chunkIndex, attempt, maxAttempts, oldPayloadHash, currentHash);
|
||||||
|
// 还要重试:这次写的对象会被下次重写,先删掉避免堆积
|
||||||
|
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
|
||||||
|
} else {
|
||||||
|
// 终局失败:保留这次写入的版本化对象,作为读路径「同槽位兄弟对象」兜底的恢复源。
|
||||||
|
// 与外观专利同一口径——行没指过去不该让该分片永久判死。
|
||||||
|
log.error("[similar-asin] chunk payload 合并冲突重试耗尽,保留兜底对象 taskId={} scopeHash={} chunk={} 期望hash={} 当前hash={} 保留对象={}",
|
||||||
|
taskId, scopeHash, chunkIndex, oldPayloadHash, currentHash, storedPayload);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
throw new IllegalStateException("相似ASIN分片载荷更新失败");
|
throw new IllegalStateException("相似ASIN分片载荷更新失败 " + conflictDetail);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 读回 chunk 行当前的 payload_hash,用于 CAS 冲突定位(行已不存在/读取失败时返回可读标记)。 */
|
||||||
|
private String currentPayloadHash(Long chunkId) {
|
||||||
|
if (chunkId == null) {
|
||||||
|
return "chunkId 为空";
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
TaskChunkEntity latest = taskChunkMapper.selectById(chunkId);
|
||||||
|
return latest == null ? "行已不存在" : latest.getPayloadHash();
|
||||||
|
} catch (Exception ex) {
|
||||||
|
return "读取失败:" + ex.getMessage();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+67
-5
@@ -30,6 +30,7 @@ import java.util.Locale;
|
|||||||
import java.util.Objects;
|
import java.util.Objects;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
import java.util.UUID;
|
import java.util.UUID;
|
||||||
|
import java.util.regex.Pattern;
|
||||||
import java.util.zip.GZIPInputStream;
|
import java.util.zip.GZIPInputStream;
|
||||||
import java.util.zip.GZIPOutputStream;
|
import java.util.zip.GZIPOutputStream;
|
||||||
|
|
||||||
@@ -150,8 +151,20 @@ public class TransientPayloadStorageService {
|
|||||||
return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey)));
|
return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey)));
|
||||||
}
|
}
|
||||||
if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) {
|
if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) {
|
||||||
return decodeStoredPayloadBytes(
|
String objectKey = pointer.substring(RUSTFS_POINTER_PREFIX.length());
|
||||||
rustfsObjectStorageService.readObjectBytes(pointer.substring(RUSTFS_POINTER_PREFIX.length())));
|
try {
|
||||||
|
return decodeStoredPayloadBytes(rustfsObjectStorageService.readObjectBytes(objectKey));
|
||||||
|
} catch (RuntimeException readException) {
|
||||||
|
String sibling = findVersionedChunkSibling(objectKey, readException);
|
||||||
|
if (sibling == null) {
|
||||||
|
throw readException;
|
||||||
|
}
|
||||||
|
// 2026-09-17 线上任务 28459:行指向的普通 key 被误删,但同一分片槽位的版本化对象还在。
|
||||||
|
// 读出它即可让任务按已有数据出结果,不必整单失败。
|
||||||
|
log.warn("[transient-payload] chunk 载荷对象不存在,回退同槽位版本化对象 pointer={} sibling={}",
|
||||||
|
pointer, sibling);
|
||||||
|
return decodeStoredPayloadBytes(rustfsObjectStorageService.readObjectBytes(sibling));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if (pointer.startsWith(OSS_POINTER_PREFIX)) {
|
if (pointer.startsWith(OSS_POINTER_PREFIX)) {
|
||||||
return decodeStoredPayloadBytes(
|
return decodeStoredPayloadBytes(
|
||||||
@@ -164,6 +177,53 @@ public class TransientPayloadStorageService {
|
|||||||
return value;
|
return value;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** chunk 载荷槽位的 entryKey 形态:{@code chunk-<index>}(版本化写入则形如 {@code chunk-<index>-<uuid>})。 */
|
||||||
|
private static final Pattern CHUNK_ENTRY_KEY_PATTERN = Pattern.compile("chunk-\\d+");
|
||||||
|
|
||||||
|
/** 兄弟对象查找上限:只为找回同槽位对象,不需要列全。 */
|
||||||
|
private static final int MAX_SIBLING_LOOKUP_KEYS = 50;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 指针对象已不存在时,尝试找回同一分片槽位的版本化兄弟对象。
|
||||||
|
*
|
||||||
|
* <p>两个条件同时满足才兜底,避免读到无关对象或掩盖真实故障:
|
||||||
|
* <ol>
|
||||||
|
* <li>末段 entryKey 是 {@code chunk-<index>} 形态——只有这种槽位才有「版本化兄弟」语义;</li>
|
||||||
|
* <li>失败原因是对象确实不存在(NoSuchKey)——权限/网络类失败照旧上抛以便重试。</li>
|
||||||
|
* </ol>
|
||||||
|
*/
|
||||||
|
private String findVersionedChunkSibling(String objectKey, Throwable cause) {
|
||||||
|
if (!isObjectMissing(cause)) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
int slash = objectKey.lastIndexOf('/');
|
||||||
|
String directory = slash < 0 ? "" : objectKey.substring(0, slash + 1);
|
||||||
|
String fileName = slash < 0 ? objectKey : objectKey.substring(slash + 1);
|
||||||
|
if (!fileName.endsWith(".json")) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
String entryKey = fileName.substring(0, fileName.length() - ".json".length());
|
||||||
|
if (!CHUNK_ENTRY_KEY_PATTERN.matcher(entryKey).matches()) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
List<String> candidates = rustfsObjectStorageService.listObjectKeysNewestFirst(
|
||||||
|
directory + entryKey + "-", MAX_SIBLING_LOOKUP_KEYS);
|
||||||
|
return candidates.isEmpty() ? null : candidates.getFirst();
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 对象确已不存在:RustFS 返回 NoSuchKey,message 为 "The specified key does not exist."。 */
|
||||||
|
private static boolean isObjectMissing(Throwable error) {
|
||||||
|
Throwable cursor = error;
|
||||||
|
while (cursor != null) {
|
||||||
|
String message = cursor.getMessage();
|
||||||
|
if (message != null && message.contains("does not exist")) {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
cursor = cursor.getCause();
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
public void deletePayloadIfPresent(String value) {
|
public void deletePayloadIfPresent(String value) {
|
||||||
String pointer = extractPointer(value);
|
String pointer = extractPointer(value);
|
||||||
if (pointer == null) {
|
if (pointer == null) {
|
||||||
@@ -209,8 +269,10 @@ public class TransientPayloadStorageService {
|
|||||||
*
|
*
|
||||||
* <p>判断口径:
|
* <p>判断口径:
|
||||||
* <ul>
|
* <ul>
|
||||||
* <li>{@code biz_task_chunk.payload_json} 命中 > 1 行(> 1 表示除了 caller 视角下
|
* <li>{@code biz_task_chunk.payload_json} 命中任意行(≥ 1)→ 视为仍被引用。
|
||||||
* 自己即将释放的那一行之外,至少还有别的 chunk 行也指向同一对象)→ 视为仍被引用。</li>
|
* 曾用 {@code > 1} 作判据,等于放行「恰好还有 1 行引用」的情况,会把对方仍在用的对象
|
||||||
|
* 删掉(2026-09-17 线上任务 28459:合并成功后指针未落库 + 旧对象被删 → 该分片永久读不到)。
|
||||||
|
* 物理删除本就约定在 DB 行删除之后执行,故调用方正常路径下引用数必然为 0。</li>
|
||||||
* <li>{@code biz_task_scope_state.parsed_payload_json}/{@code state_json} 命中 > 0 行 →
|
* <li>{@code biz_task_scope_state.parsed_payload_json}/{@code state_json} 命中 > 0 行 →
|
||||||
* 视为仍被引用(这两个字段不是 caller 自身行的常见持有者,命中即非自我引用)。</li>
|
* 视为仍被引用(这两个字段不是 caller 自身行的常见持有者,命中即非自我引用)。</li>
|
||||||
* </ul>
|
* </ul>
|
||||||
@@ -248,7 +310,7 @@ public class TransientPayloadStorageService {
|
|||||||
// 解析不出 taskId 时按原口径全局查,行为与改造前一致。
|
// 解析不出 taskId 时按原口径全局查,行为与改造前一致。
|
||||||
Long pointerTaskId = extractTaskId(pointer);
|
Long pointerTaskId = extractTaskId(pointer);
|
||||||
Long chunkCount = referencedChunkCount(pointerTaskId, values);
|
Long chunkCount = referencedChunkCount(pointerTaskId, values);
|
||||||
if (chunkCount != null && chunkCount > 1L) {
|
if (chunkCount != null && chunkCount > 0L) {
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
Long scopeStateCount = referencedScopeStateCount(pointerTaskId, values);
|
Long scopeStateCount = referencedScopeStateCount(pointerTaskId, values);
|
||||||
|
|||||||
+218
@@ -0,0 +1,218 @@
|
|||||||
|
package com.nanri.aiimage.modules.appearancepatent.service;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
||||||
|
import com.nanri.aiimage.config.AppearancePatentProperties;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentLlmClient;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto;
|
||||||
|
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||||
|
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
import org.mockito.InjectMocks;
|
||||||
|
import org.mockito.Mock;
|
||||||
|
import org.mockito.Spy;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
import org.springframework.test.util.ReflectionTestUtils;
|
||||||
|
import org.springframework.transaction.PlatformTransactionManager;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.lenient;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.times;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 外观专利分片载荷合并的 CAS 冲突处理(2026-09-17 线上任务 28459 的遗留项)。
|
||||||
|
*
|
||||||
|
* <p>原实现在每次 CAS 冲突后都删掉刚写入的版本化对象,重试耗尽即抛异常、行仍指向旧指针——
|
||||||
|
* 一旦旧对象也不在,该分片就永久读不到(28459 的 chunk-462/473/484 正是这个形态)。
|
||||||
|
* 现在:冲突时读回行上的当前哈希以便定位;**终局失败保留最后一个兜底对象**,
|
||||||
|
* 让读路径的「同槽位兄弟对象」兜底仍有数据可取。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class AppearancePatentChunkMergeConflictTest {
|
||||||
|
|
||||||
|
private static final Long TASK_ID = 28459L;
|
||||||
|
private static final String SCOPE_HASH = "2248d39710545b47b7c7035fc60c4e33924a16174abf65dd1c5a230918e6f61e";
|
||||||
|
private static final int CHUNK_INDEX = 462;
|
||||||
|
|
||||||
|
@Mock private LocalFileStorageService localFileStorageService;
|
||||||
|
@Mock private OssStorageService ossStorageService;
|
||||||
|
@Mock private StorageProperties storageProperties;
|
||||||
|
@Mock private FileTaskMapper fileTaskMapper;
|
||||||
|
@Mock private FileResultMapper fileResultMapper;
|
||||||
|
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||||
|
@Mock private TaskChunkMapper taskChunkMapper;
|
||||||
|
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||||
|
@Mock private AppearancePatentLlmClient llmClient;
|
||||||
|
@Mock private AppearancePatentTaskCacheService taskCacheService;
|
||||||
|
@Mock private AppearancePatentProperties properties;
|
||||||
|
@Mock private TaskFileJobService taskFileJobService;
|
||||||
|
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
|
||||||
|
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||||
|
@Mock private PlatformTransactionManager transactionManager;
|
||||||
|
@Mock private DistributedJobLockService distributedJobLockService;
|
||||||
|
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||||
|
@Mock private InstanceMetadata instanceMetadata;
|
||||||
|
@Mock private TaskProgressLightAssembler taskProgressLightAssembler;
|
||||||
|
|
||||||
|
@InjectMocks private AppearancePatentTaskService service;
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void initializeMybatisMetadata() {
|
||||||
|
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||||
|
}
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() {
|
||||||
|
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
|
||||||
|
// 共享写开启才会走版本化对象存储;否则落到本地兜底路径直接报「RustFS 未配置」
|
||||||
|
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 首次冲突后重试成功:中间那次写的对象要删(会被下次重写),且最终行被改到新对象。 */
|
||||||
|
@Test
|
||||||
|
void retryAfterConflictDeletesSupersededObject() {
|
||||||
|
stubChunk("hash-old", "ptr-chunk-462.json");
|
||||||
|
AtomicInteger stores = new AtomicInteger();
|
||||||
|
stubVersionedStore(stores);
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(0, 1);
|
||||||
|
when(taskChunkMapper.selectById(anyLong())).thenReturn(chunk("hash-other", "ptr-chunk-462.json"));
|
||||||
|
|
||||||
|
invokeMerge();
|
||||||
|
|
||||||
|
verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
|
||||||
|
// 第 1 次冲突写的对象被删;第 2 次成功,走的是「替换旧对象」而不是删新对象
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1");
|
||||||
|
verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-2");
|
||||||
|
verify(transientPayloadStorageService).deleteReplacedPayloadIfNeeded(eq("ptr-chunk-462.json"), eq("sibling-2"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 冲突重试耗尽:**保留**最后一次写入的兜底对象(本次要修的形态),并抛异常带出两个哈希。 */
|
||||||
|
@Test
|
||||||
|
void exhaustedConflictKeepsLastStoredObjectAsFallback() {
|
||||||
|
stubChunk("hash-old", "ptr-chunk-462.json");
|
||||||
|
AtomicInteger stores = new AtomicInteger();
|
||||||
|
stubVersionedStore(stores);
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(0);
|
||||||
|
when(taskChunkMapper.selectById(anyLong())).thenReturn(chunk("hash-other", "ptr-chunk-462.json"));
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class, this::invokeMerge);
|
||||||
|
|
||||||
|
verify(transientPayloadStorageService, times(3)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
|
||||||
|
// 前两次冲突的对象照旧删除;第三次(终局)的对象必须保留
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1");
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2");
|
||||||
|
verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3");
|
||||||
|
assertTrue(ex.getMessage().contains("hash-old"), "异常需带出期望哈希,实际: " + ex.getMessage());
|
||||||
|
assertTrue(ex.getMessage().contains("hash-other"), "异常需带出当前哈希,实际: " + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 冲突时读回行上的当前哈希,供定位(此前只有一句 update conflict,线上无法定位)。 */
|
||||||
|
@Test
|
||||||
|
void conflictReadsBackCurrentHashForDiagnostics() {
|
||||||
|
stubChunk("hash-old", "ptr-chunk-462.json");
|
||||||
|
stubVersionedStore(new AtomicInteger());
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(0);
|
||||||
|
when(taskChunkMapper.selectById(7L)).thenReturn(chunk("hash-changed-by-other-writer", "ptr-chunk-462.json"));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class, this::invokeMerge);
|
||||||
|
|
||||||
|
verify(taskChunkMapper, times(3)).selectById(7L);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 读回行失败不得掩盖原始冲突:异常信息里给出可读标记。 */
|
||||||
|
@Test
|
||||||
|
void readBackFailureDoesNotMaskConflict() {
|
||||||
|
stubChunk("hash-old", "ptr-chunk-462.json");
|
||||||
|
stubVersionedStore(new AtomicInteger());
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(0);
|
||||||
|
when(taskChunkMapper.selectById(anyLong())).thenThrow(new IllegalStateException("db down"));
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class, this::invokeMerge);
|
||||||
|
|
||||||
|
assertTrue(ex.getMessage().contains("读取失败"), "实际: " + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 辅助 =====
|
||||||
|
|
||||||
|
private void invokeMerge() {
|
||||||
|
AppearancePatentResultRowDto row = new AppearancePatentResultRowDto();
|
||||||
|
row.setRowToken("r1");
|
||||||
|
row.setAsin("B0A0000001");
|
||||||
|
ReflectionTestUtils.invokeMethod(service, "mergeChunkPayload", TASK_ID, SCOPE_HASH, CHUNK_INDEX, List.of(row));
|
||||||
|
}
|
||||||
|
|
||||||
|
private void stubChunk(String payloadHash, String payloadPointer) {
|
||||||
|
when(taskChunkMapper.selectOne(any())).thenReturn(chunk(payloadHash, payloadPointer));
|
||||||
|
lenient().when(transientPayloadStorageService.resolvePayload(eq(payloadPointer), anyString()))
|
||||||
|
.thenReturn("[{\"rowToken\":\"r1\",\"asin\":\"B0A0000001\"}]");
|
||||||
|
}
|
||||||
|
|
||||||
|
private void stubVersionedStore(AtomicInteger stores) {
|
||||||
|
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||||
|
anyString(), anyLong(), anyString(), any(), anyString()))
|
||||||
|
.thenAnswer(inv -> "sibling-" + stores.incrementAndGet());
|
||||||
|
}
|
||||||
|
|
||||||
|
private TaskChunkEntity chunk(String payloadHash, String payloadPointer) {
|
||||||
|
TaskChunkEntity chunk = new TaskChunkEntity();
|
||||||
|
chunk.setId(7L);
|
||||||
|
chunk.setTaskId(TASK_ID);
|
||||||
|
chunk.setModuleType(AppearancePatentTaskService.MODULE_TYPE);
|
||||||
|
chunk.setScopeHash(SCOPE_HASH);
|
||||||
|
chunk.setChunkIndex(CHUNK_INDEX);
|
||||||
|
chunk.setPayloadJson(payloadPointer);
|
||||||
|
chunk.setPayloadHash(payloadHash);
|
||||||
|
return chunk;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 断言辅助:确认存的对象数(避免误用未使用的 import)。 */
|
||||||
|
@Test
|
||||||
|
void storeIsCalledOnceWhenUpdateSucceeds() {
|
||||||
|
stubChunk("hash-old", "ptr-chunk-462.json");
|
||||||
|
AtomicInteger stores = new AtomicInteger();
|
||||||
|
stubVersionedStore(stores);
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(1);
|
||||||
|
|
||||||
|
invokeMerge();
|
||||||
|
|
||||||
|
assertEquals(1, stores.get());
|
||||||
|
verify(taskChunkMapper, never()).selectById(anyLong());
|
||||||
|
}
|
||||||
|
}
|
||||||
+243
@@ -0,0 +1,243 @@
|
|||||||
|
package com.nanri.aiimage.modules.appearancepatent.service;
|
||||||
|
|
||||||
|
import com.baomidou.mybatisplus.core.MybatisConfiguration;
|
||||||
|
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.common.service.DistributedJobLockService;
|
||||||
|
import com.nanri.aiimage.config.AppearancePatentProperties;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.client.AppearancePatentLlmClient;
|
||||||
|
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentSubmitResultRequest;
|
||||||
|
import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
|
||||||
|
import org.apache.ibatis.builder.MapperBuilderAssistant;
|
||||||
|
import org.junit.jupiter.api.BeforeAll;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.extension.ExtendWith;
|
||||||
|
import org.mockito.InjectMocks;
|
||||||
|
import org.mockito.Mock;
|
||||||
|
import org.mockito.Spy;
|
||||||
|
import org.mockito.junit.jupiter.MockitoExtension;
|
||||||
|
import org.springframework.test.util.ReflectionTestUtils;
|
||||||
|
import org.springframework.transaction.PlatformTransactionManager;
|
||||||
|
import org.springframework.transaction.TransactionDefinition;
|
||||||
|
import org.springframework.transaction.TransactionStatus;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.doAnswer;
|
||||||
|
import static org.mockito.Mockito.lenient;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 外观专利「补传恢复」入口(2026-09-17 线上任务 28459 的遗留项)。
|
||||||
|
*
|
||||||
|
* <p>分片缺失导致组装 job 重试耗尽后,客户端补传缺口应能自动把该 job 重置重跑。
|
||||||
|
* 该能力在 {@code TaskFileJobService} 里早就有了,但只有删除品牌模块接了入口,
|
||||||
|
* 外观专利没有——28459 补齐分片后仍需人工重置 job 才能出结果。
|
||||||
|
*/
|
||||||
|
@ExtendWith(MockitoExtension.class)
|
||||||
|
class AppearancePatentTerminalFailedRecoveryTest {
|
||||||
|
|
||||||
|
private static final Long TASK_ID = 28459L;
|
||||||
|
private static final Long RESULT_ID = 31490L;
|
||||||
|
private static final String MODULE_TYPE = "APPEARANCE_PATENT";
|
||||||
|
|
||||||
|
@Mock private LocalFileStorageService localFileStorageService;
|
||||||
|
@Mock private OssStorageService ossStorageService;
|
||||||
|
@Mock private StorageProperties storageProperties;
|
||||||
|
@Mock private FileTaskMapper fileTaskMapper;
|
||||||
|
@Mock private FileResultMapper fileResultMapper;
|
||||||
|
@Mock private TaskScopeStateMapper taskScopeStateMapper;
|
||||||
|
@Mock private TaskChunkMapper taskChunkMapper;
|
||||||
|
@Spy private ObjectMapper objectMapper = new ObjectMapper();
|
||||||
|
@Mock private AppearancePatentLlmClient llmClient;
|
||||||
|
@Mock private AppearancePatentTaskCacheService taskCacheService;
|
||||||
|
@Mock private AppearancePatentProperties properties;
|
||||||
|
@Mock private TaskFileJobService taskFileJobService;
|
||||||
|
@Mock private TaskProgressSnapshotService taskProgressSnapshotService;
|
||||||
|
@Mock private TransientPayloadStorageService transientPayloadStorageService;
|
||||||
|
@Mock private PlatformTransactionManager transactionManager;
|
||||||
|
@Mock private DistributedJobLockService distributedJobLockService;
|
||||||
|
@Mock private TaskDistributedLockService taskDistributedLockService;
|
||||||
|
@Mock private InstanceMetadata instanceMetadata;
|
||||||
|
@Mock private TaskProgressLightAssembler taskProgressLightAssembler;
|
||||||
|
@Mock private TransactionStatus transactionStatus;
|
||||||
|
|
||||||
|
@InjectMocks private AppearancePatentTaskService service;
|
||||||
|
|
||||||
|
@BeforeAll
|
||||||
|
static void initializeMybatisMetadata() {
|
||||||
|
MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, FileResultEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
|
||||||
|
TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
|
||||||
|
}
|
||||||
|
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() {
|
||||||
|
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
|
||||||
|
lenient().when(transactionManager.getTransaction(any(TransactionDefinition.class))).thenReturn(transactionStatus);
|
||||||
|
lenient().doAnswer(inv -> null).when(transactionManager).commit(transactionStatus);
|
||||||
|
lenient().doAnswer(inv -> null).when(transactionManager).rollback(transactionStatus);
|
||||||
|
lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
lenient().when(taskChunkMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
lenient().when(taskChunkMapper.selectOne(any())).thenReturn(null);
|
||||||
|
lenient().when(taskScopeStateMapper.selectOne(any())).thenReturn(null);
|
||||||
|
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||||
|
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||||
|
lenient().when(transientPayloadStorageService.storeChunkPayload(
|
||||||
|
eq(MODULE_TYPE), eq(TASK_ID), anyString(), any(), anyString()))
|
||||||
|
.thenReturn("\"rustfs:task-chunk/appearance_patent/28459/hash/chunk-11.json\"");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 直接覆盖恢复判定 =====
|
||||||
|
|
||||||
|
/** 已有成功的组装 job → 不恢复。 */
|
||||||
|
@Test
|
||||||
|
void successfulAssembleJobSkipsRecovery() {
|
||||||
|
stubResultRow();
|
||||||
|
when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true);
|
||||||
|
|
||||||
|
invokeRecovery();
|
||||||
|
|
||||||
|
verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 组装 job 不是「重试耗尽的终态失败」→ 不恢复。 */
|
||||||
|
@Test
|
||||||
|
void nonTerminalFailedJobSkipsRecovery() {
|
||||||
|
stubResultRow();
|
||||||
|
when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false);
|
||||||
|
when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false);
|
||||||
|
|
||||||
|
invokeRecovery();
|
||||||
|
|
||||||
|
verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 终态失败 + 分片已补传完整 → 重置 job 重新派发(本次要修的场景)。 */
|
||||||
|
@Test
|
||||||
|
void terminalFailedJobWithCompleteChunksIsReset() {
|
||||||
|
stubResultRow();
|
||||||
|
when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false);
|
||||||
|
when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true);
|
||||||
|
when(taskScopeStateMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
|
invokeRecovery();
|
||||||
|
|
||||||
|
verify(taskFileJobService).resetTerminalFailedForRecovery(TASK_ID, MODULE_TYPE, RESULT_ID);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 终态失败但分片尚未补齐 → 不恢复(否则又会读到缺失分片再失败一次)。 */
|
||||||
|
@Test
|
||||||
|
void terminalFailedJobWithIncompleteChunksIsNotReset() {
|
||||||
|
stubResultRow();
|
||||||
|
when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false);
|
||||||
|
when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true);
|
||||||
|
when(taskScopeStateMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
|
||||||
|
invokeRecovery();
|
||||||
|
|
||||||
|
verify(taskFileJobService, never()).resetTerminalFailedForRecovery(anyLong(), anyString(), anyLong());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 还没有结果行 → 不恢复。 */
|
||||||
|
@Test
|
||||||
|
void missingResultRowSkipsRecovery() {
|
||||||
|
when(fileResultMapper.selectList(any())).thenReturn(List.of());
|
||||||
|
|
||||||
|
invokeRecovery();
|
||||||
|
|
||||||
|
verify(taskFileJobService, never()).isTerminalFailedAssembleJob(anyLong(), anyString(), anyLong());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** taskId 非法 → 直接返回,不查库。 */
|
||||||
|
@Test
|
||||||
|
void invalidTaskIdSkipsLookup() {
|
||||||
|
ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", 0L);
|
||||||
|
ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", (Object) null);
|
||||||
|
|
||||||
|
verify(fileResultMapper, never()).selectList(any());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 恢复检查自身抛异常 → 吞掉,不影响补传结果(best-effort)。 */
|
||||||
|
@Test
|
||||||
|
void recoveryFailureDoesNotPropagate() {
|
||||||
|
when(fileResultMapper.selectList(any())).thenThrow(new IllegalStateException("db down"));
|
||||||
|
|
||||||
|
assertDoesNotThrow(this::invokeRecovery);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 走完整提交路径 =====
|
||||||
|
|
||||||
|
/** 分片提交成功后触发恢复检查(接线正确)。 */
|
||||||
|
@Test
|
||||||
|
void submitResultTriggersRecoveryCheck() {
|
||||||
|
stubRunningTask();
|
||||||
|
stubResultRow();
|
||||||
|
when(taskFileJobService.hasSuccessfulAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(false);
|
||||||
|
when(taskFileJobService.isTerminalFailedAssembleJob(TASK_ID, MODULE_TYPE, RESULT_ID)).thenReturn(true);
|
||||||
|
when(taskScopeStateMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
|
service.submitResult(TASK_ID, request());
|
||||||
|
|
||||||
|
verify(taskFileJobService).resetTerminalFailedForRecovery(TASK_ID, MODULE_TYPE, RESULT_ID);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 辅助 =====
|
||||||
|
|
||||||
|
private void invokeRecovery() {
|
||||||
|
ReflectionTestUtils.invokeMethod(service, "maybeRecoverTerminalFailedAssemble", TASK_ID);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void stubResultRow() {
|
||||||
|
FileResultEntity result = new FileResultEntity();
|
||||||
|
result.setId(RESULT_ID);
|
||||||
|
result.setTaskId(TASK_ID);
|
||||||
|
result.setModuleType(MODULE_TYPE);
|
||||||
|
lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of(result));
|
||||||
|
}
|
||||||
|
|
||||||
|
private void stubRunningTask() {
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(TASK_ID);
|
||||||
|
task.setModuleType(MODULE_TYPE);
|
||||||
|
task.setStatus("RUNNING");
|
||||||
|
task.setUserId(1121L);
|
||||||
|
task.setResultJson("{\"parsedPayloadRef\":\"rustfs:task-parsed/x.json\",\"ownerInstanceId\":\"instance-a\"}");
|
||||||
|
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
|
||||||
|
}
|
||||||
|
|
||||||
|
private AppearancePatentSubmitResultRequest request() {
|
||||||
|
AppearancePatentSubmitResultRequest request = new AppearancePatentSubmitResultRequest();
|
||||||
|
request.setSubmissionId("appearance-patent-" + TASK_ID);
|
||||||
|
request.setChunkIndex(11);
|
||||||
|
request.setChunkTotal(500);
|
||||||
|
request.setDone(false);
|
||||||
|
return request;
|
||||||
|
}
|
||||||
|
}
|
||||||
+12
-6
@@ -297,16 +297,19 @@ class RustfsObjectStorageServiceTest {
|
|||||||
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
properties, emptyProvider(), provider(retryService), () -> client);
|
properties, emptyProvider(), provider(retryService), () -> client);
|
||||||
|
|
||||||
|
// 补偿删除只对版本化 key(本次写入独占)生效,故这里用 versioned key 覆盖该路径
|
||||||
|
String firstKey = "task/a-11111111-2222-3333-4444-555555555555.json";
|
||||||
|
String secondKey = "task/b-66666666-7777-8888-9999-000000000000.json";
|
||||||
IllegalStateException firstFailure = assertThrows(IllegalStateException.class,
|
IllegalStateException firstFailure = assertThrows(IllegalStateException.class,
|
||||||
() -> service.uploadText("task/a.json", "{}", true));
|
() -> service.uploadText(firstKey, "{}", true));
|
||||||
IllegalStateException secondFailure = assertThrows(IllegalStateException.class,
|
IllegalStateException secondFailure = assertThrows(IllegalStateException.class,
|
||||||
() -> service.uploadText("task/b.json", "{}", true));
|
() -> service.uploadText(secondKey, "{}", true));
|
||||||
|
|
||||||
assertTrue(firstFailure.getMessage().contains("not visible"));
|
assertTrue(firstFailure.getMessage().contains("not visible"));
|
||||||
assertTrue(secondFailure.getMessage().contains("not visible"));
|
assertTrue(secondFailure.getMessage().contains("not visible"));
|
||||||
verify(client, times(5)).statObject(any(StatObjectArgs.class));
|
verify(client, times(5)).statObject(any(StatObjectArgs.class));
|
||||||
verify(retryService).enqueue(eq("task/a.json"), same(firstFailure));
|
verify(retryService).enqueue(eq(firstKey), same(firstFailure));
|
||||||
verify(retryService).enqueue(eq("task/b.json"), same(secondFailure));
|
verify(retryService).enqueue(eq(secondKey), same(secondFailure));
|
||||||
IllegalStateException rejected = assertThrows(IllegalStateException.class,
|
IllegalStateException rejected = assertThrows(IllegalStateException.class,
|
||||||
() -> service.uploadText("task/c.json", "{}", false));
|
() -> service.uploadText("task/c.json", "{}", false));
|
||||||
assertTrue(rejected.getMessage().contains("cooldown active"));
|
assertTrue(rejected.getMessage().contains("cooldown active"));
|
||||||
@@ -360,6 +363,9 @@ class RustfsObjectStorageServiceTest {
|
|||||||
|
|
||||||
@Test
|
@Test
|
||||||
void completedUploadIsEnqueuedWhenDeadlineExpiresAfterPut() throws Exception {
|
void completedUploadIsEnqueuedWhenDeadlineExpiresAfterPut() throws Exception {
|
||||||
|
// 只有版本化 key(本次写入独占)才允许补偿删除;共享 key 的守卫见
|
||||||
|
// RustfsUploadCompensationGuardTest#deterministicKeyDerivedFromShapeDoesNotEnqueueCompensation
|
||||||
|
String versionedKey = "task/chunk-1-0ffc254b-afad-4279-aab2-e85e3ff955e9.json";
|
||||||
TransientStorageProperties properties = configuredProperties();
|
TransientStorageProperties properties = configuredProperties();
|
||||||
properties.setOperationTimeoutSeconds(1);
|
properties.setOperationTimeoutSeconds(1);
|
||||||
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
@@ -372,11 +378,11 @@ class RustfsObjectStorageServiceTest {
|
|||||||
properties, emptyProvider(), provider(retryService), () -> client);
|
properties, emptyProvider(), provider(retryService), () -> client);
|
||||||
|
|
||||||
IllegalStateException failure = assertThrows(IllegalStateException.class,
|
IllegalStateException failure = assertThrows(IllegalStateException.class,
|
||||||
() -> service.uploadText("task/a.json", "{}", false));
|
() -> service.uploadText(versionedKey, "{}", false));
|
||||||
|
|
||||||
assertTrue(failure.getMessage().contains("operation timeout"));
|
assertTrue(failure.getMessage().contains("operation timeout"));
|
||||||
verify(client).putObject(any(PutObjectArgs.class));
|
verify(client).putObject(any(PutObjectArgs.class));
|
||||||
verify(retryService).enqueue(eq("task/a.json"), same(failure));
|
verify(retryService).enqueue(eq(versionedKey), same(failure));
|
||||||
}
|
}
|
||||||
|
|
||||||
private static TransientStorageProperties configuredProperties() {
|
private static TransientStorageProperties configuredProperties() {
|
||||||
|
|||||||
+217
@@ -0,0 +1,217 @@
|
|||||||
|
package com.nanri.aiimage.modules.file.service.object;
|
||||||
|
|
||||||
|
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||||
|
import io.minio.MinioClient;
|
||||||
|
import io.minio.PutObjectArgs;
|
||||||
|
import io.minio.StatObjectArgs;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.beans.factory.ObjectProvider;
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 上传失败补偿删除的准入守卫(2026-09-17 线上任务 28459 事故)。
|
||||||
|
*
|
||||||
|
* <p>事故形态:客户端重传复用**确定性 key**({@code chunk-462.json})。第一次上传 put 已成功、
|
||||||
|
* 但随后的 {@code verifyObjectVisible} 抖动 → 旧代码无条件
|
||||||
|
* {@code enqueueDeleteRetry(objectKey)};而删除重试队列对该 key 不再做任何引用校验
|
||||||
|
* ({@code RustfsDeleteRetryService} → {@code deleteObjectFromRetry} → {@code removeObject}),
|
||||||
|
* 于是把重传成功后 DB 行仍指向的对象删掉,下游读 chunk 直接 404,整个任务组装失败。
|
||||||
|
*
|
||||||
|
* <p>口径:只有**本次写入独占的对象**(版本化 key,末段以 UUID 结尾)才允许补偿删除;
|
||||||
|
* 确定性 key 会被重写复用,删它就可能删掉别的行正在引用的对象。
|
||||||
|
*/
|
||||||
|
class RustfsUploadCompensationGuardTest {
|
||||||
|
|
||||||
|
// ===== 正常路径 =====
|
||||||
|
|
||||||
|
/** 唯一 key(版本化)+ 验证失败 + 显式允许 → 补偿删除入队。 */
|
||||||
|
@Test
|
||||||
|
void uniqueKeyVerifyFailureEnqueuesCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-462-0ffc254b-afad-4279-aab2-e85e3ff955e9.json",
|
||||||
|
"{}".getBytes(), true, true));
|
||||||
|
|
||||||
|
verify(retryService).enqueue(eq("task/chunk-462-0ffc254b-afad-4279-aab2-e85e3ff955e9.json"),
|
||||||
|
any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 三参重载按 key 形态判定:版本化 key 自动允许补偿删除。 */
|
||||||
|
@Test
|
||||||
|
void versionedKeyDerivedFromShapeEnqueuesCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-77-d03bf3b0-abbb-469d-9bb8-a7c01c309477.json",
|
||||||
|
"{}".getBytes(), true));
|
||||||
|
|
||||||
|
verify(retryService).enqueue(anyString(), any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 事故根因:共享 key 必须不删 =====
|
||||||
|
|
||||||
|
/** 确定性(共享)key + 显式禁止 → 不入队(异常仍照旧抛出,失败归属不变)。 */
|
||||||
|
@Test
|
||||||
|
void sharedKeyVerifyFailureDoesNotEnqueueCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-462.json", "{}".getBytes(), true, false));
|
||||||
|
|
||||||
|
verify(retryService, never()).enqueue(anyString(), any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 三参重载按 key 形态判定:确定性 key 自动禁止补偿删除(本次事故的直接修复点)。 */
|
||||||
|
@Test
|
||||||
|
void deterministicKeyDerivedFromShapeDoesNotEnqueueCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-462.json", "{}".getBytes(), true));
|
||||||
|
|
||||||
|
verify(retryService, never()).enqueue(anyString(), any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** verify=false 上传即成功,不得产生任何补偿删除。 */
|
||||||
|
@Test
|
||||||
|
void verifyDisabledNeverEnqueuesCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
String key = service.uploadBytes("task/chunk-462-uuid-x.json", "{}".getBytes(), false, true);
|
||||||
|
|
||||||
|
assertEquals("task/chunk-462-uuid-x.json", key);
|
||||||
|
verify(retryService, never()).enqueue(anyString(), any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 异常路径 =====
|
||||||
|
|
||||||
|
/** put 从未成功过:对象本就不存在,不该产生删除动作(与开关无关)。 */
|
||||||
|
@Test
|
||||||
|
void putNeverCompletedDoesNotEnqueueCompensation() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
MinioClient client = mock(MinioClient.class);
|
||||||
|
when(client.putObject(any(PutObjectArgs.class))).thenThrow(new IOException("put failed"));
|
||||||
|
TransientStorageProperties properties = configuredProperties();
|
||||||
|
properties.setUploadMaxRetries(1);
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
properties, emptyProvider(), provider(retryService), () -> client);
|
||||||
|
|
||||||
|
assertThrows(RuntimeException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-462-uuid-x.json", "{}".getBytes(), true, true));
|
||||||
|
|
||||||
|
verify(retryService, never()).enqueue(anyString(), any(Throwable.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 补偿删除入队本身失败:不得把上传失败改写成入队异常,原异常必须继续上抛。 */
|
||||||
|
@Test
|
||||||
|
void enqueueFailureDoesNotMaskUploadFailure() throws Exception {
|
||||||
|
RustfsDeleteRetryService retryService = mock(RustfsDeleteRetryService.class);
|
||||||
|
org.mockito.Mockito.doThrow(new IllegalStateException("queue down"))
|
||||||
|
.when(retryService).enqueue(anyString(), any(Throwable.class));
|
||||||
|
RustfsObjectStorageService service = new RustfsObjectStorageService(
|
||||||
|
configuredProperties(), emptyProvider(), provider(retryService), RustfsUploadCompensationGuardTest::putOkButStatFails);
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class,
|
||||||
|
() -> service.uploadBytes("task/chunk-9-11111111-2222-3333-4444-555555555555.json",
|
||||||
|
"{}".getBytes(), true, true));
|
||||||
|
|
||||||
|
assertFalse(ex.getMessage() == null || ex.getMessage().contains("queue down"),
|
||||||
|
"入队异常不应成为对外错误;实际=" + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== key 形态判定的边界 =====
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void versionedKeyShapeIsRecognized() {
|
||||||
|
assertTrue(RustfsObjectStorageService.isVersionedObjectKey(
|
||||||
|
"task-chunk/appearance_patent/28459/hash/chunk-462-0ffc254b-afad-4279-aab2-e85e3ff955e9.json"));
|
||||||
|
assertTrue(RustfsObjectStorageService.isVersionedObjectKey(
|
||||||
|
"task-chunk/x/1/hash/0ffc254b-afad-4279-aab2-e85e3ff955e9.json"),
|
||||||
|
"整个 entryKey 就是 UUID 时也应识别为唯一 key");
|
||||||
|
assertTrue(RustfsObjectStorageService.isVersionedObjectKey(
|
||||||
|
"task-chunk/x/1/hash/chunk-1-0ffc254b-afad-4279-aab2-e85e3ff955e9"),
|
||||||
|
"无 .json 后缀但末段是 UUID,同样视为唯一");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void deterministicKeyShapeIsNotRecognized() {
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey("task-chunk/x/1/hash/chunk-462.json"));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey("task-parsed/x/1/hash/latest.json"));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey("task-result-payload/x/1/hash/28459.json"),
|
||||||
|
"提交号这类 key 会被重写复用,不能当成唯一 key");
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey(
|
||||||
|
"task-chunk/x/1/0ffc254b-afad-4279-aab2-e85e3ff955e9/chunk-462.json"),
|
||||||
|
"UUID 出现在中间段不算唯一——重写复用的判据只看末段");
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void malformedKeyShapeDoesNotThrowAndIsTreatedAsShared() {
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey(null));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey(""));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey(" "));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey("chunk-462.json"));
|
||||||
|
assertFalse(RustfsObjectStorageService.isVersionedObjectKey("-afad-4279-aab2-e85e3ff955e9.json".substring(0, 20)));
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 辅助 =====
|
||||||
|
|
||||||
|
private static MinioClient putOkButStatFails() {
|
||||||
|
MinioClient client = mock(MinioClient.class);
|
||||||
|
try {
|
||||||
|
when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
|
||||||
|
when(client.statObject(any(StatObjectArgs.class))).thenThrow(new IOException("stat 抖动"));
|
||||||
|
} catch (Exception ex) {
|
||||||
|
throw new IllegalStateException(ex);
|
||||||
|
}
|
||||||
|
return client;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static TransientStorageProperties configuredProperties() {
|
||||||
|
TransientStorageProperties properties = new TransientStorageProperties();
|
||||||
|
properties.setEndpoint("http://127.0.0.1:9000");
|
||||||
|
properties.setBucket("bucket");
|
||||||
|
properties.setAccessKeyId("ak");
|
||||||
|
properties.setAccessKeySecret("sk");
|
||||||
|
properties.setBaseRetryDelayMillis(0);
|
||||||
|
properties.setRetryJitterMillis(0);
|
||||||
|
return properties;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static <T> ObjectProvider<T> provider(T value) {
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
ObjectProvider<T> provider = mock(ObjectProvider.class);
|
||||||
|
when(provider.getIfAvailable()).thenReturn(value);
|
||||||
|
return provider;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static <T> ObjectProvider<T> emptyProvider() {
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
ObjectProvider<T> provider = mock(ObjectProvider.class);
|
||||||
|
when(provider.getIfAvailable()).thenReturn(null);
|
||||||
|
return provider;
|
||||||
|
}
|
||||||
|
}
|
||||||
+97
@@ -33,6 +33,7 @@ import java.nio.charset.StandardCharsets;
|
|||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
import java.util.concurrent.atomic.AtomicLong;
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
import java.util.concurrent.atomic.AtomicReference;
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
|
||||||
@@ -42,10 +43,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
|
|||||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
import static org.mockito.ArgumentMatchers.any;
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyLong;
|
||||||
import static org.mockito.ArgumentMatchers.anyString;
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
import static org.mockito.ArgumentMatchers.eq;
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
import static org.mockito.Mockito.doAnswer;
|
import static org.mockito.Mockito.doAnswer;
|
||||||
import static org.mockito.Mockito.lenient;
|
import static org.mockito.Mockito.lenient;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
import static org.mockito.Mockito.times;
|
import static org.mockito.Mockito.times;
|
||||||
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.verify;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
@@ -397,4 +400,98 @@ class SimilarAsinTaskServiceChunkMergeLimitTest {
|
|||||||
verify(taskChunkMapper, times(1)).update(any(), any());
|
verify(taskChunkMapper, times(1)).update(any(), any());
|
||||||
verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class));
|
verify(taskScopeStateMapper, times(0)).insert(any(TaskScopeStateEntity.class));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ===== CAS 冲突处理(2026-09-17 线上任务 28459 遗留项,与外观专利同一口径) =====
|
||||||
|
|
||||||
|
/** 冲突重试耗尽:**保留**最后一次写入的对象作读兜底,异常带出期望/当前哈希。 */
|
||||||
|
@Test
|
||||||
|
void casConflictExhaustedKeepsLastStoredObjectAsFallback() throws Exception {
|
||||||
|
TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A");
|
||||||
|
chunk.setPayloadHash("hashA");
|
||||||
|
stubConflictMerge(chunk, 0);
|
||||||
|
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(9004L);
|
||||||
|
Throwable ex = invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1")));
|
||||||
|
|
||||||
|
assertTrue(ex instanceof IllegalStateException, "实际: " + ex);
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1");
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2");
|
||||||
|
verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3");
|
||||||
|
assertTrue(ex.getMessage().contains("hashA"), "需带出期望哈希,实际: " + ex.getMessage());
|
||||||
|
assertTrue(ex.getMessage().contains("hashB"), "需带出当前哈希,实际: " + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 冲突后重试成功:被顶替的中间对象删除,最终对象写入行(不删)。 */
|
||||||
|
@Test
|
||||||
|
void casConflictThenSuccessDeletesSupersededObjects() throws Exception {
|
||||||
|
TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A");
|
||||||
|
chunk.setPayloadHash("hashA");
|
||||||
|
stubConflictMerge(chunk, 0, 0, 1);
|
||||||
|
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(9004L);
|
||||||
|
invokeMerge(service, task, "hashA", 1, List.of(row("r1", "B0A0000001", "标题1")));
|
||||||
|
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-1");
|
||||||
|
verify(transientPayloadStorageService).deletePayloadIfPresent("sibling-2");
|
||||||
|
verify(transientPayloadStorageService, never()).deletePayloadIfPresent("sibling-3");
|
||||||
|
verify(transientPayloadStorageService).deleteReplacedPayloadIfNeeded(eq("ptr:chunk-A"), eq("sibling-3"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 冲突时读回行上的当前哈希供定位(此前只有一句 conflict,线上无法定位)。 */
|
||||||
|
@Test
|
||||||
|
void casConflictReadsBackCurrentHashForDiagnostics() throws Exception {
|
||||||
|
TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A");
|
||||||
|
chunk.setPayloadHash("hashA");
|
||||||
|
stubConflictMerge(chunk, 0);
|
||||||
|
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(9004L);
|
||||||
|
invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1")));
|
||||||
|
|
||||||
|
verify(taskChunkMapper, times(3)).selectById(7L);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 读回行失败不得掩盖原始冲突:异常信息里给出可读标记。 */
|
||||||
|
@Test
|
||||||
|
void casConflictReadBackFailureDoesNotMaskConflict() throws Exception {
|
||||||
|
TaskChunkEntity chunk = chunk(7L, "hashA", 1, "ptr:chunk-A");
|
||||||
|
chunk.setPayloadHash("hashA");
|
||||||
|
stubConflictMerge(chunk, 0);
|
||||||
|
when(taskChunkMapper.selectById(anyLong())).thenThrow(new IllegalStateException("db down"));
|
||||||
|
|
||||||
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
|
task.setId(9004L);
|
||||||
|
Throwable ex = invokeMergeExpectingFailure(task, List.of(row("r1", "B0A0000001", "标题1")));
|
||||||
|
|
||||||
|
assertTrue(ex.getMessage().contains("读取失败"), "实际: " + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 反射调用会包一层 InvocationTargetException,取根因以便断言业务异常。 */
|
||||||
|
private Throwable invokeMergeExpectingFailure(FileTaskEntity task, List<SimilarAsinResultRowDto> rows) throws Exception {
|
||||||
|
try {
|
||||||
|
invokeMerge(service, task, "hashA", 1, rows);
|
||||||
|
return null;
|
||||||
|
} catch (java.lang.reflect.InvocationTargetException ex) {
|
||||||
|
return ex.getCause();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 冲突场景公共 stub:每次尝试写一个不同的版本化对象,update 按传入序列返回。 */
|
||||||
|
private void stubConflictMerge(TaskChunkEntity chunk, Integer... updateResults) throws Exception {
|
||||||
|
when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk));
|
||||||
|
when(taskChunkMapper.selectOne(any())).thenReturn(chunk);
|
||||||
|
when(transientPayloadStorageService.resolvePayload(eq("ptr:chunk-A"), anyString()))
|
||||||
|
.thenReturn(rowsJson(List.of(row("r0", "B0A0000000", "存量行"))));
|
||||||
|
AtomicInteger stores = new AtomicInteger();
|
||||||
|
when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||||
|
anyString(), any(), anyString(), any(), anyString()))
|
||||||
|
.thenAnswer(inv -> "sibling-" + stores.incrementAndGet());
|
||||||
|
when(taskChunkMapper.update(any(), any())).thenReturn(updateResults[0], java.util.Arrays.copyOfRange(updateResults, 1, updateResults.length));
|
||||||
|
|
||||||
|
TaskChunkEntity current = chunk(7L, "hashB", 1, "ptr:chunk-A");
|
||||||
|
current.setPayloadHash("hashB");
|
||||||
|
lenient().when(taskChunkMapper.selectById(7L)).thenReturn(current);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+9
-3
@@ -33,8 +33,13 @@ import static org.mockito.Mockito.when;
|
|||||||
/**
|
/**
|
||||||
* task-150:清理前引用检查契约(plan 09)。
|
* task-150:清理前引用检查契约(plan 09)。
|
||||||
* payload 被 biz_task_chunk / biz_task_scope_state 引用则不清理;
|
* payload 被 biz_task_chunk / biz_task_scope_state 引用则不清理;
|
||||||
* chunk 仅剩自身一行(count=1)不算共享引用;查询异常保守保留;
|
* chunk 命中任意行(含仅剩 1 行)都不算「可安全清理」;查询异常保守保留;
|
||||||
* 检查只读无副作用;候选值批量反查。
|
* 检查只读无副作用;候选值批量反查。
|
||||||
|
*
|
||||||
|
* <p>2026-09-17 线上任务 28459 之后收紧:原契约把 count==1 当作「调用方自己那行」放行删除,
|
||||||
|
* 但调用方无法证明那一行就是自己(如合并已把自己那行指向新对象、另一行仍指向旧对象时 count 恰为 1),
|
||||||
|
* 删除即造成该分片永久读不到 → 整单失败。物理删除本就约定在 DB 行删除之后执行,正常路径引用数必然为 0;
|
||||||
|
* 宁可留下孤儿对象(有保留期清理兜底),也不删掉可能仍被引用的对象。
|
||||||
*/
|
*/
|
||||||
@ExtendWith(MockitoExtension.class)
|
@ExtendWith(MockitoExtension.class)
|
||||||
class TaskPayloadReferenceCheckTest {
|
class TaskPayloadReferenceCheckTest {
|
||||||
@@ -100,13 +105,14 @@ class TaskPayloadReferenceCheckTest {
|
|||||||
verify(rustfsObjectStorageService, never()).deleteObject(any());
|
verify(rustfsObjectStorageService, never()).deleteObject(any());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** count==1 无法证明那一行就是调用方自己(可能正是别的行仍在用)→ 保守不删。 */
|
||||||
@Test
|
@Test
|
||||||
void singleChunkRowIsOwnRowNotSharedReference() {
|
void singleChunkRowBlocksDeleteBecauseOwnershipCannotBeProven() {
|
||||||
when(taskChunkMapper.selectCount(any())).thenReturn(1L);
|
when(taskChunkMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
service.deletePayloadIfPresent(RUSTFS_VALUE);
|
service.deletePayloadIfPresent(RUSTFS_VALUE);
|
||||||
|
|
||||||
verify(rustfsObjectStorageService).deleteObject("payload-key");
|
verify(rustfsObjectStorageService, never()).deleteObject(any());
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|||||||
+230
@@ -0,0 +1,230 @@
|
|||||||
|
package com.nanri.aiimage.modules.task.service;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.file.service.object.RustfsObjectStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyInt;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* chunk 载荷对象缺失时的「同槽位版本化兄弟对象」读兜底(2026-09-17 线上任务 28459)。
|
||||||
|
*
|
||||||
|
* <p>事故后 DB 行指向的普通 key({@code chunk-462.json})已被误删,但同槽位的版本化对象
|
||||||
|
* ({@code chunk-462-<uuid>.json})还在。兜底让读路径自动改用兄弟对象,任务即可按已有数据出结果,
|
||||||
|
* 而不是整单 FAILED——这是防止「删错对象 → 任务永久失败」的最后一道网。
|
||||||
|
*/
|
||||||
|
class TransientPayloadChunkSiblingFallbackTest {
|
||||||
|
|
||||||
|
private static final String DIR = "task-chunk/appearance_patent/28459/"
|
||||||
|
+ "2248d39710545b47b7c7035fc60c4e33924a16174abf65dd1c5a230918e6f61e/";
|
||||||
|
private static final String CHUNK_KEY = DIR + "chunk-462.json";
|
||||||
|
private static final String CHUNK_POINTER = "rustfs:" + CHUNK_KEY;
|
||||||
|
private static final String SIBLING_KEY = DIR + "chunk-462-0ffc254b-afad-4279-aab2-e85e3ff955e9.json";
|
||||||
|
|
||||||
|
// ===== 正常路径 =====
|
||||||
|
|
||||||
|
/** 对象在:直接读,不得触发列对象(正常读路径不能因为兜底变慢)。 */
|
||||||
|
@Test
|
||||||
|
void presentObjectIsReadWithoutListing() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenReturn(bytes("[\"原始\"]"));
|
||||||
|
|
||||||
|
String payload = service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed");
|
||||||
|
|
||||||
|
assertEquals("[\"原始\"]", payload);
|
||||||
|
verify(rustfs, never()).listObjectKeysNewestFirst(anyString(), anyInt());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 对象缺失 + 存在兄弟对象 → 读兄弟对象(本次事故的自动恢复路径)。 */
|
||||||
|
@Test
|
||||||
|
void missingChunkObjectFallsBackToVersionedSibling() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(DIR + "chunk-462-", 50)).thenReturn(List.of(SIBLING_KEY));
|
||||||
|
when(rustfs.readObjectBytes(SIBLING_KEY)).thenReturn(bytes("[\"合并后\"]"));
|
||||||
|
|
||||||
|
String payload = service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed");
|
||||||
|
|
||||||
|
assertEquals("[\"合并后\"]", payload);
|
||||||
|
verify(rustfs).listObjectKeysNewestFirst(DIR + "chunk-462-", 50);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 多个兄弟对象:取列表首个(实现约定按最后修改时间倒序,最新在前)。 */
|
||||||
|
@Test
|
||||||
|
void newestSiblingIsPreferred() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
String newest = DIR + "chunk-462-ffffffff-ffff-ffff-ffff-ffffffffffff.json";
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(anyString(), anyInt())).thenReturn(List.of(newest, SIBLING_KEY));
|
||||||
|
when(rustfs.readObjectBytes(newest)).thenReturn(bytes("[\"最新\"]"));
|
||||||
|
|
||||||
|
assertEquals("[\"最新\"]", service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed"));
|
||||||
|
verify(rustfs).readObjectBytes(newest);
|
||||||
|
verify(rustfs, never()).readObjectBytes(SIBLING_KEY);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 指针是 JSON 编码形式(DB 里的常见写法)时兜底同样生效。 */
|
||||||
|
@Test
|
||||||
|
void jsonQuotedPointerStillFallsBack() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(anyString(), anyInt())).thenReturn(List.of(SIBLING_KEY));
|
||||||
|
when(rustfs.readObjectBytes(SIBLING_KEY)).thenReturn(bytes("[\"带引号\"]"));
|
||||||
|
|
||||||
|
assertEquals("[\"带引号\"]",
|
||||||
|
service(rustfs).resolvePayload('"' + CHUNK_POINTER + '"', "read chunk failed"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 非指针内容(直接就是载荷 JSON)原样返回,不触碰对象存储。 */
|
||||||
|
@Test
|
||||||
|
void rawContentIsReturnedAsIs() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
|
||||||
|
assertEquals("[1,2,3]", service(rustfs).resolvePayload("[1,2,3]", "read chunk failed"));
|
||||||
|
verify(rustfs, never()).readObjectBytes(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 边界:不该兜底的场景 =====
|
||||||
|
|
||||||
|
/** 非 chunk 槽位(如 row-1.json)缺失 → 不兜底,照旧抛错(避免误配无关对象)。 */
|
||||||
|
@Test
|
||||||
|
void nonChunkSlotDoesNotFallBack() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
String key = "task-result-item/similar-asin/9/hash/row-1.json";
|
||||||
|
when(rustfs.readObjectBytes(key)).thenThrow(missingKey());
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload("rustfs:" + key, "read item failed"));
|
||||||
|
|
||||||
|
verify(rustfs, never()).listObjectKeysNewestFirst(anyString(), anyInt());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** parsed 槽位(latest.json)缺失 → 不兜底(该槽位没有版本化兄弟语义)。 */
|
||||||
|
@Test
|
||||||
|
void parsedSlotDoesNotFallBack() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
String key = "task-parsed/appearance_patent/28459/hash/latest.json";
|
||||||
|
when(rustfs.readObjectBytes(key)).thenThrow(missingKey());
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload("rustfs:" + key, "read parsed failed"));
|
||||||
|
|
||||||
|
verify(rustfs, never()).listObjectKeysNewestFirst(anyString(), anyInt());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 非缺失类失败(权限/网络)→ 不兜底,照旧上抛以便重试。 */
|
||||||
|
@Test
|
||||||
|
void nonMissingFailureDoesNotFallBack() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY))
|
||||||
|
.thenThrow(new IllegalStateException("Access Denied"));
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed"));
|
||||||
|
|
||||||
|
verify(rustfs, never()).listObjectKeysNewestFirst(anyString(), anyInt());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 异常路径 =====
|
||||||
|
|
||||||
|
/** 没有兄弟对象 → 抛错,且错误信息必须保留原 errorMessage 与原指针(便于定位)。 */
|
||||||
|
@Test
|
||||||
|
void noSiblingRethrowsWithOriginalPointer() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(anyString(), anyInt())).thenReturn(List.of());
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed"));
|
||||||
|
|
||||||
|
assertTrue(ex.getMessage().contains("read chunk failed"), "错误信息=" + ex.getMessage());
|
||||||
|
assertTrue(ex.getMessage().contains("chunk-462.json"), "错误信息=" + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 兄弟对象读取也失败 → 异常上抛,不得静默返回空内容。 */
|
||||||
|
@Test
|
||||||
|
void siblingReadFailurePropagates() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(anyString(), anyInt())).thenReturn(List.of(SIBLING_KEY));
|
||||||
|
when(rustfs.readObjectBytes(SIBLING_KEY)).thenThrow(new IllegalStateException("read timeout"));
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed"));
|
||||||
|
|
||||||
|
assertTrue(ex.getMessage().contains("read chunk failed"), "错误信息=" + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 列对象本身失败(桶不可用)→ 仍按原错误抛出,错误信息不含兄弟查找细节。 */
|
||||||
|
@Test
|
||||||
|
void listFailureDoesNotMaskOriginalError() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
when(rustfs.readObjectBytes(CHUNK_KEY)).thenThrow(missingKey());
|
||||||
|
when(rustfs.listObjectKeysNewestFirst(anyString(), anyInt()))
|
||||||
|
.thenThrow(new IllegalStateException("bucket unavailable"));
|
||||||
|
|
||||||
|
IllegalStateException ex = assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload(CHUNK_POINTER, "read chunk failed"));
|
||||||
|
|
||||||
|
assertTrue(ex.getMessage().contains("read chunk failed"), "错误信息=" + ex.getMessage());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** local 指针不走 rustfs 兄弟兜底。 */
|
||||||
|
@Test
|
||||||
|
void localPointerDoesNotUseChunkSiblingFallback() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
|
||||||
|
assertThrows(IllegalStateException.class,
|
||||||
|
() -> service(rustfs).resolvePayload("local:i/server-121/" + CHUNK_KEY, "read chunk failed"));
|
||||||
|
|
||||||
|
verify(rustfs, never()).listObjectKeysNewestFirst(anyString(), anyInt());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 辅助 =====
|
||||||
|
|
||||||
|
private static IllegalStateException missingKey() {
|
||||||
|
return new IllegalStateException("The specified key does not exist.");
|
||||||
|
}
|
||||||
|
|
||||||
|
private static byte[] bytes(String value) {
|
||||||
|
return value.getBytes(StandardCharsets.UTF_8);
|
||||||
|
}
|
||||||
|
|
||||||
|
private TransientPayloadStorageService service(RustfsObjectStorageService rustfs) {
|
||||||
|
TransientStorageProperties transientProperties = new TransientStorageProperties();
|
||||||
|
transientProperties.setEnabled(true);
|
||||||
|
transientProperties.setEndpoint("http://127.0.0.1:9000");
|
||||||
|
transientProperties.setBucket("json-server");
|
||||||
|
transientProperties.setAccessKeyId("ak");
|
||||||
|
transientProperties.setAccessKeySecret("sk");
|
||||||
|
StorageProperties storageProperties = new StorageProperties();
|
||||||
|
storageProperties.setLocalTempDir(System.getProperty("java.io.tmpdir"));
|
||||||
|
return new TransientPayloadStorageService(
|
||||||
|
transientProperties,
|
||||||
|
storageProperties,
|
||||||
|
rustfs,
|
||||||
|
mock(OssStorageService.class),
|
||||||
|
new ObjectMapper(),
|
||||||
|
new InstanceMetadata("server-121"),
|
||||||
|
mock(TaskChunkMapper.class),
|
||||||
|
mock(TaskScopeStateMapper.class));
|
||||||
|
}
|
||||||
|
}
|
||||||
+180
@@ -0,0 +1,180 @@
|
|||||||
|
package com.nanri.aiimage.modules.task.service;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import com.nanri.aiimage.config.InstanceMetadata;
|
||||||
|
import com.nanri.aiimage.config.StorageProperties;
|
||||||
|
import com.nanri.aiimage.config.TransientStorageProperties;
|
||||||
|
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
|
||||||
|
import com.nanri.aiimage.modules.file.service.object.RustfsObjectStorageService;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
|
||||||
|
import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.ArgumentMatchers.anyString;
|
||||||
|
import static org.mockito.ArgumentMatchers.eq;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.never;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 临时载荷删除前的引用守卫(2026-09-17 线上任务 28459 事故的第二条同族缺陷)。
|
||||||
|
*
|
||||||
|
* <p>守卫本意是「还有别的行引用同一 pointer 就不物理删」。原实现用
|
||||||
|
* {@code chunkCount > 1} 作判据,等价于「恰好还有 1 行引用时可以删」——
|
||||||
|
* 这在 caller 已经把自己那行挪走、但**另一行**仍指向旧对象的场景下,会把
|
||||||
|
* 对方仍在用的对象删掉。docs 里 CollectDataDeleteConsistencyTest 已明确口径是
|
||||||
|
* 「跨任务共享指针仍引用 → 跳过删除」,故判据收紧为「命中任意一行即不删」。
|
||||||
|
*/
|
||||||
|
class TransientPayloadDeleteReferenceGuardTest {
|
||||||
|
|
||||||
|
private static final String RUSTFS_CHUNK_POINTER =
|
||||||
|
"rustfs:task-chunk/appearance_patent/28459/2248d39710545b47b7c7035fc60c4e33924a16174abf65dd1c5a230918e6f61e/chunk-462.json";
|
||||||
|
|
||||||
|
// ===== 正常路径 =====
|
||||||
|
|
||||||
|
/** 没有任何行引用 → 允许物理删除。 */
|
||||||
|
@Test
|
||||||
|
void zeroReferencesAllowsPhysicalDelete() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
TaskScopeStateMapper scopeMapper = mock(TaskScopeStateMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
when(scopeMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, scopeMapper).deletePayloadIfPresent(RUSTFS_CHUNK_POINTER);
|
||||||
|
|
||||||
|
verify(rustfs).deleteObject(eq("task-chunk/appearance_patent/28459/2248d39710545b47b7c7035fc60c4e33924a16174abf65dd1c5a230918e6f61e/chunk-462.json"));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 多条行引用 → 不删(原有行为必须保持)。 */
|
||||||
|
@Test
|
||||||
|
void multipleReferencesBlockDelete() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(3L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class)).deletePayloadIfPresent(RUSTFS_CHUNK_POINTER);
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 本次修复点 =====
|
||||||
|
|
||||||
|
/** 恰好 1 行仍引用 → 必须不删(原实现会删掉这唯一一行仍在用的对象)。 */
|
||||||
|
@Test
|
||||||
|
void singleReferencingChunkRowBlocksDelete() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class)).deletePayloadIfPresent(RUSTFS_CHUNK_POINTER);
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** scope_state 单行引用同样阻断删除(该字段命中即非自我引用)。 */
|
||||||
|
@Test
|
||||||
|
void singleScopeStateReferenceBlocksDelete() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
TaskScopeStateMapper scopeMapper = mock(TaskScopeStateMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
when(scopeMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, scopeMapper).deletePayloadIfPresent(RUSTFS_CHUNK_POINTER);
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 边界 =====
|
||||||
|
|
||||||
|
/** 传入 JSON 编码(带引号)的指针——DB 里常见写法,必须能解析并同样受守卫约束。 */
|
||||||
|
@Test
|
||||||
|
void jsonQuotedPointerIsParsedAndStillGuarded() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(1L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class))
|
||||||
|
.deletePayloadIfPresent('"' + RUSTFS_CHUNK_POINTER + '"');
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 非指针值原样传入 → 不解析、不查库、不删。 */
|
||||||
|
@Test
|
||||||
|
void nonPointerValueIsIgnoredWithoutLookup() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
|
||||||
|
TransientPayloadStorageService service = service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class));
|
||||||
|
assertDoesNotThrow(() -> service.deletePayloadIfPresent("{\"rows\":[1,2,3]}"));
|
||||||
|
|
||||||
|
verify(chunkMapper, never()).selectCount(any());
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 空值不得抛异常、不得触发任何删除。 */
|
||||||
|
@Test
|
||||||
|
void nullOrBlankValueIsIgnored() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
TransientPayloadStorageService service = service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class));
|
||||||
|
|
||||||
|
assertDoesNotThrow(() -> service.deletePayloadIfPresent(null));
|
||||||
|
assertDoesNotThrow(() -> service.deletePayloadIfPresent(" "));
|
||||||
|
|
||||||
|
verify(chunkMapper, never()).selectCount(any());
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
// ===== 异常路径 =====
|
||||||
|
|
||||||
|
/** 反查引用的查询抛异常 → 保守不删(原实现即如此,必须保持)。 */
|
||||||
|
@Test
|
||||||
|
void referenceQueryFailureIsConservativeAndSkipsDelete() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenThrow(new IllegalStateException("db down"));
|
||||||
|
|
||||||
|
assertDoesNotThrow(() ->
|
||||||
|
service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class))
|
||||||
|
.deletePayloadIfPresent(RUSTFS_CHUNK_POINTER));
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 跨实例的 local 指针:不物理删(留给持有实例的清理任务)。 */
|
||||||
|
@Test
|
||||||
|
void crossInstanceLocalPointerIsNotDeleted() {
|
||||||
|
RustfsObjectStorageService rustfs = mock(RustfsObjectStorageService.class);
|
||||||
|
TaskChunkMapper chunkMapper = mock(TaskChunkMapper.class);
|
||||||
|
when(chunkMapper.selectCount(any())).thenReturn(0L);
|
||||||
|
|
||||||
|
service(rustfs, chunkMapper, mock(TaskScopeStateMapper.class))
|
||||||
|
.deletePayloadIfPresent("local:i/server-110/task-chunk/x/1/hash/chunk-1.json");
|
||||||
|
|
||||||
|
verify(rustfs, never()).deleteObject(anyString());
|
||||||
|
}
|
||||||
|
|
||||||
|
private TransientPayloadStorageService service(RustfsObjectStorageService rustfs,
|
||||||
|
TaskChunkMapper chunkMapper,
|
||||||
|
TaskScopeStateMapper scopeMapper) {
|
||||||
|
TransientStorageProperties transientProperties = new TransientStorageProperties();
|
||||||
|
transientProperties.setEnabled(true);
|
||||||
|
StorageProperties storageProperties = new StorageProperties();
|
||||||
|
storageProperties.setLocalTempDir(System.getProperty("java.io.tmpdir"));
|
||||||
|
return new TransientPayloadStorageService(
|
||||||
|
transientProperties,
|
||||||
|
storageProperties,
|
||||||
|
rustfs,
|
||||||
|
mock(OssStorageService.class),
|
||||||
|
new ObjectMapper(),
|
||||||
|
new InstanceMetadata("server-121"),
|
||||||
|
chunkMapper,
|
||||||
|
scopeMapper);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -24,6 +24,14 @@ export interface ClientChangelogEntry {
|
|||||||
|
|
||||||
/** 更新日志数据(新版本在前;发版时在数组最前追加一条) */
|
/** 更新日志数据(新版本在前;发版时在数组最前追加一条) */
|
||||||
export const CLIENT_CHANGELOG: ClientChangelogEntry[] = [
|
export const CLIENT_CHANGELOG: ClientChangelogEntry[] = [
|
||||||
|
{
|
||||||
|
version: '4.0.28',
|
||||||
|
date: '2026-09-17',
|
||||||
|
items: [
|
||||||
|
'修复上架、跟价等任务「多个任务同时操作同一个店铺导致打开店铺失败」的问题,同一店铺改为排队执行',
|
||||||
|
'紫鸟更新内核期间不再直接报「打开店铺失败」,会等待更新完成并显示等待进度',
|
||||||
|
],
|
||||||
|
},
|
||||||
{
|
{
|
||||||
version: '4.0.27',
|
version: '4.0.27',
|
||||||
date: '2026-09-17',
|
date: '2026-09-17',
|
||||||
|
|||||||
Reference in New Issue
Block a user