[FSSDK-12028] add ttl to events in event store (#1103)

Add configurable ttl support to stored events. This will be helpful if for some reason, events are failing permanently (for example, the event enpoint is blocked by adblocker/isp).

Also, refactored the code to move the responsibility of limiting the number of saved event to the store implementation from the event processor.

Author: raju-opti

lib/event_processor/batch_event_processor.react_native.spec.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { vi, describe, it, expect, beforeEach } from 'vitest';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
 
 const mockNetInfo = vi.hoisted(() => {
   const netInfo = {
@@ -45,13 +45,13 @@ vi.mock('@react-native-community/netinfo', () => {
   };
 });
 
-import { ReactNativeNetInfoEventProcessor } from './batch_event_processor.react_native';
-import { getMockRepeater } from '../tests/mock/mock_repeater';
 import { getMockAsyncCache } from '../tests/mock/mock_cache';
+import { getMockRepeater } from '../tests/mock/mock_repeater';
+import { ReactNativeNetInfoEventProcessor } from './batch_event_processor.react_native';
 
+import { createImpressionEvent } from '../tests/mock/create_event';
 import { EventWithId } from './batch_event_processor';
 import { buildLogEvent } from './event_builder/log_event';
-import { createImpressionEvent } from '../tests/mock/create_event';
 import { ProcessableEvent } from './event_processor';
 
 const getMockDispatcher = () => {
@@ -77,6 +77,10 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     const dispatchRepeater = getMockRepeater();
     const failedEventRepeater = getMockRepeater();
 
+        // fail the first dispatch attempt that is made on start
+    eventDispatcher.dispatchEvent = vi.fn().mockResolvedValueOnce({ statusCode: 500 })
+      .mockResolvedValue({ statusCode: 200 });
+
     const cache = getMockAsyncCache<EventWithId>();
     const events: ProcessableEvent[] = [];
 
@@ -98,18 +102,27 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     processor.start();
     await processor.onRunning();
 
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
+
     mockNetInfo.pushState(true);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
 
     mockNetInfo.pushState(true);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
   });
 
-  it('should retry failed events when network becomes reachable', async () => {
+  it('should retry failed events immediately when network becomes reachable', async () => {
     const eventDispatcher = getMockDispatcher();
     const dispatchRepeater = getMockRepeater();
     const failedEventRepeater = getMockRepeater();
 
+    // fail the first dispatch attempt that is made on start
+    eventDispatcher.dispatchEvent = vi.fn().mockResolvedValueOnce({ statusCode: 500 })
+      .mockResolvedValue({ statusCode: 200 });
+
     const cache = getMockAsyncCache<EventWithId>();
     const events: ProcessableEvent[] = [];
 
@@ -131,14 +144,17 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     processor.start();
     await processor.onRunning();
 
-    mockNetInfo.pushState(false);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
 
-    mockNetInfo.pushState(true);
+    mockNetInfo.pushState(false);
+    mockNetInfo.pushState(true); 
 
     await exhaustMicrotasks();
 
-    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledWith(buildLogEvent(events));
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
+    expect(eventDispatcher.dispatchEvent).toHaveBeenNthCalledWith(2, buildLogEvent(events));
   });
 
   it('should unsubscribe from netinfo listener when stopped', async () => {

lib/event_processor/batch_event_processor.spec.ts

@@ -368,124 +368,6 @@ describe('BatchEventProcessor', async () => {
       expect(events).toEqual(eventsInStore);
     });
 
-    it('should not store the event in the eventStore but still dispatch if the \
-        number of pending events is greater than the limit', async () => {
-      const eventDispatcher = getMockDispatcher();
-      const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;
-      mockDispatch.mockResolvedValue(resolvablePromise().promise);
-
-      const eventStore = getMockSyncCache<EventWithId>();
-
-      const idGenerator = new IdGenerator();
-
-      for (let i = 0; i < 505; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        const cacheId = idGenerator.getId();
-        await eventStore.set(cacheId, { id: cacheId, event });
-      }
-
-      expect(eventStore.size()).toEqual(505);
-
-      const processor = new BatchEventProcessor({
-        eventDispatcher,
-        dispatchRepeater: getMockRepeater(),
-        batchSize: 1,
-        eventStore,
-      });
-
-      processor.start();
-      await processor.onRunning();
-
-      const events: ProcessableEvent[] = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(505);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(507);
-      expect(eventDispatcher.dispatchEvent.mock.calls[505][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[506][0]).toEqual(buildLogEvent([events[1]]));
-    });
-
-    it('should store events in the eventStore when the number of events in the store\
-        becomes lower than the limit', async () => {
-      const eventDispatcher = getMockDispatcher();
-
-      const dispatchResponses: ResolvablePromise<any>[] = [];
-
-      const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;
-      mockDispatch.mockImplementation((arg) => {
-        const dispatchResponse = resolvablePromise();
-        dispatchResponses.push(dispatchResponse);
-        return dispatchResponse.promise;
-      });
-
-      const eventStore = getMockSyncCache<EventWithId>();
-
-      const idGenerator = new IdGenerator();
-
-      for (let i = 0; i < 502; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        const cacheId = String(i);
-        await eventStore.set(cacheId, { id: cacheId, event });
-      }
-
-      expect(eventStore.size()).toEqual(502);
-
-      const processor = new BatchEventProcessor({
-        eventDispatcher,
-        dispatchRepeater: getMockRepeater(),
-        batchSize: 1,
-        eventStore,
-      });
-
-      processor.start();
-      await processor.onRunning();
-
-      let events: ProcessableEvent[] = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i + 502}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(502);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(504);
-
-      expect(eventDispatcher.dispatchEvent.mock.calls[502][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[503][0]).toEqual(buildLogEvent([events[1]]));
-
-      // resolve the dispatch for events not saved in the store
-      dispatchResponses[502].resolve({ statusCode: 200 });
-      dispatchResponses[503].resolve({ statusCode: 200 });
-
-      await exhaustMicrotasks();
-      expect(eventStore.size()).toEqual(502);
-
-      // resolve the dispatch for 3 events in store, making the store size 499 which is lower than the limit
-      dispatchResponses[0].resolve({ statusCode: 200 });
-      dispatchResponses[1].resolve({ statusCode: 200 });
-      dispatchResponses[2].resolve({ statusCode: 200 });
-
-      await exhaustMicrotasks();
-      expect(eventStore.size()).toEqual(499);
-
-      // process 2 more events
-      events = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i + 504}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(500);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(506);
-      expect(eventDispatcher.dispatchEvent.mock.calls[504][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[505][0]).toEqual(buildLogEvent([events[1]]));      
-    });
-
     it('should still dispatch events even if the store save fails', async () => {
       const eventDispatcher = getMockDispatcher();
       const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;

lib/event_processor/batch_event_processor.ts

@@ -31,7 +31,6 @@ import { FAILED_TO_DISPATCH_EVENTS, SERVICE_NOT_RUNNING } from "error_message";
 import { OptimizelyError } from "../error/optimizly_error";
 import { sprintf } from "../utils/fns";
 import { SERVICE_STOPPED_BEFORE_RUNNING } from "../service";
-import { EVENT_STORE_FULL } from "../message/log_message";
 
 export const DEFAULT_MIN_BACKOFF = 1000;
 export const DEFAULT_MAX_BACKOFF = 32000;
@@ -40,7 +39,6 @@ export const MAX_EVENTS_IN_STORE = 500;
 export type EventWithId = {
   id: string;
   event: ProcessableEvent;
-  notStored?: boolean;
 };
 
 export type RetryConfig = {
@@ -73,9 +71,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
   private eventQueue: EventWithId[] = [];
   private batchSize: number;
   private eventStore?: Store<EventWithId>;
-  private eventCountInStore: Maybe<number> = undefined;
-  private eventCountWaitPromise: Promise<unknown> = Promise.resolve();
-  private maxEventsInStore: number = MAX_EVENTS_IN_STORE;
   private dispatchRepeater: Repeater;
   private failedEventRepeater?: Repeater;
   private idGenerator: IdGenerator = new IdGenerator();
@@ -96,7 +91,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     this.dispatchRepeater = config.dispatchRepeater;
     this.dispatchRepeater.setTask(() => this.flush());
 
-    this.maxEventsInStore = Math.max(2 * config.batchSize, MAX_EVENTS_IN_STORE);
     this.failedEventRepeater = config.failedEventRepeater;
     this.failedEventRepeater?.setTask(() => this.retryFailedEvents());
     if (config.logger) {
@@ -215,9 +209,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     runResult.result.then((res) => {
       events.forEach((event) => {
         this.eventStore?.remove(event.id);
-        if (!event.notStored && this.eventCountInStore) {
-          this.eventCountInStore--;
-        }
       });
       return Promise.resolve();
     }).catch((err) => {
@@ -250,7 +241,10 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
       event: event,
     };
 
-    await this.storeEvent(eventWithId);
+    // store in eventStore first to avoid data loss
+    await Promise.resolve(this.eventStore?.set(eventWithId.id, eventWithId)).catch((e) => {
+      this.logger?.error(e);
+    });
 
     if (this.eventQueue.length > 0 && !areEventContextsEqual(this.eventQueue[0].event, event)) {
       this.flush();
@@ -265,46 +259,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     }
   }
 
-  private async readEventCountInStore(store: Store<EventWithId>): Promise<void> {
-    if (this.eventCountInStore !== undefined) {
-      return;
-    }
-    
-    try {
-      const keys = await store.getKeys();
-      this.eventCountInStore = keys.length;
-    } catch (e) {
-      this.logger?.error(e);
-    }
-  }
-
-  private async findEventCountInStore(): Promise<unknown> {
-    if (this.eventStore && this.eventCountInStore === undefined) {
-      const store = this.eventStore;
-      this.eventCountWaitPromise = this.eventCountWaitPromise.then(() => this.readEventCountInStore(store));
-      return this.eventCountWaitPromise;
-    }
-    return Promise.resolve();
-  }
-
-  private async storeEvent(eventWithId: EventWithId): Promise<void> {
-    await this.findEventCountInStore();
-    if (this.eventCountInStore !== undefined && this.eventCountInStore >= this.maxEventsInStore) {
-      this.logger?.info(EVENT_STORE_FULL, eventWithId.event.uuid);
-      eventWithId.notStored = true;
-      return;
-    }
-
-    await Promise.resolve(this.eventStore?.set(eventWithId.id, eventWithId)).then(() => {
-      if (this.eventCountInStore !== undefined) {
-        this.eventCountInStore++;
-      }
-    }).catch((e) => {
-      eventWithId.notStored = true;
-      this.logger?.error(e);
-    });
-  }
-
   start(): void {
     if (!this.isNew()) {
       return;