diff --git a/services/history-v1/storage/lib/chunk_store/redis.js b/services/history-v1/storage/lib/chunk_store/redis.js index 0ae7cee2e5..9163536342 100644 --- a/services/history-v1/storage/lib/chunk_store/redis.js +++ b/services/history-v1/storage/lib/chunk_store/redis.js @@ -501,6 +501,11 @@ rclient.defineCommand('set_persisted_version', { return 'too_low' end + -- Refuse to set a persisted version that is higher than the head version + if newPersistedVersion > headVersion then + return 'too_high' + end + -- Set the persisted version redis.call('SET', persistedVersionKey, newPersistedVersion) @@ -541,6 +546,13 @@ async function setPersistedVersion(projectId, persistedVersion) { status, }) + if (status === 'too_high') { + throw new VersionOutOfBoundsError( + 'Persisted version cannot be higher than head version', + { projectId, persistedVersion } + ) + } + return status } catch (err) { metrics.inc('chunk_store.redis.set_persisted_version', 1, { diff --git a/services/history-v1/test/acceptance/js/storage/chunk_store_redis_backend.test.js b/services/history-v1/test/acceptance/js/storage/chunk_store_redis_backend.test.js index 2b13343fc4..04d801c73d 100644 --- a/services/history-v1/test/acceptance/js/storage/chunk_store_redis_backend.test.js +++ b/services/history-v1/test/acceptance/js/storage/chunk_store_redis_backend.test.js @@ -714,10 +714,20 @@ describe('chunk buffer Redis backend', function () { }) it('should set the persisted version', async function () { - await redisBackend.setPersistedVersion(projectId, 3) + const status = await redisBackend.setPersistedVersion(projectId, 3) + expect(status).to.equal('ok') const state = await redisBackend.getState(projectId) expect(state.persistedVersion).to.equal(3) }) + + it('should refuse to set a persisted version greater than the head version', async function () { + await expect( + redisBackend.setPersistedVersion(projectId, 10) + ).to.be.rejectedWith(VersionOutOfBoundsError) + // Ensure persisted version remains unchanged + const state = await redisBackend.getState(projectId) + expect(state.persistedVersion).to.be.null + }) }) describe('when the persisted version is set', function () { @@ -730,13 +740,24 @@ describe('chunk buffer Redis backend', function () { }) it('should set the persisted version', async function () { - await redisBackend.setPersistedVersion(projectId, 5) + const status = await redisBackend.setPersistedVersion(projectId, 5) + expect(status).to.equal('ok') const state = await redisBackend.getState(projectId) expect(state.persistedVersion).to.equal(5) }) it('should not decrease the persisted version', async function () { - await redisBackend.setPersistedVersion(projectId, 2) + const status = await redisBackend.setPersistedVersion(projectId, 2) + expect(status).to.equal('too_low') + const state = await redisBackend.getState(projectId) + expect(state.persistedVersion).to.equal(3) + }) + + it('should refuse to set a persisted version greater than the head version', async function () { + await expect( + redisBackend.setPersistedVersion(projectId, 10) + ).to.be.rejectedWith(VersionOutOfBoundsError) + // Ensure persisted version remains unchanged const state = await redisBackend.getState(projectId) expect(state.persistedVersion).to.equal(3) })