Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • getanthill/datastore
  • jeremylvln/datastore
2 results
Show changes
Commits on Source (2)
Hide whitespace changes
Inline Side-by-side
Showing
with 402 additions and 26 deletions
CHANGELOG.md
View file @ 735c7bd3
## [0.85.5](https://gitlab.com/getanthill/datastore/compare/v0.85.4...v0.85.5) (2024-06-26)
### Bug Fixes
* **walk:** fix behavior om walk multi for events with the exact same created timestamp ([1f8ceb1](https://gitlab.com/getanthill/datastore/commit/1f8ceb11bceb9141255ab5d14c4212810729147e))
## [0.85.4](https://gitlab.com/getanthill/datastore/compare/v0.85.3...v0.85.4) (2024-06-20) ## [0.85.4](https://gitlab.com/getanthill/datastore/compare/v0.85.3...v0.85.4) (2024-06-20)
......
package-lock.json
View file @ 735c7bd3
{ {
"name": "@getanthill/datastore", "name": "@getanthill/datastore",
"version": "0.85.4", "version": "0.85.5",
"lockfileVersion": 2, "lockfileVersion": 2,
"requires": true, "requires": true,
"packages": { "packages": {
"": { "": {
"name": "@getanthill/datastore", "name": "@getanthill/datastore",
"version": "0.85.4", "version": "0.85.5",
"license": "MIT", "license": "MIT",
"dependencies": { "dependencies": {
"@getanthill/api-validators": "1.3.0", "@getanthill/api-validators": "1.3.0",
package.json
View file @ 735c7bd3
{ {
"name": "@getanthill/datastore", "name": "@getanthill/datastore",
"description": "Event-Sourced Datastore", "description": "Event-Sourced Datastore",
"version": "0.85.4", "version": "0.85.5",
"main": "dist/sdk/index.js", "main": "dist/sdk/index.js",
"bin": { "bin": {
"datastore": "scripts/cli.js", "datastore": "scripts/cli.js",
......
src/App.ts
View file @ 735c7bd3
...@@ -90,7 +90,7 @@ export default class App { ...@@ -90,7 +90,7 @@ export default class App {
); );
res.header( res.header(
'Access-Control-Allow-Headers', 'Access-Control-Allow-Headers',
`${this.services.config.features.cors.allowHeaders},authorization,csrf-token,page,page-size,decrypt,content-type,cache-control,cursor-last-id`, `${this.services.config.features.cors.allowHeaders},authorization,csrf-token,page,page-size,decrypt,content-type,cache-control,cursor-last-id,cursor-last-correlation-id`,
); );
this.services.config.features.cors.allowMethods !== undefined && this.services.config.features.cors.allowMethods !== undefined &&
......
src/api/models/controllers.test.ts
View file @ 735c7bd3
...@@ -2001,6 +2001,8 @@ describe('controllers/models', () => { ...@@ -2001,6 +2001,8 @@ describe('controllers/models', () => {
req.params.model = 'users'; req.params.model = 'users';
req.headers['cursor-last-id'] = headers['cursor-last-id']; req.headers['cursor-last-id'] = headers['cursor-last-id'];
req.headers['cursor-last-correlation-id'] =
headers['cursor-last-correlation-id'];
req.headers['page-size'] = 2; req.headers['page-size'] = 2;
await controller(req, res, next); await controller(req, res, next);
...@@ -2049,6 +2051,8 @@ describe('controllers/models', () => { ...@@ -2049,6 +2051,8 @@ describe('controllers/models', () => {
req.params.model = 'users'; req.params.model = 'users';
req.headers['cursor-last-id'] = headers['cursor-last-id']; req.headers['cursor-last-id'] = headers['cursor-last-id'];
req.headers['cursor-last-correlation-id'] =
headers['cursor-last-correlation-id'];
req.headers['page-size'] = 2; req.headers['page-size'] = 2;
await controller(req, res, next); await controller(req, res, next);
...@@ -2901,6 +2905,8 @@ describe('controllers/models', () => { ...@@ -2901,6 +2905,8 @@ describe('controllers/models', () => {
delete res.body; delete res.body;
req.headers['cursor-last-id'] = cursorHeaders['cursor-last-id']; req.headers['cursor-last-id'] = cursorHeaders['cursor-last-id'];
req.headers['cursor-last-correlation-id'] =
cursorHeaders['cursor-last-correlation-id'];
await controller(req, res, next); await controller(req, res, next);
...@@ -2934,6 +2940,8 @@ describe('controllers/models', () => { ...@@ -2934,6 +2940,8 @@ describe('controllers/models', () => {
delete res.body; delete res.body;
req.headers['cursor-last-id'] = cursorHeaders['cursor-last-id']; req.headers['cursor-last-id'] = cursorHeaders['cursor-last-id'];
req.headers['cursor-last-correlation-id'] =
cursorHeaders['cursor-last-correlation-id'];
await controller(req, res, next); await controller(req, res, next);
......
src/api/models/controllers.ts
View file @ 735c7bd3
...@@ -254,6 +254,7 @@ export function find(services: Services) { ...@@ -254,6 +254,7 @@ export function find(services: Services) {
services.metrics.incrementApiFind, services.metrics.incrementApiFind,
async (req: Request, res: Response, next: NextFunction) => { async (req: Request, res: Response, next: NextFunction) => {
const Model = services.models.getModel(req.params.model); const Model = services.models.getModel(req.params.model);
const correlationField = Model.getModelConfig().correlation_field;
// Map the query parameters to request services.MongoDb accordingly // Map the query parameters to request services.MongoDb accordingly
const { query: mappedQuery, options } = mapFindQuery(Model, req.query); const { query: mappedQuery, options } = mapFindQuery(Model, req.query);
...@@ -263,6 +264,9 @@ export function find(services: Services) { ...@@ -263,6 +264,9 @@ export function find(services: Services) {
const page: number = parseInt(req.header('page') ?? '0', 10); const page: number = parseInt(req.header('page') ?? '0', 10);
const pageSize: number = parseInt(req.header('page-size') ?? '1000', 10); const pageSize: number = parseInt(req.header('page-size') ?? '1000', 10);
const cursorLastId: string = req.header('cursor-last-id') as string; const cursorLastId: string = req.header('cursor-last-id') as string;
const cursorLastCorrelationId: string = req.header(
'cursor-last-correlation-id',
) as string;
const withResponseValidation: boolean = const withResponseValidation: boolean =
req.header('with-response-validation') !== 'false'; req.header('with-response-validation') !== 'false';
const mustDecrypt: boolean = const mustDecrypt: boolean =
...@@ -281,8 +285,9 @@ export function find(services: Services) { ...@@ -281,8 +285,9 @@ export function find(services: Services) {
if (mappedQuery === null) { if (mappedQuery === null) {
res.set({ res.set({
'correlation-field': Model.getModelConfig().correlation_field, 'correlation-field': correlationField,
'cursor-last-id': cursorLastId, 'cursor-last-id': cursorLastId,
'cursor-last-correlation-id': cursorLastCorrelationId,
'page-count': 0, 'page-count': 0,
page, page,
'page-size': pageSize, 'page-size': pageSize,
...@@ -304,7 +309,12 @@ export function find(services: Services) { ...@@ -304,7 +309,12 @@ export function find(services: Services) {
_mappedQuery = { _mappedQuery = {
$and: [ $and: [
_mappedQuery, _mappedQuery,
getQueryFromCursorLastId(Model, options, cursorLastId), getQueryFromCursorLastId(
Model,
options,
cursorLastId,
cursorLastCorrelationId,
),
], ],
}; };
} }
...@@ -321,17 +331,26 @@ export function find(services: Services) { ...@@ -321,17 +331,26 @@ export function find(services: Services) {
cursor.skip(pageSize * page); cursor.skip(pageSize * page);
} }
cursor.batchSize(pageSize); cursor.batchSize(pageSize * 2);
let entities = []; let entities = [];
let i = 0; let i = 0;
let lastReturnedFound = false;
// i < pageSize useful if pageSize === 0; // i < pageSize useful if pageSize === 0;
while (i < pageSize && (await cursor.hasNext())) { while (i < pageSize && (await cursor.hasNext())) {
const entity = await cursor.next(); const entity = await cursor.next();
entities.push(omit(entity, '_id'));
if (cursorLastCorrelationId && lastReturnedFound === false) {
lastReturnedFound =
entity[correlationField] === cursorLastCorrelationId;
continue;
}
entities.push(omit(entity, '_id'));
i++; i++;
if (i >= pageSize) { if (i >= pageSize) {
break; break;
} }
...@@ -347,12 +366,11 @@ export function find(services: Services) { ...@@ -347,12 +366,11 @@ export function find(services: Services) {
* Adding the correlation field to the response headers * Adding the correlation field to the response headers
* @todo need to apply this to every request and in the API documentation * @todo need to apply this to every request and in the API documentation
*/ */
const lastEntity = entities[entities.length - 1];
res.set({ res.set({
'correlation-field': Model.getModelConfig().correlation_field, 'correlation-field': correlationField,
'cursor-last-id': buildCursorLastId( 'cursor-last-id': buildCursorLastId(lastEntity, options),
entities[entities.length - 1], 'cursor-last-correlation-id': lastEntity?.[correlationField] ?? '',
options,
),
}); });
// Pagination headers if not requested with cursor last ID: // Pagination headers if not requested with cursor last ID:
...@@ -413,11 +431,15 @@ export function getEvents(services: Services) { ...@@ -413,11 +431,15 @@ export function getEvents(services: Services) {
services.metrics.incrementApiEvents, services.metrics.incrementApiEvents,
async (req: Request, res: Response, _next: NextFunction) => { async (req: Request, res: Response, _next: NextFunction) => {
const Model = services.models.getModel(req.params.model); const Model = services.models.getModel(req.params.model);
const correlationField = Model.getModelConfig().correlation_field;
// Retrieve pagination parameters: // Retrieve pagination parameters:
const page: number = parseInt(req.header('page') ?? '0', 10); const page: number = parseInt(req.header('page') ?? '0', 10);
const pageSize: number = parseInt(req.header('page-size') ?? '1000', 10); const pageSize: number = parseInt(req.header('page-size') ?? '1000', 10);
const cursorLastId: string = req.header('cursor-last-id') as string; const cursorLastId: string = req.header('cursor-last-id') as string;
const cursorLastCorrelationId: string = req.header(
'cursor-last-correlation-id',
) as string;
let cursor; let cursor;
let count; let count;
...@@ -454,7 +476,12 @@ export function getEvents(services: Services) { ...@@ -454,7 +476,12 @@ export function getEvents(services: Services) {
_mappedQuery = { _mappedQuery = {
$and: [ $and: [
_mappedQuery, _mappedQuery,
getQueryFromCursorLastId(Model, _options, cursorLastId), getQueryFromCursorLastId(
Model,
_options,
cursorLastId,
cursorLastCorrelationId,
),
], ],
}; };
} }
...@@ -478,8 +505,9 @@ export function getEvents(services: Services) { ...@@ -478,8 +505,9 @@ export function getEvents(services: Services) {
if (mappedQuery === null) { if (mappedQuery === null) {
res.set({ res.set({
'correlation-field': Model.getModelConfig().correlation_field, 'correlation-field': correlationField,
'cursor-last-id': '', 'cursor-last-id': '',
'cursor-last-correlation-id': '',
'page-count': 0, 'page-count': 0,
page: 0, page: 0,
'page-size': pageSize, 'page-size': pageSize,
...@@ -497,7 +525,12 @@ export function getEvents(services: Services) { ...@@ -497,7 +525,12 @@ export function getEvents(services: Services) {
_mappedQuery = { _mappedQuery = {
$and: [ $and: [
_mappedQuery, _mappedQuery,
getQueryFromCursorLastId(Model, _options, cursorLastId), getQueryFromCursorLastId(
Model,
_options,
cursorLastId,
cursorLastCorrelationId,
),
], ],
}; };
} }
...@@ -519,15 +552,25 @@ export function getEvents(services: Services) { ...@@ -519,15 +552,25 @@ export function getEvents(services: Services) {
cursor.skip(pageSize * page); cursor.skip(pageSize * page);
} }
cursor.batchSize(pageSize); cursor.batchSize(pageSize * 2);
const events = []; const events = [];
let i = 0; let i = 0;
let lastReturnedFound = false;
while (i < pageSize && (await cursor.hasNext())) { while (i < pageSize && (await cursor.hasNext())) {
const entity = await cursor.next(); const entity = await cursor.next();
events.push(entity);
if (cursorLastCorrelationId && lastReturnedFound === false) {
lastReturnedFound = true;
lastReturnedFound =
`${entity[correlationField]}:${entity.version}` ===
cursorLastCorrelationId;
continue;
}
events.push(entity);
i++; i++;
if (i >= pageSize) { if (i >= pageSize) {
break; break;
...@@ -544,12 +587,13 @@ export function getEvents(services: Services) { ...@@ -544,12 +587,13 @@ export function getEvents(services: Services) {
* Adding the correlation field to the response headers * Adding the correlation field to the response headers
* @todo need to apply this to every request and in the API documentation * @todo need to apply this to every request and in the API documentation
*/ */
const lastEvent = events[events.length - 1];
res.set({ res.set({
'correlation-field': Model.getModelConfig().correlation_field, 'correlation-field': correlationField,
'cursor-last-id': buildCursorLastId( 'cursor-last-id': buildCursorLastId(lastEvent, _options),
events[events.length - 1], 'cursor-last-correlation-id': lastEvent
_options, ? `${lastEvent?.[correlationField]}:${lastEvent?.version}`
), : '',
}); });
// Pagination headers if not requested with cursor last ID: // Pagination headers if not requested with cursor last ID:
......
src/api/utils/index.ts
View file @ 735c7bd3
...@@ -238,15 +238,21 @@ export function getQueryFromCursorLastId( ...@@ -238,15 +238,21 @@ export function getQueryFromCursorLastId(
Model: GenericType, Model: GenericType,
options: any, options: any,
cursorLastId: string, cursorLastId: string,
cursorLastCorrelationId: string,
): any { ): any {
const _revertId = JSON.parse(Buffer.from(cursorLastId, 'hex').toString()); const _revertId = JSON.parse(Buffer.from(cursorLastId, 'hex').toString());
const { query: _lastIdQuery } = mapFindQuery(Model, _revertId); const { query: _lastIdQuery } = mapFindQuery(Model, _revertId);
const cursorQuery = mapValues(_lastIdQuery, (val: any, key: string) => { const cursorQuery = mapValues(_lastIdQuery, (val: any, key: string) => {
if (options.sort[key] === 1) { if (options.sort[key] === 1) {
return { $gt: val }; // Following condition for backward compatibility:
// `2024-06-26`: Nominal case should be `$gte` with `cursorLastCorrelationId`
return { [cursorLastCorrelationId ? '$gte' : '$gt']: val };
} }
return { $lt: val }; // Following condition for backward compatibility:
// `2024-06-26`: Nominal case should be `$gte` with `cursorLastCorrelationId`
return { [cursorLastCorrelationId ? '$lte' : '$lt']: val };
}); });
return cursorQuery; return cursorQuery;
......
src/sdk/Datastore.ts
View file @ 735c7bd3
...@@ -785,6 +785,7 @@ export default class Datastore extends EventEmitter { ...@@ -785,6 +785,7 @@ export default class Datastore extends EventEmitter {
current_version: number; current_version: number;
version_ordered?: boolean; version_ordered?: boolean;
cursor_last_id?: string; cursor_last_id?: string;
cursor_last_correlation_id: string;
headers?: any; headers?: any;
}, },
) { ) {
...@@ -798,6 +799,7 @@ export default class Datastore extends EventEmitter { ...@@ -798,6 +799,7 @@ export default class Datastore extends EventEmitter {
const headers = { const headers = {
...opts?.headers, ...opts?.headers,
'cursor-last-id': opts?.cursor_last_id, 'cursor-last-id': opts?.cursor_last_id,
'cursor-last-correlation-id': opts?.cursor_last_correlation_id,
}; };
if (source === 'events') { if (source === 'events') {
......
src/sdk/aggregator/Aggregator.test.ts
View file @ 735c7bd3
...@@ -442,6 +442,7 @@ describe('sdk/Aggregator', () => { ...@@ -442,6 +442,7 @@ describe('sdk/Aggregator', () => {
data: 0, data: 0,
}, },
cursor_last_id: '', cursor_last_id: '',
cursor_last_correlation_id: '',
headers: undefined, headers: undefined,
version_ordered: false, version_ordered: false,
}, },
......
src/sdk/utils.integration.test.ts
View file @ 735c7bd3
...@@ -348,6 +348,305 @@ describe('sdk/utils (integration)', () => { ...@@ -348,6 +348,305 @@ describe('sdk/utils (integration)', () => {
]); ]);
}); });
it('(backward) walk over entities in order with exact same `created_at` timestamp', async () => {
const createdAt = new Date();
const { data: entityA } = await ds.create(
modelNames[0],
{
firstname: 'Alice',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityC } = await ds.create(
modelNames[0],
{
firstname: 'Eve',
},
{
'created-at': createdAt.toISOString(),
},
);
const _walkNext = ds.walkNext;
function mockWalkNext(
model: string,
query: object,
source: string,
page: number,
pageSize: number,
opts: {
current_version: number;
version_ordered?: boolean;
cursor_last_id?: string;
cursor_last_correlation_id: string;
headers?: any;
},
) {
opts.cursor_last_correlation_id = '';
return _walkNext.call(ds, model, query, source, page, pageSize, opts);
}
ds.walkNext = mockWalkNext.bind(ds);
const queryA = {
datastore: 'default',
model: modelNames[0],
query: {},
source: 'entities',
};
// @ts-ignore
await utils.walkMulti(datastores, [queryA], 1, handler);
expect(res).toHaveLength(2);
ds.walkNext = _walkNext;
});
it('(backward) walk over entities in order with exact same `created_at` timestamp', async () => {
const createdAt = new Date();
const { data: entityA } = await ds.create(
modelNames[0],
{
firstname: 'Alice',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityC } = await ds.create(
modelNames[0],
{
firstname: 'Eve',
},
{
'created-at': createdAt.toISOString(),
},
);
const _walkNext = ds.walkNext;
function mockWalkNext(
model: string,
query: object,
source: string,
page: number,
pageSize: number,
opts: {
current_version: number;
version_ordered?: boolean;
cursor_last_id?: string;
cursor_last_correlation_id: string;
headers?: any;
},
) {
opts.cursor_last_correlation_id = '';
return _walkNext.call(ds, model, query, source, page, pageSize, opts);
}
ds.walkNext = mockWalkNext.bind(ds);
const queryA = {
datastore: 'default',
model: modelNames[0],
query: {
_sort: {
created_at: -1,
},
},
source: 'entities',
};
// @ts-ignore
await utils.walkMulti(datastores, [queryA], 1, handler);
expect(res).toHaveLength(2);
ds.walkNext = _walkNext;
});
it('(backward) walk over events in order with exact same `created_at` timestamp', async () => {
const createdAt = new Date();
const { data: entityA } = await ds.create(
modelNames[0],
{
firstname: 'Alice',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityC } = await ds.create(
modelNames[0],
{
firstname: 'Eve',
},
{
'created-at': createdAt.toISOString(),
},
);
const _walkNext = ds.walkNext;
function mockWalkNext(
model: string,
query: object,
source: string,
page: number,
pageSize: number,
opts: {
current_version: number;
version_ordered?: boolean;
cursor_last_id?: string;
cursor_last_correlation_id: string;
headers?: any;
},
) {
opts.cursor_last_correlation_id = '';
return _walkNext.call(ds, model, query, source, page, pageSize, opts);
}
ds.walkNext = mockWalkNext.bind(ds);
const queryA = {
datastore: 'default',
model: modelNames[0],
query: {},
source: 'events',
};
// @ts-ignore
await utils.walkMulti(datastores, [queryA], 1, handler);
expect(res).toHaveLength(2);
ds.walkNext = _walkNext;
});
it('walk over entities in order with exact same `created_at` timestamp', async () => {
const createdAt = new Date();
const { data: entityA } = await ds.create(
modelNames[0],
{
firstname: 'Alice',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityB } = await ds.create(
modelNames[1],
{
firstname: 'Bernard',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityC } = await ds.create(
modelNames[0],
{
firstname: 'Eve',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: eventsA } = await ds.allEvents(modelNames[0], {});
const { data: eventsB } = await ds.allEvents(modelNames[1], {});
const queryA = {
datastore: 'default',
model: modelNames[0],
query: {
_sort: {
created_at: -1,
},
},
source: 'entities',
};
const queryB = {
datastore: 'default',
model: modelNames[1],
query: {
_sort: {
created_at: -1,
},
},
source: 'entities',
};
// @ts-ignore
await utils.walkMulti(datastores, [queryB, queryA], 1, handler);
expect(res).toHaveLength(3);
});
it('walk over events in order with exact same `created_at` timestamp', async () => {
const createdAt = new Date();
const { data: entityA } = await ds.create(
modelNames[0],
{
firstname: 'Alice',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityB } = await ds.create(
modelNames[1],
{
firstname: 'Bernard',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: entityC } = await ds.create(
modelNames[0],
{
firstname: 'Eve',
},
{
'created-at': createdAt.toISOString(),
},
);
const { data: eventsA } = await ds.allEvents(modelNames[0], {});
const { data: eventsB } = await ds.allEvents(modelNames[1], {});
const queryA = {
datastore: 'default',
model: modelNames[0],
query: {},
source: 'events',
};
const queryB = {
datastore: 'default',
model: modelNames[1],
query: {},
source: 'events',
};
// @ts-ignore
await utils.walkMulti(datastores, [queryB, queryA], 1, handler);
expect(res).toHaveLength(3);
});
it('walk over events in temporal order and not version order', async () => { it('walk over events in temporal order and not version order', async () => {
const { data: entityA } = await ds.create(modelNames[0], { const { data: entityA } = await ds.create(modelNames[0], {
firstname: 'Alice', firstname: 'Alice',
......
src/sdk/utils.test.ts
View file @ 735c7bd3
...@@ -563,6 +563,7 @@ describe('utils', () => { ...@@ -563,6 +563,7 @@ describe('utils', () => {
headers: undefined, headers: undefined,
version_ordered: false, version_ordered: false,
cursor_last_id: '', cursor_last_id: '',
cursor_last_correlation_id: '',
}); });
expect(processedEntities).toEqual(entities); expect(processedEntities).toEqual(entities);
...@@ -617,6 +618,7 @@ describe('utils', () => { ...@@ -617,6 +618,7 @@ describe('utils', () => {
headers: undefined, headers: undefined,
version_ordered: false, version_ordered: false,
cursor_last_id: '', cursor_last_id: '',
cursor_last_correlation_id: '',
}, },
); );
......
src/sdk/utils.ts
View file @ 735c7bd3
...@@ -20,6 +20,7 @@ export interface Iteration { ...@@ -20,6 +20,7 @@ export interface Iteration {
max_version: number; max_version: number;
is_exhausted: boolean; is_exhausted: boolean;
cursor_last_id: string; cursor_last_id: string;
cursor_last_correlation_id: string;
results: AnyObject[]; results: AnyObject[];
} }
...@@ -154,6 +155,7 @@ export async function fetchResults( ...@@ -154,6 +155,7 @@ export async function fetchResults(
2 * pageSize, 2 * pageSize,
{ {
cursor_last_id: queryIteration.cursor_last_id, cursor_last_id: queryIteration.cursor_last_id,
cursor_last_correlation_id: queryIteration.cursor_last_correlation_id,
current_version: queryIteration.version, current_version: queryIteration.version,
version_ordered: isVersionOrdered, version_ordered: isVersionOrdered,
headers: query.headers, headers: query.headers,
...@@ -171,11 +173,13 @@ export async function fetchResults( ...@@ -171,11 +173,13 @@ export async function fetchResults(
} }
const cursorLastId = headers?.['cursor-last-id'] ?? ''; const cursorLastId = headers?.['cursor-last-id'] ?? '';
const cursorLastCorrelationId = headers?.['cursor-last-correlation-id'] ?? '';
if ( if (
!!cursorLastId && !!cursorLastId &&
!!queryIteration.cursor_last_id && !!queryIteration.cursor_last_id &&
cursorLastId === queryIteration.cursor_last_id cursorLastId === queryIteration.cursor_last_id &&
cursorLastCorrelationId === queryIteration.cursor_last_correlation_id
) { ) {
throw new Error('Same cursor last id after iteration'); throw new Error('Same cursor last id after iteration');
} }
...@@ -183,6 +187,7 @@ export async function fetchResults( ...@@ -183,6 +187,7 @@ export async function fetchResults(
queryIteration.results.push(...results); queryIteration.results.push(...results);
queryIteration.page += 1; queryIteration.page += 1;
queryIteration.cursor_last_id = cursorLastId; queryIteration.cursor_last_id = cursorLastId;
queryIteration.cursor_last_correlation_id = cursorLastCorrelationId;
if ( if (
results.length < pageSize && results.length < pageSize &&
...@@ -208,6 +213,7 @@ export async function fetchResults( ...@@ -208,6 +213,7 @@ export async function fetchResults(
queryIteration.version = nextVersion; queryIteration.version = nextVersion;
queryIteration.cursor_last_id = ''; queryIteration.cursor_last_id = '';
queryIteration.cursor_last_correlation_id = '';
queryIteration.page = 0; queryIteration.page = 0;
} }
...@@ -329,6 +335,7 @@ export async function walkMulti( ...@@ -329,6 +335,7 @@ export async function walkMulti(
max_version: maxVersions[i], max_version: maxVersions[i],
is_exhausted: false, is_exhausted: false,
cursor_last_id: '', cursor_last_id: '',
cursor_last_correlation_id: '',
results: [], results: [],
}, },
]), ]),
......