feat(jobs): add cancellation and queued source transfers - #666
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. 🗂️ Base branches to auto review (1)
Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
253f1ff to
57ff3f7
Compare
| const staleBaseCondition = and( | ||
| eq(jobs.status, "in_progress"), | ||
| lt(jobs.updatedAt, olderThan), | ||
| notInArray(jobs.type, [ |
There was a problem hiding this comment.
[P1] 長時間 transfer job を stale recovery から保護してください
updatedAt は claim/start と終了時にしか更新されず、JobWorker に実行中の heartbeat がありません。それにもかかわらず、この除外リストには source_export / source_restore が含まれていないため、1時間を超える export/restore がまだ実行中でも pending に戻され、別workerが同じ処理を開始できます。lease/tokenによる所有権確認、実行中heartbeat、または少なくとも非冪等なtransfer jobの除外が必要です。
| }); | ||
| } | ||
|
|
||
| await JobRepository.requestCancellation(input.id); |
There was a problem hiding this comment.
[P1] batch parent のキャンセルを子jobへ伝播するか、UI/APIで拒否してください
このendpointは batch_ccip_parent、bulk_tagging_parent、thumbnail_generation_parent も受け付けますが、parentはworkerにclaimされず、requestCancellation はフラグを立てるだけです。子jobは継続し、各finalizerがparentを無条件に completed/failed へ更新します。現状のCancel操作は成功したように見えて実処理を止めません。parent種別を拒否するか、子jobのcascade cancelとfinalizer側のcancel優先処理を追加してください。
| payload.inputPath, | ||
| ); | ||
| } finally { | ||
| await removeJobTransferFile(payload.inputPath); |
There was a problem hiding this comment.
[P1] restore入力をretry可能な期間まで保持してください
jobs.retry は失敗jobのpayloadをそのまま再利用するため、retry時も同じ inputPath を参照します。しかしこの finally は最初のrestoreが失敗しても入力ファイルを削除します。結果としてretryは必ずmissing-inputになり、再アップロードや入力artifactの再生成もできません。入力をretry用に保持するか、retry時に新しい入力を受け取る設計にしてください。なお、importがエラー配列を返すだけのケースもworker側でcompleted扱いになるため、部分失敗の状態表現も合わせて見直しが必要です。
| await this.markCancelled(job); | ||
| return; | ||
| } else { | ||
| await this.jobRepo.markAsCompleted(job.id, safeResult); |
There was a problem hiding this comment.
[P1] 通常のcompleted/failed遷移でもjob eventをpublishしてください
このPRで job-cancelled はpublishされますが、通常の markAsCompleted / markAsFailed の直後には job-completed / job-failed がありません。source transfer jobにはparent finalizerによる代替通知もないため、jobs画面はイベントを受け取れず、別クライアントの表示がpendingのまま残ります。DB更新の成否を確認したうえで、完了・失敗・retryの各遷移を既存event streamへ通知してください。
| ".cache", | ||
| "job-transfers", | ||
| ); | ||
| const JobArtifactTtlMs = 24 * 60 * 60 * 1000; |
There was a problem hiding this comment.
[P2] 期限切れartifactと中断されたinputのcleanupを追加してください
ここでTTLを計算していますが、artifact endpointは期限後に410を返すだけでファイルを削除しません。さらに、processor開始前にcancelされたrestoreのinputも削除されない経路があります。startup/workerの定期cleanup、または明示的なcancel/finalize cleanupを追加しないと、長期運用で .cache/job-transfers が増え続けます。
| status: MediaSourceSyncState, | ||
| message?: string, | ||
| ): Promise<void> { | ||
| publishSyncStatus(mediaSourceId, status, message); |
There was a problem hiding this comment.
[P2] sync status は永続化後にpublishするか、event payloadでUI cacheを直接更新してください
現在はDB更新より先にeventをpublishしています。UI側はevent payloadをcacheへ反映せずinvalidateして再取得するだけなので、event直後のrefetchが古いDB状態を読み、syncing/idle/errorの表示が取りこぼされるraceがあります。永続化を先にするか、受信側でeventのstatusを直接適用してください。
| }); | ||
| }), | ||
|
|
||
| enqueueExport: os |
There was a problem hiding this comment.
[P2] enqueueExport/enqueueImport に共有output schemaを設定してください
これらの新しいhandlerは toJobDto(job) を返しますが、.output(...) がないため、生成されたOpenAPIの200 responseが空の anyOf になります。#663のjobs endpointと同じ jobDtoSchema を共有して、transfer APIの契約と生成clientの型を実行時・OpenAPIの両方で保証してください。
1 participant
Stack created with GitHub Stacks CLI • Give Feedback 💬