openvibers/openvibe.sources Commits
NYC HPD open housing-code violations per building (weekly aggregate, the 4,000 buildings with the most open class C) for Reviews' landlord pages, and a per-endpoint timeout_ms (NYC computes that aggregate in 12 to 60 s) (#76)
Changes
@@@ -150,6 +150,9 @@ A value is literal text or `$rec.<path>`, read from the record itself. No match`max_bytes` (an integer from 1 KiB to 32 MiB) sets one endpoint's body cap instead of `SOURCES_MAX_BYTES`. OVHcloud'scatalog is about 12 MB.++`timeout_ms` (an integer from 1 s to 120 s) sets one endpoint's deadline instead of `SOURCES_FETCH_TIMEOUT_MS`. NYC Open++Data computes the HPD per-building violation tally (`nyc-hpd-open-violations`) in 12 to 60 seconds.++`request_content_type` (`application/json` only) is sent as the Content-Type of the endpoint's GET, which has no body.Open Collective's public GraphQL API answers a GET only with it (its CSRF guard): `opencollective-accounts` (category`crowdfunding`) reads the 100 most active and the 100 newest collectives, projects and funds every six hours, each a@@@ -146,6 +146,12 @@"maximum": 33554432,"description": "The body cap for this endpoint, in bytes (1 KiB to 32 MiB); the deployment's cap otherwise. OVHcloud's VPS catalog is about 12 MB (Sources#49)."},++ "timeout_ms": {++ "type": "integer",++ "minimum": 1000,++ "maximum": 120000,++ "description": "This endpoint's fetch deadline, in milliseconds (1 s to 120 s); the deployment's otherwise. NYC Open Data computes the HPD per-building violation tally in 12 to 60 seconds."++ },"fields": {"type": "object","additionalProperties": {@@@ -1393,6 +1393,33 @@"review_required": true,"sensitivity": "none","default_indexability": "noindex"++ },++ {++ "key": "nyc-hpd-open-violations",++ "name": "NYC HPD: open housing-code violations per building (NYC Open Data)",++ "type": "api",++ "category": "reviews",++ "homepage_url": "https://data.cityofnewyork.us/Housing-Development/Housing-Maintenance-Code-Violations/wvxf-dwi5",++ "endpoints": [{++ "url": "https://data.cityofnewyork.us/resource/wvxf-dwi5.json?$select=buildingid,max(boro)%20AS%20boro,max(housenumber)%20AS%20housenumber,max(streetname)%20AS%20streetname,max(zip)%20AS%20zip,count(*)%20AS%20open,sum(case(class=%27C%27,1,true,0))%20AS%20c,sum(case(class=%27B%27,1,true,0))%20AS%20b,sum(case(class=%27A%27,1,true,0))%20AS%20a,max(inspectiondate)%20AS%20last&$where=violationstatus=%27Open%27&$group=buildingid&$having=sum(case(class=%27C%27,1,true,0))%3E=10&$order=c%20DESC,buildingid&$limit=4000",++ "format": "json",++ "items_path": "",++ "item_kind": "housing_violation_tally",++ "timeout_ms": 120000,++ "fields": { "identity": "buildingid", "title": "streetname" },++ "extra": { "buildingid": "buildingid", "boro": "boro", "housenumber": "housenumber", "streetname": "streetname", "zip": "zip", "open": "open", "c": "c", "b": "b", "a": "a", "last": "last" }++ }],++ "robots_note": "robots.txt checked 2026-10-17: https://data.cityofnewyork.us/robots.txt does not disallow /resource/ (Crawl-delay 1; this is one request a week).",++ "terms_note": "NYC Open Data's SODA API, dataset wvxf-dwi5, published by the Department of Housing Preservation and Development for public use under the NYC Open Data Terms of Use. One aggregate query a week: for every building with 10 or more open class C (immediately hazardous) violations, how many violations HPD lists as open, by class (C immediately hazardous, B hazardous, A non-hazardous), and the latest inspection day. The 4,000 buildings with the most class C. Open means HPD has not recorded the violation as corrected; Reviews says so beside the numbers. NYC computes the aggregate in 12 to 60 seconds, hence the endpoint's own deadline.",++ "license_note": "NYC Open Data Terms of Use: free to use. Cite 'NYC Department of Housing Preservation and Development, via NYC Open Data'.",++ "min_interval_ms": 60000,++ "poll_interval_sec": 604800,++ "stale_after_sec": 1814400,++ "max_items": 4000,++ "enabled": false,++ "review_required": true,++ "sensitivity": "none",++ "default_indexability": "noindex"}]}@@@ -48,6 +48,9 @@* `max_bytes` (an integer from 1 KiB to 32 MiB) raises or lowers the body cap for one endpoint* (SOURCES_MAX_BYTES otherwise); OVHcloud's VPS catalog is about 12 MB.*++ * `timeout_ms` (an integer from 1 s to 120 s) gives one endpoint its own deadline (SOURCES_FETCH_TIMEOUT_MS otherwise):++ * NYC Open Data computes the HPD per-building tally in 12 to 60 seconds.++ ** `request_content_type` ("application/json" only) is sent as the GET's Content-Type (the request has no body): Open* Collective's public GraphQL API answers a GET only with it, as its CSRF guard.*@@@ -231,6 +234,7 @@ function checkMapping(ep) {}if (ep.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(ep.item_kind)) return 'item_kind is malformed';if (ep.max_bytes !== undefined && (!Number.isInteger(ep.max_bytes) || ep.max_bytes < 1024 || ep.max_bytes > 32 * 1024 * 1024)) return 'max_bytes is an integer from 1024 to 33554432';++ if (ep.timeout_ms !== undefined && (!Number.isInteger(ep.timeout_ms) || ep.timeout_ms < 1000 || ep.timeout_ms > 120000)) return 'timeout_ms is an integer from 1000 to 120000';if (ep.request_content_type !== undefined && ep.request_content_type !== 'application/json') return 'request_content_type may only be application/json';return null;}@@@ -141,6 +141,8 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, emsensitiveHeaders: sensitive,// an api endpoint may set its own body cap (adapters/api.js `max_bytes`)capBytes: endpoint.max_bytes || undefined,++ // and its own deadline (`timeout_ms`)++ timeoutMs: endpoint.timeout_ms || undefined,// every redirect hop is checked against its own origin's robots.txt and, on a new// host, spaced like any other request to that hostbeforeHop: async (next) => {@@@ -19,7 +19,7 @@ class FetchError extends Error {}function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 1024 * 1024, maxRedirects = 5 }) {-- function once(url, headers, deadline, capBytes) {++ function once(url, headers, deadline, capBytes, limitMs = timeoutMs) {return new Promise((resolve, reject) => {const mod = url.protocol === 'https:' ? https : http;const remaining = deadline - Date.now();@@@ -65,11 +65,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102});const timer = setTimeout(() => {req.destroy();-- done(reject, new FetchError('timeout', `no complete response within ${timeoutMs} ms`));++ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));}, remaining);req.on('timeout', () => {req.destroy();-- done(reject, new FetchError('timeout', `no complete response within ${timeoutMs} ms`));++ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));});req.on('error', (err) => {if (err instanceof GuardError) return done(reject, new FetchError(err.code, err.message));@@@ -79,8 +79,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102});}-- async function fetchUrl(input, { headers = {}, etag = null, lastModified = null, beforeHop = null, capBytes = maxBytes, sensitiveHeaders = [] } = {}) {-- const deadline = Date.now() + timeoutMs;++ async function fetchUrl(input, { headers = {}, etag = null, lastModified = null, beforeHop = null, capBytes = maxBytes, sensitiveHeaders = [], timeoutMs: ownTimeout = null } = {}) {++ // An api endpoint may set its own deadline (adapters/api.js `timeout_ms`): a city's open-data aggregate can++ // take a minute to compute.++ const limitMs = ownTimeout || timeoutMs;++ const deadline = Date.now() + limitMs;let url;try { url = guard.checkUrl(input); } catch (err) { throw new FetchError(err.code || 'bad_url', err.message); }const h = {@@@ -92,7 +95,7 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102if (etag) h['If-None-Match'] = etag;if (lastModified) h['If-Modified-Since'] = lastModified;for (let hop = 0; ; hop++) {-- const r = await once(url, h, deadline, capBytes);++ const r = await once(url, h, deadline, capBytes, limitMs);if (!r.redirect) return { kind: 'response', status: r.status, headers: r.headers, body: r.body, bytes: r.bytes, finalUrl: url.toString() };if (hop >= maxRedirects) throw new FetchError('too_many_redirects', `more than ${maxRedirects} redirects`);let next;@@@ -78,6 +78,7 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {// `where` is only copied when given, so records without conditions stay byte-identical.if (e.where) out.where = e.where;if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;++ if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;if (e.request_content_type !== undefined) out.request_content_type = e.request_content_type;} else {// A feed may read structured facts out of its items' prose (adapters/extract.js).@@@ -187,6 +187,10 @@ t('the registry stores max_bytes only when an endpoint gives it', () => {const base = { key: 'catalog', name: 'A catalog', type: 'api', category: 'hosting', endpoints: [endpoint] };assert.ok(!('max_bytes' in validateSource(base).endpoints[0]));assert.strictEqual(validateSource({ ...base, endpoints: [{ ...endpoint, max_bytes: 25165824 }] }).endpoints[0].max_bytes, 25165824);++ // timeout_ms likewise, from 1 s to 120 s.++ assert.ok(!('timeout_ms' in validateSource(base).endpoints[0]));++ assert.strictEqual(validateSource({ ...base, endpoints: [{ ...endpoint, timeout_ms: 120000 }] }).endpoints[0].timeout_ms, 120000);++ for (const bad of [999, 120001, 1.5, '60000']) assert.throws(() => validateSource({ ...base, endpoints: [{ ...endpoint, timeout_ms: bad }] }), /timeout_ms is an integer from 1000 to 120000/, String(bad));});t('the registry stores `where` only when an endpoint gives it', () => {@@@ -160,6 +160,7 @@ t('timeouts and oversize bodies are explicit states', async () => {'/slow': () => ({ delayMs: 3000, body: FEED_V1 }),'/huge': () => ({ body: 'x'.repeat(4096) }),'/huge.json': () => ({ body: JSON.stringify(Array.from({ length: 60 }, (_, i) => ({ id: `p${i}`, pad: 'x'.repeat(50) }))) }),++ '/slow.json': () => ({ delayMs: 3000, body: JSON.stringify([{ id: 'b1' }, { id: 'b2' }]) }),});try {const key = await create(sourceDef({ endpoints: [`${s.origin}/slow`] }));@@@ -180,6 +181,14 @@ t('timeouts and oversize bodies are explicit states', async () => {assert.deepStrictEqual([capped.runs[0].state, capped.runs[0].error_code], ['http_error', 'too_large']);const lifted = await small.ingest.run('lifted', { trigger: 'manual' });assert.deepStrictEqual([lifted.runs[0].state, lifted.runs[0].items_seen], ['ok', 60]);++ // And its own timeout_ms gives a slow aggregate (NYC's HPD tally) longer than everyone else.++ const slow = (key, extra) => sourceDef({ key, type: 'api', category: 'reviews', endpoints: [{ url: `${s.origin}/slow.json`, items_path: '', fields: { identity: 'id' }, ...extra }] });++ await small.registry.create(slow('hurried', {}), 'test');++ await small.registry.create(slow('patient', { timeout_ms: 10000 }), 'test');++ const hurried = await small.ingest.run('hurried', { trigger: 'manual' });++ assert.deepStrictEqual([hurried.runs[0].state, hurried.runs[0].error_code], ['timeout', 'timeout']);++ const patient = await small.ingest.run('patient', { trigger: 'manual' });++ assert.deepStrictEqual([patient.runs[0].state, patient.runs[0].items_seen], ['ok', 2]);} finally { await small.stop(); }} finally { await s.close(); }});@@@ -8,6 +8,9 @@* nyc-restaurant-inspections NYC Health's inspection results on NYC Open Data, the newest 4,000 rows each poll:* one inspection_violation item per row, keyed by Socrata's row id, with the restaurant* (camis, name, borough, address, cuisine) and what the inspector recorded++ * nyc-hpd-open-violations NYC HPD's housing-code violations, one aggregate query a week: one++ * housing_violation_tally item per building with 10 or more open class C violations++ * (open by class, the address, the latest inspection), the 4,000 with the most** Both validate, start disabled, sit in the reviews category, and fit under the per-fetch cap.*/@@@ -21,6 +24,7 @@ const t = suite('official-records-sources');const seeds = require('../seeds/sources.json').sources;const cfpb = seeds.find((s) => s.key === 'cfpb-complaints-by-company');const nyc = seeds.find((s) => s.key === 'nyc-restaurant-inspections');++const hpd = seeds.find((s) => s.key === 'nyc-hpd-open-violations');const CFPB_BODY = JSON.stringify({took: 12, hits: { total: { value: 18486121, relation: 'eq' }, hits: [] },@@@ -48,8 +52,8 @@ const NYC_BODY = JSON.stringify([grade_date: '2026-10-14T00:00:00.000', inspection_type: 'Cycle Inspection / Re-inspection' },]);--t('both seeds validate, start disabled, are reviews sources and fit under the per-fetch cap', () => {-- for (const s of [cfpb, nyc]) {++t('the seeds validate, start disabled, are reviews sources and fit under the per-fetch cap', () => {++ for (const s of [cfpb, nyc, hpd]) {assert.ok(s, 'seeded');const rec = validateSource(s);assert.strictEqual(rec.enabled, false);@@@ -91,4 +95,25 @@ t('NYC: one inspection_violation item per row, keyed by its row id, with the resassert.strictEqual(u.searchParams.get('$order'), 'inspection_date DESC');});++const HPD_BODY = JSON.stringify([++ { buildingid: '288857', boro: 'BROOKLYN', housenumber: '760', streetname: 'ELDERT LANE', zip: '11208', open: '1794', c: '423', b: '1118', a: '252', last: '2026-03-24T00:00:00.000' },++ { buildingid: '60898', boro: 'BRONX', housenumber: '3067', streetname: 'DECATUR AVENUE', zip: '10467', open: '58', c: '27', b: '21', a: '10', last: '2026-10-12T00:00:00.000' },++]);++++t('HPD: one housing_violation_tally item per building, open counts by class, a minute to answer', () => {++ const rec = validateSource(hpd);++ assert.strictEqual(rec.endpoints[0].timeout_ms, 120000, 'NYC computes the aggregate in 12 to 60 s');++ assert.strictEqual(hpd.poll_interval_sec, 604800, 'one request a week');++ const r = api.parse(HPD_BODY, { url: hpd.endpoints[0].url, endpoint: hpd.endpoints[0], maxItems: hpd.max_items });++ assert.strictEqual(r.items.length, 2);++ const [eldert] = r.items;++ assert.strictEqual(eldert.kind, 'housing_violation_tally');++ assert.strictEqual(eldert.identity, '288857');++ assert.deepStrictEqual(eldert.fields, { buildingid: '288857', boro: 'BROOKLYN', housenumber: '760', streetname: 'ELDERT LANE', zip: '11208', open: '1794', c: '423', b: '1118', a: '252', last: '2026-03-24T00:00:00.000' });++ const u = new URL(hpd.endpoints[0].url);++ assert.strictEqual(u.searchParams.get('$where'), "violationstatus='Open'");++ assert.strictEqual(u.searchParams.get('$having'), "sum(case(class='C',1,true,0))>=10");++ assert.strictEqual(u.searchParams.get('$limit'), '4000');++});++t.run();
@@ -150,6 +150,9 @@ A value is literal text or `$rec.<path>`, read from the record itself. No match
@@ -150,6 +150,9 @@ A value is literal text or `$rec.<path>`, read from the record itself. No match
`max_bytes` (an integer from 1 KiB to 32 MiB) sets one endpoint's body cap instead of `SOURCES_MAX_BYTES`. OVHcloud's
`max_bytes` (an integer from 1 KiB to 32 MiB) sets one endpoint's body cap instead of `SOURCES_MAX_BYTES`. OVHcloud's
catalog is about 12 MB.
catalog is about 12 MB.
+`timeout_ms` (an integer from 1 s to 120 s) sets one endpoint's deadline instead of `SOURCES_FETCH_TIMEOUT_MS`. NYC Open
+Data computes the HPD per-building violation tally (`nyc-hpd-open-violations`) in 12 to 60 seconds.
+
`request_content_type` (`application/json` only) is sent as the Content-Type of the endpoint's GET, which has no body.
`request_content_type` (`application/json` only) is sent as the Content-Type of the endpoint's GET, which has no body.
Open Collective's public GraphQL API answers a GET only with it (its CSRF guard): `opencollective-accounts` (category
Open Collective's public GraphQL API answers a GET only with it (its CSRF guard): `opencollective-accounts` (category
`crowdfunding`) reads the 100 most active and the 100 newest collectives, projects and funds every six hours, each a
`crowdfunding`) reads the 100 most active and the 100 newest collectives, projects and funds every six hours, each a
@@ -146,6 +146,12 @@
@@ -146,6 +146,12 @@
"maximum": 33554432,
"maximum": 33554432,
"description": "The body cap for this endpoint, in bytes (1 KiB to 32 MiB); the deployment's cap otherwise. OVHcloud's VPS catalog is about 12 MB (Sources#49)."
"description": "The body cap for this endpoint, in bytes (1 KiB to 32 MiB); the deployment's cap otherwise. OVHcloud's VPS catalog is about 12 MB (Sources#49)."
},
},
+ "timeout_ms": {
+ "type": "integer",
+ "minimum": 1000,
+ "maximum": 120000,
+ "description": "This endpoint's fetch deadline, in milliseconds (1 s to 120 s); the deployment's otherwise. NYC Open Data computes the HPD per-building violation tally in 12 to 60 seconds."
+ },
"fields": {
"fields": {
"type": "object",
"type": "object",
"additionalProperties": {
"additionalProperties": {
@@ -1393,6 +1393,33 @@
@@ -1393,6 +1393,33 @@
"review_required": true,
"review_required": true,
"sensitivity": "none",
"sensitivity": "none",
"default_indexability": "noindex"
"default_indexability": "noindex"
+ },
+ {
+ "key": "nyc-hpd-open-violations",
+ "name": "NYC HPD: open housing-code violations per building (NYC Open Data)",
+ "type": "api",
+ "category": "reviews",
+ "homepage_url": "https://data.cityofnewyork.us/Housing-Development/Housing-Maintenance-Code-Violations/wvxf-dwi5",
+ "endpoints": [{
+ "url": "https://data.cityofnewyork.us/resource/wvxf-dwi5.json?$select=buildingid,max(boro)%20AS%20boro,max(housenumber)%20AS%20housenumber,max(streetname)%20AS%20streetname,max(zip)%20AS%20zip,count(*)%20AS%20open,sum(case(class=%27C%27,1,true,0))%20AS%20c,sum(case(class=%27B%27,1,true,0))%20AS%20b,sum(case(class=%27A%27,1,true,0))%20AS%20a,max(inspectiondate)%20AS%20last&$where=violationstatus=%27Open%27&$group=buildingid&$having=sum(case(class=%27C%27,1,true,0))%3E=10&$order=c%20DESC,buildingid&$limit=4000",
+ "format": "json",
+ "items_path": "",
+ "item_kind": "housing_violation_tally",
+ "timeout_ms": 120000,
+ "fields": { "identity": "buildingid", "title": "streetname" },
+ "extra": { "buildingid": "buildingid", "boro": "boro", "housenumber": "housenumber", "streetname": "streetname", "zip": "zip", "open": "open", "c": "c", "b": "b", "a": "a", "last": "last" }
+ }],
+ "robots_note": "robots.txt checked 2026-10-17: https://data.cityofnewyork.us/robots.txt does not disallow /resource/ (Crawl-delay 1; this is one request a week).",
+ "terms_note": "NYC Open Data's SODA API, dataset wvxf-dwi5, published by the Department of Housing Preservation and Development for public use under the NYC Open Data Terms of Use. One aggregate query a week: for every building with 10 or more open class C (immediately hazardous) violations, how many violations HPD lists as open, by class (C immediately hazardous, B hazardous, A non-hazardous), and the latest inspection day. The 4,000 buildings with the most class C. Open means HPD has not recorded the violation as corrected; Reviews says so beside the numbers. NYC computes the aggregate in 12 to 60 seconds, hence the endpoint's own deadline.",
+ "license_note": "NYC Open Data Terms of Use: free to use. Cite 'NYC Department of Housing Preservation and Development, via NYC Open Data'.",
+ "min_interval_ms": 60000,
+ "poll_interval_sec": 604800,
+ "stale_after_sec": 1814400,
+ "max_items": 4000,
+ "enabled": false,
+ "review_required": true,
+ "sensitivity": "none",
+ "default_indexability": "noindex"
}
}
]
]
}
}
@@ -48,6 +48,9 @@
@@ -48,6 +48,9 @@
* `max_bytes` (an integer from 1 KiB to 32 MiB) raises or lowers the body cap for one endpoint
* `max_bytes` (an integer from 1 KiB to 32 MiB) raises or lowers the body cap for one endpoint
* (SOURCES_MAX_BYTES otherwise); OVHcloud's VPS catalog is about 12 MB.
* (SOURCES_MAX_BYTES otherwise); OVHcloud's VPS catalog is about 12 MB.
*
*
+ * `timeout_ms` (an integer from 1 s to 120 s) gives one endpoint its own deadline (SOURCES_FETCH_TIMEOUT_MS otherwise):
+ * NYC Open Data computes the HPD per-building tally in 12 to 60 seconds.
+ *
* `request_content_type` ("application/json" only) is sent as the GET's Content-Type (the request has no body): Open
* `request_content_type` ("application/json" only) is sent as the GET's Content-Type (the request has no body): Open
* Collective's public GraphQL API answers a GET only with it, as its CSRF guard.
* Collective's public GraphQL API answers a GET only with it, as its CSRF guard.
*
*
@@ -231,6 +234,7 @@ function checkMapping(ep) {
@@ -231,6 +234,7 @@ function checkMapping(ep) {
}
}
if (ep.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(ep.item_kind)) return 'item_kind is malformed';
if (ep.item_kind !== undefined && !/^[a-z][a-z0-9_]{1,39}$/.test(ep.item_kind)) return 'item_kind is malformed';
if (ep.max_bytes !== undefined && (!Number.isInteger(ep.max_bytes) || ep.max_bytes < 1024 || ep.max_bytes > 32 * 1024 * 1024)) return 'max_bytes is an integer from 1024 to 33554432';
if (ep.max_bytes !== undefined && (!Number.isInteger(ep.max_bytes) || ep.max_bytes < 1024 || ep.max_bytes > 32 * 1024 * 1024)) return 'max_bytes is an integer from 1024 to 33554432';
+ if (ep.timeout_ms !== undefined && (!Number.isInteger(ep.timeout_ms) || ep.timeout_ms < 1000 || ep.timeout_ms > 120000)) return 'timeout_ms is an integer from 1000 to 120000';
if (ep.request_content_type !== undefined && ep.request_content_type !== 'application/json') return 'request_content_type may only be application/json';
if (ep.request_content_type !== undefined && ep.request_content_type !== 'application/json') return 'request_content_type may only be application/json';
return null;
return null;
}
}
@@ -141,6 +141,8 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
@@ -141,6 +141,8 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
sensitiveHeaders: sensitive,
sensitiveHeaders: sensitive,
// an api endpoint may set its own body cap (adapters/api.js `max_bytes`)
// an api endpoint may set its own body cap (adapters/api.js `max_bytes`)
capBytes: endpoint.max_bytes || undefined,
capBytes: endpoint.max_bytes || undefined,
+ // and its own deadline (`timeout_ms`)
+ timeoutMs: endpoint.timeout_ms || undefined,
// every redirect hop is checked against its own origin's robots.txt and, on a new
// every redirect hop is checked against its own origin's robots.txt and, on a new
// host, spaced like any other request to that host
// host, spaced like any other request to that host
beforeHop: async (next) => {
beforeHop: async (next) => {
@@ -19,7 +19,7 @@ class FetchError extends Error {
@@ -19,7 +19,7 @@ class FetchError extends Error {
}
}
function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 1024 * 1024, maxRedirects = 5 }) {
function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 1024 * 1024, maxRedirects = 5 }) {
- function once(url, headers, deadline, capBytes) {
+ function once(url, headers, deadline, capBytes, limitMs = timeoutMs) {
return new Promise((resolve, reject) => {
return new Promise((resolve, reject) => {
const mod = url.protocol === 'https:' ? https : http;
const mod = url.protocol === 'https:' ? https : http;
const remaining = deadline - Date.now();
const remaining = deadline - Date.now();
@@ -65,11 +65,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
@@ -65,11 +65,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
});
});
const timer = setTimeout(() => {
const timer = setTimeout(() => {
req.destroy();
req.destroy();
- done(reject, new FetchError('timeout', `no complete response within ${timeoutMs} ms`));
+ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));
}, remaining);
}, remaining);
req.on('timeout', () => {
req.on('timeout', () => {
req.destroy();
req.destroy();
- done(reject, new FetchError('timeout', `no complete response within ${timeoutMs} ms`));
+ done(reject, new FetchError('timeout', `no complete response within ${limitMs} ms`));
});
});
req.on('error', (err) => {
req.on('error', (err) => {
if (err instanceof GuardError) return done(reject, new FetchError(err.code, err.message));
if (err instanceof GuardError) return done(reject, new FetchError(err.code, err.message));
@@ -79,8 +79,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
@@ -79,8 +79,11 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
});
});
}
}
- async function fetchUrl(input, { headers = {}, etag = null, lastModified = null, beforeHop = null, capBytes = maxBytes, sensitiveHeaders = [] } = {}) {
- const deadline = Date.now() + timeoutMs;
+ async function fetchUrl(input, { headers = {}, etag = null, lastModified = null, beforeHop = null, capBytes = maxBytes, sensitiveHeaders = [], timeoutMs: ownTimeout = null } = {}) {
+ // An api endpoint may set its own deadline (adapters/api.js `timeout_ms`): a city's open-data aggregate can
+ // take a minute to compute.
+ const limitMs = ownTimeout || timeoutMs;
+ const deadline = Date.now() + limitMs;
let url;
let url;
try { url = guard.checkUrl(input); } catch (err) { throw new FetchError(err.code || 'bad_url', err.message); }
try { url = guard.checkUrl(input); } catch (err) { throw new FetchError(err.code || 'bad_url', err.message); }
const h = {
const h = {
@@ -92,7 +95,7 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
@@ -92,7 +95,7 @@ function createFetcher({ guard, userAgent, timeoutMs = 20000, maxBytes = 5 * 102
if (etag) h['If-None-Match'] = etag;
if (etag) h['If-None-Match'] = etag;
if (lastModified) h['If-Modified-Since'] = lastModified;
if (lastModified) h['If-Modified-Since'] = lastModified;
for (let hop = 0; ; hop++) {
for (let hop = 0; ; hop++) {
- const r = await once(url, h, deadline, capBytes);
+ const r = await once(url, h, deadline, capBytes, limitMs);
if (!r.redirect) return { kind: 'response', status: r.status, headers: r.headers, body: r.body, bytes: r.bytes, finalUrl: url.toString() };
if (!r.redirect) return { kind: 'response', status: r.status, headers: r.headers, body: r.body, bytes: r.bytes, finalUrl: url.toString() };
if (hop >= maxRedirects) throw new FetchError('too_many_redirects', `more than ${maxRedirects} redirects`);
if (hop >= maxRedirects) throw new FetchError('too_many_redirects', `more than ${maxRedirects} redirects`);
let next;
let next;
@@ -78,6 +78,7 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
@@ -78,6 +78,7 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
// `where` is only copied when given, so records without conditions stay byte-identical.
// `where` is only copied when given, so records without conditions stay byte-identical.
if (e.where) out.where = e.where;
if (e.where) out.where = e.where;
if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
if (e.max_bytes !== undefined) out.max_bytes = e.max_bytes;
+ if (e.timeout_ms !== undefined) out.timeout_ms = e.timeout_ms;
if (e.request_content_type !== undefined) out.request_content_type = e.request_content_type;
if (e.request_content_type !== undefined) out.request_content_type = e.request_content_type;
} else {
} else {
// A feed may read structured facts out of its items' prose (adapters/extract.js).
// A feed may read structured facts out of its items' prose (adapters/extract.js).
@@ -187,6 +187,10 @@ t('the registry stores max_bytes only when an endpoint gives it', () => {
@@ -187,6 +187,10 @@ t('the registry stores max_bytes only when an endpoint gives it', () => {
const base = { key: 'catalog', name: 'A catalog', type: 'api', category: 'hosting', endpoints: [endpoint] };
const base = { key: 'catalog', name: 'A catalog', type: 'api', category: 'hosting', endpoints: [endpoint] };
assert.ok(!('max_bytes' in validateSource(base).endpoints[0]));
assert.ok(!('max_bytes' in validateSource(base).endpoints[0]));
assert.strictEqual(validateSource({ ...base, endpoints: [{ ...endpoint, max_bytes: 25165824 }] }).endpoints[0].max_bytes, 25165824);
assert.strictEqual(validateSource({ ...base, endpoints: [{ ...endpoint, max_bytes: 25165824 }] }).endpoints[0].max_bytes, 25165824);
+ // timeout_ms likewise, from 1 s to 120 s.
+ assert.ok(!('timeout_ms' in validateSource(base).endpoints[0]));
+ assert.strictEqual(validateSource({ ...base, endpoints: [{ ...endpoint, timeout_ms: 120000 }] }).endpoints[0].timeout_ms, 120000);
+ for (const bad of [999, 120001, 1.5, '60000']) assert.throws(() => validateSource({ ...base, endpoints: [{ ...endpoint, timeout_ms: bad }] }), /timeout_ms is an integer from 1000 to 120000/, String(bad));
});
});
t('the registry stores `where` only when an endpoint gives it', () => {
t('the registry stores `where` only when an endpoint gives it', () => {
@@ -160,6 +160,7 @@ t('timeouts and oversize bodies are explicit states', async () => {
@@ -160,6 +160,7 @@ t('timeouts and oversize bodies are explicit states', async () => {
'/slow': () => ({ delayMs: 3000, body: FEED_V1 }),
'/slow': () => ({ delayMs: 3000, body: FEED_V1 }),
'/huge': () => ({ body: 'x'.repeat(4096) }),
'/huge': () => ({ body: 'x'.repeat(4096) }),
'/huge.json': () => ({ body: JSON.stringify(Array.from({ length: 60 }, (_, i) => ({ id: `p${i}`, pad: 'x'.repeat(50) }))) }),
'/huge.json': () => ({ body: JSON.stringify(Array.from({ length: 60 }, (_, i) => ({ id: `p${i}`, pad: 'x'.repeat(50) }))) }),
+ '/slow.json': () => ({ delayMs: 3000, body: JSON.stringify([{ id: 'b1' }, { id: 'b2' }]) }),
});
});
try {
try {
const key = await create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
const key = await create(sourceDef({ endpoints: [`${s.origin}/slow`] }));
@@ -180,6 +181,14 @@ t('timeouts and oversize bodies are explicit states', async () => {
@@ -180,6 +181,14 @@ t('timeouts and oversize bodies are explicit states', async () => {
assert.deepStrictEqual([capped.runs[0].state, capped.runs[0].error_code], ['http_error', 'too_large']);
assert.deepStrictEqual([capped.runs[0].state, capped.runs[0].error_code], ['http_error', 'too_large']);
const lifted = await small.ingest.run('lifted', { trigger: 'manual' });
const lifted = await small.ingest.run('lifted', { trigger: 'manual' });
assert.deepStrictEqual([lifted.runs[0].state, lifted.runs[0].items_seen], ['ok', 60]);
assert.deepStrictEqual([lifted.runs[0].state, lifted.runs[0].items_seen], ['ok', 60]);
+ // And its own timeout_ms gives a slow aggregate (NYC's HPD tally) longer than everyone else.
+ const slow = (key, extra) => sourceDef({ key, type: 'api', category: 'reviews', endpoints: [{ url: `${s.origin}/slow.json`, items_path: '', fields: { identity: 'id' }, ...extra }] });
+ await small.registry.create(slow('hurried', {}), 'test');
+ await small.registry.create(slow('patient', { timeout_ms: 10000 }), 'test');
+ const hurried = await small.ingest.run('hurried', { trigger: 'manual' });
+ assert.deepStrictEqual([hurried.runs[0].state, hurried.runs[0].error_code], ['timeout', 'timeout']);
+ const patient = await small.ingest.run('patient', { trigger: 'manual' });
+ assert.deepStrictEqual([patient.runs[0].state, patient.runs[0].items_seen], ['ok', 2]);
} finally { await small.stop(); }
} finally { await small.stop(); }
} finally { await s.close(); }
} finally { await s.close(); }
});
});
@@ -8,6 +8,9 @@
@@ -8,6 +8,9 @@
* nyc-restaurant-inspections NYC Health's inspection results on NYC Open Data, the newest 4,000 rows each poll:
* nyc-restaurant-inspections NYC Health's inspection results on NYC Open Data, the newest 4,000 rows each poll:
* one inspection_violation item per row, keyed by Socrata's row id, with the restaurant
* one inspection_violation item per row, keyed by Socrata's row id, with the restaurant
* (camis, name, borough, address, cuisine) and what the inspector recorded
* (camis, name, borough, address, cuisine) and what the inspector recorded
+ * nyc-hpd-open-violations NYC HPD's housing-code violations, one aggregate query a week: one
+ * housing_violation_tally item per building with 10 or more open class C violations
+ * (open by class, the address, the latest inspection), the 4,000 with the most
*
*
* Both validate, start disabled, sit in the reviews category, and fit under the per-fetch cap.
* Both validate, start disabled, sit in the reviews category, and fit under the per-fetch cap.
*/
*/
@@ -21,6 +24,7 @@ const t = suite('official-records-sources');
@@ -21,6 +24,7 @@ const t = suite('official-records-sources');
const seeds = require('../seeds/sources.json').sources;
const seeds = require('../seeds/sources.json').sources;
const cfpb = seeds.find((s) => s.key === 'cfpb-complaints-by-company');
const cfpb = seeds.find((s) => s.key === 'cfpb-complaints-by-company');
const nyc = seeds.find((s) => s.key === 'nyc-restaurant-inspections');
const nyc = seeds.find((s) => s.key === 'nyc-restaurant-inspections');
+const hpd = seeds.find((s) => s.key === 'nyc-hpd-open-violations');
const CFPB_BODY = JSON.stringify({
const CFPB_BODY = JSON.stringify({
took: 12, hits: { total: { value: 18486121, relation: 'eq' }, hits: [] },
took: 12, hits: { total: { value: 18486121, relation: 'eq' }, hits: [] },
@@ -48,8 +52,8 @@ const NYC_BODY = JSON.stringify([
@@ -48,8 +52,8 @@ const NYC_BODY = JSON.stringify([
grade_date: '2026-10-14T00:00:00.000', inspection_type: 'Cycle Inspection / Re-inspection' },
grade_date: '2026-10-14T00:00:00.000', inspection_type: 'Cycle Inspection / Re-inspection' },
]);
]);
-t('both seeds validate, start disabled, are reviews sources and fit under the per-fetch cap', () => {
- for (const s of [cfpb, nyc]) {
+t('the seeds validate, start disabled, are reviews sources and fit under the per-fetch cap', () => {
+ for (const s of [cfpb, nyc, hpd]) {
assert.ok(s, 'seeded');
assert.ok(s, 'seeded');
const rec = validateSource(s);
const rec = validateSource(s);
assert.strictEqual(rec.enabled, false);
assert.strictEqual(rec.enabled, false);
@@ -91,4 +95,25 @@ t('NYC: one inspection_violation item per row, keyed by its row id, with the res
@@ -91,4 +95,25 @@ t('NYC: one inspection_violation item per row, keyed by its row id, with the res
assert.strictEqual(u.searchParams.get('$order'), 'inspection_date DESC');
assert.strictEqual(u.searchParams.get('$order'), 'inspection_date DESC');
});
});
+const HPD_BODY = JSON.stringify([
+ { buildingid: '288857', boro: 'BROOKLYN', housenumber: '760', streetname: 'ELDERT LANE', zip: '11208', open: '1794', c: '423', b: '1118', a: '252', last: '2026-03-24T00:00:00.000' },
+ { buildingid: '60898', boro: 'BRONX', housenumber: '3067', streetname: 'DECATUR AVENUE', zip: '10467', open: '58', c: '27', b: '21', a: '10', last: '2026-10-12T00:00:00.000' },
+]);
+
+t('HPD: one housing_violation_tally item per building, open counts by class, a minute to answer', () => {
+ const rec = validateSource(hpd);
+ assert.strictEqual(rec.endpoints[0].timeout_ms, 120000, 'NYC computes the aggregate in 12 to 60 s');
+ assert.strictEqual(hpd.poll_interval_sec, 604800, 'one request a week');
+ const r = api.parse(HPD_BODY, { url: hpd.endpoints[0].url, endpoint: hpd.endpoints[0], maxItems: hpd.max_items });
+ assert.strictEqual(r.items.length, 2);
+ const [eldert] = r.items;
+ assert.strictEqual(eldert.kind, 'housing_violation_tally');
+ assert.strictEqual(eldert.identity, '288857');
+ assert.deepStrictEqual(eldert.fields, { buildingid: '288857', boro: 'BROOKLYN', housenumber: '760', streetname: 'ELDERT LANE', zip: '11208', open: '1794', c: '423', b: '1118', a: '252', last: '2026-03-24T00:00:00.000' });
+ const u = new URL(hpd.endpoints[0].url);
+ assert.strictEqual(u.searchParams.get('$where'), "violationstatus='Open'");
+ assert.strictEqual(u.searchParams.get('$having'), "sum(case(class='C',1,true,0))>=10");
+ assert.strictEqual(u.searchParams.get('$limit'), '4000');
+});
+
t.run();
t.run();