fix: move cache invalidation to after transaction commit (#77)

packages/entity-database-adapter-knex/src/PostgresEntityQueryContextProvider.ts

@@ -1,25 +1,21 @@
-import {
-  IEntityQueryContextProvider,
-  EntityNonTransactionalQueryContext,
-  EntityTransactionalQueryContext,
-} from '@expo/entity';
+import { EntityQueryContextProvider } from '@expo/entity';
 import Knex from 'knex';
 
 /**
  * Query context provider for knex (postgres).
  */
-export default class PostgresEntityQueryContextProvider implements IEntityQueryContextProvider {
-  constructor(private readonly knexInstance: Knex) {}
+export default class PostgresEntityQueryContextProvider extends EntityQueryContextProvider {
+  constructor(private readonly knexInstance: Knex) {
+    super();
+  }
 
-  getQueryContext(): EntityNonTransactionalQueryContext {
-    return new EntityNonTransactionalQueryContext(this.knexInstance, this);
+  protected getQueryInterface(): any {
+    return this.knexInstance;
   }
 
-  async runInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
-  ): Promise<T> {
-    return await this.knexInstance.transaction(async (trx) => {
-      return await transactionScope(new EntityTransactionalQueryContext(trx));
-    });
+  protected createTransactionRunner<T>(): (
+    transactionScope: (trx: any) => Promise<T>
+  ) => Promise<T> {
+    return (transactionScope) => this.knexInstance.transaction(transactionScope);
   }
 }

packages/entity-database-adapter-knex/src/testfixtures/PostgresTriggerTestEntity.ts

@@ -11,6 +11,7 @@ import {
   Entity,
   EntityMutationTrigger,
   EntityQueryContext,
+  EntityNonTransactionalMutationTrigger,
 } from '@expo/entity';
 import Knex from 'knex';
 
@@ -118,6 +119,26 @@ class ThrowConditionallyTrigger extends EntityMutationTrigger<
   }
 }
 
+class ThrowConditionallyNonTransactionalTrigger extends EntityNonTransactionalMutationTrigger<
+  PostgresTriggerTestEntityFields,
+  string,
+  ViewerContext,
+  PostgresTriggerTestEntity
+> {
+  constructor(private fieldName: keyof PostgresTriggerTestEntityFields, private badValue: string) {
+    super();
+  }
+
+  async executeAsync(
+    _viewerContext: ViewerContext,
+    entity: PostgresTriggerTestEntity
+  ): Promise<void> {
+    if (entity.getField(this.fieldName) === this.badValue) {
+      throw new Error(`${this.fieldName} cannot have value ${this.badValue}`);
+    }
+  }
+}
+
 export const postgresTestEntityConfiguration = new EntityConfiguration<
   PostgresTriggerTestEntityFields
 >({
@@ -149,6 +170,6 @@ const postgresTestEntityCompanionDefinition = new EntityCompanionDefinition({
     afterDelete: [new ThrowConditionallyTrigger('name', 'afterDelete')],
     beforeAll: [new ThrowConditionallyTrigger('name', 'beforeAll')],
     afterAll: [new ThrowConditionallyTrigger('name', 'afterAll')],
-    afterCommit: [new ThrowConditionallyTrigger('name', 'afterCommit')],
+    afterCommit: [new ThrowConditionallyNonTransactionalTrigger('name', 'afterCommit')],
   },
 });

packages/entity-example/src/adapters/InMemoryQueryContextProvider.ts

@@ -1,17 +1,13 @@
-import {
-  IEntityQueryContextProvider,
-  EntityNonTransactionalQueryContext,
-  EntityTransactionalQueryContext,
-} from '@expo/entity';
+import { EntityQueryContextProvider } from '@expo/entity';
 
-export default class InMemoryQueryContextProvider implements IEntityQueryContextProvider {
-  getQueryContext(): EntityNonTransactionalQueryContext {
-    return new EntityNonTransactionalQueryContext({}, this);
+export default class InMemoryQueryContextProvider extends EntityQueryContextProvider {
+  protected getQueryInterface(): any {
+    return {};
   }
 
-  async runInTransactionAsync<T>(
-    transactionScope: (queryContext: EntityTransactionalQueryContext) => Promise<T>
-  ): Promise<T> {
-    return await transactionScope(new EntityTransactionalQueryContext({}));
+  protected createTransactionRunner<T>(): (
+    transactionScope: (queryInterface: any) => Promise<T>
+  ) => Promise<T> {
+    return (transactionScope) => Promise.resolve(transactionScope({}));
   }
 }

packages/entity-full-integration-tests/CHANGELOG.md

@@ -0,0 +1,5 @@
+# Change Log
+
+All notable changes to this project will be documented in this file.
+See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
+

packages/entity-full-integration-tests/README.md

@@ -0,0 +1,5 @@
+# `@expo/entity-full-integration-tests`
+
+Full integration tests for testing combinations of database and cache adapters.
+
+[Documentation](https://expo.github.io/entity/modules/_expo_entity_full_integration_tests.html)

packages/entity-full-integration-tests/package.json

@@ -0,0 +1,33 @@
+{
+  "name": "@expo/entity-full-integration-tests",
+  "private": true,
+  "version": "0.6.0",
+  "description": "Full redis and knex integration tests for the entity framework",
+  "scripts": {
+    "tsc": "tsc",
+    "clean": "rm -rf build coverage coverage-integration",
+    "lint": "eslint src --ext '.ts'",
+    "test": "jest --rootDir --config ../../resources/jest.config.js --passWithNoTests",
+    "integration": "../../resources/run-with-docker yarn integration-no-setup",
+    "integration-no-setup": "jest --config ../../resources/jest-integration.config.js --rootDir --runInBand --passWithNoTests",
+    "barrelsby": "barrelsby --directory src --location top --exclude tests__ --singleQuotes --exportDefault --delete"
+  },
+  "engines": {
+    "node": ">=12"
+  },
+  "keywords": [
+    "entity"
+  ],
+  "author": "Expo",
+  "license": "MIT",
+  "peerDependencies": {
+    "@expo/entity": "*",
+    "@expo/entity-cache-adapter-redis": "*",
+    "@expo/entity-database-adapter-knex": "*"
+  },
+  "devDependencies": {
+    "@expo/entity": "^0.6.0",
+    "@expo/entity-cache-adapter-redis": "^0.6.0",
+    "@expo/entity-database-adapter-knex": "^0.6.0"
+  }
+}

packages/entity-full-integration-tests/src/__integration-tests__/EntityCacheInconsistency-test.ts

@@ -0,0 +1,205 @@
+import {
+  EntityPrivacyPolicy,
+  ViewerContext,
+  AlwaysAllowPrivacyPolicyRule,
+  Entity,
+  EntityCompanionDefinition,
+  EntityConfiguration,
+  DatabaseAdapterFlavor,
+  CacheAdapterFlavor,
+  UUIDField,
+} from '@expo/entity';
+import { RedisCacheAdapterContext } from '@expo/entity-cache-adapter-redis';
+import Redis from 'ioredis';
+import Knex from 'knex';
+import { URL } from 'url';
+
+import { createFullIntegrationTestEntityCompanionProvider } from '../testfixtures/createFullIntegrationTestEntityCompanionProvider';
+
+interface TestFields {
+  id: string;
+  other_string: string;
+  third_string: string;
+}
+
+class TestEntityPrivacyPolicy extends EntityPrivacyPolicy<
+  TestFields,
+  string,
+  ViewerContext,
+  TestEntity
+> {
+  protected readonly readRules = [new AlwaysAllowPrivacyPolicyRule()];
+  protected readonly createRules = [new AlwaysAllowPrivacyPolicyRule()];
+  protected readonly updateRules = [new AlwaysAllowPrivacyPolicyRule()];
+  protected readonly deleteRules = [new AlwaysAllowPrivacyPolicyRule()];
+}
+
+class TestEntity extends Entity<TestFields, string, ViewerContext> {
+  static getCompanionDefinition(): EntityCompanionDefinition<
+    TestFields,
+    string,
+    ViewerContext,
+    TestEntity,
+    TestEntityPrivacyPolicy
+  > {
+    return testEntityCompanion;
+  }
+}
+
+const testEntityConfiguration = new EntityConfiguration<TestFields>({
+  idField: 'id',
+  tableName: 'testentities',
+  schema: {
+    id: new UUIDField({
+      columnName: 'id',
+      cache: true,
+    }),
+    other_string: new UUIDField({
+      columnName: 'other_string',
+      cache: true,
+    }),
+    third_string: new UUIDField({
+      columnName: 'third_string',
+    }),
+  },
+  databaseAdapterFlavor: DatabaseAdapterFlavor.POSTGRES,
+  cacheAdapterFlavor: CacheAdapterFlavor.REDIS,
+});
+
+const testEntityCompanion = new EntityCompanionDefinition({
+  entityClass: TestEntity,
+  entityConfiguration: testEntityConfiguration,
+  privacyPolicyClass: TestEntityPrivacyPolicy,
+});
+
+async function createOrTruncatePostgresTables(knex: Knex): Promise<void> {
+  await knex.raw('CREATE EXTENSION IF NOT EXISTS "uuid-ossp"'); // for uuid_generate_v4()
+
+  await knex.schema.createTable('testentities', (table) => {
+    table.uuid('id').defaultTo(knex.raw('uuid_generate_v4()')).primary();
+    table.string('other_string').notNullable();
+    table.string('third_string').notNullable();
+  });
+  await knex.into('testentities').truncate();
+}
+
+async function dropPostgresTable(knex: Knex): Promise<void> {
+  if (await knex.schema.hasTable('children')) {
+    await knex.schema.dropTable('children');
+  }
+  if (await knex.schema.hasTable('parents')) {
+    await knex.schema.dropTable('parents');
+  }
+}
+
+describe('Entity cache inconsistency', () => {
+  let knexInstance: Knex;
+  let redisCacheAdapterContext: RedisCacheAdapterContext;
+
+  beforeAll(() => {
+    knexInstance = Knex({
+      client: 'pg',
+      connection: {
+        user: process.env.PGUSER,
+        password: process.env.PGPASSWORD,
+        host: 'localhost',
+        port: parseInt(process.env.PGPORT!, 10),
+        database: process.env.PGDATABASE,
+      },
+    });
+    redisCacheAdapterContext = {
+      redisClient: new Redis(new URL(process.env.REDIS_URL!).toString()),
+      makeKeyFn(...parts: string[]): string {
+        const delimiter = ':';
+        const escapedParts = parts.map((part) =>
+          part.replace('\\', '\\\\').replace(delimiter, `\\${delimiter}`)
+        );
+        return escapedParts.join(delimiter);
+      },
+      cacheKeyPrefix: 'test-',
+      cacheKeyVersion: 1,
+      ttlSecondsPositive: 86400, // 1 day
+      ttlSecondsNegative: 600, // 10 minutes
+    };
+  });
+
+  beforeEach(async () => {
+    await createOrTruncatePostgresTables(knexInstance);
+    await redisCacheAdapterContext.redisClient.flushdb();
+  });
+
+  afterAll(async () => {
+    await dropPostgresTable(knexInstance);
+    knexInstance.destroy();
+    redisCacheAdapterContext.redisClient.disconnect();
+  });
+
+  test('lots of updates in long-ish running transactions', async () => {
+    const viewerContext = new ViewerContext(
+      createFullIntegrationTestEntityCompanionProvider(knexInstance, redisCacheAdapterContext)
+    );
+
+    const entity1 = await TestEntity.creator(viewerContext)
+      .setField('other_string', 'hello')
+      .setField('third_string', 'initial')
+      .enforceCreateAsync();
+
+    // put entities in cache and dataloader
+    await TestEntity.loader(viewerContext).enforcing().loadByIDAsync(entity1.getID());
+    await TestEntity.loader(viewerContext)
+      .enforcing()
+      .loadByFieldEqualingAsync('other_string', 'hello');
+
+    let openBarrier1: () => void;
+    const barrier1 = new Promise((resolve) => {
+      openBarrier1 = resolve;
+    });
+
+    let openBarrier2: () => void;
+    const barrier2 = new Promise((resolve) => {
+      openBarrier2 = resolve;
+    });
+
+    await Promise.all([
+      // do a load after the transaction below updates the entity but before transaction commits to ensure
+      // that the cache is cleared after the transaction commits rather than in the middle where the changes
+      // may not be visible by other requests (ViewerContexts) which would cache the incorrect
+      // value during the read-through cache.
+      (async () => {
+        await barrier1;
+
+        const viewerContextInternal = new ViewerContext(
+          createFullIntegrationTestEntityCompanionProvider(knexInstance, redisCacheAdapterContext)
+        );
+        await TestEntity.loader(viewerContextInternal).enforcing().loadByIDAsync(entity1.getID());
+
+        openBarrier2!();
+      })(),
+      TestEntity.runInTransactionAsync(viewerContext, async (queryContext) => {
+        await TestEntity.updater(entity1, queryContext)
+          .setField('third_string', 'updated')
+          .enforceUpdateAsync();
+
+        openBarrier1();
+
+        // wait for to ensure the transaction isn't committed until after load above occurs
+        await barrier2;
+      }),
+    ]);
+
+    // ensure cache consistency
+    const viewerContextLast = new ViewerContext(
+      createFullIntegrationTestEntityCompanionProvider(knexInstance, redisCacheAdapterContext)
+    );
+
+    const loadedById = await TestEntity.loader(viewerContextLast)
+      .enforcing()
+      .loadByIDAsync(entity1.getID());
+    const loadedByField = await TestEntity.loader(viewerContextLast)
+      .enforcing()
+      .loadByFieldEqualingAsync('other_string', 'hello');
+
+    expect(loadedById.getField('third_string')).toEqual('updated');
+    expect(loadedByField!.getField('third_string')).toEqual('updated');
+  });
+});