Method tryFlush

packages/db/src/buffers/base-buffer.ts:378–488

(options: { trigger?: FlushTrigger } = {})

  }

  async tryFlush(options: { trigger?: FlushTrigger } = {}) {
    const trigger: FlushTrigger = options.trigger ?? 'cron';
    const startedAt = performance.now();
    this.inflightStats = {};

    let llenAtStart: number | undefined;
    try {
      llenAtStart = await this.getBufferSize();
    } catch {
      // best-effort
    }

    const isCronQueuePaused = await cronQueue.isPaused();
    if (isCronQueuePaused) {
      this.emitFlushObservation({
        buffer: this.name,
        result: 'paused',
        trigger,
        totalMs: performance.now() - startedAt,
        llenAtStart,
      });
      return;
    }

    if (this.enableParallelProcessing) {
      const onFlushStarted = performance.now();
      try {
        await this.onFlush();
        this.emitFlushObservation({
          buffer: this.name,
          result: 'success',
          trigger,
          totalMs: performance.now() - startedAt,
          rowsProcessed: this.inflightStats.rowsProcessed,
          llenAtStart,
          phases: {
            ...this.inflightStats.phases,
            onFlushMs: performance.now() - onFlushStarted,
          },
        });
      } catch (error) {
        this.emitFlushObservation({
          buffer: this.name,
          result: 'error',
          trigger,
          totalMs: performance.now() - startedAt,
          rowsProcessed: this.inflightStats.rowsProcessed,
          llenAtStart,
          phases: {
            ...this.inflightStats.phases,
            onFlushMs: performance.now() - onFlushStarted,
          },
          err: error,
        });
      }
      return;
    }