feat: add forShare and skipLocked selection modifiers to knex loader

packages/entity-database-adapter-knex-testing-utils/src/StubPostgresDatabaseAdapter.ts

@@ -212,7 +212,14 @@ export class StubPostgresDatabaseAdapter<
       tableName,
       tableFieldSingleValueEqualityOperands,
       tableFieldMultiValueEqualityOperands,
-      { orderBy: undefined, offset: undefined, limit: undefined, forUpdate: undefined },
+      {
+        orderBy: undefined,
+        offset: undefined,
+        limit: undefined,
+        forUpdate: undefined,
+        forShare: undefined,
+        skipLocked: undefined,
+      },
     );
     return results.length;
   }

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

@@ -88,9 +88,21 @@ export interface EntityLoaderQuerySelectionModifiers<
 
   /**
    * Lock the selected rows for update using `SELECT ... FOR UPDATE`. The lock is held until the
-   * end of the transaction, so the query context must be transactional.
+   * end of the transaction, so the query context must be transactional. Mutually exclusive with `forShare`.
    */
   forUpdate?: boolean;
+
+  /**
+   * Lock the selected rows in share mode using `SELECT ... FOR SHARE`. The lock is held until the
+   * end of the transaction, so the query context must be transactional. Mutually exclusive with `forUpdate`.
+   */
+  forShare?: boolean;
+
+  /**
+   * Skip rows that are already locked by another transaction using `SKIP LOCKED` instead of waiting for them.
+   * Requires `forUpdate` or `forShare`.
+   */
+  skipLocked?: boolean;
 }
 
 /**

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

@@ -136,9 +136,19 @@ export interface PostgresQuerySelectionModifiers<TFields extends Record<string,
   limit?: number;
 
   /**
-   * Lock the selected rows for update using `SELECT ... FOR UPDATE`.
+   * Lock the selected rows for update using `SELECT ... FOR UPDATE`. Mutually exclusive with `forShare`.
    */
   forUpdate?: boolean;
+
+  /**
+   * Lock the selected rows in share mode using `SELECT ... FOR SHARE`. Mutually exclusive with `forUpdate`.
+   */
+  forShare?: boolean;
+
+  /**
+   * Skip rows that are already locked by another transaction using `SKIP LOCKED`. Requires `forUpdate` or `forShare`.
+   */
+  skipLocked?: boolean;
 }
 
 export type TableOrderByClause<TFields extends Record<string, any>> =
@@ -158,6 +168,8 @@ export interface TableQuerySelectionModifiers<TFields extends Record<string, any
   offset: number | undefined;
   limit: number | undefined;
   forUpdate: boolean | undefined;
+  forShare: boolean | undefined;
+  skipLocked: boolean | undefined;
 }
 
 export abstract class BasePostgresEntityDatabaseAdapter<
@@ -350,6 +362,8 @@ export abstract class BasePostgresEntityDatabaseAdapter<
       offset: querySelectionModifiers.offset,
       limit: querySelectionModifiers.limit,
       forUpdate: querySelectionModifiers.forUpdate,
+      forShare: querySelectionModifiers.forShare,
+      skipLocked: querySelectionModifiers.skipLocked,
     };
   }
 }

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

@@ -23,6 +23,8 @@ export abstract class BaseSQLQueryBuilder<
       offset?: number;
       orderBy?: readonly EntityLoaderOrderByClause<TFields, TSelectedFields>[];
       forUpdate?: boolean;
+      forShare?: boolean;
+      skipLocked?: boolean;
     },
   ) {}
 
@@ -45,12 +47,32 @@ export abstract class BaseSQLQueryBuilder<
   /**
    * Lock the selected rows for update using `SELECT ... FOR UPDATE`. The lock is held until the
    * end of the transaction, so the query must be executed in a transactional query context.
+   * Mutually exclusive with `forShare`.
    */
   forUpdate(): this {
     this.modifiers.forUpdate = true;
     return this;
   }
 
+  /**
+   * Lock the selected rows in share mode using `SELECT ... FOR SHARE`. The lock is held until the
+   * end of the transaction, so the query must be executed in a transactional query context.
+   * Mutually exclusive with `forUpdate`.
+   */
+  forShare(): this {
+    this.modifiers.forShare = true;
+    return this;
+  }
+
+  /**
+   * Skip rows that are already locked by another transaction using `SKIP LOCKED` instead of waiting for them.
+   * Requires `forUpdate()` or `forShare()`.
+   */
+  skipLocked(): this {
+    this.modifiers.skipLocked = true;
+    return this;
+  }
+
   /**
    * Order by a field. Can be called multiple times to add multiple order bys.
    */

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

@@ -116,7 +116,14 @@ export class PostgresEntityDatabaseAdapter<
         tableValue: tableTuple[index],
       })),
       [],
-      { limit: 1, orderBy: undefined, offset: undefined, forUpdate: undefined },
+      {
+        limit: 1,
+        orderBy: undefined,
+        offset: undefined,
+        forUpdate: undefined,
+        forShare: undefined,
+        skipLocked: undefined,
+      },
     );
     return results[0] ?? null;
   }
@@ -125,7 +132,7 @@ export class PostgresEntityDatabaseAdapter<
     query: Knex.QueryBuilder,
     querySelectionModifiers: TableQuerySelectionModifiers<TFields>,
   ): Knex.QueryBuilder {
-    const { orderBy, offset, limit, forUpdate } = querySelectionModifiers;
+    const { orderBy, offset, limit, forUpdate, forShare, skipLocked } = querySelectionModifiers;
 
     let ret = query;
 
@@ -165,6 +172,14 @@ export class PostgresEntityDatabaseAdapter<
       ret = ret.forUpdate();
     }
 
+    if (forShare) {
+      ret = ret.forShare();
+    }
+
+    if (skipLocked) {
+      ret = ret.skipLocked();
+    }
+
     return ret;
   }
 

packages/entity-database-adapter-knex/src/__integration-tests__/PostgresEntityIntegration-test.ts

@@ -211,13 +211,16 @@ describe('postgres entity integration', () => {
     );
   });
 
-  describe('forUpdate', () => {
+  describe('row locking modifiers (forUpdate, forShare, skipLocked)', () => {
     // Attempts to lock the row from a separate connection without waiting. Postgres raises
-    // lock_not_available (55P03) when another transaction already holds a FOR UPDATE lock on the row.
-    const tryLockRowFromOtherConnectionAsync = async (id: string): Promise<string | null> => {
+    // lock_not_available (55P03) when another transaction already holds a conflicting lock on the row.
+    const tryLockRowFromOtherConnectionAsync = async (
+      id: string,
+      lockMode: 'FOR UPDATE' | 'FOR SHARE' = 'FOR UPDATE',
+    ): Promise<string | null> => {
       try {
         await knexInstance.raw(
-          'SELECT * FROM postgres_test_entities WHERE id = ? FOR UPDATE NOWAIT',
+          `SELECT * FROM postgres_test_entities WHERE id = ? ${lockMode} NOWAIT`,
           [id],
         );
         return null;
@@ -353,21 +356,145 @@ describe('postgres entity integration', () => {
       expect(reloaded.getField('name')).toBe('counter,a,b');
     });
 
-    it('throws when forUpdate is used outside of a transaction', async () => {
+    it('forShare allows concurrent share locks but blocks update locks', async () => {
+      const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));
+      const entity = await enforceAsyncResult(
+        PostgresTestEntity.creatorWithAuthorizationResults(vc1)
+          .setField('name', 'shared')
+          .createAsync(),
+      );
+
+      await vc1.runInTransactionForDatabaseAdapterFlavorAsync('postgres', async (queryContext) => {
+        const results = await PostgresTestEntity.knexLoader(
+          vc1,
+          queryContext,
+        ).loadManyByFieldEqualityConjunctionAsync([{ fieldName: 'name', fieldValue: 'shared' }], {
+          forShare: true,
+        });
+        expect(results).toHaveLength(1);
+
+        // another share lock is compatible, an update lock is not
+        expect(await tryLockRowFromOtherConnectionAsync(entity.getID(), 'FOR SHARE')).toBeNull();
+        expect(await tryLockRowFromOtherConnectionAsync(entity.getID(), 'FOR UPDATE')).toBe(
+          '55P03',
+        );
+
+        // via the fluent builder method
+        const builderResults = await PostgresTestEntity.knexLoader(vc1, queryContext)
+          .loadManyBySQL(sql`name = ${'shared'}`)
+          .forShare()
+          .executeAsync();
+        expect(builderResults).toHaveLength(1);
+      });
+
+      expect(await tryLockRowFromOtherConnectionAsync(entity.getID(), 'FOR UPDATE')).toBeNull();
+    });
+
+    it('skipLocked skips rows locked by another transaction', async () => {
+      const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));
+      const entityA = await enforceAsyncResult(
+        PostgresTestEntity.creatorWithAuthorizationResults(vc1)
+          .setField('name', 'queue')
+          .createAsync(),
+      );
+      const entityB = await enforceAsyncResult(
+        PostgresTestEntity.creatorWithAuthorizationResults(vc1)
+          .setField('name', 'queue')
+          .createAsync(),
+      );
+
+      await vc1.runInTransactionForDatabaseAdapterFlavorAsync(
+        'postgres',
+        async (outerQueryContext) => {
+          // lock entityA in the outer transaction
+          const locked = await PostgresTestEntity.knexLoader(vc1, outerQueryContext)
+            .loadManyBySQL(sql`id = ${entityA.getID()}`)
+            .forUpdate()
+            .executeAsync();
+          expect(locked.map((e) => e.getID())).toEqual([entityA.getID()]);
+
+          // a second transaction on another connection sees only the unlocked row
+          const vc2 = new ViewerContext(
+            createKnexIntegrationTestEntityCompanionProvider(knexInstance),
+          );
+          await vc2.runInTransactionForDatabaseAdapterFlavorAsync(
+            'postgres',
+            async (innerQueryContext) => {
+              const skipLockedResults = await PostgresTestEntity.knexLoader(
+                vc2,
+                innerQueryContext,
+              ).loadManyByFieldEqualityConjunctionAsync(
+                [{ fieldName: 'name', fieldValue: 'queue' }],
+                { forUpdate: true, skipLocked: true },
+              );
+              expect(skipLockedResults.map((e) => e.getID())).toEqual([entityB.getID()]);
+
+              const builderResults = await PostgresTestEntity.knexLoader(vc2, innerQueryContext)
+                .loadManyBySQL(sql`name = ${'queue'}`)
+                .forShare()
+                .skipLocked()
+                .executeAsync();
+              expect(builderResults.map((e) => e.getID())).toEqual([entityB.getID()]);
+            },
+          );
+        },
+      );
+
+      // both rows are visible once the outer transaction commits
+      await vc1.runInTransactionForDatabaseAdapterFlavorAsync('postgres', async (queryContext) => {
+        const results = await PostgresTestEntity.knexLoader(
+          vc1,
+          queryContext,
+        ).loadManyByFieldEqualityConjunctionAsync([{ fieldName: 'name', fieldValue: 'queue' }], {
+          forUpdate: true,
+          skipLocked: true,
+        });
+        expect(results).toHaveLength(2);
+      });
+    });
+
+    it('throws when forUpdate or forShare is used outside of a transaction', async () => {
       const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));
 
       await expect(
         PostgresTestEntity.knexLoader(vc1).loadManyByFieldEqualityConjunctionAsync([], {
           forUpdate: true,
         }),
-      ).rejects.toThrow('forUpdate requires a transactional query context');
+      ).rejects.toThrow('require a transactional query context');
+
+      await expect(
+        PostgresTestEntity.knexLoader(vc1).loadManyByFieldEqualityConjunctionAsync([], {
+          forShare: true,
+        }),
+      ).rejects.toThrow('require a transactional query context');
 
       await expect(
         PostgresTestEntity.knexLoader(vc1)
           .loadManyBySQL(sql`TRUE`)
           .forUpdate()
           .executeAsync(),
-      ).rejects.toThrow('forUpdate requires a transactional query context');
+      ).rejects.toThrow('require a transactional query context');
+    });
+
+    it('throws for invalid row locking modifier combinations', async () => {
+      const vc1 = new ViewerContext(createKnexIntegrationTestEntityCompanionProvider(knexInstance));
+
+      await vc1.runInTransactionForDatabaseAdapterFlavorAsync('postgres', async (queryContext) => {
+        await expect(
+          PostgresTestEntity.knexLoader(vc1, queryContext)
+            .loadManyBySQL(sql`TRUE`)
+            .forUpdate()
+            .forShare()
+            .executeAsync(),
+        ).rejects.toThrow('forUpdate and forShare are mutually exclusive');
+
+        await expect(
+          PostgresTestEntity.knexLoader(vc1, queryContext)
+            .loadManyBySQL(sql`TRUE`)
+            .skipLocked()
+            .executeAsync(),
+        ).rejects.toThrow('skipLocked requires forUpdate or forShare');
+      });
     });
   });
 

packages/entity-database-adapter-knex/src/__tests__/BasePostgresEntityDatabaseAdapter-test.ts

@@ -164,19 +164,32 @@ describe(BasePostgresEntityDatabaseAdapter, () => {
       expect(results).toEqual([{ stringField: 'hello' }]);
     });
 
-    it('converts query selection modifiers including forUpdate', async () => {
+    it('converts query selection modifiers including row locking modifiers', async () => {
       const queryContext = instance(mock(EntityQueryContext));
       const adapter = new TestEntityDatabaseAdapter({});
       await adapter.fetchManyByFieldEqualityConjunctionAsync(queryContext, [], {
         limit: 2,
         offset: 1,
         forUpdate: true,
+        skipLocked: true,
       });
       expect(adapter.lastEqualityConditionQuerySelectionModifiers).toEqual({
         orderBy: undefined,
         limit: 2,
         offset: 1,
         forUpdate: true,
+        forShare: undefined,
+        skipLocked: true,
+      });
+
+      await adapter.fetchManyByFieldEqualityConjunctionAsync(queryContext, [], { forShare: true });
+      expect(adapter.lastEqualityConditionQuerySelectionModifiers).toEqual({
+        orderBy: undefined,
+        limit: undefined,
+        offset: undefined,
+        forUpdate: undefined,
+        forShare: true,
+        skipLocked: undefined,
       });
 
       await adapter.fetchManyByFieldEqualityConjunctionAsync(queryContext, [], {});
@@ -185,20 +198,27 @@ describe(BasePostgresEntityDatabaseAdapter, () => {
         limit: undefined,
         offset: undefined,
         forUpdate: undefined,
+        forShare: undefined,
+        skipLocked: undefined,
       });
     });
   });
 
   describe('fetchManyBySQLFragmentAsync', () => {
-    it('converts query selection modifiers including forUpdate', async () => {
+    it('converts query selection modifiers including row locking modifiers', async () => {
       const queryContext = instance(mock(EntityQueryContext));
       const adapter = new TestEntityDatabaseAdapter({});
-      await adapter.fetchManyBySQLFragmentAsync(queryContext, sql`TRUE`, { forUpdate: true });
+      await adapter.fetchManyBySQLFragmentAsync(queryContext, sql`TRUE`, {
+        forShare: true,
+        skipLocked: true,
+      });
       expect(adapter.lastSQLFragmentQuerySelectionModifiers).toEqual({
         orderBy: undefined,
         limit: undefined,
         offset: undefined,
-        forUpdate: true,
+        forUpdate: undefined,
+        forShare: true,
+        skipLocked: true,
       });
     });
   });