super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(临时载荷): 收口删除守卫 + chunk 读兜底，修「对象被误删→任务永久失败」

Browse Source 
线上任务 28459（外观专利）三个分片的载荷对象被删、DB 行仍指向已删的确定性
key，组装读不到 → 整单 FAILED。数据其实还在同槽位的版本化对象里，是行指错了。

三条收口：
- 上传补偿删除加准入：只有末段带 UUID 的版本化 key（本次写入独占）才允许入队。
  确定性 key 会被重传复用，删它就删掉了行仍在引用的对象；而删除队列
  （deleteObjectFromRetry -> removeObject）本身不做引用反查，准入必须卡在入队处。
- 引用守卫阈值 >1 收紧为 >0：原来把「恰好 1 行引用」当作调用方自己那行而放行，
  但调用方无法证明归属。物理删除本就约定在 DB 行删除之后执行，正常路径引用数
  必为 0；宁可留孤儿（有保留期清理兜底），也不删可能仍被引用的对象。
- chunk 读兜底：指针对象确已缺失（NoSuchKey）时回退同槽位版本化对象 chunk-N-*。
  仅限 chunk 槽位 + 确为缺失两个条件，避免误配无关对象或掩盖真实故障。

测试：新增 31 个用例（补偿删除准入 / 引用守卫 / 兄弟对象兜底），均先确认 RED
再实现；另更新 3 个断言旧行为的既有用例。全量 mvn test 3082 个 0 失败。
This commit is contained in:
黄自达
2026-09-17 10:42:53 +08:00
parent 1fe3368c5a
commit 3137299bfe
7 changed files with 812 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java
+97 -1
View File
@@ -5,11 +5,14 @@ import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.Timer;
import io.minio.GetObjectArgs;
import io.minio.ListObjectsArgs;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
import io.minio.RemoveObjectArgs;
import io.minio.Result;
import io.minio.StatObjectArgs;
import io.minio.errors.ErrorResponseException;
import io.minio.messages.Item;
import lombok.extern.slf4j.Slf4j;
import okhttp3.ConnectionPool;
import okhttp3.Dispatcher;
@@ -21,8 +24,11 @@ import org.springframework.stereotype.Service;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
@@ -45,6 +51,12 @@ public class RustfsObjectStorageService {
"NoSuchKey", "NoSuchBucket", "NoSuchVersion",
"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 ObjectProvider<MeterRegistry> meterRegistryProvider;
private final ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider;
@@ -106,7 +118,22 @@ public class RustfsObjectStorageService {
return uploadBytes(objectKey, bytes, verifyAfterUpload);
}
/**
* 三参重载：是否做「上传失败补偿删除」按对象 key 形态自动判定。
*
* <p>只有版本化 key（末段以 UUID 结尾）是本次写入独占的；确定性 key 会被重传重写复用，
* 删它就可能删掉别的 DB 行仍在引用的对象（2026-09-17 线上任务 28459 的载荷对象就是这么丢的）。
*/
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();
if (!isConfigured()) {
throw new IllegalStateException("transient storage is not configured");
@@ -136,13 +163,45 @@ public class RustfsObjectStorageService {
}
return uploadedObjectKey;
} catch (RuntimeException ex) {
if (putCompleted.get()) {
if (putCompleted.get() && compensateDeleteOnFailure) {
enqueueDeleteRetry(objectKey, ex);
} else if (putCompleted.get()) {
// 共享 key 会被重传重写：此处删除可能删掉别的行正在引用的对象，交给保留期清理兜底。
// 线上任务 28459 的 chunk-462/473/484 就是被这条无条件删除队列删掉的。
log.warn("[rustfs] 跳过上传失败补偿删除（对象非本次独占，可能被复用）objectKey={} err={}",
objectKey, ex.getMessage());
}
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) {
byte[] bytes = readObjectBytes(objectKey);
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) {
deleteObject(objectKey, true, operationDeadlineNanos());
}
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java
+67 -5
View File
@@ -30,6 +30,7 @@ import java.util.Locale;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.regex.Pattern;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
@@ -150,8 +151,20 @@ public class TransientPayloadStorageService {
return decodeStoredPayloadBytes(readLocalPayloadBytes(stripLocalInstanceId(localKey)));
}
if (pointer.startsWith(RUSTFS_POINTER_PREFIX)) {
return decodeStoredPayloadBytes(
rustfsObjectStorageService.readObjectBytes(pointer.substring(RUSTFS_POINTER_PREFIX.length())));
String objectKey = 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)) {
return decodeStoredPayloadBytes(
@@ -164,6 +177,53 @@ public class TransientPayloadStorageService {
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) {
String pointer = extractPointer(value);
if (pointer == null) {
@@ -209,8 +269,10 @@ public class TransientPayloadStorageService {
*
* <p>判断口径：
* <ul>
* <li>{@code biz_task_chunk.payload_json} 命中 &gt; 1 行（&gt; 1 表示除了 caller 视角下
* 自己即将释放的那一行之外，至少还有别的 chunk 行也指向同一对象）→ 视为仍被引用。</li>
* <li>{@code biz_task_chunk.payload_json} 命中任意行（&ge; 1）→ 视为仍被引用。
* 曾用 {@code > 1} 作判据，等于放行「恰好还有 1 行引用」的情况，会把对方仍在用的对象
* 删掉（2026-09-17 线上任务 28459：合并成功后指针未落库 + 旧对象被删 → 该分片永久读不到）。
* 物理删除本就约定在 DB 行删除之后执行，故调用方正常路径下引用数必然为 0。</li>
* <li>{@code biz_task_scope_state.parsed_payload_json}/{@code state_json} 命中 &gt; 0 行 →
* 视为仍被引用（这两个字段不是 caller 自身行的常见持有者，命中即非自我引用）。</li>
* </ul>
@@ -248,7 +310,7 @@ public class TransientPayloadStorageService {
// 解析不出 taskId 时按原口径全局查，行为与改造前一致。
Long pointerTaskId = extractTaskId(pointer);
Long chunkCount = referencedChunkCount(pointerTaskId, values);
if (chunkCount != null && chunkCount > 1L) {
if (chunkCount != null && chunkCount > 0L) {
return true;
}
Long scopeStateCount = referencedScopeStateCount(pointerTaskId, values);