Added support for "Refresh Ahead" caching strategy (#19499)
The main changes are: - Updating the pipeline to allow for doing a background refresh of the cache - Remove the use of the EventAwareCacheWrapper for the posts public cache ### Background refresh This is just an initial implementation, and tbh it doesn't sit right with me that the logic for this is in the pipeline - I think this should sit in the cache implementation itself, and then we call out to it with something like: `cache.get(key, fetchData)` and then the updates can happen internally. The `cache-manager` project actually has a method like this called `wrap` - but every time I've used it it hangs, and debugging was a pain, so I don't really trust it. ### EventAwareCacheWrapper This is such a small amount of logic, I don't think it's worth creating an entire wrapper for it, at least not a class based one. I would be happy to refactor this to use a `Proxy` too, so that we don't have to add methods to it each time we wanna change the underlying cache implementation.
This commit is contained in:
Fabien 'egg' O'Carroll
GitHub
parent
59ebe1919e
commit
aaaa3ba797
@ -17,6 +17,7 @@ class AdapterCacheRedis extends BaseCacheAdapter {
|
||||
* @param {Object} [config.clusterConfig] - redis cluster config used in case no cache instance provided
|
||||
* @param {Object} [config.storeConfig] - extra redis client config used in case no cache instance provided
|
||||
* @param {Number} [config.ttl] - default cached value Time To Live (expiration) in *seconds*
|
||||
* @param {Number} [config.refreshAheadFactor] - 0-1 number to use to determine how old (as a percentage of ttl) an entry should be before refreshing it
|
||||
* @param {String} [config.keyPrefix] - prefix to use when building a unique cache key, e.g.: 'some_id:image-sizes:'
|
||||
* @param {Boolean} [config.reuseConnection] - specifies if the redis store/connection should be reused within the process
|
||||
*/
|
||||
@ -57,6 +58,9 @@ class AdapterCacheRedis extends BaseCacheAdapter {
|
||||
});
|
||||
}
|
||||
|
||||
this.ttl = config.ttl;
|
||||
this.refreshAheadFactor = config.refreshAheadFactor || 0;
|
||||
this.currentlyExecutingBackgroundRefreshes = new Set();
|
||||
this.keyPrefix = config.keyPrefix;
|
||||
this._keysPattern = config.keyPrefix ? `${config.keyPrefix}*` : '';
|
||||
this.redisClient = this.cache.store.getClient();
|
||||
@ -136,9 +140,66 @@ class AdapterCacheRedis extends BaseCacheAdapter {
|
||||
*
|
||||
* @param {String} key
|
||||
*/
|
||||
async get(key) {
|
||||
async shouldRefresh(key) {
|
||||
const internalKey = this._buildKey(key);
|
||||
if (this.refreshAheadFactor === 0) {
|
||||
debug(`shouldRefresh ${internalKey}: false - refreshAheadFactor = 0`);
|
||||
return false;
|
||||
}
|
||||
if (this.refreshAheadFactor === 1) {
|
||||
debug(`shouldRefresh ${internalKey}: true - refreshAheadFactor = 1`);
|
||||
return true;
|
||||
}
|
||||
try {
|
||||
return await this.cache.get(this._buildKey(key));
|
||||
const ttlRemainingForEntry = await this.cache.ttl(internalKey);
|
||||
const shouldRefresh = ttlRemainingForEntry < this.refreshAheadFactor * this.ttl;
|
||||
debug(`shouldRefresh ${internalKey}: ${shouldRefresh} - TTL remaining = ${ttlRemainingForEntry}`);
|
||||
return shouldRefresh;
|
||||
} catch (err) {
|
||||
logging.error(err);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param {string} key
|
||||
* @param {() => Promise<any>} [fetchData] An optional function to fetch the data, which will be used in the case of a cache MISS or a background refresh
|
||||
*/
|
||||
async get(key, fetchData) {
|
||||
const internalKey = this._buildKey(key);
|
||||
try {
|
||||
const result = await this.cache.get(internalKey);
|
||||
debug(`get ${internalKey}: Cache ${result ? 'HIT' : 'MISS'}`);
|
||||
if (!fetchData) {
|
||||
return result;
|
||||
}
|
||||
if (result) {
|
||||
const shouldRefresh = await this.shouldRefresh(internalKey);
|
||||
const isRefreshing = this.currentlyExecutingBackgroundRefreshes.has(internalKey);
|
||||
if (isRefreshing) {
|
||||
debug(`Background refresh already happening for ${internalKey}`);
|
||||
}
|
||||
if (!isRefreshing && shouldRefresh) {
|
||||
debug(`Doing background refresh for ${internalKey}`);
|
||||
this.currentlyExecutingBackgroundRefreshes.add(internalKey);
|
||||
fetchData().then(async (data) => {
|
||||
await this.set(key, data); // We don't use `internalKey` here because `set` handles it
|
||||
this.currentlyExecutingBackgroundRefreshes.delete(internalKey);
|
||||
}).catch((error) => {
|
||||
this.currentlyExecutingBackgroundRefreshes.delete(internalKey);
|
||||
logging.error({
|
||||
message: 'There was an error refreshing cache data in the background',
|
||||
error: error
|
||||
});
|
||||
});
|
||||
}
|
||||
return result;
|
||||
} else {
|
||||
const data = await fetchData();
|
||||
await this.set(key, data); // We don't use `internalKey` here because `set` handles it
|
||||
return data;
|
||||
}
|
||||
} catch (err) {
|
||||
logging.error(err);
|
||||
}
|
||||
@ -150,9 +211,10 @@ class AdapterCacheRedis extends BaseCacheAdapter {
|
||||
* @param {*} value
|
||||
*/
|
||||
async set(key, value) {
|
||||
debug('set', key);
|
||||
const internalKey = this._buildKey(key);
|
||||
debug('set', internalKey);
|
||||
try {
|
||||
return await this.cache.set(this._buildKey(key), value);
|
||||
return await this.cache.set(internalKey, value);
|
||||
} catch (err) {
|
||||
logging.error(err);
|
||||
}
|
||||
|
||||
@ -59,6 +59,127 @@ describe('Adapter Cache Redis', function () {
|
||||
|
||||
assert.equal(value, 'value from cache');
|
||||
});
|
||||
it('can update the cache in the case of a cache miss', async function () {
|
||||
const KEY = 'update-cache-on-miss';
|
||||
let cachedValue = null;
|
||||
const redisCacheInstanceStub = {
|
||||
get: function (key) {
|
||||
assert(key === KEY);
|
||||
return cachedValue;
|
||||
},
|
||||
set: function (key, value) {
|
||||
assert(key === KEY);
|
||||
cachedValue = value;
|
||||
},
|
||||
store: {
|
||||
getClient: sinon.stub().returns({
|
||||
on: sinon.stub()
|
||||
})
|
||||
}
|
||||
};
|
||||
const cache = new RedisCache({
|
||||
cache: redisCacheInstanceStub
|
||||
});
|
||||
|
||||
const fetchData = sinon.stub().resolves('Da Value');
|
||||
|
||||
checkFirstRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 1);
|
||||
assert.equal(value, 'Da Value');
|
||||
break checkFirstRead;
|
||||
}
|
||||
|
||||
checkSecondRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 1);
|
||||
assert.equal(value, 'Da Value');
|
||||
break checkSecondRead;
|
||||
}
|
||||
});
|
||||
|
||||
it('Can do a background update of the cache', async function () {
|
||||
const KEY = 'update-cache-in-background';
|
||||
let cachedValue = null;
|
||||
let remainingTTL = 100;
|
||||
|
||||
const redisCacheInstanceStub = {
|
||||
get: function (key) {
|
||||
assert(key === KEY);
|
||||
return cachedValue;
|
||||
},
|
||||
set: function (key, value) {
|
||||
assert(key === KEY);
|
||||
cachedValue = value;
|
||||
},
|
||||
ttl: function (key) {
|
||||
assert(key === KEY);
|
||||
return remainingTTL;
|
||||
},
|
||||
store: {
|
||||
getClient: sinon.stub().returns({
|
||||
on: sinon.stub()
|
||||
})
|
||||
}
|
||||
};
|
||||
const cache = new RedisCache({
|
||||
cache: redisCacheInstanceStub,
|
||||
ttl: 100,
|
||||
refreshAheadFactor: 0.2
|
||||
});
|
||||
|
||||
const fetchData = sinon.stub();
|
||||
fetchData.onFirstCall().resolves('First Value');
|
||||
fetchData.onSecondCall().resolves('Second Value');
|
||||
|
||||
checkFirstRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 1);
|
||||
assert.equal(value, 'First Value');
|
||||
break checkFirstRead;
|
||||
}
|
||||
|
||||
// We simulate having been in the cache for 15 seconds
|
||||
remainingTTL = 85;
|
||||
|
||||
checkSecondRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 1);
|
||||
assert.equal(value, 'First Value');
|
||||
break checkSecondRead;
|
||||
}
|
||||
|
||||
// We simulate having been in the cache for 30 seconds
|
||||
remainingTTL = 70;
|
||||
|
||||
checkThirdRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 1);
|
||||
assert.equal(value, 'First Value');
|
||||
break checkThirdRead;
|
||||
}
|
||||
|
||||
// We simulate having been in the cache for 85 seconds
|
||||
// This should trigger a background refresh
|
||||
remainingTTL = 15;
|
||||
|
||||
checkFourthRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 2);
|
||||
assert.equal(value, 'First Value');
|
||||
break checkFourthRead;
|
||||
}
|
||||
|
||||
// We reset the TTL to 100 for the most recent write
|
||||
remainingTTL = 100;
|
||||
|
||||
checkFifthRead: {
|
||||
const value = await cache.get(KEY, fetchData);
|
||||
assert.equal(fetchData.callCount, 2);
|
||||
assert.equal(value, 'Second Value');
|
||||
break checkFifthRead;
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('set', function () {
|
||||
|
||||
@ -238,35 +238,28 @@ const pipeline = (apiController, apiUtils, apiType) => {
|
||||
const cacheKey = stringify(cacheKeyData);
|
||||
|
||||
if (apiImpl.cache) {
|
||||
const response = await apiImpl.cache.get(cacheKey);
|
||||
|
||||
const response = await apiImpl.cache.get(cacheKey, getResponse);
|
||||
if (response) {
|
||||
return Promise.resolve(response);
|
||||
}
|
||||
}
|
||||
|
||||
return Promise.resolve()
|
||||
.then(() => {
|
||||
return STAGES.validation.input(apiUtils, apiConfig, apiImpl, frame);
|
||||
})
|
||||
.then(() => {
|
||||
return STAGES.serialisation.input(apiUtils, apiConfig, apiImpl, frame);
|
||||
})
|
||||
.then(() => {
|
||||
return STAGES.permissions(apiUtils, apiConfig, apiImpl, frame);
|
||||
})
|
||||
.then(() => {
|
||||
return STAGES.query(apiUtils, apiConfig, apiImpl, frame);
|
||||
})
|
||||
.then((response) => {
|
||||
return STAGES.serialisation.output(response, apiUtils, apiConfig, apiImpl, frame);
|
||||
})
|
||||
.then(async () => {
|
||||
if (apiImpl.cache) {
|
||||
await apiImpl.cache.set(cacheKey, frame.response);
|
||||
}
|
||||
return frame.response;
|
||||
});
|
||||
async function getResponse() {
|
||||
await STAGES.validation.input(apiUtils, apiConfig, apiImpl, frame);
|
||||
await STAGES.serialisation.input(apiUtils, apiConfig, apiImpl, frame);
|
||||
await STAGES.permissions(apiUtils, apiConfig, apiImpl, frame);
|
||||
const response = await STAGES.query(apiUtils, apiConfig, apiImpl, frame);
|
||||
await STAGES.serialisation.output(response, apiUtils, apiConfig, apiImpl, frame);
|
||||
return frame.response;
|
||||
}
|
||||
|
||||
const response = await getResponse();
|
||||
|
||||
if (apiImpl.cache) {
|
||||
await apiImpl.cache.set(cacheKey, response);
|
||||
}
|
||||
|
||||
return response;
|
||||
};
|
||||
|
||||
Object.assign(obj[method], apiImpl);
|
||||
|
||||
@ -8,34 +8,18 @@ class PostsPublicServiceWrapper {
|
||||
// Wire up all the dependencies
|
||||
const adapterManager = require('../adapter-manager');
|
||||
const config = require('../../../shared/config');
|
||||
const EventAwareCacheWrapper = require('@tryghost/event-aware-cache-wrapper');
|
||||
const EventRegistry = require('../../lib/common/events');
|
||||
|
||||
let postsCache;
|
||||
if (config.get('hostSettings:postsPublicCache:enabled')) {
|
||||
const cache = adapterManager.getAdapter('cache:postsPublic');
|
||||
postsCache = new EventAwareCacheWrapper({
|
||||
cache: cache,
|
||||
resetEvents: ['site.changed'],
|
||||
eventRegistry: EventRegistry
|
||||
postsCache = adapterManager.getAdapter('cache:postsPublic');
|
||||
EventRegistry.on('site.changed', () => {
|
||||
postsCache.reset();
|
||||
});
|
||||
}
|
||||
|
||||
let cache;
|
||||
if (postsCache) {
|
||||
// @NOTE: exposing cache through getter and setter to not loose the context of "this"
|
||||
cache = {
|
||||
get() {
|
||||
return postsCache.get(...arguments);
|
||||
},
|
||||
set() {
|
||||
return postsCache.set(...arguments);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
this.api = {
|
||||
cache: cache
|
||||
cache: postsCache
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@ -8,33 +8,18 @@ class TagsPublicServiceWrapper {
|
||||
// Wire up all the dependencies
|
||||
const adapterManager = require('../adapter-manager');
|
||||
const config = require('../../../shared/config');
|
||||
const EventAwareCacheWrapper = require('@tryghost/event-aware-cache-wrapper');
|
||||
const EventRegistry = require('../../lib/common/events');
|
||||
|
||||
let tagsCache;
|
||||
if (config.get('hostSettings:tagsPublicCache:enabled')) {
|
||||
let tagsPublicCache = adapterManager.getAdapter('cache:tagsPublic');
|
||||
tagsCache = new EventAwareCacheWrapper({
|
||||
cache: tagsPublicCache,
|
||||
resetEvents: ['site.changed'],
|
||||
eventRegistry: EventRegistry
|
||||
tagsCache = adapterManager.getAdapter('cache:tagsPublic');
|
||||
EventRegistry.on('site.changed', () => {
|
||||
tagsCache.reset();
|
||||
});
|
||||
}
|
||||
|
||||
let cache;
|
||||
if (tagsCache) {
|
||||
// @NOTE: exposing cache through getter and setter to not loose the context of "this"
|
||||
cache = {
|
||||
get() {
|
||||
return tagsCache.get(...arguments);
|
||||
},
|
||||
set() {
|
||||
return tagsCache.set(...arguments);
|
||||
}
|
||||
};
|
||||
}
|
||||
this.api = {
|
||||
cache: cache
|
||||
cache: tagsCache
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@ -87,7 +87,6 @@
|
||||
"@tryghost/email-service": "0.0.0",
|
||||
"@tryghost/email-suppression-list": "0.0.0",
|
||||
"@tryghost/errors": "1.2.26",
|
||||
"@tryghost/event-aware-cache-wrapper": "0.0.0",
|
||||
"@tryghost/express-dynamic-redirects": "0.0.0",
|
||||
"@tryghost/external-media-inliner": "0.0.0",
|
||||
"@tryghost/helpers": "1.1.88",
|
||||
|
||||
@ -1,6 +0,0 @@
|
||||
module.exports = {
|
||||
plugins: ['ghost'],
|
||||
extends: [
|
||||
'plugin:ghost/node'
|
||||
]
|
||||
};
|
||||
@ -1,23 +0,0 @@
|
||||
# Event Aware Cache Wrapper
|
||||
|
||||
Cache wrapper allowing to reset the cache after certain events
|
||||
|
||||
|
||||
## Usage
|
||||
|
||||
|
||||
## Develop
|
||||
|
||||
This is a monorepo package.
|
||||
|
||||
Follow the instructions for the top-level repo.
|
||||
1. `git clone` this repo & `cd` into it as usual
|
||||
2. Run `yarn` to install top-level dependencies.
|
||||
|
||||
|
||||
|
||||
## Test
|
||||
|
||||
- `yarn lint` run just eslint
|
||||
- `yarn test` run lint and tests
|
||||
|
||||
@ -1 +0,0 @@
|
||||
module.exports = require('./lib/EventAwareCacheWrapper');
|
||||
@ -1,38 +0,0 @@
|
||||
class EventAwareCacheWrapper {
|
||||
#cache;
|
||||
/**
|
||||
* @param {Object} deps
|
||||
* @param {Object} deps.cache - cache instance extending adapter-base-cache
|
||||
* @param {Object} [deps.eventRegistry] - event registry instance
|
||||
* @param {String[]} [deps.resetEvents] - event to listen to triggering reset
|
||||
*/
|
||||
constructor(deps) {
|
||||
this.#cache = deps.cache;
|
||||
|
||||
if (deps.resetEvents && deps.eventRegistry) {
|
||||
this.#initListeners(deps.eventRegistry, deps.resetEvents);
|
||||
}
|
||||
}
|
||||
|
||||
#initListeners(eventRegistry, eventsToResetOn) {
|
||||
eventsToResetOn.forEach((event) => {
|
||||
eventRegistry.on(event, () => {
|
||||
this.reset();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async get(key) {
|
||||
return this.#cache.get(key);
|
||||
}
|
||||
|
||||
async set(key, value) {
|
||||
return this.#cache.set(key, value);
|
||||
}
|
||||
|
||||
reset() {
|
||||
return this.#cache.reset();
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = EventAwareCacheWrapper;
|
||||
@ -1,27 +0,0 @@
|
||||
{
|
||||
"name": "@tryghost/event-aware-cache-wrapper",
|
||||
"version": "0.0.0",
|
||||
"repository": "https://github.com/TryGhost/Ghost/tree/main/packages/event-aware-cache-wrapper",
|
||||
"author": "Ghost Foundation",
|
||||
"private": true,
|
||||
"main": "index.js",
|
||||
"scripts": {
|
||||
"dev": "echo \"Implement me!\"",
|
||||
"test:unit": "NODE_ENV=testing c8 --all --check-coverage --100 --reporter text --reporter cobertura -- mocha --reporter dot './test/**/*.test.js'",
|
||||
"test": "yarn test:unit",
|
||||
"lint:code": "eslint *.js lib/ --ext .js --cache",
|
||||
"lint": "yarn lint:code && yarn lint:test",
|
||||
"lint:test": "eslint -c test/.eslintrc.js test/ --ext .js --cache"
|
||||
},
|
||||
"files": [
|
||||
"index.js",
|
||||
"lib"
|
||||
],
|
||||
"devDependencies": {
|
||||
"@tryghost/adapter-cache-memory-ttl": "0.0.0",
|
||||
"c8": "8.0.1",
|
||||
"mocha": "10.2.0",
|
||||
"sinon": "15.2.0"
|
||||
},
|
||||
"dependencies": {}
|
||||
}
|
||||
@ -1,6 +0,0 @@
|
||||
module.exports = {
|
||||
plugins: ['ghost'],
|
||||
extends: [
|
||||
'plugin:ghost/test'
|
||||
]
|
||||
};
|
||||
@ -1,48 +0,0 @@
|
||||
const assert = require('assert/strict');
|
||||
const InMemoryCache = require('@tryghost/adapter-cache-memory-ttl');
|
||||
|
||||
const EventAwareCacheWrapper = require('../index');
|
||||
const {EventEmitter} = require('stream');
|
||||
|
||||
describe('EventAwareCacheWrapper', function () {
|
||||
it('Can initialize', function () {
|
||||
const cache = new InMemoryCache();
|
||||
const wrappedCache = new EventAwareCacheWrapper({
|
||||
cache
|
||||
});
|
||||
assert.ok(wrappedCache);
|
||||
});
|
||||
|
||||
describe('get', function () {
|
||||
it('calls a wrapped cache with extra key', async function () {
|
||||
const cache = new InMemoryCache();
|
||||
const wrapper = new EventAwareCacheWrapper({
|
||||
cache: cache
|
||||
});
|
||||
|
||||
await wrapper.set('a', 'b');
|
||||
assert.equal(await wrapper.get('a'), 'b');
|
||||
assert.equal(await cache.get('a'), 'b');
|
||||
});
|
||||
});
|
||||
|
||||
describe('listens to reset events', function () {
|
||||
it('resets the cache when reset event is triggered', async function () {
|
||||
const cache = new InMemoryCache();
|
||||
|
||||
const eventRegistry = new EventEmitter();
|
||||
const wrapper = new EventAwareCacheWrapper({
|
||||
cache: cache,
|
||||
resetEvents: ['site.changed'],
|
||||
eventRegistry: eventRegistry
|
||||
});
|
||||
|
||||
await wrapper.set('a', 'b');
|
||||
assert.equal(await wrapper.get('a'), 'b');
|
||||
|
||||
eventRegistry.emit('site.changed');
|
||||
|
||||
assert.equal(await wrapper.get('a'), undefined);
|
||||
});
|
||||
});
|
||||
});
|
||||
Loading…
Reference in New Issue
Block a user