diff --git a/docs/education/migration/issues/EDU-011-import-export-assets.md b/docs/education/migration/issues/EDU-011-import-export-assets.md index d4374bb5..006590ed 100644 --- a/docs/education/migration/issues/EDU-011-import-export-assets.md +++ b/docs/education/migration/issues/EDU-011-import-export-assets.md @@ -7,7 +7,7 @@ ## Delivered tenant-admin outcome -Education owns tenant-scoped import asset metadata and durable import jobs while reusing only public Infra APIs for private-file operations and other platform primitives. The delivered job lifecycle has exactly five states: `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, and `FAILED`. +Education owns tenant-scoped import asset metadata and durable import jobs while reusing only public Infra APIs for private-file operations and other platform primitives. The delivered job lifecycle has exactly five states: `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, and `FAILED`. The bounded capability provides: @@ -34,7 +34,7 @@ The bounded capability provides: ## Acceptance record - [x] Education-owned import assets and durable jobs are represented by V4130. -- [x] Jobs use the five-state lifecycle `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, `FAILED`. +- [x] Jobs use the five-state lifecycle `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, `FAILED`. - [x] Claim, lease, heartbeat, expired-lease recovery, retry bounds, and duplicate safety are defined. - [x] Scanning fails closed, with default `UNAVAILABLE`. - [x] CSV/XLSX can return metadata-only preview when the parser is unavailable. diff --git a/yudao-module-education/CONTEXT.md b/yudao-module-education/CONTEXT.md index 480cbb3f..f729ba8f 100644 --- a/yudao-module-education/CONTEXT.md +++ b/yudao-module-education/CONTEXT.md @@ -109,11 +109,11 @@ Tenant-owned metadata that identifies one private Infra-managed source object fo _Avoid_: Uploaded file owned by Education, public asset URL **Content Import Job**: -A tenant-owned durable Education job that previews and, only when safe and executable, imports content from one Content Import Asset. Its five states are `PREVIEW`, `PENDING`, `PROCESSING`, `COMPLETED`, and `FAILED`. +A tenant-owned durable Education job that previews and, only when safe and executable, imports content from one Content Import Asset. Its five states are `PREVIEW_PENDING`, `PREVIEW_READY`, `EXECUTE_PENDING`, `COMPLETED`, and `FAILED`. _Avoid_: Generic Infra job, export job **Import Job Lease**: -A fenced, expiring claim on a `PROCESSING` Content Import Job, identified by worker and lease token and kept alive by heartbeat. Expiry permits bounded recovery; it is not proof that the prior worker stopped. +A fenced, expiring claim on a pending Content Import Job, identified by worker and lease token and kept alive by heartbeat. The aggregate remains `PREVIEW_PENDING` or `EXECUTE_PENDING` while claimed; expiry permits bounded recovery and is not proof that the prior worker stopped. _Avoid_: Redis lock as job ownership, permanent worker assignment **Executable Import Preview**: diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java index 73bffc9e..0187b69f 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java @@ -40,7 +40,8 @@ public class EducationCapabilityController { .capabilities(List.of("shell", "catalog", "questions", "practice-preview", "answer-save", "session-submit", "practice-report", "vocabulary-review", "exam-reminder", "learning-award", "student-feedback", "learning-summary", - "tenant-leaderboard")) + "tenant-leaderboard", "content-import-assets", "content-import-jobs", + "content-export-policy")) .themes(describeThemes()) .catalogReadEnabled(educationProperties.isCatalogReadEnabled()) .practiceWriteEnabled(educationProperties.isPracticeWriteEnabled()) @@ -53,10 +54,10 @@ public class EducationCapabilityController { List catalogDependencies = educationProperties.getCatalogMode() == CatalogProviderMode.SCALAR_READ ? List.of("scalar-catalog") : List.of(); return List.of( - theme("EDU-011", "content-import-export-assets", "BLOCKED", + theme("EDU-011", "content-import-export-assets", "PARTIAL", List.of("education", "infra"), - List.of("asset-admission-contract", "scanner-ownership", "durable-job-lease"), - List.of(), "FEATURE_FLAG", "CONTRACT_ONLY"), + List.of("production-scanner-adapter", "production-import-parser", "export-artifact-generation"), + List.of(), "FEATURE_FLAG", "POSTGRESQL_INTEGRATION_TESTED"), theme("EDU-012", "classes-relationships", "BLOCKED", List.of("education", "member", "system"), List.of("relationship-role-model", "class-data-scope", "invitation-contract"), diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/EducationImportAssetDO.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/EducationImportAssetDO.java index 66b39cb9..b55febfb 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/EducationImportAssetDO.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/EducationImportAssetDO.java @@ -2,23 +2,29 @@ package cn.iocoder.yudao.module.education.dal.dataobject; import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO; import com.baomidou.mybatisplus.annotation.KeySequence; +import com.baomidou.mybatisplus.annotation.TableField; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import lombok.Data; import lombok.EqualsAndHashCode; -@TableName("education_import_asset") -@KeySequence("education_import_asset_seq") +@TableName("education_content_import_asset") +@KeySequence("education_content_import_asset_seq") @Data @EqualsAndHashCode(callSuper = true) public class EducationImportAssetDO extends TenantBaseDO { @TableId private Long id; private Long ownerUserId; + @TableField("file_name") private String originalName; + @TableField("mime_type") private String contentType; + @TableField("file_size_bytes") private Long size; + @TableField("checksum_sha256") private String sha256; + @TableField("object_key") private String fileReference; private String scanStatus; } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportAssetDO.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportAssetDO.java index 81cb55d6..3445429f 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportAssetDO.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportAssetDO.java @@ -14,9 +14,11 @@ import lombok.EqualsAndHashCode; public class ContentImportAssetDO extends TenantBaseDO { @TableId private Long id; + private Long ownerUserId; private String objectKey; private String fileName; private String mimeType; private Long fileSizeBytes; private String checksumSha256; + private String scanStatus; } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportJobDO.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportJobDO.java index 4671b66e..0da7ef61 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportJobDO.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/importjob/ContentImportJobDO.java @@ -15,12 +15,17 @@ import java.time.LocalDateTime; @Data @EqualsAndHashCode(callSuper = true) public class ContentImportJobDO extends TenantBaseDO { - @TableId - private Long id; + @TableId private Long id; private Long assetId; - private String previewKey; + private Long actorId; + @TableField("preview_key") private String previewKey; + private String previewRequestHash; + @TableField("execute_key") private String executeKey; + private String executeRequestHash; private String importType; private String status; + private String scanStatus; + private String parserStatus; private Integer attemptCount; private Integer maxAttempts; private LocalDateTime nextAttemptAt; @@ -30,9 +35,11 @@ public class ContentImportJobDO extends TenantBaseDO { private LocalDateTime lastHeartbeatAt; private LocalDateTime startedAt; private LocalDateTime finishedAt; - private String errorMessage; - @TableField("preview_payload") + private String failureCode; + private String failureMessage; private String previewPayload; - @TableField("result_summary") + private String parsedPayload; private String resultSummary; + private Integer previewQuestionCount; + private Integer importedQuestionCount; } diff --git a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java index fc8c7164..22172b7c 100644 --- a/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java +++ b/yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapper.java @@ -11,10 +11,12 @@ public interface ContentImportJobMapper extends BaseMapperX @Insert(""" INSERT INTO education_content_import_job - (tenant_id, asset_id, preview_key, import_type, status, attempt_count, max_attempts, + (tenant_id, asset_id, actor_id, preview_key, preview_request_hash, import_type, status, + scan_status, parser_status, attempt_count, max_attempts, next_attempt_at, preview_payload, creator, create_time, updater, update_time, deleted) VALUES - (#{tenantId}, #{assetId}, #{previewKey}, #{importType}, #{status}, 0, #{maxAttempts}, + (#{tenantId}, #{assetId}, #{actorId}, #{previewKey}, #{previewRequestHash}, #{importType}, + 'PREVIEW_PENDING', 'PENDING', 'PENDING', 0, #{maxAttempts}, CURRENT_TIMESTAMP, CAST(#{previewPayload} AS JSONB), '', CURRENT_TIMESTAMP, '', CURRENT_TIMESTAMP, FALSE) ON CONFLICT (tenant_id, preview_key) DO NOTHING """) @@ -23,71 +25,67 @@ public interface ContentImportJobMapper extends BaseMapperX @Update(""" UPDATE education_content_import_job - SET status = 'PENDING', next_attempt_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP - WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PREVIEW' AND deleted = FALSE + SET status='EXECUTE_PENDING', execute_key=#{executeKey}, execute_request_hash=#{executeRequestHash}, + next_attempt_at=CURRENT_TIMESTAMP, update_time=CURRENT_TIMESTAMP + WHERE tenant_id=#{tenantId} AND id=#{id} AND status='PREVIEW_READY' + AND scan_status='CLEAN' AND parser_status='PARSED' AND deleted=false """) - int executePreview(@Param("tenantId") Long tenantId, @Param("id") Long id); + int requestExecute(@Param("tenantId") Long tenantId, @Param("id") Long id, + @Param("executeKey") String executeKey, @Param("executeRequestHash") String executeRequestHash); @Select(""" - WITH candidates AS ( + WITH candidate AS ( SELECT id FROM education_content_import_job - WHERE deleted = FALSE AND attempt_count < max_attempts - AND ((status = 'PENDING' AND (next_attempt_at IS NULL OR next_attempt_at <= CURRENT_TIMESTAMP)) - OR (status = 'PROCESSING' AND lease_expires_at <= CURRENT_TIMESTAMP)) - ORDER BY CASE WHEN status = 'PROCESSING' THEN 0 ELSE 1 END, - COALESCE(lease_expires_at, next_attempt_at, create_time), id + WHERE deleted=false AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING') + AND attempt_count < max_attempts + AND (next_attempt_at IS NULL OR next_attempt_at <= CURRENT_TIMESTAMP) + AND (lease_expires_at IS NULL OR lease_expires_at <= CURRENT_TIMESTAMP) + ORDER BY COALESCE(next_attempt_at, create_time), id LIMIT #{limit} FOR UPDATE SKIP LOCKED ) UPDATE education_content_import_job job - SET status = 'PROCESSING', locked_by = #{workerId}, lease_token = #{leaseToken}, - lease_expires_at = CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL), - last_heartbeat_at = CURRENT_TIMESTAMP, attempt_count = job.attempt_count + 1, - next_attempt_at = NULL, started_at = COALESCE(job.started_at, CURRENT_TIMESTAMP), - finished_at = NULL, error_message = NULL, update_time = CURRENT_TIMESTAMP - FROM candidates WHERE job.id = candidates.id + SET locked_by=#{workerId}, lease_token=#{leaseToken}, + lease_expires_at=CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL), + last_heartbeat_at=CURRENT_TIMESTAMP, attempt_count=job.attempt_count + 1, + next_attempt_at=NULL, started_at=COALESCE(job.started_at,CURRENT_TIMESTAMP), update_time=CURRENT_TIMESTAMP + FROM candidate WHERE job.id=candidate.id RETURNING job.* """) - List claimReady(@Param("workerId") String workerId, - @Param("leaseToken") String leaseToken, - @Param("leaseSeconds") long leaseSeconds, - @Param("limit") int limit); + List claimReady(@Param("workerId") String workerId, @Param("leaseToken") String leaseToken, + @Param("leaseSeconds") long leaseSeconds, @Param("limit") int limit); @Update(""" UPDATE education_content_import_job - SET lease_expires_at = CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL), - last_heartbeat_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP - WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PROCESSING' - AND lease_token = #{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP + SET lease_expires_at=CURRENT_TIMESTAMP + CAST(#{leaseSeconds} || ' seconds' AS INTERVAL), + last_heartbeat_at=CURRENT_TIMESTAMP, update_time=CURRENT_TIMESTAMP + WHERE tenant_id=#{tenantId} AND id=#{id} AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING') + AND lease_token=#{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP """) int heartbeat(@Param("tenantId") Long tenantId, @Param("id") Long id, @Param("leaseToken") String leaseToken, @Param("leaseSeconds") long leaseSeconds); - @Update(""" - WITH expired AS ( - SELECT id FROM education_content_import_job - WHERE status = 'PROCESSING' AND lease_expires_at <= CURRENT_TIMESTAMP - AND attempt_count < max_attempts - ORDER BY lease_expires_at, id - LIMIT #{limit} FOR UPDATE SKIP LOCKED - ) - UPDATE education_content_import_job job - SET status = 'PENDING', locked_by = NULL, lease_token = NULL, lease_expires_at = NULL, - last_heartbeat_at = NULL, next_attempt_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP - FROM expired WHERE job.id = expired.id - """) - int recoverExpired(@Param("limit") int limit); - @Update(""" UPDATE education_content_import_job - SET status = 'COMPLETED', result_summary = CAST(#{resultSummary} AS JSONB), - locked_by = NULL, lease_token = NULL, lease_expires_at = NULL, last_heartbeat_at = NULL, - finished_at = CURRENT_TIMESTAMP, update_time = CURRENT_TIMESTAMP - WHERE tenant_id = #{tenantId} AND id = #{id} AND status = 'PROCESSING' - AND lease_token = #{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP + SET status=#{nextStatus}, scan_status=COALESCE(#{scanStatus},scan_status), + parser_status=COALESCE(#{parserStatus},parser_status), + preview_payload=COALESCE(CAST(#{previewPayload} AS JSONB),preview_payload), + parsed_payload=CAST(#{parsedPayload} AS JSONB), preview_question_count=#{previewQuestionCount}, + result_summary=COALESCE(CAST(#{resultSummary} AS JSONB),result_summary), + imported_question_count=#{importedQuestionCount}, failure_code=#{failureCode}, + failure_message=#{failureMessage}, locked_by=NULL, lease_token=NULL, lease_expires_at=NULL, + last_heartbeat_at=NULL, finished_at=CASE WHEN #{nextStatus} IN ('COMPLETED','FAILED') THEN CURRENT_TIMESTAMP ELSE finished_at END, + update_time=CURRENT_TIMESTAMP + WHERE tenant_id=#{tenantId} AND id=#{id} AND status=#{expectedStatus} + AND lease_token=#{leaseToken} AND lease_expires_at > CURRENT_TIMESTAMP """) - int complete(@Param("tenantId") Long tenantId, @Param("id") Long id, - @Param("leaseToken") String leaseToken, @Param("resultSummary") String resultSummary); + int finishClaim(@Param("tenantId") Long tenantId, @Param("id") Long id, + @Param("expectedStatus") String expectedStatus, @Param("nextStatus") String nextStatus, + @Param("leaseToken") String leaseToken, @Param("scanStatus") String scanStatus, + @Param("parserStatus") String parserStatus, @Param("previewPayload") String previewPayload, + @Param("parsedPayload") String parsedPayload, @Param("previewQuestionCount") Integer previewQuestionCount, + @Param("resultSummary") String resultSummary, @Param("importedQuestionCount") Integer importedQuestionCount, + @Param("failureCode") String failureCode, @Param("failureMessage") String failureMessage); - @Update("UPDATE education_content_import_job SET lease_expires_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = #{id}") + @Update("UPDATE education_content_import_job SET lease_expires_at=CURRENT_TIMESTAMP-INTERVAL '1 second' WHERE id=#{id}") int expireLeaseForTest(@Param("id") Long id); } diff --git a/yudao-module-education/src/main/resources/db/migration/education/V4130__add_content_import_jobs.sql b/yudao-module-education/src/main/resources/db/migration/education/V4130__add_content_import_jobs.sql index 69fe9f14..dd136483 100644 --- a/yudao-module-education/src/main/resources/db/migration/education/V4130__add_content_import_jobs.sql +++ b/yudao-module-education/src/main/resources/db/migration/education/V4130__add_content_import_jobs.sql @@ -1,35 +1,48 @@ --- EDU-011 persistence slice: tenant-owned import assets and durable preview/execute jobs. +-- EDU-011: tenant-bound import assets and durable preview/execute jobs. CREATE TABLE education_content_import_asset ( id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, tenant_id BIGINT NOT NULL, - object_key VARCHAR(512) NOT NULL, + owner_user_id BIGINT NOT NULL, + object_key VARCHAR(1024) NOT NULL, file_name VARCHAR(255) NOT NULL, mime_type VARCHAR(128) NOT NULL, file_size_bytes BIGINT NOT NULL, checksum_sha256 VARCHAR(64) NOT NULL, - creator VARCHAR(64) DEFAULT '', - create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - updater VARCHAR(64) DEFAULT '', - update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + scan_status VARCHAR(16) NOT NULL DEFAULT 'UNAVAILABLE', + creator VARCHAR(64) DEFAULT '', create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updater VARCHAR(64) DEFAULT '', update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, deleted BOOLEAN NOT NULL DEFAULT false, - CONSTRAINT ck_education_content_import_asset_size CHECK (file_size_bytes >= 0), + CONSTRAINT ck_education_content_import_asset_size CHECK (file_size_bytes > 0 AND file_size_bytes <= 10485760), + CONSTRAINT ck_education_content_import_asset_scan CHECK (scan_status IN ('PENDING','CLEAN','INFECTED','ERROR','UNAVAILABLE')), CONSTRAINT uk_education_content_import_asset_tenant_id UNIQUE (tenant_id, id), CONSTRAINT uk_education_content_import_asset_object UNIQUE (tenant_id, object_key) ); - -COMMENT ON TABLE education_content_import_asset IS '教育内容导入源文件元数据；文件存储与扫描由 Infra 持有'; -COMMENT ON COLUMN education_content_import_asset.object_key IS 'Infra 私有对象键，不存储公开 URL'; +COMMENT ON TABLE education_content_import_asset IS 'Education-owned tenant binding for an Infra file reference admitted for content import'; +COMMENT ON COLUMN education_content_import_asset.object_key IS 'Opaque reference returned by the public Infra File API; never client supplied'; CREATE TABLE education_content_import_job ( id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, tenant_id BIGINT NOT NULL, asset_id BIGINT NOT NULL, + actor_id BIGINT NOT NULL, + import_type VARCHAR(32) NOT NULL DEFAULT 'QUESTION_DRAFT', + status VARCHAR(24) NOT NULL DEFAULT 'PREVIEW_PENDING', preview_key VARCHAR(128) NOT NULL, - import_type VARCHAR(32) NOT NULL, - status VARCHAR(16) NOT NULL DEFAULT 'PREVIEW', + preview_request_hash VARCHAR(64) NOT NULL, + execute_key VARCHAR(128), + execute_request_hash VARCHAR(64), + scan_status VARCHAR(16) NOT NULL DEFAULT 'PENDING', + parser_status VARCHAR(16) NOT NULL DEFAULT 'PENDING', + preview_payload JSONB NOT NULL DEFAULT '{}'::JSONB, + parsed_payload JSONB, + result_summary JSONB NOT NULL DEFAULT '{}'::JSONB, + preview_question_count INTEGER, + imported_question_count INTEGER, + failure_code VARCHAR(64), + failure_message VARCHAR(500), attempt_count INTEGER NOT NULL DEFAULT 0, - max_attempts INTEGER NOT NULL DEFAULT 3, + max_attempts INTEGER NOT NULL DEFAULT 5, next_attempt_at TIMESTAMP, locked_by VARCHAR(128), lease_token VARCHAR(64), @@ -37,38 +50,32 @@ CREATE TABLE education_content_import_job ( last_heartbeat_at TIMESTAMP, started_at TIMESTAMP, finished_at TIMESTAMP, - error_message VARCHAR(2000), - preview_payload JSONB NOT NULL DEFAULT '{}'::JSONB, - result_summary JSONB NOT NULL DEFAULT '{}'::JSONB, - creator VARCHAR(64) DEFAULT '', - create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - updater VARCHAR(64) DEFAULT '', - update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + creator VARCHAR(64) DEFAULT '', create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updater VARCHAR(64) DEFAULT '', update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, deleted BOOLEAN NOT NULL DEFAULT false, - CONSTRAINT fk_education_content_import_job_asset_tenant - FOREIGN KEY (tenant_id, asset_id) + CONSTRAINT fk_education_content_import_job_asset_tenant FOREIGN KEY (tenant_id, asset_id) REFERENCES education_content_import_asset (tenant_id, id), CONSTRAINT uk_education_content_import_job_preview UNIQUE (tenant_id, preview_key), - CONSTRAINT ck_education_content_import_job_status - CHECK (status IN ('PREVIEW', 'PENDING', 'PROCESSING', 'COMPLETED', 'FAILED')), - CONSTRAINT ck_education_content_import_job_attempts - CHECK (attempt_count >= 0 AND max_attempts > 0 AND attempt_count <= max_attempts), + CONSTRAINT uk_education_content_import_job_execute UNIQUE (tenant_id, execute_key), + CONSTRAINT ck_education_content_import_job_status CHECK (status IN + ('PREVIEW_PENDING','PREVIEW_READY','EXECUTE_PENDING','COMPLETED','FAILED')), + CONSTRAINT ck_education_content_import_job_attempts CHECK + (attempt_count >= 0 AND max_attempts > 0 AND attempt_count <= max_attempts), + CONSTRAINT ck_education_content_import_job_counts CHECK + (COALESCE(preview_question_count, 0) >= 0 AND COALESCE(imported_question_count, 0) >= 0), CONSTRAINT ck_education_content_import_job_lease CHECK ( - (status = 'PROCESSING' AND locked_by IS NOT NULL AND lease_token IS NOT NULL - AND lease_expires_at IS NOT NULL AND last_heartbeat_at IS NOT NULL) - OR - (status <> 'PROCESSING' AND locked_by IS NULL AND lease_token IS NULL - AND lease_expires_at IS NULL AND last_heartbeat_at IS NULL) - ) + (lease_token IS NULL AND locked_by IS NULL AND lease_expires_at IS NULL AND last_heartbeat_at IS NULL) + OR (lease_token IS NOT NULL AND locked_by IS NOT NULL AND lease_expires_at IS NOT NULL AND last_heartbeat_at IS NOT NULL + AND status IN ('PREVIEW_PENDING','EXECUTE_PENDING'))), + CONSTRAINT ck_education_content_import_job_execute CHECK + (status <> 'EXECUTE_PENDING' OR (execute_key IS NOT NULL AND execute_request_hash IS NOT NULL)), + CONSTRAINT ck_education_content_import_job_failed CHECK + (status <> 'FAILED' OR failure_code IS NOT NULL) ); +COMMENT ON TABLE education_content_import_job IS 'Education import aggregate with preview/execute lifecycle and token-fenced worker lease'; +COMMENT ON COLUMN education_content_import_job.lease_token IS 'Fencing token required by heartbeat and terminal worker writes'; -COMMENT ON TABLE education_content_import_job IS '教育内容导入预览与执行作业'; -COMMENT ON COLUMN education_content_import_job.preview_key IS '租户内重复预览幂等键'; -COMMENT ON COLUMN education_content_import_job.lease_token IS '处理租约 fencing token'; - -CREATE INDEX idx_education_content_import_job_ready - ON education_content_import_job (next_attempt_at, create_time, id) - WHERE status = 'PENDING' AND deleted = false; -CREATE INDEX idx_education_content_import_job_expired - ON education_content_import_job (lease_expires_at, create_time, id) - WHERE status = 'PROCESSING' AND deleted = false; +CREATE INDEX idx_education_content_import_job_claim ON education_content_import_job + (next_attempt_at, lease_expires_at, create_time, id) + WHERE status IN ('PREVIEW_PENDING','EXECUTE_PENDING') AND deleted=false; +CREATE INDEX idx_education_content_import_job_tenant ON education_content_import_job (tenant_id, create_time DESC, id DESC); diff --git a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapperPostgreSqlIntegrationTest.java b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapperPostgreSqlIntegrationTest.java index c54e9ff5..46b08fb5 100644 --- a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapperPostgreSqlIntegrationTest.java +++ b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/dal/mysql/importjob/ContentImportJobMapperPostgreSqlIntegrationTest.java @@ -5,9 +5,7 @@ import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportA import cn.iocoder.yudao.module.education.dal.dataobject.importjob.ContentImportJobDO; import cn.iocoder.yudao.module.education.test.PostgreSqlDbIntegrationTest; import jakarta.annotation.Resource; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.*; import org.springframework.dao.DataIntegrityViolationException; import java.time.LocalDateTime; @@ -16,92 +14,58 @@ import java.util.List; import static org.junit.jupiter.api.Assertions.*; class ContentImportJobMapperPostgreSqlIntegrationTest extends PostgreSqlDbIntegrationTest { + @Resource(name="contentImportAssetMapper") ContentImportAssetMapper assetMapper; + @Resource(name="contentImportJobMapper") ContentImportJobMapper jobMapper; - @Resource(name = "contentImportAssetMapper") private ContentImportAssetMapper assetMapper; - @Resource(name = "contentImportJobMapper") private ContentImportJobMapper jobMapper; + @BeforeEach void setUp(){ TenantContextHolder.setTenantId(10L); } + @AfterEach void clear(){ TenantContextHolder.clear(); } - @BeforeEach - void setUp() { - TenantContextHolder.setTenantId(10L); - } + @Test void tenantBindingDuplicateCommandsAndLeaseFencingAreAtomic() { + ContentImportAssetDO asset=asset(10L,"ref-10"); assetMapper.insert(asset); + ContentImportJobDO preview=preview(10L,asset.getId(),"preview-1"); + assertEquals(1,jobMapper.insertPreviewIgnore(preview)); + assertEquals(0,jobMapper.insertPreviewIgnore(preview(10L,asset.getId(),"preview-1"))); + ContentImportAssetDO other=asset(20L,"ref-20"); assetMapper.insert(other); + assertThrows(DataIntegrityViolationException.class,()->jobMapper.insertPreviewIgnore(preview(10L,other.getId(),"cross"))); - @AfterEach - void tearDown() { - TenantContextHolder.clear(); - } - - @Test - void shouldEnforceTenantBoundJobLifecycleAndLeaseFencing() { - ContentImportAssetDO asset = asset(10L, "imports/10/questions.xlsx"); - assetMapper.insert(asset); - - ContentImportJobDO preview = preview(10L, asset.getId(), "preview-1"); - assertEquals(1, jobMapper.insertPreviewIgnore(preview)); - assertEquals(0, jobMapper.insertPreviewIgnore(preview(10L, asset.getId(), "preview-1"))); - - ContentImportAssetDO otherTenantAsset = asset(20L, "imports/20/questions.xlsx"); - assetMapper.insert(otherTenantAsset); - assertThrows(DataIntegrityViolationException.class, - () -> jobMapper.insertPreviewIgnore(preview(10L, otherTenantAsset.getId(), "cross-tenant"))); - - assertEquals(1, jobMapper.executePreview(10L, preview.getId())); - assertEquals(0, jobMapper.executePreview(10L, preview.getId())); - - List claimed = jobMapper.claimReady("worker-a", "token-a", 60, 1); - assertEquals(1, claimed.size()); - ContentImportJobDO job = claimed.getFirst(); - assertEquals("PROCESSING", job.getStatus()); - assertEquals(1, job.getAttemptCount()); - assertEquals("token-a", job.getLeaseToken()); - assertTrue(jobMapper.claimReady("worker-b", "token-b", 60, 1).isEmpty()); - - LocalDateTime firstExpiry = job.getLeaseExpiresAt(); - assertEquals(1, jobMapper.heartbeat(10L, job.getId(), "token-a", 120)); + List claimed=jobMapper.claimReady("worker-a","token-a",60,1); + assertEquals(1,claimed.size()); + ContentImportJobDO job=claimed.getFirst(); + assertEquals("PREVIEW_PENDING",job.getStatus()); + assertEquals(1,job.getAttemptCount()); + assertTrue(jobMapper.claimReady("worker-b","token-b",60,1).isEmpty()); + LocalDateTime firstExpiry=job.getLeaseExpiresAt(); + assertEquals(1,jobMapper.heartbeat(10L,job.getId(),"token-a",120)); assertTrue(jobMapper.selectById(job.getId()).getLeaseExpiresAt().isAfter(firstExpiry)); - assertEquals(0, jobMapper.heartbeat(10L, job.getId(), "stale-token", 120)); - assertEquals(0, jobMapper.complete(10L, job.getId(), "stale-token", "{}")); + assertEquals(0,jobMapper.heartbeat(10L,job.getId(),"stale",120)); + assertEquals(0,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","stale", + "CLEAN","UNAVAILABLE","{}",null,null,null,null,null,null)); jobMapper.expireLeaseForTest(job.getId()); - assertEquals(1, jobMapper.recoverExpired(10)); - assertEquals("PENDING", jobMapper.selectById(job.getId()).getStatus()); - - ContentImportJobDO reclaimed = jobMapper.claimReady("worker-b", "token-b", 60, 1).getFirst(); - assertEquals(2, reclaimed.getAttemptCount()); - assertEquals(0, jobMapper.complete(10L, job.getId(), "token-a", "{}")); - assertEquals(1, jobMapper.complete(10L, job.getId(), "token-b", "{\"inserted\":1}")); - assertEquals("COMPLETED", jobMapper.selectById(job.getId()).getStatus()); + ContentImportJobDO recovered=jobMapper.claimReady("worker-b","token-b",60,1).getFirst(); + assertEquals(2,recovered.getAttemptCount()); + assertEquals(0,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","token-a", + "CLEAN","UNAVAILABLE","{}",null,null,null,null,null,null)); + assertEquals(1,jobMapper.finishClaim(10L,job.getId(),"PREVIEW_PENDING","PREVIEW_READY","token-b", + "CLEAN","UNAVAILABLE","{\"mode\":\"METADATA_ONLY\",\"executable\":false}",null,null,null,null,null,null)); + assertEquals("PREVIEW_READY",jobMapper.selectById(job.getId()).getStatus()); + assertEquals(0,jobMapper.requestExecute(10L,job.getId(),"execute-1","b".repeat(64))); } - @Test - void shouldAllowOnlyFiveJobStates() { - ContentImportAssetDO asset = asset(10L, "imports/10/states.xlsx"); - assetMapper.insert(asset); - ContentImportJobDO job = preview(10L, asset.getId(), "states"); - job.setStatus("REJECTED"); - - assertThrows(DataIntegrityViolationException.class, () -> jobMapper.insert(job)); + @Test void onlyRequiredFiveStatesAreAccepted(){ + ContentImportAssetDO asset=asset(10L,"states"); assetMapper.insert(asset); + ContentImportJobDO job=preview(10L,asset.getId(),"states"); job.setStatus("PROCESSING"); + assertThrows(DataIntegrityViolationException.class,()->jobMapper.insert(job)); } - private ContentImportAssetDO asset(Long tenantId, String objectKey) { - ContentImportAssetDO asset = new ContentImportAssetDO(); - asset.setTenantId(tenantId); - asset.setObjectKey(objectKey); - asset.setFileName("questions.xlsx"); - asset.setMimeType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"); - asset.setFileSizeBytes(128L); - asset.setChecksumSha256("a".repeat(64)); - return asset; + private ContentImportAssetDO asset(Long tenant,String key){ + ContentImportAssetDO a=new ContentImportAssetDO(); a.setTenantId(tenant); a.setOwnerUserId(7L); a.setObjectKey(key); + a.setFileName("questions.xlsx"); a.setMimeType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"); + a.setFileSizeBytes(128L); a.setChecksumSha256("a".repeat(64)); a.setScanStatus("UNAVAILABLE"); return a; } - - private ContentImportJobDO preview(Long tenantId, Long assetId, String previewKey) { - ContentImportJobDO job = new ContentImportJobDO(); - job.setTenantId(tenantId); - job.setAssetId(assetId); - job.setPreviewKey(previewKey); - job.setImportType("QUESTIONS"); - job.setStatus("PREVIEW"); - job.setMaxAttempts(3); - job.setPreviewPayload("{}"); - return job; + private ContentImportJobDO preview(Long tenant,Long asset,String key){ + ContentImportJobDO j=new ContentImportJobDO(); j.setTenantId(tenant); j.setAssetId(asset); j.setActorId(7L); + j.setPreviewKey(key); j.setPreviewRequestHash("a".repeat(64)); j.setImportType("QUESTION_DRAFT"); + j.setStatus("PREVIEW_PENDING"); j.setMaxAttempts(5); j.setPreviewPayload("{}"); return j; } } diff --git a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/asset/EducationAssetAdmissionServiceImplTest.java b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/asset/EducationAssetAdmissionServiceImplTest.java index 4a09df0d..4bb0f0c1 100644 --- a/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/asset/EducationAssetAdmissionServiceImplTest.java +++ b/yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/asset/EducationAssetAdmissionServiceImplTest.java @@ -28,7 +28,7 @@ class EducationAssetAdmissionServiceImplTest { @BeforeEach void setUp() { TenantContextHolder.setTenantId(42L); - when(mapper.insert(any())).thenAnswer(invocation -> { + when(mapper.insert(any(EducationImportAssetDO.class))).thenAnswer(invocation -> { EducationImportAssetDO asset = invocation.getArgument(0); asset.setId(99L); return 1;