From 66c20fd21a588b196435f514fd22029e9f259a6b Mon Sep 17 00:00:00 2001 From: Omid Mirzaei Date: Mon, 28 Sep 2026 11:20:13 +0400 Subject: [PATCH] fix(core): do not store a snapshot when every reducer skips fetchEntitySnapshot wrote the loaded snapshot back when every pending event returned ReducerAction.Skip. That write copies the same data and does not move the snapshot cursor. Store a snapshot only when a reducer produced a new one. Fixes #700 --- ...p-unchanged-snapshot_2026-09-28-07-24.json | 11 +++ packages/core/src/services/event-store.ts | 7 ++ .../core/test/services/event-store.test.ts | 74 +++++++++++++++++-- 3 files changed, 85 insertions(+), 7 deletions(-) create mode 100644 common/changes/@magek/core/fix-skip-unchanged-snapshot_2026-09-28-07-24.json diff --git a/common/changes/@magek/core/fix-skip-unchanged-snapshot_2026-09-28-07-24.json b/common/changes/@magek/core/fix-skip-unchanged-snapshot_2026-09-28-07-24.json new file mode 100644 index 000000000..9ab4bbb5a --- /dev/null +++ b/common/changes/@magek/core/fix-skip-unchanged-snapshot_2026-09-28-07-24.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "comment": "Do not store a snapshot when every pending event is skipped.", + "type": "patch", + "packageName": "@magek/core" + } + ], + "packageName": "@magek/core", + "email": "omidomirzaei@gmail.com" +} \ No newline at end of file diff --git a/packages/core/src/services/event-store.ts b/packages/core/src/services/event-store.ts index 7220a9e87..2d599217e 100644 --- a/packages/core/src/services/event-store.ts +++ b/packages/core/src/services/event-store.ts @@ -56,6 +56,7 @@ export class EventStore { logger.debug(`Looking for the reducer for entity ${entityName} with ID ${entityID}`) // In this assignment we discard the `createdAt` field because it's not needed in the reduction process let newEntitySnapshot: NonPersistedEntitySnapshotEnvelope | undefined = latestSnapshotEnvelope + let snapshotChanged = false for (const pendingEvent of pendingEvents) { // We double-check that what we are reducing is an event if (pendingEvent.kind === 'event') { @@ -64,6 +65,7 @@ export class EventStore { // If reducer returns ReducerAction.Skip, keep the current snapshot unchanged if (reducerResult !== ReducerAction.Skip) { newEntitySnapshot = reducerResult + snapshotChanged = true } } catch (e) { if (e instanceof InvalidEventError) { @@ -85,6 +87,11 @@ export class EventStore { } } + if (!snapshotChanged) { + logger.debug('All pending events were skipped, returning the latest snapshot without storing it') + return latestSnapshotEnvelope + } + if (!newEntitySnapshot) { logger.debug('No snapshot was found or reduced, returning') diff --git a/packages/core/test/services/event-store.test.ts b/packages/core/test/services/event-store.test.ts index 34e1e9a7f..4eea87359 100644 --- a/packages/core/test/services/event-store.test.ts +++ b/packages/core/test/services/event-store.test.ts @@ -757,20 +757,54 @@ describe('EventStore', () => { replace(eventStore, 'loadLatestSnapshot', fake.resolves(someSnapshotEnvelope)) replace(eventStore, 'loadEventStreamSince', fake.resolves([someEventEnvelope])) - // Mock entityReducer to return ReducerAction.Skip - const entityReducer = stub() - .onFirstCall() - .returns(ReducerAction.Skip) + const entityReducer = stub().onFirstCall().returns(ReducerAction.Skip) replace(eventStore, 'entityReducer', entityReducer) - replace(eventStore, 'storeSnapshot', fake.resolves(someSnapshotEnvelope)) + const storedSnapshot = snapshotEnvelopeFor({ id: '99', count: 7 }) + replace(eventStore, 'storeSnapshot', fake.resolves(storedSnapshot)) const entityName = AnEntity.name const entityID = '42' const entity = await eventStore.fetchEntitySnapshot(entityName, entityID) expect(eventStore.entityReducer).to.have.been.calledOnce - expect(eventStore.storeSnapshot).to.have.been.calledOnceWith(someSnapshotEnvelope) - expect(entity).to.be.deep.equal(someSnapshotEnvelope) + expect(eventStore.storeSnapshot).not.to.have.been.called + expect(entity).to.equal(someSnapshotEnvelope) + }) + + it('does not store a snapshot when every pending event is skipped', async () => { + const eventStore = new EventStore(config) as any + const someSnapshotEnvelope = snapshotEnvelopeFor(someEntity) + const firstEventEnvelope = eventEnvelopeFor(someEvent, AnEvent.name, new Date()) + const secondEventEnvelope = eventEnvelopeFor(otherEvent, AnEvent.name, new Date()) + + replace(eventStore, 'loadLatestSnapshot', fake.resolves(someSnapshotEnvelope)) + replace(eventStore, 'loadEventStreamSince', fake.resolves([firstEventEnvelope, secondEventEnvelope])) + + const entityReducer = stub().returns(ReducerAction.Skip) + replace(eventStore, 'entityReducer', entityReducer) + const storedSnapshot = snapshotEnvelopeFor({ id: '99', count: 7 }) + replace(eventStore, 'storeSnapshot', fake.resolves(storedSnapshot)) + + const entity = await eventStore.fetchEntitySnapshot(AnEntity.name, '42') + + expect(eventStore.entityReducer).to.have.been.calledTwice + expect(eventStore.storeSnapshot).not.to.have.been.called + expect(entity).to.equal(someSnapshotEnvelope) + }) + + it('returns undefined without storing when there is no snapshot and every event is skipped', async () => { + const eventStore = new EventStore(config) as any + const someEventEnvelope = eventEnvelopeFor(someEvent, AnEvent.name, new Date()) + + replace(eventStore, 'loadLatestSnapshot', fake.resolves(undefined)) + replace(eventStore, 'loadEventStreamSince', fake.resolves([someEventEnvelope])) + replace(eventStore, 'entityReducer', stub().returns(ReducerAction.Skip)) + replace(eventStore, 'storeSnapshot', fake.resolves(snapshotEnvelopeFor(someEntity))) + + const entity = await eventStore.fetchEntitySnapshot(AnEntity.name, '42') + + expect(eventStore.storeSnapshot).not.to.have.been.called + expect(entity).to.equal(undefined) }) it('keeps the previous snapshot when Skip is returned for an event', async () => { @@ -802,6 +836,32 @@ describe('EventStore', () => { expect(eventStore.storeSnapshot).to.have.been.calledOnceWith(newSnapshot) expect(entity).to.be.deep.equal(newSnapshot) }) + + it('stores the reduced snapshot when a skipped event is followed by a reduced one', async () => { + const eventStore = new EventStore(config) as any + const initialSnapshot = snapshotEnvelopeFor(someEntity) + const firstEventEnvelope = eventEnvelopeFor(someEvent, AnEvent.name, new Date()) + const secondEventEnvelope = eventEnvelopeFor(otherEvent, AnEvent.name, new Date()) + const reducedSnapshot = snapshotEnvelopeFor(new AnEntity('42', 100)) + + replace(eventStore, 'loadLatestSnapshot', fake.resolves(initialSnapshot)) + replace(eventStore, 'loadEventStreamSince', fake.resolves([firstEventEnvelope, secondEventEnvelope])) + + const entityReducer = stub() + .onFirstCall() + .returns(ReducerAction.Skip) + .onSecondCall() + .returns(reducedSnapshot) + replace(eventStore, 'entityReducer', entityReducer) + replace(eventStore, 'storeSnapshot', fake.resolves(reducedSnapshot)) + + const entity = await eventStore.fetchEntitySnapshot(AnEntity.name, '42') + + expect(eventStore.entityReducer).to.have.been.calledTwice + expect(eventStore.entityReducer.getCall(1).args[1]).to.equal(initialSnapshot) + expect(eventStore.storeSnapshot).to.have.been.calledOnceWith(reducedSnapshot) + expect(entity).to.deep.equal(reducedSnapshot) + }) }) })