open method

Future<void> open()

Loads the manifest metadata and replays pending WAL objects.

Segment payloads are not downloaded here; the first read of a segment fetches it.

Implementation

Future<void> open() async {
  final manifestBytes = await store.get(_key(manifestKey));
  if (manifestBytes != null) {
    final manifest = Manifest.decode(manifestBytes);
    _generation = manifest.generation;
    for (final info in manifest.segments) {
      _segmentInfos.add(info);
      _trackSegmentId(info.name);
    }
  }

  final walKeys = await store.list(_key(walPrefix));
  walKeys.sort();
  // Fetch pending WAL objects in small concurrent bursts: an object store's
  // cost is round trips, not bandwidth, and replay order is preserved below.
  const burst = 8;
  final payloads = <String, Uint8List?>{};
  for (var start = 0; start < walKeys.length; start += burst) {
    final end = start + burst < walKeys.length
        ? start + burst
        : walKeys.length;
    final slice = walKeys.sublist(start, end);
    final fetched = await Future.wait(<Future<Uint8List?>>[
      for (final key in slice) store.get(key),
    ]);
    for (var i = 0; i < slice.length; i++) {
      payloads[slice[i]] = fetched[i];
    }
  }
  for (final key in walKeys) {
    final payload = payloads[key];
    if (payload == null) {
      // The key index listed an object that does not exist. Silently
      // skipping it would drop a committed write; refuse to open instead.
      throw VelxoCorruptException(
        'object store lost a write-ahead object: $key',
      );
    }
    WalBatch.decode(payload, (kind, walKey, value) {
      if (kind == WalOpKind.put) {
        final ref = _arena.append(value!);
        _memtable.put(walKey, ref.chunk, ref.offset, ref.length);
      } else {
        _memtable.putTombstone(walKey);
      }
    });
    _walKeys.add(key);
    _trackWalSeq(key);
  }
}