From 6dd968d03060d7734e097e9374c771c41b4d2d37 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 21 Apr 2026 15:19:58 -0500 Subject: [PATCH] fix(memory-jobs): address review and CI feedback --- .../0063_motionless_gertrude_yorkes.sql | 44 +++---------------- .../db/src/migrations/meta/0063_snapshot.json | 12 ++--- .../db/src/schema/memory_extraction_jobs.ts | 12 ++--- packages/shared/src/validators/memory.ts | 7 +++ .../src/__tests__/memory-job-capture.test.ts | 18 +++++++- .../__tests__/memory-job-dispatcher.test.ts | 21 ++++++++- .../src/__tests__/memory-job-routes.test.ts | 19 ++++++++ .../server-startup-feedback-export.test.ts | 26 +++++++++++ server/src/services/memory-job-capture.ts | 4 +- server/src/services/memory-job-dispatcher.ts | 18 ++++---- 10 files changed, 118 insertions(+), 63 deletions(-) diff --git a/packages/db/src/migrations/0063_motionless_gertrude_yorkes.sql b/packages/db/src/migrations/0063_motionless_gertrude_yorkes.sql index 994a2f2285a..b1dc110039b 100644 --- a/packages/db/src/migrations/0063_motionless_gertrude_yorkes.sql +++ b/packages/db/src/migrations/0063_motionless_gertrude_yorkes.sql @@ -32,36 +32,6 @@ CREATE TABLE IF NOT EXISTS "memory_extraction_jobs" ( "updated_at" timestamp with time zone DEFAULT now() NOT NULL ); --> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "company_id" uuid NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "binding_id" uuid NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "binding_key" text NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "operation_type" text NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "status" text DEFAULT 'queued' NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_agent_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_issue_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_project_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_goal_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_heartbeat_run_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "hook_kind" text;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "provider_job_id" text;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "submitted_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "started_at" timestamp with time zone;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "finished_at" timestamp with time zone;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "attribution_mode" text DEFAULT 'untracked' NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "cost_cents" integer DEFAULT 0 NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "result_summary" text;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "error_code" text;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "error" text;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_kind" text NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "source_ref_json" jsonb;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "retry_of_job_id" uuid;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "attempt_number" integer DEFAULT 1 NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "dispatcher_kind" text DEFAULT 'in_process' NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "lease_expires_at" timestamp with time zone;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "usage_json" jsonb;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "result_json" jsonb;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "created_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint -ALTER TABLE "memory_extraction_jobs" ADD COLUMN IF NOT EXISTS "updated_at" timestamp with time zone DEFAULT now() NOT NULL;--> statement-breakpoint DROP INDEX IF EXISTS "issues_open_routine_execution_uq";--> statement-breakpoint ALTER TABLE "issues" ADD COLUMN IF NOT EXISTS "origin_fingerprint" text DEFAULT 'default' NOT NULL;--> statement-breakpoint ALTER TABLE "routine_runs" ADD COLUMN IF NOT EXISTS "dispatch_fingerprint" text;--> statement-breakpoint @@ -71,7 +41,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_company_id_companies_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -80,7 +50,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_source_agent_id_agents_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_agent_id_agents_id_fk" FOREIGN KEY ("source_agent_id") REFERENCES "public"."agents"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_agent_id_agents_id_fk" FOREIGN KEY ("source_agent_id") REFERENCES "public"."agents"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -89,7 +59,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_source_issue_id_issues_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_issue_id_issues_id_fk" FOREIGN KEY ("source_issue_id") REFERENCES "public"."issues"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_issue_id_issues_id_fk" FOREIGN KEY ("source_issue_id") REFERENCES "public"."issues"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -98,7 +68,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_source_project_id_projects_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_project_id_projects_id_fk" FOREIGN KEY ("source_project_id") REFERENCES "public"."projects"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_project_id_projects_id_fk" FOREIGN KEY ("source_project_id") REFERENCES "public"."projects"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -107,7 +77,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_source_goal_id_goals_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_goal_id_goals_id_fk" FOREIGN KEY ("source_goal_id") REFERENCES "public"."goals"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_goal_id_goals_id_fk" FOREIGN KEY ("source_goal_id") REFERENCES "public"."goals"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -116,7 +86,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("source_heartbeat_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("source_heartbeat_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint DO $$ BEGIN @@ -125,7 +95,7 @@ DO $$ BEGIN WHERE conname = 'memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk' AND conrelid = 'memory_extraction_jobs'::regclass ) THEN - ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk" FOREIGN KEY ("retry_of_job_id") REFERENCES "public"."memory_extraction_jobs"("id") ON DELETE no action ON UPDATE no action; + ALTER TABLE "memory_extraction_jobs" ADD CONSTRAINT "memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk" FOREIGN KEY ("retry_of_job_id") REFERENCES "public"."memory_extraction_jobs"("id") ON DELETE set null ON UPDATE no action; END IF; END $$;--> statement-breakpoint CREATE INDEX IF NOT EXISTS "memory_extraction_jobs_company_status_submitted_idx" ON "memory_extraction_jobs" USING btree ("company_id","status","submitted_at");--> statement-breakpoint diff --git a/packages/db/src/migrations/meta/0063_snapshot.json b/packages/db/src/migrations/meta/0063_snapshot.json index 201ea25dcb0..f5c02d2bedc 100644 --- a/packages/db/src/migrations/meta/0063_snapshot.json +++ b/packages/db/src/migrations/meta/0063_snapshot.json @@ -10704,7 +10704,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "cascade", "onUpdate": "no action" }, "memory_extraction_jobs_source_agent_id_agents_id_fk": { @@ -10717,7 +10717,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "set null", "onUpdate": "no action" }, "memory_extraction_jobs_source_issue_id_issues_id_fk": { @@ -10730,7 +10730,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "set null", "onUpdate": "no action" }, "memory_extraction_jobs_source_project_id_projects_id_fk": { @@ -10743,7 +10743,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "set null", "onUpdate": "no action" }, "memory_extraction_jobs_source_goal_id_goals_id_fk": { @@ -10756,7 +10756,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "set null", "onUpdate": "no action" }, "memory_extraction_jobs_source_heartbeat_run_id_heartbeat_runs_id_fk": { @@ -10769,7 +10769,7 @@ "columnsTo": [ "id" ], - "onDelete": "no action", + "onDelete": "set null", "onUpdate": "no action" }, "memory_extraction_jobs_retry_of_job_id_memory_extraction_jobs_id_fk": { diff --git a/packages/db/src/schema/memory_extraction_jobs.ts b/packages/db/src/schema/memory_extraction_jobs.ts index 025d92a3fb7..354cf7b3b81 100644 --- a/packages/db/src/schema/memory_extraction_jobs.ts +++ b/packages/db/src/schema/memory_extraction_jobs.ts @@ -27,16 +27,16 @@ export const memoryExtractionJobs = pgTable( "memory_extraction_jobs", { id: uuid("id").primaryKey().defaultRandom(), - companyId: uuid("company_id").notNull().references(() => companies.id), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), bindingId: uuid("binding_id").notNull(), bindingKey: text("binding_key").notNull(), operationType: text("operation_type").$type().notNull(), status: text("status").$type().notNull().default("queued"), - sourceAgentId: uuid("source_agent_id").references(() => agents.id), - sourceIssueId: uuid("source_issue_id").references(() => issues.id), - sourceProjectId: uuid("source_project_id").references(() => projects.id), - sourceGoalId: uuid("source_goal_id").references(() => goals.id), - sourceHeartbeatRunId: uuid("source_heartbeat_run_id").references(() => heartbeatRuns.id), + sourceAgentId: uuid("source_agent_id").references(() => agents.id, { onDelete: "set null" }), + sourceIssueId: uuid("source_issue_id").references(() => issues.id, { onDelete: "set null" }), + sourceProjectId: uuid("source_project_id").references(() => projects.id, { onDelete: "set null" }), + sourceGoalId: uuid("source_goal_id").references(() => goals.id, { onDelete: "set null" }), + sourceHeartbeatRunId: uuid("source_heartbeat_run_id").references(() => heartbeatRuns.id, { onDelete: "set null" }), hookKind: text("hook_kind").$type(), providerJobId: text("provider_job_id"), submittedAt: timestamp("submitted_at", { withTimezone: true }).notNull().defaultNow(), diff --git a/packages/shared/src/validators/memory.ts b/packages/shared/src/validators/memory.ts index f7253eeec3f..1f6433a826a 100644 --- a/packages/shared/src/validators/memory.ts +++ b/packages/shared/src/validators/memory.ts @@ -40,6 +40,13 @@ export const listMemoryExtractionJobsQuerySchema = z }) .strict() .superRefine((value, ctx) => { + if (value.status && value.effectiveState) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["effectiveState"], + message: "effectiveState and status cannot be used together", + }); + } if (value.submittedAfter && value.submittedBefore && value.submittedAfter > value.submittedBefore) { ctx.addIssue({ code: z.ZodIssueCode.custom, diff --git a/server/src/__tests__/memory-job-capture.test.ts b/server/src/__tests__/memory-job-capture.test.ts index efd571afb4c..44588e2e723 100644 --- a/server/src/__tests__/memory-job-capture.test.ts +++ b/server/src/__tests__/memory-job-capture.test.ts @@ -35,6 +35,22 @@ async function insertCompany(db: ReturnType, companyId: string) }); } +async function waitForJobStatus( + store: ReturnType, + companyId: string, + jobId: string, + status: string, +) { + for (let attempt = 0; attempt < 20; attempt += 1) { + const detail = await store.getDetail(companyId, jobId); + if (detail?.status === status) { + return detail; + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return store.getDetail(companyId, jobId); +} + describeEmbeddedPostgres("memory job capture", () => { let db!: ReturnType; let tempDb: Awaited> | null = null; @@ -193,7 +209,7 @@ describeEmbeddedPostgres("memory job capture", () => { await dispatcher.tick(); - const completed = await store.getDetail(companyId, queued.id); + const completed = await waitForJobStatus(store, companyId, queued.id, "succeeded"); expect(completed).toMatchObject({ id: queued.id, status: "succeeded", diff --git a/server/src/__tests__/memory-job-dispatcher.test.ts b/server/src/__tests__/memory-job-dispatcher.test.ts index 9e3384f079e..47f33512a13 100644 --- a/server/src/__tests__/memory-job-dispatcher.test.ts +++ b/server/src/__tests__/memory-job-dispatcher.test.ts @@ -30,6 +30,23 @@ async function insertCompany(db: ReturnType, companyId: string) }); } +async function waitForJobStatus( + store: ReturnType, + companyId: string, + jobId: string, + status: string, + now: Date, +) { + for (let attempt = 0; attempt < 20; attempt += 1) { + const detail = await store.getDetail(companyId, jobId, { now }); + if (detail?.status === status) { + return detail; + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + return store.getDetail(companyId, jobId, { now }); +} + describeEmbeddedPostgres("memory job dispatcher", () => { let db!: ReturnType; let tempDb: Awaited> | null = null; @@ -83,7 +100,7 @@ describeEmbeddedPostgres("memory job dispatcher", () => { await dispatcher.tick(); - const detail = await store.getDetail(companyId, queued.id, { now: currentTime }); + const detail = await waitForJobStatus(store, companyId, queued.id, "succeeded", currentTime); expect(detail).toMatchObject({ id: queued.id, @@ -132,7 +149,7 @@ describeEmbeddedPostgres("memory job dispatcher", () => { await dispatcher.tick(); - const runningDetail = await store.getDetail(companyId, queued.id, { now: currentTime }); + const runningDetail = await waitForJobStatus(store, companyId, queued.id, "running", currentTime); expect(runningDetail).toMatchObject({ id: queued.id, status: "running", diff --git a/server/src/__tests__/memory-job-routes.test.ts b/server/src/__tests__/memory-job-routes.test.ts index 8b5bb9b6711..26364f2d7bc 100644 --- a/server/src/__tests__/memory-job-routes.test.ts +++ b/server/src/__tests__/memory-job-routes.test.ts @@ -140,6 +140,25 @@ describeEmbeddedPostgres("memory job routes", () => { }); }); + it("rejects ambiguous status and effective-state filters", async () => { + const companyId = randomUUID(); + await insertCompany(db, companyId); + + const app = createApp(db, { + type: "board", + userId: "board", + source: "local_implicit", + isInstanceAdmin: true, + }); + + const res = await request(app).get( + `/api/companies/${companyId}/memory/jobs?status=succeeded&effectiveState=stuck`, + ); + + expect(res.status).toBe(400); + expect(res.body.error).toContain("effectiveState and status cannot be used together"); + }); + it("returns job detail scoped to the requested company", async () => { const companyId = randomUUID(); const jobId = randomUUID(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 863922afd89..806679105c4 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -144,6 +144,32 @@ vi.mock("../services/index.js", () => ({ })), })); +vi.mock("../services/memory-job-capture.js", () => ({ + createPostRunMemoryCaptureHandler: vi.fn(() => vi.fn()), +})); + +vi.mock("../services/memory-job-dispatcher.js", () => ({ + createMemoryJobDispatcher: vi.fn(() => ({ + start: vi.fn(), + stop: vi.fn(), + tick: vi.fn(async () => undefined), + sweepExpiredLeases: vi.fn(async () => ({ recovered: 0, jobIds: [] })), + diagnostics: vi.fn(() => ({ + running: false, + activeJobCount: 0, + activeJobIds: [], + tickCount: 0, + lastTickAt: null, + recoveryCount: 0, + lastRecoveryAt: null, + })), + })), +})); + +vi.mock("../services/memory-job-store.js", () => ({ + memoryJobStore: vi.fn(() => ({})), +})); + vi.mock("../storage/index.js", () => ({ createStorageServiceFromConfig: vi.fn(() => ({ id: "storage-service" })), })); diff --git a/server/src/services/memory-job-capture.ts b/server/src/services/memory-job-capture.ts index f84c4125c78..b0cad9e172d 100644 --- a/server/src/services/memory-job-capture.ts +++ b/server/src/services/memory-job-capture.ts @@ -69,7 +69,9 @@ function readFiniteField(record: Record | null, key: string): n } function formatUuidFromHex(hex: string): string { - return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20, 32)}`; + const version = `4${hex.slice(13, 16)}`; + const variant = `${((parseInt(hex[16] ?? "0", 16) & 0x3) | 0x8).toString(16)}${hex.slice(17, 20)}`; + return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${version}-${variant}-${hex.slice(20, 32)}`; } export function buildSyntheticMemoryBindingId(companyId: string, bindingKey: string): string { diff --git a/server/src/services/memory-job-dispatcher.ts b/server/src/services/memory-job-dispatcher.ts index 7d5b94b5adb..1eed98176dc 100644 --- a/server/src/services/memory-job-dispatcher.ts +++ b/server/src/services/memory-job-dispatcher.ts @@ -58,6 +58,10 @@ export interface MemoryJobDispatcher { diagnostics(): MemoryJobDispatcherDiagnostics; } +function childLogger(bindings: Record): typeof logger { + return typeof logger.child === "function" ? logger.child(bindings) : logger; +} + export function createMemoryJobDispatcher( options: MemoryJobDispatcherOptions, ): MemoryJobDispatcher { @@ -66,7 +70,7 @@ export function createMemoryJobDispatcher( const leaseDurationMs = options.leaseDurationMs ?? DEFAULT_LEASE_DURATION_MS; const maxConcurrentJobs = options.maxConcurrentJobs ?? DEFAULT_MAX_CONCURRENT_JOBS; const now = options.now ?? (() => new Date()); - const log = logger.child({ service: "memory_job_dispatcher" }); + const log = childLogger({ service: "memory_job_dispatcher" }); let pollTimer: ReturnType | null = null; let recoveryTimer: ReturnType | null = null; @@ -81,12 +85,12 @@ export function createMemoryJobDispatcher( const activeJobs = new Set(); async function runClaimedJob(job: ClaimedMemoryJob): Promise { - const jobLog = log.child({ + const jobLog = typeof log.child === "function" ? log.child({ jobId: job.id, companyId: job.companyId, bindingKey: job.bindingKey, operationType: job.operationType, - }); + }) : log; try { const handler = await options.resolveHandler(job); @@ -185,8 +189,6 @@ export function createMemoryJobDispatcher( tickCount += 1; lastTickAt = now(); - const dispatches: Promise[] = []; - try { while (activeJobs.size < maxConcurrentJobs) { const claimed = await options.store.claimNext({ @@ -200,11 +202,7 @@ export function createMemoryJobDispatcher( } activeJobs.add(claimed.id); - dispatches.push(runClaimedJob(claimed)); - } - - if (dispatches.length > 0) { - await Promise.allSettled(dispatches); + void runClaimedJob(claimed); } } catch (error) { log.error(