From 7909189ee59f635c195a9fa54a9614d4a4d5408e Mon Sep 17 00:00:00 2001 From: Koko Bhadra Date: Tue, 9 Jun 2026 22:01:07 -0400 Subject: [PATCH] Add log_watcher: block-scoped log watching (closes #36) On each newHeads notification, fetch eth_getLogs scoped to the planned block range. Blocks missed across reconnects are back-filled (the gap between the last processed head and the new one), and reorgs detected via parent-hash mismatch re-fetch the replaced range. Re-announced old heights re-fetch only that height, since the new chain may be shorter than the previous head. The range planning lives in a pure planRange() function with unit tests for first-head lag, sequential heads, gaps, both reorg shapes, and the handle_reorgs=false paths. LogWatcher.pollOnce() is the pull-based API; watchLogs() is the callback loop from the issue. Also fixes a latent 0.15.2 incompatibility this feature exposed: subscription.parseBlockFromNotification used std.json.stringifyAlloc (removed in 0.15) to re-wrap the notification for parseBlockHeader. The header object is now parsed directly via the new provider.parseBlockHeaderObject, with no stringify round-trip. Verified: zig build test (0.15.2 and 0.17.0-dev), integration tests against Anvil including a new LogWatcher cursor-tracking test, docs build. --- CHANGELOG.md | 9 + README.md | 17 ++ docs/content/docs/websockets.mdx | 41 +++++ src/log_watcher.zig | 285 +++++++++++++++++++++++++++++++ src/provider.zig | 6 +- src/root.zig | 2 + src/subscription.zig | 10 +- tests/integration_tests.zig | 32 ++++ 8 files changed, 393 insertions(+), 9 deletions(-) create mode 100644 src/log_watcher.zig diff --git a/CHANGELOG.md b/CHANGELOG.md index b80db8a..6ecdaf5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ All notable changes to eth.zig will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added +- `log_watcher`: block-scoped log watching (#36). `LogWatcher.pollOnce()` drives per-block `eth_getLogs` from a `newHeads` subscription, back-fills blocks missed across reconnects, and re-fetches reorged ranges on parent-hash mismatch; `watchLogs` offers a callback loop +- `provider.parseBlockHeaderObject`: parse a `BlockHeader` from a bare JSON object + +### Fixed +- `subscription.parseBlockFromNotification` referenced `std.json.stringifyAlloc`, which does not exist in Zig 0.15.2; it now parses the notification object directly without a stringify round-trip + ## [0.4.0] - 2026-06-09 ### Added diff --git a/README.md b/README.md index fffd871..3b14d14 100644 --- a/README.md +++ b/README.md @@ -174,6 +174,23 @@ while (true) { Lower-level building blocks remain available: `ws_transport.WsTransport` for raw frames and `ws_transport.connectWithReconnect()` for a callback-style reconnect loop without subscription state. +### Block-scoped log watching (keepers and searchers) + +`log_watcher.LogWatcher` packages the canonical bot loop -- on each new head, fetch filtered logs for that block -- on top of `WsClient` + `eth_getLogs`. Blocks missed across reconnects are back-filled and reorged ranges are re-fetched automatically: + +```zig +var watcher = try eth.log_watcher.LogWatcher.init(allocator, &provider, client, .{ + .address = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", +}, .{}); +defer watcher.deinit(); + +while (true) { + const logs = try watcher.pollOnce(); // blocks until the next head + defer eth.log_watcher.freeLogs(allocator, logs); + for (logs) |log| { /* liquidate, arb, ... */ } +} +``` + ## Built with eth.zig Real production bots and SDKs built on eth.zig: diff --git a/docs/content/docs/websockets.mdx b/docs/content/docs/websockets.mdx index 6cec490..bcb8666 100644 --- a/docs/content/docs/websockets.mdx +++ b/docs/content/docs/websockets.mdx @@ -70,3 +70,44 @@ wss://host:port/path -- TLS encrypted (recommended) ``` Default ports: 80 for `ws://`, 443 for `wss://`. + +## Watching Logs Per Block (LogWatcher) + +The pattern "on each new block, fetch filtered logs and process them" is the core +of most on-chain bots (liquidation keepers, arbitrage, searchers). `log_watcher` +packages it: a `newHeads` subscription drives block-scoped `eth_getLogs` calls, +blocks missed across reconnects are back-filled, and reorgs (parent-hash mismatch +on sequential heads) re-fetch the replaced range. + +```zig +const eth = @import("eth"); + +var transport = eth.http_transport.HttpTransport.init(allocator, "https://rpc.example.com"); +defer transport.deinit(); +var provider = eth.provider.Provider.init(allocator, &transport); + +const client = try eth.ws_client.WsClient.connect(allocator, "wss://rpc.example.com/ws", .{}); +defer client.deinit(); + +var watcher = try eth.log_watcher.LogWatcher.init(allocator, &provider, client, .{ + .address = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2", // WETH + .topics = &.{ "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" }, // Transfer +}, .{ .start_block_lag = 0, .handle_reorgs = true }); +defer watcher.deinit(); + +while (true) { + const logs = try watcher.pollOnce(); // blocks until the next head + defer eth.log_watcher.freeLogs(allocator, logs); + for (logs) |log| { + // process log.address / log.topics / log.data + } +} +``` + +Notes: + +- On reorgs, logs for the replaced range are **re-delivered**; key your processing + on `(block_hash, log_index)` to deduplicate. +- `start_block_lag` lets the first poll start N blocks behind the first observed + head (useful for warm-up). +- A callback-style wrapper `eth.log_watcher.watchLogs(...)` is also available. diff --git a/src/log_watcher.zig b/src/log_watcher.zig new file mode 100644 index 0000000..3cffadf --- /dev/null +++ b/src/log_watcher.zig @@ -0,0 +1,285 @@ +//! Block-scoped log watching: "on each new block, fetch filtered logs". +//! +//! Combines a `WsClient` newHeads subscription with `Provider.getLogs`, +//! back-filling blocks missed across reconnects and re-fetching ranges +//! when a reorg is detected (parent-hash mismatch on sequential heads). +//! +//! Two APIs are provided: +//! - `LogWatcher`: pull-based; call `pollOnce()` to block until the next +//! head and receive the logs for the planned block range. +//! - `watchLogs`: callback-based convenience loop per issue #36. + +const std = @import("std"); +const block_mod = @import("block.zig"); +const receipt_mod = @import("receipt.zig"); +const json_rpc = @import("json_rpc.zig"); +const provider_mod = @import("provider.zig"); +const ws_client_mod = @import("ws_client.zig"); +const subscription_mod = @import("subscription.zig"); + +pub const WatchOpts = struct { + /// How many blocks behind the first observed head to start fetching + /// from. Default: 0 (only the current head). + start_block_lag: u64 = 0, + /// Re-fetch logs for reorged ranges when a parent-hash mismatch is + /// detected. Logs may be delivered more than once in that case; + /// consumers should key on (block_hash, log_index). Default: true. + handle_reorgs: bool = true, +}; + +/// The last block the watcher fully processed. +pub const Cursor = struct { + number: u64, + hash: [32]u8, +}; + +pub const RangePlan = struct { + from: u64, + to: u64, + /// True when this range re-fetches blocks that were already + /// delivered, because the chain reorged underneath us. + reorg: bool, +}; + +/// Decide which block range to fetch logs for, given the previously +/// processed head (`last`) and a newly received header. Returns null when +/// the header should be skipped entirely (duplicate head with reorg +/// handling disabled). +pub fn planRange(last: ?Cursor, header: *const block_mod.BlockHeader, opts: WatchOpts) ?RangePlan { + const cur = last orelse { + const lag = @min(opts.start_block_lag, header.number); + return .{ .from = header.number - lag, .to = header.number, .reorg = false }; + }; + + if (header.number <= cur.number) { + // The node re-announced an old height: a reorg replaced a block we + // already processed. Re-fetch only that height -- the new chain may + // be shorter than our previous head, and replacement heads for the + // remaining heights arrive as their own notifications (the cursor + // follows them once this header is processed). + if (!opts.handle_reorgs) return null; + return .{ .from = header.number, .to = header.number, .reorg = true }; + } + + if (header.number == cur.number + 1) { + if (opts.handle_reorgs and !std.mem.eql(u8, &header.parent_hash, &cur.hash)) { + // Sequential head whose parent is not the block we processed: + // our head was reorged out. Re-fetch it together with the new + // block. + return .{ .from = cur.number, .to = header.number, .reorg = true }; + } + return .{ .from = header.number, .to = header.number, .reorg = false }; + } + + // Gap (e.g. reconnect): back-fill everything we missed. Parent hashes + // cannot be verified across a gap, so no reorg detection here. + return .{ .from = cur.number + 1, .to = header.number, .reorg = false }; +} + +/// Free a slice of logs returned by `pollOnce`. +pub fn freeLogs(allocator: std.mem.Allocator, logs: []receipt_mod.Log) void { + for (logs) |log| { + allocator.free(log.data); + if (log.topics.len > 0) allocator.free(log.topics); + } + if (logs.len > 0) allocator.free(logs); +} + +pub const LogWatcher = struct { + allocator: std.mem.Allocator, + provider: *provider_mod.Provider, + client: *ws_client_mod.WsClient, + sub: *ws_client_mod.Subscription, + /// Only `address` and `topics` are used; block bounds are set per + /// fetched range. + filter: json_rpc.LogFilter, + opts: WatchOpts, + cursor: ?Cursor, + + /// Subscribe to newHeads on `client` and watch logs matching `filter`. + /// The watcher does not own `provider` or `client`; it does own the + /// subscription it creates. + pub fn init( + allocator: std.mem.Allocator, + provider: *provider_mod.Provider, + client: *ws_client_mod.WsClient, + filter: json_rpc.LogFilter, + opts: WatchOpts, + ) !LogWatcher { + const sub = try client.subscribe(.{ .new_heads = {} }); + return .{ + .allocator = allocator, + .provider = provider, + .client = client, + .sub = sub, + .filter = filter, + .opts = opts, + .cursor = null, + }; + } + + pub fn deinit(self: *LogWatcher) void { + self.client.unsubscribe(self.sub) catch {}; + } + + /// Block until the next newHeads notification for this watcher's + /// subscription, then fetch and return the logs for the planned block + /// range. Returns an empty slice when the range contains no matching + /// logs. Caller owns the result; free it with `freeLogs`. + /// + /// Notifications belonging to other subscriptions multiplexed on the + /// same client are dropped; do not share the client's event stream + /// with other consumers while a watcher is polling. + pub fn pollOnce(self: *LogWatcher) ![]receipt_mod.Log { + while (true) { + const event = try self.client.next(); + defer self.allocator.free(event.payload); + if (event.sub != self.sub) continue; + + const header = subscription_mod.parseBlockFromNotification(self.allocator, event.payload) catch { + continue; + }; + defer self.allocator.free(header.extra_data); + + const plan = planRange(self.cursor, &header, self.opts) orelse continue; + const logs = try self.fetchRange(plan.from, plan.to); + self.cursor = .{ .number = header.number, .hash = header.hash }; + return logs; + } + } + + fn fetchRange(self: *LogWatcher, from: u64, to: u64) ![]receipt_mod.Log { + const from_hex = try std.fmt.allocPrint(self.allocator, "0x{x}", .{from}); + defer self.allocator.free(from_hex); + const to_hex = try std.fmt.allocPrint(self.allocator, "0x{x}", .{to}); + defer self.allocator.free(to_hex); + + return self.provider.getLogs(.{ + .fromBlock = from_hex, + .toBlock = to_hex, + .address = self.filter.address, + .topics = self.filter.topics, + }); + } +}; + +/// Callback-based convenience wrapper (issue #36): subscribes to newHeads +/// and invokes `callback` for every matching log, forever. Only returns on +/// error (e.g. `error.Disconnected` once the client's retry budget is +/// exhausted). The log passed to the callback is freed after the callback +/// returns; copy anything that must outlive it. +pub fn watchLogs( + allocator: std.mem.Allocator, + provider: *provider_mod.Provider, + client: *ws_client_mod.WsClient, + filter: json_rpc.LogFilter, + opts: WatchOpts, + callback: *const fn (log: *const receipt_mod.Log) void, +) !void { + var watcher = try LogWatcher.init(allocator, provider, client, filter, opts); + defer watcher.deinit(); + + while (true) { + const logs = try watcher.pollOnce(); + defer freeLogs(allocator, logs); + for (logs) |*log| callback(log); + } +} + +// -- Tests -- + +fn testHeader(number: u64, hash_byte: u8, parent_byte: u8) block_mod.BlockHeader { + return .{ + .number = number, + .hash = @splat(hash_byte), + .parent_hash = @splat(parent_byte), + .nonce = null, + .sha3_uncles = @splat(0), + .miner = @splat(0), + .state_root = @splat(0), + .transactions_root = @splat(0), + .receipts_root = @splat(0), + .logs_bloom = @splat(0), + .difficulty = 0, + .gas_limit = 0, + .gas_used = 0, + .timestamp = 0, + .extra_data = &.{}, + .mix_hash = @splat(0), + .base_fee_per_gas = null, + .blob_gas_used = null, + .excess_blob_gas = null, + }; +} + +test "planRange - first head with no lag" { + const header = testHeader(100, 0xaa, 0xbb); + const plan = planRange(null, &header, .{}).?; + try std.testing.expectEqual(@as(u64, 100), plan.from); + try std.testing.expectEqual(@as(u64, 100), plan.to); + try std.testing.expect(!plan.reorg); +} + +test "planRange - first head applies start_block_lag" { + const header = testHeader(100, 0xaa, 0xbb); + const plan = planRange(null, &header, .{ .start_block_lag = 10 }).?; + try std.testing.expectEqual(@as(u64, 90), plan.from); + try std.testing.expectEqual(@as(u64, 100), plan.to); +} + +test "planRange - start_block_lag saturates at genesis" { + const header = testHeader(3, 0xaa, 0xbb); + const plan = planRange(null, &header, .{ .start_block_lag = 10 }).?; + try std.testing.expectEqual(@as(u64, 0), plan.from); +} + +test "planRange - sequential head fetches exactly one block" { + const header = testHeader(101, 0xcc, 0xaa); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + const plan = planRange(cursor, &header, .{}).?; + try std.testing.expectEqual(@as(u64, 101), plan.from); + try std.testing.expectEqual(@as(u64, 101), plan.to); + try std.testing.expect(!plan.reorg); +} + +test "planRange - gap back-fills missed blocks" { + const header = testHeader(105, 0xcc, 0xdd); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + const plan = planRange(cursor, &header, .{}).?; + try std.testing.expectEqual(@as(u64, 101), plan.from); + try std.testing.expectEqual(@as(u64, 105), plan.to); + try std.testing.expect(!plan.reorg); +} + +test "planRange - parent mismatch re-fetches reorged head" { + // New head 101 whose parent is NOT our processed block 100. + const header = testHeader(101, 0xcc, 0xee); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + const plan = planRange(cursor, &header, .{}).?; + try std.testing.expectEqual(@as(u64, 100), plan.from); + try std.testing.expectEqual(@as(u64, 101), plan.to); + try std.testing.expect(plan.reorg); +} + +test "planRange - parent mismatch ignored when handle_reorgs is off" { + const header = testHeader(101, 0xcc, 0xee); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + const plan = planRange(cursor, &header, .{ .handle_reorgs = false }).?; + try std.testing.expectEqual(@as(u64, 101), plan.from); + try std.testing.expect(!plan.reorg); +} + +test "planRange - re-announced old height re-fetches that block" { + const header = testHeader(99, 0xcc, 0xdd); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + const plan = planRange(cursor, &header, .{}).?; + try std.testing.expectEqual(@as(u64, 99), plan.from); + try std.testing.expectEqual(@as(u64, 99), plan.to); + try std.testing.expect(plan.reorg); +} + +test "planRange - old height skipped when handle_reorgs is off" { + const header = testHeader(99, 0xcc, 0xdd); + const cursor = Cursor{ .number = 100, .hash = @splat(0xaa) }; + try std.testing.expect(planRange(cursor, &header, .{ .handle_reorgs = false }) == null); +} diff --git a/src/provider.zig b/src/provider.zig index e93bd0f..b10f477 100644 --- a/src/provider.zig +++ b/src/provider.zig @@ -965,8 +965,12 @@ pub fn parseBlockHeader(allocator: std.mem.Allocator, raw: []const u8) !?block_m if (result_val == .null) return null; if (result_val != .object) return error.InvalidResponse; - const obj = result_val.object; + return try parseBlockHeaderObject(allocator, result_val.object); +} +/// Parse a BlockHeader from a bare JSON object, as found in the `result` +/// of eth_getBlockByNumber or a newHeads subscription notification. +pub fn parseBlockHeaderObject(allocator: std.mem.Allocator, obj: std.json.ObjectMap) !block_mod.BlockHeader { // Parse required fields const number = try parseHexU64(jsonGetString(obj, "number") orelse return error.InvalidResponse); const hash = try parseHash(jsonGetString(obj, "hash") orelse return error.InvalidResponse); diff --git a/src/root.zig b/src/root.zig index b6580c3..8ef28f6 100644 --- a/src/root.zig +++ b/src/root.zig @@ -55,6 +55,7 @@ pub const flashbots = @import("flashbots.zig"); pub const contract = @import("contract.zig"); pub const multicall = @import("multicall.zig"); pub const event = @import("event.zig"); +pub const log_watcher = @import("log_watcher.zig"); pub const erc20 = @import("erc20.zig"); pub const erc721 = @import("erc721.zig"); @@ -123,6 +124,7 @@ test { _ = @import("contract.zig"); _ = @import("multicall.zig"); _ = @import("event.zig"); + _ = @import("log_watcher.zig"); _ = @import("erc20.zig"); _ = @import("erc721.zig"); // Layer 9 diff --git a/src/subscription.zig b/src/subscription.zig index 87aaaf4..b7b846e 100644 --- a/src/subscription.zig +++ b/src/subscription.zig @@ -197,15 +197,9 @@ pub fn parseBlockFromNotification(allocator: std.mem.Allocator, raw: []const u8) defer parsed.deinit(); const result_val = getNotificationResult(parsed.value) orelse return error.InvalidNotification; + if (result_val != .object) return error.InvalidNotification; - // Serialize result back and wrap as {"result":...} for reuse with parseBlockHeader. - const result_json = try std.json.stringifyAlloc(allocator, result_val, .{}); - defer allocator.free(result_json); - - const wrapped = try std.fmt.allocPrint(allocator, "{{\"result\":{s}}}", .{result_json}); - defer allocator.free(wrapped); - - return (try provider_mod.parseBlockHeader(allocator, wrapped)) orelse error.NullResult; + return try provider_mod.parseBlockHeaderObject(allocator, result_val.object); } /// Parse a `logs` notification payload into a Log. diff --git a/tests/integration_tests.zig b/tests/integration_tests.zig index 3364965..93ae71d 100644 --- a/tests/integration_tests.zig +++ b/tests/integration_tests.zig @@ -562,3 +562,35 @@ test "WsClient subscribe pending full streams an RpcTransaction" { } if (!found) return error.NoPendingTxObserved; } + +// --------------------------------------------------------------------------- +// LogWatcher: block-scoped log watching (issue #36) +// --------------------------------------------------------------------------- + +test "LogWatcher pollOnce tracks mined blocks" { + if (!isAnvilAvailable()) return; + const allocator = std.testing.allocator; + + var transport = eth.http_transport.HttpTransport.init(allocator, ANVIL_URL); + defer transport.deinit(); + var provider = eth.provider.Provider.init(allocator, &transport); + + const client = try eth.ws_client.WsClient.connect(allocator, ANVIL_WS_URL, .{ .ping_interval_ms = 0 }); + defer client.deinit(); + + var watcher = try eth.log_watcher.LogWatcher.init(allocator, &provider, client, .{}, .{}); + defer watcher.deinit(); + + try anvilMineOne(allocator); + const logs = try watcher.pollOnce(); + defer eth.log_watcher.freeLogs(allocator, logs); + + // An empty mined block carries no logs, but the cursor must advance. + try std.testing.expectEqual(@as(usize, 0), logs.len); + const first = watcher.cursor.?; + + try anvilMineOne(allocator); + const logs2 = try watcher.pollOnce(); + defer eth.log_watcher.freeLogs(allocator, logs2); + try std.testing.expectEqual(first.number + 1, watcher.cursor.?.number); +}