diff --git a/v2/lib/accounts_db.zig b/v2/lib/accounts_db.zig index 5e337070d0..6c49e993e9 100644 --- a/v2/lib/accounts_db.zig +++ b/v2/lib/accounts_db.zig @@ -9,14 +9,18 @@ comptime { } } +const rooted = @import("accounts_db/rooted.zig"); + const Pubkey = lib.solana.Pubkey; const Hash = lib.solana.Hash; const Slot = lib.solana.Slot; pub const AccountPool = @import("accounts_db/pool.zig").AccountPool; -pub const Rooted = @import("accounts_db/rooted.zig").Rooted; +pub const Rooted = rooted.Rooted; pub const Table = @import("accounts_db/table.zig").Table; +pub const RootedTestState = rooted.RootedTestState; + pub const RootedConfig = extern struct { file_len: u32, file_path: [std.fs.max_path_bytes]u8, @@ -26,11 +30,24 @@ pub const RootedConfig = extern struct { }; pub const AccountLookups = extern struct { - in: lib.ipc.Ring(256, Request), - out: lib.ipc.Ring(256, Result), + pub const capacity = 256; + + in: lib.ipc.Ring(capacity, Request), + out: lib.ipc.Ring(capacity, Result), + + /// Request-supplied id to match responses to requests. + /// Opaque to accounts_db service, so it's usage is up to the callers. + pub const RequestUserData = u32; + + pub const Request = extern struct { + /// Opaque to accounts_db service, so it's usage is up to the callers. + req_user_data: RequestUserData, + pubkey: Pubkey, + }; - pub const Request = Pubkey; pub const Result = extern struct { + /// Matches the `req_user_data` field of the Request that this Result is responding to. + req_user_data: RequestUserData, pubkey: Pubkey, account_index: AccountPool.AccountRef, // .invalid if not found }; diff --git a/v2/lib/accounts_db/rooted.zig b/v2/lib/accounts_db/rooted.zig index ab5516786c..87879d084b 100644 --- a/v2/lib/accounts_db/rooted.zig +++ b/v2/lib/accounts_db/rooted.zig @@ -689,7 +689,7 @@ pub const Rooted = struct { pub fn queueRead( self: *Rooted, logger: tel.Logger("Rooted.queueRead"), - pubkey: *const Pubkey, + request: *const AccountLookups.Request, ) !bool { const lookup_idx = self.free_lookups; if (lookup_idx == invalid_lookup_index) { @@ -704,16 +704,27 @@ pub const Rooted = struct { self.free_lookups = lookup_idx; } - const entry = self.table.get(pubkey); + node.result.req_user_data = request.req_user_data; + node.result.pubkey = request.pubkey; + + const entry = self.table.get(&request.pubkey); if (entry.isEmpty()) { // not found. complete immediately. - node.result = .{ .pubkey = pubkey.*, .account_index = .invalid }; + node.result = .{ + .req_user_data = request.req_user_data, + .pubkey = request.pubkey, + .account_index = .invalid, + }; node.next = self.ready_lookups; self.ready_lookups = lookup_idx; return true; } const acc_idx = try self.account_pool.alloc(entry.len); - node.result = .{ .pubkey = pubkey.*, .account_index = acc_idx }; + node.result = .{ + .req_user_data = request.req_user_data, + .pubkey = request.pubkey, + .account_index = acc_idx, + }; // prepare the account (these must be set before self.account_pool.free() on the same index) const account = self.account_pool.getAccount(acc_idx); @@ -732,7 +743,10 @@ pub const Rooted = struct { } // Consume - pub fn pollRead(self: *Rooted, logger: tel.Logger("Rooted.pollReady")) !?LookupResult { + pub fn pollRead( + self: *Rooted, + logger: tel.Logger("Rooted.pollReady"), + ) !?LookupResult { // check if lookup in ready queue var lookup_idx = self.ready_lookups; if (lookup_idx >= self.lookup_nodes.len) { @@ -931,3 +945,206 @@ pub const Rooted = struct { } } }; + +pub const RootedTestState = struct { + tmp: std.testing.TmpDir, + rooted: *Rooted, + table_memory: []u8, + account_pool_memory: []align(@alignOf(AccountPool)) u8, + account_pool: *AccountPool, + runtime_metadata: RuntimeMetadata, + + const test_table_memory_len = 64 * 1024; + const test_account_pool_memory_len = 64 * 1024; + const test_file_name = "rooted-test.db"; + + pub const Account = struct { + pubkey: Pubkey, + owner: Pubkey, + lamports: u64, + rent_epoch: Epoch, + executable: bool, + data: []const u8, + }; + + pub fn init(logger: tel.Logger("Rooted.test")) !RootedTestState { + const gpa = std.testing.allocator; + + var tmp = std.testing.tmpDir(.{}); + errdefer tmp.cleanup(); + + const rooted = try gpa.create(Rooted); + errdefer gpa.destroy(rooted); + + const table_memory = try gpa.alloc(u8, test_table_memory_len); + errdefer gpa.free(table_memory); + @memset(table_memory, 0); + + const account_pool_memory = try gpa.alignedAlloc( + u8, + .of(AccountPool), + @sizeOf(AccountPool) + test_account_pool_memory_len, + ); + errdefer gpa.free(account_pool_memory); + + const account_pool: *AccountPool = @ptrCast(account_pool_memory.ptr); + account_pool.init(test_account_pool_memory_len); + + var runtime_metadata: RuntimeMetadata = undefined; + runtime_metadata.init(); + + var activity: lib.runner.Activity = .{}; + var service_view = activity.serviceView(); + try rooted.init( + .from(logger), + .{ .activity = &service_view }, + tmp.dir, + test_file_name, + table_memory, + account_pool, + &runtime_metadata, + ); + + return .{ + .tmp = tmp, + .rooted = rooted, + .table_memory = table_memory, + .account_pool_memory = account_pool_memory, + .account_pool = account_pool, + .runtime_metadata = runtime_metadata, + }; + } + + pub fn deinit(self: *RootedTestState) void { + const gpa = std.testing.allocator; + + self.rooted.deinit(); + gpa.destroy(self.rooted); + gpa.free(self.table_memory); + gpa.free(self.account_pool_memory); + self.tmp.cleanup(); + } + + pub fn putAccounts( + self: *RootedTestState, + logger: tel.Logger("Rooted.test"), + accounts: []const Account, + ) !void { + try self.rooted.beginTransaction(.from(logger), 1); + for (accounts) |account| { + var data_reader = std.Io.Reader.fixed(account.data); + try self.rooted.put(.from(logger), &data_reader, .{ + .slot = 1, + .pubkey = account.pubkey, + .owner = account.owner, + .lamports = account.lamports, + .rent_epoch = account.rent_epoch, + .executable = account.executable, + .data_len = account.data.len, + }); + } + try self.rooted.commitTransaction(.from(logger)); + } +}; + +test "rooted lookup preserves request id for missing account" { + const logger = tel.Logger("Rooted.test").noop; + var state = try RootedTestState.init(logger); + defer state.deinit(); + + const id: AccountLookups.RequestUserData = 1234; + const missing_pk = Pubkey.parse("SysvarC1ock11111111111111111111111111111111"); + + try std.testing.expect(try state.rooted.queueRead(.from(logger), &.{ + .req_user_data = id, + .pubkey = missing_pk, + })); + + const result = (try state.rooted.pollRead(.from(logger))).?; + try std.testing.expectEqual(id, result.req_user_data); + try std.testing.expectEqual(AccountPool.AccountRef.invalid, result.account_index); +} + +test "rooted preserves user data across multiple concurrent reads" { + const logger = tel.Logger("Rooted.test").noop; + var state = try RootedTestState.init(logger); + defer state.deinit(); + + const user_data = [_]AccountLookups.RequestUserData{ 10, 20, 30 }; + const accounts = [_]RootedTestState.Account{ + .{ + .pubkey = Pubkey.parse("F4GpAFr6vrxU3Y887F3XWkXRgybCVjZNk63m72f6pump"), + .owner = Pubkey.parse("11111111111111111111111111111111"), + .lamports = 42, + .rent_epoch = 0, + .executable = false, + .data = "first rooted account", + }, + .{ + .pubkey = Pubkey.parse("9oDndFiC7RW42vZcmSzacTKMWE9kgeqnzwXDGLSkpump"), + .owner = Pubkey.parse("ComputeBudget111111111111111111111111111111"), + .lamports = 84, + .rent_epoch = 1, + .executable = false, + .data = "second rooted account data", + }, + .{ + .pubkey = Pubkey.parse("USD1ttGY1N17NEEHLmELoaybftRBUSErhqYiQzvEmuB"), + .owner = Pubkey.parse("SysvarRent111111111111111111111111111111111"), + .lamports = 126, + .rent_epoch = 2, + .executable = true, + .data = "third rooted account payload", + }, + }; + + try state.putAccounts(logger, &accounts); + + // Queue all three before polling. + for (accounts, user_data) |account, req_user_data| { + try std.testing.expect(try state.rooted.queueRead(.from(logger), &.{ + .req_user_data = req_user_data, + .pubkey = account.pubkey, + })); + } + + var seen: [3]bool = @splat(false); + var released: usize = 0; + + var completed: usize = 0; + while (completed < user_data.len) { + const result = try state.rooted.pollRead(.from(logger)) orelse continue; + + const account_index = for (user_data, 0..) |req_user_data, i| { + if (result.req_user_data == req_user_data) { + try std.testing.expect(!seen[i]); + seen[i] = true; + break i; + } + } else { + return error.UnexpectedRequestUserData; + }; + const expected = accounts[account_index]; + + completed += 1; + + try std.testing.expect(result.account_index != .invalid); + const account = state.account_pool.getAccount(result.account_index); + defer if (account.unref()) state.account_pool.free(result.account_index); + released += 1; + + try std.testing.expect(result.pubkey.equals(&expected.pubkey)); + try std.testing.expect(account.pubkey.equals(&expected.pubkey)); + try std.testing.expect(account.owner.equals(&expected.owner)); + try std.testing.expectEqual(expected.lamports, account.lamports); + try std.testing.expectEqual(expected.rent_epoch, account.rent_epoch); + try std.testing.expectEqual(expected.executable, account.data.executable); + try std.testing.expectEqualStrings(expected.data, account.getData()); + } + + try std.testing.expectEqual(accounts.len, released); + try std.testing.expectEqual( + [_]bool{ true, true, true }, + seen, + ); +} diff --git a/v2/lib/replay.zig b/v2/lib/replay.zig index ece1feb509..23236b0462 100644 --- a/v2/lib/replay.zig +++ b/v2/lib/replay.zig @@ -5,6 +5,17 @@ const ipc = @import("ipc.zig"); const util = @import("util.zig"); const accounts_db = @import("accounts_db.zig"); +const unrooted = @import("replay/unrooted.zig"); +const account_fetcher = @import("replay/account_fetcher.zig"); +const account_resolver = @import("replay/account_resolver.zig"); +comptime { + if (@import("builtin").is_test) { + _ = @import("replay/account_fetcher.zig"); + _ = @import("replay/account_resolver.zig"); + _ = @import("replay/unrooted.zig"); + } +} + const VersionedTransaction = solana.transaction.VersionedTransaction; // This is a bit large currently because of the unrooted store @@ -14,6 +25,14 @@ pub const TransactionPool = collections.SharedPool(TransactionRecord, 10_000); pub const BlockPool = collections.SharedPool(Node, 1024); +pub const Unrooted = unrooted.Unrooted; +pub const UnrootedConfig = unrooted.Config; +pub const UnrootedType = unrooted.UnrootedType; + +pub const AccountFetcher = account_fetcher.AccountFetcher; +pub const AccountFetcherType = account_fetcher.AccountFetcherType; +pub const AccountResolver = account_resolver.AccountResolver; + /// Transaction bytes plus their validated wire layout. /// /// This struct itself is safe to share between processes. Consumers can construct transient diff --git a/v2/lib/replay/account_fetcher.zig b/v2/lib/replay/account_fetcher.zig new file mode 100644 index 0000000000..d75bf06bd3 --- /dev/null +++ b/v2/lib/replay/account_fetcher.zig @@ -0,0 +1,1007 @@ +//! Asynchronous account fetch deduplication for Replay. +//! +//! `AccountFetcher` resolves individual account reads against Replay's account view: +//! unrooted fork state first, then rooted AccountsDB storage. Requests for the same +//! `(block_ref, pubkey)` share one fetch entry, and each requester receives a separate +//! completion tagged with its opaque `user_data`. +//! +//! Higher-level transaction resolution is intentionally kept outside this module. +//! +//! The resolver decides which accounts a transaction needs, including lookup-table and +//! program-derived dependencies, while this module only fetches accounts and reports +//! whether each account was found. +const std = @import("std"); +const lib = @import("../lib.zig"); + +const AccountPool = lib.accounts_db.AccountPool; +const AccountLookups = lib.accounts_db.AccountLookups; +const BlockPool = lib.replay.BlockPool; +const BlockRef = lib.replay.BlockRef; + +const Unrooted = lib.replay.Unrooted; +const Pubkey = lib.solana.Pubkey; +const AccountRef = AccountPool.AccountRef; + +const RootedTestState = lib.accounts_db.RootedTestState; + +pub const UserData = u64; +const UserDataType = UserData; + +pub const AccountFetcher = AccountFetcherType(Unrooted); + +/// Deduplicates and drives asynchronous account reads for Replay. +/// +/// The deduplication is implemented by internally maintaining a map keyed by `(block_ref, pubkey)`. +/// When a request is submitted, if an entry already exists for the same key, the request is appended +/// to the existing entry's waiter list. +/// +/// The fetcher maintains two queues: one for entries that are queued for rooted lookups and another for +/// entries that are ready with results. The `pollCompletions` method processes these queues, +/// submitting requests to the `AccountLookups` service and draining completed results. +pub fn AccountFetcherType(comptime UnrootedStore: type) type { + return struct { + const Self = @This(); + + const entry_capacity = 512; + const waiter_capacity = 512; + + account_pool: *AccountPool, + account_lookups: *AccountLookups, + unrooted: *UnrootedStore, + block_pool: *BlockPool, + + // TODO: remove this, implement a simple map. + allocator: std.mem.Allocator, + /// In-flight or ready fetch entries keyed by `(block_ref, pubkey)`. + active_fetches: FetchMap, + + // NOTE: rational for using separate lists for entries and waiters instead of a single list of + // union enums is for simplicity mostly, but also since a single FetchEntry can have multiple waiters, + // we're wasting less space. Though perhaps there's good reason to change this in the future? + + // TODO: flatten both of these by having a pool of nodes. + + /// Backing storage for fetch entries. + entries: [entry_capacity]FetchEntry, + /// Backing storage for per-request completion waiters. + waiters: [waiter_capacity]Waiter, + + /// Pool of waiter slots stored in `waiters`. + waiter_pool: WaiterPool, + /// Pool of fetch-entry slots stored in `entries`. + entry_pool: EntryPool, + + /// Head of entries waiting to be submitted to rooted AccountsDB. + rooted_head: EntryId.Optional, + /// Tail of entries waiting to be submitted to rooted AccountsDB. + rooted_tail: EntryId.Optional, + + /// Head of entries with results ready to deliver to waiters. + ready_head: EntryId.Optional, + /// Tail of entries with results ready to deliver to waiters. + ready_tail: EntryId.Optional, + + const EntryPool = lib.collections.Pool(FetchEntry, u16); + const EntryId = EntryPool.ItemId; + + const WaiterPool = lib.collections.Pool(Waiter, u16); + const WaiterId = WaiterPool.ItemId; + + // TODO: custom map. + // Maps the fetch request (block_ref, pubkey) to the ID of the fetch entry in the `entries` array. + const FetchMap = std.HashMapUnmanaged( + FetchKey, + EntryId, + std.hash_map.AutoContext(FetchKey), + 80, + ); + + const FetchKey = extern struct { + block_ref: BlockRef, + pubkey: Pubkey, + }; + + const Waiter = extern struct { + user_data: UserDataType, + next: WaiterId.Optional, + }; + + pub const UserData = UserDataType; + + pub const Request = struct { + block_ref: BlockRef, + pubkey: Pubkey, + + // TODO: do we need this? I don;t think the resolver reallt cares about this since pubkey should be enough? + /// Opaque to AccountFetcher. + user_data: UserDataType, + }; + + pub const Completion = struct { + user_data: UserDataType, + pubkey: Pubkey, + /// `.invalid` means the account was not found. + account_ref: AccountRef, + }; + + const FetchEntry = extern struct { + state: State, + + key: FetchKey, + + waiter_head: WaiterId.Optional, + waiter_tail: WaiterId.Optional, + + /// links `FetchEntry`s together in the rooted and ready queues. + /// It's either the next entry waiting to be sent to rooted, or + /// the entry that has completed and whose result is ready for waiters. + queue_next: EntryId.Optional, + + result: AccountRef = .invalid, + + const State = enum(u8) { + free, + queued_rooted, + fetching_rooted, + ready, + }; + }; + + pub fn init( + self: *Self, + allocator: std.mem.Allocator, + account_pool: *AccountPool, + account_lookups: *AccountLookups, + unrooted: *UnrootedStore, + block_pool: *BlockPool, + ) void { + var active_fetches: FetchMap = .empty; + active_fetches.ensureTotalCapacity( + allocator, + @intCast(entry_capacity), + ) catch @panic("failed to allocate active_fetches map"); + + self.* = .{ + .allocator = allocator, + + .account_pool = account_pool, + .account_lookups = account_lookups, + .unrooted = unrooted, + .block_pool = block_pool, + + .active_fetches = active_fetches, + + .entries = undefined, + .waiters = undefined, + + .entry_pool = undefined, + .waiter_pool = undefined, + + .rooted_head = .null, + .rooted_tail = .null, + + .ready_head = .null, + .ready_tail = .null, + }; + + self.entry_pool = .init(self.entries[0..]); + self.waiter_pool = .init(self.waiters[0..]); + } + + pub fn deinit(self: *Self) void { + std.debug.assert(self.active_fetches.count() == 0); + + std.debug.assert(self.rooted_head == .null); + std.debug.assert(self.rooted_tail == .null); + + std.debug.assert(self.ready_head == .null); + std.debug.assert(self.ready_tail == .null); + + // Every waiter and entry should have been returned to its pool. + std.debug.assert(self.entry_pool.free_list.opt() != null); + std.debug.assert(self.waiter_pool.free_list.opt() != null); + + self.active_fetches.deinit(self.allocator); + self.* = undefined; + } + + /// Submits one account fetch request and attaches it to any existing fetch + /// for the same `(block_ref, pubkey)`. + /// + /// New fetches check unrooted state immediately. Misses are queued for rooted + /// AccountsDB lookup and later driven by `pollCompletions`. + pub fn submit(self: *Self, request: Request) error{Full}!void { + // Create a new waiter for this request + const waiter_id = self.waiter_pool.createId() catch return error.Full; + errdefer self.waiter_pool.destroyId(waiter_id); + + const waiter = self.waiter_pool.indexToPtr(waiter_id); + waiter.* = .{ + .user_data = request.user_data, + .next = .null, + }; + + // Create a key for this request to check if an entry already exists + // (i.e a fetch is already in progress for this pubkey and block_ref) + const key: FetchKey = .{ + .block_ref = request.block_ref, + .pubkey = request.pubkey, + }; + + // Check if there's already a fetch tracked for this key. + if (self.active_fetches.get(key)) |entry_id| { + // If there is, append this request's waiter to the existing entry's waiter list. + // The entry will be completed when the fetch completes, and this request will + // receive its own completion. + self.appendWaiter(entry_id, waiter_id); + return; + } + + // If there isn't, create a new fetch entry for this key and start the fetch process. + const entry_id = self.entry_pool.createId() catch return error.Full; + errdefer self.entry_pool.destroyId(entry_id); + + const entry = self.entry_pool.indexToPtr(entry_id); + entry.* = .{ + .key = key, + // There's one one waiter (this request) for the new entry, so both the + // head and tail point to the same waiter. + .waiter_head = .init(waiter_id), + .waiter_tail = .init(waiter_id), + // This starts as null since its not linked into any queue yet. + .queue_next = .null, + // TODO: remove undefineds and add new state. + .state = undefined, + .result = .invalid, + }; + + // Track this new fetch entry in the active fetches map. + self.active_fetches.putAssumeCapacityNoClobber(key, entry_id); + errdefer std.debug.assert(self.active_fetches.remove(key)); + + // Check if the account is already available in the unrooted state. + // If it is, we can complete the fetch immediately without needing to query the rooted storage. + const unrooted_ref = self.unrooted.fetch( + &request.pubkey, + request.block_ref, + self.block_pool, + self.account_pool, + ); + + if (unrooted_ref != .invalid) { + // FetchEntry takes ownership of the reference returned by fetch(). + entry.result = unrooted_ref; + + // Unrooted had the account, mark ready and return. + entry.state = .ready; + self.enqueueReady(entry_id); + return; + } + + // If the account isn't available in the unrooted state, we need to query the rooted storage. + entry.state = .queued_rooted; + self.enqueueRooted(entry_id); + } + + /// Drives the fetcher by submitting queued requests to the rooted AccountsDB and draining completed results. + /// + /// Returns a slice of completions. + pub fn pollCompletions( + self: *Self, + out: []Completion, + ) []Completion { + self.drainRootedResults(); + self.submitRootedRequests(); + + var len: usize = 0; + while (len < out.len) : (len += 1) { + out[len] = self.popReadyCompletion() orelse break; + } + + return out[0..len]; + } + + fn popReadyCompletion(self: *Self) ?Completion { + const entry_id = self.popReady() orelse return null; + const entry = entry_id.ptr(&self.entry_pool); + + std.debug.assert(entry.state == .ready); + + // Pop the first waiter from the entry's waiter list. + const waiter_id = entry.waiter_head.opt() orelse unreachable; + const waiter = waiter_id.ptr(&self.waiter_pool); + + // Move waiter head forward to the next waiter in this entry's list. + entry.waiter_head = waiter.next; + + // If there are no more waiters, set the tail to null as well. + if (entry.waiter_head == .null) + entry.waiter_tail = .null; + + const completion: Completion = .{ + .user_data = waiter.user_data, + .pubkey = entry.key.pubkey, + .account_ref = entry.result, + }; + + if (completion.account_ref != .invalid) { + // The caller receives its own reference. + self.account_pool + .getAccount(completion.account_ref) + .ref(); + } + + self.waiter_pool.destroyId(waiter_id); + + // If there are more waiters for this entry, re-enqueue it to the ready queue + // so the next waiter can receive its completion. + // TODO: do we want batched-completitions? + if (entry.waiter_head != .null) { + // Round-robin completion delivery between ready accounts. + self.enqueueReady(entry_id); + } else { + // No more waiters for this entry, retire it and free its resources. + self.retireEntry(entry_id); + } + + return completion; + } + + fn enqueueRooted(self: *Self, entry_id: EntryId) void { + const entry = entry_id.ptr(&self.entry_pool); + std.debug.assert(entry.queue_next == .null); + + if (self.rooted_tail.opt()) |tail_id| { + tail_id.ptr(&self.entry_pool).queue_next = .init(entry_id); + } else { + self.rooted_head = .init(entry_id); + } + + self.rooted_tail = .init(entry_id); + } + + /// Dequeue the next entry from the rooted queue. + fn popRooted(self: *Self) ?EntryId { + const entry_id = self.rooted_head.opt() orelse return null; + const entry = entry_id.ptr(&self.entry_pool); + + self.rooted_head = entry.queue_next; + if (self.rooted_head == .null) + self.rooted_tail = .null; + + entry.queue_next = .null; + return entry_id; + } + + /// Enqueue a ready entry to the ready queue, which is used to deliver completions to waiters. + fn enqueueReady(self: *Self, entry_id: EntryId) void { + const entry = self.entry_pool.indexToPtr(entry_id); + std.debug.assert(entry.queue_next == .null); + std.debug.assert(entry.state == .ready); + + // Add the entry to the end of the ready queue. + if (self.ready_tail.opt()) |tail_id| { + // update current tail's next pointer to the new entry. + tail_id.ptr(&self.entry_pool).queue_next = .init(entry_id); + } else { + // Empty queue, so set the head to the new entry. + self.ready_head = .init(entry_id); + } + + // Update the tail to the new entry. + self.ready_tail = .init(entry_id); + } + + /// Dequeue the next entry from the ready queue. + fn popReady(self: *Self) ?EntryId { + const entry_id = self.ready_head.opt() orelse return null; + const entry = entry_id.ptr(&self.entry_pool); + + // update the head to the next entry in the queue. + self.ready_head = entry.queue_next; + if (self.ready_head == .null) + self.ready_tail = .null; + + entry.queue_next = .null; + return entry_id; + } + + fn appendWaiter(self: *Self, entry_id: EntryId, waiter_id: WaiterId) void { + const entry = entry_id.ptr(&self.entry_pool); + const waiter = waiter_id.ptr(&self.waiter_pool); + + std.debug.assert(waiter.next == .null); + + if (entry.waiter_tail.opt()) |tail_id| { + tail_id.ptr(&self.waiter_pool).next = .init(waiter_id); + } else { + entry.waiter_head = .init(waiter_id); + } + + entry.waiter_tail = .init(waiter_id); + } + + fn retireEntry(self: *Self, entry_id: EntryId) void { + const entry = entry_id.ptr(&self.entry_pool); + + std.debug.assert(entry.state == .ready); + std.debug.assert(entry.waiter_head == .null); + std.debug.assert(entry.waiter_tail == .null); + + std.debug.assert(self.active_fetches.remove(entry.key)); + + self.releaseAccount(entry.result); + + self.entry_pool.destroyId(entry_id); + } + + /// Empty rooted queue of requests by submitting them to rooted. + fn submitRootedRequests(self: *Self) void { + var writer = self.account_lookups.in.get(.writer); + var submitted: usize = 0; + + while (self.rooted_head != .null) { + const request_out = writer.next() orelse break; + + // NOTE: safe to unwrap since checked in loop condition. + const entry_id = self.popRooted().?; + const entry = entry_id.ptr(&self.entry_pool); + + std.debug.assert(entry.state == .queued_rooted); + + request_out.* = .{ + .req_user_data = @intCast(entry_id.index()), + .pubkey = entry.key.pubkey, + }; + + entry.state = .fetching_rooted; + submitted += 1; + } + + writer.markUsed(); + } + + /// Drains results from the rooted DB. + /// + /// For each account drained, the corresponding fetch entry is + /// updated with the result and moved to the ready queue. + /// + /// Returns true if any results were processed, false if the queue was empty. + fn drainRootedResults(self: *Self) void { + var reader = self.account_lookups.out.get(.reader); + var consumed: usize = 0; + + while (reader.next()) |response| { + consumed += 1; + self.processRootedResult(response.*); + } + + reader.markUsed(); + } + + /// Processes a single result from the rooted DB, updating the corresponding fetch entry + /// and moving it to the ready queue. + fn processRootedResult( + self: *Self, + response: AccountLookups.Result, + ) void { + const entry_index = response.req_user_data; + + // Totally unexpected if the entry index is out of bounds, this should never happen. + // TODO: make this a panic? + if (entry_index >= self.entry_pool.len) { + self.releaseAccount(response.account_index); + return; + } + + const entry_id = EntryId.fromInt(@intCast(entry_index)); + const entry = entry_id.ptr(&self.entry_pool); + + // We don't expect to receive a result for an entry that isn't in the fetching state. + // TODO: make this a panic? + if (entry.state != .fetching_rooted) { + self.releaseAccount(response.account_index); + return; + } + + std.debug.assert(response.pubkey.equals(&entry.key.pubkey)); + + // Mark the entry as ready and store the result. + entry.result = response.account_index; + entry.state = .ready; + + // Move the entry into the ready queue. + self.enqueueReady(entry_id); + } + + /// Release an account reference back to the account pool, if it's valid. + fn releaseAccount(self: *Self, account_ref: AccountRef) void { + if (account_ref == .invalid) return; + const account = self.account_pool.getAccount(account_ref); + if (account.unref()) self.account_pool.free(account_ref); + } + }; +} + +// Smaller for unit tests. +const TestUnrooted = lib.replay.UnrootedType(.{ + .max_blocks = 4, + .max_mutations_per_block = 8, +}); + +const TestFetcher = AccountFetcherType(TestUnrooted); + +const FetcherTestState = struct { + account_lookups: AccountLookups, + + block_pool_memory: [BlockPool.size()]u8 align(@alignOf(BlockPool)), + block_pool: *BlockPool, + + unrooted: TestUnrooted, + fetcher: TestFetcher, + + fn init( + self: *FetcherTestState, + account_pool: *AccountPool, + ) void { + self.account_lookups.init(); + + self.block_pool = @ptrCast(&self.block_pool_memory); + self.block_pool.init(); + + self.unrooted.init(); + + self.fetcher.init( + std.testing.allocator, + account_pool, + &self.account_lookups, + &self.unrooted, + self.block_pool, + ); + } + + fn deinit(self: *FetcherTestState) void { + self.fetcher.deinit(); + } + + fn addBlock( + self: *FetcherTestState, + parent: ?BlockRef, + slot: u64, + ) !BlockRef { + const block_ref = try self.block_pool.createId(); + block_ref.ptr(self.block_pool).* = .{ + .parent = .init(parent), + .slot = .init(slot), + }; + return block_ref; + } + + fn respond( + self: *FetcherTestState, + request: AccountLookups.Request, + account_ref: AccountRef, + ) !void { + var writer = self.account_lookups.out.get(.writer); + const response = writer.next() orelse + return error.ResponseRingFull; + + response.* = .{ + .req_user_data = request.req_user_data, + .pubkey = request.pubkey, + .account_index = account_ref, + }; + writer.markUsed(); + } +}; + +test "rooted miss completes as not found" { + var account_pool: AccountPool = undefined; + account_pool.init(0); + + var account_lookups: AccountLookups = undefined; + account_lookups.init(); + + var block_pool_memory: [BlockPool.size()]u8 align(@alignOf(BlockPool)) = undefined; + const block_pool: *BlockPool = @ptrCast(&block_pool_memory); + block_pool.init(); + + const block_ref = try block_pool.createId(); + block_ref.ptr(block_pool).* = .{ + .slot = .init(1), + }; + + var unrooted: TestUnrooted = undefined; + unrooted.init(); + + var fetcher: TestFetcher = undefined; + fetcher.init( + std.testing.allocator, + &account_pool, + &account_lookups, + &unrooted, + block_pool, + ); + defer fetcher.deinit(); + + const pubkey: Pubkey = .parse("AUCuaE1ZfgKAReZedngX55iW1NaCjFcDQ1pRvP4caix8"); + + try fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = pubkey, + .user_data = 42, + }); + + // Publish the queued Rooted request. + var completions_buf: [1]TestFetcher.Completion = undefined; + const completions = fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(0, completions.len); + + var request_reader = account_lookups.in.get(.reader); + const rooted_request = (request_reader.next() orelse + return error.MissingRootedRequest).*; + request_reader.markUsed(); + + try std.testing.expect(rooted_request.pubkey.equals(&pubkey)); + + // Simulate AccountsDB returning not-found. + var response_writer = account_lookups.out.get(.writer); + const response = response_writer.next() orelse + return error.ResponseRingFull; + + response.* = .{ + .req_user_data = rooted_request.req_user_data, + .pubkey = rooted_request.pubkey, + .account_index = .invalid, + }; + response_writer.markUsed(); + + const rooted_completions = fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(1, rooted_completions.len); + + const completion = rooted_completions[0]; + + try std.testing.expectEqual(42, completion.user_data); + try std.testing.expect(completion.pubkey.equals(&pubkey)); + try std.testing.expectEqual(AccountRef.invalid, completion.account_ref); + try std.testing.expectEqual(0, fetcher.pollCompletions(&completions_buf).len); +} + +test "duplicate requests share rooted fetch and receive owned references" { + const logger = lib.telemetry.Logger("Rooted.test").noop; + + var rooted_state = try RootedTestState.init(logger); + defer rooted_state.deinit(); + + const expected: RootedTestState.Account = .{ + .pubkey = Pubkey.parse("F4GpAFr6vrxU3Y887F3XWkXRgybCVjZNk63m72f6pump"), + .owner = Pubkey.parse("11111111111111111111111111111111"), + .lamports = 42, + .rent_epoch = 3, + .executable = false, + .data = "rooted account data", + }; + + try rooted_state.putAccounts(logger, &.{expected}); + + var state: FetcherTestState = undefined; + state.init(rooted_state.account_pool); + defer state.deinit(); + + const block_ref = try state.addBlock(null, 2); + + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = expected.pubkey, + .user_data = 10, + }); + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = expected.pubkey, + .user_data = 20, + }); + + // Both submissions share one fetch entry and one queued Rooted request. + try std.testing.expectEqual(1, state.fetcher.active_fetches.count()); + + const entry_id = state.fetcher.rooted_head.opt() orelse + return error.MissingRootedFetchEntry; + try std.testing.expectEqual(entry_id, state.fetcher.rooted_tail.opt().?); + + const entry = entry_id.ptr(&state.fetcher.entry_pool); + try std.testing.expectEqual(.queued_rooted, entry.state); + + const first_waiter_id = entry.waiter_head.opt() orelse + return error.MissingFirstWaiter; + const second_waiter_id = first_waiter_id.ptr(&state.fetcher.waiter_pool).next.opt() orelse + return error.MissingSecondWaiter; + try std.testing.expectEqual(second_waiter_id, entry.waiter_tail.opt().?); + try std.testing.expectEqual(.null, second_waiter_id.ptr(&state.fetcher.waiter_pool).next); + + var completions_buf: [2]TestFetcher.Completion = undefined; + const published_completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(0, published_completions.len); + + var request_reader = state.account_lookups.in.get(.reader); + const request = request_reader.next() orelse + return error.MissingRootedRequest; + + try std.testing.expect(request.pubkey.equals(&expected.pubkey)); + try std.testing.expect(request_reader.next() == null); + + try std.testing.expect(try rooted_state.rooted.queueRead( + .from(logger), + request, + )); + request_reader.markUsed(); + + const rooted_result = while (true) { + break try rooted_state.rooted.pollRead(.from(logger)) orelse + continue; + }; + + var response_writer = state.account_lookups.out.get(.writer); + response_writer.next().?.* = rooted_result; + response_writer.markUsed(); + + const completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(2, completions.len); + + const first = completions[0]; + const second = completions[1]; + + try std.testing.expectEqual(10, first.user_data); + try std.testing.expectEqual(20, second.user_data); + try std.testing.expectEqual(first.account_ref, second.account_ref); + try std.testing.expect(first.account_ref != .invalid); + + const account = + rooted_state.account_pool.getAccount(first.account_ref); + + try std.testing.expect(account.pubkey.equals(&expected.pubkey)); + try std.testing.expectEqual(expected.lamports, account.lamports); + try std.testing.expectEqualStrings(expected.data, account.getData()); + + // FetchEntry has retired; both completion callers own one ref. + try std.testing.expectEqual( + 2, + account.ref_count.load(.monotonic), + ); + + try std.testing.expect(!account.unref()); + try std.testing.expect(account.unref()); + rooted_state.account_pool.free(first.account_ref); +} + +test "unrooted accounts bypass rooted and zero-lamport accounts return refs" { + const memory_len = 64 * 1024; + const memory = try std.testing.allocator.alignedAlloc( + u8, + .of(AccountPool), + @sizeOf(AccountPool) + memory_len, + ); + defer std.testing.allocator.free(memory); + + const account_pool: *AccountPool = @ptrCast(memory.ptr); + account_pool.init(memory_len); + + var state: FetcherTestState = undefined; + state.init(account_pool); + defer state.deinit(); + + const block_ref = try state.addBlock(null, 1); + + const found_pk = Pubkey.parse("9oDndFiC7RW42vZcmSzacTKMWE9kgeqnzwXDGLSkpump"); + const tombstone_pk = Pubkey.parse("USD1ttGY1N17NEEHLmELoaybftRBUSErhqYiQzvEmuB"); + + const found_ref = try account_pool.alloc(0); + account_pool.getAccount(found_ref).* = .{ + .ref_count = .init(1), + .pubkey = found_pk, + .owner = .ZEROES, + .lamports = 100, + .rent_epoch = 0, + .data = .{ + .executable = false, + .len = 0, + }, + }; + + const tombstone_ref = try account_pool.alloc(0); + account_pool.getAccount(tombstone_ref).* = .{ + .ref_count = .init(1), + .pubkey = tombstone_pk, + .owner = .ZEROES, + .lamports = 0, + .rent_epoch = 0, + .data = .{ + .executable = false, + .len = 0, + }, + }; + + try std.testing.expectEqual( + AccountRef.invalid, + state.unrooted.put(block_ref, account_pool, found_ref), + ); + try std.testing.expectEqual( + AccountRef.invalid, + state.unrooted.put(block_ref, account_pool, tombstone_ref), + ); + + // Drop the original owners; Unrooted now owns one reference each. + try std.testing.expect(!account_pool.getAccount(found_ref).unref()); + try std.testing.expect(!account_pool.getAccount(tombstone_ref).unref()); + + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = found_pk, + .user_data = 1, + }); + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = tombstone_pk, + .user_data = 2, + }); + + // Neither request should reach Rooted. + var rooted_reader = state.account_lookups.in.get(.reader); + try std.testing.expect(rooted_reader.next() == null); + + var completions_buf: [2]TestFetcher.Completion = undefined; + const completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(2, completions.len); + + const found = completions[0]; + const tombstone = completions[1]; + + try std.testing.expectEqual(found_ref, found.account_ref); + try std.testing.expectEqual(tombstone_ref, tombstone.account_ref); + + // Release completion ownership. + try std.testing.expect( + !account_pool.getAccount(found.account_ref).unref(), + ); + try std.testing.expect( + !account_pool.getAccount(tombstone.account_ref).unref(), + ); + + // Release the references owned by Unrooted before ending the test. + try std.testing.expect(account_pool.getAccount(found_ref).unref()); + account_pool.free(found_ref); + + try std.testing.expect(account_pool.getAccount(tombstone_ref).unref()); + account_pool.free(tombstone_ref); +} + +test "rooted responses complete by request id out of order" { + var account_pool: AccountPool = undefined; + account_pool.init(0); + + var state: FetcherTestState = undefined; + state.init(&account_pool); + defer state.deinit(); + + const block_ref = try state.addBlock(null, 1); + const first_pk = Pubkey.parse("SysvarC1ock11111111111111111111111111111111"); + const second_pk = Pubkey.parse("SysvarRent111111111111111111111111111111111"); + + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = first_pk, + .user_data = 11, + }); + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = second_pk, + .user_data = 22, + }); + + var completions_buf: [2]TestFetcher.Completion = undefined; + const published_completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(0, published_completions.len); + + var reader = state.account_lookups.in.get(.reader); + const first_request = reader.next().?.*; + const second_request = reader.next().?.*; + reader.markUsed(); + + // Return the second lookup first. + try state.respond(second_request, .invalid); + try state.respond(first_request, .invalid); + + const completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(2, completions.len); + + const first_completion = completions[0]; + const second_completion = completions[1]; + + try std.testing.expectEqual( + 22, + first_completion.user_data, + ); + try std.testing.expect( + first_completion.pubkey.equals(&second_pk), + ); + + try std.testing.expectEqual( + 11, + second_completion.user_data, + ); + try std.testing.expect( + second_completion.pubkey.equals(&first_pk), + ); +} + +test "rooted request remains queued while lookup ring is full" { + var account_pool: AccountPool = undefined; + account_pool.init(0); + + var state: FetcherTestState = undefined; + state.init(&account_pool); + defer state.deinit(); + + // Occupy the entire Rooted request ring. + var filler = state.account_lookups.in.get(.writer); + for (0..AccountLookups.capacity) |i| { + filler.next().?.* = .{ + .req_user_data = @intCast(i), + .pubkey = .ZEROES, + }; + } + filler.markUsed(); + + const block_ref = try state.addBlock(null, 1); + const pubkey = Pubkey.parse("SysvarC1ock11111111111111111111111111111111"); + + try state.fetcher.submit(.{ + .block_ref = block_ref, + .pubkey = pubkey, + .user_data = 77, + }); + + // The entry remains on rooted_head because no ring slot is available. + var completions_buf: [1]TestFetcher.Completion = undefined; + try std.testing.expectEqual( + 0, + state.fetcher.pollCompletions(&completions_buf).len, + ); + try std.testing.expect(state.fetcher.rooted_head != .null); + try std.testing.expectEqual( + 0, + state.fetcher.pollCompletions(&completions_buf).len, + ); + + // Drain the filler requests. + var reader = state.account_lookups.in.get(.reader); + for (0..AccountLookups.capacity) |_| { + _ = reader.next() orelse return error.MissingFillerRequest; + } + reader.markUsed(); + + // The next poll can now publish the real request. + try std.testing.expectEqual( + 0, + state.fetcher.pollCompletions(&completions_buf).len, + ); + + var actual_reader = state.account_lookups.in.get(.reader); + const request = actual_reader.next().?.*; + actual_reader.markUsed(); + + try std.testing.expect(request.pubkey.equals(&pubkey)); + + try state.respond(request, .invalid); + const completions = state.fetcher.pollCompletions(&completions_buf); + try std.testing.expectEqual(1, completions.len); + + const completion = completions[0]; + try std.testing.expectEqual( + 77, + completion.user_data, + ); + try std.testing.expectEqual( + AccountRef.invalid, + completion.account_ref, + ); +} diff --git a/v2/lib/replay/account_resolver.zig b/v2/lib/replay/account_resolver.zig new file mode 100644 index 0000000000..f7b1dc72ec --- /dev/null +++ b/v2/lib/replay/account_resolver.zig @@ -0,0 +1,276 @@ +const std = @import("std"); +const lib = @import("../lib.zig"); + +const Unrooted = lib.replay.Unrooted; +const AccountPool = lib.accounts_db.AccountPool; +const AccountLookups = lib.accounts_db.AccountLookups; +const BlockPool = lib.replay.BlockPool; +const BlockRef = lib.replay.BlockRef; +const TransactionPool = lib.replay.TransactionPool; +const VersionedTransaction = lib.solana.transaction.VersionedTransaction; +const AccountRef = AccountPool.AccountRef; + +const TransactionRef = TransactionPool.ItemId; + +const account_fetcher = @import("account_fetcher.zig"); + +const AccountFetcher = account_fetcher.AccountFetcher; + +pub const AccountResolver = struct { + const Self = @This(); + + // Arbitrarily chosen for now. + const MAX_PENDING_TRANSACTIONS = 256; + + /// 2x the max loaded accounts per transaction, to allow for some extra fetches for alt lookups. + const MAX_FETCH_WORK = 256; + + fetcher: AccountFetcher, + transaction_pool: *TransactionPool, + + pending_pool: PendingPool, + + /// Used to track the set of transactions with unsubmitted work waiting to be sent to the AccountFetcher. + active: [MAX_PENDING_TRANSACTIONS]ResolutionId, + active_len: usize, + + const PendingPool = lib.collections.SharedPool( + PendingTransaction, + MAX_PENDING_TRANSACTIONS, + ); + + pub const ResolutionId = PendingPool.ItemId; + + const PendingTransaction = struct { + block_ref: BlockRef, + tx_ref: TransactionRef, + + /// The number of valid entries in `work`. + work_len: u16, + + /// First work item not accepted by AccountFetcher (due to backpressure). + next_submit: u16, + + /// Number of completed fetch requests (completions that have + /// been processed by the AccountResolver) + completed: u16, + + work: [MAX_FETCH_WORK]FetchWork, + + account_refs: [VersionedTransaction.MAX_ACCOUNT_KEYS]AccountRef, + + //Lowest descriptor index failure observed. + // lookup_failure: ?LookupFailure, + }; + + const FetchWork = union(enum) { + /// Index into transaction's static account keys array. + static: u8, + /// Byte offset of LUT's pubkey in transaction payload. + lut: u16, + // TODO: program, program_data + }; + + const FetchTicket = packed struct(u64) { + resolution_index: u16, + work_index: u16, + _reserved: u32 = 0, + }; + + pub fn init( + self: *Self, + allocator: std.mem.Allocator, + account_pool: *AccountPool, + account_lookups: *AccountLookups, + unrooted: *Unrooted, + block_pool: *BlockPool, + transaction_pool: *TransactionPool, + ) void { + self.* = .{ + .fetcher = undefined, + .transaction_pool = transaction_pool, + .pending_pool = undefined, + .active = undefined, + .active_len = 0, + }; + + self.fetcher.init( + allocator, + account_pool, + account_lookups, + unrooted, + block_pool, + ); + + self.pending_pool.init(); + } + + pub fn resolve( + self: *Self, + block_ref: BlockRef, + tx_id: TransactionPool.ItemId, + ) error{Full}!ResolutionId { + const resolution_id = self.pending_pool.createId() catch return error.Full; + errdefer self.pending_pool.destroyId(resolution_id); + + const transaction = self.transaction_pool.indexToConstPtr(tx_id).view(); + + const static_keys = transaction.staticAccountKeys(); + const lookup_count = transaction.layout.address_table_lookup_count; + + const work_len = static_keys.len + lookup_count; + std.debug.assert(work_len <= MAX_FETCH_WORK); + + const pending = resolution_id.ptr(&self.pending_pool); + pending.* = .{ + .block_ref = block_ref, + .tx_ref = tx_id, + .work_len = @intCast(work_len), + .next_submit = 0, + .completed = 0, + .work = undefined, + .account_refs = @splat(.invalid), + }; + + for (0..static_keys.len) |i| { + pending.work[i] = .{ + .static = @intCast(i), + }; + } + + var lookups = transaction.addressTableLookups(); + var lookup_index: usize = 0; + while (lookups.next() catch unreachable) |lookup| { + pending.work[static_keys.len + lookup_index] = .{ + .lut = lookup.pubkey_byte_offset, + }; + lookup_index += 1; + } + std.debug.assert(lookup_index == lookup_count); + + self.active[self.active_len] = resolution_id; + self.active_len += 1; + + _ = self.submitWork(); + + return resolution_id; + } + + fn submitWork(self: *Self) bool { + var made_progress = false; + + for (self.active[0..self.active_len]) |resolution_id| { + const pending = resolution_id.ptr(&self.pending_pool); + const transaction = self.transaction_pool.indexToConstPtr(pending.tx_ref).view(); + + // Submit as much work as possible to the AccountFetcher, until it returns Full. + while (pending.next_submit < pending.work_len) { + const work_index = pending.next_submit; + const work = pending.work[work_index]; + + const pubkey = switch (work) { + .static => |i| transaction.staticAccountKeys()[i], + .lut => |offset| transaction.pubkeyAtByteOffset(offset).*, + }; + + const ticket = FetchTicket{ + .resolution_index = @intCast(resolution_id.index()), + .work_index = work_index, + }; + + self.fetcher.submit(.{ + .block_ref = pending.block_ref, + .pubkey = pubkey, + .user_data = @bitCast(ticket), + }) catch |err| switch (err) { + error.Full => break, + }; + + pending.next_submit += 1; + made_progress = true; + } + } + + return made_progress; + } + + fn processFetchCompletion( + self: *Self, + completion: AccountFetcher.Completion, + ) void { + const ticket: FetchTicket = @bitCast(completion.user_data); + const resolution_id = ResolutionId.fromInt(@intCast(ticket.resolution_index)); + + const pending = resolution_id.ptr(&self.pending_pool); + + std.debug.assert(ticket.work_index < pending.next_submit); + std.debug.assert(ticket.work_index < pending.work_len); + + const work = pending.work[ticket.work_index]; + + switch (work) { + .static => |account_index| { + std.debug.assert(pending.account_refs[account_index] == .invalid); + + // NOTE: `.invalid` is a valid result for a missing transaction account. + pending.account_refs[account_index] = completion.account_ref; + + // TODO: verify completion.pubkey against the expected static transaction key. + }, + .lut => |pubkey_byte_offset| { + self.processLookupTableCompletion( + pending, + pubkey_byte_offset, + completion.account_ref, + ); + }, + } + + pending.completed += 1; + } + + fn processLookupTableCompletion( + self: *Self, + pending: *PendingTransaction, + pubkey_byte_offset: u16, + account_ref: AccountRef, + ) void { + _ = pending; + _ = pubkey_byte_offset; + + defer self.releaseAccount(account_ref); + + // TODO: reject a missing lookup-table account. + // TODO: validate the lookup-table owner. + // TODO: deserialize the lookup table. + // TODO: append loaded-account work. + } + + fn releaseAccount( + self: *Self, + account_ref: AccountRef, + ) void { + if (account_ref == .invalid) return; + const account = self.account_pool.getAccount(account_ref); + if (account.unref()) self.account_pool.free(account_ref); + } + + pub fn pollResolvedTransactions(self: *Self) void { + var completion_buf: [64]AccountFetcher.Completion = undefined; + + while (true) { + const completions = self.fetcher.pollCompletions(&completion_buf); + + for (completions) |completion| { + self.processFetchCompletion(completion); + } + + // Completions may have released fetcher capacity. + const submitted = self.submitWork(); + + // Newly submitted unrooted hits are immediately + // available on the next iteration. + if (completions.len == 0 and !submitted) break; + } + } +}; diff --git a/v2/lib/replay/unrooted.zig b/v2/lib/replay/unrooted.zig new file mode 100644 index 0000000000..53eabd0700 --- /dev/null +++ b/v2/lib/replay/unrooted.zig @@ -0,0 +1,179 @@ +const std = @import("std"); + +const lib = @import("../lib.zig"); +const tracy = @import("tracy"); + +const replay = lib.replay; + +const Pubkey = lib.solana.Pubkey; +const AccountRef = lib.accounts_db.AccountPool.AccountRef; + +// [firedancer] https://github.com/firedancer-io/firedancer/blob/c2050b9c7fb8787b1eaaf9e50cac421a7281f70f/src/flamenco/runtime/fd_cost_tracker.h#L78 +// TODO: calculate this constant ourselves / keep it up to date +pub const Unrooted = UnrootedType(.{ + .max_blocks = lib.replay.BlockPool.capacity, + .max_mutations_per_block = 367_535, +}); + +pub const Config = struct { + max_blocks: usize, + max_mutations_per_block: usize, +}; + +/// Holds the accounts mutated for each tracked Block. +pub fn UnrootedType(comptime config: Config) type { + return extern struct { + const Self = @This(); + + pub const max_blocks = config.max_blocks; + pub const max_mutations_per_block = config.max_mutations_per_block; + + seed: u64, + maps: [max_blocks]Map, // we could initialise with `= @splat(.{})`, but lld disagrees + + const Map = extern struct { + len: u32 = 0, // only used to assert `max_mutations_per_block` holds true + data: [max_mutations_per_block]AccountRef = @splat(.invalid), + + fn EntryPtr(comptime SelfPtr: type) type { + return switch (SelfPtr) { + *Map => *AccountRef, + *const Map => *const AccountRef, + else => unreachable, + }; + } + + fn entry( + self: anytype, + seed: u64, + account_pool: *lib.accounts_db.AccountPool, + pubkey: *const Pubkey, + ) EntryPtr(@TypeOf(self)) { + var i: usize = @intCast(pubkey.hash(seed) % max_mutations_per_block); + + while (true) : (i = (i + 1) % max_mutations_per_block) { + if (self.data[i] == .invalid) + return &self.data[i]; + if (pubkey.equals(&account_pool.getAccount(self.data[i]).pubkey)) + return &self.data[i]; + } + } + + fn get( + self: *const Map, + seed: u64, + account_pool: *lib.accounts_db.AccountPool, + pubkey: *const Pubkey, + ) AccountRef { + return self.entry(seed, account_pool, pubkey).*; + } + + // The map takes a ref to the new account. + // Returns the replaced entry, which the caller is expected to unref/free. + // Entries are replaced when an account of the inserted pubkey already exists in the map. + // lint: allow_unused + fn put( + self: *Map, + seed: u64, + account_pool: *lib.accounts_db.AccountPool, + new_account_ref: AccountRef, + ) AccountRef { + const zone = tracy.Zone.init(@src(), .{ .name = "Map.put" }); + defer zone.deinit(); + + std.debug.assert(new_account_ref != .invalid); + const new_account = account_pool.getAccount(new_account_ref); + const pubkey: *const Pubkey = &new_account.pubkey; + + const found_entry: *AccountRef = self.entry(seed, account_pool, pubkey); + + // don't "replace" an accountref with itself! + std.debug.assert(found_entry.* != new_account_ref); + + const old_account_ref = found_entry.*; + if (old_account_ref != .invalid) { + zone.text("replace"); + + std.debug.assert( + pubkey.equals(&account_pool.getAccount(old_account_ref).pubkey), + ); + } else { + zone.text("insert"); + + self.len += 1; + if (self.len > max_mutations_per_block) + @panic("max_mutations_per_block exceeded"); + } + + found_entry.* = new_account_ref; + new_account.ref(); + + return old_account_ref; + } + }; + + pub fn init(self: *Self) void { + // TODO: create randomly + secretly at startup, to avoid performance degradation from + // attackers using pre-made keys to cause bad clustering + self.seed = 123; + for (&self.maps) |*map| map.* = .{}; + } + + /// Insert an account into the unrooted store for one block. + /// Returns the replaced account ref, which the caller must unref/free. + pub fn put( + self: *Self, + block: lib.replay.BlockRef, + account_pool: *lib.accounts_db.AccountPool, + new_account_ref: AccountRef, + ) AccountRef { + const block_index = block.index(); + std.debug.assert(block_index < max_blocks); + + return self.maps[block_index].put( + self.seed, + account_pool, + new_account_ref, + ); + } + + /// Get an account purely from the unrooted store. + /// For internal/testing usage only. + /// NOTE: caller is responsible for freeing the account + pub fn fetch( + self: *Self, + key: *const lib.solana.Pubkey, + + // current block + pool for ancestor lookups + block: lib.replay.BlockRef, + block_pool: *lib.replay.BlockPool, + + // account storage + account_pool: *lib.accounts_db.AccountPool, + ) AccountRef { + const zone = tracy.Zone.init(@src(), .{ .name = "Unrooted.fetch" }); + defer zone.deinit(); + + var current: ?*replay.Node = block.ptr(block_pool); + while (current) |ancestor_block| { + const block_index = block_pool.ptrToIndex(ancestor_block).index(); + std.debug.assert(block_index < max_blocks); + + const current_map: *const Map = &self.maps[block_index]; + + const account_ref = current_map.get(self.seed, account_pool, key); + if (account_ref != .invalid) { + const account = account_pool.getAccount(account_ref); + account.ref(); + + zone.text("found"); + + return account_ref; + } + current = if (ancestor_block.parent.opt()) |p| p.ptr(block_pool) else null; + } + + return .invalid; + } + }; +} diff --git a/v2/lib/solana.zig b/v2/lib/solana.zig index 4dff682180..60d34741fd 100644 --- a/v2/lib/solana.zig +++ b/v2/lib/solana.zig @@ -1,5 +1,6 @@ comptime { if (@import("builtin").is_test) { + _ = @import("solana/account_lookup_table.zig"); _ = @import("solana/bincode.zig"); _ = @import("solana/cluster.zig"); _ = @import("solana/epoch_schedule.zig"); @@ -22,6 +23,7 @@ pub const features = @import("solana/features.zig"); pub const snapshot = @import("solana/snapshot.zig"); pub const transaction = @import("solana/transaction.zig"); pub const verify_ticks = @import("solana/verify_ticks.zig"); +pub const account_lookup_table = @import("solana/account_lookup_table.zig"); pub const Hash = @import("solana/hash.zig").Hash; pub const Pubkey = @import("solana/pubkey.zig").Pubkey; diff --git a/v2/lib/solana/account_lookup_table.zig b/v2/lib/solana/account_lookup_table.zig new file mode 100644 index 0000000000..2dadf4ea0f --- /dev/null +++ b/v2/lib/solana/account_lookup_table.zig @@ -0,0 +1,205 @@ +//! Ported only what was needed for basic ALT parsing from shared/runtime/program/address_lookup_table/lib.zig +//! +//! TODO: we should get this from the shared runtime lib once it's ready for use with v2. +//! TODO: re-implement skipped portions as needed. +const std = @import("std"); +const lib = @import("../lib.zig"); + +const Pubkey = lib.solana.Pubkey; +const Slot = lib.solana.Slot; + +pub const ID: Pubkey = .parse("AddressLookupTab1e1111111111111111111111111"); + +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L30 +/// The maximum number of addresses that a lookup table can hold +pub const LOOKUP_TABLE_MAX_ADDRESSES: usize = 256; + +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L33 +/// The serialized size of lookup table metadata +// note - this is actually the size of ProgramState? +pub const LOOKUP_TABLE_META_SIZE: usize = 56; + +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L125 +/// Program account states +pub const ProgramState = union(enum(u32)) { + /// Account is not initialized. + Uninitialized, + /// Initialized `LookupTable` account. + LookupTable: LookupTableMeta, +}; + +// [agave] https://github.com/anza-xyz/agave/blob/a00f1b5cdea9a7d5a70f8d24b86ea3ae66feff11/sdk/slot-hashes/src/lib.rs#L21 +pub const MAX_ENTRIES: usize = 512; // about 2.5 minutes to get your vote in + +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L46 +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L66 +/// Address lookup table metadata +pub const LookupTableMeta = struct { + /// Lookup tables cannot be closed until the deactivation slot is + /// no longer "recent" (not accessible in the `SlotHashes` sysvar). + deactivation_slot: Slot = std.math.maxInt(Slot), + /// The slot that the table was last extended. Address tables may + /// only be used to lookup addresses that were extended before + /// the current bank's slot. + last_extended_slot: Slot = 0, + /// The start index where the table was last extended from during + /// the `last_extended_slot`. + last_extended_slot_start_index: u8 = 0, + /// Authority address which must sign for each modification. + authority: ?Pubkey = null, + // Padding to keep addresses 8-byte aligned + _padding: u16 = 0, + // Raw list of addresses follows this serialized structure in + // the account's data, starting from `LOOKUP_TABLE_META_SIZE`. +}; + +// [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L133-L134 +pub const AddressLookupTable = struct { + meta: LookupTableMeta, + addresses: []const Pubkey, + + pub const MAX_SERIALIZED_SIZE = LOOKUP_TABLE_META_SIZE + LOOKUP_TABLE_MAX_ADDRESSES * 32; + + pub const DeserializeError = error{ + UninitializedAccount, + InvalidAccountData, + }; + + // [agave] https://github.com/anza-xyz/agave/blob/d300f3733f45d64a3b6b9fdb5a1157f378e181c2/sdk/program/src/address_lookup_table/state.rs#L224 + pub fn deserialize( + data: []const u8, + ) DeserializeError!AddressLookupTable { + if (data.len < LOOKUP_TABLE_META_SIZE) + return error.InvalidAccountData; + + // ProgramState contains no slices or other allocating fields. + var noalloc_memory: [0]u8 = .{}; + var noalloc = std.heap.FixedBufferAllocator.init(&noalloc_memory); + + var reader = std.Io.Reader.fixed(data); + const state = lib.solana.bincode.read( + &noalloc, + &reader, + ProgramState, + ) catch return error.InvalidAccountData; + + const meta = switch (state) { + .Uninitialized => return error.UninitializedAccount, + .LookupTable => |meta| meta, + }; + + const address_bytes = data[LOOKUP_TABLE_META_SIZE..]; + if (address_bytes.len % Pubkey.SIZE != 0) + return error.InvalidAccountData; + + return .{ + .meta = meta, + .addresses = std.mem.bytesAsSlice(Pubkey, address_bytes), + }; + } +}; + +fn writeProgramState(data: []u8, state: ProgramState) !void { + std.debug.assert(data.len >= LOOKUP_TABLE_META_SIZE); + + @memset(data[0..LOOKUP_TABLE_META_SIZE], 0); + var writer: std.Io.Writer = .fixed(data[0..LOOKUP_TABLE_META_SIZE]); + try lib.solana.bincode.write(&writer, state); +} + +fn writeLookupTableData( + data: []u8, + meta: LookupTableMeta, + addresses: []const Pubkey, +) !void { + std.debug.assert(data.len == LOOKUP_TABLE_META_SIZE + addresses.len * Pubkey.SIZE); + + try writeProgramState(data, .{ .LookupTable = meta }); + for (addresses, 0..) |address, i| { + const start = LOOKUP_TABLE_META_SIZE + i * Pubkey.SIZE; + @memcpy(data[start..][0..Pubkey.SIZE], &address.data); + } +} + +test "account lookup table deserializes initialized metadata and addresses" { + const authority = Pubkey.parse("SyndicAgdEphcy5xhAKZAomTYhcF8xhC7za2UD9xeug"); + const addresses = [_]Pubkey{ + Pubkey.parse("11111111111111111111111111111111"), + Pubkey.parse("ComputeBudget111111111111111111111111111111"), + Pubkey.parse("SysvarRent111111111111111111111111111111111"), + }; + const meta: LookupTableMeta = .{ + .deactivation_slot = 999, + .last_extended_slot = 123, + .last_extended_slot_start_index = 7, + .authority = authority, + ._padding = 0, + }; + + var data: [LOOKUP_TABLE_META_SIZE + addresses.len * Pubkey.SIZE]u8 = undefined; + try writeLookupTableData(&data, meta, &addresses); + + const table = try AddressLookupTable.deserialize(&data); + try std.testing.expectEqual(meta.deactivation_slot, table.meta.deactivation_slot); + try std.testing.expectEqual(meta.last_extended_slot, table.meta.last_extended_slot); + try std.testing.expectEqual( + meta.last_extended_slot_start_index, + table.meta.last_extended_slot_start_index, + ); + try std.testing.expect(table.meta.authority.?.equals(&authority)); + try std.testing.expectEqualSlices(Pubkey, &addresses, table.addresses); +} + +test "account lookup table deserializes zero and one address" { + const no_addresses = [_]Pubkey{}; + var empty_data: [LOOKUP_TABLE_META_SIZE]u8 = undefined; + try writeLookupTableData(&empty_data, .{}, &no_addresses); + const empty_table = try AddressLookupTable.deserialize(&empty_data); + try std.testing.expectEqual(@as(usize, 0), empty_table.addresses.len); + + const one_address = [_]Pubkey{ + Pubkey.parse("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"), + }; + var one_data: [LOOKUP_TABLE_META_SIZE + Pubkey.SIZE]u8 = undefined; + try writeLookupTableData(&one_data, .{}, &one_address); + const one_table = try AddressLookupTable.deserialize(&one_data); + try std.testing.expectEqualSlices(Pubkey, &one_address, one_table.addresses); +} + +test "account lookup table rejects uninitialized and invalid data" { + var uninitialized: [LOOKUP_TABLE_META_SIZE]u8 = undefined; + try writeProgramState(&uninitialized, .{ .Uninitialized = {} }); + try std.testing.expectError( + error.UninitializedAccount, + AddressLookupTable.deserialize(&uninitialized), + ); + + var too_short: [LOOKUP_TABLE_META_SIZE - 1]u8 = @splat(0); + try std.testing.expectError( + error.InvalidAccountData, + AddressLookupTable.deserialize(&too_short), + ); + + var bad_trailing: [LOOKUP_TABLE_META_SIZE + 1]u8 = undefined; + try writeProgramState(&bad_trailing, .{ .LookupTable = .{} }); + bad_trailing[LOOKUP_TABLE_META_SIZE] = 0; + try std.testing.expectError( + error.InvalidAccountData, + AddressLookupTable.deserialize(&bad_trailing), + ); +} + +test "account lookup table addresses borrow input buffer" { + const original = Pubkey.parse("SysvarC1ock11111111111111111111111111111111"); + const replacement = Pubkey.parse("SysvarRent111111111111111111111111111111111"); + const addresses = [_]Pubkey{original}; + + var data: [LOOKUP_TABLE_META_SIZE + Pubkey.SIZE]u8 = undefined; + try writeLookupTableData(&data, .{}, &addresses); + + const table = try AddressLookupTable.deserialize(&data); + try std.testing.expect(table.addresses[0].equals(&original)); + + @memcpy(data[LOOKUP_TABLE_META_SIZE..][0..Pubkey.SIZE], &replacement.data); + try std.testing.expect(table.addresses[0].equals(&replacement)); +} diff --git a/v2/lib/solana/transaction.zig b/v2/lib/solana/transaction.zig index 78201bee86..d2e861dc01 100644 --- a/v2/lib/solana/transaction.zig +++ b/v2/lib/solana/transaction.zig @@ -415,6 +415,18 @@ pub const VersionedTransaction = struct { return ptr[0..count]; } + pub fn pubkeyAtByteOffset( + self: View, + byte_offset: u16, + ) *const Pubkey { + const offset: usize = byte_offset; + + std.debug.assert(offset <= self.payload.len); + std.debug.assert(Pubkey.SIZE <= self.payload.len - offset); + + return @ptrCast(self.payload[offset..].ptr); + } + pub fn recentBlockhash(self: View) *const Hash { const offset: usize = self.layout.recent_blockhash_off; @@ -486,6 +498,7 @@ pub const VersionedTransaction = struct { remaining: u8, pub const Item = struct { + pubkey_byte_offset: u16, account_key: *const Pubkey, writable_indexes: []const u8, readonly_indexes: []const u8, @@ -509,9 +522,15 @@ pub const VersionedTransaction = struct { const readonly_indexes = try self.reader.takeBytes(readonly_count); + const pubkey_byte_offset = + account_key_bytes.ptr - self.reader.bytes.ptr; + + std.debug.assert(pubkey_byte_offset <= std.math.maxInt(u16)); + self.remaining -= 1; return .{ + .pubkey_byte_offset = @intCast(pubkey_byte_offset), .account_key = account_key, .writable_indexes = writable_indexes, .readonly_indexes = readonly_indexes, diff --git a/v2/services/accounts_db.zig b/v2/services/accounts_db.zig index 2e4640fe2a..a2482a2c9c 100644 --- a/v2/services/accounts_db.zig +++ b/v2/services/accounts_db.zig @@ -21,6 +21,8 @@ pub const std_options = start.options; pub const ReadOnly = services.accounts_db.ReadOnly; pub const ReadWrite = services.accounts_db.ReadWrite; +pub const max_burst_drain_size = 32; + pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !noreturn { const logger = rw.tel.acquireLogger(@tagName(name), "main"); rw.tel.signalReady(); @@ -98,7 +100,12 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n pub fn load(self: @This(), pubkey: *const lib.solana.Pubkey) ?AccountRef { errdefer |err| std.debug.panic("AccountReader: {}", .{err}); - if (!(try self.r.queueRead(.from(self.l), pubkey))) + const req = lib.accounts_db.AccountLookups.Request{ + .req_user_data = 0, // TODO(Preston): id management + .pubkey = pubkey.*, + }; + + if (!(try self.r.queueRead(.from(self.l), &req))) return error.RootedQueueFull; const result: Rooted.LookupResult = while (true) : (std.atomic.spinLoopHint()) break (try self.r.pollRead(.from(self.l))) orelse continue; @@ -140,18 +147,21 @@ pub fn serviceMain(runner: lib.runner.Connection, _: ReadOnly, rw: ReadWrite) !n var replay_in = rw.replay_lookups.in.get(.reader); var replay_out = rw.replay_lookups.out.get(.writer); while (true) : (std.atomic.spinLoopHint()) { - if (replay_in.peek()) |pubkey| { - if (try rooted.queueRead(.from(logger), pubkey)) { - _ = replay_in.next(); - replay_in.markUsed(); - } + var queue_count: usize = 0; + while (queue_count < max_burst_drain_size) : (queue_count += 1) { + const request = replay_in.peek() orelse break; + if (!(try rooted.queueRead(.from(logger), request))) break; + _ = replay_in.next(); } - if (replay_out.peek()) |result| { - if (try rooted.pollRead(.from(logger))) |res| { - result.* = res; - _ = replay_out.next(); - replay_out.markUsed(); - } + if (queue_count > 0) replay_in.markUsed(); + + var poll_count: usize = 0; + while (poll_count < max_burst_drain_size) : (poll_count += 1) { + const result = replay_out.peek() orelse break; + const res = (try rooted.pollRead(.from(logger))) orelse break; + result.* = res; + _ = replay_out.next(); } + if (poll_count > 0) replay_out.markUsed(); } } diff --git a/v2/services/replay.zig b/v2/services/replay.zig index 0d52633b41..6bb91364e1 100644 --- a/v2/services/replay.zig +++ b/v2/services/replay.zig @@ -94,6 +94,8 @@ const FecSetId = lib.shred.FecSetId; const Pool = lib.collections.Pool; +const Unrooted = lib.replay.Unrooted; + comptime { _ = start; } @@ -376,143 +378,6 @@ fn bootstrap( ); } -/// Holds the accounts mutated for each tracked Block. -const Unrooted = extern struct { - seed: u64, - maps: [max_blocks]Map, // we could initialise with `= @splat(.{})`, but lld disagrees - - // [firedancer] https://github.com/firedancer-io/firedancer/blob/c2050b9c7fb8787b1eaaf9e50cac421a7281f70f/src/flamenco/runtime/fd_cost_tracker.h#L78 - // TODO: calculate this constant ourselves / keep it up to date - const max_mutations_per_block = 367_535; - - const max_blocks = lib.replay.BlockPool.capacity; - - const Map = extern struct { - len: u32 = 0, // only used to assert `max_mutations_per_block` holds true - data: [N]AccountRef = @splat(.invalid), // ~1.4MiB - - // NOTE: might be a good idea to oversize this for performance reasons - const N = max_mutations_per_block; - - fn EntryPtr(comptime SelfPtr: type) type { - return switch (SelfPtr) { - *Map => *AccountRef, - *const Map => *const AccountRef, - else => unreachable, - }; - } - - fn entry( - self: anytype, - seed: u64, - account_pool: *lib.accounts_db.AccountPool, - pubkey: *const Pubkey, - ) EntryPtr(@TypeOf(self)) { - var i: usize = @intCast(pubkey.hash(seed) % N); - - while (true) : (i = (i + 1) % N) { - if (self.data[i] == .invalid) - return &self.data[i]; - if (pubkey.equals(&account_pool.getAccount(self.data[i]).pubkey)) - return &self.data[i]; - } - } - - fn get( - self: *const Map, - seed: u64, - account_pool: *lib.accounts_db.AccountPool, - pubkey: *const Pubkey, - ) AccountRef { - return self.entry(seed, account_pool, pubkey).*; - } - - // The map takes a ref to the new account. - // Returns the replaced entry, which the caller is expected to unref/free. - // Entries are replaced when an account of the inserted pubkey already exists in the map. - // lint: allow_unused - fn put( - self: *Map, - seed: u64, - account_pool: *lib.accounts_db.AccountPool, - new_account_ref: AccountRef, - ) AccountRef { - const zone = tracy.Zone.init(@src(), .{ .name = "Map.put" }); - defer zone.deinit(); - - std.debug.assert(new_account_ref != .invalid); - const new_account = account_pool.getAccount(new_account_ref); - const pubkey: *const Pubkey = &new_account.pubkey; - - const found_entry: *AccountRef = self.entry(seed, account_pool, pubkey); - - // don't "replace" an accountref with itself! - std.debug.assert(found_entry.* != new_account_ref); - - const old_account_ref = found_entry.*; - if (old_account_ref != .invalid) { - zone.text("replace"); - - std.debug.assert(pubkey.equals(&account_pool.getAccount(old_account_ref).pubkey)); - } else { - zone.text("insert"); - - self.len += 1; - if (self.len > max_mutations_per_block) @panic("max_mutations_per_block exceeded"); - } - - found_entry.* = new_account_ref; - new_account.ref(); - - return old_account_ref; - } - }; - - fn init(self: *Unrooted) void { - // TODO: create randomly + secretly at startup, to avoid performance degradation from - // attackers using pre-made keys to cause bad clustering - self.seed = 123; - for (&self.maps) |*map| map.* = .{}; - } - - /// Get an account purely from the unrooted store. - /// For internal/testing usage only. - /// NOTE: caller is responsible for freeing the account - fn fetch( - self: *Unrooted, - key: *const lib.solana.Pubkey, - - // current block + pool for ancestor lookups - block: lib.replay.BlockRef, - block_pool: *lib.replay.BlockPool, - - // account storage - account_pool: *lib.accounts_db.AccountPool, - ) AccountRef { - const zone = tracy.Zone.init(@src(), .{ .name = "Unrooted.fetch" }); - defer zone.deinit(); - - var current: ?*replay.Node = block.ptr(block_pool); - while (current) |ancestor_block| { - const current_map: *const Map = - &self.maps[block_pool.ptrToIndex(ancestor_block).index()]; - - const account_ref = current_map.get(self.seed, account_pool, key); - if (account_ref != .invalid) { - const account = account_pool.getAccount(account_ref); - account.ref(); - - zone.text("found"); - - return account_ref; - } - current = if (ancestor_block.parent.opt()) |p| p.ptr(block_pool) else null; - } - - return .invalid; - } -}; - // TODO: // 1) *never* block the replay thread (remove this function) // 2) introduce a basic transaction scheduler @@ -546,7 +411,8 @@ fn fetchBlocking( var response_queue = rooted_lookups.out.get(.reader); const request_buf = requester.next() orelse @panic("out of space"); - request_buf.* = key.*; + // TODO(Preston): id management + request_buf.* = .{ .req_user_data = 0, .pubkey = key.* }; requester.markUsed(); // blocking the thread - do not do this