refactor(ledger): blockstore database - misc improvements (#264)

## RocksDB upgrade

Upgrade dependency to latest version of rocksdb-zig which includes:
- increment underlying rocksdb version
- add `destroy` method to wipe the entire database from the disk
- rename function for consistency with c++ code

## `Database.count` method

Add `count` method to Database to return the number of entries in a column family.

Previously the cleanup service circumvented the Database interface and directly called the rocksdb live files to count the number of entries. Now it can just call `count`.

## tests

Add tests for Database write batches and iterators, which were missing test coverage. This revealed some bugs in HashmapDB and SortedMap which are now fixed. Also added tests to utils/collections.zig to increase test coverage for the bug there.

## `Database.get` allocator parameter

Database.get now takes an allocator parameter. this is to improve memory safety. This means the calling scope:
  - has clear ownership over the returned item
  - has control over how its own data is allocated
  - can clean up the data properly with its own allocator

build.zig.zon

@@ -32,8 +32,8 @@
             .hash = "1220f70ac854b59315a8512861e039648d677feb4f9677bd873d6b9b7074a5906485",
         },
         .rocksdb = .{
-            .url = "https://github.com/Syndica/rocksdb-zig/archive/6fa5e95875941236768e87a8e0c5ef45c75c9b56.tar.gz",
-            .hash = "122067ae3d6a64b0e1401dd40059535cb782d0c1a99c39a83f24c2385391e0b22876",
+            .url = "https://github.com/Syndica/rocksdb-zig/archive/9c09659a5e41f226b6b8f3fa21149247eb26dfae.tar.gz",
+            .hash = "1220aeb80b2f8bb48c131ef306fe48ddfcb537210c5f77742e921cbf40fc4c19b41e",
         },
     },
 }

src/cmd/cmd.zig

@@ -691,7 +691,6 @@ fn validator() !void {
     );
 
     var cleanup_service_handle = try std.Thread.spawn(.{}, sig.ledger.cleanup_service.run, .{
-        allocator,
         app_base.logger,
         blockstore_reader,
         blockstore_writer,
@@ -789,7 +788,6 @@ fn shredCollector() !void {
     );
 
     var cleanup_service_handle = try std.Thread.spawn(.{}, sig.ledger.cleanup_service.run, .{
-        allocator,
         app_base.logger,
         blockstore_reader,
         blockstore_writer,

src/ledger/cleanup_service.zig

@@ -26,7 +26,6 @@ const DEFAULT_CLEANUP_SLOT_INTERVAL: u64 = 512;
 const LOOP_LIMITER = Duration.fromMillis(DEFAULT_CLEANUP_SLOT_INTERVAL * DEFAULT_MS_PER_SLOT / 10);
 
 pub fn run(
-    allocator: std.mem.Allocator,
     logger: sig.trace.Logger,
     blockstore_reader: *BlockstoreReader,
     blockstore_writer: *BlockstoreWriter,
@@ -38,7 +37,6 @@ pub fn run(
     logger.info("Starting blockstore cleanup service");
     while (!exit.load(.unordered)) {
         last_purge_slot = try cleanBlockstore(
-            allocator,
             logger,
             blockstore_reader,
             blockstore_writer,
@@ -71,7 +69,6 @@ pub fn run(
 ///
 /// Analogous to the [`cleanup_ledger`](https://github.com/anza-xyz/agave/blob/6476d5fac0c30d1f49d13eae118b89be78fb15d2/ledger/src/blockstore_cleanup_service.rs#L198) in agave:
 pub fn cleanBlockstore(
-    allocator: std.mem.Allocator,
     logger: sig.trace.Logger,
     blockstore_reader: *BlockstoreReader,
     blockstore_writer: *BlockstoreWriter,
@@ -87,12 +84,7 @@ pub fn cleanBlockstore(
 
     // NOTE: this will clean everything past the lowest slot in the blockstore
     const root: Slot = try blockstore_reader.lowestSlot();
-    const result = try findSlotsToClean(
-        allocator,
-        blockstore_reader,
-        root,
-        max_ledger_shreds,
-    );
+    const result = try findSlotsToClean(blockstore_reader, root, max_ledger_shreds);
     logger.infof("findSlotsToClean result: {any}", .{result});
 
     if (result.should_clean) {
@@ -127,7 +119,6 @@ pub fn cleanBlockstore(
 ///
 /// Analogous to the [`find_slots_to_clean`](https://github.com/anza-xyz/agave/blob/6476d5fac0c30d1f49d13eae118b89be78fb15d2/ledger/src/blockstore_cleanup_service.rs#L103)
 fn findSlotsToClean(
-    allocator: std.mem.Allocator,
     blockstore_reader: *BlockstoreReader,
     max_root: Slot,
     max_ledger_shreds: u64,
@@ -136,17 +127,7 @@ fn findSlotsToClean(
     highest_slot_to_purge: Slot,
     total_shreds: u64,
 } {
-    const data_shred_cf_name = Schema.data_shred.name;
-
-    const live_files = try blockstore_reader.db.db.liveFiles(allocator);
-    defer live_files.deinit();
-
-    var num_shreds: u64 = 0;
-    for (live_files.items) |live_file| {
-        if (std.mem.eql(u8, live_file.column_family_name, data_shred_cf_name)) {
-            num_shreds += live_file.num_entries;
-        }
-    }
+    const num_shreds = try blockstore_reader.db.count(Schema.data_shred);
 
     // Using the difference between the lowest and highest slot seen will
     // result in overestimating the number of slots in the blockstore since
@@ -229,12 +210,13 @@ test "findSlotsToClean" {
 
     {
         var write_batch = try db.initWriteBatch();
+        defer write_batch.deinit();
         try write_batch.put(ledger.schema.schema.slot_meta, lowest_slot_meta.slot, lowest_slot_meta);
         try write_batch.put(ledger.schema.schema.slot_meta, highest_slot_meta.slot, highest_slot_meta);
         try db.commit(write_batch);
     }
 
-    const r = try findSlotsToClean(allocator, &reader, 0, 100);
+    const r = try findSlotsToClean(&reader, 0, 100);
     try std.testing.expectEqual(false, r.should_clean);
     try std.testing.expectEqual(0, r.total_shreds);
     try std.testing.expectEqual(0, r.highest_slot_to_purge);