compact method

Future<int> compact({ required Duration horizon, })

Permanently removes tombstoned records (deleted documents and removed fields) whose HLC wall-clock is older than horizon. Returns the number of records physically reclaimed.

Resurrection caveat

A peer that was offline for longer than horizon and never observed the delete may, on reconnection, resurrect the supposedly deleted data. This is inherent to CRDT convergence — once a tombstone is physically removed there is no record of the deletion to re-sync. Choose a horizon that comfortably exceeds the longest time any peer is expected to stay offline.

Idempotent

Calling compact repeatedly with the same horizon is safe: records already removed are simply not present and the reclaimed count is 0.

Watchers

Existing watch subscriptions survive compact: the engine is rebuilt under the hood and every active watcher is re-pointed at the new engine. A fresh snapshot is emitted to all watchers after compaction.

Implementation

Future<int> compact({required Duration horizon}) async {
  await _ready;
  var reclaimed = 0;
  await _serialize(() async {
    if (_engine.isPersistent) {
      reclaimed = await _engine.compact(_name, horizon: horizon);
      if (reclaimed > 0) {
        // Not awaited: a query watcher's refresh samples through
        // [_serialize], and compact is the serialized action right now —
        // awaiting it here would deadlock the chain on itself.
        for (final refresh in _watcherRefresh.values.toList()) {
          unawaited(refresh());
        }
      }
      return;
    }

    final cutoff = DateTime.now().subtract(horizon);
    final changeset = await _engine.changeset(_name);
    final allRows = changeset[_name] ?? const [];
    if (allRows.isEmpty) return;

    // Phase 1 — identify tombstoned existence registers beyond the horizon.
    final deadDocs = <String>{};
    for (final r in allRows) {
      final key = r['key'] as String;
      final sep = key.indexOf(_sep);
      if (sep < 0 && (r['is_deleted'] as bool)) {
        final hlc = r['hlc'] as Hlc;
        if (hlc.dateTime.isBefore(cutoff)) {
          deadDocs.add(key);
          reclaimed++;
        }
      }
    }

    // Phase 2 — keep only records not belonging to dead or beyond-horizon
    // tombstones.
    final survivors = <Map<String, Object?>>[];
    for (final r in allRows) {
      final key = r['key'] as String;
      final sep = key.indexOf(_sep);
      if (sep >= 0) {
        final docId = key.substring(0, sep);
        if (deadDocs.contains(docId)) {
          // Field of a deleted doc whose existence tombstone was compacted.
          reclaimed++;
          continue;
        }
        if (r['is_deleted'] as bool) {
          final hlc = r['hlc'] as Hlc;
          if (hlc.dateTime.isBefore(cutoff)) {
            // Field tombstone older than the horizon.
            reclaimed++;
            continue;
          }
        }
      } else if (deadDocs.contains(key)) {
        // Existence register already marked as dead in phase 1 — omit.
        continue;
      }
      survivors.add(r);
    }

    if (reclaimed == 0) return;

    // Rebuild engine with the survivor set.
    final newEngine = TypedCrdtTable(MapCrdtTable(_name));
    if (survivors.isNotEmpty) {
      await newEngine.merge({_name: survivors});
    }

    // Re-point watchers at the replacement engine.
    for (final entry in _watcherSubs.entries) {
      final controller = entry.key;
      await entry.value?.cancel();
      final refresh = _watcherRefresh[controller];
      _watcherSubs[controller] =
          newEngine.watch(_name).listen((_) => refresh?.call());
    }

    // Swap engine.
    _engine = newEngine;

    // Emit fresh snapshots so watchers see the post-compact state. Not
    // awaited: a query watcher's refresh samples through [_serialize], and
    // compact is the serialized action right now.
    for (final refresh in _watcherRefresh.values.toList()) {
      unawaited(refresh());
    }
  });
  return reclaimed;
}