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

add zfin cache refresh command to force re-download

Browse source
This commit is contained in:
Emil Lerch 2026-08-11 12:26:52 -07:00
parent 937822f9b0
commit 97c4e3a2e8
Signed by: lobo
GPG key ID: A7B62D657EF764F8
2 changed files with 244 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
src/cache/freshness.zig vendored
View file

@ -59,9 +59,11 @@ pub const Entry = struct {
/// behind peers holding Tuesday's. More than that is not lag.
///
/// The split exists because the two have different answers. Inside the window a
/// refresh is the remedy. Outside it, a refresh has most likely already been
/// attempted and failed, so offering one as the fix sends the operator in
/// circles.
/// refresh is the routine remedy. Outside it, the gap is past anything lag
/// explains, so the cause is elsewhere and worth naming as a separate finding -
/// but NOT because a refresh is futile. Observed counter-example: a watchlist
/// symbol sat 42 days behind purely because no refresh path had ever included
/// it, and a single forced fetch brought it fully current.
///
/// This module does NOT infer a cause, and callers must not either. The
/// candidates are many and none of them are visible from here: the ticker

268
src/commands/cache.zig
View file

@ -7,11 +7,16 @@ const freshness = @import("../cache/freshness.zig");
const Store = zfin.cache.Store;
const DataType = zfin.cache.DataType;
const Date = zfin.Date;
pub const Subcommand = enum { stats, stale, clear };
pub const Subcommand = enum { stats, stale, refresh, clear };
pub const ParsedArgs = struct {
sub: Subcommand,
/// Symbols for `refresh`. Empty means "whatever the sweep found", which is
/// the point of the command - the operator should not have to know which
/// symbols are wrong, since not knowing is the original problem.
symbols: []const []const u8 = &.{},
};
pub const meta: framework.Meta = .{
@ -19,13 +24,16 @@ pub const meta: framework.Meta = .{
.group = .infra,
.synopsis = "Inspect or clear the local provider-data cache",
.help =
\\Usage: zfin cache <stats|stale|clear>
\\Usage: zfin cache <stats|stale|clear> | zfin cache refresh [SYMBOL...]
\\
\\Subcommands:
\\ stats List every cached symbol with per-data-type size,
\\ age, and freshness state. Stale entries (past TTL)
\\ are flagged. Includes the cusip_tickers.srf file
\\ if present.
\\ refresh Force-refresh candle data, bypassing the TTL and the
\\ shared server. With no arguments, refreshes exactly what
\\ `stale` reports. With symbols, refreshes those.
\\ stale Find symbols whose newest candle is behind their
\\ peers'. Compares each symbol against others of the
\\ same kind (equity vs mutual fund) rather than against
@ -40,7 +48,7 @@ pub const meta: framework.Meta = .{
\\
,
.uppercase_first_arg = false,
.user_errors = error{ MissingSubcommand, UnexpectedArg, UnknownSubcommand },
.user_errors = error{ MissingSubcommand, UnexpectedArg, UnknownSubcommand, RefreshDisabled, NoDataService },
};
/// Data types to show in the stats table (skip candles_meta and meta - internal bookkeeping).
@ -62,14 +70,18 @@ const display_labels = [_][]const u8{
pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedArgs {
if (cmd_args.len < 1) {
cli.stderrPrint(ctx.io, "Error: 'cache' requires a subcommand (stats, stale, clear)\n");
cli.stderrPrint(ctx.io, "Error: 'cache' requires a subcommand (stats, stale, refresh, clear)\n");
return error.MissingSubcommand;
}
const sub_str = cmd_args[0];
// `refresh` is the only subcommand that takes operands.
if (std.mem.eql(u8, sub_str, "refresh")) {
return .{ .sub = .refresh, .symbols = cmd_args[1..] };
}
if (cmd_args.len > 1) {
cli.stderrPrint(ctx.io, "Error: 'cache' takes a single subcommand\n");
return error.UnexpectedArg;
}
const sub_str = cmd_args[0];
if (std.mem.eql(u8, sub_str, "stats")) {
return .{ .sub = .stats };
}
@ -81,7 +93,7 @@ pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedAr
}
cli.stderrPrint(ctx.io, "Error: unknown cache subcommand '");
cli.stderrPrint(ctx.io, sub_str);
cli.stderrPrint(ctx.io, "'. Use 'stats', 'stale' or 'clear'.\n");
cli.stderrPrint(ctx.io, "'. Use 'stats', 'stale', 'refresh' or 'clear'.\n");
return error.UnknownSubcommand;
}
@ -89,6 +101,7 @@ pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
switch (parsed.sub) {
.stats => try runStats(ctx),
.stale => try runStale(ctx),
.refresh => try runRefresh(ctx, parsed.symbols),
.clear => try runClear(ctx),
}
}
@ -207,33 +220,47 @@ fn runStats(ctx: *framework.RunCtx) !void {
/// Read-only by design: it reports and never fetches, so it is safe to run
/// against a shared cache and safe to put in front of an operator who has not
/// decided what to do yet. `zfin cache refresh` is the acting half.
fn runStale(ctx: *framework.RunCtx) !void {
/// Everything the sweep needs, allocated in one arena the caller owns.
///
/// Shared by `stale` (report it) and `refresh` (act on it), because the whole
/// point of `refresh` with no arguments is that it acts on exactly what `stale`
/// reports - two builds of this would be two answers to the same question.
const Sweep = struct {
arena_state: std.heap.ArenaAllocator,
store: Store,
report: freshness.Report,
/// True when there was nothing to look at.
empty: bool,
fn deinit(self: *Sweep) void {
self.arena_state.deinit();
}
};
fn sweep(ctx: *framework.RunCtx) !Sweep {
const io = ctx.io;
const allocator = ctx.allocator;
const out = ctx.out;
// wall-clock required: the sweep compares each symbol's newest bar against
// the market calendar's notion of what should be published by now.
// Captured once so every symbol is judged against the same instant.
const now_s = std.Io.Timestamp.now(io, .real).toSeconds();
// Arena for the transient string work below. Not a nicety: the tracked-set
// keys are borrowed by a hashmap that outlives the block that builds them,
// and a scoped `defer free` there dangles every key before the sweep reads
// it. One arena at function scope removes that hazard and the two smaller
// leaks around it.
// Arena for the transient string work. Not a nicety: the tracked-set keys
// are borrowed by a hashmap that outlives the block that builds them, and a
// scoped `defer free` there dangles every key before the sweep reads it.
var arena_state = std.heap.ArenaAllocator.init(allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
var store = Store.init(io, allocator, ctx.config.cache_dir);
const keys = store.cacheKeys(arena) catch {
try out.print("Cache is empty; nothing to check.\n", .{});
return;
};
const keys = store.cacheKeys(arena) catch &.{};
if (keys.len == 0) {
try out.print("Cache is empty; nothing to check.\n", .{});
return;
return .{
.arena_state = arena_state,
.store = store,
.report = .{ .stale = &.{}, .far_behind = &.{}, .orphans = &.{}, .missing = &.{}, .groups = &.{} },
.empty = true,
};
}
// The set zfin intends to keep fresh. Absent a portfolio every cached
@ -247,9 +274,23 @@ fn runStale(ctx: *framework.RunCtx) !void {
}
const entries = try freshness.collect(arena, &store, keys, &tracked);
return .{
.arena_state = arena_state,
.store = store,
.report = try freshness.scan(arena, entries, now_s),
.empty = false,
};
}
var report = try freshness.scan(arena, entries, now_s);
defer report.deinit(arena);
fn runStale(ctx: *framework.RunCtx) !void {
const out = ctx.out;
var sw = try sweep(ctx);
defer sw.deinit();
if (sw.empty) {
try out.print("Cache is empty; nothing to check.\n", .{});
return;
}
const report = sw.report;
for (report.groups) |g| {
const label = switch (g.kind) {
@ -292,13 +333,13 @@ fn runStale(ctx: *framework.RunCtx) !void {
for (report.far_behind) |f| {
try out.print(" {s:<10} {f} {d}d behind {f}\n", .{ f.symbol, f.last_date, f.days_behind, f.peer_date });
}
try out.print(" A refresh has probably already been tried on these. Worth checking:\n", .{});
try out.print(" - whether whatever refreshes this cache includes the symbol at all\n", .{});
try out.print(" Try `zfin cache refresh` first - a gap this size often means nothing has\n", .{});
try out.print(" ever refreshed the symbol, not that a refresh cannot. If it does not move:\n", .{});
try out.print(" - check whatever refreshes this cache actually includes it\n", .{});
try out.print(" (a shared server refreshes ITS symbol set, not yours)\n", .{});
try out.print(" - whether the ticker changed, or the provider wants a different form\n", .{});
try out.print(" - whether that one symbol is failing auth or being rate-limited\n", .{});
try out.print(" - whether the provider still covers it, or it stopped trading\n", .{});
try out.print(" `zfin cache stats` shows the per-symbol fetch state.\n", .{});
try out.print(" - check whether the ticker changed, or the provider wants another form\n", .{});
try out.print(" - check whether that one symbol is failing auth or being rate-limited\n", .{});
try out.print(" - check whether the provider still covers it, or it stopped trading\n", .{});
}
if (report.missing.len > 0) {
@ -314,6 +355,109 @@ fn runStale(ctx: *framework.RunCtx) !void {
}
}
/// Symbols a bare `cache refresh` should act on, worst first within each bucket.
///
/// BOTH buckets, deliberately. `far_behind` is not "a refresh is futile" - a
/// watchlist symbol was observed 42 days behind purely because no refresh path
/// had ever included it, and one forced fetch brought it fully current.
/// Attempting the refresh is how the operator learns which case they are in, and
/// it costs a single fetch per symbol.
///
/// Strings are duped into `arena` because the report borrows from a `Sweep` whose
/// arena the caller may outlive.
fn refreshTargets(arena: std.mem.Allocator, report: freshness.Report) ![][]const u8 {
var list: std.ArrayList([]const u8) = .empty;
errdefer list.deinit(arena);
for (report.stale) |f| try list.append(arena, try arena.dupe(u8, f.symbol));
for (report.far_behind) |f| try list.append(arena, try arena.dupe(u8, f.symbol));
return list.toOwnedSlice(arena);
}
/// Force-refresh candle data for specific symbols, or for whatever the sweep
/// found.
///
/// `force_refresh` is the point: it bypasses both the TTL and the shared server,
/// so it recovers a symbol whose local copy is fresh-stamped but stale-dated -
/// which is precisely the state a lagging server leaves clients in.
///
/// With no arguments it acts on exactly what `zfin cache stale` reports, via the
/// same `sweep`. That matters because not knowing which symbols are wrong is the
/// original problem; requiring the operator to name them would hand the problem
/// back.
fn runRefresh(ctx: *framework.RunCtx, symbols: []const []const u8) !void {
const io = ctx.io;
const out = ctx.out;
const svc = ctx.svc orelse {
cli.stderrPrint(io, "Error: 'cache refresh' needs data-service access\n");
return error.NoDataService;
};
// `skip_network` beats `force_refresh` inside the service, so a refresh
// under `--refresh-data=never` would silently do nothing. Say so instead.
if (ctx.globals.refresh_policy == .never) {
cli.stderrPrint(io, "Error: 'cache refresh' fetches, which --refresh-data=never forbids\n");
return error.RefreshDisabled;
}
var arena_state = std.heap.ArenaAllocator.init(ctx.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Resolve the target list.
var targets: []const []const u8 = symbols;
var sw: ?Sweep = null;
defer if (sw) |*w| w.deinit();
if (symbols.len == 0) {
sw = try sweep(ctx);
targets = try refreshTargets(arena, sw.?.report);
if (targets.len == 0) {
try out.print("Nothing behind its peers; no refresh needed.\n", .{});
return;
}
try out.print("Refreshing {d} symbol(s) the sweep flagged.\n\n", .{targets.len});
} else {
try out.print("Refreshing {d} named symbol(s).\n\n", .{targets.len});
}
var store = Store.init(io, ctx.allocator, ctx.config.cache_dir);
var moved: usize = 0;
var failed: usize = 0;
for (targets) |sym| {
const before: ?Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null;
if (svc.getCandles(sym, .{ .force_refresh = true })) |r| {
r.deinit();
} else |err| {
// Name the error. "refresh failed" alone leaves the operator unable
// to tell a rate limit from a bad symbol from a dead key.
try out.print(" {s:<10} FAILED: {s}\n", .{ sym, @errorName(err) });
failed += 1;
continue;
}
const after: ?Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null;
if (before != null and after != null and before.?.lessThan(after.?)) {
try out.print(" {s:<10} {f} -> {f}\n", .{ sym, before.?, after.? });
moved += 1;
} else if (after) |a| {
// The provider had nothing newer. Worth stating plainly rather than
// printing an unchanged line and letting it read as success.
try out.print(" {s:<10} still {f} - the provider had nothing newer\n", .{ sym, a });
} else {
try out.print(" {s:<10} no candle data after refresh\n", .{sym});
}
}
try out.print("\n{d} of {d} moved forward", .{ moved, targets.len });
if (failed > 0) try out.print(", {d} failed", .{failed});
try out.print(".\n", .{});
if (moved < targets.len - failed) {
// No claim about why. A symbol can fail to move because the provider
// genuinely has nothing newer, or because the request never reaches the
// right data at all - and this command cannot tell those apart.
try out.print("Symbols that did not move need a look at the provider side; `zfin cache stale` lists what to check.\n", .{});
}
}
fn runClear(ctx: *framework.RunCtx) !void {
var store = Store.init(ctx.io, ctx.allocator, ctx.config.cache_dir);
try store.clearAll();
@ -523,3 +667,69 @@ test "parseArgs: 'stale' resolves to .stale" {
const parsed = try parseArgs(&ctx, &args);
try std.testing.expectEqual(Subcommand.stale, parsed.sub);
}
test "parseArgs: 'refresh' accepts an optional symbol list" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;
// No operands: acts on whatever the sweep found, which is the point - not
// knowing which symbols are wrong is the original problem.
const bare = try parseArgs(&ctx, &[_][]const u8{"refresh"});
try std.testing.expectEqual(Subcommand.refresh, bare.sub);
try std.testing.expectEqual(@as(usize, 0), bare.symbols.len);
const named = try parseArgs(&ctx, &[_][]const u8{ "refresh", "AMZN", "NKE" });
try std.testing.expectEqual(Subcommand.refresh, named.sub);
try std.testing.expectEqual(@as(usize, 2), named.symbols.len);
try std.testing.expectEqualStrings("AMZN", named.symbols[0]);
try std.testing.expectEqualStrings("NKE", named.symbols[1]);
}
test "parseArgs: only 'refresh' takes operands" {
var ctx: framework.RunCtx = undefined;
ctx.io = std.testing.io;
// `stats`/`stale`/`clear` are still single-word, so a stray operand is a
// typo worth rejecting rather than silently ignoring.
try std.testing.expectError(error.UnexpectedArg, parseArgs(&ctx, &[_][]const u8{ "stats", "AMZN" }));
try std.testing.expectError(error.UnexpectedArg, parseArgs(&ctx, &[_][]const u8{ "clear", "AMZN" }));
}
test "refreshTargets: includes both buckets, stale first" {
var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// `far_behind` is included on purpose. SPCX sat 42 days behind because
// neither the server's set nor the client's watchlist path fetched it, and a
// single forced refresh brought it current - so excluding that bucket would
// have withheld the fix from the symbol that needed it most.
const stale = [_]freshness.Finding{
.{ .symbol = "AMZN", .kind = .equity, .last_date = Date.fromYmd(2026, 8, 7), .peer_date = Date.fromYmd(2026, 8, 10), .days_behind = 3 },
};
const far = [_]freshness.Finding{
.{ .symbol = "SPCX", .kind = .equity, .last_date = Date.fromYmd(2026, 6, 29), .peer_date = Date.fromYmd(2026, 8, 10), .days_behind = 42 },
};
const targets = try refreshTargets(arena, .{
.stale = @constCast(&stale),
.far_behind = @constCast(&far),
.orphans = &.{},
.missing = &.{},
.groups = &.{},
});
try std.testing.expectEqual(@as(usize, 2), targets.len);
try std.testing.expectEqualStrings("AMZN", targets[0]);
try std.testing.expectEqualStrings("SPCX", targets[1]);
}
test "refreshTargets: a clean sweep yields nothing to do" {
var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator);
defer arena_state.deinit();
const targets = try refreshTargets(arena_state.allocator(), .{
.stale = &.{},
.far_behind = &.{},
.orphans = &.{},
.missing = &.{},
.groups = &.{},
});
try std.testing.expectEqual(@as(usize, 0), targets.len);
}