From f4847f9ef7be3a1f5dbf8650276fd8e8209feffc Mon Sep 17 00:00:00 2001 From: let5sne Date: Sun, 12 Jul 2026 03:55:15 +0800 Subject: [PATCH] fix(personal): fence ingestion workers and cleanup --- .../service/PersonalIngestionService.java | 184 ++++++++++++++---- .../service/PersonalIngestionWorker.java | 81 +++++--- .../support/PersonalKnowledgeProperties.java | 1 + .../PersonalIngestionServiceTest.java | 133 ++++++++++--- .../personal/PersonalIngestionWorkerTest.java | 97 ++++++--- .../personal/PersonalSpaceServiceTest.java | 1 + 6 files changed, 378 insertions(+), 119 deletions(-) diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java index b91f1e1e..7cd419c3 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java @@ -44,6 +44,7 @@ import java.util.regex.Pattern; public class PersonalIngestionService { private static final String ITEM_NOT_FOUND = "PERSONAL_ITEM_NOT_FOUND"; + private static final int MIN_UPLOAD_CLEANUP_AGE_MINUTES = 5; private static final Pattern SAFE_TENANT = Pattern.compile("[A-Za-z0-9_-]{1,20}"); private static final Set SUPPORTED_FILE_SUFFIXES = Set.of( "txt", "md", "markdown", "pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx" @@ -77,6 +78,7 @@ public class PersonalIngestionService { this.objectStore = objectStore; this.idSupplier = idSupplier; this.phaseTransaction = phaseTransaction; + validateRecoveryWindows(properties); } public static PersonalIngestionService forTest(JdbcTemplate jdbcTemplate, PersonalSpaceService spaceService, @@ -144,8 +146,9 @@ public class PersonalIngestionService { long itemId = positiveId(idSupplier.getAsLong()); long ossId = positiveId(idSupplier.getAsLong()); String objectKey = objectKey(owner, itemId, suffix); + String uploadToken = UUID.randomUUID().toString(); UploadIntent draft = new UploadIntent(owner, 0L, itemId, ossId, objectKey, suffix, mimeType, - bytes.length, serviceKey); + bytes.length, serviceKey, uploadToken); String hash = sha256(bytes); PhaseOne phaseOne = phaseTransaction.execute(status -> phaseOne( @@ -158,13 +161,18 @@ public class PersonalIngestionService { } UploadIntent intent = phaseOne.intent(); + String url; try { - String url = objectStore.uploadPhysical(intent.serviceKey(), intent.objectKey(), mimeType, bytes); - UploadIntent finalIntent = intent; - phaseTransaction.executeWithoutResult(status -> activate(finalIntent, url)); + url = objectStore.uploadPhysical(intent.serviceKey(), intent.objectKey(), mimeType, bytes); + } catch (RuntimeException ex) { + beginCleanup(intent, null); + throw ex; + } + try { + activate(intent, url); return new ItemCreatedResponse(intent.itemId(), "QUEUED", null); } catch (RuntimeException ex) { - cleanupIntent(intent, false, null); + reconcileActivationFailure(intent); throw ex; } } @@ -184,7 +192,7 @@ public class PersonalIngestionService { create_time, create_by, update_time, update_by, service) values (?, ?, ?, ?, ?, '', ?, now(), ?, now(), ?, ?) """, intent.ossId(), intent.owner().tenantId(), intent.objectKey(), safeName, - "." + intent.suffix(), uploadExt(intent.itemId(), "PENDING"), intent.owner().userId(), + "." + intent.suffix(), uploadExt(intent, "PENDING"), intent.owner().userId(), intent.owner().userId(), intent.serviceKey()); int itemInserted = jdbcTemplate.update(""" insert into aihr_personal_item @@ -209,6 +217,23 @@ public class PersonalIngestionService { if (url == null || url.isBlank()) { throw new ServiceException("PERSONAL_OSS_UPLOAD_FAILED"); } + for (int attempt = 0; attempt < 2; attempt++) { + Integer activated = phaseTransaction.execute(status -> activateOnce(intent, url)); + if (activated != null && activated == 1) { + return; + } + UploadState current = uploadState(intent); + if (current == UploadState.READY) { + return; + } + if (current != UploadState.PENDING) { + throw new ServiceException("PERSONAL_OSS_ACTIVATION_FAILED"); + } + } + throw new ServiceException("PERSONAL_OSS_ACTIVATION_FAILED"); + } + + private int activateOnce(UploadIntent intent, String url) { int activated = jdbcTemplate.update(""" update sys_oss o join aihr_personal_item i on i.oss_id = o.oss_id and binary i.tenant_id = binary o.tenant_id @@ -216,43 +241,63 @@ public class PersonalIngestionService { set o.url = ?, o.ext1 = ?, o.update_time = now(), o.update_by = ? where o.tenant_id = ? and o.oss_id = ? and o.create_by = ? and o.file_name = ? and i.id = ? and i.status = 'QUEUED' + and json_unquote(json_extract(o.ext1, '$.uploadToken')) = ? and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'PENDING' - """, url, uploadExt(intent.itemId(), "READY"), intent.owner().userId(), - intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), intent.itemId()); - if (activated != 1) { - throw new ServiceException("PERSONAL_OSS_ACTIVATION_FAILED"); - } + """, url, uploadExt(intent, "READY"), intent.owner().userId(), + intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), intent.itemId(), + intent.uploadToken()); + return activated; } @Scheduled(fixedDelayString = "${aihr.personal.upload-cleanup-delay-ms:60000}") public void recoverStaleUploadIntents() { - LocalDateTime cutoff = LocalDateTime.now().minusMinutes(properties.getUploadCleanupAgeMinutes()); - List> rows = jdbcTemplate.queryForList(""" + LocalDateTime now = LocalDateTime.now(); + LocalDateTime pendingCutoff = now.minusMinutes(properties.getUploadCleanupAgeMinutes()); + for (Map row : staleUploadRows("PENDING", pendingCutoff)) { + beginCleanup(intent(row), pendingCutoff); + } + LocalDateTime cleaningCutoff = now.minusMinutes(properties.getCleanupFinalizeGraceMinutes()); + for (Map row : staleUploadRows("CLEANING", cleaningCutoff)) { + finalizeStaleCleanup(intent(row), cleaningCutoff); + } + } + + private List> staleUploadRows(String state, LocalDateTime cutoff) { + return jdbcTemplate.queryForList(""" select i.tenant_id, i.owner_user_id, i.space_id, i.id item_id, i.oss_id, - i.size_bytes, i.mime_type, o.file_name, o.service + i.size_bytes, i.mime_type, o.file_name, o.service, + json_unquote(json_extract(o.ext1, '$.uploadToken')) upload_token from aihr_personal_item i join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id and o.create_by = i.owner_user_id where i.status = 'QUEUED' and o.update_time < ? - and json_unquote(json_extract(o.ext1, '$.uploadState')) in ('PENDING', 'CLEANING') + and json_unquote(json_extract(o.ext1, '$.uploadState')) = ? + and json_unquote(json_extract(o.ext1, '$.uploadToken')) is not null order by o.update_time limit 20 - """, cutoff); - for (Map row : rows) { - UploadIntent intent = intent(row); - cleanupIntent(intent, true, cutoff); + """, cutoff, state); + } + + private void beginCleanup(UploadIntent intent, LocalDateTime cutoff) { + Integer claimed = phaseTransaction.execute(status -> claimState( + intent, UploadState.PENDING, UploadState.CLEANING, cutoff)); + if (claimed != null && claimed == 1) { + deleteKnownObject(intent); + return; + } + UploadState current = uploadState(intent); + if (current == UploadState.CLEANING || current == UploadState.MISSING) { + deleteKnownObject(intent); } } - private void cleanupIntent(UploadIntent intent, boolean includeCleaning, LocalDateTime cutoff) { - Integer claimed = phaseTransaction.execute(status -> claimCleanup(intent, includeCleaning, cutoff)); + private void finalizeStaleCleanup(UploadIntent intent, LocalDateTime cutoff) { + Integer claimed = phaseTransaction.execute(status -> claimState( + intent, UploadState.CLEANING, UploadState.CLEANING, cutoff)); if (claimed == null || claimed != 1) { return; } - try { - objectStore.deletePhysical(intent.serviceKey(), intent.objectKey()); - } catch (RuntimeException ex) { - log.warn("Personal upload-intent physical cleanup failed itemId={}", intent.itemId()); + if (!deleteKnownObject(intent)) { return; } try { @@ -262,22 +307,63 @@ public class PersonalIngestionService { } } - private int claimCleanup(UploadIntent intent, boolean includeCleaning, LocalDateTime cutoff) { - String states = includeCleaning ? "('PENDING', 'CLEANING')" : "('PENDING')"; + private int claimState(UploadIntent intent, UploadState expected, UploadState target, LocalDateTime cutoff) { String cutoffClause = cutoff == null ? "" : " and update_time < ?"; String sql = """ update sys_oss - set ext1 = json_set(ext1, '$.uploadState', 'CLEANING'), update_time = now() + set ext1 = json_set(ext1, '$.uploadState', ?), update_time = now() where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? and json_unquote(json_extract(ext1, '$.itemId')) = ? - and json_unquote(json_extract(ext1, '$.uploadState')) in %s%s - """.formatted(states, cutoffClause); + and json_unquote(json_extract(ext1, '$.uploadToken')) = ? + and json_unquote(json_extract(ext1, '$.uploadState')) = ?%s + """.formatted(cutoffClause); if (cutoff == null) { - return jdbcTemplate.update(sql, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), - intent.objectKey(), String.valueOf(intent.itemId())); + return jdbcTemplate.update(sql, target.name(), intent.owner().tenantId(), intent.ossId(), + intent.owner().userId(), intent.objectKey(), String.valueOf(intent.itemId()), intent.uploadToken(), + expected.name()); + } + return jdbcTemplate.update(sql, target.name(), intent.owner().tenantId(), intent.ossId(), + intent.owner().userId(), intent.objectKey(), String.valueOf(intent.itemId()), intent.uploadToken(), + expected.name(), cutoff); + } + + private void reconcileActivationFailure(UploadIntent intent) { + UploadState current = uploadState(intent); + if (current == UploadState.PENDING) { + beginCleanup(intent, null); + } else if (current == UploadState.CLEANING || current == UploadState.MISSING) { + deleteKnownObject(intent); + } + } + + private UploadState uploadState(UploadIntent intent) { + List> rows = jdbcTemplate.queryForList(""" + select json_unquote(json_extract(ext1, '$.uploadState')) upload_state + from sys_oss + where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? + and json_unquote(json_extract(ext1, '$.itemId')) = ? + and json_unquote(json_extract(ext1, '$.uploadToken')) = ? + limit 1 + """, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), + String.valueOf(intent.itemId()), intent.uploadToken()); + if (rows.isEmpty()) { + return UploadState.MISSING; + } + try { + return UploadState.valueOf(String.valueOf(rows.get(0).get("upload_state"))); + } catch (IllegalArgumentException ex) { + throw new ServiceException("PERSONAL_UPLOAD_STATE_INVALID"); + } + } + + private boolean deleteKnownObject(UploadIntent intent) { + try { + objectStore.deletePhysical(intent.serviceKey(), intent.objectKey()); + return true; + } catch (RuntimeException ex) { + log.warn("Personal upload-intent physical cleanup failed itemId={}", intent.itemId()); + return false; } - return jdbcTemplate.update(sql, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), - intent.objectKey(), String.valueOf(intent.itemId()), cutoff); } private void finalizeCleanup(UploadIntent intent) { @@ -304,9 +390,10 @@ public class PersonalIngestionService { delete from sys_oss where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? and json_unquote(json_extract(ext1, '$.itemId')) = ? + and json_unquote(json_extract(ext1, '$.uploadToken')) = ? and json_unquote(json_extract(ext1, '$.uploadState')) = 'CLEANING' """, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), - String.valueOf(intent.itemId())); + String.valueOf(intent.itemId()), intent.uploadToken()); if (itemDeleted != 1 || counterUpdated != 1 || ossDeleted != 1) { throw new ServiceException("PERSONAL_UPLOAD_CLEANUP_FAILED"); } @@ -323,10 +410,11 @@ public class PersonalIngestionService { return new ItemCreatedResponse(id, String.valueOf(rows.get(0).get("status")), id); } - private String uploadExt(long itemId, String state) { + private String uploadExt(UploadIntent intent, String state) { try { return objectMapper.writeValueAsString(Map.of( - "source", "personal", "itemId", itemId, "uploadState", state)); + "source", "personal", "itemId", intent.itemId(), "uploadState", state, + "uploadToken", intent.uploadToken())); } catch (JsonProcessingException ex) { throw new ServiceException("PERSONAL_OSS_BIND_FAILED"); } @@ -346,7 +434,8 @@ public class PersonalIngestionService { PersonalOwner owner = new PersonalOwner(String.valueOf(row.get("tenant_id")), number(row, "owner_user_id"), null); return new UploadIntent(owner, number(row, "space_id"), number(row, "item_id"), number(row, "oss_id"), String.valueOf(row.get("file_name")), suffix(String.valueOf(row.get("file_name"))), - String.valueOf(row.get("mime_type")), number(row, "size_bytes"), String.valueOf(row.get("service"))); + String.valueOf(row.get("mime_type")), number(row, "size_bytes"), String.valueOf(row.get("service")), + String.valueOf(row.get("upload_token"))); } private static long number(Map row, String key) { @@ -376,6 +465,13 @@ public class PersonalIngestionService { return template; } + private static void validateRecoveryWindows(PersonalKnowledgeProperties properties) { + if (properties.getUploadCleanupAgeMinutes() < MIN_UPLOAD_CLEANUP_AGE_MINUTES + || properties.getCleanupFinalizeGraceMinutes() <= 0) { + throw new IllegalArgumentException("invalid personal upload recovery windows"); + } + } + private static String objectKey(PersonalOwner owner, long itemId, String suffix) { validateOwner(owner); positiveId(itemId); String safeSuffix = suffix == null ? "" : suffix.toLowerCase(Locale.ROOT); @@ -447,7 +543,9 @@ public class PersonalIngestionService { return result.getUrl(); } @Override public void deletePhysical(String serviceKey, String objectKey) { - OssClient storage = clients.get(serviceKey); requirePrivate(storage); storage.delete(objectKey); + OssClient storage = clients.get(serviceKey); + if (storage == null) throw new ServiceException("PERSONAL_OSS_UNAVAILABLE"); + storage.delete(objectKey); } } @@ -459,10 +557,14 @@ public class PersonalIngestionService { } private record PhaseOne(ItemCreatedResponse duplicate, UploadIntent intent) {} + private enum UploadState { PENDING, READY, CLEANING, MISSING } + private record UploadIntent(PersonalOwner owner, long spaceId, long itemId, long ossId, String objectKey, - String suffix, String mimeType, long sizeBytes, String serviceKey) { + String suffix, String mimeType, long sizeBytes, String serviceKey, + String uploadToken) { private UploadIntent withSpaceId(long value) { - return new UploadIntent(owner, value, itemId, ossId, objectKey, suffix, mimeType, sizeBytes, serviceKey); + return new UploadIntent(owner, value, itemId, ossId, objectKey, suffix, mimeType, sizeBytes, serviceKey, + uploadToken); } } } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java index e90d0c31..17e64bf0 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionWorker.java @@ -87,29 +87,44 @@ public class PersonalIngestionWorker { public void recoverStaleParsing() { java.time.LocalDateTime cutoff = java.time.LocalDateTime.now().minusMinutes(parsingLeaseMinutes); - jdbcTemplate.update(""" - update aihr_personal_item i - join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id + List> stale = jdbcTemplate.queryForList(""" + select i.id, i.tenant_id, i.owner_user_id, i.attempt_count + from aihr_personal_item i + join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id and o.create_by = i.owner_user_id - set i.status = 'FAILED', i.error_code = 'PERSONAL_PARSE_RETRY_EXHAUSTED', - i.error_message = '资料处理重试次数已用尽', i.update_time = now() - where i.status = 'PARSING' and i.attempt_count >= ? and i.update_time < ? + where i.status = 'PARSING' and i.update_time < ? and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'READY' - """, maxParseAttempts, cutoff); - jdbcTemplate.update(""" - update aihr_personal_item i - join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id - and o.create_by = i.owner_user_id - set i.status = 'QUEUED', i.error_code = null, i.error_message = null, i.update_time = now() - where i.status = 'PARSING' and i.attempt_count < ? and i.update_time < ? - and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'READY' - """, maxParseAttempts, cutoff); + order by i.update_time + limit 100 + """, cutoff); + for (Map row : stale) { + long id = number(row, "id"); + String tenantId = String.valueOf(row.get("tenant_id")); + long ownerUserId = number(row, "owner_user_id"); + int attemptVersion = Math.toIntExact(number(row, "attempt_count")); + if (attemptVersion >= maxParseAttempts) { + jdbcTemplate.update(""" + update aihr_personal_item + set status = 'FAILED', error_code = 'PERSONAL_PARSE_RETRY_EXHAUSTED', + error_message = '资料处理重试次数已用尽', update_time = now() + where tenant_id = ? and owner_user_id = ? and id = ? + and status = 'PARSING' and attempt_count = ? and update_time < ? + """, tenantId, ownerUserId, id, attemptVersion, cutoff); + } else { + jdbcTemplate.update(""" + update aihr_personal_item + set status = 'QUEUED', error_code = null, error_message = null, update_time = now() + where tenant_id = ? and owner_user_id = ? and id = ? + and status = 'PARSING' and attempt_count = ? and update_time < ? + """, tenantId, ownerUserId, id, attemptVersion, cutoff); + } + } } public boolean processNext() { List> queued = jdbcTemplate.queryForList(""" select i.id, i.tenant_id, i.space_id, i.owner_user_id, i.source_type, i.title, - i.oss_id, i.mime_type, i.tags_json, i.captured_at + i.oss_id, i.mime_type, i.tags_json, i.captured_at, i.attempt_count from aihr_personal_item i join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id and o.create_by = i.owner_user_id @@ -127,15 +142,16 @@ public class PersonalIngestionWorker { set status = 'PARSING', attempt_count = attempt_count + 1, error_code = null, error_message = null, update_time = now() where tenant_id = ? and owner_user_id = ? and id = ? - and status = 'QUEUED' + and status = 'QUEUED' and attempt_count = ? and exists (select 1 from sys_oss o where o.oss_id = aihr_personal_item.oss_id and binary o.tenant_id = binary aihr_personal_item.tenant_id and o.create_by = aihr_personal_item.owner_user_id and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'READY') - """, item.tenantId(), item.ownerUserId(), item.id()); + """, item.tenantId(), item.ownerUserId(), item.id(), item.attemptCount()); if (claimed != 1) { return false; } + int attemptVersion = Math.addExact(item.attemptCount(), 1); try { StoredObject stored = objectReader.read( @@ -147,7 +163,7 @@ public class PersonalIngestionWorker { KnowledgeDocumentParser.Failure.EMPTY, "document contains no text"); } transactionTemplate.execute(status -> { - persistSuccess(item, document, chunks); + persistSuccess(item, attemptVersion, document, chunks); return null; }); } catch (Exception ex) { @@ -156,15 +172,26 @@ public class PersonalIngestionWorker { update aihr_personal_item set status = 'FAILED', error_code = ?, error_message = ?, update_time = now() where tenant_id = ? and owner_user_id = ? and id = ? - and status = 'PARSING' - """, failure.code(), failure.message(), item.tenantId(), item.ownerUserId(), item.id()); + and status = 'PARSING' and attempt_count = ? + """, failure.code(), failure.message(), item.tenantId(), item.ownerUserId(), item.id(), + attemptVersion); log.warn("Personal ingestion failed itemId={} ownerUserId={} code={}", item.id(), item.ownerUserId(), failure.code()); } return true; } - private void persistSuccess(Item item, ParsedDocument document, List chunks) { + private void persistSuccess(Item item, int attemptVersion, ParsedDocument document, List chunks) { + Map locked = jdbcTemplate.queryForMap(""" + select status, attempt_count + from aihr_personal_item + where tenant_id = ? and owner_user_id = ? and id = ? + for update + """, item.tenantId(), item.ownerUserId(), item.id()); + if (!"PARSING".equals(String.valueOf(locked.get("status"))) + || number(locked, "attempt_count") != attemptVersion) { + throw new IllegalStateException("personal item attempt changed while parsing"); + } jdbcTemplate.update(""" delete from aihr_personal_fragment where tenant_id = ? and owner_user_id = ? and item_id = ? @@ -197,8 +224,9 @@ public class PersonalIngestionWorker { set status = 'READY', parsed_at = now(), summary = ?, tags_json = ?, error_code = null, error_message = null, update_time = now() where tenant_id = ? and owner_user_id = ? and id = ? - and status = 'PARSING' - """, summary(document.text()), item.tagsJson(), item.tenantId(), item.ownerUserId(), item.id()); + and status = 'PARSING' and attempt_count = ? + """, summary(document.text()), item.tagsJson(), item.tenantId(), item.ownerUserId(), item.id(), + attemptVersion); if (updated != 1) { throw new IllegalStateException("personal item state changed while parsing"); } @@ -239,7 +267,8 @@ public class PersonalIngestionWorker { String.valueOf(row.get("title")), number(row, "oss_id"), String.valueOf(row.get("mime_type")), - tags == null ? "[]" : String.valueOf(tags) + tags == null ? "[]" : String.valueOf(tags), + Math.toIntExact(number(row, "attempt_count")) ); } @@ -330,7 +359,7 @@ public class PersonalIngestionWorker { } private record Item(long id, String tenantId, long spaceId, long ownerUserId, String sourceType, - String title, long ossId, String mimeType, String tagsJson) { + String title, long ossId, String mimeType, String tagsJson, int attemptCount) { } private record Failure(String code, String message) { diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java index 81e7720a..378bce29 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java @@ -22,4 +22,5 @@ public class PersonalKnowledgeProperties { private int parsingLeaseMinutes = 15; private int maxParseAttempts = 3; private int uploadCleanupAgeMinutes = 15; + private int cleanupFinalizeGraceMinutes = 15; } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java index f1c41628..242ba6b0 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java @@ -66,7 +66,7 @@ class PersonalIngestionServiceTest { return "https://private.invalid/" + invocation.getArgument(1); }); when(fixture.jdbc.update(contains("set o.url ="), anyString(), anyString(), eq(101L), eq("000000"), - eq(101L), eq(101L), anyString(), eq(100L))).thenReturn(1); + eq(101L), eq(101L), anyString(), eq(100L), anyString())).thenReturn(1); ItemCreatedResponse response = fixture.service.createText(OWNER, new TextItemRequest("周报", "保洁巡检记录", null, List.of("保洁"))); @@ -81,11 +81,13 @@ class PersonalIngestionServiceTest { eq(key.getValue()), originalName.capture(), eq(".txt"), pendingExt.capture(), eq(101L), eq(101L), eq("personal-private")); assertTrue(originalName.getValue().matches("[0-9a-f]{32}\\.txt")); - assertUploadState(pendingExt.getValue(), "PENDING"); + JsonNode pending = uploadExt(pendingExt.getValue(), "PENDING"); + String uploadToken = pending.path("uploadToken").asText(); + assertTrue(uploadToken.matches("[0-9a-f-]{36}")); ArgumentCaptor readyExt = ArgumentCaptor.forClass(String.class); verify(fixture.jdbc).update(contains("set o.url ="), anyString(), readyExt.capture(), eq(101L), - eq("000000"), eq(101L), eq(101L), eq(key.getValue()), eq(100L)); - assertUploadState(readyExt.getValue(), "READY"); + eq("000000"), eq(101L), eq(101L), eq(key.getValue()), eq(100L), eq(uploadToken)); + assertEquals(uploadToken, uploadExt(readyExt.getValue(), "READY").path("uploadToken").asText()); assertEquals(2, fixture.transactions.commits); assertEquals(0, fixture.transactions.rollbacks); } @@ -97,7 +99,7 @@ class PersonalIngestionServiceTest { when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) .thenReturn("https://private.invalid/object"); when(fixture.jdbc.update(contains("set o.url ="), anyString(), anyString(), eq(101L), eq("000000"), - eq(101L), eq(101L), anyString(), eq(100L))).thenReturn(1); + eq(101L), eq(101L), anyString(), eq(100L), anyString())).thenReturn(1); MockMultipartFile file = new MockMultipartFile("file", "13800138000-secret.txt", "text/plain", "secret".getBytes(StandardCharsets.UTF_8)); @@ -132,16 +134,14 @@ class PersonalIngestionServiceTest { } @Test - void uploadFailureClaimsCleaningDeletesPhysicalAndReversesCountersOnce() { + void uploadFailureClaimsCleaningAndDeletesPhysicalButRetainsDurableIntent() { Fixture fixture = fixture(); stubPhaseOne(fixture, 4L); AtomicBoolean deleteInTransaction = new AtomicBoolean(true); when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) .thenThrow(new ServiceException("PERSONAL_OSS_UPLOAD_FAILED")); - when(fixture.jdbc.update(contains("json_set"), eq("000000"), eq(101L), eq(101L), anyString(), eq("100"))) - .thenReturn(1); - when(fixture.spaces.lockForUpdate(any(PersonalOwner.class))).thenReturn(7L); - stubFinalizeCleanup(fixture); + when(fixture.jdbc.update(contains("json_set"), eq("CLEANING"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), anyString(), eq("PENDING"))).thenReturn(1); org.mockito.Mockito.doAnswer(invocation -> { deleteInTransaction.set(TransactionSynchronizationManager.isActualTransactionActive()); return null; @@ -152,13 +152,12 @@ class PersonalIngestionServiceTest { assertEquals("PERSONAL_OSS_UPLOAD_FAILED", error.getMessage()); assertFalse(deleteInTransaction.get()); - verify(fixture.jdbc).update(contains("status = 'DELETED'"), eq("000000"), eq(101L), eq(7L), eq(100L), - eq(101L)); - verify(fixture.jdbc).update(contains("used_bytes = used_bytes -"), eq(4L), eq("000000"), eq(101L), - eq(7L), eq(4L)); - verify(fixture.jdbc).update(contains("delete from sys_oss"), eq("000000"), eq(101L), eq(101L), - anyString(), eq("100")); - assertEquals(3, fixture.transactions.commits); + verify(fixture.jdbc, never()).update(contains("status = 'DELETED'"), any(), any(), any(), any(), any()); + verify(fixture.jdbc, never()).update(contains("used_bytes = used_bytes -"), any(), any(), any(), any(), + any()); + verify(fixture.jdbc, never()).update(contains("delete from sys_oss"), any(), any(), any(), any(), any(), + any()); + assertEquals(2, fixture.transactions.commits); assertEquals(0, fixture.transactions.rollbacks); } @@ -166,17 +165,24 @@ class PersonalIngestionServiceTest { void staleCleanupIsIdempotentAndDoesNotDecrementCountersTwice() { Fixture fixture = fixture(); Map stale = staleIntent(); - when(fixture.jdbc.queryForList(contains("uploadState')) in ('PENDING', 'CLEANING')"), - any(LocalDateTime.class))).thenReturn(List.of(stale)); - when(fixture.jdbc.update(contains("json_set"), eq("000000"), eq(101L), eq(101L), eq("personal/key.txt"), - eq("100"), any(LocalDateTime.class))).thenReturn(1, 0); + when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("PENDING"))) + .thenReturn(List.of(stale), List.of(), List.of()); + when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("CLEANING"))) + .thenReturn(List.of(), List.of(stale), List.of(stale)); + when(fixture.jdbc.update(contains("json_set"), eq("CLEANING"), eq("000000"), eq(101L), eq(101L), + eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("PENDING"), any(LocalDateTime.class))) + .thenReturn(1); + when(fixture.jdbc.update(contains("json_set"), eq("CLEANING"), eq("000000"), eq(101L), eq(101L), + eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("CLEANING"), any(LocalDateTime.class))) + .thenReturn(1, 0); when(fixture.spaces.lockForUpdate(any(PersonalOwner.class))).thenReturn(7L); stubFinalizeCleanup(fixture); + fixture.service.recoverStaleUploadIntents(); fixture.service.recoverStaleUploadIntents(); fixture.service.recoverStaleUploadIntents(); - verify(fixture.store, times(1)).deletePhysical("personal-private", "personal/key.txt"); + verify(fixture.store, times(2)).deletePhysical("personal-private", "personal/key.txt"); verify(fixture.jdbc, times(1)).update(contains("used_bytes = used_bytes -"), eq(4L), eq("000000"), eq(101L), eq(7L), eq(4L)); } @@ -223,7 +229,7 @@ class PersonalIngestionServiceTest { } @Test - void publicStorageIsRejectedAndPrivateStorageUploadsByPhysicalKey() { + void uploadRequiresPrivatePolicyButCleanupSurvivesPolicyDrift() { PersonalKnowledgeProperties properties = new PersonalKnowledgeProperties(); PersonalIngestionService.OssClientProvider clients = mock(PersonalIngestionService.OssClientProvider.class); OssClient publicClient = mock(OssClient.class); @@ -232,8 +238,8 @@ class PersonalIngestionServiceTest { PersonalObjectStore publicStore = PersonalIngestionService.objectStoreForTest(properties, clients); assertEquals("PERSONAL_OSS_NOT_PRIVATE", assertThrows(ServiceException.class, publicStore::requirePrivateService).getMessage()); - assertEquals("PERSONAL_OSS_NOT_PRIVATE", assertThrows(ServiceException.class, - () -> publicStore.deletePhysical("", "personal/key.txt")).getMessage()); + publicStore.deletePhysical("", "personal/key.txt"); + verify(publicClient).delete("personal/key.txt"); verify(publicClient, never()).upload(any(java.io.InputStream.class), anyString(), anyLong(), anyString()); properties.setOssConfigKey(" personal-private "); @@ -251,7 +257,74 @@ class PersonalIngestionServiceTest { verify(privateClient).upload(any(java.io.InputStream.class), eq("personal/key.txt"), eq(1L), eq("text/plain")); } + @Test + void alreadyReadyActivationIsIdempotentAndNeverDeletesConfirmedObject() { + Fixture fixture = fixture(); + stubPhaseOne(fixture, 4L); + when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) + .thenReturn("https://private.invalid/object"); + when(fixture.jdbc.queryForList(contains("select json_unquote"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), anyString())).thenReturn(List.of(Map.of("upload_state", "READY"))); + + ItemCreatedResponse response = fixture.service.createFile(OWNER, + new MockMultipartFile("file", "notes.txt", "text/plain", new byte[]{1, 2, 3, 4}), null, null); + + assertEquals(100L, response.itemId()); + verify(fixture.store, never()).deletePhysical(anyString(), anyString()); + } + + @Test + void pendingActivationRetriesWithSameTokenBeforeCleanup() { + Fixture fixture = fixture(); + stubPhaseOne(fixture, 4L); + when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) + .thenReturn("https://private.invalid/object"); + when(fixture.jdbc.update(contains("set o.url ="), anyString(), anyString(), eq(101L), eq("000000"), + eq(101L), eq(101L), anyString(), eq(100L), anyString())).thenReturn(0, 1); + when(fixture.jdbc.queryForList(contains("select json_unquote"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), anyString())).thenReturn(List.of(Map.of("upload_state", "PENDING"))); + + ItemCreatedResponse response = fixture.service.createFile(OWNER, + new MockMultipartFile("file", "notes.txt", "text/plain", new byte[]{1, 2, 3, 4}), null, null); + + assertEquals(100L, response.itemId()); + verify(fixture.jdbc, times(2)).update(contains("set o.url ="), anyString(), anyString(), eq(101L), + eq("000000"), eq(101L), eq(101L), anyString(), eq(100L), anyString()); + verify(fixture.store, never()).deletePhysical(anyString(), anyString()); + } + + @Test + void cleaningActivationFailureDeletesKnownObjectWithoutReclaimingIntent() { + Fixture fixture = fixture(); + stubPhaseOne(fixture, 4L); + when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) + .thenReturn("https://private.invalid/object"); + when(fixture.jdbc.queryForList(contains("select json_unquote"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), anyString())).thenReturn(List.of(Map.of("upload_state", "CLEANING"))); + + assertThrows(ServiceException.class, () -> fixture.service.createFile(OWNER, + new MockMultipartFile("file", "notes.txt", "text/plain", new byte[]{1, 2, 3, 4}), null, null)); + + verify(fixture.store).deletePhysical(eq("personal-private"), anyString()); + verify(fixture.jdbc, never()).update(contains("json_set"), any(), any(), any(), any(), any(), any(), any(), + any()); + } + + @Test + void unsafeCleanupWindowsAreRejected() { + PersonalKnowledgeProperties properties = new PersonalKnowledgeProperties(); + properties.setUploadCleanupAgeMinutes(4); + assertThrows(IllegalArgumentException.class, () -> fixture(properties)); + properties.setUploadCleanupAgeMinutes(15); + properties.setCleanupFinalizeGraceMinutes(0); + assertThrows(IllegalArgumentException.class, () -> fixture(properties)); + } + private static Fixture fixture() { + return fixture(new PersonalKnowledgeProperties()); + } + + private static Fixture fixture(PersonalKnowledgeProperties properties) { JdbcTemplate jdbc = mock(JdbcTemplate.class); PersonalSpaceService spaces = mock(PersonalSpaceService.class); PersonalObjectStore store = mock(PersonalObjectStore.class); @@ -260,7 +333,7 @@ class PersonalIngestionServiceTest { TransactionTemplate template = new TransactionTemplate(transactions); template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); PersonalIngestionService service = PersonalIngestionService.forTest(jdbc, spaces, - new PersonalKnowledgeProperties(), new ObjectMapper(), store, + properties, new ObjectMapper(), store, new AtomicLong(100L)::getAndIncrement, template); return new Fixture(jdbc, spaces, store, service, transactions); } @@ -284,22 +357,24 @@ class PersonalIngestionServiceTest { when(fixture.jdbc.update(contains("used_bytes = used_bytes -"), eq(4L), eq("000000"), eq(101L), eq(7L), eq(4L))).thenReturn(1); when(fixture.jdbc.update(contains("delete from sys_oss"), eq("000000"), eq(101L), eq(101L), - anyString(), eq("100"))).thenReturn(1); + anyString(), eq("100"), eq("upload-token"))).thenReturn(1); } private static Map staleIntent() { return Map.of( "tenant_id", "000000", "owner_user_id", 101L, "space_id", 7L, "item_id", 100L, "oss_id", 101L, "size_bytes", 4L, "mime_type", "text/plain", - "file_name", "personal/key.txt", "service", "personal-private" + "file_name", "personal/key.txt", "service", "personal-private", + "upload_token", "upload-token" ); } - private static void assertUploadState(String ext, String state) throws Exception { + private static JsonNode uploadExt(String ext, String state) throws Exception { JsonNode json = new ObjectMapper().readTree(ext); assertEquals("personal", json.path("source").asText()); assertEquals(100L, json.path("itemId").asLong()); assertEquals(state, json.path("uploadState").asText()); + return json; } private record Fixture(JdbcTemplate jdbc, PersonalSpaceService spaces, PersonalObjectStore store, diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java index 4c0b4642..afb4ecbb 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionWorkerTest.java @@ -25,6 +25,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; 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.contains; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; @@ -54,6 +55,12 @@ class PersonalIngestionWorkerTest { @Test void staleParsingUsesLeaseAndExhaustionThreshold() { JdbcTemplate jdbc = mock(JdbcTemplate.class); + Map exhausted = Map.of( + "id", 9L, "tenant_id", "000000", "owner_user_id", 101L, "attempt_count", 3); + Map retryable = Map.of( + "id", 10L, "tenant_id", "000000", "owner_user_id", 101L, "attempt_count", 2); + when(jdbc.queryForList(contains("i.status = 'PARSING'"), any(LocalDateTime.class))) + .thenReturn(List.of(exhausted, retryable)); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), mock(KnowledgeDocumentParser.class), immediateTransactions(), (ossId, prefix, ownerUserId, maxBytes) -> @@ -61,8 +68,10 @@ class PersonalIngestionWorkerTest { worker.recoverStaleParsing(); - verify(jdbc).update(contains("PERSONAL_PARSE_RETRY_EXHAUSTED"), eq(3), any(LocalDateTime.class)); - verify(jdbc).update(contains("set i.status = 'QUEUED'"), eq(3), any(LocalDateTime.class)); + verify(jdbc).update(contains("PERSONAL_PARSE_RETRY_EXHAUSTED"), eq("000000"), eq(101L), eq(9L), + eq(3), any(LocalDateTime.class)); + verify(jdbc).update(contains("set status = 'QUEUED'"), eq("000000"), eq(101L), eq(10L), eq(2), + any(LocalDateTime.class)); } @Test @@ -72,10 +81,12 @@ class PersonalIngestionWorkerTest { TransactionTemplate transactions = immediateTransactions(); Map item = item(); when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item)); - when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L))).thenReturn(1); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); + when(jdbc.queryForMap(contains("for update"), eq("000000"), eq(101L), eq(9L))) + .thenReturn(Map.of("status", "PARSING", "attempt_count", 1)); when(parser.parse(eq("notes.txt"), eq("text/plain"), any(byte[].class))) .thenReturn(new ParsedDocument("一二三四五六", "text/plain", Map.of())); - when(jdbc.update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L))) + when(jdbc.update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L), eq(1))) .thenReturn(1); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), parser, transactions, @@ -88,10 +99,11 @@ class PersonalIngestionWorkerTest { assertTrue(worker.processNext()); - verify(jdbc).update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L)); + verify(jdbc).update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0)); verify(jdbc).update(contains("delete from aihr_personal_fragment"), eq("000000"), eq(101L), eq(9L)); verify(jdbc).batchUpdate(contains("insert into aihr_personal_fragment"), any(BatchPreparedStatementSetter.class)); - verify(jdbc).update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L)); + verify(jdbc).update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L), + eq(1)); } @Test @@ -99,11 +111,13 @@ class PersonalIngestionWorkerTest { JdbcTemplate jdbc = mock(JdbcTemplate.class); KnowledgeDocumentParser parser = mock(KnowledgeDocumentParser.class); when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); - when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L))).thenReturn(1); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); + when(jdbc.queryForMap(contains("for update"), eq("000000"), eq(101L), eq(9L))) + .thenReturn(Map.of("status", "PARSING", "attempt_count", 1)); String content = "字".repeat(900); when(parser.parse(any(), any(), any(byte[].class))) .thenReturn(new ParsedDocument(content, "text/plain", Map.of())); - when(jdbc.update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L))) + when(jdbc.update(contains("status = 'READY'"), any(), eq("[]"), eq("000000"), eq(101L), eq(9L), eq(1))) .thenReturn(1); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, mock(ISysOssService.class), parser, immediateTransactions(), @@ -133,12 +147,54 @@ class PersonalIngestionWorkerTest { verify(jdbc, never()).update(contains("status = 'READY'"), any(), any(), any(), any(), any()); } + @Test + void expiredWorkerSuccessCannotOverwriteNewAttemptOrFragments() throws Exception { + JdbcTemplate jdbc = mock(JdbcTemplate.class); + KnowledgeDocumentParser parser = mock(KnowledgeDocumentParser.class); + when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); + when(parser.parse(any(), any(), any(byte[].class))) + .thenReturn(new ParsedDocument("A worker parsed this", "text/plain", Map.of())); + when(jdbc.queryForMap(contains("for update"), eq("000000"), eq(101L), eq(9L))) + .thenReturn(Map.of("status", "PARSING", "attempt_count", 2)); + PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( + jdbc, mock(ISysOssService.class), parser, immediateTransactions(), + (ossId, prefix, ownerUserId, maxBytes) -> + new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); + + assertTrue(worker.processNext()); + + verify(jdbc, never()).update(contains("delete from aihr_personal_fragment"), any(), any(), any()); + verify(jdbc, never()).batchUpdate(any(String.class), any(BatchPreparedStatementSetter.class)); + verify(jdbc).update(contains("status = 'FAILED'"), eq("PERSONAL_PARSE_FAILED"), anyString(), + eq("000000"), eq(101L), eq(9L), eq(1)); + } + + @Test + void expiredWorkerFailureCannotOverwriteNewAttempt() throws Exception { + JdbcTemplate jdbc = mock(JdbcTemplate.class); + KnowledgeDocumentParser parser = mock(KnowledgeDocumentParser.class); + when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); + when(parser.parse(any(), any(), any(byte[].class))).thenThrow(new IllegalStateException("late failure")); + PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( + jdbc, mock(ISysOssService.class), parser, immediateTransactions(), + (ossId, prefix, ownerUserId, maxBytes) -> + new PersonalIngestionWorker.StoredObject("notes.txt", new byte[]{1})); + + assertTrue(worker.processNext()); + + verify(jdbc).update(contains("status = 'FAILED'"), eq("PERSONAL_PARSE_FAILED"), anyString(), + eq("000000"), eq(101L), eq(9L), eq(1)); + verify(jdbc, never()).update(contains("delete from aihr_personal_fragment"), any(), any(), any()); + } + @Test void workerPersistsOnlyStablePublicFailure() throws Exception { JdbcTemplate jdbc = mock(JdbcTemplate.class); KnowledgeDocumentParser parser = mock(KnowledgeDocumentParser.class); when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); - when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L))).thenReturn(1); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); when(parser.parse(any(), any(), any(byte[].class))) .thenThrow(new KnowledgeDocumentParser.ParseException( KnowledgeDocumentParser.Failure.INVALID, "secret parser detail")); @@ -150,7 +206,7 @@ class PersonalIngestionWorkerTest { assertTrue(worker.processNext()); verify(jdbc).update(contains("status = 'FAILED'"), eq("PERSONAL_PARSE_INVALID"), - eq("资料解析失败，请检查文件后重试"), eq("000000"), eq(101L), eq(9L)); + eq("资料解析失败，请检查文件后重试"), eq("000000"), eq(101L), eq(9L), eq(1)); } @Test @@ -169,7 +225,7 @@ class PersonalIngestionWorkerTest { mock(PersonalIngestionWorker.OssClientProvider.class); when(clients.get("public-client")).thenReturn(publicClient); when(jdbc.queryForList(contains("status = 'QUEUED'"))).thenReturn(List.of(item())); - when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L))).thenReturn(1); + when(jdbc.update(contains("status = 'PARSING'"), eq("000000"), eq(101L), eq(9L), eq(0))).thenReturn(1); PersonalIngestionWorker worker = PersonalIngestionWorker.forTest( jdbc, ossService, mock(KnowledgeDocumentParser.class), immediateTransactions(), PersonalIngestionWorker.objectReaderForTest(ossService, clients)); @@ -177,22 +233,17 @@ class PersonalIngestionWorkerTest { assertTrue(worker.processNext()); verify(jdbc).update(contains("status = 'FAILED'"), eq("PERSONAL_OSS_NOT_PRIVATE"), - eq("个人资料存储策略不可用"), eq("000000"), eq(101L), eq(9L)); + eq("个人资料存储策略不可用"), eq("000000"), eq(101L), eq(9L), eq(1)); verify(publicClient, never()).getObjectContent(any(String.class)); } private static Map item() { - return Map.of( - "id", 9L, - "tenant_id", "000000", - "space_id", 7L, - "owner_user_id", 101L, - "source_type", "TEXT", - "title", "周报", - "oss_id", 81L, - "mime_type", "text/plain", - "tags_json", "[]", - "captured_at", Timestamp.valueOf(LocalDateTime.of(2026, 7, 12, 9, 0)) + return Map.ofEntries( + Map.entry("id", 9L), Map.entry("tenant_id", "000000"), Map.entry("space_id", 7L), + Map.entry("owner_user_id", 101L), Map.entry("source_type", "TEXT"), Map.entry("title", "周报"), + Map.entry("oss_id", 81L), Map.entry("mime_type", "text/plain"), Map.entry("tags_json", "[]"), + Map.entry("captured_at", Timestamp.valueOf(LocalDateTime.of(2026, 7, 12, 9, 0))), + Map.entry("attempt_count", 0) ); } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java index b5b655f6..57df43ec 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java @@ -55,6 +55,7 @@ class PersonalSpaceServiceTest { assertEquals(15, properties.getParsingLeaseMinutes()); assertEquals(3, properties.getMaxParseAttempts()); assertEquals(15, properties.getUploadCleanupAgeMinutes()); + assertEquals(15, properties.getCleanupFinalizeGraceMinutes()); } @Test