joe/kami-parse-server
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle shutdown for RedisCacheAdapter (#6658)

Browse Source 
* Handle shutdown for RedisCacheAdapter

* connected value need to be tested in setTimeout

Co-authored-by: Promise Xu <promise@klido.me>
This commit is contained in:
promisenxu
2020-04-29 22:51:58 -04:00
committed by GitHub
parent 800b0392a3
commit 67bf868208
3 changed files with 74 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spec/RedisCacheAdapter.spec.js
View File

@@ -9,17 +9,17 @@ and make sure a redis server is available on the default port
*/ */
describe_only(() => { describe_only(() => {
return process.env.PARSE_SERVER_TEST_CACHE === 'redis'; return process.env.PARSE_SERVER_TEST_CACHE === 'redis';
})('RedisCacheAdapter', function() { })('RedisCacheAdapter', function () {
const KEY = 'hello'; const KEY = 'hello';
const VALUE = 'world'; const VALUE = 'world';
function wait(sleep) { function wait(sleep) {
return new Promise(function(resolve) { return new Promise(function (resolve) {
setTimeout(resolve, sleep); setTimeout(resolve, sleep);
}); });
} }
it('should get/set/clear', done => { it('should get/set/clear', (done) => {
const cache = new RedisCacheAdapter({ const cache = new RedisCacheAdapter({
ttl: NaN, ttl: NaN,
}); });
@@ -27,85 +27,94 @@ describe_only(() => {
cache cache
.put(KEY, VALUE) .put(KEY, VALUE)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(() => cache.clear()) .then(() => cache.clear())
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(null)) .then((value) => expect(value).toEqual(null))
.then(done); .then(done);
}); });
it('should expire after ttl', done => { it('should expire after ttl', (done) => {
const cache = new RedisCacheAdapter(null, 50); const cache = new RedisCacheAdapter(null, 50);
cache cache
.put(KEY, VALUE) .put(KEY, VALUE)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(wait.bind(null, 52)) .then(wait.bind(null, 52))
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(null)) .then((value) => expect(value).toEqual(null))
.then(done); .then(done);
}); });
it('should not store value for ttl=0', done => { it('should not store value for ttl=0', (done) => {
const cache = new RedisCacheAdapter(null, 5); const cache = new RedisCacheAdapter(null, 5);
cache cache
.put(KEY, VALUE, 0) .put(KEY, VALUE, 0)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(null)) .then((value) => expect(value).toEqual(null))
.then(done); .then(done);
}); });
it('should not expire when ttl=Infinity', done => { it('should not expire when ttl=Infinity', (done) => {
const cache = new RedisCacheAdapter(null, 1); const cache = new RedisCacheAdapter(null, 1);
cache cache
.put(KEY, VALUE, Infinity) .put(KEY, VALUE, Infinity)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(wait.bind(null, 5)) .then(wait.bind(null, 5))
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(done); .then(done);
}); });
it('should fallback to default ttl', done => { it('should fallback to default ttl', (done) => {
const cache = new RedisCacheAdapter(null, 1); const cache = new RedisCacheAdapter(null, 1);
let promise = Promise.resolve(); let promise = Promise.resolve();
[-100, null, undefined, 'not number', true].forEach(ttl => { [-100, null, undefined, 'not number', true].forEach((ttl) => {
promise = promise.then(() => promise = promise.then(() =>
cache cache
.put(KEY, VALUE, ttl) .put(KEY, VALUE, ttl)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(wait.bind(null, 5)) .then(wait.bind(null, 5))
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(null)) .then((value) => expect(value).toEqual(null))
); );
}); });
promise.then(done); promise.then(done);
}); });
it('should find un-expired records', done => { it('should find un-expired records', (done) => {
const cache = new RedisCacheAdapter(null, 5); const cache = new RedisCacheAdapter(null, 5);
cache cache
.put(KEY, VALUE) .put(KEY, VALUE)
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).toEqual(VALUE)) .then((value) => expect(value).toEqual(VALUE))
.then(wait.bind(null, 1)) .then(wait.bind(null, 1))
.then(() => cache.get(KEY)) .then(() => cache.get(KEY))
.then(value => expect(value).not.toEqual(null)) .then((value) => expect(value).not.toEqual(null))
.then(done); .then(done);
}); });
it('handleShutdown, close connection', async () => {
const cache = new RedisCacheAdapter(null, 5);
await cache.handleShutdown();
setTimeout(() => {
expect(cache.client.connected).toBe(false);
}, 0);
});
}); });
describe_only(() => { describe_only(() => {
return process.env.PARSE_SERVER_TEST_CACHE === 'redis'; return process.env.PARSE_SERVER_TEST_CACHE === 'redis';
})('RedisCacheAdapter/KeyPromiseQueue', function() { })('RedisCacheAdapter/KeyPromiseQueue', function () {
const KEY1 = 'key1'; const KEY1 = 'key1';
const KEY2 = 'key2'; const KEY2 = 'key2';
const VALUE = 'hello'; const VALUE = 'hello';
@@ -120,7 +129,7 @@ describe_only(() => {
return Object.keys(cache.queue.queue).length; return Object.keys(cache.queue.queue).length;
} }
it('it should clear completed operations from queue', done => { it('it should clear completed operations from queue', (done) => {
const cache = new RedisCacheAdapter({ ttl: NaN }); const cache = new RedisCacheAdapter({ ttl: NaN });
// execute a bunch of operations in sequence // execute a bunch of operations in sequence
@@ -142,7 +151,7 @@ describe_only(() => {
promise.then(() => expect(getQueueCount(cache)).toEqual(0)).then(done); promise.then(() => expect(getQueueCount(cache)).toEqual(0)).then(done);
}); });
it('it should count per key chained operations correctly', done => { it('it should count per key chained operations correctly', (done) => {
const cache = new RedisCacheAdapter({ ttl: NaN }); const cache = new RedisCacheAdapter({ ttl: NaN });
let key1Promise = Promise.resolve(); let key1Promise = Promise.resolve();
@@ -168,7 +177,7 @@ describe_only(() => {
describe_only(() => { describe_only(() => {
return process.env.PARSE_SERVER_TEST_CACHE === 'redis'; return process.env.PARSE_SERVER_TEST_CACHE === 'redis';
})('Redis Performance', function() { })('Redis Performance', function () {
let cacheAdapter; let cacheAdapter;
let getSpy; let getSpy;
let putSpy; let putSpy;

36
src/Adapters/Cache/RedisCacheAdapter/index.js
View File

@@ -9,7 +9,7 @@ function debug() {
logger.debug.apply(logger, ['RedisCacheAdapter', ...arguments]); logger.debug.apply(logger, ['RedisCacheAdapter', ...arguments]);
} }
const isValidTTL = ttl => typeof ttl === 'number' && ttl > 0; const isValidTTL = (ttl) => typeof ttl === 'number' && ttl > 0;
export class RedisCacheAdapter { export class RedisCacheAdapter {
constructor(redisCtx, ttl = DEFAULT_REDIS_TTL) { constructor(redisCtx, ttl = DEFAULT_REDIS_TTL) {
@@ -18,13 +18,27 @@ export class RedisCacheAdapter {
this.queue = new KeyPromiseQueue(); this.queue = new KeyPromiseQueue();
} }
handleShutdown() {
if (!this.client) {
return Promise.resolve();
}
return new Promise((resolve) => {
this.client.quit((err) => {
if (err) {
logger.error('RedisCacheAdapter error on shutdown', { error: err });
}
resolve();
});
});
}
get(key) { get(key) {
debug('get', key); debug('get', key);
return this.queue.enqueue( return this.queue.enqueue(
key, key,
() => () =>
new Promise(resolve => { new Promise((resolve) => {
this.client.get(key, function(err, res) { this.client.get(key, function (err, res) {
debug('-> get', key, res); debug('-> get', key, res);
if (!res) { if (!res) {
return resolve(null); return resolve(null);
@@ -48,8 +62,8 @@ export class RedisCacheAdapter {
return this.queue.enqueue( return this.queue.enqueue(
key, key,
() => () =>
new Promise(resolve => { new Promise((resolve) => {
this.client.set(key, value, function() { this.client.set(key, value, function () {
resolve(); resolve();
}); });
}) })
@@ -63,8 +77,8 @@ export class RedisCacheAdapter {
return this.queue.enqueue( return this.queue.enqueue(
key, key,
() => () =>
new Promise(resolve => { new Promise((resolve) => {
this.client.psetex(key, ttl, value, function() { this.client.psetex(key, ttl, value, function () {
resolve(); resolve();
}); });
}) })
@@ -76,8 +90,8 @@ export class RedisCacheAdapter {
return this.queue.enqueue( return this.queue.enqueue(
key, key,
() => () =>
new Promise(resolve => { new Promise((resolve) => {
this.client.del(key, function() { this.client.del(key, function () {
resolve(); resolve();
}); });
}) })
@@ -89,8 +103,8 @@ export class RedisCacheAdapter {
return this.queue.enqueue( return this.queue.enqueue(
FLUSH_DB_KEY, FLUSH_DB_KEY,
() => () =>
new Promise(resolve => { new Promise((resolve) => {
this.client.flushdb(function() { this.client.flushdb(function () {
resolve(); resolve();
}); });
}) })

28
src/ParseServer.js
View File

@@ -85,7 +85,7 @@ class ParseServer {
serverStartComplete(); serverStartComplete();
} }
}) })
.catch(error => { .catch((error) => {
if (serverStartComplete) { if (serverStartComplete) {
serverStartComplete(error); serverStartComplete(error);
} else { } else {
@@ -126,6 +126,10 @@ class ParseServer {
if (fileAdapter && typeof fileAdapter.handleShutdown === 'function') { if (fileAdapter && typeof fileAdapter.handleShutdown === 'function') {
promises.push(fileAdapter.handleShutdown()); promises.push(fileAdapter.handleShutdown());
} }
const { adapter: cacheAdapter } = this.config.cacheController;
if (cacheAdapter && typeof cacheAdapter.handleShutdown === 'function') {
promises.push(cacheAdapter.handleShutdown());
}
return (promises.length > 0 return (promises.length > 0
? Promise.all(promises) ? Promise.all(promises)
: Promise.resolve() : Promise.resolve()
@@ -154,7 +158,7 @@ class ParseServer {
}) })
); );
api.use('/health', function(req, res) { api.use('/health', function (req, res) {
res.json({ res.json({
status: 'ok', status: 'ok',
}); });
@@ -179,7 +183,7 @@ class ParseServer {
if (!process.env.TESTING) { if (!process.env.TESTING) {
//This causes tests to spew some useless warnings, so disable in test //This causes tests to spew some useless warnings, so disable in test
/* istanbul ignore next */ /* istanbul ignore next */
process.on('uncaughtException', err => { process.on('uncaughtException', (err) => {
if (err.code === 'EADDRINUSE') { if (err.code === 'EADDRINUSE') {
// user-friendly message for this common error // user-friendly message for this common error
process.stderr.write( process.stderr.write(
@@ -192,7 +196,7 @@ class ParseServer {
}); });
// verify the server url after a 'mount' event is received // verify the server url after a 'mount' event is received
/* istanbul ignore next */ /* istanbul ignore next */
api.on('mount', function() { api.on('mount', function () {
ParseServer.verifyServerUrl(); ParseServer.verifyServerUrl();
}); });
} }
@@ -334,8 +338,8 @@ class ParseServer {
if (Parse.serverURL) { if (Parse.serverURL) {
const request = require('./request'); const request = require('./request');
request({ url: Parse.serverURL.replace(/\/$/, '') + '/health' }) request({ url: Parse.serverURL.replace(/\/$/, '') + '/health' })
.catch(response => response) .catch((response) => response)
.then(response => { .then((response) => {
const json = response.data || null; const json = response.data || null;
if ( if (
response.status !== 200 || response.status !== 200 ||
@@ -368,7 +372,7 @@ function addParseCloud() {
} }
function injectDefaults(options: ParseServerOptions) { function injectDefaults(options: ParseServerOptions) {
Object.keys(defaults).forEach(key => { Object.keys(defaults).forEach((key) => {
if (!Object.prototype.hasOwnProperty.call(options, key)) { if (!Object.prototype.hasOwnProperty.call(options, key)) {
options[key] = defaults[key]; options[key] = defaults[key];
} }
@@ -424,12 +428,12 @@ function injectDefaults(options: ParseServerOptions) {
} }
// Merge protectedFields options with defaults. // Merge protectedFields options with defaults.
Object.keys(defaults.protectedFields).forEach(c => { Object.keys(defaults.protectedFields).forEach((c) => {
const cur = options.protectedFields[c]; const cur = options.protectedFields[c];
if (!cur) { if (!cur) {
options.protectedFields[c] = defaults.protectedFields[c]; options.protectedFields[c] = defaults.protectedFields[c];
} else { } else {
Object.keys(defaults.protectedFields[c]).forEach(r => { Object.keys(defaults.protectedFields[c]).forEach((r) => {
const unq = new Set([ const unq = new Set([
...(options.protectedFields[c][r] || []), ...(options.protectedFields[c][r] || []),
...defaults.protectedFields[c][r], ...defaults.protectedFields[c][r],
@@ -453,7 +457,7 @@ function configureListeners(parseServer) {
const sockets = {}; const sockets = {};
/* Currently, express doesn't shut down immediately after receiving SIGINT/SIGTERM if it has client connections that haven't timed out. (This is a known issue with node - https://github.com/nodejs/node/issues/2642) /* Currently, express doesn't shut down immediately after receiving SIGINT/SIGTERM if it has client connections that haven't timed out. (This is a known issue with node - https://github.com/nodejs/node/issues/2642)
This function, along with `destroyAliveConnections()`, intend to fix this behavior such that parse server will close all open connections and initiate the shutdown process as soon as it receives a SIGINT/SIGTERM signal. */ This function, along with `destroyAliveConnections()`, intend to fix this behavior such that parse server will close all open connections and initiate the shutdown process as soon as it receives a SIGINT/SIGTERM signal. */
server.on('connection', socket => { server.on('connection', (socket) => {
const socketId = socket.remoteAddress + ':' + socket.remotePort; const socketId = socket.remoteAddress + ':' + socket.remotePort;
sockets[socketId] = socket; sockets[socketId] = socket;
socket.on('close', () => { socket.on('close', () => {
@@ -461,7 +465,7 @@ function configureListeners(parseServer) {
}); });
}); });
const destroyAliveConnections = function() { const destroyAliveConnections = function () {
for (const socketId in sockets) { for (const socketId in sockets) {
try { try {
sockets[socketId].destroy(); sockets[socketId].destroy();
@@ -471,7 +475,7 @@ function configureListeners(parseServer) {
} }
}; };
const handleShutdown = function() { const handleShutdown = function () {
process.stdout.write('Termination signal received. Shutting down.'); process.stdout.write('Termination signal received. Shutting down.');
destroyAliveConnections(); destroyAliveConnections();
server.close(); server.close();