lobo/zfin
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

move to thread safe allocator to address issues on parallel fetch

Browse source
This commit is contained in:
Emil Lerch 2026-04-24 06:20:23 -07:00
parent 8b010eb592
commit 742c237515
1 changed files with 105 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

169
src/service.zig
View file

@ -91,7 +91,37 @@ fn earningsPostProcess(ev: *EarningsEvent, _: std.mem.Allocator) anyerror!void {
} }
pub const DataService = struct { pub const DataService = struct {
allocator: std.mem.Allocator, /// Thread-safe wrapper over the caller-provided base allocator.
///
/// Why this exists: `parallelServerSync` spawns worker threads that
/// each allocate through `DataService` — HTTP client init, TLS cert
/// bundle parsing, request/response buffers, and `Store.writeRaw`
/// path joins. The CLI's root allocator is an `ArenaAllocator`
/// (`src/main.zig`), which is NOT thread-safe. Unsynchronized
/// concurrent allocs from workers corrupt the arena's free list.
/// Symptoms seen in the wild:
///
/// thread N panic: reached unreachable code
/// std/mem/Allocator.zig:147 grow
/// std/hash_map.zig:1296 addCertsFromFile
/// std/crypto/Certificate/Bundle.zig:206 request
/// std/http/Client.zig:1789 request
/// src/net/http.zig:43 syncFromServer
///
/// and bare segfaults mid-heap on whatever pointer the arena
/// scrambled that run.
///
/// The wrapper serializes every allocation with a mutex. Cost is
/// one lock acquire/release per alloc — negligible next to the I/O
/// these allocations feed (HTTP requests, cache writes). The
/// alternative (threading per-worker arenas through every
/// transitive callsite) was rejected as error-prone.
///
/// DO NOT add an "unwrap" method or store the child allocator
/// directly. The point is that internal callers don't need to
/// know whether they're running under threads — every path goes
/// through the lock by construction.
thread_safe: std.heap.ThreadSafeAllocator,
config: Config, config: Config,
// Lazily initialized providers (null until first use) // Lazily initialized providers (null until first use)
@ -103,15 +133,26 @@ pub const DataService = struct {
yh: ?Yahoo = null, yh: ?Yahoo = null,
tg: ?Tiingo = null, tg: ?Tiingo = null,
pub fn init(allocator: std.mem.Allocator, config: Config) DataService { pub fn init(base_allocator: std.mem.Allocator, config: Config) DataService {
const self = DataService{ const self = DataService{
.allocator = allocator, .thread_safe = .{ .child_allocator = base_allocator },
.config = config, .config = config,
}; };
self.logMissingKeys(); self.logMissingKeys();
return self; return self;
} }
/// Return the thread-safe allocator. Always go through this, never
/// access the child allocator directly — see the doc-comment on
/// `thread_safe` for why.
///
/// Safe to call from any method that holds `*DataService`. The
/// returned `std.mem.Allocator` embeds `&self.thread_safe`, which
/// is stable for as long as `self` is.
pub fn allocator(self: *DataService) std.mem.Allocator {
return self.thread_safe.allocator();
}
/// Log warnings for missing API keys so users know which features are unavailable. /// Log warnings for missing API keys so users know which features are unavailable.
fn logMissingKeys(self: DataService) void { fn logMissingKeys(self: DataService) void {
// Primary candle provider // Primary candle provider
@ -157,7 +198,7 @@ pub const DataService = struct {
if (@field(self, field_name)) |*p| return p; if (@field(self, field_name)) |*p| return p;
if (T == Cboe or T == Yahoo) { if (T == Cboe or T == Yahoo) {
// CBOE and Yahoo have no API key // CBOE and Yahoo have no API key
@field(self, field_name) = T.init(self.allocator); @field(self, field_name) = T.init(self.allocator());
} else { } else {
// All we're doing here is lower casing the type name, then // All we're doing here is lower casing the type name, then
// appending _key to it, so AlphaVantage -> alphavantage_key // appending _key to it, so AlphaVantage -> alphavantage_key
@ -174,7 +215,7 @@ pub const DataService = struct {
break :blk buf[0 .. short.len + 4]; break :blk buf[0 .. short.len + 4];
}; };
const key = @field(self.config, config_key) orelse return DataError.NoApiKey; const key = @field(self.config, config_key) orelse return DataError.NoApiKey;
@field(self, field_name) = T.init(self.allocator, key); @field(self, field_name) = T.init(self.allocator(), key);
} }
return &@field(self, field_name).?; return &@field(self, field_name).?;
} }
@ -189,7 +230,7 @@ pub const DataService = struct {
// ── Cache helper ───────────────────────────────────────────── // ── Cache helper ─────────────────────────────────────────────
fn store(self: *DataService) cache.Store { fn store(self: *DataService) cache.Store {
return cache.Store.init(self.allocator, self.config.cache_dir); return cache.Store.init(self.allocator(), self.config.cache_dir);
} }
/// Generic fetch-or-cache for simple data types (dividends, splits, options). /// Generic fetch-or-cache for simple data types (dividends, splits, options).
@ -243,15 +284,15 @@ pub const DataService = struct {
return switch (T) { return switch (T) {
Dividend => { Dividend => {
var pg = try self.getProvider(Polygon); var pg = try self.getProvider(Polygon);
return pg.fetchDividends(self.allocator, symbol, null, null); return pg.fetchDividends(self.allocator(), symbol, null, null);
}, },
Split => { Split => {
var pg = try self.getProvider(Polygon); var pg = try self.getProvider(Polygon);
return pg.fetchSplits(self.allocator, symbol); return pg.fetchSplits(self.allocator(), symbol);
}, },
OptionsChain => { OptionsChain => {
var cboe = try self.getProvider(Cboe); var cboe = try self.getProvider(Cboe);
return cboe.fetchOptionsChain(self.allocator, symbol); return cboe.fetchOptionsChain(self.allocator(), symbol);
}, },
else => @compileError("unsupported type for fetchFromProvider"), else => @compileError("unsupported type for fetchFromProvider"),
}; };
@ -288,7 +329,7 @@ pub const DataService = struct {
// If preferred is Yahoo (degraded symbol), try Yahoo first // If preferred is Yahoo (degraded symbol), try Yahoo first
if (preferred == .yahoo) { if (preferred == .yahoo) {
if (self.getProvider(Yahoo)) |yh| { if (self.getProvider(Yahoo)) |yh| {
if (yh.fetchCandles(self.allocator, symbol, from, to)) |candles| { if (yh.fetchCandles(self.allocator(), symbol, from, to)) |candles| {
log.debug("{s}: candles from Yahoo (preferred)", .{symbol}); log.debug("{s}: candles from Yahoo (preferred)", .{symbol});
return .{ .candles = candles, .provider = .yahoo }; return .{ .candles = candles, .provider = .yahoo };
} else |err| { } else |err| {
@ -299,7 +340,7 @@ pub const DataService = struct {
// Primary: Tiingo // Primary: Tiingo
if (self.getProvider(Tiingo)) |tg| { if (self.getProvider(Tiingo)) |tg| {
if (tg.fetchCandles(self.allocator, symbol, from, to)) |candles| { if (tg.fetchCandles(self.allocator(), symbol, from, to)) |candles| {
log.debug("{s}: candles from Tiingo", .{symbol}); log.debug("{s}: candles from Tiingo", .{symbol});
return .{ .candles = candles, .provider = .tiingo }; return .{ .candles = candles, .provider = .tiingo };
} else |err| { } else |err| {
@ -314,7 +355,7 @@ pub const DataService = struct {
// Rate limited: back off and retry — this is expected, not a failure // Rate limited: back off and retry — this is expected, not a failure
log.info("{s}: Tiingo rate limited, backing off", .{symbol}); log.info("{s}: Tiingo rate limited, backing off", .{symbol});
self.rateLimitBackoff(); self.rateLimitBackoff();
if (tg.fetchCandles(self.allocator, symbol, from, to)) |candles| { if (tg.fetchCandles(self.allocator(), symbol, from, to)) |candles| {
log.debug("{s}: candles from Tiingo (after rate limit backoff)", .{symbol}); log.debug("{s}: candles from Tiingo (after rate limit backoff)", .{symbol});
return .{ .candles = candles, .provider = .tiingo }; return .{ .candles = candles, .provider = .tiingo };
} else |retry_err| { } else |retry_err| {
@ -322,7 +363,7 @@ pub const DataService = struct {
if (retry_err == error.RateLimited) { if (retry_err == error.RateLimited) {
// Still rate limited after backoff — one more try // Still rate limited after backoff — one more try
self.rateLimitBackoff(); self.rateLimitBackoff();
if (tg.fetchCandles(self.allocator, symbol, from, to)) |candles| { if (tg.fetchCandles(self.allocator(), symbol, from, to)) |candles| {
log.debug("{s}: candles from Tiingo (after second backoff)", .{symbol}); log.debug("{s}: candles from Tiingo (after second backoff)", .{symbol});
return .{ .candles = candles, .provider = .tiingo }; return .{ .candles = candles, .provider = .tiingo };
} else |_| {} } else |_| {}
@ -347,7 +388,7 @@ pub const DataService = struct {
// Fallback: Yahoo (symbol not on Tiingo) // Fallback: Yahoo (symbol not on Tiingo)
if (preferred != .yahoo) { if (preferred != .yahoo) {
if (self.getProvider(Yahoo)) |yh| { if (self.getProvider(Yahoo)) |yh| {
if (yh.fetchCandles(self.allocator, symbol, from, to)) |candles| { if (yh.fetchCandles(self.allocator(), symbol, from, to)) |candles| {
log.info("{s}: candles from Yahoo (Tiingo fallback)", .{symbol}); log.info("{s}: candles from Yahoo (Tiingo fallback)", .{symbol});
return .{ .candles = candles, .provider = .yahoo }; return .{ .candles = candles, .provider = .yahoo };
} else |err| { } else |err| {
@ -437,7 +478,7 @@ pub const DataService = struct {
if (new_candles.len == 0) { if (new_candles.len == 0) {
// No new candles (weekend/holiday) — refresh TTL, reset fail_count // No new candles (weekend/holiday) — refresh TTL, reset fail_count
self.allocator.free(new_candles); self.allocator().free(new_candles);
s.updateCandleMeta(symbol, m.last_close, m.last_date, result.provider, 0); s.updateCandleMeta(symbol, m.last_close, m.last_date, result.provider, 0);
if (s.read(Candle, symbol, null, .any)) |r| if (s.read(Candle, symbol, null, .any)) |r|
return .{ .data = r.data, .source = .cached, .timestamp = std.time.timestamp() }; return .{ .data = r.data, .source = .cached, .timestamp = std.time.timestamp() };
@ -445,7 +486,7 @@ pub const DataService = struct {
// Append new candles to existing file + update meta, reset fail_count // Append new candles to existing file + update meta, reset fail_count
s.appendCandles(symbol, new_candles, result.provider, 0); s.appendCandles(symbol, new_candles, result.provider, 0);
if (s.read(Candle, symbol, null, .any)) |r| { if (s.read(Candle, symbol, null, .any)) |r| {
self.allocator.free(new_candles); self.allocator().free(new_candles);
return .{ .data = r.data, .source = .fetched, .timestamp = std.time.timestamp() }; return .{ .data = r.data, .source = .fetched, .timestamp = std.time.timestamp() };
} }
return .{ .data = new_candles, .source = .fetched, .timestamp = std.time.timestamp() }; return .{ .data = new_candles, .source = .fetched, .timestamp = std.time.timestamp() };
@ -528,7 +569,7 @@ pub const DataService = struct {
return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp }; return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp };
} }
// Stale: free cached events and re-fetch below // Stale: free cached events and re-fetch below
self.allocator.free(cached.data); self.allocator().free(cached.data);
} }
// Try server sync before hitting Finnhub // Try server sync before hitting Finnhub
@ -545,10 +586,10 @@ pub const DataService = struct {
const from = today.subtractYears(5); const from = today.subtractYears(5);
const to = today.addDays(365); const to = today.addDays(365);
const fetched = fh.fetchEarnings(self.allocator, symbol, from, to) catch |err| blk: { const fetched = fh.fetchEarnings(self.allocator(), symbol, from, to) catch |err| blk: {
if (err == error.RateLimited) { if (err == error.RateLimited) {
self.rateLimitBackoff(); self.rateLimitBackoff();
break :blk fh.fetchEarnings(self.allocator, symbol, from, to) catch { break :blk fh.fetchEarnings(self.allocator(), symbol, from, to) catch {
return DataError.FetchFailed; return DataError.FetchFailed;
}; };
} }
@ -570,10 +611,10 @@ pub const DataService = struct {
return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp }; return .{ .data = cached.data, .source = .cached, .timestamp = cached.timestamp };
var av = try self.getProvider(AlphaVantage); var av = try self.getProvider(AlphaVantage);
const fetched = av.fetchEtfProfile(self.allocator, symbol) catch |err| blk: { const fetched = av.fetchEtfProfile(self.allocator(), symbol) catch |err| blk: {
if (err == error.RateLimited) { if (err == error.RateLimited) {
self.rateLimitBackoff(); self.rateLimitBackoff();
break :blk av.fetchEtfProfile(self.allocator, symbol) catch { break :blk av.fetchEtfProfile(self.allocator(), symbol) catch {
return DataError.FetchFailed; return DataError.FetchFailed;
}; };
} }
@ -592,7 +633,7 @@ pub const DataService = struct {
pub fn getQuote(self: *DataService, symbol: []const u8) DataError!Quote { pub fn getQuote(self: *DataService, symbol: []const u8) DataError!Quote {
// Primary: Yahoo Finance (free, real-time) // Primary: Yahoo Finance (free, real-time)
if (self.getProvider(Yahoo)) |yh| { if (self.getProvider(Yahoo)) |yh| {
if (yh.fetchQuote(self.allocator, symbol)) |quote| { if (yh.fetchQuote(self.allocator(), symbol)) |quote| {
log.debug("{s}: quote from Yahoo", .{symbol}); log.debug("{s}: quote from Yahoo", .{symbol});
return quote; return quote;
} else |_| {} } else |_| {}
@ -601,7 +642,7 @@ pub const DataService = struct {
// Fallback: TwelveData (requires API key, may be 15-min delayed) // Fallback: TwelveData (requires API key, may be 15-min delayed)
var td = try self.getProvider(TwelveData); var td = try self.getProvider(TwelveData);
log.debug("{s}: quote fallback to TwelveData", .{symbol}); log.debug("{s}: quote fallback to TwelveData", .{symbol});
return td.fetchQuote(self.allocator, symbol) catch return td.fetchQuote(self.allocator(), symbol) catch
return DataError.FetchFailed; return DataError.FetchFailed;
} }
@ -609,7 +650,7 @@ pub const DataService = struct {
/// No cache -- always fetches fresh. Caller must free the returned string fields. /// No cache -- always fetches fresh. Caller must free the returned string fields.
pub fn getCompanyOverview(self: *DataService, symbol: []const u8) DataError!CompanyOverview { pub fn getCompanyOverview(self: *DataService, symbol: []const u8) DataError!CompanyOverview {
var av = try self.getProvider(AlphaVantage); var av = try self.getProvider(AlphaVantage);
return av.fetchCompanyOverview(self.allocator, symbol) catch return av.fetchCompanyOverview(self.allocator(), symbol) catch
return DataError.FetchFailed; return DataError.FetchFailed;
} }
@ -818,7 +859,7 @@ pub const DataService = struct {
// 2. Try API fetch // 2. Try API fetch
if (self.getCandles(sym)) |candle_result| { if (self.getCandles(sym)) |candle_result| {
defer self.allocator.free(candle_result.data); defer self.allocator().free(candle_result.data);
if (candle_result.data.len > 0) { if (candle_result.data.len > 0) {
const last = candle_result.data[candle_result.data.len - 1]; const last = candle_result.data[candle_result.data.len - 1];
prices.put(sym, last.close) catch {}; prices.put(sym, last.close) catch {};
@ -939,7 +980,7 @@ pub const DataService = struct {
symbol_progress: ?ProgressCallback, symbol_progress: ?ProgressCallback,
) LoadAllResult { ) LoadAllResult {
var result = LoadAllResult{ var result = LoadAllResult{
.prices = std.StringHashMap(f64).init(self.allocator), .prices = std.StringHashMap(f64).init(self.allocator()),
.cached_count = 0, .cached_count = 0,
.server_synced_count = 0, .server_synced_count = 0,
.provider_fetched_count = 0, .provider_fetched_count = 0,
@ -956,13 +997,13 @@ pub const DataService = struct {
if (total_count == 0) return result; if (total_count == 0) return result;
// Build combined symbol list // Build combined symbol list
var all_symbols = std.ArrayList([]const u8).initCapacity(self.allocator, total_count) catch return result; var all_symbols = std.ArrayList([]const u8).initCapacity(self.allocator(), total_count) catch return result;
defer all_symbols.deinit(self.allocator); defer all_symbols.deinit(self.allocator());
if (portfolio_syms) |ps| { if (portfolio_syms) |ps| {
for (ps) |sym| all_symbols.append(self.allocator, sym) catch {}; for (ps) |sym| all_symbols.append(self.allocator(), sym) catch {};
} }
for (watch_syms) |sym| all_symbols.append(self.allocator, sym) catch {}; for (watch_syms) |sym| all_symbols.append(self.allocator(), sym) catch {};
// Invalidate cache if force refresh // Invalidate cache if force refresh
if (config.force_refresh) { if (config.force_refresh) {
@ -973,7 +1014,7 @@ pub const DataService = struct {
// Phase 1: Check local cache (fast path) // Phase 1: Check local cache (fast path)
var needs_fetch: std.ArrayList([]const u8) = .empty; var needs_fetch: std.ArrayList([]const u8) = .empty;
defer needs_fetch.deinit(self.allocator); defer needs_fetch.deinit(self.allocator());
if (aggregate_progress) |p| p.emit(0, total_count, .cache_check); if (aggregate_progress) |p| p.emit(0, total_count, .cache_check);
@ -985,7 +1026,7 @@ pub const DataService = struct {
} }
result.cached_count += 1; result.cached_count += 1;
} else { } else {
needs_fetch.append(self.allocator, sym) catch {}; needs_fetch.append(self.allocator(), sym) catch {};
} }
} }
@ -998,7 +1039,7 @@ pub const DataService = struct {
// Phase 2: Server sync (parallel if server configured) // Phase 2: Server sync (parallel if server configured)
var server_failures: std.ArrayList([]const u8) = .empty; var server_failures: std.ArrayList([]const u8) = .empty;
defer server_failures.deinit(self.allocator); defer server_failures.deinit(self.allocator());
if (self.config.server_url != null) { if (self.config.server_url != null) {
self.parallelServerSync( self.parallelServerSync(
@ -1012,7 +1053,7 @@ pub const DataService = struct {
} else { } else {
// No server — all need provider fetch // No server — all need provider fetch
for (needs_fetch.items) |sym| { for (needs_fetch.items) |sym| {
server_failures.append(self.allocator, sym) catch {}; server_failures.append(self.allocator(), sym) catch {};
} }
} }
@ -1054,12 +1095,12 @@ pub const DataService = struct {
// Shared state for worker threads // Shared state for worker threads
var completed = AtomicCounter{}; var completed = AtomicCounter{};
var next_index = AtomicCounter{}; var next_index = AtomicCounter{};
const sync_results = self.allocator.alloc(ServerSyncResult, symbols.len) catch { const sync_results = self.allocator().alloc(ServerSyncResult, symbols.len) catch {
// Allocation failed — fall back to marking all as failures // Allocation failed — fall back to marking all as failures
for (symbols) |sym| failures.append(self.allocator, sym) catch {}; for (symbols) |sym| failures.append(self.allocator(), sym) catch {};
return; return;
}; };
defer self.allocator.free(sync_results); defer self.allocator().free(sync_results);
// Initialize results // Initialize results
for (sync_results, 0..) |*sr, i| { for (sync_results, 0..) |*sr, i| {
@ -1067,11 +1108,11 @@ pub const DataService = struct {
} }
// Spawn worker threads // Spawn worker threads
var threads = self.allocator.alloc(std.Thread, thread_count) catch { var threads = self.allocator().alloc(std.Thread, thread_count) catch {
for (symbols) |sym| failures.append(self.allocator, sym) catch {}; for (symbols) |sym| failures.append(self.allocator(), sym) catch {};
return; return;
}; };
defer self.allocator.free(threads); defer self.allocator().free(threads);
const WorkerContext = struct { const WorkerContext = struct {
svc: *DataService, svc: *DataService,
@ -1133,10 +1174,10 @@ pub const DataService = struct {
result.server_synced_count += 1; result.server_synced_count += 1;
} else { } else {
// Sync said success but can't read cache — treat as failure // Sync said success but can't read cache — treat as failure
failures.append(self.allocator, sr.symbol) catch {}; failures.append(self.allocator(), sr.symbol) catch {};
} }
} else { } else {
failures.append(self.allocator, sr.symbol) catch {}; failures.append(self.allocator(), sr.symbol) catch {};
} }
} }
} }
@ -1159,7 +1200,7 @@ pub const DataService = struct {
// Try provider fetch // Try provider fetch
if (self.getCandles(sym)) |candle_result| { if (self.getCandles(sym)) |candle_result| {
defer self.allocator.free(candle_result.data); defer self.allocator().free(candle_result.data);
if (candle_result.data.len > 0) { if (candle_result.data.len > 0) {
const last = candle_result.data[candle_result.data.len - 1]; const last = candle_result.data[candle_result.data.len - 1];
result.prices.put(sym, last.close) catch {}; result.prices.put(sym, last.close) catch {};
@ -1201,7 +1242,7 @@ pub const DataService = struct {
/// Results array is parallel to the input cusips array (same length, same order). /// Results array is parallel to the input cusips array (same length, same order).
/// Caller owns the returned slice and all strings within each CusipResult. /// Caller owns the returned slice and all strings within each CusipResult.
pub fn lookupCusips(self: *DataService, cusips: []const []const u8) DataError![]CusipResult { pub fn lookupCusips(self: *DataService, cusips: []const []const u8) DataError![]CusipResult {
return OpenFigi.lookupCusips(self.allocator, cusips, self.config.openfigi_key) catch return OpenFigi.lookupCusips(self.allocator(), cusips, self.config.openfigi_key) catch
return DataError.FetchFailed; return DataError.FetchFailed;
} }
@ -1213,10 +1254,10 @@ pub const DataService = struct {
if (self.getCachedCusipTicker(cusip)) |t| return t; if (self.getCachedCusipTicker(cusip)) |t| return t;
// Try OpenFIGI // Try OpenFIGI
const result = OpenFigi.lookupCusip(self.allocator, cusip, self.config.openfigi_key) catch return null; const result = OpenFigi.lookupCusip(self.allocator(), cusip, self.config.openfigi_key) catch return null;
defer { defer {
if (result.name) |n| self.allocator.free(n); if (result.name) |n| self.allocator().free(n);
if (result.security_type) |s| self.allocator.free(s); if (result.security_type) |s| self.allocator().free(s);
} }
if (result.ticker) |ticker| { if (result.ticker) |ticker| {
@ -1237,20 +1278,20 @@ pub const DataService = struct {
/// Read a cached CUSIP->ticker mapping. Returns null if not cached. /// Read a cached CUSIP->ticker mapping. Returns null if not cached.
/// Caller owns the returned string. /// Caller owns the returned string.
fn getCachedCusipTicker(self: *DataService, cusip: []const u8) ?[]const u8 { fn getCachedCusipTicker(self: *DataService, cusip: []const u8) ?[]const u8 {
const path = std.fs.path.join(self.allocator, &.{ self.config.cache_dir, "cusip_tickers.srf" }) catch return null; const path = std.fs.path.join(self.allocator(), &.{ self.config.cache_dir, "cusip_tickers.srf" }) catch return null;
defer self.allocator.free(path); defer self.allocator().free(path);
const data = std.fs.cwd().readFileAlloc(self.allocator, path, 64 * 1024) catch return null; const data = std.fs.cwd().readFileAlloc(self.allocator(), path, 64 * 1024) catch return null;
defer self.allocator.free(data); defer self.allocator().free(data);
var reader = std.Io.Reader.fixed(data); var reader = std.Io.Reader.fixed(data);
var it = srf.iterator(&reader, self.allocator, .{ .alloc_strings = false }) catch return null; var it = srf.iterator(&reader, self.allocator(), .{ .alloc_strings = false }) catch return null;
defer it.deinit(); defer it.deinit();
while (it.next() catch return null) |fields| { while (it.next() catch return null) |fields| {
const entry = fields.to(CusipEntry) catch continue; const entry = fields.to(CusipEntry) catch continue;
if (std.mem.eql(u8, entry.cusip, cusip) and entry.ticker.len > 0) { if (std.mem.eql(u8, entry.cusip, cusip) and entry.ticker.len > 0) {
return self.allocator.dupe(u8, entry.ticker) catch null; return self.allocator().dupe(u8, entry.ticker) catch null;
} }
} }
return null; return null;
@ -1258,8 +1299,8 @@ pub const DataService = struct {
/// Append a CUSIP->ticker mapping to the cache file. /// Append a CUSIP->ticker mapping to the cache file.
pub fn cacheCusipTicker(self: *DataService, cusip: []const u8, ticker: []const u8) void { pub fn cacheCusipTicker(self: *DataService, cusip: []const u8, ticker: []const u8) void {
const path = std.fs.path.join(self.allocator, &.{ self.config.cache_dir, "cusip_tickers.srf" }) catch return; const path = std.fs.path.join(self.allocator(), &.{ self.config.cache_dir, "cusip_tickers.srf" }) catch return;
defer self.allocator.free(path); defer self.allocator().free(path);
// Ensure cache dir exists // Ensure cache dir exists
if (std.fs.path.dirnamePosix(path)) |dir| { if (std.fs.path.dirnamePosix(path)) |dir| {
@ -1278,7 +1319,7 @@ pub const DataService = struct {
const entry = [_]CusipEntry{.{ .cusip = cusip, .ticker = ticker }}; const entry = [_]CusipEntry{.{ .cusip = cusip, .ticker = ticker }};
var buf: [256]u8 = undefined; var buf: [256]u8 = undefined;
var writer = file.writer(&buf); var writer = file.writer(&buf);
writer.interface.print("{f}", .{srf.fmtFrom(CusipEntry, self.allocator, &entry, .{ .emit_directives = emit_directives })}) catch return; writer.interface.print("{f}", .{srf.fmtFrom(CusipEntry, self.allocator(), &entry, .{ .emit_directives = emit_directives })}) catch return;
writer.interface.flush() catch {}; writer.interface.flush() catch {};
} }
@ -1312,12 +1353,12 @@ pub const DataService = struct {
.meta => return false, .meta => return false,
}; };
const full_url = std.fmt.allocPrint(self.allocator, "{s}/{s}{s}", .{ server_url, symbol, endpoint }) catch return false; const full_url = std.fmt.allocPrint(self.allocator(), "{s}/{s}{s}", .{ server_url, symbol, endpoint }) catch return false;
defer self.allocator.free(full_url); defer self.allocator().free(full_url);
log.debug("{s}: syncing {s} from server", .{ symbol, @tagName(data_type) }); log.debug("{s}: syncing {s} from server", .{ symbol, @tagName(data_type) });
var client = http.Client.init(self.allocator); var client = http.Client.init(self.allocator());
defer client.deinit(); defer client.deinit();
var response = client.get(full_url) catch |err| { var response = client.get(full_url) catch |err| {
@ -1356,13 +1397,13 @@ pub const DataService = struct {
/// Caller owns the returned AccountMap and must call deinit(). /// Caller owns the returned AccountMap and must call deinit().
pub fn loadAccountMap(self: *DataService, portfolio_path: []const u8) ?analysis.AccountMap { pub fn loadAccountMap(self: *DataService, portfolio_path: []const u8) ?analysis.AccountMap {
const dir_end = if (std.mem.lastIndexOfScalar(u8, portfolio_path, std.fs.path.sep)) |idx| idx + 1 else 0; const dir_end = if (std.mem.lastIndexOfScalar(u8, portfolio_path, std.fs.path.sep)) |idx| idx + 1 else 0;
const acct_path = std.fmt.allocPrint(self.allocator, "{s}accounts.srf", .{portfolio_path[0..dir_end]}) catch return null; const acct_path = std.fmt.allocPrint(self.allocator(), "{s}accounts.srf", .{portfolio_path[0..dir_end]}) catch return null;
defer self.allocator.free(acct_path); defer self.allocator().free(acct_path);
const data = std.fs.cwd().readFileAlloc(self.allocator, acct_path, 1024 * 1024) catch return null; const data = std.fs.cwd().readFileAlloc(self.allocator(), acct_path, 1024 * 1024) catch return null;
defer self.allocator.free(data); defer self.allocator().free(data);
return analysis.parseAccountsFile(self.allocator, data) catch null; return analysis.parseAccountsFile(self.allocator(), data) catch null;
} }
}; };