fix: populate short interest cache directly instead of relying on clogged drain queue
The stale-while-revalidate pattern queues background fetches via the AdapterQueue, but with 227+ pending jobs, short interest keys take too long to process. Users saw empty panels on every symbol because the fetch never completed before the response returned. Fix: - shortInterest procedure: when Yahoo + NASDAQ caches are both empty, fetch directly via the adapters, store in cache, and return — all within the request. Subsequent requests hit the cache. Bypasses the clogged queue entirely. - FinraBulkAdapter: guard against per-symbol keys (non-'latest' and non-date IDs) by returning a skipped result instead of trying to download a file named after the ticker. - finraShortInterest handler: return isStale=false when no data is in the table, so per-symbol reads never trigger queue fetches (FINRA is schedule-only). Router stale computation now excludes FINRA from the combined stale flag.
This commit is contained in:
@@ -23,6 +23,17 @@ export class FinraBulkAdapter implements SourceFetch {
|
|||||||
const { kind, id } = parseCacheKey(key);
|
const { kind, id } = parseCacheKey(key);
|
||||||
if (kind !== 'finraShortinterest') throw new Error(`FinraBulkAdapter: unknown kind '${kind}'`);
|
if (kind !== 'finraShortinterest') throw new Error(`FinraBulkAdapter: unknown kind '${kind}'`);
|
||||||
|
|
||||||
|
// Only handle bulk-ingest keys ('latest' or explicit date).
|
||||||
|
// Per-symbol keys (e.g. 'AAPL') come from cache.get stale reads and
|
||||||
|
// must not trigger a bulk download — FINRA data is populated via schedule only.
|
||||||
|
if (id !== 'latest' && !/^\d{4}-\d{2}-\d{2}$/.test(id)) {
|
||||||
|
return {
|
||||||
|
value: { ok: true, skipped: true, reason: 'per-symbol key; FINRA is bulk-ingest only' },
|
||||||
|
ttlClass: 'short_interest',
|
||||||
|
provenance: { fetchedAt: new Date().toISOString(), sourceKind: 'finra-bulk', rawSourceId: id },
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
const settlementDate = id === 'latest' ? this._guessSettlementDate() : id;
|
const settlementDate = id === 'latest' ? this._guessSettlementDate() : id;
|
||||||
|
|
||||||
const result = await downloadAndIngestFinra(this._db, settlementDate);
|
const result = await downloadAndIngestFinra(this._db, settlementDate);
|
||||||
|
|||||||
+4
-1
@@ -340,7 +340,10 @@ const finraShortInterestHandler: KindHandler = {
|
|||||||
write(_d, _symbol, _value, _provenance) {
|
write(_d, _symbol, _value, _provenance) {
|
||||||
// FINRA data is bulk-ingested via FinraIngestService; no per-symbol write needed.
|
// FINRA data is bulk-ingested via FinraIngestService; no per-symbol write needed.
|
||||||
},
|
},
|
||||||
isStale(ts, now) { return tsAgeMs(ts, now) > TTL_MS.short_interest; },
|
isStale(ts, now) {
|
||||||
|
if (ts === null) return false; // schedule-only; never queue per-symbol fetches
|
||||||
|
return tsAgeMs(ts, now) > TTL_MS.short_interest;
|
||||||
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
const HANDLERS = new Map<string, KindHandler>([
|
const HANDLERS = new Map<string, KindHandler>([
|
||||||
|
|||||||
@@ -1928,10 +1928,33 @@ const institutionalRouter = router({
|
|||||||
ctx.cache.get<Record<string, unknown>>(`nasdaq:nasdaqShortinterest:${symbol}`),
|
ctx.cache.get<Record<string, unknown>>(`nasdaq:nasdaqShortinterest:${symbol}`),
|
||||||
ctx.cache.get<Record<string, unknown>>(`finra-bulk:finraShortinterest:${symbol}`),
|
ctx.cache.get<Record<string, unknown>>(`finra-bulk:finraShortinterest:${symbol}`),
|
||||||
]);
|
]);
|
||||||
const yf = yfEntry.value;
|
let yf = yfEntry.value;
|
||||||
const ndq = ndqEntry.value;
|
let ndq = ndqEntry.value;
|
||||||
const fr = frEntry.value;
|
const fr = frEntry.value;
|
||||||
|
|
||||||
|
// Direct fetch when cache is empty — the drain queue can be clogged with
|
||||||
|
// hundreds of pending jobs, making stale-while-revalidate too slow.
|
||||||
|
if (!yf && !ndq) {
|
||||||
|
const { YFinanceAdapter } = await import('../adapters/YFinanceAdapter.ts');
|
||||||
|
const { NasdaqAdapter } = await import('../adapters/NasdaqAdapter.ts');
|
||||||
|
const results = await Promise.allSettled([
|
||||||
|
(async () => {
|
||||||
|
const ya = new YFinanceAdapter();
|
||||||
|
const r = await ya.fetchOne(`yfinance:shortinterest:${symbol}`);
|
||||||
|
await ctx.cache.set(`yfinance:shortinterest:${symbol}`, r.value, r.ttlClass, r.provenance);
|
||||||
|
return r.value as Record<string, unknown>;
|
||||||
|
})(),
|
||||||
|
(async () => {
|
||||||
|
const na = new NasdaqAdapter();
|
||||||
|
const r = await na.fetchOne(`nasdaq:nasdaqShortinterest:${symbol}`);
|
||||||
|
await ctx.cache.set(`nasdaq:nasdaqShortinterest:${symbol}`, r.value, r.ttlClass, r.provenance);
|
||||||
|
return r.value as Record<string, unknown>;
|
||||||
|
})(),
|
||||||
|
]);
|
||||||
|
if (results[0].status === 'fulfilled') yf = results[0].value;
|
||||||
|
if (results[1].status === 'fulfilled') ndq = results[1].value;
|
||||||
|
}
|
||||||
|
|
||||||
// Collect sharesShort from all sources for discrepancy detection
|
// Collect sharesShort from all sources for discrepancy detection
|
||||||
const sharesSources: Array<{ source: string; sharesShort: number | null }> = [
|
const sharesSources: Array<{ source: string; sharesShort: number | null }> = [
|
||||||
{ source: 'finra', sharesShort: (fr?.shortVolume as number | undefined) ?? null },
|
{ source: 'finra', sharesShort: (fr?.shortVolume as number | undefined) ?? null },
|
||||||
@@ -1980,7 +2003,7 @@ const institutionalRouter = router({
|
|||||||
nasdaq: { available: !!ndq, stale: ndqEntry.isStale },
|
nasdaq: { available: !!ndq, stale: ndqEntry.isStale },
|
||||||
finra: { available: !!fr, stale: frEntry.isStale },
|
finra: { available: !!fr, stale: frEntry.isStale },
|
||||||
},
|
},
|
||||||
stale: yfEntry.isStale && ndqEntry.isStale && frEntry.isStale,
|
stale: yfEntry.isStale && ndqEntry.isStale, // FINRA is schedule-only, not per-symbol
|
||||||
};
|
};
|
||||||
}),
|
}),
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user