Class RuntimeQueueWorker

runtime/api-server/src/queue-worker.ts:112–755

}

export class RuntimeQueueWorker implements QueueWorkerLike {
  readonly #store: RuntimeStateStore;
  readonly #logger: RuntimeQueueWorkerOptions["logger"];
  readonly #executeClaimedInput: (record: SessionInputRecord, options?: { signal?: AbortSignal }) => Promise<void>;
  readonly #captureRuntimeException: typeof captureRuntimeException;
  readonly #claimedBy: string;
  readonly #leaseSeconds: number;
  readonly #pollIntervalMs: number;
  readonly #maxConcurrency: number;
  readonly #claimStaleHeartbeatMs: number;
  #stopped = false;
  #task: Promise<void> | null = null;
  #wakeResolver: (() => void) | null = null;
  #activeRuns = new Map<
    string,
    {
      controller: AbortController;
      record: SessionInputRecord;
      promise: Promise<void>;
    }
  >();

  constructor(options: RuntimeQueueWorkerOptions) {
    this.#store = options.store;
    this.#logger = options.logger;
    this.#claimedBy = options.claimedBy ?? DEFAULT_CLAIMED_BY;
    this.#captureRuntimeException =
      options.captureRuntimeException ?? captureRuntimeException;
    this.#executeClaimedInput =
      options.executeClaimedInput ??
      ((record, executionOptions) =>
        processClaimedInput({
          store: this.#store,
          record,
          claimedBy: this.#claimedBy,
          leaseSeconds: this.#leaseSeconds,
          memoryService: options.memoryService ?? null,
          wakeDurableMemoryWorker: options.wakeDurableMemoryWorker ?? null,
          abortSignal: executionOptions?.signal,
        }));
    this.#leaseSeconds = options.leaseSeconds ?? DEFAULT_LEASE_SECONDS;
    this.#pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS;
    this.#maxConcurrency = options.maxConcurrency ?? queueWorkerMaxConcurrency();
    this.#claimStaleHeartbeatMs =
      options.claimStaleHeartbeatMs ?? queueWorkerClaimStaleHeartbeatMs();
  }

  async start(): Promise<void> {
    if (this.#task) {
      return;
    }
    this.#stopped = false;
    this.#task = this.#runLoop();
  }

  wake(): void {
    const resolve = this.#wakeResolver;
    this.#wakeResolver = null;