feat: add changes delay (#1739)

* 新增 changesDelay 配置，调用 /-/all/changes 接口时默认返回 delay 之前的 changes
* 防止出现 since 和当前时间接近时，changes 异步插入，导致 changes 计算失败的问题

.github/workflows/nodejs.yml

@@ -28,7 +28,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        node-version: [10, 12, 14, 16]
+        node-version: [14, 16]
         os: [ubuntu-latest]
 
     steps:

config/index.js

@@ -201,6 +201,11 @@ var config = {
   officialNpmReplicate: 'https://replicate.npmjs.com',
   cnpmRegistry: 'https://r.cnpmjs.com',
 
+  // /-/all/changes
+  // since different changes are aggregated through many tables
+  // prevent changesStream changes collisions
+  changesDelay: 5000,
+
   // sync source, upstream registry
   // If you want to directly sync from official npm's registry
   // please drop them an email first

services/package.js

@@ -203,16 +203,23 @@ exports.listPublicModuleNamesByUser = function* (username) {
 };
 
 exports.listModelSince = function(Model, attributes, mapper) {
+
   return function*(since, limit) {
+
     var start = ensureSinceIsDate(since);
     var findCondition = {
       attributes: attributes,
       where: {
         gmt_modified: {
-          gte: start
+          gte: start,
+          // 添加延时，防止同一时间多个数据未同步
+          lte: new Date(Date.now() - config.changesDelay || 5000),
         },
       },
-      order: [['gmt_modified', 'ASC'], ['id', 'ASC']],
+      order: [
+        ["gmt_modified", "ASC"],
+        ["id", "ASC"],
+      ],
     };
     if (limit) {
       findCondition.limit = limit;

test/controllers/registry/package/changes.test.js

@@ -3,6 +3,7 @@
 var should = require('should');
 var request = require('supertest');
 var mm = require('mm');
+var config = require('../../../../config');
 var app = require('../../../../servers/registry');
 var utils = require('../../../utils');
 var CHANGE_TYPE = require('../../../../services/common').CHANGE_TYPE;
@@ -12,6 +13,7 @@ describe('test/controllers/registry/package/changes.test.js', function () {
 
   var since;
   before(function (done) {
+    mm(config, 'changesDelay', 0);
     setTimeout(() => {
       since = Date.now();
       var pkg = utils.getPackage('@cnpmtest/test_changes', '0.0.1', utils.admin, 'alpha');
@@ -39,6 +41,7 @@ describe('test/controllers/registry/package/changes.test.js', function () {
         .expect(200, function (err, res) {
           should.not.exist(err);
           res.body.results.should.be.an.Array();
+
           res.body.results
             .filter(function (item) {
               return item.type === CHANGE_TYPE.PACKAGE_VERSION_ADDED;
@@ -53,6 +56,22 @@ describe('test/controllers/registry/package/changes.test.js', function () {
         });
     });
 
+    it('changes delay should work', function(done) {
+      mm(config, 'changesDelay', 10000);
+      request(app)
+        .get("/-/all/changes?since=" + since)
+        .expect(200, function (err, res) {
+          should.not.exist(err);
+          res.body.results.should.be.an.Array();
+          res.body.results
+            .filter(function (item) {
+              return item.type === CHANGE_TYPE.PACKAGE_VERSION_ADDED;
+            })
+            .length.should.equal(0);
+          done();
+        });
+    })
+
     it('since should work', function (done) {
       var now = Date.now();
       request(app)
@@ -66,6 +85,7 @@ describe('test/controllers/registry/package/changes.test.js', function () {
     });
 
     it('limit should work', function (done) {
+      mm(config, 'changesDelay', 0);
       request(app)
       .get('/-/all/changes?limit=1&since=' + since)
       .expect(200, function (err, res) {

test/services/package.test.js

@@ -5,6 +5,8 @@ var sleep = require('co-sleep');
 var Package = require('../../services/package');
 var utils = require('../utils');
 var common = require('../../services/common');
+var config = require('../../config');
+var mm = require('mm');
 
 describe('test/services/package.test.js', function () {
   describe('addModuleTag()', function () {
@@ -132,6 +134,7 @@ describe('test/services/package.test.js', function () {
 
   describe('listModelSince()', function () {
     it('list tags since', function* () {
+      mm(config, 'changesDelay', 0);
       yield utils.createModule('test-listModuleSince-module-0', '1.0.0');
       yield sleep(2100);
       var start = Date.now() - 1000;
@@ -162,6 +165,7 @@ describe('test/services/package.test.js', function () {
         ]);
     });
     it('list package version since', function* () {
+      mm(config, 'changesDelay', 0);
       yield utils.createModule('test-listModuleSince-module-0', '1.0.0');
       yield sleep(2100);
       var start = Date.now() - 1000;