TTL-cache slow read-only CKAN actions (organization_list, group_list)

config/index.js

@@ -33,6 +33,11 @@ nconf.defaults({
   SESSION_SECRET: process.env.SESSION_SECRET || 'keyboard cat',
   // CKAN pages PLUGIN
   CKAN_PAGES_URL: process.env.CKAN_PAGES_URL || api_url,
+  // Seconds to cache the slow, slow-changing CKAN actions listed in
+  // API_CACHE_ACTIONS. 0 disables caching entirely.
+  API_CACHE_TTL: process.env.API_CACHE_TTL || 0,
+  API_CACHE_ACTIONS: process.env.API_CACHE_ACTIONS ||
+    'organization_list,group_list',
   // dashboard and maps PLUGIN
   GIT_BASE_URL: process.env.GIT_BASE_URL || 'https://raw.githubusercontent.com',
   // carto plugin

env.template

@@ -20,6 +20,12 @@ REDIS_URL=
 
 GA_ID=
 CKAN_PAGES_URL=
+
+# Seconds to cache slow, slow-changing CKAN actions (0 = off).
+# organization_list/group_list with all_fields are an N+1 in CKAN and can
+# take seconds; see datopian/tech-devops#668.
+API_CACHE_TTL=
+API_CACHE_ACTIONS=organization_list,group_list
 GIT_BASE_URL=
 GIT_OWNER=
 

lib/cache.js

@@ -0,0 +1,80 @@
+'use strict'
+
+const logger = require('../utils/logger')
+
+// In-process TTL cache for slow, read-only CKAN actions.
+//
+// Exists because organization_list/group_list with all_fields=true are an N+1 in
+// CKAN: it calls organization_show once per org, each running its own Solr query
+// for the dataset count. On oddk-prod that is 51 sequential queries and ~9.3s,
+// paid on every render of the home page and /organization. See
+// datopian/tech-devops#668 for the upstream fix that would make this unnecessary.
+//
+// Deliberately in-process, not Redis: the cache is per-pod and cold after a
+// restart, which is accepted - the point is that the Nth visitor does not pay
+// what the 1st does. Entries store the in-flight promise, so concurrent misses
+// share one upstream call instead of each starting its own.
+
+const MAX_ENTRIES = 200
+
+class TtlCache {
+  constructor() {
+    this.entries = new Map()
+  }
+
+  // Stable across param ordering, so equivalent calls share an entry.
+  static key(action, params) {
+    const parts = Object.keys(params || {})
+      .sort()
+      .map(name => `${name}=${params[name]}`)
+    return `${action}?${parts.join('&')}`
+  }
+
+  // `load` is only invoked on a miss. Its promise is what gets cached, so
+  // callers arriving mid-flight await the same upstream request.
+  async wrap(key, ttlSeconds, load) {
+    const now = Date.now()
+    const cached = this.entries.get(key)
+
+    if (cached && cached.expires > now) {
+      return cached.promise
+    }
+
+    const promise = load().catch(err => {
+      this.entries.delete(key)
+      // A refresh failure should not blank a page that has rendered before.
+      if (cached && 'value' in cached) {
+        logger.warn({
+          message: `cache: refresh failed for ${key}, serving stale copy`
+        })
+        return cached.value
+      }
+      throw err
+    })
+
+    if (this.entries.size >= MAX_ENTRIES && !this.entries.has(key)) {
+      this.entries.delete(this.entries.keys().next().value)
+    }
+
+    const entry = { expires: now + ttlSeconds * 1000, promise }
+    if (cached && 'value' in cached) {
+      entry.value = cached.value
+    }
+    this.entries.set(key, entry)
+
+    // Retain the resolved value so the stale fallback above has something to
+    // serve. Guarded so a superseded promise cannot overwrite a newer entry.
+    promise
+      .then(value => {
+        const current = this.entries.get(key)
+        if (current && current.promise === promise) {
+          current.value = value
+        }
+      })
+      .catch(() => {})
+
+    return promise
+  }
+}
+
+module.exports = TtlCache

lib/dms.js

@@ -5,14 +5,40 @@ const fetch = require('node-fetch')
 const utils = require('../utils')
 const querystring = require("querystring");
 const logger = require('../utils/logger')
+const TtlCache = require('./cache')
+
+const apiCache = new TtlCache()
 
 class DmsModel {
   constructor(config) {
     this.config = config
     this.api = config.get('INTERNAL_API_URL') || config.get('API_URL')
+    this.cacheTtl = Number(config.get('API_CACHE_TTL')) || 0
+    this.cachedActions = String(config.get('API_CACHE_ACTIONS') || '')
+      .split(',')
+      .map(action => action.trim())
+      .filter(Boolean)
+  }
+
+  // Only the allowlisted actions are cached, so anything user- or query-specific
+  // (package_search above all) still goes straight to CKAN.
+  shouldCache(action) {
+    return this.cacheTtl > 0 && this.cachedActions.includes(action)
   }
 
   async getJsonResponse(params, action) {
+    if (!this.shouldCache(action)) {
+      return this.fetchJsonResponse(params, action)
+    }
+
+    return apiCache.wrap(
+      TtlCache.key(action, params),
+      this.cacheTtl,
+      () => this.fetchJsonResponse(params, action)
+    )
+  }
+
+  async fetchJsonResponse(params, action) {
     let url
 
     if (params !== {}) {

tests/lib/cache.test.js

@@ -0,0 +1,98 @@
+const test = require('ava')
+const TtlCache = require('../../lib/cache')
+
+test('key is stable regardless of param ordering', t => {
+  t.is(
+    TtlCache.key('organization_list', { all_fields: true, sort: 'name' }),
+    TtlCache.key('organization_list', { sort: 'name', all_fields: true })
+  )
+})
+
+test('key separates different actions and params', t => {
+  t.not(
+    TtlCache.key('organization_list', { all_fields: true }),
+    TtlCache.key('group_list', { all_fields: true })
+  )
+  t.not(
+    TtlCache.key('organization_list', { all_fields: true }),
+    TtlCache.key('organization_list', { all_fields: false })
+  )
+})
+
+test('a hit within the TTL does not call the loader again', async t => {
+  const cache = new TtlCache()
+  let calls = 0
+  const load = async () => {
+    calls += 1
+    return calls
+  }
+
+  t.is(await cache.wrap('k', 60, load), 1)
+  t.is(await cache.wrap('k', 60, load), 1)
+  t.is(calls, 1)
+})
+
+test('an expired entry refetches', async t => {
+  const cache = new TtlCache()
+  let calls = 0
+  const load = async () => {
+    calls += 1
+    return calls
+  }
+
+  t.is(await cache.wrap('k', -1, load), 1)
+  t.is(await cache.wrap('k', -1, load), 2)
+  t.is(calls, 2)
+})
+
+test('concurrent misses share a single upstream call', async t => {
+  const cache = new TtlCache()
+  let calls = 0
+  const load = async () => {
+    calls += 1
+    await new Promise(resolve => setTimeout(resolve, 20))
+    return 'value'
+  }
+
+  const results = await Promise.all([
+    cache.wrap('k', 60, load),
+    cache.wrap('k', 60, load),
+    cache.wrap('k', 60, load)
+  ])
+
+  t.deepEqual(results, ['value', 'value', 'value'])
+  t.is(calls, 1, 'the thundering herd must collapse into one request')
+})
+
+test('a failed refresh serves the last good value', async t => {
+  const cache = new TtlCache()
+  t.is(await cache.wrap('k', -1, async () => 'good'), 'good')
+
+  const stale = await cache.wrap('k', -1, async () => {
+    throw new Error('CKAN is down')
+  })
+  t.is(stale, 'good')
+})
+
+test('a failure with nothing cached propagates', async t => {
+  const cache = new TtlCache()
+  await t.throwsAsync(
+    cache.wrap('k', 60, async () => {
+      throw new Error('CKAN is down')
+    }),
+    { message: 'CKAN is down' }
+  )
+})
+
+test('a failure does not pin the entry, so the next call retries', async t => {
+  const cache = new TtlCache()
+  let calls = 0
+  const load = async () => {
+    calls += 1
+    if (calls === 1) throw new Error('transient')
+    return 'recovered'
+  }
+
+  await t.throwsAsync(cache.wrap('k', 60, load))
+  t.is(await cache.wrap('k', 60, load), 'recovered')
+})