History rework

spec/engine/historySpec.js

@@ -0,0 +1,112 @@
+const q = require('q'),
+    utils = require('../../src/utils'),
+    driver = utils.getDriver(),
+    history = require('../../src/history');
+
+describe('history', () => {
+    const originalHistory = driver.history;
+    const originalChunkSize = driver.config.historyChunkSize;
+    const originalGetActiveRooms = driver.getActiveRooms;
+    const originalGetRoomObjects = driver.getRoomObjects;
+
+    beforeEach(() => {
+        driver.config.historyChunkSize = 20;
+        driver.history = {
+            saveTick: jasmine.createSpy('saveTick').and.callFake(() => q.when()),
+            upload: jasmine.createSpy('upload').and.callFake(() => q.when()),
+            markPendingHistory: jasmine.createSpy('markPendingHistory').and.callFake(() => q.when()),
+            takePendingHistory: jasmine.createSpy('takePendingHistory').and.callFake(() => q.when([]))
+        };
+        driver.getActiveRooms = jasmine.createSpy('getActiveRooms').and.callFake(() => q.when([]));
+        driver.getRoomObjects = jasmine.createSpy('getRoomObjects').and.callFake(() => q.when({objects: {}}));
+    });
+
+    afterEach(() => {
+        driver.history = originalHistory;
+        driver.config.historyChunkSize = originalChunkSize;
+        driver.getActiveRooms = originalGetActiveRooms;
+        driver.getRoomObjects = originalGetRoomObjects;
+    });
+
+    describe('buildHistoryPayload', () => {
+        it('skips flags and strips private say', () => {
+            const payload = history.buildHistoryPayload({
+                a: {_id: 'a', type: 'source'},
+                b: {_id: 'b', type: 'flag'},
+                c: {
+                    _id: 'c',
+                    type: 'creep',
+                    actionLog: {say: {message: 'hi', isPublic: false}}
+                }
+            });
+
+            expect(payload.a).toEqual({_id: 'a', type: 'source'});
+            expect(payload.b).toBeUndefined();
+            expect(payload.c._id).toBe('c');
+            expect(payload.c.actionLog.say).toBeUndefined();
+        });
+    });
+
+    describe('saveRoomHistory', () => {
+        it('does not upload mid-chunk', () => {
+            return history.saveRoomHistory('W1N1', {a: 1}, 10).then(() => {
+                expect(driver.history.upload).not.toHaveBeenCalled();
+                expect(driver.history.saveTick).toHaveBeenCalledWith('W1N1', 10, JSON.stringify({a: 1}));
+                expect(driver.history.markPendingHistory).toHaveBeenCalledWith('W1N1', 0);
+            });
+        });
+
+        it('uploads the previous chunk on a boundary then saveTicks the new chunk', () => {
+            return history.saveRoomHistory('W1N1', {a: 1}, 20).then(() => {
+                expect(driver.history.upload).toHaveBeenCalledWith('W1N1', 0);
+                expect(driver.history.saveTick).toHaveBeenCalledWith('W1N1', 20, JSON.stringify({a: 1}));
+                expect(driver.history.markPendingHistory).toHaveBeenCalledWith('W1N1', 20);
+                expect(driver.history.upload.calls.count()).toBe(1);
+            });
+        });
+    });
+
+    describe('uploadPendingChunks', () => {
+        it('does not flush while the chunk can still receive ticks', () => {
+            return history.uploadPendingChunks(20).then(() => {
+                expect(driver.history.takePendingHistory).not.toHaveBeenCalled();
+                expect(driver.history.upload).not.toHaveBeenCalled();
+            });
+        });
+
+        it('uploads leftover rooms after active rooms have closed the chunk', () => {
+            driver.history.takePendingHistory.and.callFake(() => q.when(['W1N1', 'W2N2', 'W3N3']));
+            return history.uploadPendingChunks(21, ['W1N1']).then(() => {
+                expect(driver.history.takePendingHistory).toHaveBeenCalledWith(0);
+                expect(driver.history.upload).not.toHaveBeenCalledWith('W1N1', 0);
+                expect(driver.history.upload).toHaveBeenCalledWith('W2N2', 0);
+                expect(driver.history.upload).toHaveBeenCalledWith('W3N3', 0);
+            });
+        });
+    });
+
+    describe('saveDeactivatedRoomsHistory', () => {
+        it('saves the last tick for rooms that will not run next tick', () => {
+            driver.getActiveRooms.and.callFake(() => q.when(['W1N1']));
+            driver.getRoomObjects.and.callFake(roomId => q.when({
+                objects: {src: {_id: 'src', type: 'source', room: roomId}}
+            }));
+            return history.saveDeactivatedRoomsHistory(['W1N1', 'W2N2'], 10).then(() => {
+                expect(driver.getRoomObjects).toHaveBeenCalledWith('W2N2');
+                expect(driver.getRoomObjects).not.toHaveBeenCalledWith('W1N1');
+                expect(driver.history.saveTick).toHaveBeenCalledWith(
+                    'W2N2', 10, JSON.stringify({src: {_id: 'src', type: 'source', room: 'W2N2'}}));
+                expect(driver.history.upload).not.toHaveBeenCalled();
+            });
+        });
+
+        it('closes the previous chunk when the last tick is a boundary', () => {
+            driver.getActiveRooms.and.callFake(() => q.when([]));
+            driver.getRoomObjects.and.callFake(() => q.when({objects: {}}));
+            return history.saveDeactivatedRoomsHistory(['W2N2'], 20).then(() => {
+                expect(driver.history.upload).toHaveBeenCalledWith('W2N2', 0);
+                expect(driver.history.saveTick).toHaveBeenCalledWith('W2N2', 20, JSON.stringify({}));
+            });
+        });
+    });
+});

src/history.js

@@ -0,0 +1,87 @@
+var q = require('q'),
+    _ = require('lodash'),
+    utils = require('./utils'),
+    driver = utils.getDriver();
+
+var currentHistoryPromise = q.when();
+
+exports.buildHistoryPayload = function(roomObjects) {
+    var historyPayload = {};
+    _.forEach(roomObjects, (object) => {
+        if (!object || object.type === 'flag') {
+            return;
+        }
+        if (object.type === 'creep' || object.type === 'powerCreep') {
+            var clone = JSON.parse(JSON.stringify(object));
+            clone._id = '' + object._id;
+            if (clone.actionLog && clone.actionLog.say && !clone.actionLog.say.isPublic) {
+                delete clone.actionLog.say;
+            }
+            historyPayload[clone._id] = clone;
+        } else {
+            historyPayload[object._id] = object;
+        }
+    });
+    return historyPayload;
+};
+
+exports.saveRoomHistory = function(roomId, objects, gameTime) {
+
+    var data = JSON.stringify(objects);
+    var chunkSize = driver.config.historyChunkSize;
+
+    return currentHistoryPromise.then(() => {
+        var promise = q.when();
+
+        if (!(gameTime % chunkSize)) {
+            var prevBase = Math.floor((gameTime - 1) / chunkSize) * chunkSize;
+            promise = driver.history.upload(roomId, prevBase);
+        }
+
+        var baseTime = gameTime - (gameTime % chunkSize);
+        currentHistoryPromise = promise.then(() => driver.history.saveTick(roomId, gameTime, data)
+            .then(() => driver.history.markPendingHistory(roomId, baseTime)));
+        return currentHistoryPromise;
+    });
+};
+
+exports.uploadPendingChunks = function(gameTime, processedRooms) {
+    var chunkSize = driver.config.historyChunkSize;
+    if ((gameTime - 1) % chunkSize) {
+        return q.when();
+    }
+    var prevBase = Math.floor((gameTime - 2) / chunkSize) * chunkSize;
+    if (prevBase < 0) {
+        return q.when();
+    }
+    var skip = {};
+    _.forEach(processedRooms, roomId => {
+        skip[roomId] = true;
+    });
+    return driver.history.takePendingHistory(prevBase)
+        .then(rooms => q.all(_.map(rooms || [], roomId => {
+            if (skip[roomId]) {
+                return;
+            }
+            return driver.history.upload(roomId, prevBase);
+        })));
+};
+
+exports.saveDeactivatedRoomsHistory = function(processedRooms, gameTime) {
+    if (!processedRooms || !processedRooms.length) {
+        return q.when();
+    }
+    return driver.getActiveRooms()
+        .then(active => {
+            var stillActive = {};
+            _.forEach(active || [], roomId => {
+                stillActive[roomId] = true;
+            });
+            var deactivated = _.filter(processedRooms, roomId => !stillActive[roomId]);
+            return q.all(_.map(deactivated, roomId =>
+                driver.getRoomObjects(roomId).then(result =>
+                    exports.saveRoomHistory(roomId, exports.buildHistoryPayload(result && result.objects), gameTime)
+                )
+            ));
+        });
+};

src/main.js

@@ -3,15 +3,18 @@ var q = require('q'),
     _ = require('lodash'),
     utils = require('./utils'),
     driver = utils.getDriver(),
-    config = require('./config');
+    config = require('./config')
+    history = require('./history');
 
 var lastAccessibleRoomsUpdate = 0;
 var roomsQueue, usersQueue;
 
 function loop() {
 
     var resetInterval, startLoopTime = process.hrtime ? process.hrtime() : Date.now(),
-        stage = 'start';
+        stage = 'start',
+        processedRooms,
+        tickGameTime;
 
     driver.config.emit('mainLoopStage',stage);
 
@@ -45,6 +48,7 @@ function loop() {
             return driver.getAllRoomsNames();
         })
         .then((rooms) => {
+            processedRooms = rooms;
             stage = 'addRoomsToQueue';
             driver.config.emit('mainLoopStage',stage, rooms);
             return roomsQueue.addMulti(rooms);
@@ -54,6 +58,13 @@ function loop() {
             driver.config.emit('mainLoopStage',stage);
             return roomsQueue.whenAllDone();
         })
+        .then(() => driver.getGameTime())
+        .then((gameTime) => {
+            tickGameTime = gameTime;
+            stage = 'flushHistory';
+            driver.config.emit('mainLoopStage',stage);
+            return history.uploadPendingChunks(gameTime, processedRooms);
+        })
         .then(() => {
             stage = 'commit1';
             driver.config.emit('mainLoopStage',stage);
@@ -69,6 +80,11 @@ function loop() {
             driver.config.emit('mainLoopStage',stage);
             return driver.commitDbBulk();
         })
+        .then(() => {
+            stage = 'saveDeactivatedRoomHistory';
+            driver.config.emit('mainLoopStage',stage);
+            return history.saveDeactivatedRoomsHistory(processedRooms, tickGameTime);
+        })
         .then(() => {
             stage = 'incrementGameTime';
             driver.config.emit('mainLoopStage',stage);

src/processor.js

@@ -6,9 +6,10 @@ var q = require('q'),
     driver = utils.getDriver(),
     C = driver.constants,
     config = require('./config'),
-    fakeRuntime = require('./processor/common/fake-runtime');
+    fakeRuntime = require('./processor/common/fake-runtime'),
+    history = require('./history');
 
-var roomsQueue, usersQueue, lastRoomsStatsSaveTime = 0, currentHistoryPromise = q.when();
+var roomsQueue, usersQueue, lastRoomsStatsSaveTime = 0;
 
 const KEEPER_ID = "3";
 const INVADER_ID = "2";
@@ -17,14 +18,17 @@ function processRoom(roomId, {intents, roomObjects, users, roomTerrain, gameTime
 
     return q.when().then(() => {
 
+        if (gameTime > 0) {
+            history.saveRoomHistory(roomId, history.buildHistoryPayload(roomObjects), gameTime - 1);
+        }
+
         var bulk = driver.bulkObjectsWrite(),
             bulkUsers = driver.bulkUsersWrite(),
             bulkFlags = driver.bulkFlagsWrite(),
             bulkUsersPowerCreeps = driver.bulkUsersPowerCreeps(),
             oldObjects = {},
             hasNewbieWalls = false,
             stats = driver.getRoomStatsUpdater(roomId),
-            objectsToHistory = {},
             roomSpawns = [], roomExtensions = [], roomNukes = [], keepers = [], invaders = [], invaderCore = null,
             oldRoomInfo = _.clone(roomInfo);
 
@@ -425,20 +429,6 @@ function processRoom(roomId, {intents, roomObjects, users, roomTerrain, gameTime
                 }
             }
 
-            if (object.type != 'flag') {
-                objectsToHistory[object._id] = object;
-
-                if (object.type == 'creep' || object.type == 'powerCreep') {
-                    objectsToHistory[object._id] = JSON.parse(JSON.stringify(object));
-                    objectsToHistory[object._id]._id = "" + object._id;
-                    delete objectsToHistory[object._id]._actionLog;
-                    delete objectsToHistory[object._id]._ticksToLive;
-                    if (object.actionLog.say && !object.actionLog.say.isPublic) {
-                        delete objectsToHistory[object._id].actionLog.say;
-                    }
-                }
-            }
-
             if (object.user) {
                 //userVisibility[object.user] = true;
 
@@ -503,7 +493,6 @@ function processRoom(roomId, {intents, roomObjects, users, roomTerrain, gameTime
 
         if(activateRoom) {
             driver.activateRoom(roomId);
-            saveRoomHistory(roomId, objectsToHistory, gameTime);
         }
 
         if(!_.isEqual(roomInfo, oldRoomInfo)) {
@@ -519,23 +508,6 @@ function processRoom(roomId, {intents, roomObjects, users, roomTerrain, gameTime
     });
 }
 
-function saveRoomHistory(roomId, objects, gameTime) {
-
-    return currentHistoryPromise.then(() => {
-        var promise = q.when();
-
-        if (!(gameTime % driver.config.historyChunkSize)) {
-            var baseTime = Math.floor((gameTime - 1) / driver.config.historyChunkSize) * driver.config.historyChunkSize;
-            promise = driver.history.upload(roomId, baseTime);
-        }
-
-        var data = JSON.stringify(objects);
-        currentHistoryPromise = promise.then(() => driver.history.saveTick(roomId, gameTime, data));
-        return currentHistoryPromise;
-    });
-}
-
-
 driver.connect('processor')
     .then(() => driver.queue.create('rooms', 'read'))
     .catch((error) => {