From fc839edb3c12bc0ed312a6ecf66268d8756ec129 Mon Sep 17 00:00:00 2001 From: Matt Krick Date: Mon, 29 Apr 2024 09:53:35 -0700 Subject: [PATCH 1/2] fix: handle 0 discussions Signed-off-by: Matt Krick --- packages/embedder/workflows/relatedDiscussionsStart.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/embedder/workflows/relatedDiscussionsStart.ts b/packages/embedder/workflows/relatedDiscussionsStart.ts index e62a986a182..a9881f2999c 100644 --- a/packages/embedder/workflows/relatedDiscussionsStart.ts +++ b/packages/embedder/workflows/relatedDiscussionsStart.ts @@ -31,6 +31,7 @@ export const relatedDiscussionsStart: JobQueueStepRun< // Not techincally updatedAt since discussions are updated after they get created refUpdatedAt: createdAt })) + if (metadataRows.length === 0) return false const inserts = await pg .insertInto('EmbeddingsMetadata') .values(metadataRows) From 92cecbfbd40bde882144991a8f605a45ba1a9dca Mon Sep 17 00:00:00 2001 From: Matt Krick Date: Mon, 29 Apr 2024 10:02:28 -0700 Subject: [PATCH 2/2] fix: prevent excess retries Signed-off-by: Matt Krick --- packages/embedder/EmbeddingsJobQueueStream.ts | 5 ++++- packages/embedder/WorkflowOrchestrator.ts | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/embedder/EmbeddingsJobQueueStream.ts b/packages/embedder/EmbeddingsJobQueueStream.ts index 49c3e4c9e58..1be69480856 100644 --- a/packages/embedder/EmbeddingsJobQueueStream.ts +++ b/packages/embedder/EmbeddingsJobQueueStream.ts @@ -29,7 +29,10 @@ export class EmbeddingsJobQueueStream implements AsyncIterableIterator { .orderBy(['priority']) .$if(!isFailed, (db) => db.where('state', '=', 'queued')) .$if(isFailed, (db) => - db.where('state', '=', 'failed').where('retryAfter', '<', new Date()) + db + .where('state', '=', 'failed') + .where('retryAfter', 'is not', null) + .where('retryAfter', '<', new Date()) ) .limit(1) .forUpdate() diff --git a/packages/embedder/WorkflowOrchestrator.ts b/packages/embedder/WorkflowOrchestrator.ts index 704320542be..f81f67bcd1a 100644 --- a/packages/embedder/WorkflowOrchestrator.ts +++ b/packages/embedder/WorkflowOrchestrator.ts @@ -40,7 +40,7 @@ export class WorkflowOrchestrator { Logger.error(error) const pg = getKysely() const {message, retryDelay, jobData} = error - const maxRetries = error.maxRetries ?? 1e6 + const maxRetries = error.maxRetries ?? 10 await pg .updateTable('EmbeddingsJobQueue') .set((eb) => ({