diff --git a/src/App.zig b/src/App.zig index c357b52..2fbcdb2 100644 --- a/src/App.zig +++ b/src/App.zig @@ -35,6 +35,17 @@ pub const App = struct { /// Without it two simultaneous adds could both read the old file and /// the second writer would clobber the first's new symbol. watch_mutex: std.Io.Mutex = .init, + /// Serializes `POST /refresh`, and does so with a NON-blocking acquire so a + /// concurrent request is rejected (409) rather than queued. + /// + /// Queueing would be worse than refusing: each refresh spends provider quota, + /// the request is synchronous, and zfin's HTTP client retries up to three + /// times on a 5xx or transient failure. A refresh slow enough to time out + /// would therefore be retried while the first was still running, and a + /// blocking lock would dutifully run every one of them in turn - turning one + /// slow request into a provider stampede. Refusing tells the caller the truth: + /// a refresh is already in flight, so wait for it. + refresh_mutex: std.Io.Mutex = .init, pub fn init(io: std.Io, allocator: std.mem.Allocator, environ: *const std.process.Environ.Map) App { const config = zfin.Config.fromEnv(io, allocator, environ); @@ -218,6 +229,11 @@ test "pathIsPublic: gated surface (key required)" { // the cost of that assumption being wrong is silent disclosure. try std.testing.expect(!pathIsPublic("/AAPL/diagnostics")); try std.testing.expect(!pathIsPublic("/BRK.B/diagnostics")); + // `/refresh` spends provider quota and writes the cache. It needs no entry + // here to be gated - the default is closed - and this asserts the default + // rather than trusting it, because an unauthenticated refresh trigger is a + // free denial-of-quota for anyone who finds the URL. + try std.testing.expect(!pathIsPublic("/refresh")); } test "pathIsPublic: returns look-alikes do not slip through" { diff --git a/src/handlers.zig b/src/handlers.zig index f1cd24b..a1b25af 100644 --- a/src/handlers.zig +++ b/src/handlers.zig @@ -84,6 +84,9 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void { \\ GET /{SYMBOL}/quote Latest quote (JSON) \\ GET /{SYMBOL}/candles Raw SRF cache file \\ GET /{SYMBOL}/candles_meta Candle freshness metadata (SRF) + \\ GET /{SYMBOL}/diagnostics Whether the server TRACKS the symbol, + \\ how far behind its peers it is, and + \\ whether its copy is stamped fresh (JSON) \\ GET /{SYMBOL}/dividends Raw SRF cache file \\ GET /{SYMBOL}/splits Raw SRF cache file \\ GET /{SYMBOL}/earnings Raw SRF cache file @@ -94,6 +97,8 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void { \\ GET /_edgar/tickers_funds EDGAR mutual-fund ticker map (SRF) \\ GET /_edgar/tickers_companies EDGAR company ticker map (SRF) \\ GET /symbols List of tracked symbols + \\ POST /refresh?symbols=A,B Force-refresh candles now (max 25; + \\ 409 if a refresh is already running) \\ \\Auth: \\ All endpoints except /, /help, and /{SYMBOL}/returns require an @@ -105,6 +110,13 @@ pub fn handleHelp(_: *App, _: *httpz.Request, res: *httpz.Response) !void { \\ fetches once from the provider, fills the cache, then serves \\ (404 only if that fetch also fails). \\ + \\Freshness (why POST for one and GET for the other): + \\ A read never triggers a refetch - a present file is served as-is and + \\ cron is the freshness authority. POST /refresh is the one way to make + \\ the server fetch on demand, and it is a POST because it spends provider + \\ quota and writes the cache. /{SYMBOL}/watch is a GET only because + \\ LibreOffice's WEBSERVICE cannot issue anything else. + \\ \\Watchlist (add a symbol to the cron refresh set): \\ GET /{SYMBOL}/watch authenticated; for your own tooling \\ GET /{SYMBOL}/returns?watch=true public, but only LibreOffice's @@ -518,6 +530,147 @@ pub fn handleDiagnostics(app: *App, req: *httpz.Request, res: *httpz.Response) ! res.body = aw.written(); } +/// Cap on symbols per `POST /refresh`. +/// +/// Each one costs a forced provider fetch, so an unbounded list is an unbounded +/// synchronous request. Generous for the intended use - one or two symbols +/// breaking a reconciliation - while keeping any single request bounded. Exceeding +/// it is a 400 naming the limit, never a silent truncation: quietly refreshing the +/// first 25 of 40 would report success for symbols nobody touched. +pub const max_refresh_symbols: usize = 25; + +/// Outcome of parsing a `symbols=` value. A union rather than an error set so the +/// 400 can name the offending token - "Invalid symbol" without saying which is a +/// worse message than the operator's own typo. +pub const SymbolList = union(enum) { + ok: []const []const u8, + /// No usable token at all. + empty, + /// Distinct symbol count after dedupe. + too_many: usize, + /// The token that failed `isPlausibleSymbol`. + malformed: []const u8, +}; + +/// Parse a comma-separated `symbols=` value into a deduped, upper-cased list. +/// +/// Empty tokens are skipped rather than rejected, so a trailing comma is harmless +/// - punctuation is not a typo'd ticker, and there is nothing an operator could +/// mean by it. A genuinely malformed token IS rejected, and rejects the whole +/// request: refreshing the valid remainder would report overall success while the +/// symbol the operator actually cared about was never touched. +/// +/// Deduping happens BEFORE the cap check, so `AAPL,AAPL,...` cannot trip a limit +/// it does not really exceed. +pub fn parseSymbolList(arena: std.mem.Allocator, raw: []const u8) !SymbolList { + var out = std.ArrayList([]const u8).empty; + var seen = std.StringHashMap(void).init(arena); + var it = std.mem.splitScalar(u8, raw, ','); + while (it.next()) |tok| { + const trimmed = std.mem.trim(u8, tok, " \t\r\n"); + if (trimmed.len == 0) continue; + const sym = try upperDupe(arena, trimmed); + if (!isPlausibleSymbol(sym)) return .{ .malformed = sym }; + if (seen.contains(sym)) continue; + try seen.put(sym, {}); + try out.append(arena, sym); + } + if (out.items.len == 0) return .empty; + if (out.items.len > max_refresh_symbols) return .{ .too_many = out.items.len }; + return .{ .ok = out.items }; +} + +/// Force-refresh candles for the given symbols, synchronously. +/// +/// The acting counterpart to `/:symbol/diagnostics`: diagnostics says a symbol is +/// behind and nothing routine will move it, and this is the lever that moves it. +/// Deliberately a POST, and deliberately NOT a `?refresh=1` parameter on the data +/// endpoints - the L2 contract is that a present file is served as-is and a read +/// never triggers a refetch, with cron as the freshness authority. Triggering the +/// refresh path is consistent with that; making reads authoritative is not. +/// +/// Reports whether each symbol's newest bar actually MOVED, which is the question +/// the caller has. A successful fetch that changes nothing is the common outcome +/// when the provider simply has no newer data, and calling that "refreshed" would +/// hide the finding. +pub fn handleRefresh(app: *App, req: *httpz.Request, res: *httpz.Response) !void { + const arena = res.arena; + const q = try req.query(); + const raw = q.get("symbols") orelse { + res.status = 400; + res.body = "Missing symbols (use ?symbols=AAPL,MSFT)"; + return; + }; + + const symbols = switch (try parseSymbolList(arena, raw)) { + .ok => |list| list, + .empty => { + res.status = 400; + res.body = "No symbols given"; + return; + }, + .malformed => |bad| { + res.status = 400; + res.body = try std.fmt.allocPrint(arena, "Invalid symbol: {s}", .{bad}); + return; + }, + .too_many => |n| { + res.status = 400; + res.body = try std.fmt.allocPrint( + arena, + "Too many symbols: {d} (limit {d})", + .{ n, max_refresh_symbols }, + ); + return; + }, + }; + + // Non-blocking on purpose - see `App.refresh_mutex`. + if (!app.refresh_mutex.tryLock()) { + res.status = 409; + res.body = "A refresh is already running; retry when it finishes"; + return; + } + defer app.refresh_mutex.unlock(app.io); + + var store = zfin.cache.Store.init(app.io, arena, app.config.cache_dir); + + var aw: std.Io.Writer.Allocating = .init(arena); + try aw.writer.writeAll("{\"results\":["); + for (symbols, 0..) |sym, i| { + if (i > 0) try aw.writer.writeByte(','); + + // Read before and after rather than trusting the fetch's return value: + // what matters is what a client will now be served, which is whatever + // landed in the cache. + const before: ?zfin.Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null; + + if (app.svc.getCandles(sym, .{ .force_refresh = true })) |result| { + result.deinit(); + const after: ?zfin.Date = if (store.readCandleMeta(sym)) |m| m.meta.last_date else null; + const moved = if (after) |a| (if (before) |b| b.lessThan(a) else true) else false; + try aw.writer.print("{{\"symbol\":\"{s}\",\"ok\":true,\"moved\":{}", .{ sym, moved }); + if (after) |a| { + try aw.writer.print(",\"last_date\":\"{f}\"}}", .{a}); + } else { + try aw.writer.writeAll(",\"last_date\":null}"); + } + } else |err| { + // The error NAME, not a generic flag: an auth failure, a rate limit + // and a delisted ticker send the operator to different places. + try aw.writer.print( + "{{\"symbol\":\"{s}\",\"ok\":false,\"error\":\"{s}\"}}", + .{ sym, @errorName(err) }, + ); + log.warn("POST /refresh: {s} failed: {t}", .{ sym, err }); + } + } + try aw.writer.writeAll("]}"); + + res.content_type = httpz.ContentType.JSON; + res.body = aw.written(); +} + pub fn handleSrfFile(app: *App, req: *httpz.Request, res: *httpz.Response, filename: []const u8, kind: SrfKind) !void { return handleSrfFileByKey(app, req, res, "symbol", filename, kind); } @@ -938,3 +1091,87 @@ test "groupPeerDate: an inconclusive group has no peer date" { // A kind with no group at all is not an error, just unanswerable. try std.testing.expectEqual(@as(?zfin.Date, null), groupPeerDate(report, .mutual_fund)); } + +test "parseSymbolList: upper-cases, dedupes, tolerates spacing and trailing commas" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + + // The realistic invocation, including the trailing comma a shell loop leaves + // behind and the space a human types after one. + const r = try parseSymbolList(a, "nke, AMZN ,nke,"); + const list = r.ok; + try std.testing.expectEqual(@as(usize, 2), list.len); + try std.testing.expectEqualStrings("NKE", list[0]); + try std.testing.expectEqualStrings("AMZN", list[1]); +} + +test "parseSymbolList: nothing usable is `empty`, not a zero-length success" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + // Punctuation only. Refreshing nothing and reporting success would tell the + // operator their symbols were handled. + try std.testing.expect((try parseSymbolList(a, ",,,")) == .empty); + try std.testing.expect((try parseSymbolList(a, " ")) == .empty); + try std.testing.expect((try parseSymbolList(a, "")) == .empty); +} + +test "parseSymbolList: a malformed token rejects the whole request, and is named" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + + // A path-shaped segment is the case worth blocking - these symbols reach the + // cache as directory names. + switch (try parseSymbolList(a, "AMZN,../../etc/passwd")) { + .malformed => |bad| try std.testing.expectEqualStrings("../../ETC/PASSWD", bad), + else => return error.TestUnexpectedResult, + } + // Rejecting the WHOLE list matters: refreshing AMZN and quietly dropping the + // typo would report success for a run that never touched what was asked for. + try std.testing.expect((try parseSymbolList(a, "AMZN,NK E")) == .malformed); + try std.testing.expect((try parseSymbolList(a, "THISTICKERISWAYTOOLONG")) == .malformed); +} + +test "parseSymbolList: the cap counts DISTINCT symbols" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const a = arena.allocator(); + + // Exactly at the limit is allowed. + var at: std.ArrayList(u8) = .empty; + defer at.deinit(a); + for (0..max_refresh_symbols) |i| { + if (i > 0) try at.append(a, ','); + try at.appendSlice(a, try std.fmt.allocPrint(a, "SY{d}", .{i})); + } + try std.testing.expect((try parseSymbolList(a, at.items)) == .ok); + + // One past it is refused, and the count reported is the distinct count. + try at.appendSlice(a, ",EXTRA"); + switch (try parseSymbolList(a, at.items)) { + .too_many => |n| try std.testing.expectEqual(max_refresh_symbols + 1, n), + else => return error.TestUnexpectedResult, + } + + // Duplicates must NOT push a legitimate list over the cap - dedupe happens + // first, so this is 1 distinct symbol however many times it appears. + var dupes: std.ArrayList(u8) = .empty; + defer dupes.deinit(a); + for (0..max_refresh_symbols + 10) |i| { + if (i > 0) try dupes.append(a, ','); + try dupes.appendSlice(a, "AMZN"); + } + const r = try parseSymbolList(a, dupes.items); + try std.testing.expectEqual(@as(usize, 1), r.ok.len); +} + +test "refresh_mutex: a second acquire fails rather than queueing" { + // The 409 path. A blocking lock would serialize retries into repeated full + // refreshes; `tryLock` is what makes the second caller get an answer instead + // of a turn in the queue. + var m: std.Io.Mutex = .init; + try std.testing.expect(m.tryLock()); + try std.testing.expect(!m.tryLock()); +} diff --git a/src/main.zig b/src/main.zig index 90f6b18..0213cca 100644 --- a/src/main.zig +++ b/src/main.zig @@ -59,6 +59,12 @@ pub fn main(init: std.process.Init) !u8 { router.get("/", handlers.handleIndex, .{}); router.get("/help", handlers.handleHelp, .{}); router.get("/symbols", handlers.handleSymbols, .{}); + // POST, not GET: this spends provider quota and writes the cache, so it is + // not `safe` in the HTTP sense and must not sit on a verb that proxies, + // prefetchers and link checkers feel free to replay. (`/:symbol/watch` + // stays GET only because LibreOffice's WEBSERVICE cannot issue anything + // else - a constraint, not a precedent.) + router.post("/refresh", handlers.handleRefresh, .{}); // Symbol routes router.get("/:symbol/returns", handlers.handleReturns, .{});