diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts index e1a60f9f05..7086d90cbf 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.test.ts @@ -210,7 +210,7 @@ describe('DatabaseEventBusStore', () => { // in a real-world scenario given our count window size is 10,000. await db.raw(` INSERT INTO event_bus_events (id, created_by, topic, data_json, notified_subscribers) - SELECT id, 'abc', CONCAT('test-', id), '{}', '{"${String( + SELECT id, 'abc', CONCAT('test-', MOD(id, 10)), CONCAT('{"payload":{"id":"', id, '"}}'), '{"${String( Math.random(), ).slice(2, 6)}"}' FROM generate_series(1, ${COUNT}) AS id @@ -219,18 +219,26 @@ describe('DatabaseEventBusStore', () => { id: 'tester', created_by: 'abc', read_until: 0, - topics: ['test-1000'], + topics: ['test-5'], }); const start = Date.now(); const { events } = await store.readSubscription('tester'); const duration = Date.now() - start; - expect(events).toEqual([{ topic: 'test-1000', eventPayload: undefined }]); + expect(events).toEqual([ + { topic: 'test-5', eventPayload: { id: '5' } }, + { topic: 'test-5', eventPayload: { id: '15' } }, + { topic: 'test-5', eventPayload: { id: '25' } }, + { topic: 'test-5', eventPayload: { id: '35' } }, + { topic: 'test-5', eventPayload: { id: '45' } }, + { topic: 'test-5', eventPayload: { id: '55' } }, + { topic: 'test-5', eventPayload: { id: '65' } }, + { topic: 'test-5', eventPayload: { id: '75' } }, + { topic: 'test-5', eventPayload: { id: '85' } }, + { topic: 'test-5', eventPayload: { id: '95' } }, + ]); - // When not run in an optimized way this takes anywhere from 10ms to - // 100ms. When optimized it's closed to 1ms, but we leave a bit of room - // for CI load spikes. expect(duration).toBeLessThan(20); }, ); diff --git a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts index b1f26ab841..107dabfc27 100644 --- a/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts +++ b/plugins/events-backend/src/service/hub/DatabaseEventBusStore.ts @@ -472,15 +472,6 @@ export class DatabaseEventBusStore implements EventBusStore { // an update, reads events for the subscription up to the limit, and then // updates the pointer to the last read event. // - // The query for selected_events is written in this particular way to force - // evaluation of the notified subscribers last. Without this, the query - // planner loves to first do a sequential scan of the events table to filter - // out the events that have already been consumed, but that is often a small - // subset and extremely wasteful. We instead want to make sure that the - // query is executed such that we select the events that are relevant to the - // subscription first, and then filter out any events that have already been - // consumed last. - // // This is written as a plain SQL query to spare us all the horrors of // expressing this in knex. @@ -499,13 +490,8 @@ export class DatabaseEventBusStore implements EventBusStore { FROM event_bus_events INNER JOIN subscription ON event_bus_events.topic = ANY(subscription.topics) - WHERE ( - CASE WHEN event_bus_events.id > subscription.read_until THEN - CASE WHEN NOT :id = ANY(event_bus_events.notified_subscribers) - THEN 1 - END - END - ) = 1 + WHERE event_bus_events.id > subscription.read_until + AND NOT :id = ANY(event_bus_events.notified_subscribers) ORDER BY event_bus_events.id ASC LIMIT :limit ), last_event_id AS (