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

add server refresh command

Browse source
This commit is contained in:
Emil Lerch 2026-08-13 16:15:22 -07:00
parent 914dfca356
commit a9b923f766
Signed by: lobo
GPG key ID: A7B62D657EF764F8
6 changed files with 396 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
src/cache/freshness.zig vendored
View file

@ -406,6 +406,15 @@ test "scan: kinds are judged separately, so fund NAV lag is not staleness" {
defer r.deinit(a);
try testing.expectEqual(@as(usize, 0), r.stale.len);
// And not merely absent from `stale` - absent from BOTH lists. A fund that
// slipped into `far_behind` would exit the server's refresh non-zero and mail
// the operator every evening, which is a worse failure than a noisy list.
try testing.expectEqual(@as(usize, 0), r.far_behind.len);
// The peer maximum is what protects this: each kind's reference moves with
// its own group, so a schedule shared by every member of a group can never
// put any member behind it.
try testing.expect(r.groups[0].peer_date.?.eql(fri));
try testing.expect(r.groups[1].peer_date.?.eql(thu));
// Both groups are caught up against their own schedules.
try testing.expectEqual(market.CandleFreshness.current, r.groups[0].freshness.?);
try testing.expectEqual(market.CandleFreshness.current, r.groups[1].freshness.?);

12
src/commands/cache.zig
View file

@ -31,9 +31,15 @@ pub const meta: framework.Meta = .{
\\ 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.
\\ refresh Force-refresh candle data in the LOCAL cache, bypassing
\\ the TTL and the shared server. With no arguments,
\\ refreshes exactly what `stale` reports. With symbols,
\\ refreshes those.
\\
\\ Note the direction: this deliberately does NOT go through
\\ ZFIN_SERVER. To refresh the SERVER's copy instead, use
\\ `zfin server refresh SYMBOL...`. `zfin diagnose SYMBOL`
\\ says which side is actually behind.
\\ 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

355
src/commands/server.zig Normal file
View file

@ -0,0 +1,355 @@
//! `zfin server refresh SYMBOL...` - ask the shared server to force-refresh
//! symbols in ITS cache, right now.
//!
//! The distinction from `zfin cache refresh` matters and is easy to get backwards:
//!
//! `cache refresh` acts on the LOCAL cache, bypassing the TTL *and* the
//! shared server, so it fetches straight from the provider.
//! `server refresh` asks the SERVER to refresh its own copy. Nothing local
//! changes until the next run syncs from it.
//!
//! Which one you want depends on where the stale bar lives. `zfin diagnose SYMBOL`
//! answers that: when it reports the server offering a bar behind your local copy,
//! or tracking the symbol at all, this is the lever. When the server is fine and
//! only your copy is behind, `cache refresh` is.
//!
//! Exists because a symbol the server serves but does not refresh will hand every
//! client the same stale bar indefinitely, and until now the only fix was an ssh
//! session and a cron run.
const std = @import("std");
const cli = @import("common.zig");
const framework = @import("framework.zig");
const http = @import("../net/http.zig");
pub const ParsedArgs = struct {
symbols: []const []const u8,
};
pub const meta: framework.Meta = .{
.name = "server",
.group = .infra,
.synopsis = "Ask the shared server to refresh its own cache",
.help =
\\Usage: zfin server refresh SYMBOL [SYMBOL...]
\\
\\Force-refreshes candle data in the SHARED SERVER's cache (ZFIN_SERVER),
\\bypassing its TTL. Reports, per symbol, whether its newest bar actually
\\moved - a fetch can succeed and change nothing when the provider has no
\\newer data, which is a finding rather than a success.
\\
\\Not the same as `zfin cache refresh`, which acts on your LOCAL cache and
\\deliberately bypasses the server. Use `zfin diagnose SYMBOL` to see which
\\side is behind before picking one.
\\
\\Nothing local changes: your next normal run picks up the server's new copy.
\\
\\Requires ZFIN_SERVER, and ZFIN_SERVER_API_KEY when the server enforces one.
\\The server caps a single request (currently 25 symbols) and refuses a second
\\concurrent refresh rather than queueing it.
\\
,
.uppercase_first_arg = false,
.user_errors = error{ MissingSubcommand, UnknownSubcommand, MissingSymbol },
};
pub fn parseArgs(ctx: *framework.RunCtx, cmd_args: []const []const u8) !ParsedArgs {
if (cmd_args.len < 1) {
cli.stderrPrint(ctx.io, "Error: 'server' requires a subcommand (refresh)\n");
return error.MissingSubcommand;
}
// Only one subcommand today. Matched explicitly rather than ignored so a typo
// is an error instead of silently refreshing whatever came next - `zfin server
// referesh AMZN` must not treat "referesh" as a symbol.
if (!std.mem.eql(u8, cmd_args[0], "refresh")) {
cli.stderrPrint(ctx.io, "Error: unknown 'server' subcommand (expected: refresh)\n");
return error.UnknownSubcommand;
}
if (cmd_args.len < 2) {
cli.stderrPrint(ctx.io, "Error: 'server refresh' requires at least one symbol\n");
return error.MissingSymbol;
}
return .{ .symbols = cmd_args[1..] };
}
/// One symbol's outcome, as the server reported it.
pub const Outcome = struct {
symbol: []const u8,
ok: bool,
/// Did the newest bar actually advance? The question the caller has: a
/// successful fetch that moves nothing means the provider had nothing newer,
/// which is a different situation from a fix.
moved: bool = false,
last_date: ?[]const u8 = null,
/// Server-side error NAME when `ok` is false, passed through rather than
/// reworded - an auth failure, a rate limit and a delisted ticker each send
/// you somewhere different.
err: ?[]const u8 = null,
};
/// Parse the server's `{"results":[...]}` body.
///
/// Returns null only when the body is not the expected shape at all. Individual
/// fields are read defensively: a newer server adding fields must not cost an
/// older client the ones it understands. Separated from the request so the shape
/// is testable without a server.
pub fn parseResults(arena: std.mem.Allocator, body: []const u8) ?[]const Outcome {
const parsed = std.json.parseFromSlice(std.json.Value, arena, body, .{}) catch return null;
const obj = switch (parsed.value) {
.object => |o| o,
else => return null,
};
const results = switch (obj.get("results") orelse return null) {
.array => |a| a,
else => return null,
};
var out = std.ArrayList(Outcome).empty;
for (results.items) |item| {
const r = switch (item) {
.object => |o| o,
else => continue,
};
const sym = switch (r.get("symbol") orelse continue) {
.string => |s| s,
else => continue,
};
var o = Outcome{ .symbol = sym, .ok = false };
if (r.get("ok")) |v| if (v == .bool) {
o.ok = v.bool;
};
if (r.get("moved")) |v| if (v == .bool) {
o.moved = v.bool;
};
if (r.get("last_date")) |v| if (v == .string) {
o.last_date = v.string;
};
if (r.get("error")) |v| if (v == .string) {
o.err = v.string;
};
out.append(arena, o) catch return null;
}
return out.items;
}
/// Human-readable guidance for a transport-level failure.
///
/// Every branch names the actual condition rather than "request failed": the
/// whole point of `HttpError.Conflict` existing is that "a refresh is already
/// running" and "your key is wrong" are different problems, and collapsing them
/// sends you to the wrong place.
pub fn failureAdvice(err: anyerror) []const u8 {
return switch (err) {
error.Conflict => "a refresh is already running there - retry once it finishes",
error.Unauthorized => "rejected: check ZFIN_SERVER_API_KEY",
error.NotFound => "this server has no /refresh endpoint - it predates the feature",
error.RateLimited => "the server is rate-limiting; retry shortly",
error.ServerError => "the server errored; check its logs",
// 400 lands here, and the server's 400s are specific ("Too many symbols:
// 26 (limit 25)", "Invalid symbol: NK E"). That text is already emitted by
// the transport as an `http rejection body` warning, so point at it rather
// than reprinting "request failed" directly beneath the real reason.
//
// Deliberately NOT solved by mapping 400 to its own HttpError variant:
// `InvalidResponse` is load-bearing elsewhere - `service.zig` routes it to
// the Yahoo fallback and `isPermanentProviderFailure` counts it transient -
// so renaming it would alter the provider chain to improve one CLI message.
error.InvalidResponse => "the server rejected the request - its reason is on the `http rejection body` line above",
else => "request failed",
};
}
pub fn run(ctx: *framework.RunCtx, parsed: ParsedArgs) !void {
const io = ctx.io;
const out = ctx.out;
const base = ctx.config.server_url orelse {
cli.stderrPrint(io, "Error: ZFIN_SERVER is not set - there is no server to ask\n");
return;
};
var arena_state = std.heap.ArenaAllocator.init(ctx.allocator);
defer arena_state.deinit();
const arena = arena_state.allocator();
// Uppercased here so the request matches what the server stores and what the
// response echoes; the framework's `uppercase_first_arg` only covers the first
// operand, which for this command is the subcommand.
var joined = std.ArrayList(u8).empty;
for (parsed.symbols, 0..) |sym, i| {
if (i > 0) try joined.append(arena, ',');
const upper = try arena.alloc(u8, sym.len);
for (sym, 0..) |c, j| upper[j] = std.ascii.toUpper(c);
try joined.appendSlice(arena, upper);
}
// Percent-encoded rather than interpolated. A hand-built query string put a
// malformed symbol's space straight into the request line, so httpz rejected
// the request at the HTTP layer and answered with its own generic "Invalid
// Request" - the operator saw a protocol complaint instead of the server's
// "Invalid symbol: NK E". An `&` would have been worse: it would silently
// truncate the symbol list. `isQueryValueChar` leaves `,` unencoded, which is
// what the server splits on.
const endpoint = try std.fmt.allocPrint(arena, "{s}/refresh", .{base});
const url = try http.buildUrl(arena, endpoint, &.{.{ "symbols", joined.items }});
var client = http.Client.init(io, arena);
defer client.deinit();
var hdr: [1]std.http.Header = .{.{ .name = "", .value = "" }};
const extra: []const std.http.Header = if (ctx.config.server_api_key) |k| blk: {
hdr[0] = .{ .name = "X-API-Key", .value = k };
break :blk hdr[0..1];
} else &.{};
try out.print("Asking {s} to refresh {d} symbol(s)...\n", .{ base, parsed.symbols.len });
try out.flush();
// POST, with an empty body: the symbols travel as query parameters, matching
// the server's other operator endpoints and keeping the call curl-able.
var resp = client.request(.POST, url, "", extra) catch |err| {
// Both the advice and the underlying error name: the advice is what to do,
// the name is what actually happened, and dropping either leaves the
// operator guessing at one of them.
const msg = std.fmt.allocPrint(arena, "Error: {s} ({s})\n", .{
failureAdvice(err),
@errorName(err),
}) catch "Error: request failed\n";
cli.stderrPrint(io, msg);
return;
};
defer resp.deinit();
const results = parseResults(arena, resp.body) orelse {
cli.stderrPrint(io, "Error: could not parse the server's reply\n");
return;
};
var moved: usize = 0;
var failed: usize = 0;
for (results) |r| {
if (!r.ok) {
failed += 1;
try out.print(" {s:<10} FAILED {s}\n", .{ r.symbol, r.err orelse "unknown error" });
continue;
}
if (r.moved) moved += 1;
// "unchanged" rather than "ok": a successful fetch that moved nothing is
// the shape of a provider with no newer data, and calling it ok would hide
// exactly what the operator came to find out.
try out.print(" {s:<10} {s:<10} {s}\n", .{
r.symbol,
if (r.moved) "moved" else "unchanged",
r.last_date orelse "no bar",
});
}
try out.print("\n{d} moved, {d} unchanged, {d} failed\n", .{
moved,
results.len - moved - failed,
failed,
});
if (moved > 0) {
try out.print("Your local cache is untouched - a normal run will sync the new copy.\n", .{});
}
try out.flush();
}
// ── tests ────────────────────────────────────────────────────
const testing = std.testing;
test "parseResults: the shape zfin-server actually returns" {
// Verbatim from `POST /refresh?symbols=NKE,AMZN,ZZZZQQ` against a live server.
var arena = std.heap.ArenaAllocator.init(testing.allocator);
defer arena.deinit();
const body =
\\{"results":[{"symbol":"NKE","ok":true,"moved":true,"last_date":"2026-08-12"},{"symbol":"AMZN","ok":true,"moved":false,"last_date":"2026-08-12"},{"symbol":"ZZZZQQ","ok":false,"error":"FetchFailed"}]}
;
const r = parseResults(arena.allocator(), body).?;
try testing.expectEqual(@as(usize, 3), r.len);
try testing.expectEqualStrings("NKE", r[0].symbol);
try testing.expect(r[0].ok);
try testing.expect(r[0].moved);
try testing.expectEqualStrings("2026-08-12", r[0].last_date.?);
// The distinction the command exists to surface: fetched fine, moved nothing.
try testing.expect(r[1].ok);
try testing.expect(!r[1].moved);
// A failure carries the server's error name, not a reworded summary.
try testing.expect(!r[2].ok);
try testing.expectEqualStrings("FetchFailed", r[2].err.?);
}
test "parseResults: a null last_date survives, and junk is rejected" {
var arena = std.heap.ArenaAllocator.init(testing.allocator);
defer arena.deinit();
const a = arena.allocator();
// An uncached symbol reports success with no bar. `null` must not become the
// string "null" or an empty date.
const r = parseResults(a, "{\"results\":[{\"symbol\":\"X\",\"ok\":true,\"moved\":false,\"last_date\":null}]}").?;
try testing.expectEqual(@as(?[]const u8, null), r[0].last_date);
// Not JSON at all - e.g. an HTML error page from a proxy in front of it.
try testing.expectEqual(@as(?[]const Outcome, null), parseResults(a, "<html>502</html>"));
// JSON, wrong shape.
try testing.expectEqual(@as(?[]const Outcome, null), parseResults(a, "[1,2,3]"));
try testing.expectEqual(@as(?[]const Outcome, null), parseResults(a, "{\"other\":[]}"));
// An empty result set is valid, not an error.
try testing.expectEqual(@as(usize, 0), parseResults(a, "{\"results\":[]}").?.len);
}
test "parseResults: an unknown field does not blind an older client" {
// Forward compatibility, the same reason `diagnose` reads fields individually
// rather than by struct coercion.
var arena = std.heap.ArenaAllocator.init(testing.allocator);
defer arena.deinit();
const body =
\\{"results":[{"symbol":"NKE","ok":true,"moved":true,"last_date":"2026-08-12","queued_at":123,"extra":{"a":1}}],"summary":"whatever"}
;
const r = parseResults(arena.allocator(), body).?;
try testing.expectEqual(@as(usize, 1), r.len);
try testing.expect(r[0].moved);
}
test "failureAdvice: each condition points somewhere different" {
// The justification for adding `HttpError.Conflict` at all. Before it, 409
// collapsed into InvalidResponse and this command would have told the operator
// their request was malformed when it just needed a retry.
try testing.expect(std.mem.indexOf(u8, failureAdvice(error.Conflict), "already running") != null);
try testing.expect(std.mem.indexOf(u8, failureAdvice(error.Unauthorized), "ZFIN_SERVER_API_KEY") != null);
try testing.expect(std.mem.indexOf(u8, failureAdvice(error.NotFound), "predates") != null);
// 400 arrives as InvalidResponse. The advice must send the operator to the
// logged rejection body, which carries the server's exact complaint - observed
// returning "Too many symbols: 26 (limit 25)" while this line said only
// "request failed".
try testing.expect(std.mem.indexOf(u8, failureAdvice(error.InvalidResponse), "rejection body") != null);
try testing.expect(!std.mem.eql(u8, failureAdvice(error.InvalidResponse), failureAdvice(error.ConnectionRefused)));
// Every branch must say something concrete; none may be empty.
for ([_]anyerror{
error.Conflict, error.Unauthorized, error.NotFound,
error.RateLimited, error.ServerError, error.InvalidResponse,
error.ConnectionRefused,
}) |e| {
try testing.expect(failureAdvice(e).len > 0);
}
// And the advice must never be the same for Conflict and Unauthorized, which
// is the exact collapse this replaced.
try testing.expect(!std.mem.eql(u8, failureAdvice(error.Conflict), failureAdvice(error.Unauthorized)));
}
test "the symbols parameter is percent-encoded" {
// Guards the bug this shipped with: an unencoded space made the request line
// malformed, so the server never saw the symbol and could not explain what was
// wrong with it. An `&` would have truncated the list without any error at all.
const a = testing.allocator;
const url = try http.buildUrl(a, "https://h/refresh", &.{.{ "symbols", "NK E,A&B" }});
defer a.free(url);
try testing.expectEqualStrings("https://h/refresh?symbols=NK%20E,A%26B", url);
// The comma must NOT be encoded - it is the separator the server splits on.
try testing.expect(std.mem.indexOfScalar(u8, url, ',') != null);
}

1
src/main.zig
View file

@ -47,6 +47,7 @@ const command_modules = .{
// Infrastructure
.cache = @import("commands/cache.zig"),
.diagnose = @import("commands/diagnose.zig"),
.server = @import("commands/server.zig"),
.doctor = @import("commands/doctor.zig"),
.version = @import("commands/version.zig"),
};

10
src/net/http.zig
View file

@ -48,6 +48,14 @@ pub const HttpError = std.Uri.ParseError ||
/// by the caller's current plan. Providers should translate this into
/// "no data" rather than a hard failure.
PaymentRequired,
/// HTTP 409 Conflict - the request cannot run because an equivalent one
/// is already in flight server-side. Distinct from `ServerError` because
/// it must NOT be retried by the transport: the correct response is to
/// wait for the in-flight operation, and a retry would only be refused
/// again. `zfin-server`'s `POST /refresh` returns this while a refresh is
/// running, and collapsing it into `InvalidResponse` reported a
/// wait-and-retry condition as a malformed request.
Conflict,
ServerError,
InvalidResponse,
};
@ -426,6 +434,7 @@ pub const Client = struct {
.unauthorized, .forbidden => HttpError.Unauthorized,
.payment_required => HttpError.PaymentRequired,
.not_found => HttpError.NotFound,
.conflict => HttpError.Conflict,
.internal_server_error, .bad_gateway, .service_unavailable, .gateway_timeout => HttpError.ServerError,
else => HttpError.InvalidResponse,
};
@ -501,6 +510,7 @@ test "classifyResponse maps each HTTP status to its HttpError" {
.{ .status = .forbidden, .expected = HttpError.Unauthorized },
.{ .status = .payment_required, .expected = HttpError.PaymentRequired },
.{ .status = .not_found, .expected = HttpError.NotFound },
.{ .status = .conflict, .expected = HttpError.Conflict },
.{ .status = .internal_server_error, .expected = HttpError.ServerError },
.{ .status = .bad_gateway, .expected = HttpError.ServerError },
.{ .status = .service_unavailable, .expected = HttpError.ServerError },

12
src/service.zig
View file

@ -3300,6 +3300,18 @@ pub const DataService = struct {
/// returns the next boundary when it is, so a genuinely caught-up fetch
/// behaves exactly as before.
///
/// TWO EARLIER DIAGNOSES OF THIS WERE WRONG, recorded so they are not
/// re-derived at the same cost:
///
/// - It is NOT `market.staleCandleExpiry` routing `.overdue` to the next
/// boundary. That path was never reached on the observed run:
/// `created=Mon 17:00` beside `expires=Tue 16:55` proves it, because five
/// minutes past the target is well inside `provider_lag_grace_s`, so the
/// verdict there was `.lagging`, not `.overdue`.
/// - It is NOT the 90-minute grace window being too short. It behaved
/// exactly as designed; the defect was upstream of it, in treating any
/// non-empty fetch result as proof of catching up.
///
/// Takes the maximum rather than the last element: provider ordering is not
/// something this decision should depend on.
fn expiryAfterFetch(now_s: i64, kind: market.InstrumentKind, candles: []const Candle) i64 {