tick static method

Future<void> tick({ DVQueues queues = const DVQueues(), bool work = false, })

One pass of everything capture does on a schedule: changes a crash left staged are published, a destination or a data model that has never been copied is backfilled, pending changes are dispatched for delivery, lag is measured, and the log is pruned to its retention.

With work, this process also runs the jobs it dispatched: it is the whole deployment, with no worker to run them.

Implementation

static Future<void> tick({
  DVQueues queues = const DVQueues(),
  bool work = false,
}) async {
  final _Runtime? runtime = _runtime;
  if (runtime == null || runtime.ticking) return;
  runtime.ticking = true;
  runtime.queues = queues;
  runtime.works = work;
  try {
    await runtime.log.publishStranded();
    for (final DVCaptureConsumer consumer in runtime.consumers) {
      await _backfillWhatIsMissing(runtime, consumer, queues);
    }
    await _dispatchDeliveries();
    if (work) await _work(runtime, queues);
    for (final DVCaptureConsumer consumer in runtime.consumers) {
      if (await consumer.position() == null) continue;
      runtime.lag[consumer.name] = await consumer.lag();
    }
    final DateTime now = DateTime.now();
    final DateTime? pruned = runtime.prunedAt;
    if (pruned == null || now.difference(pruned) >= pruneEvery) {
      runtime.prunedAt = now;
      await runtime.log.prune();
    }
  } on Object catch (error) {
    DVObservability.log(
      'Change capture could not complete its scheduled pass: $error',
      level: DVLogLevel.warn,
    );
  } finally {
    runtime.ticking = false;
  }
}