mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 16:27:18 +02:00
Fix async import job polling cleanup
This commit is contained in:
1 parent
114181e374
commit
2214b2c784
2 files changed
+26
-1
No files matched your search
@@ -507,6 +507,7 @@ describe.sequential("company portability routes", () => {
|
||||
expect(succeeded.status).toBe(200);
|
||||
expect(succeeded.body.job.status).toBe("succeeded");
|
||||
expect(succeeded.body.job.result.companyId).toBe(companyId);
|
||||
expect(succeeded.body.retryAfterMs).toBeUndefined();
|
||||
expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
|
||||
action: "company.imported",
|
||||
companyId,
|
||||
@@ -516,6 +517,15 @@ describe.sequential("company portability routes", () => {
|
||||
companyAction: "updated",
|
||||
}),
|
||||
}));
|
||||
|
||||
const nowSpy = vi.spyOn(Date, "now").mockReturnValue(Date.parse(succeeded.body.job.completedAt) + (5 * 60 * 1000) + 1);
|
||||
try {
|
||||
const expired = await request(app).get(accepted.body.statusUrl).set(cloudHeaders);
|
||||
expect(expired.status).toBe(404);
|
||||
expect(expired.body.error).toBe("Import job not found");
|
||||
} finally {
|
||||
nowSpy.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it.sequential("reports trusted Cloud async import job failures with the tenant error message", async () => {
|
||||
@@ -534,6 +544,7 @@ describe.sequential("company portability routes", () => {
|
||||
expect(failed.status).toBe(200);
|
||||
expect(failed.body.job.status).toBe("failed");
|
||||
expect(failed.body.job.error.message).toBe("tenant import exploded");
|
||||
expect(failed.body.retryAfterMs).toBeUndefined();
|
||||
expect(failed.body.message).toBe("tenant import exploded");
|
||||
expect(mockLogActivity).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
@@ -37,6 +37,7 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
const budgets = budgetService(db);
|
||||
const feedback = feedbackService(db);
|
||||
const importJobs = new Map<string, ImportJobRecord>();
|
||||
const importJobTerminalRetentionMs = 5 * 60 * 1000;
|
||||
|
||||
function parseBooleanQuery(value: unknown) {
|
||||
return value === true || value === "true" || value === "1";
|
||||
@@ -181,6 +182,7 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
|
||||
router.get("/import/jobs/:jobId", async (req, res) => {
|
||||
assertCloudTenantCaller(req);
|
||||
cleanupTerminalImportJobs(importJobs, importJobTerminalRetentionMs);
|
||||
const job = importJobs.get(req.params.jobId as string);
|
||||
if (!job || job.cloudTenantKey !== cloudTenantRequestKey(req)) {
|
||||
res.status(404).json({ error: "Import job not found" });
|
||||
@@ -196,6 +198,7 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
const boardUserId = req.actor.type === "board" ? req.actor.userId : null;
|
||||
if (req.header("x-paperclip-cloud-async-import") === "1") {
|
||||
assertCloudTenantCaller(req);
|
||||
cleanupTerminalImportJobs(importJobs, importJobTerminalRetentionMs);
|
||||
const job = createImportJob(cloudTenantRequestKey(req));
|
||||
importJobs.set(job.id, job);
|
||||
const operation = async () => {
|
||||
@@ -561,6 +564,7 @@ function importJobAcceptedResponse(job: ImportJobRecord) {
|
||||
}
|
||||
|
||||
function importJobResponse(job: ImportJobRecord) {
|
||||
const isTerminal = job.status === "succeeded" || job.status === "failed";
|
||||
const response: Record<string, unknown> = {
|
||||
job: {
|
||||
id: job.id,
|
||||
@@ -571,7 +575,7 @@ function importJobResponse(job: ImportJobRecord) {
|
||||
...(job.error ? { error: job.error } : {}),
|
||||
...(job.result ? { result: job.result } : {}),
|
||||
},
|
||||
retryAfterMs: 1000,
|
||||
...(isTerminal ? {} : { retryAfterMs: 1000 }),
|
||||
};
|
||||
if (job.error?.message) {
|
||||
response.error = job.error.message;
|
||||
@@ -581,6 +585,16 @@ function importJobResponse(job: ImportJobRecord) {
|
||||
return response;
|
||||
}
|
||||
|
||||
function cleanupTerminalImportJobs(importJobs: Map<string, ImportJobRecord>, terminalRetentionMs: number) {
|
||||
const now = Date.now();
|
||||
for (const [jobId, job] of importJobs) {
|
||||
if (job.status === "running" || !job.completedAt) continue;
|
||||
if (now - Date.parse(job.completedAt) > terminalRetentionMs) {
|
||||
importJobs.delete(jobId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function errorMessage(error: unknown) {
|
||||
return error instanceof Error && error.message.trim() ? error.message : String(error);
|
||||
}
|
||||
Reference in new issue
Block a user