issue 879: add composite index constraint to block concurrent inserts…

… of the same item to actions table

.gitignore

@@ -119,3 +119,4 @@ yalc.lock
 # IDE
 .idea/
 .envrc
+.vscode

packages/indexer-agent/src/db/migrations/12-add-actions-composite-constraint.ts

@@ -0,0 +1,32 @@
+import { Logger } from '@graphprotocol/common-ts'
+import { QueryInterface } from 'sequelize'
+
+interface MigrationContext {
+  queryInterface: QueryInterface
+  logger: Logger
+}
+
+interface Context {
+  context: MigrationContext
+}
+
+export async function up({ context }: Context): Promise<void> {
+  const { queryInterface, logger } = context
+
+  logger.info('Adding composite unique constraint to Actions table')
+  // Add the new composite primary key
+  await queryInterface.addConstraint('Actions', {
+    fields: ['deploymentID', 'source'],
+    type: 'unique',
+    name: 'Actions_ckey_unique_deploymentID_source',
+  })
+}
+
+export async function down({ context }: Context): Promise<void> {
+  const { queryInterface, logger } = context
+  logger.info('Removing composite uniqie constraint from Actions table')
+  await queryInterface.removeConstraint(
+    'Actions',
+    'Actions_ckey_unique_deploymentID_source',
+  )
+}

packages/indexer-common/src/indexer-management/__tests__/resolvers/actions.test.ts

@@ -25,7 +25,7 @@ import {
   OrderDirection,
   QueryFeeModels,
 } from '@graphprotocol/indexer-common'
-import { CombinedError } from '@urql/core'
+import { CombinedError, OperationResult } from '@urql/core'
 import { GraphQLError } from 'graphql'
 import {
   createTestManagementClient,
@@ -532,28 +532,51 @@ describe('Actions', () => {
         .toPromise(),
     ).resolves.toHaveProperty('data.actions', [])
   })
-
-  test('Rejects duplicate queued actions', async () => {
+
+  test('Rejects duplicate actions', async function rejectsDuplicateActions() {
+    console.log('running Rejects duplicate actions test')
+    class ResultWrapper {
+      constructor(
+        public actionNum: number,
+        public inner: OperationResult<Action>,
+      ) {}
+    }
     const inputAction = queuedAllocateAction
-    const mutations: Promise<any>[] = []
-    for (let i = 0; i < 50; i++) {
-      mutations.push(client.mutation(QUEUE_ACTIONS_MUTATION, { actions: [inputAction] }).toPromise())
+    const mutations: Promise<ResultWrapper>[] = []
+    for (let i = 0; i < 5; i++) {
+      mutations.push(
+        (async function pushMutation() {
+          return new ResultWrapper(
+            i,
+            await client
+              .mutation(QUEUE_ACTIONS_MUTATION, { actions: [inputAction] })
+              .toPromise(),
+          )
+        })(),
+      )
     }
 
     const results = await Promise.all(mutations)
-    for (const result of results) {
-      expect(result.data).not.toBeNull()
+    for (const actionResult of results) {
+      expect(actionResult).not.toBeNull()
+      if (actionResult.inner.data === null) {
+        expect(actionResult.inner.error).not.toBeNull()
+        console.log(
+          `action ${actionResult.actionNum} result.error:`,
+          actionResult.inner.error,
+        )
+      } else {
+        expect(actionResult.inner.data).not.toBeNull()
+      }
     }
 
     const actions = await client
       .query(ACTIONS_QUERY, { filter: { type: ActionType.ALLOCATE } })
       .toPromise()
 
     expect(actions.data.actions).toHaveLength(1)
-    console.log('actions.data.actions', actions.data.actions)
   })
 
-
   test('Reject duplicate queued action from different source', async () => {
     const inputAction = queuedAllocateAction
     const expected = await actionInputToExpected(inputAction, 1)

packages/indexer-common/src/indexer-management/models/action.ts

@@ -155,6 +155,13 @@ export const defineActionModels = (sequelize: Sequelize): ActionModels => {
     {
       modelName: 'Action',
       sequelize,
+      indexes: [
+        {
+          fields: ['deploymentID', 'source'],
+          unique: true,
+          name: 'Actions_ckey_unique_deploymentID_source',
+        },
+      ],
       validate: {
         requiredActionParams() {
           switch (this.type) {