diff --git a/apps/worker/worker.ts b/apps/worker/worker.ts index e58c6553..43be035e 100644 --- a/apps/worker/worker.ts +++ b/apps/worker/worker.ts @@ -109,6 +109,7 @@ async function startWorkerProcess() { { deleteGraphFiles }, { processFile, processFiles }, { updateDescriptions }, + { processDescriptionsGroups }, ] = await Promise.all([ import("."), import("./env"), @@ -116,6 +117,7 @@ async function startWorkerProcess() { import("./workflows/delete-graph-files"), import("./workflows/process-file"), import("./workflows/update-descriptions"), + import("./workflows/process-descriptions-group"), ]); const parentPid = process.ppid; @@ -126,6 +128,7 @@ async function startWorkerProcess() { ow.implementWorkflow(deleteProjectFile.spec, deleteProjectFile.fn); ow.implementWorkflow(deleteGraphFiles.spec, deleteGraphFiles.fn); ow.implementWorkflow(updateDescriptions.spec, updateDescriptions.fn); + ow.implementWorkflow(processDescriptionsGroups.spec, processDescriptionsGroups.fn); const worker = ow.newWorker({ concurrency }); let shuttingDown = false; diff --git a/apps/worker/workflows/process-descriptions-group-spec.ts b/apps/worker/workflows/process-descriptions-group-spec.ts new file mode 100644 index 00000000..1e56df88 --- /dev/null +++ b/apps/worker/workflows/process-descriptions-group-spec.ts @@ -0,0 +1,26 @@ +import { defineWorkflowSpec } from "openworkflow"; +import z from "zod"; + +/** + * Maximum number of update-descriptions sub-workflow calls per group. + * Keep this low enough that the group workflow itself stays well under + * the OpenWorkflow step limit (1000), while ensuring the number of groups + * spawned by process-files also stays bounded. + */ +export const DESCRIPTION_BATCHES_PER_GROUP = 50; + +export const processDescriptionsGroupsSpec = defineWorkflowSpec({ + name: "process-descriptions-groups", + version: "1.0.0", + retryPolicy: { + initialInterval: "1s", + backoffCoefficient: 2, + maximumInterval: "30s", + maximumAttempts: 3, + }, + schema: z.object({ + graphId: z.string(), + entityIds: z.array(z.string()), + relationshipIds: z.array(z.string()), + }), +}); \ No newline at end of file diff --git a/apps/worker/workflows/process-descriptions-group.ts b/apps/worker/workflows/process-descriptions-group.ts new file mode 100644 index 00000000..3dd4ef75 --- /dev/null +++ b/apps/worker/workflows/process-descriptions-group.ts @@ -0,0 +1,41 @@ +import { defineWorkflow } from "openworkflow"; +import { updateDescriptionsSpec } from "./update-descriptions-spec"; +import { processDescriptionsGroupsSpec } from "./process-descriptions-group-spec"; +import { chunkItems } from "../lib/chunk"; +import { DESCRIPTION_BATCH_SIZE } from "../lib/description-workflow"; + +/** + * Spawns update-descriptions sub-workflows for a slice of entity/relationship IDs. + * Each process-descriptions-group call handles at most DESCRIPTION_BATCHES_PER_GROUP + * worth of IDs (pre-sliced by process-files), so this simply batches them for + * update-descriptions without additional grouping. + */ +export const processDescriptionsGroups = defineWorkflow( + processDescriptionsGroupsSpec, + async ({ input, step }) => { + const entityIdBatches = chunkItems(input.entityIds, DESCRIPTION_BATCH_SIZE); + const relationshipIdBatches = chunkItems(input.relationshipIds, DESCRIPTION_BATCH_SIZE); + + const entityPromises = entityIdBatches.map((entityIds) => + step.runWorkflow(updateDescriptionsSpec, { + graphId: input.graphId, + entityIds, + relationshipIds: [], + }) + ); + + const relationshipPromises = relationshipIdBatches.map((relationshipIds) => + step.runWorkflow(updateDescriptionsSpec, { + graphId: input.graphId, + entityIds: [], + relationshipIds, + }) + ); + + const results = await Promise.allSettled([...entityPromises, ...relationshipPromises]); + const failures = results.filter((r) => r.status === "rejected"); + if (failures.length > 0) { + throw new Error(`${failures.length} of ${results.length} description batches failed`); + } + } +); \ No newline at end of file diff --git a/apps/worker/workflows/process-file.ts b/apps/worker/workflows/process-file.ts index f5d486b6..ecd2083e 100644 --- a/apps/worker/workflows/process-file.ts +++ b/apps/worker/workflows/process-file.ts @@ -35,7 +35,7 @@ import { chunkItems } from "../lib/chunk"; import { processFilesSpec } from "./process-files-spec"; import { deleteGraphFileProcessingArtifacts, getGraphFileArtifactPaths } from "../lib/derived-files"; import { buildMetadata, buildMetadataExcerpt } from "../lib/metadata"; -import { updateDescriptionsSpec } from "./update-descriptions-spec"; +import { processDescriptionsGroupsSpec, DESCRIPTION_BATCHES_PER_GROUP } from "./process-descriptions-group-spec"; import { DESCRIPTION_BATCH_SIZE } from "../lib/description-workflow"; import { getFileTypeProcessingConfig } from "../lib/file-type-config"; import { toTextUnitRows } from "../lib/text-unit-rows"; @@ -200,20 +200,42 @@ export const processFiles = defineWorkflow(processFilesSpec, async ({ input, ste }; }); - await Promise.all([ - ...chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE).map((entityIds) => - step.runWorkflow(updateDescriptionsSpec, { + // Two-level fan-out: spawn groups of description batches to avoid exceeding + // the OpenWorkflow step limit (1000) on large projects with many entities/relationships. + // Each group spawns update-descriptions sub-workflows internally, so the parent workflow + // only counts one step per group instead of one step per batch. + const totalEntityBatches = chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE).length; + const totalRelationshipBatches = chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE).length; + const totalBatches = totalEntityBatches + totalRelationshipBatches; + const groupsCount = Math.ceil(totalBatches / DESCRIPTION_BATCHES_PER_GROUP); + + const entityIdBatches = chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE); + const relationshipIdBatches = chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE); + const allBatches = [ + ...entityIdBatches.map((ids) => ({ entityIds: ids, relationshipIds: [] as string[] })), + ...relationshipIdBatches.map((ids) => ({ entityIds: [] as string[], relationshipIds: ids })), + ]; + + const groupPromises = []; + for (let i = 0; i < groupsCount; i++) { + const groupBatches = allBatches.slice(i * DESCRIPTION_BATCHES_PER_GROUP, (i + 1) * DESCRIPTION_BATCHES_PER_GROUP); + const groupEntityIds = groupBatches.flatMap((b) => b.entityIds); + const groupRelationshipIds = groupBatches.flatMap((b) => b.relationshipIds); + + groupPromises.push( + step.runWorkflow(processDescriptionsGroupsSpec, { graphId: input.graphId, - entityIds, + entityIds: groupEntityIds, + relationshipIds: groupRelationshipIds, }) - ), - ...chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE).map((relationshipIds) => - step.runWorkflow(updateDescriptionsSpec, { - graphId: input.graphId, - relationshipIds, - }) - ), - ]); + ); + } + + const groupResults = await Promise.allSettled(groupPromises); + const failedGroups = groupResults.filter((r) => r.status === "rejected"); + if (failedGroups.length > 0) { + throw new Error(`${failedGroups.length} of ${groupResults.length} description groups failed`); + } await step.run({ name: "finalize-project-status" }, async () => { await db.update(graphTable).set({ state: "ready" }).where(eq(graphTable.id, input.graphId));