openvibers/openvibe.sources Commits

A source of jobs or listings retires what its feed stopped showing (retire_after_sec, contracts 0.211.0) (#90)

OP OpenVibers committed 21 days ago · 12b1bf4210 · 11 files, +220 −21

Changes

ddiff --git a/README.md b/README.md
iindex b3e5ffe..ef19f4b 100644
---- a/README.md
++++ b/README.md
@@@ -59,7 +59,7 @@ provider), `auth` (`none|header|bearer|query` with the **name** of an environmen
  names must start with `SOURCES_CRED_`, so an entry can never point at this service's own
  secrets), `robots_note`, `terms_note`, `license_note`, the rate limit (`min_interval_ms`),
  `poll_interval_sec`, `stale_after_sec`, `max_items`, `enabled`, `review_required`,
--`sensitivity`, `default_indexability` and `search_visibility`.
++`sensitivity`, `default_indexability`, `search_visibility` and `retire_after_sec` (since contracts 0.211.0).
  
  A source cannot be enabled without a terms note (and a robots note if it is fetched): an adapter
  existing is not permission to ingest (anti-goal 26). A source with items cannot be deleted —
@@@ -177,7 +177,13 @@ fragments removed), `title`, `summary`, `authors`, `published_at`, `source_updat
  `revision`, and `provenance` (`retrieved_at`, `first_seen_at`, `content_hash`, `raw_body_hash`,
  `parser_version`, `fetch_run_id`, `license_note`, `terms_note`, `entered_by`). Earlier revisions
  are kept in `item_revisions`. Removal (takedown, licence) is explicit, needs a reason, and is
--sticky: a later fetch that still lists the item does not bring it back. Contract (released in openvibe-contracts v0.12.0):
++sticky: a later fetch that still lists the item does not bring it back.
++
++A source of jobs or listings sets `retire_after_sec` (an hour to a year; null, the default, keeps every item, as news
++and datasets should). After a run in which every endpoint answered (`ok` or `not_modified`) and none failed, an item
++the feed has not shown for that long is removed with reason `gone_from_source`. That is a change like any removal, so a
++consumer following `GET /api/v1/items?include_removed=1` learns that the job is gone. A 304 counts as seeing again what
++the endpoint's last good parse saw. At most 500 items are retired a run; the rest go on the next. Contract (released in openvibe-contracts v0.12.0):
  [`sources.item@1`](docs/contracts-proposal/).
  
  ## API (service tokens, one capability per route)
@@@ -408,7 +414,7 @@ vhost serves it.
  Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
  
  <!-- versions:start -->
--- openvibe-contracts: v0.210.0
++- openvibe-contracts: v0.211.0
  - openvibe-sdk: v0.60.0
  - openvibe-shared: v4.8.0
  <!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 7fdb68a..8cd06d7 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,7 +7,7 @@
    "runtime": "node 22, express, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9, fast-xml-parser",
    "port": 4720,
    "domain": "sources.openvibe.network (vhost in deploy/nginx: an internal-service page, health and JSON 404s; the API is loopback-only; README \"Public host\")",
--  "contracts": "openvibe-contracts v0.210.0",
++  "contracts": "openvibe-contracts v0.211.0",
    "sdk": "openvibe-sdk v0.60.0",
    "shared": "openvibe-shared v4.8.0",
    "capabilities": [
@@@ -43,6 +43,7 @@
      "SSRF-guarded fetcher",
      "failed fetches recorded, never fabricated",
      "item revisions and sticky removals with provenance",
++    "retire_after_sec: items a feed stopped showing are removed as gone_from_source (jobs, listings)",
      "cursor item API by capability",
      "openvibe-sdk transactional outbox to Events (service_outbox)",
      "staff-only index documents for Search",
ddiff --git a/docs/contracts-proposal/contracts/sources/source.v1.json b/docs/contracts-proposal/contracts/sources/source.v1.json
iindex 5975e46..fd09659 100644
---- a/docs/contracts-proposal/contracts/sources/source.v1.json
++++ b/docs/contracts-proposal/contracts/sources/source.v1.json
@@@ -296,6 +296,12 @@
        "type": "integer",
        "minimum": 1
      },
++    "retire_after_sec": {
++      "type": ["integer", "null"],
++      "minimum": 3600,
++      "maximum": 31536000,
++      "description": "Retire what the feed stopped showing (jobs, listings): once a fetch of this source succeeds, an item it has not seen for this many seconds is removed with reason \"gone_from_source\" (a change, so consumers following the change feed learn it). null (the default): items stay, as a news archive or a dataset should."
++    },
      "max_items": {
        "type": "integer",
        "minimum": 1
ddiff --git a/migrations/0004_source_retire_after.sql b/migrations/0004_source_retire_after.sql
nnew file mode 100644
iindex 0000000..6de1b19
---- /dev/null
++++ b/migrations/0004_source_retire_after.sql
@@@ -0,0 +1,8 @@
++-- phase: expand
++-- A source may retire what its feed stopped showing (sources.source@1 retire_after_sec, openvibe-contracts 0.211.0; plan
++-- 2026-10-22, track F2): once a fetch of the source succeeds, an item not seen for this many seconds is removed with
++-- reason gone_from_source. NULL, every existing source's value, keeps every item as before.
++ALTER TABLE sources ADD COLUMN IF NOT EXISTS retire_after_sec bigint;
++
++-- The 304 path refreshes the items the endpoint's last good parse saw (server/ingest.js).
++CREATE INDEX IF NOT EXISTS idx_items_last_run ON items (source_key, last_fetch_run_id);
ddiff --git a/package-lock.json b/package-lock.json
iindex d926201..691d624 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -12,7 +12,7 @@
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
          "fast-xml-parser": "^4.5.3",
--        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.210.0",
++        "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.211.0",
          "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.60.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.8.0",
          "pg": "^8.23.0"
@@@ -769,9 +769,9 @@
        }
      },
      "node_modules/openvibe-contracts": {
--      "version": "0.210.0",
--      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.210.0",
--      "integrity": "sha512-9PnhWV0DxtqK7eogKfGPLf8A9qEyXjtlZNoAypYIluF6ynYXUJNHtsB/dFAwAwXq6dvqR/ilTJbe+HxgHb5CuQ==",
++      "version": "0.211.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.211.0",
++      "integrity": "sha512-M/DHcyQcxmwrT65s64D/mJn77ANF6Je5SeDgpWTZKKhxPffKVuhV+SzRZf8grZWvD3flmSqWNTTPOcKUWTjEug==",
        "license": "AGPL-3.0-only",
        "dependencies": {
          "ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 0a09162..eb2839f 100644
---- a/package.json
++++ b/package.json
@@@ -27,7 +27,7 @@
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
      "fast-xml-parser": "^4.5.3",
--    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.210.0",
++    "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.211.0",
      "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.60.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.8.0",
      "pg": "^8.23.0"
ddiff --git a/seeds/sources.json b/seeds/sources.json
iindex bcfb020..5bab631 100644
---- a/seeds/sources.json
++++ b/seeds/sources.json
@@@ -1768,6 +1768,7 @@
        "min_interval_ms": 60000,
        "poll_interval_sec": 21600,
        "max_items": 300,
++      "retire_after_sec": 604800,
        "enabled": false,
        "review_required": true,
        "sensitivity": "none",
ddiff --git a/server/ingest.js b/server/ingest.js
iindex 5c10338..e7943f0 100644
---- a/server/ingest.js
++++ b/server/ingest.js
@@@ -16,6 +16,10 @@
   * A failed run never creates or modifies an item: only the `ok` path calls items.ingest(), in the
   * same transaction as its run row. Validators (ETag/Last-Modified) are stored only after a good
   * parse, so a body that failed to parse is fetched and parsed again next time, not skipped as 304.
++ *
++ * A source with retire_after_sec (jobs, listings) retires what its feed stopped showing: after a run
++ * in which every endpoint answered (ok or 304) and none failed, its items not seen for that long are
++ * removed as gone_from_source. A 304 counts as seeing again what the endpoint's last good parse saw.
   */
  const { ids } = require('openvibe-contracts');
  const { ADAPTERS } = require('./adapters');
@@@ -47,6 +51,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
          touchEndpoint: db.prepare(`INSERT INTO endpoint_state (source_key, url, last_status, last_fetch_at) VALUES (@k, @u, @status, @at)
              ON CONFLICT(source_key, url) DO UPDATE SET last_status = excluded.last_status, last_fetch_at = excluded.last_fetch_at`),
          requested: db.prepare('UPDATE sources SET last_request_at = ? WHERE key = ?'),
++        lastOk: db.prepare(`SELECT id FROM fetch_runs WHERE source_key = ? AND endpoint_url = ? AND state = 'ok' ORDER BY rid DESC LIMIT 1`),
          finish: db.prepare(`UPDATE sources SET last_run_at = @at, last_state = @state, consecutive_failures = @failures,
              last_success_at = COALESCE(@success_at, last_success_at), next_due_at = @next_due, not_before = @not_before WHERE key = @key`),
      };
@@@ -166,6 +171,10 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
  
          if (res.status === 304) {
              Object.assign(run, { state: 'not_modified' });
++            if (source.retire_after_sec) {
++                const last = await st.lastOk.get(source.key, endpoint.url);
++                if (last) await items.seenAgain(source, last.id, run.finished_at);
++            }
              return { run, failed: false, success: true };
          }
          if (res.status === 429) {
@@@ -247,6 +256,7 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
  
              let anyFailure = false;
              let anySuccess = false;
++            let allAnswered = true;     // every endpoint ok or 304: what retirement needs to trust the feed
              let notBefore = 0;
              let failures = row.consecutive_failures;
              for (const endpoint of source.endpoints) {
@@@ -262,12 +272,22 @@ function createIngest({ db, config, registry, items, robots, fetcher, spacer, em
                  }
                  if (out.failed) { anyFailure = true; failures = row.consecutive_failures + 1; }
                  if (out.success) anySuccess = true;
++                else allAnswered = false;
                  if (out.notBefore) notBefore = Math.max(notBefore, out.notBefore);
                  if (!out.recorded) await recordNoItems(source, out.run, { countsAsFailure: out.failed, failures });
                  runs.push(out.run);
                  if (out.run.state === 'rate_limited') break;   // the site asked us to stop
              }
  
++            if (source.retire_after_sec && anySuccess && allAnswered && !anyFailure) {
++                try {
++                    const retired = await items.retireUnseen(source, now() - source.retire_after_sec * 1000);
++                    if (retired) log.log(`[ingest] ${key}: retired ${retired} item(s) the feed stopped showing`);
++                } catch (err) {
++                    log.error(`[ingest] ${key}: retiring unseen items failed: ${err.stack || err}`);
++                }
++            }
++
              const finishedAt = now();
              const newFailures = anyFailure ? row.consecutive_failures + 1 : 0;
              const backoff = anyFailure
ddiff --git a/server/items.js b/server/items.js
iindex 8fd9e91..758c056 100644
---- a/server/items.js
++++ b/server/items.js
@@@ -8,7 +8,8 @@
   * version and the source's license/terms notes as they were at ingestion. A changed content hash
   * is a new revision (the previous one stays in item_revisions); an unchanged item only has its
   * retrieval time refreshed. Removal (a takedown, a licence problem) is explicit, audited and
-- * sticky: a later fetch that still lists the item does not bring it back.
++ * sticky: a later fetch that still lists the item does not bring it back. A source with
++ * retire_after_sec also removes, as gone_from_source, what its feed has not shown for that long.
   *
   * Only successful fetches reach ingest(); nothing in this module runs for a failed fetch.
   */
@@@ -45,6 +46,9 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
          revision: db.prepare(`INSERT INTO item_revisions (item_id, revision, content_hash, raw_body_hash, parser_version, fetch_run_id, retrieved_at, snapshot)
              VALUES (?, ?, ?, ?, ?, ?, ?, ?)`),
          revisions: db.prepare('SELECT * FROM item_revisions WHERE item_id = ? ORDER BY revision'),
++        unseen: db.prepare(`SELECT * FROM items WHERE source_key = ? AND removed_at IS NULL AND retrieved_at < ?
++            ORDER BY retrieved_at, id LIMIT ?`),
++        seenAgain: db.prepare('UPDATE items SET retrieved_at = ? WHERE source_key = ? AND last_fetch_run_id = ? AND removed_at IS NULL'),
      };
  
      function view(row) {
@@@ -192,22 +196,51 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
          return counts;
      }
  
++    // One removal: the row, its revision, sources.item.removed and the Search delete. Inside a transaction.
++    async function markRemoved(source, cur, reason, stored, t) {
++        await st.remove.run(t, stored.slice(0, 500), await nextSeq(db, 'items'), cur.id);
++        const saved = await st.byId.get(cur.id);
++        await st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, null, t, JSON.stringify({ removed: true, reason }));
++        await itemEvent('sources.item.removed', saved, { reason });
++        await indexEvent(source, saved, true);
++        return saved;
++    }
++
      /** Remove an item (takedown, licence, error). Sticky; emits sources.item.removed. */
      async function remove(source, id, reason, by) {
          return await db.tx(async () => {
              const cur = await st.byId.get(id);
              if (!cur || cur.source_key !== source.key) return null;
              if (cur.removed_at) return view(cur);
++            return view(await markRemoved(source, cur, reason, `${reason}${by ? ` (by ${by})` : ''}`, now()));
++        });
++    }
++
++    /**
++     * Retire what the source stopped showing (sources.source@1 retire_after_sec): its live items not seen since `cutoff`
++     * are removed with reason gone_from_source, a change in the feed like any removal, so a consumer following it learns
++     * the job or listing is gone. Called after a fetch of the source succeeded; at most `limit` items a run, the rest
++     * on the next. Sticky like any removal. → the number retired
++     */
++    async function retireUnseen(source, cutoff, { limit = 500 } = {}) {
++        return await db.tx(async () => {
++            const rows = await st.unseen.all(source.key, cutoff, limit);
              const t = now();
--            await st.remove.run(t, `${reason}${by ? ` (by ${by})` : ''}`.slice(0, 500), await nextSeq(db, 'items'), id);
--            const saved = await st.byId.get(id);
--            await st.revision.run(saved.id, saved.revision, saved.content_hash, saved.raw_body_hash, saved.parser_version, null, t, JSON.stringify({ removed: true, reason }));
--            await itemEvent('sources.item.removed', saved, { reason });
--            await indexEvent(source, saved, true);
--            return view(saved);
++            for (const cur of rows) await markRemoved(source, cur, 'gone_from_source', 'gone_from_source', t);
++            return rows.length;
          });
      }
  
++    /**
++     * A 304 says the endpoint still lists what its last good parse (`runId`) saw: those items count as seen at `at`, so
++     * retirement does not take them for gone. Only for a source that retires; the others keep the parse's time.
++     */
++    async function seenAgain(source, runId, at) {
++        if (!runId) return 0;
++        const r = await st.seenAgain.run(at, source.key, runId);
++        return r && r.changes != null ? r.changes : 0;
++    }
++
      async function get(id, { revisions = false } = {}) {
          const row = await st.byId.get(id);
          if (!row) return null;
@@@ -233,7 +266,7 @@ function createItems({ db, emitEvent, now = () => Date.now() }) {
          return { items: page.map(view), next_after: page.length ? page[page.length - 1].change_seq : after, more };
      }
  
--    return { ingest, remove, get, list, view, indexDocument, contentHash };
++    return { ingest, remove, retireUnseen, seenAgain, get, list, view, indexDocument, contentHash };
  }
  
  module.exports = { createItems, contentHash, STAFF_GROUPS };
ddiff --git a/server/registry.js b/server/registry.js
iindex 882ac0d..598fc20 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -57,7 +57,7 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
      if (!input || typeof input !== 'object' || Array.isArray(input)) throw new RegistryError('body must be a source object');
      const known = new Set(['key', 'name', 'type', 'category', 'homepage_url', 'endpoints', 'auth', 'robots_note', 'terms_note', 'license_note',
          'min_interval_ms', 'poll_interval_sec', 'stale_after_sec', 'max_items', 'enabled', 'review_required', 'sensitivity',
--        'default_indexability', 'search_visibility']);
++        'default_indexability', 'search_visibility', 'retire_after_sec']);
      for (const k of Object.keys(input)) if (!known.has(k)) throw new RegistryError(`unknown field ${k}`);
      if (!KEY_RE.test(String(input.key || ''))) throw new RegistryError('key must match ^[a-z0-9][a-z0-9-]{1,63}$');
      const name = text(input.name, 200, 'name');
@@@ -134,6 +134,8 @@ function validateSource(input, { maxItemsCap = 12000 } = {}) {
          sensitivity: input.sensitivity == null ? 'none' : input.sensitivity,
          default_indexability: input.default_indexability == null ? 'noindex' : input.default_indexability,
          search_visibility: input.search_visibility == null ? null : input.search_visibility,
++        // Retire what the feed stopped showing (jobs, listings): an hour to a year, or null to keep every item (server/ingest.js).
++        retire_after_sec: intIn(input.retire_after_sec, 3600, 365 * 86400, null, 'retire_after_sec'),
      };
      if (!SENSITIVITY.includes(record.sensitivity)) throw new RegistryError(`sensitivity must be one of ${SENSITIVITY.join('|')}`);
      if (!['index', 'noindex'].includes(record.default_indexability)) throw new RegistryError('default_indexability must be index or noindex');
@@@ -152,15 +154,15 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
          all: db.prepare('SELECT * FROM sources ORDER BY key'),
          insert: db.prepare(`INSERT INTO sources (key, name, type, category, homepage_url, endpoints, auth, robots_note, terms_note, license_note,
              min_interval_ms, poll_interval_sec, stale_after_sec, max_items, enabled, review_required, sensitivity, default_indexability,
--            search_visibility, created_at, updated_at, updated_by, next_due_at)
++            search_visibility, retire_after_sec, created_at, updated_at, updated_by, next_due_at)
              VALUES (@key, @name, @type, @category, @homepage_url, @endpoints, @auth, @robots_note, @terms_note, @license_note,
              @min_interval_ms, @poll_interval_sec, @stale_after_sec, @max_items, @enabled, @review_required, @sensitivity, @default_indexability,
--            @search_visibility, @created_at, @updated_at, @updated_by, 0)`),
++            @search_visibility, @retire_after_sec, @created_at, @updated_at, @updated_by, 0)`),
          update: db.prepare(`UPDATE sources SET name = @name, type = @type, category = @category, homepage_url = @homepage_url, endpoints = @endpoints,
              auth = @auth, robots_note = @robots_note, terms_note = @terms_note, license_note = @license_note, min_interval_ms = @min_interval_ms,
              poll_interval_sec = @poll_interval_sec, stale_after_sec = @stale_after_sec, max_items = @max_items, enabled = @enabled,
              review_required = @review_required, sensitivity = @sensitivity, default_indexability = @default_indexability,
--            search_visibility = @search_visibility, updated_at = @updated_at, updated_by = @updated_by WHERE key = @key`),
++            search_visibility = @search_visibility, retire_after_sec = @retire_after_sec, updated_at = @updated_at, updated_by = @updated_by WHERE key = @key`),
          del: db.prepare('DELETE FROM sources WHERE key = ?'),
          itemCount: db.prepare('SELECT COUNT(*) AS n FROM items WHERE source_key = ?'),
      };
@@@ -187,6 +189,7 @@ function createRegistry({ db, now = () => Date.now(), maxItemsCap = 12000 }) {
              min_interval_ms: row.min_interval_ms, poll_interval_sec: row.poll_interval_sec, stale_after_sec: row.stale_after_sec,
              max_items: row.max_items, enabled: Boolean(row.enabled), review_required: Boolean(row.review_required),
              sensitivity: row.sensitivity, default_indexability: row.default_indexability, search_visibility: row.search_visibility,
++            retire_after_sec: row.retire_after_sec == null ? null : Number(row.retire_after_sec),
          };
      }
  
ddiff --git a/test/retire.test.js b/test/retire.test.js
nnew file mode 100644
iindex 0000000..2f7d6fa
---- /dev/null
++++ b/test/retire.test.js
@@@ -0,0 +1,121 @@
++'use strict';
++/**
++ * retire_after_sec (sources.source@1, openvibe-contracts 0.211.0; plan 2026-10-22, track F2): a source of jobs or
++ * listings retires what its feed stopped showing, so a consumer following the change feed learns the job is gone.
++ *   - the registry takes an hour to a year, or null, and keeps it across a patch;
++ *   - after a good run, an item not seen for that long is removed as gone_from_source (a change, an event);
++ *   - an item seen within the window stays; a 304 counts as seeing again what the last good parse saw;
++ *   - a failed run retires nothing, and a source without the setting keeps every item;
++ *   - like any removal it is sticky: the item showing up again does not bring it back.
++ */
++const assert = require('assert');
++const { boot, site, sourceDef, rss, suite } = require('./helpers');
++
++const t = suite('retire');
++const HOUR = 3600 * 1000;
++let clock = Date.parse('2026-10-23T00:00:00Z');
++let svc;
++let web;
++let listed;
++let status;
++let etag;
++
++const feed = () => rss(listed.map((g) => ({ guid: g, title: `Job ${g}`, link: `https://jobs.example.org/${g}` })));
++const run = async (key) => await svc.ingest.run(key, { trigger: 'manual' });
++const itemsOf = async (key) => await svc.db.prepare('SELECT * FROM items WHERE source_key = ? ORDER BY identity').all(key);
++const live = async (key) => (await itemsOf(key)).filter((i) => !i.removed_at).map((i) => i.identity);
++const removedEvents = async (key) => (await svc.db.prepare('SELECT envelope FROM service_outbox ORDER BY id').all())
++    .map((r) => r.envelope).filter((e) => e.event_type === 'sources.item.removed' && e.payload.source_key === key);
++const show = (guids, tag) => { listed = guids; etag = tag; status = 200; };
++
++t('boot', async () => {
++    svc = await boot({ now: () => clock });
++    web = await site({
++        '/robots.txt': () => ({ status: 404 }),
++        '/jobs.xml': (req) => {
++            if (status !== 200) return { status, body: 'down' };
++            if (etag && req.headers['if-none-match'] === etag) return { status: 304 };
++            return { headers: { 'Content-Type': 'application/rss+xml', ETag: etag }, body: feed() };
++        },
++    });
++});
++
++t('the registry takes an hour to a year or null, and a patch keeps it', async () => {
++    for (const bad of [3599, 365 * 86400 + 1, 1.5, '3600']) {
++        await assert.rejects(svc.registry.create(sourceDef({ endpoints: [`${web.origin}/jobs.xml`], retire_after_sec: bad }), 'test'), /retire_after_sec/);
++    }
++    const plain = await svc.registry.create(sourceDef({ endpoints: [`${web.origin}/jobs.xml`] }), 'test');
++    assert.strictEqual(svc.registry.view(plain).retire_after_sec, null);
++    const row = await svc.registry.create(sourceDef({ key: 'jobs-a', category: 'jobs', endpoints: [`${web.origin}/jobs.xml`], retire_after_sec: 3600 }), 'test');
++    assert.strictEqual(svc.registry.view(row).retire_after_sec, 3600);
++    const patched = await svc.registry.patch('jobs-a', { name: 'Jobs A' }, 'test');
++    assert.strictEqual(svc.registry.view(patched).retire_after_sec, 3600, 'a patch that does not name it keeps it');
++});
++
++t('an item the feed stopped showing is retired once unseen for the window, as gone_from_source', async () => {
++    show(['a', 'b', 'c'], '"1"');
++    assert.deepStrictEqual((await run('jobs-a')).runs.map((r) => r.state), ['ok']);
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'b', 'c']);
++
++    clock += 0.5 * HOUR;
++    show(['a', 'c'], '"2"');
++    await run('jobs-a');
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'b', 'c'], 'b was seen half an hour ago: it stays');
++
++    clock += 0.6 * HOUR;
++    show(['a', 'c'], '"3"');
++    await run('jobs-a');
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'c']);
++    const b = (await itemsOf('jobs-a')).find((i) => i.identity === 'b');
++    assert.strictEqual(b.removed_reason, 'gone_from_source');
++    const events = await removedEvents('jobs-a');
++    assert.deepStrictEqual(events.map((e) => [e.payload.item_id, e.payload.reason]), [[b.id, 'gone_from_source']]);
++
++    // The change feed carries it after everything before it, so a consumer following it removes the job.
++    const page = await svc.items.list({ source: 'jobs-a', includeRemoved: true });
++    const last = page.items[page.items.length - 1];
++    assert.deepStrictEqual([last.id, last.removed && last.removed.reason], [b.id, 'gone_from_source']);
++});
++
++t('a 304 counts as seeing again what the last good parse saw', async () => {
++    clock += 5 * HOUR;
++    const out = await run('jobs-a');
++    assert.deepStrictEqual(out.runs.map((r) => r.state), ['not_modified']);
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'c'], 'the feed did not change: nothing is gone');
++    for (const i of (await itemsOf('jobs-a')).filter((x) => !x.removed_at)) assert.strictEqual(Number(i.retrieved_at), clock);
++});
++
++t('a failed run retires nothing', async () => {
++    clock += 5 * HOUR;
++    status = 503;
++    assert.deepStrictEqual((await run('jobs-a')).runs.map((r) => r.state), ['http_error']);
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'c']);
++});
++
++t('retirement is sticky: the item showing up again stays removed', async () => {
++    await svc.db.prepare('UPDATE sources SET last_request_at = NULL, not_before = 0 WHERE key = ?').run('jobs-a');
++    show(['a', 'b', 'c'], '"4"');
++    const [r] = (await run('jobs-a')).runs;
++    assert.strictEqual(r.state, 'ok');
++    assert.deepStrictEqual(await live('jobs-a'), ['a', 'c']);
++    assert.strictEqual(r.items_skipped, 1);
++});
++
++t('a source without the setting keeps every item, however long ago it was seen', async () => {
++    const row = await svc.registry.create(sourceDef({ key: 'news-keep', endpoints: [`${web.origin}/jobs.xml`] }), 'test');
++    assert.strictEqual(svc.registry.view(row).retire_after_sec, null);
++    show(['x', 'y'], '"5"');
++    await run('news-keep');
++    clock += 400 * 24 * HOUR;
++    show(['x'], '"6"');
++    await run('news-keep');
++    assert.deepStrictEqual(await live('news-keep'), ['x', 'y']);
++    assert.strictEqual((await removedEvents('news-keep')).length, 0);
++});
++
++t('close', async () => {
++    await web.close();
++    await svc.stop();
++});
++
++t.run();