publishStranded method

Future<int> publishStranded()

Publishes changes a crash left staged: written, never given a sequence.

The rows they describe are in the database -- a compensation-based transaction that dies before compensating leaves its writes in place -- so the honest thing is to publish what the store holds. Changes of a transaction still open in this process are left alone.

Implementation

Future<int> publishStranded() async {
  await _ensureReady();
  final String cutoff = _stamp(_clock().subtract(strandedAfter));
  final List<Map<String, Object?>> rows = await _records.find(
    logTable,
    where: DVFilter.all(<DVFilter>[
      const DVFilter.isNull('change_seq'),
      DVFilter.compare('occurred_at', DVCompare.less, cutoff),
    ]),
    orderBy: const <DVSort>[DVSort('write_order')],
    fields: const <String>['change_id', 'transaction_id', 'write_order'],
  );
  final List<String> ids = <String>[
    for (final Map<String, Object?> row in rows)
      if (!_staged.containsKey(row['transaction_id'])) '${row['change_id']}',
  ];
  if (ids.isEmpty) return 0;
  return _publish(ids);
}