feat(rpc): WIP implement getSlot

Author: ajw221

src/cmd.zig

@@ -1314,6 +1314,9 @@ fn validator(
     else
         null;
 
+    var latest_processed_slot: sig.replay.trackers.ForkChoiceProcessedSlot = .{};
+    var latest_confirmed_slot: sig.replay.trackers.OptimisticallyConfirmedSlot = .{};
+
     var replay_service_state: ReplayAndConsensusServiceState = try .init(allocator, .{
         .app_base = &app_base,
         .account_store = .{ .accounts_db_two = &new_db },
@@ -1325,9 +1328,15 @@ fn validator(
         .voting_enabled = voting_enabled,
         .vote_account_address = maybe_vote_pubkey,
         .stop_at_slot = cfg.stop_at_slot,
+        .latest_processed_slot = &latest_processed_slot,
+        .latest_confirmed_slot = &latest_confirmed_slot,
     });
     defer replay_service_state.deinit(allocator);
 
+    try app_base.rpc_hooks.set(allocator, sig.rpc.methods.SlotHookContext{
+        .slot_tracker = &replay_service_state.replay_state.slot_tracker,
+    });
+
     const replay_thread = try replay_service_state.spawnService(
         &app_base,
         if (maybe_vote_sockets) |*vs| vs else null,
@@ -1529,6 +1538,9 @@ fn replayOffline(
         });
     }
 
+    var latest_processed_slot: sig.replay.trackers.ForkChoiceProcessedSlot = .{};
+    var latest_confirmed_slot: sig.replay.trackers.OptimisticallyConfirmedSlot = .{};
+
     var replay_service_state: ReplayAndConsensusServiceState = try .init(allocator, .{
         .app_base = &app_base,
         .account_store = .{ .accounts_db_two = &new_db },
@@ -1540,6 +1552,8 @@ fn replayOffline(
         .voting_enabled = false,
         .vote_account_address = null,
         .stop_at_slot = cfg.stop_at_slot,
+        .latest_processed_slot = &latest_processed_slot,
+        .latest_confirmed_slot = &latest_confirmed_slot,
     });
     defer replay_service_state.deinit(allocator);
 
@@ -2240,6 +2254,8 @@ const ReplayAndConsensusServiceState = struct {
             voting_enabled: bool,
             vote_account_address: ?Pubkey,
             stop_at_slot: ?Slot,
+            latest_processed_slot: *sig.replay.trackers.ForkChoiceProcessedSlot,
+            latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
         },
     ) !ReplayAndConsensusServiceState {
         var replay_state: replay.service.ReplayState = replay_state: {
@@ -2310,6 +2326,8 @@ const ReplayAndConsensusServiceState = struct {
                 .hard_forks = hard_forks,
                 .replay_threads = params.replay_threads,
                 .stop_at_slot = params.stop_at_slot,
+                .latest_processed_slot = params.latest_processed_slot,
+                .latest_confirmed_slot = params.latest_confirmed_slot,
             }, if (params.disable_consensus) .disabled else .enabled);
         };
         errdefer replay_state.deinit();

src/consensus/vote_listener.zig

@@ -28,7 +28,7 @@ pub const SlotDataProvider = struct {
     epoch_tracker: *const EpochTracker,
 
     pub fn rootSlot(self: *const SlotDataProvider) Slot {
-        return self.slot_tracker.root;
+        return self.slot_tracker.root.load(.monotonic);
     }
 
     fn getSlotHash(self: *const SlotDataProvider, slot: Slot) ?Hash {
@@ -317,6 +317,7 @@ pub const VoteCollector = struct {
     latest_vote_slot_per_validator: sig.utils.collections.PubkeyMap(Slot),
     last_process_root: sig.time.Instant,
     vote_processing_time: VoteProcessingTiming,
+    latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
     metrics: VoteListenerMetrics,
 
     pub fn deinit(self: *VoteCollector, allocator: std.mem.Allocator) void {
@@ -331,6 +332,7 @@ pub const VoteCollector = struct {
         now: sig.time.Instant,
         root_slot: Slot,
         registry: *sig.prometheus.Registry(.{}),
+        latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
     ) !VoteCollector {
         return .{
             .gossip_vote_receptor = .INIT,
@@ -339,6 +341,7 @@ pub const VoteCollector = struct {
             .latest_vote_slot_per_validator = .empty,
             .last_process_root = now,
             .vote_processing_time = .ZEROES,
+            .latest_confirmed_slot = latest_confirmed_slot,
             .metrics = try .init(registry),
         };
     }
@@ -397,6 +400,7 @@ pub const VoteCollector = struct {
             gossip_vote_txs,
             &self.vote_processing_time,
             &self.latest_vote_slot_per_validator,
+            self.latest_confirmed_slot,
             self.metrics,
         );
         defer allocator.free(confirmed_slots);
@@ -420,6 +424,7 @@ fn listenAndConfirmVotes(
     gossip_vote_txs: []const vote_parser.ParsedVote,
     vote_processing_time: ?*VoteProcessingTiming,
     latest_vote_slot_per_validator: *sig.utils.collections.PubkeyMap(Slot),
+    latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
     metrics: VoteListenerMetrics,
 ) std.mem.Allocator.Error![]const ThresholdConfirmedSlot {
     var replay_votes_buffer: std.ArrayListUnmanaged(vote_parser.ParsedVote) = .empty;
@@ -454,6 +459,7 @@ fn listenAndConfirmVotes(
         latest_vote_slot_per_validator,
         gossip_vote_txs,
         replay_votes,
+        latest_confirmed_slot,
         metrics,
     );
 }
@@ -468,6 +474,7 @@ fn filterAndConfirmWithNewVotes(
     latest_vote_slot_per_validator: *sig.utils.collections.PubkeyMap(Slot),
     gossip_vote_txs: []const vote_parser.ParsedVote,
     replayed_votes: []const vote_parser.ParsedVote,
+    latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
     metrics: VoteListenerMetrics,
 ) std.mem.Allocator.Error![]const ThresholdConfirmedSlot {
     const root_slot = slot_data_provider.rootSlot();
@@ -503,6 +510,7 @@ fn filterAndConfirmWithNewVotes(
                 &new_optimistic_confirmed_slots,
                 is_gossip,
                 latest_vote_slot_per_validator,
+                latest_confirmed_slot,
             );
             if (is_gossip)
                 metrics.gossip_votes_processed.inc()
@@ -767,6 +775,7 @@ fn trackNewVotesAndNotifyConfirmations(
     new_optimistic_confirmed_slots: *std.ArrayListUnmanaged(ThresholdConfirmedSlot),
     is_gossip_vote: bool,
     latest_vote_slot_per_validator: *sig.utils.collections.PubkeyMap(Slot),
+    latest_confirmed_slot: *sig.replay.trackers.OptimisticallyConfirmedSlot,
 ) std.mem.Allocator.Error!void {
     if (vote.isEmpty()) return;
     const root = slot_data_provider.rootSlot();
@@ -867,6 +876,7 @@ fn trackNewVotesAndNotifyConfirmations(
                     .slot = slot,
                     .hash = hash,
                 });
+                latest_confirmed_slot.update(slot);
                 // Notify subscribers about new optimistic confirmation
                 if (senders.bank_notification) |sender| {
                     sender.send(.{ .optimistically_confirmed = slot }) catch |err| {

src/replay/consensus/cluster_sync.zig

@@ -492,7 +492,7 @@ fn processAncestorHashesDuplicateSlots(
     slot_tracker: *const SlotTracker,
     duplicate_slots_to_repair: *SlotData.DuplicateSlotsToRepair,
 ) !void {
-    const root = slot_tracker.root;
+    const root = slot_tracker.root.load(.monotonic);
 
     while (ancestor_duplicate_slots_receiver.tryReceive()) |ancestor_dupe_slot_to_repair| {
         const request_type = ancestor_dupe_slot_to_repair.request_type;
@@ -560,7 +560,7 @@ fn processDuplicateConfirmedSlots(
     ancestor_hashes_replay_update_sender: *sig.sync.Channel(AncestorHashesReplayUpdate),
     purge_repair_slot_counter: *SlotData.PurgeRepairSlotCounters,
 ) !void {
-    const root = slot_tracker.root;
+    const root = slot_tracker.root.load(.monotonic);
     for (duplicate_confirmed_slots_received) |new_duplicate_confirmed_slot| {
         const confirmed_slot, const duplicate_confirmed_hash = new_duplicate_confirmed_slot.tuple();
         if (confirmed_slot <= root) {
@@ -642,7 +642,7 @@ fn processPrunedButPopularForks(
     slot_tracker: *const SlotTracker,
     ancestor_hashes_replay_update_sender: *sig.sync.Channel(AncestorHashesReplayUpdate),
 ) !void {
-    const root = slot_tracker.root;
+    const root = slot_tracker.root.load(.monotonic);
     while (pruned_but_popular_forks_receiver.tryReceive()) |new_popular_pruned_slot| {
         if (new_popular_pruned_slot <= root) {
             continue;
@@ -700,7 +700,7 @@ fn processDuplicateSlots(
             });
         }
 
-        break :blk .{ slot_tracker.root, slots_hashes };
+        break :blk .{ slot_tracker.root.load(.monotonic), slots_hashes };
     };
     for (new_duplicate_slots.constSlice(), slots_hashes.constSlice()) |duplicate_slot, slot_hash| {
         // WindowService should only send the signal once per slot
@@ -1563,7 +1563,7 @@ test "apply state changes" {
 
     // MarkSlotDuplicate should mark progress map and remove
     // the slot from fork choice
-    const duplicate_slot = slot_tracker.root + 1;
+    const duplicate_slot = slot_tracker.root.load(.monotonic) + 1;
     const duplicate_slot_hash = slot_tracker.get(duplicate_slot).?.state.hash.readCopy().?;
     // AKA: `ResultingStateChange::MarkSlotDuplicate` in agave
     try heaviest_subtree_fork_choice.markForkInvalidCandidate(allocator, &.{
@@ -1623,7 +1623,7 @@ test "apply state changes slot frozen" {
     var ledger = try ledger_tests.initTestLedger(allocator, @src(), .FOR_TESTS);
     defer ledger.deinit();
 
-    const duplicate_slot = slot_tracker.root + 1;
+    const duplicate_slot = slot_tracker.root.load(.monotonic) + 1;
     const duplicate_slot_hash = slot_tracker.get(duplicate_slot).?.state.hash.readCopy().?;
 
     // Simulate ReplayStage freezing a Slot with the given hash.
@@ -1659,9 +1659,10 @@ test "apply state changes slot frozen" {
     // version in blockstore.
     const new_slot_hash: Hash = .initRandom(random);
     const root_slot_hash: sig.core.hash.SlotAndHash = rsh: {
-        const root_slot_info = slot_tracker.get(slot_tracker.root).?;
+        const root_slot = slot_tracker.root.load(.monotonic);
+        const root_slot_info = slot_tracker.get(root_slot).?;
         break :rsh .{
-            .slot = slot_tracker.root,
+            .slot = root_slot,
             .hash = root_slot_info.state.hash.readCopy().?,
         };
     };
@@ -1711,7 +1712,7 @@ test "apply state changes duplicate confirmed matches frozen" {
     var ledger = try ledger_tests.initTestLedger(allocator, @src(), .FOR_TESTS);
     defer ledger.deinit();
 
-    const duplicate_slot = slot_tracker.root + 1;
+    const duplicate_slot = slot_tracker.root.load(.monotonic) + 1;
     const our_duplicate_slot_hash = slot_tracker.get(duplicate_slot).?.state.hash.readCopy().?;
 
     var duplicate_slots_to_repair: SlotData.DuplicateSlotsToRepair = .empty;
@@ -1807,7 +1808,7 @@ test "apply state changes slot frozen and duplicate confirmed matches frozen" {
     var purge_repair_slot_counter: SlotData.PurgeRepairSlotCounters = .empty;
     defer purge_repair_slot_counter.deinit(allocator);
 
-    const duplicate_slot = slot_tracker.root + 1;
+    const duplicate_slot = slot_tracker.root.load(.monotonic) + 1;
     const our_duplicate_slot_hash = slot_tracker.get(duplicate_slot).?.state.hash.readCopy().?;
 
     // Setup and check the state that is about to change.

src/replay/consensus/core.zig

@@ -155,7 +155,8 @@ pub const TowerConsensus = struct {
         );
         errdefer fork_choice.deinit(allocator);
 
-        const root_ref = deps.slot_tracker.get(deps.slot_tracker.root).?;
+        const root = deps.slot_tracker.root.load(.monotonic);
+        const root_ref = deps.slot_tracker.get(root).?;
         const root_ancestors = &root_ref.constants.ancestors;
 
         var tower: Tower = if (deps.identity.vote_account) |vote_account_address|
@@ -166,8 +167,8 @@ pub const TowerConsensus = struct {
                 vote_account_address,
             )
         else
-            .{ .root = deps.slot_tracker.root };
-        tower.setRoot(deps.slot_tracker.root);
+            .{ .root = root };
+        tower.setRoot(root);
 
         const replay_tower: ReplayTower = try .init(
             .from(deps.logger),
@@ -178,7 +179,7 @@ pub const TowerConsensus = struct {
         errdefer replay_tower.deinit(allocator);
 
         var vote_collector: sig.consensus.VoteCollector =
-            try .init(deps.now, deps.slot_tracker.root, deps.registry);
+            try .init(deps.now, root, deps.registry, deps.slot_tracker.latest_confirmed_slot);
         errdefer vote_collector.deinit(allocator);
 
         return .{
@@ -207,7 +208,7 @@ pub const TowerConsensus = struct {
     ) !HeaviestSubtreeForkChoice {
         const root_slot, const root_hash = blk: {
             const root = slot_tracker.getRoot();
-            const root_slot = slot_tracker.root;
+            const root_slot = slot_tracker.root.load(.monotonic);
             const root_hash = root.state.hash.readCopy();
             break :blk .{ root_slot, root_hash.? };
         };
@@ -586,7 +587,7 @@ pub const TowerConsensus = struct {
             }
             // Update cluster with the duplicate confirmation status.
             // Analogous to [ReplayStage::mark_slots_duplicate_confirmed](https://github.com/anza-xyz/agave/blob/47c0383f2301e5a739543c1af9992ae182b7e06c/core/src/replay_stage.rs#L3876)
-            const root_slot = slot_tracker.root;
+            const root_slot = slot_tracker.root.load(.monotonic);
             for (duplicate_confirmed_forks.items) |duplicate_confirmed_fork| {
                 const slot, const frozen_hash = duplicate_confirmed_fork.tuple();
                 try self.handleDuplicateConfirmedFork(
@@ -1606,7 +1607,7 @@ fn checkAndHandleNewRoot(
     // Audit: The rest of the code maps to Self::handle_new_root in Agave.
     // Update the slot tracker.
     // Set new root.
-    slot_tracker.root = new_root;
+    slot_tracker.root.store(new_root, .monotonic);
     // Prune non rooted slots
     slot_tracker.pruneNonRooted(allocator);
 
@@ -4771,12 +4772,10 @@ test "edge cases - duplicate slot" {
 
     const slot_tracker = &replay_state.slot_tracker;
     const progress_map = &replay_state.progress_map;
-    std.debug.assert(slot_tracker.root == 0);
+    const root_slot0 = slot_tracker.root.load(.monotonic);
+    std.debug.assert(root_slot0 == 0);
 
-    const root_slot0 = slot_tracker.root;
     const root_slot0_hash = slot_tracker.getRoot().state.hash.readCopy().?;
-
-    std.debug.assert(root_slot0 == 0); // assert initial root value
     std.debug.assert(root_slot0_hash.eql(.ZEROES)); // assert initial root hash
 
     // -- slot1 -- //
@@ -4930,12 +4929,11 @@ test "edge cases - duplicate confirmed slot" {
 
     const slot_tracker = &replay_state.slot_tracker;
     const progress_map = &replay_state.progress_map;
-    std.debug.assert(slot_tracker.root == 0);
+    const root_slot0 = slot_tracker.root.load(.monotonic);
+    std.debug.assert(root_slot0 == 0);
 
-    const root_slot0 = slot_tracker.root;
     const root_slot0_hash = slot_tracker.getRoot().state.hash.readCopy().?;
-
-    std.debug.assert(root_slot0 == 0); // assert initial root value
+    std.debug.assert(root_slot0_hash.eql(.ZEROES)); // assert initial root hash
     std.debug.assert(root_slot0_hash.eql(.ZEROES)); // assert initial root hash
 
     // -- slot1 -- //
@@ -5090,16 +5088,14 @@ test "edge cases - gossip verified vote hashes" {
 
     const slot_tracker = &replay_state.slot_tracker;
     const progress_map = &replay_state.progress_map;
-    std.debug.assert(slot_tracker.root == 0);
+    const root_slot0 = slot_tracker.root.load(.monotonic);
+    std.debug.assert(root_slot0 == 0); // assert initial root value
 
     var vote_collector: sig.consensus.vote_listener.VoteCollector =
-        try .init(.EPOCH_ZERO, slot_tracker.root, &registry);
+        try .init(.EPOCH_ZERO, root_slot0, &registry);
     defer vote_collector.deinit(gpa);
 
-    const root_slot0 = slot_tracker.root;
     const root_slot0_hash = slot_tracker.getRoot().state.hash.readCopy().?;
-
-    std.debug.assert(root_slot0 == 0); // assert initial root value
     std.debug.assert(root_slot0_hash.eql(.ZEROES)); // assert initial root hash
 
     // -- slot1 -- //
@@ -6019,7 +6015,7 @@ test "root advances after vote satisfies lockouts" {
             consensus.replay_tower.tower.votes.len,
         );
 
-        try std.testing.expectEqual(1, slot_tracker.root);
+        try std.testing.expectEqual(1, slot_tracker.root.load(.monotonic));
         // No longer tracking slot 0
         try std.testing.expect(!slot_tracker.contains(0));
         // Still tracking slot 1
@@ -6106,7 +6102,7 @@ test "root advances after vote satisfies lockouts" {
         const last_voted = consensus.replay_tower.tower.lastVotedSlot();
         try std.testing.expectEqual(33, last_voted);
 
-        try std.testing.expectEqual(2, slot_tracker.root);
+        try std.testing.expectEqual(2, slot_tracker.root.load(.monotonic));
         // No longer tracking slot 0
         try std.testing.expect(!slot_tracker.contains(0));
         // No longer tracking slot 1

src/replay/consensus/process_result.zig

@@ -117,7 +117,7 @@ fn markDeadSlot(
         params.allocator,
         .from(params.logger),
         dead_slot,
-        params.slot_tracker.root,
+        params.slot_tracker.root.load(.monotonic),
         params.duplicate_slots_to_repair,
         ancestor_hashes_replay_update_sender,
         dead_state,
@@ -148,7 +148,7 @@ fn markDeadSlot(
             params.allocator,
             .from(params.logger),
             dead_slot,
-            params.slot_tracker.root,
+            params.slot_tracker.root.load(.monotonic),
             params.duplicate_slots_tracker,
             params.fork_choice,
             duplicate_state,
@@ -194,7 +194,7 @@ fn updateConsensusForFrozenSlot(params: ProcessResultParams, slot: Slot) !void {
         params.allocator,
         .from(params.logger),
         slot,
-        params.slot_tracker.root,
+        params.slot_tracker.root.load(.monotonic),
         params.ledger.resultWriter(),
         params.fork_choice,
         params.duplicate_slots_to_repair,